This is an automated email from the ASF dual-hosted git repository.
polyzos pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 68ad16c79 [FIP-37] Add bitmap infrastructure: BitmapUtils,
RoaringBitmapSerializer, AbstractRbAggFunction (#3319)
68ad16c79 is described below
commit 68ad16c799a4bbf19be9582c1f5b2209beca2503
Author: Prajwal Banakar <[email protected]>
AuthorDate: Tue May 26 14:48:58 2026 +0530
[FIP-37] Add bitmap infrastructure: BitmapUtils, RoaringBitmapSerializer,
AbstractRbAggFunction (#3319)
* [FIP-37] Add bitmap infrastructure: BitmapUtils, RoaringBitmapSerializer,
AbstractRbAggFunction
* Add RoaringBitmapSerializer tests and fix jacoco coverage
* Add missing RoaringBitmapTypeInfo coverage tests
* Address review comments for bitmap infrastructure
* small improvements
---------
Co-authored-by: ipolyzos <[email protected]>
---
fluss-flink/fluss-flink-common/pom.xml | 7 +
.../functions/bitmap/AbstractRbAggFunction.java | 79 ++++++
.../fluss/flink/functions/bitmap/BitmapUtils.java | 70 +++++
.../functions/bitmap/RoaringBitmapSerializer.java | 114 +++++++++
.../functions/bitmap/RoaringBitmapTypeInfo.java | 98 +++++++
.../bitmap/AbstractRbAggFunctionITCase.java | 91 +++++++
.../flink/functions/bitmap/BitmapUtilsTest.java | 100 ++++++++
.../bitmap/RoaringBitmapSerializerTest.java | 281 +++++++++++++++++++++
8 files changed, 840 insertions(+)
diff --git a/fluss-flink/fluss-flink-common/pom.xml
b/fluss-flink/fluss-flink-common/pom.xml
index 026649f54..c1835466f 100644
--- a/fluss-flink/fluss-flink-common/pom.xml
+++ b/fluss-flink/fluss-flink-common/pom.xml
@@ -95,6 +95,13 @@
<scope>provided</scope>
</dependency>
+ <!-- RoaringBitmap for bitmap SQL functions (FIP-37) -->
+ <dependency>
+ <groupId>org.roaringbitmap</groupId>
+ <artifactId>RoaringBitmap</artifactId>
+ <version>${roaringbitmap.version}</version>
+ </dependency>
+
<!-- test dependency -->
<dependency>
<groupId>org.apache.fluss</groupId>
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/AbstractRbAggFunction.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/AbstractRbAggFunction.java
new file mode 100644
index 000000000..1690a757f
--- /dev/null
+++
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/AbstractRbAggFunction.java
@@ -0,0 +1,79 @@
+/*
+ * 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.fluss.exception.FlussRuntimeException;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.FunctionHint;
+import org.apache.flink.table.functions.AggregateFunction;
+import org.roaringbitmap.RoaringBitmap;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+/**
+ * Shared base for bitmap aggregate UDFs that use {@link RoaringBitmap} as the
accumulator.
+ *
+ * <p>The {@code @FunctionHint} annotation with {@code accumulator =
@DataTypeHint("RAW")} tells
+ * Flink's Table planner to skip reflection-based POJO extraction and instead
use the {@link
+ * TypeInformation} returned by {@link #getAccumulatorType()}, which provides
the custom {@link
+ * RoaringBitmapSerializer}. Without this annotation, Flink attempts POJO
field extraction on
+ * RoaringBitmap and fails.
+ */
+@FunctionHint(accumulator = @DataTypeHint(value = "RAW", bridgedTo =
RoaringBitmap.class))
+abstract class AbstractRbAggFunction extends AggregateFunction<byte[],
RoaringBitmap> {
+
+ @Override
+ public RoaringBitmap createAccumulator() {
+ return new RoaringBitmap();
+ }
+
+ /** Merges partial accumulators, required for two-phase aggregation in the
Flink Table API. */
+ public void merge(RoaringBitmap acc, Iterable<RoaringBitmap> it) {
+ for (RoaringBitmap other : it) {
+ if (other != null) {
+ acc.or(other);
+ }
+ }
+ }
+
+ public void resetAccumulator(RoaringBitmap acc) {
+ acc.clear();
+ }
+
+ @Override
+ @Nullable
+ public byte[] getValue(RoaringBitmap acc) {
+ if (acc == null || acc.isEmpty()) {
+ return null;
+ }
+ try {
+ return BitmapUtils.toBytes(acc);
+ } catch (IOException e) {
+ throw new FlussRuntimeException("Failed to serialize bitmap
accumulator.", e);
+ }
+ }
+
+ @Override
+ public TypeInformation<RoaringBitmap> getAccumulatorType() {
+ return RoaringBitmapTypeInfo.INSTANCE;
+ }
+}
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/BitmapUtils.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/BitmapUtils.java
new file mode 100644
index 000000000..48c249d64
--- /dev/null
+++
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/BitmapUtils.java
@@ -0,0 +1,70 @@
+/*
+ * 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;
+import java.nio.ByteBuffer;
+
+/**
+ * Utility methods for serializing and deserializing {@link RoaringBitmap}.
+ *
+ * <p>Uses the ByteBuffer-based serialization approach, which is the preferred
method recommended by
+ * the RoaringBitmap library. This format is compatible with the server-side
{@code
+ * RoaringBitmapUtils.serializeRoaringBitmap32} used by {@code
FieldRoaringBitmap32Agg}.
+ */
+public final class BitmapUtils {
+
+ private BitmapUtils() {}
+
+ /**
+ * Serializes a {@link RoaringBitmap} to a byte array.
+ *
+ * @param bitmap the bitmap to serialize; null returns null
+ * @return serialized byte array, or null if input is null
+ */
+ @Nullable
+ public static byte[] toBytes(@Nullable RoaringBitmap bitmap) throws
IOException {
+ if (bitmap == null) {
+ return null;
+ }
+ bitmap.runOptimize();
+ ByteBuffer buffer =
ByteBuffer.allocate(bitmap.serializedSizeInBytes());
+ bitmap.serialize(buffer);
+ return buffer.array();
+ }
+
+ /**
+ * Deserializes a {@link RoaringBitmap} from a byte array.
+ *
+ * @param bytes the serialized bitmap bytes; null returns null
+ * @return deserialized RoaringBitmap, or null if input is null
+ */
+ @Nullable
+ public static RoaringBitmap fromBytes(@Nullable byte[] bytes) throws
IOException {
+ if (bytes == null) {
+ return null;
+ }
+ RoaringBitmap bitmap = new RoaringBitmap();
+ bitmap.deserialize(ByteBuffer.wrap(bytes));
+ return bitmap;
+ }
+}
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/RoaringBitmapSerializer.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/RoaringBitmapSerializer.java
new file mode 100644
index 000000000..27ad9e927
--- /dev/null
+++
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/RoaringBitmapSerializer.java
@@ -0,0 +1,114 @@
+/*
+ * 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.api.common.typeutils.SimpleTypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.base.TypeSerializerSingleton;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.roaringbitmap.RoaringBitmap;
+
+import javax.annotation.concurrent.ThreadSafe;
+
+import java.io.IOException;
+
+/**
+ * Flink {@link org.apache.flink.api.common.typeutils.TypeSerializer} for
{@link RoaringBitmap}.
+ *
+ * <p>Used as the accumulator serializer for bitmap aggregate functions to
ensure correct
+ * checkpoint/savepoint behavior. Without a custom serializer, Flink falls
back to Kryo which is
+ * sensitive to internal class layout changes across RoaringBitmap library
versions.
+ */
+@ThreadSafe
+public final class RoaringBitmapSerializer extends
TypeSerializerSingleton<RoaringBitmap> {
+
+ public static final RoaringBitmapSerializer INSTANCE = new
RoaringBitmapSerializer();
+
+ private static final long serialVersionUID = 1L;
+
+ private RoaringBitmapSerializer() {}
+
+ @Override
+ public boolean isImmutableType() {
+ return false;
+ }
+
+ @Override
+ public RoaringBitmap createInstance() {
+ return new RoaringBitmap();
+ }
+
+ @Override
+ public RoaringBitmap copy(RoaringBitmap from) {
+ return from.clone();
+ }
+
+ @Override
+ public RoaringBitmap copy(RoaringBitmap from, RoaringBitmap reuse) {
+ return from.clone();
+ }
+
+ @Override
+ public int getLength() {
+ return -1;
+ }
+
+ @Override
+ public void serialize(RoaringBitmap record, DataOutputView target) throws
IOException {
+ byte[] bytes = BitmapUtils.toBytes(record);
+ target.writeInt(bytes.length);
+ target.write(bytes);
+ }
+
+ @Override
+ public RoaringBitmap deserialize(DataInputView source) throws IOException {
+ int size = source.readInt();
+ byte[] bytes = new byte[size];
+ source.readFully(bytes);
+ return BitmapUtils.fromBytes(bytes);
+ }
+
+ @Override
+ public RoaringBitmap deserialize(RoaringBitmap reuse, DataInputView
source) throws IOException {
+ return deserialize(source);
+ }
+
+ @Override
+ public void copy(DataInputView source, DataOutputView target) throws
IOException {
+ int size = source.readInt();
+ target.writeInt(size);
+ byte[] buffer = new byte[size];
+ source.readFully(buffer);
+ target.write(buffer);
+ }
+
+ @Override
+ public TypeSerializerSnapshot<RoaringBitmap> snapshotConfiguration() {
+ return new RoaringBitmapSerializerSnapshot();
+ }
+
+ /** Snapshot for {@link RoaringBitmapSerializer}. */
+ public static final class RoaringBitmapSerializerSnapshot
+ extends SimpleTypeSerializerSnapshot<RoaringBitmap> {
+
+ public RoaringBitmapSerializerSnapshot() {
+ super(() -> INSTANCE);
+ }
+ }
+}
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/RoaringBitmapTypeInfo.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/RoaringBitmapTypeInfo.java
new file mode 100644
index 000000000..fe0905523
--- /dev/null
+++
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/RoaringBitmapTypeInfo.java
@@ -0,0 +1,98 @@
+/*
+ * 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.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.roaringbitmap.RoaringBitmap;
+
+import javax.annotation.concurrent.ThreadSafe;
+
+import java.util.Objects;
+
+/**
+ * {@link TypeInformation} for {@link RoaringBitmap}.
+ *
+ * <p>Provides the custom {@link RoaringBitmapSerializer} to Flink's type
system, ensuring correct
+ * checkpoint and savepoint behavior for bitmap aggregate function
accumulators.
+ */
+@ThreadSafe
+public final class RoaringBitmapTypeInfo extends
TypeInformation<RoaringBitmap> {
+
+ public static final RoaringBitmapTypeInfo INSTANCE = new
RoaringBitmapTypeInfo();
+
+ private static final long serialVersionUID = 1L;
+
+ private RoaringBitmapTypeInfo() {}
+
+ @Override
+ public boolean isBasicType() {
+ return false;
+ }
+
+ @Override
+ public boolean isTupleType() {
+ return false;
+ }
+
+ @Override
+ public int getArity() {
+ return 1;
+ }
+
+ @Override
+ public int getTotalFields() {
+ return 1;
+ }
+
+ @Override
+ public Class<RoaringBitmap> getTypeClass() {
+ return RoaringBitmap.class;
+ }
+
+ @Override
+ public boolean isKeyType() {
+ return false;
+ }
+
+ @Override
+ public TypeSerializer<RoaringBitmap> createSerializer(ExecutionConfig
config) {
+ return RoaringBitmapSerializer.INSTANCE;
+ }
+
+ @Override
+ public String toString() {
+ return "RoaringBitmapTypeInfo";
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ return obj instanceof RoaringBitmapTypeInfo &&
((RoaringBitmapTypeInfo) obj).canEqual(this);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(getTypeClass());
+ }
+
+ @Override
+ public boolean canEqual(Object obj) {
+ return obj instanceof RoaringBitmapTypeInfo;
+ }
+}
diff --git
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/functions/bitmap/AbstractRbAggFunctionITCase.java
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/functions/bitmap/AbstractRbAggFunctionITCase.java
new file mode 100644
index 000000000..51a3e2a50
--- /dev/null
+++
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/functions/bitmap/AbstractRbAggFunctionITCase.java
@@ -0,0 +1,91 @@
+/*
+ * 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.api.EnvironmentSettings;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CollectionUtil;
+import org.junit.jupiter.api.Test;
+import org.roaringbitmap.RoaringBitmap;
+
+import java.util.List;
+
+import static org.apache.flink.table.api.Expressions.$;
+import static org.apache.flink.table.api.Expressions.call;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * End-to-end smoke test that validates the Flink Table planner accepts the
{@link
+ * AbstractRbAggFunction} contract — namely the {@code
@FunctionHint(accumulator
+ * = @DataTypeHint(value = "RAW", bridgedTo = RoaringBitmap.class))} hint
together with the custom
+ * {@link RoaringBitmapTypeInfo} returned from {@link
AbstractRbAggFunction#getAccumulatorType()}.
+ *
+ * <p>Direct method-level invocation of the aggregate function does not
exercise the Flink planner's
+ * type-extraction path, so this IT test wires a minimal concrete subclass
through a batch {@link
+ * TableEnvironment} to verify the planner accepts and runs it.
+ */
+class AbstractRbAggFunctionITCase {
+
+ /** Minimal concrete subclass used to exercise the planner. */
+ public static final class TestRbAggFunction extends AbstractRbAggFunction {
+ public void accumulate(RoaringBitmap acc, Integer value) {
+ if (value != null) {
+ acc.add(value);
+ }
+ }
+ }
+
+ @Test
+ void testPlannerAcceptsRawAccumulatorHintAndProducesCorrectResult() throws
Exception {
+ TableEnvironment tEnv =
TableEnvironment.create(EnvironmentSettings.inBatchMode());
+ tEnv.createTemporarySystemFunction("test_rb_agg", new
TestRbAggFunction());
+
+ Table source = tEnv.fromValues(1, 2, 3, 1, 2).as("v");
+ Table result = source.select(call("test_rb_agg", $("v")));
+
+ List<Row> rows =
CollectionUtil.iteratorToList(result.execute().collect());
+ assertThat(rows).hasSize(1);
+
+ byte[] bytes = (byte[]) rows.get(0).getField(0);
+ assertThat(bytes).isNotNull();
+
+ RoaringBitmap restored = BitmapUtils.fromBytes(bytes);
+ assertThat(restored).isNotNull();
+ assertThat(restored.getLongCardinality()).isEqualTo(3L);
+ assertThat(restored.contains(1)).isTrue();
+ assertThat(restored.contains(2)).isTrue();
+ assertThat(restored.contains(3)).isTrue();
+ }
+
+ @Test
+ void testPlannerReturnsNullForEmptyInput() throws Exception {
+ TableEnvironment tEnv =
TableEnvironment.create(EnvironmentSettings.inBatchMode());
+ tEnv.createTemporarySystemFunction("test_rb_agg", new
TestRbAggFunction());
+
+ // Filter eliminates all rows; non-grouped aggregation over zero rows
must emit
+ // one row whose value is the result of getValue() on an empty
accumulator (null).
+ Table source = tEnv.fromValues(1, 2,
3).as("v").filter($("v").isGreater(100));
+ Table result = source.select(call("test_rb_agg", $("v")));
+
+ List<Row> rows =
CollectionUtil.iteratorToList(result.execute().collect());
+ assertThat(rows).hasSize(1);
+ assertThat(rows.get(0).getField(0)).isNull();
+ }
+}
diff --git
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/functions/bitmap/BitmapUtilsTest.java
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/functions/bitmap/BitmapUtilsTest.java
new file mode 100644
index 000000000..fa0a42ab6
--- /dev/null
+++
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/functions/bitmap/BitmapUtilsTest.java
@@ -0,0 +1,100 @@
+/*
+ * 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.junit.jupiter.api.Test;
+import org.roaringbitmap.RoaringBitmap;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Unit tests for {@link BitmapUtils}. */
+class BitmapUtilsTest {
+
+ @Test
+ void testNullInputToBytes() throws IOException {
+ assertThat(BitmapUtils.toBytes(null)).isNull();
+ }
+
+ @Test
+ void testNullInputFromBytes() throws IOException {
+ assertThat(BitmapUtils.fromBytes(null)).isNull();
+ }
+
+ @Test
+ void testEmptyBitmapRoundTrip() throws IOException {
+ RoaringBitmap bitmap = new RoaringBitmap();
+ byte[] bytes = BitmapUtils.toBytes(bitmap);
+ assertThat(bytes).isNotNull();
+ RoaringBitmap result = BitmapUtils.fromBytes(bytes);
+ assertThat(result).isNotNull();
+ assertThat(result.isEmpty()).isTrue();
+ }
+
+ @Test
+ void testKnownValuesRoundTrip() throws IOException {
+ RoaringBitmap bitmap = new RoaringBitmap();
+ bitmap.add(1);
+ bitmap.add(100);
+ bitmap.add(1000);
+ bitmap.add(Integer.MAX_VALUE);
+
+ byte[] bytes = BitmapUtils.toBytes(bitmap);
+ assertThat(bytes).isNotNull();
+
+ RoaringBitmap result = BitmapUtils.fromBytes(bytes);
+ assertThat(result).isNotNull();
+ assertThat(result.getLongCardinality()).isEqualTo(4L);
+ assertThat(result.contains(1)).isTrue();
+ assertThat(result.contains(100)).isTrue();
+ assertThat(result.contains(1000)).isTrue();
+ assertThat(result.contains(Integer.MAX_VALUE)).isTrue();
+ assertThat(result.contains(2)).isFalse();
+ }
+
+ @Test
+ void testLargeCardinality() throws IOException {
+ RoaringBitmap bitmap = new RoaringBitmap();
+ for (int i = 0; i < 100_000; i++) {
+ bitmap.add(i);
+ }
+ byte[] bytes = BitmapUtils.toBytes(bitmap);
+ RoaringBitmap result = BitmapUtils.fromBytes(bytes);
+ assertThat(result.getLongCardinality()).isEqualTo(100_000L);
+ }
+
+ @Test
+ void testProducesSameBytesAsServerSerializationRecipe() throws IOException
{
+ // Use a bitmap that benefits from runOptimize so the test would catch
+ // removal of runOptimize from one side but not the other.
+ RoaringBitmap original = new RoaringBitmap();
+ original.add(0L, 50_000L);
+
+ byte[] clientBytes = BitmapUtils.toBytes(original.clone());
+
+ RoaringBitmap forServer = original.clone();
+ forServer.runOptimize();
+ ByteBuffer buffer =
ByteBuffer.allocate(forServer.serializedSizeInBytes());
+ forServer.serialize(buffer);
+ byte[] serverRecipeBytes = buffer.array();
+
+ assertThat(clientBytes).isEqualTo(serverRecipeBytes);
+ }
+}
diff --git
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/functions/bitmap/RoaringBitmapSerializerTest.java
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/functions/bitmap/RoaringBitmapSerializerTest.java
new file mode 100644
index 000000000..75e0309aa
--- /dev/null
+++
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/functions/bitmap/RoaringBitmapSerializerTest.java
@@ -0,0 +1,281 @@
+/*
+ * 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.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputDeserializer;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.junit.jupiter.api.Test;
+import org.roaringbitmap.RoaringBitmap;
+
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Unit tests for {@link RoaringBitmapSerializer} and {@link
RoaringBitmapTypeInfo}. */
+class RoaringBitmapSerializerTest {
+
+ private final RoaringBitmapSerializer serializer =
RoaringBitmapSerializer.INSTANCE;
+
+ /** Minimal concrete implementation used only for testing
AbstractRbAggFunction. */
+ private static final class TestRbAggFunction extends AbstractRbAggFunction
{
+
+ public void accumulate(RoaringBitmap acc, Integer value) {
+ if (value != null) {
+ acc.add(value);
+ }
+ }
+ }
+
+ private final TestRbAggFunction aggFunction = new TestRbAggFunction();
+
+ @Test
+ void testCreateInstance() {
+ RoaringBitmap instance = serializer.createInstance();
+ assertThat(instance).isNotNull();
+ assertThat(instance.isEmpty()).isTrue();
+ }
+
+ @Test
+ void testIsNotImmutable() {
+ assertThat(serializer.isImmutableType()).isFalse();
+ }
+
+ @Test
+ void testGetLengthIsMinusOne() {
+ assertThat(serializer.getLength()).isEqualTo(-1);
+ }
+
+ @Test
+ void testSerializeDeserializeRoundTrip() throws Exception {
+ RoaringBitmap original = new RoaringBitmap();
+ original.add(1);
+ original.add(100);
+ original.add(100_000);
+
+ DataOutputSerializer out = new DataOutputSerializer(256);
+ serializer.serialize(original, out);
+
+ DataInputDeserializer in = new
DataInputDeserializer(out.getCopyOfBuffer());
+ RoaringBitmap restored = serializer.deserialize(in);
+
+ assertThat(restored).isEqualTo(original);
+ assertThat(restored.getLongCardinality()).isEqualTo(3L);
+ }
+
+ @Test
+ void testDeserializeWithReuse() throws Exception {
+ RoaringBitmap original = RoaringBitmap.bitmapOf(42, 99, 1000);
+
+ DataOutputSerializer out = new DataOutputSerializer(256);
+ serializer.serialize(original, out);
+
+ DataInputDeserializer in = new
DataInputDeserializer(out.getCopyOfBuffer());
+ RoaringBitmap reuse = new RoaringBitmap();
+ RoaringBitmap restored = serializer.deserialize(reuse, in);
+
+ assertThat(restored).isEqualTo(original);
+ }
+
+ @Test
+ void testCopy() {
+ RoaringBitmap original = RoaringBitmap.bitmapOf(1, 2, 3);
+ RoaringBitmap copy = serializer.copy(original);
+
+ assertThat(copy).isEqualTo(original);
+ copy.add(999);
+ assertThat(original.contains(999)).isFalse();
+ }
+
+ @Test
+ void testCopyWithReuse() {
+ RoaringBitmap original = RoaringBitmap.bitmapOf(10, 20, 30);
+ RoaringBitmap reuse = new RoaringBitmap();
+ RoaringBitmap copy = serializer.copy(original, reuse);
+
+ assertThat(copy).isEqualTo(original);
+ }
+
+ @Test
+ void testSerializeWithRunOptimizableBitmap() throws Exception {
+ RoaringBitmap original = new RoaringBitmap();
+ original.add(0L, 50_000L);
+
+ DataOutputSerializer out = new DataOutputSerializer(64 * 1024);
+ serializer.serialize(original, out);
+
+ DataInputDeserializer in = new
DataInputDeserializer(out.getCopyOfBuffer());
+ RoaringBitmap restored = serializer.deserialize(in);
+
+ assertThat(restored).isEqualTo(original);
+ assertThat(restored.getLongCardinality()).isEqualTo(50_000L);
+ // The deserializer must have consumed exactly the bytes we wrote.
+ assertThat(in.available()).isZero();
+ }
+
+ @Test
+ void testCopyStreamWithRunOptimizableBitmap() throws Exception {
+ RoaringBitmap original = new RoaringBitmap();
+ original.add(0L, 50_000L);
+
+ DataOutputSerializer out = new DataOutputSerializer(64 * 1024);
+ serializer.serialize(original, out);
+
+ DataInputDeserializer in = new
DataInputDeserializer(out.getCopyOfBuffer());
+ DataOutputSerializer copied = new DataOutputSerializer(64 * 1024);
+ serializer.copy(in, copied);
+
+ RoaringBitmap restored =
+ serializer.deserialize(new
DataInputDeserializer(copied.getCopyOfBuffer()));
+ assertThat(restored).isEqualTo(original);
+ }
+
+ @Test
+ void testEmptyBitmapRoundTrip() throws Exception {
+ RoaringBitmap empty = new RoaringBitmap();
+
+ DataOutputSerializer out = new DataOutputSerializer(64);
+ serializer.serialize(empty, out);
+
+ DataInputDeserializer in = new
DataInputDeserializer(out.getCopyOfBuffer());
+ RoaringBitmap restored = serializer.deserialize(in);
+
+ assertThat(restored.isEmpty()).isTrue();
+ }
+
+ @Test
+ void testSnapshotConfiguration() {
+ assertThat(serializer.snapshotConfiguration()).isNotNull();
+ }
+
+ @Test
+ void testTypeInfoGetTypeClass() {
+
assertThat(RoaringBitmapTypeInfo.INSTANCE.getTypeClass()).isEqualTo(RoaringBitmap.class);
+ }
+
+ @Test
+ void testTypeInfoCreateSerializer() {
+ TypeSerializer<RoaringBitmap> s =
+ RoaringBitmapTypeInfo.INSTANCE.createSerializer(new
ExecutionConfig());
+ assertThat(s).isInstanceOf(RoaringBitmapSerializer.class);
+ }
+
+ @Test
+ void testTypeInfoEquality() {
+
assertThat(RoaringBitmapTypeInfo.INSTANCE.equals(RoaringBitmapTypeInfo.INSTANCE)).isTrue();
+ assertThat(RoaringBitmapTypeInfo.INSTANCE.equals("other")).isFalse();
+ }
+
+ @Test
+ void testTypeInfoIsNotKeyType() {
+ assertThat(RoaringBitmapTypeInfo.INSTANCE.isKeyType()).isFalse();
+ }
+
+ @Test
+ void testTypeInfoArity() {
+ assertThat(RoaringBitmapTypeInfo.INSTANCE.getArity()).isEqualTo(1);
+
assertThat(RoaringBitmapTypeInfo.INSTANCE.getTotalFields()).isEqualTo(1);
+ }
+
+ @Test
+ void testTypeInfoIsNotBasicType() {
+ assertThat(RoaringBitmapTypeInfo.INSTANCE.isBasicType()).isFalse();
+ }
+
+ @Test
+ void testTypeInfoIsNotTupleType() {
+ assertThat(RoaringBitmapTypeInfo.INSTANCE.isTupleType()).isFalse();
+ }
+
+ @Test
+ void testTypeInfoToString() {
+
assertThat(RoaringBitmapTypeInfo.INSTANCE.toString()).isEqualTo("RoaringBitmapTypeInfo");
+ }
+
+ @Test
+ void testTypeInfoHashCode() {
+ assertThat(RoaringBitmapTypeInfo.INSTANCE.hashCode()).isNotZero();
+ }
+
+ @Test
+ void testTypeInfoCanEqual() {
+
assertThat(RoaringBitmapTypeInfo.INSTANCE.canEqual(RoaringBitmapTypeInfo.INSTANCE))
+ .isTrue();
+ assertThat(RoaringBitmapTypeInfo.INSTANCE.canEqual("other")).isFalse();
+ }
+
+ @Test
+ void testAggCreateAccumulator() {
+ RoaringBitmap acc = aggFunction.createAccumulator();
+ assertThat(acc).isNotNull();
+ assertThat(acc.isEmpty()).isTrue();
+ }
+
+ @Test
+ void testAggGetValue() throws Exception {
+ RoaringBitmap acc = aggFunction.createAccumulator();
+ aggFunction.accumulate(acc, 1);
+ aggFunction.accumulate(acc, 2);
+
+ byte[] result = aggFunction.getValue(acc);
+
+ assertThat(result).isNotNull();
+ RoaringBitmap restored = BitmapUtils.fromBytes(result);
+ assertThat(restored).isNotNull();
+ assertThat(restored.getLongCardinality()).isEqualTo(2L);
+ assertThat(restored.contains(1)).isTrue();
+ assertThat(restored.contains(2)).isTrue();
+ }
+
+ @Test
+ void testAggGetValueNullOnEmpty() {
+ RoaringBitmap acc = aggFunction.createAccumulator();
+ assertThat(aggFunction.getValue(acc)).isNull();
+ }
+
+ @Test
+ void testAggMerge() {
+ RoaringBitmap acc1 = aggFunction.createAccumulator();
+ aggFunction.accumulate(acc1, 1);
+ aggFunction.accumulate(acc1, 2);
+
+ RoaringBitmap acc2 = aggFunction.createAccumulator();
+ aggFunction.accumulate(acc2, 3);
+
+ aggFunction.merge(acc1, Collections.singletonList(acc2));
+ assertThat(acc1.getLongCardinality()).isEqualTo(3L);
+ assertThat(acc1.contains(3)).isTrue();
+ }
+
+ @Test
+ void testAggResetAccumulator() {
+ RoaringBitmap acc = aggFunction.createAccumulator();
+ acc.add(1);
+ acc.add(2);
+
+ aggFunction.resetAccumulator(acc);
+
+ assertThat(acc.isEmpty()).isTrue();
+ }
+
+ @Test
+ void testAggGetAccumulatorType() {
+
assertThat(aggFunction.getAccumulatorType()).isInstanceOf(RoaringBitmapTypeInfo.class);
+ }
+}
