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/05/11 08:33:28 UTC
[incubator-paimon] branch master updated: [core] paimon table scan listPartition support (#1122)
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 9f480b5c4 [core] paimon table scan listPartition support (#1122)
9f480b5c4 is described below
commit 9f480b5c491d16af057a4da08e91486bc7347aca
Author: YeJunHao <41...@users.noreply.github.com>
AuthorDate: Thu May 11 16:33:22 2023 +0800
[core] paimon table scan listPartition support (#1122)
---
.../apache/paimon/operation/PartitionExpire.java | 2 +
.../table/source/AbstractInnerTableScan.java | 6 +++
.../paimon/table/source/ReadOnceTableScan.java | 9 ++++
.../org/apache/paimon/table/source/TableScan.java | 4 ++
.../table/source/snapshot/SnapshotSplitReader.java | 4 ++
.../source/snapshot/SnapshotSplitReaderImpl.java | 19 +++++++
.../apache/paimon/table/system/AuditLogTable.java | 16 ++++++
.../org/apache/paimon/catalog/CatalogTestBase.java | 13 +++++
.../paimon/catalog/FileSystemCatalogTest.java | 47 ----------------
.../table/source/TableScanListPartitionsTest.java | 62 ++++++++++++++++++++++
.../source/ContinuousFileSplitEnumeratorTest.java | 6 +++
11 files changed, 141 insertions(+), 47 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/PartitionExpire.java b/paimon-core/src/main/java/org/apache/paimon/operation/PartitionExpire.java
index 05ab8ec22..e20c47eea 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/PartitionExpire.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/PartitionExpire.java
@@ -113,6 +113,8 @@ public class PartitionExpire {
private List<BinaryRow> readPartitions() {
// TODO optimize this to read partition only
+ // NOTE: avro projection push down can't reach expected cause row-oriented design, see pull
+ // request #1114
return scan.plan().files().stream()
.map(ManifestEntry::partition)
.distinct()
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractInnerTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractInnerTableScan.java
index 250131467..863ba9f3f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractInnerTableScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractInnerTableScan.java
@@ -23,6 +23,7 @@ import org.apache.paimon.CoreOptions.ChangelogProducer;
import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.consumer.Consumer;
import org.apache.paimon.consumer.ConsumerManager;
+import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.operation.FileStoreScan;
import org.apache.paimon.table.source.snapshot.CompactedStartingScanner;
import org.apache.paimon.table.source.snapshot.ContinuousCompactorStartingScanner;
@@ -37,6 +38,7 @@ import org.apache.paimon.table.source.snapshot.StaticFromSnapshotStartingScanner
import org.apache.paimon.table.source.snapshot.StaticFromTimestampStartingScanner;
import org.apache.paimon.utils.Preconditions;
+import java.util.List;
import java.util.Optional;
import static org.apache.paimon.CoreOptions.FULL_COMPACTION_DELTA_COMMITS;
@@ -128,4 +130,8 @@ public abstract class AbstractInnerTableScan implements InnerTableScan {
"Unknown startup mode " + startupMode.name());
}
}
+
+ public List<BinaryRow> listPartitions() {
+ return snapshotSplitReader.partitions();
+ }
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/ReadOnceTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/ReadOnceTableScan.java
index 7efe2c4ff..b93f8468b 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/ReadOnceTableScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/ReadOnceTableScan.java
@@ -18,6 +18,10 @@
package org.apache.paimon.table.source;
+import org.apache.paimon.data.BinaryRow;
+
+import java.util.List;
+
/** An {@link InnerTableScan} for reading only once, this is for batch scan. */
public abstract class ReadOnceTableScan implements InnerTableScan {
@@ -34,4 +38,9 @@ public abstract class ReadOnceTableScan implements InnerTableScan {
}
protected abstract Plan innerPlan();
+
+ @Override
+ public List<BinaryRow> listPartitions() {
+ throw new UnsupportedOperationException();
+ }
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/TableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/TableScan.java
index 193d6e8be..ba8b0b5e1 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/TableScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/TableScan.java
@@ -19,6 +19,7 @@
package org.apache.paimon.table.source;
import org.apache.paimon.annotation.Public;
+import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.table.Table;
import java.util.List;
@@ -34,6 +35,9 @@ public interface TableScan {
/** Plan splits, throws {@link EndOfScanException} if the scan is ended. */
Plan plan();
+ /** Get partitions from simple manifest entries. */
+ List<BinaryRow> listPartitions();
+
/**
* Plan of scan.
*
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotSplitReader.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotSplitReader.java
index 87def5fa1..703386886 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotSplitReader.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotSplitReader.java
@@ -20,6 +20,7 @@ package org.apache.paimon.table.source.snapshot;
import org.apache.paimon.Snapshot;
import org.apache.paimon.consumer.ConsumerManager;
+import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.operation.ScanKind;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.table.source.DataSplit;
@@ -47,4 +48,7 @@ public interface SnapshotSplitReader {
/** Get splits from an overwrite snapshot. */
List<DataSplit> overwriteSplits();
+
+ /** Get partitions from a snapshot. */
+ List<BinaryRow> partitions();
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotSplitReaderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotSplitReaderImpl.java
index 6ab594501..4ceb8ecff 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotSplitReaderImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotSplitReaderImpl.java
@@ -27,6 +27,7 @@ import org.apache.paimon.consumer.ConsumerManager;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.operation.FileStoreScan;
import org.apache.paimon.operation.ScanKind;
import org.apache.paimon.predicate.Predicate;
@@ -43,6 +44,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.function.BiConsumer;
+import java.util.stream.Collectors;
import static org.apache.paimon.predicate.PredicateBuilder.transformFieldMapping;
@@ -159,6 +161,23 @@ public class SnapshotSplitReaderImpl implements SnapshotSplitReader {
files);
}
+ @Override
+ public List<BinaryRow> partitions() {
+ List<ManifestEntry> entryList = scan.plan().files();
+
+ return entryList.stream()
+ .collect(
+ Collectors.groupingBy(
+ ManifestEntry::partition,
+ LinkedHashMap::new,
+ Collectors.reducing((a, b) -> b)))
+ .values()
+ .stream()
+ .map(Optional::get)
+ .map(ManifestEntry::partition)
+ .collect(Collectors.toList());
+ }
+
/**
* Get splits from an overwrite snapshot files. The {@link FileKind#DELETE} part will be marked
* with reverseRowKind = true (see {@link DataSplit}).
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
index 27d4efbb4..a6247095d 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
@@ -20,6 +20,7 @@ package org.apache.paimon.table.system;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.consumer.ConsumerManager;
+import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.fs.FileIO;
@@ -222,6 +223,11 @@ public class AuditLogTable implements DataTable, ReadonlyTable {
public List<DataSplit> overwriteSplits() {
return snapshotSplitReader.overwriteSplits();
}
+
+ @Override
+ public List<BinaryRow> partitions() {
+ return snapshotSplitReader.partitions();
+ }
}
private class AuditLogBatchScan implements InnerTableScan {
@@ -242,6 +248,11 @@ public class AuditLogTable implements DataTable, ReadonlyTable {
public Plan plan() {
return batchScan.plan();
}
+
+ @Override
+ public List<BinaryRow> listPartitions() {
+ return batchScan.listPartitions();
+ }
}
private class AuditLogStreamScan implements InnerStreamTableScan {
@@ -263,6 +274,11 @@ public class AuditLogTable implements DataTable, ReadonlyTable {
return streamScan.plan();
}
+ @Override
+ public List<BinaryRow> listPartitions() {
+ return streamScan.listPartitions();
+ }
+
@Nullable
@Override
public Long checkpoint() {
diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogTestBase.java b/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogTestBase.java
index b5892f729..ba3fcc0d6 100644
--- a/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogTestBase.java
+++ b/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogTestBase.java
@@ -37,6 +37,7 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
@@ -63,6 +64,18 @@ public abstract class CatalogTestBase {
Maps.newHashMap(),
"");
+ protected static final Schema PARTITION_SCHEMA =
+ new Schema(
+ Lists.newArrayList(
+ new DataField(0, "pk1", DataTypes.INT()),
+ new DataField(1, "pk2", DataTypes.STRING()),
+ new DataField(3, "pk3", DataTypes.STRING()),
+ new DataField(4, "col", DataTypes.STRING())),
+ Arrays.asList("pk1", "pk2"),
+ Arrays.asList("pk1", "pk2", "pk3"),
+ Maps.newHashMap(),
+ "");
+
@BeforeEach
public void setUp() throws Exception {
warehouse = tempFile.toUri().toString();
diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java
deleted file mode 100644
index 426d088d4..000000000
--- a/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java
+++ /dev/null
@@ -1,47 +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.catalog;
-
-import org.apache.paimon.fs.Path;
-
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-
-import java.util.List;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** Tests for {@link FileSystemCatalog}. */
-public class FileSystemCatalogTest extends CatalogTestBase {
-
- @Override
- @BeforeEach
- public void setUp() throws Exception {
- super.setUp();
- catalog = new FileSystemCatalog(fileIO, new Path(warehouse));
- }
-
- @Test
- @Override
- public void testListDatabasesWhenNoDatabases() {
- // List databases returns an empty list when there are no databases
- List<String> databases = catalog.listDatabases();
- assertThat(databases).isEmpty();
- }
-}
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanListPartitionsTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanListPartitionsTest.java
new file mode 100644
index 000000000..0356fd3e1
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanListPartitionsTest.java
@@ -0,0 +1,62 @@
+/*
+ * 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;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.source.snapshot.ScannerTestBase;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link TableScan} listPartitions. */
+public class TableScanListPartitionsTest extends ScannerTestBase {
+
+ @Test
+ public void testListPartitions() throws Exception {
+ BatchTableWrite write = table.newWrite(commitUser);
+
+ for (int i = 0; i < 1000; i++) {
+ InternalRow row = GenericRow.of(i, i, Long.valueOf(i));
+ write.write(row);
+ }
+ List<CommitMessage> result = write.prepareCommit();
+ table.newCommit(commitUser).commit(result);
+
+ AtomicInteger ai = new AtomicInteger(0);
+
+ BinaryRow[] rows =
+ table.newReadBuilder().newScan().listPartitions().toArray(new BinaryRow[0]);
+ assertThat(rows.length).isEqualTo(1000);
+ Arrays.sort(rows, Comparator.comparing(o -> o.getInt(0)));
+
+ for (BinaryRow row : rows) {
+ assertThat(row.getInt(0)).isEqualTo(ai.getAndIncrement());
+ }
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/ContinuousFileSplitEnumeratorTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/ContinuousFileSplitEnumeratorTest.java
index a0ab1fc41..ed9ce6b90 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/ContinuousFileSplitEnumeratorTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/ContinuousFileSplitEnumeratorTest.java
@@ -18,6 +18,7 @@
package org.apache.paimon.flink.source;
+import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.table.source.DataFilePlan;
import org.apache.paimon.table.source.DataSplit;
@@ -314,6 +315,11 @@ public class ContinuousFileSplitEnumeratorTest {
return plan;
}
+ @Override
+ public List<BinaryRow> listPartitions() {
+ throw new UnsupportedOperationException();
+ }
+
@Override
public Long checkpoint() {
return null;