Skip to content

Commit aab656b

Browse files
SteNicholasSteNicholas
committed
feat: support commit metrics of FileStoreCommitImpl to align with CommitMetrics
1 parent 95940ad commit aab656b

4 files changed

Lines changed: 178 additions & 6 deletions

File tree

‎src/paimon/core/operation/file_store_commit_impl.cpp‎

Lines changed: 138 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -262,9 +262,15 @@ Status FileStoreCommitImpl::Overwrite(
262262
std::shared_ptr<ManifestCommittable> committable =
263263
CreateManifestCommittable(identifier, commit_messages, watermark);
264264
std::vector<ManifestEntry> append_table_files;
265+
std::vector<ManifestEntry> append_changelog_files;
266+
std::vector<ManifestEntry> compact_table_files;
267+
std::vector<ManifestEntry> compact_changelog_files;
265268
std::vector<IndexManifestEntry> append_table_index_files;
269+
std::vector<IndexManifestEntry> compact_table_index_files;
266270
PAIMON_RETURN_NOT_OK(CollectChanges(committable->FileCommittables(), &append_table_files,
267-
&append_table_index_files));
271+
&append_changelog_files, &compact_table_files,
272+
&compact_changelog_files, &append_table_index_files,
273+
&compact_table_index_files));
268274
if (!append_table_index_files.empty()) {
269275
return Status::NotImplemented("Overwrite not support index for now");
270276
}
@@ -283,9 +289,15 @@ Result<int32_t> FileStoreCommitImpl::FilterAndOverwrite(
283289
FilterCommitted(committables));
284290
if (!actual_committables.empty()) {
285291
std::vector<ManifestEntry> append_table_files;
292+
std::vector<ManifestEntry> append_changelog_files;
293+
std::vector<ManifestEntry> compact_table_files;
294+
std::vector<ManifestEntry> compact_changelog_files;
286295
std::vector<IndexManifestEntry> append_table_index_files;
296+
std::vector<IndexManifestEntry> compact_table_index_files;
287297
PAIMON_RETURN_NOT_OK(CollectChanges(actual_committables[0]->FileCommittables(),
288-
&append_table_files, &append_table_index_files));
298+
&append_table_files, &append_changelog_files,
299+
&compact_table_files, &compact_changelog_files,
300+
&append_table_index_files, &compact_table_index_files));
289301
if (!append_table_index_files.empty()) {
290302
return Status::NotImplemented("FilterAndOverwrite not support index for now");
291303
}
@@ -355,20 +367,72 @@ Status FileStoreCommitImpl::TryOverwrite(
355367
Status FileStoreCommitImpl::Commit(const std::shared_ptr<ManifestCommittable>& committable,
356368
bool check_append_files) {
357369
std::vector<ManifestEntry> append_table_files;
370+
std::vector<ManifestEntry> append_changelog_files;
371+
std::vector<ManifestEntry> compact_table_files;
372+
std::vector<ManifestEntry> compact_changelog_files;
358373
std::vector<IndexManifestEntry> append_table_index_files;
374+
std::vector<IndexManifestEntry> compact_table_index_files;
359375
PAIMON_RETURN_NOT_OK(CollectChanges(committable->FileCommittables(), &append_table_files,
360-
&append_table_index_files));
376+
&append_changelog_files, &compact_table_files,
377+
&compact_changelog_files, &append_table_index_files,
378+
&compact_table_index_files));
361379

362380
int32_t attempt = 0;
381+
int32_t generated_snapshot = 0;
382+
const auto started = std::chrono::high_resolution_clock::now();
363383
if (!ignore_empty_commit_ || !append_table_files.empty() || !append_table_index_files.empty()) {
364384
PAIMON_ASSIGN_OR_RAISE(int32_t cnt,
365385
TryCommit(append_table_files, append_table_index_files,
366386
committable->Identifier(), committable->Watermark(),
367387
committable->LogOffsets(), committable->Properties(),
368388
Snapshot::CommitKind::Append(), check_append_files));
369389
attempt += cnt;
390+
++generated_snapshot;
391+
}
392+
auto table_files_added = static_cast<int32_t>(append_table_files.size());
393+
int32_t table_files_deleted = 0;
394+
int64_t compaction_input_file_size = 0;
395+
int64_t compaction_output_file_size = 0;
396+
for (const auto& entry : compact_table_files) {
397+
const auto& kind = entry.Kind();
398+
if (kind == FileKind::Add()) {
399+
++table_files_added;
400+
compaction_output_file_size += entry.File()->file_size;
401+
} else if (kind == FileKind::Delete()) {
402+
++table_files_deleted;
403+
compaction_input_file_size += entry.File()->file_size;
404+
}
370405
}
406+
metrics_->SetCounter(CommitMetrics::LAST_COMMIT_DURATION,
407+
std::chrono::duration_cast<std::chrono::nanoseconds>(
408+
std::chrono::high_resolution_clock::now() - started)
409+
.count());
371410
metrics_->SetCounter(CommitMetrics::LAST_COMMIT_ATTEMPTS, attempt);
411+
metrics_->SetCounter(CommitMetrics::LAST_TABLE_FILES_ADDED, table_files_added);
412+
metrics_->SetCounter(CommitMetrics::LAST_TABLE_FILES_DELETED, table_files_deleted);
413+
metrics_->SetCounter(CommitMetrics::LAST_TABLE_FILES_APPENDED, append_table_files.size());
414+
metrics_->SetCounter(CommitMetrics::LAST_TABLE_FILES_COMMIT_COMPACTED,
415+
compact_table_files.size());
416+
metrics_->SetCounter(CommitMetrics::LAST_CHANGELOG_FILES_APPENDED,
417+
append_changelog_files.size());
418+
metrics_->SetCounter(CommitMetrics::LAST_CHANGELOG_FILES_COMMIT_COMPACTED,
419+
compact_changelog_files.size());
420+
metrics_->SetCounter(CommitMetrics::LAST_GENERATED_SNAPSHOTS, generated_snapshot);
421+
metrics_->SetCounter(CommitMetrics::LAST_DELTA_RECORDS_APPENDED, RowCounts(append_table_files));
422+
metrics_->SetCounter(CommitMetrics::LAST_CHANGELOG_RECORDS_APPENDED,
423+
RowCounts(append_changelog_files));
424+
metrics_->SetCounter(CommitMetrics::LAST_DELTA_RECORDS_COMMIT_COMPACTED,
425+
RowCounts(compact_table_files));
426+
metrics_->SetCounter(CommitMetrics::LAST_CHANGELOG_RECORDS_COMMIT_COMPACTED,
427+
RowCounts(compact_changelog_files));
428+
metrics_->SetCounter(CommitMetrics::LAST_PARTITIONS_WRITTEN,
429+
NumChangedPartitions({append_table_files, compact_changelog_files}));
430+
metrics_->SetCounter(CommitMetrics::LAST_BUCKETS_WRITTEN,
431+
NumChangedBuckets({append_table_files, compact_changelog_files}));
432+
metrics_->SetCounter(CommitMetrics::LAST_COMPACTION_INPUT_FILE_SIZE,
433+
compaction_input_file_size);
434+
metrics_->SetCounter(CommitMetrics::LAST_COMPACTION_OUTPUT_FILE_SIZE,
435+
compaction_output_file_size);
372436
return Status::OK();
373437
}
374438

@@ -784,7 +848,11 @@ std::shared_ptr<ManifestCommittable> FileStoreCommitImpl::CreateManifestCommitta
784848
Status FileStoreCommitImpl::CollectChanges(
785849
const std::vector<std::shared_ptr<CommitMessage>>& commit_messages,
786850
std::vector<ManifestEntry>* append_table_files,
787-
std::vector<IndexManifestEntry>* append_table_index_files) {
851+
std::vector<ManifestEntry>* append_changelog_files,
852+
std::vector<ManifestEntry>* compact_table_files,
853+
std::vector<ManifestEntry>* compact_changelog_files,
854+
std::vector<IndexManifestEntry>* append_table_index_files,
855+
std::vector<IndexManifestEntry>* compact_table_index_files) {
788856
for (const auto& message : commit_messages) {
789857
auto commit_message = std::dynamic_pointer_cast<CommitMessageImpl>(message);
790858
if (commit_message) {
@@ -797,6 +865,11 @@ Status FileStoreCommitImpl::CollectChanges(
797865
append_table_files->push_back(
798866
MakeEntry(FileKind::Delete(), commit_message, deleted_file));
799867
}
868+
for (const std::shared_ptr<DataFileMeta>& changelog_file :
869+
new_files_increment.ChangelogFiles()) {
870+
append_changelog_files->push_back(
871+
MakeEntry(FileKind::Add(), commit_message, changelog_file));
872+
}
800873
for (const std::shared_ptr<IndexFileMeta>& deleted_index_file :
801874
new_files_increment.DeletedIndexFiles()) {
802875
append_table_index_files->emplace_back(
@@ -808,6 +881,34 @@ Status FileStoreCommitImpl::CollectChanges(
808881
append_table_index_files->emplace_back(FileKind::Add(), commit_message->Partition(),
809882
commit_message->Bucket(), new_index_file);
810883
}
884+
CompactIncrement compact_increment = commit_message->GetCompactIncrement();
885+
for (const std::shared_ptr<DataFileMeta>& compact_before :
886+
compact_increment.CompactBefore()) {
887+
compact_table_files->push_back(
888+
MakeEntry(FileKind::Delete(), commit_message, compact_before));
889+
}
890+
for (const std::shared_ptr<DataFileMeta>& compact_after :
891+
compact_increment.CompactAfter()) {
892+
compact_table_files->push_back(
893+
MakeEntry(FileKind::Add(), commit_message, compact_after));
894+
}
895+
for (const std::shared_ptr<DataFileMeta>& changelog_file :
896+
compact_increment.ChangelogFiles()) {
897+
compact_changelog_files->push_back(
898+
MakeEntry(FileKind::Add(), commit_message, changelog_file));
899+
}
900+
for (const std::shared_ptr<IndexFileMeta>& deleted_index_file :
901+
compact_increment.DeletedIndexFiles()) {
902+
compact_table_index_files->emplace_back(
903+
FileKind::Delete(), commit_message->Partition(), commit_message->Bucket(),
904+
deleted_index_file);
905+
}
906+
for (const std::shared_ptr<IndexFileMeta>& new_index_file :
907+
compact_increment.NewIndexFiles()) {
908+
compact_table_index_files->emplace_back(FileKind::Add(),
909+
commit_message->Partition(),
910+
commit_message->Bucket(), new_index_file);
911+
}
811912
} else {
812913
return Status::Invalid("fail to cast commit message to commit message impl");
813914
}
@@ -825,4 +926,37 @@ ManifestEntry FileStoreCommitImpl::MakeEntry(
825926
file);
826927
}
827928

929+
int64_t FileStoreCommitImpl::RowCounts(const std::vector<ManifestEntry>& files) {
930+
return std::accumulate(files.begin(), files.end(), 0L,
931+
[](int64_t row_count, const ManifestEntry& entry) {
932+
return row_count + entry.File()->row_count;
933+
});
934+
}
935+
936+
int64_t FileStoreCommitImpl::NumChangedPartitions(
937+
const std::vector<std::vector<ManifestEntry>>& changes) {
938+
std::unordered_set<BinaryRow> changed_partitions;
939+
for (const auto& change : changes) {
940+
for (const auto& entry : change) {
941+
changed_partitions.insert(entry.Partition());
942+
}
943+
}
944+
return static_cast<int64_t>(changed_partitions.size());
945+
}
946+
947+
int64_t FileStoreCommitImpl::NumChangedBuckets(
948+
const std::vector<std::vector<ManifestEntry>>& changes) {
949+
std::unordered_map<BinaryRow, std::unordered_set<int>> changed_partition_buckets;
950+
for (const auto& change : changes) {
951+
for (const auto& entry : change) {
952+
changed_partition_buckets[entry.Partition()].insert(entry.Bucket());
953+
}
954+
}
955+
return std::accumulate(changed_partition_buckets.begin(), changed_partition_buckets.end(),
956+
int64_t{0}, [](int64_t num_changed_buckets, const auto& bucket) {
957+
return num_changed_buckets +
958+
static_cast<int64_t>(bucket.second.size());
959+
});
960+
}
961+
828962
} // namespace paimon

‎src/paimon/core/operation/file_store_commit_impl.h‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -145,7 +145,11 @@ class FileStoreCommitImpl : public FileStoreCommit {
145145

146146
Status CollectChanges(const std::vector<std::shared_ptr<CommitMessage>>& commit_messages,
147147
std::vector<ManifestEntry>* append_table_files,
148-
std::vector<IndexManifestEntry>* append_table_index_files);
148+
std::vector<ManifestEntry>* append_changelog_files,
149+
std::vector<ManifestEntry>* compact_table_files,
150+
std::vector<ManifestEntry>* compact_changelog_files,
151+
std::vector<IndexManifestEntry>* append_table_index_files,
152+
std::vector<IndexManifestEntry>* compact_table_index_files);
149153

150154
Result<int32_t> TryCommit(const std::vector<ManifestEntry>& delta_files,
151155
const std::vector<IndexManifestEntry>& index_entries,
@@ -192,6 +196,12 @@ class FileStoreCommitImpl : public FileStoreCommit {
192196
const std::vector<ManifestEntry>& data_files,
193197
const std::vector<IndexManifestEntry>& index_files) const;
194198

199+
int64_t RowCounts(const std::vector<ManifestEntry>& files);
200+
201+
int64_t NumChangedPartitions(const std::vector<std::vector<ManifestEntry>>& changes);
202+
203+
int64_t NumChangedBuckets(const std::vector<std::vector<ManifestEntry>>& changes);
204+
195205
private:
196206
std::shared_ptr<MemoryPool> memory_pool_;
197207
std::shared_ptr<Executor> executor_;

‎src/paimon/core/operation/file_store_commit_impl_test.cpp‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1280,10 +1280,20 @@ TEST_F(FileStoreCommitImplTest, TestCollectChanges) {
12801280
auto commit_impl = std::dynamic_pointer_cast<FileStoreCommitImpl>(
12811281
std::shared_ptr<FileStoreCommit>(std::move(commit)));
12821282
std::vector<ManifestEntry> append_table_files;
1283+
std::vector<ManifestEntry> append_changelog_files;
1284+
std::vector<ManifestEntry> compact_table_files;
1285+
std::vector<ManifestEntry> compact_changelog_files;
12831286
std::vector<IndexManifestEntry> append_table_index_files;
1284-
ASSERT_OK(commit_impl->CollectChanges(msgs, &append_table_files, &append_table_index_files));
1287+
std::vector<IndexManifestEntry> compact_table_index_files;
1288+
ASSERT_OK(commit_impl->CollectChanges(msgs, &append_table_files, &append_changelog_files,
1289+
&compact_table_files, &compact_changelog_files,
1290+
&append_table_index_files, &compact_table_index_files));
12851291
ASSERT_EQ(append_table_files.size(), 3u);
1292+
ASSERT_EQ(append_changelog_files.size(), 0u);
1293+
ASSERT_EQ(compact_table_files.size(), 0u);
1294+
ASSERT_EQ(compact_changelog_files.size(), 0u);
12861295
ASSERT_EQ(append_table_index_files.size(), 0u);
1296+
ASSERT_EQ(compact_table_index_files.size(), 0u);
12871297
ASSERT_EQ(append_table_files[0].Kind(), FileKind::Add());
12881298
ASSERT_EQ(append_table_files[0].Bucket(), 0);
12891299
ASSERT_EQ(append_table_files[0].TotalBuckets(), 10);

‎src/paimon/core/operation/metrics/commit_metrics.h‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,25 @@ namespace paimon {
2121
/// Metrics to measure a commit.
2222
class CommitMetrics {
2323
public:
24+
static constexpr char LAST_COMMIT_DURATION[] = "lastCommitDuration";
2425
static constexpr char LAST_COMMIT_ATTEMPTS[] = "lastCommitAttempts";
26+
static constexpr char LAST_TABLE_FILES_ADDED[] = "lastTableFilesAdded";
27+
static constexpr char LAST_TABLE_FILES_DELETED[] = "lastTableFilesDeleted";
28+
static constexpr char LAST_TABLE_FILES_APPENDED[] = "lastTableFilesAppended";
29+
static constexpr char LAST_TABLE_FILES_COMMIT_COMPACTED[] = "lastTableFilesCommitCompacted";
30+
static constexpr char LAST_CHANGELOG_FILES_APPENDED[] = "lastChangelogFilesAppended";
31+
static constexpr char LAST_CHANGELOG_FILES_COMMIT_COMPACTED[] =
32+
"lastChangelogFileCommitCompacted";
33+
static constexpr char LAST_GENERATED_SNAPSHOTS[] = "lastGeneratedSnapshots";
34+
static constexpr char LAST_DELTA_RECORDS_APPENDED[] = "lastDeltaRecordsAppended";
35+
static constexpr char LAST_CHANGELOG_RECORDS_APPENDED[] = "lastChangelogRecordsAppended";
36+
static constexpr char LAST_DELTA_RECORDS_COMMIT_COMPACTED[] = "lastDeltaRecordsCommitCompacted";
37+
static constexpr char LAST_CHANGELOG_RECORDS_COMMIT_COMPACTED[] =
38+
"lastChangelogRecordsCommitCompacted";
39+
static constexpr char LAST_PARTITIONS_WRITTEN[] = "lastPartitionsWritten";
40+
static constexpr char LAST_BUCKETS_WRITTEN[] = "lastBucketsWritten";
41+
static constexpr char LAST_COMPACTION_INPUT_FILE_SIZE[] = "lastCompactionInputFileSize";
42+
static constexpr char LAST_COMPACTION_OUTPUT_FILE_SIZE[] = "lastCompactionOutputFileSize";
2543
};
2644

2745
} // namespace paimon

0 commit comments

Comments
 (0)