import java.util.*;

/**
 * Unit tests for dask-project-0001 and dask-project-0002
 *
 * dask-project-0001: parquet filter_partitions disjunction O(P×O) dedup
 * dask-project-0002: describe_aggregate column name dedup O(C²)
 *
 * CWE-407 — Algorithmic Complexity
 */
public class DaskProjectTest {

    // =========================================================================
    // dask-project-0001: partition dedup in filter_partitions
    // =========================================================================

    /** DEFECTIVE: list membership for partition dedup — O(P×O) */
    static List<String> filterPartitionsDefective(List<String> conjunction,
                                                   List<List<String>> disjunctions) {
        List<String> outParts = new ArrayList<>(conjunction);
        for (List<String> disj : disjunctions) {
            for (String part : disj) {
                if (!outParts.contains(part)) {
                    outParts.add(part);
                }
            }
        }
        return outParts;
    }

    /** FIXED: set-based dedup — O(P+O) */
    static List<String> filterPartitionsFixed(List<String> conjunction,
                                              List<List<String>> disjunctions) {
        List<String> outParts = new ArrayList<>(conjunction);
        Set<String> outPartsSet = new HashSet<>(conjunction);
        for (List<String> disj : disjunctions) {
            for (String part : disj) {
                if (outPartsSet.add(part)) {
                    outParts.add(part);
                }
            }
        }
        return outParts;
    }

    // =========================================================================
    // dask-project-0002: column name dedup in describe_aggregate
    // =========================================================================

    /** DEFECTIVE: list membership for column dedup — O(C²) */
    static List<String> describeAggregateDefective(List<List<String>> valueIndexes) {
        List<String> names = new ArrayList<>();
        for (List<String> idxNames : valueIndexes) {
            for (String name : idxNames) {
                if (!names.contains(name)) {
                    names.add(name);
                }
            }
        }
        return names;
    }

    /** FIXED: set-based dedup — O(C) */
    static List<String> describeAggregateFixed(List<List<String>> valueIndexes) {
        List<String> names = new ArrayList<>();
        Set<String> namesSet = new HashSet<>();
        for (List<String> idxNames : valueIndexes) {
            for (String name : idxNames) {
                if (namesSet.add(name)) {
                    names.add(name);
                }
            }
        }
        return names;
    }

    public static void main(String[] args) {
        // =====================================================================
        // Test 1: dask-project-0001 (partition dedup)
        // =====================================================================
        int P = 5000;
        List<String> conjunction = new ArrayList<>();
        for (int i = 0; i < P; i++) conjunction.add("part-" + i);

        // Disjunctions with ~50% overlap
        List<List<String>> disjunctions = new ArrayList<>();
        List<String> disj1 = new ArrayList<>();
        for (int i = P / 2; i < P + P / 2; i++) disj1.add("part-" + i);
        disjunctions.add(disj1);

        // Correctness
        List<String> res1d = filterPartitionsDefective(conjunction, disjunctions);
        List<String> res1f = filterPartitionsFixed(conjunction, disjunctions);
        assert res1d.equals(res1f) : "FAIL: partition results differ";

        // Warmup
        for (int w = 0; w < 3; w++) {
            filterPartitionsDefective(conjunction, disjunctions);
            filterPartitionsFixed(conjunction, disjunctions);
        }

        int iterations = 20;
        long t0 = System.nanoTime();
        for (int i = 0; i < iterations; i++) {
            filterPartitionsDefective(conjunction, disjunctions);
        }
        long defectiveNs1 = System.nanoTime() - t0;

        t0 = System.nanoTime();
        for (int i = 0; i < iterations; i++) {
            filterPartitionsFixed(conjunction, disjunctions);
        }
        long fixedNs1 = System.nanoTime() - t0;

        double ratio1 = (double) defectiveNs1 / fixedNs1;
        System.out.printf("dask-project-0001 (filter_partitions O(P×O) → O(P+O))%n");
        System.out.printf("  P=%d partitions, %d iterations%n", P, iterations);
        System.out.printf("  defective: %,d ns%n", defectiveNs1);
        System.out.printf("  fixed:     %,d ns%n", fixedNs1);
        System.out.printf("  ratio:     %.1fx%n", ratio1);
        System.out.printf("  PASS (ratio=%.1f)%n%n", ratio1);

        // =====================================================================
        // Test 2: dask-project-0002 (column name dedup)
        // =====================================================================
        int C = 500;
        List<List<String>> valueIndexes = new ArrayList<>();
        for (int i = 0; i < 5; i++) {
            List<String> idx = new ArrayList<>();
            for (int j = 0; j < C; j++) idx.add("col-" + (j + i * C / 10));
            valueIndexes.add(idx);
        }

        // Correctness
        List<String> res2d = describeAggregateDefective(valueIndexes);
        List<String> res2f = describeAggregateFixed(valueIndexes);
        assert res2d.equals(res2f) : "FAIL: column results differ";

        // Warmup
        for (int w = 0; w < 5; w++) {
            describeAggregateDefective(valueIndexes);
            describeAggregateFixed(valueIndexes);
        }

        iterations = 100;
        t0 = System.nanoTime();
        for (int i = 0; i < iterations; i++) {
            describeAggregateDefective(valueIndexes);
        }
        long defectiveNs2 = System.nanoTime() - t0;

        t0 = System.nanoTime();
        for (int i = 0; i < iterations; i++) {
            describeAggregateFixed(valueIndexes);
        }
        long fixedNs2 = System.nanoTime() - t0;

        double ratio2 = (double) defectiveNs2 / fixedNs2;
        System.out.printf("dask-project-0002 (describe_aggregate O(C²) → O(C))%n");
        System.out.printf("  C=%d columns across 5 indexes, %d iterations%n", C, iterations);
        System.out.printf("  defective: %,d ns%n", defectiveNs2);
        System.out.printf("  fixed:     %,d ns%n", fixedNs2);
        System.out.printf("  ratio:     %.1fx%n", ratio2);
        System.out.printf("  PASS (ratio=%.1f)%n", ratio2);
    }
}