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 8ef362f848 [Fix][Connector-V2] Fix parquet read failure when column
name contains Avro-illegal characters (#10960)
8ef362f848 is described below
commit 8ef362f84882ebc1c195d9b0ca279d5267710eb4
Author: shown <[email protected]>
AuthorDate: Mon Jun 1 10:36:00 2026 +0800
[Fix][Connector-V2] Fix parquet read failure when column name contains
Avro-illegal characters (#10960)
Signed-off-by: yuluo-yx <[email protected]>
---
.../file/source/reader/ParquetReadStrategy.java | 427 +++++++++++++++++++++
.../file/writer/ParquetReadStrategyTest.java | 158 ++++++++
2 files changed, 585 insertions(+)
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/ParquetReadStrategy.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/ParquetReadStrategy.java
index 0b7698b04b..d444dc84f7 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/ParquetReadStrategy.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/ParquetReadStrategy.java
@@ -36,6 +36,7 @@ import
org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorExc
import
org.apache.seatunnel.connectors.seatunnel.file.source.split.FileSourceSplit;
import org.apache.avro.Conversions;
+import org.apache.avro.SchemaParseException;
import org.apache.avro.data.TimeConversions;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
@@ -43,9 +44,11 @@ import org.apache.avro.util.Utf8;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.Path;
import org.apache.parquet.avro.AvroParquetReader;
+import org.apache.parquet.example.data.Group;
import org.apache.parquet.example.data.simple.NanoTime;
import org.apache.parquet.hadoop.ParquetFileReader;
import org.apache.parquet.hadoop.ParquetReader;
+import org.apache.parquet.hadoop.example.GroupReadSupport;
import org.apache.parquet.hadoop.metadata.FileMetaData;
import org.apache.parquet.hadoop.metadata.ParquetMetadata;
import org.apache.parquet.hadoop.util.HadoopInputFile;
@@ -54,16 +57,19 @@ import org.apache.parquet.schema.GroupType;
import org.apache.parquet.schema.LogicalTypeAnnotation;
import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.OriginalType;
+import org.apache.parquet.schema.PrimitiveType;
import org.apache.parquet.schema.Type;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.math.BigDecimal;
+import java.math.BigInteger;
import java.math.RoundingMode;
import java.nio.ByteBuffer;
import java.sql.Timestamp;
import java.time.Instant;
+import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.ArrayList;
@@ -96,6 +102,23 @@ public class ParquetReadStrategy extends
AbstractReadStrategy {
@Override
public void read(FileSourceSplit split, Collector<SeaTunnelRow> output)
throws IOException, FileConnectorException {
+ try {
+ readWithAvro(split, output);
+ } catch (RuntimeException e) {
+ if (!isIllegalAvroFieldNameException(e)) {
+ throw e;
+ }
+ log.warn(
+ "Failed to read parquet file [{}] with Avro reader due to
illegal Avro field"
+ + " name, fallback to native parquet reader",
+ split.getFilePath(),
+ e);
+ readWithNativeParquet(split, output);
+ }
+ }
+
+ private void readWithAvro(FileSourceSplit split, Collector<SeaTunnelRow>
output)
+ throws IOException, FileConnectorException {
String tableId = split.getTableId();
String path = split.getFilePath();
if (Boolean.FALSE.equals(checkFileType(path))) {
@@ -150,6 +173,410 @@ public class ParquetReadStrategy extends
AbstractReadStrategy {
}
}
+ private void readWithNativeParquet(FileSourceSplit split,
Collector<SeaTunnelRow> output)
+ throws IOException {
+ String tableId = split.getTableId();
+ String path = split.getFilePath();
+ Path filePath = new Path(path);
+ Map<String, String> partitionsMap = parsePartitionsByPath(path);
+ int fieldsCount = seaTunnelRowType.getTotalFields();
+ final boolean useSplitRange =
+ enableSplitFile && split.getStart() >= 0 && split.getLength()
> 0;
+ ParquetReader<Group> reader =
+ hadoopFileSystemProxy.doWithHadoopAuth(
+ (configuration, userGroupInformation) -> {
+ ParquetReader.Builder<Group> builder =
+ ParquetReader.builder(new
GroupReadSupport(), filePath)
+ .withConf(configuration);
+ if (useSplitRange) {
+ long start = split.getStart();
+ long end = start + split.getLength();
+ builder.withFileRange(start, end);
+ }
+ return builder.build();
+ });
+ try (ParquetReader<Group> closeableReader = reader) {
+ Group record;
+ while ((record = closeableReader.read()) != null) {
+ Object[] fields;
+ if (isMergePartition) {
+ int index = fieldsCount;
+ fields = new Object[fieldsCount + partitionsMap.size()];
+ for (String value : partitionsMap.values()) {
+ fields[index++] = value;
+ }
+ } else {
+ fields = new Object[fieldsCount];
+ }
+ for (int i = 0; i < fieldsCount; i++) {
+ fields[i] =
+ resolveGroupObject(
+ record,
+ record.getType().getType(indexes[i]),
+ indexes[i],
+ seaTunnelRowType.getFieldType(i));
+ }
+ SeaTunnelRow seaTunnelRow = new SeaTunnelRow(fields);
+ seaTunnelRow.setTableId(tableId);
+ output.collect(seaTunnelRow);
+ }
+ }
+ }
+
+ private boolean isIllegalAvroFieldNameException(Throwable throwable) {
+ while (throwable != null) {
+ String message = throwable.getMessage();
+ if ((throwable instanceof SchemaParseException
+ ||
throwable.getClass().getName().endsWith(".SchemaParseException"))
+ && message != null
+ && message.contains("Illegal character in")) {
+ return true;
+ }
+ throwable = throwable.getCause();
+ }
+ return false;
+ }
+
+ private Object resolveGroupObject(
+ Group group, Type parquetType, int fieldIndex,
SeaTunnelDataType<?> fieldType) {
+ if (group.getFieldRepetitionCount(fieldIndex) == 0) {
+ return null;
+ }
+ if (!parquetType.isPrimitive()) {
+ return resolveGroupType(group, parquetType, fieldIndex, fieldType);
+ }
+ Object field = readPrimitiveGroupObject(group,
parquetType.asPrimitiveType(), fieldIndex);
+ if (field instanceof LocalDateTime || field instanceof LocalDate) {
+ return field;
+ }
+ return resolveObject(field, fieldType);
+ }
+
+ private Object resolveGroupType(
+ Group group, Type parquetType, int fieldIndex,
SeaTunnelDataType<?> fieldType) {
+ GroupType groupType = parquetType.asGroupType();
+ LogicalTypeAnnotation logicalTypeAnnotation =
groupType.getLogicalTypeAnnotation();
+ if (logicalTypeAnnotation == null) {
+ SeaTunnelRowType rowType = (SeaTunnelRowType) fieldType;
+ Group childGroup = group.getGroup(fieldIndex, 0);
+ Object[] objects = new Object[rowType.getTotalFields()];
+ for (int i = 0; i < rowType.getTotalFields(); i++) {
+ objects[i] =
+ resolveGroupObject(
+ childGroup, groupType.getType(i), i,
rowType.getFieldType(i));
+ }
+ return new SeaTunnelRow(objects);
+ }
+ OriginalType originalType = logicalTypeAnnotation.toOriginalType();
+ if (originalType == OriginalType.LIST) {
+ return readList(group, groupType, fieldIndex, fieldType);
+ }
+ if (originalType == OriginalType.MAP) {
+ return readMap(group, groupType, fieldIndex, fieldType);
+ }
+ throw CommonError.convertToSeaTunnelTypeError(
+ PARQUET, parquetType.toString(), parquetType.getName());
+ }
+
+ /**
+ * Reads a LIST field from a Parquet Group using the native Group API.
Handles both the standard
+ * 3-level LIST encoding (group → repeated list → element) and the legacy
2-level encoding
+ * (group → repeated element directly).
+ */
+ private Object readList(
+ Group group,
+ GroupType parentGroupType,
+ int listFieldIndex,
+ SeaTunnelDataType<?> fieldType) {
+ // Check if the LIST field is present
+ if (group.getFieldRepetitionCount(listFieldIndex) == 0) {
+ return new Object[0];
+ }
+
+ ArrayType<?, ?> arrayType = (ArrayType<?, ?>) fieldType;
+ SeaTunnelDataType<?> elementType = arrayType.getElementType();
+ Group listGroup = group.getGroup(listFieldIndex, 0);
+ GroupType listGroupType = parentGroupType;
+
+ Type repeatedType = listGroupType.getType(0);
+ // Number of repeated elements within the LIST group
+ int numElements = listGroup.getFieldRepetitionCount(0);
+
+ // Determine 3-level vs 2-level LIST encoding.
+ // 3-level: LIST group → REPEATED group (1 field) → element
+ // 2-level: LIST group → REPEATED element directly (legacy)
+ boolean isThreeLevel = false;
+ if (!repeatedType.isPrimitive()) {
+ GroupType repeatedGroupType = repeatedType.asGroupType();
+ LogicalTypeAnnotation repeatedAnnotation =
repeatedGroupType.getLogicalTypeAnnotation();
+ if (repeatedAnnotation == null
+ || (repeatedAnnotation.toOriginalType() !=
OriginalType.LIST
+ && repeatedAnnotation.toOriginalType() !=
OriginalType.MAP)) {
+ if (repeatedGroupType.getFieldCount() == 1) {
+ isThreeLevel = true;
+ }
+ }
+ }
+
+ List<Object> result = new ArrayList<>(numElements);
+ if (isThreeLevel) {
+ read3LevelList(
+ listGroup, listGroupType, repeatedType, numElements,
elementType, result);
+ } else {
+ read2LevelList(
+ listGroup, repeatedType, numElements, elementType,
listGroupType, result);
+ }
+ return convertListResult(result, elementType);
+ }
+
+ private Object convertListResult(List<Object> result, SeaTunnelDataType<?>
elementType) {
+ switch (elementType.getSqlType()) {
+ case STRING:
+ return result.toArray(TYPE_ARRAY_STRING);
+ case BOOLEAN:
+ return result.toArray(TYPE_ARRAY_BOOLEAN);
+ case TINYINT:
+ return result.toArray(TYPE_ARRAY_BYTE);
+ case SMALLINT:
+ return result.toArray(TYPE_ARRAY_SHORT);
+ case INT:
+ return result.toArray(TYPE_ARRAY_INTEGER);
+ case BIGINT:
+ return result.toArray(TYPE_ARRAY_LONG);
+ case FLOAT:
+ return result.toArray(TYPE_ARRAY_FLOAT);
+ case DOUBLE:
+ return result.toArray(TYPE_ARRAY_DOUBLE);
+ case DECIMAL:
+ return result.toArray(TYPE_ARRAY_BIG_DECIMAL);
+ case DATE:
+ return result.toArray(TYPE_ARRAY_LOCAL_DATE);
+ case TIMESTAMP:
+ return result.toArray(TYPE_ARRAY_LOCAL_DATETIME);
+ case BYTES:
+ byte[][] bytesArray = new byte[result.size()][];
+ for (int i = 0; i < result.size(); i++) {
+ Object element = result.get(i);
+ if (element instanceof ByteBuffer) {
+ ByteBuffer buffer = (ByteBuffer) element;
+ byte[] bytes = new byte[buffer.remaining()];
+ buffer.get(bytes, 0, bytes.length);
+ bytesArray[i] = bytes;
+ } else if (element instanceof byte[]) {
+ bytesArray[i] = (byte[]) element;
+ }
+ }
+ return bytesArray;
+ default:
+ return result.toArray(new Object[0]);
+ }
+ }
+
+ /**
+ * Reads elements from a 3-level encoded Parquet LIST. Structure:
listGroup → REPEATED wrapper
+ * (field 0) → actual element (field 0 within wrapper).
+ */
+ private void read3LevelList(
+ Group listGroup,
+ GroupType listGroupType,
+ Type repeatedType,
+ int numElements,
+ SeaTunnelDataType<?> elementType,
+ List<Object> result) {
+ Type elementTypeInParquet = repeatedType.asGroupType().getType(0);
+ for (int i = 0; i < numElements; i++) {
+ Group repeatedInstance = listGroup.getGroup(0, i);
+ if (repeatedInstance.getFieldRepetitionCount(0) == 0) {
+ result.add(null);
+ } else {
+ result.add(
+ resolveGroupObject(repeatedInstance,
elementTypeInParquet, 0, elementType));
+ }
+ }
+ }
+
+ /**
+ * Reads elements from a 2-level encoded Parquet LIST (legacy). Structure:
listGroup → REPEATED
+ * element directly at field index 0, with each instance accessed by
repetition index.
+ */
+ private void read2LevelList(
+ Group listGroup,
+ Type repeatedType,
+ int numElements,
+ SeaTunnelDataType<?> elementType,
+ GroupType listGroupType,
+ List<Object> result) {
+ if (repeatedType.isPrimitive()) {
+ // 2-level LIST with primitive repeated elements: read each
instance by index
+ PrimitiveType primitiveElementType =
repeatedType.asPrimitiveType();
+ for (int i = 0; i < numElements; i++) {
+ result.add(readPrimitiveGroupObjectByIndex(listGroup,
primitiveElementType, 0, i));
+ }
+ } else {
+ // 2-level LIST with group repeated elements (e.g., LIST of ROW)
+ GroupType elementGroupType = repeatedType.asGroupType();
+ LogicalTypeAnnotation elemAnnotation =
elementGroupType.getLogicalTypeAnnotation();
+ if (elemAnnotation != null) {
+ throw CommonError.convertToSeaTunnelTypeError(
+ PARQUET,
+ "2-level LIST with annotated group element",
+ listGroupType.getName());
+ }
+ SeaTunnelRowType rowType = (SeaTunnelRowType) elementType;
+ for (int i = 0; i < numElements; i++) {
+ Group elementGroup = listGroup.getGroup(0, i);
+ Object[] objects = new Object[rowType.getTotalFields()];
+ for (int j = 0; j < rowType.getTotalFields(); j++) {
+ objects[j] =
+ resolveGroupObject(
+ elementGroup,
+ elementGroupType.getType(j),
+ j,
+ rowType.getFieldType(j));
+ }
+ result.add(new SeaTunnelRow(objects));
+ }
+ }
+ }
+
+ /**
+ * Reads a single primitive value from a Parquet Group at the given field
index and repetition
+ * index. This is needed for 2-level LIST encoding where repeated
primitive elements are
+ * accessed by repetition index.
+ */
+ private Object readPrimitiveGroupObjectByIndex(
+ Group group, PrimitiveType parquetType, int fieldIndex, int
repetitionIndex) {
+ switch (parquetType.getPrimitiveTypeName()) {
+ case BOOLEAN:
+ return group.getBoolean(fieldIndex, repetitionIndex);
+ case INT32:
+ if (parquetType.getOriginalType() == OriginalType.DATE) {
+ return LocalDate.ofEpochDay(group.getInteger(fieldIndex,
repetitionIndex));
+ }
+ return group.getInteger(fieldIndex, repetitionIndex);
+ case INT64:
+ return group.getLong(fieldIndex, repetitionIndex);
+ case INT96:
+ return int96ToLocalDateTime(group.getInt96(fieldIndex,
repetitionIndex));
+ case FLOAT:
+ return group.getFloat(fieldIndex, repetitionIndex);
+ case DOUBLE:
+ return group.getDouble(fieldIndex, repetitionIndex);
+ case BINARY:
+ return readBinaryGroupObject(
+ group.getBinary(fieldIndex, repetitionIndex),
parquetType);
+ case FIXED_LEN_BYTE_ARRAY:
+ return readFixedLenByteArrayGroupObject(
+ group.getBinary(fieldIndex, repetitionIndex),
parquetType);
+ default:
+ throw CommonError.convertToSeaTunnelTypeError(
+ PARQUET, parquetType.toString(),
parquetType.getName());
+ }
+ }
+
+ /**
+ * Reads a MAP field from a Parquet Group using the native Group API.
Parquet MAP structure:
+ * mapGroup → repeated key_value (field 0) → {key (field 0), value (field
1)}
+ */
+ private Object readMap(
+ Group group,
+ GroupType parentGroupType,
+ int mapFieldIndex,
+ SeaTunnelDataType<?> fieldType) {
+ MapType<?, ?> mapType = (MapType<?, ?>) fieldType;
+ SeaTunnelDataType<?> keyType = mapType.getKeyType();
+ SeaTunnelDataType<?> valueType = mapType.getValueType();
+
+ if (group.getFieldRepetitionCount(mapFieldIndex) == 0) {
+ return new HashMap<>();
+ }
+
+ Group mapGroup = group.getGroup(mapFieldIndex, 0);
+ GroupType mapGroupType =
parentGroupType.getType(mapFieldIndex).asGroupType();
+ GroupType keyValueGroupType = mapGroupType.getType(0).asGroupType();
+
+ int numEntries = mapGroup.getFieldRepetitionCount(0);
+ Map<Object, Object> result = new HashMap<>();
+ for (int i = 0; i < numEntries; i++) {
+ Group keyValue = mapGroup.getGroup(0, i);
+ Object key = resolveGroupObject(keyValue,
keyValueGroupType.getType(0), 0, keyType);
+ Object value = null;
+ if (keyValue.getFieldRepetitionCount(1) > 0) {
+ value = resolveGroupObject(keyValue,
keyValueGroupType.getType(1), 1, valueType);
+ }
+ result.put(key, value);
+ }
+ return result;
+ }
+
+ private Object readPrimitiveGroupObject(
+ Group group, PrimitiveType parquetType, int fieldIndex) {
+ switch (parquetType.getPrimitiveTypeName()) {
+ case BOOLEAN:
+ return group.getBoolean(fieldIndex, 0);
+ case INT32:
+ if (parquetType.getOriginalType() == OriginalType.DATE) {
+ return LocalDate.ofEpochDay(group.getInteger(fieldIndex,
0));
+ }
+ return group.getInteger(fieldIndex, 0);
+ case INT64:
+ return group.getLong(fieldIndex, 0);
+ case INT96:
+ return int96ToLocalDateTime(group.getInt96(fieldIndex, 0));
+ case FLOAT:
+ return group.getFloat(fieldIndex, 0);
+ case DOUBLE:
+ return group.getDouble(fieldIndex, 0);
+ case BINARY:
+ return readBinaryGroupObject(group.getBinary(fieldIndex, 0),
parquetType);
+ case FIXED_LEN_BYTE_ARRAY:
+ return readFixedLenByteArrayGroupObject(
+ group.getBinary(fieldIndex, 0), parquetType);
+ default:
+ throw CommonError.convertToSeaTunnelTypeError(
+ PARQUET, parquetType.toString(),
parquetType.getName());
+ }
+ }
+
+ private Object readBinaryGroupObject(Binary binary, PrimitiveType
parquetType) {
+ if (parquetType.getOriginalType() == OriginalType.DECIMAL) {
+ return binaryToDecimal(binary, parquetType);
+ }
+ if (parquetType.getOriginalType() == null) {
+ return binary.toByteBuffer();
+ }
+ return binary.toStringUsingUTF8();
+ }
+
+ private Object readFixedLenByteArrayGroupObject(Binary binary,
PrimitiveType parquetType) {
+ if (parquetType.getLogicalTypeAnnotation()
+ instanceof LogicalTypeAnnotation.DecimalLogicalTypeAnnotation)
{
+ return binaryToDecimal(binary, parquetType);
+ }
+ return binary.toByteBuffer();
+ }
+
+ private BigDecimal binaryToDecimal(Binary binary, PrimitiveType
parquetType) {
+ int scale =
+ parquetType.getLogicalTypeAnnotation()
+ instanceof
LogicalTypeAnnotation.DecimalLogicalTypeAnnotation
+ ? ((LogicalTypeAnnotation.DecimalLogicalTypeAnnotation)
+ parquetType.getLogicalTypeAnnotation())
+ .getScale()
+ : parquetType.getDecimalMetadata().getScale();
+ return new BigDecimal(new BigInteger(binary.getBytes()), scale);
+ }
+
+ private LocalDateTime int96ToLocalDateTime(Binary binary) {
+ NanoTime nanoTime = NanoTime.fromBinary(binary);
+ int julianDay = nanoTime.getJulianDay();
+ long nanosOfDay = nanoTime.getTimeOfDayNanos();
+ long timestamp =
+ (julianDay - JULIAN_DAY_NUMBER_FOR_UNIX_EPOCH) * MILLIS_PER_DAY
+ + nanosOfDay / NANOS_PER_MILLISECOND;
+ return new Timestamp(timestamp).toLocalDateTime();
+ }
+
private Object resolveObject(Object field, SeaTunnelDataType<?> fieldType)
{
if (field == null) {
return null;
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/writer/ParquetReadStrategyTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/writer/ParquetReadStrategyTest.java
index 1806d8bfa1..fb9d416bbb 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/writer/ParquetReadStrategyTest.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/writer/ParquetReadStrategyTest.java
@@ -77,6 +77,7 @@ import java.util.Map;
import java.util.TimeZone;
import static
org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_DEFAULT_NAME_DEFAULT;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BINARY;
import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32;
@Slf4j
@@ -240,6 +241,65 @@ public class ParquetReadStrategyTest {
parquetReadStrategy.read(NativeParquetWriter.DATA_FILE_PATH, "",
testCollector);
}
+ @DisabledOnOs(OS.WINDOWS)
+ @Test
+ public void testParquetReadColumnNameNotCompatibleWithAvro() throws
Exception {
+ NativeParquetWriterWithAvroIncompatibleColumn.generateTestData();
+ try {
+ ParquetReadStrategy parquetReadStrategy = new
ParquetReadStrategy();
+ LocalConf localConf = new LocalConf(FS_DEFAULT_NAME_DEFAULT);
+ parquetReadStrategy.init(localConf);
+ SeaTunnelRowType rowType =
+ parquetReadStrategy.getSeaTunnelRowTypeInfo(
+
NativeParquetWriterWithAvroIncompatibleColumn.DATA_FILE_PATH);
+ Assertions.assertEquals("job_blue-collar",
rowType.getFieldName(1));
+
+ TestCollector testCollector = new TestCollector();
+ parquetReadStrategy.read(
+
NativeParquetWriterWithAvroIncompatibleColumn.DATA_FILE_PATH,
+ "",
+ testCollector);
+ Assertions.assertEquals(1, testCollector.getRows().size());
+ SeaTunnelRow row = testCollector.getRows().get(0);
+ Assertions.assertEquals(1, row.getField(0));
+ Assertions.assertEquals("engineer", row.getField(1));
+ } finally {
+ NativeParquetWriterWithAvroIncompatibleColumn.deleteFile();
+ }
+ }
+
+ @DisabledOnOs(OS.WINDOWS)
+ @Test
+ public void testParquetReadColumnNameNotCompatibleWithAvroWithList()
throws Exception {
+
NativeParquetWriterWithAvroIncompatibleColumnAndList.generateTestData();
+ try {
+ ParquetReadStrategy parquetReadStrategy = new
ParquetReadStrategy();
+ LocalConf localConf = new LocalConf(FS_DEFAULT_NAME_DEFAULT);
+ parquetReadStrategy.init(localConf);
+ SeaTunnelRowType rowType =
+ parquetReadStrategy.getSeaTunnelRowTypeInfo(
+
NativeParquetWriterWithAvroIncompatibleColumnAndList.DATA_FILE_PATH);
+ Assertions.assertEquals("job_blue-collar",
rowType.getFieldName(1));
+ Assertions.assertEquals("skill-tags", rowType.getFieldName(2));
+
+ TestCollector testCollector = new TestCollector();
+ parquetReadStrategy.read(
+
NativeParquetWriterWithAvroIncompatibleColumnAndList.DATA_FILE_PATH,
+ "",
+ testCollector);
+ Assertions.assertEquals(1, testCollector.getRows().size());
+ SeaTunnelRow row = testCollector.getRows().get(0);
+ Assertions.assertEquals(1, row.getField(0));
+ Assertions.assertEquals("engineer", row.getField(1));
+ String[] tags = (String[]) row.getField(2);
+ Assertions.assertEquals(2, tags.length);
+ Assertions.assertEquals("java", tags[0]);
+ Assertions.assertEquals("python", tags[1]);
+ } finally {
+ NativeParquetWriterWithAvroIncompatibleColumnAndList.deleteFile();
+ }
+ }
+
@DisabledOnOs(OS.WINDOWS)
@Test
public void testParquetWithUserConfigRowType() throws Exception {
@@ -932,4 +992,102 @@ public class ParquetReadStrategyTest {
}
}
}
+
+ public static class NativeParquetWriterWithAvroIncompatibleColumn {
+
+ public static final String DATA_FILE_PATH =
"/tmp/data_invalid_avro_column.parquet";
+
+ public static MessageType createSchema() {
+ return Types.buildMessage()
+ .required(INT32)
+ .named("id")
+ .required(BINARY)
+ .as(LogicalTypeAnnotation.stringType())
+ .named("job_blue-collar")
+ .named("User");
+ }
+
+ public static void generateTestData() throws IOException {
+ deleteFile();
+ MessageType schema = createSchema();
+ Configuration conf = new Configuration();
+ GroupWriteSupport.setSchema(schema, conf);
+
+ Path file = new Path(DATA_FILE_PATH);
+ try (ParquetWriter<Group> writer =
+ ExampleParquetWriter.builder(file)
+ .withConf(conf)
+ .withCompressionCodec(CompressionCodecName.SNAPPY)
+ .build()) {
+ Group record = new SimpleGroup(schema);
+ record.add("id", 1);
+ record.add("job_blue-collar", "engineer");
+ writer.write(record);
+ }
+ }
+
+ private static void deleteFile() {
+ File parquetFile = new File(DATA_FILE_PATH);
+ if (parquetFile.exists()) {
+ parquetFile.delete();
+ }
+ File checksumFile = new File(DATA_FILE_PATH + ".crc");
+ if (checksumFile.exists()) {
+ checksumFile.delete();
+ }
+ }
+ }
+
+ public static class NativeParquetWriterWithAvroIncompatibleColumnAndList {
+
+ public static final String DATA_FILE_PATH =
+ "/tmp/data_invalid_avro_column_with_list.parquet";
+
+ public static MessageType createSchema() {
+ return Types.buildMessage()
+ .required(INT32)
+ .named("id")
+ .required(BINARY)
+ .as(LogicalTypeAnnotation.stringType())
+ .named("job_blue-collar")
+ .optionalList()
+ .optionalElement(BINARY)
+ .as(LogicalTypeAnnotation.stringType())
+ .named("skill-tags")
+ .named("User");
+ }
+
+ public static void generateTestData() throws IOException {
+ deleteFile();
+ MessageType schema = createSchema();
+ Configuration conf = new Configuration();
+ GroupWriteSupport.setSchema(schema, conf);
+
+ Path file = new Path(DATA_FILE_PATH);
+ try (ParquetWriter<Group> writer =
+ ExampleParquetWriter.builder(file)
+ .withConf(conf)
+ .withCompressionCodec(CompressionCodecName.SNAPPY)
+ .build()) {
+ Group record = new SimpleGroup(schema);
+ record.add("id", 1);
+ record.add("job_blue-collar", "engineer");
+ Group listGroup = record.addGroup("skill-tags");
+ listGroup.addGroup(0).append("element", "java");
+ listGroup.addGroup(0).append("element", "python");
+ writer.write(record);
+ }
+ }
+
+ private static void deleteFile() {
+ File parquetFile = new File(DATA_FILE_PATH);
+ if (parquetFile.exists()) {
+ parquetFile.delete();
+ }
+ File checksumFile = new File(DATA_FILE_PATH + ".crc");
+ if (checksumFile.exists()) {
+ checksumFile.delete();
+ }
+ }
+ }
}
