apache/flink · error · FlinkRuntimeException

Parquet with case insensitive mode should have no duplicate

Error message

Parquet with case insensitive mode should have no duplicate key: {}

What it means

Thrown by ParquetColumnarRowSplitReader when building the requested schema in case-insensitive mode (the default for Flink/Hive catalogs). The reader folds every field of the Parquet file schema into a map keyed by lowercased name; if two file columns collide (e.g. 'Id' and 'ID'), a FlinkRuntimeException is thrown because the reader cannot decide which physical column a query name refers to. This is detected purely from the file schema, before any data is read.

Source

Thrown at flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetColumnarRowSplitReader.java:178

    private static MessageType clipParquetSchema(
            GroupType parquetSchema, String[] fieldNames, boolean caseSensitive) {
        Type[] types = new Type[fieldNames.length];
        if (caseSensitive) {
            for (int i = 0; i < fieldNames.length; ++i) {
                String fieldName = fieldNames[i];
                if (parquetSchema.getFieldIndex(fieldName) < 0) {
                    throw new IllegalArgumentException(fieldName + " does not exist");
                }
                types[i] = parquetSchema.getType(fieldName);
            }
        } else {
            Map<String, Type> caseInsensitiveFieldMap = new HashMap<>();
            for (Type type : parquetSchema.getFields()) {
                caseInsensitiveFieldMap.compute(
                        type.getName().toLowerCase(Locale.ROOT),
                        (key, previousType) -> {
                            if (previousType != null) {
                                throw new FlinkRuntimeException(
                                        "Parquet with case insensitive mode should have no duplicate key: "
                                                + key);
                            }
                            return type;
                        });
            }
            for (int i = 0; i < fieldNames.length; ++i) {
                Type type = caseInsensitiveFieldMap.get(fieldNames[i].toLowerCase(Locale.ROOT));
                if (type == null) {
                    throw new IllegalArgumentException(fieldNames[i] + " does not exist");
                }
                // TODO clip for array,map,row types.
                types[i] = type;
            }
        }

        return Types.buildMessage().addFields(types).named("flink-parquet");
    }

View on GitHub (pinned to 2f3c205e92)

Solutions

Rewrite/rename the duplicate columns in the Parquet file so their lowercased names are unique
Set the table/format option to case-sensitive name matching (e.g. Flink Hive table with parquet column names matching exactly, or read via a projection that avoids ambiguity is NOT enough - the check runs over the whole file schema)
If the table is external, recreate the underlying data with normalized lower-case column names
As a last resort, read the file with a case-sensitive reader path (ParquetVectorReader with caseSensitive=true) where fieldNames[i] must match exactly

Example fix

// file schema: id: INT, ID: STRING  -> case-insensitive collision
// fix: rewrite file with unique names
// before (file):  message flink { required INT32 id; required BINARY ID (UTF8); }
// after (file):   message flink { required INT32 id; required BINARY id_str (UTF8); }

Defensive patterns

Strategy: validation

Validate before calling

MessageType fileSchema = ParquetFileReader.readFooter(conf, path).getFileMetaData().getSchema();
Set<String> seen = new HashSet<>();
for (Type f : fileSchema.getFields()) {
    if (!seen.add(f.getName().toLowerCase(Locale.ROOT))) {
        throw new IllegalStateException("Case-insensitive duplicate column in file: " + f.getName());
    }
}

Try / catch

catch (FlinkRuntimeException e) { if (e.getMessage() != null && e.getMessage().contains("no duplicate key")) { /* normalize/rename columns in file */ } else throw e; }

Prevention

Standardize all column names to lower case in upstream writers
Validate incoming Parquet files against the catalog schema in an ingestion test job before registering partitions

When it happens

Trigger: Reading a Parquet file whose top-level schema contains two field names that are equal after toLowerCase(Locale.ROOT), while the reader was constructed with caseSensitive=false (ParquetColumnarRowSplitReader builds caseInsensitiveFieldMap and finds previousType != null).

Common situations: Tables written by tools that preserve mixed-case column names (Spark with spark.sql.caseSensitive=true, Avro-to-Parquet conversions, hand-crafted files); Hive tables defined case-insensitively over such files; merging data from sources with inconsistent column casing.

Related errors

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/1db0f2293e08d907. Report an issue: GitHub.