wgtmac commented on code in PR #701:
URL: https://github.com/apache/iceberg-cpp/pull/701#discussion_r3537717970
##########
src/iceberg/delete_file_index.cc:
##########
@@ -638,8 +644,10 @@ Result<std::vector<ManifestEntry>>
DeleteFileIndex::Builder::LoadDeleteFiles() {
ICEBERG_ASSIGN_OR_RAISE(auto should_match,
manifest_evaluator->Evaluate(manifest));
if (!should_match) {
+ if (scan_metrics_)
scan_metrics_->skipped_delete_manifests->Increment(1);
Review Comment:
This only counts delete manifests skipped by the partition evaluator. A
delete manifest with no added or existing files returns earlier and is not
counted as `skipped_delete_manifests`, even though Java counts that
manifest-level skip.
##########
src/iceberg/table.cc:
##########
@@ -219,7 +255,11 @@ Result<std::shared_ptr<UpdateLocation>>
Table::NewUpdateLocation() {
Result<std::shared_ptr<FastAppend>> Table::NewFastAppend() {
ICEBERG_ASSIGN_OR_RAISE(
auto ctx, TransactionContext::Make(shared_from_this(),
TransactionKind::kUpdate));
- return FastAppend::Make(name().name, std::move(ctx));
+ ICEBERG_ASSIGN_OR_RAISE(auto op, FastAppend::Make(name().name,
std::move(ctx)));
+ if (reporter_) {
+ op->ReportWith(reporter_);
Review Comment:
This only propagates the table reporter to fast append. Other
snapshot-producing operations such as merge append, row delta, overwrite,
delete files, and rewrite files will not emit commit reports unless the caller
sets a reporter on each operation. Java table operations inherit the table
reporter consistently.
##########
src/iceberg/delete_file_index.cc:
##########
@@ -638,8 +644,10 @@ Result<std::vector<ManifestEntry>>
DeleteFileIndex::Builder::LoadDeleteFiles() {
ICEBERG_ASSIGN_OR_RAISE(auto should_match,
manifest_evaluator->Evaluate(manifest));
if (!should_match) {
+ if (scan_metrics_)
scan_metrics_->skipped_delete_manifests->Increment(1);
return manifest_result;
}
+ if (scan_metrics_)
scan_metrics_->scanned_delete_manifests->Increment(1);
Review Comment:
The delete manifest entry filters below are still not wired to a
`SkipCounter`, so delete files filtered by partition stats or the partition set
are not counted in `skipped_delete_files`. Java wraps those entry-level filters
with the skipped-delete-files counter.
##########
src/iceberg/transaction.cc:
##########
@@ -275,6 +277,10 @@ Status Transaction::ApplyUpdateSnapshot(SnapshotUpdate&
update) {
ICEBERG_ASSIGN_OR_RAISE(auto result, update.Apply());
+ if (const auto& override_reporter = update.reporter()) {
Review Comment:
This stores the per-update reporter on the transaction and never clears it.
If one snapshot update calls `ReportWith(custom)`, later snapshot updates in
the same transaction inherit that reporter even when they did not opt in. The
override should be scoped to the update that produced the snapshot.
##########
src/iceberg/test/rest_metrics_reporter_test.cc:
##########
@@ -0,0 +1,74 @@
+/*
+ * 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/catalog/rest/rest_metrics_reporter.h"
+
+#include <memory>
+
+#include <gtest/gtest.h>
+
+#include "iceberg/catalog/rest/auth/auth_session.h"
+#include "iceberg/catalog/rest/http_client.h"
+#include "iceberg/metrics/commit_report.h"
+#include "iceberg/metrics/metrics_reporter.h"
+#include "iceberg/metrics/scan_report.h"
+#include "iceberg/test/matchers.h"
+
+namespace iceberg::rest {
+
+class RestMetricsReporterTest : public ::testing::Test {
+ protected:
+ void SetUp() override {
+ client_ = std::make_shared<HttpClient>();
+ session_ = auth::AuthSession::MakeDefault({});
+ }
+
+ std::shared_ptr<HttpClient> client_;
+ std::shared_ptr<auth::AuthSession> session_;
+};
+
+// Report() must return OK even when the HTTP call fails (connection refused).
+// This validates the fire-and-forget error-suppression contract matching Java
behavior.
+TEST_F(RestMetricsReporterTest, ReportSuppressesHttpErrorsForScanReport) {
Review Comment:
This only verifies that errors are suppressed. It does not assert that the
reporter posts to the configured metrics endpoint, includes the required
`report-type`, or serializes scan vs commit bodies correctly. A mock or local
test server would catch REST payload regressions here.
##########
src/iceberg/transaction.cc:
##########
@@ -394,6 +412,30 @@ Result<std::shared_ptr<Table>> Transaction::Commit() {
committed_ = true;
ctx_->table = std::move(commit_result.value());
+ // Fire CommitReport only when a new snapshot was produced (not for
property-only
+ // commits). A SnapshotUpdate's own ReportWith() override (captured into
+ // snapshot_reporter_ by ApplyUpdateSnapshot()) takes precedence over the
table's
+ // reporter.
+ std::shared_ptr<MetricsReporter> reporter =
+ snapshot_reporter_ ? snapshot_reporter_ : ctx_->table->reporter();
+ if (reporter) {
+ auto snapshot_result = ctx_->table->metadata()->Snapshot();
Review Comment:
This derives the commit report from the table's current snapshot after the
commit. That misses `StageOnly()` updates, where the new snapshot is not
current, and it can report rollback/set-current operations as if they produced
a new commit. Java reports from the snapshot-producing operation event, so this
should use the `SnapshotUpdate` result instead of the post-commit current
snapshot.
##########
src/iceberg/util/content_file_util.cc:
##########
@@ -83,6 +83,10 @@ std::string ContentFileUtil::DVDesc(const DataFile& file) {
file.referenced_data_file.value_or(""));
}
+int64_t ContentFileUtil::ContentSizeInBytes(const DataFile& file) {
Review Comment:
This uses `content_size_in_bytes` for every content file when it is present.
Java only uses content size for deletion vectors; normal data/delete files
should report `file_size_in_bytes`. Otherwise any non-DV file that carries this
optional field will inflate or change delete metrics.
##########
src/iceberg/catalog/rest/rest_metrics_reporter.cc:
##########
@@ -0,0 +1,74 @@
+/*
+ * 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/catalog/rest/rest_metrics_reporter.h"
+
+#include <cstdio>
+#include <utility>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/auth/auth_session.h"
+#include "iceberg/catalog/rest/error_handlers.h"
+#include "iceberg/catalog/rest/http_client.h"
+#include "iceberg/metrics/json_serde_internal.h"
+#include "iceberg/metrics/metrics_reporter.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+constexpr std::string_view kReportType = "report-type";
+constexpr std::string_view kScanReportType = "scan-report";
+constexpr std::string_view kCommitReportType = "commit-report";
+
+} // namespace
+
+RestMetricsReporter::RestMetricsReporter(std::shared_ptr<HttpClient> client,
+ std::string metrics_endpoint,
+ std::shared_ptr<auth::AuthSession>
session)
+ : client_(std::move(client)),
+ metrics_endpoint_(std::move(metrics_endpoint)),
+ session_(std::move(session)) {}
+
+Status RestMetricsReporter::Report(const MetricsReport& report) {
+ try {
+ // Serialize the report variant to JSON.
+ Result<nlohmann::json> json_result = std::visit(
+ [](const auto& r) -> Result<nlohmann::json> { return ToJson(r); },
report);
+ if (!json_result) {
+ return {};
+ }
+
+ // Inject "report-type" required by the REST spec (not included in core
ToJson).
+ auto& json = json_result.value();
+ json[kReportType] =
+ std::holds_alternative<ScanReport>(report) ? kScanReportType :
kCommitReportType;
+
+ // POST to the metrics endpoint; suppress errors to match Java
fire-and-forget
+ // behavior.
+ std::ignore = client_->Post(metrics_endpoint_, json.dump(), /*headers=*/{},
Review Comment:
Nit/TODO: this POST is still synchronous. Java sends REST metrics
asynchronously and fire-and-forget, so a slow metrics endpoint should not block
scan planning or commits.
##########
src/iceberg/test/CMakeLists.txt:
##########
@@ -223,6 +223,7 @@ if(ICEBERG_BUILD_BUNDLE)
file_scan_task_test.cc
incremental_append_scan_test.cc
incremental_changelog_scan_test.cc
+ scan_planning_metrics_test.cc
Review Comment:
This adds `scan_planning_metrics_test.cc` to CMake only.
`src/iceberg/test/meson.build` should include the same test, otherwise Meson
test runs miss this coverage.
##########
src/iceberg/table_scan.cc:
##########
@@ -557,11 +586,52 @@ Result<std::vector<std::shared_ptr<FileScanTask>>>
DataTableScan::PlanFiles() co
.FilterData(filter())
.IgnoreDeleted()
.ColumnsToKeepStats(context_.columns_to_keep_stats)
- .PlanWith(context_.plan_executor);
+ .PlanWith(context_.plan_executor)
+ .ScanMetrics(scan_metrics);
if (context_.ignore_residuals) {
manifest_group->IgnoreResiduals();
}
- return manifest_group->PlanFiles();
+ ICEBERG_ASSIGN_OR_RAISE(auto tasks, manifest_group->PlanFiles());
+
+ timed.Stop();
+
+ if (context_.metrics_reporter) {
+ ICEBERG_ASSIGN_OR_RAISE(auto projected_schema, ResolveProjectedSchema());
+ const auto& schema_ptr = projected_schema.get();
+
+ ICEBERG_ASSIGN_OR_RAISE(auto projected_id_set,
+ GetProjectedIdsVisitor::GetProjectedIds(
+ *schema_ptr, /*include_struct_ids=*/true));
+ std::vector<int32_t> projected_field_ids(projected_id_set.begin(),
+ projected_id_set.end());
+ std::ranges::sort(projected_field_ids);
+
+ std::vector<std::string> projected_field_names;
+ projected_field_names.reserve(projected_field_ids.size());
+ for (int32_t field_id : projected_field_ids) {
+ ICEBERG_ASSIGN_OR_RAISE(auto field_name,
schema_ptr->FindColumnNameById(field_id));
+ ICEBERG_CHECK(field_name.has_value(), "Projected field {} not found in
schema",
+ field_id);
+ projected_field_names.emplace_back(*field_name);
+ }
+
+ ICEBERG_ASSIGN_OR_RAISE(auto sanitized_filter,
Review Comment:
This sanitizes the raw unbound filter without the table schema or
case-sensitivity setting. Java binds the expression before sanitizing, which
preserves resolved field names and transform terms. Case-insensitive scans or
filters with renamed/case-variant fields can report a different sanitized
filter here.
##########
src/iceberg/expression/sanitize_expression.cc:
##########
@@ -0,0 +1,358 @@
+/*
+ * 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/expression/sanitize_expression.h"
+
+#include <chrono>
+#include <cmath>
+#include <cstdint>
+#include <format>
+#include <limits>
+#include <regex>
+#include <string>
+#include <string_view>
+#include <utility>
+#include <vector>
+
+#include "iceberg/expression/binder.h"
+#include "iceberg/expression/literal.h"
+#include "iceberg/expression/predicate.h"
+#include "iceberg/expression/term.h"
+#include "iceberg/transform.h"
+#include "iceberg/type.h"
+#include "iceberg/util/bucket_util.h"
+#include "iceberg/util/checked_cast.h"
+#include "iceberg/util/temporal_util.h"
+
+namespace iceberg {
+
+namespace {
+
+std::string SanitizeDate(int32_t days, int32_t today) {
+ std::string is_past = today > days ? "ago" : "from-now";
+ // Compute the unsigned distance directly instead of subtracting then taking
abs():
+ // `today - days` can overflow int32_t (UB) when the values are near the
type's
+ // limits, since the sign of the operands isn't known ahead of time.
+ int32_t diff = today > days ?
static_cast<int32_t>(static_cast<uint32_t>(today) -
+
static_cast<uint32_t>(days))
+ :
static_cast<int32_t>(static_cast<uint32_t>(days) -
+
static_cast<uint32_t>(today));
+ if (diff == 0) {
+ return "(date-today)";
+ } else if (diff < 90) {
+ return "(date-" + std::to_string(diff) + "-days-" + is_past + ")";
+ }
+
+ return "(date)";
+}
+
+std::string SanitizeTimestamp(int64_t micros, int64_t now) {
+ constexpr int64_t kMicrosPerHour = 60LL * 60LL * 1'000'000LL;
+ constexpr int64_t kFiveMinutesInMicros = 5LL * 60LL * 1'000'000LL;
+ constexpr int64_t kThreeDaysInHours = 3LL * 24LL;
+ constexpr int64_t kNinetyDaysInHours = 90LL * 24LL;
+
+ std::string is_past = now > micros ? "ago" : "from-now";
+ // See SanitizeDate() for why this avoids `std::abs(now - micros)`: the
subtraction
+ // can overflow int64_t (UB) when the values are near the type's limits.
+ int64_t diff = now > micros ?
static_cast<int64_t>(static_cast<uint64_t>(now) -
Review Comment:
The unsigned subtraction avoids overflow, but casting the result back to
`int64_t` can wrap when the distance is greater than `INT64_MAX`. Extreme
timestamp/date values can then look negative and be classified as `about-now`
or as a small distance. This should saturate or keep the distance unsigned
before comparing.
--
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]
