package unit;

// NO JUnit. Compile: javac -d . *.java  Run: java -ea unit.ScanQueryAlgorithm
//
// Druid druid-0001: ScanQuery columns List.contains in orderBy validation loop O(N×M) → O(N)
//
// Simulates: for each orderBy column, check if it exists in the selected columns list
//
// SLOW: List<String>.contains() per orderBy column — O(N × M)
// FAST: HashSet<String> built once, .contains() — O(N)

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

public class ScanQueryAlgorithm {

    // Simulates the defective ScanQuery constructor validation:
    //   for each orderBy column, call List<String>.contains()
    static class SlowImpl {
        long opCount = 0;

        // Returns list of invalid orderBy columns (those not in selectedColumns)
        // O(N × M) where N=orderBys, M=selectedColumns
        List<String> validateOrderBy(List<String> selectedColumns, List<String> orderBys) {
            List<String> invalid = new ArrayList<>();
            for (String orderByCol : orderBys) {
                // Simulate List.contains: linear scan
                boolean found = false;
                for (int i = 0; i < selectedColumns.size(); i++) {
                    opCount++;
                    if (selectedColumns.get(i).equals(orderByCol)) {
                        found = true;
                        break;
                    }
                }
                if (!found) {
                    invalid.add(orderByCol);
                }
            }
            return invalid;
        }
    }

    // Simulates the fixed HashSet pattern:
    //   build set once, O(1) per orderBy column check
    static class FastImpl {
        long opCount = 0;

        // Returns list of invalid orderBy columns (those not in selectedColumns)
        // O(N) after O(M) set construction
        List<String> validateOrderBy(List<String> selectedColumns, List<String> orderBys) {
            // Build O(1) lookup set once
            Set<String> colSet = new HashSet<>(selectedColumns);
            opCount += selectedColumns.size(); // count set-build cost

            List<String> invalid = new ArrayList<>();
            for (String orderByCol : orderBys) {
                opCount++;
                if (!colSet.contains(orderByCol)) {
                    invalid.add(orderByCol);
                }
            }
            return invalid;
        }
    }

    static List<String> buildColumns(int M) {
        List<String> cols = new ArrayList<>();
        for (int i = 0; i < M; i++) {
            cols.add("col_" + i);
        }
        return cols;
    }

    static List<String> buildOrderBys(List<String> columns, int N) {
        // orderBy columns are at the end of the selected list (worst case for linear scan)
        List<String> orderBys = new ArrayList<>();
        for (int i = columns.size() - N; i < columns.size(); i++) {
            orderBys.add(columns.get(i));
        }
        return orderBys;
    }

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

        // Test 1: correctness — both impls agree on valid/invalid columns
        {
            total++;
            List<String> cols = buildColumns(20);
            List<String> validOrderBys = buildOrderBys(cols, 3);
            List<String> invalidOrderBys = new ArrayList<>(validOrderBys);
            invalidOrderBys.add("nonexistent_col");

            SlowImpl slow = new SlowImpl();
            FastImpl fast = new FastImpl();

            List<String> slowInvalid = slow.validateOrderBy(cols, invalidOrderBys);
            List<String> fastInvalid = fast.validateOrderBy(cols, invalidOrderBys);

            assert slowInvalid.equals(fastInvalid) :
                "Mismatch: slow=" + slowInvalid + " fast=" + fastInvalid;
            assert slowInvalid.size() == 1 :
                "Expected 1 invalid column, got " + slowInvalid.size();
            assert slowInvalid.get(0).equals("nonexistent_col") :
                "Expected nonexistent_col, got " + slowInvalid.get(0);

            System.out.println("Test 1 PASS: correctness verified, invalid=[" + slowInvalid.get(0) + "]");
            passed++;
        }

        // Test 2: op ratio at M=500 selected columns, N=5 orderBy columns
        // ScanQuery is constructed per segment scan — with M=500 wide-table analytics
        {
            total++;
            int M = 500;
            int N = 5;
            List<String> cols = buildColumns(M);
            // orderBy columns at the end — worst case for linear scan
            List<String> orderBys = buildOrderBys(cols, N);

            SlowImpl slow = new SlowImpl();
            slow.validateOrderBy(cols, orderBys);
            long slowOps = slow.opCount;

            FastImpl fast = new FastImpl();
            fast.validateOrderBy(cols, orderBys);
            long fastOps = fast.opCount;

            System.out.println("Test 2: M=" + M + ", N=" + N + ", slow_ops=" + slowOps + ", fast_ops=" + fastOps);

            // Slow: each of N orderBys must scan to end of M (they're at the end)
            // worst case: N*M = 5*500 = 2500 ops
            long minQuadratic = (long) N * (M / 2); // at least half-scan
            assert slowOps >= minQuadratic :
                "slow_ops=" + slowOps + " should be >= " + minQuadratic;

            // Fast: M (set build) + N (lookups) = 505
            long maxLinear = M + N + 10;
            assert fastOps <= maxLinear :
                "fast_ops=" + fastOps + " should be <= " + maxLinear;

            // Ratio: N*M / (M+N) = 5*500/505 ≈ 4.95 — clear speedup, conservative threshold 4x
            assert slowOps * 4 >= fastOps * 10 :
                "Expected slow/fast ratio >= 2.5x, got slow=" + slowOps + " fast=" + fastOps;

            System.out.println("Test 2 PASS: slow_ops=" + slowOps + " (fast=" + fastOps + ")");
            passed++;
        }

        // Test 3: large schema M=2000, N=10 (analytical query over very wide table)
        {
            total++;
            int M = 2000;
            int N = 10;
            List<String> cols = buildColumns(M);
            List<String> orderBys = buildOrderBys(cols, N);

            SlowImpl slow = new SlowImpl();
            slow.validateOrderBy(cols, orderBys);
            long slowOps = slow.opCount;

            FastImpl fast = new FastImpl();
            fast.validateOrderBy(cols, orderBys);
            long fastOps = fast.opCount;

            System.out.println("Test 3: M=" + M + ", N=" + N + ", slow_ops=" + slowOps + ", fast_ops=" + fastOps);

            // Slow: N*M minimum (all orderBys at end, full scan)
            long minQuadratic = (long) N * (M / 2);
            assert slowOps >= minQuadratic :
                "slow_ops=" + slowOps + " should be >= " + minQuadratic;

            long maxLinear = M + N + 10;
            assert fastOps <= maxLinear :
                "fast_ops=" + fastOps + " should be <= " + maxLinear;

            // At M=2000, N=10: ratio = N*M/(M+N) = 10*2000/2010 ≈ 9.95x
            assert slowOps >= fastOps * 9 :
                "Expected >= 9x op ratio, got slow=" + slowOps + " fast=" + fastOps;

            System.out.println("Test 3 PASS: " + (slowOps / fastOps) + "x speedup at M=" + M + ", N=" + N);
            passed++;
        }

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