class HyperLogLogDistinctAggregationBenchmark {

    private final List<Row1> rows = IntStream.range(0, 10_000).mapToObj(i -> new Row1(String.valueOf(i))).toList();

    private HyperLogLogPlusPlus hyperLogLogPlusPlus;
    private AggregateCollector onHeapCollector;
    private OnHeapMemoryManager onHeapMemoryManager;
    private OffHeapMemoryManager offHeapMemoryManager;
    private AggregateCollector offHeapCollector;
    private MurmurHash3.Hash128 hash;

    @SuppressWarnings("unchecked")
    @Setup
    public void setUp() throws Exception {
        hash = new MurmurHash3.Hash128();
        final RowCollectExpression inExpr0 = new RowCollectExpression(0);
        Functions functions = Functions.load(Settings.EMPTY, new SessionSettingRegistry(Set.of()));
        final HyperLogLogDistinctAggregation hllAggregation = (HyperLogLogDistinctAggregation) functions.getQualified(
            Signature.aggregate(
                HyperLogLogDistinctAggregation.NAME,
                DataTypes.STRING.getTypeSignature(),
                DataTypes.LONG.getTypeSignature()
            ),
            List.of(DataTypes.STRING),
            DataTypes.STRING
        );
        onHeapMemoryManager = new OnHeapMemoryManager(bytes -> {});
        offHeapMemoryManager = new OffHeapMemoryManager();
        hyperLogLogPlusPlus = new HyperLogLogPlusPlus(HyperLogLogPlusPlus.DEFAULT_PRECISION, onHeapMemoryManager::allocate);
        onHeapCollector = new AggregateCollector(
            Collections.singletonList(inExpr0),
            RamAccounting.NO_ACCOUNTING,
            onHeapMemoryManager,
            Version.CURRENT,
            AggregateMode.ITER_FINAL,
            new AggregationFunction[] { hllAggregation },
            Version.CURRENT,
            new Input[][] { { inExpr0 } },
            new Input[] { Literal.BOOLEAN_TRUE }
        );
        offHeapCollector = new AggregateCollector(
            Collections.singletonList(inExpr0),
            RamAccounting.NO_ACCOUNTING,
            offHeapMemoryManager,
            Version.CURRENT,
            AggregateMode.ITER_FINAL,
            new AggregationFunction[] { hllAggregation },
            Version.CURRENT,
            new Input[][] { { inExpr0 } },
            new Input[] { Literal.BOOLEAN_TRUE }
        );
    }

    @TearDown
    public void tearDown() {
        onHeapMemoryManager.close();
        offHeapMemoryManager.close();
    }

    @Benchmark
    public long benchmarkHLLPlusPlusMurmur128() throws Exception {<FILL_FUNCTION_BODY>}

    @Benchmark
    public long benchmarkHLLPlusPlusMurmur64() throws Exception {
        for (int i = 0; i < rows.size(); i++) {
            String value = DataTypes.STRING.sanitizeValue(rows.get(i).get(0));
            byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
            hyperLogLogPlusPlus.collect(MurmurHash3.hash64(bytes, 0, bytes.length));
        }
        return hyperLogLogPlusPlus.cardinality();
    }

    @Benchmark
    public Iterable<Row> benchmarkHLLAggregationOnHeap() throws Exception {
        Object[] state = onHeapCollector.supplier().get();
        BiConsumer<Object[], Row> accumulator = onHeapCollector.accumulator();
        Function<Object[], Iterable<Row>> finisher = onHeapCollector.finisher();
        for (int i = 0; i < rows.size(); i++) {
            accumulator.accept(state, rows.get(i));
        }
        return finisher.apply(state);
    }

    @Benchmark
    public Iterable<Row> benchmarkHLLAggregationOffHeap() throws Exception {
        Object[] state = offHeapCollector.supplier().get();
        BiConsumer<Object[], Row> accumulator = offHeapCollector.accumulator();
        Function<Object[], Iterable<Row>> finisher = offHeapCollector.finisher();
        for (int i = 0; i < rows.size(); i++) {
            accumulator.accept(state, rows.get(i));
        }
        return finisher.apply(state);
    }
}

        for (int i = 0; i < rows.size(); i++) {
            String value = DataTypes.STRING.sanitizeValue(rows.get(i).get(0));
            byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
            MurmurHash3.Hash128 hash128 = MurmurHash3.hash128(bytes, 0, bytes.length, 0, hash);
            hyperLogLogPlusPlus.collect(hash128.h1);
        }
        return hyperLogLogPlusPlus.cardinality();
    
<no_super_class>