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 2eaf657691 [flink] Do not fail partition_idle_time compaction on
partitions written after the job is built (#10612)
2eaf657691 is described below
commit 2eaf657691602d04e458f90f9c656db5def79bb0
Author: YangJie <[email protected]>
AuthorDate: Sun Oct 18 04:51:44 2026 -0400
[flink] Do not fail partition_idle_time compaction on partitions written
after the job is built (#10612)
---
.../paimon/flink/compact/AppendTableCompact.java | 7 ++-
.../flink/source/CompactorSourceBuilder.java | 6 ++-
.../paimon/flink/action/CompactActionITCase.java | 61 ++++++++++++++++++++++
3 files changed, 71 insertions(+), 3 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/AppendTableCompact.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/AppendTableCompact.java
index a9037e8428..c51f310d3e 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/AppendTableCompact.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/AppendTableCompact.java
@@ -101,8 +101,11 @@ public class AppendTableCompact {
SingleOutputStreamOperator<AppendCompactTask> filterStream =
source.filter(
task -> {
- BinaryRow partition = task.partition();
- return partitionInfo.get(partition) <=
historyMilli;
+ // a partition missing from the map was first
written after the job
+ // was built
+ Long lastFileCreationTime =
partitionInfo.get(task.partition());
+ return lastFileCreationTime != null
+ && lastFileCreationTime <=
historyMilli;
});
source = new DataStreamSource<>(filterStream);
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java
index 9bc38ebbbb..d72f263375 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java
@@ -169,7 +169,11 @@ public class CompactorSourceBuilder {
dataStream.filter(
rowData -> {
BinaryRow partition =
deserializeBinaryRow(rowData.getBinary(1));
- return partitionInfo.get(partition) <=
historyMilli;
+ // a partition missing from the map was first
written after the job
+ // was built
+ Long lastFileCreationTime =
partitionInfo.get(partition);
+ return lastFileCreationTime != null
+ && lastFileCreationTime <=
historyMilli;
});
if (parallelism != null) {
filterStream.setParallelism(parallelism);
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactActionITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactActionITCase.java
index d1bcf1db3a..feb84676df 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactActionITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactActionITCase.java
@@ -313,6 +313,67 @@ public class CompactActionITCase extends
CompactActionITCaseBase {
}
}
+ @ParameterizedTest(name = "mode = {0}")
+ @ValueSource(booleans = {true, false})
+ @Timeout(60)
+ public void
testHistoryPartitionCompactWithPartitionAddedAfterBuild(boolean mode)
+ throws Exception {
+ FileStoreTable table;
+ if (mode) {
+ table =
+ prepareTable(
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ Collections.emptyList(),
+
Collections.singletonMap(CoreOptions.WRITE_ONLY.key(), "true"));
+ } else {
+ // for unaware bucket table
+ Map<String, String> tableOptions = new HashMap<>();
+ tableOptions.put(CoreOptions.BUCKET.key(), "-1");
+ tableOptions.put(CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "2");
+
+ table =
+ prepareTable(
+ Arrays.asList("dt", "hh"),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ tableOptions);
+ }
+
+ writeData(rowData(1, 100, 15, BinaryString.fromString("20221208")));
+ writeData(rowData(2, 100, 15, BinaryString.fromString("20221208")));
+ Thread.sleep(5000);
+
+ CompactAction action =
+ createAction(
+ CompactAction.class,
+ "compact",
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ tableName,
+ "--partition_idle_time",
+ "5s");
+ StreamExecutionEnvironment env =
streamExecutionEnvironmentBuilder().batchMode().build();
+ action.withStreamExecutionEnvironment(env).build();
+
+ // the job is built, the partitions are planned when it runs
+ writeData(rowData(1, 100, 15, BinaryString.fromString("20221210")));
+ writeData(rowData(2, 100, 15, BinaryString.fromString("20221210")));
+ env.execute();
+
+ checkLatestSnapshot(table, 5, Snapshot.CommitKind.COMPACT);
+ List<DataSplit> splits = table.newSnapshotReader().read().dataSplits();
+ assertThat(splits).hasSize(2);
+ for (DataSplit split : splits) {
+ String dt = split.partition().getString(0).toString();
+ // the idle partition is compacted, the one written after the
build is not
+ assertThat(split.dataFiles()).hasSize(dt.equals("20221208") ? 1 :
2);
+ }
+ }
+
@Test
public void testStreamingCompactWithChangedExternalPath() throws Exception
{
String externalPath1 = getTempDirPath();
