[core] Manifest and Statistic support query by timestamp #4304
Conversation
| @@ -200,6 +200,8 @@ private static List<ManifestFileMeta> allManifests(FileStoreTable dataTable) { | |||
| SnapshotManager snapshotManager = dataTable.snapshotManager(); | |||
| Long snapshotId = coreOptions.scanSnapshotId(); | |||
| String tagName = coreOptions.scanTagName(); | |||
| Long timestampMills = coreOptions.scanTimestampMills(); | |||
|
|
|||
| Snapshot snapshot = null; | |||
There was a problem hiding this comment.
Variable 'snapshot' initializer 'null' is redundant.
Change "Snapshot snapshot = null;" to "Snapshot snapshot;".
| } else if (timestampMills != null) { | ||
| snapshot = snapshotManager.earlierOrEqualTimeMills(timestampMills); | ||
| } else { | ||
| snapshot = snapshotManager.latestSnapshot(); | ||
| } | ||
|
|
||
| if (snapshot == null) { |
| snapshot = snapshotManager.latestSnapshot(); | ||
| } else if (!StringUtils.isEmpty(tagName)) { | ||
| if (!dataTable.tagManager().tagExists(tagName)) { | ||
| throw new RuntimeException( |
There was a problem hiding this comment.
throw new RuntimeException(
String.format("Specified parameter %s = %s is not exist.", SCAN_TAG_NAME.key(), tagName));
| @@ -212,12 +214,16 @@ private static List<ManifestFileMeta> allManifests(FileStoreTable dataTable) { | |||
| snapshotId, earliestSnapshotId, latestSnapshotId)); | |||
There was a problem hiding this comment.
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));
| @@ -135,6 +135,24 @@ public void testReadManifestsFromSpecifiedTagName() throws Exception { | |||
| assertThat(result).containsExactlyElementsOf(expectedRow); | |||
| } | |||
|
|
|||
| @Test | |||
| public void testReadManifestsFromSpecifiedTimestampMillis() throws Exception { | |||
| Thread.sleep(10); | |||
| public void testReadManifestsFromSpecifiedTimestampMillis() throws Exception { | ||
| Thread.sleep(10); | ||
| write(table, GenericRow.of(3, 1, 1), GenericRow.of(3, 2, 1)); | ||
| long time = System.currentTimeMillis(); |
There was a problem hiding this comment.
inline "long time = System.currentTimeMillis();"
| spark.sql(s"ANALYZE TABLE T COMPUTE STATISTICS") | ||
| val time1 = System.currentTimeMillis |
There was a problem hiding this comment.
inline "val time1 = System.currentTimeMillis"
| spark.sql(s"ANALYZE TABLE T COMPUTE STATISTICS") | ||
| val time2 = System.currentTimeMillis |
There was a problem hiding this comment.
inline "val time2 = System.currentTimeMillis"
| @@ -118,6 +121,12 @@ abstract class AnalyzeTableTestBase extends PaimonSparkTestBase { | |||
| Row(2, 0, 2, "{ }")) | |||
| } | |||
|
|
|||
| withSQLConf("spark.paimon.scan.timestamp-millis" -> time1.toString) { | |||
| checkAnswer( | |||
| sql("SELECT snapshot_id, schema_id, mergedRecordCount, colstat FROM `T$statistics`"), | |||
There was a problem hiding this comment.
I think you can modify "colstat" to "colStat" in StatisticTable. And then modify other relevant codes.
|
@wwj6591812 thanks for your review,I have completed the modification. |
|
+1 |
| @@ -72,7 +72,7 @@ public class StatisticTable implements ReadonlyTable { | |||
| new DataField(1, "schema_id", new BigIntType(false)), | |||
| new DataField(2, "mergedRecordCount", new BigIntType(true)), | |||
| new DataField(3, "mergedRecordSize", new BigIntType(true)), | |||
| new DataField(4, "colstat", SerializationUtils.newStringType(true)))); | |||
There was a problem hiding this comment.
please don't modify this, it is better to keep compability.
|
+1 |
Purpose
fllow up #4291
Linked issue: close #xxx
Tests
API and Format
Documentation