cardinality optimization

Author: zeromax007

roaringbitmap--0.2.65.1--0.2.65.2.sql

@@ -50,8 +50,13 @@ DROP FUNCTION IF EXISTS rb_cardinality(roaringbitmap, integer, integer, integer)
 CREATE
   OR REPLACE FUNCTION rb_cardinality(roaringbitmap, integer, integer, integer) RETURNS BIGINT AS 'MODULE_PATHNAME',
   'rb_cardinality_step' LANGUAGE C STRICT IMMUTABLE;
+DROP FUNCTION IF EXISTS rb_cardinality(roaringbitmap, integer, integer, integer, integer[]);
 CREATE
   OR REPLACE FUNCTION rb_cardinality(roaringbitmap, integer, integer, integer, integer[]) RETURNS BIGINT AS 'MODULE_PATHNAME',
+  'rb_cardinality_step_array' LANGUAGE C STRICT IMMUTABLE;
+DROP FUNCTION IF EXISTS rb_cardinality(roaringbitmap, integer, integer, integer, integer, integer);
+CREATE
+  OR REPLACE FUNCTION rb_cardinality(roaringbitmap, integer, integer, integer, integer, integer) RETURNS BIGINT AS 'MODULE_PATHNAME',
   'rb_cardinality_step_intval' LANGUAGE C STRICT IMMUTABLE;
 CREATE
   OR REPLACE FUNCTION rb_is_empty(roaringbitmap) RETURNS bool AS 'MODULE_PATHNAME',

roaringbitmap--0.2.65.2.sql

@@ -73,6 +73,9 @@ CREATE
   'rb_cardinality_step' LANGUAGE C STRICT IMMUTABLE;
 CREATE
   OR REPLACE FUNCTION rb_cardinality(roaringbitmap, integer, integer, integer, integer[]) RETURNS BIGINT AS 'MODULE_PATHNAME',
+  'rb_cardinality_step_array' LANGUAGE C STRICT IMMUTABLE;
+CREATE
+  OR REPLACE FUNCTION rb_cardinality(roaringbitmap, integer, integer, integer, integer, integer) RETURNS BIGINT AS 'MODULE_PATHNAME',
   'rb_cardinality_step_intval' LANGUAGE C STRICT IMMUTABLE;
 CREATE
   OR REPLACE FUNCTION rb_is_empty(roaringbitmap) RETURNS bool AS 'MODULE_PATHNAME',

roaringbitmap.c

@@ -389,23 +389,58 @@ Datum
     uint32_t rmin = roaring_bitmap_minimum(r1);
     uint32_t rmax = roaring_bitmap_maximum(r1);
 
-    max = max == -1 ? rmax : max;
+    max = max == -1 ? rmax : (rmax < max ? rmax : max);
+    uint32_t p = min;
+    int64 card1 = 0;
+    while (p <= max)
+    {
+        if (p + step >= rmin)
+        {
+            if (roaring_bitmap_range_cardinality(r1, p, p + step) > 0)
+                card1++;
+        }
+        p += step;
+    }
+
+    roaring_bitmap_free(r1);
+    PG_RETURN_INT64(card1);
+}
+
+//bitmap cardinality step interval
+PG_FUNCTION_INFO_V1(rb_cardinality_step_interval);
+Datum rb_cardinality_step_interval(PG_FUNCTION_ARGS);
+
+Datum
+    rb_cardinality_step_interval(PG_FUNCTION_ARGS)
+{
+    bytea *data = PG_GETARG_BYTEA_P(0);
+    uint32_t min = PG_GETARG_INT32(1);
+    uint32_t max = PG_GETARG_INT32(2);
+    int step = PG_GETARG_INT32(3);
+    uint32_t start = PG_GETARG_INT32(4);
+    uint32_t end = PG_GETARG_INT32(5);
+
+    if (step < 2)
+        ereport(ERROR, (errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE), errmsg("step must be greater than 1")));
+    if (end < start)
+        ereport(ERROR, (errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE), errmsg("range end must be greater than range start")));
+
+    roaring_bitmap_t *r1 = roaring_bitmap_portable_deserialize(VARDATA(data));
+    if (!r1)
+        ereport(ERROR, (errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED), errmsg("bitmap format is error")));
 
+    uint32_t rmin = roaring_bitmap_minimum(r1);
+    uint32_t rmax = roaring_bitmap_maximum(r1);
+
+    max = max == -1 ? rmax : (rmax < max ? rmax : max);
     uint32_t p = min;
-    uint32_t n;
     int64 card1 = 0;
     while (p <= max)
     {
         if (p + step >= rmin)
         {
-            for (n = 0; n < step; n++)
-            {
-                if (roaring_bitmap_contains(r1, n + p))
-                {
-                    card1++;
-                    break;
-                }
-            }
+            if (roaring_bitmap_range_cardinality(r1, p + start, p + end) > 0)
+                card1++;
         }
         p += step;
     }
@@ -414,12 +449,12 @@ Datum
     PG_RETURN_INT64(card1);
 }
 
-//bitmap cardinality step intval
-PG_FUNCTION_INFO_V1(rb_cardinality_step_intval);
-Datum rb_cardinality_step_intval(PG_FUNCTION_ARGS);
+//bitmap cardinality step array
+PG_FUNCTION_INFO_V1(rb_cardinality_step_array);
+Datum rb_cardinality_step_array(PG_FUNCTION_ARGS);
 
 Datum
-    rb_cardinality_step_intval(PG_FUNCTION_ARGS)
+    rb_cardinality_step_array(PG_FUNCTION_ARGS)
 {
     bytea *data = PG_GETARG_BYTEA_P(0);
     uint32_t min = PG_GETARG_INT32(1);