Replace EXISTS/ROWS FROM with = ANY() for target row filtering

jhf · claude · jhf · commit 6facdb0053dc · 2026-02-22T12:34:54.000+01:00
The parameterized target-read WHERE clause used EXISTS with unnest/ROWS FROM
and IS NOT DISTINCT FROM, which PostgreSQL planned as Nested Loop with O(N*M)
comparisons. Replace with = ANY() per column, enabling Index Scan/Bitmap Index
Scan. EXPLAIN ANALYZE confirms 347x speedup (1.8ms vs 625ms) for 5K legal_unit.

Also simplifies extract_filter_values from tuple-pair deduplication to
independent per-column distinct values using BTreeSet. The cross-product may
over-fetch a few target rows, but the sweep handles extras at zero cost.

Co-Authored-By: Claude Opus 4.6 &lt;noreply@anthropic.com&gt;
diff --git a/native/src/reader.rs b/native/src/reader.rs
@@ -523,8 +523,10 @@ fn read_source_ordinals(
 }
 
 /// Try to build a parameterized WHERE clause for the target read query.
-/// Single-column key sets: WHERE t."col" = ANY($N::text::type[])
-/// Multi-column key sets: WHERE (t."c1", t."c2") IN (SELECT c1, c2 FROM unnest($N1::text::type1[], $N2::text::type2[]) AS u(c1, c2))
+/// Uses = ANY() per column, enabling Bitmap Index Scan instead of Nested Loop.
+/// Single-column: WHERE t."col" = ANY($N::text::type[])
+/// Multi-column:  WHERE (t."c1" = ANY($1::text::type1[]) AND t."c2" = ANY($2::text::type2[]))
+/// Nullable:      WHERE (t."col" IS NOT NULL AND t."c1" = ANY($1) AND t."col" = ANY($2)) OR ...
 /// Returns None only if any column type is unknown.
 /// Returns Some(("", [])) for full-scan modes (no WHERE clause needed).
 fn try_build_parameterized_filter(
@@ -583,16 +585,15 @@ fn try_build_parameterized_filter(
         return Some((String::new(), vec![])); // No filter needed
     }
 
-    // Build parameterized WHERE clause — handles both single and multi-column key sets.
+    // Build parameterized WHERE clause using = ANY() per column.
     // For key sets with nullable columns, uses partition-by-NULL optimization:
-    // generates separate EXISTS per nullable column with IS NOT NULL + = joins,
-    // enabling Hash/Merge joins instead of Nested Loop with IS NOT DISTINCT FROM.
+    // generates separate (IS NOT NULL AND col = ANY(...)) per nullable column.
     let mut params = Vec::new();
     let mut where_parts = Vec::new();
     let mut param_index = 1usize;
     let mut key_set_id = 0usize;
 
-    for (_original_idx, (cols, is_identity)) in filter_key_sets.iter().enumerate() {
+    for (cols, is_identity) in &filter_key_sets {
         // Verify all columns have known types
         for col in cols {
             if !target_col_types.contains_key(col) {
@@ -606,9 +607,9 @@ fn try_build_parameterized_filter(
             .collect();
 
         if nullable_cols.is_empty() {
-            // All columns are NOT NULL — use existing approach
-            if cols.len() == 1 {
-                let col = &cols[0];
+            // All columns NOT NULL — = ANY() per column
+            let mut conditions = Vec::new();
+            for col in cols {
                 let pg_type = target_col_types.get(col).unwrap();
                 params.push(FilterParam {
                     col_name: col.clone(),
@@ -618,132 +619,55 @@ fn try_build_parameterized_filter(
                     key_set_id,
                     partition_not_null_cols: vec![],
                 });
-                where_parts.push(format!(
-                    "EXISTS (SELECT 1 FROM unnest(${idx}::text::{typ}[]) AS _u(v) WHERE t.{col} IS NOT DISTINCT FROM _u.v)",
+                conditions.push(format!(
+                    "t.{col} = ANY(${idx}::text::{typ}[])",
                     col = qi(col),
                     idx = param_index,
                     typ = pg_type,
                 ));
                 param_index += 1;
+            }
+            if conditions.len() == 1 {
+                where_parts.push(conditions.into_iter().next().unwrap());
             } else {
-                // Multi-column, all NOT NULL: EXISTS with IS NOT DISTINCT FROM
-                let mut unnest_calls = Vec::new();
-                let mut u_col_names = Vec::new();
-                let mut conditions = Vec::new();
-
-                for (ci, col) in cols.iter().enumerate() {
-                    let pg_type = target_col_types.get(col).unwrap();
-                    params.push(FilterParam {
-                        col_name: col.clone(),
-                        pg_type: pg_type.clone(),
-                        param_index,
-                        is_identity: *is_identity,
-                        key_set_id,
-                        partition_not_null_cols: vec![],
-                    });
-                    let u_alias = format!("_c{}", ci);
-                    unnest_calls.push(format!(
-                        "unnest(${idx}::text::{typ}[])",
-                        idx = param_index,
-                        typ = pg_type,
-                    ));
-                    u_col_names.push(u_alias.clone());
-                    conditions.push(format!(
-                        "t.{col} IS NOT DISTINCT FROM _u.{u_alias}",
-                        col = qi(col),
-                        u_alias = u_alias,
-                    ));
-                    param_index += 1;
-                }
-
-                where_parts.push(format!(
-                    "EXISTS (SELECT 1 FROM ROWS FROM({fns}) AS _u({cols}) WHERE {conds})",
-                    fns = unnest_calls.join(", "),
-                    cols = u_col_names.join(", "),
-                    conds = conditions.join(" AND "),
-                ));
+                where_parts.push(format!("({})", conditions.join(" AND ")));
             }
             key_set_id += 1;
         } else {
-            // Has nullable columns — partition-by-NULL optimization.
-            // For each nullable column, generate a separate EXISTS clause with:
-            // - IS NOT NULL filter on target column
-            // - = equality joins (Hash/Merge join eligible)
-            // - Source params filtered to rows where that column IS NOT NULL
-            // Mirrors PL/pgSQL partition-by-NULL pattern (src/2k_temporal_merge_plan.sql:1183-1254).
+            // Has nullable columns — partition-by-NULL with = ANY() per column.
             let notnull_cols: Vec<&String> = cols.iter()
                 .filter(|c| target_col_notnull.get(*c).copied().unwrap_or(true))
                 .collect();
 
             for nullable_col in &nullable_cols {
-                // Partition columns = NOT NULL cols + this nullable col
                 let partition_cols: Vec<&String> = notnull_cols.iter()
                     .copied()
                     .chain(std::iter::once(*nullable_col))
                     .collect();
 
-                if partition_cols.len() == 1 {
-                    // Single-column partition (nullable col only, no NOT NULL cols)
-                    let col = partition_cols[0];
-                    let pg_type = target_col_types.get(col).unwrap();
+                let mut conditions = Vec::new();
+                conditions.push(format!("t.{} IS NOT NULL", qi(nullable_col)));
+
+                for col in &partition_cols {
+                    let pg_type = target_col_types.get(*col).unwrap();
                     params.push(FilterParam {
-                        col_name: col.clone(),
+                        col_name: (*col).clone(),
                         pg_type: pg_type.clone(),
                         param_index,
                         is_identity: *is_identity,
                         key_set_id,
                         partition_not_null_cols: vec![(*nullable_col).clone()],
                     });
-                    where_parts.push(format!(
-                        "EXISTS (SELECT 1 FROM unnest(${idx}::text::{typ}[]) AS _u(v) \
-                         WHERE t.{col} IS NOT NULL AND t.{col} = _u.v)",
+                    conditions.push(format!(
+                        "t.{col} = ANY(${idx}::text::{typ}[])",
                         col = qi(col),
                         idx = param_index,
                         typ = pg_type,
                     ));
                     param_index += 1;
-                } else {
-                    // Multi-column partition: ROWS FROM with = equality joins
-                    let mut unnest_calls = Vec::new();
-                    let mut u_col_names = Vec::new();
-                    let mut conditions = Vec::new();
-
-                    // Add IS NOT NULL condition for the nullable column on target
-                    conditions.push(format!("t.{} IS NOT NULL", qi(nullable_col)));
-
-                    for (ci, col) in partition_cols.iter().enumerate() {
-                        let pg_type = target_col_types.get(*col).unwrap();
-                        params.push(FilterParam {
-                            col_name: (*col).clone(),
-                            pg_type: pg_type.clone(),
-                            param_index,
-                            is_identity: *is_identity,
-                            key_set_id,
-                            partition_not_null_cols: vec![(*nullable_col).clone()],
-                        });
-                        let u_alias = format!("_c{}", ci);
-                        unnest_calls.push(format!(
-                            "unnest(${idx}::text::{typ}[])",
-                            idx = param_index,
-                            typ = pg_type,
-                        ));
-                        u_col_names.push(u_alias.clone());
-                        // Use = for equality (all columns in partition are NOT NULL by construction)
-                        conditions.push(format!(
-                            "t.{col} = _u.{u_alias}",
-                            col = qi(col),
-                            u_alias = u_alias,
-                        ));
-                        param_index += 1;
-                    }
-
-                    where_parts.push(format!(
-                        "EXISTS (SELECT 1 FROM ROWS FROM({fns}) AS _u({cols}) WHERE {conds})",
-                        fns = unnest_calls.join(", "),
-                        cols = u_col_names.join(", "),
-                        conds = conditions.join(" AND "),
-                    ));
                 }
+
+                where_parts.push(format!("({})", conditions.join(" AND ")));
                 key_set_id += 1;
             }
         }
@@ -756,108 +680,54 @@ fn try_build_parameterized_filter(
 /// Extract distinct filter values from source rows for parameterized target read.
 /// Returns one PostgreSQL array literal string per FilterParam.
 ///
-/// For single-column key sets: extracts distinct values for that column.
-/// For multi-column key sets: extracts distinct tuples as parallel arrays,
-/// ensuring the arrays stay aligned (same index = same tuple).
+/// Each column collects its distinct values independently (no tuple pairing).
+/// This matches the = ANY() per-column WHERE clause: the cross-product may
+/// over-fetch a few target rows, but the sweep handles extras at zero cost.
 pub fn extract_filter_values(
     source_rows: &[SourceRow],
     filter_params: &[FilterParam],
 ) -> Vec<String> {
-    // Group params by key_set_id to handle multi-column correctly
-    let mut key_set_ids: Vec<usize> = filter_params.iter().map(|p| p.key_set_id).collect();
-    key_set_ids.sort_unstable();
-    key_set_ids.dedup();
-
-    // For each key_set, extract distinct tuples (Option<String> to support NULLs)
-    let mut key_set_values: std::collections::HashMap<usize, Vec<Vec<Option<String>>>> =
-        std::collections::HashMap::new();
-
-    for &ks_id in &key_set_ids {
-        let ks_params: Vec<&FilterParam> = filter_params
-            .iter()
-            .filter(|p| p.key_set_id == ks_id)
-            .collect();
-
-        // Get partition NOT NULL filter columns (same for all params in a key_set)
-        let partition_not_null = ks_params.first()
-            .map(|p| &p.partition_not_null_cols)
-            .cloned()
-            .unwrap_or_default();
-
-        let mut seen = std::collections::HashSet::new();
-        let n_cols = ks_params.len();
-        let mut columns: Vec<Vec<Option<String>>> = vec![Vec::new(); n_cols];
-
-        for row in source_rows {
-            // Partition-by-NULL filter: skip rows where required columns are NULL
-            if !partition_not_null.is_empty() {
-                let passes = partition_not_null.iter().all(|col| {
-                    let val = row.identity_keys.get(col)
-                        .or_else(|| row.lookup_keys.get(col));
-                    val.map_or(false, |v| !v.is_null())
-                });
-                if !passes {
-                    continue;
+    // Each FilterParam gets independent per-column distinct values.
+    filter_params
+        .iter()
+        .map(|param| {
+            let mut distinct = std::collections::BTreeSet::new();
+
+            for row in source_rows {
+                // Partition-by-NULL filter: skip rows where required columns are NULL
+                if !param.partition_not_null_cols.is_empty() {
+                    let passes = param.partition_not_null_cols.iter().all(|col| {
+                        let val = row.identity_keys.get(col)
+                            .or_else(|| row.lookup_keys.get(col));
+                        val.map_or(false, |v| !v.is_null())
+                    });
+                    if !passes {
+                        continue;
+                    }
                 }
-            }
-
-            let mut tuple: Vec<Option<String>> = Vec::with_capacity(n_cols);
 
-            for param in &ks_params {
-                // Check both maps: a column might be in identity_keys or lookup_keys
-                // depending on how PL/pgSQL wrapper classified it. For filter purposes,
-                // we just need the value from whichever map has it.
-                let val_opt = row
-                    .identity_keys
-                    .get(&param.col_name)
+                let val_opt = row.identity_keys.get(&param.col_name)
                     .or_else(|| row.lookup_keys.get(&param.col_name));
 
                 if let Some(val) = val_opt {
-                    if val.is_null() {
-                        tuple.push(None); // NULL — included for IS NOT DISTINCT FROM matching
-                    } else {
+                    if !val.is_null() {
                         let text_val = match val {
                             serde_json::Value::String(s) => s.clone(),
                             serde_json::Value::Number(n) => n.to_string(),
                             serde_json::Value::Bool(b) => b.to_string(),
                             _ => val.to_string(),
                         };
-                        tuple.push(Some(text_val));
+                        distinct.insert(text_val);
                     }
-                } else {
-                    tuple.push(None); // Missing column treated as NULL
                 }
             }
 
-            // Deduplicate by tuple (represent NULL as sentinel for dedup key)
-            let tuple_key: Vec<String> = tuple.iter().map(|v| match v {
-                Some(s) => s.clone(),
-                None => "\x00NULL\x00".to_string(),
-            }).collect();
-            if seen.insert(tuple_key) {
-                for (i, val) in tuple.into_iter().enumerate() {
-                    columns[i].push(val);
-                }
-            }
-        }
-
-        key_set_values.insert(ks_id, columns);
-    }
-
-    // Map back to per-FilterParam array literals (in order)
-    filter_params
-        .iter()
-        .map(|param| {
-            let columns = key_set_values.get(&param.key_set_id).unwrap();
-            let ks_params: Vec<&FilterParam> = filter_params
-                .iter()
-                .filter(|p| p.key_set_id == param.key_set_id)
+            // Convert to PostgreSQL array literal (no NULLs — = ANY() doesn't match NULLs)
+            let values: Vec<Option<String>> = distinct
+                .into_iter()
+                .map(Some)
                 .collect();
-            let col_idx = ks_params
-                .iter()
-                .position(|p| p.param_index == param.param_index)
-                .unwrap();
-            format_pg_array_literal(&columns[col_idx])
+            format_pg_array_literal(&values)
         })
         .collect()
 }
