@ScalarFunction
    @SqlType(StandardTypes.BIGINT)
    public static long intersectionCardinality(@SqlType(KHyperLogLogType.NAME) Slice slice1, @SqlType(KHyperLogLogType.NAME) Slice slice2)
    {
        KHyperLogLog khll1 = KHyperLogLog.newInstance(slice1);
        KHyperLogLog khll2 = KHyperLogLog.newInstance(slice2);

        if (khll1.isExact() && khll2.isExact()) {
            return KHyperLogLog.exactIntersectionCardinality(khll1, khll2);
        }

        long lowestCardinality = Math.min(khll1.cardinality(), khll2.cardinality());
        double jaccard = KHyperLogLog.jaccardIndex(khll1, khll2);
        KHyperLogLog setUnion = KHyperLogLog.merge(khll1, khll2);
        long result = Math.round(jaccard * setUnion.cardinality());

        // When one of the sets is much smaller than the other and approaches being a true
        // subset of the other, the computed cardinality may exceed the cardinality estimate
        // of the smaller set. When this happens the cardinality of the smaller set is obviously
        // a better estimate of the one computed with the Jaccard Index.
        return Math.min(result, lowestCardinality);
    }
@Test
    public void testIntersectionCardinality()
    {
        int blockSize = 10;
        long uniqueElements = 10000 * blockSize;
        double error = uniqueElements * 0.05;

        List<KHyperLogLog> list1 = buildKHyperLogLogs(blockSize, uniqueElements, threshold, potential);

        List<KHyperLogLog> list2 = buildKHyperLogLogs(15, (uniqueElements * 15) / blockSize, threshold, potential);

        String projection = getIntersectionCardinalityProjection(list1, list2);

        functionAssertions.assertFunctionWithError(projection, BIGINT, uniqueElements, error);
    }