danny0405 commented on code in PR #19351:
URL: https://github.com/apache/hudi/pull/19351#discussion_r3636463762
##########
hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/model/TestHoodieFlinkRecord.java:
##########
@@ -266,4 +278,75 @@ public void testUpdateMetaFieldWithoutOperationField() {
HoodieFlinkRecord updatedRecord = (HoodieFlinkRecord)
record.updateMetaField(schema, 0, "20240101000001");
assertEquals(HoodieOperation.INSERT, updatedRecord.getOperation());
}
+
+ @Test
+ public void testRecordContractAndLogicalTypeConversions() {
+ HoodieKey key = new HoodieKey("id-001", "partition-1");
+ GenericRowData row = GenericRowData.of(StringData.fromString("id-001"));
+ HoodieFlinkRecord record = new HoodieFlinkRecord(
+ key, HoodieOperation.INSERT, 100L, row, true);
+
+ assertEquals(HoodieRecord.HoodieRecordType.FLINK, record.getRecordType());
+ assertEquals(key, record.newInstance().getKey());
+ assertEquals("new-key", record.newInstance(
+ new HoodieKey("new-key", "p"),
HoodieOperation.UPDATE_AFTER).getRecordKey());
+ assertEquals("another-key", record.newInstance(
+ new HoodieKey("another-key", "p")).getRecordKey());
+ assertFalse(record.shouldIgnore(null, new Properties()));
+ assertSame(record, record.copy());
+ assertTrue(record.getMetadata().isEmpty());
+
+ assertNull(record.convertColumnValueForLogicalType(
+ HoodieSchema.create(HoodieSchemaType.STRING), null, true));
+ assertEquals(LocalDate.ofEpochDay(2),
record.convertColumnValueForLogicalType(
+ HoodieSchema.createDate(), 2, true));
+ TimestampData timestamp =
TimestampData.fromInstant(Instant.parse("2025-02-03T04:05:06.123456Z"));
+ assertEquals(timestamp.getMillisecond(),
record.convertColumnValueForLogicalType(
+ HoodieSchema.createTimestampMillis(), timestamp, true));
+ assertEquals(timestamp.getMillisecond() / 1000,
record.convertColumnValueForLogicalType(
+ HoodieSchema.createTimestampMicros(), timestamp, true));
+ assertEquals(new BigDecimal("12.34"),
record.convertColumnValueForLogicalType(
+ HoodieSchema.createDecimal("decimal", null, null, 10, 2, 5),
+ DecimalData.fromBigDecimal(new BigDecimal("12.34"), 10, 2), true));
+ assertSame(timestamp, record.convertColumnValueForLogicalType(
+ HoodieSchema.createTimestampMillis(), timestamp, false));
+
+ assertThrows(UnsupportedOperationException.class,
+ () -> record.writeRecordPayload(row, null, null));
+ assertThrows(UnsupportedOperationException.class,
+ () -> record.readRecordPayload(null, null));
+ assertThrows(UnsupportedOperationException.class,
+ () -> record.getColumnValues(null, null, false));
+ assertThrows(UnsupportedOperationException.class,
+ () -> record.joinWith(record, null));
+ assertThrows(UnsupportedOperationException.class,
+ () -> record.wrapIntoHoodieRecordPayloadWithParams(
+ null, null, Option.empty(), false, Option.empty(), false,
Option.empty()));
+ assertThrows(UnsupportedOperationException.class,
+ () -> record.wrapIntoHoodieRecordPayloadWithKeyGen(null, null,
Option.empty()));
+ assertThrows(UnsupportedOperationException.class,
+ () -> record.truncateRecordKey(null, null, null));
+
+ HoodieFlinkRecord empty = new HoodieFlinkRecord(
+ key, HoodieOperation.INSERT, (org.apache.flink.table.data.RowData)
null);
+ assertTrue(empty.checkIsDelete(null, new Properties()));
+ }
+
+ @Test
+ public void testKeyLookupAndAvroMaterialization() {
+ HoodieSchema schema = HoodieSchema.createRecord("test", null, null,
Arrays.asList(
+ HoodieSchemaField.of("id",
HoodieSchema.create(HoodieSchemaType.STRING)),
+ HoodieSchemaField.of("value",
HoodieSchema.create(HoodieSchemaType.INT))));
+ HoodieFlinkRecord record = new HoodieFlinkRecord(GenericRowData.of(
+ StringData.fromString("id-001"), 42));
+
+ assertEquals("id-001", record.getRecordKey(schema, "id"));
+ assertEquals("id-001", record.getRecordKey(schema, "id"));
+ assertTrue(record.toIndexedRecord(schema, new Properties()).isPresent());
+ assertEquals("id-001", record.toIndexedRecord(schema, new Properties())
Review Comment:
Addressed in `4b400756d4ee`: retained the second `getRecordKey` call with a
comment explaining that it exercises the cached-key path, and removed the
redundant first `toIndexedRecord` call.
##########
hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestRowDataAvroRoundTrip.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * 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.hudi.util;
+
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaType;
+
+import org.apache.avro.generic.GenericRecord;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.GenericMapData;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.logical.RowType;
+import org.joda.time.DateTime;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.nio.ByteBuffer;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalTime;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestRowDataAvroRoundTrip {
Review Comment:
Addressed in `4b400756d4ee`: `TestRowDataAvroRoundTrip` and its test methods
are now public, matching the module's existing style.
##########
hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/model/TestHoodieFlinkRecord.java:
##########
@@ -266,4 +278,75 @@ public void testUpdateMetaFieldWithoutOperationField() {
HoodieFlinkRecord updatedRecord = (HoodieFlinkRecord)
record.updateMetaField(schema, 0, "20240101000001");
assertEquals(HoodieOperation.INSERT, updatedRecord.getOperation());
}
+
+ @Test
+ public void testRecordContractAndLogicalTypeConversions() {
Review Comment:
Addressed in `4b400756d4ee`: split this into `testRecordContract`,
`testConvertColumnValueForLogicalType`, and `testUnsupportedOperations` so
failures are isolated by behavior.
##########
hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/model/TestHoodieFlinkRecord.java:
##########
@@ -266,4 +278,75 @@ public void testUpdateMetaFieldWithoutOperationField() {
HoodieFlinkRecord updatedRecord = (HoodieFlinkRecord)
record.updateMetaField(schema, 0, "20240101000001");
assertEquals(HoodieOperation.INSERT, updatedRecord.getOperation());
}
+
+ @Test
+ public void testRecordContractAndLogicalTypeConversions() {
+ HoodieKey key = new HoodieKey("id-001", "partition-1");
+ GenericRowData row = GenericRowData.of(StringData.fromString("id-001"));
+ HoodieFlinkRecord record = new HoodieFlinkRecord(
+ key, HoodieOperation.INSERT, 100L, row, true);
+
+ assertEquals(HoodieRecord.HoodieRecordType.FLINK, record.getRecordType());
+ assertEquals(key, record.newInstance().getKey());
+ assertEquals("new-key", record.newInstance(
+ new HoodieKey("new-key", "p"),
HoodieOperation.UPDATE_AFTER).getRecordKey());
+ assertEquals("another-key", record.newInstance(
+ new HoodieKey("another-key", "p")).getRecordKey());
+ assertFalse(record.shouldIgnore(null, new Properties()));
+ assertSame(record, record.copy());
+ assertTrue(record.getMetadata().isEmpty());
+
+ assertNull(record.convertColumnValueForLogicalType(
+ HoodieSchema.create(HoodieSchemaType.STRING), null, true));
+ assertEquals(LocalDate.ofEpochDay(2),
record.convertColumnValueForLogicalType(
+ HoodieSchema.createDate(), 2, true));
+ TimestampData timestamp =
TimestampData.fromInstant(Instant.parse("2025-02-03T04:05:06.123456Z"));
+ assertEquals(timestamp.getMillisecond(),
record.convertColumnValueForLogicalType(
+ HoodieSchema.createTimestampMillis(), timestamp, true));
+ assertEquals(timestamp.getMillisecond() / 1000,
record.convertColumnValueForLogicalType(
+ HoodieSchema.createTimestampMicros(), timestamp, true));
Review Comment:
Fixed in `4b400756d4ee`. `HoodieFlinkRecord` now consumes the `Long`
returned by `getColumnValueAsJava`, preserves millis values, and divides micros
by 1,000, matching the Spark contract. The test now exercises the real caller
values for both timestamp precisions, so it covers both the former CCE and the
unit conversion.
##########
hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/model/TestHoodieFlinkRecord.java:
##########
@@ -266,4 +278,75 @@ public void testUpdateMetaFieldWithoutOperationField() {
HoodieFlinkRecord updatedRecord = (HoodieFlinkRecord)
record.updateMetaField(schema, 0, "20240101000001");
assertEquals(HoodieOperation.INSERT, updatedRecord.getOperation());
}
+
+ @Test
+ public void testRecordContractAndLogicalTypeConversions() {
+ HoodieKey key = new HoodieKey("id-001", "partition-1");
+ GenericRowData row = GenericRowData.of(StringData.fromString("id-001"));
+ HoodieFlinkRecord record = new HoodieFlinkRecord(
+ key, HoodieOperation.INSERT, 100L, row, true);
+
+ assertEquals(HoodieRecord.HoodieRecordType.FLINK, record.getRecordType());
+ assertEquals(key, record.newInstance().getKey());
+ assertEquals("new-key", record.newInstance(
+ new HoodieKey("new-key", "p"),
HoodieOperation.UPDATE_AFTER).getRecordKey());
+ assertEquals("another-key", record.newInstance(
+ new HoodieKey("another-key", "p")).getRecordKey());
+ assertFalse(record.shouldIgnore(null, new Properties()));
+ assertSame(record, record.copy());
+ assertTrue(record.getMetadata().isEmpty());
+
+ assertNull(record.convertColumnValueForLogicalType(
+ HoodieSchema.create(HoodieSchemaType.STRING), null, true));
+ assertEquals(LocalDate.ofEpochDay(2),
record.convertColumnValueForLogicalType(
+ HoodieSchema.createDate(), 2, true));
+ TimestampData timestamp =
TimestampData.fromInstant(Instant.parse("2025-02-03T04:05:06.123456Z"));
+ assertEquals(timestamp.getMillisecond(),
record.convertColumnValueForLogicalType(
+ HoodieSchema.createTimestampMillis(), timestamp, true));
+ assertEquals(timestamp.getMillisecond() / 1000,
record.convertColumnValueForLogicalType(
+ HoodieSchema.createTimestampMicros(), timestamp, true));
+ assertEquals(new BigDecimal("12.34"),
record.convertColumnValueForLogicalType(
+ HoodieSchema.createDecimal("decimal", null, null, 10, 2, 5),
+ DecimalData.fromBigDecimal(new BigDecimal("12.34"), 10, 2), true));
+ assertSame(timestamp, record.convertColumnValueForLogicalType(
+ HoodieSchema.createTimestampMillis(), timestamp, false));
+
+ assertThrows(UnsupportedOperationException.class,
+ () -> record.writeRecordPayload(row, null, null));
+ assertThrows(UnsupportedOperationException.class,
+ () -> record.readRecordPayload(null, null));
+ assertThrows(UnsupportedOperationException.class,
+ () -> record.getColumnValues(null, null, false));
+ assertThrows(UnsupportedOperationException.class,
+ () -> record.joinWith(record, null));
+ assertThrows(UnsupportedOperationException.class,
+ () -> record.wrapIntoHoodieRecordPayloadWithParams(
+ null, null, Option.empty(), false, Option.empty(), false,
Option.empty()));
+ assertThrows(UnsupportedOperationException.class,
+ () -> record.wrapIntoHoodieRecordPayloadWithKeyGen(null, null,
Option.empty()));
+ assertThrows(UnsupportedOperationException.class,
+ () -> record.truncateRecordKey(null, null, null));
+
+ HoodieFlinkRecord empty = new HoodieFlinkRecord(
+ key, HoodieOperation.INSERT, (org.apache.flink.table.data.RowData)
null);
Review Comment:
Addressed in `4b400756d4ee`: imported `RowData` and now use `(RowData) null`.
##########
hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/parquet/TestParquetRowDataWriter.java:
##########
@@ -0,0 +1,173 @@
+/*
+ * 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.hudi.io.storage.row.parquet;
+
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.util.HoodieSchemaConverter;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.GenericMapData;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.parquet.io.api.RecordConsumer;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.time.Instant;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+class TestParquetRowDataWriter {
+
+ @Test
+ void testWritePrimitiveNestedArrayMapDecimalAndTimestampValues() {
+ RowType rowType = (RowType) DataTypes.ROW(
+ DataTypes.FIELD("text", DataTypes.STRING()),
+ DataTypes.FIELD("flag", DataTypes.BOOLEAN()),
+ DataTypes.FIELD("bytes", DataTypes.BYTES()),
+ DataTypes.FIELD("tiny", DataTypes.TINYINT()),
+ DataTypes.FIELD("small", DataTypes.SMALLINT()),
+ DataTypes.FIELD("number", DataTypes.INT()),
+ DataTypes.FIELD("big", DataTypes.BIGINT()),
+ DataTypes.FIELD("ratio", DataTypes.FLOAT()),
+ DataTypes.FIELD("score", DataTypes.DOUBLE()),
+ DataTypes.FIELD("day", DataTypes.DATE()),
+ DataTypes.FIELD("time", DataTypes.TIME(3)),
+ DataTypes.FIELD("small_decimal", DataTypes.DECIMAL(10, 2)),
+ DataTypes.FIELD("large_decimal", DataTypes.DECIMAL(30, 4)),
+ DataTypes.FIELD("timestamp3", DataTypes.TIMESTAMP(3)),
+ DataTypes.FIELD("timestamp6", DataTypes.TIMESTAMP_LTZ(6)),
+ DataTypes.FIELD("items", DataTypes.ARRAY(DataTypes.STRING())),
+ DataTypes.FIELD("empty_items", DataTypes.ARRAY(DataTypes.INT())),
+ DataTypes.FIELD("attributes", DataTypes.MAP(DataTypes.STRING(),
DataTypes.INT())),
+ DataTypes.FIELD("nested", DataTypes.ROW(
+ DataTypes.FIELD("id", DataTypes.BIGINT()),
+ DataTypes.FIELD("label", DataTypes.STRING()),
+ DataTypes.FIELD("optional", DataTypes.INT()))),
+ DataTypes.FIELD("null_field", DataTypes.STRING()))
+ .notNull().getLogicalType();
+ HoodieSchema schema = HoodieSchemaConverter.convertToSchema(rowType,
"writer_record");
+
+ Map<Object, Object> attributes = new LinkedHashMap<>();
+ attributes.put(StringData.fromString("present"), 1);
+ attributes.put(StringData.fromString("missing"), null);
+ GenericRowData row = GenericRowData.of(
+ StringData.fromString("hello"), true, new byte[] {1, 2},
+ 3, 4, 5, 6L, 7.5f, 8.25d, 9, 10,
+ DecimalData.fromBigDecimal(new BigDecimal("12.34"), 10, 2),
+ DecimalData.fromBigDecimal(new
BigDecimal("12345678901234567890.1234"), 30, 4),
+ TimestampData.fromInstant(Instant.parse("2025-02-03T04:05:06.123Z")),
+
TimestampData.fromInstant(Instant.parse("2025-02-03T04:05:06.123456Z")),
+ new GenericArrayData(new Object[] {StringData.fromString("a"), null,
StringData.fromString("c")}),
+ new GenericArrayData(new int[0]),
+ new GenericMapData(attributes),
+ GenericRowData.of(99L, StringData.fromString("nested"), null),
+ null);
+
+ RecordConsumer consumer = mock(RecordConsumer.class);
+ new ParquetRowDataWriter(consumer, true, schema).write(row);
+
+ verify(consumer).startMessage();
+ verify(consumer).endMessage();
+ verify(consumer, atLeastOnce()).startField(any(String.class), anyInt());
+ verify(consumer, atLeastOnce()).addBoolean(true);
+ verify(consumer, atLeastOnce()).addInteger(5);
+ verify(consumer, atLeastOnce()).addLong(6L);
+ verify(consumer, atLeastOnce()).addFloat(7.5f);
+ verify(consumer, atLeastOnce()).addDouble(8.25d);
+ verify(consumer, atLeastOnce()).addBinary(any());
Review Comment:
Addressed in `4b400756d4ee`: added `verify(consumer,
never()).startField(eq("null_field"), anyInt())` to pin null-field omission.
##########
hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestRowDataAvroRoundTrip.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * 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.hudi.util;
+
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaType;
+
+import org.apache.avro.generic.GenericRecord;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.GenericMapData;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.logical.RowType;
+import org.joda.time.DateTime;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.nio.ByteBuffer;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalTime;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestRowDataAvroRoundTrip {
+
+ private static final RowType ROW_TYPE = (RowType) DataTypes.ROW(
+ DataTypes.FIELD("tiny", DataTypes.TINYINT().notNull()),
+ DataTypes.FIELD("small", DataTypes.SMALLINT().notNull()),
+ DataTypes.FIELD("flag", DataTypes.BOOLEAN().notNull()),
+ DataTypes.FIELD("number", DataTypes.INT().notNull()),
+ DataTypes.FIELD("big", DataTypes.BIGINT().notNull()),
+ DataTypes.FIELD("ratio", DataTypes.FLOAT().notNull()),
+ DataTypes.FIELD("score", DataTypes.DOUBLE().notNull()),
+ DataTypes.FIELD("day", DataTypes.DATE().notNull()),
+ DataTypes.FIELD("time", DataTypes.TIME(3).notNull()),
+ DataTypes.FIELD("name", DataTypes.STRING()),
+ DataTypes.FIELD("payload", DataTypes.BYTES()),
+ DataTypes.FIELD("amount", DataTypes.DECIMAL(20, 4)),
+ DataTypes.FIELD("timestamp3", DataTypes.TIMESTAMP(3)),
+ DataTypes.FIELD("timestamp6", DataTypes.TIMESTAMP(6)),
+ DataTypes.FIELD("local_timestamp3", DataTypes.TIMESTAMP_LTZ(3)),
+ DataTypes.FIELD("local_timestamp6", DataTypes.TIMESTAMP_LTZ(6)),
+ DataTypes.FIELD("items", DataTypes.ARRAY(DataTypes.STRING())),
+ DataTypes.FIELD("attributes", DataTypes.MAP(DataTypes.STRING(),
DataTypes.INT())),
+ DataTypes.FIELD("nested", DataTypes.ROW(
+ DataTypes.FIELD("id", DataTypes.BIGINT()),
+ DataTypes.FIELD("label", DataTypes.STRING()))),
+ DataTypes.FIELD("nullable", DataTypes.STRING()))
+ .notNull().getLogicalType();
+
+ @Test
+ void testAllSupportedTypesRoundTripValueByValue() {
+ Instant instant = Instant.parse("2025-02-03T04:05:06.123456Z");
+ Map<Object, Object> attributes = new LinkedHashMap<>();
+ attributes.put(StringData.fromString("one"), 1);
+ attributes.put(StringData.fromString("missing"), null);
+ GenericRowData input = GenericRowData.of(
+ (byte) 1, (short) 2, true, 3, 4L, 5.5f, 6.25d, 20, 1234,
+ StringData.fromString("alice"), new byte[] {7, 8},
+ DecimalData.fromBigDecimal(new BigDecimal("123456789012.3400"), 20, 4),
+ TimestampData.fromInstant(instant), TimestampData.fromInstant(instant),
+ TimestampData.fromInstant(instant), TimestampData.fromInstant(instant),
+ new GenericArrayData(new Object[] {StringData.fromString("a"), null,
StringData.fromString("c")}),
+ new GenericMapData(attributes),
+ GenericRowData.of(99L, StringData.fromString("nested")),
+ null);
+
+ HoodieSchema schema = HoodieSchemaConverter.convertToSchema(ROW_TYPE,
"AllTypes");
+ GenericRecord avro = (GenericRecord)
RowDataToAvroConverters.createConverter(ROW_TYPE)
+ .convert(schema, input);
+ RowData output = (RowData) AvroToRowDataConverters
+ .createRowConverter(schema, ROW_TYPE, true).convert(avro);
+ AvroToRowDataConverters.createRowConverter(ROW_TYPE).convert(avro);
+ AvroToRowDataConverters.createRowConverter(ROW_TYPE, false).convert(avro);
+ AvroToRowDataConverters.createRowConverter(schema).convert(avro);
+
+ assertEquals((byte) 1, output.getByte(0));
+ assertEquals((short) 2, output.getShort(1));
+ assertTrue(output.getBoolean(2));
+ assertEquals(3, output.getInt(3));
+ assertEquals(4L, output.getLong(4));
+ assertEquals(5.5f, output.getFloat(5));
+ assertEquals(6.25d, output.getDouble(6));
+ assertEquals(20, output.getInt(7));
+ assertEquals(1234, output.getInt(8));
+ assertEquals("alice", output.getString(9).toString());
+ assertArrayEquals(new byte[] {7, 8}, output.getBinary(10));
+ assertEquals(new BigDecimal("123456789012.3400"), output.getDecimal(11,
20, 4).toBigDecimal());
+ assertEquals(instant.toEpochMilli(), output.getTimestamp(12,
3).getMillisecond());
+ assertEquals(instant, output.getTimestamp(13, 6).toInstant());
+ assertEquals(instant.toEpochMilli(), output.getTimestamp(14,
3).getMillisecond());
+ assertEquals(instant, output.getTimestamp(15, 6).toInstant());
+ assertEquals("a", output.getArray(16).getString(0).toString());
+ assertTrue(output.getArray(16).isNullAt(1));
+ assertEquals("c", output.getArray(16).getString(2).toString());
+ assertMap(output.getMap(17));
+ assertEquals(99L, output.getRow(18, 2).getLong(0));
+ assertEquals("nested", output.getRow(18, 2).getString(1).toString());
+ assertTrue(output.isNullAt(19));
+ }
+
+ @Test
+ void testConvertersAcceptAvroLogicalRuntimeRepresentations() {
+ HoodieSchema intSchema =
HoodieSchemaConverter.convertToSchema(DataTypes.INT().getLogicalType());
+
assertNull(AvroToRowDataConverters.createConverter(DataTypes.NULL().getLogicalType(),
true)
+ .convert("ignored"));
+ assertEquals((byte) 3, AvroToRowDataConverters.createConverter(
+ DataTypes.TINYINT().getLogicalType(), true).convert(3));
+ assertEquals((short) 4, AvroToRowDataConverters.createConverter(
+ DataTypes.SMALLINT().getLogicalType(), true).convert(4));
+ assertEquals(5, AvroToRowDataConverters.createConverter(
+ DataTypes.INT().getLogicalType(), true).convert(5));
+ assertEquals(2, AvroToRowDataConverters.createConverter(
+ DataTypes.DATE().getLogicalType(),
true).convert(LocalDate.ofEpochDay(2)));
+ assertEquals(1234, AvroToRowDataConverters.createConverter(
+ DataTypes.TIME(3).getLogicalType(),
true).convert(LocalTime.ofNanoOfDay(1_234_000_000L)));
+ assertArrayEquals(new byte[] {1, 2}, (byte[])
AvroToRowDataConverters.createConverter(
+ DataTypes.BYTES().getLogicalType(), true).convert(new byte[] {1, 2}));
+ byte[] unscaled = new BigInteger("1234").toByteArray();
+ assertEquals(new BigDecimal("12.34"), ((DecimalData)
AvroToRowDataConverters.createConverter(
+ DataTypes.DECIMAL(8, 2).getLogicalType(),
true).convert(ByteBuffer.wrap(unscaled))).toBigDecimal());
+ assertEquals(new BigDecimal("12.34"), ((DecimalData)
AvroToRowDataConverters.createConverter(
+ DataTypes.DECIMAL(8, 2).getLogicalType(),
true).convert(unscaled)).toBigDecimal());
+
+ Instant instant = Instant.parse("2025-02-03T04:05:06.123Z");
+ assertEquals(instant, ((TimestampData)
AvroToRowDataConverters.createConverter(
+ DataTypes.TIMESTAMP(3).getLogicalType(),
true).convert(instant)).toInstant());
+ assertEquals(instant.toEpochMilli(), ((TimestampData)
AvroToRowDataConverters.createConverter(
+ DataTypes.TIMESTAMP(3).getLogicalType(), true).convert(new
DateTime(instant.toEpochMilli())))
+ .getMillisecond());
+ AvroToRowDataConverters.createConverter(
+ DataTypes.TIMESTAMP(3).getLogicalType(),
false).convert(instant.toEpochMilli());
+ AvroToRowDataConverters.createConverter(DataTypes.DATE().getLogicalType(),
true)
+ .convert(new org.joda.time.LocalDate(2025, 2, 3));
+
AvroToRowDataConverters.createConverter(DataTypes.TIME(3).getLogicalType(),
true)
+ .convert(new org.joda.time.LocalTime(4, 5, 6, 123));
+ AvroToRowDataConverters.JodaConverter.getConverter();
+ AvroToRowDataConverters.JodaConverter.getConverter();
Review Comment:
Addressed in `4b400756d4ee`: added assertions for the non-UTC timestamp,
Joda date/time conversions, and cached Joda converter instance. The date
assertion also exposed an existing epoch-millis/truncation bug, so
`JodaConverter.convertDate` now returns epoch days.
--
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]
