Skip to content

feat: Add manifests and files system tables #309

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
suxiaogang223 wants to merge 10 commits into
base: main
Choose a base branch
from
+816 −5
Open

feat: Add manifests and files system tables #309

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
ff2b7da
Implement manifests and files system tables
suxiaogang223 May 27, 2026
fc58d0c
Refactor system table implementations
suxiaogang223 May 27, 2026
866f817
Narrow system table utilities
suxiaogang223 May 27, 2026
5d9dddb
Revert "Narrow system table utilities"
suxiaogang223 May 28, 2026
ff9be72
Revert "Refactor system table implementations"
suxiaogang223 May 28, 2026
efe2273
Merge branch 'main' into codex/system-table-files-manifests-pr4
lxy-9602 May 31, 2026
4d34c89
Address files system table review comments
suxiaogang223 Jun 1, 2026
3f8e317
Merge branch 'main' into codex/system-table-files-manifests-pr4
lxy-9602 Jun 2, 2026
1a2d85c
Fix manifests TODO lint
suxiaogang223 Jun 3, 2026
03b6184
Simplify local timestamp handling
suxiaogang223 Jun 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 51 additions & 2 deletions src/paimon/core/catalog/file_system_catalog_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -300,8 +300,8 @@ TEST(FileSystemCatalogTest, TestMetadataSystemTableCatalog) {
/*ignore_if_exists=*/false));
ArrowSchemaRelease(&schema);

std::vector<std::string> metadata_tables = {"snapshots", "schemas", "tags", "branches",
"consumers"};
std::vector<std::string> metadata_tables = {"snapshots", "schemas", "tags", "branches",
"consumers", "manifests", "files"};
for (const auto& table_name : metadata_tables) {
Identifier system_identifier("db1", "tbl1$" + table_name);
ASSERT_OK_AND_ASSIGN(bool exists, catalog.TableExists(system_identifier));
Expand Down Expand Up @@ -363,6 +363,55 @@ TEST(FileSystemCatalogTest, TestMetadataSystemTableCatalog) {
(std::vector<std::string>{"consumer_id", "next_snapshot_id"}));
ASSERT_FALSE(consumers_arrow_schema->field(1)->nullable());

ASSERT_OK_AND_ASSIGN(std::shared_ptr<Schema> manifests_schema,
catalog.LoadTableSchema(Identifier("db1", "tbl1$manifests")));
ASSERT_OK_AND_ASSIGN(auto manifests_c_schema, manifests_schema->GetArrowSchema());
auto manifests_arrow_schema = arrow::ImportSchema(manifests_c_schema.get()).ValueUnsafe();
ASSERT_EQ(manifests_arrow_schema->field_names(),
(std::vector<std::string>{"file_name", "file_size", "num_added_files",
"num_deleted_files", "schema_id", "min_partition_stats",
"max_partition_stats", "min_row_id", "max_row_id"}));
ASSERT_FALSE(manifests_arrow_schema->field(0)->nullable());
ASSERT_EQ(manifests_arrow_schema->field(1)->type()->id(), arrow::Type::INT64);
ASSERT_FALSE(manifests_arrow_schema->field(4)->nullable());
ASSERT_TRUE(manifests_arrow_schema->field(5)->nullable());
ASSERT_TRUE(manifests_arrow_schema->field(8)->nullable());

ASSERT_OK_AND_ASSIGN(std::shared_ptr<Schema> files_schema,
catalog.LoadTableSchema(Identifier("db1", "tbl1$files")));
ASSERT_OK_AND_ASSIGN(auto files_c_schema, files_schema->GetArrowSchema());
auto files_arrow_schema = arrow::ImportSchema(files_c_schema.get()).ValueUnsafe();
ASSERT_EQ(files_arrow_schema->field_names(), (std::vector<std::string>{"partition",
"bucket",
"file_path",
"file_format",
"schema_id",
"level",
"record_count",
"file_size_in_bytes",
"min_key",
"max_key",
"null_value_counts",
"min_value_stats",
"max_value_stats",
"min_sequence_number",
"max_sequence_number",
"creation_time",
"deleteRowCount",
"file_source",
"first_row_id",
"write_cols"}));
ASSERT_TRUE(files_arrow_schema->field(0)->nullable());
ASSERT_FALSE(files_arrow_schema->field(1)->nullable());
ASSERT_FALSE(files_arrow_schema->field(2)->nullable());
ASSERT_FALSE(files_arrow_schema->field(10)->nullable());
ASSERT_EQ(files_arrow_schema->field(15)->type()->id(), arrow::Type::TIMESTAMP);
ASSERT_EQ(files_arrow_schema->field(19)->type()->id(), arrow::Type::LIST);
auto write_cols_type =
std::dynamic_pointer_cast<arrow::ListType>(files_arrow_schema->field(19)->type());
ASSERT_TRUE(write_cols_type);
ASSERT_EQ(write_cols_type->value_type()->id(), arrow::Type::STRING);

Identifier snapshots_identifier("db1", "tbl1$snapshots");
::ArrowSchema system_create_schema;
ASSERT_TRUE(arrow::ExportSchema(*typed_schema, &system_create_schema).ok());
Expand Down
3 changes: 3 additions & 0 deletions src/paimon/core/table/system/in_memory_system_table.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,9 @@ class InMemorySystemTableBatchReader : public BatchReader {
emitted_ = true;
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Schema> schema, table_->ArrowSchema());
PAIMON_ASSIGN_OR_RAISE(std::vector<GenericRow> rows, table_->BuildRows());
if (rows.empty()) {
return BatchReader::MakeEofBatch();
}
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<GenericRowToArrowArrayConverter> converter,
GenericRowToArrowArrayConverter::Create(schema, arrow_pool_.get()));
return converter->NextBatch(rows);
Expand Down
Loading
Loading