Skip to content

Commit 8f57240

Browse files
suxiaogang223suxiaogang223
committed
fix: align global table metadata with Java
1 parent ce662b9 commit 8f57240

2 files changed

Lines changed: 32 additions & 39 deletions

File tree

‎src/paimon/core/table/system/global_system_tables.cpp‎

Lines changed: 8 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -421,11 +421,10 @@ Result<std::vector<GenericRow>> TablesSystemTable::BuildRows() const {
421421
LoadAllDataTables(*context_.catalog));
422422
for (const CatalogTableInfo& table : tables) {
423423
const std::shared_ptr<DataSchema>& data_schema = table.schema;
424-
Identifier id(table.database_name, table.table_name);
425424

426425
const auto& opts = data_schema->Options();
427426
auto table_type = opts.find("type");
428-
const std::string table_type_str = table_type == opts.end() ? "TABLE" : table_type->second;
427+
const std::string table_type_str = table_type == opts.end() ? "table" : table_type->second;
429428

430429
bool partitioned = !data_schema->PartitionKeys().empty();
431430

@@ -443,38 +442,13 @@ Result<std::vector<GenericRow>> TablesSystemTable::BuildRows() const {
443442
row.SetField(8, std::move(updated_at));
444443
row.SetField(9, OptionalStringValue(opts, "updatedBy"));
445444

446-
// Get table path and aggregate file stats from manifest entries. Ignore only concurrent
447-
// table deletion; corrupted metadata, unsupported aggregation, and I/O errors must fail the
448-
// query.
449-
PAIMON_ASSIGN_OR_RAISE(std::string table_path, context_.catalog->GetTableLocation(id));
450-
451-
Result<AggregatedFileStats> file_stats_result =
452-
AggregateFileStats(context_.fs, table_path, *data_schema);
453-
if (!file_stats_result.ok() && !file_stats_result.status().IsNotExist()) {
454-
return file_stats_result.status();
455-
}
456-
if (file_stats_result.ok() && file_stats_result.value().has_snapshot) {
457-
auto& all_stats = file_stats_result.value().by_partition;
458-
int64_t total_record = 0, total_size = 0, total_files = 0, max_creation = 0;
459-
for (const auto& [key, stats] : all_stats) {
460-
total_record += stats.record_count;
461-
total_size += stats.file_size_in_bytes;
462-
total_files += stats.file_count;
463-
if (stats.last_file_creation_time_millis > max_creation) {
464-
max_creation = stats.last_file_creation_time_millis;
465-
}
466-
}
467-
row.SetField(10, VariantType(total_record));
468-
row.SetField(11, VariantType(total_size));
469-
row.SetField(12, VariantType(total_files));
470-
row.SetField(13,
471-
max_creation > 0 ? VariantType(max_creation) : VariantType(NullType()));
472-
} else {
473-
row.SetField(10, NullType());
474-
row.SetField(11, NullType());
475-
row.SetField(12, NullType());
476-
row.SetField(13, NullType());
477-
}
445+
// Match Java CatalogUtils::toTableAndSnapshots when version management is unsupported.
446+
// The C++ Catalog API currently has no version-management capability, so leave snapshot
447+
// statistics null instead of deriving different live-file semantics from manifests.
448+
row.SetField(10, NullType());
449+
row.SetField(11, NullType());
450+
row.SetField(12, NullType());
451+
row.SetField(13, NullType());
478452

479453
rows.push_back(std::move(row));
480454
}

‎test/inte/read_inte_test.cpp‎

Lines changed: 24 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -3905,6 +3905,17 @@ TEST(SystemTableReadInteTest, TestReadGlobalTables) {
39053905
/*ignore_if_exists=*/false));
39063906
ArrowSchemaRelease(&schema);
39073907

3908+
ASSERT_OK_AND_ASSIGN(std::string table_path,
3909+
catalog->GetTableLocation(Identifier("test_db", "test_tbl")));
3910+
ASSERT_OK_AND_ASSIGN(auto helper, TestHelper::Create(table_path, options,
3911+
/*is_streaming_mode=*/true));
3912+
ASSERT_OK_AND_ASSIGN(
3913+
std::unique_ptr<RecordBatch> batch,
3914+
TestHelper::MakeRecordBatch(arrow::struct_(typed_schema->fields()), R"([["k", 1]])",
3915+
/*partition_map=*/{}, /*bucket=*/0, {}));
3916+
ASSERT_OK(helper->WriteAndCommit(std::move(batch), /*commit_identifier=*/0,
3917+
/*expected_commit_messages=*/std::nullopt));
3918+
39083919
std::map<std::string, std::string> no_pk_options = options;
39093920
no_pk_options[Options::BUCKET_KEY] = "pk";
39103921
::ArrowSchema no_pk_schema;
@@ -3936,6 +3947,10 @@ TEST(SystemTableReadInteTest, TestReadGlobalTables) {
39363947
auto updated_at_array = std::dynamic_pointer_cast<arrow::Int64Array>(struct_array->field(8));
39373948
auto updated_by_array = std::dynamic_pointer_cast<arrow::StringArray>(struct_array->field(9));
39383949
auto record_count_array = std::dynamic_pointer_cast<arrow::Int64Array>(struct_array->field(10));
3950+
auto file_size_array = std::dynamic_pointer_cast<arrow::Int64Array>(struct_array->field(11));
3951+
auto file_count_array = std::dynamic_pointer_cast<arrow::Int64Array>(struct_array->field(12));
3952+
auto last_creation_time_array =
3953+
std::dynamic_pointer_cast<arrow::Int64Array>(struct_array->field(13));
39393954
ASSERT_TRUE(db_array);
39403955
ASSERT_TRUE(tbl_array);
39413956
ASSERT_TRUE(type_array);
@@ -3947,14 +3962,17 @@ TEST(SystemTableReadInteTest, TestReadGlobalTables) {
39473962
ASSERT_TRUE(updated_at_array);
39483963
ASSERT_TRUE(updated_by_array);
39493964
ASSERT_TRUE(record_count_array);
3965+
ASSERT_TRUE(file_size_array);
3966+
ASSERT_TRUE(file_count_array);
3967+
ASSERT_TRUE(last_creation_time_array);
39503968

39513969
// Find our table by table name
39523970
bool found = false;
39533971
bool found_no_pk = false;
39543972
for (int64_t i = 0; i < struct_array->length(); ++i) {
39553973
if (std::string(tbl_array->GetString(i)) == "test_tbl") {
39563974
EXPECT_EQ(std::string(db_array->GetString(i)), "test_db");
3957-
EXPECT_EQ(std::string(type_array->GetString(i)), "TABLE");
3975+
EXPECT_EQ(std::string(type_array->GetString(i)), "table");
39583976
EXPECT_FALSE(part_array->Value(i));
39593977
EXPECT_TRUE(pk_array->Value(i));
39603978
EXPECT_EQ(owner_array->GetString(i), "alice");
@@ -3963,6 +3981,9 @@ TEST(SystemTableReadInteTest, TestReadGlobalTables) {
39633981
EXPECT_EQ(updated_at_array->Value(i), 2000);
39643982
EXPECT_EQ(updated_by_array->GetString(i), "updater");
39653983
EXPECT_TRUE(record_count_array->IsNull(i));
3984+
EXPECT_TRUE(file_size_array->IsNull(i));
3985+
EXPECT_TRUE(file_count_array->IsNull(i));
3986+
EXPECT_TRUE(last_creation_time_array->IsNull(i));
39663987
found = true;
39673988
} else if (std::string(tbl_array->GetString(i)) == "test_no_pk_tbl") {
39683989
EXPECT_FALSE(pk_array->Value(i));
@@ -4060,7 +4081,7 @@ TEST(SystemTableReadInteTest, TestGlobalSystemTablesPropagateCorruptSchema) {
40604081
}
40614082
}
40624083

4063-
TEST(SystemTableReadInteTest, TestGlobalSystemTablesPropagateCorruptSnapshot) {
4084+
TEST(SystemTableReadInteTest, TestPartitionsSystemTablePropagatesCorruptSnapshot) {
40644085
std::map<std::string, std::string> options = {{Options::FILE_SYSTEM, "local"},
40654086
{Options::FILE_FORMAT, "orc"},
40664087
{Options::MANIFEST_FORMAT, "orc"},
@@ -4092,11 +4113,10 @@ TEST(SystemTableReadInteTest, TestGlobalSystemTablesPropagateCorruptSnapshot) {
40924113
ASSERT_OK(fs->WriteFile(PathUtil::JoinPath(table_path, "snapshot/snapshot-1"), "{invalid-json",
40934114
/*overwrite=*/true));
40944115

4095-
ASSERT_NOK(ReadGlobalSystemTable("tables", catalog.get(), fs, warehouse, options));
40964116
ASSERT_NOK(ReadGlobalSystemTable("partitions", catalog.get(), fs, warehouse, options));
40974117
}
40984118

4099-
TEST(SystemTableReadInteTest, TestGlobalSystemTablesPropagateCorruptManifest) {
4119+
TEST(SystemTableReadInteTest, TestPartitionsSystemTablePropagatesCorruptManifest) {
41004120
std::map<std::string, std::string> options = {{Options::FILE_SYSTEM, "local"},
41014121
{Options::FILE_FORMAT, "orc"},
41024122
{Options::MANIFEST_FORMAT, "orc"},
@@ -4132,7 +4152,6 @@ TEST(SystemTableReadInteTest, TestGlobalSystemTablesPropagateCorruptManifest) {
41324152
PathUtil::JoinPath(table_path, "manifest/" + snapshot->BaseManifestList()), "corrupt",
41334153
/*overwrite=*/true));
41344154

4135-
ASSERT_NOK(ReadGlobalSystemTable("tables", catalog.get(), fs, warehouse, options));
41364155
ASSERT_NOK(ReadGlobalSystemTable("partitions", catalog.get(), fs, warehouse, options));
41374156
}
41384157

0 commit comments

Comments
 (0)