Skip to content

Commit 94a2781

Browse files
suxiaogang223claude
suxiaogang223
and
claude
committed
feat: populate sys.tables file stats via manifest reading
Replace the TODO stubs in TablesSystemTable::BuildRows() with actual manifest entry aggregation. The new AggregateFileStats() helper reads the latest snapshot data files and computes record_count, file_size, file_count, and last_file_creation_time. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
1 parent 92de892 commit 94a2781

1 file changed

Lines changed: 137 additions & 16 deletions

File tree

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

Lines changed: 137 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -34,11 +34,21 @@
3434
#include "paimon/common/utils/path_util.h"
3535
#include "paimon/core/core_options.h"
3636
#include "paimon/defs.h"
37+
#include "paimon/core/io/data_file_meta.h"
38+
#include "paimon/core/manifest/file_entry.h"
39+
#include "paimon/core/manifest/file_kind.h"
40+
#include "paimon/core/manifest/manifest_entry.h"
41+
#include "paimon/core/manifest/manifest_file.h"
42+
#include "paimon/core/manifest/manifest_file_meta.h"
43+
#include "paimon/core/manifest/manifest_list.h"
3744
#include "paimon/core/schema/schema_manager.h"
3845
#include "paimon/core/schema/table_schema.h"
3946
#include "paimon/core/snapshot.h"
4047
#include "paimon/core/utils/branch_manager.h"
48+
#include "paimon/core/utils/field_mapping.h"
49+
#include "paimon/core/utils/file_store_path_factory.h"
4150
#include "paimon/core/utils/snapshot_manager.h"
51+
#include "paimon/data/timestamp.h"
4252
#include "paimon/memory/memory_pool.h"
4353
#include "paimon/status.h"
4454

@@ -87,6 +97,111 @@ VariantType StringValue(const std::string& value) {
8797
return BinaryString::FromString(value, GetDefaultPool().get());
8898
}
8999

100+
// Aggregated file-level statistics for a table or partition.
101+
struct FileStats {
102+
int64_t record_count = 0;
103+
int64_t file_size_in_bytes = 0;
104+
int64_t file_count = 0;
105+
int64_t last_file_creation_time_millis = 0;
106+
};
107+
108+
// Read the latest snapshot's data files and aggregate statistics.
109+
// Returns an empty map if no snapshot or no data files exist.
110+
Result<std::map<std::string, FileStats>> AggregateFileStats(
111+
const std::shared_ptr<FileSystem>& fs, const std::string& table_path,
112+
const std::map<std::string, std::string>& options) {
113+
std::map<std::string, FileStats> result;
114+
115+
SnapshotManager snapshot_manager(fs, table_path,
116+
BranchManager::DEFAULT_MAIN_BRANCH);
117+
PAIMON_ASSIGN_OR_RAISE(std::optional<Snapshot> snapshot,
118+
snapshot_manager.LatestSnapshot());
119+
if (!snapshot) {
120+
return result;
121+
}
122+
123+
PAIMON_ASSIGN_OR_RAISE(CoreOptions core_options,
124+
CoreOptions::FromMap(options));
125+
126+
// Use SchemaManager to load the latest schema for field/partition info
127+
SchemaManager schema_mgr(fs, table_path, BranchManager::DEFAULT_MAIN_BRANCH);
128+
auto latest_schema_result = schema_mgr.Latest();
129+
if (!latest_schema_result.ok() || !latest_schema_result.value()) {
130+
return result;
131+
}
132+
auto table_schema = *latest_schema_result.value();
133+
134+
auto pool = GetDefaultPool();
135+
136+
std::shared_ptr<arrow::Schema> arrow_schema =
137+
DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields());
138+
PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> external_paths,
139+
core_options.CreateExternalPaths());
140+
PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> global_index_external_path,
141+
core_options.CreateGlobalIndexExternalPath());
142+
PAIMON_ASSIGN_OR_RAISE(
143+
std::shared_ptr<FileStorePathFactory> path_factory,
144+
FileStorePathFactory::Create(
145+
table_path, arrow_schema, table_schema->PartitionKeys(),
146+
core_options.GetPartitionDefaultName(),
147+
core_options.GetFileFormat()->Identifier(),
148+
core_options.DataFilePrefix(),
149+
core_options.LegacyPartitionNameEnabled(), external_paths,
150+
global_index_external_path, core_options.IndexFileInDataFileDir(), pool));
151+
152+
PAIMON_ASSIGN_OR_RAISE(
153+
std::unique_ptr<ManifestList> manifest_list,
154+
ManifestList::Create(fs, core_options.GetManifestFormat(),
155+
core_options.GetManifestCompression(), path_factory, pool));
156+
157+
std::vector<ManifestFileMeta> manifests;
158+
PAIMON_RETURN_NOT_OK(
159+
manifest_list->ReadDataManifests(*snapshot, &manifests));
160+
161+
PAIMON_ASSIGN_OR_RAISE(
162+
std::shared_ptr<arrow::Schema> partition_schema,
163+
FieldMapping::GetPartitionSchema(arrow_schema, table_schema->PartitionKeys()));
164+
PAIMON_ASSIGN_OR_RAISE(
165+
std::unique_ptr<ManifestFile> manifest_file,
166+
ManifestFile::Create(fs, core_options.GetManifestFormat(),
167+
core_options.GetManifestCompression(), path_factory,
168+
core_options.GetManifestTargetFileSize(), pool,
169+
core_options, partition_schema));
170+
171+
std::vector<ManifestEntry> entries;
172+
for (const auto& manifest : manifests) {
173+
PAIMON_RETURN_NOT_OK(
174+
manifest_file->Read(manifest.FileName(), /*filter=*/nullptr, &entries));
175+
}
176+
177+
std::vector<ManifestEntry> merged_entries;
178+
PAIMON_RETURN_NOT_OK(FileEntry::MergeEntries(entries, &merged_entries));
179+
180+
for (const auto& entry : merged_entries) {
181+
if (!(entry.Kind() == FileKind::Add())) {
182+
continue;
183+
}
184+
const auto& file = entry.File();
185+
186+
// Use empty string key for unpartitioned tables
187+
std::string partition_key;
188+
if (entry.Partition().GetFieldCount() > 0) {
189+
partition_key = "partitioned";
190+
}
191+
192+
auto& stats = result[partition_key];
193+
stats.record_count += file->row_count;
194+
stats.file_size_in_bytes += file->file_size;
195+
stats.file_count++;
196+
int64_t creation_millis = file->creation_time.GetMillisecond();
197+
if (creation_millis > stats.last_file_creation_time_millis) {
198+
stats.last_file_creation_time_millis = creation_millis;
199+
}
200+
}
201+
202+
return result;
203+
}
204+
90205
} // namespace
91206

92207
// =============================================================================
@@ -278,24 +393,30 @@ Result<std::vector<GenericRow>> TablesSystemTable::BuildRows() const {
278393
? VariantType(NullType())
279394
: VariantType(StringValue(primary_keys_str)));
280395

281-
// Try to get stats from latest snapshot
396+
// Get table path and aggregate file stats from manifest entries
282397
PAIMON_ASSIGN_OR_RAISE(std::string table_path,
283398
context_.catalog->GetTableLocation(id));
284-
SnapshotManager snapshot_manager(context_.fs, table_path,
285-
BranchManager::DEFAULT_MAIN_BRANCH);
286-
auto snapshot_result = snapshot_manager.LatestSnapshot();
287-
if (snapshot_result.ok() && snapshot_result.value()) {
288-
const auto& snapshot = *snapshot_result.value();
289-
auto total_count = snapshot.TotalRecordCount();
290-
row.SetField(5, total_count ? VariantType(total_count.value())
291-
: VariantType(NullType()));
292-
// TODO(suxiaogang223): Populate file_size_in_bytes, file_count, and
293-
// last_file_creation_time by reading manifest entries. This requires
294-
// the manifest reading infrastructure from the files/manifests system
295-
// tables PR (codex/system-table-files-manifests-pr4).
296-
row.SetField(6, NullType());
297-
row.SetField(7, NullType());
298-
row.SetField(8, NullType());
399+
400+
auto file_stats_result =
401+
AggregateFileStats(context_.fs, table_path, data_schema->Options());
402+
if (file_stats_result.ok()) {
403+
auto& all_stats = file_stats_result.value();
404+
int64_t total_record = 0, total_size = 0, total_files = 0,
405+
max_creation = 0;
406+
for (const auto& [key, stats] : all_stats) {
407+
total_record += stats.record_count;
408+
total_size += stats.file_size_in_bytes;
409+
total_files += stats.file_count;
410+
if (stats.last_file_creation_time_millis > max_creation) {
411+
max_creation = stats.last_file_creation_time_millis;
412+
}
413+
}
414+
row.SetField(5, VariantType(total_record));
415+
row.SetField(6, VariantType(total_size));
416+
row.SetField(7, VariantType(total_files));
417+
row.SetField(8, max_creation > 0
418+
? VariantType(Timestamp::FromEpochMillis(max_creation))
419+
: VariantType(NullType()));
299420
} else {
300421
row.SetField(5, NullType());
301422
row.SetField(6, NullType());

0 commit comments

Comments
 (0)