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 efed7dbb0d [core][flink] Add option to skip expired partitions in
compaction job (#7537)
efed7dbb0d is described below
commit efed7dbb0d8eef775fffbbb9b1871a31012d9176
Author: sanshi <[email protected]>
AuthorDate: Tue Jun 23 18:58:41 2026 +0800
[core][flink] Add option to skip expired partitions in compaction job
(#7537)
Add a new option `compaction.skip-expired-partitions` (default: false) that
allows the Flink compaction job to skip partitions already expired under the
`values-time` partition expiration strategy, avoiding unnecessary compaction
work on soon-to-be-deleted dataPlease continue.
---
docs/docs/maintenance/dedicated-compaction.mdx | 90 +++++++++++++++
docs/generated/core_configuration.html | 42 ++++---
.../main/java/org/apache/paimon/CoreOptions.java | 35 ++++++
.../paimon/partition/PartitionExpireStrategy.java | 5 +-
.../paimon/partition/PartitionTimeExtractor.java | 5 +-
.../PartitionUpdateTimeExpireStrategy.java | 2 +
.../PartitionValuesTimeExpireStrategy.java | 2 +
.../flink/source/CompactorSourceBuilder.java | 21 ++++
.../paimon/flink/action/CompactActionITCase.java | 127 +++++++++++++++++++++
9 files changed, 309 insertions(+), 20 deletions(-)
diff --git a/docs/docs/maintenance/dedicated-compaction.mdx
b/docs/docs/maintenance/dedicated-compaction.mdx
index 70a1d68740..7695b09970 100644
--- a/docs/docs/maintenance/dedicated-compaction.mdx
+++ b/docs/docs/maintenance/dedicated-compaction.mdx
@@ -450,4 +450,94 @@ Example: compact historical partitions for tables in
database
</TabItem>
+</Tabs>
+
+## Skip Expired Partitions in Compaction Job
+
+When [partition expiration](./manage-partitions#expiring-partitions) is
configured with
+`partition.expiration-strategy = values-time`, a compaction job may waste
resources compacting
+partitions that are already expired and will soon be deleted.
+
+You can set `compaction.skip-expired-partitions = true` to make the compaction
job skip these
+already-expired partitions, avoiding unnecessary compaction work on data that
will be deleted shortly.
+
+:::info
+
+This option requires `partition.expiration-strategy = values-time` to be
configured on the table.
+It only supports primary key tables with fixed or dynamic bucket mode. Unaware
bucket (append-only) tables
+and `compact_database` in `combined` mode are not supported — the option will
be silently ignored in those cases.
+
+Even when this option is enabled, expired partitions may still be deleted
during the compaction commit
+phase as a side effect of partition expiration triggered by committing the
remaining active partitions.
+This is expected behavior — the option avoids unnecessary compaction IO on
expired partitions, but does
+not prevent those partitions from being expired.
+
+:::
+
+### For Table
+
+<Tabs groupId="skip-expired-partitions-compact">
+
+<TabItem value="flink-sql" label="Flink SQL">
+
+Run the following sql:
+
+```sql
+-- skip expired partitions compact table
+CALL sys.compact(`table` => 'default.T', options =>
'compaction.skip-expired-partitions=true')
+```
+
+</TabItem>
+
+<TabItem value="flink-action-jar" label="Flink Action Jar">
+
+```bash
+<FLINK_HOME>/bin/flink run \
+ /path/to/paimon-flink-action-@@VERSION@@.jar \
+ compact \
+ --warehouse <warehouse-path> \
+ --database <database-name> \
+ --table <table-name> \
+ --table_conf compaction.skip-expired-partitions=true \
+ [--catalog_conf <paimon-catalog-conf> [--catalog_conf
<paimon-catalog-conf> ...]]
+```
+
+</TabItem>
+
+</Tabs>
+
+### For Databases
+
+<Tabs groupId="skip-expired-partitions-compact-database">
+
+<TabItem value="flink-sql" label="Flink SQL">
+
+Run the following sql:
+
+```sql
+-- skip expired partitions compact database
+CALL sys.compact_database(
+ including_databases => 'includingDatabases',
+ mode => 'divided',
+ table_options => 'compaction.skip-expired-partitions=true'
+)
+```
+
+</TabItem>
+
+<TabItem value="flink-action-jar" label="Flink Action Jar">
+
+```bash
+<FLINK_HOME>/bin/flink run \
+ /path/to/paimon-flink-action-@@VERSION@@.jar \
+ compact_database \
+ --warehouse <warehouse-path> \
+ --including_databases <database-name|name-regular-expr> \
+ --mode divided \
+ --table_conf compaction.skip-expired-partitions=true \
+ [--catalog_conf <paimon-catalog-conf> [--catalog_conf
<paimon-catalog-conf> ...]]
+```
+
+</TabItem>
+
</Tabs>
\ No newline at end of file
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index 9d1a962463..f525a00b64 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -404,6 +404,12 @@ under the License.
<td>Integer</td>
<td>Percentage flexibility while comparing sorted run size for
changelog mode table. If the candidate sorted run(s) size is 1% smaller than
the next sorted run's size, then include next sorted run into this candidate
set.</td>
</tr>
+ <tr>
+ <td><h5>compaction.skip-expired-partitions</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to skip compacting partitions that are already expired
according to 'partition.expiration-time'. Only effective when
'partition.expiration-time' is set and 'partition.expiration-strategy' is
'values-time'. Note: even when this option is enabled, expired partitions may
still be deleted during the compaction commit phase as a side effect of
partition expiration triggered by committing the remaining active
partitions.</td>
+ </tr>
<tr>
<td><h5>compaction.small-file-ratio</h5></td>
<td style="word-wrap: break-word;">0.7</td>
@@ -909,6 +915,24 @@ Mainly to resolve data skew on primary keys. We recommend
starting with 64 mb wh
<td>Integer</td>
<td>Level threshold of lookup to generate remote lookup files.
Level files below this threshold will not generate remote lookup files.</td>
</tr>
+ <tr>
+ <td><h5>manifest-sort.enabled</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to invoke manifest sort rewrite during commit.<br
/>Note: enabling this changes the semantics of 'manifest.merge-min-count'. In
the sort rewrite path, small manifest files within the rewrite budget are
sorted and merged directly, so the minimum-count gate no longer prevents
merging a small number of under-budget manifest files when full compaction is
not triggered.</td>
+ </tr>
+ <tr>
+ <td><h5>manifest-sort.max-rewrite-size</h5></td>
+ <td style="word-wrap: break-word;">256 mb</td>
+ <td>MemorySize</td>
+ <td>Maximum total size of manifest files to rewrite in a single
sort rewrite pass. Sections exceeding this limit are skipped. Set to a larger
value to allow more aggressive sort rewriting. The cap only limits the sorted
rewrite portion and full/minor cleanup may still happen beyond it.</td>
+ </tr>
+ <tr>
+ <td><h5>manifest-sort.partition-field</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Partition field name to sort manifest entries by. Validated by
schema validation, if not configured, defaults to the first partition
field.</td>
+ </tr>
<tr>
<td><h5>manifest.compression</h5></td>
<td style="word-wrap: break-word;">"zstd"</td>
@@ -939,24 +963,6 @@ Mainly to resolve data skew on primary keys. We recommend
starting with 64 mb wh
<td>Integer</td>
<td>To avoid frequent manifest merges, this parameter specifies
the minimum number of ManifestFileMeta to merge.<br />Note: when
'manifest-sort.enabled' is true, this minimum-count gate is only applied to the
trailing sub-segment of a section that exceeds
'manifest-sort.max-rewrite-size'. Small under-budget sections are sorted and
rewritten directly, so two small manifest files may be merged into one even
when their count is below this threshold and full compaction is not tri [...]
</tr>
- <tr>
- <td><h5>manifest-sort.enabled</h5></td>
- <td style="word-wrap: break-word;">false</td>
- <td>Boolean</td>
- <td>Whether to invoke manifest sort rewrite during commit.<br
/>Note: enabling this changes the semantics of 'manifest.merge-min-count'. In
the sort rewrite path, small manifest files within the rewrite budget are
sorted and merged directly, so the minimum-count gate no longer prevents
merging a small number of under-budget manifest files when full compaction is
not triggered.</td>
- </tr>
- <tr>
- <td><h5>manifest-sort.partition-field</h5></td>
- <td style="word-wrap: break-word;">(none)</td>
- <td>String</td>
- <td>Partition field name to sort manifest entries by. Validated by
schema validation, if not configured, defaults to the first partition
field.</td>
- </tr>
- <tr>
- <td><h5>manifest-sort.max-rewrite-size</h5></td>
- <td style="word-wrap: break-word;">256 mb</td>
- <td>MemorySize</td>
- <td>Maximum total size of manifest files to rewrite in a single
sort rewrite pass. Sections exceeding this limit are skipped. Set to a larger
value to allow more aggressive sort rewriting. The cap only limits the sorted
rewrite portion and full/minor cleanup may still happen beyond it.</td>
- </tr>
<tr>
<td><h5>manifest.target-file-size</h5></td>
<td style="word-wrap: break-word;">8 mb</td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index bf85dd66dc..5eb86a56bc 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -1221,6 +1221,23 @@ public class CoreOptions implements Serializable {
"Whether only overwrite dynamic partition when
overwriting a partitioned table with "
+ "dynamic partition columns. Works only
when the table has partition keys.");
+ /** The strategy for partition expiration. */
+ public enum PartitionExpireStrategy {
+ VALUES_TIME("values-time"),
+ UPDATE_TIME("update-time");
+
+ private final String value;
+
+ PartitionExpireStrategy(String value) {
+ this.value = value;
+ }
+
+ @Override
+ public String toString() {
+ return value;
+ }
+ }
+
public static final ConfigOption<String> PARTITION_EXPIRATION_STRATEGY =
key("partition.expiration-strategy")
.stringType()
@@ -1266,6 +1283,20 @@ public class CoreOptions implements Serializable {
+ "By default, all partitions to be
expired will be expired together, which may cause a risk of out-of-memory. "
+ "Use this parameter to divide partition
expiration process and mitigate memory pressure.");
+ public static final ConfigOption<Boolean>
COMPACTION_SKIP_EXPIRED_PARTITIONS =
+ key("compaction.skip-expired-partitions")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to skip compacting partitions that are
already expired "
+ + "according to
'partition.expiration-time'. "
+ + "Only effective when
'partition.expiration-time' is set "
+ + "and 'partition.expiration-strategy' is
'values-time'. "
+ + "Note: even when this option is enabled,
expired partitions "
+ + "may still be deleted during the
compaction commit phase "
+ + "as a side effect of partition
expiration triggered by "
+ + "committing the remaining active
partitions.");
+
public static final ConfigOption<String> PARTITION_TIMESTAMP_FORMATTER =
key("partition.timestamp-formatter")
.stringType()
@@ -3597,6 +3628,10 @@ public class CoreOptions implements Serializable {
return options.get(PARTITION_EXPIRATION_STRATEGY);
}
+ public boolean compactionSkipExpiredPartitions() {
+ return options.get(COMPACTION_SKIP_EXPIRED_PARTITIONS);
+ }
+
@Nullable
public String dataFileExternalPaths() {
return options.get(DATA_FILE_EXTERNAL_PATHS);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionExpireStrategy.java
b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionExpireStrategy.java
index 0921a65697..4371fe86fa 100644
---
a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionExpireStrategy.java
+++
b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionExpireStrategy.java
@@ -29,6 +29,7 @@ import org.apache.paimon.utils.RowDataToObjectArrayConverter;
import javax.annotation.Nullable;
+import java.io.Serializable;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.LinkedHashMap;
@@ -37,7 +38,9 @@ import java.util.Map;
import java.util.Optional;
/** Strategy for partition expiration. */
-public abstract class PartitionExpireStrategy {
+public abstract class PartitionExpireStrategy implements Serializable {
+
+ private static final long serialVersionUID = 1L;
protected final List<String> partitionKeys;
protected final String partitionDefaultName;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionTimeExtractor.java
b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionTimeExtractor.java
index 0016619e21..afc58c8513 100644
---
a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionTimeExtractor.java
+++
b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionTimeExtractor.java
@@ -20,6 +20,7 @@ package org.apache.paimon.partition;
import javax.annotation.Nullable;
+import java.io.Serializable;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
@@ -43,7 +44,9 @@ import static java.time.temporal.ChronoField.SECOND_OF_MINUTE;
import static java.time.temporal.ChronoField.YEAR;
/** Time extractor to extract time from partition values. */
-public class PartitionTimeExtractor {
+public class PartitionTimeExtractor implements Serializable {
+
+ private static final long serialVersionUID = 1L;
private static final DateTimeFormatter TIMESTAMP_FORMATTER =
new DateTimeFormatterBuilder()
diff --git
a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionUpdateTimeExpireStrategy.java
b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionUpdateTimeExpireStrategy.java
index 3cb7a405d2..628f4c3733 100644
---
a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionUpdateTimeExpireStrategy.java
+++
b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionUpdateTimeExpireStrategy.java
@@ -34,6 +34,8 @@ import java.util.stream.Collectors;
*/
public class PartitionUpdateTimeExpireStrategy extends PartitionExpireStrategy
{
+ private static final long serialVersionUID = 1L;
+
public PartitionUpdateTimeExpireStrategy(CoreOptions options, RowType
partitionType) {
super(partitionType, options.partitionDefaultName());
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionValuesTimeExpireStrategy.java
b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionValuesTimeExpireStrategy.java
index 94e26d6f37..6238817da4 100644
---
a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionValuesTimeExpireStrategy.java
+++
b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionValuesTimeExpireStrategy.java
@@ -45,6 +45,8 @@ import static
org.apache.paimon.CoreOptions.PARTITION_EXPIRATION_STRATEGY;
*/
public class PartitionValuesTimeExpireStrategy extends PartitionExpireStrategy
{
+ private static final long serialVersionUID = 1L;
+
private static final Logger LOG =
LoggerFactory.getLogger(PartitionValuesTimeExpireStrategy.class);
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 d4edb5cbaa..83486a8624 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
@@ -25,6 +25,7 @@ import org.apache.paimon.flink.LogicalTypeConversion;
import org.apache.paimon.manifest.PartitionEntry;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.partition.PartitionValuesTimeExpireStrategy;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.system.CompactBucketsTable;
@@ -141,6 +142,26 @@ public class CompactorSourceBuilder {
});
dataStream = new DataStreamSource<>(filterStream);
}
+ CoreOptions coreOptions = table.coreOptions();
+ if (coreOptions.compactionSkipExpiredPartitions()
+ && coreOptions.partitionExpireTime() != null
+ && CoreOptions.PartitionExpireStrategy.VALUES_TIME
+ .toString()
+ .equals(coreOptions.partitionExpireStrategy())) {
+ RowType partitionType = table.schema().logicalPartitionType();
+ Duration expireTime = coreOptions.partitionExpireTime();
+ PartitionValuesTimeExpireStrategy expireStrategy =
+ new PartitionValuesTimeExpireStrategy(coreOptions,
partitionType);
+ SingleOutputStreamOperator<RowData> filterStream =
+ dataStream.filter(
+ rowData -> {
+ LocalDateTime expireDateTime =
+ LocalDateTime.now().minus(expireTime);
+ BinaryRow partition =
deserializeBinaryRow(rowData.getBinary(1));
+ return
!expireStrategy.isExpired(expireDateTime, partition);
+ });
+ dataStream = new DataStreamSource<>(filterStream);
+ }
Integer parallelism =
Options.fromMap(table.options()).get(FlinkConnectorOptions.SCAN_PARALLELISM);
if (parallelism != null) {
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 646a20ac70..fedd98ceea 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
@@ -62,6 +62,8 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import java.time.Duration;
+import java.time.LocalDate;
+import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -930,6 +932,131 @@ public class CompactActionITCase extends
CompactActionITCaseBase {
assertThat(value).isEqualTo(30000);
}
+ @Test
+ @Timeout(60)
+ public void testSkipExpiredPartitions() throws Exception {
+ // Use a date far in the past (expired) and today's date (not expired)
+ String expiredDt =
+
LocalDate.now().minusDays(30).format(DateTimeFormatter.ofPattern("yyyyMMdd"));
+ String activeDt =
LocalDate.now().format(DateTimeFormatter.ofPattern("yyyyMMdd"));
+
+ Map<String, String> tableOptions = new HashMap<>();
+ tableOptions.put(CoreOptions.WRITE_ONLY.key(), "true");
+ tableOptions.put(CoreOptions.PARTITION_EXPIRATION_TIME.key(), "7 d");
+ tableOptions.put(
+ CoreOptions.PARTITION_EXPIRATION_STRATEGY.key(),
+ CoreOptions.PartitionExpireStrategy.VALUES_TIME.toString());
+ tableOptions.put(CoreOptions.PARTITION_TIMESTAMP_FORMATTER.key(),
"yyyyMMdd");
+ // Prevent partition expiration from running during the compact commit.
+
tableOptions.put(CoreOptions.PARTITION_EXPIRATION_CHECK_INTERVAL.key(), "999
d");
+
+ FileStoreTable table =
+ prepareTable(
+ Collections.singletonList("dt"),
+ Arrays.asList("dt", "k"),
+ Collections.emptyList(),
+ tableOptions);
+
+ // Write two batches to each partition so each has multiple files
+ writeData(
+ rowData(1, 100, 15, BinaryString.fromString(expiredDt)),
+ rowData(1, 100, 15, BinaryString.fromString(activeDt)));
+
+ writeData(
+ rowData(2, 100, 15, BinaryString.fromString(expiredDt)),
+ rowData(2, 100, 15, BinaryString.fromString(activeDt)));
+
+ checkLatestSnapshot(table, 2, Snapshot.CommitKind.APPEND);
+
+ CompactAction action =
+ createAction(
+ CompactAction.class,
+ "compact",
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ tableName,
+ "--table_conf",
+ CoreOptions.COMPACTION_SKIP_EXPIRED_PARTITIONS.key() +
"=true");
+ StreamExecutionEnvironment env =
streamExecutionEnvironmentBuilder().batchMode().build();
+ action.withStreamExecutionEnvironment(env).build();
+ env.execute();
+
+ checkLatestSnapshot(table, 3, Snapshot.CommitKind.COMPACT);
+
+ List<DataSplit> splits = table.newSnapshotReader().read().dataSplits();
+ for (DataSplit split : splits) {
+ String dt = split.partition().getString(0).toString();
+ if (dt.equals(activeDt)) {
+ // active partition should be compacted into 1 file
+ assertThat(split.dataFiles().size()).isEqualTo(1);
+ } else {
+ // expired partition should be skipped, still has 2 files
+ assertThat(split.dataFiles().size()).isEqualTo(2);
+ }
+ }
+ }
+
+ @Test
+ @Timeout(60)
+ public void testNotSkipExpiredPartitionsByDefault() throws Exception {
+ String expiredDt =
+
LocalDate.now().minusDays(30).format(DateTimeFormatter.ofPattern("yyyyMMdd"));
+ String activeDt =
LocalDate.now().format(DateTimeFormatter.ofPattern("yyyyMMdd"));
+
+ Map<String, String> tableOptions = new HashMap<>();
+ tableOptions.put(CoreOptions.WRITE_ONLY.key(), "true");
+ tableOptions.put(CoreOptions.PARTITION_EXPIRATION_TIME.key(), "7 d");
+ tableOptions.put(
+ CoreOptions.PARTITION_EXPIRATION_STRATEGY.key(),
+ CoreOptions.PartitionExpireStrategy.VALUES_TIME.toString());
+ tableOptions.put(CoreOptions.PARTITION_TIMESTAMP_FORMATTER.key(),
"yyyyMMdd");
+ // COMPACTION_SKIP_EXPIRED_PARTITIONS is not set, default is false
+ // Prevent partition expiration from running during the compact commit.
+
tableOptions.put(CoreOptions.PARTITION_EXPIRATION_CHECK_INTERVAL.key(), "999
d");
+
+ FileStoreTable table =
+ prepareTable(
+ Collections.singletonList("dt"),
+ Arrays.asList("dt", "k"),
+ Collections.emptyList(),
+ tableOptions);
+
+ writeData(
+ rowData(1, 100, 15, BinaryString.fromString(expiredDt)),
+ rowData(1, 100, 15, BinaryString.fromString(activeDt)));
+
+ writeData(
+ rowData(2, 100, 15, BinaryString.fromString(expiredDt)),
+ rowData(2, 100, 15, BinaryString.fromString(activeDt)));
+
+ checkLatestSnapshot(table, 2, Snapshot.CommitKind.APPEND);
+
+ CompactAction action =
+ createAction(
+ CompactAction.class,
+ "compact",
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ tableName);
+ StreamExecutionEnvironment env =
streamExecutionEnvironmentBuilder().batchMode().build();
+ action.withStreamExecutionEnvironment(env).build();
+ env.execute();
+
+ checkLatestSnapshot(table, 3, Snapshot.CommitKind.COMPACT);
+
+ // both expired and active partitions should be compacted into 1 file
+ List<DataSplit> splits = table.newSnapshotReader().read().dataSplits();
+ for (DataSplit split : splits) {
+ assertThat(split.dataFiles().size()).isEqualTo(1);
+ }
+ }
+
private void setFirstRowId(List<CommitMessage> commitables, long
firstRowId) {
commitables.forEach(
c -> {
