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/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 1ed8e5080 [Bug] Fix the unsupported retract message when projection
push down in partial-update engine (#2795)
1ed8e5080 is described below
commit 1ed8e5080cba31b3aa0c67f08b7fe1b8f7052ceb
Author: Aitozi <[email protected]>
AuthorDate: Fri Jan 26 11:18:58 2024 +0800
[Bug] Fix the unsupported retract message when projection push down in
partial-update engine (#2795)
---
.../compact/PartialUpdateMergeFunction.java | 14 ++++--
.../org/apache/paimon/flink/CatalogITCaseBase.java | 23 +++++++++
.../apache/paimon/flink/PartialUpdateITCase.java | 57 ++++++++++++++++++++++
3 files changed, 89 insertions(+), 5 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
index 8625530dc..b850248da 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
@@ -56,7 +56,7 @@ public class PartialUpdateMergeFunction implements
MergeFunction<KeyValue> {
private final InternalRow.FieldGetter[] getters;
private final boolean ignoreDelete;
private final Map<Integer, SequenceGenerator> fieldSequences;
-
+ private final boolean fieldSequenceEnabled;
private final Map<Integer, FieldAggregator> fieldAggregators;
private InternalRow currentKey;
@@ -69,11 +69,13 @@ public class PartialUpdateMergeFunction implements
MergeFunction<KeyValue> {
InternalRow.FieldGetter[] getters,
boolean ignoreDelete,
Map<Integer, SequenceGenerator> fieldSequences,
- Map<Integer, FieldAggregator> fieldAggregators) {
+ Map<Integer, FieldAggregator> fieldAggregators,
+ boolean fieldSequenceEnabled) {
this.getters = getters;
this.ignoreDelete = ignoreDelete;
this.fieldSequences = fieldSequences;
this.fieldAggregators = fieldAggregators;
+ this.fieldSequenceEnabled = fieldSequenceEnabled;
}
@Override
@@ -95,7 +97,7 @@ public class PartialUpdateMergeFunction implements
MergeFunction<KeyValue> {
return;
}
- if (fieldSequences.size() > 1) {
+ if (fieldSequenceEnabled) {
retractWithSequenceGroup(kv);
return;
}
@@ -312,13 +314,15 @@ public class PartialUpdateMergeFunction implements
MergeFunction<KeyValue> {
createFieldGetters(Projection.of(projection).project(tableTypes)),
ignoreDelete,
projectedSequences,
- projectedAggregators);
+ projectedAggregators,
+ !fieldSequences.isEmpty());
} else {
return new PartialUpdateMergeFunction(
createFieldGetters(tableTypes),
ignoreDelete,
fieldSequences,
- fieldAggregators);
+ fieldAggregators,
+ !fieldSequences.isEmpty());
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogITCaseBase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogITCaseBase.java
index fd9a72f41..20ebdd505 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogITCaseBase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogITCaseBase.java
@@ -43,6 +43,8 @@ import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.ddl.CreateCatalogOperation;
import org.apache.flink.types.Row;
import org.apache.flink.util.CloseableIterator;
+import org.assertj.core.api.Assertions;
+import org.assertj.core.api.ListAssert;
import org.junit.jupiter.api.BeforeEach;
import javax.annotation.Nullable;
@@ -55,6 +57,7 @@ import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.function.Consumer;
import java.util.stream.Collectors;
import static
org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL;
@@ -156,6 +159,26 @@ public abstract class CatalogITCaseBase extends
AbstractTestBase {
}
}
+ protected void sqlAssertWithRetry(
+ String query, Consumer<ListAssert<Row>> checker, Object... args) {
+ long start = System.currentTimeMillis();
+ while (true) {
+ try (CloseableIterator<Row> iter =
+ tEnv.executeSql(String.format(query, args)).collect()) {
+ try {
+
checker.accept(Assertions.assertThat(ImmutableList.copyOf(iter)));
+ return;
+ } catch (AssertionError e) {
+ if (System.currentTimeMillis() - start >= 3 * 60 * 1000) {
+ throw new RuntimeException(e);
+ }
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
protected CloseableIterator<Row> streamSqlIter(String query, Object...
args) {
return sEnv.executeSql(String.format(query, args)).collect();
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PartialUpdateITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PartialUpdateITCase.java
index c8172131e..9fe276a9d 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PartialUpdateITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PartialUpdateITCase.java
@@ -21,6 +21,7 @@ package org.apache.paimon.flink;
import org.apache.paimon.utils.BlockingIterator;
import org.apache.flink.table.api.config.ExecutionConfigOptions;
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
import org.apache.flink.util.CloseableIterator;
@@ -347,4 +348,60 @@ public class PartialUpdateITCase extends CatalogITCaseBase
{
.hasMessageContaining("Sink materializer must not be used
with Paimon sink.");
}
}
+
+ @Test
+ public void testPartialUpdateProjectionPushDownWithDeleteMessage() throws
Exception {
+ List<Row> input = Arrays.asList(Row.ofKind(RowKind.INSERT, 1, 1, 1));
+
+ String id = TestValuesTableFactory.registerData(input);
+ // create temp table in stream table env
+ sEnv.executeSql(
+ String.format(
+ "CREATE TEMPORARY TABLE source (k INT, a INT, g_1 INT,
PRIMARY KEY (k) NOT ENFORCED) "
+ + "WITH ('connector'='values',
'bounded'='true', 'data-id'='%s', "
+ + "'changelog-mode' = 'I,D,UA,UB')",
+ id));
+
+ sql(
+ "CREATE TABLE TEST ("
+ + "k INT, a INT, b INT, g_1 INT, g_2 INT, PRIMARY KEY
(k) NOT ENFORCED)"
+ + " WITH ("
+ + "'merge-engine'='partial-update', "
+ + "'fields.g_1.sequence-group'='a', "
+ + "'fields.g_2.sequence-group'='b');");
+
+ CloseableIterator<Row> insert1 =
+ streamSqlIter(
+ "INSERT INTO TEST SELECT k, a, CAST(NULL AS INT) AS b,
g_1,"
+ + " CAST(NULL AS INT) as g_2 FROM source");
+
+ sqlAssertWithRetry(
+ "SELECT * FROM TEST",
+ list -> list.containsExactlyInAnyOrder(Row.of(1, 1, null, 1,
null)));
+
+ // insert the delete message
+ input = Arrays.asList(Row.ofKind(RowKind.DELETE, 1, 1, 2));
+
+ id = TestValuesTableFactory.registerData(input);
+
+ // create temp table in stream table env
+ sEnv.executeSql(
+ String.format(
+ "CREATE TEMPORARY TABLE source2 (k INT, a INT, g_1
INT) "
+ + "WITH ('connector'='values',
'bounded'='true', 'data-id'='%s')",
+ id));
+
+ CloseableIterator<Row> insert2 =
+ streamSqlIter(
+ "INSERT INTO TEST SELECT k, a, CAST(NULL AS INT) AS b,
g_1,"
+ + " CAST(NULL AS INT) as g_2 FROM source2");
+
+ sqlAssertWithRetry(
+ "SELECT * FROM TEST",
+ list -> list.containsExactlyInAnyOrder(Row.of(1, null, null,
2, null)));
+
+ assertThat(sql("SELECT COUNT(*) FROM
TEST")).containsExactlyInAnyOrder(Row.of(1L));
+ insert1.close();
+ insert2.close();
+ }
}
