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 9c39e0d7e2 [core] Read the table branch schemas when rewriting file
indexes (#9468)
9c39e0d7e2 is described below
commit 9c39e0d7e2effbbeea91fb104408a0e372fb4997
Author: Jiajia Li <[email protected]>
AuthorDate: Sun Aug 30 15:17:48 2026 +0800
[core] Read the table branch schemas when rewriting file indexes (#9468)
---
.../apache/paimon/index/FileIndexProcessor.java | 3 +-
.../paimon/index/FileIndexProcessorTest.java | 105 +++++++++++++++++++++
2 files changed, 106 insertions(+), 2 deletions(-)
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 db781f9d8a..7a0c2aa8a2 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
@@ -74,8 +74,7 @@ public class FileIndexProcessor {
this.fileIO = table.fileIO();
this.pathFactory = table.store().pathFactory();
this.pathFactories = new DataFilePathFactories(pathFactory);
- this.schemaInfoCache =
- new SchemaCache(fileIndexOptions, new SchemaManager(fileIO,
table.location()));
+ this.schemaInfoCache = new SchemaCache(fileIndexOptions,
table.schemaManager());
this.sizeInMeta = table.coreOptions().fileIndexInManifestThreshold();
}
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
new file mode 100644
index 0000000000..321088047a
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/index/FileIndexProcessorTest.java
@@ -0,0 +1,105 @@
+/*
+ * 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.index;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.FileSystemCatalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.TableCommitImpl;
+import org.apache.paimon.table.sink.TableWriteImpl;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link FileIndexProcessor}. */
+public class FileIndexProcessorTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ public void testProcessReadsTheSchemasOfTheTableBranch() 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");
+ options.put(CoreOptions.FILE_INDEX + ".bloom-filter.columns", "v");
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.INT()}, new
String[] {"k", "v"});
+
+ Identifier identifier = Identifier.create("mydb", "t");
+ FileStoreTable branchTable;
+ 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);
+ FileStoreTable table = (FileStoreTable)
catalog.getTable(identifier);
+ table.branchManager().createBranch("b1");
+
+ branchTable = table.switchToBranch("b1");
+ branchTable
+ .schemaManager()
+ .commitChanges(SchemaChange.addColumn("branch_only",
DataTypes.INT()));
+ branchTable = table.switchToBranch("b1");
+ }
+
+ String commitUser = UUID.randomUUID().toString();
+ try (TableWriteImpl<?> write = branchTable.newWrite(commitUser);
+ TableCommitImpl commit = branchTable.newCommit(commitUser)) {
+ write.write(GenericRow.of(1, 10, 100));
+ commit.commit(1, write.prepareCommit(false, 1));
+ }
+
+ List<ManifestEntry> entries =
branchTable.store().newScan().plan().files();
+ assertThat(entries).isNotEmpty();
+ ManifestEntry entry = entries.get(0);
+ assertThat(entry.file().schemaId()).isEqualTo(1L);
+
+ FileIndexProcessor processor = new FileIndexProcessor(branchTable);
+ DataFileMeta processed = processor.process(entry.partition(),
entry.bucket(), entry);
+ assertThat(processed.extraFiles()).isNotEmpty();
+ }
+}
