9211 csv import perf fix 3

[ttoomey]

Merging this PR

• use the squash-merge strategy for PRs targeting a release-X branch
  Here's the PR summary:

---

Description

Improve process_existing query performance by materializing in-scope warehouse rows into a shared temp table and replacing correlated Arel EXISTS subqueries with indexed INNER JOINs.

• process_existing orchestration: Replace three independent methods (mark_unchanged, mark_incoming_older, apply_updates) with a shared temp table pattern (with_delete_pending_rows) that evaluates existing_destination_data_scope once instead of three times. Each sub-pass drains resolved rows from the temp table so subsequent passes operate on a shrinking set.
• Unchanged / incoming-older detection: Replace Arel-built correlated EXISTS subqueries against the full warehouse table with raw SQL INNER JOINs against the small indexed temp table (unchanged_ids_sql, incoming_older_ids_sql). Results are materialized into an inner temp table to avoid re-running unindexed staging joins per batch.
• apply_updates: Paginate through the shared temp table via keyset pagination instead of re-querying existing_destination_data_scope with ActiveRecord in_batches.
• Guard clause hoisting: Move ExportID and custom_augmentation? early returns from individual methods into process_existing, avoiding temp table creation for skipped file types.
• Removed methods: mark_unchanged, mark_incoming_older, local_date_cast_arel, batch_clear_pending_deletion — replaced by with_delete_pending_rows, unchanged_ids_sql, incoming_older_ids_sql, resolve_matched_rows.
• Comment cleanup: Fix typos, update class-level and method-level comments to reflect the new architecture.

Type of Change

Refactor

Checklist before requesting review

• I have performed a self-review of my code
• I have run the code that is being changed under ideal conditions, and it doesn't fail
• I have updated the documentation (or not applicable)
• I have added spec tests (or not applicable)
• I have provided testing instructions in this PR or the related issue (or not applicable)

[ttoomey]

Helpful AI artifact

# CSV Importer Refactor: `process_existing` (commit `73ec1cd`)

Context: Where process_existing sits in the pipeline

By the time process_existing runs, two earlier passes have already executed:

Pass	Method	Effect
0	mark_tree_as_dead	Sets pending_date_deleted on all in-scope warehouse rows ("guilty until proven innocent")
1	add_new_data	Upserts staging rows whose hud_key has no match in the warehouse (truly new records)

State on entry to process_existing: Every in-scope warehouse row still has pending_date_deleted set. The rows that were brand new have been inserted. What remains are warehouse rows that do have a matching key in the staging data — process_existing must decide what to do with each one.

State after process_existing: Every warehouse row that should be kept has had its pending_date_deleted cleared. Updated rows have been overwritten from staging. The only rows with pending_date_deleted still set are ones the CSV no longer contains — those get soft-deleted by the subsequent remove_pending_deletes pass.

There are three possible dispositions for each matched row:

Disposition	Condition	Action
Unchanged	source_hash in staging matches warehouse	Clear pending_date_deleted — no data update needed
Incoming older	Staging DateUpdated < warehouse DateUpdated (by local-tz date)	Clear pending_date_deleted — keep the newer warehouse version
Genuine update	Everything else (hash mismatch, incoming same or newer)	Overwrite warehouse from staging, clear pending_date_deleted, mark dirty flags

---

1) The original flow (before the commit)

process_existing called three independent methods sequentially, each of which independently queried existing_destination_data_scope(klass) — the ActiveRecord scope representing in-scope, delete-pending warehouse rows.

Methods

mark_unchanged(klass, file_name)

• Built an Arel EXISTS subquery: correlated the staging table with the warehouse table on hud_key AND source_hash equality.
• Used existing_destination_data_scope(klass).where(exists) to find matching warehouse rows.
• Called batch_clear_pending_deletion which materialized those IDs into a temp table and batch-updated pending_date_deleted = NULL.

mark_incoming_older(klass, file_name)

• Same pattern, but the EXISTS subquery compared DateUpdated (with timezone-aware date casting via the local_date_cast_arel helper) instead of source_hash.
• Again called batch_clear_pending_deletion on the matched scope.

apply_updates(klass, file_name)

• Iterated existing_destination_data_scope(klass).in_batches(of: SELECT_BATCH_SIZE).
• For each batch, plucked the hud_key values, looked up matching staging rows, transformed them via prepare_destination_for_update, and upserted.
• Also handled ExportID early-return and custom augmentation dispatch internally.

Helper methods

batch_clear_pending_deletion(klass, matched_scope, file_name)

• Materialized matched_scope IDs into a temp table.
• Batch-updated pending_date_deleted = NULL via keyset pagination on the temp table.

local_date_cast_arel(conn, arel_column)

• Built an Arel node for CAST(col AT TIME ZONE 'UTC' AT TIME ZONE '<local>' AS DATE).

flowchart TD
    subgraph PE["process_existing (per file)"]
        direction TB
        A["mark_unchanged"]
        B["mark_incoming_older"]
        C["apply_updates"]
        A --> B --> C
    end

    subgraph MU["mark_unchanged"]
        direction TB
        A1["Evaluate existing_destination_data_scope<br/>(query #1 against warehouse)"]
        A2["Build Arel EXISTS subquery:<br/>staging.hud_key = wh.hud_key<br/>AND staging.source_hash = wh.source_hash"]
        A3["batch_clear_pending_deletion<br/>↳ materialize IDs into temp table<br/>↳ batch UPDATE<br/>  pending_date_deleted = NULL<br/>↳ drop temp table"]
        A1 --> A2 --> A3
    end

    subgraph MIO["mark_incoming_older"]
        direction TB
        B1["Evaluate existing_destination_data_scope<br/>(query #2 against warehouse)"]
        B2["Build Arel EXISTS subquery:<br/>staging.hud_key = wh.hud_key<br/>AND staging.DateUpdated<br/>  < wh.DateUpdated"]
        B3["batch_clear_pending_deletion<br/>↳ materialize IDs into temp table<br/>↳ batch UPDATE<br/>  pending_date_deleted = NULL<br/>↳ drop temp table"]
        B1 --> B2 --> B3
    end

    subgraph AU["apply_updates"]
        direction TB
        C1["Evaluate existing_destination_data_scope<br/>(query #3 against warehouse)"]
        C2["in_batches → pluck hud_keys<br/>↳ find staging rows<br/>↳ prepare_destination_for_update<br/>↳ upsert"]
        C1 --> C2
    end

    %% Connect summary to details to align columns
    A -.-> A1
    B -.-> B1
    C -.-> C1

    %% Force subgraphs to stack vertically 
    A3 ---> B1
    B3 ---> C1

Loading

Key characteristics

1. existing_destination_data_scope is evaluated 3 times. Each call produces a query that scans/filters the warehouse table by data_source_id, project_ids, date_range, and pending_date_deleted.

2. Correlated EXISTS subqueries. mark_unchanged and mark_incoming_older use WHERE EXISTS (SELECT 1 FROM staging WHERE staging.hud_key = wh.hud_key AND ...). These are semi-joins — for each warehouse row, Postgres must probe the staging table. On large tables, the planner may choose a nested-loop strategy that's expensive.

3. Each temp table is independent. batch_clear_pending_deletion creates a one-shot temp table, uses it, and drops it. There's no sharing between passes — each pass starts from scratch.

4. No pruning between passes. After mark_unchanged clears pending_date_deleted on some rows, those rows may naturally fall out of existing_destination_data_scope for the next query (if the scope filters on pending_date_deleted IS NOT NULL). But the scope itself must still be re-evaluated — Postgres must re-scan to discover which rows remain.

---

2) How the original flow served its purpose

Despite the performance concerns, the original flow was correct:

• Ordering guarantee: Unchanged runs first, so hash-matched rows never reach the date comparison or update logic. Incoming-older runs second, so only hash-mismatched rows get date-compared. Apply_updates runs last and only sees rows that failed both prior checks.

• Independence: Each method was self-contained with its own scope evaluation and temp table lifecycle. This made each method easy to reason about in isolation.

• Correctness of the semi-joins: The Arel EXISTS subqueries correctly expressed "there exists a staging row with matching hud_key and matching source_hash" (or date condition). NULL source_hash on the warehouse side correctly failed the equality check, routing those rows through to apply_updates.

The trade-off was performance: three full scope evaluations and three rounds of correlated subquery execution, with no shared intermediate state between passes.

---

3) The new flow (after the commit)

The refactor replaces the three independent methods with a shared temp table pattern and progressive draining.

New orchestration in process_existing

with_delete_pending_rows(conn, klass, file_name) do |pending_rows|
  # Sub-pass 1: unchanged
  resolve_matched_rows(conn, klass, ..., unchanged_ids_sql(..., pending_rows), pending_rows, label: 'unchanged')

  # Sub-pass 2: incoming_older (skipped if most recent export)
  resolve_matched_rows(conn, klass, ..., incoming_older_ids_sql(..., pending_rows), pending_rows, label: 'incoming_older')

  # Sub-pass 3: apply_updates (uses the now-drained pending_rows directly)
  apply_updates(klass, file_name, pending_rows, conn: conn)
end

Early returns for ExportID and custom_augmentation? are hoisted into process_existing itself, before the temp table is created.

New methods

with_delete_pending_rows(conn, klass, file_name) — single materialization

• Evaluates existing_destination_data_scope(klass) once.
• Materializes the result into a temp table with four columns: id, hud_key, source_hash, DateUpdated.
• Creates indexes on id and hud_key.
• Yields the quoted temp table name. Drops it in ensure.

unchanged_ids_sql(klass, conn, pending_rows) — raw SQL, no Arel

• Returns a SQL string: SELECT DISTINCT t.id FROM <pending_rows> t INNER JOIN <staging> s ON s.hud_key = t.hud_key AND s.source_hash = t.source_hash WHERE ...
• This is a regular INNER JOIN on the small, indexed temp table — not a correlated EXISTS against the full warehouse table.

incoming_older_ids_sql(klass, conn, pending_rows) — raw SQL, no Arel

• Same pattern but with the timezone-aware date comparison inlined as raw SQL (replacing the local_date_cast_arel Arel helper).
• Joins temp table to staging on hud_key, filters where staging date < temp table date.

resolve_matched_rows(conn, klass, ..., matched_sql, pending_rows, label:) — replaces batch_clear_pending_deletion

• Materializes the matched IDs (from the SQL string) into an inner temp table.
• Batch-updates the warehouse to clear pending_date_deleted (same keyset pagination pattern as before).
• New step: Deletes resolved rows from the shared pending_rows temp table: DELETE FROM <pending_rows> WHERE id IN (SELECT id FROM <inner>).
• Drops the inner temp table in ensure.

apply_updates(klass, file_name, pending_rows, conn:) — now paginating from the temp table

• Instead of re-querying existing_destination_data_scope(klass).in_batches(...), paginates through the shared temp table: SELECT id, hud_key FROM <pending_rows> WHERE id > last_id ORDER BY id LIMIT batch_size.
• For each batch of hud_key values, looks up staging rows, transforms, and upserts (same logic as before).

flowchart TD
    subgraph PE["process_existing (per file)"]
        direction TB

        M["with_delete_pending_rows\nEvaluate existing_destination_data_scope ONCE\n→ CREATE TEMP TABLE with id, hud_key, source_hash, DateUpdated\n→ CREATE INDEX on id, hud_key"]

        subgraph PASS1["Sub-pass 1: unchanged"]
            U1["unchanged_ids_sql\nJOIN temp ↔ staging ON hud_key + source_hash"]
            U2["resolve_matched_rows\n→ materialize matched IDs into inner temp\n→ batch UPDATE wh SET pending_date_deleted = NULL\n→ DELETE matched rows from shared temp"]
            U1 --> U2
        end

        subgraph PASS2["Sub-pass 2: incoming_older"]
            IO1["incoming_older_ids_sql\nJOIN temp ↔ staging ON hud_key\nWHERE staging.date < temp.date"]
            IO2["resolve_matched_rows\n→ materialize matched IDs into inner temp\n→ batch UPDATE wh SET pending_date_deleted = NULL\n→ DELETE matched rows from shared temp"]
            IO1 --> IO2
        end

        subgraph PASS3["Sub-pass 3: apply_updates"]
            AU1["Paginate remaining rows in shared temp\nSELECT id, hud_key WHERE id > last_id"]
            AU2["For each batch of hud_keys:\nfind staging rows → prepare → upsert"]
            AU1 --> AU2
        end

        M --> PASS1 --> PASS2 --> PASS3

        DROP["DROP shared temp table (ensure)"]
        PASS3 --> DROP
    end

Loading

What changed (summary)

Aspect	Before	After
Scope evaluation	existing_destination_data_scope queried 3 times (once per method)	Queried once, materialized into a shared temp table
Join strategy	Correlated EXISTS subqueries (Arel semi-joins against full warehouse table)	INNER JOIN against small, indexed temp table
Inter-pass coordination	None — each pass independently re-queries the warehouse	Progressive draining — resolved rows are deleted from the shared temp table, so each subsequent pass operates on a shrinking set
apply_updates iteration	existing_destination_data_scope.in_batches (ActiveRecord batching against warehouse)	Keyset pagination on the shared temp table (WHERE id > last_id)
Guard clauses	Each method individually checked ExportID / custom_augmentation?	Hoisted into process_existing before temp table creation
Arel vs raw SQL	mark_unchanged and mark_incoming_older used Arel for EXISTS + date casting	unchanged_ids_sql and incoming_older_ids_sql use hand-written SQL strings
Removed methods	—	mark_unchanged, mark_incoming_older, local_date_cast_arel, batch_clear_pending_deletion
Added methods	—	with_delete_pending_rows, unchanged_ids_sql, incoming_older_ids_sql, resolve_matched_rows

Where the performance win comes from

1. Single scope evaluation. The existing_destination_data_scope query — which filters a potentially very large warehouse table by data_source_id, project_ids, date_range, and pending_date_deleted — runs once instead of three times.

2. Temp-table JOINs are cheaper. The unchanged and incoming-older checks now join the staging table against a small, indexed temp table rather than executing correlated EXISTS subqueries against the full warehouse table. Postgres can use hash joins or merge joins on the temp table.

3. Progressive draining reduces work. Each sub-pass deletes its resolved rows from the shared temp table. If 90% of rows are unchanged (common in steady-state imports), the incoming-older pass only evaluates the remaining 10%, and apply_updates only iterates the small residual.

4. Keyset pagination on temp table. apply_updates no longer uses ActiveRecord's in_batches (which can be unpredictable with large offset-based pagination). Instead it uses simple WHERE id > last_id ORDER BY id LIMIT N on the temp table — fast with an index.