You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "wuchong (via GitHub)" <gi...@apache.org> on 2025/09/19 07:20:10 UTC
Re: [PR] Introduce BinaryRowData for row data encoding [fluss]
wuchong commented on code in PR #1620:
URL: https://github.com/apache/fluss/pull/1620#discussion_r2359863225
##########
fluss-common/src/test/java/org/apache/fluss/row/BinaryRowDataTest.java:
##########
@@ -0,0 +1,728 @@
+/*
+ * 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.row;
+
+import org.apache.fluss.memory.MemorySegment;
+import org.apache.fluss.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.charset.StandardCharsets;
+import java.time.LocalDateTime;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Random;
+import java.util.Set;
+import java.util.function.Consumer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link BinaryRowData}. */
+class BinaryRowDataTest {
+
+ @Test
+ public void testBasic() {
+ // consider header 1 byte.
+ assertThat(new BinaryRowData(0).getFixedLengthPartSize()).isEqualTo(8);
+ assertThat(new BinaryRowData(1).getFixedLengthPartSize()).isEqualTo(16);
+ assertThat(new BinaryRowData(65).getFixedLengthPartSize()).isEqualTo(536);
+ assertThat(new BinaryRowData(128).getFixedLengthPartSize()).isEqualTo(1048);
+
+ MemorySegment segment = MemorySegment.wrap(new byte[100]);
+ BinaryRowData row = new BinaryRowData(2);
+ row.pointTo(segment, 10, 48);
+ assertThat(segment).isSameAs(row.getSegments()[0]);
+ row.setInt(0, 5);
+ row.setDouble(1, 5.8D);
+ }
+
+ @Test
+ public void testSetAndGet() throws IOException, ClassNotFoundException {
Review Comment:
the exception signature can be removed
##########
fluss-common/src/test/java/org/apache/fluss/row/BinaryRowDataTest.java:
##########
@@ -0,0 +1,728 @@
+/*
+ * 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.row;
+
+import org.apache.fluss.memory.MemorySegment;
+import org.apache.fluss.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.charset.StandardCharsets;
+import java.time.LocalDateTime;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Random;
+import java.util.Set;
+import java.util.function.Consumer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link BinaryRowData}. */
+class BinaryRowDataTest {
+
+ @Test
+ public void testBasic() {
+ // consider header 1 byte.
+ assertThat(new BinaryRowData(0).getFixedLengthPartSize()).isEqualTo(8);
+ assertThat(new BinaryRowData(1).getFixedLengthPartSize()).isEqualTo(16);
+ assertThat(new BinaryRowData(65).getFixedLengthPartSize()).isEqualTo(536);
+ assertThat(new BinaryRowData(128).getFixedLengthPartSize()).isEqualTo(1048);
+
+ MemorySegment segment = MemorySegment.wrap(new byte[100]);
+ BinaryRowData row = new BinaryRowData(2);
+ row.pointTo(segment, 10, 48);
+ assertThat(segment).isSameAs(row.getSegments()[0]);
+ row.setInt(0, 5);
+ row.setDouble(1, 5.8D);
+ }
+
+ @Test
+ public void testSetAndGet() throws IOException, ClassNotFoundException {
+ MemorySegment segment = MemorySegment.wrap(new byte[100]);
+ BinaryRowData row = new BinaryRowData(9);
+ row.pointTo(segment, 20, 80);
+ row.setNullAt(0);
+ row.setInt(1, 11);
+ row.setLong(2, 22);
+ row.setDouble(3, 33);
+ row.setBoolean(4, true);
+ row.setShort(5, (short) 55);
+ row.setByte(6, (byte) 66);
+ row.setFloat(7, 77f);
+
+ Consumer<BinaryRow> assertConsumer =
Review Comment:
Why use Consumer instead of assert directly on the row?
##########
fluss-common/src/test/java/org/apache/fluss/row/BinaryRowDataTest.java:
##########
@@ -0,0 +1,728 @@
+/*
+ * 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.row;
+
+import org.apache.fluss.memory.MemorySegment;
+import org.apache.fluss.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.charset.StandardCharsets;
+import java.time.LocalDateTime;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Random;
+import java.util.Set;
+import java.util.function.Consumer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link BinaryRowData}. */
+class BinaryRowDataTest {
+
+ @Test
+ public void testBasic() {
+ // consider header 1 byte.
+ assertThat(new BinaryRowData(0).getFixedLengthPartSize()).isEqualTo(8);
+ assertThat(new BinaryRowData(1).getFixedLengthPartSize()).isEqualTo(16);
+ assertThat(new BinaryRowData(65).getFixedLengthPartSize()).isEqualTo(536);
+ assertThat(new BinaryRowData(128).getFixedLengthPartSize()).isEqualTo(1048);
+
+ MemorySegment segment = MemorySegment.wrap(new byte[100]);
+ BinaryRowData row = new BinaryRowData(2);
+ row.pointTo(segment, 10, 48);
+ assertThat(segment).isSameAs(row.getSegments()[0]);
+ row.setInt(0, 5);
+ row.setDouble(1, 5.8D);
+ }
+
+ @Test
+ public void testSetAndGet() throws IOException, ClassNotFoundException {
+ MemorySegment segment = MemorySegment.wrap(new byte[100]);
+ BinaryRowData row = new BinaryRowData(9);
+ row.pointTo(segment, 20, 80);
+ row.setNullAt(0);
+ row.setInt(1, 11);
+ row.setLong(2, 22);
+ row.setDouble(3, 33);
+ row.setBoolean(4, true);
+ row.setShort(5, (short) 55);
+ row.setByte(6, (byte) 66);
+ row.setFloat(7, 77f);
+
+ Consumer<BinaryRow> assertConsumer =
+ assertRow -> {
+ assertThat((long) assertRow.getDouble(3)).isEqualTo(33L);
+ assertThat(assertRow.getInt(1)).isEqualTo(11);
+ assertThat(assertRow.isNullAt(0)).isTrue();
+ assertThat(assertRow.getShort(5)).isEqualTo((short) 55);
+ assertThat(assertRow.getLong(2)).isEqualTo(22L);
+ assertThat(assertRow.getBoolean(4)).isTrue();
+ assertThat(assertRow.getByte(6)).isEqualTo((byte) 66);
+ assertThat(assertRow.getFloat(7)).isEqualTo(77f);
+ };
+
+ assertConsumer.accept(row);
+ }
+
+ @Test
+ public void testWriter() {
+
+ int arity = 13;
+ BinaryRowData row = new BinaryRowData(arity);
+ BinaryRowWriter writer = new BinaryRowWriter(row, 20);
+
+ writer.writeString(0, BinaryString.fromString("1"));
+ writer.writeString(3, BinaryString.fromString("1234567"));
+ writer.writeString(5, BinaryString.fromString("12345678"));
+ writer.writeString(
+ 9, BinaryString.fromString("God in his heaven, alls right with the world"));
+
+ writer.writeBoolean(1, true);
+ writer.writeByte(2, (byte) 99);
+ writer.writeDouble(6, 87.1d);
+ writer.writeFloat(7, 26.1f);
+ writer.writeInt(8, 88);
+ writer.writeLong(10, 284);
+ writer.writeShort(11, (short) 292);
+ writer.setNullAt(12);
+
+ writer.complete();
+
+ assertTestWriterRow(row);
+ assertTestWriterRow(row.copy());
+
+ // test copy from var segments.
+ int subSize = row.getFixedLengthPartSize() + 10;
+ MemorySegment subMs1 = MemorySegment.wrap(new byte[subSize]);
+ MemorySegment subMs2 = MemorySegment.wrap(new byte[subSize]);
+ row.getSegments()[0].copyTo(0, subMs1, 0, subSize);
+ row.getSegments()[0].copyTo(subSize, subMs2, 0, row.getSizeInBytes() - subSize);
+
+ BinaryRowData toCopy = new BinaryRowData(arity);
+ toCopy.pointTo(new MemorySegment[] {subMs1, subMs2}, 0, row.getSizeInBytes());
+ assertThat(toCopy).isEqualTo(row);
+ assertTestWriterRow(toCopy);
+ assertTestWriterRow(toCopy.copy(new BinaryRowData(arity)));
+ }
+
+ @Test
+ public void testWriteString() {
+ {
+ // litter byte[]
+ BinaryRowData row = new BinaryRowData(1);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ char[] chars = new char[2];
+ chars[0] = 0xFFFF;
+ chars[1] = 0;
+ writer.writeString(0, BinaryString.fromString(new String(chars)));
+ writer.complete();
+
+ String str = row.getString(0).toString();
+ assertThat(str.charAt(0)).isEqualTo(chars[0]);
+ assertThat(str.charAt(1)).isEqualTo(chars[1]);
+ }
+
+ {
+ // big byte[]
+ String str = "God in his heaven, alls right with the world";
+ BinaryRowData row = new BinaryRowData(2);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ writer.writeString(0, BinaryString.fromString(str));
+ writer.writeString(1, BinaryString.fromBytes(str.getBytes(StandardCharsets.UTF_8)));
+ writer.complete();
+
+ assertThat(row.getString(0).toString()).isEqualTo(str);
+ assertThat(row.getString(1).toString()).isEqualTo(str);
+ }
+ }
+
+ private void assertTestWriterRow(BinaryRowData row) {
+ assertThat(row.getString(0).toString()).isEqualTo("1");
+ assertThat(row.getInt(8)).isEqualTo(88);
+ assertThat(row.getShort(11)).isEqualTo((short) 292);
+ assertThat(row.getLong(10)).isEqualTo(284);
+ assertThat(row.getByte(2)).isEqualTo((byte) 99);
+ assertThat(row.getDouble(6)).isEqualTo(87.1d);
+ assertThat(row.getFloat(7)).isEqualTo(26.1f);
+ assertThat(row.getBoolean(1)).isTrue();
+ assertThat(row.getString(3).toString()).isEqualTo("1234567");
+ assertThat(row.getString(5).toString()).isEqualTo("12345678");
+ assertThat(row.getString(9).toString())
+ .isEqualTo("God in his heaven, alls right with the world");
+ assertThat(row.getString(9).hashCode())
+ .isEqualTo(
+ BinaryString.fromString("God in his heaven, alls right with the world")
+ .hashCode());
+ assertThat(row.isNullAt(12)).isTrue();
+ }
+
+ @Test
+ public void testReuseWriter() {
+ BinaryRowData row = new BinaryRowData(2);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ writer.writeString(0, BinaryString.fromString("01234567"));
+ writer.writeString(1, BinaryString.fromString("012345678"));
+ writer.complete();
+ assertThat(row.getString(0).toString()).isEqualTo("01234567");
+ assertThat(row.getString(1).toString()).isEqualTo("012345678");
+
+ writer.reset();
+ writer.writeString(0, BinaryString.fromString("1"));
+ writer.writeString(1, BinaryString.fromString("0123456789"));
+ writer.complete();
+ assertThat(row.getString(0).toString()).isEqualTo("1");
+ assertThat(row.getString(1).toString()).isEqualTo("0123456789");
+ }
+
+ @Test
+ public void anyNullTest() {
+ {
+ BinaryRowData row = new BinaryRowData(3);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ assertThat(row.anyNull()).isFalse();
+
+ // test header should not compute by anyNull
+ assertThat(row.anyNull()).isFalse();
+
+ writer.setNullAt(2);
+ assertThat(row.anyNull()).isTrue();
+
+ writer.setNullAt(0);
+ assertThat(row.anyNull(new int[] {0, 1, 2})).isTrue();
+ assertThat(row.anyNull(new int[] {1})).isFalse();
+
+ writer.setNullAt(1);
+ assertThat(row.anyNull()).isTrue();
+ }
+
+ int numFields = 80;
+ for (int i = 0; i < numFields; i++) {
+ BinaryRowData row = new BinaryRowData(numFields);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ assertThat(row.anyNull()).isFalse();
+ writer.setNullAt(i);
+ assertThat(row.anyNull()).isTrue();
+ }
+ }
+
+ @Test
+ public void testSingleSegmentBinaryRowHashCode() {
+ final Random rnd = new Random(System.currentTimeMillis());
+ // test hash stabilization
+ BinaryRowData row = new BinaryRowData(13);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ for (int i = 0; i < 99; i++) {
+ writer.reset();
+ writer.writeString(0, BinaryString.fromString("" + rnd.nextInt()));
+ writer.writeString(3, BinaryString.fromString("01234567"));
+ writer.writeString(5, BinaryString.fromString("012345678"));
+ writer.writeString(
+ 9, BinaryString.fromString("God in his heaven, alls right with the world"));
+ writer.writeBoolean(1, true);
+ writer.writeByte(2, (byte) 99);
+ writer.writeDouble(6, 87.1d);
+ writer.writeFloat(7, 26.1f);
+ writer.writeInt(8, 88);
+ writer.writeLong(10, 284);
+ writer.writeShort(11, (short) 292);
+ writer.setNullAt(12);
+ writer.complete();
+ BinaryRowData copy = row.copy();
+ assertThat(copy.hashCode()).isEqualTo(row.hashCode());
+ }
+
+ // test hash distribution
+ int count = 999999;
+ Set<Integer> hashCodes = new HashSet<>(count);
+ for (int i = 0; i < count; i++) {
+ row.setInt(8, i);
+ hashCodes.add(row.hashCode());
+ }
+ assertThat(hashCodes).hasSize(count);
+ hashCodes.clear();
+ row = new BinaryRowData(1);
+ writer = new BinaryRowWriter(row);
+ for (int i = 0; i < count; i++) {
+ writer.reset();
+ writer.writeString(
+ 0, BinaryString.fromString("God in his heaven, alls right with the world" + i));
+ writer.complete();
+ hashCodes.add(row.hashCode());
+ }
+ assertThat(hashCodes.size()).isGreaterThan((int) (count * 0.997));
+ }
+
+ @Test
+ public void testHeaderSize() {
+ assertThat(BinaryRowData.calculateBitSetWidthInBytes(56)).isEqualTo(8);
+ assertThat(BinaryRowData.calculateBitSetWidthInBytes(57)).isEqualTo(16);
+ assertThat(BinaryRowData.calculateBitSetWidthInBytes(120)).isEqualTo(16);
+ assertThat(BinaryRowData.calculateBitSetWidthInBytes(121)).isEqualTo(24);
+ }
+
+ @Test
+ public void testHeader() {
+ BinaryRowData row = new BinaryRowData(2);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+
+ writer.writeInt(0, 10);
+ writer.setNullAt(1);
+ writer.complete();
+
+ BinaryRowData newRow = row.copy();
+ assertThat(newRow).isEqualTo(row);
+ }
+
+ @Test
+ public void testDecimal() {
+ // 1.compact
+ {
+ int precision = 4;
+ int scale = 2;
+ BinaryRowData row = new BinaryRowData(2);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ writer.writeDecimal(0, Decimal.fromUnscaledLong(5, precision, scale), precision);
+ writer.setNullAt(1);
+ writer.complete();
+
+ assertThat(row.getDecimal(0, precision, scale).toString()).isEqualTo("0.05");
+ assertThat(row.isNullAt(1)).isTrue();
+ row.setDecimal(0, Decimal.fromUnscaledLong(6, precision, scale), precision);
+ assertThat(row.getDecimal(0, precision, scale).toString()).isEqualTo("0.06");
+ }
+
+ // 2.not compact
+ {
+ int precision = 25;
+ int scale = 5;
+ Decimal decimal1 = Decimal.fromBigDecimal(BigDecimal.valueOf(5.55), precision, scale);
+ Decimal decimal2 = Decimal.fromBigDecimal(BigDecimal.valueOf(6.55), precision, scale);
+
+ BinaryRowData row = new BinaryRowData(2);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ writer.writeDecimal(0, decimal1, precision);
+ writer.writeDecimal(1, null, precision);
+ writer.complete();
+
+ assertThat(row.getDecimal(0, precision, scale).toString()).isEqualTo("5.55000");
+ assertThat(row.isNullAt(1)).isTrue();
+ row.setDecimal(0, decimal2, precision);
+ assertThat(row.getDecimal(0, precision, scale).toString()).isEqualTo("6.55000");
+ }
+ }
+
+ @Test
+ public void testBinary() {
+ BinaryRowData row = new BinaryRowData(2);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ byte[] bytes1 = new byte[] {1, -1, 5};
+ byte[] bytes2 = new byte[] {1, -1, 5, 5, 1, 5, 1, 5};
+ writer.writeBinary(0, bytes1);
+ writer.writeBinary(1, bytes2);
+ writer.complete();
+
+ assertThat(row.getBinary(0, bytes1.length)).isEqualTo(bytes1);
+ assertThat(row.getBinary(1, bytes2.length)).isEqualTo(bytes2);
+ }
+
+ @Test
+ public void testZeroOutPaddingString() {
+
+ Random random = new Random();
+ byte[] bytes = new byte[1024];
+
+ BinaryRowData row = new BinaryRowData(1);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+
+ writer.reset();
+ random.nextBytes(bytes);
+ writer.writeBinary(0, bytes);
+ writer.reset();
+ writer.writeString(0, BinaryString.fromString("wahahah"));
+ writer.complete();
+ int hash1 = row.hashCode();
+
+ writer.reset();
+ random.nextBytes(bytes);
+ writer.writeBinary(0, bytes);
+ writer.reset();
+ writer.writeString(0, BinaryString.fromString("wahahah"));
+ writer.complete();
+ int hash2 = row.hashCode();
+
+ assertThat(hash2).isEqualTo(hash1);
+ }
+
+ @Test
+ public void testTimestampData() {
Review Comment:
`testTimestamp`
##########
fluss-common/src/test/java/org/apache/fluss/row/BinaryRowDataTest.java:
##########
@@ -0,0 +1,728 @@
+/*
+ * 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.row;
+
+import org.apache.fluss.memory.MemorySegment;
+import org.apache.fluss.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.charset.StandardCharsets;
+import java.time.LocalDateTime;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Random;
+import java.util.Set;
+import java.util.function.Consumer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link BinaryRowData}. */
+class BinaryRowDataTest {
+
+ @Test
+ public void testBasic() {
+ // consider header 1 byte.
+ assertThat(new BinaryRowData(0).getFixedLengthPartSize()).isEqualTo(8);
+ assertThat(new BinaryRowData(1).getFixedLengthPartSize()).isEqualTo(16);
+ assertThat(new BinaryRowData(65).getFixedLengthPartSize()).isEqualTo(536);
+ assertThat(new BinaryRowData(128).getFixedLengthPartSize()).isEqualTo(1048);
+
+ MemorySegment segment = MemorySegment.wrap(new byte[100]);
+ BinaryRowData row = new BinaryRowData(2);
+ row.pointTo(segment, 10, 48);
+ assertThat(segment).isSameAs(row.getSegments()[0]);
+ row.setInt(0, 5);
+ row.setDouble(1, 5.8D);
+ }
+
+ @Test
+ public void testSetAndGet() throws IOException, ClassNotFoundException {
+ MemorySegment segment = MemorySegment.wrap(new byte[100]);
+ BinaryRowData row = new BinaryRowData(9);
+ row.pointTo(segment, 20, 80);
+ row.setNullAt(0);
+ row.setInt(1, 11);
+ row.setLong(2, 22);
+ row.setDouble(3, 33);
+ row.setBoolean(4, true);
+ row.setShort(5, (short) 55);
+ row.setByte(6, (byte) 66);
+ row.setFloat(7, 77f);
+
+ Consumer<BinaryRow> assertConsumer =
+ assertRow -> {
+ assertThat((long) assertRow.getDouble(3)).isEqualTo(33L);
+ assertThat(assertRow.getInt(1)).isEqualTo(11);
+ assertThat(assertRow.isNullAt(0)).isTrue();
+ assertThat(assertRow.getShort(5)).isEqualTo((short) 55);
+ assertThat(assertRow.getLong(2)).isEqualTo(22L);
+ assertThat(assertRow.getBoolean(4)).isTrue();
+ assertThat(assertRow.getByte(6)).isEqualTo((byte) 66);
+ assertThat(assertRow.getFloat(7)).isEqualTo(77f);
+ };
+
+ assertConsumer.accept(row);
+ }
+
+ @Test
+ public void testWriter() {
+
+ int arity = 13;
+ BinaryRowData row = new BinaryRowData(arity);
+ BinaryRowWriter writer = new BinaryRowWriter(row, 20);
+
+ writer.writeString(0, BinaryString.fromString("1"));
+ writer.writeString(3, BinaryString.fromString("1234567"));
+ writer.writeString(5, BinaryString.fromString("12345678"));
+ writer.writeString(
+ 9, BinaryString.fromString("God in his heaven, alls right with the world"));
+
+ writer.writeBoolean(1, true);
+ writer.writeByte(2, (byte) 99);
+ writer.writeDouble(6, 87.1d);
+ writer.writeFloat(7, 26.1f);
+ writer.writeInt(8, 88);
+ writer.writeLong(10, 284);
+ writer.writeShort(11, (short) 292);
+ writer.setNullAt(12);
+
+ writer.complete();
+
+ assertTestWriterRow(row);
+ assertTestWriterRow(row.copy());
+
+ // test copy from var segments.
+ int subSize = row.getFixedLengthPartSize() + 10;
+ MemorySegment subMs1 = MemorySegment.wrap(new byte[subSize]);
+ MemorySegment subMs2 = MemorySegment.wrap(new byte[subSize]);
+ row.getSegments()[0].copyTo(0, subMs1, 0, subSize);
+ row.getSegments()[0].copyTo(subSize, subMs2, 0, row.getSizeInBytes() - subSize);
+
+ BinaryRowData toCopy = new BinaryRowData(arity);
+ toCopy.pointTo(new MemorySegment[] {subMs1, subMs2}, 0, row.getSizeInBytes());
+ assertThat(toCopy).isEqualTo(row);
+ assertTestWriterRow(toCopy);
+ assertTestWriterRow(toCopy.copy(new BinaryRowData(arity)));
+ }
+
+ @Test
+ public void testWriteString() {
+ {
+ // litter byte[]
+ BinaryRowData row = new BinaryRowData(1);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ char[] chars = new char[2];
+ chars[0] = 0xFFFF;
+ chars[1] = 0;
+ writer.writeString(0, BinaryString.fromString(new String(chars)));
+ writer.complete();
+
+ String str = row.getString(0).toString();
+ assertThat(str.charAt(0)).isEqualTo(chars[0]);
+ assertThat(str.charAt(1)).isEqualTo(chars[1]);
+ }
+
+ {
+ // big byte[]
+ String str = "God in his heaven, alls right with the world";
+ BinaryRowData row = new BinaryRowData(2);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ writer.writeString(0, BinaryString.fromString(str));
+ writer.writeString(1, BinaryString.fromBytes(str.getBytes(StandardCharsets.UTF_8)));
+ writer.complete();
+
+ assertThat(row.getString(0).toString()).isEqualTo(str);
+ assertThat(row.getString(1).toString()).isEqualTo(str);
+ }
+ }
+
+ private void assertTestWriterRow(BinaryRowData row) {
+ assertThat(row.getString(0).toString()).isEqualTo("1");
+ assertThat(row.getInt(8)).isEqualTo(88);
+ assertThat(row.getShort(11)).isEqualTo((short) 292);
+ assertThat(row.getLong(10)).isEqualTo(284);
+ assertThat(row.getByte(2)).isEqualTo((byte) 99);
+ assertThat(row.getDouble(6)).isEqualTo(87.1d);
+ assertThat(row.getFloat(7)).isEqualTo(26.1f);
+ assertThat(row.getBoolean(1)).isTrue();
+ assertThat(row.getString(3).toString()).isEqualTo("1234567");
+ assertThat(row.getString(5).toString()).isEqualTo("12345678");
+ assertThat(row.getString(9).toString())
+ .isEqualTo("God in his heaven, alls right with the world");
+ assertThat(row.getString(9).hashCode())
+ .isEqualTo(
+ BinaryString.fromString("God in his heaven, alls right with the world")
+ .hashCode());
+ assertThat(row.isNullAt(12)).isTrue();
+ }
+
+ @Test
+ public void testReuseWriter() {
+ BinaryRowData row = new BinaryRowData(2);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ writer.writeString(0, BinaryString.fromString("01234567"));
+ writer.writeString(1, BinaryString.fromString("012345678"));
+ writer.complete();
+ assertThat(row.getString(0).toString()).isEqualTo("01234567");
+ assertThat(row.getString(1).toString()).isEqualTo("012345678");
+
+ writer.reset();
+ writer.writeString(0, BinaryString.fromString("1"));
+ writer.writeString(1, BinaryString.fromString("0123456789"));
+ writer.complete();
+ assertThat(row.getString(0).toString()).isEqualTo("1");
+ assertThat(row.getString(1).toString()).isEqualTo("0123456789");
+ }
+
+ @Test
+ public void anyNullTest() {
+ {
+ BinaryRowData row = new BinaryRowData(3);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ assertThat(row.anyNull()).isFalse();
+
+ // test header should not compute by anyNull
+ assertThat(row.anyNull()).isFalse();
+
+ writer.setNullAt(2);
+ assertThat(row.anyNull()).isTrue();
+
+ writer.setNullAt(0);
+ assertThat(row.anyNull(new int[] {0, 1, 2})).isTrue();
+ assertThat(row.anyNull(new int[] {1})).isFalse();
+
+ writer.setNullAt(1);
+ assertThat(row.anyNull()).isTrue();
+ }
+
+ int numFields = 80;
+ for (int i = 0; i < numFields; i++) {
+ BinaryRowData row = new BinaryRowData(numFields);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ assertThat(row.anyNull()).isFalse();
+ writer.setNullAt(i);
+ assertThat(row.anyNull()).isTrue();
+ }
+ }
+
+ @Test
+ public void testSingleSegmentBinaryRowHashCode() {
+ final Random rnd = new Random(System.currentTimeMillis());
+ // test hash stabilization
+ BinaryRowData row = new BinaryRowData(13);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ for (int i = 0; i < 99; i++) {
+ writer.reset();
+ writer.writeString(0, BinaryString.fromString("" + rnd.nextInt()));
+ writer.writeString(3, BinaryString.fromString("01234567"));
+ writer.writeString(5, BinaryString.fromString("012345678"));
+ writer.writeString(
+ 9, BinaryString.fromString("God in his heaven, alls right with the world"));
+ writer.writeBoolean(1, true);
+ writer.writeByte(2, (byte) 99);
+ writer.writeDouble(6, 87.1d);
+ writer.writeFloat(7, 26.1f);
+ writer.writeInt(8, 88);
+ writer.writeLong(10, 284);
+ writer.writeShort(11, (short) 292);
+ writer.setNullAt(12);
+ writer.complete();
+ BinaryRowData copy = row.copy();
+ assertThat(copy.hashCode()).isEqualTo(row.hashCode());
+ }
+
+ // test hash distribution
+ int count = 999999;
+ Set<Integer> hashCodes = new HashSet<>(count);
+ for (int i = 0; i < count; i++) {
+ row.setInt(8, i);
+ hashCodes.add(row.hashCode());
+ }
+ assertThat(hashCodes).hasSize(count);
+ hashCodes.clear();
+ row = new BinaryRowData(1);
+ writer = new BinaryRowWriter(row);
+ for (int i = 0; i < count; i++) {
+ writer.reset();
+ writer.writeString(
+ 0, BinaryString.fromString("God in his heaven, alls right with the world" + i));
+ writer.complete();
+ hashCodes.add(row.hashCode());
+ }
+ assertThat(hashCodes.size()).isGreaterThan((int) (count * 0.997));
+ }
+
+ @Test
+ public void testHeaderSize() {
+ assertThat(BinaryRowData.calculateBitSetWidthInBytes(56)).isEqualTo(8);
+ assertThat(BinaryRowData.calculateBitSetWidthInBytes(57)).isEqualTo(16);
+ assertThat(BinaryRowData.calculateBitSetWidthInBytes(120)).isEqualTo(16);
+ assertThat(BinaryRowData.calculateBitSetWidthInBytes(121)).isEqualTo(24);
+ }
+
+ @Test
+ public void testHeader() {
+ BinaryRowData row = new BinaryRowData(2);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+
+ writer.writeInt(0, 10);
+ writer.setNullAt(1);
+ writer.complete();
+
+ BinaryRowData newRow = row.copy();
+ assertThat(newRow).isEqualTo(row);
+ }
+
+ @Test
+ public void testDecimal() {
+ // 1.compact
+ {
+ int precision = 4;
+ int scale = 2;
+ BinaryRowData row = new BinaryRowData(2);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ writer.writeDecimal(0, Decimal.fromUnscaledLong(5, precision, scale), precision);
+ writer.setNullAt(1);
+ writer.complete();
+
+ assertThat(row.getDecimal(0, precision, scale).toString()).isEqualTo("0.05");
+ assertThat(row.isNullAt(1)).isTrue();
+ row.setDecimal(0, Decimal.fromUnscaledLong(6, precision, scale), precision);
+ assertThat(row.getDecimal(0, precision, scale).toString()).isEqualTo("0.06");
+ }
+
+ // 2.not compact
+ {
+ int precision = 25;
+ int scale = 5;
+ Decimal decimal1 = Decimal.fromBigDecimal(BigDecimal.valueOf(5.55), precision, scale);
+ Decimal decimal2 = Decimal.fromBigDecimal(BigDecimal.valueOf(6.55), precision, scale);
+
+ BinaryRowData row = new BinaryRowData(2);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ writer.writeDecimal(0, decimal1, precision);
+ writer.writeDecimal(1, null, precision);
+ writer.complete();
+
+ assertThat(row.getDecimal(0, precision, scale).toString()).isEqualTo("5.55000");
+ assertThat(row.isNullAt(1)).isTrue();
+ row.setDecimal(0, decimal2, precision);
+ assertThat(row.getDecimal(0, precision, scale).toString()).isEqualTo("6.55000");
+ }
+ }
+
+ @Test
+ public void testBinary() {
+ BinaryRowData row = new BinaryRowData(2);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ byte[] bytes1 = new byte[] {1, -1, 5};
+ byte[] bytes2 = new byte[] {1, -1, 5, 5, 1, 5, 1, 5};
+ writer.writeBinary(0, bytes1);
+ writer.writeBinary(1, bytes2);
+ writer.complete();
+
+ assertThat(row.getBinary(0, bytes1.length)).isEqualTo(bytes1);
+ assertThat(row.getBinary(1, bytes2.length)).isEqualTo(bytes2);
+ }
+
+ @Test
+ public void testZeroOutPaddingString() {
+
+ Random random = new Random();
+ byte[] bytes = new byte[1024];
+
+ BinaryRowData row = new BinaryRowData(1);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+
+ writer.reset();
+ random.nextBytes(bytes);
+ writer.writeBinary(0, bytes);
+ writer.reset();
+ writer.writeString(0, BinaryString.fromString("wahahah"));
+ writer.complete();
+ int hash1 = row.hashCode();
+
+ writer.reset();
+ random.nextBytes(bytes);
+ writer.writeBinary(0, bytes);
+ writer.reset();
+ writer.writeString(0, BinaryString.fromString("wahahah"));
+ writer.complete();
+ int hash2 = row.hashCode();
+
+ assertThat(hash2).isEqualTo(hash1);
+ }
+
+ @Test
+ public void testTimestampData() {
+ // 1. compact
+ {
+ final int precision = 3;
+ BinaryRowData row = new BinaryRowData(2);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ writer.writeTimestampNtz(0, TimestampNtz.fromMillis(123L), precision);
+ writer.setNullAt(1);
+ writer.complete();
+
+ assertThat(row.getTimestampNtz(0, 3).toString()).isEqualTo("1970-01-01T00:00:00.123");
+ assertThat(row.isNullAt(1)).isTrue();
+ row.setTimestampNtz(0, TimestampNtz.fromMillis(-123L), precision);
+ assertThat(row.getTimestampNtz(0, 3).toString()).isEqualTo("1969-12-31T23:59:59.877");
+ }
+
+ // 2. not compact
+ {
+ final int precision = 9;
+ TimestampLtz timestamp1 =
+ TimestampLtz.fromLocalDateTime(
Review Comment:
Create `TimestampLtz` from Java `Instant`.
##########
fluss-common/src/main/java/org/apache/fluss/row/BinaryRowWriter.java:
##########
@@ -0,0 +1,411 @@
+/*
+ * 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.row;
+
+import org.apache.fluss.memory.MemorySegment;
+import org.apache.fluss.types.DataType;
+import org.apache.fluss.types.DecimalType;
+import org.apache.fluss.types.LocalZonedTimestampType;
+import org.apache.fluss.types.TimestampType;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+
+import static org.apache.fluss.row.BinarySection.MAX_FIX_PART_DATA_SIZE;
+
+/**
+ * Writer for {@link BinaryRowData}.
+ *
+ * <p>Use the special format to write data to a {@link MemorySegment} (its capacity grows
+ * automatically).
+ *
+ * <p>If write a format binary: 1. New a writer. 2. Write each field by writeXX or setNullAt.
+ * (Variable length fields can not be written repeatedly.) 3. Invoke {@link #complete()}.
+ *
+ * <p>If want to reuse this writer, please invoke {@link #reset()} first.
+ */
+public final class BinaryRowWriter {
+
+ private final int nullBitsSizeInBytes;
+ private final BinaryRowData row;
+ private final int fixedSize;
+
+ protected MemorySegment segment;
+ protected int cursor;
Review Comment:
can be private.
##########
fluss-common/src/main/java/org/apache/fluss/row/BinaryRowData.java:
##########
@@ -0,0 +1,504 @@
+/*
+ * 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.row;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.memory.MemorySegment;
+import org.apache.fluss.types.DataType;
+import org.apache.fluss.types.DecimalType;
+import org.apache.fluss.types.LocalZonedTimestampType;
+import org.apache.fluss.types.TimestampType;
+
+import javax.annotation.Nullable;
+
+import java.nio.ByteOrder;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+
+/**
+ * An implementation of {@link InternalRow} which is backed by {@link MemorySegment} instead of
+ * Object. It can significantly reduce the serialization/deserialization of Java objects.
+ *
+ * <p>A Row has two part: Fixed-length part and variable-length part.
+ *
+ * <p>Fixed-length part contains 1 byte header and null bit set and field values. Null bit set is
+ * used for null tracking and is aligned to 8-byte word boundaries. `Field values` holds
+ * fixed-length primitive types and variable-length values which can be stored in 8 bytes inside. If
+ * it do not fit the variable-length field, then store the length and offset of variable-length
+ * part.
+ *
+ * <p>Fixed-length part will certainly fall into a MemorySegment, which will speed up the read and
+ * write of field. During the write phase, if the target memory segment has less space than fixed
+ * length part size, we will skip the space. So the number of fields in a single Row cannot exceed
+ * the capacity of a single MemorySegment, if there are too many fields, we suggest that user set a
+ * bigger pageSize of MemorySegment.
+ *
+ * <p>Variable-length part may fall into multiple MemorySegments.
+ */
+@Internal
+public final class BinaryRowData extends BinarySection
Review Comment:
`BinaryRowData` is a name in Flink, because it is `RowData` implementation in Flink. But it is strange if we use the same name in Fluss. Because `BinaryRowData` extends `BinaryRow` and `InternalRow`.
Considering `CompactedRow` and `IndexedRow` are both `BinaryRow` which maps `InternalRow` on a `MemorySegment`, so I suggest to use another name for `BinaryRowData`.
`BinaryRowData` aligns buffers with 8-bytes, maybe we can name it `AlignedRow`? We should also rename `BinaryRowWriter` and `BinaryRowDataTest` as well.
##########
fluss-common/src/main/java/org/apache/fluss/row/TimestampLtz.java:
##########
@@ -55,6 +57,21 @@ private TimestampLtz(long millisecond, int nanoOfMillisecond) {
this.nanoOfMillisecond = nanoOfMillisecond;
}
+ /**
+ * Creates an instance of {@link TimestampLtz} from an instance of {@link LocalDateTime}.
+ *
+ * @param dateTime an instance of {@link LocalDateTime}
+ */
+ public static TimestampLtz fromLocalDateTime(LocalDateTime dateTime) {
Review Comment:
It's on purpose we didn't introduce this static method on `TimestampLtz`, because `LocalDateTime` should be mapped to `TimestampNtz` instead of `TimestampLtz`. If converting `LocalDateTime` to `TimestampLtz`, users must to provides a local timezone parameter to convert the `LocalDateTime` into a epoch millisecond.
##########
fluss-common/src/main/java/org/apache/fluss/row/DataSetters.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.row;
+
+/**
+ * Provide type specialized setters to reduce if/else and eliminate box and unbox. This is mainly
+ * used on the binary format such as {@link BinaryRow}.
+ */
+public interface DataSetters {
Review Comment:
Better to name `TypedSetters`
--
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