ClickHouse · CheSema · Jun 3, 2026 · Jun 3, 2026 · Jun 3, 2026 · Jun 3, 2026
diff --git a/.gitignore b/.gitignore
@@ -198,3 +198,4 @@ tests/casa_del_dolor/_instances*
 ci/praktika/json_test.html
 ci/praktika/json_test.html.gz
 ci/praktika/json.html.gz
+tmp/
diff --git a/ci/jobs/scripts/clickhouse_proc.py b/ci/jobs/scripts/clickhouse_proc.py
@@ -737,7 +737,7 @@ def prepare_stateful_data(self, with_s3_storage, is_db_replicated):
 fi
 clickhouse-client --query "CREATE TABLE test.hits_s3  (WatchID UInt64, JavaEnable UInt8, Title String, GoodEvent Int16, EventTime DateTime, EventDate Date, CounterID UInt32, ClientIP UInt32, ClientIP6 FixedString(16), RegionID UInt32, UserID UInt64, CounterClass Int8, OS UInt8, UserAgent UInt8, URL String, Referer String, URLDomain String, RefererDomain String, Refresh UInt8, IsRobot UInt8, RefererCategories Array(UInt16), URLCategories Array(UInt16), URLRegions Array(UInt32), RefererRegions Array(UInt32), ResolutionWidth UInt16, ResolutionHeight UInt16, ResolutionDepth UInt8, FlashMajor UInt8, FlashMinor UInt8, FlashMinor2 String, NetMajor UInt8, NetMinor UInt8, UserAgentMajor UInt16, UserAgentMinor FixedString(2), CookieEnable UInt8, JavascriptEnable UInt8, IsMobile UInt8, MobilePhone UInt8, MobilePhoneModel String, Params String, IPNetworkID UInt32, TraficSourceID Int8, SearchEngineID UInt16, SearchPhrase String, AdvEngineID UInt8, IsArtifical UInt8, WindowClientWidth UInt16, WindowClientHeight UInt16, ClientTimeZone Int16, ClientEventTime DateTime, SilverlightVersion1 UInt8, SilverlightVersion2 UInt8, SilverlightVersion3 UInt32, SilverlightVersion4 UInt16, PageCharset String, CodeVersion UInt32, IsLink UInt8, IsDownload UInt8, IsNotBounce UInt8, FUniqID UInt64, HID UInt32, IsOldCounter UInt8, IsEvent UInt8, IsParameter UInt8, DontCountHits UInt8, WithHash UInt8, HitColor FixedString(1), UTCEventTime DateTime, Age UInt8, Sex UInt8, Income UInt8, Interests UInt16, Robotness UInt8, GeneralInterests Array(UInt16), RemoteIP UInt32, RemoteIP6 FixedString(16), WindowName Int32, OpenerName Int32, HistoryLength Int16, BrowserLanguage FixedString(2), BrowserCountry FixedString(2), SocialNetwork String, SocialAction String, HTTPError UInt16, SendTiming Int32, DNSTiming Int32, ConnectTiming Int32, ResponseStartTiming Int32, ResponseEndTiming Int32, FetchTiming Int32, RedirectTiming Int32, DOMInteractiveTiming Int32, DOMContentLoadedTiming Int32, DOMCompleteTiming Int32, LoadEventStartTiming Int32, LoadEventEndTiming Int32, NSToDOMContentLoadedTiming Int32, FirstPaintTiming Int32, RedirectCount Int8, SocialSourceNetworkID UInt8, SocialSourcePage String, ParamPrice Int64, ParamOrderID String, ParamCurrency FixedString(3), ParamCurrencyID UInt16, GoalsReached Array(UInt32), OpenstatServiceName String, OpenstatCampaignID String, OpenstatAdID String, OpenstatSourceID String, UTMSource String, UTMMedium String, UTMCampaign String, UTMContent String, UTMTerm String, FromTag String, HasGCLID UInt8, RefererHash UInt64, URLHash UInt64, CLID UInt32, YCLID UInt64, ShareService String, ShareURL String, ShareTitle String, ParsedParams Nested(Key1 String, Key2 String, Key3 String, Key4 String, Key5 String, ValueDouble Float64), IslandID FixedString(16), RequestNum UInt32, RequestTry UInt8) ENGINE = MergeTree() PARTITION BY toYYYYMM(EventDate) ORDER BY (CounterID, EventDate, intHash32(UserID)) SAMPLE BY intHash32(UserID) SETTINGS index_granularity = 8192, storage_policy='s3_cache'"
 # AWS S3 is very inefficient, so increase memory even further:
-clickhouse-client --max_estimated_execution_time 0 --max_execution_time "$MAX_EXECUTION_TIME" --max_memory_usage 30G --max_memory_usage_for_user 30G --query "INSERT INTO test.hits_s3 SELECT * FROM test.hits SETTINGS enable_filesystem_cache_on_write_operations=0, write_through_distributed_cache=0, max_insert_threads=16"
+clickhouse-client --max_estimated_execution_time 0 --max_execution_time "$MAX_EXECUTION_TIME" --max_memory_usage 30G --max_memory_usage_for_user 30G --query "INSERT INTO test.hits_s3 SELECT * FROM test.hits SETTINGS enable_filesystem_cache_on_write_operations=0, write_through_distributed_cache=0, max_insert_threads=8"
 
 clickhouse-client --query "CREATE TABLE test.hits_parquet (Title String, URL String, Referer String, SearchPhrase String, WatchID UInt64, UserID UInt64, CounterID UInt32, EventTime DateTime, EventDate Date, RegionID UInt32, ClientIP UInt32) ENGINE = S3('https://clickhouse-public-datasets.s3.eu-central-1.amazonaws.com/hits_compatible/hits.parquet', NOSIGN)"
 

diff --git a/programs/server/Server.cpp b/programs/server/Server.cpp
@@ -1,6 +1,8 @@
 #include <Server.h>
 #include <Common/CurrentThread.h>
 #include <Common/QueryScope.h>
+#include <Common/MemoryPressureMonitor.h>
+#include <IO/LongConnectionLimit.h>
 
 #include <memory>
 #include <Interpreters/ClientInfo.h>
@@ -320,6 +322,10 @@ namespace ServerSetting
     extern const ServerSettingsUInt64 max_io_thread_pool_free_size;
     extern const ServerSettingsUInt64 max_io_thread_pool_size;
     extern const ServerSettingsUInt64 max_keep_alive_requests;
+    extern const ServerSettingsUInt64 max_remote_read_connections;
+    extern const ServerSettingsUInt64 reader_executor_memory_pressure_level_1_pct;
+    extern const ServerSettingsUInt64 reader_executor_memory_pressure_level_2_pct;
+    extern const ServerSettingsUInt64 reader_executor_memory_pressure_level_3_pct;
     extern const ServerSettingsUInt64 max_outdated_parts_loading_thread_pool_size;
     extern const ServerSettingsUInt64 max_per_cpu_untracked_memory;
     extern const ServerSettingsUInt64 max_partition_size_to_drop;
@@ -2413,6 +2419,15 @@ try
             ServerSettings new_server_settings;
             new_server_settings.loadSettingsFromConfig(config());
 
+            /// Reject an invalid memory-pressure threshold triple BEFORE applying any
+            /// live setting below: `setThresholds` validates too, but it runs after the
+            /// pools/throttlers/workloads are already reloaded, so a late throw would
+            /// leave those earlier settings from the same rejected reload published.
+            validateMemoryPressureThresholds(
+                new_server_settings[ServerSetting::reader_executor_memory_pressure_level_1_pct],
+                new_server_settings[ServerSetting::reader_executor_memory_pressure_level_2_pct],
+                new_server_settings[ServerSetting::reader_executor_memory_pressure_level_3_pct]);
+
             DB::abort_on_logical_error.store(new_server_settings[ServerSetting::abort_on_logical_error], std::memory_order_relaxed);
 
             size_t max_server_memory_usage = new_server_settings[ServerSetting::max_server_memory_usage];
@@ -2674,6 +2689,17 @@ try
                 new_server_settings[ServerSetting::cpu_slot_quantum_ns],
                 new_server_settings[ServerSetting::cpu_slot_preemption_timeout_ms]);
 
+            /// Apply the thresholds (already validated above, so this won't throw)
+            /// and reset the monitor's cooldown so the next sample reclassifies
+            /// against the new ladder.
+            memoryPressureMonitor().setThresholds(
+                new_server_settings[ServerSetting::reader_executor_memory_pressure_level_1_pct],
+                new_server_settings[ServerSetting::reader_executor_memory_pressure_level_2_pct],
+                new_server_settings[ServerSetting::reader_executor_memory_pressure_level_3_pct]);
+
+            global_context->getLongConnectionLimit()->setCapacity(
+                new_server_settings[ServerSetting::max_remote_read_connections]);
+
             if (config().has("resources"))
             {
                 global_context->getResourceManager()->updateConfiguration(config());

diff --git a/src/Common/CurrentMetrics.cpp b/src/Common/CurrentMetrics.cpp
@@ -58,6 +58,7 @@
     M(Read, "Number of read (read, pread, io_getevents, etc.) syscalls in fly") \
     M(RemoteRead, "Number of read with remote reader in fly") \
     M(ReaderExecutorActive, "Number of live ReaderExecutor instances.") \
+    M(ReaderExecutorPrefetchInFlight, "Number of ReaderExecutor prefetch tasks currently queued or executing.") \
     M(ReaderExecutorChainedBufferBytes, "Bytes currently held in live ReaderExecutor chain buffers (OwnedChainedBuffer) -- live in-flight read memory, not a cumulative total.") \
     M(Write, "Number of write (write, pwrite, io_getevents, etc.) syscalls in fly") \
     M(NetworkReceive, "Number of threads receiving data from network. Only ClickHouse-related network interaction is included, not by 3rd party libraries.") \
@@ -536,6 +537,8 @@
     M(MergeTreeSnapshotCommitThreads, "Number of threads used to commit snapshot") \
     M(MergeTreeSnapshotCommitThreadsActive, "Number of active threads used to commit snapshot") \
     M(MergeTreeSnapshotCommitThreadsScheduled, "Number of scheduled threads used to commit snapshot") \
+    \
+    M(LongConnections, "Number of long source connections currently held open by ReaderExecutor for sequential read optimization.") \
 
 
 #ifdef APPLY_FOR_EXTERNAL_METRICS

diff --git a/src/Common/FailPoint.cpp b/src/Common/FailPoint.cpp
@@ -98,6 +98,8 @@ static struct InitFiu
     REGULAR(smt_sleep_in_schedule_data_processing_job) \
     REGULAR(cache_warmer_stall) \
     REGULAR(file_cache_dynamic_resize_fail_to_evict) \
+    PAUSEABLE_ONCE(reader_executor_pause_after_window) \
+    PAUSEABLE_ONCE(reader_executor_pause_after_cache_status) \
     REGULAR(file_cache_slru_downgrade_fail_before_finalize) \
     REGULAR(file_cache_modify_size_limits_fail) \
     REGULAR(check_table_query_delay_for_part) \

diff --git a/src/Common/HistogramMetrics.cpp b/src/Common/HistogramMetrics.cpp
@@ -68,6 +68,31 @@ namespace HistogramMetrics
         "Duration of S3 read request connections, from request initiation to connection close, in microseconds.",
         {1000, 10000, 50000, 200000, 500000, 1000000, 2000000, 5000000, 10000000, 60000000});
 
+    Metric & ReaderExecutorCacheReadLatency = Factory::instance().registerMetric(
+        "reader_executor_cache_read_microseconds",
+        "Per-call latency of CacheReader::read inside ReaderExecutor, in microseconds.",
+        {10, 100, 1000, 5000, 20000, 100000, 500000, 2000000});
+
+    Metric & ReaderExecutorCachePopulateLatency = Factory::instance().registerMetric(
+        "reader_executor_cache_populate_microseconds",
+        "Per-call latency of CacheWriter::write inside ReaderExecutor, in microseconds.",
+        {10, 100, 1000, 5000, 20000, 100000, 500000, 2000000});
+
+    Metric & ReaderExecutorSourceReadLatency = Factory::instance().registerMetric(
+        "reader_executor_source_read_microseconds",
+        "Per-call latency of source reads driven by ReaderExecutor, in microseconds.",
+        {100, 1000, 10000, 100000, 500000, 2000000, 5000000, 15000000});
+
+    Metric & ReaderExecutorPrefetchWaitLatency = Factory::instance().registerMetric(
+        "reader_executor_prefetch_wait_microseconds",
+        "Per-event wait time when the ReaderExecutor consumer blocked on a not-yet-ready prefetch future, in microseconds. Directly contributes to query latency.",
+        {10, 100, 1000, 10000, 100000, 500000, 2000000, 10000000});
+
+    Metric & ReaderExecutorSyncReadLatency = Factory::instance().registerMetric(
+        "reader_executor_sync_read_microseconds",
+        "Per-event latency of in-line synchronous reads when no usable prefetch was available, in microseconds. Directly contributes to query latency.",
+        {100, 1000, 10000, 100000, 500000, 2000000, 5000000, 15000000});
+
     Metric & S3ReadRequestBytes = Factory::instance().registerMetric(
         "s3_read_request_bytes",
         "Bytes read per S3 read request connection.",

diff --git a/src/Common/MemoryPressureMonitor.cpp b/src/Common/MemoryPressureMonitor.cpp
@@ -0,0 +1,218 @@
+#include <Common/MemoryPressureMonitor.h>
+#include <Common/CurrentThread.h>
+#include <Common/Exception.h>
+#include <Common/MemoryTracker.h>
+#include <Common/VariableContext.h>
+#include <base/defines.h>
+
+#include <algorithm>
+#include <chrono>
+
+namespace DB
+{
+
+namespace ErrorCodes
+{
+    extern const int BAD_ARGUMENTS;
+}
+
+uint8_t PressureLevelMachine::rawLevel(double pressure) const
+{
+    const uint32_t packed = thresholds_packed.load(std::memory_order_relaxed);
+    const double t1 = ((packed >> 16) & 0xFFu) / 100.0;
+    const double t2 = ((packed >> 8) & 0xFFu) / 100.0;
+    const double t3 = (packed & 0xFFu) / 100.0;
+
+    if (pressure >= t3)
+        return 3;
+    if (pressure >= t2)
+        return 2;
+    if (pressure >= t1)
+        return 1;
+    return 0;
+}
+
+MemoryPressureLevel PressureLevelMachine::levelForPressure(double pressure) const
+{
+    /// NOLINTNEXTLINE(clang-analyzer-optin.core.EnumCastOutOfRange)
+    return static_cast<MemoryPressureLevel>(rawLevel(pressure));
+}
+
+MemoryPressureLevel PressureLevelMachine::sample(double pressure, uint64_t now_ns)
+{
+    std::lock_guard lk(mutex);
+
+    /// Classify under the lock: a concurrent `setThresholds` swaps the ladder and
+    /// resets the cooldown under this same lock, so the thresholds can't change
+    /// between this classification and the cooldown update below.
+    const uint8_t raw_level = rawLevel(pressure);
+
+    if (raw_level >= level)
+    {
+        /// Snap up immediately; refresh the "still elevated" timestamp.
+        level = raw_level;
+        last_at_or_above_ns = now_ns;
+    }
+    else if (level > 0 && now_ns >= last_at_or_above_ns + COOLDOWN_NS)
+    {
+        /// Step down by ONE level per cooldown — a CRITICAL → NORMAL recovery
+        /// needs ≥ 3 × COOLDOWN_NS of sustained low pressure.
+        level -= 1;
+        last_at_or_above_ns = now_ns;
+    }
+
+    chassert(level <= 3);
+    /// NOLINTNEXTLINE(clang-analyzer-optin.core.EnumCastOutOfRange)
+    return static_cast<MemoryPressureLevel>(level);
+}
+
+void validateMemoryPressureThresholds(UInt64 l1_pct, UInt64 l2_pct, UInt64 l3_pct)
+{
+    /// Validate range and ordering loudly. Silent clamping/sorting (the
+    /// previous behavior) hid config typos — e.g. `300` would wrap to `44`
+    /// through `uint8_t` at the call site and then look valid here, so the
+    /// server would silently apply a wrong threshold ladder.
+    if (l1_pct > 100 || l2_pct > 100 || l3_pct > 100)
+        throw Exception(ErrorCodes::BAD_ARGUMENTS,
+            "Memory pressure thresholds must be in [0, 100], got "
+            "level_1={}, level_2={}, level_3={}",
+            l1_pct, l2_pct, l3_pct);
+    if (!(l1_pct <= l2_pct && l2_pct <= l3_pct))
+        throw Exception(ErrorCodes::BAD_ARGUMENTS,
+            "Memory pressure thresholds must satisfy level_1 <= level_2 <= level_3, got "
+            "level_1={}, level_2={}, level_3={}",
+            l1_pct, l2_pct, l3_pct);
+}
+
+void PressureLevelMachine::setThresholds(UInt64 l1_pct, UInt64 l2_pct, UInt64 l3_pct)
+{
+    validateMemoryPressureThresholds(l1_pct, l2_pct, l3_pct);
+
+    /// Publish all three as one packed value so `rawLevel` never sees a
+    /// half-applied ladder.
+    const uint32_t packed = (static_cast<uint32_t>(l1_pct) << 16)
+                          | (static_cast<uint32_t>(l2_pct) << 8)
+                          | static_cast<uint32_t>(l3_pct);
+
+    std::lock_guard lk(mutex);
+    thresholds_packed.store(packed, std::memory_order_relaxed);
+    /// Reset the cooldown under the same lock `sample` uses: a level classified
+    /// under the old ladder must not stick after the ladder changes. The next
+    /// `sample` reclassifies (snap-up is immediate, so genuinely-high pressure
+    /// re-escalates at once rather than waiting out a stale cooldown).
+    level = 0;
+    last_at_or_above_ns = 0;
+}
+
+MemoryPressureThresholds PressureLevelMachine::getThresholds() const
+{
+    const uint32_t packed = thresholds_packed.load(std::memory_order_relaxed);
+    return {(packed >> 16) & 0xFFu, (packed >> 8) & 0xFFu, packed & 0xFFu};
+}
+
+namespace
+{
+
+double readTotalPressure()
+{
+    const Int64 limit = total_memory_tracker.getHardLimit();
+    if (limit <= 0)
+        return 0.0;
+    const Int64 used = total_memory_tracker.get();
+    if (used <= 0)
+        return 0.0;
+    return static_cast<double>(used) / static_cast<double>(limit);
+}
+
+uint64_t steadyNowNs()
+{
+    return std::chrono::duration_cast<std::chrono::nanoseconds>(
+               std::chrono::steady_clock::now().time_since_epoch())
+        .count();
+}
+
+}
+
+double localMemoryPressureFromChain(MemoryTracker * start)
+{
+    double worst = 0.0;
+    for (MemoryTracker * t = start; t != nullptr; t = t->getParent())
+    {
+        /// The server total (`Global`) is handled by the cooldown-smoothed
+        /// total-pressure path; here we react only to the transient per-query
+        /// (`Process`) and per-user (`User`) limits.
+        if (t->level == VariableContext::Global)
+            continue;
+        worst = std::max(worst, t->getPressure());
+    }
+    return worst;
+}
+
+MemoryPressureLevel MemoryPressureMonitor::currentLevel()
+{
+    /// Server-wide pressure keeps its 60s sticky-downward cooldown.
+    const MemoryPressureLevel total_level = machine.sample(readTotalPressure(), steadyNowNs());
+    /// Per-query / per-user limits are transient: react immediately, with no
+    /// sticky state that could leak one query's pressure into another's.
+    const MemoryPressureLevel local_level =
+        machine.levelForPressure(localMemoryPressureFromChain(CurrentThread::getMemoryTracker()));
+    return std::max(total_level, local_level);
+}
+
+MemoryPressureLevel FakeMemoryPressureMonitor::currentLevel()
+{
+    return machine.sample(
+        pressure.load(std::memory_order_relaxed),
+        now_ns.load(std::memory_order_relaxed));
+}
+
+namespace
+{
+
+/// Atomic pointer to the active monitor. Production never writes after
+/// first init; tests swap it via `ScopedMemoryPressureMonitor`.
+std::atomic<IMemoryPressureMonitor *> & activeMonitorSlot()
+{
+    static std::atomic<IMemoryPressureMonitor *> slot{nullptr};
+    return slot;
+}
+
+MemoryPressureMonitor & productionInstance()
+{
+    static MemoryPressureMonitor instance;
+    return instance;
+}
+
+}
+
+IMemoryPressureMonitor & memoryPressureMonitor()
+{
+    auto & slot = activeMonitorSlot();
+    auto * current = slot.load(std::memory_order_acquire);
+    if (current)
+        return *current;
+
+    /// First access: install the production singleton. Concurrent first
+    /// callers all reference the same Meyers singleton; `compare_exchange`
+    /// ensures only one wins the slot, the rest pick up that value.
+    auto * production = &productionInstance();
+    IMemoryPressureMonitor * expected = nullptr;
+    slot.compare_exchange_strong(expected, production, std::memory_order_acq_rel);
+    return *slot.load(std::memory_order_acquire);
+}
+
+ScopedMemoryPressureMonitor::ScopedMemoryPressureMonitor(IMemoryPressureMonitor & override_monitor)
+{
+    /// Force the production singleton to be constructed and slotted so the
+    /// dtor can restore it even if nothing called `memoryPressureMonitor()`
+    /// before this scope was entered.
+    (void)memoryPressureMonitor();
+    prior = activeMonitorSlot().exchange(&override_monitor, std::memory_order_acq_rel);
+}
+
+ScopedMemoryPressureMonitor::~ScopedMemoryPressureMonitor()
+{
+    activeMonitorSlot().store(prior, std::memory_order_release);
+}
+
+}