Merge main into sync PR 79503

Signed-off-by: luohaha <18810541851@163.com>

Author: luohaha

be/src/base/CMakeLists.txt

@@ -83,6 +83,7 @@ ADD_BE_LIB(Base
   time/monotime.cpp
   time/time.cpp
   time/timezone_utils.cpp
+  time/tz_offset_cache.cpp
   template/mustache/mustache.cc
   # base/testutil is not the UT-only TestUtil library. Production sources
   # include its BE_TEST-gated sync-point/fault-injection macros, which compile

be/src/base/time/tz_offset_cache.cpp

@@ -0,0 +1,169 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// Licensed 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
+//
+//     https://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 "base/time/tz_offset_cache.h"
+
+#include <algorithm>
+#include <chrono>
+#include <limits>
+
+namespace starrocks {
+
+namespace {
+// civil_second interpreted as if it were UTC, in whole seconds since the Unix epoch. Used only
+// as an offset-free anchor for the additive identity civil = epoch + (unix_seconds + tz_offset);
+// not itself a valid timezone conversion.
+int64_t civil_as_utc_seconds(const cctz::civil_second& cs) {
+    return cs - cctz::civil_second(1970, 1, 1, 0, 0, 0);
+}
+
+// Past the last transition in a zone's explicit table, cctz's next_transition() returns false
+// unconditionally ("ignoring future_spec_", per its own doc comment) -- it cannot tell us whether
+// that's because the zone will hold `offset` forever (e.g. Asia/Shanghai and Asia/Kolkata, which
+// abolished DST decades ago, or America/Phoenix, which has been fixed MST since it was defined --
+// all of them reach this point on ordinary present-day lookups, not just far-future ones) or
+// because it's a still-cycling DST zone we've simply run past the enumerable table for (only
+// reachable past ~year 2437 for this build's tzdata).
+//
+// There is no zone-specific logic here -- this is a generic probe applied to whatever zone was
+// passed in, and it works by exploiting a property of DST rules in general rather than anything
+// about a particular zone: every real-world annual DST cycle has exactly two transitions, and
+// each phase (DST / standard time) lasts several months, never mere days or weeks. So sampling
+// the offset at a few points spread roughly a quarter-year apart -- +91d, +182d, +273d, i.e.
+// ~3/6/9 months out, alongside `offset` itself already sampled at `tp` -- puts one sample in
+// each of the four seasons. If the zone is still cycling, at least one of those samples is
+// guaranteed to land in a different phase than `tp` and show a different offset; if it's
+// permanently fixed, all of them trivially match. The exact day counts aren't calibrated to any
+// zone's specific transition dates (that would be pointless -- the whole point is this has to
+// work for a zone we can no longer enumerate transitions for); they just need to be spaced closer
+// together than the shortest real-world DST phase, which they are by a wide margin.
+bool offset_is_constant_beyond(const cctz::time_zone& tz, const cctz::time_point<cctz::seconds>& tp, int64_t offset) {
+    for (int64_t days : {91, 182, 273}) {
+        if (tz.lookup_offset(tp + cctz::seconds(days * 86400)).offset != offset) {
+            return false;
+        }
+    }
+    return true;
+}
+} // namespace
+
+int64_t TzOffsetCache::offset_for_unix(int64_t unix_sec, const cctz::time_zone& tz) {
+    if (_abs_window.has_value && tz == _abs_window.zone && unix_sec >= _abs_window.lo && unix_sec < _abs_window.hi) {
+        return _abs_window.offset;
+    }
+
+    static const cctz::time_point<cctz::seconds> epoch =
+            std::chrono::time_point_cast<cctz::seconds>(std::chrono::system_clock::from_time_t(0));
+    const cctz::time_point<cctz::seconds> tp = epoch + cctz::seconds(unix_sec);
+    _abs_window.offset = tz.lookup_offset(tp).offset;
+
+    int64_t lo = std::numeric_limits<int64_t>::min();
+    int64_t hi = std::numeric_limits<int64_t>::max();
+    cctz::time_zone::civil_transition ct;
+    // next_transition(tp)/prev_transition(tp) both use a *strict* inequality against tp, so
+    // probing prev_transition at tp+1s (rather than tp) is what makes it return "the largest
+    // transition <= tp" instead of skipping over a transition that lands exactly on it.
+    const bool has_prev = tz.prev_transition(tp + cctz::seconds(1), &ct);
+    if (has_prev) {
+        lo = tz.lookup(ct.to).trans.time_since_epoch().count();
+    }
+    const bool has_next = tz.next_transition(tp, &ct);
+    if (has_next) {
+        hi = tz.lookup(ct.to).trans.time_since_epoch().count();
+    }
+    if (has_prev && !has_next && !offset_is_constant_beyond(tz, tp, _abs_window.offset)) {
+        // has_prev true but has_next false, and the offset actually varies later on: this is a
+        // zone with an ongoing DST cycle that we've simply run past cctz's enumerable transition
+        // table for (see offset_is_constant_beyond's comment). Caching an unbounded window here
+        // would silently reuse this offset for arbitrarily-far future instants that could be in
+        // the opposite DST season -- so don't cache; every such row gets a fresh, authoritative
+        // lookup instead. When the offset turns out to be constant beyond this point (e.g.
+        // Asia/Shanghai, Asia/Kolkata, America/Phoenix -- the common case this branch actually
+        // hits on ordinary present-day lookups), fall through and cache with hi left at +inf.
+        _abs_window.has_value = false;
+        return _abs_window.offset;
+    }
+    _abs_window.zone = tz;
+    _abs_window.lo = lo;
+    _abs_window.hi = hi;
+    _abs_window.has_value = true;
+    return _abs_window.offset;
+}
+
+int64_t TzOffsetCache::unix_for_civil(int64_t civil_as_utc_sec, int year, int month, int day, int hour, int minute,
+                                      int second, const cctz::time_zone& tz) {
+    if (_civil_window.has_value && tz == _civil_window.zone && civil_as_utc_sec >= _civil_window.lo &&
+        civil_as_utc_sec < _civil_window.hi) {
+        return civil_as_utc_sec - _civil_window.offset;
+    }
+
+    // Cold path: only reached ~twice per DST transition (or once per call for a wildly
+    // out-of-order stream), so it's fine to pay cctz::civil_second's construction/arithmetic
+    // cost here -- unlike the hot path above, which never touches it.
+    const cctz::civil_second cs(year, month, day, hour, minute, second);
+    const cctz::time_zone::civil_lookup cl = tz.lookup(cs);
+    const cctz::time_point<cctz::seconds> answer =
+            (cl.kind == cctz::time_zone::civil_lookup::SKIPPED) ? cl.trans : cl.pre;
+    const int64_t answer_unix = answer.time_since_epoch().count();
+
+    if (cl.kind != cctz::time_zone::civil_lookup::UNIQUE) {
+        // Ambiguous wall-clock value (the skipped/repeated hour around a DST change, ~1h/year).
+        // Too rare to be worth caching a window for; always re-resolve authoritatively.
+        _civil_window.has_value = false;
+        return answer_unix;
+    }
+
+    cctz::time_zone::civil_transition prev_trans, next_trans;
+    const bool has_prev = tz.prev_transition(answer + cctz::seconds(1), &prev_trans);
+    const bool has_next = tz.next_transition(answer, &next_trans);
+
+    int64_t lo = std::numeric_limits<int64_t>::min();
+    int64_t hi = std::numeric_limits<int64_t>::max();
+    if (has_prev) {
+        // For a spring-forward (gap) transition, prev_trans.to is later and starts the segment;
+        // for a fall-back (repeat) transition, prev_trans.from is later, and cctz's "prefer pre"
+        // tie-break means the whole repeated hour still belongs to the *earlier* segment, not
+        // this one -- either way max(from, to) is where the current segment actually starts.
+        lo = civil_as_utc_seconds(std::max(prev_trans.from, prev_trans.to));
+    }
+    if (has_next) {
+        hi = civil_as_utc_seconds(std::min(next_trans.from, next_trans.to));
+    }
+    if (has_prev && !has_next && !offset_is_constant_beyond(tz, answer, civil_as_utc_sec - answer_unix)) {
+        // Same "cctz gives up enumerating past the explicit table" hazard as in offset_for_unix
+        // above, and the same disambiguation: only skip caching when the offset actually varies
+        // beyond this point (a still-cycling DST zone run past the enumerable table), not when
+        // it's simply a zone permanently fixed since its last historical transition (the common
+        // case this branch hits on ordinary present-day lookups for zones like Asia/Shanghai).
+        _civil_window.has_value = false;
+        return answer_unix;
+    }
+    if (!(lo < hi)) {
+        // Pre-existing degenerate/adjacent-transitions guard.
+        _civil_window.has_value = false;
+        return answer_unix;
+    }
+
+    _civil_window.zone = tz;
+    // civil_as_utc_sec is the caller's non-cctz computation of the same quantity
+    // civil_as_utc_seconds(cs) would give; using it here (instead of recomputing via cctz) keeps
+    // this assignment consistent with what the hot path above will compare against later.
+    _civil_window.offset = civil_as_utc_sec - answer_unix;
+    _civil_window.lo = lo;
+    _civil_window.hi = hi;
+    _civil_window.has_value = true;
+    return answer_unix;
+}
+
+} // namespace starrocks

be/src/base/time/tz_offset_cache.h

@@ -0,0 +1,93 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// Licensed 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
+//
+//     https://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 <cctz/civil_time.h>
+#include <cctz/time_zone.h>
+
+#include <cstdint>
+
+namespace starrocks {
+
+// Caches the UTC offset currently in force for a cctz::time_zone, so that repeated conversions
+// against nearby times can skip cctz's transition-table search.
+//
+// cctz::time_zone itself already keeps such a hint (TimeZoneInfo::local_time_hint_ /
+// time_local_hint_ in time_zone_info.cc), but it is a single mutable slot shared by *every*
+// caller of that zone process-wide: under any real concurrency it becomes a cross-core
+// cache-line ping-pong, and one thread's hint constantly evicts another's. This class holds the
+// same kind of window privately per caller instead.
+//
+// Not thread-safe and not meant to be shared: each execution thread (or, in exprs/, each
+// FunctionContext worker via FunctionContext::get_or_create_thread_state) should own its own
+// instance. A single instance is reused across an unbounded number of calls and automatically
+// adapts as the input strays outside its current window -- there is no reset needed between
+// unrelated calls to the *same* (offset semantics never change for a fixed IANA zone), but an
+// instance must not be reused across two different cctz::time_zone values without expecting a
+// cache miss on the first call after the switch (which is always handled correctly, just not
+// from the fast path).
+//
+// Real-world timestamp columns are almost always time-clustered within one scan batch, so most
+// consecutive calls land in the same window; for a fixed-offset zone (no DST, e.g. "+08:00") the
+// window degenerates to unbounded after the very first call, so every later call is a hit
+// regardless of the time range covered.
+class TzOffsetCache {
+public:
+    // Returns the UTC offset (seconds east) in force for `tz` at absolute unix time `unix_sec`.
+    int64_t offset_for_unix(int64_t unix_sec, const cctz::time_zone& tz);
+
+    // Interprets the wall-clock fields (year, month, day, hour, minute, second) as being in `tz`
+    // and returns the absolute unix-second instant, matching cctz::convert(civil_second, tz)'s
+    // tie-break exactly: a SKIPPED (nonexistent) civil time resolves to the transition instant, a
+    // REPEATED (ambiguous) one resolves to the earlier ("pre") instant. Ambiguous inputs are
+    // always re-resolved through the authoritative cctz lookup and are never cached (they are
+    // rare -- at most ~1 hour per DST transition).
+    //
+    // `civil_as_utc_sec` is the same wall-clock fields reinterpreted as literal UTC seconds since
+    // the epoch (i.e. what they'd mean with no timezone applied at all) -- the caller computes
+    // this via a cheap, non-cctz calendar routine (e.g. TimestampValue::to_unix_second(), which
+    // this class -- living in base/, below types/ -- cannot call itself) so the common
+    // cache-hit path never has to touch cctz::civil_second at all. Passing a value inconsistent
+    // with the other fields is a caller bug; it is only used verbatim, never re-derived.
+    int64_t unix_for_civil(int64_t civil_as_utc_sec, int year, int month, int day, int hour, int minute, int second,
+                           const cctz::time_zone& tz);
+
+private:
+    // `zone` records which cctz::time_zone the window was computed for. An instance's (from, to)
+    // pair is normally fixed for its whole lifetime, but this guards against a window silently
+    // being reused after the caller switches to a different zone.
+    struct AbsWindow {
+        bool has_value = false;
+        cctz::time_zone zone;
+        int64_t offset = 0;
+        int64_t lo = 0; // inclusive, unix seconds
+        int64_t hi = 0; // exclusive, unix seconds
+    } _abs_window;
+
+    // Bounds are in the same "civil fields reinterpreted as literal UTC seconds" domain as
+    // `unix_for_civil`'s civil_as_utc_sec parameter, not cctz::civil_second -- comparing plain
+    // int64s on the hot path avoids cctz::civil_second arithmetic (specifically
+    // cctz::detail::impl::n_min and friends, the normalization routine backing civil_second's
+    // arithmetic operators) entirely once the window is warm.
+    struct CivilWindow {
+        bool has_value = false;
+        cctz::time_zone zone;
+        int64_t offset = 0; // seconds east of UTC
+        int64_t lo = 0;     // inclusive, civil-as-utc seconds
+        int64_t hi = 0;     // exclusive, civil-as-utc seconds
+    } _civil_window;
+};
+
+} // namespace starrocks

be/src/compute_env/spill/data_stream.cpp

@@ -147,7 +147,8 @@ std::shared_ptr<SpillOutputDataStream> create_spill_output_stream(Spiller* spill
 }
 
 Status DataTranster::transfer(workgroup::YieldContext& yield_ctx, RuntimeState* state, Serde* serde,
-                              const SpillOutputDataStreamPtr& output, const InputStreamPtr& input_stream) {
+                              const SpillOutputDataStreamPtr& output, const InputStreamPtr& input_stream,
+                              RuntimeProfile::Counter* merge_timer) {
     // read data from input stream and append to output stream
     bool need_aligned = state->spill_enable_direct_io();
     auto task_context = std::any_cast<SpillIOTaskContextPtr>(yield_ctx.task_context_data);
@@ -167,7 +168,13 @@ Status DataTranster::transfer(workgroup::YieldContext& yield_ctx, RuntimeState*
             RETURN_IF_YIELD(yield_ctx.need_yield);
         }
         DCHECK(input_stream->is_ready());
-        auto chunk_st = input_stream->get_next(yield_ctx, read_ctx);
+        // Charge the read itself -- for a compaction, key comparison and chunk assembly -- to the
+        // caller's timer. The deserialize and read IO it feeds on happen in the restore task above
+        // and keep their own counters, so the two do not overlap.
+        auto chunk_st = [&]() {
+            SCOPED_TIMER(merge_timer);
+            return input_stream->get_next(yield_ctx, read_ctx);
+        }();
         RETURN_IF(!chunk_st.status().is_ok_or_eof(), chunk_st.status());
         RETURN_IF(chunk_st.status().is_end_of_file(), Status::OK());
         RETURN_IF_ERROR(serde->serialize(state, read_ctx, std::move(chunk_st.value()), output, need_aligned));

be/src/compute_env/spill/data_stream.h

@@ -15,6 +15,7 @@
 #pragma once
 
 #include "base/string/slice.h"
+#include "common/runtime_profile.h"
 #include "common/status.h"
 #include "compute_env/spill/spill_fwd.h"
 #include "runtime/runtime_state_fwd.h"
@@ -44,8 +45,12 @@ using InputStreamPtr = std::shared_ptr<SpillInputStream>;
 // This class will execution io tasks. make sure call it in io threads
 class DataTranster {
 public:
+    // merge_timer, when non-null, is charged with the time spent pulling chunks out of input_stream
+    // (for a compaction that is the multi-way merge). Callers whose work is not a compaction pass
+    // nullptr rather than mislabeling it; ScopedTimer no-ops on a null counter.
     static Status transfer(workgroup::YieldContext& yield_ctx, RuntimeState* state, Serde* serde,
-                           const SpillOutputDataStreamPtr& output, const InputStreamPtr& input_stream);
+                           const SpillOutputDataStreamPtr& output, const InputStreamPtr& input_stream,
+                           RuntimeProfile::Counter* merge_timer = nullptr);
 };
 
 } // namespace starrocks::spill

be/src/compute_env/spill/spill_components.cpp

@@ -119,6 +119,7 @@ Status RawSpillerWriter::yieldable_flush_task(workgroup::YieldContext& yield_ctx
 }
 
 Status RawSpillerWriter::_spill_mem_table(workgroup::YieldContext& yield_ctx, const MemTablePtr& mem_table) {
+    SCOPED_TIMER(_spiller->metrics().flush_mem_table_timer);
     auto io_task = std::any_cast<SpillIOTaskContextPtr>(yield_ctx.task_context_data);
     auto flush_ctx = std::static_pointer_cast<FlushContext>(io_task);
 
@@ -144,6 +145,7 @@ Status RawSpillerWriter::_spill_mem_table(workgroup::YieldContext& yield_ctx, co
 }
 
 Status RawSpillerWriter::_compact_mem_table(workgroup::YieldContext& yield_ctx) {
+    SCOPED_TIMER(_spiller->metrics().compact_timer);
     auto io_task = std::any_cast<SpillIOTaskContextPtr>(yield_ctx.task_context_data);
     auto flush_ctx = std::static_pointer_cast<FlushContext>(io_task);
     // flush_ctx->output is not nullptr means the task is resumed from yield point
@@ -159,6 +161,9 @@ Status RawSpillerWriter::_compact_mem_table(workgroup::YieldContext& yield_ctx)
 
         COUNTER_UPDATE(_spiller->metrics().compact_count, 1);
         COUNTER_UPDATE(_spiller->metrics().compact_block_count, block_groups.size());
+        COUNTER_UPDATE(_spiller->metrics().compact_bytes_read,
+                       std::accumulate(block_groups.begin(), block_groups.end(), int64_t{0},
+                                       [](int64_t sum, const auto& group) { return sum + group->data_size(); }));
 
         flush_ctx->compact_input_num_rows =
                 std::accumulate(block_groups.begin(), block_groups.end(), 0,
@@ -174,11 +179,12 @@ Status RawSpillerWriter::_compact_mem_table(workgroup::YieldContext& yield_ctx)
                                                              options().sort_exprs, options().sort_desc));
     }
     auto st = DataTranster::transfer(yield_ctx, _runtime_state, _spiller->serde().get(), flush_ctx->output,
-                                     flush_ctx->input_stream);
+                                     flush_ctx->input_stream, _spiller->metrics().compact_merge_timer);
     RETURN_IF(!st.is_ok_or_eof(), st);
     RETURN_IF_YIELD(yield_ctx.need_yield);
     RETURN_IF_ERROR(flush_ctx->output->flush());
     flush_ctx->output.reset();
+    COUNTER_UPDATE(_spiller->metrics().compact_bytes_written, flush_ctx->block_group->data_size());
     // On cancellation the transfer stops early, so the compacted block group holds fewer rows than
     // its input; only assert full compaction while the query is still running. Query cancel sets
     // the runtime state (which the transfer honors) but not necessarily the spiller flag.