package unit;

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

/**
 * kafka-0008 — ListDeserializer nullIndexList ArrayList.contains O(S×N) → HashSet O(S)
 *
 * Simulates the inner deserialization loop:
 *   for (int i = 0; i < size; i++) {
 *       if (nullIndexList.contains(i)) { ... }
 *   }
 *
 * SLOW: nullIndexList is ArrayList<Integer>  → O(S × N)
 * FAST: nullIndexList is HashSet<Integer>    → O(S)
 */
public class KafkaListDeserializerNullIndexTest {

    static long slowContains(int size, List<Integer> nullIndexList) {
        long ops = 0;
        for (int i = 0; i < size; i++) {
            ops++;
            if (nullIndexList.contains(i)) {
                // null entry — just count
            }
        }
        return ops;
    }

    static long fastContains(int size, Set<Integer> nullIndexSet) {
        long ops = 0;
        for (int i = 0; i < size; i++) {
            ops++;
            if (nullIndexSet.contains(i)) {
                // null entry — just count
            }
        }
        return ops;
    }

    static long benchSlow(int size, int nullCount) {
        List<Integer> nullIndexList = new ArrayList<>(nullCount);
        // nulls at every-other index (worst-case spread)
        for (int i = 0; i < nullCount; i++) {
            nullIndexList.add(i * 2);
        }
        long start = System.nanoTime();
        long ops = slowContains(size, nullIndexList);
        long elapsed = System.nanoTime() - start;
        return elapsed;
    }

    static long benchFast(int size, int nullCount) {
        Set<Integer> nullIndexSet = new HashSet<>(nullCount * 2);
        for (int i = 0; i < nullCount; i++) {
            nullIndexSet.add(i * 2);
        }
        long start = System.nanoTime();
        long ops = fastContains(size, nullIndexSet);
        long elapsed = System.nanoTime() - start;
        return elapsed;
    }

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

        int[][] cases = {
            {1000,  500},
            {2000, 1000},
            {5000, 2500},
        };

        // warmup
        benchSlow(500, 250);
        benchFast(500, 250);

        for (int[] c : cases) {
            int size = c[0];
            int nullCount = c[1];

            // run multiple times for stable timing
            long slowTotal = 0, fastTotal = 0;
            int reps = 5;
            for (int r = 0; r < reps; r++) {
                slowTotal += benchSlow(size, nullCount);
                fastTotal += benchFast(size, nullCount);
            }
            long slowAvg = slowTotal / reps;
            long fastAvg = fastTotal / reps;

            double ratio = fastAvg > 0 ? (double) slowAvg / fastAvg : 999.0;
            boolean ok = ratio >= minRatio;
            System.out.printf("  size=%-5d nulls=%-5d  slow=%7dns  fast=%7dns  ratio=%.1fx  %s%n",
                size, nullCount, slowAvg, fastAvg, ratio, ok ? "PASS" : "FAIL");
            if (ok) passed++; else failed++;
        }

        System.out.printf("%nTotal: %d/%d PASS%n", passed, passed + failed);
        if (failed > 0) System.exit(1);
    }
}