This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 427efb265429 fix(flink): normalize row logical conversions and improve
coverage (#19351)
427efb265429 is described below
commit 427efb2654296cd78c93e2602ca06f6f82bf0733
Author: Danny Chan <[email protected]>
AuthorDate: Thu Jul 23 17:42:02 2026 +0800
fix(flink): normalize row logical conversions and improve coverage (#19351)
* test(flink): improve row conversion coverage
---
.../hudi/client/model/HoodieFlinkRecord.java | 6 +-
.../apache/hudi/util/AvroToRowDataConverters.java | 3 +-
.../TestHoodieFlinkInternalRowSerializer.java | 125 ++++++++++++
.../hudi/client/model/TestHoodieFlinkRecord.java | 116 +++++++++++
.../row/parquet/TestParquetRowDataWriter.java | 176 +++++++++++++++++
.../row/parquet/TestParquetSchemaConverter.java | 63 ++++++
.../apache/hudi/util/TestRowDataAvroRoundTrip.java | 212 +++++++++++++++++++++
.../org/apache/hudi/util/TestRowDataUtils.java | 127 ++++++++++++
.../hudi/util/TestVectorConversionUtils.java | 156 +++++++++++++++
9 files changed, 979 insertions(+), 5 deletions(-)
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/model/HoodieFlinkRecord.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/model/HoodieFlinkRecord.java
index 5bf98cfdd2ce..913cd5e68350 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/model/HoodieFlinkRecord.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/model/HoodieFlinkRecord.java
@@ -46,7 +46,6 @@ import org.apache.flink.table.data.DecimalData;
import org.apache.flink.table.data.GenericRowData;
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.data.utils.JoinedRowData;
import java.io.ByteArrayOutputStream;
@@ -170,11 +169,10 @@ public class HoodieFlinkRecord extends
HoodieRecord<RowData> {
return LocalDate.ofEpochDay(((Integer) fieldValue).longValue());
} else if (schemaType == HoodieSchemaType.TIMESTAMP &&
keepConsistentLogicalTimestamp) {
HoodieSchema.Timestamp timestampSchema = (HoodieSchema.Timestamp)
fieldSchema;
- TimestampData ts = (TimestampData) fieldValue;
if (timestampSchema.getPrecision() == HoodieSchema.TimePrecision.MILLIS)
{
- return ts.getMillisecond();
+ return fieldValue;
} else if (timestampSchema.getPrecision() ==
HoodieSchema.TimePrecision.MICROS) {
- return ts.getMillisecond() / 1000;
+ return ((Long) fieldValue) / 1000;
}
} else if (schemaType == HoodieSchemaType.DECIMAL) {
return ((DecimalData) fieldValue).toBigDecimal();
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/AvroToRowDataConverters.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/AvroToRowDataConverters.java
index 0a7d421c2b83..cc6c90c341f8 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/AvroToRowDataConverters.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/AvroToRowDataConverters.java
@@ -412,7 +412,8 @@ public class AvroToRowDataConverters {
public long convertDate(Object object) {
final org.joda.time.LocalDate value = (org.joda.time.LocalDate) object;
- return value.toDate().getTime();
+ return LocalDate.of(
+ value.getYear(), value.getMonthOfYear(),
value.getDayOfMonth()).toEpochDay();
}
public int convertTime(Object object) {
diff --git
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/model/TestHoodieFlinkInternalRowSerializer.java
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/model/TestHoodieFlinkInternalRowSerializer.java
new file mode 100644
index 000000000000..60227372c7f1
--- /dev/null
+++
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/model/TestHoodieFlinkInternalRowSerializer.java
@@ -0,0 +1,125 @@
+/*
+ * 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.client.model;
+
+import org.apache.flink.core.memory.DataInputDeserializer;
+import org.apache.flink.core.memory.DataOutputSerializer;
+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.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.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.time.Instant;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestHoodieFlinkInternalRowSerializer {
+
+ private static final RowType ROW_TYPE = (RowType) DataTypes.ROW(
+ DataTypes.FIELD("name", DataTypes.STRING()),
+ DataTypes.FIELD("amount", DataTypes.DECIMAL(10, 2)),
+ DataTypes.FIELD("created", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(6)),
+ DataTypes.FIELD("tags", DataTypes.ARRAY(DataTypes.STRING())))
+ .getLogicalType();
+
+ @Test
+ void testRoundTripCopyAndStreamCopy() throws Exception {
+ HoodieFlinkInternalRowSerializer serializer = new
HoodieFlinkInternalRowSerializer(ROW_TYPE);
+ GenericRowData payload = GenericRowData.of(
+ StringData.fromString("alice"),
+ DecimalData.fromBigDecimal(new BigDecimal("12.34"), 10, 2),
+
TimestampData.fromInstant(Instant.parse("2025-02-03T04:05:06.123456Z")),
+ new GenericArrayData(new Object[] {StringData.fromString("x"), null}));
+ HoodieFlinkInternalRow original = new HoodieFlinkInternalRow(
+ "key", "partition", "file", "instant", "I", false, payload);
+
+ DataOutputSerializer output = new DataOutputSerializer(128);
+ serializer.serialize(original, output);
+ HoodieFlinkInternalRow restored = serializer.deserialize(
+ new DataInputDeserializer(output.getCopyOfBuffer()));
+ assertDataRecord(restored);
+
+ HoodieFlinkInternalRow copied = serializer.copy(original);
+ assertNotSame(original, copied);
+ assertNotSame(original.getRowData(), copied.getRowData());
+ assertDataRecord(copied);
+
+ DataOutputSerializer copiedOutput = new DataOutputSerializer(128);
+ serializer.copy(new DataInputDeserializer(output.getCopyOfBuffer()),
copiedOutput);
+ assertDataRecord(serializer.deserialize(new
DataInputDeserializer(copiedOutput.getCopyOfBuffer())));
+ }
+
+ @Test
+ void testIndexRecordRoundTripAndSerializerContract() throws Exception {
+ HoodieFlinkInternalRowSerializer serializer = new
HoodieFlinkInternalRowSerializer(ROW_TYPE);
+ HoodieFlinkInternalRow indexRecord =
+ new HoodieFlinkInternalRow("key", "partition", "file", "instant");
+ DataOutputSerializer output = new DataOutputSerializer(64);
+ serializer.serialize(indexRecord, output);
+
+ HoodieFlinkInternalRow restored = serializer.deserialize(
+ indexRecord, new DataInputDeserializer(output.getCopyOfBuffer()));
+ assertTrue(restored.isIndexRecord());
+ assertEquals("key", restored.getRecordKey());
+ assertEquals("partition", restored.getPartitionPath());
+ assertEquals("file", restored.getFileId());
+ assertEquals("instant", restored.getInstantTime());
+ assertEquals("", restored.getOperationType());
+
+ DataOutputSerializer copiedOutput = new DataOutputSerializer(64);
+ serializer.copy(new DataInputDeserializer(output.getCopyOfBuffer()),
copiedOutput);
+ assertTrue(serializer.deserialize(
+ new
DataInputDeserializer(copiedOutput.getCopyOfBuffer())).isIndexRecord());
+
+ assertFalse(serializer.isImmutableType());
+ assertEquals(-1, serializer.getLength());
+ assertEquals(serializer, serializer.duplicate());
+ assertEquals(serializer.hashCode(), serializer.duplicate().hashCode());
+ assertFalse(serializer.equals(null));
+ assertFalse(serializer.equals("serializer"));
+ assertThrows(UnsupportedOperationException.class,
serializer::createInstance);
+ assertThrows(UnsupportedOperationException.class,
+ () -> serializer.copy(indexRecord, indexRecord));
+ assertThrows(UnsupportedOperationException.class,
serializer::snapshotConfiguration);
+ }
+
+ private static void assertDataRecord(HoodieFlinkInternalRow record) {
+ assertFalse(record.isIndexRecord());
+ assertEquals("key", record.getRecordKey());
+ assertEquals("partition", record.getPartitionPath());
+ assertEquals("file", record.getFileId());
+ assertEquals("instant", record.getInstantTime());
+ assertEquals("I", record.getOperationType());
+ assertEquals("alice", record.getRowData().getString(0).toString());
+ assertEquals(new BigDecimal("12.34"), record.getRowData().getDecimal(1,
10, 2).toBigDecimal());
+ assertEquals(Instant.parse("2025-02-03T04:05:06.123456Z"),
+ record.getRowData().getTimestamp(2, 6).toInstant());
+ assertEquals("x", record.getRowData().getArray(3).getString(0).toString());
+ assertTrue(record.getRowData().getArray(3).isNullAt(1));
+ }
+}
diff --git
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/model/TestHoodieFlinkRecord.java
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/model/TestHoodieFlinkRecord.java
index 6744e2d14830..9753628da410 100644
---
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/model/TestHoodieFlinkRecord.java
+++
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/model/TestHoodieFlinkRecord.java
@@ -20,20 +20,33 @@ package org.apache.hudi.client.model;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieOperation;
+import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.schema.HoodieSchemaField;
import org.apache.hudi.common.schema.HoodieSchemaType;
+import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.OrderingValues;
+import org.apache.flink.table.data.DecimalData;
import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
import org.junit.jupiter.api.Test;
+import java.math.BigDecimal;
+import java.time.Instant;
+import java.time.LocalDate;
import java.util.Arrays;
import java.util.Properties;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* Unit tests for {@link HoodieFlinkRecord}.
@@ -266,4 +279,107 @@ public class TestHoodieFlinkRecord {
HoodieFlinkRecord updatedRecord = (HoodieFlinkRecord)
record.updateMetaField(schema, 0, "20240101000001");
assertEquals(HoodieOperation.INSERT, updatedRecord.getOperation());
}
+
+ @Test
+ public void testRecordContract() {
+ 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());
+
+ HoodieFlinkRecord empty = new HoodieFlinkRecord(
+ key, HoodieOperation.INSERT, (RowData) null);
+ assertTrue(empty.checkIsDelete(null, new Properties()));
+ }
+
+ @Test
+ public void testConvertColumnValueForLogicalType() {
+ TimestampData timestamp = TimestampData.fromInstant(
+ Instant.parse("2025-02-03T04:05:06.123456Z"));
+ HoodieFlinkRecord record = new
HoodieFlinkRecord(GenericRowData.of(timestamp));
+
+ assertNull(record.convertColumnValueForLogicalType(
+ HoodieSchema.create(HoodieSchemaType.STRING), null, true));
+ assertEquals(LocalDate.ofEpochDay(2),
record.convertColumnValueForLogicalType(
+ HoodieSchema.createDate(), 2, true));
+
+ HoodieSchema millisSchema = HoodieSchema.createTimestampMillis();
+ HoodieSchema millisRecordSchema = HoodieSchema.createRecord(
+ "millis_record", null, null,
+ Arrays.asList(HoodieSchemaField.of("event_time", millisSchema)));
+ Object millisValue = record.getColumnValueAsJava(
+ millisRecordSchema, "event_time", new Properties());
+ assertEquals(timestamp.getMillisecond(), millisValue);
+ assertEquals(timestamp.getMillisecond(),
record.convertColumnValueForLogicalType(
+ millisSchema, millisValue, true));
+
+ HoodieSchema microsSchema = HoodieSchema.createTimestampMicros();
+ HoodieSchema microsRecordSchema = HoodieSchema.createRecord(
+ "micros_record", null, null,
+ Arrays.asList(HoodieSchemaField.of("event_time", microsSchema)));
+ Object microsValue = record.getColumnValueAsJava(
+ microsRecordSchema, "event_time", new Properties());
+ long expectedMicros = timestamp.toInstant().getEpochSecond() * 1_000_000
+ + timestamp.toInstant().getNano() / 1_000;
+ assertEquals(expectedMicros, microsValue);
+ assertEquals(timestamp.getMillisecond(),
record.convertColumnValueForLogicalType(
+ microsSchema, microsValue, 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(millisValue, record.convertColumnValueForLogicalType(
+ millisSchema, millisValue, false));
+ }
+
+ @Test
+ public void testUnsupportedOperations() {
+ 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);
+ 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));
+ }
+
+ @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"));
+ // The second lookup exercises the cached record-key path.
+ assertEquals("id-001", record.getRecordKey(schema, "id"));
+ assertEquals("id-001", record.toIndexedRecord(schema, new Properties())
+ .get().getData().get(0).toString());
+ assertTrue(record.getAvroBytes(schema, new Properties()).size() > 0);
+ assertEquals(OrderingValues.getDefault(),
+ record.getOrderingValueAsJava(schema, new Properties(), null));
+ }
}
diff --git
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/parquet/TestParquetRowDataWriter.java
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/parquet/TestParquetRowDataWriter.java
new file mode 100644
index 000000000000..7d4c685c7a5f
--- /dev/null
+++
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/parquet/TestParquetRowDataWriter.java
@@ -0,0 +1,176 @@
+/*
+ * 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.ArgumentMatchers.eq;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+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());
+ verify(consumer, atLeastOnce()).startGroup();
+ verify(consumer, atLeastOnce()).endGroup();
+ verify(consumer, never()).startField(eq("null_field"), anyInt());
+ }
+
+ @Test
+ void testNonUtcTimestampWriter() {
+ RowType rowType = (RowType) DataTypes.ROW(
+ DataTypes.FIELD("timestamp3", DataTypes.TIMESTAMP(3)),
+ DataTypes.FIELD("timestamp6",
DataTypes.TIMESTAMP(6))).notNull().getLogicalType();
+ HoodieSchema schema = HoodieSchemaConverter.convertToSchema(rowType,
"timestamps");
+ GenericRowData row = GenericRowData.of(
+ TimestampData.fromInstant(Instant.parse("2025-02-03T04:05:06.123Z")),
+
TimestampData.fromInstant(Instant.parse("2025-02-03T04:05:06.123456Z")));
+ RecordConsumer consumer = mock(RecordConsumer.class);
+
+ new ParquetRowDataWriter(consumer, false, schema).write(row);
+
+ verify(consumer, times(2)).addLong(anyLong());
+ }
+
+ @Test
+ void testArrayElementWritersForPrimitiveAndComplexTypes() {
+ RowType rowType = (RowType) DataTypes.ROW(
+ DataTypes.FIELD("booleans", DataTypes.ARRAY(DataTypes.BOOLEAN())),
+ DataTypes.FIELD("longs", DataTypes.ARRAY(DataTypes.BIGINT())),
+ DataTypes.FIELD("floats", DataTypes.ARRAY(DataTypes.FLOAT())),
+ DataTypes.FIELD("doubles", DataTypes.ARRAY(DataTypes.DOUBLE())),
+ DataTypes.FIELD("binaries", DataTypes.ARRAY(DataTypes.BYTES())),
+ DataTypes.FIELD("small_decimals",
DataTypes.ARRAY(DataTypes.DECIMAL(10, 2))),
+ DataTypes.FIELD("large_decimals",
DataTypes.ARRAY(DataTypes.DECIMAL(30, 4))),
+ DataTypes.FIELD("timestamps", DataTypes.ARRAY(DataTypes.TIMESTAMP(6))),
+ DataTypes.FIELD("nested_arrays",
DataTypes.ARRAY(DataTypes.ARRAY(DataTypes.INT()))),
+ DataTypes.FIELD("maps", DataTypes.ARRAY(
+ DataTypes.MAP(DataTypes.STRING(), DataTypes.INT()))),
+ DataTypes.FIELD("rows", DataTypes.ARRAY(DataTypes.ROW(
+ DataTypes.FIELD("id", DataTypes.BIGINT()),
+ DataTypes.FIELD("name",
DataTypes.STRING()))))).notNull().getLogicalType();
+ HoodieSchema schema = HoodieSchemaConverter.convertToSchema(rowType,
"array_elements");
+ Map<Object, Object> map = new LinkedHashMap<>();
+ map.put(StringData.fromString("key"), 1);
+ GenericRowData row = GenericRowData.of(
+ new GenericArrayData(new boolean[] {true, false}),
+ new GenericArrayData(new long[] {1L, 2L}),
+ new GenericArrayData(new float[] {1.5f, 2.5f}),
+ new GenericArrayData(new double[] {3.5d, 4.5d}),
+ new GenericArrayData(new Object[] {new byte[] {1}, new byte[] {2}}),
+ new GenericArrayData(new Object[] {
+ DecimalData.fromBigDecimal(new BigDecimal("12.34"), 10, 2)}),
+ new GenericArrayData(new Object[] {
+ DecimalData.fromBigDecimal(new
BigDecimal("12345678901234567890.1234"), 30, 4)}),
+ new GenericArrayData(new Object[] {
+
TimestampData.fromInstant(Instant.parse("2025-02-03T04:05:06.123456Z"))}),
+ new GenericArrayData(new Object[] {new GenericArrayData(new int[] {1,
2})}),
+ new GenericArrayData(new Object[] {new GenericMapData(map)}),
+ new GenericArrayData(new Object[] {
+ GenericRowData.of(1L, StringData.fromString("nested"))}));
+ RecordConsumer consumer = mock(RecordConsumer.class);
+
+ new ParquetRowDataWriter(consumer, true, schema).write(row);
+
+ verify(consumer, atLeastOnce()).addBoolean(true);
+ verify(consumer, atLeastOnce()).addLong(1L);
+ verify(consumer, atLeastOnce()).addFloat(1.5f);
+ verify(consumer, atLeastOnce()).addDouble(3.5d);
+ verify(consumer, atLeastOnce()).addBinary(any());
+ }
+}
diff --git
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/parquet/TestParquetSchemaConverter.java
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/parquet/TestParquetSchemaConverter.java
index 23c2d6dac56e..962af8f6e04f 100644
---
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/parquet/TestParquetSchemaConverter.java
+++
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/parquet/TestParquetSchemaConverter.java
@@ -287,6 +287,69 @@ public class TestParquetSchemaConverter {
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);
+ assertEquals(LogicalTypeAnnotation.timestampType(
+ false, LogicalTypeAnnotation.TimeUnit.MILLIS),
+ parquet.getType("local_millis").getLogicalTypeAnnotation());
+ assertEquals(LogicalTypeAnnotation.timestampType(
+ false, LogicalTypeAnnotation.TimeUnit.MICROS),
+ parquet.getType("local_micros").getLogicalTypeAnnotation());
+ }
+
+ @Test
+ void testConvertBinaryDateAndTimeToParquet() {
+ RowType rowType = (RowType) DataTypes.ROW(
+ DataTypes.FIELD("payload", DataTypes.BYTES()),
+ DataTypes.FIELD("day", DataTypes.DATE()),
+ DataTypes.FIELD("time", DataTypes.TIME(3))).notNull().getLogicalType();
+ MessageType parquet =
ParquetSchemaConverter.convertToParquetMessageType("primitives", rowType);
+ assertEquals(PrimitiveType.PrimitiveTypeName.BINARY,
+ parquet.getType("payload").asPrimitiveType().getPrimitiveTypeName());
+ assertEquals(LogicalTypeAnnotation.dateType(),
+ parquet.getType("day").getLogicalTypeAnnotation());
+ assertEquals(LogicalTypeAnnotation.timeType(true,
LogicalTypeAnnotation.TimeUnit.MILLIS),
+ parquet.getType("time").getLogicalTypeAnnotation());
+ }
+
/**
* A Parquet group with metadata + value binary fields but NO VARIANT
annotation must be
* treated as a plain ROW. Only the Parquet {@code VARIANT} annotation
triggers variant
diff --git
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestRowDataAvroRoundTrip.java
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestRowDataAvroRoundTrip.java
new file mode 100644
index 000000000000..b9e53c985151
--- /dev/null
+++
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestRowDataAvroRoundTrip.java
@@ -0,0 +1,212 @@
+/*
+ * 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.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public 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
+ public 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);
+ RowData defaultOutput = (RowData) AvroToRowDataConverters
+ .createRowConverter(ROW_TYPE).convert(avro);
+ RowData nonUtcOutput = (RowData) AvroToRowDataConverters
+ .createRowConverter(ROW_TYPE, false).convert(avro);
+ RowData schemaOutput = (RowData) 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));
+ assertEquals("alice", defaultOutput.getString(9).toString());
+ assertEquals("alice", nonUtcOutput.getString(9).toString());
+ assertEquals("alice", schemaOutput.getString(9).toString());
+ }
+
+ @Test
+ public void testPrimitiveRuntimeRepresentations() {
+ 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());
+ assertEquals(7,
RowDataToAvroConverters.createConverter(DataTypes.TINYINT().getLogicalType())
+ .convert(intSchema, (byte) 7));
+ assertEquals(8,
RowDataToAvroConverters.createConverter(DataTypes.SMALLINT().getLogicalType())
+ .convert(intSchema, (short) 8));
+
assertNull(RowDataToAvroConverters.createConverter(DataTypes.NULL().getLogicalType())
+ .convert(HoodieSchema.create(HoodieSchemaType.NULL), "ignored"));
+ assertEquals(ByteBuffer.wrap(new byte[] {4, 5}),
RowDataToAvroConverters.createConverter(
+ DataTypes.BYTES().getLogicalType()).convert(
+
HoodieSchemaConverter.convertToSchema(DataTypes.BYTES().getLogicalType()), new
byte[] {4, 5}));
+ HoodieSchema.Vector vectorSchema = HoodieSchema.createVector(2);
+ Object vector = RowDataToAvroConverters.createConverter(
+ DataTypes.ARRAY(DataTypes.FLOAT()).getLogicalType()).convert(
+ vectorSchema, new GenericArrayData(new float[] {1.0f, 2.0f}));
+ assertTrue(vector instanceof org.apache.avro.generic.GenericData.Fixed);
+ }
+
+ @Test
+ public void testTimestampAndJodaRuntimeRepresentations() {
+ 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());
+ TimestampData nonUtcTimestamp = (TimestampData)
AvroToRowDataConverters.createConverter(
+ DataTypes.TIMESTAMP(3).getLogicalType(),
false).convert(instant.toEpochMilli());
+ assertEquals(instant.toEpochMilli(),
nonUtcTimestamp.toTimestamp().getTime());
+ assertEquals((int) LocalDate.of(2025, 2, 3).toEpochDay(),
+
AvroToRowDataConverters.createConverter(DataTypes.DATE().getLogicalType(), true)
+ .convert(new org.joda.time.LocalDate(2025, 2, 3)));
+ assertEquals(14_706_123,
+
AvroToRowDataConverters.createConverter(DataTypes.TIME(3).getLogicalType(),
true)
+ .convert(new org.joda.time.LocalTime(4, 5, 6, 123)));
+ assertSame(AvroToRowDataConverters.JodaConverter.getConverter(),
+ AvroToRowDataConverters.JodaConverter.getConverter());
+ assertThrows(IllegalArgumentException.class,
+ () ->
AvroToRowDataConverters.createConverter(DataTypes.TIMESTAMP(9).getLogicalType(),
true));
+ }
+
+ private static void assertMap(MapData map) {
+ ArrayData keys = map.keyArray();
+ ArrayData values = map.valueArray();
+ Map<String, Integer> actual = new LinkedHashMap<>();
+ for (int i = 0; i < map.size(); i++) {
+ actual.put(keys.getString(i).toString(), values.isNullAt(i) ? null :
values.getInt(i));
+ }
+ Map<String, Integer> expected = new LinkedHashMap<>();
+ expected.put("one", 1);
+ expected.put("missing", null);
+ assertEquals(expected, actual);
+ }
+}
diff --git
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestRowDataUtils.java
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestRowDataUtils.java
new file mode 100644
index 000000000000..0249e74028cf
--- /dev/null
+++
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestRowDataUtils.java
@@ -0,0 +1,127 @@
+/*
+ * 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.flink.table.api.DataTypes;
+import org.apache.flink.table.data.DecimalData;
+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.LocalZonedTimestampType;
+import org.apache.flink.table.types.logical.TimestampType;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.time.LocalDate;
+
+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.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class TestRowDataUtils {
+
+ @Test
+ void testJavaAndFlinkValueConverters() {
+ assertNull(RowDataUtils.NULL_GETTER.getFieldOrNull(GenericRowData.of(1)));
+ assertNull(RowDataUtils.javaValFunc(DataTypes.NULL().getLogicalType(),
true).apply("ignored"));
+ assertEquals(7,
RowDataUtils.javaValFunc(DataTypes.TINYINT().getLogicalType(),
true).apply((byte) 7));
+ assertEquals(8,
RowDataUtils.javaValFunc(DataTypes.SMALLINT().getLogicalType(),
true).apply((short) 8));
+ assertEquals(2,
RowDataUtils.javaValFunc(DataTypes.DATE().getLogicalType(), true).apply(2));
+ assertEquals("text",
RowDataUtils.javaValFunc(DataTypes.STRING().getLogicalType(), true)
+ .apply(StringData.fromString("text")));
+ assertArrayEquals(new byte[] {1, 2}, ((ByteBuffer)
RowDataUtils.javaValFunc(
+ DataTypes.BYTES().getLogicalType(), true).apply(new byte[] {1,
2})).array());
+ assertEquals(new BigDecimal("12.30"), RowDataUtils.javaValFunc(
+ DataTypes.DECIMAL(8, 2).getLogicalType(), true)
+ .apply(DecimalData.fromBigDecimal(new BigDecimal("12.30"), 8, 2)));
+
+ assertEquals((byte) 7,
RowDataUtils.flinkValFunc(DataTypes.TINYINT().getLogicalType(),
true).apply((byte) 7));
+ assertEquals((short) 8,
RowDataUtils.flinkValFunc(DataTypes.SMALLINT().getLogicalType(),
true).apply((short) 8));
+ assertEquals(3,
RowDataUtils.flinkValFunc(DataTypes.DATE().getLogicalType(), true)
+ .apply(LocalDate.ofEpochDay(3)));
+ assertEquals("text",
RowDataUtils.flinkValFunc(DataTypes.STRING().getLogicalType(), true)
+ .apply("text").toString());
+ ByteBuffer buffer = ByteBuffer.wrap(new byte[] {3, 4});
+ assertSame(buffer,
RowDataUtils.flinkValFunc(DataTypes.BYTES().getLogicalType(),
true).apply(buffer));
+ assertEquals(new BigDecimal("45.60"), ((DecimalData)
RowDataUtils.flinkValFunc(
+ DataTypes.DECIMAL(8, 2).getLogicalType(), true).apply(new
BigDecimal("45.60"))).toBigDecimal());
+ }
+
+ @Test
+ void testTimestampConvertersAtMillisAndMicros() {
+ TimestampData timestamp =
TimestampData.fromInstant(Instant.parse("2025-02-03T04:05:06.123456Z"));
+ long millis = timestamp.toInstant().toEpochMilli();
+ long micros = timestamp.toInstant().getEpochSecond() * 1_000_000 + 123456;
+
+ assertEquals(millis,
RowDataUtils.javaValFunc(DataTypes.TIMESTAMP_LTZ(3).getLogicalType(), true)
+ .apply(timestamp));
+ assertEquals(micros,
RowDataUtils.javaValFunc(DataTypes.TIMESTAMP_LTZ(6).getLogicalType(), true)
+ .apply(timestamp));
+ assertEquals(millis,
RowDataUtils.javaValFunc(DataTypes.TIMESTAMP(3).getLogicalType(), true)
+ .apply(timestamp));
+ assertEquals(micros,
RowDataUtils.javaValFunc(DataTypes.TIMESTAMP(6).getLogicalType(), true)
+ .apply(timestamp));
+ long localMillis = (long) RowDataUtils.javaValFunc(
+ DataTypes.TIMESTAMP(3).getLogicalType(), false).apply(timestamp);
+ long localMicros = (long) RowDataUtils.javaValFunc(
+ DataTypes.TIMESTAMP(6).getLogicalType(), false).apply(timestamp);
+
+ assertEquals(Instant.ofEpochMilli(millis), ((TimestampData)
RowDataUtils.flinkValFunc(
+ DataTypes.TIMESTAMP_LTZ(3).getLogicalType(),
true).apply(millis)).toInstant());
+ assertEquals(timestamp.toInstant(), ((TimestampData)
RowDataUtils.flinkValFunc(
+ DataTypes.TIMESTAMP_LTZ(6).getLogicalType(),
true).apply(micros)).toInstant());
+ assertEquals(millis, ((TimestampData) RowDataUtils.flinkValFunc(
+ DataTypes.TIMESTAMP(3).getLogicalType(),
false).apply(millis)).toTimestamp().getTime());
+ assertEquals(timestamp.toInstant(), ((TimestampData)
RowDataUtils.flinkValFunc(
+ DataTypes.TIMESTAMP(6).getLogicalType(),
true).apply(micros)).toInstant());
+ assertEquals(localMillis,
RowDataUtils.javaValFunc(DataTypes.TIMESTAMP(3).getLogicalType(), false)
+
.apply(RowDataUtils.flinkValFunc(DataTypes.TIMESTAMP(3).getLogicalType(),
false).apply(localMillis)));
+ assertEquals(localMicros,
RowDataUtils.javaValFunc(DataTypes.TIMESTAMP(6).getLogicalType(), false)
+
.apply(RowDataUtils.flinkValFunc(DataTypes.TIMESTAMP(6).getLogicalType(),
false).apply(localMicros)));
+
+ assertThrows(UnsupportedOperationException.class,
+ () ->
RowDataUtils.javaValFunc(DataTypes.TIMESTAMP(9).getLogicalType(), true));
+ assertThrows(UnsupportedOperationException.class,
+ () ->
RowDataUtils.flinkValFunc(DataTypes.TIMESTAMP_LTZ(9).getLogicalType(), true));
+ }
+
+ @Test
+ void testGenericConversionAndPrecision() {
+ assertNull(RowDataUtils.convertValueToFlinkType(null));
+ assertEquals("value",
RowDataUtils.convertValueToFlinkType("value").toString());
+ assertEquals(new BigDecimal("1.20"), ((DecimalData)
RowDataUtils.convertValueToFlinkType(
+ new BigDecimal("1.20"))).toBigDecimal());
+ Timestamp timestamp = Timestamp.valueOf("2025-02-03 04:05:06.123456");
+ assertEquals(timestamp, ((TimestampData)
RowDataUtils.convertValueToFlinkType(timestamp)).toTimestamp());
+ assertEquals(2,
RowDataUtils.convertValueToFlinkType(LocalDate.ofEpochDay(2)));
+ assertArrayEquals(new byte[] {9, 8},
+ (byte[]) RowDataUtils.convertValueToFlinkType(ByteBuffer.wrap(new
byte[] {9, 8})));
+ Object marker = new Object();
+ assertSame(marker, RowDataUtils.convertValueToFlinkType(marker));
+ assertEquals(3, RowDataUtils.precision(new TimestampType(3)));
+ assertEquals(6, RowDataUtils.precision(new LocalZonedTimestampType(6)));
+ assertThrows(AssertionError.class,
+ () -> RowDataUtils.precision(DataTypes.INT().getLogicalType()));
+ }
+}
diff --git
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestVectorConversionUtils.java
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestVectorConversionUtils.java
new file mode 100644
index 000000000000..e5a2e8478fd2
--- /dev/null
+++
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestVectorConversionUtils.java
@@ -0,0 +1,156 @@
+/*
+ * 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.HoodieSchemaField;
+import org.apache.hudi.common.schema.HoodieSchemaType;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.exception.SchemaCompatibilityException;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.types.RowKind;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+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.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestVectorConversionUtils {
+
+ private static final HoodieSchema.Vector FLOAT_VECTOR =
HoodieSchema.createVector(2);
+ private static final HoodieSchema.Vector DOUBLE_VECTOR =
HoodieSchema.createVector(
+ 2, HoodieSchema.Vector.VectorElementType.DOUBLE);
+ private static final HoodieSchema.Vector INT8_VECTOR =
HoodieSchema.createVector(
+ 3, HoodieSchema.Vector.VectorElementType.INT8);
+
+ @Test
+ void testDetectAndRewriteVectorColumns() {
+ HoodieSchema schema = vectorRecordSchema();
+ String[] names = {"id", "Float_Vec", "double_vec", "codes"};
+ int[] selected = {3, 0, 1};
+ Map<Integer, HoodieSchema.Vector> detected =
+ VectorConversionUtils.detectVectorColumns(names, selected, schema);
+ assertEquals(INT8_VECTOR, detected.get(0));
+ assertEquals(FLOAT_VECTOR, detected.get(2));
+ assertFalse(detected.containsKey(1));
+
+ DataType requested = DataTypes.ROW(
+ DataTypes.FIELD("codes",
DataTypes.ARRAY(DataTypes.TINYINT()).notNull()),
+ DataTypes.FIELD("id", DataTypes.INT().notNull()),
+ DataTypes.FIELD("float_vec",
DataTypes.ARRAY(DataTypes.FLOAT()))).notNull();
+ HoodieSchema projectedSchema = HoodieSchema.createRecord("projected",
null, null, Arrays.asList(
+ HoodieSchemaField.of("codes", INT8_VECTOR),
+ HoodieSchemaField.of("id", HoodieSchema.create(HoodieSchemaType.INT)),
+ HoodieSchemaField.of("float_vec", FLOAT_VECTOR)));
+ DataType physical = VectorConversionUtils.getParquetReadDataType(
+ requested, projectedSchema, detected);
+ RowType physicalRow = (RowType) physical.getLogicalType();
+ assertEquals(LogicalTypeRoot.VARBINARY,
physicalRow.getTypeAt(0).getTypeRoot());
+ assertEquals(LogicalTypeRoot.INTEGER,
physicalRow.getTypeAt(1).getTypeRoot());
+ assertEquals(LogicalTypeRoot.VARBINARY,
physicalRow.getTypeAt(2).getTypeRoot());
+ assertFalse(physicalRow.getTypeAt(0).isNullable());
+ assertSame(requested, VectorConversionUtils.getParquetReadDataType(
+ requested, projectedSchema, Collections.emptyMap()));
+
+ DataType[] fieldTypes = {
+ DataTypes.INT(), DataTypes.ARRAY(DataTypes.FLOAT()),
+ DataTypes.ARRAY(DataTypes.DOUBLE()),
DataTypes.ARRAY(DataTypes.TINYINT())};
+ DataType[] physicalTypes =
VectorConversionUtils.getParquetReadFieldTypes(names, fieldTypes, schema);
+ assertEquals(LogicalTypeRoot.INTEGER,
physicalTypes[0].getLogicalType().getTypeRoot());
+ assertEquals(LogicalTypeRoot.VARBINARY,
physicalTypes[1].getLogicalType().getTypeRoot());
+ assertEquals(LogicalTypeRoot.VARBINARY,
physicalTypes[2].getLogicalType().getTypeRoot());
+ assertEquals(LogicalTypeRoot.VARBINARY,
physicalTypes[3].getLogicalType().getTypeRoot());
+ }
+
+ @Test
+ void testEncodeDecodeAndIteratorConversion() throws Exception {
+ byte[] floatBytes = VectorConversionUtils.encodeVectorArrayData(
+ new GenericArrayData(new float[] {1.25f, -2.5f}), FLOAT_VECTOR);
+ byte[] doubleBytes = VectorConversionUtils.encodeVectorArrayData(
+ new GenericArrayData(new double[] {3.5d, -4.75d}), DOUBLE_VECTOR);
+ byte[] int8Bytes = VectorConversionUtils.encodeVectorArrayData(
+ new GenericArrayData(new byte[] {1, -2, 3}), INT8_VECTOR);
+
+ ArrayData floats = VectorConversionUtils.createVectorArrayData(floatBytes,
FLOAT_VECTOR);
+ assertEquals(1.25f, floats.getFloat(0));
+ assertEquals(-2.5f, floats.getFloat(1));
+ ArrayData doubles =
VectorConversionUtils.createVectorArrayData(doubleBytes, DOUBLE_VECTOR);
+ assertEquals(3.5d, doubles.getDouble(0));
+ assertEquals(-4.75d, doubles.getDouble(1));
+ ArrayData codes = VectorConversionUtils.createVectorArrayData(int8Bytes,
INT8_VECTOR);
+ assertEquals((byte) -2, codes.getByte(1));
+ assertArrayEquals(int8Bytes,
VectorConversionUtils.encodeVectorArrayData(codes, INT8_VECTOR));
+ assertThrows(IllegalArgumentException.class, () ->
VectorConversionUtils.encodeVectorArrayData(
+ new GenericArrayData(new float[] {1.0f}), FLOAT_VECTOR));
+
+ GenericRowData physical = GenericRowData.of(
+ StringData.fromString("key"), floatBytes, null);
+ physical.setRowKind(RowKind.UPDATE_AFTER);
+ RowType rowType = (RowType) DataTypes.ROW(
+ DataTypes.FIELD("id", DataTypes.STRING()),
+ DataTypes.FIELD("vector", DataTypes.BYTES()),
+ DataTypes.FIELD("optional", DataTypes.INT())).getLogicalType();
+ ClosableIterator<RowData> wrapped =
VectorConversionUtils.wrapVectorColumnIterator(
+
ClosableIterator.wrap(Collections.<RowData>singletonList(physical).iterator()),
+ rowType, Collections.singletonMap(1, FLOAT_VECTOR));
+ RowData converted = wrapped.next();
+ assertEquals(RowKind.UPDATE_AFTER, converted.getRowKind());
+ assertEquals("key", converted.getString(0).toString());
+ assertEquals(1.25f, converted.getArray(1).getFloat(0));
+ assertTrue(converted.isNullAt(2));
+ wrapped.close();
+ }
+
+ @Test
+ void testVectorLogicalTypeValidation() {
+ VectorConversionUtils.validateVectorLogicalType(
+ FLOAT_VECTOR, DataTypes.ARRAY(DataTypes.FLOAT()).getLogicalType());
+ VectorConversionUtils.validateVectorLogicalType(
+ DOUBLE_VECTOR, DataTypes.ARRAY(DataTypes.DOUBLE()).getLogicalType());
+ VectorConversionUtils.validateVectorLogicalType(
+ INT8_VECTOR, DataTypes.ARRAY(DataTypes.TINYINT()).getLogicalType());
+ assertThrows(SchemaCompatibilityException.class,
+ () -> VectorConversionUtils.validateVectorLogicalType(
+ FLOAT_VECTOR,
DataTypes.ARRAY(DataTypes.DOUBLE()).getLogicalType()));
+ }
+
+ private static HoodieSchema vectorRecordSchema() {
+ return HoodieSchema.createRecord("vectors", null, null, Arrays.asList(
+ HoodieSchemaField.of("id", HoodieSchema.create(HoodieSchemaType.INT)),
+ HoodieSchemaField.of("float_vec", FLOAT_VECTOR),
+ HoodieSchemaField.of("double_vec", DOUBLE_VECTOR),
+ HoodieSchemaField.of("codes", INT8_VECTOR)));
+ }
+}
