[core] Fix bug that overwrite does not create remote lookup file

paimon-core/src/main/java/org/apache/paimon/mergetree/compact/KvCompactionManagerFactory.java

@@ -104,5 +104,5 @@ CompactManager create(
             ExecutorService compactExecutor,
             List<DataFileMeta> restoreFiles,
             @Nullable BucketedDvMaintainer dvMaintainer,
-            boolean lookupEnabled);
+            boolean ignorePreviousFiles);
 }

paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManagerFactory.java

@@ -148,13 +148,12 @@ public CompactManager create(
             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 CompactManager create(
                         userDefinedSeqComparator,
                         levels,
                         dvMaintainer,
-                        lookupEnabled);
+                        ignorePreviousFiles);
         CompactionMetrics.Reporter metricsReporter =
                 compactionMetrics == null
                         ? null
@@ -184,18 +183,18 @@ public CompactManager create(
                 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 @@ private MergeTreeCompactRewriter createRewriter(
             @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 @@ private MergeTreeCompactRewriter createRewriter(
                     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 @@ private MergeTreeCompactRewriter createRewriter(
                     mfFactory,
                     mergeSorter,
                     wrapperFactory,
-                    lookupStrategy.produceChangelog,
+                    lookupStrategy.produceChangelog && !ignorePreviousFiles,
                     dvMaintainer,
                     options,
                     remoteLookupFileManager);

paimon-core/src/main/java/org/apache/paimon/mergetree/compact/clustering/ClusteringCompactManagerFactory.java

@@ -87,7 +87,7 @@ public CompactManager create(
             ExecutorService compactExecutor,
             List<DataFileMeta> restoreFiles,
             @Nullable BucketedDvMaintainer dvMaintainer,
-            boolean lookupEnabled) {
+            boolean ignorePreviousFiles) {
         if (options.writeOnly()) {
             return new NoopCompactManager();
         }

paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java

@@ -212,7 +212,6 @@ protected MergeTreeWriter createWriter(
                     restoreFiles);
         }
 
-        boolean lookupEnabled = !ignorePreviousFiles && options.needLookup();
         KeyValueFileWriterFactory writerFactory =
                 writerFactoryBuilder.build(partition, bucket, options);
         Comparator<InternalRow> keyComparator = keyComparatorSupplier.get();
@@ -223,7 +222,7 @@ protected MergeTreeWriter createWriter(
                         compactExecutor,
                         restoreFiles,
                         dvMaintainer,
-                        lookupEnabled);
+                        ignorePreviousFiles);
 
         return new MergeTreeWriter(
                 options.writeBufferSpillable(),

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 BatchWriteBuilderImpl(InnerTable table) {
     }
 
     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 Optional<WriteSelector> newWriteSelector() {
 
     @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 BatchTableCommit newCommit() {
     }
 
     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) {

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/LookupRemoteFileTableTest.java

@@ -268,4 +268,31 @@ public void testRemoteFileLevelThreshold() throws Exception {
         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());
+    }
 }