You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2023/07/03 09:11:35 UTC
[incubator-paimon] branch master updated: [flink] Refactor Delete Pushdown (#1488)
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 ef5794721 [flink] Refactor Delete Pushdown (#1488)
ef5794721 is described below
commit ef57947210525787e89e859d61ff07af5f609785
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Mon Jul 3 17:11:30 2023 +0800
[flink] Refactor Delete Pushdown (#1488)
---
.../predicate/AllPrimaryKeyEqualVisitor.java | 110 ++++++++++++++++++++
.../predicate/DeletePushDownPrimaryKeyVisitor.java | 114 ---------------------
...itor.java => OnlyPartitionKeyEqualVisitor.java} | 25 ++++-
.../predicate/DeletePushDownVisitorTest.java | 94 ++++++++---------
.../apache/paimon/operation/FileStoreCommit.java | 2 +
.../paimon/operation/FileStoreCommitImpl.java | 11 ++
.../apache/paimon/flink/sink/FlinkTableSink.java | 107 +++++++++++--------
.../paimon/flink/sink/FlinkTableSinkBase.java | 6 +-
.../apache/paimon/flink/ReadWriteTableITCase.java | 36 ++-----
9 files changed, 264 insertions(+), 241 deletions(-)
diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/AllPrimaryKeyEqualVisitor.java b/paimon-common/src/main/java/org/apache/paimon/predicate/AllPrimaryKeyEqualVisitor.java
new file mode 100644
index 000000000..b4bbddf30
--- /dev/null
+++ b/paimon-common/src/main/java/org/apache/paimon/predicate/AllPrimaryKeyEqualVisitor.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.predicate;
+
+import org.apache.paimon.shade.guava30.com.google.common.collect.Sets;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Visit the predicate and check if it only contains primary keys and can be push down.
+ *
+ * <p>We will check: all the primary keys are in the predicate with equal operator.
+ *
+ * <p>TODO: support IN.
+ */
+public class AllPrimaryKeyEqualVisitor implements FunctionVisitor<Set<String>> {
+
+ private final List<String> primaryKeys;
+
+ public AllPrimaryKeyEqualVisitor(List<String> primaryKeys) {
+ this.primaryKeys = primaryKeys;
+ }
+
+ @Override
+ public Set<String> visitIsNotNull(FieldRef fieldRef) {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Set<String> visitIsNull(FieldRef fieldRef) {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Set<String> visitStartsWith(FieldRef fieldRef, Object literal) {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Set<String> visitLessThan(FieldRef fieldRef, Object literal) {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Set<String> visitGreaterOrEqual(FieldRef fieldRef, Object literal) {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Set<String> visitNotEqual(FieldRef fieldRef, Object literal) {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Set<String> visitLessOrEqual(FieldRef fieldRef, Object literal) {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Set<String> visitEqual(FieldRef fieldRef, Object literal) {
+ if (primaryKeys.contains(fieldRef.name())) {
+ return Collections.singleton(fieldRef.name());
+ } else {
+ return Collections.emptySet();
+ }
+ }
+
+ @Override
+ public Set<String> visitGreaterThan(FieldRef fieldRef, Object literal) {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Set<String> visitIn(FieldRef fieldRef, List<Object> literals) {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Set<String> visitNotIn(FieldRef fieldRef, List<Object> literals) {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Set<String> visitAnd(List<Set<String>> children) {
+ return children.stream().reduce(Sets::union).get();
+ }
+
+ @Override
+ public Set<String> visitOr(List<Set<String>> children) {
+ return Collections.emptySet();
+ }
+}
diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/DeletePushDownPrimaryKeyVisitor.java b/paimon-common/src/main/java/org/apache/paimon/predicate/DeletePushDownPrimaryKeyVisitor.java
deleted file mode 100644
index dcf4cc75c..000000000
--- a/paimon-common/src/main/java/org/apache/paimon/predicate/DeletePushDownPrimaryKeyVisitor.java
+++ /dev/null
@@ -1,114 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.paimon.predicate;
-
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-/**
- * DeletePushDownPrimaryKeyVisitor visit the predicate and check if it only contains primary keys
- * and can be push down.
- *
- * <p>We will check: 1. all the primary keys are in the predicate with equal operator. 2. if step1
- * is satisfied, the other fields in the predicate could be ignored (e.g. return true).
- */
-public class DeletePushDownPrimaryKeyVisitor implements FunctionVisitor<Boolean> {
-
- private final List<String> primaryKeys;
-
- private Map<String, Boolean> hit;
-
- public DeletePushDownPrimaryKeyVisitor(List<String> primaryKeys) {
- this.primaryKeys = primaryKeys;
- this.hit = new HashMap<>();
- }
-
- public boolean isHitAll() {
- return hit.size() == primaryKeys.size();
- }
-
- @Override
- public Boolean visitIsNotNull(FieldRef fieldRef) {
- return !primaryKeys.contains(fieldRef.name());
- }
-
- @Override
- public Boolean visitIsNull(FieldRef fieldRef) {
- return !primaryKeys.contains(fieldRef.name());
- }
-
- @Override
- public Boolean visitStartsWith(FieldRef fieldRef, Object literal) {
- return !primaryKeys.contains(fieldRef.name());
- }
-
- @Override
- public Boolean visitLessThan(FieldRef fieldRef, Object literal) {
- return !primaryKeys.contains(fieldRef.name());
- }
-
- @Override
- public Boolean visitGreaterOrEqual(FieldRef fieldRef, Object literal) {
- return !primaryKeys.contains(fieldRef.name());
- }
-
- @Override
- public Boolean visitNotEqual(FieldRef fieldRef, Object literal) {
- return !primaryKeys.contains(fieldRef.name());
- }
-
- @Override
- public Boolean visitLessOrEqual(FieldRef fieldRef, Object literal) {
- return !primaryKeys.contains(fieldRef.name());
- }
-
- @Override
- public Boolean visitEqual(FieldRef fieldRef, Object literal) {
- if (primaryKeys.contains(fieldRef.name())) {
- hit.put(fieldRef.name(), true);
- }
- return true;
- }
-
- @Override
- public Boolean visitGreaterThan(FieldRef fieldRef, Object literal) {
- return !primaryKeys.contains(fieldRef.name());
- }
-
- @Override
- public Boolean visitIn(FieldRef fieldRef, List<Object> literals) {
- return !primaryKeys.contains(fieldRef.name());
- }
-
- @Override
- public Boolean visitNotIn(FieldRef fieldRef, List<Object> literals) {
- return !primaryKeys.contains(fieldRef.name());
- }
-
- @Override
- public Boolean visitAnd(List<Boolean> children) {
- return children.stream().reduce((x, y) -> x && y).get();
- }
-
- @Override
- public Boolean visitOr(List<Boolean> children) {
- return children.stream().reduce((x, y) -> x || y).get();
- }
-}
diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/DeletePushDownPartitionKeyVisitor.java b/paimon-common/src/main/java/org/apache/paimon/predicate/OnlyPartitionKeyEqualVisitor.java
similarity index 76%
rename from paimon-common/src/main/java/org/apache/paimon/predicate/DeletePushDownPartitionKeyVisitor.java
rename to paimon-common/src/main/java/org/apache/paimon/predicate/OnlyPartitionKeyEqualVisitor.java
index ca77d72a1..f085dcb53 100644
--- a/paimon-common/src/main/java/org/apache/paimon/predicate/DeletePushDownPartitionKeyVisitor.java
+++ b/paimon-common/src/main/java/org/apache/paimon/predicate/OnlyPartitionKeyEqualVisitor.java
@@ -18,18 +18,28 @@
package org.apache.paimon.predicate;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
/**
- * DeletePushDownPartitionKeyVisitor visit the predicate and check if it only contains partition
- * keys and can be push down.
+ * Visit the predicate and check if it only contains partition keys and can be push down.
+ *
+ * <p>TODO: more filters if {@code BatchWriteBuilder} supports predicate.
*/
-public class DeletePushDownPartitionKeyVisitor implements FunctionVisitor<Boolean> {
+public class OnlyPartitionKeyEqualVisitor implements FunctionVisitor<Boolean> {
private final List<String> partitionKeys;
- public DeletePushDownPartitionKeyVisitor(List<String> partitionKeys) {
+ private final Map<String, String> partitions;
+
+ public OnlyPartitionKeyEqualVisitor(List<String> partitionKeys) {
this.partitionKeys = partitionKeys;
+ partitions = new HashMap<>();
+ }
+
+ public Map<String, String> partitions() {
+ return partitions;
}
@Override
@@ -69,7 +79,12 @@ public class DeletePushDownPartitionKeyVisitor implements FunctionVisitor<Boolea
@Override
public Boolean visitEqual(FieldRef fieldRef, Object literal) {
- return partitionKeys.contains(fieldRef.name());
+ boolean contains = partitionKeys.contains(fieldRef.name());
+ if (contains) {
+ partitions.put(fieldRef.name(), literal.toString());
+ return true;
+ }
+ return false;
}
@Override
diff --git a/paimon-common/src/test/java/org/apache/paimon/predicate/DeletePushDownVisitorTest.java b/paimon-common/src/test/java/org/apache/paimon/predicate/DeletePushDownVisitorTest.java
index 623f2c3bc..9c767355c 100644
--- a/paimon-common/src/test/java/org/apache/paimon/predicate/DeletePushDownVisitorTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/predicate/DeletePushDownVisitorTest.java
@@ -24,7 +24,9 @@ import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
+import java.util.Set;
import static org.assertj.core.api.Assertions.assertThat;
@@ -89,35 +91,31 @@ public class DeletePushDownVisitorTest {
new LeafPredicate(
IsNotNull.INSTANCE, DataTypes.INT(), 6, "f", Collections.singletonList(6));
- /** filters contain all the primary keys with AND of Equal *********** */
+ /* filters contain all the primary keys with AND of Equal *********** */
// where a=1 and b=2 and c=3 and d=4, push down
- DeletePushDownPrimaryKeyVisitor visitor = new DeletePushDownPrimaryKeyVisitor(primaryKeys);
+ AllPrimaryKeyEqualVisitor visitor = new AllPrimaryKeyEqualVisitor(primaryKeys);
Predicate compoundPredicate =
PredicateBuilder.and(Arrays.asList(predicateA, predicateB, predicateC, predicateD));
- Boolean visitResult = compoundPredicate.visit(visitor);
- Boolean hit = visitor.isHitAll();
- assertThat(visitResult).isTrue();
- assertThat(hit).isTrue();
+ Set<String> visitResult = compoundPredicate.visit(visitor);
+ assertThat(visitResult).containsAnyElementsOf(primaryKeys);
// where a=1 and b=2 and c=3 and d=4 and e is not null, push down
- DeletePushDownPrimaryKeyVisitor visitor1 = new DeletePushDownPrimaryKeyVisitor(primaryKeys);
+ AllPrimaryKeyEqualVisitor visitor1 = new AllPrimaryKeyEqualVisitor(primaryKeys);
Predicate predicateABCDE =
PredicateBuilder.and(
Arrays.asList(predicateA, predicateB, predicateC, predicateD, predicateE));
- assertThat(predicateABCDE.visit(visitor1)).isTrue();
- assertThat(visitor1.isHitAll()).isTrue();
+ assertThat(predicateABCDE.visit(visitor1)).containsAnyElementsOf(primaryKeys);
// where a=1 and b=2 and c=3 and d=4 and f=6, push down
- DeletePushDownPrimaryKeyVisitor visitor2 = new DeletePushDownPrimaryKeyVisitor(primaryKeys);
+ AllPrimaryKeyEqualVisitor visitor2 = new AllPrimaryKeyEqualVisitor(primaryKeys);
Predicate predicateABCDF =
PredicateBuilder.and(
Arrays.asList(predicateA, predicateB, predicateC, predicateD, predicateF));
- assertThat(predicateABCDF.visit(visitor2)).isTrue();
- assertThat(visitor2.isHitAll()).isTrue();
+ assertThat(predicateABCDF.visit(visitor2)).containsAnyElementsOf(primaryKeys);
// where a=1 and b=2 and c=3 and d=4 and e is not null and f=6, push down
- DeletePushDownPrimaryKeyVisitor visitor3 = new DeletePushDownPrimaryKeyVisitor(primaryKeys);
+ AllPrimaryKeyEqualVisitor visitor3 = new AllPrimaryKeyEqualVisitor(primaryKeys);
Predicate predicateABCDEF =
PredicateBuilder.and(
Arrays.asList(
@@ -127,11 +125,10 @@ public class DeletePushDownVisitorTest {
predicateD,
predicateE,
predicateF));
- assertThat(predicateABCDEF.visit(visitor3)).isTrue();
- assertThat(visitor3.isHitAll()).isTrue();
+ assertThat(predicateABCDEF.visit(visitor3)).containsAnyElementsOf(primaryKeys);
// where a in (1,2) and b=2 and c=3 and d=4, push down
- DeletePushDownPrimaryKeyVisitor visitor4 = new DeletePushDownPrimaryKeyVisitor(primaryKeys);
+ AllPrimaryKeyEqualVisitor visitor4 = new AllPrimaryKeyEqualVisitor(primaryKeys);
Predicate predicateAABCD =
PredicateBuilder.and(
Arrays.asList(
@@ -140,41 +137,46 @@ public class DeletePushDownVisitorTest {
predicateC,
predicateD,
predicateE));
- assertThat(predicateAABCD.visit(visitor4)).isTrue();
- assertThat(visitor4.isHitAll()).isTrue();
+ assertThat(predicateAABCD.visit(visitor4)).containsAnyElementsOf(primaryKeys);
- /** not all the primary keys filters are of Equal func ************* */
+ /* not all the primary keys filters are of Equal func ************* */
// where a=1 and b=1 and c is not null and d=4, do not push down
- DeletePushDownPrimaryKeyVisitor visitor5 = new DeletePushDownPrimaryKeyVisitor(primaryKeys);
+ AllPrimaryKeyEqualVisitor visitor5 = new AllPrimaryKeyEqualVisitor(primaryKeys);
Predicate predicateABCNotNull =
PredicateBuilder.and(
Arrays.asList(predicateA, predicateB, predicateCIsNotNull, predicateD));
- assertThat(predicateABCNotNull.visit(visitor5)).isFalse();
- assertThat(visitor5.isHitAll()).isFalse();
+ assertThat(predicateABCNotNull.visit(visitor5)).isNotEqualTo(new HashSet<>(primaryKeys));
- /** filters not contain all the primary keys ****************** */
+ /* filters not contain all the primary keys ****************** */
// where a=1, do not push down
- DeletePushDownPrimaryKeyVisitor visitor6 = new DeletePushDownPrimaryKeyVisitor(primaryKeys);
- PredicateBuilder.and(Arrays.asList(predicateA)).visit(visitor6);
- assertThat(visitor6.isHitAll()).isFalse();
+ AllPrimaryKeyEqualVisitor visitor6 = new AllPrimaryKeyEqualVisitor(primaryKeys);
+ assertThat(PredicateBuilder.and(Collections.singletonList(predicateA)).visit(visitor6))
+ .isNotEqualTo(new HashSet<>(primaryKeys));
// where a=1 and b=2 and d=4, do not push down
- DeletePushDownPrimaryKeyVisitor visitor7 = new DeletePushDownPrimaryKeyVisitor(primaryKeys);
- PredicateBuilder.and(Arrays.asList(predicateA, predicateB, predicateD)).visit(visitor7);
- assertThat(visitor7.isHitAll()).isFalse();
+ AllPrimaryKeyEqualVisitor visitor7 = new AllPrimaryKeyEqualVisitor(primaryKeys);
+ assertThat(
+ PredicateBuilder.and(Arrays.asList(predicateA, predicateB, predicateD))
+ .visit(visitor7))
+ .isNotEqualTo(new HashSet<>(primaryKeys));
// where a=1 and c=3 and d=4, do not push down
- DeletePushDownPrimaryKeyVisitor visitor8 = new DeletePushDownPrimaryKeyVisitor(primaryKeys);
- PredicateBuilder.and(Arrays.asList(predicateA, predicateC, predicateD)).visit(visitor8);
- assertThat(visitor8.isHitAll()).isFalse();
+ AllPrimaryKeyEqualVisitor visitor8 = new AllPrimaryKeyEqualVisitor(primaryKeys);
+ assertThat(
+ PredicateBuilder.and(Arrays.asList(predicateA, predicateC, predicateD))
+ .visit(visitor8))
+ .isNotEqualTo(new HashSet<>(primaryKeys));
// where b=2 and c=3 and d=4 and f=6, not push down
- DeletePushDownPrimaryKeyVisitor visitor9 = new DeletePushDownPrimaryKeyVisitor(primaryKeys);
- PredicateBuilder.and(Arrays.asList(predicateB, predicateC, predicateD, predicateF))
- .visit(visitor9);
- assertThat(visitor9.isHitAll()).isFalse();
+ AllPrimaryKeyEqualVisitor visitor9 = new AllPrimaryKeyEqualVisitor(primaryKeys);
+ assertThat(
+ PredicateBuilder.and(
+ Arrays.asList(
+ predicateB, predicateC, predicateD, predicateF))
+ .visit(visitor9))
+ .isNotEqualTo(new HashSet<>(primaryKeys));
}
@Test
@@ -205,31 +207,27 @@ public class DeletePushDownVisitorTest {
new LeafPredicate(
Equal.INSTANCE, DataTypes.INT(), 5, "e", Collections.singletonList(5));
- /** filters only contain partition keys with func Equal ****************** */
+ /* filters only contain partition keys with func Equal ****************** */
// where b=2 and c=3 and d=4, push down
- DeletePushDownPartitionKeyVisitor visitor =
- new DeletePushDownPartitionKeyVisitor(partitionKeys);
+ OnlyPartitionKeyEqualVisitor visitor = new OnlyPartitionKeyEqualVisitor(partitionKeys);
assertThat(
PredicateBuilder.and(Arrays.asList(predicateB, predicateC, predicateD))
.visit(visitor))
.isTrue();
// where b=2 and c=3, push down
- DeletePushDownPartitionKeyVisitor visitor1 =
- new DeletePushDownPartitionKeyVisitor(partitionKeys);
+ OnlyPartitionKeyEqualVisitor visitor1 = new OnlyPartitionKeyEqualVisitor(partitionKeys);
assertThat(PredicateBuilder.and(Arrays.asList(predicateB, predicateC)).visit(visitor1))
.isTrue();
// where b=2 and d=4, push down
- DeletePushDownPartitionKeyVisitor visitor2 =
- new DeletePushDownPartitionKeyVisitor(partitionKeys);
+ OnlyPartitionKeyEqualVisitor visitor2 = new OnlyPartitionKeyEqualVisitor(partitionKeys);
assertThat(PredicateBuilder.and(Arrays.asList(predicateB, predicateD)).visit(visitor2))
.isTrue();
// where b=2 and c=3 and d=4 and e=5, do not push down
- DeletePushDownPartitionKeyVisitor visitor3 =
- new DeletePushDownPartitionKeyVisitor(partitionKeys);
+ OnlyPartitionKeyEqualVisitor visitor3 = new OnlyPartitionKeyEqualVisitor(partitionKeys);
assertThat(
PredicateBuilder.and(
Arrays.asList(
@@ -238,8 +236,7 @@ public class DeletePushDownVisitorTest {
.isFalse();
// where b=2 and c=3 or d=4, do not push down
- DeletePushDownPartitionKeyVisitor visitor4 =
- new DeletePushDownPartitionKeyVisitor(partitionKeys);
+ OnlyPartitionKeyEqualVisitor visitor4 = new OnlyPartitionKeyEqualVisitor(partitionKeys);
assertThat(
PredicateBuilder.and(
Arrays.asList(
@@ -249,8 +246,7 @@ public class DeletePushDownVisitorTest {
.isFalse();
// where b=2 and c=3 and d>5, do not push down
- DeletePushDownPartitionKeyVisitor visitor5 =
- new DeletePushDownPartitionKeyVisitor(partitionKeys);
+ OnlyPartitionKeyEqualVisitor visitor5 = new OnlyPartitionKeyEqualVisitor(partitionKeys);
assertThat(
PredicateBuilder.and(
Arrays.asList(predicateB, predicateC, predicateDgreater))
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
index d7c68dd5b..a585d6812 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
@@ -74,6 +74,8 @@ public interface FileStoreCommit {
*/
void dropPartitions(List<Map<String, String>> partitions, long commitIdentifier);
+ void purgeTable(long commitIdentifier);
+
/** Abort an unsuccessful commit. The data files will be deleted. */
void abort(List<CommitMessage> commitMessages);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index c7e242e80..3eda2bcdb 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -395,6 +395,17 @@ public class FileStoreCommitImpl implements FileStoreCommit {
Collections.emptyMap());
}
+ @Override
+ public void purgeTable(long commitIdentifier) {
+ tryOverwrite(
+ null,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ commitIdentifier,
+ null,
+ new HashMap<>());
+ }
+
@Override
public void abort(List<CommitMessage> commitMessages) {
Map<Pair<BinaryRow, Integer>, DataFilePathFactory> factoryMap = new HashMap<>();
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkTableSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkTableSink.java
index c2afaf0d3..96044bbf7 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkTableSink.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkTableSink.java
@@ -24,9 +24,8 @@ import org.apache.paimon.flink.PredicateConverter;
import org.apache.paimon.flink.log.LogStoreTableFactory;
import org.apache.paimon.operation.FileStoreCommit;
import org.apache.paimon.options.Options;
-import org.apache.paimon.predicate.DeletePushDownPartitionKeyVisitor;
-import org.apache.paimon.predicate.DeletePushDownPrimaryKeyVisitor;
-import org.apache.paimon.predicate.LeafPredicate;
+import org.apache.paimon.predicate.AllPrimaryKeyEqualVisitor;
+import org.apache.paimon.predicate.OnlyPartitionKeyEqualVisitor;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.table.AbstractFileStoreTable;
@@ -40,6 +39,7 @@ import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.connector.RowLevelModificationScanContext;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.sink.abilities.SupportsDeletePushDown;
import org.apache.flink.table.connector.sink.abilities.SupportsRowLevelDelete;
import org.apache.flink.table.connector.sink.abilities.SupportsRowLevelUpdate;
@@ -50,7 +50,8 @@ import org.apache.flink.table.types.logical.RowType;
import javax.annotation.Nullable;
import java.util.ArrayList;
-import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -65,12 +66,7 @@ import static org.apache.paimon.CoreOptions.MERGE_ENGINE;
public class FlinkTableSink extends FlinkTableSinkBase
implements SupportsRowLevelUpdate, SupportsRowLevelDelete, SupportsDeletePushDown {
- @Nullable protected List<Predicate> predicates;
- @Nullable protected Predicate predicate;
- private boolean isDeleteFilterPartitionKey = false;
- private boolean isDeleteFilterPrimaryKey = false;
-
- private static final String TABLE_PATH_KEY = "path";
+ @Nullable protected Predicate deletePredicate;
public FlinkTableSink(
ObjectIdentifier tableIdentifier,
@@ -80,6 +76,16 @@ public class FlinkTableSink extends FlinkTableSinkBase
super(tableIdentifier, table, context, logStoreTableFactory);
}
+ @Override
+ public DynamicTableSink copy() {
+ FlinkTableSink copied =
+ new FlinkTableSink(tableIdentifier, table, context, logStoreTableFactory);
+ copied.staticPartitions = new HashMap<>(staticPartitions);
+ copied.overwrite = overwrite;
+ copied.deletePredicate = deletePredicate;
+ return copied;
+ }
+
@Override
public RowLevelUpdateInfo applyRowLevelUpdate(
List<Column> updatedColumns, @Nullable RowLevelModificationScanContext context) {
@@ -138,7 +144,7 @@ public class FlinkTableSink extends FlinkTableSinkBase
@Override
public RowLevelDeleteInfo applyRowLevelDelete(
@Nullable RowLevelModificationScanContext rowLevelModificationScanContext) {
- checkDeletable();
+ validateDeletable();
return new RowLevelDeleteInfo() {};
}
@@ -146,12 +152,8 @@ public class FlinkTableSink extends FlinkTableSinkBase
@Override
public boolean applyDeleteFilters(List<ResolvedExpression> list) {
- checkDeletable();
- if (list.size() == 0) {
- return false;
- }
-
- predicates = new ArrayList<>();
+ validateDeletable();
+ List<Predicate> predicates = new ArrayList<>();
RowType rowType = LogicalTypeConversion.toLogicalType(table.rowType());
for (ResolvedExpression filter : list) {
Optional<Predicate> predicate = PredicateConverter.convert(rowType, filter);
@@ -162,35 +164,28 @@ public class FlinkTableSink extends FlinkTableSinkBase
return false;
}
}
- predicate = predicates.isEmpty() ? null : PredicateBuilder.and(predicates);
- return canPushDownDeleteFilter(predicate);
+ deletePredicate = predicates.isEmpty() ? null : PredicateBuilder.and(predicates);
+ return canPushDownDeleteFilter();
}
@Override
public Optional<Long> executeDeletion() {
- // delete partition
- if (isDeleteFilterPartitionKey) {
- Map<String, String> partitions =
- predicates.stream()
- .map(p -> (LeafPredicate) p)
- .collect(
- Collectors.toMap(
- LeafPredicate::fieldName,
- x -> String.valueOf(x.literals().get(0))));
-
- FileStoreCommit commit =
- ((AbstractFileStoreTable) table)
- .store()
- .newCommit(UUID.randomUUID().toString());
- commit.dropPartitions(Arrays.asList(partitions), BatchWriteBuilder.COMMIT_IDENTIFIER);
+ FileStoreCommit commit =
+ ((AbstractFileStoreTable) table).store().newCommit(UUID.randomUUID().toString());
+ long identifier = BatchWriteBuilder.COMMIT_IDENTIFIER;
+ if (deletePredicate == null) {
+ commit.purgeTable(identifier);
+ return Optional.empty();
+ } else if (deleteIsDropPartition()) {
+ commit.dropPartitions(Collections.singletonList(deletePartitions()), identifier);
return Optional.empty();
+ } else {
+ return Optional.of(
+ TableUtils.deleteWhere(table, Collections.singletonList(deletePredicate)));
}
-
- // delete primary key related data
- return Optional.of(TableUtils.deleteWhere(table, predicates));
}
- private void checkDeletable() {
+ private void validateDeletable() {
if (table instanceof ChangelogWithKeyFileStoreTable) {
Options options = Options.fromMap(table.options());
if (options.get(MERGE_ENGINE) == MergeEngine.DEDUPLICATE) {
@@ -201,6 +196,7 @@ public class FlinkTableSink extends FlinkTableSinkBase
"merge engine '%s' can not support delete, currently only %s can support delete.",
options.get(MERGE_ENGINE), MergeEngine.DEDUPLICATE));
} else if (table instanceof ChangelogValueCountFileStoreTable) {
+ // ChangelogValueCountFileStoreTable is OK to be deleted
return;
} else if (table instanceof AppendOnlyFileStoreTable) {
throw new UnsupportedOperationException(
@@ -215,12 +211,33 @@ public class FlinkTableSink extends FlinkTableSinkBase
}
}
- private boolean canPushDownDeleteFilter(Predicate predicate) {
- DeletePushDownPrimaryKeyVisitor pkVisitor =
- new DeletePushDownPrimaryKeyVisitor(table.primaryKeys());
- isDeleteFilterPrimaryKey = predicate.visit(pkVisitor) && pkVisitor.isHitAll();
- isDeleteFilterPartitionKey =
- predicate.visit(new DeletePushDownPartitionKeyVisitor(table.partitionKeys()));
- return isDeleteFilterPartitionKey || isDeleteFilterPrimaryKey;
+ private boolean canPushDownDeleteFilter() {
+ return deletePredicate == null || deleteIsDropPartition() || deleteInSingleNode();
+ }
+
+ private boolean deleteIsDropPartition() {
+ if (deletePredicate == null) {
+ return false;
+ }
+ return deletePredicate.visit(new OnlyPartitionKeyEqualVisitor(table.partitionKeys()));
+ }
+
+ private Map<String, String> deletePartitions() {
+ if (deletePredicate == null) {
+ return null;
+ }
+ OnlyPartitionKeyEqualVisitor visitor =
+ new OnlyPartitionKeyEqualVisitor(table.partitionKeys());
+ deletePredicate.visit(visitor);
+ return visitor.partitions();
+ }
+
+ private boolean deleteInSingleNode() {
+ if (deletePredicate == null) {
+ return false;
+ }
+ return deletePredicate
+ .visit(new AllPrimaryKeyEqualVisitor(table.primaryKeys()))
+ .containsAll(table.primaryKeys());
}
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkTableSinkBase.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkTableSinkBase.java
index b9d7326f3..d64816377 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkTableSinkBase.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkTableSinkBase.java
@@ -54,9 +54,9 @@ import static org.apache.paimon.CoreOptions.MERGE_ENGINE;
public abstract class FlinkTableSinkBase
implements DynamicTableSink, SupportsOverwrite, SupportsPartitioning {
- private final ObjectIdentifier tableIdentifier;
- private final DynamicTableFactory.Context context;
- @Nullable private final LogStoreTableFactory logStoreTableFactory;
+ protected final ObjectIdentifier tableIdentifier;
+ protected final DynamicTableFactory.Context context;
+ @Nullable protected final LogStoreTableFactory logStoreTableFactory;
protected final Table table;
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ReadWriteTableITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ReadWriteTableITCase.java
index ca94bde9c..af9736e9d 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ReadWriteTableITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ReadWriteTableITCase.java
@@ -1699,11 +1699,9 @@ public class ReadWriteTableITCase extends AbstractTestBase {
"(7, 'INR', 119, '2022-01-03')",
"(8, 'MOP', 119, '2022-01-03')");
- // Step3: prepare delete statement 'where pk = x'
+ // Test1 delete statement 'where pk = x'
String deleteStatement =
String.format("DELETE FROM %s WHERE id = 2 and dt = '2022-01-01'", table);
-
- // Step4: execute delete statement and verify result
List<Row> expectedRecords =
Arrays.asList(
changelogRow("+I", 1L, "US Dollar", 114L, "2022-01-01"),
@@ -1722,37 +1720,25 @@ public class ReadWriteTableITCase extends AbstractTestBase {
.satisfies(AssertionUtils.anyCauseMatches(UnsupportedOperationException.class));
}
- // Step5: prepare delete statement 'where pk = x or pk = y'
- String deleteStatement1 =
- String.format("DELETE FROM %s WHERE id = 3 or id = 4 and dt = '2022-01-02'", table);
- List<Row> expectedRecords1 =
- Arrays.asList(
- changelogRow("+I", 1L, "US Dollar", 114L, "2022-01-01"),
- changelogRow("+I", 5L, "HKD", 119L, "2022-01-03"),
- changelogRow("+I", 6L, "CAD", 119L, "2022-01-03"),
- changelogRow("+I", 7L, "INR", 119L, "2022-01-03"),
- changelogRow("+I", 8L, "MOP", 119L, "2022-01-03"));
+ // Test2 delete statement no where
+ String deleteStatement2 = String.format("DELETE FROM %s", table);
if (supportUpdateEngines.contains(mergeEngine)) {
- bEnv.executeSql(deleteStatement1).await();
+ bEnv.executeSql(deleteStatement2).await();
String querySql = String.format("SELECT * FROM %s", table);
- testBatchRead(querySql, expectedRecords1);
+ testBatchRead(querySql, Collections.emptyList());
} else {
- assertThatThrownBy(() -> bEnv.executeSql(deleteStatement1).await())
+ assertThatThrownBy(() -> bEnv.executeSql(deleteStatement2).await())
.satisfies(AssertionUtils.anyCauseMatches(UnsupportedOperationException.class));
}
- // Step5: prepare delete statement 'where pk in (x, y, z)'
- String deleteStatement2 =
- String.format(
- "DELETE FROM %s WHERE id in (5, 6, 7, 8) and dt = '2022-01-03'", table);
- List<Row> expectedRecords2 =
- Arrays.asList(changelogRow("+I", 1L, "US Dollar", 114L, "2022-01-01"));
+ // Test3 delete statement where pt
+ String deleteStatement3 = String.format("DELETE FROM %s WHERE dt = '2022-01-03'", table);
if (supportUpdateEngines.contains(mergeEngine)) {
- bEnv.executeSql(deleteStatement2).await();
+ bEnv.executeSql(deleteStatement3).await();
String querySql = String.format("SELECT * FROM %s", table);
- testBatchRead(querySql, expectedRecords2);
+ testBatchRead(querySql, Collections.emptyList());
} else {
- assertThatThrownBy(() -> bEnv.executeSql(deleteStatement2).await())
+ assertThatThrownBy(() -> bEnv.executeSql(deleteStatement3).await())
.satisfies(AssertionUtils.anyCauseMatches(UnsupportedOperationException.class));
}
}