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

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

@@ -27,13 +27,16 @@
 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.RbAndNotFunction;
 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.functions.bitmap.RbXorAggFunction;
+import org.apache.fluss.flink.functions.bitmap.RbXorFunction;
 import org.apache.fluss.flink.lake.LakeFlinkCatalog;
 import org.apache.fluss.flink.procedure.ProcedureManager;
 import org.apache.fluss.flink.utils.CatalogExceptionUtils;
@@ -154,13 +157,16 @@ public class FlinkCatalog extends AbstractCatalog {
         map.put("rb_build_agg", RbBuildAggFunction.class.getName());
         map.put("rb_or_agg", RbOrAggFunction.class.getName());
         map.put("rb_and_agg", RbAndAggFunction.class.getName());
+        map.put("rb_xor_agg", RbXorAggFunction.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());
+        map.put("rb_xor", RbXorFunction.class.getName());
+        map.put("rb_andnot", RbAndNotFunction.class.getName());
         BUILTIN_BITMAP_FUNCTIONS = Collections.unmodifiableMap(map);
     }
 

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

@@ -0,0 +1,52 @@
+/*
+ * 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_andnot(left BYTES, right BYTES) -> BYTES}
+ *
+ * <p>Returns the difference (AND NOT) of two serialized {@link RoaringBitmap} values. The result
+ * contains elements present in the left bitmap but not in the right bitmap. Useful for exclusion
+ * analysis such as "users who visited page A but not page B." Returns {@code null} if either
+ * argument is null.
+ */
+public class RbAndNotFunction extends ScalarFunction {
+
+    /**
+     * @param leftBytes serialized left bitmap
+     * @param rightBytes serialized right bitmap
+     * @return elements in left but not in 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.andNot(right);
+        return BitmapUtils.toBytes(left);
+    }
+}

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

@@ -0,0 +1,73 @@
+/*
+ * 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.roaringbitmap.RoaringBitmap;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+/**
+ * {@code rb_xor_agg(bitmap BYTES) -> BYTES}
+ *
+ * <p>Aggregates multiple serialized {@link RoaringBitmap} values using bitwise XOR across rows.
+ * Returns elements that appear in an odd number of input bitmaps — useful for change detection and
+ * symmetric difference analysis.
+ *
+ * <p>Note: there is no server-side {@code FieldRoaringBitmapXorAgg} counterpart. This function
+ * executes entirely in Flink. Users should be aware that combining it with {@code
+ * table.merge-engine=aggregation} may produce unexpected results during server-side compaction.
+ *
+ * <p>Null and empty inputs are ignored. Returns {@code null} when all inputs are null or the result
+ * fully cancels to an empty bitmap.
+ */
+public class RbXorAggFunction extends AbstractRbAggFunction {
+
+    /**
+     * XORs the input bitmap into the accumulator.
+     *
+     * @param acc the running bitmap accumulator
+     * @param bitmapBytes serialized RoaringBitmap bytes; null and empty arrays are ignored
+     */
+    public void accumulate(RoaringBitmap acc, @Nullable byte[] bitmapBytes) throws IOException {
+        if (bitmapBytes == null || bitmapBytes.length == 0) {
+            return;
+        }
+        acc.xor(BitmapUtils.fromBytes(bitmapBytes));
+    }
+
+    /**
+     * Retraction is not supported for bitmap XOR aggregation.
+     *
+     * @throws UnsupportedOperationException always
+     */
+    public void retract(RoaringBitmap acc, @Nullable byte[] bitmapBytes) {
+        throw new UnsupportedOperationException(
+                "rb_xor_agg does not support retraction. " + "Use it only on append-only streams.");
+    }
+
+    @Override
+    public void merge(RoaringBitmap acc, Iterable<RoaringBitmap> it) {
+        for (RoaringBitmap other : it) {
+            if (other != null) {
+                acc.xor(other);
+            }
+        }
+    }
+}

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/RbXorFunction.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_xor(left BYTES, right BYTES) -> BYTES}
+ *
+ * <p>Returns the bitwise XOR (symmetric difference) of two serialized {@link RoaringBitmap} values.
+ * The result contains elements that appear in exactly one of the two bitmaps. Returns {@code null}
+ * if either argument is null.
+ */
+public class RbXorFunction extends ScalarFunction {
+
+    /**
+     * @param leftBytes serialized left bitmap
+     * @param rightBytes serialized right bitmap
+     * @return symmetric difference 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.xor(right);
+        return BitmapUtils.toBytes(left);
+    }
+}

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java

@@ -1012,12 +1012,15 @@ void testViewsAndFunctions() throws Exception {
                         "rb_build_agg",
                         "rb_or_agg",
                         "rb_and_agg",
+                        "rb_xor_agg",
                         "rb_cardinality",
                         "rb_build",
                         "rb_contains",
                         "rb_to_array",
                         "rb_or",
-                        "rb_and");
+                        "rb_and",
+                        "rb_xor",
+                        "rb_andnot");
 
         ObjectPath functionPath = new ObjectPath(DEFAULT_DB, "testFunction");
         assertThat(catalog.functionExists(functionPath)).isFalse();
@@ -1118,7 +1121,7 @@ void testBitmapFunctionsRegistered() throws Exception {
         List<String> functions = catalog.listFunctions(DEFAULT_DB);
 
         // aggregate functions
-        assertThat(functions).contains("rb_build_agg", "rb_or_agg", "rb_and_agg");
+        assertThat(functions).contains("rb_build_agg", "rb_or_agg", "rb_and_agg", "rb_xor_agg");
         // scalar functions
         assertThat(functions)
                 .contains(
@@ -1127,7 +1130,9 @@ void testBitmapFunctionsRegistered() throws Exception {
                         "rb_contains",
                         "rb_to_array",
                         "rb_or",
-                        "rb_and");
+                        "rb_and",
+                        "rb_xor",
+                        "rb_andnot");
 
         // verify each function exists and resolves to the correct class
         assertThat(catalog.functionExists(new ObjectPath(DEFAULT_DB, "rb_cardinality"))).isTrue();
@@ -1136,6 +1141,9 @@ void testBitmapFunctionsRegistered() throws Exception {
         assertThat(catalog.functionExists(new ObjectPath(DEFAULT_DB, "rb_to_array"))).isTrue();
         assertThat(catalog.functionExists(new ObjectPath(DEFAULT_DB, "rb_or"))).isTrue();
         assertThat(catalog.functionExists(new ObjectPath(DEFAULT_DB, "rb_and"))).isTrue();
+        assertThat(catalog.functionExists(new ObjectPath(DEFAULT_DB, "rb_xor_agg"))).isTrue();
+        assertThat(catalog.functionExists(new ObjectPath(DEFAULT_DB, "rb_xor"))).isTrue();
+        assertThat(catalog.functionExists(new ObjectPath(DEFAULT_DB, "rb_andnot"))).isTrue();
 
         // verify unknown still returns false
         assertThat(catalog.functionExists(new ObjectPath(DEFAULT_DB, "unknown_fn"))).isFalse();

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/functions/bitmap/RbAggFunctionsTest.java

@@ -30,7 +30,10 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
-/** Unit tests for {@link RbBuildAggFunction}, {@link RbOrAggFunction}, {@link RbAndAggFunction}. */
+/**
+ * Unit tests for {@link RbBuildAggFunction}, {@link RbOrAggFunction}, {@link RbAndAggFunction},
+ * {@link RbXorAggFunction}.
+ */
 class RbAggFunctionsTest {
 
     // -------------------------------------------------------------------------
@@ -340,4 +343,103 @@ void testAndAggAccumulatorSerializerSnapshotNotNull() {
         assertThat(RbAndAggFunction.AccumulatorSerializer.INSTANCE.snapshotConfiguration())
                 .isNotNull();
     }
+
+    // -------------------------------------------------------------------------
+    // RbXorAggFunction
+    // -------------------------------------------------------------------------
+
+    @Test
+    void testXorAggBasic() throws IOException {
+        RbXorAggFunction fn = new RbXorAggFunction();
+        RoaringBitmap acc = fn.createAccumulator();
+        fn.accumulate(acc, BitmapUtils.toBytes(RoaringBitmap.bitmapOf(1, 2, 3)));
+        fn.accumulate(acc, BitmapUtils.toBytes(RoaringBitmap.bitmapOf(2, 3, 4)));
+        // XOR: elements in exactly one bitmap = {1, 4}
+        byte[] result = fn.getValue(acc);
+        assertThat(result).isNotNull();
+        RoaringBitmap restored = BitmapUtils.fromBytes(result);
+        assertThat(restored.getLongCardinality()).isEqualTo(2L);
+        assertThat(restored.contains(1)).isTrue();
+        assertThat(restored.contains(4)).isTrue();
+        assertThat(restored.contains(2)).isFalse();
+        assertThat(restored.contains(3)).isFalse();
+    }
+
+    @Test
+    void testXorAggNullInputIgnored() throws IOException {
+        RbXorAggFunction fn = new RbXorAggFunction();
+        RoaringBitmap acc = fn.createAccumulator();
+        fn.accumulate(acc, null);
+        fn.accumulate(acc, new byte[0]);
+        fn.accumulate(acc, BitmapUtils.toBytes(RoaringBitmap.bitmapOf(1, 2)));
+        byte[] result = fn.getValue(acc);
+        assertThat(result).isNotNull();
+        RoaringBitmap restored = BitmapUtils.fromBytes(result);
+        assertThat(restored.getLongCardinality()).isEqualTo(2L);
+    }
+
+    @Test
+    void testXorAggEmptyReturnsNull() {
+        RbXorAggFunction fn = new RbXorAggFunction();
+        assertThat(fn.getValue(fn.createAccumulator())).isNull();
+    }
+
+    @Test
+    void testXorAggMerge() throws IOException {
+        RbXorAggFunction fn = new RbXorAggFunction();
+        RoaringBitmap acc1 = fn.createAccumulator();
+        fn.accumulate(acc1, BitmapUtils.toBytes(RoaringBitmap.bitmapOf(1, 2, 3)));
+        RoaringBitmap acc2 = fn.createAccumulator();
+        fn.accumulate(acc2, BitmapUtils.toBytes(RoaringBitmap.bitmapOf(3, 4, 5)));
+        fn.merge(acc1, Collections.singletonList(acc2));
+        // XOR merge: {1,2,3} XOR {3,4,5} = {1,2,4,5}
+        byte[] result = fn.getValue(acc1);
+        assertThat(result).isNotNull();
+        RoaringBitmap restored = BitmapUtils.fromBytes(result);
+        assertThat(restored.getLongCardinality()).isEqualTo(4L);
+        assertThat(restored.contains(1)).isTrue();
+        assertThat(restored.contains(2)).isTrue();
+        assertThat(restored.contains(4)).isTrue();
+        assertThat(restored.contains(5)).isTrue();
+        assertThat(restored.contains(3)).isFalse();
+    }
+
+    @Test
+    void testXorAggRetractThrows() {
+        RbXorAggFunction fn = new RbXorAggFunction();
+        assertThatThrownBy(() -> fn.retract(fn.createAccumulator(), new byte[0]))
+                .isInstanceOf(UnsupportedOperationException.class);
+    }
+
+    @Test
+    void testXorAggMergeCancelsToEmpty() throws IOException {
+        // Two partials each holding {1} — XOR merge should cancel to empty, getValue returns null
+        RbXorAggFunction fn = new RbXorAggFunction();
+        RoaringBitmap acc1 = fn.createAccumulator();
+        fn.accumulate(acc1, BitmapUtils.toBytes(RoaringBitmap.bitmapOf(1)));
+        RoaringBitmap acc2 = fn.createAccumulator();
+        fn.accumulate(acc2, BitmapUtils.toBytes(RoaringBitmap.bitmapOf(1)));
+        fn.merge(acc1, Collections.singletonList(acc2));
+        // {1} XOR {1} = empty
+        assertThat(fn.getValue(acc1)).isNull();
+    }
+
+    @Test
+    void testXorAggThreeInputs() throws IOException {
+        // Exercises the "odd number of inputs" contract
+        // {1,2} XOR {2,3} XOR {3,4} = {1,4}
+        RbXorAggFunction fn = new RbXorAggFunction();
+        RoaringBitmap acc = fn.createAccumulator();
+        fn.accumulate(acc, BitmapUtils.toBytes(RoaringBitmap.bitmapOf(1, 2)));
+        fn.accumulate(acc, BitmapUtils.toBytes(RoaringBitmap.bitmapOf(2, 3)));
+        fn.accumulate(acc, BitmapUtils.toBytes(RoaringBitmap.bitmapOf(3, 4)));
+        byte[] result = fn.getValue(acc);
+        assertThat(result).isNotNull();
+        RoaringBitmap restored = BitmapUtils.fromBytes(result);
+        assertThat(restored.getLongCardinality()).isEqualTo(2L);
+        assertThat(restored.contains(1)).isTrue();
+        assertThat(restored.contains(4)).isTrue();
+        assertThat(restored.contains(2)).isFalse();
+        assertThat(restored.contains(3)).isFalse();
+    }
 }