package unit;

import java.util.*;

/**
 * flink-0004: DynamicSinkUtils UPDATE column resolution O(C×U) vs O(C+U)
 *
 * Simulates getUpdatedColumns() and projectColumnsForUpdate() from:
 *   flink-table/flink-table-planner/.../connectors/DynamicSinkUtils.java
 *
 * Standalone — no JUnit, no Flink deps.
 */
public class DynamicSinkUtilsAlgorithm {

    // -----------------------------------------------------------------------
    // Slow: List.contains inside loop (as in production code)
    // -----------------------------------------------------------------------

    static class SlowResult {
        final long ops;
        final List<String> updatedColumns;
        final List<Integer> columnIndexes;

        SlowResult(long ops, List<String> updatedColumns, List<Integer> columnIndexes) {
            this.ops = ops;
            this.updatedColumns = updatedColumns;
            this.columnIndexes = columnIndexes;
        }
    }

    static SlowResult getUpdatedColumnsSlow(List<String> allColumns, List<String> updatedColumnNames) {
        long ops = 0;
        List<String> result = new ArrayList<>();
        for (String column : allColumns) {            // O(C) loop
            for (String upd : updatedColumnNames) {   // O(U) scan simulating List.contains
                ops++;
                if (upd.equals(column)) {
                    result.add(column);
                    break;
                }
            }
        }
        return new SlowResult(ops, result, Collections.emptyList());
    }

    static SlowResult projectColumnsForUpdateSlow(
            List<Integer> updatedIndexes, List<String> allColumnNames, List<String> updatedColumnNames) {
        long ops = 0;
        List<Integer> result = new ArrayList<>();
        for (int index : updatedIndexes) {
            String colName = allColumnNames.get(index);
            // contains scan: O(U)
            boolean found = false;
            int foundIdx = -1;
            for (int i = 0; i < updatedColumnNames.size(); i++) {
                ops++;
                if (updatedColumnNames.get(i).equals(colName)) {
                    found = true;
                    // indexOf scan: another O(U) — simulate both as done in production
                }
            }
            // second pass simulating indexOf (production does both contains + indexOf)
            for (int i = 0; i < updatedColumnNames.size(); i++) {
                ops++;
                if (updatedColumnNames.get(i).equals(colName)) {
                    foundIdx = i;
                    break;
                }
            }
            if (found) {
                result.add(foundIdx);
            }
        }
        return new SlowResult(ops, Collections.emptyList(), result);
    }

    // -----------------------------------------------------------------------
    // Fast: HashSet/HashMap for O(1) lookup
    // -----------------------------------------------------------------------

    static class FastResult {
        final long ops;
        final List<String> updatedColumns;
        final List<Integer> columnIndexes;

        FastResult(long ops, List<String> updatedColumns, List<Integer> columnIndexes) {
            this.ops = ops;
            this.updatedColumns = updatedColumns;
            this.columnIndexes = columnIndexes;
        }
    }

    static FastResult getUpdatedColumnsFast(List<String> allColumns, List<String> updatedColumnNames) {
        long ops = 0;
        // Build set: O(U)
        Set<String> updatedSet = new HashSet<>();
        for (String name : updatedColumnNames) {
            ops++;
            updatedSet.add(name);
        }
        List<String> result = new ArrayList<>();
        for (String column : allColumns) {  // O(C) loop
            ops++;
            if (updatedSet.contains(column)) {  // O(1)
                result.add(column);
            }
        }
        return new FastResult(ops, result, Collections.emptyList());
    }

    static FastResult projectColumnsForUpdateFast(
            List<Integer> updatedIndexes, List<String> allColumnNames, List<String> updatedColumnNames) {
        long ops = 0;
        // Build index map: O(U)
        Map<String, Integer> updatedMap = new HashMap<>();
        for (int i = 0; i < updatedColumnNames.size(); i++) {
            ops++;
            updatedMap.put(updatedColumnNames.get(i), i);
        }
        List<Integer> result = new ArrayList<>();
        for (int index : updatedIndexes) {  // O(I)
            ops++;
            String colName = allColumnNames.get(index);
            Integer i = updatedMap.get(colName);  // O(1) replaces contains + indexOf
            if (i != null) {
                result.add(i);
            }
        }
        return new FastResult(ops, Collections.emptyList(), result);
    }

    // -----------------------------------------------------------------------
    // Test harness
    // -----------------------------------------------------------------------

    public static void main(String[] args) {
        int passed = 0;
        int total = 0;

        // Test parameters
        int C = 300;   // total columns
        int U = 30;    // updated columns
        int I = 150;   // updated indexes (subset of C)

        // Build column lists
        List<String> allColumns = new ArrayList<>();
        for (int i = 0; i < C; i++) {
            allColumns.add("col_" + i);
        }

        // Updated column names: every 10th column
        List<String> updatedColumnNames = new ArrayList<>();
        for (int i = 0; i < U; i++) {
            updatedColumnNames.add("col_" + (i * 10));
        }

        // Updated indexes: first I columns
        List<Integer> updatedIndexes = new ArrayList<>();
        for (int i = 0; i < I; i++) {
            updatedIndexes.add(i);
        }

        // ---- Test 1: getUpdatedColumns correctness ----
        total++;
        SlowResult slowGetCols = getUpdatedColumnsSlow(allColumns, updatedColumnNames);
        FastResult fastGetCols = getUpdatedColumnsFast(allColumns, updatedColumnNames);

        boolean colsMatch = slowGetCols.updatedColumns.equals(fastGetCols.updatedColumns);
        if (colsMatch) {
            System.out.println("PASS test1: getUpdatedColumns produces same results (" +
                slowGetCols.updatedColumns.size() + " columns found)");
            passed++;
        } else {
            System.out.println("FAIL test1: getUpdatedColumns mismatch: slow=" +
                slowGetCols.updatedColumns.size() + " fast=" + fastGetCols.updatedColumns.size());
        }

        // ---- Test 2: getUpdatedColumns ops ratio ----
        total++;
        long slowOps1 = slowGetCols.ops;
        long fastOps1 = fastGetCols.ops;
        double ratio1 = (double) slowOps1 / fastOps1;
        if (ratio1 >= 10.0) {
            System.out.printf("PASS test2: getUpdatedColumns slow=%d ops, fast=%d ops, ratio=%.1fx%n",
                slowOps1, fastOps1, ratio1);
            passed++;
        } else {
            System.out.printf("FAIL test2: getUpdatedColumns ratio=%.1fx (need >=10x) slow=%d fast=%d%n",
                ratio1, slowOps1, fastOps1);
        }

        // ---- Test 3: projectColumnsForUpdate correctness ----
        total++;
        SlowResult slowProject = projectColumnsForUpdateSlow(updatedIndexes, allColumns, updatedColumnNames);
        FastResult fastProject = projectColumnsForUpdateFast(updatedIndexes, allColumns, updatedColumnNames);

        boolean projectMatch = slowProject.columnIndexes.equals(fastProject.columnIndexes);
        if (projectMatch) {
            System.out.println("PASS test3: projectColumnsForUpdate produces same results (" +
                slowProject.columnIndexes.size() + " entries)");
            passed++;
        } else {
            System.out.println("FAIL test3: projectColumnsForUpdate mismatch: slow=" +
                slowProject.columnIndexes.size() + " fast=" + fastProject.columnIndexes.size());
        }

        // ---- Test 4: projectColumnsForUpdate ops ratio ----
        total++;
        long slowOps2 = slowProject.ops;
        long fastOps2 = fastProject.ops;
        double ratio2 = (double) slowOps2 / fastOps2;
        if (ratio2 >= 10.0) {
            System.out.printf("PASS test4: projectColumnsForUpdate slow=%d ops, fast=%d ops, ratio=%.1fx%n",
                slowOps2, fastOps2, ratio2);
            passed++;
        } else {
            System.out.printf("FAIL test4: projectColumnsForUpdate ratio=%.1fx (need >=10x) slow=%d fast=%d%n",
                ratio2, slowOps2, fastOps2);
        }

        // ---- Test 5: empty updatedColumns edge case ----
        total++;
        SlowResult slowEmpty = getUpdatedColumnsSlow(allColumns, Collections.emptyList());
        FastResult fastEmpty = getUpdatedColumnsFast(allColumns, Collections.emptyList());
        if (slowEmpty.updatedColumns.isEmpty() && fastEmpty.updatedColumns.isEmpty()) {
            System.out.println("PASS test5: empty updatedColumnNames → empty result");
            passed++;
        } else {
            System.out.println("FAIL test5: empty updatedColumnNames edge case");
        }

        System.out.println("\n" + passed + "/" + total + " PASS");

        if (passed != total) {
            System.exit(1);
        }
    }
}