duckdb/arrow: CWE-407 scan — 4 DuckDB defects, 2 Arrow defects

DuckDB (C++ query engine):
- duckdb-0001: Binder::AddCorrelatedColumn vector dedup O(C²) MEDIUM 200x
- duckdb-0002: HasCorrelatedExpressions vector scan O(N×M) MEDIUM 100x
- duckdb-0003: ComputeOverlappingBindings vector scan O(N×H) MEDIUM 219x
- duckdb-0004: Deliminator group-join binding check O(G×J) MEDIUM 125x

Apache Arrow (C++ analytics):
- arrow-0001: AsofJoin IsTimeOrKeyColumn vector scan O(F×K) MEDIUM 114x
- arrow-0002: Scanner AddFieldsNeededForFilter vector dedup O(F×C) MEDIUM 250x

All 6/6 unit tests PASS.

defects/arrow/patch/arrow-0001-asof-join-key-col-index-hashset.patch

@@ -0,0 +1,57 @@
+# UNDF:
+--- a/cpp/src/arrow/acero/asof_join_node.cc
++++ b/cpp/src/arrow/acero/asof_join_node.cc
+@@ -483,6 +483,8 @@ class InputState : public util::SerialSequencingQueue::Processor {
+   InputState(size_t index, TolType tolerance, bool must_hash, bool may_rehash,
+              KeyHasher* key_hasher, AsofJoinNode* node, BackpressureHandler handler,
+              const std::shared_ptr<arrow::Schema>& schema,
+              const col_index_t time_col_index,
+              const std::vector<col_index_t>& key_col_index)
+       : sequencer_(util::SerialSequencingQueue::Make(this)),
+         queue_(std::move(handler)),
+         schema_(schema),
+         time_col_index_(time_col_index),
+         key_col_index_(key_col_index),
++        key_col_index_set_(key_col_index.begin(), key_col_index.end()),
+         time_type_id_(schema_->fields()[time_col_index_]->type()->id()),
+         key_type_id_(key_col_index.size()),
+         key_hasher_(key_hasher),
+@@ -537,7 +539,9 @@ class InputState : public util::SerialSequencingQueue::Processor {
+
+   bool IsTimeOrKeyColumn(col_index_t i) const {
+     DCHECK_LT(i, schema_->num_fields());
+-    return (i == time_col_index_) || std_has(key_col_index_, i);
++    // key_col_index_set_ gives O(1) lookup; std_has(key_col_index_, i) was O(K)
++    // so InitSrcToDstMapping's loop over F fields was O(F×K).  Now O(F).
++    return (i == time_col_index_) || (key_col_index_set_.count(i) > 0);
+   }
+
+@@ -786,6 +790,8 @@ class InputState : public util::SerialSequencingQueue::Processor {
+   std::vector<col_index_t> key_col_index_;
++  // Shadow set of key_col_index_ for O(1) membership test (IsTimeOrKeyColumn)
++  std::unordered_set<col_index_t> key_col_index_set_;
+   // Type id of the time column
+   Type::type time_type_id_;
+
+--- a/cpp/src/arrow/acero/asof_join_node.cc
++++ b/cpp/src/arrow/acero/asof_join_node.cc
+@@ -1287,9 +1293,14 @@ static Result<std::shared_ptr<Schema>> MakeOutputSchema(
+       for (int i = 0; i < input_schema[j]->num_fields(); ++i) {
+         const auto field = input_schema[j]->field(i);
+         bool as_output;  // true if the field appears as an output
+         if (i == on_field_ix) {
+           ARROW_RETURN_NOT_OK(is_valid_on_field(field));
+           as_output = (j == 0);
+-        } else if (std_has(by_field_ix, i)) {
++        } else if (by_field_ix_set.count(i) > 0) {
+           ARROW_RETURN_NOT_OK(is_valid_by_field(field));
+           as_output = (j == 0);
+         } else {
+@@ -1255,6 +1258,10 @@ static Result<std::shared_ptr<Schema>> MakeOutputSchema(
+     for (size_t j = 0; j < input_schema.size(); ++j) {
+       const auto& on_field_ix = indices_of_on_key[j];
+       const auto& by_field_ix = indices_of_by_key[j];
++      // Build O(1) set for by-key membership; std_has(by_field_ix, i) was O(K) per field
++      // so the inner loop over F fields was O(F×K).  Now O(F+K).
++      const std::unordered_set<col_index_t> by_field_ix_set(by_field_ix.begin(),
++                                                             by_field_ix.end());

defects/arrow/patch/arrow-0002-scanner-addfields-dedup.patch

@@ -0,0 +1,29 @@
+# UNDF: (leave blank)
+# Apache Arrow CWE-407: ScanV2Options::AddFieldsNeededForFilter O(F×C)
+# File: cpp/src/arrow/dataset/scanner.cc
+# Severity: MEDIUM — dataset scanner path for filter field dedup
+# Ratio: ~250x at F=C=500 (wide schema with complex filter expression)
+#
+# AddFieldsNeededForFilter iterates over fields_referenced (F), and for each
+# calls std::find on options->columns vector (growing up to C) to dedup,
+# yielding O(F×C). With wide schemas and complex filters, both F and C grow.
+# Fix: build an unordered_set of existing columns for O(1) lookup.
+--- a/cpp/src/arrow/dataset/scanner.cc
++++ b/cpp/src/arrow/dataset/scanner.cc
+@@ -82,10 +82,13 @@
+ Status ScanV2Options::AddFieldsNeededForFilter(ScanV2Options* options) {
+   std::vector<FieldRef> fields_referenced = FieldsInExpression(options->filter);
++  // Build O(1) lookup set for existing columns; std::find was O(C) per field → O(F×C) total
++  std::unordered_set<FieldPath, FieldPath::Hash> existing_columns(
++      options->columns.begin(), options->columns.end());
+   for (const auto& field : fields_referenced) {
+     // Note: this will fail if the field reference is ambiguous or the field doesn't
+     // exist in the dataset schema
+     ARROW_ASSIGN_OR_RAISE(auto field_path, field.FindOne(*options->dataset->schema()));
+-    if (std::find(options->columns.begin(), options->columns.end(), field_path) ==
+-        options->columns.end()) {
++    if (existing_columns.find(field_path) == existing_columns.end()) {
+       options->columns.push_back(std::move(field_path));
++      existing_columns.insert(options->columns.back());
+     }
+   }

defects/arrow/unit/ArrowTest.java

@@ -0,0 +1,136 @@
+import java.util.*;
+
+/**
+ * CWE-407 simulation tests for Apache Arrow defects.
+ *
+ * arrow-0001: cpp/src/arrow/acero/asof_join_node.cc
+ *   IsTimeOrKeyColumn: std_has(key_col_index_, i) O(K) inside InitSrcToDstMapping
+ *   loop over F fields -> O(F×K). Fix: unordered_set for O(1) lookup.
+ *   Also MakeOutputSchema: std_has(by_field_ix, i) O(K) per field -> O(F×K).
+ *
+ * arrow-0002: cpp/src/arrow/dataset/scanner.cc
+ *   AddFieldsNeededForFilter: std::find in options->columns O(C) per referenced
+ *   field -> O(F×C). Fix: unordered_set for O(1) dedup.
+ */
+public class ArrowTest {
+
+    // =========================================================
+    // arrow-0001: AsofJoin IsTimeOrKeyColumn O(F×K) vs O(F)
+    // =========================================================
+
+    /** DEFECTIVE: linear scan of key columns for each field */
+    static long isTimeOrKeyDefective(int numFields, List<Integer> keyCols, int timeCol) {
+        long ops = 0;
+        for (int i = 0; i < numFields; i++) {
+            if (i == timeCol) continue;
+            // std_has: linear scan
+            for (int k : keyCols) {
+                ops++;
+                if (k == i) break;
+            }
+        }
+        return ops;
+    }
+
+    /** FIXED: HashSet for O(1) key membership */
+    static long isTimeOrKeyFixed(int numFields, List<Integer> keyCols, int timeCol) {
+        Set<Integer> keySet = new HashSet<>(keyCols);
+        long ops = keyCols.size(); // build set
+        for (int i = 0; i < numFields; i++) {
+            if (i == timeCol) continue;
+            ops++; // O(1) lookup
+            keySet.contains(i);
+        }
+        return ops;
+    }
+
+    static boolean testArrow0001() {
+        System.out.println("=== arrow-0001: AsofJoin IsTimeOrKeyColumn O(F*K) vs O(F+K) ===");
+        // Simulate wide schema with many key columns
+        int F = 500;
+        List<Integer> keyCols = new ArrayList<>();
+        for (int i = 0; i < 200; i++) keyCols.add(i + 300); // 200 key columns at high indices
+
+        long opsDefective = isTimeOrKeyDefective(F, keyCols, 0);
+        long opsFixed = isTimeOrKeyFixed(F, keyCols, 0);
+        double ratio = (double) opsDefective / opsFixed;
+        System.out.printf("  F=%d K=%d defective_ops=%,d fixed_ops=%,d ratio=%.1fx%n",
+                F, keyCols.size(), opsDefective, opsFixed, ratio);
+        boolean pass = ratio > 5.0;
+        System.out.println("  " + (pass ? "PASS" : "FAIL"));
+        return pass;
+    }
+
+    // =========================================================
+    // arrow-0002: Scanner AddFieldsNeededForFilter O(F×C) vs O(F+C)
+    // =========================================================
+
+    static class FieldPath {
+        final int index;
+        FieldPath(int i) { this.index = i; }
+        @Override public boolean equals(Object o) {
+            return o instanceof FieldPath && ((FieldPath) o).index == this.index;
+        }
+        @Override public int hashCode() { return Integer.hashCode(index); }
+    }
+
+    /** DEFECTIVE: std::find in columns vector per referenced field */
+    static long addFieldsDefective(List<FieldPath> existingColumns, List<FieldPath> fieldsReferenced) {
+        long ops = 0;
+        List<FieldPath> columns = new ArrayList<>(existingColumns);
+        for (FieldPath fp : fieldsReferenced) {
+            // linear scan for dedup
+            boolean found = false;
+            for (FieldPath c : columns) {
+                ops++;
+                if (c.equals(fp)) { found = true; break; }
+            }
+            if (!found) columns.add(fp);
+        }
+        return ops;
+    }
+
+    /** FIXED: unordered_set for O(1) dedup */
+    static long addFieldsFixed(List<FieldPath> existingColumns, List<FieldPath> fieldsReferenced) {
+        Set<FieldPath> existing = new HashSet<>(existingColumns);
+        long ops = existingColumns.size(); // build set
+        List<FieldPath> columns = new ArrayList<>(existingColumns);
+        for (FieldPath fp : fieldsReferenced) {
+            ops++; // O(1) lookup
+            if (existing.add(fp)) {
+                columns.add(fp);
+            }
+        }
+        return ops;
+    }
+
+    static boolean testArrow0002() {
+        System.out.println("=== arrow-0002: Scanner AddFieldsNeededForFilter O(F*C) vs O(F+C) ===");
+        int N = 500;
+        List<FieldPath> existingColumns = new ArrayList<>();
+        for (int i = 0; i < N; i++) existingColumns.add(new FieldPath(i));
+        List<FieldPath> fieldsReferenced = new ArrayList<>();
+        for (int i = N / 2; i < N + N / 2; i++) fieldsReferenced.add(new FieldPath(i)); // half overlap
+
+        long opsDefective = addFieldsDefective(existingColumns, fieldsReferenced);
+        long opsFixed = addFieldsFixed(existingColumns, fieldsReferenced);
+        double ratio = (double) opsDefective / opsFixed;
+        System.out.printf("  F=%d C=%d defective_ops=%,d fixed_ops=%,d ratio=%.1fx%n",
+                N, N, opsDefective, opsFixed, ratio);
+        boolean pass = ratio > 10.0;
+        System.out.println("  " + (pass ? "PASS" : "FAIL"));
+        return pass;
+    }
+
+    // =========================================================
+
+    public static void main(String[] args) {
+        int pass = 0, fail = 0;
+
+        if (testArrow0001()) pass++; else fail++;
+        if (testArrow0002()) pass++; else fail++;
+
+        System.out.printf("%nArrow CWE-407: %d/%d PASS%n", pass, pass + fail);
+        if (fail > 0) System.exit(1);
+    }
+}