diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java index 17194c05b85..1beb69288b8 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java @@ -27,6 +27,7 @@ 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; @@ -34,6 +35,8 @@ 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,6 +157,7 @@ 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()); @@ -161,6 +165,8 @@ public class FlinkCatalog extends AbstractCatalog { 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); } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/RbAndNotFunction.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/RbAndNotFunction.java new file mode 100644 index 00000000000..0be55229303 --- /dev/null +++ b/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} + * + *

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); + } +} diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/RbXorAggFunction.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/RbXorAggFunction.java new file mode 100644 index 00000000000..e08c4a7d12b --- /dev/null +++ b/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} + * + *

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. + * + *

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. + * + *

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 it) { + for (RoaringBitmap other : it) { + if (other != null) { + acc.xor(other); + } + } + } +} diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/RbXorFunction.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/RbXorFunction.java new file mode 100644 index 00000000000..3b1c01cb908 --- /dev/null +++ b/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} + * + *

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); + } +} diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java index 6b087803903..3c390a1c857 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java +++ b/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 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(); diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/functions/bitmap/RbAggFunctionsTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/functions/bitmap/RbAggFunctionsTest.java index 1f9e9149635..05cab1d8e27 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/functions/bitmap/RbAggFunctionsTest.java +++ b/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(); + } } diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/functions/bitmap/RbFunctionsCatalogITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/functions/bitmap/RbFunctionsCatalogITCase.java index f76647dc19c..b4cb568d1fd 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/functions/bitmap/RbFunctionsCatalogITCase.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/functions/bitmap/RbFunctionsCatalogITCase.java @@ -318,4 +318,68 @@ void testRbBuildFromArrayResolvedFromCatalog() throws Exception { assertThat(rows).hasSize(1); assertThat(rows.get(0).getField(0)).isEqualTo(3L); // duplicate 2 ignored } + + @Test + void testRbXorAggResolvedFromCatalog() throws Exception { + byte[] bitmap1 = BitmapUtils.toBytes(RoaringBitmap.bitmapOf(1, 2, 3)); + byte[] bitmap2 = BitmapUtils.toBytes(RoaringBitmap.bitmapOf(2, 3, 4)); + + Table source = + tEnv.fromValues( + DataTypes.ROW( + DataTypes.FIELD("k", DataTypes.INT()), + DataTypes.FIELD("bmap", DataTypes.BYTES())), + Row.of(1, bitmap1), + Row.of(1, bitmap2)); + tEnv.createTemporaryView("bitmaps", source); + + TableResult result = tEnv.executeSql("SELECT k, rb_xor_agg(bmap) FROM bitmaps GROUP BY k"); + List rows = CollectionUtil.iteratorToList(result.collect()); + assertThat(rows).hasSize(1); + + byte[] resultBytes = (byte[]) rows.get(0).getField(1); + RoaringBitmap xor = BitmapUtils.fromBytes(resultBytes); + // XOR of {1,2,3} and {2,3,4} = {1,4} + assertThat(xor.getLongCardinality()).isEqualTo(2L); + assertThat(xor.contains(1)).isTrue(); + assertThat(xor.contains(4)).isTrue(); + } + + @Test + void testRbXorScalarResolvedFromCatalog() throws Exception { + byte[] left = BitmapUtils.toBytes(RoaringBitmap.bitmapOf(1, 2, 3)); + byte[] right = BitmapUtils.toBytes(RoaringBitmap.bitmapOf(2, 3, 4)); + + Table source = + tEnv.fromValues( + DataTypes.ROW( + DataTypes.FIELD("l", DataTypes.BYTES()), + DataTypes.FIELD("r", DataTypes.BYTES())), + Row.of(left, right)); + tEnv.createTemporaryView("bitmaps", source); + + TableResult result = tEnv.executeSql("SELECT rb_cardinality(rb_xor(l, r)) FROM bitmaps"); + List rows = CollectionUtil.iteratorToList(result.collect()); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getField(0)).isEqualTo(2L); // {1, 4} + } + + @Test + void testRbAndNotResolvedFromCatalog() throws Exception { + byte[] left = BitmapUtils.toBytes(RoaringBitmap.bitmapOf(1, 2, 3, 4)); + byte[] right = BitmapUtils.toBytes(RoaringBitmap.bitmapOf(3, 4, 5)); + + Table source = + tEnv.fromValues( + DataTypes.ROW( + DataTypes.FIELD("l", DataTypes.BYTES()), + DataTypes.FIELD("r", DataTypes.BYTES())), + Row.of(left, right)); + tEnv.createTemporaryView("bitmaps", source); + + TableResult result = tEnv.executeSql("SELECT rb_cardinality(rb_andnot(l, r)) FROM bitmaps"); + List rows = CollectionUtil.iteratorToList(result.collect()); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getField(0)).isEqualTo(2L); // {1, 2} + } } diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/functions/bitmap/RbScalarFunctionsTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/functions/bitmap/RbScalarFunctionsTest.java index 0d7a4b3d05e..5918ee9cc1d 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/functions/bitmap/RbScalarFunctionsTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/functions/bitmap/RbScalarFunctionsTest.java @@ -25,7 +25,7 @@ import static org.assertj.core.api.Assertions.assertThat; -/** Unit tests for the six Phase 1 scalar bitmap functions. */ +/** Unit tests for the Phase 1 and Phase 2 scalar bitmap functions. */ class RbScalarFunctionsTest { // ------------------------------------------------------------------------- @@ -252,4 +252,118 @@ void testAndEmptyIntersectionReturnsBytes() throws IOException { RoaringBitmap intersection = BitmapUtils.fromBytes(result); assertThat(intersection.isEmpty()).isTrue(); } + + // ------------------------------------------------------------------------- + // rb_xor (scalar) + // ------------------------------------------------------------------------- + + @Test + void testXorBasic() throws IOException { + RbXorFunction fn = new RbXorFunction(); + byte[] left = BitmapUtils.toBytes(RoaringBitmap.bitmapOf(1, 2, 3)); + byte[] right = BitmapUtils.toBytes(RoaringBitmap.bitmapOf(2, 3, 4)); + byte[] result = fn.eval(left, right); + 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 testXorLeftNull() throws IOException { + byte[] right = BitmapUtils.toBytes(RoaringBitmap.bitmapOf(1, 2)); + assertThat(new RbXorFunction().eval(null, right)).isNull(); + } + + @Test + void testXorRightNull() throws IOException { + byte[] left = BitmapUtils.toBytes(RoaringBitmap.bitmapOf(1, 2)); + assertThat(new RbXorFunction().eval(left, null)).isNull(); + } + + @Test + void testXorBothNull() throws IOException { + assertThat(new RbXorFunction().eval(null, null)).isNull(); + } + + @Test + void testXorDisjointSets() throws IOException { + RbXorFunction fn = new RbXorFunction(); + byte[] left = BitmapUtils.toBytes(RoaringBitmap.bitmapOf(1, 2)); + byte[] right = BitmapUtils.toBytes(RoaringBitmap.bitmapOf(3, 4)); + // XOR of disjoint sets = union + RoaringBitmap result = BitmapUtils.fromBytes(fn.eval(left, right)); + assertThat(result.getLongCardinality()).isEqualTo(4L); + } + + @Test + void testXorIdenticalSetsProducesEmpty() throws IOException { + RbXorFunction fn = new RbXorFunction(); + byte[] left = BitmapUtils.toBytes(RoaringBitmap.bitmapOf(1, 2, 3)); + byte[] right = BitmapUtils.toBytes(RoaringBitmap.bitmapOf(1, 2, 3)); + // XOR of identical sets = empty bitmap (not null, since both inputs were provided) + byte[] result = fn.eval(left, right); + assertThat(result).isNotNull(); + RoaringBitmap restored = BitmapUtils.fromBytes(result); + assertThat(restored.isEmpty()).isTrue(); + } + + // ------------------------------------------------------------------------- + // rb_andnot (scalar) + // ------------------------------------------------------------------------- + + @Test + void testAndNotBasic() throws IOException { + RbAndNotFunction fn = new RbAndNotFunction(); + byte[] left = BitmapUtils.toBytes(RoaringBitmap.bitmapOf(1, 2, 3, 4)); + byte[] right = BitmapUtils.toBytes(RoaringBitmap.bitmapOf(3, 4, 5)); + byte[] result = fn.eval(left, right); + assertThat(result).isNotNull(); + RoaringBitmap restored = BitmapUtils.fromBytes(result); + assertThat(restored.getLongCardinality()).isEqualTo(2L); + assertThat(restored.contains(1)).isTrue(); + assertThat(restored.contains(2)).isTrue(); + assertThat(restored.contains(3)).isFalse(); + assertThat(restored.contains(4)).isFalse(); + } + + @Test + void testAndNotLeftNull() throws IOException { + byte[] right = BitmapUtils.toBytes(RoaringBitmap.bitmapOf(1, 2)); + assertThat(new RbAndNotFunction().eval(null, right)).isNull(); + } + + @Test + void testAndNotRightNull() throws IOException { + byte[] left = BitmapUtils.toBytes(RoaringBitmap.bitmapOf(1, 2)); + assertThat(new RbAndNotFunction().eval(left, null)).isNull(); + } + + @Test + void testAndNotBothNull() throws IOException { + assertThat(new RbAndNotFunction().eval(null, null)).isNull(); + } + + @Test + void testAndNotRightIsSuperset() throws IOException { + RbAndNotFunction fn = new RbAndNotFunction(); + byte[] left = BitmapUtils.toBytes(RoaringBitmap.bitmapOf(1, 2)); + byte[] right = BitmapUtils.toBytes(RoaringBitmap.bitmapOf(1, 2, 3)); + RoaringBitmap result = BitmapUtils.fromBytes(fn.eval(left, right)); + assertThat(result.isEmpty()).isTrue(); + } + + @Test + void testAndNotDisjointSets() throws IOException { + RbAndNotFunction fn = new RbAndNotFunction(); + byte[] left = BitmapUtils.toBytes(RoaringBitmap.bitmapOf(1, 2)); + byte[] right = BitmapUtils.toBytes(RoaringBitmap.bitmapOf(3, 4)); + RoaringBitmap result = BitmapUtils.fromBytes(fn.eval(left, right)); + assertThat(result.getLongCardinality()).isEqualTo(2L); + assertThat(result.contains(1)).isTrue(); + assertThat(result.contains(2)).isTrue(); + } }