package unit;

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

/**
 * hudi-0003: HoodieTableMetadataUtil.getRevivedAndDeletedKeysFromMergedLogs
 *            List<String>.contains O(N×M) → HashSet O(N+M)
 *
 * Simulates the log file path deduplication filter in the RLI update path.
 *
 * Compile: javac -d . HudiLogFilePathsAlgorithm.java
 * Run:     java -ea unit.HudiLogFilePathsAlgorithm
 */
public class HudiLogFilePathsAlgorithm {

    static long compareCount = 0;

    // Defective: List<String>.contains — O(N×M)
    static List<String> filterLogPaths_slow(List<String> allPaths, List<String> currentPaths) {
        return allPaths.stream()
            .filter(path -> {
                for (String c : currentPaths) {
                    compareCount++;
                    if (c.equals(path)) return false;
                }
                return true;
            })
            .collect(Collectors.toList());
    }

    // Fixed: HashSet<String>.contains — O(N+M)
    static List<String> filterLogPaths_fast(List<String> allPaths, List<String> currentPaths) {
        Set<String> currentSet = new HashSet<>(currentPaths);
        return allPaths.stream()
            .filter(path -> {
                compareCount++;
                return !currentSet.contains(path);
            })
            .collect(Collectors.toList());
    }

    static void assertEq(String label, Object expected, Object actual) {
        if (!expected.equals(actual)) {
            throw new AssertionError(label + ": expected " + expected + " but got " + actual);
        }
    }

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

        System.out.println("hudi-0003: HoodieTableMetadataUtil.getRevivedAndDeletedKeysFromMergedLogs");

        // Test 1: correctness — small overlap
        try {
            List<String> all = new ArrayList<>();
            List<String> current = new ArrayList<>();
            for (int i = 0; i < 20; i++) all.add("/table/part/.hoodie_meta/file_" + i + ".log");
            for (int i = 0; i < 5; i++) current.add(all.get(i));

            compareCount = 0;
            List<String> slow = filterLogPaths_slow(all, current);
            compareCount = 0;
            List<String> fast = filterLogPaths_fast(all, current);

            assertEq("size", slow.size(), fast.size());
            assertEq("content", new HashSet<>(slow), new HashSet<>(fast));
            System.out.println("  PASS correctness-small overlap=" + current.size());
            pass++;
        } catch (AssertionError e) { System.out.println("  FAIL correctness: " + e.getMessage()); fail++; }

        // Test 2: correctness — no overlap
        try {
            List<String> all = new ArrayList<>();
            List<String> current = new ArrayList<>();
            for (int i = 0; i < 10; i++) all.add("/part/file_a_" + i + ".log");
            for (int i = 0; i < 5; i++) current.add("/part/file_b_" + i + ".log");

            List<String> slow = filterLogPaths_slow(all, current);
            List<String> fast = filterLogPaths_fast(all, current);
            assertEq("no-overlap-size", slow.size(), fast.size());
            assertEq("no-overlap-all", 10, slow.size());
            System.out.println("  PASS no-overlap");
            pass++;
        } catch (AssertionError e) { System.out.println("  FAIL no-overlap: " + e.getMessage()); fail++; }

        // Test 3: correctness — full overlap
        try {
            List<String> all = new ArrayList<>();
            for (int i = 0; i < 10; i++) all.add("/part/file_" + i + ".log");
            List<String> current = new ArrayList<>(all);

            List<String> slow = filterLogPaths_slow(all, current);
            List<String> fast = filterLogPaths_fast(all, current);
            assertEq("full-overlap-size", slow.size(), fast.size());
            assertEq("full-overlap-empty", 0, slow.size());
            System.out.println("  PASS full-overlap");
            pass++;
        } catch (AssertionError e) { System.out.println("  FAIL full-overlap: " + e.getMessage()); fail++; }

        // Test 4: N=500, M=500 — operation count
        {
            List<String> all = new ArrayList<>();
            List<String> current = new ArrayList<>();
            for (int i = 0; i < 500; i++) all.add("/table/p0/.hoodie_meta/file_" + i + ".log");
            for (int i = 0; i < 250; i++) current.add(all.get(i));        // half overlap
            for (int i = 500; i < 750; i++) current.add("/table/p0/.hoodie_meta/file_" + i + ".log");  // non-overlap current

            compareCount = 0;
            List<String> slowResult = filterLogPaths_slow(all, current);
            long slowOps = compareCount;

            compareCount = 0;
            List<String> fastResult = filterLogPaths_fast(all, current);
            long fastOps = compareCount;

            assertEq("N500-result", new HashSet<>(slowResult), new HashSet<>(fastResult));
            System.out.printf("  N=500 M=500: slow=%8d fast=%6d ratio=%4dx%n",
                slowOps, fastOps, fastOps > 0 ? slowOps / fastOps : 0);

            if (slowOps > fastOps * 10) {
                System.out.println("  PASS slow > 10x fast at N=500"); pass++;
            } else {
                System.out.println("  FAIL expected slow > 10x fast"); fail++;
            }
        }

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