You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by lz...@apache.org on 2023/02/27 08:39:23 UTC
[flink-table-store] branch master updated: [FLINK-31009] Add recordCount to snapshot meta
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/flink-table-store.git
The following commit(s) were added to refs/heads/master by this push:
new d79b39a6 [FLINK-31009] Add recordCount to snapshot meta
d79b39a6 is described below
commit d79b39a63bcd550a516d6b6caa8e9e8a092dac01
Author: Nicholas Jiang <pr...@163.com>
AuthorDate: Mon Feb 27 16:39:18 2023 +0800
[FLINK-31009] Add recordCount to snapshot meta
This closes #546
---
docs/content/docs/how-to/querying-tables.md | 14 +--
.../apache/flink/table/store/file/Snapshot.java | 134 +++++++++++++++++++--
.../file/operation/AbstractFileStoreScan.java | 14 +--
.../store/file/operation/FileStoreCommitImpl.java | 14 ++-
.../store/file/operation/FileStoreExpireImpl.java | 4 +-
.../table/store/table/system/SnapshotsTable.java | 10 +-
.../flink/table/store/file/TestFileStore.java | 5 +-
.../store/file/operation/FileStoreCommitTest.java | 50 +++++++-
.../file/operation/KeyValueFileStoreScanTest.java | 2 +-
.../connector/source/LogHybridSourceFactory.java | 2 +-
.../table/store/connector/CatalogTableITCase.java | 13 ++
.../flink/table/store/spark/SparkReadITCase.java | 13 ++
12 files changed, 234 insertions(+), 41 deletions(-)
diff --git a/docs/content/docs/how-to/querying-tables.md b/docs/content/docs/how-to/querying-tables.md
index c2fde114..da8159d2 100644
--- a/docs/content/docs/how-to/querying-tables.md
+++ b/docs/content/docs/how-to/querying-tables.md
@@ -100,18 +100,18 @@ SELECT * FROM my_catalog.my_db.`MyTable$snapshots`;
### Snapshots Table
-You can query the snapshot history information of the table through snapshots table.
+You can query the snapshot history information of the table through snapshots table, including the record count occurred in the snapshot.
```sql
SELECT * FROM MyTable$snapshots;
/*
-+--------------+------------+-----------------+-------------------+--------------+-------------------------+
-| snapshot_id | schema_id | commit_user | commit_identifier | commit_kind | commit_time |
-+--------------+------------+-----------------+-------------------+--------------+-------------------------+
-| 2 | 0 | 7ca4cd28-98e... | 2 | APPEND | 2022-10-26 11:44:15.600 |
-| 1 | 0 | 870062aa-3e9... | 1 | APPEND | 2022-10-26 11:44:15.148 |
-+--------------+------------+-----------------+-------------------+--------------+-------------------------+
++--------------+------------+-----------------+-------------------+--------------+-------------------------+---------------------+---------------------+-------------------------+
+| snapshot_id | schema_id | commit_user | commit_identifier | commit_kind | commit_time | total_record_count | delta_record_count | changelog_record_count |
++--------------+------------+-----------------+-------------------+--------------+-------------------------+---------------------+---------------------+-------------------------+
+| 2 | 0 | 7ca4cd28-98e... | 2 | APPEND | 2022-10-26 11:44:15.600 | 2 | 2 | 0 |
+| 1 | 0 | 870062aa-3e9... | 1 | APPEND | 2022-10-26 11:44:15.148 | 1 | 1 | 0 |
++--------------+------------+-----------------+-------------------+--------------+-------------------------+---------------------+---------------------+-------------------------+
2 rows in set
*/
```
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/Snapshot.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/Snapshot.java
index d12720fc..21c2ecee 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/Snapshot.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/Snapshot.java
@@ -18,8 +18,10 @@
package org.apache.flink.table.store.file;
+import org.apache.flink.table.store.file.manifest.ManifestEntry;
import org.apache.flink.table.store.file.manifest.ManifestFileMeta;
import org.apache.flink.table.store.file.manifest.ManifestList;
+import org.apache.flink.table.store.file.operation.FileStoreScan;
import org.apache.flink.table.store.file.utils.JsonSerdeUtil;
import org.apache.flink.table.store.fs.FileIO;
import org.apache.flink.table.store.fs.Path;
@@ -32,6 +34,7 @@ import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -45,6 +48,8 @@ import java.util.Map;
* file.
* <li>Version 2: Introduced in table store 0.3. Add "version" field and "changelogManifestList"
* field.
+ * <li>Version 3: Introduced in table store 0.4. Add "baseRecordCount" field, "deltaRecordCount"
+ * field and "changelogRecordCount" field.
* </ul>
*
* <p>Unversioned change list:
@@ -60,8 +65,8 @@ public class Snapshot {
public static final long FIRST_SNAPSHOT_ID = 1;
- private static final int TABLE_STORE_02_VERSION = 1;
- private static final int CURRENT_VERSION = 2;
+ public static final int TABLE_STORE_02_VERSION = 1;
+ private static final int CURRENT_VERSION = 3;
private static final String FIELD_VERSION = "version";
private static final String FIELD_ID = "id";
@@ -74,6 +79,9 @@ public class Snapshot {
private static final String FIELD_COMMIT_KIND = "commitKind";
private static final String FIELD_TIME_MILLIS = "timeMillis";
private static final String FIELD_LOG_OFFSETS = "logOffsets";
+ private static final String FIELD_TOTAL_RECORD_COUNT = "totalRecordCount";
+ private static final String FIELD_DELTA_RECORD_COUNT = "deltaRecordCount";
+ private static final String FIELD_CHANGELOG_RECORD_COUNT = "changelogRecordCount";
// version of snapshot
// null for table store <= 0.2
@@ -124,6 +132,24 @@ public class Snapshot {
@JsonProperty(FIELD_LOG_OFFSETS)
private final Map<Integer, Long> logOffsets;
+ // record count of all changes occurred in this snapshot
+ // null for table store <= 0.3
+ @JsonProperty(FIELD_TOTAL_RECORD_COUNT)
+ @Nullable
+ private final Long totalRecordCount;
+
+ // record count of all new changes occurred in this snapshot
+ // null for table store <= 0.3
+ @JsonProperty(FIELD_DELTA_RECORD_COUNT)
+ @Nullable
+ private final Long deltaRecordCount;
+
+ // record count of all changelog produced in this snapshot
+ // null for table store <= 0.3
+ @JsonProperty(FIELD_CHANGELOG_RECORD_COUNT)
+ @Nullable
+ private final Long changelogRecordCount;
+
public Snapshot(
long id,
long schemaId,
@@ -134,7 +160,10 @@ public class Snapshot {
long commitIdentifier,
CommitKind commitKind,
long timeMillis,
- Map<Integer, Long> logOffsets) {
+ Map<Integer, Long> logOffsets,
+ @Nullable Long totalRecordCount,
+ @Nullable Long deltaRecordCount,
+ @Nullable Long changelogRecordCount) {
this(
CURRENT_VERSION,
id,
@@ -146,7 +175,10 @@ public class Snapshot {
commitIdentifier,
commitKind,
timeMillis,
- logOffsets);
+ logOffsets,
+ totalRecordCount,
+ deltaRecordCount,
+ changelogRecordCount);
}
@JsonCreator
@@ -161,7 +193,10 @@ public class Snapshot {
@JsonProperty(FIELD_COMMIT_IDENTIFIER) long commitIdentifier,
@JsonProperty(FIELD_COMMIT_KIND) CommitKind commitKind,
@JsonProperty(FIELD_TIME_MILLIS) long timeMillis,
- @JsonProperty(FIELD_LOG_OFFSETS) Map<Integer, Long> logOffsets) {
+ @JsonProperty(FIELD_LOG_OFFSETS) Map<Integer, Long> logOffsets,
+ @JsonProperty(FIELD_TOTAL_RECORD_COUNT) Long totalRecordCount,
+ @JsonProperty(FIELD_DELTA_RECORD_COUNT) Long deltaRecordCount,
+ @JsonProperty(FIELD_CHANGELOG_RECORD_COUNT) Long changelogRecordCount) {
this.version = version;
this.id = id;
this.schemaId = schemaId;
@@ -173,6 +208,9 @@ public class Snapshot {
this.commitKind = commitKind;
this.timeMillis = timeMillis;
this.logOffsets = logOffsets;
+ this.totalRecordCount = totalRecordCount;
+ this.deltaRecordCount = deltaRecordCount;
+ this.changelogRecordCount = changelogRecordCount;
}
@JsonGetter(FIELD_VERSION)
@@ -228,21 +266,97 @@ public class Snapshot {
}
@JsonGetter(FIELD_LOG_OFFSETS)
- public Map<Integer, Long> getLogOffsets() {
+ public Map<Integer, Long> logOffsets() {
return logOffsets;
}
- public String toJson() {
- return JsonSerdeUtil.toJson(this);
+ @JsonGetter(FIELD_TOTAL_RECORD_COUNT)
+ @Nullable
+ public Long totalRecordCount() {
+ return totalRecordCount;
+ }
+
+ @JsonGetter(FIELD_DELTA_RECORD_COUNT)
+ @Nullable
+ public Long deltaRecordCount() {
+ return deltaRecordCount;
}
- public List<ManifestFileMeta> readAllDataManifests(ManifestList manifestList) {
+ @JsonGetter(FIELD_CHANGELOG_RECORD_COUNT)
+ @Nullable
+ public Long changelogRecordCount() {
+ return changelogRecordCount;
+ }
+
+ /**
+ * Return all {@link ManifestFileMeta} instances for either data or changelog manifests in this
+ * snapshot.
+ *
+ * @param manifestList a {@link ManifestList} instance used for reading files at snapshot.
+ * @return a list of ManifestFileMeta.
+ */
+ public List<ManifestFileMeta> allManifests(ManifestList manifestList) {
+ List<ManifestFileMeta> result = new ArrayList<>();
+ result.addAll(dataManifests(manifestList));
+ result.addAll(changelogManifests(manifestList));
+ return result;
+ }
+
+ /**
+ * Return a {@link ManifestFileMeta} for each data manifest in this snapshot.
+ *
+ * @param manifestList a {@link ManifestList} instance used for reading files at snapshot.
+ * @return a list of ManifestFileMeta.
+ */
+ public List<ManifestFileMeta> dataManifests(ManifestList manifestList) {
List<ManifestFileMeta> result = new ArrayList<>();
result.addAll(manifestList.read(baseManifestList));
- result.addAll(manifestList.read(deltaManifestList));
+ result.addAll(deltaManifests(manifestList));
return result;
}
+ /**
+ * Return a {@link ManifestFileMeta} for each delta manifest in this snapshot.
+ *
+ * @param manifestList a {@link ManifestList} instance used for reading files at snapshot.
+ * @return a list of ManifestFileMeta.
+ */
+ public List<ManifestFileMeta> deltaManifests(ManifestList manifestList) {
+ return manifestList.read(deltaManifestList);
+ }
+
+ /**
+ * Return a {@link ManifestFileMeta} for each changelog manifest in this snapshot.
+ *
+ * @param manifestList a {@link ManifestList} instance used for reading files at snapshot.
+ * @return a list of ManifestFileMeta.
+ */
+ public List<ManifestFileMeta> changelogManifests(ManifestList manifestList) {
+ return changelogManifestList == null
+ ? Collections.emptyList()
+ : manifestList.read(changelogManifestList);
+ }
+
+ /**
+ * Return record count of all changes occurred in this snapshot given the scan.
+ *
+ * @param scan a {@link FileStoreScan} instance used for count of reading files at snapshot.
+ * @return total record count of Snapshot.
+ */
+ public Long totalRecordCount(FileStoreScan scan) {
+ return totalRecordCount == null
+ ? recordCount(scan.withSnapshot(id).plan().files())
+ : totalRecordCount;
+ }
+
+ public static long recordCount(List<ManifestEntry> manifestEntries) {
+ return manifestEntries.stream().mapToLong(manifest -> manifest.file().rowCount()).sum();
+ }
+
+ public String toJson() {
+ return JsonSerdeUtil.toJson(this);
+ }
+
public static Snapshot fromJson(String json) {
return JsonSerdeUtil.fromJson(json, Snapshot.class);
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/AbstractFileStoreScan.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/AbstractFileStoreScan.java
index 6fee0ead..2db95384 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/AbstractFileStoreScan.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/AbstractFileStoreScan.java
@@ -252,22 +252,18 @@ public abstract class AbstractFileStoreScan implements FileStoreScan {
private List<ManifestFileMeta> readManifests(Snapshot snapshot) {
switch (scanKind) {
case ALL:
- return snapshot.readAllDataManifests(manifestList);
+ return snapshot.dataManifests(manifestList);
case DELTA:
- return manifestList.read(snapshot.deltaManifestList());
+ return snapshot.deltaManifests(manifestList);
case CHANGELOG:
- if (snapshot.version() >= 2) {
- if (snapshot.changelogManifestList() == null) {
- return Collections.emptyList();
- } else {
- return manifestList.read(snapshot.changelogManifestList());
- }
+ if (snapshot.version() > Snapshot.TABLE_STORE_02_VERSION) {
+ return snapshot.changelogManifests(manifestList);
}
// compatible with Table Store 0.2, we'll read extraFiles in DataFileMeta
// see comments on DataFileMeta#extraFiles
if (snapshot.commitKind() == Snapshot.CommitKind.APPEND) {
- return manifestList.read(snapshot.deltaManifestList());
+ return snapshot.deltaManifests(manifestList);
}
throw new IllegalStateException(
String.format(
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreCommitImpl.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreCommitImpl.java
index 9fd6c3f2..5540ee3b 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreCommitImpl.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreCommitImpl.java
@@ -477,12 +477,16 @@ public class FileStoreCommitImpl implements FileStoreCommit {
List<ManifestFileMeta> newMetas = new ArrayList<>();
List<ManifestFileMeta> changelogMetas = new ArrayList<>();
try {
+ long previousTotalRecordCount = 0L;
if (latestSnapshot != null) {
+ previousTotalRecordCount = latestSnapshot.totalRecordCount(scan);
+ List<ManifestFileMeta> previousManifests =
+ latestSnapshot.dataManifests(manifestList);
// read all previous manifest files
- oldMetas.addAll(latestSnapshot.readAllDataManifests(manifestList));
+ oldMetas.addAll(previousManifests);
// read the last snapshot to complete the bucket's offsets when logOffsets does not
// contain all buckets
- latestSnapshot.getLogOffsets().forEach(logOffsets::putIfAbsent);
+ latestSnapshot.logOffsets().forEach(logOffsets::putIfAbsent);
}
// merge manifest files with changes
newMetas.addAll(
@@ -494,6 +498,7 @@ public class FileStoreCommitImpl implements FileStoreCommit {
previousChangesListName = manifestList.write(newMetas);
// write new changes into manifest files
+ long deltaRecordCount = Snapshot.recordCount(tableFiles);
List<ManifestFileMeta> newChangesManifests = manifestFile.write(tableFiles);
newMetas.addAll(newChangesManifests);
newChangesListName = manifestList.write(newChangesManifests);
@@ -516,7 +521,10 @@ public class FileStoreCommitImpl implements FileStoreCommit {
identifier,
commitKind,
System.currentTimeMillis(),
- logOffsets);
+ logOffsets,
+ previousTotalRecordCount + deltaRecordCount,
+ deltaRecordCount,
+ Snapshot.recordCount(changelogFiles));
} catch (Throwable e) {
// fails when preparing for commit, we should clean up
cleanUpTmpManifests(
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreExpireImpl.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreExpireImpl.java
index 3fda2887..c79b83eb 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreExpireImpl.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreExpireImpl.java
@@ -184,7 +184,7 @@ public class FileStoreExpireImpl implements FileStoreExpire {
// delete manifests
Snapshot exclusiveSnapshot = snapshotManager.snapshot(endExclusiveId);
Set<ManifestFileMeta> manifestsInUse =
- new HashSet<>(exclusiveSnapshot.readAllDataManifests(manifestList));
+ new HashSet<>(exclusiveSnapshot.dataManifests(manifestList));
// to avoid deleting twice
Set<ManifestFileMeta> deletedManifests = new HashSet<>();
for (long id = beginInclusiveId; id < endExclusiveId; id++) {
@@ -193,7 +193,7 @@ public class FileStoreExpireImpl implements FileStoreExpire {
}
Snapshot toExpire = snapshotManager.snapshot(id);
- // cannot call `toExpire.readAllDataManifests` directly, it is possible that a job is
+ // cannot call `toExpire.dataManifests` directly, it is possible that a job is
// killed during expiration, so some manifest files may have been deleted
List<ManifestFileMeta> toExpireManifests = new ArrayList<>();
toExpireManifests.addAll(tryReadManifestList(toExpire.baseManifestList()));
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/system/SnapshotsTable.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/system/SnapshotsTable.java
index 9292beb1..ace73aed 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/system/SnapshotsTable.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/system/SnapshotsTable.java
@@ -71,7 +71,10 @@ public class SnapshotsTable implements Table {
new DataField(3, "commit_identifier", new BigIntType(false)),
new DataField(
4, "commit_kind", SerializationUtils.newStringType(false)),
- new DataField(5, "commit_time", new TimestampType(false, 3))));
+ new DataField(5, "commit_time", new TimestampType(false, 3)),
+ new DataField(6, "total_record_count", new BigIntType(true)),
+ new DataField(7, "delta_record_count", new BigIntType(true)),
+ new DataField(8, "changelog_record_count", new BigIntType(true))));
private final FileIO fileIO;
private final Path location;
@@ -206,7 +209,10 @@ public class SnapshotsTable implements Table {
Timestamp.fromLocalDateTime(
LocalDateTime.ofInstant(
Instant.ofEpochMilli(snapshot.timeMillis()),
- ZoneId.systemDefault())));
+ ZoneId.systemDefault())),
+ snapshot.totalRecordCount(),
+ snapshot.deltaRecordCount(),
+ snapshot.changelogRecordCount());
}
}
}
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/TestFileStore.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/TestFileStore.java
index 23eee675..e6df05da 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/TestFileStore.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/TestFileStore.java
@@ -457,10 +457,7 @@ public class TestFileStore extends KeyValueFileStore {
}
// manifests
- List<ManifestFileMeta> manifests = snapshot.readAllDataManifests(manifestList);
- if (snapshot.changelogManifestList() != null) {
- manifests.addAll(manifestList.read(snapshot.changelogManifestList()));
- }
+ List<ManifestFileMeta> manifests = snapshot.allManifests(manifestList);
manifests.forEach(m -> result.add(pathFactory.toManifestFilePath(m.fileName())));
// data file
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/FileStoreCommitTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/FileStoreCommitTest.java
index 79656f9e..bca0ba9b 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/FileStoreCommitTest.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/FileStoreCommitTest.java
@@ -418,7 +418,7 @@ public class FileStoreCommitTest {
offsets.put(1, 3L);
Snapshot snapshot =
store.commitData(generateDataList(10), gen::getPartition, kv -> 0, offsets).get(0);
- assertThat(snapshot.getLogOffsets()).isEqualTo(offsets);
+ assertThat(snapshot.logOffsets()).isEqualTo(offsets);
// commit 2
offsets = new HashMap<>();
@@ -428,7 +428,53 @@ public class FileStoreCommitTest {
Map<Integer, Long> expected = new HashMap<>();
expected.put(0, 1L);
expected.put(1, 8L);
- assertThat(snapshot.getLogOffsets()).isEqualTo(expected);
+ assertThat(snapshot.logOffsets()).isEqualTo(expected);
+ }
+
+ @Test
+ public void testSnapshotRecordCount() throws Exception {
+ TestFileStore store = createStore(false);
+
+ // commit 1
+ Snapshot snapshot1 =
+ store.commitData(
+ generateDataList(10),
+ gen::getPartition,
+ kv -> 0,
+ Collections.emptyMap())
+ .get(0);
+ long deltaRecordCount1 = snapshot1.deltaRecordCount();
+ assertThat(deltaRecordCount1).isNotEqualTo(0L);
+ assertThat(snapshot1.totalRecordCount()).isEqualTo(deltaRecordCount1);
+ assertThat(snapshot1.changelogRecordCount()).isEqualTo(0L);
+
+ // commit 2
+ Snapshot snapshot2 =
+ store.commitData(
+ generateDataList(20),
+ gen::getPartition,
+ kv -> 0,
+ Collections.emptyMap())
+ .get(0);
+ long deltaRecordCount2 = snapshot2.deltaRecordCount();
+ assertThat(deltaRecordCount2).isNotEqualTo(0L);
+ assertThat(snapshot2.totalRecordCount())
+ .isEqualTo(snapshot1.totalRecordCount() + deltaRecordCount2);
+ assertThat(snapshot2.changelogRecordCount()).isEqualTo(0L);
+
+ // commit 3
+ Snapshot snapshot3 =
+ store.commitData(
+ generateDataList(30),
+ gen::getPartition,
+ kv -> 0,
+ Collections.emptyMap())
+ .get(0);
+ long deltaRecordCount3 = snapshot3.deltaRecordCount();
+ assertThat(deltaRecordCount3).isNotEqualTo(0L);
+ assertThat(snapshot3.totalRecordCount())
+ .isEqualTo(snapshot2.totalRecordCount() + deltaRecordCount3);
+ assertThat(snapshot3.changelogRecordCount()).isEqualTo(0L);
}
@Test
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreScanTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreScanTest.java
index 16152854..eb95046f 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreScanTest.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreScanTest.java
@@ -201,7 +201,7 @@ public class KeyValueFileStoreScanTest {
ManifestList manifestList = store.manifestListFactory().create();
long wantedSnapshotId = random.nextLong(snapshotManager.latestSnapshotId()) + 1;
Snapshot wantedSnapshot = snapshotManager.snapshot(wantedSnapshotId);
- List<ManifestFileMeta> wantedManifests = wantedSnapshot.readAllDataManifests(manifestList);
+ List<ManifestFileMeta> wantedManifests = wantedSnapshot.dataManifests(manifestList);
FileStoreScan scan = store.newScan();
scan.withManifestList(wantedManifests);
diff --git a/flink-table-store-flink/flink-table-store-flink-common/src/main/java/org/apache/flink/table/store/connector/source/LogHybridSourceFactory.java b/flink-table-store-flink/flink-table-store-flink-common/src/main/java/org/apache/flink/table/store/connector/source/LogHybridSourceFactory.java
index e73a4e34..4036bc17 100644
--- a/flink-table-store-flink/flink-table-store-flink-common/src/main/java/org/apache/flink/table/store/connector/source/LogHybridSourceFactory.java
+++ b/flink-table-store-flink/flink-table-store-flink-common/src/main/java/org/apache/flink/table/store/connector/source/LogHybridSourceFactory.java
@@ -44,7 +44,7 @@ public class LogHybridSourceFactory
Snapshot snapshot = enumerator.snapshot();
Map<Integer, Long> logOffsets = null;
if (snapshot != null) {
- logOffsets = snapshot.getLogOffsets();
+ logOffsets = snapshot.logOffsets();
}
return provider.createSource(logOffsets);
}
diff --git a/flink-table-store-flink/flink-table-store-flink-common/src/test/java/org/apache/flink/table/store/connector/CatalogTableITCase.java b/flink-table-store-flink/flink-table-store-flink-common/src/test/java/org/apache/flink/table/store/connector/CatalogTableITCase.java
index 548a35f0..51d7d408 100644
--- a/flink-table-store-flink/flink-table-store-flink-common/src/test/java/org/apache/flink/table/store/connector/CatalogTableITCase.java
+++ b/flink-table-store-flink/flink-table-store-flink-common/src/test/java/org/apache/flink/table/store/connector/CatalogTableITCase.java
@@ -57,6 +57,19 @@ public class CatalogTableITCase extends CatalogITCaseBase {
.containsExactlyInAnyOrder(Row.of(1L, 0L, "APPEND"), Row.of(2L, 0L, "APPEND"));
}
+ @Test
+ public void testSnapshotsTableWithRecordCount() throws Exception {
+ sql("CREATE TABLE T (a INT, b INT)");
+ sql("INSERT INTO T VALUES (1, 2)");
+ sql("INSERT INTO T VALUES (3, 4)");
+
+ List<Row> result =
+ sql(
+ "SELECT snapshot_id, total_record_count, delta_record_count, changelog_record_count FROM T$snapshots");
+ assertThat(result)
+ .containsExactlyInAnyOrder(Row.of(1L, 1L, 1L, 0L), Row.of(2L, 2L, 1L, 0L));
+ }
+
@Test
public void testOptionsTable() throws Exception {
sql("CREATE TABLE T (a INT, b INT)");
diff --git a/flink-table-store-spark/src/test/java/org/apache/flink/table/store/spark/SparkReadITCase.java b/flink-table-store-spark/src/test/java/org/apache/flink/table/store/spark/SparkReadITCase.java
index bee716f5..84de2f05 100644
--- a/flink-table-store-spark/src/test/java/org/apache/flink/table/store/spark/SparkReadITCase.java
+++ b/flink-table-store-spark/src/test/java/org/apache/flink/table/store/spark/SparkReadITCase.java
@@ -99,6 +99,19 @@ public class SparkReadITCase extends SparkReadTestBase {
+ "{\"id\":2,\"name\":\"c\",\"type\":\"STRING\"}]");
}
+ @Test
+ public void testSnapshotsTableWithRecordCount() {
+ List<Row> rows =
+ spark.table("tablestore.default.`t1$snapshots`")
+ .select(
+ "snapshot_id",
+ "total_record_count",
+ "delta_record_count",
+ "changelog_record_count")
+ .collectAsList();
+ assertThat(rows.toString()).isEqualTo("[[1,2,2,0]]");
+ }
+
@Test
public void testCatalogFilterPushDown() {
innerTestSimpleTypeFilterPushDown(spark.table("tablestore.default.t1"));