src/paimon/common/metrics/histogram.cpp - paimon-cpp - Git at Google

 /*
  * 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 "paimon/common/metrics/histogram.h"

 #include <algorithm>
 #include <cmath>

 namespace paimon {

 HistogramImpl::HistogramImpl() : bucket_counts_(BucketLimits().size(), 0) {}

 void HistogramImpl::Add(double value) {
     if (std::isnan(value)) {
         return;
     }
     std::lock_guard<std::mutex> guard(mu_);
     const auto& limits = BucketLimits();
     size_t idx = BucketIndex(value, limits);
     bucket_counts_[idx] += 1;
     num_ += 1;
     sum_ += value;
     sum_squares_ += value * value;
     if (value < min_) {
         min_ = value;
     }
     if (value > max_) {
         max_ = value;
     }
 }

 HistogramStats HistogramImpl::GetStats() const {
     return ToStats(GetSnapshot());
 }

 HistogramImpl::Snapshot HistogramImpl::GetSnapshot() const {
     std::lock_guard<std::mutex> guard(mu_);
     Snapshot s;
     s.count = num_;
     s.sum = sum_;
     s.sum_squares = sum_squares_;
     s.min = (num_ == 0) ? 0 : min_;
     s.max = (num_ == 0) ? 0 : max_;
     s.bucket_counts = bucket_counts_;
     return s;
 }

 void HistogramImpl::Merge(const Histogram& other) {
     const auto* other_impl = dynamic_cast<const HistogramImpl*>(&other);
     if (!other_impl) {
         return;
     }
     MergeFrom(*other_impl);
 }

 void HistogramImpl::MergeFrom(const HistogramImpl& other) {
     Snapshot other_snapshot = other.GetSnapshot();
     std::lock_guard<std::mutex> guard(mu_);
     if (bucket_counts_.size() != other_snapshot.bucket_counts.size()) {
         // Should never happen unless bucket config changes.
         return;
     }
     if (other_snapshot.count == 0) {
         return;
     }
     for (size_t i = 0; i < bucket_counts_.size(); ++i) {
         bucket_counts_[i] += other_snapshot.bucket_counts[i];
     }
     if (num_ == 0) {
         min_ = other_snapshot.min;
         max_ = other_snapshot.max;
     } else {
         min_ = std::min(min_, other_snapshot.min);
         max_ = std::max(max_, other_snapshot.max);
     }
     num_ += other_snapshot.count;
     sum_ += other_snapshot.sum;
     sum_squares_ += other_snapshot.sum_squares;
 }

 std::shared_ptr<Histogram> HistogramImpl::Clone() const {
     Snapshot s = GetSnapshot();
     auto cloned = std::make_shared<HistogramImpl>();
     std::lock_guard<std::mutex> guard(cloned->mu_);
     cloned->num_ = s.count;
     cloned->sum_ = s.sum;
     cloned->sum_squares_ = s.sum_squares;
     cloned->min_ = (s.count == 0) ? std::numeric_limits<double>::infinity() : s.min;
     cloned->max_ = s.max;
     cloned->bucket_counts_ = std::move(s.bucket_counts);
     return cloned;
 }

 HistogramStats HistogramImpl::ToStats(const Snapshot& s) {
     HistogramStats stats;
     stats.count = s.count;
     stats.sum = s.sum;
     stats.min = s.min;
     stats.max = s.max;
     stats.average = (s.count == 0) ? 0 : (s.sum / static_cast<double>(s.count));
     stats.p50 = EstimatePercentile(s, 0.50);
     stats.p90 = EstimatePercentile(s, 0.90);
     stats.p95 = EstimatePercentile(s, 0.95);
     stats.p99 = EstimatePercentile(s, 0.99);
     stats.p999 = EstimatePercentile(s, 0.999);
     stats.stddev = EstimateStddev(s);
     return stats;
 }

 const std::vector<double>& HistogramImpl::BucketLimits() {
     static const std::vector<double> limits = []() {
         // Keep bucket generation aligned with RocksDB's HistogramBucketMapper
         // (see rocksdb/monitoring/histogram.cc).
         //
         // The mapper starts from {1,2}, then grows by *1.5 and rounds each bucket boundary to
         // be human-readable by keeping the two most significant digits.
         std::vector<double> v;
         v.reserve(256);
         v.push_back(1);
         v.push_back(2);

         double bucket_val = v.back();
         const auto max_u64 = static_cast<double>(std::numeric_limits<uint64_t>::max());
         while ((bucket_val = 1.5 * bucket_val) <= max_u64) {
             auto rounded = static_cast<uint64_t>(bucket_val);
             // Keep two most significant digits (e.g., 172 -> 170).
             uint64_t pow_of_ten = 1;
             while (rounded / 10 > 10) {
                 rounded /= 10;
                 pow_of_ten *= 10;
             }
             rounded *= pow_of_ten;
             v.push_back(static_cast<double>(rounded));
         }
         return v;
     }();
     return limits;
 }

 size_t HistogramImpl::BucketIndex(double value, const std::vector<double>& limits) {
     if (value <= limits.front()) {
         return 0;
     }
     auto it = std::lower_bound(limits.begin(), limits.end(), value);
     if (it == limits.end()) {
         return limits.size() - 1;
     }
     return static_cast<size_t>(std::distance(limits.begin(), it));
 }

 double HistogramImpl::EstimatePercentile(const Snapshot& s, double p) {
     if (s.count == 0) {
         return 0;
     }
     if (p <= 0) {
         return s.min;
     }
     if (p >= 1) {
         return s.max;
     }
     const auto& limits = BucketLimits();
     const auto rank = static_cast<uint64_t>(std::ceil(p * static_cast<double>(s.count)));
     uint64_t cum = 0;
     for (size_t i = 0; i < s.bucket_counts.size(); ++i) {
         uint64_t bcnt = s.bucket_counts[i];
         if (bcnt == 0) {
             continue;
         }
         uint64_t prev = cum;
         cum += bcnt;
         if (cum < rank) {
             continue;
         }

         // Interpolate inside bucket: estimate that samples are uniformly distributed
         // between bucket lower and upper bounds.
         double upper = (i < limits.size()) ? limits[i] : s.max;
         double lower = (i == 0) ? std::min(s.min, 0.0) : limits[i - 1];
         if (std::isinf(upper)) {
             return s.max;
         }
         const double pos_in_bucket = static_cast<double>(rank - prev) / static_cast<double>(bcnt);
         double est = lower + (upper - lower) * pos_in_bucket;
         if (est < s.min) {
             est = s.min;
         }
         if (est > s.max) {
             est = s.max;
         }
         return est;
     }
     return s.max;
 }

 double HistogramImpl::EstimateStddev(const Snapshot& s) {
     if (s.count == 0) {
         return 0;
     }
     const auto n = static_cast<double>(s.count);
     const double mean = s.sum / n;
     const double ex2 = s.sum_squares / n;
     const double var = std::max(0.0, ex2 - mean * mean);
     return std::sqrt(var);
 }

 }  // namespace paimon
	/*
	* 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 "paimon/common/metrics/histogram.h"

	#include <algorithm>
	#include <cmath>

	namespace paimon {

	HistogramImpl::HistogramImpl() : bucket_counts_(BucketLimits().size(), 0) {}

	void HistogramImpl::Add(double value) {
	if (std::isnan(value)) {
	return;
	}
	std::lock_guard<std::mutex> guard(mu_);
	const auto& limits = BucketLimits();
	size_t idx = BucketIndex(value, limits);
	bucket_counts_[idx] += 1;
	num_ += 1;
	sum_ += value;
	sum_squares_ += value * value;
	if (value < min_) {
	min_ = value;
	}
	if (value > max_) {
	max_ = value;
	}
	}

	HistogramStats HistogramImpl::GetStats() const {
	return ToStats(GetSnapshot());
	}

	HistogramImpl::Snapshot HistogramImpl::GetSnapshot() const {
	std::lock_guard<std::mutex> guard(mu_);
	Snapshot s;
	s.count = num_;
	s.sum = sum_;
	s.sum_squares = sum_squares_;
	s.min = (num_ == 0) ? 0 : min_;
	s.max = (num_ == 0) ? 0 : max_;
	s.bucket_counts = bucket_counts_;
	return s;
	}

	void HistogramImpl::Merge(const Histogram& other) {
	const auto* other_impl = dynamic_cast<const HistogramImpl*>(&other);
	if (!other_impl) {
	return;
	}
	MergeFrom(*other_impl);
	}

	void HistogramImpl::MergeFrom(const HistogramImpl& other) {
	Snapshot other_snapshot = other.GetSnapshot();
	std::lock_guard<std::mutex> guard(mu_);
	if (bucket_counts_.size() != other_snapshot.bucket_counts.size()) {
	// Should never happen unless bucket config changes.
	return;
	}
	if (other_snapshot.count == 0) {
	return;
	}
	for (size_t i = 0; i < bucket_counts_.size(); ++i) {
	bucket_counts_[i] += other_snapshot.bucket_counts[i];
	}
	if (num_ == 0) {
	min_ = other_snapshot.min;
	max_ = other_snapshot.max;
	} else {
	min_ = std::min(min_, other_snapshot.min);
	max_ = std::max(max_, other_snapshot.max);
	}
	num_ += other_snapshot.count;
	sum_ += other_snapshot.sum;
	sum_squares_ += other_snapshot.sum_squares;
	}

	std::shared_ptr<Histogram> HistogramImpl::Clone() const {
	Snapshot s = GetSnapshot();
	auto cloned = std::make_shared<HistogramImpl>();
	std::lock_guard<std::mutex> guard(cloned->mu_);
	cloned->num_ = s.count;
	cloned->sum_ = s.sum;
	cloned->sum_squares_ = s.sum_squares;
	cloned->min_ = (s.count == 0) ? std::numeric_limits<double>::infinity() : s.min;
	cloned->max_ = s.max;
	cloned->bucket_counts_ = std::move(s.bucket_counts);
	return cloned;
	}

	HistogramStats HistogramImpl::ToStats(const Snapshot& s) {
	HistogramStats stats;
	stats.count = s.count;
	stats.sum = s.sum;
	stats.min = s.min;
	stats.max = s.max;
	stats.average = (s.count == 0) ? 0 : (s.sum / static_cast<double>(s.count));
	stats.p50 = EstimatePercentile(s, 0.50);
	stats.p90 = EstimatePercentile(s, 0.90);
	stats.p95 = EstimatePercentile(s, 0.95);
	stats.p99 = EstimatePercentile(s, 0.99);
	stats.p999 = EstimatePercentile(s, 0.999);
	stats.stddev = EstimateStddev(s);
	return stats;
	}

	const std::vector<double>& HistogramImpl::BucketLimits() {
	static const std::vector<double> limits = []() {
	// Keep bucket generation aligned with RocksDB's HistogramBucketMapper
	// (see rocksdb/monitoring/histogram.cc).
	//
	// The mapper starts from {1,2}, then grows by *1.5 and rounds each bucket boundary to
	// be human-readable by keeping the two most significant digits.
	std::vector<double> v;
	v.reserve(256);
	v.push_back(1);
	v.push_back(2);

	double bucket_val = v.back();
	const auto max_u64 = static_cast<double>(std::numeric_limits<uint64_t>::max());
	while ((bucket_val = 1.5 * bucket_val) <= max_u64) {
	auto rounded = static_cast<uint64_t>(bucket_val);
	// Keep two most significant digits (e.g., 172 -> 170).
	uint64_t pow_of_ten = 1;
	while (rounded / 10 > 10) {
	rounded /= 10;
	pow_of_ten *= 10;
	}
	rounded *= pow_of_ten;
	v.push_back(static_cast<double>(rounded));
	}
	return v;
	}();
	return limits;
	}

	size_t HistogramImpl::BucketIndex(double value, const std::vector<double>& limits) {
	if (value <= limits.front()) {
	return 0;
	}
	auto it = std::lower_bound(limits.begin(), limits.end(), value);
	if (it == limits.end()) {
	return limits.size() - 1;
	}
	return static_cast<size_t>(std::distance(limits.begin(), it));
	}

	double HistogramImpl::EstimatePercentile(const Snapshot& s, double p) {
	if (s.count == 0) {
	return 0;
	}
	if (p <= 0) {
	return s.min;
	}
	if (p >= 1) {
	return s.max;
	}
	const auto& limits = BucketLimits();
	const auto rank = static_cast<uint64_t>(std::ceil(p * static_cast<double>(s.count)));
	uint64_t cum = 0;
	for (size_t i = 0; i < s.bucket_counts.size(); ++i) {
	uint64_t bcnt = s.bucket_counts[i];
	if (bcnt == 0) {
	continue;
	}
	uint64_t prev = cum;
	cum += bcnt;
	if (cum < rank) {
	continue;
	}

	// Interpolate inside bucket: estimate that samples are uniformly distributed
	// between bucket lower and upper bounds.
	double upper = (i < limits.size()) ? limits[i] : s.max;
	double lower = (i == 0) ? std::min(s.min, 0.0) : limits[i - 1];
	if (std::isinf(upper)) {
	return s.max;
	}
	const double pos_in_bucket = static_cast<double>(rank - prev) / static_cast<double>(bcnt);
	double est = lower + (upper - lower) * pos_in_bucket;
	if (est < s.min) {
	est = s.min;
	}
	if (est > s.max) {
	est = s.max;
	}
	return est;
	}
	return s.max;
	}

	double HistogramImpl::EstimateStddev(const Snapshot& s) {
	if (s.count == 0) {
	return 0;
	}
	const auto n = static_cast<double>(s.count);
	const double mean = s.sum / n;
	const double ex2 = s.sum_squares / n;
	const double var = std::max(0.0, ex2 - mean * mean);
	return std::sqrt(var);
	}

	} // namespace paimon