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 8712ecd520 [core] Fix rewriting file index with multiple map keys of
one column (#9669)
8712ecd520 is described below
commit 8712ecd520a5c2f21cc2398254d71d3aa070a7e0
Author: jackylee <[email protected]>
AuthorDate: Fri Sep 11 14:13:29 2026 +0800
[core] Fix rewriting file index with multiple map keys of one column (#9669)
---
.../apache/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 7a0c2aa8a2..de65608816 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.ArrayList;
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 class FileIndexProcessor {
: createIndexNameMapping(
currentSchema.fields(),
fileSchema.getFields());
- List<String> 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<String> projectedColNames = new LinkedHashSet<>();
Set<String> projectedColFullNames = new HashSet<>();
Map<String, Set<String>> projectedIndexTypes = new HashMap<>();
for (Map.Entry<FileIndexOptions.Column, Map<String, Options>>
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 321088047a..fa6f51e34d 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 @@ package org.apache.paimon.index;
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<String, String> 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<Object, Object> 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<ManifestEntry> 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();
