Skip to content

[FLINK-38504][table] Add restore tests for delta join #27117

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

Open
wants to merge 3 commits into
base: master
Choose a base branch
from
+5,792 −37
Open

[FLINK-38504][table] Add restore tests for delta join #27117

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0404af8
[FLINK-38504][table] Add restore tests for delta join
xuyangzhong Oct 14, 2025
d2c0332
add restore test for calc on source
xuyangzhong Oct 30, 2025
13dc49f
address comment
xuyangzhong Oct 31, 2025
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
86 changes: 71 additions & 15 deletions .../flink-table-api-java/src/test/java/org/apache/flink/table/test/program/SinkTestStep.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
Expand All @@ -42,6 +43,7 @@ public final class SinkTestStep extends TableTestStep {
public final @Nullable List<Row> expectedMaterializedRows;
public final @Nullable List<String> expectedMaterializedStrings;
public final boolean testChangelogData;
public final @Nullable int[] deduplicatedFieldIndices;

SinkTestStep(
String name,
Expand All @@ -55,17 +57,16 @@ public final class SinkTestStep extends TableTestStep {
@Nullable List<String> expectedAfterRestoreStrings,
@Nullable List<Row> expectedMaterializedRows,
@Nullable List<String> expectedMaterializedStrings,
boolean testChangelogData) {
super(name, schemaComponents, distribution, partitionKeys, options);
boolean hasRowsSet =
expectedBeforeRestore != null
|| expectedAfterRestore != null
|| expectedMaterializedRows != null;
boolean hasStringsSet =
expectedBeforeRestoreStrings != null
|| expectedAfterRestoreStrings != null
|| expectedMaterializedStrings != null;
if (hasRowsSet && hasStringsSet) {
boolean testChangelogData,
@Nullable int[] deduplicatedFieldIndices) {
super(
name,
schemaComponents,
distribution,
partitionKeys,
options,
Collections.emptyList());
if (hasRowsSet() && hasStringsSet()) {
throw new IllegalArgumentException(
"You can not mix Row/String representations in restore data.");
}
Expand All @@ -76,6 +77,12 @@ public final class SinkTestStep extends TableTestStep {
this.expectedMaterializedRows = expectedMaterializedRows;
this.expectedMaterializedStrings = expectedMaterializedStrings;
this.testChangelogData = testChangelogData;
this.deduplicatedFieldIndices = deduplicatedFieldIndices;

if (deduplicatedFieldIndices != null && !hasRowsSet()) {
throw new IllegalArgumentException(
"DeduplicatedFieldIndices can only be used for Row representations in restore data.");
}
}

/** Builder for creating a {@link SinkTestStep}. */
Expand Down Expand Up @@ -108,16 +115,42 @@ public List<String> getExpectedAfterRestoreAsStrings() {
}

public List<String> getExpectedAsStrings() {
final List<String> data = new ArrayList<>(getExpectedBeforeRestoreAsStrings());
data.addAll(getExpectedAfterRestoreAsStrings());
return data;
if (hasStringsSet() || deduplicatedFieldIndices == null) {
final List<String> data = new ArrayList<>(getExpectedBeforeRestoreAsStrings());
data.addAll(getExpectedAfterRestoreAsStrings());
return data;
}
if (!hasRowsSet()) {
return Collections.emptyList();
}
final List<Row> data = new ArrayList<>();
if (expectedBeforeRestore != null) {
data.addAll(expectedBeforeRestore);
}
if (expectedAfterRestore != null) {
data.addAll(expectedAfterRestore);
}

Map<List<Object>, Row> deduplicatedMap = new HashMap<>();
Copy link
Contributor

@Au-Miner Au-Miner Oct 31, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The logic in this section, which uses deduplicatedFieldIndices to modify expectedBeforeRestore and expectedAfterRestore, can also be implemented by calling a function in getExpectedBeforeRestoreAsStrings and getExpectedAfterRestoreAsStrings. Of course, it's up to you, but I personally find the latter easier to understand

Copy link
Contributor Author

@xuyangzhong xuyangzhong Oct 31, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Currently, for expectedBeforeRestoreStrings and expectedAfterRestoreStrings, which have transformed each row into a string and lost their schema, it is not possible to deduplicate them by column. (Perhaps one way to handle this is by removing the leading and trailing parentheses and splitting the strings by commas).
However, for now, there is no need to support this in string scenarios for test, so I haven't implemented it yet.

for (Row row : data) {
List<Object> key = new ArrayList<>(deduplicatedFieldIndices.length);
for (int deduplicatedFieldIndex : deduplicatedFieldIndices) {
key.add(row.getField(deduplicatedFieldIndex));
}
deduplicatedMap.put(key, row);
}
return deduplicatedMap.values().stream().map(Row::toString).collect(Collectors.toList());
}

public List<String> getExpectedMaterializedResultsAsStrings() {
if (expectedMaterializedStrings != null) {
return expectedMaterializedStrings;
}
if (expectedMaterializedRows != null) {
if (deduplicatedFieldIndices != null) {
throw new UnsupportedOperationException(
"Unsupported to deduplicate data for materialized rows");
}
return expectedMaterializedRows.stream()
.map(Row::toString)
.collect(Collectors.toList());
Expand All @@ -138,6 +171,18 @@ public boolean shouldTestChangelogData() {
return testChangelogData;
}

private boolean hasRowsSet() {
return expectedBeforeRestore != null
|| expectedAfterRestore != null
|| expectedMaterializedRows != null;
}

private boolean hasStringsSet() {
return expectedBeforeRestoreStrings != null
|| expectedAfterRestoreStrings != null
|| expectedMaterializedStrings != null;
}

/** Builder pattern for {@link SinkTestStep}. */
public static final class Builder extends AbstractBuilder<Builder> {

Expand All @@ -151,6 +196,8 @@ public static final class Builder extends AbstractBuilder<Builder> {

private boolean testChangelogData = true;

private @Nullable int[] deduplicatedFieldIndices;

private Builder(String name) {
super(name);
}
Expand Down Expand Up @@ -203,6 +250,14 @@ public Builder testMaterializedData() {
return this;
}

public Builder deduplicatedFieldIndices(int[] deduplicatedFieldIndices) {
// TODO FLINK-38518 use pk to deduplicate data rather than specific fields.
// This task requires refactoring the current `AbstractBuilder` to separate the
// declaration of the primary key from the `List<String> schemaComponents`.
this.deduplicatedFieldIndices = deduplicatedFieldIndices;
return this;
}

public SinkTestStep build() {
return new SinkTestStep(
name,
Expand All @@ -216,7 +271,8 @@ public SinkTestStep build() {
expectedAfterRestoreStrings,
expectedMaterializedBeforeRows,
expectedMaterializedBeforeStrings,
testChangelogData);
testChangelogData,
deduplicatedFieldIndices);
}
}
}
38 changes: 35 additions & 3 deletions ...link-table-api-java/src/test/java/org/apache/flink/table/test/program/SourceTestStep.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,24 +34,42 @@ public final class SourceTestStep extends TableTestStep {
public final List<Row> dataBeforeRestore;
public final List<Row> dataAfterRestore;

public final boolean treatDataBeforeRestoreAsConsumedData;

SourceTestStep(
String name,
List<String> schemaComponents,
@Nullable TableDistribution distribution,
List<String> partitionKeys,
Map<String, String> options,
List<List<String>> indexes,
List<Row> dataBeforeRestore,
List<Row> dataAfterRestore) {
super(name, schemaComponents, distribution, partitionKeys, options);
List<Row> dataAfterRestore,
boolean treatDataBeforeRestoreAsConsumedData) {
super(name, schemaComponents, distribution, partitionKeys, options, indexes);
this.dataBeforeRestore = dataBeforeRestore;
this.dataAfterRestore = dataAfterRestore;
this.treatDataBeforeRestoreAsConsumedData = treatDataBeforeRestoreAsConsumedData;
}

/** Builder for creating a {@link SourceTestStep}. */
public static Builder newBuilder(String name) {
return new Builder(name);
}

public SourceTestStep withNewOptions(Map<String, String> newOptions) {
return new SourceTestStep(
name,
schemaComponents,
distribution,
partitionKeys,
newOptions,
indexes,
dataBeforeRestore,
dataAfterRestore,
treatDataBeforeRestoreAsConsumedData);
}

@Override
public TestKind getKind() {
return dataBeforeRestore.isEmpty()
Expand All @@ -66,6 +84,8 @@ public static final class Builder extends AbstractBuilder<Builder> {

private final List<Row> dataBeforeRestore = new ArrayList<>();
private final List<Row> dataAfterRestore = new ArrayList<>();
private final List<List<String>> indexes = new ArrayList<>();
private boolean treatDataBeforeRestoreAsConsumedData = false;

private Builder(String name) {
super(name);
Expand All @@ -85,15 +105,27 @@ public Builder producedAfterRestore(Row... data) {
return this;
}

public Builder addIndex(String... index) {
this.indexes.add(Arrays.asList(index));
return this;
}

public Builder treatDataBeforeRestoreAsConsumedData() {
this.treatDataBeforeRestoreAsConsumedData = true;
return this;
}

public SourceTestStep build() {
return new SourceTestStep(
name,
schemaComponents,
distribution,
partitionKeys,
options,
indexes,
dataBeforeRestore,
dataAfterRestore);
dataAfterRestore,
treatDataBeforeRestoreAsConsumedData);
}
}
}
20 changes: 20 additions & 0 deletions ...nk-table-api-java/src/test/java/org/apache/flink/table/test/program/TableTestProgram.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -317,6 +317,16 @@ public Builder setupTableSource(SourceTestStep sourceTestStep) {
return this;
}

/**
* Setup steps for each table source.
*
* <p>Use {@link SourceTestStep.Builder} to construct this step.
*/
public Builder setupTableSources(List<SourceTestStep> sourceTestSteps) {
setupSteps.addAll(sourceTestSteps);
return this;
}

/**
* Setup step for a table sink.
*
Expand All @@ -327,6 +337,16 @@ public Builder setupTableSink(SinkTestStep sinkTestStep) {
return this;
}

/**
* Setup steps for each table sink.
*
* <p>Use {@link SinkTestStep.Builder} to construct this step.
*/
public Builder setupTableSinks(List<SinkTestStep> sinkTestSteps) {
setupSteps.addAll(sinkTestSteps);
return this;
}

/**
* Setup step for a model.
*
Expand Down
56 changes: 50 additions & 6 deletions ...flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TableTestStep.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,11 +20,18 @@

import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigurationUtils;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.catalog.Catalog;
import org.apache.flink.table.catalog.CatalogTable;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.catalog.TableDistribution;
import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.catalog.exceptions.TableNotExistException;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.types.RowKind;
import org.apache.flink.util.Preconditions;

import javax.annotation.Nullable;

Expand Down Expand Up @@ -58,25 +65,28 @@ public abstract class TableTestStep implements TestStep {
public final @Nullable TableDistribution distribution;
public final List<String> partitionKeys;
public final Map<String, String> options;
public final List<List<String>> indexes;

TableTestStep(
String name,
List<String> schemaComponents,
@Nullable TableDistribution distribution,
List<String> partitionKeys,
Map<String, String> options) {
Map<String, String> options,
List<List<String>> indexes) {
this.name = name;
this.schemaComponents = schemaComponents;
this.distribution = distribution;
this.partitionKeys = partitionKeys;
this.options = options;
this.indexes = indexes;
}

public TableResult apply(TableEnvironment env) {
return apply(env, Collections.emptyMap());
public void apply(TableEnvironment env) {
apply(env, Collections.emptyMap());
}

public TableResult apply(TableEnvironment env, Map<String, String> extraOptions) {
public void apply(TableEnvironment env, Map<String, String> extraOptions) {
final Map<String, String> allOptions = new HashMap<>(options);
allOptions.putAll(extraOptions);

Expand All @@ -97,7 +107,41 @@ public TableResult apply(TableEnvironment env, Map<String, String> extraOptions)
.map(e -> String.format("'%s'='%s'", e.getKey(), e.getValue()))
.collect(Collectors.joining(",\n")));

return env.executeSql(createTable);
env.executeSql(createTable);
if (indexes.isEmpty()) {
return;
}

Optional<Catalog> currentCatalogOp = env.getCatalog(env.getCurrentCatalog());
Preconditions.checkState(currentCatalogOp.isPresent());
Catalog catalog = currentCatalogOp.get();

String currentDatabaseName = env.getCurrentDatabase();
ObjectPath tablePath = new ObjectPath(currentDatabaseName, name);
CatalogTable oldTable;
try {
oldTable = (CatalogTable) catalog.getTable(tablePath);
catalog.dropTable(tablePath, false);
} catch (TableNotExistException e) {
throw new IllegalStateException(e);
}
Schema schema = oldTable.getUnresolvedSchema();
Schema.Builder schemaBuilder = Schema.newBuilder().fromSchema(schema);
indexes.forEach(schemaBuilder::index);
CatalogTable newTable =
CatalogTable.newBuilder()
.schema(schemaBuilder.build())
.comment(oldTable.getComment())
.partitionKeys(oldTable.getPartitionKeys())
.options(oldTable.getOptions())
.snapshot(oldTable.getSnapshot().orElse(null))
.distribution(oldTable.getDistribution().orElse(null))
.build();
try {
catalog.createTable(tablePath, newTable, false);
} catch (TableAlreadyExistException | DatabaseNotExistException e) {
throw new IllegalStateException(e);
}
}

/** Builder pattern for {@link SourceTestStep} and {@link SinkTestStep}. */
Expand Down
Loading