Skip to content

Kafka Connect: Fix case-insensitive field lookup with name mapping - #17454

Open
annurahar wants to merge 2 commits into
apache:mainfrom
annurahar:fix_schema_case_insensitive
Open

Kafka Connect: Fix case-insensitive field lookup with name mapping#17454
annurahar wants to merge 2 commits into
apache:mainfrom
annurahar:fix_schema_case_insensitive

Conversation

@annurahar

@annurahar annurahar commented Jul 31, 2026

Copy link
Copy Markdown

Description

This PR fixes a bug where the iceberg.tables.schema-case-insensitive configuration is ignored when name mapping (schema.name-mapping.default) is present in RecordConverter.

Problem

When both name mapping and case-insensitive mode are configured, the lookupStructField method does not apply case-insensitive logic for field lookups. This causes fields with case differences (e.g., "II" vs "ii") to fail matching the existing column.

During schema evolution, the unmatched field is treated as a new field, causing a duplicate column to be added to the table. This results in unintended schema changes with redundant columns and data written to wrong columns.

Solution

Modified lookupStructField and createStructNameMap methods to respect the schemaCaseInsensitive configuration when name mapping is present:

  1. createStructNameMap: Store keys in lowercase when case-insensitive mode is enabled
  2. lookupStructField: Convert the lookup key to lowercase when case-insensitive mode is enabled

Changes

// lookupStructField - Before
return structNameMap
    .computeIfAbsent(structFieldId, notUsed -> createStructNameMap(schema))
    .get(fieldName);

// lookupStructField - After
Map<String, NestedField> nameMap =
    structNameMap.computeIfAbsent(structFieldId, notUsed -> createStructNameMap(schema));
String lookupKey = config.schemaCaseInsensitive() ? fieldName.toLowerCase() : fieldName;
return nameMap.get(lookupKey);
// createStructNameMap - Before
if (mappedField != null && !mappedField.names().isEmpty()) {
  mappedField.names().forEach(name -> map.put(name, col));
} else {
  map.put(col.name(), col);
}

// createStructNameMap - After
boolean caseInsensitive = config.schemaCaseInsensitive();
if (mappedField != null && !mappedField.names().isEmpty()) {
  mappedField
      .names()
      .forEach(name -> map.put(caseInsensitive ? name.toLowerCase() : name, col));
} else {
  map.put(caseInsensitive ? col.name().toLowerCase() : col.name(), col);
}

Testing

Added comprehensive test cases in TestRecordConverter.java:

  1. testNameMappingWithDifferentCaseAlias - Tests field lookup with alias in different case (parameterized for case-sensitive/insensitive)
  2. testNameMappingWithNestedStruct - Tests nested struct field lookup with case sensitivity (parameterized)
  3. testLookupStructFieldWithNameMappingAndCaseSensitivity - Explicit test for lookupStructField behavior with name mapping and both case sensitivity modes

Checklist

  • Bug fix (non-breaking change that fixes an issue)
  • Unit tests added
  • Code follows project style guidelines (spotlessCheck passes)
  • All existing tests pass

Related Issue

Closes: #15392

This a new PR for stale PR - #15393

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Kafka Connect: iceberg.tables.schema-case-insensitive config is ignored when name mapping is present

1 participant