java-topology/defects/flink/unit/FlinkTest.java

import java.util.*;

/**
 * CWE-407 unit tests for Apache Flink deeper scan (flink-0006, flink-0007).
 *
 * flink-0006: MultipleInputNodeCreationProcessor.createMultipleInputNode()
 *   flink-table/flink-table-planner/.../exec/processor/MultipleInputNodeCreationProcessor.java
 *   In createMultipleInputNode(), the inner loop iterates over group.members (ArrayList) and
 *   calls group.members.contains(memberInput) for each input of each member:
 *     for (ExecNodeWrapper member : group.members) {          // O(M) outer
 *         for (int i = 0; i < member.inputs.size(); i++) {   // O(I) inner
 *             if (group.members.contains(memberInput)) { ... } // O(M) linear scan
 *   Total: O(M * I * M) = O(M²×I).
 *   Fix: build a HashSet<ExecNodeWrapper> membersSet = new HashSet<>(group.members) before the
 *   outer loop. Each .contains() becomes O(1), reducing total to O(M*I).
 *
 * flink-0007: MLPredictTypeStrategy.validateTableAndDescriptorArguments()
 *   flink-table/flink-table-common/.../inference/strategies/MLPredictTypeStrategy.java
 *   At planning time for ML_PREDICT(), the validator iterates over descriptor column names and
 *   calls tableFieldNames.contains() where tableFieldNames is List<String>:
 *     for (String descriptorColumnName : descriptorColumnNames) {  // O(D)
 *         if (!tableFieldNames.contains(descriptorColumnName)) {   // O(T) linear scan
 *   Total: O(D * T) where D=descriptor columns, T=table fields.
 *   Fix: convert tableFieldNames to a HashSet<String> before the loop → O(D) total.
 */
public class FlinkTest {

    // --- flink-0006 simulation ---

    /** Simulate defect: ArrayList.contains in nested loop → O(M² * I) */
    static int simulateCreateMultipleInputNode_defect(
            List<Integer> members, Map<Integer, List<Integer>> memberInputs) {
        int ops = 0;
        for (int member : members) {
            List<Integer> inputs = memberInputs.getOrDefault(member, Collections.emptyList());
            for (int input : inputs) {
                ops++;
                if (members.contains(input)) { // O(M) per call — defect
                    continue;
                }
                // would add to result list here
            }
        }
        return ops;
    }

    /** Simulate fix: HashSet.contains is O(1) */
    static int simulateCreateMultipleInputNode_fixed(
            List<Integer> members, Map<Integer, List<Integer>> memberInputs) {
        int ops = 0;
        Set<Integer> membersSet = new HashSet<>(members); // O(M) once
        for (int member : members) {
            List<Integer> inputs = memberInputs.getOrDefault(member, Collections.emptyList());
            for (int input : inputs) {
                ops++;
                if (membersSet.contains(input)) { // O(1) per call — fix
                    continue;
                }
            }
        }
        return ops;
    }

    static void testFlink0006() throws Exception {
        // M = members in group, I = inputs per member
        int M = 300;
        int I = 10;
        List<Integer> members = new ArrayList<>();
        for (int i = 0; i < M; i++) members.add(i);

        // Each member has I inputs: half internal (in members), half external
        Map<Integer, List<Integer>> memberInputs = new HashMap<>();
        for (int m = 0; m < M; m++) {
            List<Integer> inputs = new ArrayList<>();
            for (int i = 0; i < I; i++) {
                // alternating internal/external inputs
                inputs.add(i % 2 == 0 ? m : M + i);
            }
            memberInputs.put(m, inputs);
        }

        // Warm up
        simulateCreateMultipleInputNode_defect(members, memberInputs);
        simulateCreateMultipleInputNode_fixed(members, memberInputs);

        int RUNS = 200;

        long t0 = System.nanoTime();
        for (int r = 0; r < RUNS; r++) {
            simulateCreateMultipleInputNode_defect(members, memberInputs);
        }
        long defectNs = System.nanoTime() - t0;

        long t1 = System.nanoTime();
        for (int r = 0; r < RUNS; r++) {
            simulateCreateMultipleInputNode_fixed(members, memberInputs);
        }
        long fixedNs = System.nanoTime() - t1;

        double ratio = (double) defectNs / Math.max(fixedNs, 1);
        System.out.printf("flink-0006 [M=%d I=%d RUNS=%d]: defect=%.1fms fixed=%.1fms ratio=%.1fx%n",
                M, I, RUNS, defectNs / 1e6, fixedNs / 1e6, ratio);

        if (ratio < 1.5) {
            throw new AssertionError("Expected defect to be slower; ratio=" + ratio);
        }
        System.out.println("flink-0006 PASS");
    }

    // --- flink-0007 simulation ---

    /** Simulate defect: List<String>.contains in loop → O(D*T) */
    static boolean validateDescriptors_defect(List<String> tableFieldNames, List<String> descriptorColumnNames) {
        for (String col : descriptorColumnNames) {    // O(D)
            if (!tableFieldNames.contains(col)) {     // O(T) — defect
                return false;
            }
        }
        return true;
    }

    /** Simulate fix: HashSet<String>.contains is O(1) */
    static boolean validateDescriptors_fixed(List<String> tableFieldNames, List<String> descriptorColumnNames) {
        Set<String> tableFieldSet = new HashSet<>(tableFieldNames); // O(T) once
        for (String col : descriptorColumnNames) {    // O(D)
            if (!tableFieldSet.contains(col)) {       // O(1) — fix
                return false;
            }
        }
        return true;
    }

    static void testFlink0007() throws Exception {
        int T = 500; // table columns
        int D = 200; // descriptor columns (subset of table columns)

        List<String> tableFields = new ArrayList<>();
        for (int i = 0; i < T; i++) tableFields.add("col_" + i);

        List<String> descriptorCols = new ArrayList<>();
        for (int i = 0; i < D; i++) descriptorCols.add("col_" + (i * 2 % T));

        // Warm up
        validateDescriptors_defect(tableFields, descriptorCols);
        validateDescriptors_fixed(tableFields, descriptorCols);

        int RUNS = 2000;

        long t0 = System.nanoTime();
        for (int r = 0; r < RUNS; r++) {
            validateDescriptors_defect(tableFields, descriptorCols);
        }
        long defectNs = System.nanoTime() - t0;

        long t1 = System.nanoTime();
        for (int r = 0; r < RUNS; r++) {
            validateDescriptors_fixed(tableFields, descriptorCols);
        }
        long fixedNs = System.nanoTime() - t1;

        double ratio = (double) defectNs / Math.max(fixedNs, 1);
        System.out.printf("flink-0007 [T=%d D=%d RUNS=%d]: defect=%.1fms fixed=%.1fms ratio=%.1fx%n",
                T, D, RUNS, defectNs / 1e6, fixedNs / 1e6, ratio);

        if (ratio < 1.5) {
            throw new AssertionError("Expected defect to be slower; ratio=" + ratio);
        }
        System.out.println("flink-0007 PASS");
    }

    public static void main(String[] args) throws Exception {
        testFlink0006();
        testFlink0007();
        System.out.println("ALL PASS");
    }
}