Prajwal-banakar commented on code in PR #3319: URL: https://github.com/apache/fluss/pull/3319#discussion_r3260497838
########## fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/RoaringBitmapSerializer.java: ########## @@ -0,0 +1,113 @@ +/* + * 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 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. + */ +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 { Review Comment: Fixed, removed the redundant runOptimize() call from serialize(); BitmapUtils.toBytes() handles it. ########## fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/functions/bitmap/RoaringBitmapTypeInfo.java: ########## @@ -0,0 +1,95 @@ +/* + * 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 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. + */ +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) { Review Comment: Fixed equals to delegate through canEqual. Added @ThreadSafe to both classes. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
