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 c0bc1e53851c698d6004b0169b507e5572f9fade Author: dylanhz <[email protected]> AuthorDate: Tue Mar 24 01:50:48 2026 +0800 [FLINK-39186][table] Add the built-in function BITMAP_AND, BITMAP_ANDNOT, BITMAP_OR, BITMAP_XOR --- docs/data/sql_functions.yml | 32 ++++ docs/data/sql_functions_zh.yml | 32 ++++ .../docs/reference/pyflink.table/expressions.rst | 4 + flink-python/pyflink/table/expression.py | 44 +++++ .../pyflink/table/tests/test_expression.py | 4 + .../flink/table/api/internal/BaseExpressions.java | 56 ++++++ .../functions/BuiltInFunctionDefinitions.java | 60 +++++++ .../planner/functions/BitmapFunctionsITCase.java | 197 +++++++++++++++++++++ .../functions/scalar/BitmapAndFunction.java | 45 +++++ .../functions/scalar/BitmapAndnotFunction.java | 45 +++++ .../runtime/functions/scalar/BitmapOrFunction.java | 45 +++++ .../functions/scalar/BitmapXorFunction.java | 45 +++++ 12 files changed, 609 insertions(+) diff --git a/docs/data/sql_functions.yml b/docs/data/sql_functions.yml index 0259592275d..8f92038de69 100644 --- a/docs/data/sql_functions.yml +++ b/docs/data/sql_functions.yml @@ -1350,6 +1350,22 @@ hashfunctions: 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: | @@ -1376,6 +1392,14 @@ bitmap: `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: | @@ -1408,6 +1432,14 @@ bitmap: `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) diff --git a/docs/data/sql_functions_zh.yml b/docs/data/sql_functions_zh.yml index 13a63f4df46..c28996064d7 100644 --- a/docs/data/sql_functions_zh.yml +++ b/docs/data/sql_functions_zh.yml @@ -1436,6 +1436,22 @@ hashfunctions: 第二个参数 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: | @@ -1462,6 +1478,14 @@ bitmap: `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: | @@ -1494,6 +1518,14 @@ bitmap: `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) diff --git a/flink-python/docs/reference/pyflink.table/expressions.rst b/flink-python/docs/reference/pyflink.table/expressions.rst index 480631ee554..9ad848c167f 100644 --- a/flink-python/docs/reference/pyflink.table/expressions.rst +++ b/flink-python/docs/reference/pyflink.table/expressions.rst @@ -345,9 +345,13 @@ Bitmap functions .. 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 diff --git a/flink-python/pyflink/table/expression.py b/flink-python/pyflink/table/expression.py index 81f28c76999..a251a2dfcda 100644 --- a/flink-python/pyflink/table/expression.py +++ b/flink-python/pyflink/table/expression.py @@ -2277,6 +2277,28 @@ class Expression(Generic[T]): # ---------------------------- 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. @@ -2310,6 +2332,17 @@ class Expression(Generic[T]): """ 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 \ @@ -2351,6 +2384,17 @@ class Expression(Generic[T]): """ 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() diff --git a/flink-python/pyflink/table/tests/test_expression.py b/flink-python/pyflink/table/tests/test_expression.py index 9fffd470ebf..edc0dab1770 100644 --- a/flink-python/pyflink/table/tests/test_expression.py +++ b/flink-python/pyflink/table/tests/test_expression.py @@ -266,12 +266,16 @@ class PyFlinkBatchExpressionTests(PyFlinkTestCase): 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') 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 6e49f0190b8..3704f80ea8b 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 @@ -76,12 +76,16 @@ import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.ATAN; 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_AND; +import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.BITMAP_ANDNOT; 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_OR; 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.BITMAP_XOR; 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; @@ -2610,6 +2614,32 @@ public abstract class BaseExpressions<InType, OutType> { // Bitmap functions + /** + * Computes the AND (intersection) of two bitmaps. + * + * <p>If any of the inputs are null, the result is null. + * + * @param bitmap2 the bitmap to perform AND operation with + * @return a BITMAP expression + */ + public OutType bitmapAnd(InType bitmap2) { + return toApiSpecificExpression( + unresolvedCall(BITMAP_AND, toExpr(), objectToExpression(bitmap2))); + } + + /** + * Computes the AND NOT (difference) of two bitmaps. + * + * <p>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 + */ + public OutType bitmapAndnot(InType bitmap2) { + return toApiSpecificExpression( + unresolvedCall(BITMAP_ANDNOT, toExpr(), objectToExpression(bitmap2))); + } + /** * Creates a bitmap from an array of 32-bit integers. * @@ -2647,6 +2677,19 @@ public abstract class BaseExpressions<InType, OutType> { return toApiSpecificExpression(unresolvedCall(BITMAP_FROM_BYTES, toExpr())); } + /** + * Computes the OR (union) of two bitmaps. + * + * <p>If any of the inputs are null, the result is null. + * + * @param bitmap2 the bitmap to perform OR operation with + * @return a BITMAP expression + */ + public OutType bitmapOr(InType bitmap2) { + return toApiSpecificExpression( + unresolvedCall(BITMAP_OR, toExpr(), objectToExpression(bitmap2))); + } + /** * Converts a bitmap to an array of 32-bit integers, the values are sorted by {@link * Integer#compareUnsigned}. @@ -2693,4 +2736,17 @@ public abstract class BaseExpressions<InType, OutType> { public OutType bitmapToString() { return toApiSpecificExpression(unresolvedCall(BITMAP_TO_STRING, toExpr())); } + + /** + * Computes the XOR (symmetric difference) of two bitmaps. + * + * <p>If any of the inputs are null, the result is null. + * + * @param bitmap2 the bitmap to perform XOR operation with + * @return a BITMAP expression + */ + public OutType bitmapXor(InType bitmap2) { + return toApiSpecificExpression( + unresolvedCall(BITMAP_XOR, toExpr(), objectToExpression(bitmap2))); + } } 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 5ad21acf0ec..7dac1d91cc8 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 @@ -2985,6 +2985,36 @@ public final class BuiltInFunctionDefinitions { // Bitmap functions // -------------------------------------------------------------------------------------------- + public static final BuiltInFunctionDefinition BITMAP_AND = + BuiltInFunctionDefinition.newBuilder() + .name("BITMAP_AND") + .kind(SCALAR) + .inputTypeStrategy( + sequence( + Arrays.asList("bitmap1", "bitmap2"), + Arrays.asList( + logical(LogicalTypeRoot.BITMAP), + logical(LogicalTypeRoot.BITMAP)))) + .outputTypeStrategy(nullableIfArgs(explicit(DataTypes.BITMAP()))) + .runtimeClass( + "org.apache.flink.table.runtime.functions.scalar.BitmapAndFunction") + .build(); + + public static final BuiltInFunctionDefinition BITMAP_ANDNOT = + BuiltInFunctionDefinition.newBuilder() + .name("BITMAP_ANDNOT") + .kind(SCALAR) + .inputTypeStrategy( + sequence( + Arrays.asList("bitmap1", "bitmap2"), + Arrays.asList( + logical(LogicalTypeRoot.BITMAP), + logical(LogicalTypeRoot.BITMAP)))) + .outputTypeStrategy(nullableIfArgs(explicit(DataTypes.BITMAP()))) + .runtimeClass( + "org.apache.flink.table.runtime.functions.scalar.BitmapAndnotFunction") + .build(); + public static final BuiltInFunctionDefinition BITMAP_BUILD = BuiltInFunctionDefinition.newBuilder() .name("BITMAP_BUILD") @@ -3034,6 +3064,21 @@ public final class BuiltInFunctionDefinitions { "org.apache.flink.table.runtime.functions.scalar.BitmapFromBytesFunction") .build(); + public static final BuiltInFunctionDefinition BITMAP_OR = + BuiltInFunctionDefinition.newBuilder() + .name("BITMAP_OR") + .kind(SCALAR) + .inputTypeStrategy( + sequence( + Arrays.asList("bitmap1", "bitmap2"), + Arrays.asList( + logical(LogicalTypeRoot.BITMAP), + logical(LogicalTypeRoot.BITMAP)))) + .outputTypeStrategy(nullableIfArgs(explicit(DataTypes.BITMAP()))) + .runtimeClass( + "org.apache.flink.table.runtime.functions.scalar.BitmapOrFunction") + .build(); + public static final BuiltInFunctionDefinition BITMAP_TO_ARRAY = BuiltInFunctionDefinition.newBuilder() .name("BITMAP_TO_ARRAY") @@ -3073,6 +3118,21 @@ public final class BuiltInFunctionDefinitions { "org.apache.flink.table.runtime.functions.scalar.BitmapToStringFunction") .build(); + public static final BuiltInFunctionDefinition BITMAP_XOR = + BuiltInFunctionDefinition.newBuilder() + .name("BITMAP_XOR") + .kind(SCALAR) + .inputTypeStrategy( + sequence( + Arrays.asList("bitmap1", "bitmap2"), + Arrays.asList( + logical(LogicalTypeRoot.BITMAP), + logical(LogicalTypeRoot.BITMAP)))) + .outputTypeStrategy(nullableIfArgs(explicit(DataTypes.BITMAP()))) + .runtimeClass( + "org.apache.flink.table.runtime.functions.scalar.BitmapXorFunction") + .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 708e79668e3..d69d966a14d 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 @@ -48,15 +48,116 @@ class BitmapFunctionsITCase extends BuiltInFunctionTestBase { @Override Stream<TestSetSpec> getTestSetSpecs() { final List<TestSetSpec> specs = new ArrayList<>(); + specs.addAll(bitmapAndTestCases()); + specs.addAll(bitmapAndnotTestCases()); specs.addAll(bitmapBuildTestCases()); specs.addAll(bitmapCardinalityTestCases()); specs.addAll(bitmapFromBytesTestCases()); + specs.addAll(bitmapOrTestCases()); specs.addAll(bitmapToArrayTestCases()); specs.addAll(bitmapToBytesTestCases()); specs.addAll(bitmapToStringTestCases()); + specs.addAll(bitmapXorTestCases()); return specs.stream(); } + private List<TestSetSpec> bitmapAndTestCases() { + return Arrays.asList( + TestSetSpec.forFunction(BuiltInFunctionDefinitions.BITMAP_AND) + .onFieldsWithData( + null, + Bitmap.fromArray(new int[] {-1}), + Bitmap.fromArray(new int[] {0, 2, 4, -4}), + Bitmap.fromArray(new int[] {1, 2, 3, -4})) + .andDataTypes( + DataTypes.BITMAP(), + DataTypes.BITMAP(), + DataTypes.BITMAP().notNull(), + DataTypes.BITMAP().notNull()) + // null + .testResult( + $("f0").bitmapAnd($("f1")), + "BITMAP_AND(f0, f1)", + null, + DataTypes.BITMAP()) + .testResult( + $("f1").bitmapAnd($("f0")), + "BITMAP_AND(f1, f0)", + null, + DataTypes.BITMAP()) + // normal cases + .testResult( + $("f1").bitmapAnd($("f2")), + "BITMAP_AND(f1, f2)", + Bitmap.empty(), + DataTypes.BITMAP()) + .testResult( + $("f2").bitmapAnd($("f3")), + "BITMAP_AND(f2, f3)", + Bitmap.fromArray(new int[] {2, -4}), + DataTypes.BITMAP().notNull()), + TestSetSpec.forFunction(BuiltInFunctionDefinitions.BITMAP_AND, "Validation Error") + .onFieldsWithData("{1,2}", new int[] {1, 2}) + .andDataTypes(DataTypes.STRING(), DataTypes.ARRAY(DataTypes.INT())) + .testTableApiValidationError( + $("f0").bitmapAnd($("f1")), + "Invalid input arguments. Expected signatures are:\n" + + "BITMAP_AND(bitmap1 <BITMAP>, bitmap2 <BITMAP>)") + .testSqlValidationError( + "BITMAP_AND(f1, f0)", + "Invalid input arguments. Expected signatures are:\n" + + "BITMAP_AND(bitmap1 <BITMAP>, bitmap2 <BITMAP>)")); + } + + private List<TestSetSpec> bitmapAndnotTestCases() { + return Arrays.asList( + TestSetSpec.forFunction(BuiltInFunctionDefinitions.BITMAP_ANDNOT) + .onFieldsWithData( + null, + Bitmap.fromArray(new int[] {-1}), + Bitmap.fromArray(new int[] {0, 2, 4, -4}), + Bitmap.fromArray(new int[] {1, 2, 3, -4})) + .andDataTypes( + DataTypes.BITMAP(), + DataTypes.BITMAP(), + DataTypes.BITMAP().notNull(), + DataTypes.BITMAP().notNull()) + // null + .testResult( + $("f0").bitmapAndnot($("f1")), + "BITMAP_ANDNOT(f0, f1)", + null, + DataTypes.BITMAP()) + .testResult( + $("f1").bitmapAndnot($("f0")), + "BITMAP_ANDNOT(f1, f0)", + null, + DataTypes.BITMAP()) + // normal cases + .testResult( + $("f1").bitmapAndnot($("f2")), + "BITMAP_ANDNOT(f1, f2)", + Bitmap.fromArray(new int[] {-1}), + DataTypes.BITMAP()) + .testResult( + $("f2").bitmapAndnot($("f3")), + "BITMAP_ANDNOT(f2, f3)", + Bitmap.fromArray(new int[] {0, 4}), + DataTypes.BITMAP().notNull()), + TestSetSpec.forFunction( + BuiltInFunctionDefinitions.BITMAP_ANDNOT, "Validation Error") + .onFieldsWithData("{1,2}", new int[] {1, 2}) + .andDataTypes(DataTypes.STRING(), DataTypes.ARRAY(DataTypes.INT())) + .testTableApiValidationError( + $("f0").bitmapAndnot($("f1")), + "Invalid input arguments. Expected signatures are:\n" + + "BITMAP_ANDNOT(bitmap1 <BITMAP>, bitmap2 <BITMAP>)") + .testSqlValidationError( + "BITMAP_ANDNOT(f1, f0)", + "Invalid input arguments. Expected signatures are:\n" + + "BITMAP_ANDNOT(bitmap1 <BITMAP>, bitmap2 <BITMAP>)")); + } + private List<TestSetSpec> bitmapBuildTestCases() { return Arrays.asList( TestSetSpec.forFunction(BuiltInFunctionDefinitions.BITMAP_BUILD) @@ -245,6 +346,54 @@ class BitmapFunctionsITCase extends BuiltInFunctionTestBase { + "BITMAP_FROM_BYTES(bytes <BINARY_STRING>)")); } + private List<TestSetSpec> bitmapOrTestCases() { + return Arrays.asList( + TestSetSpec.forFunction(BuiltInFunctionDefinitions.BITMAP_OR) + .onFieldsWithData( + null, + Bitmap.fromArray(new int[] {-1}), + Bitmap.fromArray(new int[] {0, 2, 4, -4}), + Bitmap.fromArray(new int[] {1, 2, 3, -4})) + .andDataTypes( + DataTypes.BITMAP(), + DataTypes.BITMAP(), + DataTypes.BITMAP().notNull(), + DataTypes.BITMAP().notNull()) + // null + .testResult( + $("f0").bitmapOr($("f1")), + "BITMAP_OR(f0, f1)", + null, + DataTypes.BITMAP()) + .testResult( + $("f1").bitmapOr($("f0")), + "BITMAP_OR(f1, f0)", + null, + DataTypes.BITMAP()) + // normal cases + .testResult( + $("f1").bitmapOr($("f2")), + "BITMAP_OR(f1, f2)", + Bitmap.fromArray(new int[] {0, 2, 4, -4, -1}), + DataTypes.BITMAP()) + .testResult( + $("f2").bitmapOr($("f3")), + "BITMAP_OR(f2, f3)", + Bitmap.fromArray(new int[] {0, 1, 2, 3, 4, -4}), + DataTypes.BITMAP().notNull()), + TestSetSpec.forFunction(BuiltInFunctionDefinitions.BITMAP_OR, "Validation Error") + .onFieldsWithData("{1,2}", new int[] {1, 2}) + .andDataTypes(DataTypes.STRING(), DataTypes.ARRAY(DataTypes.INT())) + .testTableApiValidationError( + $("f0").bitmapOr($("f1")), + "Invalid input arguments. Expected signatures are:\n" + + "BITMAP_OR(bitmap1 <BITMAP>, bitmap2 <BITMAP>)") + .testSqlValidationError( + "BITMAP_OR(f1, f0)", + "Invalid input arguments. Expected signatures are:\n" + + "BITMAP_OR(bitmap1 <BITMAP>, bitmap2 <BITMAP>)")); + } + private List<TestSetSpec> bitmapToArrayTestCases() { return Arrays.asList( TestSetSpec.forFunction(BuiltInFunctionDefinitions.BITMAP_TO_ARRAY) @@ -412,6 +561,54 @@ class BitmapFunctionsITCase extends BuiltInFunctionTestBase { + "BITMAP_TO_STRING(bitmap <BITMAP>)")); } + private List<TestSetSpec> bitmapXorTestCases() { + return Arrays.asList( + TestSetSpec.forFunction(BuiltInFunctionDefinitions.BITMAP_XOR) + .onFieldsWithData( + null, + Bitmap.fromArray(new int[] {-1}), + Bitmap.fromArray(new int[] {0, 2, 4, -4}), + Bitmap.fromArray(new int[] {1, 2, 3, -4})) + .andDataTypes( + DataTypes.BITMAP(), + DataTypes.BITMAP(), + DataTypes.BITMAP().notNull(), + DataTypes.BITMAP().notNull()) + // null + .testResult( + $("f0").bitmapXor($("f1")), + "BITMAP_XOR(f0, f1)", + null, + DataTypes.BITMAP()) + .testResult( + $("f1").bitmapXor($("f0")), + "BITMAP_XOR(f1, f0)", + null, + DataTypes.BITMAP()) + // normal cases + .testResult( + $("f1").bitmapXor($("f2")), + "BITMAP_XOR(f1, f2)", + Bitmap.fromArray(new int[] {0, 2, 4, -4, -1}), + DataTypes.BITMAP()) + .testResult( + $("f2").bitmapXor($("f3")), + "BITMAP_XOR(f2, f3)", + Bitmap.fromArray(new int[] {0, 1, 3, 4}), + DataTypes.BITMAP().notNull()), + TestSetSpec.forFunction(BuiltInFunctionDefinitions.BITMAP_XOR, "Validation Error") + .onFieldsWithData("{1,2}", new int[] {1, 2}) + .andDataTypes(DataTypes.STRING(), DataTypes.ARRAY(DataTypes.INT())) + .testTableApiValidationError( + $("f0").bitmapXor($("f1")), + "Invalid input arguments. Expected signatures are:\n" + + "BITMAP_XOR(bitmap1 <BITMAP>, bitmap2 <BITMAP>)") + .testSqlValidationError( + "BITMAP_XOR(f1, f0)", + "Invalid input arguments. Expected signatures are:\n" + + "BITMAP_XOR(bitmap1 <BITMAP>, bitmap2 <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/BitmapAndFunction.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/BitmapAndFunction.java new file mode 100644 index 00000000000..67187b36f9c --- /dev/null +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/BitmapAndFunction.java @@ -0,0 +1,45 @@ +/* + * 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_AND}. */ +@Internal +public class BitmapAndFunction extends BuiltInScalarFunction { + + public BitmapAndFunction(SpecializedContext context) { + super(BuiltInFunctionDefinitions.BITMAP_AND, context); + } + + public Bitmap eval(@Nullable Bitmap bitmap1, @Nullable Bitmap bitmap2) { + if (bitmap1 == null || bitmap2 == null) { + return null; + } + + Bitmap copy = Bitmap.from(bitmap1); + copy.and(bitmap2); + return copy; + } +} diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/BitmapAndnotFunction.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/BitmapAndnotFunction.java new file mode 100644 index 00000000000..3cd70e048b9 --- /dev/null +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/BitmapAndnotFunction.java @@ -0,0 +1,45 @@ +/* + * 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_ANDNOT}. */ +@Internal +public class BitmapAndnotFunction extends BuiltInScalarFunction { + + public BitmapAndnotFunction(SpecializedContext context) { + super(BuiltInFunctionDefinitions.BITMAP_ANDNOT, context); + } + + public Bitmap eval(@Nullable Bitmap bitmap1, @Nullable Bitmap bitmap2) { + if (bitmap1 == null || bitmap2 == null) { + return null; + } + + Bitmap copy = Bitmap.from(bitmap1); + copy.andNot(bitmap2); + return copy; + } +} diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/BitmapOrFunction.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/BitmapOrFunction.java new file mode 100644 index 00000000000..bea91f90f2b --- /dev/null +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/BitmapOrFunction.java @@ -0,0 +1,45 @@ +/* + * 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_OR}. */ +@Internal +public class BitmapOrFunction extends BuiltInScalarFunction { + + public BitmapOrFunction(SpecializedContext context) { + super(BuiltInFunctionDefinitions.BITMAP_OR, context); + } + + public Bitmap eval(@Nullable Bitmap bitmap1, @Nullable Bitmap bitmap2) { + if (bitmap1 == null || bitmap2 == null) { + return null; + } + + Bitmap copy = Bitmap.from(bitmap1); + copy.or(bitmap2); + return copy; + } +} diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/BitmapXorFunction.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/BitmapXorFunction.java new file mode 100644 index 00000000000..8977b3be12c --- /dev/null +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/BitmapXorFunction.java @@ -0,0 +1,45 @@ +/* + * 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_XOR}. */ +@Internal +public class BitmapXorFunction extends BuiltInScalarFunction { + + public BitmapXorFunction(SpecializedContext context) { + super(BuiltInFunctionDefinitions.BITMAP_XOR, context); + } + + public Bitmap eval(@Nullable Bitmap bitmap1, @Nullable Bitmap bitmap2) { + if (bitmap1 == null || bitmap2 == null) { + return null; + } + + Bitmap copy = Bitmap.from(bitmap1); + copy.xor(bitmap2); + return copy; + } +}
