Skip to content

[flink] Fix batch database compaction for Data Evolution tables #10110

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JingsongLi merged 1 commit into from
Sep 24, 2026
+246 −2
Merged

[flink] Fix batch database compaction for Data Evolution tables #10110

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 32 additions & 2 deletions ...imon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import org.apache.paimon.flink.FlinkConnectorOptions;
import org.apache.paimon.flink.FlinkConnectorOptions.CompactionBucketDistributionStrategy;
import org.apache.paimon.flink.compact.AppendTableCompact;
import org.apache.paimon.flink.compact.DataEvolutionTableCompact;
import org.apache.paimon.flink.sink.BucketsRowChannelComputer;
import org.apache.paimon.flink.sink.CombinedTableCompactorSink;
import org.apache.paimon.flink.sink.CompactorSinkBuilder;
Expand All @@ -48,6 +49,7 @@
import javax.annotation.Nullable;

import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
Expand Down Expand Up @@ -147,7 +149,7 @@ public void build() {
}
}

private void buildForDividedMode() {
private void collectTables() {
try {
List<String> databases = catalog.listDatabases();
for (String databaseName : databases) {
Expand Down Expand Up @@ -181,7 +183,10 @@ private void buildForDividedMode() {
} catch (Catalog.DatabaseNotExistException | Catalog.TableNotExistException e) {
throw new RuntimeException(e);
}
}

private void buildForDividedMode() {
collectTables();
Preconditions.checkState(
!tableMap.isEmpty(),
"no tables to be compacted. possible cause is that there are no tables detected after pattern matching");
Expand All @@ -205,12 +210,30 @@ private void buildForDividedMode() {
}

private void buildForCombinedMode() {
Pattern combinedIncludingPattern = includingPattern;
if (!isStreaming) {
collectTables();
List<String> combinedTables = new ArrayList<>();
for (Map.Entry<String, FileStoreTable> entry : tableMap.entrySet()) {
if (entry.getValue().coreOptions().dataEvolutionEnabled()) {
buildForUnawareBucketCompaction(env, entry.getKey(), entry.getValue());
} else {
combinedTables.add(Pattern.quote(entry.getKey()));
}
}
if (combinedTables.isEmpty() && !tableMap.isEmpty()) {
return;
}
// Batch table selection is fixed when building the job. Data-evolution tables
// already have dedicated compactors and must not enter the shared append path.
combinedIncludingPattern = Pattern.compile(String.join("|", combinedTables));
}

CombinedTableCompactorSourceBuilder sourceBuilder =
new CombinedTableCompactorSourceBuilder(
catalogLoader(),
databasePattern,
includingPattern,
combinedIncludingPattern,
excludingPattern,
tableOptions
.get(CoreOptions.CONTINUOUS_DISCOVERY_INTERVAL)
Expand Down Expand Up @@ -285,6 +308,13 @@ private void buildForTraditionalCompaction(

private void buildForUnawareBucketCompaction(
StreamExecutionEnvironment env, String fullName, FileStoreTable table) {
if (table.coreOptions().dataEvolutionEnabled()) {
Preconditions.checkArgument(
!isStreaming, "Data evolution table compact only supports batch mode yet.");
new DataEvolutionTableCompact(env, fullName, table).build();
return;
}

AppendTableCompact unawareBucketCompactionTopoBuilder =
new AppendTableCompact(env, fullName, table);

Expand Down
214 changes: 214 additions & 0 deletions ...link-common/src/test/java/org/apache/paimon/flink/action/CompactDatabaseActionITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,14 @@
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.flink.FlinkConnectorOptions;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.paimon.table.sink.BatchTableWrite;
import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.table.sink.StreamTableCommit;
import org.apache.paimon.table.sink.StreamTableWrite;
import org.apache.paimon.table.sink.StreamWriteBuilder;
Expand Down Expand Up @@ -58,10 +64,12 @@
import java.util.List;
import java.util.Map;
import java.util.concurrent.ThreadLocalRandom;
import java.util.regex.Pattern;
import java.util.stream.Stream;

import static org.apache.paimon.utils.CommonTestUtils.waitUtil;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

/** IT cases for {@link CompactDatabaseAction}. */
public class CompactDatabaseActionITCase extends CompactActionITCaseBase {
Expand Down Expand Up @@ -297,6 +305,212 @@ public void testBatchCompact(String mode, String invoker) throws Exception {
}
}

@ParameterizedTest
@MethodSource("testData")
@Timeout(120)
public void testDataEvolutionTableCompact(String mode, String invoker) throws Exception {
List<FileStoreTable> evolutionTables =
Arrays.asList(
createDataEvolutionTable(), createDataEvolutionTable("evolution_other"));
for (FileStoreTable table : evolutionTables) {
writeDataEvolutionRecords(table);
}
FileStoreTable table = evolutionTables.get(0);
// Compact an ordinary append table in the same job to exercise mixed-table routing.
Map<String, String> options = new HashMap<>();
options.put(CoreOptions.BUCKET.key(), "-1");
options.put(CoreOptions.WRITE_ONLY.key(), "true");
options.put(CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "2");
FileStoreTable ordinaryTable =
createFileStoreTable(
"ordinary_table",
table.rowType(),
Collections.emptyList(),
Collections.emptyList(),
Collections.emptyList(),
options);
for (int i = 1; i <= 2; i++) {
BatchWriteBuilder builder = ordinaryTable.newBatchWriteBuilder();
try (BatchTableWrite write = builder.newWrite();
BatchTableCommit commit = builder.newCommit()) {
write.write(rowData(i, i * 10));
commit.commit(write.prepareCommit());
}
}

assertThat(getResult(table.newRead(), table.newScan().plan().splits(), table.rowType()))
.containsExactlyInAnyOrder("+I[1, 30]", "+I[2, 60]");

switch (invoker) {
case "action":
StreamExecutionEnvironment env =
streamExecutionEnvironmentBuilder().batchMode().build();
createAction(
CompactDatabaseAction.class,
"compact_database",
"--warehouse",
warehouse,
"--mode",
mode)
.withStreamExecutionEnvironment(env)
.build();
env.execute();
break;
case "procedure_indexed":
executeSQL(String.format("CALL sys.compact_database('', '%s')", mode), false, true)
.close();
break;
case "procedure_named":
executeSQL(
String.format("CALL sys.compact_database(mode => '%s')", mode),
false,
true)
.close();
break;
default:
throw new UnsupportedOperationException(invoker);
}

for (FileStoreTable evolutionTable : evolutionTables) {
checkLatestSnapshot(evolutionTable, 4, Snapshot.CommitKind.COMPACT);
List<DataSplit> splits = evolutionTable.newSnapshotReader().read().dataSplits();
assertThat(splits).hasSize(1);
assertThat(splits.get(0).dataFiles()).hasSize(1);
DataFileMeta compacted = splits.get(0).dataFiles().get(0);
assertThat(compacted.nonNullFirstRowId()).isZero();
assertThat(compacted.rowCount()).isEqualTo(2);
assertThat(
getResult(
evolutionTable.newRead(),
evolutionTable.newScan().plan().splits(),
evolutionTable.rowType()))
.containsExactlyInAnyOrder("+I[1, 30]", "+I[2, 60]");
}

checkLatestSnapshot(ordinaryTable, 3, Snapshot.CommitKind.COMPACT);
assertThat(ordinaryTable.newSnapshotReader().read().dataSplits().get(0).dataFiles())
.hasSize(1);
assertThat(
getResult(
ordinaryTable.newRead(),
ordinaryTable.newScan().plan().splits(),
ordinaryTable.rowType()))
.containsExactlyInAnyOrder("+I[1, 10]", "+I[2, 20]");
}

@ParameterizedTest
@ValueSource(booleans = {false, true})
public void testDataEvolutionOnlyCombinedCompact(boolean empty) throws Exception {
FileStoreTable table = createDataEvolutionTable();
if (!empty) {
writeDataEvolutionRecords(table);
}
StreamExecutionEnvironment env = streamExecutionEnvironmentBuilder().batchMode().build();
new CompactDatabaseAction(Collections.singletonMap("warehouse", warehouse))
.withDatabaseCompactMode("combined")
// Data Evolution uses the single-table compactor without idle-time filtering.
.withPartitionIdleTime(Duration.ofDays(1))
.withStreamExecutionEnvironment(env)
.build();
env.execute();
if (empty) {
assertThat(table.snapshotManager().latestSnapshotId()).isNull();
} else {
checkLatestSnapshot(table, 4, Snapshot.CommitKind.COMPACT);
assertThat(getResult(table.newRead(), table.newScan().plan().splits(), table.rowType()))
.containsExactlyInAnyOrder("+I[1, 30]", "+I[2, 60]");
}
}

@ParameterizedTest
@ValueSource(booleans = {false, true})
public void testDataEvolutionCombinedTableFilter(boolean exclude) throws Exception {
FileStoreTable selected = createDataEvolutionTable();
FileStoreTable excluded = createDataEvolutionTable("excluded_evolution");
writeDataEvolutionRecords(selected);
writeDataEvolutionRecords(excluded);
StreamExecutionEnvironment env = streamExecutionEnvironmentBuilder().batchMode().build();
CompactDatabaseAction action =
new CompactDatabaseAction(Collections.singletonMap("warehouse", warehouse))
.includingDatabases(Pattern.quote(database))
.withDatabaseCompactMode("combined");
if (exclude) {
action.excludingTables(Pattern.quote(database + ".excluded_evolution"));
} else {
action.includingTables(Pattern.quote(database + "." + tableName));
}
action.withStreamExecutionEnvironment(env).build();
env.execute();
checkLatestSnapshot(selected, 4, Snapshot.CommitKind.COMPACT);
checkLatestSnapshot(excluded, 3, Snapshot.CommitKind.APPEND);
}

@Test
public void testDataEvolutionStreamingCompactRejected() throws Exception {
createDataEvolutionTable();
CompactDatabaseAction action =
new CompactDatabaseAction(Collections.singletonMap("warehouse", warehouse));
action.withStreamExecutionEnvironment(
streamExecutionEnvironmentBuilder().streamingMode().build());
assertThatThrownBy(action::build)
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Data evolution table compact only supports batch mode");
}

private void writeDataEvolutionRecords(FileStoreTable table) throws Exception {
BatchWriteBuilder builder = table.newBatchWriteBuilder();
try (BatchTableWrite write = builder.newWrite();
BatchTableCommit commit = builder.newCommit()) {
write.write(rowData(1, 10));
write.write(rowData(2, 20));
commit.commit(write.prepareCommit());
}

for (int multiplier : new int[] {2, 3}) {
builder = table.newBatchWriteBuilder();
try (BatchTableWrite write =
builder.newWrite()
.withWriteType(
table.rowType()
.project(Collections.singletonList("v")));
BatchTableCommit commit = builder.newCommit()) {
write.write(rowData(10 * multiplier));
write.write(rowData(20 * multiplier));
List<CommitMessage> messages = write.prepareCommit();
assertThat(messages).hasSize(1);
List<DataFileMeta> files =
((CommitMessageImpl) messages.get(0)).newFilesIncrement().newFiles();
assertThat(files).hasSize(1);
assertThat(files.get(0).writeCols()).containsExactly("v");
files.replaceAll(file -> file.assignFirstRowId(0L));
commit.commit(messages);
}
}
}

private FileStoreTable createDataEvolutionTable() throws Exception {
return createDataEvolutionTable(tableName);
}

private FileStoreTable createDataEvolutionTable(String name) throws Exception {
Map<String, String> options = new HashMap<>();
options.put(CoreOptions.BUCKET.key(), "-1");
options.put(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true");
options.put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
options.put(CoreOptions.WRITE_ONLY.key(), "true");
options.put(CoreOptions.FILE_FORMAT.key(), "parquet");
options.put(CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "2");
return createFileStoreTable(
name,
RowType.of(
new DataType[] {DataTypes.INT().notNull(), DataTypes.INT()},
new String[] {"id", "v"}),
Collections.emptyList(),
Collections.emptyList(),
Collections.emptyList(),
options);
}

@ParameterizedTest(name = "mode = {0}, invoker = {1}")
@MethodSource("testData")
public void testStreamingCompact(String mode, String invoker) throws Exception {
Expand Down
Loading