[FLINK-39187][table] Add core built-in aggregate functions for BITMAP

docs/data/sql_functions.yml

@@ -1366,6 +1366,22 @@ bitmap:
       `bitmap1 BITMAP, bitmap2 BITMAP`
 
       Returns a `BITMAP`. `NULL` if any of the arguments are `NULL`.
+  - sql: BITMAP_AND_AGG(bitmap)
+    table: bitmap.bitmapAndAgg()
+    description: |
+      Aggregates the AND (intersection) of multiple bitmaps.
+
+      `bitmap BITMAP`
+
+      Returns a `BITMAP`.
+  - sql: BITMAP_AND_CARDINALITY_AGG(bitmap)
+    table: bitmap.bitmapAndCardinalityAgg()
+    description: |
+      Aggregates the AND (intersection) of multiple bitmaps and returns its 64-bit cardinality.
+
+      `bitmap BITMAP`
+
+      Returns a `BIGINT`.
   - sql: BITMAP_BUILD(array)
     table: array.bitmapBuild()
     description: |
@@ -1374,6 +1390,22 @@ bitmap:
       `array ARRAY<INT>`
 
       Returns a `BITMAP`. `NULL` if the argument is `NULL`.
+  - sql: BITMAP_BUILD_AGG(value)
+    table: value.bitmapBuildAgg()
+    description: |
+      Aggregates 32-bit integers into a bitmap.
+
+      `value INT`
+
+      Returns a `BITMAP`.
+  - sql: BITMAP_BUILD_CARDINALITY_AGG(value)
+    table: value.bitmapBuildCardinalityAgg()
+    description: |
+      Aggregates 32-bit integers into a bitmap and returns its 64-bit cardinality.
+
+      `value INT`
+
+      Returns a `BIGINT`.
   - sql: BITMAP_CARDINALITY(bitmap)
     table: bitmap.bitmapCardinality()
     description: |
@@ -1400,6 +1432,22 @@ bitmap:
       `bitmap1 BITMAP, bitmap2 BITMAP`
 
       Returns a `BITMAP`. `NULL` if any of the arguments are `NULL`.
+  - sql: BITMAP_OR_AGG(bitmap)
+    table: bitmap.bitmapOrAgg()
+    description: |
+      Aggregates the OR (union) of multiple bitmaps.
+
+      `bitmap BITMAP`
+
+      Returns a `BITMAP`.
+  - sql: BITMAP_OR_CARDINALITY_AGG(bitmap)
+    table: bitmap.bitmapOrCardinalityAgg()
+    description: |
+      Aggregates the OR (union) of multiple bitmaps and returns its 64-bit cardinality.
+
+      `bitmap BITMAP`
+
+      Returns a `BIGINT`.
   - sql: BITMAP_TO_ARRAY(bitmap)
     table: bitmap.bitmapToArray()
     description: |
@@ -1440,6 +1488,22 @@ bitmap:
       `bitmap1 BITMAP, bitmap2 BITMAP`
 
       Returns a `BITMAP`. `NULL` if any of the arguments are `NULL`.
+  - sql: BITMAP_XOR_AGG(bitmap)
+    table: bitmap.bitmapXorAgg()
+    description: |
+      Aggregates the XOR (symmetric difference) of multiple bitmaps.
+
+      `bitmap BITMAP`
+
+      Returns a `BITMAP`.
+  - sql: BITMAP_XOR_CARDINALITY_AGG(bitmap)
+    table: bitmap.bitmapXorCardinalityAgg()
+    description: |
+      Aggregates the XOR (symmetric difference) of multiple bitmaps and returns its 64-bit cardinality.
+
+      `bitmap BITMAP`
+
+      Returns a `BIGINT`.
 
 auxilary:
   - table: callSql(STRING)

docs/data/sql_functions_zh.yml

@@ -1439,19 +1439,35 @@ bitmap:
   - sql: BITMAP_AND(bitmap1, bitmap2)
     table: bitmap1.bitmapAnd(bitmap2)
     description: |
-      计算两个位图的交集 (AND)。
+      计算两个位图的交集（AND）。
 
       `bitmap1 BITMAP, bitmap2 BITMAP`
 
       返回一个 `BITMAP`。如果任一参数为 `NULL`，则返回 `NULL`。
   - sql: BITMAP_ANDNOT(bitmap1, bitmap2)
     table: bitmap1.bitmapAndnot(bitmap2)
     description: |
-      计算两个位图的差集 (AND NOT)。
+      计算两个位图的差集（AND NOT）。
 
       `bitmap1 BITMAP, bitmap2 BITMAP`
 
       返回一个 `BITMAP`。如果任一参数为 `NULL`，则返回 `NULL`。
+  - sql: BITMAP_AND_AGG(bitmap)
+    table: bitmap.bitmapAndAgg()
+    description: |
+      聚合多个位图的交集（AND）。
+
+      `bitmap BITMAP`
+
+      返回一个 `BITMAP`。
+  - sql: BITMAP_AND_CARDINALITY_AGG(bitmap)
+    table: bitmap.bitmapAndCardinalityAgg()
+    description: |
+      聚合多个位图的交集（AND）并返回其 64 位基数。
+
+      `bitmap BITMAP`
+
+      返回一个 `BIGINT`。
   - sql: BITMAP_BUILD(array)
     table: array.bitmapBuild()
     description: |
@@ -1460,6 +1476,22 @@ bitmap:
       `array ARRAY<INT>`
 
       返回一个 `BITMAP`。如果参数为 `NULL`，则返回 `NULL`。
+  - sql: BITMAP_BUILD_AGG(value)
+    table: value.bitmapBuildAgg()
+    description: |
+      将 32 位整数聚合成位图。
+
+      `value INT`
+
+      返回一个 `BITMAP`。
+  - sql: BITMAP_BUILD_CARDINALITY_AGG(value)
+    table: value.bitmapBuildCardinalityAgg()
+    description: |
+      将 32 位整数聚合成位图并返回其 64 位基数。
+
+      `value INT`
+
+      返回一个 `BIGINT`。
   - sql: BITMAP_CARDINALITY(bitmap)
     table: bitmap.bitmapCardinality()
     description: |
@@ -1481,11 +1513,27 @@ bitmap:
   - sql: BITMAP_OR(bitmap1, bitmap2)
     table: bitmap1.bitmapOr(bitmap2)
     description: |
-      计算两个位图的并集 (OR)。
+      计算两个位图的并集（OR）。
 
       `bitmap1 BITMAP, bitmap2 BITMAP`
 
       返回一个 `BITMAP`。如果任一参数为 `NULL`，则返回 `NULL`。
+  - sql: BITMAP_OR_AGG(bitmap)
+    table: bitmap.bitmapOrAgg()
+    description: |
+      聚合多个位图的并集（OR）。
+
+      `bitmap BITMAP`
+
+      返回一个 `BITMAP`。
+  - sql: BITMAP_OR_CARDINALITY_AGG(bitmap)
+    table: bitmap.bitmapOrCardinalityAgg()
+    description: |
+      聚合多个位图的并集（OR）并返回其 64 位基数。
+
+      `bitmap BITMAP`
+
+      返回一个 `BIGINT`。
   - sql: BITMAP_TO_ARRAY(bitmap)
     table: bitmap.bitmapToArray()
     description: |
@@ -1521,11 +1569,27 @@ bitmap:
   - sql: BITMAP_XOR(bitmap1, bitmap2)
     table: bitmap1.bitmapXor(bitmap2)
     description: |
-      计算两个位图的异或 (XOR)。
+      计算两个位图的异或（XOR）。
 
       `bitmap1 BITMAP, bitmap2 BITMAP`
 
       返回一个 `BITMAP`。如果任一参数为 `NULL`，则返回 `NULL`。
+  - sql: BITMAP_XOR_AGG(bitmap)
+    table: bitmap.bitmapXorAgg()
+    description: |
+      聚合多个位图的异或（XOR）。
+
+      `bitmap BITMAP`
+
+      返回一个 `BITMAP`。
+  - sql: BITMAP_XOR_CARDINALITY_AGG(bitmap)
+    table: bitmap.bitmapXorCardinalityAgg()
+    description: |
+      聚合多个位图的异或（XOR）并返回其 64 位基数。
+
+      `bitmap BITMAP`
+
+      返回一个 `BIGINT`。
 
 auxilary:
   - table: callSql(STRING)

flink-core/src/main/java/org/apache/flink/types/bitmap/RoaringBitmapData.java

@@ -20,14 +20,14 @@
 
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.types.DeserializationException;
-import org.apache.flink.util.Preconditions;
 
 import org.roaringbitmap.IntConsumer;
 import org.roaringbitmap.RoaringBitmap;
 
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
+import java.io.Serializable;
 import java.nio.ByteBuffer;
 import java.util.Objects;
 
@@ -36,7 +36,9 @@
  * modifications.
  */
 @Internal
-public final class RoaringBitmapData implements Bitmap {
+public final class RoaringBitmapData implements Bitmap, Serializable {
+
+    private static final long serialVersionUID = 1L;
 
     private final RoaringBitmap roaringBitmap;
 
@@ -48,10 +50,6 @@ private RoaringBitmapData(RoaringBitmapData other) {
         this.roaringBitmap = other.roaringBitmap.clone();
     }
 
-    private RoaringBitmapData(RoaringBitmap roaringBitmap) {
-        this.roaringBitmap = roaringBitmap;
-    }
-
     // ~ Static Methods ----------------------------------------------------------------
 
     public static RoaringBitmapData empty() {
@@ -79,15 +77,6 @@ public static RoaringBitmapData fromArray(@Nonnull int[] values) {
         return rb32;
     }
 
-    /**
-     * Wraps the given {@link RoaringBitmap} without copying. The returned {@link RoaringBitmapData}
-     * shares the same internal object as the input.
-     */
-    public static RoaringBitmapData wrap(@Nonnull RoaringBitmap roaringBitmap) {
-        Preconditions.checkNotNull(roaringBitmap);
-        return new RoaringBitmapData(roaringBitmap);
-    }
-
     private static RoaringBitmapData toRoaringBitmapData(Bitmap bm)
             throws IllegalArgumentException {
         if (!(bm instanceof RoaringBitmapData)) {

flink-core/src/test/java/org/apache/flink/types/RoaringBitmapDataTest.java

@@ -66,12 +66,6 @@ void testStaticConstructors() {
         // fromArray
         assertThat(RoaringBitmapData.fromArray(new int[0]).toArray()).containsExactly();
         assertThat(RoaringBitmapData.fromArray(new int[] {1, 2}).toArray()).containsExactly(1, 2);
-
-        // wrap
-        assertThatThrownBy(() -> RoaringBitmapData.wrap(null))
-                .isInstanceOf(NullPointerException.class);
-        assertThat(RoaringBitmapData.wrap(RoaringBitmap.bitmapOf(1, 2)).toArray())
-                .containsExactly(1, 2);
     }
 
     @Test

flink-python/docs/reference/pyflink.table/expressions.rst

@@ -347,11 +347,19 @@ Bitmap functions
 
     Expression.bitmap_and
     Expression.bitmap_andnot
+    Expression.bitmap_and_agg
+    Expression.bitmap_and_cardinality_agg
     Expression.bitmap_build
+    Expression.bitmap_build_agg
+    Expression.bitmap_build_cardinality_agg
     Expression.bitmap_cardinality
     Expression.bitmap_from_bytes
     Expression.bitmap_or
+    Expression.bitmap_or_agg
+    Expression.bitmap_or_cardinality_agg
     Expression.bitmap_to_array
     Expression.bitmap_to_bytes
     Expression.bitmap_to_string
     Expression.bitmap_xor
+    Expression.bitmap_xor_agg
+    Expression.bitmap_xor_cardinality_agg

flink-python/pyflink/table/expression.py

@@ -2299,6 +2299,22 @@ def bitmap_andnot(self, bitmap2) -> 'Expression':
         """
         return _binary_op("bitmapAndnot")(self, bitmap2)
 
+    def bitmap_and_agg(self):
+        """
+        Aggregates the AND (intersection) of multiple bitmaps.
+
+        :return: a BITMAP expression
+        """
+        return _unary_op("bitmapAndAgg")(self)
+
+    def bitmap_and_cardinality_agg(self):
+        """
+        Aggregates the AND (intersection) of multiple bitmaps and returns its 64-bit cardinality.
+
+        :return: a BIGINT expression
+        """
+        return _unary_op("bitmapAndCardinalityAgg")(self)
+
     def bitmap_build(self) -> 'Expression':
         """
         Creates a bitmap from an array of 32-bit integers.
@@ -2309,6 +2325,22 @@ def bitmap_build(self) -> 'Expression':
         """
         return _unary_op("bitmapBuild")(self)
 
+    def bitmap_build_agg(self):
+        """
+        Aggregates 32-bit integers into a bitmap.
+
+        :return: a BITMAP expression
+        """
+        return _unary_op("bitmapBuildAgg")(self)
+
+    def bitmap_build_cardinality_agg(self):
+        """
+        Aggregates 32-bit integers into a bitmap and returns its 64-bit cardinality.
+
+        :return: a BIGINT expression
+        """
+        return _unary_op("bitmapBuildCardinalityAgg")(self)
+
     def bitmap_cardinality(self) -> 'Expression':
         """
         Returns the cardinality of a bitmap.
@@ -2343,6 +2375,22 @@ def bitmap_or(self, bitmap2) -> 'Expression':
         """
         return _binary_op("bitmapOr")(self, bitmap2)
 
+    def bitmap_or_agg(self):
+        """
+        Aggregates the OR (union) of multiple bitmaps.
+
+        :return: a BITMAP expression
+        """
+        return _unary_op("bitmapOrAgg")(self)
+
+    def bitmap_or_cardinality_agg(self):
+        """
+        Aggregates the OR (union) of multiple bitmaps and returns its 64-bit cardinality.
+
+        :return: a BIGINT expression
+        """
+        return _unary_op("bitmapOrCardinalityAgg")(self)
+
     def bitmap_to_array(self) -> 'Expression':
         """
         Converts a bitmap to an array of 32-bit integers, the values are sorted by \
@@ -2395,6 +2443,23 @@ def bitmap_xor(self, bitmap2) -> 'Expression':
         """
         return _binary_op("bitmapXor")(self, bitmap2)
 
+    def bitmap_xor_agg(self):
+        """
+        Aggregates the XOR (symmetric difference) of multiple bitmaps.
+
+        :return: a BITMAP expression
+        """
+        return _unary_op("bitmapXorAgg")(self)
+
+    def bitmap_xor_cardinality_agg(self):
+        """
+        Aggregates the XOR (symmetric difference) of multiple bitmaps and returns its 64-bit
+        cardinality.
+
+        :return: a BIGINT expression
+        """
+        return _unary_op("bitmapXorCardinalityAgg")(self)
+
 
 # add the docs
 _make_math_log_doc()