Skip to content

Commit b1daefe

Browse files
mrdrivingduckmrdrivingduck
committed
feat(catalog): expose latest snapshot metadata
Let catalog clients retrieve the latest snapshot and its schema ID without loading the full snapshot history. Reuse filesystem snapshot resolution and the REST metadata endpoint, preserving the existing branch and missing-table behavior of each catalog.
1 parent 3a45772 commit b1daefe

8 files changed

Lines changed: 145 additions & 0 deletions

File tree

‎include/paimon/catalog/catalog.h‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -233,6 +233,15 @@ class PAIMON_EXPORT Catalog {
233233
virtual Result<std::vector<SnapshotInfo>> ListSnapshots(
234234
const Identifier& identifier, const std::string& branch = "") const = 0;
235235

236+
/// Gets the latest snapshot of the specified table without loading its snapshot history.
237+
///
238+
/// @param identifier The identifier (database and table name) of the table.
239+
/// @param branch Branch name; empty string means the main branch.
240+
/// @return A result containing the latest SnapshotInfo, nullopt if there are no snapshots,
241+
/// or an error status.
242+
virtual Result<std::optional<SnapshotInfo>> GetLatestSnapshot(
243+
const Identifier& identifier, const std::string& branch = "") const = 0;
244+
236245
/// Returns whether this catalog publishes snapshots through catalog commits. Defaults to false.
237246
/// @note Implementations returning true must also implement the internal
238247
/// `VersionManagedCatalog`.

‎src/paimon/core/catalog/file_system_catalog.cpp‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -541,6 +541,22 @@ Result<std::vector<SnapshotInfo>> FileSystemCatalog::ListSnapshots(
541541
return result;
542542
}
543543

544+
Result<std::optional<SnapshotInfo>> FileSystemCatalog::GetLatestSnapshot(
545+
const Identifier& identifier, const std::string& branch) const {
546+
PAIMON_RETURN_NOT_OK(BranchManager::CheckValidBranch(branch));
547+
PAIMON_ASSIGN_OR_RAISE(bool exists, TableExists(identifier));
548+
if (!exists) {
549+
return Status::NotExist(fmt::format("table {} does not exist", identifier.ToString()));
550+
}
551+
PAIMON_ASSIGN_OR_RAISE(std::string table_path, GetTableLocation(identifier));
552+
SnapshotManager mgr(fs_, table_path, branch);
553+
PAIMON_ASSIGN_OR_RAISE(std::optional<Snapshot> snapshot, mgr.LatestSnapshot());
554+
if (!snapshot) {
555+
return std::optional<SnapshotInfo>();
556+
}
557+
return std::optional<SnapshotInfo>(snapshot->ToSnapshotInfo());
558+
}
559+
544560
Result<std::shared_ptr<FormatTable>> FileSystemCatalog::LoadFormatTable(
545561
const Identifier& identifier) const {
546562
return CatalogUtils::LoadFormatTableInTwoRequests(*this, identifier,

‎src/paimon/core/catalog/file_system_catalog.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,8 @@ class FileSystemCatalog : public Catalog {
6868
Result<std::shared_ptr<Table>> GetTable(const Identifier& identifier) const override;
6969
Result<std::vector<SnapshotInfo>> ListSnapshots(const Identifier& identifier,
7070
const std::string& branch) const override;
71+
Result<std::optional<SnapshotInfo>> GetLatestSnapshot(const Identifier& identifier,
72+
const std::string& branch) const override;
7173

7274
protected:
7375
/// This catalog keeps a table's schema under the table's own path, so the `schema` and

‎src/paimon/core/catalog/file_system_catalog_test.cpp‎

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1255,6 +1255,64 @@ TEST(FileSystemCatalogTest, TestListSnapshots) {
12551255
ASSERT_LT(snapshots[0].snapshot_id, snapshots[1].snapshot_id);
12561256
}
12571257

1258+
TEST(FileSystemCatalogTest, TestGetLatestSnapshot) {
1259+
std::map<std::string, std::string> options;
1260+
options[Options::FILE_SYSTEM] = "local";
1261+
options[Options::FILE_FORMAT] = "orc";
1262+
ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options));
1263+
auto dir = UniqueTestDirectory::Create();
1264+
ASSERT_TRUE(dir);
1265+
1266+
std::string test_data_path = GetDataDir() + "/append_table_with_multiple_file_format.db";
1267+
std::string db_path = dir->Str() + "/append_table_with_multiple_file_format.db";
1268+
ASSERT_TRUE(TestUtil::CopyDirectory(test_data_path, db_path));
1269+
1270+
FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options);
1271+
const Catalog& public_catalog = catalog;
1272+
Identifier identifier("append_table_with_multiple_file_format",
1273+
"append_table_with_multiple_file_format");
1274+
ASSERT_OK_AND_ASSIGN(std::optional<SnapshotInfo> snapshot,
1275+
public_catalog.GetLatestSnapshot(identifier));
1276+
ASSERT_TRUE(snapshot);
1277+
ASSERT_EQ(snapshot->snapshot_id, 2);
1278+
ASSERT_EQ(snapshot->schema_id, 1);
1279+
}
1280+
1281+
TEST(FileSystemCatalogTest, TestGetLatestSnapshotEmptyMissingAndBranch) {
1282+
std::map<std::string, std::string> options;
1283+
options[Options::FILE_SYSTEM] = "local";
1284+
options[Options::FILE_FORMAT] = "orc";
1285+
ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options));
1286+
auto dir = UniqueTestDirectory::Create();
1287+
ASSERT_TRUE(dir);
1288+
1289+
std::string empty_test_data_path = GetDataDir() + "/orc/append_table_with_nested_type.db";
1290+
std::string empty_db_path = dir->Str() + "/append_table_with_nested_type.db";
1291+
ASSERT_TRUE(TestUtil::CopyDirectory(empty_test_data_path, empty_db_path));
1292+
std::string branch_test_data_path = GetDataDir() + "/orc/append_table_with_rt_branch.db";
1293+
std::string branch_db_path = dir->Str() + "/append_table_with_rt_branch.db";
1294+
ASSERT_TRUE(TestUtil::CopyDirectory(branch_test_data_path, branch_db_path));
1295+
1296+
FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options);
1297+
ASSERT_OK_AND_ASSIGN(
1298+
std::optional<SnapshotInfo> empty,
1299+
catalog.GetLatestSnapshot(
1300+
Identifier("append_table_with_nested_type", "append_table_with_nested_type"), ""));
1301+
ASSERT_FALSE(empty);
1302+
Identifier identifier("append_table_with_rt_branch", "append_table_with_rt_branch");
1303+
ASSERT_OK_AND_ASSIGN(std::optional<SnapshotInfo> branch_snapshot,
1304+
catalog.GetLatestSnapshot(identifier, "rt"));
1305+
ASSERT_TRUE(branch_snapshot);
1306+
ASSERT_EQ(branch_snapshot->snapshot_id, 1);
1307+
ASSERT_EQ(branch_snapshot->total_record_count, 4);
1308+
1309+
ASSERT_TRUE(catalog.GetLatestSnapshot(Identifier(identifier.GetDatabaseName(), "missing"), "")
1310+
.status()
1311+
.IsNotExist());
1312+
ASSERT_OK_AND_ASSIGN(empty, catalog.GetLatestSnapshot(identifier, "missing"));
1313+
ASSERT_FALSE(empty);
1314+
}
1315+
12581316
namespace {
12591317
class ClaimsVersionManagementCatalog : public FileSystemCatalog {
12601318
public:

‎src/paimon/rest/rest_catalog.cpp‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -508,6 +508,18 @@ Result<std::vector<SnapshotInfo>> RestCatalog::ListSnapshots(const Identifier& i
508508
return result;
509509
}
510510

511+
Result<std::optional<SnapshotInfo>> RestCatalog::GetLatestSnapshot(
512+
const Identifier& identifier, const std::string& branch) const {
513+
PAIMON_RETURN_NOT_OK(CatalogUtils::CheckNotBranch(identifier, "getLatestSnapshot"));
514+
PAIMON_RETURN_NOT_OK(CatalogUtils::CheckNotSystemTable(identifier, "getLatestSnapshot"));
515+
Identifier load_identifier(identifier.GetDatabaseName(), identifier.GetTableName(), branch);
516+
PAIMON_ASSIGN_OR_RAISE(std::optional<Snapshot> snapshot, api_->LoadSnapshot(load_identifier));
517+
if (!snapshot) {
518+
return std::optional<SnapshotInfo>();
519+
}
520+
return std::optional<SnapshotInfo>(snapshot->ToSnapshotInfo());
521+
}
522+
511523
Result<std::optional<Snapshot>> RestCatalog::LoadSnapshot(const Identifier& identifier) const {
512524
PAIMON_RETURN_NOT_OK(CatalogUtils::CheckNotSystemTable(identifier, "loadSnapshot"));
513525
PAIMON_ASSIGN_OR_RAISE(Identifier load_identifier, ToLoadIdentifier(identifier));

‎src/paimon/rest/rest_catalog.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -92,6 +92,8 @@ class RestCatalog : public Catalog, public VersionManagedCatalog {
9292
Result<std::shared_ptr<Table>> GetTable(const Identifier& identifier) const override;
9393
Result<std::vector<SnapshotInfo>> ListSnapshots(const Identifier& identifier,
9494
const std::string& branch) const override;
95+
Result<std::optional<SnapshotInfo>> GetLatestSnapshot(const Identifier& identifier,
96+
const std::string& branch) const override;
9597

9698
bool SupportsVersionManagement() const override {
9799
return true;

‎src/paimon/rest/rest_catalog_test.cpp‎

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1518,6 +1518,48 @@ TEST_F(RestCatalogTest, ListSnapshots) {
15181518
"branch table");
15191519
}
15201520

1521+
TEST_F(RestCatalogTest, GetLatestSnapshot) {
1522+
ASSERT_OK_AND_ASSIGN(std::unique_ptr<RestCatalog> catalog, CreateRestCatalog());
1523+
ASSERT_OK(catalog->CreateDatabase("db1", {}, false));
1524+
Identifier identifier("db1", "t1");
1525+
ASSERT_OK(CreateSampleTable(catalog.get(), identifier));
1526+
const Catalog& public_catalog = *catalog;
1527+
ASSERT_OK_AND_ASSIGN(std::optional<SnapshotInfo> empty,
1528+
public_catalog.GetLatestSnapshot(identifier));
1529+
ASSERT_FALSE(empty);
1530+
{
1531+
std::lock_guard<std::mutex> lock(state_->mutex);
1532+
state_->current_snapshot = SnapshotJson(7);
1533+
}
1534+
ASSERT_OK_AND_ASSIGN(std::optional<SnapshotInfo> snapshot,
1535+
public_catalog.GetLatestSnapshot(identifier));
1536+
ASSERT_TRUE(snapshot);
1537+
ASSERT_EQ(snapshot->snapshot_id, 7);
1538+
ASSERT_EQ(snapshot->schema_id, 0);
1539+
}
1540+
1541+
TEST_F(RestCatalogTest, GetLatestSnapshotBranchAndErrors) {
1542+
ASSERT_OK_AND_ASSIGN(std::unique_ptr<RestCatalog> catalog, CreateRestCatalog());
1543+
ASSERT_OK(catalog->CreateDatabase("db1", {}, false));
1544+
Identifier identifier("db1", "t1");
1545+
ASSERT_OK(CreateSampleTable(catalog.get(), identifier));
1546+
{
1547+
std::lock_guard<std::mutex> lock(state_->mutex);
1548+
state_->current_snapshot = SnapshotJson(7);
1549+
state_->databases["db1"]["t1$branch_b1"] = state_->databases["db1"]["t1"];
1550+
}
1551+
ASSERT_OK_AND_ASSIGN(std::optional<SnapshotInfo> snapshot,
1552+
catalog->GetLatestSnapshot(identifier, "b1"));
1553+
ASSERT_TRUE(snapshot);
1554+
ASSERT_EQ(snapshot->snapshot_id, 7);
1555+
{
1556+
std::lock_guard<std::mutex> lock(state_->mutex);
1557+
ASSERT_EQ(state_->last_snapshot_table, "t1$branch_b1");
1558+
}
1559+
ASSERT_TRUE(catalog->GetLatestSnapshot(Identifier("db1", "missing"), "").status().IsNotExist());
1560+
ASSERT_TRUE(catalog->GetLatestSnapshot(identifier, "missing").status().IsNotExist());
1561+
}
1562+
15211563
TEST_F(RestCatalogTest, FileStoreCommitIsBuiltFromTheCatalog) {
15221564
std::unique_ptr<UniqueTestDirectory> dir = UniqueTestDirectory::Create();
15231565
ASSERT_TRUE(dir);

‎src/paimon/testing/mock/mock_catalog.h‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -255,6 +255,10 @@ class MockVersionManagedCatalog : public Catalog, public VersionManagedCatalog {
255255
const std::string&) const override {
256256
return Unsupported();
257257
}
258+
Result<std::optional<SnapshotInfo>> GetLatestSnapshot(const Identifier&,
259+
const std::string&) const override {
260+
return Unsupported();
261+
}
258262

259263
private:
260264
static Status Unsupported() {

0 commit comments

Comments
 (0)