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 e01147795c [core] Fix bug that overwrite does not create remote lookup
file (#8333)
e01147795c is described below
commit e01147795c8143e51a91360c376165acfee72b16
Author: tsreaper <[email protected]>
AuthorDate: Tue Jun 23 20:29:16 2026 +0800
[core] Fix bug that overwrite does not create remote lookup file (#8333)
---
.../compact/KvCompactionManagerFactory.java | 2 +-
.../compact/MergeTreeCompactManagerFactory.java | 21 ++++++++---------
.../ClusteringCompactManagerFactory.java | 2 +-
.../paimon/operation/KeyValueFileStoreWrite.java | 3 +--
.../paimon/table/sink/BatchWriteBuilderImpl.java | 13 +++--------
.../flink/lookup/LookupRemoteFileTableTest.java | 27 ++++++++++++++++++++++
6 files changed, 43 insertions(+), 25 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/KvCompactionManagerFactory.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/KvCompactionManagerFactory.java
index f512ad7b78..8ff0180bfb 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/KvCompactionManagerFactory.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/KvCompactionManagerFactory.java
@@ -104,5 +104,5 @@ public interface KvCompactionManagerFactory extends
Closeable {
ExecutorService compactExecutor,
List<DataFileMeta> restoreFiles,
@Nullable BucketedDvMaintainer dvMaintainer,
- boolean lookupEnabled);
+ boolean ignorePreviousFiles);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManagerFactory.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManagerFactory.java
index 8b858e6592..1daa51a399 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManagerFactory.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManagerFactory.java
@@ -148,13 +148,12 @@ public class MergeTreeCompactManagerFactory implements
KvCompactionManagerFactor
ExecutorService compactExecutor,
List<DataFileMeta> restoreFiles,
@Nullable BucketedDvMaintainer dvMaintainer,
- boolean lookupEnabled) {
+ boolean ignorePreviousFiles) {
if (options.writeOnly()) {
return new NoopCompactManager();
}
- CompactStrategy compactStrategy =
- createCompactStrategy(options, restoreFiles, lookupEnabled);
+ CompactStrategy compactStrategy = createCompactStrategy(options,
restoreFiles);
Comparator<InternalRow> keyComparator = keyComparatorSupplier.get();
Levels levels = new Levels(keyComparator, restoreFiles,
options.numLevels());
@Nullable FieldsComparator userDefinedSeqComparator =
udsComparatorSupplier.get();
@@ -166,7 +165,7 @@ public class MergeTreeCompactManagerFactory implements
KvCompactionManagerFactor
userDefinedSeqComparator,
levels,
dvMaintainer,
- lookupEnabled);
+ ignorePreviousFiles);
CompactionMetrics.Reporter metricsReporter =
compactionMetrics == null
? null
@@ -184,18 +183,18 @@ public class MergeTreeCompactManagerFactory implements
KvCompactionManagerFactor
rewriter,
metricsReporter,
dvMaintainer,
- lookupEnabled && options.prepareCommitWaitCompaction(),
- lookupEnabled,
+ options.prepareCommitWaitCompaction(),
+ options.needLookup(),
recordLevelExpire,
options.forceRewriteAllFiles(),
options.isChainTable());
}
private CompactStrategy createCompactStrategy(
- CoreOptions options, List<DataFileMeta> restoreFiles, boolean
lookupEnabled) {
+ CoreOptions options, List<DataFileMeta> restoreFiles) {
Long initialLastFullCompaction =
estimateLastFullCompactionTime(restoreFiles,
options.numLevels());
- if (lookupEnabled) {
+ if (options.needLookup()) {
Integer compactMaxInterval = null;
switch (options.lookupCompact()) {
case GENTLE:
@@ -251,7 +250,7 @@ public class MergeTreeCompactManagerFactory implements
KvCompactionManagerFactor
@Nullable FieldsComparator userDefinedSeqComparator,
Levels levels,
@Nullable BucketedDvMaintainer dvMaintainer,
- boolean lookupEnabled) {
+ boolean ignorePreviousFiles) {
DeletionVector.Factory dvFactory =
DeletionVector.factory(dvMaintainer);
KeyValueFileReaderFactory keyReaderFactory =
readerFactoryBuilder.build(partition, bucket, dvFactory);
@@ -277,7 +276,7 @@ public class MergeTreeCompactManagerFactory implements
KvCompactionManagerFactor
mfFactory,
mergeSorter,
logDedupEqualSupplier.get());
- } else if (lookupEnabled && lookupStrategy.needLookup) {
+ } else if (lookupStrategy.needLookup) {
PersistProcessor.Factory<?> processorFactory;
LookupMergeTreeCompactRewriter.MergeFunctionWrapperFactory<?>
wrapperFactory;
FileReaderFactory<KeyValue> lookupReaderFactory = readerFactory;
@@ -335,7 +334,7 @@ public class MergeTreeCompactManagerFactory implements
KvCompactionManagerFactor
mfFactory,
mergeSorter,
wrapperFactory,
- lookupStrategy.produceChangelog,
+ lookupStrategy.produceChangelog && !ignorePreviousFiles,
dvMaintainer,
options,
remoteLookupFileManager);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/clustering/ClusteringCompactManagerFactory.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/clustering/ClusteringCompactManagerFactory.java
index e58c172fd6..c63a9812fb 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/clustering/ClusteringCompactManagerFactory.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/clustering/ClusteringCompactManagerFactory.java
@@ -87,7 +87,7 @@ public class ClusteringCompactManagerFactory implements
KvCompactionManagerFacto
ExecutorService compactExecutor,
List<DataFileMeta> restoreFiles,
@Nullable BucketedDvMaintainer dvMaintainer,
- boolean lookupEnabled) {
+ boolean ignorePreviousFiles) {
if (options.writeOnly()) {
return new NoopCompactManager();
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
index fc46c91c4b..1bbbfd2c30 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
@@ -212,7 +212,6 @@ public class KeyValueFileStoreWrite extends
MemoryFileStoreWrite<KeyValue> {
restoreFiles);
}
- boolean lookupEnabled = !ignorePreviousFiles && options.needLookup();
KeyValueFileWriterFactory writerFactory =
writerFactoryBuilder.build(partition, bucket, options);
Comparator<InternalRow> keyComparator = keyComparatorSupplier.get();
@@ -223,7 +222,7 @@ public class KeyValueFileStoreWrite extends
MemoryFileStoreWrite<KeyValue> {
compactExecutor,
restoreFiles,
dvMaintainer,
- lookupEnabled);
+ ignorePreviousFiles);
return new MergeTreeWriter(
options.writeBufferSpillable(),
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchWriteBuilderImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchWriteBuilderImpl.java
index 989db8e655..c06bd9de42 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchWriteBuilderImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchWriteBuilderImpl.java
@@ -39,7 +39,6 @@ public class BatchWriteBuilderImpl implements
BatchWriteBuilder {
private final InnerTable table;
private final String commitUser;
- private boolean overwrite;
private Map<String, String> staticPartition;
private boolean appendCommitCheckConflict = false;
private @Nullable Long rowIdCheckFromSnapshot = null;
@@ -50,13 +49,9 @@ public class BatchWriteBuilderImpl implements
BatchWriteBuilder {
}
private BatchWriteBuilderImpl(
- InnerTable table,
- String commitUser,
- boolean overwrite,
- @Nullable Map<String, String> staticPartition) {
+ InnerTable table, String commitUser, @Nullable Map<String, String>
staticPartition) {
this.table = table;
this.commitUser = commitUser;
- this.overwrite = overwrite;
this.staticPartition = staticPartition;
}
@@ -77,14 +72,13 @@ public class BatchWriteBuilderImpl implements
BatchWriteBuilder {
@Override
public BatchWriteBuilder withOverwrite(@Nullable Map<String, String>
staticPartition) {
- this.overwrite = true;
this.staticPartition = staticPartition;
return this;
}
@Override
public BatchTableWrite newWrite() {
- return table.newWrite(commitUser).withIgnorePreviousFiles(overwrite);
+ return
table.newWrite(commitUser).withIgnorePreviousFiles(staticPartition != null);
}
@Override
@@ -102,8 +96,7 @@ public class BatchWriteBuilderImpl implements
BatchWriteBuilder {
}
public BatchWriteBuilderImpl copyWithNewTable(Table newTable) {
- return new BatchWriteBuilderImpl(
- (InnerTable) newTable, commitUser, overwrite, staticPartition);
+ return new BatchWriteBuilderImpl((InnerTable) newTable, commitUser,
staticPartition);
}
public BatchWriteBuilderImpl appendCommitCheckConflict(boolean
appendCommitCheckConflict) {
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/LookupRemoteFileTableTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/LookupRemoteFileTableTest.java
index b78b5e98d9..95b2b84fbe 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/LookupRemoteFileTableTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/LookupRemoteFileTableTest.java
@@ -268,4 +268,31 @@ public class LookupRemoteFileTableTest extends
TableTestBase {
assertThat(level5.extraFiles()).hasSize(1);
assertThat(level4.extraFiles()).hasSize(0);
}
+
+ @Test
+ public void testOverwriteGeneratesRemoteFile() throws Exception {
+ Options options = new Options();
+ options.set(CoreOptions.BUCKET, 1);
+ options.set(CoreOptions.DELETION_VECTORS_ENABLED, true);
+ options.set(CoreOptions.LOOKUP_REMOTE_FILE_ENABLED, true);
+ Identifier identifier = new Identifier("default", "t");
+ Schema schema =
+ new Schema(
+ RowType.of(new IntType(), new IntType()).getFields(),
+ Collections.emptyList(),
+ Collections.singletonList("f0"),
+ options.toMap(),
+ null);
+ catalog.createTable(identifier, schema, false);
+ FileStoreTable table = (FileStoreTable) catalog.getTable(identifier);
+ BatchWriteBuilder writeBuilder =
table.newBatchWriteBuilder().withOverwrite();
+ try (BatchTableWrite write =
writeBuilder.newWrite().withIOManager(ioManager);
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ write.write(GenericRow.of(1, 1));
+ write.write(GenericRow.of(2, 1));
+ commit.commit(write.prepareCommit());
+ }
+
+
allShouldHaveRemoteSst(table.newReadBuilder().newScan().plan().splits());
+ }
}
