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 2024/06/05 10:51:34 UTC
(paimon) branch release-0.8 updated (c2c72e3be -> 66074c28f)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a change to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
from c2c72e3be [core] Fix that cannot read UPDATE_BEFORE of ignore-delete table (#3467)
new 46d9f3d5c [core] Fisrt row merge engine supports none changelog producer (#3452)
new de5474782 [doc] Delete redundant symbols in compaction doc (#3453)
new 3b6f9ea6f [core] Optimize snapshots table when specific snapshot id (#3339)
new 1c677142b [core] Fix FilesTable splits too big to distribute (#3454)
new 9e60f8eb4 [core] added greater and less than query optimization to snapshots table queries (#3396)
new d92091c57 [core] Fix should ignore the Long.MIN value when create TAG based on the watermark. (#3457)
new 432d7e23e [spark] Spark 3.2 insert overwrite test support (#3458)
new 3119130f1 [core] Fix snapshot EARLIEST first write value (#3398)
new a0a76e6fd [doc] Fix changelog producer doc to better desc (#3469)
new 66074c28f [minor] Document specify hugo version (#3471)
The 10 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
docs/README.md | 4 +-
docs/content/engines/overview.md | 6 +-
docs/content/maintenance/dedicated-compaction.md | 2 +-
.../primary-key-table/changelog-producer.md | 2 +-
docs/content/primary-key-table/merge-engine.md | 5 +-
.../main/java/org/apache/paimon/CoreOptions.java | 3 +-
.../org/apache/paimon/lookup/LookupStrategy.java | 31 ++---
.../LeafPredicateExtractor.java} | 24 ++--
.../paimon/operation/KeyValueFileStoreWrite.java | 7 +-
.../org/apache/paimon/schema/SchemaValidation.java | 5 +-
.../apache/paimon/table/ExpireSnapshotsImpl.java | 2 +-
.../table/source/InnerStreamTableScanImpl.java | 3 +-
.../apache/paimon/table/source/SingletonSplit.java | 11 +-
.../table/system/AggregationFieldsTable.java | 18 +--
.../paimon/table/system/AllTableOptionsTable.java | 23 +---
.../apache/paimon/table/system/BranchesTable.java | 16 +--
.../paimon/table/system/CatalogOptionsTable.java | 11 +-
.../apache/paimon/table/system/ConsumersTable.java | 18 +--
.../org/apache/paimon/table/system/FilesTable.java | 132 +++++++++------------
.../apache/paimon/table/system/ManifestsTable.java | 34 ++----
.../apache/paimon/table/system/OptionsTable.java | 16 +--
.../paimon/table/system/PartitionsTable.java | 8 +-
.../apache/paimon/table/system/SchemasTable.java | 22 +---
.../apache/paimon/table/system/SnapshotsTable.java | 76 ++++++++----
.../apache/paimon/table/system/StatisticTable.java | 23 +---
.../org/apache/paimon/table/system/TagsTable.java | 16 +--
.../org/apache/paimon/tag/TagTimeExtractor.java | 2 +-
.../org/apache/paimon/utils/SnapshotManager.java | 27 +++++
.../LookupChangelogMergeFunctionWrapperTest.java | 6 +-
.../org/apache/paimon/tag/TagAutoManagerTest.java | 4 +
.../apache/paimon/flink/CatalogTableITCase.java | 30 ++++-
.../org/apache/paimon/flink/FirstRowITCase.java | 24 ++--
...DLTest.scala => InsertOverwriteTableTest.scala} | 2 +-
.../spark/sql/InsertOverwriteTableTest.scala} | 2 +-
.../spark/sql/InsertOverwriteTableTest.scala} | 2 +-
.../spark/sql/InsertOverwriteTableTest.scala} | 2 +-
...teTest.scala => InsertOverwriteTableTest.scala} | 2 +-
37 files changed, 290 insertions(+), 331 deletions(-)
copy paimon-common/src/main/java/org/apache/paimon/{utils/MapBuilder.java => predicate/LeafPredicateExtractor.java} (52%)
copy paimon-common/src/test/java/org/apache/paimon/factories/DummyFactory.java => paimon-core/src/main/java/org/apache/paimon/table/source/SingletonSplit.java (77%)
copy paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/{DDLTest.scala => InsertOverwriteTableTest.scala} (92%)
copy paimon-spark/{paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/DDLTest.scala => paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTest.scala} (92%)
copy paimon-spark/{paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/DDLTest.scala => paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTest.scala} (92%)
copy paimon-spark/{paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/DDLTest.scala => paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTest.scala} (92%)
rename paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/{InsertOverwriteTest.scala => InsertOverwriteTableTest.scala} (99%)
(paimon) 08/10: [core] Fix snapshot EARLIEST first write value (#3398)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 3119130f116c571e84a285bf14fe363ba9c535cf
Author: Askwang <13...@users.noreply.github.com>
AuthorDate: Tue Jun 4 15:51:56 2024 +0800
[core] Fix snapshot EARLIEST first write value (#3398)
---
.../src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java
index 50cf39eb8..9023cf93e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java
@@ -146,7 +146,7 @@ public class ExpireSnapshotsImpl implements ExpireSnapshots {
// write the hint file in order to see the earliest snapshot directly next time
// should avoid duplicate writes when the file exists
if (snapshotManager.readHint(SnapshotManager.EARLIEST) == null) {
- writeEarliestHint(endExclusiveId);
+ writeEarliestHint(earliestId);
}
// fast exit
(paimon) 03/10: [core] Optimize snapshots table when specific snapshot id (#3339)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 3b6f9ea6f58d763dea017a8e512df70ed7a3586a
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Thu May 16 16:23:13 2024 +0800
[core] Optimize snapshots table when specific snapshot id (#3339)
---
.../paimon/predicate/LeafPredicateExtractor.java | 44 ++++++++++++++++++++++
.../org/apache/paimon/table/system/FilesTable.java | 28 +++++---------
.../apache/paimon/table/system/SnapshotsTable.java | 42 +++++++++++++--------
.../apache/paimon/flink/CatalogTableITCase.java | 10 ++++-
4 files changed, 89 insertions(+), 35 deletions(-)
diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/LeafPredicateExtractor.java b/paimon-common/src/main/java/org/apache/paimon/predicate/LeafPredicateExtractor.java
new file mode 100644
index 000000000..a5cf772c0
--- /dev/null
+++ b/paimon-common/src/main/java/org/apache/paimon/predicate/LeafPredicateExtractor.java
@@ -0,0 +1,44 @@
+/*
+ * 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.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/** Extract leaf predicate for field names. */
+public class LeafPredicateExtractor implements PredicateVisitor<Map<String, LeafPredicate>> {
+
+ public static final LeafPredicateExtractor INSTANCE = new LeafPredicateExtractor();
+
+ @Override
+ public Map<String, LeafPredicate> visit(LeafPredicate predicate) {
+ return Collections.singletonMap(predicate.fieldName(), predicate);
+ }
+
+ @Override
+ public Map<String, LeafPredicate> visit(CompoundPredicate predicate) {
+ if (predicate.function() instanceof And) {
+ Map<String, LeafPredicate> leafPredicates = new HashMap<>();
+ predicate.children().stream().map(p -> p.visit(this)).forEach(leafPredicates::putAll);
+ return leafPredicates;
+ }
+ return Collections.emptyMap();
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/FilesTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/FilesTable.java
index 0501dab08..1f6eb2e0e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/FilesTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/FilesTable.java
@@ -28,8 +28,8 @@ import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataFilePathFactory;
import org.apache.paimon.predicate.Equal;
import org.apache.paimon.predicate.LeafPredicate;
+import org.apache.paimon.predicate.LeafPredicateExtractor;
import org.apache.paimon.predicate.Predicate;
-import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.TableSchema;
@@ -162,25 +162,15 @@ public class FilesTable implements ReadonlyTable {
@Override
public InnerTableScan withFilter(Predicate pushdown) {
- List<Predicate> predicates = PredicateBuilder.splitAnd(pushdown);
- for (Predicate predicate : predicates) {
- if (predicate instanceof LeafPredicate) {
- LeafPredicate leaf = (LeafPredicate) predicate;
- switch (leaf.fieldName()) {
- case "partition":
- this.partitionPredicate = leaf;
- break;
- case "bucket":
- this.bucketPredicate = leaf;
- break;
- case "level":
- this.levelPredicate = leaf;
- break;
- default:
- break;
- }
- }
+ if (pushdown == null) {
+ return this;
}
+
+ Map<String, LeafPredicate> leafPredicates =
+ pushdown.visit(LeafPredicateExtractor.INSTANCE);
+ this.partitionPredicate = leafPredicates.get("partition");
+ this.bucketPredicate = leafPredicates.get("bucket");
+ this.levelPredicate = leafPredicates.get("level");
return this;
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/SnapshotsTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/SnapshotsTable.java
index 8d6b545d9..b8ff9cd8b 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/SnapshotsTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/SnapshotsTable.java
@@ -26,6 +26,9 @@ import org.apache.paimon.data.Timestamp;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.predicate.Equal;
+import org.apache.paimon.predicate.LeafPredicate;
+import org.apache.paimon.predicate.LeafPredicateExtractor;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.FileStoreTable;
@@ -47,6 +50,8 @@ import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.time.Instant;
import java.time.LocalDateTime;
@@ -140,19 +145,13 @@ public class SnapshotsTable implements ReadonlyTable {
@Override
public InnerTableScan withFilter(Predicate predicate) {
- // TODO
+ // do filter in read
return this;
}
@Override
public Plan innerPlan() {
- long rowCount;
- try {
- rowCount = new SnapshotManager(fileIO, location).snapshotCount();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- return () -> Collections.singletonList(new SnapshotsSplit(rowCount, location));
+ return () -> Collections.singletonList(new SnapshotsSplit(location));
}
}
@@ -160,17 +159,16 @@ public class SnapshotsTable implements ReadonlyTable {
private static final long serialVersionUID = 1L;
- private final long rowCount;
private final Path location;
- private SnapshotsSplit(long rowCount, Path location) {
+ private SnapshotsSplit(Path location) {
this.location = location;
- this.rowCount = rowCount;
}
@Override
public long rowCount() {
- return rowCount;
+ // dummy 1, just 1 parallelism
+ return 1;
}
@Override
@@ -195,6 +193,7 @@ public class SnapshotsTable implements ReadonlyTable {
private final FileIO fileIO;
private int[][] projection;
+ @Nullable private Long specificSnapshot;
public SnapshotsRead(FileIO fileIO) {
this.fileIO = fileIO;
@@ -202,7 +201,15 @@ public class SnapshotsTable implements ReadonlyTable {
@Override
public InnerTableRead withFilter(Predicate predicate) {
- // TODO
+ if (predicate == null) {
+ return this;
+ }
+
+ LeafPredicate snapshotPred =
+ predicate.visit(LeafPredicateExtractor.INSTANCE).get("snapshot_id");
+ if (snapshotPred != null && snapshotPred.function() instanceof Equal) {
+ specificSnapshot = (Long) snapshotPred.literals().get(0);
+ }
return this;
}
@@ -222,8 +229,13 @@ public class SnapshotsTable implements ReadonlyTable {
if (!(split instanceof SnapshotsSplit)) {
throw new IllegalArgumentException("Unsupported split: " + split.getClass());
}
- Path location = ((SnapshotsSplit) split).location;
- Iterator<Snapshot> snapshots = new SnapshotManager(fileIO, location).snapshots();
+ SnapshotManager snapshotManager =
+ new SnapshotManager(fileIO, ((SnapshotsSplit) split).location);
+ Iterator<Snapshot> snapshots =
+ specificSnapshot != null
+ ? Collections.singletonList(snapshotManager.snapshot(specificSnapshot))
+ .iterator()
+ : snapshotManager.snapshots();
Iterator<InternalRow> rows = Iterators.transform(snapshots, this::toRow);
if (projection != null) {
rows =
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java
index 6b167c4e7..a94d91af6 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java
@@ -64,9 +64,17 @@ public class CatalogTableITCase extends CatalogITCaseBase {
sql("INSERT INTO T VALUES (3, 4)");
List<Row> result = sql("SELECT snapshot_id, schema_id, commit_kind FROM T$snapshots");
+ assertThat(result).containsExactly(Row.of(1L, 0L, "APPEND"), Row.of(2L, 0L, "APPEND"));
- // check correctness and sequence snapshots.
+ result =
+ sql(
+ "SELECT snapshot_id, schema_id, commit_kind FROM T$snapshots WHERE schema_id = 0");
assertThat(result).containsExactly(Row.of(1L, 0L, "APPEND"), Row.of(2L, 0L, "APPEND"));
+
+ result =
+ sql(
+ "SELECT snapshot_id, schema_id, commit_kind FROM T$snapshots WHERE snapshot_id = 2");
+ assertThat(result).containsExactly(Row.of(2L, 0L, "APPEND"));
}
@Test
(paimon) 10/10: [minor] Document specify hugo version (#3471)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 66074c28f0f414186cb3e48a6ed3e51465eee0be
Author: wgcn <10...@qq.com>
AuthorDate: Tue Jun 4 22:34:37 2024 +0800
[minor] Document specify hugo version (#3471)
---
docs/README.md | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/docs/README.md b/docs/README.md
index 439ab5d14..de0a7d16b 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -8,8 +8,8 @@ that you always have docs corresponding to your checked-out version.
### Build the site locally
-Make sure you have installed
-[Hugo](https://gohugo.io/getting-started/installing/) on your system.
+Make sure you have installed Hugo on your system.
+Note: An extended version of Hugo <= 0.124.1 is required. you can Find this at [Hugo](https://github.com/gohugoio/hugo/releases/tag/v0.124.1)
From this directory:
(paimon) 04/10: [core] Fix FilesTable splits too big to distribute (#3454)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 1c677142b4c16bc1729d1499ea45cf41321614ba
Author: YeJunHao <41...@users.noreply.github.com>
AuthorDate: Fri May 31 19:39:19 2024 +0800
[core] Fix FilesTable splits too big to distribute (#3454)
---
.../apache/paimon/table/source/SingletonSplit.java | 28 ++++++
.../table/system/AggregationFieldsTable.java | 18 +---
.../paimon/table/system/AllTableOptionsTable.java | 23 +----
.../apache/paimon/table/system/BranchesTable.java | 16 +---
.../paimon/table/system/CatalogOptionsTable.java | 11 +--
.../apache/paimon/table/system/ConsumersTable.java | 18 +---
.../org/apache/paimon/table/system/FilesTable.java | 104 ++++++++++-----------
.../apache/paimon/table/system/ManifestsTable.java | 34 ++-----
.../apache/paimon/table/system/OptionsTable.java | 16 +---
.../paimon/table/system/PartitionsTable.java | 8 +-
.../apache/paimon/table/system/SchemasTable.java | 22 +----
.../apache/paimon/table/system/SnapshotsTable.java | 9 +-
.../apache/paimon/table/system/StatisticTable.java | 23 +----
.../org/apache/paimon/table/system/TagsTable.java | 16 +---
14 files changed, 125 insertions(+), 221 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/SingletonSplit.java b/paimon-core/src/main/java/org/apache/paimon/table/source/SingletonSplit.java
new file mode 100644
index 000000000..e5a649a72
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/SingletonSplit.java
@@ -0,0 +1,28 @@
+/*
+ * 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.table.source;
+
+/** Singleton split use for system table, in which, scan always just produce one split. */
+public abstract class SingletonSplit implements Split {
+
+ @Override
+ public long rowCount() {
+ return 1;
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/AggregationFieldsTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/AggregationFieldsTable.java
index 29bec8502..0aa17b1b6 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/AggregationFieldsTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/AggregationFieldsTable.java
@@ -33,6 +33,7 @@ import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.InnerTableRead;
import org.apache.paimon.table.source.InnerTableScan;
import org.apache.paimon.table.source.ReadOnceTableScan;
+import org.apache.paimon.table.source.SingletonSplit;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableRead;
import org.apache.paimon.types.DataField;
@@ -120,32 +121,21 @@ public class AggregationFieldsTable implements ReadonlyTable {
@Override
public Plan innerPlan() {
- return () ->
- Collections.singletonList(
- new AggregationSplit(
- new SchemaManager(fileIO, location).listAllIds().size(),
- location));
+ return () -> Collections.singletonList(new AggregationSplit(location));
}
}
/** {@link Split} implementation for {@link AggregationFieldsTable}. */
- private static class AggregationSplit implements Split {
+ private static class AggregationSplit extends SingletonSplit {
private static final long serialVersionUID = 1L;
- private final long rowCount;
private final Path location;
- private AggregationSplit(long rowCount, Path location) {
- this.rowCount = rowCount;
+ private AggregationSplit(Path location) {
this.location = location;
}
- @Override
- public long rowCount() {
- return rowCount;
- }
-
@Override
public boolean equals(Object o) {
if (this == o) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/AllTableOptionsTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/AllTableOptionsTable.java
index a7b1236bf..eb50a38a5 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/AllTableOptionsTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/AllTableOptionsTable.java
@@ -32,6 +32,7 @@ import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.InnerTableRead;
import org.apache.paimon.table.source.InnerTableScan;
import org.apache.paimon.table.source.ReadOnceTableScan;
+import org.apache.paimon.table.source.SingletonSplit;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableRead;
import org.apache.paimon.types.DataField;
@@ -42,7 +43,6 @@ import org.apache.paimon.utils.ProjectedRow;
import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -121,33 +121,20 @@ public class AllTableOptionsTable implements ReadonlyTable {
@Override
public Plan innerPlan() {
- return () ->
- Collections.singletonList(
- new AllTableSplit(
- options(fileIO, allTablePaths).values().stream()
- .flatMap(t -> t.values().stream())
- .reduce(0, (a, b) -> a + b.size(), Integer::sum),
- allTablePaths));
+ return () -> Collections.singletonList(new AllTableSplit(allTablePaths));
}
}
- private static class AllTableSplit implements Split {
+ private static class AllTableSplit extends SingletonSplit {
private static final long serialVersionUID = 1L;
- private final long rowCount;
private final Map<String, Map<String, Path>> allTablePaths;
- private AllTableSplit(long rowCount, Map<String, Map<String, Path>> allTablePaths) {
- this.rowCount = rowCount;
+ private AllTableSplit(Map<String, Map<String, Path>> allTablePaths) {
this.allTablePaths = allTablePaths;
}
- @Override
- public long rowCount() {
- return rowCount;
- }
-
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -192,7 +179,7 @@ public class AllTableOptionsTable implements ReadonlyTable {
}
@Override
- public RecordReader<InternalRow> createReader(Split split) throws IOException {
+ public RecordReader<InternalRow> createReader(Split split) {
if (!(split instanceof AllTableSplit)) {
throw new IllegalArgumentException("Unsupported split: " + split.getClass());
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/BranchesTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/BranchesTable.java
index 7ae31095c..f37370641 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/BranchesTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/BranchesTable.java
@@ -35,6 +35,7 @@ import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.InnerTableRead;
import org.apache.paimon.table.source.InnerTableScan;
import org.apache.paimon.table.source.ReadOnceTableScan;
+import org.apache.paimon.table.source.SingletonSplit;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableRead;
import org.apache.paimon.types.BigIntType;
@@ -122,28 +123,19 @@ public class BranchesTable implements ReadonlyTable {
@Override
public Plan innerPlan() {
- FileStoreTable table = FileStoreTableFactory.create(fileIO, location);
- long rowCount = table.branchManager().branchCount();
- return () -> Collections.singletonList(new BranchesSplit(rowCount, location));
+ return () -> Collections.singletonList(new BranchesSplit(location));
}
}
- private static class BranchesSplit implements Split {
+ private static class BranchesSplit extends SingletonSplit {
private static final long serialVersionUID = 1L;
- private final long rowCount;
private final Path location;
- private BranchesSplit(long rowCount, Path location) {
- this.rowCount = rowCount;
+ private BranchesSplit(Path location) {
this.location = location;
}
- @Override
- public long rowCount() {
- return rowCount;
- }
-
@Override
public boolean equals(Object o) {
if (this == o) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/CatalogOptionsTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/CatalogOptionsTable.java
index 2e94edf1e..699d0324f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/CatalogOptionsTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/CatalogOptionsTable.java
@@ -30,6 +30,7 @@ import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.InnerTableRead;
import org.apache.paimon.table.source.InnerTableScan;
import org.apache.paimon.table.source.ReadOnceTableScan;
+import org.apache.paimon.table.source.SingletonSplit;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableRead;
import org.apache.paimon.types.DataField;
@@ -39,7 +40,6 @@ import org.apache.paimon.utils.ProjectedRow;
import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
-import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
import java.util.Iterator;
@@ -113,7 +113,7 @@ public class CatalogOptionsTable implements ReadonlyTable {
}
}
- private static class CatalogOptionsSplit implements Split {
+ private static class CatalogOptionsSplit extends SingletonSplit {
private static final long serialVersionUID = 1L;
@@ -123,11 +123,6 @@ public class CatalogOptionsTable implements ReadonlyTable {
this.catalogOptions = catalogOptions.toMap();
}
- @Override
- public long rowCount() {
- return catalogOptions.size();
- }
-
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -168,7 +163,7 @@ public class CatalogOptionsTable implements ReadonlyTable {
}
@Override
- public RecordReader<InternalRow> createReader(Split split) throws IOException {
+ public RecordReader<InternalRow> createReader(Split split) {
if (!(split instanceof CatalogOptionsTable.CatalogOptionsSplit)) {
throw new IllegalArgumentException("Unsupported split: " + split.getClass());
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/ConsumersTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/ConsumersTable.java
index a3ec3017e..05b3382d2 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/ConsumersTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/ConsumersTable.java
@@ -32,6 +32,7 @@ import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.InnerTableRead;
import org.apache.paimon.table.source.InnerTableScan;
import org.apache.paimon.table.source.ReadOnceTableScan;
+import org.apache.paimon.table.source.SingletonSplit;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableRead;
import org.apache.paimon.types.BigIntType;
@@ -114,32 +115,21 @@ public class ConsumersTable implements ReadonlyTable {
@Override
public Plan innerPlan() {
- return () ->
- Collections.singletonList(
- new ConsumersTable.ConsumersSplit(
- new ConsumerManager(fileIO, location).listAllIds().size(),
- location));
+ return () -> Collections.singletonList(new ConsumersTable.ConsumersSplit(location));
}
}
/** {@link Split} implementation for {@link ConsumersTable}. */
- private static class ConsumersSplit implements Split {
+ private static class ConsumersSplit extends SingletonSplit {
private static final long serialVersionUID = 1L;
- private final long rowCount;
private final Path location;
- private ConsumersSplit(long rowCount, Path location) {
- this.rowCount = rowCount;
+ private ConsumersSplit(Path location) {
this.location = location;
}
- @Override
- public long rowCount() {
- return rowCount;
- }
-
@Override
public boolean equals(Object o) {
if (this == o) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/FilesTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/FilesTable.java
index 1f6eb2e0e..df5a13881 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/FilesTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/FilesTable.java
@@ -43,6 +43,7 @@ import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.InnerTableRead;
import org.apache.paimon.table.source.InnerTableScan;
import org.apache.paimon.table.source.ReadOnceTableScan;
+import org.apache.paimon.table.source.SingletonSplit;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableRead;
import org.apache.paimon.table.source.TableScan;
@@ -62,7 +63,6 @@ import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
import javax.annotation.Nullable;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -134,7 +134,7 @@ public class FilesTable implements ReadonlyTable {
@Override
public InnerTableScan newScan() {
- return new FilesScan(storeTable);
+ return new FilesScan();
}
@Override
@@ -150,16 +150,10 @@ public class FilesTable implements ReadonlyTable {
private static class FilesScan extends ReadOnceTableScan {
- private final FileStoreTable storeTable;
-
@Nullable private LeafPredicate partitionPredicate;
@Nullable private LeafPredicate bucketPredicate;
@Nullable private LeafPredicate levelPredicate;
- private FilesScan(FileStoreTable storeTable) {
- this.storeTable = storeTable;
- }
-
@Override
public InnerTableScan withFilter(Predicate pushdown) {
if (pushdown == null) {
@@ -176,12 +170,51 @@ public class FilesTable implements ReadonlyTable {
@Override
public Plan innerPlan() {
- // plan here, just set the result of plan to split
- TableScan.Plan plan = tablePlan();
- return () -> Collections.singletonList(new FilesSplit(plan.splits()));
+ return () ->
+ Collections.singletonList(
+ new FilesSplit(partitionPredicate, bucketPredicate, levelPredicate));
+ }
+ }
+
+ private static class FilesSplit extends SingletonSplit {
+
+ @Nullable private final LeafPredicate partitionPredicate;
+ @Nullable private final LeafPredicate bucketPredicate;
+ @Nullable private final LeafPredicate levelPredicate;
+
+ private FilesSplit(
+ @Nullable LeafPredicate partitionPredicate,
+ @Nullable LeafPredicate bucketPredicate,
+ @Nullable LeafPredicate levelPredicate) {
+ this.partitionPredicate = partitionPredicate;
+ this.bucketPredicate = bucketPredicate;
+ this.levelPredicate = levelPredicate;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ FilesSplit that = (FilesSplit) o;
+ return Objects.equals(partitionPredicate, that.partitionPredicate)
+ && Objects.equals(bucketPredicate, that.bucketPredicate)
+ && Objects.equals(this.levelPredicate, that.levelPredicate);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(partitionPredicate, bucketPredicate, levelPredicate);
+ }
+
+ public List<Split> splits(FileStoreTable storeTable) {
+ return tablePlan(storeTable).splits();
}
- private TableScan.Plan tablePlan() {
+ private TableScan.Plan tablePlan(FileStoreTable storeTable) {
InnerTableScan scan = storeTable.newScan();
if (partitionPredicate != null) {
if (partitionPredicate.function() instanceof Equal) {
@@ -221,46 +254,6 @@ public class FilesTable implements ReadonlyTable {
}
}
- private static class FilesSplit implements Split {
-
- private static final long serialVersionUID = 1L;
-
- private final List<Split> splits;
-
- private FilesSplit(List<Split> splits) {
- this.splits = splits;
- }
-
- @Override
- public long rowCount() {
- return splits.stream()
- .map(s -> (DataSplit) s)
- .mapToLong(s -> s.dataFiles().size())
- .sum();
- }
-
- public List<Split> splits() {
- return splits;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- FilesSplit that = (FilesSplit) o;
- return Objects.equals(splits, that.splits);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(splits);
- }
- }
-
private static class FilesRead implements InnerTableRead {
private final SchemaManager schemaManager;
@@ -292,12 +285,13 @@ public class FilesTable implements ReadonlyTable {
}
@Override
- public RecordReader<InternalRow> createReader(Split split) throws IOException {
+ public RecordReader<InternalRow> createReader(Split split) {
if (!(split instanceof FilesSplit)) {
throw new IllegalArgumentException("Unsupported split: " + split.getClass());
}
FilesSplit filesSplit = (FilesSplit) split;
- if (filesSplit.splits().isEmpty()) {
+ List<Split> splits = filesSplit.splits(storeTable);
+ if (splits.isEmpty()) {
return new IteratorRecordReader<>(Collections.emptyIterator());
}
@@ -332,7 +326,7 @@ public class FilesTable implements ReadonlyTable {
});
}
};
- for (Split dataSplit : filesSplit.splits()) {
+ for (Split dataSplit : splits) {
iteratorList.add(
Iterators.transform(
((DataSplit) dataSplit).dataFiles().iterator(),
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/ManifestsTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/ManifestsTable.java
index 7d7a9570e..e07ff602c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/ManifestsTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/ManifestsTable.java
@@ -35,6 +35,7 @@ import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.InnerTableRead;
import org.apache.paimon.table.source.InnerTableScan;
import org.apache.paimon.table.source.ReadOnceTableScan;
+import org.apache.paimon.table.source.SingletonSplit;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableRead;
import org.apache.paimon.types.BigIntType;
@@ -48,13 +49,11 @@ import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
-import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
import static org.apache.paimon.catalog.Catalog.SYSTEM_TABLE_SPLITTER;
@@ -109,7 +108,7 @@ public class ManifestsTable implements ReadonlyTable {
return new ManifestsTable(dataTable.copy(dynamicOptions));
}
- private class ManifestsScan extends ReadOnceTableScan {
+ private static class ManifestsScan extends ReadOnceTableScan {
@Override
public InnerTableScan withFilter(Predicate predicate) {
@@ -119,41 +118,22 @@ public class ManifestsTable implements ReadonlyTable {
@Override
protected Plan innerPlan() {
- return () ->
- Collections.singletonList(new ManifestsSplit(allManifests(dataTable).size()));
+ return () -> Collections.singletonList(new ManifestsSplit());
}
}
- private static class ManifestsSplit implements Split {
+ private static class ManifestsSplit extends SingletonSplit {
private static final long serialVersionUID = 1L;
- private final long rowCount;
-
- private ManifestsSplit(long rowCount) {
- this.rowCount = rowCount;
- }
-
- @Override
- public long rowCount() {
- return rowCount;
- }
+ private ManifestsSplit() {}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- ManifestsSplit that = (ManifestsSplit) o;
- return Objects.equals(rowCount, that.rowCount);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(rowCount);
+ return o != null && getClass() == o.getClass();
}
}
@@ -185,7 +165,7 @@ public class ManifestsTable implements ReadonlyTable {
}
@Override
- public RecordReader<InternalRow> createReader(Split split) throws IOException {
+ public RecordReader<InternalRow> createReader(Split split) {
if (!(split instanceof ManifestsSplit)) {
throw new IllegalArgumentException("Unsupported split: " + split.getClass());
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/OptionsTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/OptionsTable.java
index b740ddec1..0c8ac2f25 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/OptionsTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/OptionsTable.java
@@ -32,6 +32,7 @@ import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.InnerTableRead;
import org.apache.paimon.table.source.InnerTableScan;
import org.apache.paimon.table.source.ReadOnceTableScan;
+import org.apache.paimon.table.source.SingletonSplit;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableRead;
import org.apache.paimon.types.DataField;
@@ -112,29 +113,20 @@ public class OptionsTable implements ReadonlyTable {
@Override
public Plan innerPlan() {
- return () ->
- Collections.singletonList(
- new OptionsSplit(options(fileIO, location).size(), location));
+ return () -> Collections.singletonList(new OptionsSplit(location));
}
}
- private static class OptionsSplit implements Split {
+ private static class OptionsSplit extends SingletonSplit {
private static final long serialVersionUID = 1L;
- private final long rowCount;
private final Path location;
- private OptionsSplit(long rowCount, Path location) {
- this.rowCount = rowCount;
+ private OptionsSplit(Path location) {
this.location = location;
}
- @Override
- public long rowCount() {
- return rowCount;
- }
-
@Override
public boolean equals(Object o) {
if (this == o) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/PartitionsTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/PartitionsTable.java
index c08cdfaa3..76c0768ee 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/PartitionsTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/PartitionsTable.java
@@ -32,6 +32,7 @@ import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.InnerTableRead;
import org.apache.paimon.table.source.InnerTableScan;
import org.apache.paimon.table.source.ReadOnceTableScan;
+import org.apache.paimon.table.source.SingletonSplit;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableRead;
import org.apache.paimon.types.BigIntType;
@@ -124,15 +125,10 @@ public class PartitionsTable implements ReadonlyTable {
}
}
- private static class PartitionsSplit implements Split {
+ private static class PartitionsSplit extends SingletonSplit {
private static final long serialVersionUID = 1L;
- @Override
- public long rowCount() {
- return 1;
- }
-
@Override
public boolean equals(Object o) {
if (this == o) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/SchemasTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/SchemasTable.java
index 127323b42..2dcbcd2d9 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/SchemasTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/SchemasTable.java
@@ -34,6 +34,7 @@ import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.InnerTableRead;
import org.apache.paimon.table.source.InnerTableScan;
import org.apache.paimon.table.source.ReadOnceTableScan;
+import org.apache.paimon.table.source.SingletonSplit;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableRead;
import org.apache.paimon.types.BigIntType;
@@ -47,7 +48,6 @@ import org.apache.paimon.utils.SerializationUtils;
import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
-import java.io.IOException;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
@@ -127,33 +127,21 @@ public class SchemasTable implements ReadonlyTable {
@Override
public Plan innerPlan() {
- return () ->
- Collections.singletonList(
- new SchemasSplit(
- new SchemaManager(fileIO, location).listAllIds().size(),
- location));
+ return () -> Collections.singletonList(new SchemasSplit(location));
}
}
/** {@link Split} implementation for {@link SchemasTable}. */
- private static class SchemasSplit implements Split {
+ private static class SchemasSplit extends SingletonSplit {
private static final long serialVersionUID = 1L;
- private final long rowCount;
private final Path location;
- private SchemasSplit(long rowCount, Path location) {
- this.rowCount = rowCount;
+ private SchemasSplit(Path location) {
this.location = location;
}
- @Override
- public long rowCount() {
- return rowCount;
- }
-
- @Override
public boolean equals(Object o) {
if (this == o) {
return true;
@@ -198,7 +186,7 @@ public class SchemasTable implements ReadonlyTable {
}
@Override
- public RecordReader<InternalRow> createReader(Split split) throws IOException {
+ public RecordReader<InternalRow> createReader(Split split) {
if (!(split instanceof SchemasSplit)) {
throw new IllegalArgumentException("Unsupported split: " + split.getClass());
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/SnapshotsTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/SnapshotsTable.java
index b8ff9cd8b..991129038 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/SnapshotsTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/SnapshotsTable.java
@@ -37,6 +37,7 @@ import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.InnerTableRead;
import org.apache.paimon.table.source.InnerTableScan;
import org.apache.paimon.table.source.ReadOnceTableScan;
+import org.apache.paimon.table.source.SingletonSplit;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableRead;
import org.apache.paimon.types.BigIntType;
@@ -155,7 +156,7 @@ public class SnapshotsTable implements ReadonlyTable {
}
}
- private static class SnapshotsSplit implements Split {
+ private static class SnapshotsSplit extends SingletonSplit {
private static final long serialVersionUID = 1L;
@@ -165,12 +166,6 @@ public class SnapshotsTable implements ReadonlyTable {
this.location = location;
}
- @Override
- public long rowCount() {
- // dummy 1, just 1 parallelism
- return 1;
- }
-
@Override
public boolean equals(Object o) {
if (this == o) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/StatisticTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/StatisticTable.java
index 682b5b775..600c85a6b 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/StatisticTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/StatisticTable.java
@@ -34,6 +34,7 @@ import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.InnerTableRead;
import org.apache.paimon.table.source.InnerTableScan;
import org.apache.paimon.table.source.ReadOnceTableScan;
+import org.apache.paimon.table.source.SingletonSplit;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableRead;
import org.apache.paimon.types.BigIntType;
@@ -43,7 +44,6 @@ import org.apache.paimon.utils.IteratorRecordReader;
import org.apache.paimon.utils.JsonSerdeUtil;
import org.apache.paimon.utils.ProjectedRow;
import org.apache.paimon.utils.SerializationUtils;
-import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
@@ -125,33 +125,18 @@ public class StatisticTable implements ReadonlyTable {
@Override
public Plan innerPlan() {
- long rowCount;
- try {
- rowCount = new SnapshotManager(fileIO, location).snapshotCount();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- return () ->
- Collections.singletonList(
- new StatisticTable.StatisticSplit(rowCount, location));
+ return () -> Collections.singletonList(new StatisticTable.StatisticSplit(location));
}
}
- private static class StatisticSplit implements Split {
+ private static class StatisticSplit extends SingletonSplit {
private static final long serialVersionUID = 1L;
- private final long rowCount;
private final Path location;
- private StatisticSplit(long rowCount, Path location) {
+ private StatisticSplit(Path location) {
this.location = location;
- this.rowCount = rowCount;
- }
-
- @Override
- public long rowCount() {
- return rowCount;
}
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/TagsTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/TagsTable.java
index 0c1c4aa29..9e311ae15 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/TagsTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/TagsTable.java
@@ -32,6 +32,7 @@ import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.InnerTableRead;
import org.apache.paimon.table.source.InnerTableScan;
import org.apache.paimon.table.source.ReadOnceTableScan;
+import org.apache.paimon.table.source.SingletonSplit;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableRead;
import org.apache.paimon.tag.Tag;
@@ -126,29 +127,20 @@ public class TagsTable implements ReadonlyTable {
@Override
public Plan innerPlan() {
- return () ->
- Collections.singletonList(
- new TagsSplit(new TagManager(fileIO, location).tagCount(), location));
+ return () -> Collections.singletonList(new TagsSplit(location));
}
}
- private static class TagsSplit implements Split {
+ private static class TagsSplit extends SingletonSplit {
private static final long serialVersionUID = 1L;
- private final long rowCount;
private final Path location;
- private TagsSplit(long rowCount, Path location) {
- this.rowCount = rowCount;
+ private TagsSplit(Path location) {
this.location = location;
}
- @Override
- public long rowCount() {
- return rowCount;
- }
-
@Override
public boolean equals(Object o) {
if (this == o) {
(paimon) 06/10: [core] Fix should ignore the Long.MIN value when create TAG based on the watermark. (#3457)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit d92091c57fb9453b035b999c63c63e20565ef93e
Author: HunterXHunter <13...@qq.com>
AuthorDate: Mon Jun 3 17:19:40 2024 +0800
[core] Fix should ignore the Long.MIN value when create TAG based on the watermark. (#3457)
---
paimon-core/src/main/java/org/apache/paimon/tag/TagTimeExtractor.java | 2 +-
.../src/test/java/org/apache/paimon/tag/TagAutoManagerTest.java | 4 ++++
2 files changed, 5 insertions(+), 1 deletion(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/tag/TagTimeExtractor.java b/paimon-core/src/main/java/org/apache/paimon/tag/TagTimeExtractor.java
index e017b2151..29bc34503 100644
--- a/paimon-core/src/main/java/org/apache/paimon/tag/TagTimeExtractor.java
+++ b/paimon-core/src/main/java/org/apache/paimon/tag/TagTimeExtractor.java
@@ -56,7 +56,7 @@ public interface TagTimeExtractor {
@Override
public Optional<LocalDateTime> extract(long timeMilli, @Nullable Long watermark) {
- if (watermark == null) {
+ if (watermark == null || watermark < 0) {
return Optional.empty();
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/tag/TagAutoManagerTest.java b/paimon-core/src/test/java/org/apache/paimon/tag/TagAutoManagerTest.java
index a065945ba..84678048a 100644
--- a/paimon-core/src/test/java/org/apache/paimon/tag/TagAutoManagerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/tag/TagAutoManagerTest.java
@@ -135,6 +135,10 @@ public class TagAutoManagerTest extends PrimaryKeyTableTestBase {
TableCommitImpl commit = table.newCommit(commitUser).ignoreEmptyCommit(false);
TagManager tagManager = table.store().newTagManager();
+ // test watermark is Long.MIN_VALUE.
+ commit.commit(new ManifestCommittable(0, Long.MIN_VALUE));
+ assertThat(tagManager.allTagNames()).isEmpty();
+
// test first create
commit.commit(new ManifestCommittable(0, localZoneMills("2023-07-18T12:00:09")));
assertThat(tagManager.allTagNames()).containsOnly("2023-07-18 11");
(paimon) 02/10: [doc] Delete redundant symbols in compaction doc (#3453)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit de5474782c02397f9ff928e2c0f0de6458171bff
Author: xleoken <le...@163.com>
AuthorDate: Fri May 31 18:10:44 2024 +0800
[doc] Delete redundant symbols in compaction doc (#3453)
---
docs/content/maintenance/dedicated-compaction.md | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/docs/content/maintenance/dedicated-compaction.md b/docs/content/maintenance/dedicated-compaction.md
index ae975f4c9..d77e7e41c 100644
--- a/docs/content/maintenance/dedicated-compaction.md
+++ b/docs/content/maintenance/dedicated-compaction.md
@@ -96,7 +96,7 @@ Run the following command to submit a compaction job for the table.
--table <table-name> \
[--partition <partition-name>] \
[--table_conf <table_conf>] \
- [--catalog_conf <paimon-catalog-conf> [--catalog_conf <paimon-catalog-conf> ...]] \
+ [--catalog_conf <paimon-catalog-conf> [--catalog_conf <paimon-catalog-conf> ...]]
```
Example: compact table
(paimon) 09/10: [doc] Fix changelog producer doc to better desc (#3469)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit a0a76e6fd776490e8fe175ed020a8828c89c4ab1
Author: wangwj <ho...@gmail.com>
AuthorDate: Tue Jun 4 19:12:57 2024 +0800
[doc] Fix changelog producer doc to better desc (#3469)
---
docs/content/primary-key-table/changelog-producer.md | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/docs/content/primary-key-table/changelog-producer.md b/docs/content/primary-key-table/changelog-producer.md
index 3ddfbf158..45723da1f 100644
--- a/docs/content/primary-key-table/changelog-producer.md
+++ b/docs/content/primary-key-table/changelog-producer.md
@@ -26,7 +26,7 @@ under the License.
# Changelog Producer
-Streaming queries will continuously produce the latest changes.
+Streaming write can continuously produce the latest changes for streaming read.
By specifying the `changelog-producer` table property when creating the table, users can choose the pattern of changes produced from table files.
(paimon) 05/10: [core] added greater and less than query optimization to snapshots table queries (#3396)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 9e60f8eb4645a63d65974172341838f0088bf063
Author: ctrlaltdilj <10...@users.noreply.github.com>
AuthorDate: Mon Jun 3 22:04:47 2024 -0400
[core] added greater and less than query optimization to snapshots table queries (#3396)
---
.../apache/paimon/table/system/SnapshotsTable.java | 47 +++++++++++++++++-----
.../org/apache/paimon/utils/SnapshotManager.java | 27 +++++++++++++
.../apache/paimon/flink/CatalogTableITCase.java | 20 +++++++++
3 files changed, 85 insertions(+), 9 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/SnapshotsTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/SnapshotsTable.java
index 991129038..ca7bc925d 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/SnapshotsTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/SnapshotsTable.java
@@ -27,8 +27,12 @@ import org.apache.paimon.disk.IOManager;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.predicate.Equal;
+import org.apache.paimon.predicate.GreaterOrEqual;
+import org.apache.paimon.predicate.GreaterThan;
import org.apache.paimon.predicate.LeafPredicate;
import org.apache.paimon.predicate.LeafPredicateExtractor;
+import org.apache.paimon.predicate.LessOrEqual;
+import org.apache.paimon.predicate.LessThan;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.FileStoreTable;
@@ -51,8 +55,6 @@ import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
-import javax.annotation.Nullable;
-
import java.io.IOException;
import java.time.Instant;
import java.time.LocalDateTime;
@@ -63,6 +65,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
import static org.apache.paimon.catalog.Catalog.SYSTEM_TABLE_SPLITTER;
@@ -188,7 +191,8 @@ public class SnapshotsTable implements ReadonlyTable {
private final FileIO fileIO;
private int[][] projection;
- @Nullable private Long specificSnapshot;
+ private Optional<Long> optionalFilterSnapshotIdMax = Optional.empty();
+ private Optional<Long> optionalFilterSnapshotIdMin = Optional.empty();
public SnapshotsRead(FileIO fileIO) {
this.fileIO = fileIO;
@@ -202,9 +206,35 @@ public class SnapshotsTable implements ReadonlyTable {
LeafPredicate snapshotPred =
predicate.visit(LeafPredicateExtractor.INSTANCE).get("snapshot_id");
- if (snapshotPred != null && snapshotPred.function() instanceof Equal) {
- specificSnapshot = (Long) snapshotPred.literals().get(0);
+ if (snapshotPred != null) {
+ if (snapshotPred.function() instanceof Equal) {
+ optionalFilterSnapshotIdMin =
+ Optional.of((Long) snapshotPred.literals().get(0));
+ optionalFilterSnapshotIdMax =
+ Optional.of((Long) snapshotPred.literals().get(0));
+ }
+
+ if (snapshotPred.function() instanceof GreaterThan) {
+ optionalFilterSnapshotIdMin =
+ Optional.of((Long) snapshotPred.literals().get(0) + 1);
+ }
+
+ if (snapshotPred.function() instanceof GreaterOrEqual) {
+ optionalFilterSnapshotIdMin =
+ Optional.of((Long) snapshotPred.literals().get(0));
+ }
+
+ if (snapshotPred.function() instanceof LessThan) {
+ optionalFilterSnapshotIdMax =
+ Optional.of((Long) snapshotPred.literals().get(0) - 1);
+ }
+
+ if (snapshotPred.function() instanceof LessOrEqual) {
+ optionalFilterSnapshotIdMax =
+ Optional.of((Long) snapshotPred.literals().get(0));
+ }
}
+
return this;
}
@@ -227,10 +257,9 @@ public class SnapshotsTable implements ReadonlyTable {
SnapshotManager snapshotManager =
new SnapshotManager(fileIO, ((SnapshotsSplit) split).location);
Iterator<Snapshot> snapshots =
- specificSnapshot != null
- ? Collections.singletonList(snapshotManager.snapshot(specificSnapshot))
- .iterator()
- : snapshotManager.snapshots();
+ snapshotManager.snapshotsWithinRange(
+ optionalFilterSnapshotIdMax, optionalFilterSnapshotIdMin);
+
Iterator<InternalRow> rows = Iterators.transform(snapshots, this::toRow);
if (projection != null) {
rows =
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java b/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
index dbbc8fffd..e120bab3c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
@@ -44,6 +44,7 @@ import java.util.function.BinaryOperator;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;
+import java.util.stream.LongStream;
import static org.apache.paimon.utils.BranchManager.DEFAULT_MAIN_BRANCH;
import static org.apache.paimon.utils.BranchManager.getBranchPath;
@@ -375,6 +376,32 @@ public class SnapshotManager implements Serializable {
.iterator();
}
+ public Iterator<Snapshot> snapshotsWithinRange(
+ Optional<Long> optionalMaxSnapshotId, Optional<Long> optionalMinSnapshotId)
+ throws IOException {
+ Long lowerBoundSnapshotId = earliestSnapshotId();
+ Long upperBoundSnapshotId = latestSnapshotId();
+
+ // null check on lowerBoundSnapshotId & upperBoundSnapshotId
+ if (lowerBoundSnapshotId == null || upperBoundSnapshotId == null) {
+ return Collections.emptyIterator();
+ }
+
+ if (optionalMaxSnapshotId.isPresent()) {
+ upperBoundSnapshotId = optionalMaxSnapshotId.get();
+ }
+
+ if (optionalMinSnapshotId.isPresent()) {
+ lowerBoundSnapshotId = optionalMinSnapshotId.get();
+ }
+
+ // +1 here to include the upperBoundSnapshotId
+ return LongStream.range(lowerBoundSnapshotId, upperBoundSnapshotId + 1)
+ .mapToObj(this::snapshot)
+ .sorted(Comparator.comparingLong(Snapshot::id))
+ .iterator();
+ }
+
public Iterator<Changelog> changelogs() throws IOException {
return listVersionedFiles(fileIO, changelogDirectory(), CHANGELOG_PREFIX)
.map(this::changelog)
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java
index a94d91af6..2017ae4ba 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java
@@ -75,6 +75,26 @@ public class CatalogTableITCase extends CatalogITCaseBase {
sql(
"SELECT snapshot_id, schema_id, commit_kind FROM T$snapshots WHERE snapshot_id = 2");
assertThat(result).containsExactly(Row.of(2L, 0L, "APPEND"));
+
+ result =
+ sql(
+ "SELECT snapshot_id, schema_id, commit_kind FROM T$snapshots WHERE snapshot_id > 1");
+ assertThat(result).containsExactly(Row.of(2L, 0L, "APPEND"));
+
+ result =
+ sql(
+ "SELECT snapshot_id, schema_id, commit_kind FROM T$snapshots WHERE snapshot_id < 2");
+ assertThat(result).containsExactly(Row.of(1L, 0L, "APPEND"));
+
+ result =
+ sql(
+ "SELECT snapshot_id, schema_id, commit_kind FROM T$snapshots WHERE snapshot_id >= 1");
+ assertThat(result).contains(Row.of(1L, 0L, "APPEND"), Row.of(2L, 0L, "APPEND"));
+
+ result =
+ sql(
+ "SELECT snapshot_id, schema_id, commit_kind FROM T$snapshots WHERE snapshot_id <= 2");
+ assertThat(result).contains(Row.of(1L, 0L, "APPEND"), Row.of(2L, 0L, "APPEND"));
}
@Test
(paimon) 01/10: [core] Fisrt row merge engine supports none changelog producer (#3452)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 46d9f3d5ce74c14af5144a4e00ec80275f90510a
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Fri May 31 17:31:35 2024 +0800
[core] Fisrt row merge engine supports none changelog producer (#3452)
---
docs/content/primary-key-table/merge-engine.md | 5 ++--
.../main/java/org/apache/paimon/CoreOptions.java | 3 ++-
.../org/apache/paimon/lookup/LookupStrategy.java | 31 +++++++---------------
.../paimon/operation/KeyValueFileStoreWrite.java | 7 +++--
.../org/apache/paimon/schema/SchemaValidation.java | 5 ++--
.../table/source/InnerStreamTableScanImpl.java | 3 ++-
.../LookupChangelogMergeFunctionWrapperTest.java | 6 ++---
.../org/apache/paimon/flink/FirstRowITCase.java | 24 ++++++++---------
8 files changed, 36 insertions(+), 48 deletions(-)
diff --git a/docs/content/primary-key-table/merge-engine.md b/docs/content/primary-key-table/merge-engine.md
index bc6c1ee35..24caa5b43 100644
--- a/docs/content/primary-key-table/merge-engine.md
+++ b/docs/content/primary-key-table/merge-engine.md
@@ -326,9 +326,8 @@ By specifying `'merge-engine' = 'first-row'`, users can keep the first row of th
`deduplicate` merge engine that in the `first-row` merge engine, it will generate insert only changelog.
{{< hint info >}}
-1. `first-row` merge engine must be used together with `lookup` [changelog producer]({{< ref "primary-key-table/changelog-producer" >}}).
-2. You can not specify `sequence.field`.
-3. Not accept `DELETE` and `UPDATE_BEFORE` message. You can config `ignore-delete` to ignore these two kinds records.
+1. You can not specify `sequence.field`.
+2. Not accept `DELETE` and `UPDATE_BEFORE` message. You can config `ignore-delete` to ignore these two kinds records.
{{< /hint >}}
This is of great help in replacing log deduplication in streaming computation.
diff --git a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
index 63832638b..d88b9515a 100644
--- a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
@@ -1489,7 +1489,8 @@ public class CoreOptions implements Serializable {
public LookupStrategy lookupStrategy() {
return LookupStrategy.from(
- options.get(CHANGELOG_PRODUCER).equals(ChangelogProducer.LOOKUP),
+ mergeEngine().equals(MergeEngine.FIRST_ROW),
+ changelogProducer().equals(ChangelogProducer.LOOKUP),
deletionVectorsEnabled());
}
diff --git a/paimon-common/src/main/java/org/apache/paimon/lookup/LookupStrategy.java b/paimon-common/src/main/java/org/apache/paimon/lookup/LookupStrategy.java
index 6c709bcae..24e03ebdb 100644
--- a/paimon-common/src/main/java/org/apache/paimon/lookup/LookupStrategy.java
+++ b/paimon-common/src/main/java/org/apache/paimon/lookup/LookupStrategy.java
@@ -19,38 +19,25 @@
package org.apache.paimon.lookup;
/** Strategy for lookup. */
-public enum LookupStrategy {
- NO_LOOKUP(false, false),
-
- CHANGELOG_ONLY(true, false),
-
- DELETION_VECTOR_ONLY(false, true),
-
- CHANGELOG_AND_DELETION_VECTOR(true, true);
+public class LookupStrategy {
public final boolean needLookup;
+ public final boolean isFirstRow;
+
public final boolean produceChangelog;
public final boolean deletionVector;
- LookupStrategy(boolean produceChangelog, boolean deletionVector) {
+ private LookupStrategy(boolean isFirstRow, boolean produceChangelog, boolean deletionVector) {
+ this.isFirstRow = isFirstRow;
this.produceChangelog = produceChangelog;
this.deletionVector = deletionVector;
- this.needLookup = produceChangelog || deletionVector;
+ this.needLookup = produceChangelog || deletionVector || isFirstRow;
}
- public static LookupStrategy from(boolean produceChangelog, boolean deletionVector) {
- for (LookupStrategy strategy : values()) {
- if (strategy.produceChangelog == produceChangelog
- && strategy.deletionVector == deletionVector) {
- return strategy;
- }
- }
- throw new IllegalArgumentException(
- "Invalid combination of produceChangelog : "
- + produceChangelog
- + " and deletionVector : "
- + deletionVector);
+ public static LookupStrategy from(
+ boolean isFirstRow, boolean produceChangelog, boolean deletionVector) {
+ return new LookupStrategy(isFirstRow, produceChangelog, deletionVector);
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
index b7702ef8a..1bdc3042d 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
@@ -83,7 +83,6 @@ import java.util.function.Supplier;
import static org.apache.paimon.CoreOptions.ChangelogProducer.FULL_COMPACTION;
import static org.apache.paimon.CoreOptions.MergeEngine.DEDUPLICATE;
-import static org.apache.paimon.CoreOptions.MergeEngine.FIRST_ROW;
import static org.apache.paimon.lookup.LookupStoreFactory.bfGenerator;
/** {@link FileStoreWrite} for {@link KeyValueFileStore}. */
@@ -273,6 +272,7 @@ public class KeyValueFileStoreWrite extends MemoryFileStoreWrite<KeyValue> {
int maxLevel = options.numLevels() - 1;
MergeEngine mergeEngine = options.mergeEngine();
ChangelogProducer changelogProducer = options.changelogProducer();
+ LookupStrategy lookupStrategy = options.lookupStrategy();
if (changelogProducer.equals(FULL_COMPACTION)) {
return new FullChangelogMergeTreeCompactRewriter(
maxLevel,
@@ -285,12 +285,11 @@ public class KeyValueFileStoreWrite extends MemoryFileStoreWrite<KeyValue> {
mergeSorter,
valueEqualiserSupplier.get(),
options.changelogRowDeduplicate());
- } else if (options.needLookup()) {
- LookupStrategy lookupStrategy = options.lookupStrategy();
+ } else if (lookupStrategy.needLookup) {
LookupLevels.ValueProcessor<?> processor;
LookupMergeTreeCompactRewriter.MergeFunctionWrapperFactory<?> wrapperFactory;
FileReaderFactory<KeyValue> lookupReaderFactory = readerFactory;
- if (mergeEngine == FIRST_ROW) {
+ if (lookupStrategy.isFirstRow) {
if (options.deletionVectorsEnabled()) {
throw new UnsupportedOperationException(
"First row merge engine does not need deletion vectors because there is no deletion of old data in this merge engine.");
diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index fab5f31fd..b87dac065 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -171,9 +171,10 @@ public class SchemaValidation {
}
if (options.mergeEngine() == MergeEngine.FIRST_ROW) {
- if (options.changelogProducer() != ChangelogProducer.LOOKUP) {
+ if (options.changelogProducer() != ChangelogProducer.LOOKUP
+ && options.changelogProducer() != ChangelogProducer.NONE) {
throw new IllegalArgumentException(
- "Only support 'lookup' changelog-producer on FIRST_MERGE merge engine");
+ "Only support 'none' and 'lookup' changelog-producer on FIRST_MERGE merge engine");
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/InnerStreamTableScanImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/InnerStreamTableScanImpl.java
index 3e03dc0a1..b2c74a83d 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/InnerStreamTableScanImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/InnerStreamTableScanImpl.java
@@ -145,7 +145,8 @@ public class InnerStreamTableScanImpl extends AbstractInnerTableScan
ScannedResult scannedResult = (ScannedResult) result;
currentWatermark = scannedResult.currentWatermark();
long currentSnapshotId = scannedResult.currentSnapshotId();
- if (options.lookupStrategy().equals(LookupStrategy.DELETION_VECTOR_ONLY)) {
+ LookupStrategy lookupStrategy = options.lookupStrategy();
+ if (!lookupStrategy.produceChangelog && lookupStrategy.deletionVector) {
// For DELETION_VECTOR_ONLY mode, we need to return the remaining data from level 0
// in the subsequent plan.
nextSnapshotId = currentSnapshotId;
diff --git a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapperTest.java b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapperTest.java
index d43be8aff..dbee0805a 100644
--- a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapperTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapperTest.java
@@ -73,7 +73,7 @@ public class LookupChangelogMergeFunctionWrapperTest {
highLevel::get,
EQUALISER,
changelogRowDeduplicate,
- LookupStrategy.CHANGELOG_ONLY,
+ LookupStrategy.from(false, true, false),
null,
null);
@@ -233,7 +233,7 @@ public class LookupChangelogMergeFunctionWrapperTest {
key -> null,
EQUALISER,
changelogRowDeduplicate,
- LookupStrategy.CHANGELOG_ONLY,
+ LookupStrategy.from(false, true, false),
null,
null);
@@ -322,7 +322,7 @@ public class LookupChangelogMergeFunctionWrapperTest {
highLevel::get,
EQUALISER,
false,
- LookupStrategy.CHANGELOG_ONLY,
+ LookupStrategy.from(false, true, false),
null,
UserDefinedSeqComparator.create(
RowType.builder().field("f0", DataTypes.INT()).build(),
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FirstRowITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FirstRowITCase.java
index 5d7927a97..e44aa4c63 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FirstRowITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FirstRowITCase.java
@@ -31,7 +31,6 @@ import java.util.Collections;
import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** ITCase for first row merge engine. */
public class FirstRowITCase extends CatalogITCaseBase {
@@ -45,23 +44,24 @@ public class FirstRowITCase extends CatalogITCaseBase {
}
@Test
- public void testIllegal() {
- assertThatThrownBy(
- () ->
- sql(
- "CREATE TABLE ILLEGAL_T (a INT, b INT, c STRING, PRIMARY KEY (a) NOT ENFORCED)"
- + " WITH ('merge-engine'='first-row')"))
- .hasRootCauseMessage(
- "Only support 'lookup' changelog-producer on FIRST_MERGE merge engine");
+ public void testBatchQueryNoChangelog() {
+ sql(
+ "CREATE TABLE T_NO_CHANGELOG (a INT, b INT, c STRING, PRIMARY KEY (a) NOT ENFORCED)"
+ + " WITH ('merge-engine'='first-row')");
+ testBatchQuery("T_NO_CHANGELOG");
}
@Test
public void testBatchQuery() {
- batchSql("INSERT INTO T VALUES (1, 1, '1'), (1, 2, '2')");
- List<Row> result = batchSql("SELECT * FROM T");
+ testBatchQuery("T");
+ }
+
+ private void testBatchQuery(String table) {
+ batchSql("INSERT INTO %s VALUES (1, 1, '1'), (1, 2, '2')", table);
+ List<Row> result = batchSql("SELECT * FROM %s", table);
assertThat(result).containsExactlyInAnyOrder(Row.ofKind(RowKind.INSERT, 1, 1, "1"));
- result = batchSql("SELECT c FROM T");
+ result = batchSql("SELECT c FROM %s", table);
assertThat(result).containsExactlyInAnyOrder(Row.ofKind(RowKind.INSERT, "1"));
}
(paimon) 07/10: [spark] Spark 3.2 insert overwrite test support (#3458)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 432d7e23e3cd1ad7224552a7d9c427d8882f661f
Author: xuzifu666 <12...@qq.com>
AuthorDate: Mon Jun 3 21:39:58 2024 +0800
[spark] Spark 3.2 insert overwrite test support (#3458)
---
docs/content/engines/overview.md | 6 +++---
.../paimon/spark/sql/InsertOverwriteTableTest.scala | 21 +++++++++++++++++++++
.../paimon/spark/sql/InsertOverwriteTableTest.scala | 21 +++++++++++++++++++++
.../paimon/spark/sql/InsertOverwriteTableTest.scala | 21 +++++++++++++++++++++
.../paimon/spark/sql/InsertOverwriteTableTest.scala | 21 +++++++++++++++++++++
...iteTest.scala => InsertOverwriteTableTest.scala} | 2 +-
6 files changed, 88 insertions(+), 4 deletions(-)
diff --git a/docs/content/engines/overview.md b/docs/content/engines/overview.md
index 19f3bf1c1..4a7e4b907 100644
--- a/docs/content/engines/overview.md
+++ b/docs/content/engines/overview.md
@@ -28,10 +28,10 @@ under the License.
## Compatibility Matrix
-| Engine | Version | Batch Read | Batch Write | Create Table | Alter Table | Streaming Write | Streaming Read | Batch Overwrite | DELETE & UPDATE | MERGE INTO |
-|:-------------------------------------------------------------------------------:|:-------------:|:-----------:|:-------------:|:-------------:|:-------------:|:----------------:|:----------------:|:----------------:|:------------------:|:-----------:|
+| Engine | Version | Batch Read | Batch Write | Create Table | Alter Table | Streaming Write | Streaming Read | Batch Overwrite | DELETE & UPDATE | MERGE INTO |
+|:-------------------------------------------------------------------------------:|:-------------:|:-----------:|:-------------:|:-------------:|:-------------:|:----------------:|:----------------:|:---------------:|:------------------:|:-----------:|
| Flink | 1.15 - 1.19 | ✅ | ✅ | ✅ | ✅(1.17+) | ✅ | ✅ | ✅ | ✅(1.17+) | ❌ |
-| Spark | 3.1 - 3.5 | ✅ | ✅(3.3+) | ✅ | ✅ | ✅(3.3+) | ✅(3.3+) | ✅(3.3+) | ✅(3.2+) | ✅(3.2+) |
+| Spark | 3.1 - 3.5 | ✅ | ✅(3.3+) | ✅ | ✅ | ✅(3.3+) | ✅(3.3+) | ✅(3.2+) | ✅(3.2+) | ✅(3.2+) |
| Hive | 2.1 - 3.1 | ✅ | ✅ | ✅ | ❌ | ❌ | ❌ | ❌ | ❌ | ❌ |
| Trino | 420 - 439 | ✅ | ✅(427+) | ✅(427+) | ✅(427+) | ❌ | ❌ | ❌ | ❌ | ❌ |
| Presto | 0.236 - 0.280 | ✅ | ❌ | ✅ | ✅ | ❌ | ❌ | ❌ | ❌ | ❌ |
diff --git a/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTest.scala b/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTest.scala
new file mode 100644
index 000000000..4f66584c3
--- /dev/null
+++ b/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTest.scala
@@ -0,0 +1,21 @@
+/*
+ * 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.spark.sql
+
+class InsertOverwriteTableTest extends InsertOverwriteTableTestBase {}
diff --git a/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTest.scala b/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTest.scala
new file mode 100644
index 000000000..4f66584c3
--- /dev/null
+++ b/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTest.scala
@@ -0,0 +1,21 @@
+/*
+ * 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.spark.sql
+
+class InsertOverwriteTableTest extends InsertOverwriteTableTestBase {}
diff --git a/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTest.scala b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTest.scala
new file mode 100644
index 000000000..4f66584c3
--- /dev/null
+++ b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTest.scala
@@ -0,0 +1,21 @@
+/*
+ * 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.spark.sql
+
+class InsertOverwriteTableTest extends InsertOverwriteTableTestBase {}
diff --git a/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTest.scala b/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTest.scala
new file mode 100644
index 000000000..4f66584c3
--- /dev/null
+++ b/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTest.scala
@@ -0,0 +1,21 @@
+/*
+ * 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.spark.sql
+
+class InsertOverwriteTableTest extends InsertOverwriteTableTestBase {}
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTest.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTest.scala
similarity index 99%
rename from paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTest.scala
rename to paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTest.scala
index f33c49f82..978128a3a 100644
--- a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTest.scala
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTest.scala
@@ -25,7 +25,7 @@ import org.apache.spark.sql.types._
import java.sql.Date
-class InsertOverwriteTest extends PaimonSparkTestBase {
+abstract class InsertOverwriteTableTestBase extends PaimonSparkTestBase {
withPk.foreach {
hasPk =>