package unit;

// NO JUnit. Compile: javac -d . *.java  Run: java -ea unit.StorageSystemColumnsAlgorithm
//
// ClickHouse clickhouse-0001: StorageSystemColumns find_in_vector O(C×K) → O(C)
//
// Simulates: for each column in table schema, perform linear scan of key-column names
// to determine if the column is part of partition/sort/primary key.
//
// SLOW: std::find over Names vector per column  → O(C × K)
// FAST: unordered_set built once before loop     → O(C)

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

public class StorageSystemColumnsAlgorithm {

    // Simulates the defective find_in_vector pattern:
    //   for each column, call std::find on a Names vector
    static class SlowImpl {
        long opCount = 0;

        // Returns bool[] of size C: whether each column is in keyNames
        // O(C × K) — for each of C columns, scans K key names
        boolean[] computeMembership(List<String> columns, List<String> keyNames) {
            boolean[] result = new boolean[columns.size()];
            for (int i = 0; i < columns.size(); i++) {
                String col = columns.get(i);
                // Simulate find_in_vector: linear scan over keyNames
                for (int j = 0; j < keyNames.size(); j++) {
                    opCount++;
                    if (keyNames.get(j).equals(col)) {
                        result[i] = true;
                        break;
                    }
                }
            }
            return result;
        }
    }

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

        // Returns bool[] of size C: whether each column is in keyNames
        // O(C) after O(K) set construction
        boolean[] computeMembership(List<String> columns, List<String> keyNames) {
            // Build O(1) lookup set — done once per table, not per column
            Set<String> keySet = new HashSet<>(keyNames);
            opCount += keyNames.size(); // count set-build ops

            boolean[] result = new boolean[columns.size()];
            for (int i = 0; i < columns.size(); i++) {
                opCount++;
                result[i] = keySet.contains(columns.get(i));
            }
            return result;
        }
    }

    // Build a realistic schema: C total columns, K in the key lists
    // (4 key lists: partition, sorting, primary, sampling)
    static List<String> buildSchema(int C) {
        List<String> cols = new ArrayList<>();
        for (int i = 0; i < C; i++) {
            cols.add("col_" + i);
        }
        return cols;
    }

    static List<String> buildKeyList(List<String> schema, int K) {
        // Put key columns at the END of the schema so non-key columns must scan
        // the entire key list before determining non-membership — worst case
        List<String> keys = new ArrayList<>();
        int start = Math.max(0, schema.size() - K);
        for (int i = start; i < schema.size(); i++) {
            keys.add(schema.get(i));
        }
        return keys;
    }

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

        // Test 1: correctness — both impls produce same membership result
        {
            total++;
            List<String> schema = buildSchema(20);
            List<String> keyList = buildKeyList(schema, 5);
            // add a non-existent key to ensure false-negatives handled
            keyList.add("nonexistent_col");

            SlowImpl slow = new SlowImpl();
            FastImpl fast = new FastImpl();
            boolean[] slowResult = slow.computeMembership(schema, keyList);
            boolean[] fastResult = fast.computeMembership(schema, keyList);

            boolean correct = true;
            for (int i = 0; i < slowResult.length; i++) {
                if (slowResult[i] != fastResult[i]) {
                    correct = false;
                    System.err.println("MISMATCH at col " + i + ": slow=" + slowResult[i] + " fast=" + fastResult[i]);
                }
            }
            assert correct : "correctness check failed";
            System.out.println("Test 1 PASS: correctness verified for C=20, K=5");
            passed++;
        }

        // Test 2: op ratio at C=500, K=10 (typical wide table with compound sort key)
        // StorageSystemColumns calls find_in_vector 4× per column (4 key lists).
        // Simulate 4 key lists: partition, sorting, primary, sampling
        {
            total++;
            int C = 500;
            int K = 10;
            List<String> schema = buildSchema(C);
            List<String> partitionKey = buildKeyList(schema, K);
            List<String> sortingKey = buildKeyList(schema, K);
            List<String> primaryKey = buildKeyList(schema, K);
            List<String> samplingKey = buildKeyList(schema, 2);

            SlowImpl slow = new SlowImpl();
            // 4 calls per column simulating the 4 find_in_vector invocations
            slow.computeMembership(schema, partitionKey);
            slow.computeMembership(schema, sortingKey);
            slow.computeMembership(schema, primaryKey);
            slow.computeMembership(schema, samplingKey);
            long slowOps = slow.opCount;

            FastImpl fast = new FastImpl();
            fast.computeMembership(schema, partitionKey);
            fast.computeMembership(schema, sortingKey);
            fast.computeMembership(schema, primaryKey);
            fast.computeMembership(schema, samplingKey);
            long fastOps = fast.opCount;

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

            // Keys at END of schema: non-key columns (C-K = 490) each scan full K=10 per list
            // Slow: 4 lists × [(C-K)*K + K*(K/2)] ≈ 4*(490*10 + 50) = 4*4950 = 19800
            // Fast: 4*(K+C) = 4*510 = 2040
            // Ratio ≈ 9.7x — use triangular lower bound for minQuadratic
            long minQuadratic = (long) (C - K) * K * 4 / 2; // half of worst case, 4 lists
            assert slowOps >= minQuadratic :
                "slow_ops=" + slowOps + " should be >= " + minQuadratic;

            // Fast ops are bounded by 4*(K+C)
            long maxLinear = (long) 4 * (K + C + 10); // small slack
            assert fastOps <= maxLinear :
                "fast_ops=" + fastOps + " should be <= " + maxLinear;

            // At least 7x speedup (realistic with worst-case key placement)
            assert slowOps >= fastOps * 7 :
                "Expected >= 7x op ratio, got slow=" + slowOps + " fast=" + fastOps;

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

        // Test 3: worst-case C=1000, K=20 — typical ClickHouse wide-table schema
        {
            total++;
            int C = 1000;
            int K = 20;
            List<String> schema = buildSchema(C);
            // Worst case: key columns are at the end of the schema
            // so every std::find scans the entire vector
            List<String> keyList = new ArrayList<>();
            for (int i = C - K; i < C; i++) {
                keyList.add(schema.get(i));
            }

            SlowImpl slow = new SlowImpl();
            slow.computeMembership(schema, keyList); // 1 key list for simplicity
            long slowOps = slow.opCount;

            FastImpl fast = new FastImpl();
            fast.computeMembership(schema, keyList);
            long fastOps = fast.opCount;

            System.out.println("Test 3: C=" + C + ", K=" + K + " (worst case), slow_ops=" + slowOps + ", fast_ops=" + fastOps);

            // Worst case: every column NOT in key (they're at end), slow scans full K
            // slowOps ≈ C * K = 1000 * 20 = 20000
            long expectedSlowMin = (long) C * K / 2; // at least half
            assert slowOps >= expectedSlowMin :
                "slow_ops=" + slowOps + " should be >= " + expectedSlowMin;

            assert fastOps <= C + K + 10 :
                "fast_ops=" + fastOps + " should be <= " + (C + K + 10);

            assert slowOps >= fastOps * 7 :
                "Expected >= 7x op ratio, got slow=" + slowOps + " fast=" + fastOps;

            System.out.println("Test 3 PASS: " + (slowOps / fastOps) + "x speedup at C=1000, K=20");
            passed++;
        }

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