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 5e5b2fd35 [core] Manifest and Statistic support query by timestamp
(#4304)
5e5b2fd35 is described below
commit 5e5b2fd35ac3f434f6df1bce56f2432c9dbdd805
Author: herefree <[email protected]>
AuthorDate: Tue Oct 15 12:10:41 2024 +0800
[core] Manifest and Statistic support query by timestamp (#4304)
---
docs/content/maintenance/system-tables.md | 11 +++++++
.../paimon/table/AbstractFileStoreTable.java | 19 +++++-------
.../apache/paimon/table/system/ManifestsTable.java | 36 +++++++++++++++++-----
.../paimon/table/system/ManifestsTableTest.java | 16 +++++++++-
.../paimon/spark/sql/AnalyzeTableTestBase.scala | 15 +++++++--
5 files changed, 74 insertions(+), 23 deletions(-)
diff --git a/docs/content/maintenance/system-tables.md
b/docs/content/maintenance/system-tables.md
index 05a0c0db7..a94031e48 100644
--- a/docs/content/maintenance/system-tables.md
+++ b/docs/content/maintenance/system-tables.md
@@ -283,6 +283,17 @@ SELECT * FROM my_table$manifests /*+
OPTIONS('scan.tag-name'='tag1') */;
+--------------------------------+-------------+------------------+-------------------+---------------+
1 rows in set
*/
+
+- You can also query the manifest with specified timestamp in unix milliseconds
+SELECT * FROM my_table$manifests /*+
OPTIONS('scan.timestamp-millis'='1678883047356') */;
+/*
++--------------------------------+-------------+------------------+-------------------+---------------+
+| file_name | file_size | num_added_files |
num_deleted_files | schema_id |
++--------------------------------+-------------+------------------+-------------------+---------------+
+| manifest-f4dcab43-ef6b-4713... | 12365| 40 |
0 | 0 |
++--------------------------------+-------------+------------------+-------------------+---------------+
+1 rows in set
+*/
```
### Aggregation fields Table
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 6aacef6ed..24e97e1c9 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
@@ -169,22 +169,19 @@ abstract class AbstractFileStoreTable implements
FileStoreTable {
@Override
public Optional<Statistics> statistics() {
- Snapshot latestSnapshot;
+ Snapshot latestSnapshot = null;
Long snapshotId = coreOptions().scanSnapshotId();
String tagName = coreOptions().scanTagName();
-
- if (snapshotId == null) {
- if (!StringUtils.isEmpty(tagName) &&
tagManager().tagExists(tagName)) {
- return store().newStatsFileHandler()
- .readStats(tagManager().tag(tagName).trimToSnapshot());
- } else {
- snapshotId = snapshotManager().latestSnapshotId();
- }
- }
+ Long timestampMills = coreOptions().scanTimestampMills();
if (snapshotId != null &&
snapshotManager().snapshotExists(snapshotId)) {
latestSnapshot = snapshotManager().snapshot(snapshotId);
- } else {
+ } 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();
}
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 4e07c6c58..02e0e8618 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
@@ -50,16 +50,24 @@ import org.apache.paimon.utils.StringUtils;
import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.util.Arrays;
import java.util.Collections;
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. */
public class ManifestsTable implements ReadonlyTable {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(ManifestsTable.class);
+
private static final long serialVersionUID = 1L;
public static final String MANIFESTS = "manifests";
@@ -200,7 +208,9 @@ public class ManifestsTable implements ReadonlyTable {
SnapshotManager snapshotManager = dataTable.snapshotManager();
Long snapshotId = coreOptions.scanSnapshotId();
String tagName = coreOptions.scanTagName();
- Snapshot snapshot = null;
+ Long timestampMills = coreOptions.scanTimestampMills();
+
+ Snapshot snapshot;
if (snapshotId != null) {
// reminder user with snapshot id range
if (!snapshotManager.snapshotExists(snapshotId)) {
@@ -208,19 +218,29 @@ public class ManifestsTable implements ReadonlyTable {
Long latestSnapshotId = snapshotManager.latestSnapshotId();
throw new SnapshotNotExistException(
String.format(
- "Specified scan.snapshot-id %s is not exist,
you can set it in range from %s to %s",
- snapshotId, earliestSnapshotId,
latestSnapshotId));
+ "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) &&
dataTable.tagManager().tagExists(tagName)) {
- snapshot =
dataTable.tagManager().tag(tagName).trimToSnapshot();
- } else {
- snapshot = snapshotManager.latestSnapshot();
+ } 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();
}
if (snapshot == null) {
+ LOG.warn("Check if your snapshot is empty.");
return Collections.emptyList();
}
FileStorePathFactory fileStorePathFactory =
dataTable.store().pathFactory();
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 4125a3f83..eacd40847 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
@@ -135,6 +135,20 @@ public class ManifestsTableTest extends TableTestBase {
assertThat(result).containsExactlyElementsOf(expectedRow);
}
+ @Test
+ public void testReadManifestsFromSpecifiedTimestampMillis() throws
Exception {
+ write(table, GenericRow.of(3, 1, 1), GenericRow.of(3, 2, 1));
+ List<InternalRow> expectedRow = getExpectedResult(3L);
+ manifestsTable =
+ (ManifestsTable)
+ manifestsTable.copy(
+ Collections.singletonMap(
+
CoreOptions.SCAN_TIMESTAMP_MILLIS.key(),
+
String.valueOf(System.currentTimeMillis())));
+ List<InternalRow> result = read(manifestsTable);
+ assertThat(result).containsExactlyElementsOf(expectedRow);
+ }
+
@Test
public void testReadManifestsFromNotExistSnapshot() throws Exception {
manifestsTable =
@@ -142,7 +156,7 @@ public class ManifestsTableTest extends TableTestBase {
manifestsTable.copy(
Collections.singletonMap(CoreOptions.SCAN_SNAPSHOT_ID.key(), "3"));
assertThrows(
- "Specified scan.snapshot-id 3 is not exist, you can set it in
range from 1 to 2",
+ "Specified parameter scan.snapshot-id = 3 is not exist, you
can set it in range from 1 to 2",
SnapshotNotExistException.class,
() -> read(manifestsTable));
}
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 4aa73d026..b397be24b 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
@@ -86,11 +86,22 @@ abstract class AnalyzeTableTestBase extends
PaimonSparkTestBase {
spark.sql(s"ANALYZE TABLE T COMPUTE STATISTICS")
+ withSQLConf("spark.paimon.scan.timestamp-millis" ->
System.currentTimeMillis.toString) {
+ checkAnswer(
+ sql("SELECT snapshot_id, schema_id, mergedRecordCount, colstat FROM
`T$statistics`"),
+ Row(2, 0, 2, "{ }"))
+ }
+
spark.sql(s"INSERT INTO T VALUES ('3', 'b', 2, 1)")
spark.sql(s"INSERT INTO T VALUES ('4', 'bbb', 3, 2)")
spark.sql(s"ANALYZE TABLE T COMPUTE STATISTICS")
+ withSQLConf("spark.paimon.scan.timestamp-millis" ->
System.currentTimeMillis.toString) {
+ checkAnswer(
+ sql("SELECT snapshot_id, schema_id, mergedRecordCount, colstat FROM
`T$statistics`"),
+ Row(5, 0, 4, "{ }"))
+ }
// create tag
checkAnswer(
spark.sql("CALL paimon.sys.create_tag(table => 'test.T', tag =>
'test_tag5', snapshot => 5)"),
@@ -131,9 +142,7 @@ abstract class AnalyzeTableTestBase extends
PaimonSparkTestBase {
}
withSQLConf("spark.paimon.scan.snapshot-id" -> "100") {
- checkAnswer(
- sql("SELECT snapshot_id, schema_id, mergedRecordCount, colstat FROM
`T$statistics`"),
- Row(5, 0, 4, "{ }"))
+ Assertions.assertEquals(0, sql("select * from `T$statistics`").count())
}
}
