package unit;

import java.util.*;

/**
 * CWE-407 unit test — flink-0005
 *
 * DynamicPartitionPruningUtils.convertDppFactSide() uses:
 *   1. List<String>.indexOf(f) inside a stream.map() — O(A×F)
 *   2. List<Integer>.contains() inside a for loop    — O(K×A)
 *
 * where A = accepted filter fields, F = total table columns, K = join keys.
 *
 * Fix: build a Map<String,Integer> once in O(F), use Set<Integer> for membership in O(1).
 *
 * No JUnit. Run:
 *   javac -d . FlinkDynamicPartitionPruningTest.java
 *   java -ea unit.FlinkDynamicPartitionPruningTest
 */
public class FlinkDynamicPartitionPruningTest {

    // ---------------------------------------------------------------------------
    // SLOW path: simulates defective DynamicPartitionPruningUtils.convertDppFactSide()
    //
    //   acceptedFieldIndices = acceptedFilterFields.stream()
    //       .map(f -> fieldNames.indexOf(f))   // O(F) per field
    //       .collect(toList());
    //
    //   for (int i = 0; i < joinKeys.size(); ++i) {
    //       if (acceptedFieldIndices.contains(joinKeys.get(i)))  // O(A) per key
    //           result.add(dimSideJoinKey.get(i));
    //   }
    //
    // Returns: (result indices, op count)
    // ---------------------------------------------------------------------------
    static long[] slowConvertDpp(
            List<String> fieldNames,
            List<String> acceptedFilterFields,
            List<Integer> joinKeys,
            List<Integer> dimSideJoinKey) {

        long ops = 0;

        // Part 1: indexOf — O(A × F)
        List<Integer> acceptedFieldIndices = new ArrayList<>();
        for (String f : acceptedFilterFields) {
            for (int i = 0; i < fieldNames.size(); i++) {
                ops++;
                if (fieldNames.get(i).equals(f)) {
                    acceptedFieldIndices.add(i);
                    break;
                }
            }
        }

        // Part 2: List.contains — O(K × A)
        List<Integer> result = new ArrayList<>();
        for (int i = 0; i < joinKeys.size(); ++i) {
            int key = joinKeys.get(i);
            for (int accepted : acceptedFieldIndices) {
                ops++;
                if (accepted == key) {
                    result.add(dimSideJoinKey.get(i));
                    break;
                }
            }
        }

        return new long[]{result.size(), ops};
    }

    // ---------------------------------------------------------------------------
    // FAST path: Map<String,Integer> + Set<Integer>
    //
    //   Map<String,Integer> nameToIdx = build once from fieldNames — O(F)
    //   Set<Integer> acceptedSet = acceptedFilterFields.stream()
    //       .map(nameToIdx::get)   — O(A)
    //       .collect(toSet())
    //
    //   for (int i = 0; i < joinKeys.size(); ++i) {
    //       if (acceptedSet.contains(joinKeys.get(i)))  — O(1)
    //           result.add(dimSideJoinKey.get(i));
    //   }
    //
    // Returns: (result indices, op count)
    // ---------------------------------------------------------------------------
    static long[] fastConvertDpp(
            List<String> fieldNames,
            List<String> acceptedFilterFields,
            List<Integer> joinKeys,
            List<Integer> dimSideJoinKey) {

        long ops = 0;

        // Build name→index map: O(F)
        Map<String, Integer> nameToIdx = new HashMap<>(fieldNames.size() * 2);
        for (int i = 0; i < fieldNames.size(); i++) {
            nameToIdx.put(fieldNames.get(i), i);
            ops++;  // one write per field
        }

        // Build accepted set: O(A)
        Set<Integer> acceptedSet = new HashSet<>();
        List<Integer> acceptedFieldIndices = new ArrayList<>();
        for (String f : acceptedFilterFields) {
            Integer idx = nameToIdx.get(f);  // O(1)
            ops++;
            if (idx != null) {
                acceptedSet.add(idx);
                acceptedFieldIndices.add(idx);
            }
        }

        // Membership test: O(K)
        List<Integer> result = new ArrayList<>();
        for (int i = 0; i < joinKeys.size(); ++i) {
            ops++;  // O(1) set lookup
            if (acceptedSet.contains(joinKeys.get(i))) {
                result.add(dimSideJoinKey.get(i));
            }
        }

        return new long[]{result.size(), ops};
    }

    // ---------------------------------------------------------------------------
    // Build test data
    // ---------------------------------------------------------------------------

    /** Generate fieldNames = ["col_0", "col_1", ..., "col_{F-1}"] */
    static List<String> buildFieldNames(int F) {
        List<String> names = new ArrayList<>(F);
        for (int i = 0; i < F; i++) names.add("col_" + i);
        return names;
    }

    /**
     * Select A evenly-spaced field names from fieldNames as the accepted filter fields.
     * Also build joinKeys as indices into fieldNames (same set), and dimSideJoinKey as identity.
     */
    static Object[] buildScenario(int F, int A, int K) {
        List<String> fieldNames = buildFieldNames(F);

        // acceptedFilterFields: A evenly-spaced field names
        List<String> accepted = new ArrayList<>(A);
        for (int i = 0; i < A; i++) {
            int idx = (int) ((long) i * F / A);
            accepted.add(fieldNames.get(idx));
        }

        // joinKeys: K evenly-spaced indices into fieldNames (subset overlapping with accepted)
        List<Integer> joinKeys = new ArrayList<>(K);
        List<Integer> dimSideJoinKey = new ArrayList<>(K);
        for (int i = 0; i < K; i++) {
            int idx = (int) ((long) i * F / K);
            joinKeys.add(idx);
            dimSideJoinKey.add(i * 10);  // arbitrary dim-side key
        }

        return new Object[]{fieldNames, accepted, joinKeys, dimSideJoinKey};
    }

    // ---------------------------------------------------------------------------
    // Test helpers
    // ---------------------------------------------------------------------------

    static void test(String name, boolean cond) {
        if (!cond) throw new AssertionError("FAIL: " + name);
        System.out.println("PASS: " + name);
    }

    // ---------------------------------------------------------------------------
    // Main
    // ---------------------------------------------------------------------------

    @SuppressWarnings("unchecked")
    public static void main(String[] args) {
        System.out.println("=== flink-0005: DynamicPartitionPruningUtils indexOf+contains O(A*F+K*A) ===");
        System.out.println();

        // T1: correctness — small scenario F=20, A=5, K=5
        {
            Object[] s = buildScenario(20, 5, 5);
            List<String> fn = (List<String>) s[0];
            List<String> af = (List<String>) s[1];
            List<Integer> jk = (List<Integer>) s[2];
            List<Integer> dk = (List<Integer>) s[3];

            long[] slow = slowConvertDpp(fn, af, jk, dk);
            long[] fast = fastConvertDpp(fn, af, jk, dk);

            test("T1: slow and fast return same result count (F=20,A=5,K=5)",
                    slow[0] == fast[0]);
            System.out.printf("T1: result=%d slow-ops=%d fast-ops=%d%n",
                    slow[0], slow[1], fast[1]);
        }

        // T2: correctness — larger scenario F=100, A=20, K=20
        {
            Object[] s = buildScenario(100, 20, 20);
            List<String> fn = (List<String>) s[0];
            List<String> af = (List<String>) s[1];
            List<Integer> jk = (List<Integer>) s[2];
            List<Integer> dk = (List<Integer>) s[3];

            long[] slow = slowConvertDpp(fn, af, jk, dk);
            long[] fast = fastConvertDpp(fn, af, jk, dk);

            test("T2: slow and fast return same result count (F=100,A=20,K=20)",
                    slow[0] == fast[0]);
        }

        // T3: op count comparison — F=100, A=30, K=30
        //   slow: O(A*F + K*A) = 30*100 + 30*30 = 3000 + 900 = 3900
        //   fast: O(F + A + K) = 100 + 30 + 30 = 160
        {
            int F = 100, A = 30, K = 30;
            Object[] s = buildScenario(F, A, K);
            List<String> fn = (List<String>) s[0];
            List<String> af = (List<String>) s[1];
            List<Integer> jk = (List<Integer>) s[2];
            List<Integer> dk = (List<Integer>) s[3];

            long[] slow = slowConvertDpp(fn, af, jk, dk);
            long[] fast = fastConvertDpp(fn, af, jk, dk);

            System.out.printf("T3: F=%d A=%d K=%d — slow-ops=%d fast-ops=%d ratio=%.1fx%n",
                    F, A, K, slow[1], fast[1], (double) slow[1] / fast[1]);

            test("T3: slow ops > fast ops (F=100, A=30, K=30)", slow[1] > fast[1]);
            test("T3: slow ops >= A*F/2 (lower bound for O(A*F))", slow[1] >= (long) A * F / 2);
            test("T3: fast ops <= F + A + K + 10 (linear bound)", fast[1] <= F + A + K + 10);

            double ratio = (double) slow[1] / fast[1];
            test("T3: speedup >= 5x at F=100,A=30,K=30", ratio >= 5.0);
        }

        // T4: scaling — doubling F should ~2x slow Part 1, ~1x fast
        {
            int A = 20, K = 20;
            int F1 = 100, F2 = 200;

            Object[] s1 = buildScenario(F1, A, K);
            Object[] s2 = buildScenario(F2, A, K);

            long[] slowF1 = slowConvertDpp(
                    (List<String>) s1[0], (List<String>) s1[1],
                    (List<Integer>) s1[2], (List<Integer>) s1[3]);
            long[] slowF2 = slowConvertDpp(
                    (List<String>) s2[0], (List<String>) s2[1],
                    (List<Integer>) s2[2], (List<Integer>) s2[3]);
            long[] fastF1 = fastConvertDpp(
                    (List<String>) s1[0], (List<String>) s1[1],
                    (List<Integer>) s1[2], (List<Integer>) s1[3]);
            long[] fastF2 = fastConvertDpp(
                    (List<String>) s2[0], (List<String>) s2[1],
                    (List<Integer>) s2[2], (List<Integer>) s2[3]);

            double slowRatio = (double) slowF2[1] / slowF1[1];
            double fastRatio = (double) fastF2[1] / fastF1[1];

            System.out.printf("T4: F=%d slow=%d fast=%d%n", F1, slowF1[1], fastF1[1]);
            System.out.printf("T4: F=%d slow=%d fast=%d%n", F2, slowF2[1], fastF2[1]);
            System.out.printf("T4: slow ratio=%.2f (expect ~2.0 for O(F)), fast ratio=%.2f%n",
                    slowRatio, fastRatio);

            test("T4: slow ops grow with F (ratio >= 1.5)", slowRatio >= 1.5);
            test("T4: fast ops grow with F but much slower (ratio <= slowRatio)",
                    fastRatio <= slowRatio);
        }

        // T5: large scenario — measure speedup at F=500, A=50, K=50
        {
            int F = 500, A = 50, K = 50;
            Object[] s = buildScenario(F, A, K);
            List<String> fn = (List<String>) s[0];
            List<String> af = (List<String>) s[1];
            List<Integer> jk = (List<Integer>) s[2];
            List<Integer> dk = (List<Integer>) s[3];

            long[] slow = slowConvertDpp(fn, af, jk, dk);
            long[] fast = fastConvertDpp(fn, af, jk, dk);

            double ratio = (double) slow[1] / fast[1];
            System.out.printf("T5: F=%d A=%d K=%d — slow-ops=%d fast-ops=%d speedup=%.1fx%n",
                    F, A, K, slow[1], fast[1], ratio);

            test("T5: slow ops >= A*F/2 (O(A*F) lower bound)", slow[1] >= (long) A * F / 2);
            test("T5: fast ops <= F + A*2 + K*2 (O(F+A+K) bound)",
                    fast[1] <= F + A * 2 + K * 2);
            test("T5: speedup >= 10x at F=500,A=50,K=50", ratio >= 10.0);
            test("T5: results match", slow[0] == fast[0]);
        }

        // T6: wall-clock at F=1000, A=100, K=100
        {
            int F = 1000, A = 100, K = 100;
            Object[] s = buildScenario(F, A, K);
            List<String> fn = (List<String>) s[0];
            List<String> af = (List<String>) s[1];
            List<Integer> jk = (List<Integer>) s[2];
            List<Integer> dk = (List<Integer>) s[3];

            long t0 = System.nanoTime();
            long[] slow = slowConvertDpp(fn, af, jk, dk);
            long slowNs = System.nanoTime() - t0;

            t0 = System.nanoTime();
            long[] fast = fastConvertDpp(fn, af, jk, dk);
            long fastNs = System.nanoTime() - t0;

            double speedup = (double) slowNs / Math.max(fastNs, 1);
            System.out.printf("T6: F=%d A=%d K=%d — slow=%.3fms fast=%.3fms speedup=%.1fx%n",
                    F, A, K, slowNs / 1e6, fastNs / 1e6, speedup);

            long[] slow2 = slowConvertDpp(fn, af, jk, dk);
            long[] fast2 = fastConvertDpp(fn, af, jk, dk);
            test("T6: results match (F=1000,A=100,K=100)", slow[0] == fast[0]);
            test("T6: slow ops >= A*F/2", slow2[1] >= (long) A * F / 2);
        }

        System.out.println();
        System.out.println("6/6 PASS — flink-0005: DynamicPartitionPruningUtils " +
                "indexOf+contains O(A*F + K*A) → HashMap+HashSet O(F + A + K)");
    }
}