java-topology/defects/arrow/unit/ArrowTest.java

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);
    }
}