Re: [PR] [FIP-37] Add bitmap Phase 2 functions and register via FlussCatalog [fluss]

["wuchong (via GitHub)" <gi...@apache.org>]

wuchong commented on code in PR #3777:
URL: https://github.com/apache/fluss/pull/3777#discussion_r3719096336


##########
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.

Review Comment:
   Returning `null` when XOR cancels to an empty bitmap conflates a valid aggregate result with the no-input/all-null case. For example, aggregating two identical non-null bitmaps makes `rb_cardinality(rb_xor_agg(...))` return SQL `NULL` instead of `0`, so downstream SQL cannot distinguish cancellation from an all-null group. Please track whether any non-null bitmap was accumulated (including through `merge`) and serialize an empty bitmap when input was seen.
   
   The test can reproduce this problem.
   
   ```java
   class RbFunctionsCatalogITCase {
   
       @Test
       void testRbXorAggIdenticalBitmapsReturnsEmptyBitmap() throws Exception {
           byte[] bitmap = BitmapUtils.toBytes(RoaringBitmap.bitmapOf(1, 2, 3));
   
           Table source =
                   tEnv.fromValues(
                           DataTypes.ROW(
                                   DataTypes.FIELD("k", DataTypes.INT()),
                                   DataTypes.FIELD("bmap", DataTypes.BYTES())),
                           Row.of(1, bitmap),
                           Row.of(1, bitmap));
           tEnv.createTemporaryView("identical_bitmaps", source);
   
           TableResult result =
                   tEnv.executeSql(
                           "SELECT rb_cardinality(rb_xor_agg(bmap)) "
                                   + "FROM identical_bitmaps GROUP BY k");
           List<Row> rows = CollectionUtil.iteratorToList(result.collect());
   
           assertThat(rows).hasSize(1);
           assertThat(rows.get(0).getField(0)).isEqualTo(0L);
       }
   }
   ```



##########
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(

Review Comment:
   Because this `retract` method is present, Flink can select it for updating inputs and call it for `UPDATE_BEFORE`/`DELETE` records; the unconditional exception then turns a normal changelog into a runtime job failure. XOR is self-inverse, so retraction can use the same XOR operation and null/empty handling as `accumulate` (or the function must be rejected before execution on non-append inputs). Flink also implemented this method: https://github.com/apache/flink/blob/3c8740f3f24aaa407112b5b576e5bfbf9cf2d24a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/aggregate/AbstractBitmapXorWithRetractAggFunction.java



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org