package unit;

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

/**
 * hudi-0001: BaseHoodieTimeline.appendLoadedInstants List.contains O(N×M) → HashSet O(N+M)
 *
 * Simulates the deduplication filter in appendLoadedInstants using a lightweight
 * Instant stand-in (just an integer ID) to count comparison operations.
 *
 * Compile: javac -d . HudiTimelineListContainsAlgorithm.java
 * Run:     java -ea unit.HudiTimelineListContainsAlgorithm
 */
public class HudiTimelineListContainsAlgorithm {

    // Lightweight stand-in for HoodieInstant
    static class Instant {
        final int id;
        static long compareCount = 0;

        Instant(int id) {
            this.id = id;
        }

        @Override
        public boolean equals(Object o) {
            compareCount++;
            if (this == o) return true;
            if (!(o instanceof Instant)) return false;
            return this.id == ((Instant) o).id;
        }

        @Override
        public int hashCode() {
            return Integer.hashCode(id);
        }
    }

    // Defective: List.contains — O(N×M)
    static List<Instant> appendLoadedInstants_slow(List<Instant> existing, List<Instant> loaded) {
        // existingInstants is a plain List
        List<Instant> existingInstants = new ArrayList<>(existing);
        return loaded.stream()
            .filter(instant -> !existingInstants.contains(instant))
            .collect(Collectors.toList());
    }

    // Fixed: Set.contains — O(N+M)
    static List<Instant> appendLoadedInstants_fast(List<Instant> existing, List<Instant> loaded) {
        Set<Instant> existingSet = new HashSet<>(existing);
        return loaded.stream()
            .filter(instant -> !existingSet.contains(instant))
            .collect(Collectors.toList());
    }

    static void test(String name, int existingSize, int loadedSize, int overlapSize) {
        List<Instant> existing = new ArrayList<>();
        for (int i = 0; i < existingSize; i++) existing.add(new Instant(i));

        // loaded = first overlapSize are duplicates, rest are new
        List<Instant> loaded = new ArrayList<>();
        for (int i = 0; i < overlapSize; i++) loaded.add(new Instant(i));            // duplicates
        for (int i = existingSize; i < existingSize + (loadedSize - overlapSize); i++) {
            loaded.add(new Instant(i));  // new instants
        }

        Instant.compareCount = 0;
        List<Instant> resultSlow = appendLoadedInstants_slow(existing, loaded);
        long slowCount = Instant.compareCount;

        Instant.compareCount = 0;
        List<Instant> resultFast = appendLoadedInstants_fast(existing, loaded);
        long fastCount = Instant.compareCount;

        // Correctness: both must return the same new instants
        assert resultSlow.size() == resultFast.size()
            : "Size mismatch slow=" + resultSlow.size() + " fast=" + resultFast.size();

        Set<Integer> slowIds = resultSlow.stream().map(x -> x.id).collect(Collectors.toSet());
        Set<Integer> fastIds = resultFast.stream().map(x -> x.id).collect(Collectors.toSet());
        assert slowIds.equals(fastIds) : "Result mismatch";

        long ratio = fastCount == 0 ? 0 : slowCount / fastCount;
        System.out.printf("  %-40s existing=%4d loaded=%4d overlap=%4d  slow=%8d fast=%8d ratio=%4dx%n",
            name, existingSize, loadedSize, overlapSize, slowCount, fastCount, ratio);
    }

    public static void main(String[] args) {
        int pass = 0;
        int fail = 0;

        System.out.println("hudi-0001: BaseHoodieTimeline.appendLoadedInstants");

        // Test 1: correctness with small input
        try {
            test("correctness-small", 10, 10, 5);
            pass++;
        } catch (AssertionError e) {
            System.out.println("  FAIL correctness-small: " + e.getMessage()); fail++;
        }

        // Test 2: slow path is measurably more expensive than fast path at N=500
        Instant.compareCount = 0;
        List<Instant> existing500 = new ArrayList<>();
        for (int i = 0; i < 500; i++) existing500.add(new Instant(i));
        List<Instant> loaded500 = new ArrayList<>();
        for (int i = 0; i < 250; i++) loaded500.add(new Instant(i));         // duplicates
        for (int i = 500; i < 750; i++) loaded500.add(new Instant(i));       // new

        Instant.compareCount = 0;
        appendLoadedInstants_slow(existing500, loaded500);
        long slowOps = Instant.compareCount;

        Instant.compareCount = 0;
        appendLoadedInstants_fast(existing500, loaded500);
        long fastOps = Instant.compareCount;

        System.out.printf("  %-40s slow=%8d fast=%8d%n", "N=500-ops-comparison", slowOps, fastOps);
        if (slowOps > fastOps * 10) {
            System.out.println("  PASS slow-path > 10x fast-path at N=500"); pass++;
        } else {
            System.out.println("  FAIL expected slow > 10x fast"); fail++;
        }

        // Test 3: no overlap — all new instants
        try {
            test("all-new", 200, 200, 0);
            pass++;
        } catch (AssertionError e) {
            System.out.println("  FAIL all-new: " + e.getMessage()); fail++;
        }

        // Test 4: full overlap — all duplicates
        try {
            test("all-dup", 200, 100, 100);
            pass++;
        } catch (AssertionError e) {
            System.out.println("  FAIL all-dup: " + e.getMessage()); fail++;
        }

        System.out.println(pass + "/" + (pass + fail) + " PASS");
        if (fail > 0) throw new RuntimeException(fail + " tests failed");
    }
}