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/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 2ff82cda7 [flink] extend merge_into procedure to support
notMatchedBySourceUpsert/Delete. (#4314)
2ff82cda7 is described below
commit 2ff82cda7966a4ce244fccaeda813de5f7f9f6d8
Author: liming.1018 <[email protected]>
AuthorDate: Mon Oct 14 13:03:54 2024 +0800
[flink] extend merge_into procedure to support
notMatchedBySourceUpsert/Delete. (#4314)
---
docs/content/flink/procedures.md | 7 +-
.../paimon/flink/procedure/MergeIntoProcedure.java | 36 ++++++-
.../paimon/flink/action/MergeIntoActionITCase.java | 117 +++++++++++++++++++++
3 files changed, 156 insertions(+), 4 deletions(-)
diff --git a/docs/content/flink/procedures.md b/docs/content/flink/procedures.md
index ce8c8043a..b763b7aaa 100644
--- a/docs/content/flink/procedures.md
+++ b/docs/content/flink/procedures.md
@@ -241,10 +241,13 @@ All available procedures are listed below.
matched_upsert_setting => 'matchedUpsertSetting',<br/>
not_matched_insert_condition => 'notMatchedInsertCondition',<br/>
not_matched_insert_values => 'notMatchedInsertValues',<br/>
- matched_delete_condition => 'matchedDeleteCondition') <br/><br/>
+ matched_delete_condition => 'matchedDeleteCondition',<br/>
+ not_matched_by_source_upsert_condition =>
'notMatchedBySourceUpsertCondition',<br/>
+ not_matched_by_source_upsert_setting =>
'notMatchedBySourceUpsertSetting',<br/>
+ not_matched_by_source_delete_condition =>
'notMatchedBySourceDeleteCondition') <br/><br/>
</td>
<td>
- To perform "MERGE INTO" syntax. See <a
href="/how-to/writing-tables#merging-into-table">merge_into action</a> for
+ To perform "MERGE INTO" syntax. See <a
href="/flink/action-jars#merging-into-table">merge_into action</a> for
details of arguments.
</td>
<td>
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MergeIntoProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MergeIntoProcedure.java
index cf8d71919..e297c0bdb 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MergeIntoProcedure.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MergeIntoProcedure.java
@@ -102,7 +102,19 @@ public class MergeIntoProcedure extends ProcedureBase {
@ArgumentHint(
name = "matched_delete_condition",
type = @DataTypeHint("STRING"),
- isOptional = true)
+ isOptional = true),
+ @ArgumentHint(
+ name = "not_matched_by_source_upsert_condition",
+ type = @DataTypeHint("STRING"),
+ isOptional = true),
+ @ArgumentHint(
+ name = "not_matched_by_source_upsert_setting",
+ type = @DataTypeHint("STRING"),
+ isOptional = true),
+ @ArgumentHint(
+ name = "not_matched_by_source_delete_condition",
+ type = @DataTypeHint("STRING"),
+ isOptional = true),
})
public String[] call(
ProcedureContext procedureContext,
@@ -115,7 +127,10 @@ public class MergeIntoProcedure extends ProcedureBase {
String matchedUpsertSetting,
String notMatchedInsertCondition,
String notMatchedInsertValues,
- String matchedDeleteCondition) {
+ String matchedDeleteCondition,
+ String notMatchedBySourceUpsertCondition,
+ String notMatchedBySourceUpsertSetting,
+ String notMatchedBySourceDeleteCondition) {
targetAlias = notnull(targetAlias);
sourceSqls = notnull(sourceSqls);
sourceTable = notnull(sourceTable);
@@ -125,6 +140,9 @@ public class MergeIntoProcedure extends ProcedureBase {
notMatchedInsertCondition = notnull(notMatchedInsertCondition);
notMatchedInsertValues = notnull(notMatchedInsertValues);
matchedDeleteCondition = notnull(matchedDeleteCondition);
+ notMatchedBySourceUpsertCondition =
notnull(notMatchedBySourceUpsertCondition);
+ notMatchedBySourceUpsertSetting =
notnull(notMatchedBySourceUpsertSetting);
+ notMatchedBySourceDeleteCondition =
notnull(notMatchedBySourceDeleteCondition);
String warehouse = catalog.warehouse();
Map<String, String> catalogOptions = catalog.options();
@@ -166,6 +184,20 @@ public class MergeIntoProcedure extends ProcedureBase {
action.withMatchedDelete(matchedDeleteCondition);
}
+ if (!notMatchedBySourceUpsertCondition.isEmpty()
+ || !notMatchedBySourceUpsertSetting.isEmpty()) {
+ String condition = nullable(notMatchedBySourceUpsertCondition);
+ String values = nullable(notMatchedBySourceUpsertSetting);
+ checkArgument(
+ !"*".equals(values),
+ "not-matched-by-source-upsert does not support setting
notMatchedBySourceUpsertSetting to *.");
+ action.withNotMatchedBySourceUpsert(condition, values);
+ }
+
+ if (!notMatchedBySourceDeleteCondition.isEmpty()) {
+
action.withNotMatchedBySourceDelete(notMatchedBySourceDeleteCondition);
+ }
+
action.withStreamExecutionEnvironment(procedureContext.getExecutionEnvironment());
action.validate();
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/MergeIntoActionITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/MergeIntoActionITCase.java
index 41d607fac..3907c0398 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/MergeIntoActionITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/MergeIntoActionITCase.java
@@ -571,6 +571,123 @@ public class MergeIntoActionITCase extends
ActionITCaseBase {
.satisfies(anyCauseMatches(ValidationException.class, "Object
'S' not found"));
}
+ @ParameterizedTest
+ @MethodSource("testArguments")
+ public void testNotMatchedBySourceUpsert(boolean qualified, String
invoker) throws Exception {
+ sEnv.executeSql("DROP TABLE T");
+ prepareTargetTable(CoreOptions.ChangelogProducer.INPUT);
+
+ // build MergeIntoAction
+ MergeIntoActionBuilder action = new MergeIntoActionBuilder(warehouse,
database, "T");
+ action.withSourceSqls("CREATE TEMPORARY VIEW SS AS SELECT k, v,
'unknown', dt FROM S")
+ .withSourceTable(qualified ? "default.SS" : "SS")
+ .withMergeCondition("T.k = SS.k AND T.dt = SS.dt")
+ .withNotMatchedBySourceUpsert(
+ "dt < '02-28'", "v = v || '_nmu', last_action =
'not_matched_upsert'");
+
+ String procedureStatement = "";
+ if ("procedure_indexed".equals(invoker)) {
+ procedureStatement =
+ String.format(
+ "CALL sys.merge_into('%s.T', '', '%s', '%s', 'T.k
= SS.k AND T.dt = SS.dt', '', '', '', '', '', 'dt < ''02-28''', 'v = v ||
''_nmu'', last_action = ''not_matched_upsert''')",
+ database,
+ "CREATE TEMPORARY VIEW SS AS SELECT k, v,
''unknown'', dt FROM S",
+ qualified ? "default.SS" : "SS");
+ } else if ("procedure_named".equals(invoker)) {
+ procedureStatement =
+ String.format(
+ "CALL sys.merge_into("
+ + "target_table => '%s.T', "
+ + "source_sqls => '%s', "
+ + "source_table => '%s', "
+ + "merge_condition => 'T.k = SS.k AND T.dt
= SS.dt', "
+ + "not_matched_by_source_upsert_condition
=> 'dt < ''02-28''',"
+ + "not_matched_by_source_upsert_setting =>
'v = v || ''_nmu'', last_action = ''not_matched_upsert''')",
+ database,
+ "CREATE TEMPORARY VIEW SS AS SELECT k, v,
''unknown'', dt FROM S",
+ qualified ? "default.SS" : "SS");
+ }
+
+ List<Row> streamingExpected =
+ Arrays.asList(
+ changelogRow("+U", 2, "v_2_nmu", "not_matched_upsert",
"02-27"),
+ changelogRow("+U", 3, "v_3_nmu", "not_matched_upsert",
"02-27"));
+
+ List<Row> batchExpected =
+ Arrays.asList(
+ changelogRow("+I", 1, "v_1", "creation", "02-27"),
+ changelogRow("+I", 2, "v_2_nmu", "not_matched_upsert",
"02-27"),
+ changelogRow("+I", 3, "v_3_nmu", "not_matched_upsert",
"02-27"),
+ changelogRow("+I", 4, "v_4", "creation", "02-27"),
+ changelogRow("+I", 5, "v_5", "creation", "02-28"),
+ changelogRow("+I", 6, "v_6", "creation", "02-28"),
+ changelogRow("+I", 7, "v_7", "creation", "02-28"),
+ changelogRow("+I", 8, "v_8", "creation", "02-28"),
+ changelogRow("+I", 9, "v_9", "creation", "02-28"),
+ changelogRow("+I", 10, "v_10", "creation", "02-28"));
+
+ if ("action".equals(invoker)) {
+ validateActionRunResult(action.build(), streamingExpected,
batchExpected);
+ } else {
+ validateProcedureResult(procedureStatement, streamingExpected,
batchExpected);
+ }
+ }
+
+ @ParameterizedTest
+ @MethodSource("testArguments")
+ public void testNotMatchedBySourceDelete(boolean qualified, String
invoker) throws Exception {
+ // build MergeIntoAction
+ MergeIntoActionBuilder action = new MergeIntoActionBuilder(warehouse,
database, "T");
+ action.withSourceSqls("CREATE TEMPORARY VIEW SS AS SELECT k, v,
'unknown', dt FROM S")
+ .withSourceTable(qualified ? "default.SS" : "SS")
+ .withMergeCondition("T.k = SS.k AND T.dt = SS.dt")
+ .withNotMatchedBySourceDelete(null);
+
+ String procedureStatement = "";
+ if ("procedure_indexed".equals(invoker)) {
+ procedureStatement =
+ String.format(
+ "CALL sys.merge_into('%s.T', '', '%s', '%s', 'T.k
= SS.k AND T.dt = SS.dt', '', '', '', '', '', '', '', 'TRUE')",
+ database,
+ "CREATE TEMPORARY VIEW SS AS SELECT k, v,
''unknown'', dt FROM S",
+ qualified ? "default.SS" : "SS");
+ } else if ("procedure_named".equals(invoker)) {
+ procedureStatement =
+ String.format(
+ "CALL sys.merge_into("
+ + "target_table => '%s.T', "
+ + "source_sqls => '%s', "
+ + "source_table => '%s', "
+ + "merge_condition => 'T.k = SS.k AND T.dt
= SS.dt', "
+ + "not_matched_by_source_delete_condition
=> 'TRUE')",
+ database,
+ "CREATE TEMPORARY VIEW SS AS SELECT k, v,
''unknown'', dt FROM S",
+ qualified ? "default.SS" : "SS");
+ }
+
+ List<Row> streamingExpected =
+ Arrays.asList(
+ changelogRow("-D", 2, "v_2", "creation", "02-27"),
+ changelogRow("-D", 3, "v_3", "creation", "02-27"),
+ changelogRow("-D", 5, "v_5", "creation", "02-28"),
+ changelogRow("-D", 6, "v_6", "creation", "02-28"),
+ changelogRow("-D", 9, "v_9", "creation", "02-28"),
+ changelogRow("-D", 10, "v_10", "creation", "02-28"));
+
+ List<Row> batchExpected =
+ Arrays.asList(
+ changelogRow("+I", 1, "v_1", "creation", "02-27"),
+ changelogRow("+I", 4, "v_4", "creation", "02-27"),
+ changelogRow("+I", 7, "v_7", "creation", "02-28"),
+ changelogRow("+I", 8, "v_8", "creation", "02-28"));
+
+ if ("action".equals(invoker)) {
+ validateActionRunResult(action.build(), streamingExpected,
batchExpected);
+ } else {
+ validateProcedureResult(procedureStatement, streamingExpected,
batchExpected);
+ }
+ }
+
private void validateActionRunResult(
MergeIntoAction action, List<Row> streamingExpected, List<Row>
batchExpected)
throws Exception {
