diff --git a/fluss-common/src/main/java/org/apache/fluss/record/DefaultKvRecordBatch.java b/fluss-common/src/main/java/org/apache/fluss/record/DefaultKvRecordBatch.java index 1fe83650d19..852c7ebbf25 100644 --- a/fluss-common/src/main/java/org/apache/fluss/record/DefaultKvRecordBatch.java +++ b/fluss-common/src/main/java/org/apache/fluss/record/DefaultKvRecordBatch.java @@ -23,6 +23,7 @@ import org.apache.fluss.memory.MemorySegment; import org.apache.fluss.record.bytesview.BytesView; import org.apache.fluss.utils.CloseableIterator; +import org.apache.fluss.utils.MurmurHashUtils; import org.apache.fluss.utils.crc.Crc32C; import java.nio.ByteBuffer; @@ -219,6 +220,11 @@ public boolean equals(Object o) { && segment.equalTo(that.segment, position, that.position, sizeInBytes); } + @Override + public int hashCode() { + return MurmurHashUtils.hashBytes(segment, position, sizeInBytes()); + } + abstract class KvRecordIterator implements CloseableIterator { private final int numRecords; private int readRecords = 0; diff --git a/fluss-common/src/main/java/org/apache/fluss/record/DefaultValueRecordBatch.java b/fluss-common/src/main/java/org/apache/fluss/record/DefaultValueRecordBatch.java index 3bb7539f615..55877215481 100644 --- a/fluss-common/src/main/java/org/apache/fluss/record/DefaultValueRecordBatch.java +++ b/fluss-common/src/main/java/org/apache/fluss/record/DefaultValueRecordBatch.java @@ -21,6 +21,7 @@ import org.apache.fluss.memory.MemorySegmentOutputView; import org.apache.fluss.row.BinaryRow; import org.apache.fluss.utils.CloseableIterator; +import org.apache.fluss.utils.MurmurHashUtils; import java.io.IOException; import java.nio.ByteBuffer; @@ -132,6 +133,11 @@ public boolean equals(Object o) { && segment.equalTo(that.segment, position, that.position, sizeInBytes); } + @Override + public int hashCode() { + return MurmurHashUtils.hashBytes(segment, position, sizeInBytes()); + } + // ------------------------------------------------------------------------------------------ abstract class ValueRecordIterator implements CloseableIterator { diff --git a/fluss-common/src/test/java/org/apache/fluss/record/DefaultKvRecordBatchTest.java b/fluss-common/src/test/java/org/apache/fluss/record/DefaultKvRecordBatchTest.java index 6c8b483a0bf..cbdceced5f0 100644 --- a/fluss-common/src/test/java/org/apache/fluss/record/DefaultKvRecordBatchTest.java +++ b/fluss-common/src/test/java/org/apache/fluss/record/DefaultKvRecordBatchTest.java @@ -27,7 +27,9 @@ import java.util.ArrayList; import java.util.List; +import static org.apache.fluss.record.TestData.DATA1_ROW_TYPE; import static org.apache.fluss.record.TestData.DATA1_SCHEMA; +import static org.apache.fluss.testutils.DataTestUtils.compactedRow; import static org.assertj.core.api.Assertions.assertThat; /** Test for {@link DefaultKvRecordBatch}. */ @@ -76,4 +78,39 @@ KvFormat.COMPACTED, new TestingSchemaGetter(1, DATA1_SCHEMA)))) { builder.close(); } + + @Test + void testEqualsAndHashCode() throws Exception { + DefaultKvRecordBatch batch1 = buildBatch(); + DefaultKvRecordBatch batch2 = buildBatch(); + + assertThat(batch1).isEqualTo(batch2); + assertThat(batch1).hasSameHashCodeAs(batch2); + } + + @Test + void testHashCodeDiffersForDifferentContents() throws Exception { + DefaultKvRecordBatch batch1 = buildBatch(); + DefaultKvRecordBatch batch2 = buildBatch(new byte[] {(byte) 9, (byte) 9}); + + assertThat(batch1).isNotEqualTo(batch2); + assertThat(batch1.hashCode()).isNotEqualTo(batch2.hashCode()); + } + + private DefaultKvRecordBatch buildBatch() throws Exception { + return buildBatch(new byte[] {(byte) 1, (byte) 1}); + } + + private DefaultKvRecordBatch buildBatch(byte[] key) throws Exception { + KvRecordBatchBuilder builder = + KvRecordBatchBuilder.builder( + schemaId, + Integer.MAX_VALUE, + new UnmanagedPagedOutputView(100), + KvFormat.COMPACTED); + builder.append(key, compactedRow(DATA1_ROW_TYPE, new Object[] {1, "a1"})); + DefaultKvRecordBatch batch = DefaultKvRecordBatch.pointToBytesView(builder.build()); + builder.close(); + return batch; + } } diff --git a/fluss-common/src/test/java/org/apache/fluss/record/DefaultValueRecordBatchTest.java b/fluss-common/src/test/java/org/apache/fluss/record/DefaultValueRecordBatchTest.java new file mode 100644 index 00000000000..5558ec7c04a --- /dev/null +++ b/fluss-common/src/test/java/org/apache/fluss/record/DefaultValueRecordBatchTest.java @@ -0,0 +1,69 @@ +/* + * 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.record; + +import org.junit.jupiter.api.Test; + +import static org.apache.fluss.record.TestData.DATA1_ROW_TYPE; +import static org.apache.fluss.record.TestData.DEFAULT_SCHEMA_ID; +import static org.apache.fluss.testutils.DataTestUtils.compactedRow; +import static org.assertj.core.api.Assertions.assertThat; + +/** Test for {@link DefaultValueRecordBatch}. */ +class DefaultValueRecordBatchTest { + + @Test + void testEqualsAndHashCode() throws Exception { + DefaultValueRecordBatch batch1 = buildBatch(); + DefaultValueRecordBatch batch2 = buildBatch(); + + assertThat(batch1).isEqualTo(batch2); + assertThat(batch1).hasSameHashCodeAs(batch2); + } + + @Test + void testHashCodeDiffersForDifferentContents() throws Exception { + DefaultValueRecordBatch.Builder builder = DefaultValueRecordBatch.builder(); + builder.append(DEFAULT_SCHEMA_ID, compactedRow(DATA1_ROW_TYPE, new Object[] {1, "a1"})); + DefaultValueRecordBatch batch1 = builder.build(); + + DefaultValueRecordBatch.Builder otherBuilder = DefaultValueRecordBatch.builder(); + otherBuilder.append( + DEFAULT_SCHEMA_ID, compactedRow(DATA1_ROW_TYPE, new Object[] {2, "a2"})); + DefaultValueRecordBatch batch2 = otherBuilder.build(); + + assertThat(batch1).isNotEqualTo(batch2); + assertThat(batch1.hashCode()).isNotEqualTo(batch2.hashCode()); + } + + @Test + void testEmptyBatchesAreEqualAndShareHashCode() throws Exception { + DefaultValueRecordBatch batch1 = DefaultValueRecordBatch.builder().build(); + DefaultValueRecordBatch batch2 = DefaultValueRecordBatch.builder().build(); + + assertThat(batch1).isEqualTo(batch2); + assertThat(batch1).hasSameHashCodeAs(batch2); + } + + private static DefaultValueRecordBatch buildBatch() throws Exception { + DefaultValueRecordBatch.Builder builder = DefaultValueRecordBatch.builder(); + builder.append(DEFAULT_SCHEMA_ID, compactedRow(DATA1_ROW_TYPE, new Object[] {1, "a1"})); + builder.append(DEFAULT_SCHEMA_ID, compactedRow(DATA1_ROW_TYPE, new Object[] {2, "a2"})); + return builder.build(); + } +}