[FLINK-39186][table] Add core built-in scalar functions for BITMAP

docs/content.zh/docs/sql/functions/built-in-functions.md

@@ -107,6 +107,10 @@ JSON 函数使用符合 ISO/IEC TR 19075-6 SQL标准的 JSON 路径表达式。
 
 {{< sql_functions_zh "hashfunctions" >}}
 
+### 位图函数
+
+{{< sql_functions_zh "bitmap" >}}
+
 ### 辅助函数
 
 {{< sql_functions_zh "auxiliary" >}}

docs/content/docs/sql/functions/built-in-functions.md

@@ -110,6 +110,10 @@ Known Limitations:
 
 {{< sql_functions "hashfunctions" >}}
 
+### Bitmap Functions
+
+{{< sql_functions "bitmap" >}}
+
 ### Auxiliary Functions
 
 {{< sql_functions "auxiliary" >}}

docs/data/sql_functions.yml

@@ -1349,6 +1349,98 @@ hashfunctions:
     table: STRING.sha2(INT)
     description: Returns the hash using the SHA-2 family of hash functions (SHA-224, SHA-256, SHA-384, or SHA-512). The first argument string is the string to be hashed and the second argument hashLength is the bit length of the result (224, 256, 384, or 512). Returns NULL if string or hashLength is NULL.
 
+bitmap:
+  - sql: BITMAP_AND(bitmap1, bitmap2)
+    table: bitmap1.bitmapAnd(bitmap2)
+    description: |
+      Computes the AND (intersection) of two bitmaps.
+
+      `bitmap1 BITMAP, bitmap2 BITMAP`
+
+      Returns a `BITMAP`. `NULL` if any of the arguments are `NULL`.
+  - sql: BITMAP_ANDNOT(bitmap1, bitmap2)
+    table: bitmap1.bitmapAndnot(bitmap2)
+    description: |
+      Computes the AND NOT (difference) of two bitmaps.
+
+      `bitmap1 BITMAP, bitmap2 BITMAP`
+
+      Returns a `BITMAP`. `NULL` if any of the arguments are `NULL`.
+  - sql: BITMAP_BUILD(array)
+    table: array.bitmapBuild()
+    description: |
+      Creates a bitmap from an array of 32-bit integers.
+
+      `array ARRAY<INT>`
+
+      Returns a `BITMAP`. `NULL` if the argument is `NULL`.
+  - sql: BITMAP_CARDINALITY(bitmap)
+    table: bitmap.bitmapCardinality()
+    description: |
+      Returns the cardinality of a bitmap.
+
+      `bitmap BITMAP`
+
+      Returns a `BIGINT`. `NULL` if the argument is `NULL`.
+  - sql: BITMAP_FROM_BYTES(bytes)
+    table: bytes.bitmapFromBytes()
+    description: |
+      Converts an array of bytes to a bitmap.
+
+      Following the format defined in [32-bit RoaringBitmap format specification](https://github.com/RoaringBitmap/RoaringFormatSpec).
+
+      `bytes <BINARY | VARBINARY>`
+
+      Returns a `BITMAP`. `NULL` if the argument is `NULL`.
+  - sql: BITMAP_OR(bitmap1, bitmap2)
+    table: bitmap1.bitmapOr(bitmap2)
+    description: |
+      Computes the OR (union) of two bitmaps.
+
+      `bitmap1 BITMAP, bitmap2 BITMAP`
+
+      Returns a `BITMAP`. `NULL` if any of the arguments are `NULL`.
+  - sql: BITMAP_TO_ARRAY(bitmap)
+    table: bitmap.bitmapToArray()
+    description: |
+      Converts a bitmap to an array of 32-bit integers, the values are sorted by `Integer.compareUnsigned`.
+
+      `bitmap BITMAP`
+
+      Returns an `ARRAY<INT>`. `NULL` if the argument is `NULL`.
+  - sql: BITMAP_TO_BYTES(bitmap)
+    table: bitmap.bitmapToBytes()
+    description: |
+      Converts a bitmap to an array of bytes.
+
+      Following the format defined in [32-bit RoaringBitmap format specification](https://github.com/RoaringBitmap/RoaringFormatSpec).
+
+      `bitmap BITMAP`
+
+      Returns a `VARBINARY`. `NULL` if the argument is `NULL`.
+  - sql: BITMAP_TO_STRING(bitmap)
+    table: bitmap.bitmapToString()
+    description: |
+      Converts a bitmap to a string, the values are sorted by `Integer.compareUnsigned`. The string will be truncated and end with "..." if it is too long.
+
+      For example:
+
+      - `"{}"`, `"{1,2,3,4,5}"`
+      - Negative values (converted to unsigned): `"{0,1,4294967294,4294967295}"`
+      - String too long: `"{1,2,3,...}"`
+
+      `bitmap BITMAP`
+
+      Returns a `STRING`. `NULL` if the argument is `NULL`.
+  - sql: BITMAP_XOR(bitmap1, bitmap2)
+    table: bitmap1.bitmapXor(bitmap2)
+    description: |
+      Computes the XOR (symmetric difference) of two bitmaps.
+
+      `bitmap1 BITMAP, bitmap2 BITMAP`
+
+      Returns a `BITMAP`. `NULL` if any of the arguments are `NULL`.
+
 auxilary:
   - table: callSql(STRING)
     description: |

docs/data/sql_functions_zh.yml

@@ -1435,6 +1435,98 @@ hashfunctions:
       使用 SHA-2 系列散列函数（SHA-224，SHA-256，SHA-384 或 SHA-512）返回散列值。第一个参数 string 是要散列的字符串，
       第二个参数 hashLength 是结果的位数（224，256，384 或 512）。如果 string 或 hashLength 为 `NULL`，则返回 `NULL`。
 
+bitmap:
+  - sql: BITMAP_AND(bitmap1, bitmap2)
+    table: bitmap1.bitmapAnd(bitmap2)
+    description: |
+      计算两个位图的交集 (AND)。
+
+      `bitmap1 BITMAP, bitmap2 BITMAP`
+
+      返回一个 `BITMAP`。如果任一参数为 `NULL`，则返回 `NULL`。
+  - sql: BITMAP_ANDNOT(bitmap1, bitmap2)
+    table: bitmap1.bitmapAndnot(bitmap2)
+    description: |
+      计算两个位图的差集 (AND NOT)。
+
+      `bitmap1 BITMAP, bitmap2 BITMAP`
+
+      返回一个 `BITMAP`。如果任一参数为 `NULL`，则返回 `NULL`。
+  - sql: BITMAP_BUILD(array)
+    table: array.bitmapBuild()
+    description: |
+      从 32 位整数数组创建位图。
+
+      `array ARRAY<INT>`
+
+      返回一个 `BITMAP`。如果参数为 `NULL`，则返回 `NULL`。
+  - sql: BITMAP_CARDINALITY(bitmap)
+    table: bitmap.bitmapCardinality()
+    description: |
+      返回位图的基数。
+
+      `bitmap BITMAP`
+
+      返回一个 `BIGINT`。如果参数为 `NULL`，则返回 `NULL`。
+  - sql: BITMAP_FROM_BYTES(bytes)
+    table: bytes.bitmapFromBytes()
+    description: |
+      将字节数组转换为位图。
+
+      遵循 [32-bit RoaringBitmap format specification](https://github.com/RoaringBitmap/RoaringFormatSpec) 定义的格式。
+
+      `bytes <BINARY | VARBINARY>`
+
+      返回一个 `BITMAP`。如果参数为 `NULL`，则返回 `NULL`。
+  - sql: BITMAP_OR(bitmap1, bitmap2)
+    table: bitmap1.bitmapOr(bitmap2)
+    description: |
+      计算两个位图的并集 (OR)。
+
+      `bitmap1 BITMAP, bitmap2 BITMAP`
+
+      返回一个 `BITMAP`。如果任一参数为 `NULL`，则返回 `NULL`。
+  - sql: BITMAP_TO_ARRAY(bitmap)
+    table: bitmap.bitmapToArray()
+    description: |
+      将位图转换为 32 位整数数组，值按 `Integer.compareUnsigned` 排序。
+
+      `bitmap BITMAP`
+
+      返回一个 `ARRAY<INT>`。如果参数为 `NULL`，则返回 `NULL`。
+  - sql: BITMAP_TO_BYTES(bitmap)
+    table: bitmap.bitmapToBytes()
+    description: |
+      将位图转换为字节数组。
+
+      遵循 [32-bit RoaringBitmap format specification](https://github.com/RoaringBitmap/RoaringFormatSpec) 定义的格式。
+
+      `bitmap BITMAP`
+
+      返回一个 `VARBINARY`。如果参数为 `NULL`，则返回 `NULL`。
+  - sql: BITMAP_TO_STRING(bitmap)
+    table: bitmap.bitmapToString()
+    description: |
+      将位图转换为字符串，值按 `Integer.compareUnsigned` 排序。如果字符串过长，将被截断并以 "..." 结尾。
+
+      例如：
+
+      - `"{}"`, `"{1,2,3,4,5}"`
+      - 负值（转换为无符号）: `"{0,1,4294967294,4294967295}"`
+      - 字符串过长: `"{1,2,3,...}"`
+
+      `bitmap BITMAP`
+
+      返回一个 `STRING`。如果参数为 `NULL`，则返回 `NULL`。
+  - sql: BITMAP_XOR(bitmap1, bitmap2)
+    table: bitmap1.bitmapXor(bitmap2)
+    description: |
+      计算两个位图的异或 (XOR)。
+
+      `bitmap1 BITMAP, bitmap2 BITMAP`
+
+      返回一个 `BITMAP`。如果任一参数为 `NULL`，则返回 `NULL`。
+
 auxilary:
   - table: callSql(STRING)
     description: |

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

@@ -336,3 +336,22 @@ value modification functions
     :toctree: api/
 
     Expression.object_update
+
+Bitmap functions
+----------------
+
+.. currentmodule:: pyflink.table.expression
+
+.. autosummary::
+    :toctree: api/
+
+    Expression.bitmap_and
+    Expression.bitmap_andnot
+    Expression.bitmap_build
+    Expression.bitmap_cardinality
+    Expression.bitmap_from_bytes
+    Expression.bitmap_or
+    Expression.bitmap_to_array
+    Expression.bitmap_to_bytes
+    Expression.bitmap_to_string
+    Expression.bitmap_xor

flink-python/pyflink/table/expression.py

@@ -2275,6 +2275,126 @@ def object_update(self, *kv) -> "Expression":
         """
         return _varargs_op("objectUpdate")(self, *kv)
 
+    # ---------------------------- Bitmap functions -----------------------------
+
+    def bitmap_and(self, bitmap2) -> 'Expression':
+        """
+        Computes the AND (intersection) of two bitmaps.
+
+        If any of the inputs are null, the result is null.
+
+        :param bitmap2: the bitmap to perform AND operation with
+        :return: a BITMAP expression
+        """
+        return _binary_op("bitmapAnd")(self, bitmap2)
+
+    def bitmap_andnot(self, bitmap2) -> 'Expression':
+        """
+        Computes the AND NOT (difference) of two bitmaps.
+
+        If any of the inputs are null, the result is null.
+
+        :param bitmap2: the bitmap to perform AND NOT operation with
+        :return: a BITMAP expression
+        """
+        return _binary_op("bitmapAndnot")(self, bitmap2)
+
+    def bitmap_build(self) -> 'Expression':
+        """
+        Creates a bitmap from an array of 32-bit integers.
+
+        If the input is null, the result is null.
+
+        :return: a BITMAP expression
+        """
+        return _unary_op("bitmapBuild")(self)
+
+    def bitmap_cardinality(self) -> 'Expression':
+        """
+        Returns the cardinality of a bitmap.
+
+        If the input is null, the result is null.
+
+        :return: a BIGINT expression
+        """
+        return _unary_op("bitmapCardinality")(self)
+
+    def bitmap_from_bytes(self) -> 'Expression':
+        """
+        Converts an array of bytes to a bitmap.
+
+        Following the format defined in `32-bit RoaringBitmap format specification \
+        <https://github.com/RoaringBitmap/RoaringFormatSpec>`_.
+
+        If the input is null, the result is null.
+
+        :return: a BITMAP expression
+        """
+        return _unary_op("bitmapFromBytes")(self)
+
+    def bitmap_or(self, bitmap2) -> 'Expression':
+        """
+        Computes the OR (union) of two bitmaps.
+
+        If any of the inputs are null, the result is null.
+
+        :param bitmap2: the bitmap to perform OR operation with
+        :return: a BITMAP expression
+        """
+        return _binary_op("bitmapOr")(self, bitmap2)
+
+    def bitmap_to_array(self) -> 'Expression':
+        """
+        Converts a bitmap to an array of 32-bit integers, the values are sorted by \
+        :py:meth:`Integer.compareUnsigned`.
+
+        If the input is null, the result is null.
+
+        :return: an ARRAY<INT> expression
+        """
+        return _unary_op("bitmapToArray")(self)
+
+    def bitmap_to_bytes(self) -> 'Expression':
+        """
+        Converts a bitmap to an array of bytes.
+
+        Following the format defined in `32-bit RoaringBitmap format specification \
+        <https://github.com/RoaringBitmap/RoaringFormatSpec>`_.
+
+        If the input is null, the result is null.
+
+        :return: a VARBINARY expression
+        """
+        return _unary_op("bitmapToBytes")(self)
+
+    def bitmap_to_string(self) -> 'Expression':
+        """
+        Converts a bitmap to a string, the values are sorted by `Integer.compareUnsigned` in Java.
+        The string will be truncated and end with "..." if it is too long.
+
+        For example:
+
+        - ``"{}"``, ``"{1,2,3,4,5}"``
+        - Negative values (converted to unsigned): ``"{0,1,4294967294,4294967295}"``
+        - String too long: ``"{1,2,3,...}"``
+
+        If the input is null, the result is null.
+
+        :return: a STRING expression
+        """
+        return _unary_op("bitmapToString")(self)
+
+    def bitmap_xor(self, bitmap2) -> 'Expression':
+        """
+        Computes the XOR (symmetric difference) of two bitmaps.
+
+        If any of the inputs are null, the result is null.
+
+        :param bitmap2: the bitmap to perform XOR operation with
+        :return: a BITMAP expression
+        """
+        return _binary_op("bitmapXor")(self, bitmap2)
+
 
 # add the docs
 _make_math_log_doc()

flink-python/pyflink/table/tests/test_expression.py

@@ -265,6 +265,18 @@ def test_expression(self):
                                                   JsonQueryOnEmptyOrError.NULL,
                                                   JsonQueryOnEmptyOrError.EMPTY_ARRAY)))
 
+        # bitmap functions
+        self.assertEqual("BITMAP_AND(a, b)", str(expr1.bitmap_and(expr2)))
+        self.assertEqual("BITMAP_ANDNOT(a, b)", str(expr1.bitmap_andnot(expr2)))
+        self.assertEqual("BITMAP_BUILD(a)", str(expr1.bitmap_build()))
+        self.assertEqual("BITMAP_CARDINALITY(a)", str(expr1.bitmap_cardinality()))
+        self.assertEqual("BITMAP_FROM_BYTES(a)", str(expr1.bitmap_from_bytes()))
+        self.assertEqual("BITMAP_OR(a, b)", str(expr1.bitmap_or(expr2)))
+        self.assertEqual("BITMAP_TO_ARRAY(a)", str(expr1.bitmap_to_array()))
+        self.assertEqual("BITMAP_TO_BYTES(a)", str(expr1.bitmap_to_bytes()))
+        self.assertEqual("BITMAP_TO_STRING(a)", str(expr1.bitmap_to_string()))
+        self.assertEqual("BITMAP_XOR(a, b)", str(expr1.bitmap_xor(expr2)))
+
     def test_expressions(self):
         expr1 = col('a')
         expr2 = col('b')