[FIP-37] Add bitmap scalar functions and register via FlussCatalog

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java

@@ -26,8 +26,14 @@
 import org.apache.fluss.flink.FlinkConnectorOptions;
 import org.apache.fluss.flink.adapter.CatalogTableAdapter;
 import org.apache.fluss.flink.functions.bitmap.RbAndAggFunction;
+import org.apache.fluss.flink.functions.bitmap.RbAndFunction;
 import org.apache.fluss.flink.functions.bitmap.RbBuildAggFunction;
+import org.apache.fluss.flink.functions.bitmap.RbBuildFunction;
+import org.apache.fluss.flink.functions.bitmap.RbCardinalityFunction;
+import org.apache.fluss.flink.functions.bitmap.RbContainsFunction;
 import org.apache.fluss.flink.functions.bitmap.RbOrAggFunction;
+import org.apache.fluss.flink.functions.bitmap.RbOrFunction;
+import org.apache.fluss.flink.functions.bitmap.RbToArrayFunction;
 import org.apache.fluss.flink.lake.LakeFlinkCatalog;
 import org.apache.fluss.flink.procedure.ProcedureManager;
 import org.apache.fluss.flink.utils.CatalogExceptionUtils;
@@ -144,9 +150,17 @@ public class FlinkCatalog extends AbstractCatalog {
 
     static {
         Map<String, String> map = new HashMap<>();
+        // aggregate functions
         map.put("rb_build_agg", RbBuildAggFunction.class.getName());
         map.put("rb_or_agg", RbOrAggFunction.class.getName());
         map.put("rb_and_agg", RbAndAggFunction.class.getName());
+        // scalar functions
+        map.put("rb_cardinality", RbCardinalityFunction.class.getName());
+        map.put("rb_build", RbBuildFunction.class.getName());
+        map.put("rb_contains", RbContainsFunction.class.getName());
+        map.put("rb_to_array", RbToArrayFunction.class.getName());
+        map.put("rb_or", RbOrFunction.class.getName());
+        map.put("rb_and", RbAndFunction.class.getName());
         BUILTIN_BITMAP_FUNCTIONS = Collections.unmodifiableMap(map);
     }
 

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/RbAndFunction.java

@@ -0,0 +1,51 @@
+/*
+ * 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.fluss.flink.functions.bitmap;
+
+import org.apache.flink.table.functions.ScalarFunction;
+import org.roaringbitmap.RoaringBitmap;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+/**
+ * {@code rb_and(left BYTES, right BYTES) -> BYTES}
+ *
+ * <p>Returns the bitwise AND (intersection) of two serialized {@link RoaringBitmap} values. Returns
+ * {@code null} if either argument is null. Returns the intersection bytes even if the result is an
+ * empty bitmap, since both inputs were explicitly provided by the caller.
+ */
+public class RbAndFunction extends ScalarFunction {
+
+    /**
+     * @param leftBytes serialized left bitmap
+     * @param rightBytes serialized right bitmap
+     * @return intersection of left and right, or null if either argument is null
+     */
+    @Nullable
+    public byte[] eval(@Nullable byte[] leftBytes, @Nullable byte[] rightBytes) throws IOException {
+        if (leftBytes == null || rightBytes == null) {
+            return null;
+        }
+        RoaringBitmap left = BitmapUtils.fromBytes(leftBytes);
+        RoaringBitmap right = BitmapUtils.fromBytes(rightBytes);
+        left.and(right);
+        return BitmapUtils.toBytes(left);
+    }
+}

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/RbBuildFunction.java

@@ -0,0 +1,54 @@
+/*
+ * 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.fluss.flink.functions.bitmap;
+
+import org.apache.flink.table.functions.ScalarFunction;
+import org.roaringbitmap.RoaringBitmap;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+/**
+ * {@code rb_build(v1 INT, v2 INT, ...) -> BYTES}
+ *
+ * <p>Builds a serialized {@link RoaringBitmap} from a variadic list of 32-bit integer values within
+ * a single row. Unlike {@code rb_build_agg}, this function operates on individual column values in
+ * the same row rather than aggregating across rows. Null values are ignored. Returns {@code null}
+ * if all inputs are null or no inputs are provided.
+ */
+public class RbBuildFunction extends ScalarFunction {
+
+    /**
+     * @param values variadic integer values to add to the bitmap; null values are ignored
+     * @return serialized bitmap, or null if all inputs are null
+     */
+    @Nullable
+    public byte[] eval(@Nullable Integer... values) throws IOException {
+        if (values == null || values.length == 0) {
+            return null;
+        }
+        RoaringBitmap bitmap = new RoaringBitmap();
+        for (Integer v : values) {
+            if (v != null) {
+                bitmap.add(v);
+            }
+        }
+        return bitmap.isEmpty() ? null : BitmapUtils.toBytes(bitmap);
+    }
+}

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/RbCardinalityFunction.java

@@ -0,0 +1,47 @@
+/*
+ * 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.fluss.flink.functions.bitmap;
+
+import org.apache.flink.table.functions.ScalarFunction;
+import org.roaringbitmap.RoaringBitmap;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+/**
+ * {@code rb_cardinality(bitmap BYTES) -> BIGINT}
+ *
+ * <p>Returns the number of distinct integers in the serialized {@link RoaringBitmap}. Returns
+ * {@code null} for null or empty input.
+ */
+public class RbCardinalityFunction extends ScalarFunction {
+
+    /**
+     * @param bitmapBytes serialized RoaringBitmap; null returns null
+     * @return number of distinct integers, or null if input is null
+     */
+    @Nullable
+    public Long eval(@Nullable byte[] bitmapBytes) throws IOException {
+        if (bitmapBytes == null || bitmapBytes.length == 0) {
+            return null;
+        }
+        RoaringBitmap bitmap = BitmapUtils.fromBytes(bitmapBytes);
+        return bitmap.getLongCardinality();
+    }
+}

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/RbContainsFunction.java

@@ -0,0 +1,48 @@
+/*
+ * 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.fluss.flink.functions.bitmap;
+
+import org.apache.flink.table.functions.ScalarFunction;
+import org.roaringbitmap.RoaringBitmap;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+/**
+ * {@code rb_contains(bitmap BYTES, value INT) -> BOOLEAN}
+ *
+ * <p>Returns {@code true} if the serialized {@link RoaringBitmap} contains the given integer.
+ * Returns {@code null} if either argument is null.
+ */
+public class RbContainsFunction extends ScalarFunction {
+
+    /**
+     * @param bitmapBytes serialized RoaringBitmap
+     * @param value the integer to check
+     * @return true if the bitmap contains the value, null if either argument is null
+     */
+    @Nullable
+    public Boolean eval(@Nullable byte[] bitmapBytes, @Nullable Integer value) throws IOException {
+        if (bitmapBytes == null || value == null) {
+            return null;
+        }
+        RoaringBitmap bitmap = BitmapUtils.fromBytes(bitmapBytes);
+        return bitmap.contains(value);
+    }
+}

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/RbOrFunction.java

@@ -0,0 +1,56 @@
+/*
+ * 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.fluss.flink.functions.bitmap;
+
+import org.apache.flink.table.functions.ScalarFunction;
+import org.roaringbitmap.RoaringBitmap;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+/**
+ * {@code rb_or(left BYTES, right BYTES) -> BYTES}
+ *
+ * <p>Returns the bitwise OR (union) of two serialized {@link RoaringBitmap} values. If one argument
+ * is null, the other is returned as-is. Returns {@code null} only if both are null.
+ */
+public class RbOrFunction extends ScalarFunction {
+
+    /**
+     * @param leftBytes serialized left bitmap
+     * @param rightBytes serialized right bitmap
+     * @return union of left and right, or null if both are null
+     */
+    @Nullable
+    public byte[] eval(@Nullable byte[] leftBytes, @Nullable byte[] rightBytes) throws IOException {
+        if (leftBytes == null && rightBytes == null) {
+            return null;
+        }
+        if (leftBytes == null) {
+            return rightBytes;
+        }
+        if (rightBytes == null) {
+            return leftBytes;
+        }
+        RoaringBitmap left = BitmapUtils.fromBytes(leftBytes);
+        RoaringBitmap right = BitmapUtils.fromBytes(rightBytes);
+        left.or(right);
+        return BitmapUtils.toBytes(left);
+    }
+}

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/RbToArrayFunction.java

@@ -0,0 +1,51 @@
+/*
+ * 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.fluss.flink.functions.bitmap;
+
+import org.apache.flink.table.functions.ScalarFunction;
+import org.roaringbitmap.RoaringBitmap;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+/**
+ * {@code rb_to_array(bitmap BYTES) -> ARRAY<INT>}
+ *
+ * <p>Expands a serialized {@link RoaringBitmap} into an array of its integer values in ascending
+ * order. Returns {@code null} for null input. Returns an empty array for an empty bitmap.
+ *
+ * <p>Note: The return type must be {@code Integer[]} (boxed), not {@code int[]} (primitive).
+ * Flink's type system maps {@code int[]} to {@code BYTES}, whereas {@code Integer[]} correctly maps
+ * to {@code ARRAY<INT>}.
+ */
+public class RbToArrayFunction extends ScalarFunction {
+
+    /**
+     * @param bitmapBytes serialized RoaringBitmap; null returns null
+     * @return array of integers in ascending order, or null if input is null
+     */
+    @Nullable
+    public Integer[] eval(@Nullable byte[] bitmapBytes) throws IOException {
+        if (bitmapBytes == null) {
+            return null;
+        }
+        RoaringBitmap bitmap = BitmapUtils.fromBytes(bitmapBytes);
+        return bitmap.stream().boxed().toArray(Integer[]::new);
+    }
+}