This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 55f2102ed [core] Use TimeTravelUtil to stats read and manifests table
55f2102ed is described below
commit 55f2102edffce4b1bafc9dd1698544d3a66d623e
Author: Jingsong <[email protected]>
AuthorDate: Tue Oct 15 13:46:42 2024 +0800
[core] Use TimeTravelUtil to stats read and manifests table
---
.../paimon/table/AbstractFileStoreTable.java | 23 ++---------
.../table/source/snapshot/TimeTravelUtil.java | 35 +++++++++++-----
.../apache/paimon/table/system/ManifestsTable.java | 47 +++-------------------
.../apache/paimon/table/system/StatisticTable.java | 25 ++++--------
.../paimon/table/system/ManifestsTableTest.java | 2 +-
.../paimon/spark/sql/AnalyzeTableTestBase.scala | 5 ---
6 files changed, 43 insertions(+), 94 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
index 24e97e1c9..b33cf6922 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
@@ -58,6 +58,7 @@ import org.apache.paimon.table.source.snapshot.SnapshotReader;
import org.apache.paimon.table.source.snapshot.SnapshotReaderImpl;
import
org.apache.paimon.table.source.snapshot.StaticFromTimestampStartingScanner;
import
org.apache.paimon.table.source.snapshot.StaticFromWatermarkStartingScanner;
+import org.apache.paimon.table.source.snapshot.TimeTravelUtil;
import org.apache.paimon.tag.TagPreview;
import org.apache.paimon.utils.BranchManager;
import org.apache.paimon.utils.Preconditions;
@@ -65,7 +66,6 @@ import org.apache.paimon.utils.SegmentsCache;
import org.apache.paimon.utils.SimpleFileReader;
import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.utils.SnapshotNotExistException;
-import org.apache.paimon.utils.StringUtils;
import org.apache.paimon.utils.TagManager;
import javax.annotation.Nullable;
@@ -169,24 +169,9 @@ abstract class AbstractFileStoreTable implements
FileStoreTable {
@Override
public Optional<Statistics> statistics() {
- Snapshot latestSnapshot = null;
- Long snapshotId = coreOptions().scanSnapshotId();
- String tagName = coreOptions().scanTagName();
- Long timestampMills = coreOptions().scanTimestampMills();
-
- if (snapshotId != null &&
snapshotManager().snapshotExists(snapshotId)) {
- latestSnapshot = snapshotManager().snapshot(snapshotId);
- } else if (!StringUtils.isEmpty(tagName) &&
tagManager().tagExists(tagName)) {
- return store().newStatsFileHandler()
- .readStats(tagManager().tag(tagName).trimToSnapshot());
- } else if (timestampMills != null) {
- latestSnapshot =
snapshotManager().earlierOrEqualTimeMills(timestampMills);
- } else if (snapshotId == null && StringUtils.isEmpty(tagName)) {
- latestSnapshot = snapshotManager().latestSnapshot();
- }
-
- if (latestSnapshot != null) {
- return store().newStatsFileHandler().readStats(latestSnapshot);
+ Snapshot snapshot = TimeTravelUtil.resolveSnapshot(this);
+ if (snapshot != null) {
+ return store().newStatsFileHandler().readStats(snapshot);
}
return Optional.empty();
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java
index 2ef0d5153..4c8b41aa4 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java
@@ -20,40 +20,42 @@ package org.apache.paimon.table.source.snapshot;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.utils.Preconditions;
import org.apache.paimon.utils.SnapshotManager;
+import org.apache.paimon.utils.SnapshotNotExistException;
import org.apache.paimon.utils.TagManager;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import java.util.ArrayList;
import java.util.List;
+import static org.apache.paimon.CoreOptions.SCAN_SNAPSHOT_ID;
+
/** The util class of resolve snapshot from scan params for time travel. */
public class TimeTravelUtil {
- private static String[] scanKeys = {
+ private static final String[] SCAN_KEYS = {
CoreOptions.SCAN_SNAPSHOT_ID.key(),
CoreOptions.SCAN_TAG_NAME.key(),
CoreOptions.SCAN_WATERMARK.key(),
CoreOptions.SCAN_TIMESTAMP_MILLIS.key()
};
- private static final Logger LOG =
LoggerFactory.getLogger(TimeTravelUtil.class);
+ public static Snapshot resolveSnapshot(FileStoreTable table) {
+ return resolveSnapshotFromOptions(table.coreOptions(),
table.snapshotManager());
+ }
public static Snapshot resolveSnapshotFromOptions(
CoreOptions options, SnapshotManager snapshotManager) {
List<String> scanHandleKey = new ArrayList<>(1);
- for (String key : scanKeys) {
+ for (String key : SCAN_KEYS) {
if (options.toConfiguration().containsKey(key)) {
scanHandleKey.add(key);
}
}
if (scanHandleKey.size() == 0) {
- LOG.warn("Not set any time travel parameter.");
- return null;
+ return snapshotManager.latestSnapshot();
}
Preconditions.checkArgument(
@@ -76,13 +78,28 @@ public class TimeTravelUtil {
} else if (key.equals(CoreOptions.SCAN_TAG_NAME.key())) {
snapshot = resolveSnapshotByTagName(snapshotManager, options);
}
+
+ if (snapshot == null) {
+ snapshot = snapshotManager.latestSnapshot();
+ }
return snapshot;
}
private static Snapshot resolveSnapshotBySnapshotId(
SnapshotManager snapshotManager, CoreOptions options) {
Long snapshotId = options.scanSnapshotId();
- if (snapshotId != null && snapshotManager.snapshotExists(snapshotId)) {
+ if (snapshotId != null) {
+ if (!snapshotManager.snapshotExists(snapshotId)) {
+ Long earliestSnapshotId = snapshotManager.earliestSnapshotId();
+ Long latestSnapshotId = snapshotManager.latestSnapshotId();
+ throw new SnapshotNotExistException(
+ String.format(
+ "Specified parameter %s = %s is not exist, you
can set it in range from %s to %s.",
+ SCAN_SNAPSHOT_ID.key(),
+ snapshotId,
+ earliestSnapshotId,
+ latestSnapshotId));
+ }
return snapshotManager.snapshot(snapshotId);
}
return null;
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 02e0e8618..d88636d02 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
@@ -37,6 +37,7 @@ 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.snapshot.TimeTravelUtil;
import org.apache.paimon.types.BigIntType;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;
@@ -44,9 +45,6 @@ import org.apache.paimon.utils.FileStorePathFactory;
import org.apache.paimon.utils.IteratorRecordReader;
import org.apache.paimon.utils.ProjectedRow;
import org.apache.paimon.utils.SerializationUtils;
-import org.apache.paimon.utils.SnapshotManager;
-import org.apache.paimon.utils.SnapshotNotExistException;
-import org.apache.paimon.utils.StringUtils;
import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
@@ -59,8 +57,6 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
-import static org.apache.paimon.CoreOptions.SCAN_SNAPSHOT_ID;
-import static org.apache.paimon.CoreOptions.SCAN_TAG_NAME;
import static org.apache.paimon.catalog.Catalog.SYSTEM_TABLE_SPLITTER;
/** A {@link Table} for showing committing snapshots of table. */
@@ -204,41 +200,8 @@ public class ManifestsTable implements ReadonlyTable {
}
private static List<ManifestFileMeta> allManifests(FileStoreTable
dataTable) {
- CoreOptions coreOptions = CoreOptions.fromMap(dataTable.options());
- SnapshotManager snapshotManager = dataTable.snapshotManager();
- Long snapshotId = coreOptions.scanSnapshotId();
- String tagName = coreOptions.scanTagName();
- Long timestampMills = coreOptions.scanTimestampMills();
-
- Snapshot snapshot;
- if (snapshotId != null) {
- // reminder user with snapshot id range
- if (!snapshotManager.snapshotExists(snapshotId)) {
- Long earliestSnapshotId = snapshotManager.earliestSnapshotId();
- Long latestSnapshotId = snapshotManager.latestSnapshotId();
- throw new SnapshotNotExistException(
- String.format(
- "Specified parameter %s = %s is not exist, you
can set it in range from %s to %s.",
- SCAN_SNAPSHOT_ID.key(),
- snapshotId,
- earliestSnapshotId,
- latestSnapshotId));
- }
- snapshot = snapshotManager.snapshot(snapshotId);
- } else if (!StringUtils.isEmpty(tagName)) {
- if (!dataTable.tagManager().tagExists(tagName)) {
- throw new RuntimeException(
- String.format(
- "Specified parameter %s = %s is not exist.",
- SCAN_TAG_NAME.key(), tagName));
- }
- snapshot = dataTable.tagManager().tag(tagName).trimToSnapshot();
- } else if (timestampMills != null) {
- snapshot = snapshotManager.earlierOrEqualTimeMills(timestampMills);
- } else {
- snapshot = snapshotManager.latestSnapshot();
- }
-
+ CoreOptions options = dataTable.coreOptions();
+ Snapshot snapshot = TimeTravelUtil.resolveSnapshot(dataTable);
if (snapshot == null) {
LOG.warn("Check if your snapshot is empty.");
return Collections.emptyList();
@@ -247,8 +210,8 @@ public class ManifestsTable implements ReadonlyTable {
ManifestList manifestList =
new ManifestList.Factory(
dataTable.fileIO(),
- coreOptions.manifestFormat(),
- coreOptions.manifestCompression(),
+ options.manifestFormat(),
+ options.manifestCompression(),
fileStorePathFactory,
null)
.create();
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 faaa654eb..f01804449 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
@@ -22,7 +22,6 @@ import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManager;
-import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.reader.EmptyRecordReader;
@@ -47,7 +46,6 @@ import org.apache.paimon.utils.SerializationUtils;
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;
@@ -74,18 +72,9 @@ public class StatisticTable implements ReadonlyTable {
new DataField(3, "mergedRecordSize", new
BigIntType(true)),
new DataField(4, "colstat",
SerializationUtils.newStringType(true))));
- private final FileIO fileIO;
- private final Path location;
-
private final FileStoreTable dataTable;
public StatisticTable(FileStoreTable dataTable) {
- this(dataTable.fileIO(), dataTable.location(), dataTable);
- }
-
- public StatisticTable(FileIO fileIO, Path location, FileStoreTable
dataTable) {
- this.fileIO = fileIO;
- this.location = location;
this.dataTable = dataTable;
}
@@ -111,12 +100,12 @@ public class StatisticTable implements ReadonlyTable {
@Override
public InnerTableRead newRead() {
- return new StatisticRead(fileIO, dataTable);
+ return new StatisticRead(dataTable);
}
@Override
public Table copy(Map<String, String> dynamicOptions) {
- return new StatisticTable(fileIO, location,
dataTable.copy(dynamicOptions));
+ return new StatisticTable(dataTable.copy(dynamicOptions));
}
private class StatisticScan extends ReadOnceTableScan {
@@ -129,7 +118,9 @@ public class StatisticTable implements ReadonlyTable {
@Override
public Plan innerPlan() {
- return () -> Collections.singletonList(new
StatisticTable.StatisticSplit(location));
+ return () ->
+ Collections.singletonList(
+ new
StatisticTable.StatisticSplit(dataTable.location()));
}
}
@@ -163,13 +154,11 @@ public class StatisticTable implements ReadonlyTable {
private static class StatisticRead implements InnerTableRead {
- private final FileIO fileIO;
private RowType readType;
private final FileStoreTable dataTable;
- public StatisticRead(FileIO fileIO, FileStoreTable dataTable) {
- this.fileIO = fileIO;
+ public StatisticRead(FileStoreTable dataTable) {
this.dataTable = dataTable;
}
@@ -191,7 +180,7 @@ public class StatisticTable implements ReadonlyTable {
}
@Override
- public RecordReader<InternalRow> createReader(Split split) throws
IOException {
+ public RecordReader<InternalRow> createReader(Split split) {
if (!(split instanceof StatisticTable.StatisticSplit)) {
throw new IllegalArgumentException("Unsupported split: " +
split.getClass());
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/system/ManifestsTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/system/ManifestsTableTest.java
index eacd40847..a39e6f6fa 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/system/ManifestsTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/system/ManifestsTableTest.java
@@ -150,7 +150,7 @@ public class ManifestsTableTest extends TableTestBase {
}
@Test
- public void testReadManifestsFromNotExistSnapshot() throws Exception {
+ public void testReadManifestsFromNotExistSnapshot() {
manifestsTable =
(ManifestsTable)
manifestsTable.copy(
diff --git
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/AnalyzeTableTestBase.scala
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/AnalyzeTableTestBase.scala
index b397be24b..238dd0399 100644
---
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/AnalyzeTableTestBase.scala
+++
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/AnalyzeTableTestBase.scala
@@ -140,11 +140,6 @@ abstract class AnalyzeTableTestBase extends
PaimonSparkTestBase {
sql("SELECT snapshot_id, schema_id, mergedRecordCount, colstat FROM
`T$statistics`"),
Row(5, 0, 4, "{ }"))
}
-
- withSQLConf("spark.paimon.scan.snapshot-id" -> "100") {
- Assertions.assertEquals(0, sql("select * from `T$statistics`").count())
- }
-
}
test("Paimon analyze: analyze table without snapshot") {
