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 30131ba675 [flink] Fix asynchronous partition refresh generation race
(#9318)
30131ba675 is described below
commit 30131ba675d999903a61d2a51bf235422aa4cfe7
Author: QuakeWang <[email protected]>
AuthorDate: Fri Aug 21 18:04:27 2026 +0800
[flink] Fix asynchronous partition refresh generation race (#9318)
---
.../flink/lookup/FileStoreLookupFunction.java | 3 +-
.../paimon/flink/lookup/PartitionRefresher.java | 79 ++++++++----
.../flink/lookup/PartitionRefresherTest.java | 133 +++++++++++++++++++++
3 files changed, 189 insertions(+), 26 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
index 1545a092c4..aa647f167c 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
@@ -399,8 +399,7 @@ public class FileStoreLookupFunction implements
Serializable, Closeable {
// 2. check if async partition refresh has completed, and switch if so
if (partitionRefresher != null &&
partitionRefresher.isPartitionRefreshAsync()) {
- LookupTable newLookupTable =
-
partitionRefresher.getNewLookupTable(partitionLoader.partitions());
+ LookupTable newLookupTable =
partitionRefresher.getNewLookupTable();
if (newLookupTable != null) {
lookupTable.close();
lookupTable = newLookupTable;
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PartitionRefresher.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PartitionRefresher.java
index a9caa0f6fb..b10ad23780 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PartitionRefresher.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PartitionRefresher.java
@@ -18,6 +18,7 @@
package org.apache.paimon.flink.lookup;
+import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.options.Options;
@@ -54,7 +55,7 @@ public class PartitionRefresher implements Closeable {
private final String tmpDirectory;
private volatile File path;
private ExecutorService partitionRefreshExecutor;
- private AtomicReference<LookupTable> pendingLookupTable;
+ private AtomicReference<RefreshResult> pendingRefresh;
private AtomicReference<Exception> partitionRefreshException;
/** Current partitions being used for lookup. Updated when partition
refresh completes. */
@@ -72,7 +73,7 @@ public class PartitionRefresher implements Closeable {
if (!partitionRefreshAsync) {
return;
}
- this.pendingLookupTable = new AtomicReference<>(null);
+ this.pendingRefresh = new AtomicReference<>(null);
this.partitionRefreshException = new AtomicReference<>(null);
this.partitionRefreshExecutor =
Executors.newSingleThreadExecutor(
@@ -139,14 +140,15 @@ public class PartitionRefresher implements Closeable {
partitionRefreshExecutor.submit(
() -> {
+ File newPath = new File(tmpDirectory, "lookup-" +
UUID.randomUUID());
+ FullCacheLookupTable newTable;
try {
- this.path = new File(tmpDirectory, "lookup-" +
UUID.randomUUID());
- if (!path.mkdirs()) {
- throw new RuntimeException("Failed to create dir:
" + path);
+ if (!newPath.mkdirs()) {
+ throw new RuntimeException("Failed to create dir:
" + newPath);
}
- FullCacheLookupTable.Context newContext =
context.copy(path);
+ FullCacheLookupTable.Context newContext =
context.copy(newPath);
Options options =
Options.fromMap(context.table.options());
- FullCacheLookupTable newTable =
+ newTable =
FullCacheLookupTable.create(
newContext,
options.get(LOOKUP_CACHE_ROWS));
if (cacheRowFilter != null) {
@@ -154,28 +156,43 @@ public class PartitionRefresher implements Closeable {
}
newTable.specifyPartitions(newPartitions,
partitionFilter);
newTable.open();
-
- pendingLookupTable.set(newTable);
- LOG.info("Async partition refresh completed for table
{}.", tableName);
} catch (Exception e) {
LOG.error("Async partition refresh failed for table
{}.", tableName, e);
partitionRefreshException.set(e);
- if (path != null) {
- FileIOUtils.deleteDirectoryQuietly(path);
- }
+ FileIOUtils.deleteDirectoryQuietly(newPath);
+ return;
+ }
+
+ try {
+ publishRefreshResult(newTable, newPartitions, newPath);
+ LOG.info("Async partition refresh completed for table
{}.", tableName);
+ } catch (IOException e) {
+ LOG.error(
+ "Failed to close superseded async partition
refresh for table {}.",
+ tableName,
+ e);
+ partitionRefreshException.set(e);
}
});
}
+ @VisibleForTesting
+ void publishRefreshResult(LookupTable lookupTable, List<BinaryRow>
partitions, File refreshPath)
+ throws IOException {
+ RefreshResult previous =
+ pendingRefresh.getAndSet(new RefreshResult(lookupTable,
partitions, refreshPath));
+ if (previous != null) {
+ previous.lookupTable.close();
+ }
+ }
+
/**
* Check if an async partition refresh has completed.
*
- * @param newPartitions the new partitions to update after refresh
completes
- * @return a Pair containing the new lookup table and its temp path if
ready, or null if no
- * switch is needed
+ * @return the new lookup table if ready, or null if no switch is needed
*/
@Nullable
- public LookupTable getNewLookupTable(List<BinaryRow> newPartitions) throws
Exception {
+ public LookupTable getNewLookupTable() throws Exception {
if (!partitionRefreshAsync) {
return null;
}
@@ -189,14 +206,15 @@ public class PartitionRefresher implements Closeable {
throw asyncException;
}
- LookupTable newTable = pendingLookupTable.getAndSet(null);
- if (newTable == null) {
+ RefreshResult refreshResult = pendingRefresh.getAndSet(null);
+ if (refreshResult == null) {
return null;
}
- this.currentPartitions = newPartitions;
+ this.currentPartitions = refreshResult.partitions;
+ this.path = refreshResult.path;
LOG.info("Switched to new lookup table for table {} with new
partitions.", tableName);
- return newTable;
+ return refreshResult.lookupTable;
}
/** Close partition refresh resources. */
@@ -205,10 +223,10 @@ public class PartitionRefresher implements Closeable {
if (partitionRefreshExecutor != null) {
ExecutorUtils.gracefulShutdown(1L, TimeUnit.MINUTES,
partitionRefreshExecutor);
}
- if (pendingLookupTable != null) {
- LookupTable pending = pendingLookupTable.getAndSet(null);
+ if (pendingRefresh != null) {
+ RefreshResult pending = pendingRefresh.getAndSet(null);
if (pending != null) {
- pending.close();
+ pending.lookupTable.close();
}
}
}
@@ -220,4 +238,17 @@ public class PartitionRefresher implements Closeable {
public File path() {
return path;
}
+
+ private static final class RefreshResult {
+
+ private final LookupTable lookupTable;
+ private final List<BinaryRow> partitions;
+ private final File path;
+
+ private RefreshResult(LookupTable lookupTable, List<BinaryRow>
partitions, File path) {
+ this.lookupTable = lookupTable;
+ this.partitions = partitions;
+ this.path = path;
+ }
+ }
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/PartitionRefresherTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/PartitionRefresherTest.java
new file mode 100644
index 0000000000..21b3e7dbde
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/PartitionRefresherTest.java
@@ -0,0 +1,133 @@
+/*
+ * 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.flink.lookup;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryRowWriter;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.utils.Filter;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link PartitionRefresher}. */
+class PartitionRefresherTest {
+
+ @TempDir private Path tempDir;
+
+ @Test
+ void testRefreshResultKeepsGenerationConsistent() throws Exception {
+ List<BinaryRow> initialPartitions = partitions(1);
+ List<BinaryRow> refreshedPartitions = partitions(2);
+ File refreshPath = tempDir.resolve("refresh-2").toFile();
+ TestingLookupTable refreshedTable = new TestingLookupTable();
+ PartitionRefresher refresher = createRefresher(initialPartitions);
+
+ try {
+ refresher.publishRefreshResult(refreshedTable,
refreshedPartitions, refreshPath);
+
+ assertThat(refresher.getNewLookupTable()).isSameAs(refreshedTable);
+
assertThat(refresher.currentPartitions()).isSameAs(refreshedPartitions);
+ assertThat(refresher.path()).isEqualTo(refreshPath);
+ assertThat(refreshedTable.closed).isFalse();
+ } finally {
+ refresher.close();
+ refreshedTable.close();
+ }
+ }
+
+ @Test
+ void testPublishingNewResultClosesUnconsumedTable() throws Exception {
+ List<BinaryRow> partitionsB = partitions(2);
+ List<BinaryRow> partitionsC = partitions(3);
+ File pathB = tempDir.resolve("refresh-2").toFile();
+ File pathC = tempDir.resolve("refresh-3").toFile();
+ TestingLookupTable tableB = new TestingLookupTable();
+ TestingLookupTable tableC = new TestingLookupTable();
+ PartitionRefresher refresher = createRefresher(partitions(1));
+
+ try {
+ refresher.publishRefreshResult(tableB, partitionsB, pathB);
+ refresher.publishRefreshResult(tableC, partitionsC, pathC);
+
+ assertThat(tableB.closed).isTrue();
+ assertThat(tableC.closed).isFalse();
+ assertThat(refresher.getNewLookupTable()).isSameAs(tableC);
+ assertThat(refresher.currentPartitions()).isSameAs(partitionsC);
+ assertThat(refresher.path()).isEqualTo(pathC);
+ } finally {
+ refresher.close();
+ tableB.close();
+ tableC.close();
+ }
+ }
+
+ private PartitionRefresher createRefresher(List<BinaryRow>
initialPartitions) {
+ return new PartitionRefresher(true, "table", tempDir.toString(),
initialPartitions);
+ }
+
+ private static List<BinaryRow> partitions(int value) {
+ BinaryRow row = new BinaryRow(1);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ writer.writeInt(0, value);
+ writer.complete();
+ return Collections.singletonList(row);
+ }
+
+ private static class TestingLookupTable implements LookupTable {
+
+ private boolean closed;
+
+ @Override
+ public void specifyPartitions(List<BinaryRow> scanPartitions,
Predicate partitionFilter) {}
+
+ @Override
+ public void open() {}
+
+ @Override
+ public List<InternalRow> get(InternalRow key) {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public void refresh() {}
+
+ @Override
+ public void specifyCacheRowFilter(Filter<InternalRow> filter) {}
+
+ @Override
+ public Long nextSnapshotId() {
+ return null;
+ }
+
+ @Override
+ public void close() throws IOException {
+ closed = true;
+ }
+ }
+}
