From 78e76935d417f3a27f2d17afe85c2167a8b4059b Mon Sep 17 00:00:00 2001 From: jackylee-ch Date: Mon, 7 Sep 2026 11:46:03 +0800 Subject: [PATCH] [core] Fix rewriting file index with multiple map keys of one column FileIndexOptions keys nested index entries by Column, which includes the nested name, so m[k1] and m[k2] are two entries reporting the same top level column m. FileIndexProcessor collected that name once per entry into a list, so the projection carried m twice and RowType.project rejected it with "Field names must be unique. Found duplicates: [m]". DataFileIndexWriter already keys its maintainer by the top level column and adds each nested key to it, so the configuration is valid on write and only rewriting the index failed. --- .../paimon/index/FileIndexProcessor.java | 5 +- .../paimon/index/FileIndexProcessorTest.java | 72 +++++++++++++++++++ 2 files changed, 76 insertions(+), 1 deletion(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/index/FileIndexProcessor.java b/paimon-core/src/main/java/org/apache/paimon/index/FileIndexProcessor.java index 7a0c2aa8a269..de6560881655 100644 --- a/paimon-core/src/main/java/org/apache/paimon/index/FileIndexProcessor.java +++ b/paimon-core/src/main/java/org/apache/paimon/index/FileIndexProcessor.java @@ -49,6 +49,7 @@ import java.util.Collections; import java.util.HashMap; import java.util.HashSet; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Set; @@ -211,7 +212,9 @@ public SchemaInfo schemaInfo(long schemaId) { : createIndexNameMapping( currentSchema.fields(), fileSchema.getFields()); - List projectedColNames = new ArrayList<>(); + // several nested columns can share one top level map column, and the projection + // must not repeat it: RowType rejects duplicate field names + Set projectedColNames = new LinkedHashSet<>(); Set projectedColFullNames = new HashSet<>(); Map> projectedIndexTypes = new HashMap<>(); for (Map.Entry> entry : diff --git a/paimon-core/src/test/java/org/apache/paimon/index/FileIndexProcessorTest.java b/paimon-core/src/test/java/org/apache/paimon/index/FileIndexProcessorTest.java index 321088047a0f..fa6f51e34d81 100644 --- a/paimon-core/src/test/java/org/apache/paimon/index/FileIndexProcessorTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/index/FileIndexProcessorTest.java @@ -21,10 +21,14 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.catalog.FileSystemCatalog; import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.GenericMap; import org.apache.paimon.data.GenericRow; +import org.apache.paimon.fileindex.FileIndexFormat; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.io.DataFilePathFactory; import org.apache.paimon.manifest.ManifestEntry; import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaChange; @@ -51,6 +55,74 @@ public class FileIndexProcessorTest { @TempDir java.nio.file.Path tempDir; + @Test + public void testProcessIndexesTwoKeysOfOneMapColumn() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path warehouse = new Path(tempDir.toString()); + Map options = new HashMap<>(); + options.put(CoreOptions.BUCKET.key(), "1"); + options.put(CoreOptions.FILE_FORMAT.key(), "parquet"); + // both entries share the top level column "m" + options.put(CoreOptions.FILE_INDEX + ".bloom-filter.columns", "m[k1],m[k2]"); + RowType rowType = + RowType.of( + new DataType[] { + DataTypes.INT(), DataTypes.MAP(DataTypes.STRING(), DataTypes.INT()) + }, + new String[] {"k", "m"}); + + Identifier identifier = Identifier.create("mydb", "t"); + FileStoreTable table; + try (FileSystemCatalog catalog = new FileSystemCatalog(fileIO, warehouse)) { + catalog.createDatabase("mydb", false); + catalog.createTable( + identifier, + new Schema( + rowType.getFields(), + Collections.emptyList(), + Collections.singletonList("k"), + options, + ""), + false); + table = (FileStoreTable) catalog.getTable(identifier); + } + + Map map = new HashMap<>(); + map.put(BinaryString.fromString("k1"), 1); + map.put(BinaryString.fromString("k2"), 2); + + String commitUser = UUID.randomUUID().toString(); + try (TableWriteImpl write = table.newWrite(commitUser); + TableCommitImpl commit = table.newCommit(commitUser)) { + write.write(GenericRow.of(1, new GenericMap(map))); + commit.commit(1, write.prepareCommit(false, 1)); + } + + List entries = table.store().newScan().plan().files(); + assertThat(entries).isNotEmpty(); + ManifestEntry entry = entries.get(0); + + FileIndexProcessor processor = new FileIndexProcessor(table); + DataFileMeta processed = processor.process(entry.partition(), entry.bucket(), entry); + assertThat(processed.extraFiles()).isNotEmpty(); + + // both keys have to survive: deduplicating the entries instead of the column names + // would silently drop one of them + String indexFile = + processed.extraFiles().stream() + .filter(name -> name.endsWith(DataFilePathFactory.INDEX_PATH_SUFFIX)) + .findFirst() + .orElseThrow(() -> new AssertionError("no file index was written")); + Path indexPath = + new Path( + table.store().pathFactory().bucketPath(entry.partition(), entry.bucket()), + indexFile); + try (FileIndexFormat.Reader reader = + FileIndexFormat.createReader(fileIO.newInputStream(indexPath), rowType)) { + assertThat(reader.readAll().keySet()).containsExactlyInAnyOrder("m[k1]", "m[k2]"); + } + } + @Test public void testProcessReadsTheSchemasOfTheTableBranch() throws Exception { LocalFileIO fileIO = LocalFileIO.create();