This is an automated email from the ASF dual-hosted git repository.

lincoln pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit 198816c4115ee7a6a588d8763ca37038b451d150
Author: dylanhz <[email protected]>
AuthorDate: Tue Mar 24 01:47:50 2026 +0800

    [FLINK-39186][table] Add the built-in function BITMAP_TO_ARRAY, 
BITMAP_TO_STRING
---
 docs/data/sql_functions.yml                        |  22 ++++
 docs/data/sql_functions_zh.yml                     |  22 ++++
 .../docs/reference/pyflink.table/expressions.rst   |   2 +
 flink-python/pyflink/table/expression.py           |  28 +++++
 .../pyflink/table/tests/test_expression.py         |   2 +
 .../flink/table/api/internal/BaseExpressions.java  |  34 ++++++
 .../functions/BuiltInFunctionDefinitions.java      |  26 ++++
 .../planner/functions/BitmapFunctionsITCase.java   | 132 ++++++++++++++++++++-
 .../functions/scalar/BitmapToArrayFunction.java    |  50 ++++++++
 .../functions/scalar/BitmapToStringFunction.java   |  43 +++++++
 10 files changed, 360 insertions(+), 1 deletion(-)

diff --git a/docs/data/sql_functions.yml b/docs/data/sql_functions.yml
index 3b5e5a48d25..0259592275d 100644
--- a/docs/data/sql_functions.yml
+++ b/docs/data/sql_functions.yml
@@ -1376,6 +1376,14 @@ bitmap:
       `bytes <BINARY | VARBINARY>`
       
       Returns a `BITMAP`. `NULL` if the argument is `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: |
@@ -1386,6 +1394,20 @@ bitmap:
       `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`.
 
 auxilary:
   - table: callSql(STRING)
diff --git a/docs/data/sql_functions_zh.yml b/docs/data/sql_functions_zh.yml
index 9cf9602a26c..13a63f4df46 100644
--- a/docs/data/sql_functions_zh.yml
+++ b/docs/data/sql_functions_zh.yml
@@ -1462,6 +1462,14 @@ bitmap:
       `bytes <BINARY | VARBINARY>`
       
       返回一个 `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: |
@@ -1472,6 +1480,20 @@ bitmap:
       `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`。
 
 auxilary:
   - table: callSql(STRING)
diff --git a/flink-python/docs/reference/pyflink.table/expressions.rst 
b/flink-python/docs/reference/pyflink.table/expressions.rst
index 21a70cb62e8..480631ee554 100644
--- a/flink-python/docs/reference/pyflink.table/expressions.rst
+++ b/flink-python/docs/reference/pyflink.table/expressions.rst
@@ -348,4 +348,6 @@ Bitmap functions
     Expression.bitmap_build
     Expression.bitmap_cardinality
     Expression.bitmap_from_bytes
+    Expression.bitmap_to_array
     Expression.bitmap_to_bytes
+    Expression.bitmap_to_string
diff --git a/flink-python/pyflink/table/expression.py 
b/flink-python/pyflink/table/expression.py
index 5d9c766f4c8..81f28c76999 100644
--- a/flink-python/pyflink/table/expression.py
+++ b/flink-python/pyflink/table/expression.py
@@ -2310,6 +2310,17 @@ class Expression(Generic[T]):
         """
         return _unary_op("bitmapFromBytes")(self)
 
+    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.
@@ -2323,6 +2334,23 @@ class Expression(Generic[T]):
         """
         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)
+
 
 # add the docs
 _make_math_log_doc()
diff --git a/flink-python/pyflink/table/tests/test_expression.py 
b/flink-python/pyflink/table/tests/test_expression.py
index 1db5b8af8cd..9fffd470ebf 100644
--- a/flink-python/pyflink/table/tests/test_expression.py
+++ b/flink-python/pyflink/table/tests/test_expression.py
@@ -269,7 +269,9 @@ class PyFlinkBatchExpressionTests(PyFlinkTestCase):
         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_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()))
 
     def test_expressions(self):
         expr1 = col('a')
diff --git 
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
 
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
index 07f4e85837c..6e49f0190b8 100644
--- 
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
+++ 
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
@@ -79,7 +79,9 @@ import static 
org.apache.flink.table.functions.BuiltInFunctionDefinitions.BIN;
 import static 
org.apache.flink.table.functions.BuiltInFunctionDefinitions.BITMAP_BUILD;
 import static 
org.apache.flink.table.functions.BuiltInFunctionDefinitions.BITMAP_CARDINALITY;
 import static 
org.apache.flink.table.functions.BuiltInFunctionDefinitions.BITMAP_FROM_BYTES;
+import static 
org.apache.flink.table.functions.BuiltInFunctionDefinitions.BITMAP_TO_ARRAY;
 import static 
org.apache.flink.table.functions.BuiltInFunctionDefinitions.BITMAP_TO_BYTES;
+import static 
org.apache.flink.table.functions.BuiltInFunctionDefinitions.BITMAP_TO_STRING;
 import static 
org.apache.flink.table.functions.BuiltInFunctionDefinitions.BTRIM;
 import static 
org.apache.flink.table.functions.BuiltInFunctionDefinitions.CARDINALITY;
 import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.CAST;
@@ -2645,6 +2647,18 @@ public abstract class BaseExpressions<InType, OutType> {
         return toApiSpecificExpression(unresolvedCall(BITMAP_FROM_BYTES, 
toExpr()));
     }
 
+    /**
+     * Converts a bitmap to an array of 32-bit integers, the values are sorted 
by {@link
+     * Integer#compareUnsigned}.
+     *
+     * <p>If the input is null, the result is null.
+     *
+     * @return an ARRAY&lt;INT&gt; expression
+     */
+    public OutType bitmapToArray() {
+        return toApiSpecificExpression(unresolvedCall(BITMAP_TO_ARRAY, 
toExpr()));
+    }
+
     /**
      * Converts a bitmap to an array of bytes.
      *
@@ -2659,4 +2673,24 @@ public abstract class BaseExpressions<InType, OutType> {
     public OutType bitmapToBytes() {
         return toApiSpecificExpression(unresolvedCall(BITMAP_TO_BYTES, 
toExpr()));
     }
+
+    /**
+     * Converts a bitmap to a string, the values are sorted by {@link 
Integer#compareUnsigned}. The
+     * string will be truncated and end with "..." if it is too long.
+     *
+     * <p>For example:
+     *
+     * <ul>
+     *   <li>{@code "{}"}, {@code "{1,2,3,4,5}"}
+     *   <li>Negative values (converted to unsigned): {@code 
"{0,1,4294967294,4294967295}"}
+     *   <li>String too long: {@code "{1,2,3,...}"}
+     * </ul>
+     *
+     * <p>If the input is null, the result is null.
+     *
+     * @return a STRING expression
+     */
+    public OutType bitmapToString() {
+        return toApiSpecificExpression(unresolvedCall(BITMAP_TO_STRING, 
toExpr()));
+    }
 }
diff --git 
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
 
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
index 8cf4179605e..5ad21acf0ec 100644
--- 
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
+++ 
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
@@ -3034,6 +3034,19 @@ public final class BuiltInFunctionDefinitions {
                             
"org.apache.flink.table.runtime.functions.scalar.BitmapFromBytesFunction")
                     .build();
 
+    public static final BuiltInFunctionDefinition BITMAP_TO_ARRAY =
+            BuiltInFunctionDefinition.newBuilder()
+                    .name("BITMAP_TO_ARRAY")
+                    .kind(SCALAR)
+                    .inputTypeStrategy(
+                            sequence(
+                                    Collections.singletonList("bitmap"),
+                                    
Collections.singletonList(logical(LogicalTypeRoot.BITMAP))))
+                    
.outputTypeStrategy(nullableIfArgs(explicit(DataTypes.ARRAY(DataTypes.INT()))))
+                    .runtimeClass(
+                            
"org.apache.flink.table.runtime.functions.scalar.BitmapToArrayFunction")
+                    .build();
+
     public static final BuiltInFunctionDefinition BITMAP_TO_BYTES =
             BuiltInFunctionDefinition.newBuilder()
                     .name("BITMAP_TO_BYTES")
@@ -3047,6 +3060,19 @@ public final class BuiltInFunctionDefinitions {
                             
"org.apache.flink.table.runtime.functions.scalar.BitmapToBytesFunction")
                     .build();
 
+    public static final BuiltInFunctionDefinition BITMAP_TO_STRING =
+            BuiltInFunctionDefinition.newBuilder()
+                    .name("BITMAP_TO_STRING")
+                    .kind(SCALAR)
+                    .inputTypeStrategy(
+                            sequence(
+                                    Collections.singletonList("bitmap"),
+                                    
Collections.singletonList(logical(LogicalTypeRoot.BITMAP))))
+                    
.outputTypeStrategy(nullableIfArgs(explicit(DataTypes.STRING())))
+                    .runtimeClass(
+                            
"org.apache.flink.table.runtime.functions.scalar.BitmapToStringFunction")
+                    .build();
+
     // 
--------------------------------------------------------------------------------------------
     // Other functions
     // 
--------------------------------------------------------------------------------------------
diff --git 
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/BitmapFunctionsITCase.java
 
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/BitmapFunctionsITCase.java
index 4744067a8a5..708e79668e3 100644
--- 
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/BitmapFunctionsITCase.java
+++ 
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/BitmapFunctionsITCase.java
@@ -51,7 +51,9 @@ class BitmapFunctionsITCase extends BuiltInFunctionTestBase {
         specs.addAll(bitmapBuildTestCases());
         specs.addAll(bitmapCardinalityTestCases());
         specs.addAll(bitmapFromBytesTestCases());
+        specs.addAll(bitmapToArrayTestCases());
         specs.addAll(bitmapToBytesTestCases());
+        specs.addAll(bitmapToStringTestCases());
         return specs.stream();
     }
 
@@ -98,7 +100,16 @@ class BitmapFunctionsITCase extends BuiltInFunctionTestBase 
{
                                 $("f3").bitmapBuild(),
                                 "BITMAP_BUILD(f3)",
                                 Bitmap.fromArray(new int[] {1, 2, 3, -4}),
-                                DataTypes.BITMAP().notNull()),
+                                DataTypes.BITMAP().notNull())
+                        // round-trip: BUILD -> TO_BYTES -> FROM_BYTES -> 
TO_ARRAY
+                        .testResult(
+                                $("f3").bitmapBuild()
+                                        .bitmapToBytes()
+                                        .bitmapFromBytes()
+                                        .bitmapToArray(),
+                                
"BITMAP_TO_ARRAY(BITMAP_FROM_BYTES(BITMAP_TO_BYTES(BITMAP_BUILD(f3))))",
+                                new Integer[] {1, 2, 3, -4},
+                                DataTypes.ARRAY(DataTypes.INT()).notNull()),
                 
TestSetSpec.forFunction(BuiltInFunctionDefinitions.BITMAP_BUILD, "Validation 
Error")
                         .onFieldsWithData(1024, new long[] {1L, 2L})
                         .andDataTypes(DataTypes.INT(), 
DataTypes.ARRAY(DataTypes.BIGINT()))
@@ -234,6 +245,67 @@ class BitmapFunctionsITCase extends 
BuiltInFunctionTestBase {
                                         + "BITMAP_FROM_BYTES(bytes 
<BINARY_STRING>)"));
     }
 
+    private List<TestSetSpec> bitmapToArrayTestCases() {
+        return Arrays.asList(
+                
TestSetSpec.forFunction(BuiltInFunctionDefinitions.BITMAP_TO_ARRAY)
+                        .onFieldsWithData(
+                                null,
+                                Bitmap.empty(),
+                                Bitmap.fromArray(new int[] {-1}),
+                                Bitmap.fromArray(new int[] {-1, 1, 3, 2, 
Integer.MIN_VALUE}))
+                        .andDataTypes(
+                                DataTypes.BITMAP(),
+                                DataTypes.BITMAP(),
+                                DataTypes.BITMAP(),
+                                DataTypes.BITMAP().notNull())
+                        // null
+                        .testResult(
+                                $("f0").bitmapToArray(),
+                                "BITMAP_TO_ARRAY(f0)",
+                                null,
+                                DataTypes.ARRAY(DataTypes.INT()))
+                        // empty
+                        .testResult(
+                                $("f1").bitmapToArray(),
+                                "BITMAP_TO_ARRAY(f1)",
+                                new Integer[0],
+                                DataTypes.ARRAY(DataTypes.INT()))
+                        // normal cases
+                        .testResult(
+                                $("f2").bitmapToArray(),
+                                "BITMAP_TO_ARRAY(f2)",
+                                new Integer[] {-1},
+                                DataTypes.ARRAY(DataTypes.INT()))
+                        .testResult(
+                                $("f3").bitmapToArray(),
+                                "BITMAP_TO_ARRAY(f3)",
+                                new Integer[] {1, 2, 3, Integer.MIN_VALUE, -1},
+                                DataTypes.ARRAY(DataTypes.INT()).notNull()),
+                
TestSetSpec.forFunction(BuiltInFunctionDefinitions.BITMAP_TO_ARRAY, "Runtime 
Error")
+                        .onFieldsWithData(OVERSIZE_BITMAP)
+                        .andDataTypes(DataTypes.BITMAP().notNull())
+                        .testTableApiRuntimeError(
+                                $("f0").bitmapToArray(),
+                                TableRuntimeException.class,
+                                "Failed to convert bitmap to an array.")
+                        .testSqlRuntimeError(
+                                "BITMAP_TO_ARRAY(f0)",
+                                TableRuntimeException.class,
+                                "Failed to convert bitmap to an array."),
+                TestSetSpec.forFunction(
+                                BuiltInFunctionDefinitions.BITMAP_TO_ARRAY, 
"Validation Error")
+                        .onFieldsWithData(1024, new int[] {1, 2})
+                        .andDataTypes(DataTypes.INT(), 
DataTypes.ARRAY(DataTypes.INT()))
+                        .testTableApiValidationError(
+                                $("f0").bitmapToArray(),
+                                "Invalid input arguments. Expected signatures 
are:\n"
+                                        + "BITMAP_TO_ARRAY(bitmap <BITMAP>)")
+                        .testSqlValidationError(
+                                "BITMAP_TO_ARRAY(f1)",
+                                "Invalid input arguments. Expected signatures 
are:\n"
+                                        + "BITMAP_TO_ARRAY(bitmap <BITMAP>)"));
+    }
+
     private List<TestSetSpec> bitmapToBytesTestCases() {
         return Arrays.asList(
                 
TestSetSpec.forFunction(BuiltInFunctionDefinitions.BITMAP_TO_BYTES)
@@ -282,6 +354,64 @@ class BitmapFunctionsITCase extends 
BuiltInFunctionTestBase {
                                         + "BITMAP_TO_BYTES(bitmap <BITMAP>)"));
     }
 
+    private List<TestSetSpec> bitmapToStringTestCases() {
+        return Arrays.asList(
+                
TestSetSpec.forFunction(BuiltInFunctionDefinitions.BITMAP_TO_STRING)
+                        .onFieldsWithData(
+                                null,
+                                Bitmap.empty(),
+                                Bitmap.fromArray(new int[] {-1}),
+                                Bitmap.fromArray(new int[] {Integer.MIN_VALUE, 
-1, 1, 3, 2}),
+                                OVERSIZE_BITMAP)
+                        .andDataTypes(
+                                DataTypes.BITMAP(),
+                                DataTypes.BITMAP(),
+                                DataTypes.BITMAP(),
+                                DataTypes.BITMAP().notNull(),
+                                DataTypes.BITMAP().notNull())
+                        // null
+                        .testResult(
+                                $("f0").bitmapToString(),
+                                "BITMAP_TO_STRING(f0)",
+                                null,
+                                DataTypes.STRING())
+                        // empty
+                        .testResult(
+                                $("f1").bitmapToString(),
+                                "BITMAP_TO_STRING(f1)",
+                                "{}",
+                                DataTypes.STRING())
+                        // normal cases
+                        .testResult(
+                                $("f2").bitmapToString(),
+                                "BITMAP_TO_STRING(f2)",
+                                String.format("{%s}", 0xFFFFFFFFL),
+                                DataTypes.STRING())
+                        .testResult(
+                                $("f3").bitmapToString(),
+                                "BITMAP_TO_STRING(f3)",
+                                String.format("{1,2,3,%s,%s}", 0x80000000L, 
0xFFFFFFFFL),
+                                DataTypes.STRING().notNull())
+                        // oversize
+                        .testResult(
+                                $("f4").bitmapToString().endsWith(",...}"),
+                                "ENDSWITH(BITMAP_TO_STRING(f4), ',...}')",
+                                true,
+                                DataTypes.BOOLEAN().notNull()),
+                TestSetSpec.forFunction(
+                                BuiltInFunctionDefinitions.BITMAP_TO_STRING, 
"Validation Error")
+                        .onFieldsWithData(1024, new int[] {1, 2})
+                        .andDataTypes(DataTypes.INT(), 
DataTypes.ARRAY(DataTypes.INT()))
+                        .testTableApiValidationError(
+                                $("f0").bitmapToString(),
+                                "Invalid input arguments. Expected signatures 
are:\n"
+                                        + "BITMAP_TO_STRING(bitmap <BITMAP>)")
+                        .testSqlValidationError(
+                                "BITMAP_TO_STRING(f1)",
+                                "Invalid input arguments. Expected signatures 
are:\n"
+                                        + "BITMAP_TO_STRING(bitmap 
<BITMAP>)"));
+    }
+
     // ~ Utils 
--------------------------------------------------------------------
 
     private byte[] toSerializedBytes(int... values) {
diff --git 
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/BitmapToArrayFunction.java
 
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/BitmapToArrayFunction.java
new file mode 100644
index 00000000000..aeae71fe4ab
--- /dev/null
+++ 
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/BitmapToArrayFunction.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.TableRuntimeException;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction.SpecializedContext;
+import org.apache.flink.types.bitmap.Bitmap;
+
+import javax.annotation.Nullable;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#BITMAP_TO_ARRAY}. */
+@Internal
+public class BitmapToArrayFunction extends BuiltInScalarFunction {
+
+    public BitmapToArrayFunction(SpecializedContext context) {
+        super(BuiltInFunctionDefinitions.BITMAP_TO_ARRAY, context);
+    }
+
+    public ArrayData eval(@Nullable Bitmap bitmap) {
+        if (bitmap == null) {
+            return null;
+        }
+
+        try {
+            return new GenericArrayData(bitmap.toArray());
+        } catch (Throwable e) {
+            throw new TableRuntimeException("Failed to convert bitmap to an 
array.", e);
+        }
+    }
+}
diff --git 
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/BitmapToStringFunction.java
 
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/BitmapToStringFunction.java
new file mode 100644
index 00000000000..04f0323282f
--- /dev/null
+++ 
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/BitmapToStringFunction.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction.SpecializedContext;
+import org.apache.flink.types.bitmap.Bitmap;
+
+import javax.annotation.Nullable;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#BITMAP_TO_STRING}. */
+@Internal
+public class BitmapToStringFunction extends BuiltInScalarFunction {
+
+    public BitmapToStringFunction(SpecializedContext context) {
+        super(BuiltInFunctionDefinitions.BITMAP_TO_STRING, context);
+    }
+
+    public StringData eval(@Nullable Bitmap bitmap) {
+        if (bitmap == null) {
+            return null;
+        }
+        return StringData.fromString(bitmap.toString());
+    }
+}

Reply via email to