voonhous commented on code in PR #19351:
URL: https://github.com/apache/hudi/pull/19351#discussion_r3636268639
##########
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:
nit: import `RowData` and write `(RowData) null`.
##########
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:
The duplicated `getRecordKey` / `toIndexedRecord` calls read like a
copy-paste slip. If the second calls are meant to hit the cached-key path, add
a short `// second call hits the cached key` comment so they survive future
cleanups; otherwise drop them.
##########
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);
Review Comment:
These three converts assert nothing, so they only catch exceptions. Capture
the results and reuse a couple of the value checks below, e.g.
```java
assertEquals("alice", ((RowData)
AvroToRowDataConverters.createRowConverter(ROW_TYPE)
.convert(avro)).getString(9).toString());
```
or drop them.
##########
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:
Same in this block: the non-UTC convert, the joda conversions, and the
double `JodaConverter.getConverter()` have no assertions. Concrete fixes:
- `assertEquals((int) LocalDate.of(2025, 2, 3).toEpochDay(), ...convert(new
org.joda.time.LocalDate(2025, 2, 3)))`
- `assertEquals(14706123, ...convert(new org.joda.time.LocalTime(4, 5, 6,
123)))`
- `assertSame(JodaConverter.getConverter(), JodaConverter.getConverter())`
if the point is the cached instance.
##########
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:
nit: this method mixes the throws-contract, `newInstance`, and logical-type
conversions, so the first failure hides the rest. Consider splitting into
`testUnsupportedOperations` and `testConvertColumnValueForLogicalType`.
##########
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:
This assertion pins a bug rather than a contract. In `HoodieSparkRecord` the
MICROS value is a raw micros `Long`, so `/ 1000` normalizes it to millis; here
the input is `TimestampData` and `getMillisecond()` is already millis, so the
MICROS branch returns **seconds** while MILLIS returns millis. That skews
`_hoodie_event_time` 1000x for micros fields.
On top of that, the real caller (`HoodieWriteHandle#getRecordMetadata`)
feeds this method the output of `getColumnValueAsJava`, which is a `Long` for
timestamps, so the `(TimestampData)` cast would CCE on the production path.
Can we fix the MICROS branch to `return ts.getMillisecond();` (matching
Spark's micros-to-millis semantics) and assert that here? If that's out of
scope, better to drop this assertion and file an issue than enshrine the
current output.
##########
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:
The `atLeastOnce()` checks prove some values got written, but the branches
this fixture sets up (null field, empty array) aren't asserted. Cheap
tightening:
```java
verify(consumer, never()).startField(eq("null_field"), anyInt());
```
The writer skips null fields entirely, and that's worth locking in.
##########
hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/parquet/TestParquetSchemaConverter.java:
##########
@@ -287,6 +287,65 @@ void testConvertTimestampTypes() {
assertThat(messageType.toString(), is(expected));
}
+ @Test
+ void testConvertAnnotatedPrimitiveParquetTypes() {
+ MessageType parquet = new MessageType("annotated",
+ Types.primitive(PrimitiveType.PrimitiveTypeName.INT32,
Type.Repetition.REQUIRED)
+ .as(LogicalTypeAnnotation.decimalType(2, 8)).named("decimal32"),
+ Types.primitive(PrimitiveType.PrimitiveTypeName.INT32,
Type.Repetition.OPTIONAL)
+ .as(LogicalTypeAnnotation.intType(8, true)).named("tiny"),
+ Types.primitive(PrimitiveType.PrimitiveTypeName.INT32,
Type.Repetition.OPTIONAL)
+ .as(LogicalTypeAnnotation.intType(16, true)).named("small"),
+ Types.primitive(PrimitiveType.PrimitiveTypeName.INT32,
Type.Repetition.OPTIONAL)
+ .as(LogicalTypeAnnotation.intType(32, true)).named("number"),
+ Types.primitive(PrimitiveType.PrimitiveTypeName.INT32,
Type.Repetition.OPTIONAL)
+ .as(LogicalTypeAnnotation.dateType()).named("day"),
+ Types.primitive(PrimitiveType.PrimitiveTypeName.INT32,
Type.Repetition.OPTIONAL)
+ .as(LogicalTypeAnnotation.timeType(true,
LogicalTypeAnnotation.TimeUnit.MILLIS)).named("time"),
+ Types.primitive(PrimitiveType.PrimitiveTypeName.INT64,
Type.Repetition.OPTIONAL)
+ .as(LogicalTypeAnnotation.decimalType(3, 12)).named("decimal64"),
+ Types.primitive(PrimitiveType.PrimitiveTypeName.INT96,
Type.Repetition.OPTIONAL)
+ .named("timestamp96"),
+ Types.primitive(PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY,
Type.Repetition.OPTIONAL)
+ .length(8).as(LogicalTypeAnnotation.decimalType(4,
16)).named("decimal_fixed"));
+
+ RowType converted = ParquetSchemaConverter.convertToRowType(parquet);
+ assertEquals("DECIMAL(8, 2) NOT NULL",
converted.getTypeAt(0).asSummaryString());
+ assertEquals("TINYINT", converted.getTypeAt(1).asSummaryString());
+ assertEquals("SMALLINT", converted.getTypeAt(2).asSummaryString());
+ assertEquals("INT", converted.getTypeAt(3).asSummaryString());
+ assertEquals("DATE", converted.getTypeAt(4).asSummaryString());
+ assertEquals("TIME(0)", converted.getTypeAt(5).asSummaryString());
+ assertEquals("DECIMAL(12, 3)", converted.getTypeAt(6).asSummaryString());
+ assertEquals("TIMESTAMP(9)", converted.getTypeAt(7).asSummaryString());
+ assertEquals("DECIMAL(16, 4)", converted.getTypeAt(8).asSummaryString());
+ }
+
+ @Test
+ void testConvertLocalZonedTimestampToParquet() {
+ RowType rowType = (RowType) DataTypes.ROW(
+ DataTypes.FIELD("local_millis", DataTypes.TIMESTAMP_LTZ(3)),
+ DataTypes.FIELD("local_micros",
DataTypes.TIMESTAMP_LTZ(6))).notNull().getLogicalType();
+ MessageType parquet =
ParquetSchemaConverter.convertToParquetMessageType("local", rowType);
+ assertTrue(parquet.toString().contains("TIMESTAMP(MILLIS,false)"));
+ assertTrue(parquet.toString().contains("TIMESTAMP(MICROS,false)"));
Review Comment:
String-matching `MessageType.toString()` ties the test to Parquet's
rendering. Assert the annotation instead, like
`testConvertBinaryDateAndTimeToParquet` below:
```java
assertEquals(LogicalTypeAnnotation.timestampType(false,
LogicalTypeAnnotation.TimeUnit.MILLIS),
parquet.getType("local_millis").getLogicalTypeAnnotation());
```
##########
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:
nit: existing tests in this module use `public class` / `public void`; worth
keeping one style per module.
--
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]
