wgtmac commented on code in PR #589:
URL: https://github.com/apache/iceberg-cpp/pull/589#discussion_r3101340090
##########
src/iceberg/table_scan.h:
##########
@@ -148,8 +151,12 @@ class ICEBERG_EXPORT TableScanBuilder : public
ErrorCollector {
/// \brief Constructs a TableScanBuilder for the given table.
/// \param metadata Current table metadata.
/// \param io FileIO instance for reading manifests files.
+ /// \param reporter Optional metrics reporter for scan metrics.
+ /// \param table_name Optional table name for metrics reporting.
static Result<std::unique_ptr<TableScanBuilder<ScanType>>> Make(
Review Comment:
The builder stores a reporter internally, but there is no public scan-level
API to add an extra reporter the way Java's `Scan.metricsReporter(...)` does.
That means C++ can only use the reporter injected from the table/catalog path,
while Java supports composing an additional reporter for a particular scan. For
observability integrations and tests, that is a meaningful flexibility gap.
##########
src/iceberg/metrics/metrics_reporters.cc:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/metrics/metrics_reporters.h"
+
+#include <iostream>
+#include <unordered_set>
+
+#include "iceberg/util/string_util.h"
+
+namespace iceberg {
+
+namespace {
+
+/// \brief Registry type for MetricsReporter factories.
+using MetricsReporterRegistry = std::unordered_map<std::string,
MetricsReporterFactory>;
+
+/// \brief Get the set of known built-in metrics reporter types.
+const std::unordered_set<std::string>& DefaultReporterTypes() {
+ static const std::unordered_set<std::string> kReporterTypes = {
+ std::string(kMetricsReporterTypeNoop),
+ };
+ return kReporterTypes;
+}
+
+/// \brief Infer the reporter type from properties.
+std::string InferReporterType(
+ const std::unordered_map<std::string, std::string>& properties) {
+ auto it = properties.find(std::string(kMetricsReporterImpl));
Review Comment:
This loader treats `metrics-reporter-impl` as a lowercase registry key and
defaults to `noop`, while Java and the Iceberg docs treat the same property as
a fully qualified class name whose instance is initialized with catalog
properties. On top of that, the catalog constructors ignore `Load()` errors and
`Table` falls back to `noop`, so a Java-style config string or typo quietly
disables metrics instead of surfacing a configuration error. That is a
public-contract drift, not just an implementation detail.
##########
src/iceberg/metrics/commit_report.h:
##########
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <string>
+#include <unordered_map>
+
+#include "iceberg/constants.h"
+#include "iceberg/iceberg_export.h"
+
+namespace iceberg {
+
+/// \brief Metrics collected during a table commit (snapshot creation).
Review Comment:
`CommitReport` only carries summary-derived counters, but Java's commit
reporting model also has live commit metrics (`totalDuration`, `attempts`)
collected through `CommitMetrics` and serialized through `CommitMetricsResult`.
Because those fields are absent from the C++ public model entirely, even a
future implementation cannot fully mirror Java's commit report without another
API change.
##########
src/iceberg/table_scan.cc:
##########
@@ -533,7 +542,75 @@ Result<std::vector<std::shared_ptr<FileScanTask>>>
DataTableScan::PlanFiles() co
if (context_.ignore_residuals) {
manifest_group->IgnoreResiduals();
}
- return manifest_group->PlanFiles();
+
+ ICEBERG_ASSIGN_OR_RAISE(auto tasks, manifest_group->PlanFiles());
+
+ // Report scan metrics if a reporter is configured
+ if (context_.reporter) {
+ auto scan_end = std::chrono::steady_clock::now();
+ auto duration = std::chrono::duration_cast<DurationNs>(scan_end -
scan_start);
+
+ ScanReport report;
+ report.table_name = context_.table_name;
+ report.snapshot_id = snapshot->snapshot_id;
+ report.schema_id = schema_ ? schema_->schema_id() : kInvalidSchemaId;
+ if (context_.filter) {
+ report.filter = context_.filter;
+ }
+ report.scan_metrics.total_planning_duration = duration;
+
+ // Manifest counts from ManifestGroup counters
+ const auto& counters = manifest_group->scan_counters();
+ report.scan_metrics.total_data_manifests =
+ static_cast<int64_t>(data_manifests.size());
+ report.scan_metrics.total_delete_manifests =
+ static_cast<int64_t>(delete_manifests.size());
+ report.scan_metrics.scanned_data_manifests =
counters.scanned_data_manifests;
+ report.scan_metrics.skipped_data_manifests =
counters.skipped_data_manifests;
+ report.scan_metrics.scanned_delete_manifests =
counters.scanned_delete_manifests;
+ report.scan_metrics.skipped_delete_manifests =
counters.skipped_delete_manifests;
+ report.scan_metrics.skipped_data_files = counters.skipped_data_files;
+ report.scan_metrics.skipped_delete_files = counters.skipped_delete_files;
+
+ // Result counts and file sizes from tasks
+ report.scan_metrics.result_data_files = static_cast<int64_t>(tasks.size());
+ for (const auto& task : tasks) {
+ report.scan_metrics.total_file_size_in_bytes +=
+ task->data_file()->file_size_in_bytes;
+ for (const auto& del_file : task->delete_files()) {
Review Comment:
The scan report counts equality and positional delete files, but it never
increments `indexed_delete_files` and it never separates deletion vectors from
ordinary positional deletes. Java's `ScanMetricsUtil.indexedDeleteFile(...)`
increments all three counters (`indexedDeleteFiles`,
`positionalDeleteFiles`/`dvs`, `equalityDeleteFiles`). Because the C++ report
schema already exposes `indexed_delete_files` and `dvs`, leaving them unset
produces parity drift and misleading metrics for DV-heavy tables.
##########
src/iceberg/catalog/rest/rest_catalog.cc:
##########
@@ -168,10 +169,17 @@ Result<std::shared_ptr<RestCatalog>> RestCatalog::Make(
ICEBERG_ASSIGN_OR_RAISE(auto catalog_session,
auth_manager->CatalogSession(*client,
final_config.configs()));
+ // Load metrics reporter from catalog properties
+ std::shared_ptr<MetricsReporter> reporter;
+ auto reporter_result = MetricsReporters::Load(final_config.configs());
Review Comment:
The REST catalog now loads a local reporter from catalog properties, but it
never constructs the built-in REST reporter for the `/metrics` endpoint and
never combines the two. As a result, scan/commit reports from tables loaded
through `RestCatalog` are never POSTed back to the server even when the server
advertises `ReportMetrics` support. Java's `RESTSessionCatalog` explicitly
combines the catalog reporter with `RESTMetricsReporter`, so this leaves the
feature only partially implemented on the C++ side.
##########
src/iceberg/metrics/scan_report.h:
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+#include <chrono>
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+#include "iceberg/constants.h"
+#include "iceberg/expression/expression.h"
+#include "iceberg/iceberg_export.h"
+
+namespace iceberg {
+
+/// \brief Duration type for scan metrics reporting (nanosecond precision).
+using DurationNs = std::chrono::nanoseconds;
+
+/// \brief Metrics collected during the planning and execution of a table scan.
+///
+/// Embedded in ScanReport and populated by DataTableScan after PlanFiles()
+/// completes. Mirrors the fields in Java's ScanMetricsResult.
+struct ICEBERG_EXPORT ScanMetrics {
Review Comment:
`ScanReport` embeds a plain `ScanMetrics` struct of raw integers and
`std::chrono::nanoseconds`, but Java's contract is layered: live
counters/timers are collected in `ScanMetrics`, converted into
`ScanMetricsResult`, and then wrapped by `ScanReport`. That layering preserves
units, optional presence, noop semantics, and serialization behavior. In C++,
every metric defaults to `0`, so consumers cannot distinguish "not collected"
from a real zero value, and the report shape is no longer compatible with
Java's richer contract.
##########
src/iceberg/transaction.cc:
##########
@@ -374,7 +375,7 @@ Result<std::shared_ptr<Table>> Transaction::Commit() {
// Mark as committed and update table reference
committed_ = true;
ctx_->table = std::move(commit_result.value());
-
+ ReportCommitMetrics();
Review Comment:
`Transaction::Commit()` always calls `ReportCommitMetrics()` after a
successful metadata commit, even for metadata-only transactions. That helper
then reads `current_snapshot()` from the post-commit table and reports it as if
this transaction created that snapshot, which can re-emit stale snapshot
metrics and a misleading operation name. Java only reports commit metrics from
snapshot-producing commits (`CreateSnapshotEvent` in `SnapshotProducer`), so
the reporting hook should move closer to snapshot-producing updates or
otherwise prove that this transaction actually created a new snapshot before
emitting a `CommitReport`.
##########
src/iceberg/transaction.cc:
##########
@@ -463,4 +464,70 @@ Result<std::shared_ptr<SnapshotManager>>
Transaction::NewSnapshotManager() {
return SnapshotManager::Make(shared_from_this());
}
+void Transaction::ReportCommitMetrics() const {
+ const auto& reporter = ctx_->table->reporter();
+ if (!reporter) return;
+
+ auto snapshot_result = ctx_->table->current_snapshot();
+ if (!snapshot_result.has_value() || !snapshot_result.value()) return;
+
+ const auto& snapshot = snapshot_result.value();
+ const auto& summary = snapshot->summary;
+
+ auto parse_int64 = [&summary](const std::string& key) -> int64_t {
+ auto it = summary.find(key);
+ if (it != summary.end()) {
+ auto res = StringUtils::ParseNumber<int64_t>(it->second);
+ return res.has_value() ? res.value() : 0;
+ }
+ return 0;
+ };
+
+ CommitReport report;
+ report.table_name = ctx_->table->name().ToString();
+ report.snapshot_id = snapshot->snapshot_id;
+ report.sequence_number = snapshot->sequence_number;
+
+ // Operation from summary
+ if (auto it = summary.find(SnapshotSummaryFields::kOperation); it !=
summary.end()) {
+ report.operation = it->second;
+ }
+ CommitMetrics& metric = report.commit_metrics;
+ metric.added_data_files =
parse_int64(SnapshotSummaryFields::kAddedDataFiles);
+ metric.removed_data_files =
parse_int64(SnapshotSummaryFields::kDeletedDataFiles);
+ metric.total_data_files =
parse_int64(SnapshotSummaryFields::kTotalDataFiles);
+ metric.added_delete_files =
parse_int64(SnapshotSummaryFields::kAddedDeleteFiles);
+ metric.removed_delete_files =
parse_int64(SnapshotSummaryFields::kRemovedDeleteFiles);
+ metric.total_delete_files =
parse_int64(SnapshotSummaryFields::kTotalDeleteFiles);
+ metric.added_records = parse_int64(SnapshotSummaryFields::kAddedRecords);
+ metric.removed_records = parse_int64(SnapshotSummaryFields::kDeletedRecords);
+ metric.added_files_size_bytes =
parse_int64(SnapshotSummaryFields::kAddedFileSize);
+ metric.removed_files_size_bytes =
parse_int64(SnapshotSummaryFields::kRemovedFileSize);
+ metric.total_records = parse_int64(SnapshotSummaryFields::kTotalRecords);
+ metric.total_files_size_bytes =
parse_int64(SnapshotSummaryFields::kTotalFileSize);
+ metric.added_equality_delete_files =
+ parse_int64(SnapshotSummaryFields::kAddedEqDeleteFiles);
+ metric.removed_equality_delete_files =
+ parse_int64(SnapshotSummaryFields::kRemovedEqDeleteFiles);
+ metric.added_positional_delete_files =
+ parse_int64(SnapshotSummaryFields::kAddedPosDeleteFiles);
+ metric.removed_positional_delete_files =
+ parse_int64(SnapshotSummaryFields::kRemovedPosDeleteFiles);
+ metric.added_positional_deletes =
parse_int64(SnapshotSummaryFields::kAddedPosDeletes);
+ metric.removed_positional_deletes =
+ parse_int64(SnapshotSummaryFields::kRemovedPosDeletes);
+ metric.total_positional_deletes =
parse_int64(SnapshotSummaryFields::kTotalPosDeletes);
+ metric.added_equality_deletes =
parse_int64(SnapshotSummaryFields::kAddedEqDeletes);
+ metric.removed_equality_deletes =
parse_int64(SnapshotSummaryFields::kRemovedEqDeletes);
+ metric.total_equality_deletes =
parse_int64(SnapshotSummaryFields::kTotalEqDeletes);
+ metric.added_dvs = parse_int64(SnapshotSummaryFields::kAddedDVs);
+ metric.removed_dvs = parse_int64(SnapshotSummaryFields::kRemovedDVs);
+ metric.created_manifest_count =
parse_int64(SnapshotSummaryFields::kManifestsCreated);
Review Comment:
`CommitMetrics` exposes `replaced_manifest_count`, but
`ReportCommitMetrics()` never reads `SnapshotSummaryFields::kManifestsReplaced`
into it. Java's `CommitMetricsResult.from(...)` does populate the corresponding
`manifestsReplaced` field from the snapshot summary. Any consumer relying on
this report will currently observe `0` even when the commit actually replaced
manifests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
