This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-table-store.git
The following commit(s) were added to refs/heads/master by this push:
new 8e63f1f6 [FLINK-29554] Add partial-update.ignore-delete option to
avoid exception after join
8e63f1f6 is described below
commit 8e63f1f67fcfe83773a7264068687c7340bc01cb
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Oct 13 19:52:38 2022 +0800
[FLINK-29554] Add partial-update.ignore-delete option to avoid exception
after join
This closes #311
---
.../shortcodes/generated/core_configuration.html | 6 +++
.../table/store/connector/PartialUpdateITCase.java | 57 ++++++++++++++++++++--
.../org/apache/flink/table/store/CoreOptions.java | 6 +++
.../compact/PartialUpdateMergeFunction.java | 37 +++++++++-----
.../table/ChangelogWithKeyFileStoreTable.java | 3 +-
.../table/ChangelogWithKeyFileStoreTableTest.java | 25 ++++++++++
6 files changed, 119 insertions(+), 15 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/core_configuration.html
b/docs/layouts/shortcodes/generated/core_configuration.html
index 81aa677d..6e6da2bd 100644
--- a/docs/layouts/shortcodes/generated/core_configuration.html
+++ b/docs/layouts/shortcodes/generated/core_configuration.html
@@ -170,6 +170,12 @@
<td>MemorySize</td>
<td>Memory page size.</td>
</tr>
+ <tr>
+ <td><h5>partial-update.ignore-delete</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to ignore delete records in partial-update mode.</td>
+ </tr>
<tr>
<td><h5>partition.default-name</h5></td>
<td style="word-wrap: break-word;">"__DEFAULT_PARTITION__"</td>
diff --git
a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/PartialUpdateITCase.java
b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/PartialUpdateITCase.java
index 2ec08e17..bd36bd55 100644
---
a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/PartialUpdateITCase.java
+++
b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/PartialUpdateITCase.java
@@ -18,12 +18,17 @@
package org.apache.flink.table.store.connector;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+import org.apache.flink.util.CloseableIterator;
import org.junit.Test;
-import java.util.Collections;
+import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
+import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -33,10 +38,15 @@ public class PartialUpdateITCase extends
FileStoreTableITCase {
@Override
protected List<String> ddl() {
- return Collections.singletonList(
+ return Arrays.asList(
"CREATE TABLE IF NOT EXISTS T ("
+ "j INT, k INT, a INT, b INT, c STRING, PRIMARY KEY
(j,k) NOT ENFORCED)"
- + " WITH ('merge-engine'='partial-update');");
+ + " WITH ('merge-engine'='partial-update');",
+ "CREATE TABLE IF NOT EXISTS dwd_orders ("
+ + "OrderID INT, OrderNumber INT, PersonID INT,
LastName STRING, FirstName STRING, Age INT, PRIMARY KEY (OrderID) NOT ENFORCED)"
+ + " WITH ('merge-engine'='partial-update',
'partial-update.ignore-delete'='true');",
+ "CREATE TABLE IF NOT EXISTS ods_orders (OrderID INT,
OrderNumber INT, PersonID INT, PRIMARY KEY (OrderID) NOT ENFORCED) WITH
('changelog-producer'='input');",
+ "CREATE TABLE IF NOT EXISTS dim_persons (PersonID INT,
LastName STRING, FirstName STRING, Age INT, PRIMARY KEY (PersonID) NOT
ENFORCED) WITH ('changelog-producer'='input');");
}
@Test
@@ -78,6 +88,47 @@ public class PartialUpdateITCase extends
FileStoreTableITCase {
.containsExactlyInAnyOrder(Row.of(1, 2, 4, 5, "6"), Row.of(1,
3, 2, 4, "1"));
}
+ @Test
+ public void testForeignKeyJoin() throws Exception {
+ sEnv.getConfig()
+ .set(
+
ExecutionConfigOptions.TABLE_EXEC_SINK_UPSERT_MATERIALIZE,
+ ExecutionConfigOptions.UpsertMaterialize.NONE);
+ CloseableIterator<Row> iter =
+ streamSqlIter(
+ "INSERT INTO dwd_orders "
+ + "SELECT OrderID, OrderNumber, PersonID,
CAST(NULL AS STRING), CAST(NULL AS STRING), CAST(NULL AS INT) FROM ods_orders "
+ + "UNION ALL "
+ + "SELECT OrderID, CAST(NULL AS INT),
dim_persons.PersonID, LastName, FirstName, Age FROM dim_persons JOIN ods_orders
ON dim_persons.PersonID = ods_orders.PersonID;");
+
+ batchSql("INSERT INTO ods_orders VALUES (1, 2, 3)");
+ batchSql("INSERT INTO dim_persons VALUES (3, 'snow', 'jon', 23)");
+ Thread.sleep(1000);
+ assertThat(rowsToList(batchSql("SELECT * FROM dwd_orders")))
+ .containsExactly(Arrays.asList(1, 2, 3, "snow", "jon", 23));
+
+ batchSql("INSERT INTO ods_orders VALUES (1, 4, 3)");
+ batchSql("INSERT INTO dim_persons VALUES (3, 'snow', 'targaryen',
23)");
+ Thread.sleep(1000);
+ assertThat(rowsToList(batchSql("SELECT * FROM dwd_orders")))
+ .containsExactly(Arrays.asList(1, 4, 3, "snow", "targaryen",
23));
+
+ iter.close();
+ }
+
+ private List<List<Object>> rowsToList(List<Row> rows) {
+ return rows.stream().map(this::toList).collect(Collectors.toList());
+ }
+
+ private List<Object> toList(Row row) {
+ assertThat(row.getKind()).isIn(RowKind.INSERT, RowKind.UPDATE_AFTER);
+ List<Object> result = new ArrayList<>();
+ for (int i = 0; i < row.getArity(); i++) {
+ result.add(row.getField(i));
+ }
+ return result;
+ }
+
@Test
public void testStreamingRead() {
assertThatThrownBy(
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/CoreOptions.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/CoreOptions.java
index 72bce71a..de190ccf 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/CoreOptions.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/CoreOptions.java
@@ -149,6 +149,12 @@ public class CoreOptions implements Serializable {
.defaultValue(MergeEngine.DEDUPLICATE)
.withDescription("Specify the merge engine for table with
primary key.");
+ public static final ConfigOption<Boolean> PARTIAL_UPDATE_IGNORE_DELETE =
+ ConfigOptions.key("partial-update.ignore-delete")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Whether to ignore delete records in
partial-update mode.");
+
@Immutable
public static final ConfigOption<WriteMode> WRITE_MODE =
ConfigOptions.key("write-mode")
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/PartialUpdateMergeFunction.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/PartialUpdateMergeFunction.java
index ed3a2bcc..fe9049bc 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/PartialUpdateMergeFunction.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/PartialUpdateMergeFunction.java
@@ -25,9 +25,6 @@ import org.apache.flink.types.RowKind;
import javax.annotation.Nullable;
-import static org.apache.flink.util.Preconditions.checkArgument;
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
/**
* A {@link MergeFunction} where key is primary key (unique) and value is the
partial record, update
* non-null fields on merge.
@@ -37,13 +34,15 @@ public class PartialUpdateMergeFunction implements
MergeFunction<KeyValue> {
private static final long serialVersionUID = 1L;
private final RowData.FieldGetter[] getters;
+ private final boolean ignoreDelete;
private transient KeyValue latestKv;
private transient GenericRowData row;
private transient KeyValue reused;
- public PartialUpdateMergeFunction(RowData.FieldGetter[] getters) {
+ public PartialUpdateMergeFunction(RowData.FieldGetter[] getters, boolean
ignoreDelete) {
this.getters = getters;
+ this.ignoreDelete = ignoreDelete;
}
@Override
@@ -54,9 +53,20 @@ public class PartialUpdateMergeFunction implements
MergeFunction<KeyValue> {
@Override
public void add(KeyValue kv) {
- checkArgument(
- kv.valueKind() == RowKind.INSERT || kv.valueKind() ==
RowKind.UPDATE_AFTER,
- "Partial update can not accept delete records. Partial delete
is not supported!");
+ if (kv.valueKind() == RowKind.UPDATE_BEFORE || kv.valueKind() ==
RowKind.DELETE) {
+ if (ignoreDelete) {
+ return;
+ }
+
+ if (kv.valueKind() == RowKind.UPDATE_BEFORE) {
+ throw new IllegalArgumentException(
+ "Partial update can not accept update_before records,
it is a bug.");
+ }
+
+ throw new IllegalArgumentException(
+ "Partial update can not accept delete records. Partial
delete is not supported!");
+ }
+
latestKv = kv;
for (int i = 0; i < getters.length; i++) {
Object field = getters[i].getFieldOrNull(kv.value());
@@ -69,9 +79,14 @@ public class PartialUpdateMergeFunction implements
MergeFunction<KeyValue> {
@Override
@Nullable
public KeyValue getResult() {
- checkNotNull(
- latestKv,
- "Trying to get result from merge function without any input.
This is unexpected.");
+ if (latestKv == null) {
+ if (ignoreDelete) {
+ return null;
+ }
+
+ throw new IllegalArgumentException(
+ "Trying to get result from merge function without any
input. This is unexpected.");
+ }
if (reused == null) {
reused = new KeyValue();
@@ -82,6 +97,6 @@ public class PartialUpdateMergeFunction implements
MergeFunction<KeyValue> {
@Override
public MergeFunction<KeyValue> copy() {
// RowData.FieldGetter is thread safe
- return new PartialUpdateMergeFunction(getters);
+ return new PartialUpdateMergeFunction(getters, ignoreDelete);
}
}
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTable.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTable.java
index 18f44f75..ed464435 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTable.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTable.java
@@ -86,7 +86,8 @@ public class ChangelogWithKeyFileStoreTable extends
AbstractFileStoreTable {
mergeFunction = new DeduplicateMergeFunction();
break;
case PARTIAL_UPDATE:
- mergeFunction = new PartialUpdateMergeFunction(fieldGetters);
+ boolean ignoreDelete =
conf.get(CoreOptions.PARTIAL_UPDATE_IGNORE_DELETE);
+ mergeFunction = new PartialUpdateMergeFunction(fieldGetters,
ignoreDelete);
break;
case AGGREGATE:
List<String> fieldNames = tableSchema.fieldNames();
diff --git
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTableTest.java
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTableTest.java
index 3f4e908d..598269bc 100644
---
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTableTest.java
+++
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTableTest.java
@@ -283,6 +283,31 @@ public class ChangelogWithKeyFileStoreTableTest extends
FileStoreTableTestBase {
"1|60|500|binary|varbinary"));
}
+ @Test
+ public void testPartialUpdateIgnoreDelete() throws Exception {
+ FileStoreTable table =
+ createFileStoreTable(
+ conf -> {
+ conf.set(
+ CoreOptions.MERGE_ENGINE,
+ CoreOptions.MergeEngine.PARTIAL_UPDATE);
+ conf.set(CoreOptions.PARTIAL_UPDATE_IGNORE_DELETE,
true);
+ });
+ TableWrite write = table.newWrite();
+ TableCommit commit = table.newCommit("user");
+ write.write(rowData(1, 10, 200L));
+ write.write(rowDataWithKind(RowKind.DELETE, 1, 10, 100L));
+ write.write(rowDataWithKind(RowKind.UPDATE_BEFORE, 1, 10, 100L));
+ write.write(rowDataWithKind(RowKind.DELETE, 1, 20, 400L));
+ commit.commit("0", write.prepareCommit(true));
+ write.close();
+
+ List<Split> splits = table.newScan().plan().splits;
+ TableRead read = table.newRead();
+ assertThat(getResult(read, splits, binaryRow(1), 0,
BATCH_ROW_TO_STRING))
+
.isEqualTo(Collections.singletonList("1|10|200|binary|varbinary"));
+ }
+
@Override
protected FileStoreTable createFileStoreTable(Consumer<Configuration>
configure)
throws Exception {
