This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new b38601a074 [Fix][Connector-V2] Fix Parquet INT96 mixed-case field
matching (#11067)
b38601a074 is described below
commit b38601a074494c53006ca14cd7b353873e4052a9
Author: Jast <[email protected]>
AuthorDate: Thu Jun 18 08:43:08 2026 +0800
[Fix][Connector-V2] Fix Parquet INT96 mixed-case field matching (#11067)
---
.../file/sink/writer/ParquetWriteStrategy.java | 32 ++++++++++------
.../file/writer/ParquetWriteStrategyTest.java | 44 ++++++++++++++++++++++
2 files changed, 65 insertions(+), 11 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/ParquetWriteStrategy.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/ParquetWriteStrategy.java
index d8ed92d5fb..8537fc688b 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/ParquetWriteStrategy.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/ParquetWriteStrategy.java
@@ -65,9 +65,9 @@ import java.time.temporal.JulianFields;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
-import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Locale;
import java.util.Set;
import java.util.TimeZone;
import java.util.concurrent.TimeUnit;
@@ -99,12 +99,15 @@ public class ParquetWriteStrategy extends
AbstractWriteStrategy<ParquetWriter<Ge
public void init(HadoopConf conf, String jobId, String uuidPrefix, int
subTaskIndex) {
super.init(conf, jobId, uuidPrefix, subTaskIndex);
Configuration configuration = getConfiguration(hadoopConf);
- writePathsAsInt96 = new
HashSet<>(fileSinkConfig.getParquetAvroWriteFixedAsInt96());
+ writePathsAsInt96 =
+ fileSinkConfig.getParquetAvroWriteFixedAsInt96().stream()
+ .map(this::normalizeFieldName)
+ .collect(Collectors.toSet());
if (fileSinkConfig.getParquetWriteTimestampAsInt96()) {
List<String> timestampFields = new ArrayList<>();
for (int i = 0; i < seaTunnelRowType.getTotalFields(); i++) {
if
(SqlType.TIMESTAMP.equals(seaTunnelRowType.getFieldType(i).getSqlType())) {
- timestampFields.add(seaTunnelRowType.getFieldName(i));
+
timestampFields.add(normalizeFieldName(seaTunnelRowType.getFieldName(i)));
}
}
writePathsAsInt96.addAll(timestampFields);
@@ -244,7 +247,7 @@ public class ParquetWriteStrategy extends
AbstractWriteStrategy<ParquetWriter<Ge
case DATE:
return data;
case TIMESTAMP:
- if (writePathsAsInt96.contains(name)) {
+ if (writePathsAsInt96.contains(normalizeFieldName(name))) {
LocalDateTime localDateTime = (LocalDateTime) data;
Calendar calendar =
Calendar.getInstance(TimeZone.getTimeZone("UTC"));
calendar.setTime(
@@ -264,7 +267,7 @@ public class ParquetWriteStrategy extends
AbstractWriteStrategy<ParquetWriter<Ge
calendar.get(Calendar.MILLISECOND));
NanoTime nanoTime = new NanoTime(julianDays,
timeOfDayNanos);
return new GenericData.Fixed(
- schema.getField(name.toLowerCase()).schema(),
+ schema.getField(normalizeFieldName(name)).schema(),
nanoTime.toBinary().getBytes());
}
return ((LocalDateTime) data)
@@ -272,9 +275,9 @@ public class ParquetWriteStrategy extends
AbstractWriteStrategy<ParquetWriter<Ge
.toInstant()
.toEpochMilli();
case BYTES:
- if (writePathsAsInt96.contains(name)) {
+ if (writePathsAsInt96.contains(normalizeFieldName(name))) {
return new GenericData.Fixed(
- schema.getField(name.toLowerCase()).schema(),
(byte[]) data);
+
schema.getField(normalizeFieldName(name)).schema(), (byte[]) data);
}
return ByteBuffer.wrap((byte[]) data);
case ROW:
@@ -365,7 +368,7 @@ public class ParquetWriteStrategy extends
AbstractWriteStrategy<ParquetWriter<Ge
PrimitiveType.PrimitiveTypeName.INT64,
Type.Repetition.OPTIONAL)
.named(fieldName);
case TIMESTAMP:
- if (writePathsAsInt96.contains(fieldName)) {
+ if (writePathsAsInt96.contains(normalizeFieldName(fieldName)))
{
return Types.primitive(
PrimitiveType.PrimitiveTypeName.INT96,
Type.Repetition.OPTIONAL)
.named(fieldName);
@@ -392,7 +395,7 @@ public class ParquetWriteStrategy extends
AbstractWriteStrategy<ParquetWriter<Ge
.scale(scale)
.named(fieldName);
case BYTES:
- if (writePathsAsInt96.contains(fieldName)) {
+ if (writePathsAsInt96.contains(normalizeFieldName(fieldName)))
{
return Types.primitive(
PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY,
Type.Repetition.OPTIONAL)
@@ -430,10 +433,13 @@ public class ParquetWriteStrategy extends
AbstractWriteStrategy<ParquetWriter<Ge
// are included. Without this, new TIMESTAMP columns get INT64
encoding instead of INT96
// when parquetWriteTimestampAsInt96=true — silent data corruption.
if (fileSinkConfig.getParquetWriteTimestampAsInt96()) {
- writePathsAsInt96 = new
HashSet<>(fileSinkConfig.getParquetAvroWriteFixedAsInt96());
+ writePathsAsInt96 =
+ fileSinkConfig.getParquetAvroWriteFixedAsInt96().stream()
+ .map(this::normalizeFieldName)
+ .collect(Collectors.toSet());
for (int i = 0; i < seaTunnelRowType.getTotalFields(); i++) {
if
(SqlType.TIMESTAMP.equals(seaTunnelRowType.getFieldType(i).getSqlType())) {
- writePathsAsInt96.add(seaTunnelRowType.getFieldName(i));
+
writePathsAsInt96.add(normalizeFieldName(seaTunnelRowType.getFieldName(i)));
}
}
}
@@ -457,4 +463,8 @@ public class ParquetWriteStrategy extends
AbstractWriteStrategy<ParquetWriter<Ge
Types.buildMessage().addFields(types.toArray(new
Type[0])).named("SeaTunnelRecord");
return schemaConverter.convert(seaTunnelRow);
}
+
+ private String normalizeFieldName(String fieldName) {
+ return fieldName.toLowerCase(Locale.ROOT);
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/writer/ParquetWriteStrategyTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/writer/ParquetWriteStrategyTest.java
index f24def6288..eea84414c9 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/writer/ParquetWriteStrategyTest.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/writer/ParquetWriteStrategyTest.java
@@ -149,4 +149,48 @@ public class ParquetWriteStrategyTest {
Assertions.assertEquals(1, readRows.size());
readStrategy.close();
}
+
+ @DisabledOnOs(OS.WINDOWS)
+ @Test
+ public void testParquetWriteInt96WithMixedCaseTimestampColumn() throws
Exception {
+ Map<String, Object> writeConfig = new HashMap<>();
+ writeConfig.put("tmp_path", TMP_PATH + "-mixed-case");
+ writeConfig.put("path",
"file:///tmp/seatunnel/parquet/int96-mixed-case");
+ writeConfig.put("file_format_type", FileFormat.PARQUET.name());
+ writeConfig.put("parquet_avro_write_timestamp_as_int96", "true");
+
+ SeaTunnelRowType writeRowType =
+ new SeaTunnelRowType(
+ new String[] {"createTime"},
+ new SeaTunnelDataType[]
{LocalTimeType.LOCAL_DATE_TIME_TYPE});
+ FileSinkConfig writeSinkConfig =
+ new FileSinkConfig(ReadonlyConfig.fromMap(writeConfig),
writeRowType);
+ ParquetWriteStrategy writeStrategy = new
ParquetWriteStrategy(writeSinkConfig);
+ LocalFileSystemConf.LocalConf hadoopConf =
+ new LocalFileSystemConf.LocalConf(FS_DEFAULT_NAME_DEFAULT);
+ writeStrategy.setCatalogTable(
+ CatalogTableUtil.getCatalogTable("test", null, null, "test",
writeRowType));
+ writeStrategy.init(hadoopConf, "test1", "test1", 0);
+ writeStrategy.beginTransaction(1L);
+ writeStrategy.write(new SeaTunnelRow(new Object[]
{LocalDateTime.now()}));
+ writeStrategy.finishAndCloseFile();
+ writeStrategy.close();
+
+ ParquetReadStrategy readStrategy = new ParquetReadStrategy();
+ readStrategy.init(hadoopConf);
+ List<String> readFiles = readStrategy.getFileNamesByPath(TMP_PATH +
"-mixed-case");
+ Assertions.assertEquals(1, readFiles.size());
+ try (ParquetFileReader reader =
+ ParquetFileReader.open(
+ HadoopInputFile.fromPath(
+ new
org.apache.hadoop.fs.Path(readFiles.get(0)),
+ new Configuration()))) {
+ FileMetaData metadata = reader.getFooter().getFileMetaData();
+ Type createTimeType = metadata.getSchema().getType("createtime");
+ Assertions.assertEquals(
+ PrimitiveType.PrimitiveTypeName.INT96,
+ createTimeType.asPrimitiveType().getPrimitiveTypeName());
+ }
+ readStrategy.close();
+ }
}
