Function bodies 231 total

    static void setUpHBase() throws Exception {
        Path dockerDir = Path.of(System.getProperty("project.dir"), "docker");
        hbaseContainer = new GenericContainer<>(
                new ImageFromDockerfile()
                        .withDockerfile(dockerDir.resolve("Dockerfile")))
                .withCreateContainerCmdModifier(cmd -> {
                    cmd.withHostName("localhost");
                    cmd.getHostConfig()
                            .withPortBindings(
                                    new PortBinding(Ports.Binding.bindPort(ZK_PORT), new ExposedPort(ZK_PORT)),
                                    new PortBinding(Ports.Binding.bindPort(MASTER_PORT), new ExposedPort(MASTER_PORT)),
                                    new PortBinding(Ports.Binding.bindPort(RS_PORT), new ExposedPort(RS_PORT)),
                                    new PortBinding(Ports.Binding.bindPort(MASTER_UI_PORT), new ExposedPort(MASTER_UI_PORT))
                            );
             

    static void tearDownHBase() throws Exception {
        if (asyncConnection != null) asyncConnection.close();
        if (syncConnection != null) syncConnection.close();
        if (hbaseContainer != null) hbaseContainer.stop();
    }

    private static void createNamespacesAndTables() throws Exception {
        try (Admin admin = syncConnection.getAdmin()) {
            try {
                admin.createNamespace(NamespaceDescriptor.create("govt").build());
            } catch (Exception e) {
                // namespace may already exist
            }

            createTableFromMapper(admin, new CitizenHBMapper(new BestSuitCodec()));
            createTableFromMapper(admin, new EmployeeHBMapper(new BestSuitCodec()));
            createTableFromMapper(admin, new CrawlHBMapper(new BestSuitCodec()));
        }
    }

    private static void createTableFromMapper(Admin admin, HBaseMapper<?, ?> mapper) throws Exception {
        TableName tableName = TableName.valueOf(mapper.getTableName());
        if (admin.tableExists(tableName)) {
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
        }

        TableDescriptorBuilder tableBuilder = TableDescriptorBuilder.newBuilder(tableName);
        for (Map.Entry<String, Integer> entry : mapper.getColumnFamiliesAndVersions().entrySet()) {
            ColumnFamilyDescriptorBuilder cfBuilder =
                    ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(entry.getKey()));
            cfBuilder.setMaxVersions(entry.getValue());
            tableBuilder.setColumnFamily(cfBuilder.build());
        }
        admin.createTable(tableBuilder.build());
    }

    private static void waitForZookeeper(String host, int port, int maxRetries) throws Exception {
        for (int i = 0; i < maxRetries; i++) {
            try (Socket socket = new Socket(host, port)) {
                socket.getOutputStream().write("ruok".getBytes());
                socket.getOutputStream().flush();
                byte[] buf = new byte[4];
                int read = socket.getInputStream().read(buf);
                if (read == 4 && new String(buf).equals("imok")) {
                    System.out.println("ZooKeeper ready after " + (i + 1) + " attempts");
                    return;
                }
            } catch (Exception e) {
                // not ready yet
            }
            Thread.sleep(1000);
        }
        throw new RuntimeException("ZooKeeper did not become ready at " + host + ":" + port);
    }

public final class PerfResultReporter {

    private static final List<Entry> entries = Collections.synchronizedList(new ArrayList<>());

    private PerfResultReporter() {}

    /**
     * Run a benchmark: warmup, then measured iterations. Records result for the final report.
     */
    public static BenchmarkResult run(String name, int warmupCount, int measuredCount, Runnable operation) {
        // Warmup
        for (int i = 0; i < warmupCount; i++) {
            operation.run();
        }

        // Measured
        long[] latencies = new long[measuredCount];
        long start = System.nanoTime();
        for (int i = 0; i < measuredCount; i++) {
            long opStart = System.nanoTime();
            operation.run();
            latencies[i] = System.nanoTime() - opStart;
        }
        long totalNanos = System.nanoTime() - start;

        Arrays.sort(latencies);
        BenchmarkResult result = new BenchmarkResult(
                name,
                measuredCount,
   

    public static BenchmarkResult run(String name, int warmupCount, int measuredCount, Runnable operation) {
        // Warmup
        for (int i = 0; i < warmupCount; i++) {
            operation.run();
        }

        // Measured
        long[] latencies = new long[measuredCount];
        long start = System.nanoTime();
        for (int i = 0; i < measuredCount; i++) {
            long opStart = System.nanoTime();
            operation.run();
            latencies[i] = System.nanoTime() - opStart;
        }
        long totalNanos = System.nanoTime() - start;

        Arrays.sort(latencies);
        BenchmarkResult result = new BenchmarkResult(
                name,
                measuredCount,
                percentile(latencies, 50),
                percentile(latencies, 95),
                percentile(latencies, 99),
                mean(latencies),
                (measuredCount * 1_000_000_000.0) / totalNanos
        );

        entries.add(new Entry(name, result));
      

    public static void writeReport(Path projectDir) throws IOException {
        Path reportDir = projectDir.resolve("build/reports/perfTest");
        Files.createDirectories(reportDir);
        Path reportFile = reportDir.resolve("results.json");

        StringBuilder sb = new StringBuilder("[\n");
        for (int i = 0; i < entries.size(); i++) {
            Entry e = entries.get(i);
            BenchmarkResult r = e.result;
            sb.append("  {\n");
            sb.append("    \"name\": \"").append(escapeJson(r.name)).append("\",\n");
            sb.append("    \"ops\": ").append(r.ops).append(",\n");
            sb.append("    \"p50_us\": ").append(String.format("%.2f", r.p50Nanos / 1000.0)).append(",\n");
            sb.append("    \"p95_us\": ").append(String.format("%.2f", r.p95Nanos / 1000.0)).append(",\n");
            sb.append("    \"p99_us\": ").append(String.format("%.2f", r.p99Nanos / 1000.0)).append(",\n");
            sb.append("    \"mean_us\": ").append(String

    public static void clear() {
        entries.clear();
    }

    private static long percentile(long[] sorted, int pct) {
        int idx = (int) Math.ceil(pct / 100.0 * sorted.length) - 1;
        return sorted[Math.max(0, idx)];
    }

    private static double mean(long[] values) {
        long sum = 0;
        for (long v : values) sum += v;
        return (double) sum / values.length;
    }

    private static String escapeJson(String s) {
        return s.replace("\\", "\\\\").replace("\"", "\\\"");
    }

    public record BenchmarkResult(
            String name,
            int ops,
            long p50Nanos,
            long p95Nanos,
            long p99Nanos,
            double meanNanos,
            double throughputOpsPerSec
    ) {
        @Override
        public String toString() {
            return String.format("%s: ops=%d, p50=%.1fus, p95=%.1fus, p99=%.1fus, mean=%.1fus, throughput=%.1f ops/s",
                    name, ops, p50Nanos / 1000.0, p95Nanos / 1000.0, p99Nanos / 1000.0,
                    meanNanos / 1000.0, throughputOpsPerSec);
        }
    }

    public record BenchmarkResult(
            String name,
            int ops,
            long p50Nanos,
            long p95Nanos,
            long p99Nanos,
            double meanNanos,
            double throughputOpsPerSec
    ) {
        @Override
        public String toString() {
            return String.format("%s: ops=%d, p50=%.1fus, p95=%.1fus, p99=%.1fus, mean=%.1fus, throughput=%.1f ops/s",
                    name, ops, p50Nanos / 1000.0, p95Nanos / 1000.0, p99Nanos / 1000.0,
                    meanNanos / 1000.0, throughputOpsPerSec);
        }
    }

        public String toString() {
            return String.format("%s: ops=%d, p50=%.1fus, p95=%.1fus, p99=%.1fus, mean=%.1fus, throughput=%.1f ops/s",
                    name, ops, p50Nanos / 1000.0, p95Nanos / 1000.0, p99Nanos / 1000.0,
                    meanNanos / 1000.0, throughputOpsPerSec);
        }

class SyncDAOPerfTest extends HBasePerfTestBase {

    @BeforeAll
    static void init() {
        PerfResultReporter.clear();
    }

    @AfterAll
    static void report() throws Exception {
        PerfResultReporter.writeReport(Path.of(System.getProperty("project.dir")));
    }

    @Test
    @Order(1)
    void singlePutGetLatency() throws Exception {
        BenchmarkResult result = PerfResultReporter.run("sync.single.put_get", 50, 1000, () -> {
            try {
                Citizen c = new Citizen("PERF", 1, "SingleTest");
                c.setAge((short) 25);
                citizenDAO.persist(c);
                Citizen fetched = citizenDAO.get("PERF#1");
                assertNotNull(fetched);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
        System.out.println(result);
    }

    @ParameterizedTest
    @ValueSource(ints = {10, 100, 1000})
    @Order(2)
    void bulkPutThroughput(int batchSize) throws Except

    static void init() {
        PerfResultReporter.clear();
    }

    static void report() throws Exception {
        PerfResultReporter.writeReport(Path.of(System.getProperty("project.dir")));
    }

    void singlePutGetLatency() throws Exception {
        BenchmarkResult result = PerfResultReporter.run("sync.single.put_get", 50, 1000, () -> {
            try {
                Citizen c = new Citizen("PERF", 1, "SingleTest");
                c.setAge((short) 25);
                citizenDAO.persist(c);
                Citizen fetched = citizenDAO.get("PERF#1");
                assertNotNull(fetched);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
        System.out.println(result);
    }

    void bulkPutThroughput(int batchSize) throws Exception {
        List<Citizen> batch = new ArrayList<>(batchSize);
        for (int i = 0; i < batchSize; i++) {
            Citizen c = new Citizen("BULK", i, "Citizen-" + i);
            c.setAge((short) (20 + i % 50));
            batch.add(c);
        }

        BenchmarkResult result = PerfResultReporter.run(
                "sync.bulk_put.batch_" + batchSize, 3, 20, () -> {
                    try {
                        citizenDAO.persist(batch);
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                });
        System.out.println(result);
    }

    void bulkGetThroughput(int batchSize) throws Exception {
        // Seed data
        List<Citizen> batch = new ArrayList<>(batchSize);
        List<String> keys = new ArrayList<>(batchSize);
        for (int i = 0; i < batchSize; i++) {
            Citizen c = new Citizen("BGET", i, "Citizen-" + i);
            batch.add(c);
            keys.add("BGET#" + i);
        }
        citizenDAO.persist(batch);

        BenchmarkResult result = PerfResultReporter.run(
                "sync.bulk_get.batch_" + batchSize, 3, 20, () -> {
                    try {
                        citizenDAO.get(keys);
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                });
        System.out.println(result);
    }

    void prefixScanThroughput(int resultSize) throws Exception {
        // Seed data with unique prefix
        String prefix = "SCAN" + resultSize;
        List<Citizen> batch = new ArrayList<>(resultSize);
        for (int i = 0; i < resultSize; i++) {
            batch.add(new Citizen(prefix, i, "Citizen-" + i));
        }
        citizenDAO.persist(batch);

        BenchmarkResult result = PerfResultReporter.run(
                "sync.prefix_scan.size_" + resultSize, 3, 20, () -> {
                    try {
                        citizenDAO.getByPrefix(Bytes.toBytes(prefix + "#"));
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                });
        System.out.println(result);
    }

    void rangeScanLatency() throws Exception {
        // Seed contiguous range
        List<Citizen> batch = new ArrayList<>(100);
        for (int i = 0; i < 100; i++) {
            batch.add(new Citizen("RNG", i, "Citizen-" + i));
        }
        citizenDAO.persist(batch);

        BenchmarkResult result = PerfResultReporter.run("sync.range_scan", 5, 50, () -> {
            try {
                citizenDAO.get("RNG#0", "RNG#99");
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
        System.out.println(result);
    }

    void multiVersionWriteRead() throws Exception {
        BenchmarkResult result = PerfResultReporter.run("sync.multi_version.write_read", 5, 50, () -> {
            try {
                Crawl crawl = new Crawl("perf-mv-" + System.nanoTime());
                for (int v = 0; v < 10; v++) {
                    crawl.addF1((long) (v + 1) * 1000, v * 1.1);
                }
                crawlDAO.persist(crawl);
                Crawl fetched = crawlDAO.get(crawl.getKey(), 10);
                assertNotNull(fetched);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
        System.out.println(result);
    }

class SyncVsAsyncComparisonTest extends HBasePerfTestBase {

    @BeforeAll
    static void init() {
        PerfResultReporter.clear();
    }

    @AfterAll
    static void report() throws Exception {
        PerfResultReporter.writeReport(Path.of(System.getProperty("project.dir")));
    }

    @Test
    @Order(1)
    void singlePutGet() {
        BenchmarkResult syncResult = PerfResultReporter.run("compare.sync.single_put_get", 50, 500, () -> {
            try {
                Citizen c = new Citizen("CSYNC", 1, "CompareSync");
                citizenDAO.persist(c);
                citizenDAO.get("CSYNC#1");
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });

        BenchmarkResult asyncResult = PerfResultReporter.run("compare.async.single_put_get", 50, 500, () -> {
            try {
                Citizen c = new Citizen("CASYNC", 1, "CompareAsync");
                asyncCitizenDAO.persist(c).get(10, TimeUnit.SECONDS);
   

    static void init() {
        PerfResultReporter.clear();
    }

    static void report() throws Exception {
        PerfResultReporter.writeReport(Path.of(System.getProperty("project.dir")));
    }

    void singlePutGet() {
        BenchmarkResult syncResult = PerfResultReporter.run("compare.sync.single_put_get", 50, 500, () -> {
            try {
                Citizen c = new Citizen("CSYNC", 1, "CompareSync");
                citizenDAO.persist(c);
                citizenDAO.get("CSYNC#1");
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });

        BenchmarkResult asyncResult = PerfResultReporter.run("compare.async.single_put_get", 50, 500, () -> {
            try {
                Citizen c = new Citizen("CASYNC", 1, "CompareAsync");
                asyncCitizenDAO.persist(c).get(10, TimeUnit.SECONDS);
                asyncCitizenDAO.get("CASYNC#1").get(10, TimeUnit.SECONDS);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });

        System.out.println("=== Single Put+Get Comparison ===");
        System.out.println("  Sync:  " + syncResult);
        System.o

    void bulkPut100() {
        List<Citizen> batch = new ArrayList<>(100);
        for (int i = 0; i < 100; i++) {
            batch.add(new Citizen("CBULK", i, "Citizen-" + i));
        }

        BenchmarkResult syncResult = PerfResultReporter.run("compare.sync.bulk_put_100", 3, 20, () -> {
            try {
                citizenDAO.persist(batch);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });

        BenchmarkResult asyncResult = PerfResultReporter.run("compare.async.bulk_put_100", 3, 20, () -> {
            try {
                asyncCitizenDAO.persistAll(batch).get(30, TimeUnit.SECONDS);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });

        System.out.println("=== Bulk Put 100 Comparison ===");
        System.out.println("  Sync:  " + syncResult);
        System.out.println("  Async: " + asyncResult);
    }

    void prefixScan100() throws Exception {
        // Seed data
        List<Citizen> batch = new ArrayList<>(100);
        for (int i = 0; i < 100; i++) {
            batch.add(new Citizen("CPSCAN", i, "Citizen-" + i));
        }
        citizenDAO.persist(batch);

        BenchmarkResult syncResult = PerfResultReporter.run("compare.sync.prefix_scan_100", 3, 30, () -> {
            try {
                citizenDAO.getByPrefix(Bytes.toBytes("CPSCAN#"));
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });

        BenchmarkResult asyncResult = PerfResultReporter.run("compare.async.prefix_scan_100", 3, 30, () -> {
            try {
                asyncCitizenDAO.getByPrefix(Bytes.toBytes("CPSCAN#")).get(30, TimeUnit.SECONDS);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });

        System.out.println("=== Prefix Scan 100 Comparison ===");
        System.out.println("  S

    void bulkGet100() throws Exception {
        // Seed data
        List<Citizen> batch = new ArrayList<>(100);
        List<String> keys = new ArrayList<>(100);
        for (int i = 0; i < 100; i++) {
            batch.add(new Citizen("CBGET", i, "Citizen-" + i));
            keys.add("CBGET#" + i);
        }
        citizenDAO.persist(batch);

        BenchmarkResult syncResult = PerfResultReporter.run("compare.sync.bulk_get_100", 3, 30, () -> {
            try {
                citizenDAO.get(keys);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });

        BenchmarkResult asyncResult = PerfResultReporter.run("compare.async.bulk_get_100", 3, 30, () -> {
            try {
                asyncCitizenDAO.getAll(keys).get(30, TimeUnit.SECONDS);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });

        System.out.println("=== Bulk Get 100 Comparison ===");
        Sys