This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 0e65589efb [format] Support case-insensitive column matching for
Parquet and ORC readers (#8337)
0e65589efb is described below
commit 0e65589efbd69ee51d4e73b2f86d38a0482d047c
Author: Zouxxyy <[email protected]>
AuthorDate: Wed Jun 24 09:12:04 2026 +0800
[format] Support case-insensitive column matching for Parquet and ORC
readers (#8337)
Support case-insensitive column matching when reading Parquet and ORC
files.
When `case-sensitive` is set to `false` at the catalog level, the format
readers now resolve columns by case-insensitive name matching. This is
useful for reading external data files whose column names differ only in
case from the table schema (e.g. `Event_Name` in the file vs
`event_name` in the table).
---
.../apache/paimon/format/FileFormatFactory.java | 5 +
.../paimon/table/format/FormatReadBuilder.java | 22 +-
.../apache/paimon/format/orc/OrcFileFormat.java | 2 +
.../apache/paimon/format/orc/OrcReaderFactory.java | 5 +-
.../paimon/format/parquet/ParquetFileFormat.java | 5 +
.../format/parquet/ParquetReaderFactory.java | 41 +++-
.../format/parquet/reader/ParquetReaderUtil.java | 20 +-
.../format/orc/OrcCaseInsensitiveReadTest.java | 124 ++++++++++
.../parquet/ParquetCaseInsensitiveReadTest.java | 267 +++++++++++++++++++++
.../table/FormatTableCaseInsensitiveTest.scala | 125 ++++++++++
10 files changed, 606 insertions(+), 10 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/format/FileFormatFactory.java
b/paimon-common/src/main/java/org/apache/paimon/format/FileFormatFactory.java
index 79354b9c26..b5cba150fe 100644
---
a/paimon-common/src/main/java/org/apache/paimon/format/FileFormatFactory.java
+++
b/paimon-common/src/main/java/org/apache/paimon/format/FileFormatFactory.java
@@ -19,6 +19,7 @@
package org.apache.paimon.format;
import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.options.CatalogOptions;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
@@ -94,5 +95,9 @@ public interface FileFormatFactory {
public MemorySize blockSize() {
return blockSize;
}
+
+ public boolean caseSensitive() {
+ return
options.getOptional(CatalogOptions.CASE_SENSITIVE).orElse(true);
+ }
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
index 05ad36da72..30a567b436 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
@@ -19,6 +19,7 @@
package org.apache.paimon.table.format;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.format.FileFormatDiscover;
@@ -26,6 +27,8 @@ import org.apache.paimon.format.FormatReaderContext;
import org.apache.paimon.format.FormatReaderFactory;
import org.apache.paimon.io.DataFileRecordReader;
import org.apache.paimon.mergetree.compact.ConcatRecordReader;
+import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.options.Options;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.partition.PartitionUtils;
import org.apache.paimon.predicate.Predicate;
@@ -78,7 +81,24 @@ public class FormatReadBuilder implements ReadBuilder {
public FormatReadBuilder(FormatTable table) {
this.table = table;
this.readType = this.table.rowType();
- this.options = new CoreOptions(table.options());
+ this.options = mergeCaseSensitive(table);
+ }
+
+ /**
+ * Case-sensitivity is a catalog-level property carried by {@link
CatalogContext}, not part of
+ * the table options. Merge it onto the format options bus once so it
flows naturally to the
+ * format readers via {@link
org.apache.paimon.format.FileFormatFactory.FormatContext}.
+ */
+ private static CoreOptions mergeCaseSensitive(FormatTable table) {
+ CatalogContext ctx = table.catalogContext();
+ if (ctx == null) {
+ return new CoreOptions(table.options());
+ }
+ Options merged = new Options(table.options());
+ ctx.options()
+ .getOptional(CatalogOptions.CASE_SENSITIVE)
+ .ifPresent(v -> merged.set(CatalogOptions.CASE_SENSITIVE, v));
+ return new CoreOptions(merged);
}
@Override
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/orc/OrcFileFormat.java
b/paimon-format/src/main/java/org/apache/paimon/format/orc/OrcFileFormat.java
index 1072f036d0..01c2fa9517 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/orc/OrcFileFormat.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/orc/OrcFileFormat.java
@@ -78,6 +78,8 @@ public class OrcFileFormat extends FileFormat {
this.orcProperties = getOrcProperties(formatContext.options(),
formatContext);
this.readerConf = new org.apache.hadoop.conf.Configuration(false);
this.orcProperties.forEach((k, v) -> readerConf.set(k.toString(),
v.toString()));
+ OrcConf.IS_SCHEMA_EVOLUTION_CASE_SENSITIVE.setBoolean(
+ readerConf, formatContext.caseSensitive());
this.writerConf = new org.apache.hadoop.conf.Configuration(false);
this.orcProperties.forEach((k, v) -> writerConf.set(k.toString(),
v.toString()));
this.readBatchSize = formatContext.readBatchSize();
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/orc/OrcReaderFactory.java
b/paimon-format/src/main/java/org/apache/paimon/format/orc/OrcReaderFactory.java
index 81bfdd3185..b1de74242a 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/orc/OrcReaderFactory.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/orc/OrcReaderFactory.java
@@ -288,8 +288,9 @@ public class OrcReaderFactory implements
FormatReaderFactory {
.range(offsetAndLength.getLeft(),
offsetAndLength.getRight())
.useZeroCopy(OrcConf.USE_ZEROCOPY.getBoolean(conf))
.skipCorruptRecords(OrcConf.SKIP_CORRUPT_DATA.getBoolean(conf))
- .tolerateMissingSchema(
-
OrcConf.TOLERATE_MISSING_SCHEMA.getBoolean(conf));
+
.tolerateMissingSchema(OrcConf.TOLERATE_MISSING_SCHEMA.getBoolean(conf))
+ .isSchemaEvolutionCaseAware(
+
OrcConf.IS_SCHEMA_EVOLUTION_CASE_SENSITIVE.getBoolean(conf));
if (!conjunctPredicates.isEmpty() && !deletionVectorsEnabled &&
selection == null) {
// row group filter push down will make row number change
incorrect
// so deletion vectors mode and bitmap index cannot work with
row group push down
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetFileFormat.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetFileFormat.java
index 9ccf88bd17..66ab9565e7 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetFileFormat.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetFileFormat.java
@@ -27,6 +27,7 @@ import org.apache.paimon.format.SimpleStatsExtractor;
import org.apache.paimon.format.parquet.writer.RowDataParquetBuilder;
import org.apache.paimon.format.variant.VariantInferenceConfig;
import org.apache.paimon.format.variant.VariantInferenceWriterFactory;
+import org.apache.paimon.options.CatalogOptions;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.Predicate;
@@ -106,6 +107,10 @@ public class ParquetFileFormat extends FileFormat {
ParquetOutputFormat.BLOCK_SIZE,
String.valueOf(blockSize.getBytes()));
}
+ // case-sensitive is not a parquet.* key, so it is dropped by
getIdentifierPrefixOptions;
+ // carry the resolved value onto the reader options bus for
ParquetReaderFactory to read.
+ parquetOptions.set(CatalogOptions.CASE_SENSITIVE,
context.caseSensitive());
+
return parquetOptions;
}
}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java
index fad603e74c..43c516a48c 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java
@@ -28,6 +28,7 @@ import org.apache.paimon.data.variant.VariantPathSegment;
import org.apache.paimon.format.FormatReaderFactory;
import org.apache.paimon.format.parquet.reader.VectorizedParquetRecordReader;
import org.apache.paimon.format.parquet.type.ParquetField;
+import org.apache.paimon.options.CatalogOptions;
import org.apache.paimon.options.Options;
import org.apache.paimon.reader.FileRecordReader;
import org.apache.paimon.types.ArrayType;
@@ -82,6 +83,7 @@ public class ParquetReaderFactory implements
FormatReaderFactory {
private final Options conf;
private final DataField[] readFields;
private final int batchSize;
+ private final boolean caseSensitive;
@Nullable private final FilterCompat.Filter filter;
/**
@@ -102,6 +104,7 @@ public class ParquetReaderFactory implements
FormatReaderFactory {
this.conf = conf;
this.readFields = readType.getFields().toArray(new DataField[0]);
this.batchSize = batchSize;
+ this.caseSensitive =
conf.getOptional(CatalogOptions.CASE_SENSITIVE).orElse(true);
this.filter = filter;
}
@@ -175,21 +178,47 @@ public class ParquetReaderFactory implements
FormatReaderFactory {
Type[] types = new Type[readFields.length];
for (int i = 0; i < readFields.length; ++i) {
String fieldName = readFields[i].name();
- if (!parquetSchema.containsField(fieldName)) {
+ Type matched = matchParquetField(parquetSchema, fieldName);
+ if (matched != null) {
+ types[i] = clipParquetType(readFields[i].type(), matched);
+ } else {
LOG.warn(
"{} does not exist in {}, will fill the field with
null.",
fieldName,
parquetSchema);
types[i] =
ParquetSchemaConverter.convertToParquetType(readFields[i]);
- } else {
- Type parquetType = parquetSchema.getType(fieldName);
- types[i] = clipParquetType(readFields[i].type(), parquetType);
}
}
return Types.buildMessage().addFields(types).named(PAIMON_SCHEMA);
}
+ /**
+ * Resolves a field of {@code group} by {@code fieldName}, returning
{@code null} when no field
+ * matches. In case-sensitive mode only an exact-name match is accepted.
In case-insensitive
+ * mode a name that matches more than one parquet field (differing only by
case) is ambiguous
+ * and fails, mirroring Spark's case-insensitive Parquet resolution.
+ */
+ @Nullable
+ private Type matchParquetField(GroupType group, String fieldName) {
+ if (caseSensitive) {
+ return group.containsField(fieldName) ? group.getType(fieldName) :
null;
+ }
+ Type matched = null;
+ for (Type field : group.getFields()) {
+ if (field.getName().equalsIgnoreCase(fieldName)) {
+ if (matched != null) {
+ throw new RuntimeException(
+ String.format(
+ "Found duplicate field(s) \"%s\": [%s, %s]
in case-insensitive mode",
+ fieldName, matched.getName(),
field.getName()));
+ }
+ matched = field;
+ }
+ }
+ return matched;
+ }
+
/** Clips `parquetType` by `readType`. */
private Type clipParquetType(DataType readType, Type parquetType) {
switch (readType.getTypeRoot()) {
@@ -202,8 +231,8 @@ public class ParquetReaderFactory implements
FormatReaderFactory {
List<Type> rowGroupFields = new ArrayList<>();
for (DataField field : rowType.getFields()) {
String fieldName = field.name();
- if (rowGroup.containsField(fieldName)) {
- Type type = rowGroup.getType(fieldName);
+ Type type = matchParquetField(rowGroup, fieldName);
+ if (type != null) {
rowGroupFields.add(clipParquetType(field.type(),
type));
} else {
// todo: support nested field missing
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/ParquetReaderUtil.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/ParquetReaderUtil.java
index 316cc2e4fe..02ffe69931 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/ParquetReaderUtil.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/ParquetReaderUtil.java
@@ -62,6 +62,7 @@ import org.apache.parquet.io.ColumnIO;
import org.apache.parquet.io.GroupColumnIO;
import org.apache.parquet.io.MessageColumnIO;
import org.apache.parquet.io.PrimitiveColumnIO;
+import org.apache.parquet.schema.GroupType;
import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.Type;
@@ -275,7 +276,7 @@ public class ParquetReaderUtil {
constructField(
children.get(i),
lookupColumnByName(groupColumnIO, childName),
- parquetType.asGroupType().getType(childName)));
+ getTypeIgnoreCase(parquetType.asGroupType(),
childName)));
}
return new ParquetGroupField(
@@ -414,6 +415,23 @@ public class ParquetReaderUtil {
columnName, String.join(".",
groupColumnIO.getFieldPath())));
}
+ /**
+ * Resolves a child {@link Type} by name, first by exact match then
case-insensitively,
+ * mirroring {@link #lookupColumnByName}. Falls back to {@link
GroupType#getType(String)} (which
+ * throws) so a genuinely missing field keeps the original failure
behavior.
+ */
+ private static Type getTypeIgnoreCase(GroupType groupType, String
fieldName) {
+ if (groupType.containsField(fieldName)) {
+ return groupType.getType(fieldName);
+ }
+ for (Type field : groupType.getFields()) {
+ if (field.getName().equalsIgnoreCase(fieldName)) {
+ return field;
+ }
+ }
+ return groupType.getType(fieldName);
+ }
+
public static GroupColumnIO getMapKeyValueColumn(GroupColumnIO
groupColumnIO) {
while (groupColumnIO.getChildrenCount() == 1) {
groupColumnIO = (GroupColumnIO) groupColumnIO.getChild(0);
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/orc/OrcCaseInsensitiveReadTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/orc/OrcCaseInsensitiveReadTest.java
new file mode 100644
index 0000000000..1af3e0b8d1
--- /dev/null
+++
b/paimon-format/src/test/java/org/apache/paimon/format/orc/OrcCaseInsensitiveReadTest.java
@@ -0,0 +1,124 @@
+/*
+ * 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.paimon.format.orc;
+
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.format.FormatReaderContext;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.orc.OrcConf;
+import org.apache.orc.OrcFile;
+import org.apache.orc.TypeDescription;
+import org.apache.orc.Writer;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test case-insensitive column matching in {@link OrcReaderFactory}. */
+class OrcCaseInsensitiveReadTest {
+
+ @TempDir File folder;
+
+ @Test
+ void testCaseInsensitiveColumnMatching() throws Exception {
+ Path path = writeMixedCaseOrc();
+
+ // Read with lowercase column names + caseSensitive=false.
+ RowType readType =
+ RowType.builder()
+ .field("event_name", DataTypes.STRING())
+ .field("campaign_id", DataTypes.STRING())
+ .field("amount", DataTypes.BIGINT())
+ .build();
+
+ Configuration conf = new Configuration(false);
+ OrcConf.IS_SCHEMA_EVOLUTION_CASE_SENSITIVE.setBoolean(conf, false);
+
+ OrcReaderFactory factory =
+ new OrcReaderFactory(conf, readType, Collections.emptyList(),
1024, false, false);
+
+ LocalFileIO fileIO = new LocalFileIO();
+ List<InternalRow> rows = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+ factory.createReader(
+ new FormatReaderContext(fileIO, path,
fileIO.getFileSize(path)))) {
+ reader.forEachRemaining(
+ row ->
+ rows.add(
+ GenericRow.of(
+ row.isNullAt(0) ? null :
row.getString(0).copy(),
+ row.isNullAt(1) ? null :
row.getString(1).copy(),
+ row.isNullAt(2) ? null : (Object)
row.getLong(2))));
+ }
+
+ assertThat(rows).hasSize(3);
+ assertThat(rows.get(0).getString(0).toString()).isEqualTo("install");
+ assertThat(rows.get(0).getString(1).toString()).isEqualTo("c001");
+ assertThat(rows.get(0).getLong(2)).isEqualTo(100L);
+ }
+
+ private Path writeMixedCaseOrc() throws Exception {
+ Path path = new Path(folder.getPath(), UUID.randomUUID().toString());
+ TypeDescription schema =
+ TypeDescription.fromString(
+
"struct<Event_Name:string,Campaign_ID:string,Amount:bigint>");
+ Configuration conf = new Configuration(false);
+
+ try (Writer writer =
+ OrcFile.createWriter(
+ new org.apache.hadoop.fs.Path(path.toString()),
+ OrcFile.writerOptions(conf).setSchema(schema))) {
+ VectorizedRowBatch batch = schema.createRowBatch();
+ BytesColumnVector eventName = (BytesColumnVector) batch.cols[0];
+ BytesColumnVector campaignId = (BytesColumnVector) batch.cols[1];
+ LongColumnVector amount = (LongColumnVector) batch.cols[2];
+
+ String[][] data = {{"install", "c001"}, {"purchase", "c002"},
{"login", "c001"}};
+ long[] amounts = {100L, 200L, 300L};
+
+ for (int i = 0; i < 3; i++) {
+ int row = batch.size++;
+ byte[] eventBytes =
data[i][0].getBytes(StandardCharsets.UTF_8);
+ byte[] cidBytes = data[i][1].getBytes(StandardCharsets.UTF_8);
+ eventName.setVal(row, eventBytes);
+ campaignId.setVal(row, cidBytes);
+ amount.vector[row] = amounts[i];
+ }
+ writer.addRowBatch(batch);
+ }
+ return path;
+ }
+}
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetCaseInsensitiveReadTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetCaseInsensitiveReadTest.java
new file mode 100644
index 0000000000..f8c22de58b
--- /dev/null
+++
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetCaseInsensitiveReadTest.java
@@ -0,0 +1,267 @@
+/*
+ * 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.paimon.format.parquet;
+
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.format.FormatReaderContext;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.parquet.column.ParquetProperties;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.SimpleGroupFactory;
+import org.apache.parquet.filter2.compat.FilterCompat;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.hadoop.util.HadoopOutputFile;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.MessageTypeParser;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test case-insensitive column matching in {@link ParquetReaderFactory}. */
+class ParquetCaseInsensitiveReadTest {
+
+ @TempDir File folder;
+
+ @Test
+ void testCaseInsensitiveColumnMatching() throws Exception {
+ Path path = writeMixedCaseParquet();
+
+ // Read with lowercase column names + caseSensitive=false.
+ RowType readType =
+ RowType.builder()
+ .field("event_name", DataTypes.STRING())
+ .field("campaign_id", DataTypes.STRING())
+ .field("amount", DataTypes.BIGINT())
+ .build();
+
+ List<InternalRow> rows = read(path, readType, false);
+
+ assertThat(rows).hasSize(3);
+ assertThat(rows.get(0).getString(0).toString()).isEqualTo("install");
+ assertThat(rows.get(0).getString(1).toString()).isEqualTo("c001");
+ assertThat(rows.get(0).getLong(2)).isEqualTo(100L);
+ }
+
+ @Test
+ void testCaseSensitiveReturnsNull() throws Exception {
+ Path path = writeMixedCaseParquet();
+
+ RowType readType =
+ RowType.builder()
+ .field("event_name", DataTypes.STRING())
+ .field("amount", DataTypes.BIGINT())
+ .build();
+
+ // caseSensitive=true: lowercase names won't match mixed-case parquet
columns.
+ List<InternalRow> rows = read(path, readType, true);
+
+ assertThat(rows).hasSize(3);
+ assertThat(rows.get(0).isNullAt(0)).isTrue();
+ assertThat(rows.get(0).isNullAt(1)).isTrue();
+ }
+
+ @Test
+ void testNestedCaseInsensitiveColumnMatching() throws Exception {
+ Path path = writeNestedMixedCaseParquet();
+
+ // Lowercase top-level and nested field names + caseSensitive=false.
+ RowType readType =
+ RowType.builder()
+ .field("event_name", DataTypes.STRING())
+ .field(
+ "user_info",
+ RowType.builder()
+ .field("user_name", DataTypes.STRING())
+ .field("user_age", DataTypes.BIGINT())
+ .build())
+ .build();
+
+ List<InternalRow> rows = read(path, readType, false);
+
+ assertThat(rows).hasSize(2);
+ assertThat(rows.get(0).getString(0).toString()).isEqualTo("install");
+ InternalRow nested = rows.get(0).getRow(1, 2);
+ assertThat(nested.getString(0).toString()).isEqualTo("alice");
+ assertThat(nested.getLong(1)).isEqualTo(20L);
+ }
+
+ @Test
+ void testAmbiguousCaseInsensitiveMatchFails() throws Exception {
+ Path path = writeDuplicateCaseParquet();
+
+ RowType readType = RowType.builder().field("col",
DataTypes.STRING()).build();
+
+ assertThatThrownBy(() -> read(path, readType, false))
+ .hasMessageContaining("Found duplicate field(s)")
+ .hasMessageContaining("col");
+ }
+
+ private List<InternalRow> read(Path path, RowType readType, boolean
caseSensitive)
+ throws Exception {
+ Options options = new Options();
+ options.set(CatalogOptions.CASE_SENSITIVE, caseSensitive);
+ ParquetReaderFactory factory =
+ new ParquetReaderFactory(options, readType, 1024,
FilterCompat.NOOP);
+ LocalFileIO fileIO = new LocalFileIO();
+ List<InternalRow> rows = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+ factory.createReader(
+ new FormatReaderContext(fileIO, path,
fileIO.getFileSize(path)))) {
+ // Materialize a copy because InternalRow instances are reused
across iterations.
+ reader.forEachRemaining(row -> rows.add(copy(row, readType)));
+ }
+ return rows;
+ }
+
+ private static InternalRow copy(InternalRow row, RowType rowType) {
+ Object[] values = new Object[rowType.getFieldCount()];
+ for (int i = 0; i < values.length; i++) {
+ if (row.isNullAt(i)) {
+ continue;
+ }
+ switch (rowType.getTypeAt(i).getTypeRoot()) {
+ case VARCHAR:
+ case CHAR:
+ values[i] = row.getString(i).copy();
+ break;
+ case BIGINT:
+ values[i] = row.getLong(i);
+ break;
+ case ROW:
+ RowType child = (RowType) rowType.getTypeAt(i);
+ values[i] = copy(row.getRow(i, child.getFieldCount()),
child);
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ "Unhandled type in test: " + rowType.getTypeAt(i));
+ }
+ }
+ return GenericRow.of(values);
+ }
+
+ private Path writeMixedCaseParquet() throws Exception {
+ MessageType schema =
+ MessageTypeParser.parseMessageType(
+ "message root {\n"
+ + " optional binary Event_Name (UTF8);\n"
+ + " optional binary Campaign_ID (UTF8);\n"
+ + " optional int64 Amount;\n"
+ + "}");
+
+ String[][] data = {{"install", "c001"}, {"purchase", "c002"},
{"login", "c001"}};
+ long[] amounts = {100L, 200L, 300L};
+ return write(
+ schema,
+ factory -> {
+ List<Group> groups = new ArrayList<>();
+ for (int i = 0; i < data.length; i++) {
+ groups.add(
+ factory.newGroup()
+ .append("Event_Name", data[i][0])
+ .append("Campaign_ID", data[i][1])
+ .append("Amount", amounts[i]));
+ }
+ return groups;
+ });
+ }
+
+ private Path writeNestedMixedCaseParquet() throws Exception {
+ MessageType schema =
+ MessageTypeParser.parseMessageType(
+ "message root {\n"
+ + " optional binary Event_Name (UTF8);\n"
+ + " optional group User_Info {\n"
+ + " optional binary User_Name (UTF8);\n"
+ + " optional int64 User_Age;\n"
+ + " }\n"
+ + "}");
+
+ String[][] data = {{"install", "alice", "20"}, {"login", "bob", "30"}};
+ return write(
+ schema,
+ factory -> {
+ List<Group> groups = new ArrayList<>();
+ for (String[] row : data) {
+ Group g = factory.newGroup().append("Event_Name",
row[0]);
+ g.addGroup("User_Info")
+ .append("User_Name", row[1])
+ .append("User_Age", Long.parseLong(row[2]));
+ groups.add(g);
+ }
+ return groups;
+ });
+ }
+
+ private Path writeDuplicateCaseParquet() throws Exception {
+ // An exact match ("col") coexists with a case-variant ("COL");
reading "col" must still be
+ // ambiguous in case-insensitive mode, mirroring Spark.
+ MessageType schema =
+ MessageTypeParser.parseMessageType(
+ "message root {\n"
+ + " optional binary col (UTF8);\n"
+ + " optional binary COL (UTF8);\n"
+ + "}");
+
+ return write(
+ schema,
+ factory ->
+ Collections.singletonList(
+ factory.newGroup().append("col",
"a").append("COL", "b")));
+ }
+
+ private interface GroupSupplier {
+ List<Group> get(SimpleGroupFactory factory);
+ }
+
+ private Path write(MessageType schema, GroupSupplier supplier) throws
Exception {
+ Path path = new Path(folder.getPath(), UUID.randomUUID().toString());
+ Configuration conf = new Configuration();
+ try (ParquetWriter<Group> writer =
+ ExampleParquetWriter.builder(
+ HadoopOutputFile.fromPath(
+ new
org.apache.hadoop.fs.Path(path.toString()), conf))
+ .withType(schema)
+ .withConf(conf)
+
.withWriterVersion(ParquetProperties.WriterVersion.PARQUET_1_0)
+ .build()) {
+ for (Group group : supplier.get(new SimpleGroupFactory(schema))) {
+ writer.write(group);
+ }
+ }
+ return path;
+ }
+}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/FormatTableCaseInsensitiveTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/FormatTableCaseInsensitiveTest.scala
new file mode 100644
index 0000000000..d857819710
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/FormatTableCaseInsensitiveTest.scala
@@ -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.paimon.spark.table
+
+import org.apache.paimon.spark.PaimonSparkTestWithRestCatalogBase
+import org.apache.paimon.spark.SparkCatalog
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.Row
+
+class FormatTableCaseInsensitiveTest extends
PaimonSparkTestWithRestCatalogBase {
+
+ override protected def sparkConf: SparkConf = {
+ val conf = super.sparkConf
+ .set("spark.sql.catalog.paimon.case-sensitive", "false")
+ Seq("metastore", "uri", "token", "warehouse", "token.provider").foreach {
+ key =>
+ conf.getOption(s"spark.sql.catalog.paimon.$key").foreach {
+ value => conf.set(s"spark.sql.catalog.paimon_cs.$key", value)
+ }
+ }
+ conf
+ .set("spark.sql.catalog.paimon_cs", classOf[SparkCatalog].getName)
+ .set("spark.sql.catalog.paimon_cs.case-sensitive", "true")
+ }
+
+ override protected def beforeEach(): Unit = {
+ super.beforeEach()
+ sql("USE paimon")
+ sql("CREATE DATABASE IF NOT EXISTS test_db")
+ sql("USE test_db")
+ }
+
+ test("format table: case-insensitive read all and partial columns") {
+ Seq("parquet", "orc").foreach {
+ format =>
+ val dataPath = s"${tempDBDir.getCanonicalPath}/${format}_ci"
+ withTable("writer_tbl", "reader_tbl") {
+ sql(s"""CREATE TABLE writer_tbl (Event_Name STRING, Campaign_ID
STRING, Amount BIGINT)
+ |USING $format LOCATION '$dataPath'
+ |TBLPROPERTIES ('format-table.implementation'='paimon')
+ |""".stripMargin)
+ sql("INSERT INTO writer_tbl VALUES ('install', 'c001', 100),
('purchase', 'c002', 200)")
+
+ sql(s"""CREATE TABLE reader_tbl (event_name STRING, campaign_id
STRING, amount BIGINT)
+ |USING $format LOCATION '$dataPath'
+ |TBLPROPERTIES ('format-table.implementation'='paimon')
+ |""".stripMargin)
+
+ // case-sensitive=false: all columns
+ checkAnswer(
+ sql("SELECT * FROM paimon.test_db.reader_tbl ORDER BY amount"),
+ Seq(Row("install", "c001", 100L), Row("purchase", "c002", 200L))
+ )
+ // case-sensitive=true: all columns return nulls
+ checkAnswer(
+ sql("SELECT * FROM paimon_cs.test_db.reader_tbl ORDER BY amount"),
+ Seq(Row(null, null, null), Row(null, null, null))
+ )
+
+ // case-sensitive=false: partial columns
+ checkAnswer(
+ sql("SELECT event_name, amount FROM paimon.test_db.reader_tbl
ORDER BY amount"),
+ Seq(Row("install", 100L), Row("purchase", 200L))
+ )
+ // case-sensitive=true: partial columns return nulls
+ checkAnswer(
+ sql("SELECT event_name, amount FROM paimon_cs.test_db.reader_tbl
ORDER BY amount"),
+ Seq(Row(null, null), Row(null, null))
+ )
+ }
+ }
+ }
+
+ test("parquet format table: case-insensitive read nested struct columns") {
+ val dataPath = s"${tempDBDir.getCanonicalPath}/parquet_ci_nested"
+ withTable("nested_writer", "nested_reader") {
+ sql(s"""CREATE TABLE nested_writer (
+ | Event_Name STRING,
+ | User_Info STRUCT<User_Name: STRING, User_Age: BIGINT>
+ |) USING parquet LOCATION '$dataPath'
+ |TBLPROPERTIES ('format-table.implementation'='paimon')
+ |""".stripMargin)
+ sql(
+ "INSERT INTO nested_writer VALUES ('install',
named_struct('User_Name', 'alice', 'User_Age', 20L)),"
+ + " ('login', named_struct('User_Name', 'bob', 'User_Age', 30L))")
+
+ sql(s"""CREATE TABLE nested_reader (
+ | event_name STRING,
+ | user_info STRUCT<user_name: STRING, user_age: BIGINT>
+ |) USING parquet LOCATION '$dataPath'
+ |TBLPROPERTIES ('format-table.implementation'='paimon')
+ |""".stripMargin)
+
+ // case-sensitive=false: nested struct fields match
+ checkAnswer(
+ sql(
+ "SELECT event_name, user_info.user_name, user_info.user_age FROM
paimon.test_db.nested_reader ORDER BY event_name"),
+ Seq(Row("install", "alice", 20L), Row("login", "bob", 30L))
+ )
+ // case-sensitive=true: returns nulls
+ checkAnswer(
+ sql(
+ "SELECT event_name, user_info.user_name, user_info.user_age FROM
paimon_cs.test_db.nested_reader ORDER BY event_name"),
+ Seq(Row(null, null, null), Row(null, null, null))
+ )
+ }
+ }
+}
