You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by li...@apache.org on 2026/03/25 09:26:32 UTC
(flink) 04/06: [FLINK-39186][table] Add the built-in function BITMAP_CARDINALITY
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 20af677612fd767dfcb062b0107b25dab54e3cf3
Author: dylanhz <dy...@163.com>
AuthorDate: Tue Mar 24 01:44:13 2026 +0800
[FLINK-39186][table] Add the built-in function BITMAP_CARDINALITY
---
docs/data/sql_functions.yml | 8 +++
docs/data/sql_functions_zh.yml | 8 +++
.../docs/reference/pyflink.table/expressions.rst | 1 +
flink-python/pyflink/table/expression.py | 10 ++++
.../pyflink/table/tests/test_expression.py | 1 +
.../flink/table/api/internal/BaseExpressions.java | 12 ++++
.../functions/BuiltInFunctionDefinitions.java | 13 +++++
.../planner/functions/BitmapFunctionsITCase.java | 68 ++++++++++++++++++++++
.../scalar/BitmapCardinalityFunction.java | 42 +++++++++++++
9 files changed, 163 insertions(+)
diff --git a/docs/data/sql_functions.yml b/docs/data/sql_functions.yml
index 56d5491b77e..3b5e5a48d25 100644
--- a/docs/data/sql_functions.yml
+++ b/docs/data/sql_functions.yml
@@ -1358,6 +1358,14 @@ bitmap:
`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: |
diff --git a/docs/data/sql_functions_zh.yml b/docs/data/sql_functions_zh.yml
index 1b56accd1d0..9cf9602a26c 100644
--- a/docs/data/sql_functions_zh.yml
+++ b/docs/data/sql_functions_zh.yml
@@ -1444,6 +1444,14 @@ bitmap:
`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: |
diff --git a/flink-python/docs/reference/pyflink.table/expressions.rst b/flink-python/docs/reference/pyflink.table/expressions.rst
index 61bf4a8eef3..21a70cb62e8 100644
--- a/flink-python/docs/reference/pyflink.table/expressions.rst
+++ b/flink-python/docs/reference/pyflink.table/expressions.rst
@@ -346,5 +346,6 @@ Bitmap functions
:toctree: api/
Expression.bitmap_build
+ Expression.bitmap_cardinality
Expression.bitmap_from_bytes
Expression.bitmap_to_bytes
diff --git a/flink-python/pyflink/table/expression.py b/flink-python/pyflink/table/expression.py
index 8a26a887b45..5d9c766f4c8 100644
--- a/flink-python/pyflink/table/expression.py
+++ b/flink-python/pyflink/table/expression.py
@@ -2287,6 +2287,16 @@ class Expression(Generic[T]):
"""
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.
diff --git a/flink-python/pyflink/table/tests/test_expression.py b/flink-python/pyflink/table/tests/test_expression.py
index c6385bd488f..1db5b8af8cd 100644
--- a/flink-python/pyflink/table/tests/test_expression.py
+++ b/flink-python/pyflink/table/tests/test_expression.py
@@ -267,6 +267,7 @@ class PyFlinkBatchExpressionTests(PyFlinkTestCase):
# bitmap functions
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_BYTES(a)", str(expr1.bitmap_to_bytes()))
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 c742b7ca958..07f4e85837c 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
@@ -77,6 +77,7 @@ import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.AVG;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.BETWEEN;
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_BYTES;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.BTRIM;
@@ -2618,6 +2619,17 @@ public abstract class BaseExpressions<InType, OutType> {
return toApiSpecificExpression(unresolvedCall(BITMAP_BUILD, toExpr()));
}
+ /**
+ * Returns the cardinality of a bitmap.
+ *
+ * <p>If the input is null, the result is null.
+ *
+ * @return a BIGINT expression
+ */
+ public OutType bitmapCardinality() {
+ return toApiSpecificExpression(unresolvedCall(BITMAP_CARDINALITY, toExpr()));
+ }
+
/**
* Converts an array of bytes to a bitmap.
*
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 ddbe1887fce..8cf4179605e 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
@@ -3007,6 +3007,19 @@ public final class BuiltInFunctionDefinitions {
"org.apache.flink.table.runtime.functions.scalar.BitmapBuildFunction")
.build();
+ public static final BuiltInFunctionDefinition BITMAP_CARDINALITY =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("BITMAP_CARDINALITY")
+ .kind(SCALAR)
+ .inputTypeStrategy(
+ sequence(
+ Collections.singletonList("bitmap"),
+ Collections.singletonList(logical(LogicalTypeRoot.BITMAP))))
+ .outputTypeStrategy(nullableIfArgs(explicit(DataTypes.BIGINT())))
+ .runtimeClass(
+ "org.apache.flink.table.runtime.functions.scalar.BitmapCardinalityFunction")
+ .build();
+
public static final BuiltInFunctionDefinition BITMAP_FROM_BYTES =
BuiltInFunctionDefinition.newBuilder()
.name("BITMAP_FROM_BYTES")
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 121675f2318..4744067a8a5 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
@@ -36,10 +36,20 @@ import static org.apache.flink.table.api.Expressions.$;
/** Test bitmap functions correct behaviour. */
class BitmapFunctionsITCase extends BuiltInFunctionTestBase {
+ private static final Bitmap OVERSIZE_BITMAP;
+
+ static {
+ // size 0x80000000L
+ OVERSIZE_BITMAP = Bitmap.empty();
+ OVERSIZE_BITMAP.add(0L, Integer.MAX_VALUE);
+ OVERSIZE_BITMAP.add(Integer.MAX_VALUE);
+ }
+
@Override
Stream<TestSetSpec> getTestSetSpecs() {
final List<TestSetSpec> specs = new ArrayList<>();
specs.addAll(bitmapBuildTestCases());
+ specs.addAll(bitmapCardinalityTestCases());
specs.addAll(bitmapFromBytesTestCases());
specs.addAll(bitmapToBytesTestCases());
return specs.stream();
@@ -104,6 +114,64 @@ class BitmapFunctionsITCase extends BuiltInFunctionTestBase {
+ "BITMAP_BUILD(array ARRAY<INT>)"));
}
+ private List<TestSetSpec> bitmapCardinalityTestCases() {
+ return Arrays.asList(
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.BITMAP_CARDINALITY)
+ .onFieldsWithData(
+ null,
+ Bitmap.empty(),
+ Bitmap.fromArray(new int[] {-1}),
+ Bitmap.fromArray(new int[] {1, 2, 3, -4}),
+ OVERSIZE_BITMAP)
+ .andDataTypes(
+ DataTypes.BITMAP(),
+ DataTypes.BITMAP(),
+ DataTypes.BITMAP(),
+ DataTypes.BITMAP().notNull(),
+ DataTypes.BITMAP().notNull())
+ // null
+ .testResult(
+ $("f0").bitmapCardinality(),
+ "BITMAP_CARDINALITY(f0)",
+ null,
+ DataTypes.BIGINT())
+ // empty
+ .testResult(
+ $("f1").bitmapCardinality(),
+ "BITMAP_CARDINALITY(f1)",
+ 0L,
+ DataTypes.BIGINT())
+ // normal cases
+ .testResult(
+ $("f2").bitmapCardinality(),
+ "BITMAP_CARDINALITY(f2)",
+ 1L,
+ DataTypes.BIGINT())
+ .testResult(
+ $("f3").bitmapCardinality(),
+ "BITMAP_CARDINALITY(f3)",
+ 4L,
+ DataTypes.BIGINT().notNull())
+ // oversize
+ .testResult(
+ $("f4").bitmapCardinality(),
+ "BITMAP_CARDINALITY(f4)",
+ 0x80000000L,
+ DataTypes.BIGINT().notNull()),
+ TestSetSpec.forFunction(
+ BuiltInFunctionDefinitions.BITMAP_CARDINALITY, "Validation Error")
+ .onFieldsWithData(1024, new int[] {1, 2})
+ .andDataTypes(DataTypes.INT(), DataTypes.ARRAY(DataTypes.INT()))
+ .testTableApiValidationError(
+ $("f0").bitmapCardinality(),
+ "Invalid input arguments. Expected signatures are:\n"
+ + "BITMAP_CARDINALITY(bitmap <BITMAP>)")
+ .testSqlValidationError(
+ "BITMAP_CARDINALITY(f1)",
+ "Invalid input arguments. Expected signatures are:\n"
+ + "BITMAP_CARDINALITY(bitmap <BITMAP>)"));
+ }
+
private List<TestSetSpec> bitmapFromBytesTestCases() {
return Arrays.asList(
TestSetSpec.forFunction(BuiltInFunctionDefinitions.BITMAP_FROM_BYTES)
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/BitmapCardinalityFunction.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/BitmapCardinalityFunction.java
new file mode 100644
index 00000000000..ab0328d04e8
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/BitmapCardinalityFunction.java
@@ -0,0 +1,42 @@
+/*
+ * 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.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_CARDINALITY}. */
+@Internal
+public class BitmapCardinalityFunction extends BuiltInScalarFunction {
+
+ public BitmapCardinalityFunction(SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.BITMAP_CARDINALITY, context);
+ }
+
+ public Long eval(@Nullable Bitmap bitmap) {
+ if (bitmap == null) {
+ return null;
+ }
+ return bitmap.getLongCardinality();
+ }
+}