diff --git a/.github/import_generation.txt b/.github/import_generation.txt
index 21e72e8ac3d..95f9650f015 100644
--- a/.github/import_generation.txt
+++ b/.github/import_generation.txt
@@ -1 +1 @@
-48
+49
diff --git a/.github/last_commit.txt b/.github/last_commit.txt
index e0b76008046..a2a0a3576eb 100644
--- a/.github/last_commit.txt
+++ b/.github/last_commit.txt
@@ -1 +1 @@
-a7781966132cf3d3f84cc320cd002257c6220f5b
+d4e67d2428cc0f9dd329065a3c38476fbd8b7592
diff --git a/.github/workflows/slo.yml b/.github/workflows/slo.yml
index ef1e8b5fe38..85be622e883 100644
--- a/.github/workflows/slo.yml
+++ b/.github/workflows/slo.yml
@@ -12,7 +12,7 @@ jobs:
     if: contains(github.event.pull_request.labels.*.name, 'SLO')
 
     name: Cache SLO SDK (${{ matrix.sdk.name }})
-    runs-on: ubuntu-latest
+    runs-on: large-runner-cpp-sdk
     timeout-minutes: 90
 
     strategy:
@@ -86,7 +86,7 @@ jobs:
     needs: sdk-cache
 
     name: Run YDB SLO Tests (${{ matrix.sdk.name }})
-    runs-on: ubuntu-latest
+    runs-on: large-runner-cpp-sdk
     timeout-minutes: 120
 
     strategy:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 77a91361126..39cb457b8e7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,5 @@
+* Added `EQ_HEIGHT_HISTOGRAM` to `EMultiColumnStatisticsType`.
+
 # v3.22.0
 
 * Added `IWriteSession::Flush` to asynchronously wait until all previously accepted topic writes are acknowledged.
diff --git a/cmake/public_headers.txt b/cmake/public_headers.txt
index c80c556a5e7..133386bfa38 100644
--- a/cmake/public_headers.txt
+++ b/cmake/public_headers.txt
@@ -103,6 +103,7 @@ util/generic/array_ref.h
 util/generic/bitops.h
 util/generic/buffer.h
 util/generic/cast.h
+util/generic/constant_evaluation.h
 util/generic/deque.h
 util/generic/explicit_type.h
 util/generic/flags.h
diff --git a/include/ydb-cpp-sdk/client/query/query.h b/include/ydb-cpp-sdk/client/query/query.h
index fa0c557f038..c4acd8c1f72 100644
--- a/include/ydb-cpp-sdk/client/query/query.h
+++ b/include/ydb-cpp-sdk/client/query/query.h
@@ -111,6 +111,7 @@ struct TExecuteQuerySettings : public TRequestSettings<TExecuteQuerySettings> {
     FLUENT_SETTING_DEFAULT(ESyntax, Syntax, ESyntax::YqlV1);
     FLUENT_SETTING_DEFAULT(EExecMode, ExecMode, EExecMode::Execute);
     FLUENT_SETTING_DEFAULT(EStatsMode, StatsMode, EStatsMode::None);
+    FLUENT_SETTING_DEFAULT(bool, CollectAffectedRows, false);
     FLUENT_SETTING_OPTIONAL(bool, ConcurrentResultSets);
     FLUENT_SETTING(std::string, ResourcePool);
     FLUENT_SETTING_OPTIONAL(std::chrono::milliseconds, StatsCollectPeriod);
diff --git a/include/ydb-cpp-sdk/client/query/stats.h b/include/ydb-cpp-sdk/client/query/stats.h
index a1c944eb710..91b5b54390b 100644
--- a/include/ydb-cpp-sdk/client/query/stats.h
+++ b/include/ydb-cpp-sdk/client/query/stats.h
@@ -42,6 +42,7 @@ class TTableAccessStats {
     const TOperationStats& GetUpdates() const;
     const TOperationStats& GetDeletes() const;
     uint64_t GetPartitionsCount() const;
+    std::optional<uint64_t> GetAffectedRows() const;
 
 private:
     std::string Name_;
@@ -49,6 +50,7 @@ class TTableAccessStats {
     TOperationStats Updates_;
     TOperationStats Deletes_;
     uint64_t PartitionsCount_ = 0;
+    std::optional<uint64_t> AffectedRows_;
 };
 
 class TQueryPhaseStats {
diff --git a/include/ydb-cpp-sdk/client/table/table.h b/include/ydb-cpp-sdk/client/table/table.h
index f6c39b6b8eb..877df18de35 100644
--- a/include/ydb-cpp-sdk/client/table/table.h
+++ b/include/ydb-cpp-sdk/client/table/table.h
@@ -407,9 +407,11 @@ struct TFulltextIndexSettings {
         std::optional<int32_t> FilterLengthMin;
         std::optional<int32_t> FilterLengthMax;
         std::optional<bool> UseFilterSnowball;
+        std::optional<bool> UseFilterSuperLemmer;
 
         static TAnalyzers Standard();
         static TAnalyzers Snowball(std::string language);
+        static TAnalyzers SuperLemmer(std::string language);
         static TAnalyzers Keyword();
     };
 
@@ -1044,6 +1046,7 @@ enum class EStoreType {
 enum class EMultiColumnStatisticsType {
     Unknown = 0,
     CountMinSketch = 1,
+    EqHeightHistogram = 2,
 };
 
 //! Represents multi-column table statistics description
diff --git a/library/cpp/http/io/stream.cpp b/library/cpp/http/io/stream.cpp
index e4a00b54d9e..a97721c722a 100644
--- a/library/cpp/http/io/stream.cpp
+++ b/library/cpp/http/io/stream.cpp
@@ -134,10 +134,12 @@ class THttpInput::TImpl {
     typedef THashSet<TString> TAcceptCodings;
 
 public:
-    inline TImpl(IInputStream* slave)
+    inline TImpl(IInputStream* slave, const THttpInput::TOptions& options = {})
         : Slave_(slave)
+        , Options_(options)
         , Buffered_(Slave_, SuggestBufferSize())
         , ChunkedInput_(nullptr)
+        , LengthLimitedInput_(nullptr)
         , Input_(nullptr)
         , FirstLine_(ReadFirstLine(Buffered_))
         , Headers_(&Buffered_)
@@ -208,12 +210,22 @@ class THttpInput::TImpl {
         return Expect100Continue_;
     }
 
+    inline ui64 ContentLengthLeft() const noexcept {
+        return LengthLimitedInput_ ? LengthLimitedInput_->Left() : 0;
+    }
+
 private:
     template <class Operation>
     inline size_t Perform(size_t len, const Operation& operation) {
         size_t processed = operation(len);
         if (processed == 0 && len > 0) {
             if (!ChunkedInput_) {
+                if (Options_.StrictContentLength) {
+                    if (const ui64 left = ContentLengthLeft()) {
+                        ythrow THttpTruncatedBodyException() << "Body ended after " << (ContentLength_ - left)
+                                                            << " of " << ContentLength_ << " byte(s) declared in Content-Length";
+                    }
+                }
                 Trailers_.ConstructInPlace();
             } else {
                 // Read the header of the trailing chunk. It remains in
@@ -345,7 +357,8 @@ class THttpInput::TImpl {
                 /*
                  * TODO - we have other cases
                  */
-                Input_ = Streams_.Add(new TLengthLimitedInput(Input_, ContentLength_));
+                LengthLimitedInput_ = Streams_.Add(new TLengthLimitedInput(Input_, ContentLength_));
+                Input_ = LengthLimitedInput_;
             }
         }
 
@@ -359,6 +372,7 @@ class THttpInput::TImpl {
 
 private:
     IInputStream* Slave_;
+    THttpInput::TOptions Options_;
 
     /*
      * input helpers
@@ -366,6 +380,7 @@ class THttpInput::TImpl {
     TBufferedInput Buffered_;
     TStreams<IInputStream, 8> Streams_;
     IInputStream* ChunkedInput_;
+    TLengthLimitedInput* LengthLimitedInput_;
 
     /*
      * final input stream
@@ -391,6 +406,11 @@ THttpInput::THttpInput(IInputStream* slave)
 {
 }
 
+THttpInput::THttpInput(IInputStream* slave, const TOptions& options)
+    : Impl_(new TImpl(slave, options))
+{
+}
+
 THttpInput::THttpInput(THttpInput&& httpInput) = default;
 
 THttpInput::~THttpInput() {
@@ -445,6 +465,10 @@ bool THttpInput::ContentEncoded() const noexcept {
     return Impl_->ContentEncoded();
 }
 
+ui64 THttpInput::ContentLengthLeft() const noexcept {
+    return Impl_->ContentLengthLeft();
+}
+
 bool THttpInput::HasContent() const noexcept {
     return Impl_->HasContent();
 }
diff --git a/library/cpp/http/io/stream.h b/library/cpp/http/io/stream.h
index 78ca4fc814c..5ea5e18270e 100644
--- a/library/cpp/http/io/stream.h
+++ b/library/cpp/http/io/stream.h
@@ -21,10 +21,24 @@ struct THttpParseException: public THttpException {
 struct THttpReadException: public THttpException {
 };
 
+// Body ended before Content-Length. Thrown only under TOptions::StrictContentLength.
+struct THttpTruncatedBodyException: public THttpReadException {
+};
+
 /// Чтение ответа HTTP-сервера.
 class THttpInput: public IInputStream {
 public:
+    struct TOptions {
+        // If Content-Length is present, throw THttpTruncatedBodyException once the underlying
+        // stream reaches EOF with fewer bytes read than announced. A caller that stops reading
+        // early and destroys the stream does not trigger it.
+        // Do not enable for HEAD responses: they announce Content-Length but carry no body,
+        // which is indistinguishable from truncation at this level.
+        bool StrictContentLength = false;
+    };
+
     THttpInput(IInputStream* slave);
+    THttpInput(IInputStream* slave, const TOptions& options);
     THttpInput(THttpInput&& httpInput);
     ~THttpInput() override;
 
@@ -81,6 +95,11 @@ class THttpInput: public IInputStream {
     /// показывает объём запакованных данных, а из THttpInput мы будем вычитывать уже распакованные.
     bool ContentEncoded() const noexcept;
 
+    /// Сколько байт из заявленных в Content-Length ещё не вычитано из тела (до распаковки).
+    /// Всегда 0, если Content-Length в ответе нет или используется chunked encoding.
+    /// После полного вычитывания тела ненулевое значение означает обрыв ответа.
+    ui64 ContentLengthLeft() const noexcept;
+
     /// Returns true if Content-Length or Transfer-Encoding header received
     bool HasContent() const noexcept;
 
diff --git a/library/cpp/http/io/stream_ut.cpp b/library/cpp/http/io/stream_ut.cpp
index 02401fffd9d..9eb59acab74 100644
--- a/library/cpp/http/io/stream_ut.cpp
+++ b/library/cpp/http/io/stream_ut.cpp
@@ -537,6 +537,208 @@ Y_UNIT_TEST_SUITE(THttpStreamTest) {
         UNIT_ASSERT_VALUES_EQUAL(trailers.GetRef().Count(), 0);
     }
 
+    // A response whose body stops short of its Content-Length. TLengthLimitedInput reports
+    // socket EOF by returning 0 without checking that Content-Length was reached, so by
+    // default this is indistinguishable from a complete body. Existing callers depend on
+    // that, hence the opt-in flag.
+    Y_UNIT_TEST(TruncatedBodyIsToleratedByDefault) {
+        TMemoryInput response(
+            "HTTP/1.1 200 OK\r\n"
+            "Content-Length: 10\r\n"
+            "\r\n"
+            "bar");
+        THttpInput i(&response);
+        UNIT_ASSERT_VALUES_EQUAL(i.ReadAll(), "bar");
+        UNIT_ASSERT_VALUES_EQUAL(i.ContentLengthLeft(), 7u);
+    }
+
+    Y_UNIT_TEST(TruncatedBodyThrowsWithStrictContentLength) {
+        TMemoryInput response(
+            "HTTP/1.1 200 OK\r\n"
+            "Content-Length: 10\r\n"
+            "\r\n"
+            "bar");
+        THttpInput i(&response, {.StrictContentLength = true});
+        UNIT_ASSERT_EXCEPTION_CONTAINS(i.ReadAll(), THttpTruncatedBodyException, "declared in Content-Length");
+    }
+
+    Y_UNIT_TEST(CompleteBodyPassesStrictContentLength) {
+        TMemoryInput response(
+            "HTTP/1.1 200 OK\r\n"
+            "Content-Length: 3\r\n"
+            "\r\n"
+            "bar");
+        THttpInput i(&response, {.StrictContentLength = true});
+        UNIT_ASSERT_VALUES_EQUAL(i.ReadAll(), "bar");
+        UNIT_ASSERT_VALUES_EQUAL(i.ContentLengthLeft(), 0u);
+    }
+
+    static TString GzipBody() {
+        TString body;
+        for (size_t i = 0; i < 512; ++i) {
+            body += "the quick brown fox jumps over the lazy dog ";
+        }
+        return body;
+    }
+
+    // Content-Length always advertises the whole encoded body, even when only half of it is
+    // actually sent: that is what a connection dying mid-response looks like on the wire.
+    static TString GzipHttpResponse(TStringBuf body, bool truncate) {
+        TString compressed;
+        {
+            TStringOutput so(compressed);
+            TZLibCompress c(&so, ZLib::GZip);
+            c << body;
+            c.Finish();
+        }
+
+        const TString contentLength = ToString(compressed.size());
+        if (truncate) {
+            compressed.resize(compressed.size() / 2);
+        }
+
+        return TString::Join(
+            "HTTP/1.1 200 OK\r\n"
+            "Content-Encoding: gzip\r\n"
+            "Content-Length: ", contentLength, "\r\n"
+            "\r\n",
+            compressed);
+    }
+
+    // Content-Length measures the encoded body and TLengthLimitedInput sits below the
+    // decoder, so both are in the same units and the check covers encoded bodies too.
+    // TZLibDecompress reads until its slave EOFs rather than stopping at Z_STREAM_END, so a
+    // complete body leaves nothing behind.
+    Y_UNIT_TEST(CompleteContentEncodedBodyPassesStrictContentLength) {
+        const TString body = GzipBody();
+        const TString response = GzipHttpResponse(body, false);
+        TMemoryInput in(response.data(), response.size());
+        THttpInput i(&in, {.StrictContentLength = true});
+        UNIT_ASSERT(i.ContentEncoded());
+        UNIT_ASSERT_VALUES_EQUAL(i.ReadAll(), body);
+        // Proves the decoder drained TLengthLimitedInput, which is what makes the check
+        // meaningful for encoded bodies.
+        UNIT_ASSERT_VALUES_EQUAL(i.ContentLengthLeft(), 0u);
+    }
+
+    // A truncated gzip body is the worst case: zlib does not complain, it just stops
+    // producing output once its input runs dry, so without the check this yields a short
+    // body and reports success.
+    Y_UNIT_TEST(TruncatedContentEncodedBodyIsToleratedByDefault) {
+        const TString body = GzipBody();
+        const TString response = GzipHttpResponse(body, true);
+        TMemoryInput in(response.data(), response.size());
+        THttpInput i(&in);
+        UNIT_ASSERT(i.ReadAll().size() < body.size());
+        UNIT_ASSERT(i.ContentLengthLeft() > 0);
+    }
+
+    Y_UNIT_TEST(TruncatedContentEncodedBodyThrowsWithStrictContentLength) {
+        const TString response = GzipHttpResponse(GzipBody(), true);
+        TMemoryInput in(response.data(), response.size());
+        THttpInput i(&in, {.StrictContentLength = true});
+        UNIT_ASSERT_EXCEPTION_CONTAINS(i.ReadAll(), THttpTruncatedBodyException, "declared in Content-Length");
+    }
+
+    // Everything below pins the "must not throw" half of the contract: strict mode may only
+    // fire on a genuinely short Content-Length body.
+
+    Y_UNIT_TEST(ChunkedBodyIgnoresStrictContentLength) {
+        TMemoryInput response(
+            "HTTP/1.1 200 OK\r\n"
+            "Transfer-Encoding: chunked\r\n"
+            "\r\n"
+            "3\r\nbar\r\n"
+            "0\r\n"
+            "\r\n");
+        THttpInput i(&response, {.StrictContentLength = true});
+        UNIT_ASSERT_VALUES_EQUAL(i.ReadAll(), "bar");
+        UNIT_ASSERT_VALUES_EQUAL(i.ContentLengthLeft(), 0u);
+    }
+
+    // Body framed by connection close: no Content-Length, so nothing to verify.
+    Y_UNIT_TEST(NoContentLengthIgnoresStrictContentLength) {
+        TMemoryInput response(
+            "HTTP/1.1 200 OK\r\n"
+            "Connection: close\r\n"
+            "\r\n"
+            "bar");
+        THttpInput i(&response, {.StrictContentLength = true});
+        UNIT_ASSERT_VALUES_EQUAL(i.ReadAll(), "bar");
+        UNIT_ASSERT_VALUES_EQUAL(i.ContentLengthLeft(), 0u);
+    }
+
+    Y_UNIT_TEST(ZeroContentLengthPassesStrictContentLength) {
+        TMemoryInput response(
+            "HTTP/1.1 200 OK\r\n"
+            "Content-Length: 0\r\n"
+            "\r\n");
+        THttpInput i(&response, {.StrictContentLength = true});
+        UNIT_ASSERT_VALUES_EQUAL(i.ReadAll(), "");
+        UNIT_ASSERT_VALUES_EQUAL(i.ContentLengthLeft(), 0u);
+    }
+
+    // Trailing garbage past Content-Length is a keep-alive framing concern, not truncation.
+    Y_UNIT_TEST(BodyLongerThanContentLengthPassesStrictContentLength) {
+        TMemoryInput response(
+            "HTTP/1.1 200 OK\r\n"
+            "Content-Length: 3\r\n"
+            "\r\n"
+            "barbaz");
+        THttpInput i(&response, {.StrictContentLength = true});
+        UNIT_ASSERT_VALUES_EQUAL(i.ReadAll(), "bar");
+        UNIT_ASSERT_VALUES_EQUAL(i.ContentLengthLeft(), 0u);
+    }
+
+    // Requests take the IsRequest() branch, which builds TLengthLimitedInput with length 0.
+    Y_UNIT_TEST(RequestWithoutContentLengthPassesStrictContentLength) {
+        TMemoryInput request(
+            "GET / HTTP/1.1\r\n"
+            "Host: yandex.ru\r\n"
+            "\r\n");
+        THttpInput i(&request, {.StrictContentLength = true});
+        UNIT_ASSERT_VALUES_EQUAL(i.ReadAll(), "");
+        UNIT_ASSERT_VALUES_EQUAL(i.ContentLengthLeft(), 0u);
+    }
+
+    // Strict mode reacts to EOF, not to the caller losing interest.
+    Y_UNIT_TEST(PartialReadThenDestroyPassesStrictContentLength) {
+        TMemoryInput response(
+            "HTTP/1.1 200 OK\r\n"
+            "Content-Length: 10\r\n"
+            "\r\n"
+            "0123456789");
+        char buf[4];
+        THttpInput i(&response, {.StrictContentLength = true});
+        UNIT_ASSERT_VALUES_EQUAL(i.Load(buf, sizeof(buf)), sizeof(buf));
+        UNIT_ASSERT_VALUES_EQUAL(i.ContentLengthLeft(), 6u);
+    }
+
+    // Skip() reaches the same Perform() as Read(). A short skip is not itself EOF -- the
+    // check only fires once a call comes back with nothing at all.
+    Y_UNIT_TEST(SkipDetectsTruncationWithStrictContentLength) {
+        TMemoryInput response(
+            "HTTP/1.1 200 OK\r\n"
+            "Content-Length: 10\r\n"
+            "\r\n"
+            "bar");
+        THttpInput i(&response, {.StrictContentLength = true});
+        UNIT_ASSERT_VALUES_EQUAL(i.Skip(10), 3u);
+        UNIT_ASSERT_EXCEPTION_CONTAINS(i.Skip(7), THttpTruncatedBodyException, "declared in Content-Length");
+    }
+
+    // THttpInput does not know the request method, so a HEAD response -- Content-Length set,
+    // body legitimately absent -- looks exactly like truncation. Clients must not enable
+    // strict mode for HEAD; TKeepAliveHttpClient does that for us, see http_ut.cpp.
+    Y_UNIT_TEST(HeadResponseCannotBeDistinguishedFromTruncation) {
+        TMemoryInput response(
+            "HTTP/1.1 200 OK\r\n"
+            "Content-Length: 1024\r\n"
+            "\r\n");
+        THttpInput i(&response, {.StrictContentLength = true});
+        UNIT_ASSERT_EXCEPTION_CONTAINS(i.ReadAll(), THttpTruncatedBodyException, "declared in Content-Length");
+    }
+
     Y_UNIT_TEST(RequestWithoutContentLength) {
         TStringStream request;
         {
diff --git a/library/cpp/http/simple/http_client.cpp b/library/cpp/http/simple/http_client.cpp
index 87bbabc3f83..993ee62f76a 100644
--- a/library/cpp/http/simple/http_client.cpp
+++ b/library/cpp/http/simple/http_client.cpp
@@ -3,6 +3,7 @@
 #include <library/cpp/string_utils/url/url.h>
 
 #include <util/stream/output.h>
+#include <util/string/ascii.h>
 #include <util/string/cast.h>
 #include <util/string/join.h>
 #include <util/string/split.h>
@@ -18,13 +19,15 @@ TKeepAliveHttpClient::TKeepAliveHttpClient(const TString& host,
                                            TDuration socketTimeout,
                                            TDuration connectTimeout,
                                            bool useKeepAlive,
-                                           bool useConnectionPool)
+                                           bool useConnectionPool,
+                                           bool strictContentLength)
     : Host(CutHttpPrefix(host))
     , Port(port)
     , SocketTimeout(socketTimeout)
     , ConnectTimeout(connectTimeout)
     , UseKeepAlive(useKeepAlive)
     , UseConnectionPool(useConnectionPool)
+    , StrictContentLength(strictContentLength)
     , IsHttps(host.StartsWith("https"))
     , IsClosingRequired(false)
     , HttpsVerification(TVerifyCert{Host})
@@ -69,14 +72,22 @@ TKeepAliveHttpClient::THttpCode TKeepAliveHttpClient::DoRequest(const TStringBuf
                                                                 THttpHeaders* outHeaders,
                                                                 NThreading::TCancellationToken cancellation) {
     const TString contentLength = IntToString<10, size_t>(body.size());
-    return DoRequestReliable(FormRequest(method, relativeUrl, body, inHeaders, contentLength), output, outHeaders, std::move(cancellation));
+    return DoRequestReliable(FormRequest(method, relativeUrl, body, inHeaders, contentLength),
+                             output,
+                             outHeaders,
+                             std::move(cancellation),
+                             ResponseBodyExpected(method));
 }
 
 TKeepAliveHttpClient::THttpCode TKeepAliveHttpClient::DoRequestRaw(const TStringBuf raw,
                                                                    IOutputStream* output,
                                                                    THttpHeaders* outHeaders,
                                                                    NThreading::TCancellationToken cancellation) {
-    return DoRequestReliable(raw, output, outHeaders, std::move(cancellation));
+    return DoRequestReliable(raw, output, outHeaders, std::move(cancellation), ResponseBodyExpected(raw.Before(' ')));
+}
+
+bool TKeepAliveHttpClient::ResponseBodyExpected(TStringBuf method) {
+    return !AsciiEqualsIgnoreCase(method, TStringBuf("HEAD"));
 }
 
 void TKeepAliveHttpClient::DisableVerificationForHttps() {
@@ -129,7 +140,8 @@ TVector<IOutputStream::TPart> TKeepAliveHttpClient::FormRequest(TStringBuf metho
 
 TKeepAliveHttpClient::THttpCode TKeepAliveHttpClient::ReadAndTransferHttp(THttpInput& input,
                                                                           IOutputStream* output,
-                                                                          THttpHeaders* outHeaders) const {
+                                                                          THttpHeaders* outHeaders,
+                                                                          bool responseBodyExpected) const {
     TKeepAliveHttpClient::THttpCode statusCode;
     try {
         statusCode = ParseHttpRetCode(input.FirstLine());
@@ -140,8 +152,8 @@ TKeepAliveHttpClient::THttpCode TKeepAliveHttpClient::ReadAndTransferHttp(THttpI
                                        << rest;
     }
 
-    auto canContainBody = [](auto statusCode) {
-        return statusCode != HTTP_NOT_MODIFIED && statusCode != HTTP_NO_CONTENT;
+    auto canContainBody = [responseBodyExpected](auto statusCode) {
+        return responseBodyExpected && statusCode != HTTP_NOT_MODIFIED && statusCode != HTTP_NO_CONTENT;
     };
 
     if (output && canContainBody(statusCode) && IfResponseRequired(input)) {
@@ -170,7 +182,8 @@ bool TKeepAliveHttpClient::CreateNewConnectionIfNeeded() {
                                                            IsHttps,
                                                            ClientCertificate,
                                                            HttpsVerification,
-                                                           UseKeepAlive);
+                                                           UseKeepAlive,
+                                                           StrictContentLength);
         IsClosingRequired = false;
         return true;
     }
@@ -208,6 +221,7 @@ TSimpleHttpClient::TSimpleHttpClient(const TOptions& options)
     , ConnectTimeout(options.ConnectTimeout())
     , UseKeepAlive(options.UseKeepAlive())
     , UseConnectionPool(options.UseConnectionPool())
+    , StrictContentLength(options.StrictContentLength())
 {
 }
 
@@ -258,8 +272,10 @@ namespace NPrivate {
                                      bool isHttps,
                                      const TMaybe<TOpenSslClientIO::TOptions::TClientCert>& clientCert,
                                      const TMaybe<TOpenSslClientIO::TOptions::TVerifyCert>& verifyCert,
-                                     bool keepAlive)
-        : Addr(Resolve(host, port))
+                                     bool keepAlive,
+                                     bool strictContentLength)
+        : StrictContentLength(strictContentLength)
+        , Addr(Resolve(host, port))
         , Socket(Connect(Addr, sockTimeout, connTimeout, host, port))
         , SocketIn(Socket)
         , SocketOut(Socket)
@@ -312,7 +328,7 @@ namespace NPrivate {
 
 void TSimpleHttpClient::ProcessResponse(const TStringBuf relativeUrl, THttpInput& input, IOutputStream*, const unsigned statusCode) const {
     if (!(statusCode >= 200 && statusCode < 300)) {
-        TString rest = input.ReadAll();
+        TString rest = ReadDiagnosticBody(input, statusCode);
         ythrow THttpRequestException(statusCode) << "Got " << statusCode << " at " << Host << relativeUrl << "\nFull http response:\n"
                                                  << rest;
     }
@@ -322,7 +338,7 @@ TSimpleHttpClient::~TSimpleHttpClient() {
 }
 
 TKeepAliveHttpClient TSimpleHttpClient::CreateClient() const {
-    TKeepAliveHttpClient cl(Host, Port, SocketTimeout, ConnectTimeout, UseKeepAlive, UseConnectionPool);
+    TKeepAliveHttpClient cl(Host, Port, SocketTimeout, ConnectTimeout, UseKeepAlive, UseConnectionPool, StrictContentLength);
 
     if (!HttpsVerification) {
         cl.DisableVerificationForHttps();
@@ -336,6 +352,13 @@ TKeepAliveHttpClient TSimpleHttpClient::CreateClient() const {
 void TSimpleHttpClient::PrepareClient(TKeepAliveHttpClient&) const {
 }
 
+TString TSimpleHttpClient::ReadDiagnosticBody(THttpInput& input, unsigned statusCode) {
+    if (statusCode == HTTP_NOT_MODIFIED || statusCode == HTTP_NO_CONTENT) {
+        return {}; // no body to read, and Content-Length may still be set
+    }
+    return input.ReadAll();
+}
+
 TRedirectableHttpClient::TRedirectableHttpClient(const TOptions& options)
     : TSimpleHttpClient(options)
     , Opts(options)
@@ -390,7 +413,7 @@ void TRedirectableHttpClient::ProcessResponse(const TStringBuf relativeUrl, THtt
         }
     }
     if (!(statusCode >= 200 && statusCode < 300)) {
-        TString rest = input.ReadAll();
+        TString rest = ReadDiagnosticBody(input, statusCode);
         ythrow THttpRequestException(statusCode) << "Got " << statusCode << " at " << Host << relativeUrl << "\nFull http response:\n"
                                                  << rest;
     }
diff --git a/library/cpp/http/simple/http_client.h b/library/cpp/http/simple/http_client.h
index d208e0ae055..3e05aeae612 100644
--- a/library/cpp/http/simple/http_client.h
+++ b/library/cpp/http/simple/http_client.h
@@ -54,7 +54,8 @@ class TKeepAliveHttpClient {
                          TDuration socketTimeout = TDuration::Seconds(5),
                          TDuration connectTimeout = TDuration::Seconds(30),
                          bool useKeepAlive = true,
-                         bool useConnectionPool = false);
+                         bool useConnectionPool = false,
+                         bool strictContentLength = false);
 
     TKeepAliveHttpClient(TKeepAliveHttpClient&&) = default;
     ~TKeepAliveHttpClient();
@@ -106,13 +107,17 @@ class TKeepAliveHttpClient {
     THttpCode DoRequestReliable(const T& raw,
                                 IOutputStream* output,
                                 THttpHeaders* outHeaders,
-                                NThreading::TCancellationToken cancellation);
+                                NThreading::TCancellationToken cancellation,
+                                bool responseBodyExpected = true);
+
+    // False for HEAD, whose answer announces Content-Length but carries no body.
+    static bool ResponseBodyExpected(TStringBuf method);
 
     TVector<IOutputStream::TPart> FormRequest(TStringBuf method, const TStringBuf relativeUrl,
                                               TStringBuf body,
                                               const THeaders& headers, TStringBuf contentLength) const;
 
-    THttpCode ReadAndTransferHttp(THttpInput& input, IOutputStream* output, THttpHeaders* outHeaders) const;
+    THttpCode ReadAndTransferHttp(THttpInput& input, IOutputStream* output, THttpHeaders* outHeaders, bool responseBodyExpected) const;
 
     bool CreateNewConnectionIfNeeded(); // Returns true if now we have a new connection.
 
@@ -126,6 +131,7 @@ class TKeepAliveHttpClient {
     const TDuration ConnectTimeout;
     const bool UseKeepAlive;
     const bool UseConnectionPool;
+    const bool StrictContentLength;
     const bool IsHttps;
 
     static TSpinLock ConnectionQuarantineMutex;
@@ -172,6 +178,7 @@ class TSimpleHttpClient {
     const TDuration ConnectTimeout;
     const bool UseKeepAlive = true;
     const bool UseConnectionPool = false;
+    const bool StrictContentLength = false;
     bool HttpsVerification = false;
 
 public:
@@ -196,6 +203,11 @@ class TSimpleHttpClient {
 
     virtual ~TSimpleHttpClient();
 
+protected:
+    // 304 and 204 carry no body even when Content-Length is set, so reading one for the
+    // error message would hide the status code behind a read failure.
+    static TString ReadDiagnosticBody(THttpInput& input, unsigned statusCode);
+
 private:
     TKeepAliveHttpClient CreateClient() const;
 
@@ -230,7 +242,8 @@ namespace NPrivate {
                         bool isHttps,
                         const TMaybe<TOpenSslClientIO::TOptions::TClientCert>& clientCert,
                         const TMaybe<TOpenSslClientIO::TOptions::TVerifyCert>& verifyCert,
-                        bool keepAlive = true);
+                        bool keepAlive = true,
+                        bool strictContentLength = false);
 
         bool IsOk() const {
             return IsNotSocketClosedByOtherSide(Socket);
@@ -239,8 +252,9 @@ namespace NPrivate {
         template <typename TContainer>
         void Write(const TContainer& request) {
             HttpOut->Write(request.data(), request.size());
-            HttpIn = Ssl ? MakeHolder<THttpInput>(Ssl.Get())
-                         : MakeHolder<THttpInput>(&SocketIn);
+            const THttpInput::TOptions inOptions{.StrictContentLength = StrictContentLength};
+            HttpIn = Ssl ? MakeHolder<THttpInput>(Ssl.Get(), inOptions)
+                         : MakeHolder<THttpInput>(&SocketIn, inOptions);
             HttpOut->Flush();
         }
 
@@ -262,6 +276,7 @@ namespace NPrivate {
                                ui32 port);
 
     private:
+        const bool StrictContentLength;
         TNetworkAddress Addr;
         TSocket Socket;
         TSocketInput SocketIn;
@@ -276,7 +291,8 @@ template <class T>
 TKeepAliveHttpClient::THttpCode TKeepAliveHttpClient::DoRequestReliable(const T& raw,
                                                                         IOutputStream* output,
                                                                         THttpHeaders* outHeaders,
-                                                                        NThreading::TCancellationToken cancellation) {
+                                                                        NThreading::TCancellationToken cancellation,
+                                                                        bool responseBodyExpected) {
 
     for (int i = 0; i < 2; ++i) {
         const bool haveNewConnection = CreateNewConnectionIfNeeded();
@@ -295,7 +311,7 @@ TKeepAliveHttpClient::THttpCode TKeepAliveHttpClient::DoRequestReliable(const T&
         try {
             Connection->Write(raw);
 
-            THttpCode code = ReadAndTransferHttp(*Connection->GetHttpInput(), output, outHeaders);
+            THttpCode code = ReadAndTransferHttp(*Connection->GetHttpInput(), output, outHeaders, responseBodyExpected);
             if (!Connection->GetHttpInput()->IsKeepAlive()) {
                 IsClosingRequired = true;
             }
@@ -309,6 +325,15 @@ TKeepAliveHttpClient::THttpCode TKeepAliveHttpClient::DoRequestReliable(const T&
             if (!couldRetry || e.Status() != EPIPE) {
                 throw;
             }
+        } catch (const THttpTruncatedBodyException&) {
+            // Must precede THttpReadException: part of the body already reached `output`,
+            // so a retry would append a second copy instead of reporting the failure.
+            if (cancellation.IsCancellationRequested()) {
+                cancellationEndEvent->WaitI();
+                cancellation.ThrowIfCancellationRequested();
+            }
+            Connection.Reset();
+            throw;
         } catch (const THttpReadException&) { // Actually old connection is already closed by server
             if (cancellation.IsCancellationRequested()) {
                 cancellationEndEvent->WaitI();
diff --git a/library/cpp/http/simple/http_client_options.h b/library/cpp/http/simple/http_client_options.h
index f237a3770a9..6242b0958b0 100644
--- a/library/cpp/http/simple/http_client_options.h
+++ b/library/cpp/http/simple/http_client_options.h
@@ -79,6 +79,18 @@ class TSimpleHttpClientOptions {
         return UseConnectionPool_;
     }
 
+    /// Fail with THttpTruncatedBodyException when a response body ends before its
+    /// Content-Length instead of silently yielding a short body. Off by default.
+    /// HEAD is handled for you; see THttpInput::TOptions for the raw-stream caveats.
+    TSelf& StrictContentLength(bool strictContentLength) {
+        StrictContentLength_ = strictContentLength;
+        return *this;
+    }
+
+    bool StrictContentLength() const noexcept {
+        return StrictContentLength_;
+    }
+
 private:
     TString Host_;
     ui16 Port_;
@@ -87,4 +99,5 @@ class TSimpleHttpClientOptions {
     int MaxRedirectCount_ = INT_MAX;
     bool UseKeepAlive_ = true;
     bool UseConnectionPool_ = false;
+    bool StrictContentLength_ = false;
 };
diff --git a/library/cpp/http/simple/ut/http_ut.cpp b/library/cpp/http/simple/ut/http_ut.cpp
index e175f1e795c..927becf8da0 100644
--- a/library/cpp/http/simple/ut/http_ut.cpp
+++ b/library/cpp/http/simple/ut/http_ut.cpp
@@ -9,9 +9,12 @@
 #include <library/cpp/threading/future/async.h>
 #include <util/thread/pool.h>
 
+#include <util/network/sock.h>
+#include <util/network/socket.h>
 #include <util/system/event.h>
 #include <util/system/thread.h>
 
+#include <atomic>
 #include <thread>
 
 Y_UNIT_TEST_SUITE(SimpleHttp) {
@@ -613,4 +616,222 @@ Y_UNIT_TEST_SUITE(SimpleHttp) {
         Sleep(TDuration::MilliSeconds(500));
         UNIT_ASSERT_NO_EXCEPTION(cl.DoGet("/ping"));
     }
+
+    // Byte-exact server. THttpServer frames replies through THttpOutput and therefore cannot
+    // emit a body shorter than the Content-Length it announces, which is the case under test.
+    // One thread per connection: a redirect keeps the original connection open while it
+    // fetches the target, so serving them one at a time would deadlock.
+    class TRawHttpServer {
+    public:
+        struct TReply {
+            TString Bytes;
+            bool CloseAfter = false;
+        };
+
+        TRawHttpServer(ui16 port, TVector<TReply> script)
+            : Script_(std::move(script))
+            , Port_(port)
+        {
+            CheckedSetSockOpt((SOCKET)Listener_, SOL_SOCKET, SO_REUSEADDR, 1, "TRawHttpServer");
+            TSockAddrInet addr("127.0.0.1", port);
+            TBaseSocket::Check(Listener_.Bind(&addr), "bind");
+            TBaseSocket::Check(Listener_.Listen(4), "listen");
+            Acceptor_ = std::thread([this] { Accept(); });
+        }
+
+        ~TRawHttpServer() {
+            Stop_.store(true);
+            WakeUpAcceptor();
+            Acceptor_.join();
+            for (auto& conn : Connections_) {
+                conn.join();
+            }
+        }
+
+        size_t Served() const {
+            return Served_.load();
+        }
+
+    private:
+        static constexpr long PollMs = 100;
+
+        // Accept() blocks; a self-connect lands in the backlog and releases it.
+        void WakeUpAcceptor() {
+            TInetStreamSocket waker;
+            TSockAddrInet addr("127.0.0.1", Port_);
+            waker.Connect(&addr);
+        }
+
+        void Accept() {
+            while (!Stop_.load()) {
+                auto conn = MakeAtomicShared<TStreamSocket>();
+                if (Listener_.Accept(conn.Get()) < 0 || Stop_.load()) {
+                    return;
+                }
+                SetSocketTimeout((SOCKET)*conn, 0, PollMs);
+                Connections_.emplace_back([this, conn] { Serve(*conn); });
+            }
+        }
+
+        void Serve(TStreamSocket& conn) {
+            while (ReadRequest(conn)) {
+                const size_t idx = Served_.fetch_add(1);
+                if (idx >= Script_.size()) {
+                    return;
+                }
+
+                const TReply& reply = Script_[idx];
+                conn.Send(reply.Bytes.data(), reply.Bytes.size());
+                if (reply.CloseAfter) {
+                    try {
+                        conn.ShutDown(SHUT_WR); // clean FIN mid-body
+                    } catch (const TSystemError&) {
+                        // peer hung up first, nothing left to half-close
+                    }
+                    return;
+                }
+            }
+        }
+
+        // The receive timeout keeps this responsive to Stop_ on an idle connection.
+        bool ReadRequest(TStreamSocket& conn) {
+            TString head;
+            char c = 0;
+            while (!head.EndsWith("\r\n\r\n")) {
+                if (Stop_.load()) {
+                    return false;
+                }
+
+                const ssize_t got = conn.Recv(&c, 1);
+                if (got == 1) {
+                    head += c;
+                } else if (got == -EAGAIN || got == -EWOULDBLOCK) {
+                    continue;
+                } else {
+                    return false; // peer closed, or a real error
+                }
+            }
+            return true;
+        }
+
+    private:
+        TVector<TReply> Script_;
+        ui16 Port_ = 0;
+        TInetStreamSocket Listener_;
+        std::atomic<size_t> Served_{0};
+        std::atomic<bool> Stop_{false};
+        std::thread Acceptor_;
+        TVector<std::thread> Connections_;
+    };
+
+    static TString Truncated(size_t announced, TStringBuf sent) {
+        return TStringBuilder() << "HTTP/1.1 200 OK\r\nContent-Length: " << announced
+                                << "\r\nConnection: Keep-Alive\r\n\r\n" << sent;
+    }
+
+    Y_UNIT_TEST(truncatedBodyIsToleratedByDefault) {
+        TPortManager pm;
+        ui16 port = pm.GetPort(80);
+        TRawHttpServer server(port, {{Truncated(100, "truncated"), true}});
+
+        TSimpleHttpClient cl("127.0.0.1", port);
+
+        TStringStream s;
+        UNIT_ASSERT_NO_EXCEPTION(cl.DoGet("/ping", &s));
+        UNIT_ASSERT_VALUES_EQUAL("truncated", s.Str());
+    }
+
+    Y_UNIT_TEST(strictContentLengthViaOptions) {
+        TPortManager pm;
+        ui16 port = pm.GetPort(80);
+        TRawHttpServer server(port, {{Truncated(100, "truncated"), true}});
+
+        TSimpleHttpClient cl(TSimpleHttpClientOptions().Host("127.0.0.1").Port(port).StrictContentLength(true));
+
+        TStringStream s;
+        UNIT_ASSERT_EXCEPTION(cl.DoGet("/ping", &s), THttpTruncatedBodyException);
+    }
+
+    Y_UNIT_TEST(strictContentLengthSurvivesRedirect) {
+        TPortManager pm;
+        ui16 port = pm.GetPort(80);
+        TRawHttpServer server(port, {
+                                        {TStringBuilder() << "HTTP/1.1 302 Found\r\nLocation: http://127.0.0.1:" << port
+                                                          << "/ping2\r\nContent-Length: 0\r\nConnection: Keep-Alive\r\n\r\n",
+                                         false},
+                                        {Truncated(100, "truncated"), true},
+                                    });
+
+        TRedirectableHttpClient cl(TSimpleHttpClientOptions().Host("127.0.0.1").Port(port).StrictContentLength(true));
+
+        TStringStream s;
+        UNIT_ASSERT_EXCEPTION(cl.DoGet("/ping", &s), THttpTruncatedBodyException);
+    }
+
+    // A truncation on a reused connection must not be mistaken for the stale-connection case
+    // that the THttpReadException handler retries: the body is already partly in `output`.
+    Y_UNIT_TEST(strictContentLengthDoesNotRetryTruncatedBody) {
+        TPortManager pm;
+        ui16 port = pm.GetPort(80);
+        TRawHttpServer server(port, {
+                                        {"HTTP/1.1 200 OK\r\nContent-Length: 4\r\nConnection: Keep-Alive\r\n\r\nfull", false},
+                                        {Truncated(100, "truncated"), true},
+                                        {"HTTP/1.1 200 OK\r\nContent-Length: 5\r\nConnection: Keep-Alive\r\n\r\nAGAIN", false},
+                                    });
+
+        TKeepAliveHttpClient cl("127.0.0.1", port, TDuration::Seconds(5), TDuration::Seconds(30), true, false, true);
+
+        {
+            TStringStream s;
+            UNIT_ASSERT_VALUES_EQUAL(200u, cl.DoGet("/ping", &s));
+            UNIT_ASSERT_VALUES_EQUAL("full", s.Str());
+        }
+        {
+            TStringStream s;
+            UNIT_ASSERT_EXCEPTION(cl.DoGet("/ping", &s), THttpTruncatedBodyException);
+            UNIT_ASSERT_VALUES_EQUAL("truncated", s.Str());
+        }
+
+        UNIT_ASSERT_VALUES_EQUAL(2u, server.Served());
+    }
+
+    // HEAD answers carry Content-Length with no body; strict mode must not read it as truncation.
+    Y_UNIT_TEST(strictContentLengthAllowsHeadResponse) {
+        TPortManager pm;
+        ui16 port = pm.GetPort(80);
+        TRawHttpServer server(port, {{"HTTP/1.1 200 OK\r\nContent-Length: 1024\r\nConnection: Keep-Alive\r\n\r\n", false}});
+
+        TKeepAliveHttpClient cl("127.0.0.1", port, TDuration::Seconds(5), TDuration::Seconds(30), true, false, true);
+
+        TStringStream s;
+        TKeepAliveHttpClient::THttpCode code = 0;
+        UNIT_ASSERT_NO_EXCEPTION(code = cl.DoRequest("HEAD", "/ping", "", &s));
+        UNIT_ASSERT_VALUES_EQUAL(200u, code);
+        UNIT_ASSERT_VALUES_EQUAL("", s.Str());
+    }
+
+    Y_UNIT_TEST(strictContentLengthAllowsRawHeadRequest) {
+        TPortManager pm;
+        ui16 port = pm.GetPort(80);
+        TRawHttpServer server(port, {{"HTTP/1.1 200 OK\r\nContent-Length: 1024\r\nConnection: Keep-Alive\r\n\r\n", false}});
+
+        TKeepAliveHttpClient cl("127.0.0.1", port, TDuration::Seconds(5), TDuration::Seconds(30), true, false, true);
+
+        TStringStream s;
+        const TString raw = "HEAD /ping HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Length: 0\r\n\r\n";
+        UNIT_ASSERT_NO_EXCEPTION(cl.DoRequestRaw(raw, &s));
+        UNIT_ASSERT_VALUES_EQUAL("", s.Str());
+    }
+
+    // 304 may carry Content-Length with no body; the status code must survive strict mode.
+    Y_UNIT_TEST(strictContentLengthKeepsStatusCodeForNotModified) {
+        TPortManager pm;
+        ui16 port = pm.GetPort(80);
+        TRawHttpServer server(port, {{"HTTP/1.1 304 Not Modified\r\nContent-Length: 42\r\nConnection: Keep-Alive\r\n\r\n", false}});
+
+        TSimpleHttpClient cl(TSimpleHttpClientOptions().Host("127.0.0.1").Port(port).StrictContentLength(true));
+
+        TStringStream s;
+        UNIT_ASSERT_EXCEPTION_CONTAINS(cl.DoGet("/ping", &s), THttpRequestException, "304");
+    }
 }
diff --git a/library/cpp/monlib/encode/json/json.h b/library/cpp/monlib/encode/json/json.h
index 21530f20c37..657b3d67fcc 100644
--- a/library/cpp/monlib/encode/json/json.h
+++ b/library/cpp/monlib/encode/json/json.h
@@ -8,6 +8,9 @@ class IOutputStream;
 
 namespace NMonitoring {
 
+    class TJsonEncodeError: public yexception {
+    };
+
     class TJsonDecodeError: public yexception {
     };
 
diff --git a/library/cpp/monlib/encode/json/json_encoder.cpp b/library/cpp/monlib/encode/json/json_encoder.cpp
index 4394685512c..0e450841eb5 100644
--- a/library/cpp/monlib/encode/json/json_encoder.cpp
+++ b/library/cpp/monlib/encode/json/json_encoder.cpp
@@ -170,15 +170,15 @@ namespace NMonitoring {
                         break;
 
                     case EMetricValueType::UNKNOWN:
-                        ythrow yexception() << "unknown metric value type";
+                        ythrow TJsonEncodeError() << "unknown metric value type";
                 }
             }
 
             void WriteLabel(TStringBuf name, TStringBuf value) {
                 if (!IsUtf(name)) {
-                    ythrow yexception() << "label name is not valid UTF-8 string: '" << EscapeC(name.SubStr(0, 100)) << "'";
+                    ythrow TJsonEncodeError() << "label name is not valid UTF-8 string: '" << EscapeC(name.SubStr(0, 100)) << "'";
                 } else if (!IsUtf(value)) {
-                    ythrow yexception() << "label value is not valid UTF-8 string, name: '" << name << "', value: '" << EscapeC(value.SubStr(0, 100)) << "'";
+                    ythrow TJsonEncodeError() << "label value is not valid UTF-8 string, name: '" << name << "', value: '" << EscapeC(value.SubStr(0, 100)) << "'";
                 }
 
                 if (Style_ == EJsonStyle::Cloud && name == MetricNameLabel_) {
@@ -204,7 +204,7 @@ namespace NMonitoring {
                     return;
                 }
                 if (CurrentMetricName_.empty()) {
-                    ythrow yexception() << "label '" << MetricNameLabel_ << "' is not defined";
+                    ythrow TJsonEncodeError() << "label '" << MetricNameLabel_ << "' is not defined";
                 }
                 Buf_.WriteKey("name");
                 Buf_.WriteString(CurrentMetricName_);
@@ -223,7 +223,7 @@ namespace NMonitoring {
                     case EMetricType::IGAUGE:
                         return TStringBuf("IGAUGE");
                     default:
-                        ythrow yexception() << "metric type '" << type << "' is not supported by cloud json format";
+                        ythrow TJsonEncodeError() << "metric type '" << type << "' is not supported by cloud json format";
                 }
             }
 
diff --git a/library/cpp/monlib/encode/json/json_ut.cpp b/library/cpp/monlib/encode/json/json_ut.cpp
index 78103a42b5c..bfe2c4774cb 100644
--- a/library/cpp/monlib/encode/json/json_ut.cpp
+++ b/library/cpp/monlib/encode/json/json_ut.cpp
@@ -295,7 +295,7 @@ Y_UNIT_TEST_SUITE(TJsonTest) {
             auto encoder = buffered ? BufferedEncoderCloudJson(&out, 2) : EncoderCloudJson(&out, 2);
             const TString expectedMessage = TStringBuilder()
                 << "metric type '" << metricType << "' is not supported by cloud json format";
-            UNIT_ASSERT_EXCEPTION_CONTAINS_C(emit(encoder.Get(), metricType), yexception, expectedMessage,
+            UNIT_ASSERT_EXCEPTION_CONTAINS_C(emit(encoder.Get(), metricType), TJsonEncodeError, expectedMessage,
                                              TString("buffered: ") + ToString(buffered));
         };
 
@@ -307,6 +307,60 @@ Y_UNIT_TEST_SUITE(TJsonTest) {
         doTest(true, EMetricType::DSUMMARY);
     }
 
+    Y_UNIT_TEST(InvalidUtfLabelsRaiseJsonEncodeError) {
+        const TString invalidUtf("\xff", 1);
+
+        auto doTest = [&](bool buffered, TStringBuf name, TStringBuf value) {
+            TString json;
+            TStringOutput out(json);
+            auto encoder = buffered ? BufferedEncoderCloudJson(&out) : EncoderCloudJson(&out);
+
+            encoder->OnStreamBegin();
+            encoder->OnMetricBegin(EMetricType::GAUGE);
+            encoder->OnLabelsBegin();
+            if (buffered) {
+                encoder->OnLabel(name, value);
+                encoder->OnLabelsEnd();
+                encoder->OnDouble(now, 1.0);
+                encoder->OnMetricEnd();
+                encoder->OnStreamEnd();
+                UNIT_ASSERT_EXCEPTION(encoder->Close(), TJsonEncodeError);
+            } else {
+                UNIT_ASSERT_EXCEPTION(encoder->OnLabel(name, value), TJsonEncodeError);
+            }
+        };
+
+        doTest(false, invalidUtf, "value");
+        doTest(false, "name", invalidUtf);
+        doTest(true, invalidUtf, "value");
+        doTest(true, "name", invalidUtf);
+    }
+
+    Y_UNIT_TEST(MissingMetricNameRaisesJsonEncodeError) {
+        auto doTest = [&](bool buffered) {
+            TString json;
+            TStringOutput out(json);
+            auto encoder = buffered ? BufferedEncoderCloudJson(&out) : EncoderCloudJson(&out);
+
+            encoder->OnStreamBegin();
+            encoder->OnMetricBegin(EMetricType::GAUGE);
+            encoder->OnLabelsBegin();
+            encoder->OnLabel("label", "value");
+            if (buffered) {
+                encoder->OnLabelsEnd();
+                encoder->OnDouble(now, 1.0);
+                encoder->OnMetricEnd();
+                encoder->OnStreamEnd();
+                UNIT_ASSERT_EXCEPTION_CONTAINS(encoder->Close(), TJsonEncodeError, "label 'name' is not defined");
+            } else {
+                UNIT_ASSERT_EXCEPTION_CONTAINS(encoder->OnLabelsEnd(), TJsonEncodeError, "label 'name' is not defined");
+            }
+        };
+
+        doTest(false);
+        doTest(true);
+    }
+
     Y_UNIT_TEST(MetricsWithDifferentLabelOrderGetMerged) {
         TString json;
         TStringOutput out(json);
diff --git a/library/cpp/threading/chunk_queue/queue.h b/library/cpp/threading/chunk_queue/queue.h
index 7cb4be4b4ef..ed3cf57b887 100644
--- a/library/cpp/threading/chunk_queue/queue.h
+++ b/library/cpp/threading/chunk_queue/queue.h
@@ -1,11 +1,8 @@
 #pragma once
 
-#include <library/cpp/deprecated/atomic/atomic.h> // AtomicGet
-
 #include <util/datetime/base.h>
 #include <util/generic/noncopyable.h>
 #include <util/generic/ptr.h>
-#include <util/generic/typetraits.h>
 #include <util/generic/ylimits.h>
 #include <util/system/guard.h>
 #include <util/system/spinlock.h>
@@ -22,11 +19,11 @@ namespace NThreading {
 #endif
 
 #if !defined(PLATFORM_PAGE_SIZE)
-    #define PLATFORM_PAGE_SIZE 4 * 1024
+    #define PLATFORM_PAGE_SIZE (4 * 1024)
 #endif
 
     template <typename T, size_t PadSize = PLATFORM_CACHE_LINE>
-    struct TPadded: public T {
+    struct alignas(PadSize) TPadded: public T {
         char Pad[PadSize - sizeof(T) % PadSize];
 
         TPadded() {
@@ -43,87 +40,59 @@ namespace NThreading {
         }
     };
 
-    ////////////////////////////////////////////////////////////////////////////////
-    // Type helpers
-
-    namespace NImpl {
-        template <typename T>
-        struct TPodTypeHelper {
-            template <typename TT>
-            static void Write(T* ptr, TT&& value) {
-                *ptr = value;
-            }
-
-            static T Read(T* ptr) {
-                return *ptr;
-            }
-
-            static void Destroy(T* ptr) {
-                Y_UNUSED(ptr);
-            }
-        };
-
-        template <typename T>
-        struct TNonPodTypeHelper {
-            template <typename TT>
-            static void Write(T* ptr, TT&& value) {
-                new (ptr) T(std::forward<TT>(value));
-            }
-
-            static T Read(T* ptr) {
-                return std::move(*ptr);
-            }
-
-            static void Destroy(T* ptr) {
-                (void)ptr; /* Make MSVC happy. */
-                ptr->~T();
-            }
-        };
-
-        template <typename T>
-        using TTypeHelper = std::conditional_t<
-            TTypeTraits<T>::IsPod,
-            TPodTypeHelper<T>,
-            TNonPodTypeHelper<T>>;
-
-    } // namespace NImpl
-
     ////////////////////////////////////////////////////////////////////////////////
     // One producer/one consumer chunked queue.
 
     template <typename T, size_t ChunkSize = PLATFORM_PAGE_SIZE>
     class TOneOneQueue: private TNonCopyable {
-        using TTypeHelper = NImpl::TTypeHelper<T>;
-
         struct TChunk;
 
         struct TChunkHeader {
-            size_t Count = 0;
-            TChunk* Next = nullptr;
+            // Incremented by the producer (release) after writing an entry, read by
+            // the consumer (acquire) — publishes the entry data written so far.
+            std::atomic<size_t> Count = 0;
+            // Set by the producer (release) when the chunk is exhausted, read by the
+            // consumer (acquire) — publishes the next chunk and makes it safe for the
+            // consumer to delete the exhausted one.
+            std::atomic<TChunk*> Next = nullptr;
         };
 
         struct TChunk: public TChunkHeader {
-            static constexpr size_t MaxCount = (ChunkSize - sizeof(TChunkHeader)) / sizeof(T);
+            // Offset of Entries inside TChunk: the header size rounded up to the
+            // alignment of T, so that every slot is properly aligned.
+            static constexpr size_t EntriesOffset = (sizeof(TChunkHeader) + alignof(T) - 1) / alignof(T) * alignof(T);
+            static constexpr size_t MaxCount = ChunkSize > EntriesOffset ? (ChunkSize - EntriesOffset) / sizeof(T) : 0;
+            static_assert(MaxCount > 0, "ChunkSize is too small to hold at least one element of T");
 
-            char Entries[MaxCount * sizeof(T)];
+            alignas(T) char Entries[MaxCount * sizeof(T)];
 
             TChunk() {
                 Y_UNUSED(Entries); // uninitialized
             }
 
-            ~TChunk() {
-                for (size_t i = 0; i < this->Count; ++i) {
-                    TTypeHelper::Destroy(GetPtr(i));
+            // No concurrent access: the chunk is destroyed after the producer
+            // finished with it (or in the queue destructor).
+            void DestroyRangeFrom(size_t start) requires (!std::is_trivially_destructible_v<T>) {
+                const size_t end = this->Count.load(std::memory_order_relaxed);
+                Y_ASSERT(start <= end);
+                T* const endPtr = GetPtr(end);
+
+                for (T* ptr = GetPtr(start); ptr != endPtr; ++ptr) {
+                    ptr->~T();
                 }
             }
 
             T* GetPtr(size_t i) {
-                return (T*)Entries + i;
+                return reinterpret_cast<T*>(Entries) + i;
             }
         };
 
         struct TWriterState {
             TChunk* Chunk = nullptr;
+            // Producer-local mirror of Chunk->Count: the number of entries
+            // already published in the current chunk. Only the producer
+            // touches it, so it needs no synchronization.
+            size_t Pos = 0;
         };
 
         struct TReaderState {
@@ -143,20 +112,36 @@ namespace NThreading {
         }
 
         ~TOneOneQueue() {
-            DeleteChunks(Reader.Chunk);
+            auto chunk = Reader.Chunk->Next.load(std::memory_order_relaxed);
+            if constexpr (!std::is_trivially_destructible_v<T>) {
+                Reader.Chunk->DestroyRangeFrom(Reader.Count);
+            }
+            delete Reader.Chunk;
+
+            while (chunk) {
+                if constexpr (!std::is_trivially_destructible_v<T>) {
+                    chunk->DestroyRangeFrom(0);
+                }
+                auto next = chunk->Next.load(std::memory_order_relaxed);
+                delete chunk;
+                chunk = next;
+            }
         }
 
         template <typename TT>
         void Enqueue(TT&& value) {
             T* ptr = PrepareWrite();
             Y_ASSERT(ptr);
-            TTypeHelper::Write(ptr, std::forward<TT>(value));
+            new (ptr) T(std::forward<TT>(value));
             CompleteWrite();
         }
 
         bool Dequeue(T& value) {
-            if (T* ptr = PrepareRead()) {
-                value = TTypeHelper::Read(ptr);
+            if (T* ptr = PrepareRead(); ptr) {
+                value = std::move(*ptr);
+                if constexpr (!std::is_trivially_destructible_v<T>) {
+                    ptr->~T();
+                }
                 CompleteRead();
                 return true;
             }
@@ -170,20 +155,25 @@ namespace NThreading {
     protected:
         T* PrepareWrite() {
             TChunk* chunk = Writer.Chunk;
-            Y_ASSERT(chunk && !chunk->Next);
+            Y_ASSERT(chunk && chunk->Next.load(std::memory_order_relaxed) == nullptr);
 
-            if (chunk->Count != TChunk::MaxCount) {
-                return chunk->GetPtr(chunk->Count);
+            if (Writer.Pos != TChunk::MaxCount) {
+                return chunk->GetPtr(Writer.Pos);
             }
 
             chunk = new TChunk();
-            AtomicSet(Writer.Chunk->Next, chunk);
+            // Release-publishes the new chunk to the consumer.
+            Writer.Chunk->Next.store(chunk, std::memory_order_release);
             Writer.Chunk = chunk;
+            Writer.Pos = 0;
             return chunk->GetPtr(0);
         }
 
         void CompleteWrite() {
-            AtomicSet(Writer.Chunk->Count, Writer.Chunk->Count + 1);
+            // Release-publishes the entry written by the preceding PrepareWrite().
+            // A plain store suffices: Count of the current chunk is only ever
+            // written by the (single) producer, so no atomic RMW is needed.
+            Writer.Chunk->Count.store(++Writer.Pos, std::memory_order_release);
         }
 
         T* PrepareRead() {
@@ -191,7 +181,8 @@ namespace NThreading {
             Y_ASSERT(chunk);
 
             for (;;) {
-                size_t writerCount = AtomicGet(chunk->Count);
+                // Acquire-syncs with CompleteWrite(), making the published entry visible.
+                size_t writerCount = chunk->Count.load(std::memory_order_acquire);
                 if (Reader.Count != writerCount) {
                     return chunk->GetPtr(Reader.Count);
                 }
@@ -200,7 +191,10 @@ namespace NThreading {
                     return nullptr;
                 }
 
-                chunk = AtomicGet(chunk->Next);
+                // Acquire-syncs with the release-store in PrepareWrite(); after this
+                // load the producer is known to be done with the exhausted chunk,
+                // so it is safe to delete it below.
+                chunk = chunk->Next.load(std::memory_order_acquire);
                 if (!chunk) {
                     return nullptr;
                 }
@@ -214,15 +208,6 @@ namespace NThreading {
         void CompleteRead() {
             ++Reader.Count;
         }
-
-    private:
-        static void DeleteChunks(TChunk* chunk) {
-            while (chunk) {
-                TChunk* next = chunk->Next;
-                delete chunk;
-                chunk = next;
-            }
-        }
     };
 
     ////////////////////////////////////////////////////////////////////////////////
@@ -231,15 +216,13 @@ namespace NThreading {
 
     template <typename T, size_t Concurrency = 4, size_t ChunkSize = PLATFORM_PAGE_SIZE>
     class TManyOneQueue: private TNonCopyable {
-        using TTypeHelper = NImpl::TTypeHelper<T>;
-
         struct TEntry {
             T Value;
             ui64 Tag;
         };
 
         struct TQueueType: public TOneOneQueue<TEntry, ChunkSize> {
-            TSpinLock WriteLock;
+            TPadded<TSpinLock> WriteLock;
 
             using TOneOneQueue<TEntry, ChunkSize>::PrepareWrite;
             using TOneOneQueue<TEntry, ChunkSize>::CompleteWrite;
@@ -266,7 +249,11 @@ namespace NThreading {
         bool Dequeue(T& value) {
             size_t index = 0;
             if (TEntry* entry = PrepareRead(index)) {
-                value = TTypeHelper::Read(&entry->Value);
+                T* valuePtr = &entry->Value;
+                value = std::move(*valuePtr);
+                if constexpr (!std::is_trivially_destructible_v<T>) {
+                    valuePtr->~T();
+                }
                 Queues[index].CompleteRead();
                 return true;
             }
@@ -302,7 +289,7 @@ namespace NThreading {
                 }
                 TEntry* entry = queue.PrepareWrite();
                 Y_ASSERT(entry);
-                TTypeHelper::Write(&entry->Value, std::forward<TT>(value));
+                new (&entry->Value) T(std::forward<TT>(value));
                 entry->Tag = tag;
                 queue.CompleteWrite();
                 return true;
@@ -382,7 +369,7 @@ namespace NThreading {
     template <typename T, size_t Concurrency = 4, size_t ChunkSize = PLATFORM_PAGE_SIZE>
     class TRelaxedManyOneQueue: private TNonCopyable {
         struct TQueueType: public TOneOneQueue<T, ChunkSize> {
-            TSpinLock WriteLock;
+            TPadded<TSpinLock> WriteLock;
         };
 
     private:
diff --git a/library/cpp/threading/chunk_queue/queue_ut.cpp b/library/cpp/threading/chunk_queue/queue_ut.cpp
index 717ce9f047e..d5623a97f06 100644
--- a/library/cpp/threading/chunk_queue/queue_ut.cpp
+++ b/library/cpp/threading/chunk_queue/queue_ut.cpp
@@ -3,6 +3,8 @@
 #include <library/cpp/testing/unittest/registar.h>
 
 #include <util/generic/set.h>
+#include <util/generic/string.h>
+#include <util/string/cast.h>
 
 namespace NThreading {
     ////////////////////////////////////////////////////////////////////////////////
@@ -52,6 +54,55 @@ Y_UNIT_TEST(ShouldStoreMultipleChunks) {
         UNIT_ASSERT_EQUAL(result, i);
     }
 }
+
+struct alignas(64) TOverAligned {
+    size_t Value = 0;
+
+    TOverAligned() = default;
+
+    explicit TOverAligned(size_t value)
+        : Value(value)
+    {
+        UNIT_ASSERT(reinterpret_cast<uintptr_t>(this) % alignof(TOverAligned) == 0);
+    }
+};
+
+// ChunkSize = 128, alignof = 64: EntriesOffset = 64, sizeof = 64, so
+// MaxCount = 1 — every Enqueue exercises the chunk hand-off.
+Y_UNIT_TEST(ShouldKeepOverAlignedEntriesAligned) {
+    TOneOneQueue<TOverAligned, 128> queue;
+
+    for (size_t i = 0; i < 10; ++i) {
+        queue.Enqueue(TOverAligned{i});
+    }
+
+    for (size_t i = 0; i < 10; ++i) {
+        TOverAligned result;
+        UNIT_ASSERT(queue.Dequeue(result));
+        UNIT_ASSERT_EQUAL(result.Value, i);
+    }
+
+    UNIT_ASSERT(queue.IsEmpty());
+}
+
+Y_UNIT_TEST(ShouldDestroyNonTrivialEntriesOnDestruction) {
+    // Small chunks force the queue to span several chunks, and half of the
+    // entries are still alive when the queue is destroyed: ~TOneOneQueue()
+    // must destroy the leftovers in every chunk exactly once.
+    TOneOneQueue<TString, 128> queue;
+
+    for (int i = 0; i < 100; ++i) {
+        queue.Enqueue(ToString(i));
+    }
+
+    for (int i = 0; i < 50; ++i) {
+        TString result;
+        UNIT_ASSERT(queue.Dequeue(result));
+        UNIT_ASSERT_EQUAL(result, ToString(i));
+    }
+
+    UNIT_ASSERT(!queue.IsEmpty());
+}
 }
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -87,6 +138,22 @@ Y_UNIT_TEST(ShouldReturnEntries) {
     UNIT_ASSERT(queue.IsEmpty());
     UNIT_ASSERT(!queue.Dequeue(result));
 }
+
+Y_UNIT_TEST(ShouldHandleNonTrivialEntries) {
+    TManyOneQueue<TString> queue;
+
+    for (int i = 0; i < 100; ++i) {
+        queue.Enqueue(ToString(i));
+    }
+
+    for (int i = 0; i < 100; ++i) {
+        TString result;
+        UNIT_ASSERT(queue.Dequeue(result));
+        UNIT_ASSERT_EQUAL(result, ToString(i));
+    }
+
+    UNIT_ASSERT(queue.IsEmpty());
+}
 }
 
 ////////////////////////////////////////////////////////////////////////////////
diff --git a/library/cpp/yt/memory/poison-inl.h b/library/cpp/yt/memory/poison-inl.h
index 0f2998fdc04..1f9db9a67b8 100644
--- a/library/cpp/yt/memory/poison-inl.h
+++ b/library/cpp/yt/memory/poison-inl.h
@@ -31,7 +31,7 @@ Y_FORCE_INLINE void RecycleFreedMemory(TMutableRef ref)
     __asan_unpoison_memory_region(ref.data(), ref.size());
 }
 
-Y_FORCE_INLINE void PoisonUnitializedOrFreedMemory(TMutableRef /*ref*/)
+Y_FORCE_INLINE void PoisonUninitializedOrFreedMemory(TMutableRef /*ref*/)
 { }
 
 #elif defined(_msan_enabled_)
@@ -53,7 +53,7 @@ Y_FORCE_INLINE void PoisonFreedMemory(TMutableRef ref)
 Y_FORCE_INLINE void RecycleFreedMemory(TMutableRef /*ref*/)
 { }
 
-Y_FORCE_INLINE void PoisonUnitializedOrFreedMemory(TMutableRef /*ref*/)
+Y_FORCE_INLINE void PoisonUninitializedOrFreedMemory(TMutableRef /*ref*/)
 { }
 
 #elif defined(NDEBUG)
@@ -67,7 +67,7 @@ Y_FORCE_INLINE void PoisonFreedMemory(TMutableRef /*ref*/)
 Y_FORCE_INLINE void RecycleFreedMemory(TMutableRef /*ref*/)
 { }
 
-Y_FORCE_INLINE void PoisonUnitializedOrFreedMemory(TMutableRef /*ref*/)
+Y_FORCE_INLINE void PoisonUninitializedOrFreedMemory(TMutableRef /*ref*/)
 { }
 
 #endif
diff --git a/library/cpp/yt/memory/poison.cpp b/library/cpp/yt/memory/poison.cpp
index dbb561ea39f..a5bc7ec1ac7 100644
--- a/library/cpp/yt/memory/poison.cpp
+++ b/library/cpp/yt/memory/poison.cpp
@@ -51,7 +51,7 @@ void RecycleFreedMemory(TMutableRef ref)
     ClobberMemory<'\xc0', '\x01', '\xb1', '\xba'>(ref.data(), ref.size());
 }
 
-void PoisonUnitializedOrFreedMemory(TMutableRef ref)
+void PoisonUninitializedOrFreedMemory(TMutableRef ref)
 {
     // BADBLOOD
     ClobberMemory<'\xba', '\xdb', '\x10', '\x0d'>(ref.data(), ref.size());
diff --git a/library/cpp/yt/memory/poison.h b/library/cpp/yt/memory/poison.h
index d28b5464551..a83ec49aaa4 100644
--- a/library/cpp/yt/memory/poison.h
+++ b/library/cpp/yt/memory/poison.h
@@ -42,7 +42,7 @@ void RecycleFreedMemory(TMutableRef ref);
  * In ASAN builds, does nothing.
  * In MSAN builds, does nothing.
  */
-void PoisonUnitializedOrFreedMemory(TMutableRef ref);
+void PoisonUninitializedOrFreedMemory(TMutableRef ref);
 
 ////////////////////////////////////////////////////////////////////////////////
 
diff --git a/library/cpp/yt/memory/ref.cpp b/library/cpp/yt/memory/ref.cpp
index eddae004ca6..8a247a2a698 100644
--- a/library/cpp/yt/memory/ref.cpp
+++ b/library/cpp/yt/memory/ref.cpp
@@ -144,7 +144,7 @@ class TAllocationHolderBase
         if (options.InitializeStorage) {
             ::memset(static_cast<TDerived*>(this)->GetBegin(), 0, Size_);
         } else {
-            PoisonUnitializedOrFreedMemory(GetRef());
+            PoisonUninitializedOrFreedMemory(GetRef());
         }
 #ifdef YT_ENABLE_REF_COUNTED_TRACKING
         TRefCountedTrackerFacade::AllocateTagInstance(Cookie_);
@@ -154,7 +154,7 @@ class TAllocationHolderBase
 
     void Finalize()
     {
-        PoisonUnitializedOrFreedMemory(GetRef());
+        PoisonUninitializedOrFreedMemory(GetRef());
     }
 };
 
diff --git a/src/api/protos/ydb_query.proto b/src/api/protos/ydb_query.proto
index 6263cbfe612..beaeeda48bd 100644
--- a/src/api/protos/ydb_query.proto
+++ b/src/api/protos/ydb_query.proto
@@ -237,6 +237,13 @@ message ExecuteQueryRequest {
 
     // Format settings, only used for Ydb.ResultSet.Format.FORMAT_ARROW
     Ydb.Formats.ArrowFormatSettings arrow_format_settings = 14;
+
+    // Enable PostgreSQL-like affected_rows statistics collection.
+    // When true, the server counts logical base-table rows affected by DML
+    // (INSERT, UPSERT, REPLACE, UPDATE, DELETE), excluding secondary-index
+    // maintenance. Zero for read-only queries and non-matching DML.
+    // Enabling this setting can lead to overheads.
+    bool collect_affected_rows = 15;
 }
 
 message ResultSetMeta {
diff --git a/src/api/protos/ydb_query_stats.proto b/src/api/protos/ydb_query_stats.proto
index 34f4f49bdb1..c8c3676f6ce 100644
--- a/src/api/protos/ydb_query_stats.proto
+++ b/src/api/protos/ydb_query_stats.proto
@@ -18,6 +18,10 @@ message TableAccessStats {
     OperationStats updates = 4;
     OperationStats deletes = 5;
     uint64 partitions_count = 6;
+    // Set only when collect_affected_rows is enabled. Counts logical base-table rows
+    // affected by DML (INSERT, UPSERT, REPLACE, UPDATE, DELETE), excluding
+    // secondary-index maintenance. Zero for read-only queries and non-matching DML.
+    optional uint64 affected_rows = 7;
 }
 
 message QueryPhaseStats {
diff --git a/src/api/protos/ydb_table.proto b/src/api/protos/ydb_table.proto
index 5a40929e7ff..6f3cda7065e 100644
--- a/src/api/protos/ydb_table.proto
+++ b/src/api/protos/ydb_table.proto
@@ -225,6 +225,15 @@ message FulltextIndexSettings {
         //   Tokens: ["cars", "beautifully", "conspired"]
         //   Output: ["car", "beauti", "conspir"]
         optional bool use_filter_snowball = 140;
+
+        // Whether to apply superlemmer suffix replacement to each token
+        // Generalizes lemmas by suffix replacement using a built-in dictionary
+        // Works on UTF-8 text, primarily optimized for Russian
+        // Cannot be used together with use_filter_snowball, use_filter_ngram, or use_filter_edge_ngram
+        // Example:
+        //   Tokens: ["мороженое", "уже"]
+        //   Output: ["мороженый", "узкий"]
+        optional bool use_filter_superlemmer = 141;
     }
 
     // Represents text analyzers settings for a specific column
@@ -445,6 +454,7 @@ message TableMultiColumnStatistics {
     enum MultiColumnStatisticsType {
         STATISTIC_TYPE_UNSPECIFIED = 0;
         COUNT_MIN_SKETCH = 1;
+        EQ_HEIGHT_HISTOGRAM = 2;
     }
     // Name of statistics
     string name = 1;
diff --git a/src/client/impl/observability/constants.h b/src/client/impl/observability/constants.h
index 26a4a99c936..f5a3011b0be 100644
--- a/src/client/impl/observability/constants.h
+++ b/src/client/impl/observability/constants.h
@@ -14,7 +14,7 @@ namespace NYdb::inline V3::NObservability {
 // SDK build-info chain versions. Bump these when the corresponding
 // observability integration changes incompatibly.
 inline constexpr std::string_view kTracingChainVersion = "0.1.0";
-inline constexpr std::string_view kMetricsChainVersion = "0.1.0";
+inline constexpr std::string_view kMetricsChainVersion = "0.2.0";
 
 // ---------------------------------------------------------------------------
 // OTel Semconv attribute keys shared between span attributes and metric labels.
@@ -120,6 +120,7 @@ inline constexpr std::string_view kSessionPrefixGeneric       = "ydb.session";
 
 // Leaf names for session-pool metrics (combined as "<prefix>.<leaf>").
 inline constexpr std::string_view kSessionLeafCount           = "count";
+inline constexpr std::string_view kSessionLeafClosed          = "closed";
 inline constexpr std::string_view kSessionLeafCreateTime      = "create_time";
 inline constexpr std::string_view kSessionLeafPendingRequests = "pending_requests";
 inline constexpr std::string_view kSessionLeafTimeouts        = "timeouts";
diff --git a/src/client/impl/session/kqp_session_common.cpp b/src/client/impl/session/kqp_session_common.cpp
index b2c5f38f818..bb7c4f2a2aa 100644
--- a/src/client/impl/session/kqp_session_common.cpp
+++ b/src/client/impl/session/kqp_session_common.cpp
@@ -56,31 +56,45 @@ const TEndpointKey& TKqpSessionCommon::GetEndpointKey() const {
 }
 
 // Can be called from interceptor, need lock
-void TKqpSessionCommon::MarkBroken() {
+bool TKqpSessionCommon::MarkBroken() {
     std::lock_guard guard(Lock_);
+    const bool firstTerminal = State_ != EState::S_BROKEN && State_ != EState::S_CLOSING;
     if (State_ == EState::S_ACTIVE) {
         NeedUpdateActiveCounter_ = true;
     }
     State_ = EState::S_BROKEN;
+    return firstTerminal;
 }
 
-void TKqpSessionCommon::MarkAsClosing() {
+bool TKqpSessionCommon::MarkAsClosing() {
     std::lock_guard guard(Lock_);
+    const bool firstTerminal = State_ != EState::S_BROKEN && State_ != EState::S_CLOSING;
     if (State_ == EState::S_ACTIVE) {
         NeedUpdateActiveCounter_ = true;
     }
 
     State_ = EState::S_CLOSING;
+    return firstTerminal;
 }
 
-void TKqpSessionCommon::MarkActive() {
+bool TKqpSessionCommon::MarkActive() {
+    std::lock_guard guard(Lock_);
+    if (State_ == EState::S_BROKEN || State_ == EState::S_CLOSING) {
+        return false;
+    }
     State_ = EState::S_ACTIVE;
     NeedUpdateActiveCounter_ = false;
+    return true;
 }
 
-void TKqpSessionCommon::MarkIdle() {
+bool TKqpSessionCommon::MarkIdle() {
+    std::lock_guard guard(Lock_);
+    if (State_ == EState::S_BROKEN || State_ == EState::S_CLOSING) {
+        return false;
+    }
     State_ = EState::S_IDLE;
     NeedUpdateActiveCounter_ = false;
+    return true;
 }
 
 bool TKqpSessionCommon::IsOwnedBySessionPool() const {
@@ -138,7 +152,7 @@ void TKqpSessionCommon::UpdateServerCloseHandler(IServerCloseHandler* handler) {
     CloseHandler_.store(handler);
 }
 
-void TKqpSessionCommon::CloseFromServer(std::weak_ptr<ISessionClient> client) noexcept {
+void TKqpSessionCommon::CloseFromServer(std::weak_ptr<ISessionClient> client, std::string_view reason) noexcept {
     auto strong = client.lock();
     if (!strong) {
         // Session closed on the server after stopping client - do nothing
@@ -148,7 +162,7 @@ void TKqpSessionCommon::CloseFromServer(std::weak_ptr<ISessionClient> client) no
 
     IServerCloseHandler* h = CloseHandler_.load();
     if (h) {
-        h->OnCloseSession(this, strong);
+        h->OnCloseSession(this, strong, reason);
     }
 }
 
diff --git a/src/client/impl/session/kqp_session_common.h b/src/client/impl/session/kqp_session_common.h
index 3cdff81a1e0..a552702531c 100644
--- a/src/client/impl/session/kqp_session_common.h
+++ b/src/client/impl/session/kqp_session_common.h
@@ -20,7 +20,8 @@ class IServerCloseHandler {
 public:
     virtual ~IServerCloseHandler() = default;
     // called when session should be closed by server signal
-    virtual void OnCloseSession(const TKqpSessionCommon*, std::shared_ptr<ISessionClient>) = 0;
+    virtual void OnCloseSession(const TKqpSessionCommon*, std::shared_ptr<ISessionClient>,
+        std::string_view) = 0;
 };
 
 class TKqpSessionCommon : public TEndpointObj {
@@ -42,14 +43,15 @@ class TKqpSessionCommon : public TEndpointObj {
     const std::string& GetId() const;
     const std::string& GetEndpoint() const;
     const TEndpointKey& GetEndpointKey() const;
-    void MarkBroken();
-    void MarkAsClosing();
-    void MarkActive();
-    void MarkIdle();
+    bool MarkBroken();
+    bool MarkAsClosing();
+    bool MarkActive();
+    bool MarkIdle();
     bool IsOwnedBySessionPool() const;
     EState GetState() const;
     void SetNeedUpdateActiveCounter(bool flag);
     bool NeedUpdateActiveCounter() const;
+    virtual std::shared_ptr<ISessionClient> GetSessionClient() const { return {}; }
     void InvalidateQueryInCache(const std::string& key);
     void InvalidateQueryCache();
     void ScheduleTimeToTouch(TDuration interval, bool updateTimeInPast);
@@ -63,7 +65,7 @@ class TKqpSessionCommon : public TEndpointObj {
     void UpdateServerCloseHandler(IServerCloseHandler*);
 
     // Called asynchronously from grpc thread.
-    void CloseFromServer(std::weak_ptr<ISessionClient> client) noexcept;
+    void CloseFromServer(std::weak_ptr<ISessionClient> client, std::string_view reason) noexcept;
 
 public:
     std::optional<TDeadline> PropagatedDeadline_;
diff --git a/src/client/impl/session/session_client.h b/src/client/impl/session/session_client.h
index 5054a690b46..3da209058e7 100644
--- a/src/client/impl/session/session_client.h
+++ b/src/client/impl/session/session_client.h
@@ -1,6 +1,7 @@
 #pragma once
 
 #include <cstdint>
+#include <string_view>
 
 namespace NYdb::inline V3 {
 
@@ -14,6 +15,9 @@ class ISessionClient {
 
     virtual void PessimizeNode(std::uint64_t nodeId) = 0;
 
+    virtual void RecordSessionClosed(std::string_view) {
+    }
+
     // TODO: Try to remove from ISessionClient
     virtual bool ReturnSession(TKqpSessionCommon* sessionImpl) = 0;
 };
diff --git a/src/client/impl/session/session_pool.cpp b/src/client/impl/session/session_pool.cpp
index 18a5979c75f..9ce4d9600c4 100644
--- a/src/client/impl/session/session_pool.cpp
+++ b/src/client/impl/session/session_pool.cpp
@@ -9,6 +9,8 @@
 
 #include <util/random/random.h>
 
+#include <algorithm>
+
 namespace NYdb::inline V3 {
 namespace NSessionPool {
 
@@ -59,6 +61,48 @@ bool IsSessionCloseRequested(const TStatus& status) {
     return false;
 }
 
+void TSessionCloseCommand::Execute(TKqpSessionCommon& session, ISessionClient* client) const {
+    if (Transition(session) && client && session.IsOwnedBySessionPool()) {
+        client->RecordSessionClosed(Reason);
+    }
+}
+
+namespace NSessionCloseCommands {
+
+namespace {
+
+template<EStatus Code>
+bool HasStatus(const TStatus& status) { return status.GetStatus() == Code; }
+
+bool IsBreakingTransport(const TStatus& status) {
+    const auto code = status.GetStatus();
+    return status.IsTransportError()
+        && code != EStatus::CLIENT_RESOURCE_EXHAUSTED
+        && code != EStatus::CLIENT_OUT_OF_RANGE;
+}
+
+using TStatusCloseCommand =
+    std::pair<std::function<bool(const TStatus&)>, const TSessionCloseCommand*>;
+
+} // namespace
+
+const TSessionCloseCommand* FromStatus(const TStatus& status) {
+    static const TStatusCloseCommand Commands[] = {
+        {&HasStatus<EStatus::CLIENT_DEADLINE_EXCEEDED>, &ClientTimeout},
+        {&HasStatus<EStatus::CLIENT_CANCELLED>, &ClientCancelled},
+        {&IsBreakingTransport, &TransportError},
+        {&HasStatus<EStatus::SESSION_BUSY>, &SessionBusy},
+        {&HasStatus<EStatus::BAD_SESSION>, &BadSession},
+        {&IsSessionCloseRequested, &SessionShutdown},
+    };
+    const auto command = std::ranges::find_if(
+        Commands,
+        [&status](const auto& entry) { return entry.first(status); });
+    return command == std::ranges::end(Commands) ? nullptr : command->second;
+}
+
+} // namespace NSessionCloseCommands
+
 TSessionPool::TWaitersQueue::TWaitersQueue(std::uint32_t maxQueueSize)
     : MaxQueueSize_(maxQueueSize)
 {
@@ -121,7 +165,7 @@ void TSessionPool::ReplySessionToUser(
     std::unique_ptr<IGetSessionCtx> ctx)
 {
     Y_ABORT_UNLESS(session->GetState() == TKqpSessionCommon::S_IDLE);
-    session->MarkActive();
+    Y_ABORT_UNLESS(session->MarkActive());
     session->SetNeedUpdateActiveCounter(true);
     ctx->ReplySessionToUser(session);
 }
@@ -224,7 +268,7 @@ bool TSessionPool::ReturnSession(TKqpSessionCommon* impl, bool active) {
     std::unique_ptr<IGetSessionCtx> getSessionCtx;
     {
         std::lock_guard guard(Mtx_);
-        if (Closed_)
+        if (Closed_ || impl->GetState() != TKqpSessionCommon::S_IDLE)
             return false;
 
         if (auto maybeCtx = WaitersQueue_.TryGet()) {
@@ -233,6 +277,10 @@ bool TSessionPool::ReturnSession(TKqpSessionCommon* impl, bool active) {
                 IncrementActiveCounterUnsafe();
         } else {
             impl->UpdateServerCloseHandler(this);
+            if (impl->GetState() != TKqpSessionCommon::S_IDLE) {
+                impl->UpdateServerCloseHandler(nullptr);
+                return false;
+            }
             Sessions_.emplace(std::make_pair(
                 impl->GetTimeToTouchFast(),
                 impl));
@@ -356,6 +404,7 @@ TPeriodicCb TSessionPool::CreatePeriodicTask(std::weak_ptr<ISessionClient> weakC
             for (auto& sessionImpl : sessionsToDelete) {
                 if (sessionImpl) {
                     Y_ABORT_UNLESS(sessionImpl->GetState() == TKqpSessionCommon::S_IDLE);
+                    NSessionCloseCommands::PoolIdleTimeout.Execute(*sessionImpl, strongClient.get());
                     CloseAndDeleteSession(std::move(sessionImpl), strongClient);
                 }
             }
@@ -386,7 +435,8 @@ std::int64_t TSessionPool::GetCurrentPoolSize() const {
     return Sessions_.size();
 }
 
-void TSessionPool::OnCloseSession(const TKqpSessionCommon* s, std::shared_ptr<ISessionClient> client) {
+void TSessionPool::OnCloseSession(const TKqpSessionCommon* s, std::shared_ptr<ISessionClient> client,
+    std::string_view reason) {
     std::unique_ptr<TKqpSessionCommon> session;
     {
         std::lock_guard guard(Mtx_);
@@ -408,6 +458,7 @@ void TSessionPool::OnCloseSession(const TKqpSessionCommon* s, std::shared_ptr<IS
 
     if (session) {
         Y_ABORT_UNLESS(session->GetState() == TKqpSessionCommon::S_IDLE);
+        RecordSessionClosed(reason);
         CloseAndDeleteSession(std::move(session), client);
     }
 }
@@ -438,6 +489,10 @@ void TSessionPool::RecordConnectionCreateTime(double seconds) {
     ExternalStatCollector_.RecordConnectionCreateTime(seconds);
 }
 
+void TSessionPool::RecordSessionClosed(std::string_view reason) {
+    ExternalStatCollector_.IncSessionClosed(reason);
+}
+
 void TSessionPool::UpdateStats() {
     ActiveSessionsCounter_.Apply(ActiveSessions_);
     InPoolSessionsCounter_.Apply(Sessions_.size());
diff --git a/src/client/impl/session/session_pool.h b/src/client/impl/session/session_pool.h
index 7f48b80d482..882e51b26ee 100644
--- a/src/client/impl/session/session_pool.h
+++ b/src/client/impl/session/session_pool.h
@@ -32,6 +32,28 @@ TStatus GetStatus(const TStatus& status);
 TDuration RandomizeThreshold(TDuration duration);
 bool IsSessionCloseRequested(const TStatus& status);
 
+struct TSessionCloseCommand {
+    std::string_view Reason;
+    std::function<bool(TKqpSessionCommon&)> Transition;
+
+    void Execute(TKqpSessionCommon& session, ISessionClient* client) const;
+};
+
+namespace NSessionCloseCommands {
+inline const TSessionCloseCommand PoolIdleTimeout{"pool_idle_timeout", &TKqpSessionCommon::MarkBroken};
+inline const TSessionCloseCommand PoolGracefulShutdown{"pool_graceful_shutdown", &TKqpSessionCommon::MarkBroken};
+inline const TSessionCloseCommand ClientTimeout{"client_timeout", &TKqpSessionCommon::MarkBroken};
+inline const TSessionCloseCommand ClientCancelled{"client_cancelled", &TKqpSessionCommon::MarkBroken};
+inline const TSessionCloseCommand AttachClosed{"attach_closed", &TKqpSessionCommon::MarkBroken};
+inline const TSessionCloseCommand TransportError{"transport_error", &TKqpSessionCommon::MarkBroken};
+inline const TSessionCloseCommand NodeShutdown{"node_shutdown", &TKqpSessionCommon::MarkAsClosing};
+inline const TSessionCloseCommand SessionShutdown{"session_shutdown", &TKqpSessionCommon::MarkAsClosing};
+inline const TSessionCloseCommand BadSession{"bad_session", &TKqpSessionCommon::MarkBroken};
+inline const TSessionCloseCommand SessionBusy{"session_busy", &TKqpSessionCommon::MarkBroken};
+
+const TSessionCloseCommand* FromStatus(const TStatus& status);
+}
+
 template<typename TResponse>
 NThreading::TFuture<TResponse> InjectSessionStatusInterception(
         std::shared_ptr<::NYdb::TKqpSessionCommon> impl, NThreading::TFuture<TResponse> asyncResponse,
@@ -50,19 +72,9 @@ NThreading::TFuture<TResponse> InjectSessionStatusInterception(
         TResponse value = std::move(future.ExtractValue());
 
         const TStatus& status = GetStatus(value);
-        // Exclude CLIENT_RESOURCE_EXHAUSTED from transport errors which can cause to session disconnect
-        // since we have guarantee this request wasn't been started to execute.
-
-        if (status.IsTransportError()
-            && status.GetStatus() != EStatus::CLIENT_RESOURCE_EXHAUSTED && status.GetStatus() != EStatus::CLIENT_OUT_OF_RANGE)
-        {
-            impl->MarkBroken();
-        } else if (status.GetStatus() == EStatus::SESSION_BUSY) {
-            impl->MarkBroken();
-        } else if (status.GetStatus() == EStatus::BAD_SESSION) {
-            impl->MarkBroken();
-        } else if (IsSessionCloseRequested(status)) {
-            impl->MarkAsClosing();
+        if (const auto* command = NSessionCloseCommands::FromStatus(status)) {
+            const auto client = impl->GetSessionClient();
+            command->Execute(*impl, client.get());
         } else {
             // NOTE: About GetState and lock
             // Simultanious call multiple requests on the same session make no sence, due to server limitation.
@@ -129,8 +141,10 @@ class TSessionPool : public IServerCloseHandler {
     void SetStatCollector(NSdkStats::TStatCollector::TSessionPoolStatCollector collector);
 
     void RecordConnectionCreateTime(double seconds);
+    void RecordSessionClosed(std::string_view reason);
 
-    void OnCloseSession(const TKqpSessionCommon*, std::shared_ptr<ISessionClient> client) override;
+    void OnCloseSession(const TKqpSessionCommon*, std::shared_ptr<ISessionClient>,
+        std::string_view) override;
 
 private:
     void UpdateStats();
diff --git a/src/client/impl/stats/stats.h b/src/client/impl/stats/stats.h
index 53f48bd2c0f..b24a6a15e15 100644
--- a/src/client/impl/stats/stats.h
+++ b/src/client/impl/stats/stats.h
@@ -238,6 +238,17 @@ struct TStatCollector {
             )->Inc();
         }
 
+        void IncSessionClosed(std::string_view reason) {
+            if (!ExternalRegistry_) {
+                return;
+            }
+            auto labels = BasePoolLabels();
+            labels["reason"] = std::string(reason);
+            ExternalRegistry_->Counter(MetricName(NObservability::MetricName::kSessionLeafClosed), labels,
+                "Number of closed sessions, split by reason.",
+                std::string(NObservability::MetricUnit::kSession))->Inc();
+        }
+
         void RecordConnectionCreateTime(double seconds) {
             if (!ExternalRegistry_) {
                 return;
diff --git a/src/client/persqueue_public/ut/read_session_ut.cpp b/src/client/persqueue_public/ut/read_session_ut.cpp
index 92b45da0a53..30e18b65d3c 100644
--- a/src/client/persqueue_public/ut/read_session_ut.cpp
+++ b/src/client/persqueue_public/ut/read_session_ut.cpp
@@ -926,7 +926,7 @@ Y_UNIT_TEST_SUITE(ReadSessionImplTest) {
                 }
             });
         EXPECT_CALL(*setup.MockProcessor, OnInitRequest(_))
-            .WillOnce(Invoke([](const Ydb::PersQueue::V1::MigrationStreamingReadClientMessage::InitRequest& req) {
+            .WillOnce([](const Ydb::PersQueue::V1::MigrationStreamingReadClientMessage::InitRequest& req) {
                 UNIT_ASSERT_STRINGS_EQUAL(req.consumer(), "TestConsumer");
                 UNIT_ASSERT_VALUES_EQUAL(req.max_lag_duration_ms(), 32000);
                 UNIT_ASSERT_VALUES_EQUAL(req.start_from_written_at_ms(), 42000);
@@ -936,7 +936,7 @@ Y_UNIT_TEST_SUITE(ReadSessionImplTest) {
                 UNIT_ASSERT_VALUES_EQUAL(req.topics_read_settings(0).partition_group_ids_size(), 2);
                 UNIT_ASSERT_VALUES_EQUAL(req.topics_read_settings(0).partition_group_ids(0), 100);
                 UNIT_ASSERT_VALUES_EQUAL(req.topics_read_settings(0).partition_group_ids(1), 101);
-            }));
+            });
         setup.GetSession()->Start();
         setup.MockProcessorFactory->Wait();
 
@@ -1100,14 +1100,14 @@ Y_UNIT_TEST_SUITE(ReadSessionImplTest) {
             UNIT_ASSERT(stream);
 
             EXPECT_CALL(*setup.MockProcessor, OnStartReadRequest(_))
-                .WillOnce(Invoke([](const Ydb::PersQueue::V1::MigrationStreamingReadClientMessage::StartRead& req) {
+                .WillOnce([](const Ydb::PersQueue::V1::MigrationStreamingReadClientMessage::StartRead& req) {
                     UNIT_ASSERT_STRINGS_EQUAL(req.topic().path(), "TestTopic");
                     UNIT_ASSERT_STRINGS_EQUAL(req.cluster(), "TestCluster");
                     UNIT_ASSERT_VALUES_EQUAL(req.partition(), 1);
                     UNIT_ASSERT_VALUES_EQUAL(req.assign_id(), 1);
                     UNIT_ASSERT_VALUES_EQUAL(req.read_offset(), 13);
                     UNIT_ASSERT_VALUES_EQUAL(req.commit_offset(), 31);
-                }));
+                });
 
             event.Confirm(13, 31);
         }
@@ -1140,12 +1140,12 @@ Y_UNIT_TEST_SUITE(ReadSessionImplTest) {
             UNIT_ASSERT_EQUAL(destroyEvent.GetPartitionStream(), stream);
 
             EXPECT_CALL(*setup.MockProcessor, OnReleasedRequest(_))
-                .WillOnce(Invoke([](const Ydb::PersQueue::V1::MigrationStreamingReadClientMessage::Released& req) {
+                .WillOnce([](const Ydb::PersQueue::V1::MigrationStreamingReadClientMessage::Released& req) {
                     UNIT_ASSERT_STRINGS_EQUAL(req.topic().path(), "TestTopic");
                     UNIT_ASSERT_STRINGS_EQUAL(req.cluster(), "TestCluster");
                     UNIT_ASSERT_VALUES_EQUAL(req.partition(), 1);
                     UNIT_ASSERT_VALUES_EQUAL(req.assign_id(), 1);
-                }));
+                });
 
             destroyEvent.Confirm();
         }
@@ -1319,7 +1319,7 @@ Y_UNIT_TEST_SUITE(ReadSessionImplTest) {
         THashSet<ui64> committedCookies;
         THashSet<ui64> committedOffsets;
         EXPECT_CALL(*setup.MockProcessor, OnCommitRequest(_))
-            .WillRepeatedly(Invoke([&committedCookies, &committedOffsets](const Ydb::PersQueue::V1::MigrationStreamingReadClientMessage::Commit& req) {
+            .WillRepeatedly([&committedCookies, &committedOffsets](const Ydb::PersQueue::V1::MigrationStreamingReadClientMessage::Commit& req) {
                 for (const auto& commit : req.cookies()) {
                     committedCookies.insert(commit.partition_cookie());
                 }
@@ -1329,7 +1329,7 @@ Y_UNIT_TEST_SUITE(ReadSessionImplTest) {
                         committedOffsets.insert(i);
                     }
                 }
-            }));
+            });
 
         for (ui64 i = 1; i <= serverBatchesCount; ++i) {
             TMockReadSessionProcessor::TServerReadInfo resp;
@@ -1512,12 +1512,12 @@ Y_UNIT_TEST_SUITE(ReadSessionImplTest) {
         setup.SuccessfulInit();
         TPartitionStream::TPtr stream = setup.CreatePartitionStream();
         EXPECT_CALL(*setup.MockProcessor, OnStatusRequest(_))
-            .WillOnce(Invoke([](const Ydb::PersQueue::V1::MigrationStreamingReadClientMessage::Status& req) {
+            .WillOnce([](const Ydb::PersQueue::V1::MigrationStreamingReadClientMessage::Status& req) {
                 UNIT_ASSERT_VALUES_EQUAL(req.topic().path(), "TestTopic");
                 UNIT_ASSERT_VALUES_EQUAL(req.cluster(), "TestCluster");
                 UNIT_ASSERT_VALUES_EQUAL(req.partition(), 1);
                 UNIT_ASSERT_VALUES_EQUAL(req.assign_id(), 1);
-            }));
+            });
         // Another assign id.
         setup.MockProcessor->AddServerResponse(TMockReadSessionProcessor::TServerReadInfo()
                                                .PartitionStreamStatus(11, 34, TInstant::Seconds(4), "TestTopic", "TestCluster", 1 /*partition*/, 13/*assign id to ignore*/));
@@ -1558,7 +1558,7 @@ Y_UNIT_TEST_SUITE(ReadSessionImplTest) {
         bool has1 = false;
         bool has2 = false;
         EXPECT_CALL(*setup.MockProcessor, OnCommitRequest(_))
-            .WillRepeatedly(Invoke([&](const Ydb::PersQueue::V1::MigrationStreamingReadClientMessage::Commit& req) {
+            .WillRepeatedly([&](const Ydb::PersQueue::V1::MigrationStreamingReadClientMessage::Commit& req) {
                 Cerr << "Got commit req " << req << "\n";
                 for (const auto& commit : req.cookies()) {
                     if (commit.partition_cookie() == 1) {
@@ -1575,7 +1575,7 @@ Y_UNIT_TEST_SUITE(ReadSessionImplTest) {
                     else if (range.start_offset() == 0 && range.end_offset() == 3) has2 = true;
                     else UNIT_ASSERT(false);
                 }
-            }));
+            });
 
         for (int i = 0; i < 2; ) {
             std::optional<TReadSessionEvent::TEvent> event = setup.EventsQueue->GetEvent(true);
diff --git a/src/client/query/client.cpp b/src/client/query/client.cpp
index 4a8b378f1df..c917e2b909a 100644
--- a/src/client/query/client.cpp
+++ b/src/client/query/client.cpp
@@ -92,7 +92,10 @@ class TQueryClient::TImpl: public TClientImplCommon<TQueryClient::TImpl>, public
     }
 
     ~TImpl() {
-        // TODO: Drain sessions.
+        auto sessions = SessionPool_.GetCurrentPoolSize();
+        while (sessions-- > 0) {
+            SessionPool_.RecordSessionClosed(NSessionPool::NSessionCloseCommands::PoolGracefulShutdown.Reason);
+        }
     }
 
     void SetStatCollector(const NSdkStats::TStatCollector::TClientStatCollector& collector) {
@@ -443,13 +446,17 @@ class TQueryClient::TImpl: public TClientImplCommon<TQueryClient::TImpl>, public
     }
 
     bool ReturnSession(TKqpSessionCommon* sessionImpl) override {
-        Y_ABORT_UNLESS(sessionImpl->GetState() == TSession::TImpl::S_ACTIVE ||
-            sessionImpl->GetState() == TSession::TImpl::S_IDLE);
+        const auto state = sessionImpl->GetState();
+        if (state != TSession::TImpl::S_ACTIVE && state != TSession::TImpl::S_IDLE) {
+            return false;
+        }
 
         //TODO: Remove this copy-paste from table client
         bool needUpdateCounter = sessionImpl->NeedUpdateActiveCounter();
         // Also removes NeedUpdateActiveCounter flag
-        sessionImpl->MarkIdle();
+        if (!sessionImpl->MarkIdle()) {
+            return false;
+        }
         if (!SessionPool_.ReturnSession(sessionImpl, needUpdateCounter)) {
             sessionImpl->SetNeedUpdateActiveCounter(needUpdateCounter);
             return false;
@@ -461,6 +468,10 @@ class TQueryClient::TImpl: public TClientImplCommon<TQueryClient::TImpl>, public
         DbDriverState_->EndpointPool.BanNodeId(nodeId);
     }
 
+    void RecordSessionClosed(std::string_view reason) override {
+        SessionPool_.RecordSessionClosed(reason);
+    }
+
     void DoAttachSession(Ydb::Query::CreateSessionResponse* resp
         , NThreading::TPromise<TCreateSessionResult> promise
         , const std::string& endpoint
diff --git a/src/client/query/impl/client_session.cpp b/src/client/query/impl/client_session.cpp
index 6049ecb1ac6..83a85457347 100644
--- a/src/client/query/impl/client_session.cpp
+++ b/src/client/query/impl/client_session.cpp
@@ -3,6 +3,7 @@
 
 #define INCLUDE_YDB_INTERNAL_H
 #include <src/client/impl/internal/plain_status/status.h>
+#include <src/client/impl/session/session_pool.h>
 #undef INCLUDE_YDB_INTERNAL_H
 
 #include <src/library/issue/yql_issue_message.h>
@@ -91,7 +92,10 @@ void TSession::TImpl::StartAsyncRead(TStreamProcessorPtr ptr, std::weak_ptr<ISes
             default: {
                 auto impl = holder->TrySharedOwning();
                 if (impl) {
-                    impl->CloseFromServer(client);
+                    const auto& closeCommand = grpcStatus.GRpcStatusCode == grpc::StatusCode::OUT_OF_RANGE
+                        ? NSessionPool::NSessionCloseCommands::AttachClosed
+                        : NSessionPool::NSessionCloseCommands::TransportError;
+                    impl->CloseFromServer(client, closeCommand.Reason);
                     holder->Release();
                 }
             }
@@ -102,18 +106,23 @@ void TSession::TImpl::StartAsyncRead(TStreamProcessorPtr ptr, std::weak_ptr<ISes
 TSession::TImpl::TImpl(TStreamProcessorPtr ptr, const std::string& sessionId, const std::string& endpoint, std::weak_ptr<ISessionClient> client)
     : TKqpSessionCommon(sessionId, endpoint, true)
     , StreamProcessor_(ptr)
+    , SessionClient_(client)
     , SessionHolder(std::make_shared<TSafeTSessionImplHolder>(this))
 {
     if (ptr) {
         MarkActive();
         SetNeedUpdateActiveCounter(true);
-        StartAsyncRead(StreamProcessor_, client, SessionHolder);
+        StartAsyncRead(StreamProcessor_, SessionClient_, SessionHolder);
     } else {
         MarkBroken();
         SetNeedUpdateActiveCounter(true);
     }
 }
 
+std::shared_ptr<ISessionClient> TSession::TImpl::GetSessionClient() const {
+    return SessionClient_.lock();
+}
+
 TSession::TImpl::~TImpl() {
     if (StreamProcessor_) {
         StreamProcessor_->Cancel();
diff --git a/src/client/query/impl/client_session.h b/src/client/query/impl/client_session.h
index 5bda823a0de..69b8974299e 100644
--- a/src/client/query/impl/client_session.h
+++ b/src/client/query/impl/client_session.h
@@ -38,6 +38,8 @@ class TSession::TImpl : public TKqpSessionCommon {
 
     static void MakeImplAsync(TStreamProcessorPtr processor, std::shared_ptr<TAttachSessionArgs> args);
 
+    std::shared_ptr<ISessionClient> GetSessionClient() const override;
+
 private:
     static void NewSmartShared(TStreamProcessorPtr ptr, std::shared_ptr<TAttachSessionArgs> args, NYdb::TStatus status);
 
@@ -45,6 +47,7 @@ class TSession::TImpl : public TKqpSessionCommon {
 
 private:
     TStreamProcessorPtr StreamProcessor_;
+    std::weak_ptr<ISessionClient> SessionClient_;
     std::shared_ptr<TSafeTSessionImplHolder> SessionHolder;
 };
 
diff --git a/src/client/query/impl/exec_query.cpp b/src/client/query/impl/exec_query.cpp
index 3654d396dce..98ae334553c 100644
--- a/src/client/query/impl/exec_query.cpp
+++ b/src/client/query/impl/exec_query.cpp
@@ -322,6 +322,7 @@ class TExecQueryInternal {
         auto request = MakeRequest<Ydb::Query::ExecuteQueryRequest>();
         request.set_exec_mode(::Ydb::Query::ExecMode(settings.ExecMode_));
         request.set_stats_mode(::Ydb::Query::StatsMode(settings.StatsMode_));
+        request.set_collect_affected_rows(settings.CollectAffectedRows_);
         request.set_pool_id(TStringType{settings.ResourcePool_});
         request.mutable_query_content()->set_text(TStringType{query});
         request.mutable_query_content()->set_syntax(::Ydb::Query::Syntax(settings.Syntax_));
diff --git a/src/client/query/impl/session_state_handler.cpp b/src/client/query/impl/session_state_handler.cpp
index a5e64063b17..1dae07e81b4 100644
--- a/src/client/query/impl/session_state_handler.cpp
+++ b/src/client/query/impl/session_state_handler.cpp
@@ -1,5 +1,7 @@
 #include "session_state_handler.h"
 
+#include <src/client/impl/session/session_pool.h>
+
 namespace NYdb::inline V3::NQuery {
 
 EAttachStreamReadAction HandleAttachSessionState(
@@ -11,6 +13,9 @@ EAttachStreamReadAction HandleAttachSessionState(
         if (!session) {
             return EAttachStreamReadAction::Stop;
         }
+        const auto& closeCommand = state.has_node_shutdown()
+            ? NSessionPool::NSessionCloseCommands::NodeShutdown
+            : NSessionPool::NSessionCloseCommands::SessionShutdown;
         if (state.has_node_shutdown()) {
             const auto nodeId = session->GetEndpointKey().GetNodeId();
             if (nodeId != 0 && client) {
@@ -19,10 +24,10 @@ EAttachStreamReadAction HandleAttachSessionState(
         }
         if (session->GetState() == TKqpSessionCommon::S_IDLE) {
             if (client) {
-                session->CloseFromServer(client);
+                session->CloseFromServer(client, closeCommand.Reason);
             }
         } else {
-            session->MarkAsClosing();
+            closeCommand.Execute(*session, client.get());
         }
         return EAttachStreamReadAction::Stop;
     }
diff --git a/src/client/query/stats.cpp b/src/client/query/stats.cpp
index 22a47b21ebe..daa2bc01a4e 100644
--- a/src/client/query/stats.cpp
+++ b/src/client/query/stats.cpp
@@ -34,6 +34,7 @@ TTableAccessStats::TTableAccessStats(const Ydb::TableStats::TableAccessStats& pr
     , Updates_(proto.updates())
     , Deletes_(proto.deletes())
     , PartitionsCount_(proto.partitions_count())
+    , AffectedRows_(proto.has_affected_rows() ? std::optional<uint64_t>(proto.affected_rows()) : std::nullopt)
 {}
 
 const std::string& TTableAccessStats::GetName() const {
@@ -56,6 +57,10 @@ uint64_t TTableAccessStats::GetPartitionsCount() const {
     return PartitionsCount_;
 }
 
+std::optional<uint64_t> TTableAccessStats::GetAffectedRows() const {
+    return AffectedRows_;
+}
+
 TQueryPhaseStats::TQueryPhaseStats(const Ydb::TableStats::QueryPhaseStats& proto)
     : DurationUs_(proto.duration_us())
     , CpuTimeUs_(proto.cpu_time_us())
diff --git a/src/client/table/out.cpp b/src/client/table/out.cpp
index 718d95cf8e8..408bfa416cb 100644
--- a/src/client/table/out.cpp
+++ b/src/client/table/out.cpp
@@ -141,6 +141,9 @@ Y_DECLARE_OUT_SPEC(, NYdb::NTable::TFulltextIndexSettings::TAnalyzers, stream, v
     if (value.UseFilterSnowball.has_value()) {
         stream << ", use_filter_snowball: " << (*value.UseFilterSnowball ? "true" : "false");
     }
+    if (value.UseFilterSuperLemmer.has_value()) {
+        stream << ", use_filter_superlemmer: " << (*value.UseFilterSuperLemmer ? "true" : "false");
+    }
     stream << " }";
 }
 
diff --git a/src/client/table/table.cpp b/src/client/table/table.cpp
index f3d7aba1ca9..31fff8e66ba 100644
--- a/src/client/table/table.cpp
+++ b/src/client/table/table.cpp
@@ -3093,6 +3093,9 @@ TFulltextIndexSettings::TAnalyzers FromProto(const Ydb::Table::FulltextIndexSett
     if (proto.has_use_filter_snowball()) {
         result.UseFilterSnowball = proto.use_filter_snowball();
     }
+    if (proto.has_use_filter_superlemmer()) {
+        result.UseFilterSuperLemmer = proto.use_filter_superlemmer();
+    }
 
     return result;
 }
@@ -3153,6 +3156,9 @@ Ydb::Table::FulltextIndexSettings::Analyzers ToProto(const TFulltextIndexSetting
     if (analyzers.UseFilterSnowball.has_value()) {
         proto.set_use_filter_snowball(*analyzers.UseFilterSnowball);
     }
+    if (analyzers.UseFilterSuperLemmer.has_value()) {
+        proto.set_use_filter_superlemmer(*analyzers.UseFilterSuperLemmer);
+    }
 
     return proto;
 }
@@ -3172,6 +3178,13 @@ TFulltextIndexSettings::TAnalyzers TFulltextIndexSettings::TAnalyzers::Snowball(
     return result;
 }
 
+TFulltextIndexSettings::TAnalyzers TFulltextIndexSettings::TAnalyzers::SuperLemmer(std::string language) {
+    TAnalyzers result = Standard();
+    result.Language = std::move(language);
+    result.UseFilterSuperLemmer = true;
+    return result;
+}
+
 TFulltextIndexSettings::TAnalyzers TFulltextIndexSettings::TAnalyzers::Keyword() {
     TAnalyzers result;
     result.Tokenizer = ETokenizer::Keyword;
@@ -3503,6 +3516,9 @@ TMultiColumnStatisticsDescription TMultiColumnStatisticsDescription::FromProto(c
         case Ydb::Table::TableMultiColumnStatistics::COUNT_MIN_SKETCH:
             types.push_back(EMultiColumnStatisticsType::CountMinSketch);
             break;
+        case Ydb::Table::TableMultiColumnStatistics::EQ_HEIGHT_HISTOGRAM:
+            types.push_back(EMultiColumnStatisticsType::EqHeightHistogram);
+            break;
         default:
             types.push_back(EMultiColumnStatisticsType::Unknown);
             break;
@@ -3533,6 +3549,9 @@ void TMultiColumnStatisticsDescription::SerializeTo(Ydb::Table::TableMultiColumn
         case EMultiColumnStatisticsType::CountMinSketch:
             proto.add_types(Ydb::Table::TableMultiColumnStatistics::COUNT_MIN_SKETCH);
             break;
+        case EMultiColumnStatisticsType::EqHeightHistogram:
+            proto.add_types(Ydb::Table::TableMultiColumnStatistics::EQ_HEIGHT_HISTOGRAM);
+            break;
         case EMultiColumnStatisticsType::Unknown:
             proto.add_types(Ydb::Table::TableMultiColumnStatistics::STATISTIC_TYPE_UNSPECIFIED);
             break;
diff --git a/tests/integration/topic/direct_read_it.cpp b/tests/integration/topic/direct_read_it.cpp
index af60431c792..a795833d445 100644
--- a/tests/integration/topic/direct_read_it.cpp
+++ b/tests/integration/topic/direct_read_it.cpp
@@ -1040,7 +1040,7 @@ void SuccessfulInitImpl(bool thenTimeout) {
             });
 
         EXPECT_CALL(*setup.MockReadProcessor, OnInitRequest(_))
-            .WillOnce(Invoke([&setup](const Ydb::Topic::StreamReadMessage::InitRequest& req) {
+            .WillOnce([&setup](const Ydb::Topic::StreamReadMessage::InitRequest& req) {
                 ASSERT_EQ(req.consumer(), setup.ReadSessionSettings.ConsumerName_);
                 ASSERT_TRUE(req.direct_read());
                 ASSERT_EQ(req.topics_read_settings_size(), 1);
@@ -1049,7 +1049,7 @@ void SuccessfulInitImpl(bool thenTimeout) {
                 ASSERT_EQ(req.topics_read_settings(0).partition_ids_size(), 2);
                 ASSERT_EQ(req.topics_read_settings(0).partition_ids(0), 100);
                 ASSERT_EQ(req.topics_read_settings(0).partition_ids(1), 101);
-            }));
+            });
 
         EXPECT_CALL(*setup.MockReadProcessor, OnReadRequest(_));
     }
@@ -1098,20 +1098,20 @@ TEST_F(DirectReadWithControlSession, StopPartitionSessionGracefully) {
                 });
 
             EXPECT_CALL(*setup.MockReadProcessor, OnInitRequest(_))
-                .WillOnce(Invoke([&](const Ydb::Topic::StreamReadMessage::InitRequest& req) {
+                .WillOnce([&](const Ydb::Topic::StreamReadMessage::InitRequest& req) {
                     ASSERT_TRUE(req.direct_read());
                     ASSERT_EQ(req.topics_read_settings_size(), 1);
                     ASSERT_EQ(req.topics_read_settings(0).path(), setup.ReadSessionSettings.Topics_[0].Path_);
                     ASSERT_EQ(req.topics_read_settings(0).partition_ids_size(), 1);
                     ASSERT_EQ(req.topics_read_settings(0).partition_ids(0), startPartitionSessionRequest.PartitionId);
-                }));
+                });
 
             EXPECT_CALL(*setup.MockReadProcessor, OnReadRequest(_));
 
             EXPECT_CALL(*setup.MockReadProcessor, OnStartPartitionSessionResponse(_))
-                .WillOnce(Invoke([&startPartitionSessionRequest](const Ydb::Topic::StreamReadMessage::StartPartitionSessionResponse& resp) {
+                .WillOnce([&startPartitionSessionRequest](const Ydb::Topic::StreamReadMessage::StartPartitionSessionResponse& resp) {
                     ASSERT_EQ(static_cast<std::uint64_t>(resp.partition_session_id()), startPartitionSessionRequest.PartitionSessionId);
-                }));
+                });
 
             EXPECT_CALL(*setup.MockReadProcessor, OnDirectReadAck(_))
                 .Times(4);
@@ -1129,18 +1129,18 @@ TEST_F(DirectReadWithControlSession, StopPartitionSessionGracefully) {
                 });
 
             EXPECT_CALL(*setup.MockDirectReadProcessor, OnInitRequest(_))
-                .WillOnce(Invoke([&setup](const Ydb::Topic::StreamDirectReadMessage::InitRequest& req) {
+                .WillOnce([&setup](const Ydb::Topic::StreamDirectReadMessage::InitRequest& req) {
                     ASSERT_EQ(req.session_id(), SERVER_SESSION_ID);
                     ASSERT_EQ(static_cast<std::size_t>(req.topics_read_settings_size()), setup.ReadSessionSettings.Topics_.size());
                     ASSERT_EQ(req.topics_read_settings(0).path(), setup.ReadSessionSettings.Topics_[0].Path_);
                     ASSERT_EQ(req.consumer(), setup.ReadSessionSettings.ConsumerName_);
-                }));
+                });
 
             EXPECT_CALL(*setup.MockDirectReadProcessor, OnStartDirectReadPartitionSessionRequest(_))
-                .WillOnce(Invoke([&startPartitionSessionRequest](const Ydb::Topic::StreamDirectReadMessage::StartDirectReadPartitionSessionRequest& request) {
+                .WillOnce([&startPartitionSessionRequest](const Ydb::Topic::StreamDirectReadMessage::StartDirectReadPartitionSessionRequest& request) {
                     ASSERT_EQ(static_cast<std::uint64_t>(request.partition_session_id()), startPartitionSessionRequest.PartitionSessionId);
                     ASSERT_EQ(request.generation(), startPartitionSessionRequest.Generation);
-                }));
+                });
 
             // Expect OnReadRequest in case it is called before the test ends.
             // TODO(qyryq) Fix number, not 10.
@@ -1246,20 +1246,20 @@ TEST_F(DirectReadWithControlSession, StopPartitionSession) {
                 });
 
             EXPECT_CALL(*setup.MockReadProcessor, OnInitRequest(_))
-                .WillOnce(Invoke([&](const Ydb::Topic::StreamReadMessage::InitRequest& req) {
+                .WillOnce([&](const Ydb::Topic::StreamReadMessage::InitRequest& req) {
                     ASSERT_TRUE(req.direct_read());
                     ASSERT_EQ(req.topics_read_settings_size(), 1);
                     ASSERT_EQ(req.topics_read_settings(0).path(), setup.ReadSessionSettings.Topics_[0].Path_);
                     ASSERT_EQ(req.topics_read_settings(0).partition_ids_size(), 1);
                     ASSERT_EQ(req.topics_read_settings(0).partition_ids(0), startPartitionSessionRequest.PartitionId);
-                }));
+                });
 
             EXPECT_CALL(*setup.MockReadProcessor, OnReadRequest(_));
 
             EXPECT_CALL(*setup.MockReadProcessor, OnStartPartitionSessionResponse(_))
-                .WillOnce(Invoke([&startPartitionSessionRequest](const Ydb::Topic::StreamReadMessage::StartPartitionSessionResponse& resp) {
+                .WillOnce([&startPartitionSessionRequest](const Ydb::Topic::StreamReadMessage::StartPartitionSessionResponse& resp) {
                     ASSERT_EQ(static_cast<std::uint64_t>(resp.partition_session_id()), startPartitionSessionRequest.PartitionSessionId);
-                }));
+                });
 
             EXPECT_CALL(*setup.MockReadProcessor, OnDirectReadAck(_))
                 .Times(4);
@@ -1277,18 +1277,18 @@ TEST_F(DirectReadWithControlSession, StopPartitionSession) {
                 });
 
             EXPECT_CALL(*setup.MockDirectReadProcessor, OnInitRequest(_))
-                .WillOnce(Invoke([&setup](const Ydb::Topic::StreamDirectReadMessage::InitRequest& req) {
+                .WillOnce([&setup](const Ydb::Topic::StreamDirectReadMessage::InitRequest& req) {
                     ASSERT_EQ(req.session_id(), SERVER_SESSION_ID);
                     ASSERT_EQ(static_cast<std::size_t>(req.topics_read_settings_size()), setup.ReadSessionSettings.Topics_.size());
                     ASSERT_EQ(req.topics_read_settings(0).path(), setup.ReadSessionSettings.Topics_[0].Path_);
                     ASSERT_EQ(req.consumer(), setup.ReadSessionSettings.ConsumerName_);
-                }));
+                });
 
             EXPECT_CALL(*setup.MockDirectReadProcessor, OnStartDirectReadPartitionSessionRequest(_))
-                .WillOnce(Invoke([&startPartitionSessionRequest](const Ydb::Topic::StreamDirectReadMessage::StartDirectReadPartitionSessionRequest& request) {
+                .WillOnce([&startPartitionSessionRequest](const Ydb::Topic::StreamDirectReadMessage::StartDirectReadPartitionSessionRequest& request) {
                     ASSERT_EQ(static_cast<std::uint64_t>(request.partition_session_id()), startPartitionSessionRequest.PartitionSessionId);
                     ASSERT_EQ(request.generation(), startPartitionSessionRequest.Generation);
-                }));
+                });
 
             // Expect OnReadRequest in case it is called before the test ends.
             // TODO(qyryq) Fix number, not 10.
@@ -1421,28 +1421,28 @@ TEST_F(DirectReadWithControlSession, EmptyDirectReadResponse) {
                 });
 
             EXPECT_CALL(*setup.MockReadProcessor, OnInitRequest(_))
-                .WillOnce(Invoke([&](const Ydb::Topic::StreamReadMessage::InitRequest& req) {
+                .WillOnce([&](const Ydb::Topic::StreamReadMessage::InitRequest& req) {
                     ASSERT_TRUE(req.direct_read());
                     ASSERT_EQ(req.topics_read_settings_size(), 1);
                     ASSERT_EQ(req.topics_read_settings(0).path(), setup.ReadSessionSettings.Topics_[0].Path_);
                     ASSERT_EQ(req.topics_read_settings(0).partition_ids_size(), 1);
                     ASSERT_EQ(req.topics_read_settings(0).partition_ids(0), startPartitionSessionRequest.PartitionId);
-                }));
+                });
 
             EXPECT_CALL(*setup.MockReadProcessor, OnReadRequest(_));
 
             EXPECT_CALL(*setup.MockReadProcessor, OnStartPartitionSessionResponse(_))
-                .WillOnce(Invoke([&startPartitionSessionRequest](const Ydb::Topic::StreamReadMessage::StartPartitionSessionResponse& resp) {
+                .WillOnce([&startPartitionSessionRequest](const Ydb::Topic::StreamReadMessage::StartPartitionSessionResponse& resp) {
                     ASSERT_EQ(static_cast<std::uint64_t>(resp.partition_session_id()), startPartitionSessionRequest.PartitionSessionId);
-                }));
+                });
 
             EXPECT_CALL(*setup.MockReadProcessor, OnDirectReadAck(_))
                 .Times(1);
 
             EXPECT_CALL(*setup.MockReadProcessor, OnReadRequest(_))
-                .WillOnce(Invoke([&](const Ydb::Topic::StreamReadMessage::ReadRequest& req) {
+                .WillOnce([&](const Ydb::Topic::StreamReadMessage::ReadRequest& req) {
                     ASSERT_EQ(req.bytes_size(), bytesSize);
-                }));
+                });
         }
 
         // There are two sequences, because OnCreateProcessor from the second sequence may be called
@@ -1457,18 +1457,18 @@ TEST_F(DirectReadWithControlSession, EmptyDirectReadResponse) {
                 });
 
             EXPECT_CALL(*setup.MockDirectReadProcessor, OnInitRequest(_))
-                .WillOnce(Invoke([&setup](const Ydb::Topic::StreamDirectReadMessage::InitRequest& req) {
+                .WillOnce([&setup](const Ydb::Topic::StreamDirectReadMessage::InitRequest& req) {
                     ASSERT_EQ(req.session_id(), SERVER_SESSION_ID);
                     ASSERT_EQ(static_cast<std::size_t>(req.topics_read_settings_size()), setup.ReadSessionSettings.Topics_.size());
                     ASSERT_EQ(req.topics_read_settings(0).path(), setup.ReadSessionSettings.Topics_[0].Path_);
                     ASSERT_EQ(req.consumer(), setup.ReadSessionSettings.ConsumerName_);
-                }));
+                });
 
             EXPECT_CALL(*setup.MockDirectReadProcessor, OnStartDirectReadPartitionSessionRequest(_))
-                .WillOnce(Invoke([&startPartitionSessionRequest](const Ydb::Topic::StreamDirectReadMessage::StartDirectReadPartitionSessionRequest& request) {
+                .WillOnce([&startPartitionSessionRequest](const Ydb::Topic::StreamDirectReadMessage::StartDirectReadPartitionSessionRequest& request) {
                     ASSERT_EQ(static_cast<std::uint64_t>(request.partition_session_id()), startPartitionSessionRequest.PartitionSessionId);
                     ASSERT_EQ(request.generation(), startPartitionSessionRequest.Generation);
-                }));
+                });
         }
     }
 
@@ -1539,16 +1539,16 @@ TEST_F(DirectReadSession, InitAndStartPartitionSession) {
             .WillOnce([&]() { setup.MockDirectReadProcessorFactory->CreateProcessor(setup.MockDirectReadProcessor); });
 
         EXPECT_CALL(*setup.MockDirectReadProcessor, OnInitRequest(_))
-            .WillOnce(Invoke([&](const Ydb::Topic::StreamDirectReadMessage::InitRequest& req) {
+            .WillOnce([&](const Ydb::Topic::StreamDirectReadMessage::InitRequest& req) {
                 ASSERT_EQ(req.session_id(), SERVER_SESSION_ID);
                 ASSERT_EQ(req.consumer(), setup.ReadSessionSettings.ConsumerName_);
-            }));
+            });
 
         EXPECT_CALL(*setup.MockDirectReadProcessor, OnStartDirectReadPartitionSessionRequest(_))
-            .WillOnce(Invoke([&](const Ydb::Topic::StreamDirectReadMessage::StartDirectReadPartitionSessionRequest& req) {
+            .WillOnce([&](const Ydb::Topic::StreamDirectReadMessage::StartDirectReadPartitionSessionRequest& req) {
                 ASSERT_EQ(req.partition_session_id(), static_cast<std::int64_t>(partitionSessionId));
                 gotStart.SetValue();
-            }));
+            });
     }
 
     session->Start();
diff --git a/tests/unit/client/observability/metrics_ut.cpp b/tests/unit/client/observability/metrics_ut.cpp
index 625850f6521..1b354b97119 100644
--- a/tests/unit/client/observability/metrics_ut.cpp
+++ b/tests/unit/client/observability/metrics_ut.cpp
@@ -445,6 +445,15 @@ TEST_F(QueryPoolMetricsTest, TimeoutsIncrement) {
     EXPECT_EQ(counter->Get(), 3);
 }
 
+TEST_F(QueryPoolMetricsTest, ClosedSessionReason) {
+    Collector.IncSessionClosed("transport_error");
+    auto labels = QueryPoolLabels(kTestPoolName);
+    labels["reason"] = "transport_error";
+    auto counter = Registry->GetCounter("ydb.query.session.closed", labels);
+    ASSERT_NE(counter, nullptr);
+    EXPECT_EQ(counter->Get(), 1);
+}
+
 TEST_F(QueryPoolMetricsTest, SessionCountSplitsByState) {
     Collector.UpdateConnectionCount(/*idle=*/8, /*used=*/3);
 
diff --git a/tests/unit/client/query/client_session_ut.cpp b/tests/unit/client/query/client_session_ut.cpp
index cf9610f2390..c287cb2e976 100644
--- a/tests/unit/client/query/client_session_ut.cpp
+++ b/tests/unit/client/query/client_session_ut.cpp
@@ -1,5 +1,9 @@
 #include <src/client/query/impl/session_state_handler.h>
 
+#define INCLUDE_YDB_INTERNAL_H
+#include <src/client/impl/session/session_pool.h>
+#undef INCLUDE_YDB_INTERNAL_H
+
 #include <ydb-cpp-sdk/library/operation_id/operation_id.h>
 
 #include <library/cpp/testing/unittest/registar.h>
@@ -24,13 +28,22 @@ class TMockSessionClient : public ISessionClient {
         return true;
     }
 
+    void RecordSessionClosed(std::string_view reason) override {
+        LastReason = std::string(reason);
+        ++CloseMetrics;
+    }
+
     std::uint64_t PessimizedNodeId = 0;
     int PessimizeCalls = 0;
+    std::string LastReason;
+    int CloseMetrics = 0;
 };
 
 class TMockServerCloseHandler : public IServerCloseHandler {
 public:
-    void OnCloseSession(const TKqpSessionCommon*, std::shared_ptr<ISessionClient>) override {
+    void OnCloseSession(const TKqpSessionCommon*, std::shared_ptr<ISessionClient>,
+        std::string_view) override
+    {
         ++CloseCalls;
     }
 
@@ -133,4 +146,32 @@ Y_UNIT_TEST(SessionShutdownNullSessionStopsReading) {
     UNIT_ASSERT_VALUES_EQUAL(client->PessimizeCalls, 0);
 }
 
+Y_UNIT_TEST(CloseReasonCommandsAreCompleteAndDeduplicated) {
+    const std::pair<const NSessionPool::TSessionCloseCommand*, std::string_view> commands[] = {
+        {&NSessionPool::NSessionCloseCommands::PoolIdleTimeout, "pool_idle_timeout"},
+        {&NSessionPool::NSessionCloseCommands::PoolGracefulShutdown, "pool_graceful_shutdown"},
+        {&NSessionPool::NSessionCloseCommands::ClientTimeout, "client_timeout"},
+        {&NSessionPool::NSessionCloseCommands::ClientCancelled, "client_cancelled"},
+        {&NSessionPool::NSessionCloseCommands::AttachClosed, "attach_closed"},
+        {&NSessionPool::NSessionCloseCommands::TransportError, "transport_error"},
+        {&NSessionPool::NSessionCloseCommands::NodeShutdown, "node_shutdown"},
+        {&NSessionPool::NSessionCloseCommands::SessionShutdown, "session_shutdown"},
+        {&NSessionPool::NSessionCloseCommands::BadSession, "bad_session"},
+        {&NSessionPool::NSessionCloseCommands::SessionBusy, "session_busy"},
+    };
+    auto client = std::make_shared<TMockSessionClient>();
+    for (const auto& [command, reason] : commands) {
+        TTestKqpSession session("", "");
+        command->Execute(session, client.get());
+        command->Execute(session, client.get());
+        UNIT_ASSERT_VALUES_EQUAL(client->LastReason, reason);
+    }
+    UNIT_ASSERT_VALUES_EQUAL(client->CloseMetrics, 10);
+
+    TKqpSessionCommon standalone("", "", false);
+    NSessionPool::NSessionCloseCommands::BadSession.Execute(standalone, client.get());
+    UNIT_ASSERT_VALUES_EQUAL(client->CloseMetrics, 10);
+    UNIT_ASSERT(!NSessionPool::NSessionCloseCommands::FromStatus(TStatus(EStatus::SESSION_EXPIRED, {})));
+}
+
 }
diff --git a/tests/unit/client/table/table_ut.cpp b/tests/unit/client/table/table_ut.cpp
index 023bcc68274..61fc95843a0 100644
--- a/tests/unit/client/table/table_ut.cpp
+++ b/tests/unit/client/table/table_ut.cpp
@@ -1,11 +1,14 @@
 #include <ydb-cpp-sdk/client/driver/driver.h>
+#include <ydb-cpp-sdk/client/proto/accessor.h>
 #include <ydb-cpp-sdk/client/table/table.h>
 
 #include <library/cpp/testing/common/network.h>
 
 #include <util/string/builder.h>
+#include <util/string/cast.h>
 
 #include <src/api/grpc/ydb_table_v1.grpc.pb.h>
+#include <src/api/protos/ydb_table.pb.h>
 
 #include <grpcpp/server.h>
 #include <grpcpp/server_builder.h>
@@ -203,6 +206,30 @@ namespace {
 
 } // namespace <anonymous>
 
+TEST(TableTest, FulltextSuperLemmerAnalyzerRoundTrip) {
+    NTable::TFulltextIndexSettings settings;
+    NTable::TFulltextIndexSettings::TColumnAnalyzers column;
+    column.Column = "Text";
+    column.Analyzers = NTable::TFulltextIndexSettings::TAnalyzers::SuperLemmer("russian");
+    settings.Columns.push_back(column);
+
+    Ydb::Table::FulltextIndexSettings proto;
+    settings.SerializeTo(proto);
+    ASSERT_EQ(proto.columns_size(), 1);
+    ASSERT_TRUE(proto.columns(0).has_analyzers());
+    ASSERT_TRUE(proto.columns(0).analyzers().use_filter_superlemmer());
+
+    const auto restored = NTable::TFulltextIndexSettings::FromProto(proto);
+    ASSERT_EQ(restored.Columns.size(), 1);
+    ASSERT_TRUE(restored.Columns[0].Analyzers.has_value());
+    const auto& analyzers = *restored.Columns[0].Analyzers;
+    ASSERT_EQ(analyzers.Language.value_or(""), "russian");
+    ASSERT_TRUE(analyzers.UseFilterLowercase.value_or(false));
+    ASSERT_TRUE(analyzers.UseFilterStopwords.value_or(false));
+    ASSERT_TRUE(analyzers.UseFilterSuperLemmer.value_or(false));
+    ASSERT_NE(ToString(restored).find("use_filter_superlemmer: true"), TString::npos);
+}
+
 TEST(TableTest, SessionHandleDestructionSendsDeleteSession) {
     TMockTableService tableService;
     std::unique_ptr<grpc::Server> grpcServer;
@@ -756,6 +783,29 @@ TEST(TableTest, AlterTableDroppedMetricsSettings) {
     ASSERT_TRUE(tableService.LastAlterTableRequest->has_drop_metrics_settings());
 }
 
+/**
+ * Verify proto round-trip for equi-height histogram multi-column statistics.
+ */
+TEST(TableTest, MultiColumnStatisticsEqHeightHistogramRoundTrip) {
+    NTable::TMultiColumnStatisticsDescription desc(
+        "h1",
+        {"a", "b"},
+        {NTable::EMultiColumnStatisticsType::EqHeightHistogram});
+
+    Ydb::Table::TableMultiColumnStatistics proto;
+    desc.SerializeTo(proto);
+    ASSERT_EQ(proto.name(), "h1");
+    ASSERT_EQ(proto.columns_size(), 2);
+    ASSERT_EQ(proto.types_size(), 1);
+    ASSERT_EQ(proto.types(0), Ydb::Table::TableMultiColumnStatistics::EQ_HEIGHT_HISTOGRAM);
+
+    auto roundTrip = TProtoAccessor::FromProto(proto);
+    ASSERT_EQ(roundTrip.GetName(), "h1");
+    ASSERT_EQ(roundTrip.GetColumns().size(), 2u);
+    ASSERT_EQ(roundTrip.GetTypes().size(), 1u);
+    ASSERT_EQ(roundTrip.GetTypes()[0], NTable::EMultiColumnStatisticsType::EqHeightHistogram);
+}
+
 /**
  * Verify that the SDK creates the ALTER TABLE request correctly,
  * when the metrics configuration is explicitly set.
diff --git a/util/datetime/base.h b/util/datetime/base.h
index f5e9c3ff147..729d3eeec06 100644
--- a/util/datetime/base.h
+++ b/util/datetime/base.h
@@ -5,6 +5,7 @@
 #include <util/str_stl.h>
 #include <util/system/platform.h>
 #include <util/system/datetime.h>
+#include <util/generic/constant_evaluation.h>
 #include <util/generic/string.h>
 #include <util/generic/strbuf.h>
 #include <util/generic/ylimits.h>
@@ -115,8 +116,8 @@ namespace NDateTimeHelpers {
     template <ui64 b>
     constexpr ui64 MulWithSaturation(ui64 a) {
         constexpr ui64 maxMultiplicand = Max<ui64>() / b;
-#if defined(__GNUC__) && defined(__cpp_lib_is_constant_evaluated)
-        if (!std::is_constant_evaluated()) {
+#if Y_HAS_BUILTIN(__builtin_umull_overflow) && Y_HAS_BUILTIN(__builtin_umulll_overflow)
+        if (!IsConstantEvaluated()) {
             if constexpr (std::is_same<ui64, unsigned long>::value) {
                 unsigned long r = 0;
                 return __builtin_umull_overflow(a, b, &r) ? Max<ui64>() : r;
diff --git a/util/generic/bitops.h b/util/generic/bitops.h
index 601daf7a309..72dd18e662e 100644
--- a/util/generic/bitops.h
+++ b/util/generic/bitops.h
@@ -1,11 +1,14 @@
 #pragma once
 
+#include "constant_evaluation.h"
 #include "ylimits.h"
 #include "typelist.h"
 
 #include <util/system/compiler.h>
 #include <util/system/yassert.h>
 
+#include <type_traits>
+
 #ifdef _MSC_VER
     #include <intrin.h>
 #endif
@@ -129,7 +132,7 @@ namespace NBitOps {
         }
 
         template <typename T>
-        Y_FORCE_INLINE T RotateBitsRightImpl(T value, const ui8 shift) noexcept {
+        Y_FORCE_INLINE constexpr T RotateBitsRightImpl(T value, const ui8 shift) noexcept {
             constexpr ui8 bits = sizeof(T) * 8;
             constexpr ui8 mask = bits - 1;
             Y_ASSERT(shift <= mask);
@@ -404,9 +407,14 @@ Y_FORCE_INLINE T RotateBitsLeft(T value, const ui8 shift) noexcept {
 /* Rotate bits right. Also known as right circular shift.
  */
 template <typename T>
-Y_FORCE_INLINE T RotateBitsRight(T value, const ui8 shift) noexcept {
+Y_FORCE_INLINE constexpr T RotateBitsRight(T value, const ui8 shift) noexcept {
     static_assert(std::is_unsigned<T>::value, "must be unsigned arithmetic type");
-    return ::NBitOps::NPrivate::RotateBitsRightImpl((TFixedWidthUnsignedInt<T>)value, shift);
+    using TFixedWidth = TFixedWidthUnsignedInt<T>;
+    if (!IsConstantEvaluated()) {
+        return ::NBitOps::NPrivate::RotateBitsRightImpl(static_cast<TFixedWidth>(value), shift);
+    }
+    // Explicit template arguments force the constexpr fallback; runtime overload resolution prefers non-template assembly overloads.
+    return ::NBitOps::NPrivate::RotateBitsRightImpl<TFixedWidth>(static_cast<TFixedWidth>(value), shift);
 }
 
 /* Rotate bits left. Also known as left circular shift.
@@ -423,10 +431,7 @@ constexpr T RotateBitsLeftCT(T value, const ui8 shift) noexcept {
  */
 template <typename T>
 constexpr T RotateBitsRightCT(T value, const ui8 shift) noexcept {
-    static_assert(std::is_unsigned<T>::value, "must be unsigned arithmetic type");
-
-    // do trick with mask to avoid undefined behaviour
-    return (value >> shift) | (value << ((-shift) & (sizeof(T) * 8 - 1)));
+    return RotateBitsRight(value, shift);
 }
 
 /* Remain `size` bits to current `offset` of `value`
diff --git a/util/generic/bitops_ut.cpp b/util/generic/bitops_ut.cpp
index 90f4bc1c3ce..737ffbcc174 100644
--- a/util/generic/bitops_ut.cpp
+++ b/util/generic/bitops_ut.cpp
@@ -222,6 +222,7 @@ Y_UNIT_TEST_SUITE(TBitOpsTest) {
 
     Y_UNIT_TEST(TestRotateBitsRight) {
         static_assert(RotateBitsRightCT<ui8>(0b00000000u, 0) == 0b00000000u, "");
+        static_assert(RotateBitsRight<ui8>(0b00000000u, 0) == 0b00000000u, "");
         static_assert(RotateBitsRightCT<ui8>(0b00000001u, 0) == 0b00000001u, "");
         static_assert(RotateBitsRightCT<ui8>(0b10000000u, 0) == 0b10000000u, "");
         static_assert(RotateBitsRightCT<ui8>(0b00000001u, 1) == 0b10000000u, "");
@@ -240,6 +241,7 @@ Y_UNIT_TEST_SUITE(TBitOpsTest) {
         UNIT_ASSERT_VALUES_EQUAL(RotateBitsRight<ui8>(0b00000001u, 7), 0b00000010u);
 
         static_assert(RotateBitsRightCT<ui16>(0b0000000000000000u, 0) == 0b0000000000000000u, "");
+        static_assert(RotateBitsRight<ui16>(0b0000000000000000u, 0) == 0b0000000000000000u, "");
         static_assert(RotateBitsRightCT<ui16>(0b0000000000000001u, 0) == 0b0000000000000001u, "");
         static_assert(RotateBitsRightCT<ui16>(0b1000000000000000u, 0) == 0b1000000000000000u, "");
         static_assert(RotateBitsRightCT<ui16>(0b0000000000000001u, 1) == 0b1000000000000000u, "");
@@ -258,6 +260,7 @@ Y_UNIT_TEST_SUITE(TBitOpsTest) {
         UNIT_ASSERT_VALUES_EQUAL(RotateBitsRight<ui16>(0b0000000000000001u, 15), 0b0000000000000010u);
 
         static_assert(RotateBitsRightCT<ui32>(0b00000000000000000000000000000000u, 0) == 0b00000000000000000000000000000000u, "");
+        static_assert(RotateBitsRight<ui32>(0b00000000000000000000000000000000u, 0) == 0b00000000000000000000000000000000u, "");
         static_assert(RotateBitsRightCT<ui32>(0b00000000000000000000000000000001u, 0) == 0b00000000000000000000000000000001u, "");
         static_assert(RotateBitsRightCT<ui32>(0b10000000000000000000000000000000u, 0) == 0b10000000000000000000000000000000u, "");
         static_assert(RotateBitsRightCT<ui32>(0b00000000000000000000000000000001u, 1) == 0b10000000000000000000000000000000u, "");
@@ -276,6 +279,7 @@ Y_UNIT_TEST_SUITE(TBitOpsTest) {
         UNIT_ASSERT_VALUES_EQUAL(RotateBitsRight<ui32>(0b00000000000000000000000000000001u, 31), 0b00000000000000000000000000000010u);
 
         static_assert(RotateBitsRightCT<ui64>(0b0000000000000000000000000000000000000000000000000000000000000000u, 0) == 0b0000000000000000000000000000000000000000000000000000000000000000u, "");
+        static_assert(RotateBitsRight<ui64>(0b0000000000000000000000000000000000000000000000000000000000000000u, 0) == 0b0000000000000000000000000000000000000000000000000000000000000000u, "");
         static_assert(RotateBitsRightCT<ui64>(0b0000000000000000000000000000000000000000000000000000000000000001u, 0) == 0b0000000000000000000000000000000000000000000000000000000000000001u, "");
         static_assert(RotateBitsRightCT<ui64>(0b1000000000000000000000000000000000000000000000000000000000000000u, 0) == 0b1000000000000000000000000000000000000000000000000000000000000000u, "");
         static_assert(RotateBitsRightCT<ui64>(0b0000000000000000000000000000000000000000000000000000000000000001u, 1) == 0b1000000000000000000000000000000000000000000000000000000000000000u, "");
diff --git a/util/generic/constant_evaluation.cpp b/util/generic/constant_evaluation.cpp
new file mode 100644
index 00000000000..202ba46dce2
--- /dev/null
+++ b/util/generic/constant_evaluation.cpp
@@ -0,0 +1 @@
+#include "constant_evaluation.h"
diff --git a/util/generic/constant_evaluation.h b/util/generic/constant_evaluation.h
new file mode 100644
index 00000000000..bab53771c17
--- /dev/null
+++ b/util/generic/constant_evaluation.h
@@ -0,0 +1,19 @@
+#pragma once
+
+#include <util/system/compiler.h>
+
+#include <type_traits>
+
+/// Reports whether the current expression is evaluated at compile time.
+///
+/// Prefers the standard API, uses the compiler builtin in pre-C++20 modes, and
+/// conservatively returns true when neither check is available.
+constexpr bool IsConstantEvaluated() noexcept {
+#if defined(__cpp_lib_is_constant_evaluated)
+    return std::is_constant_evaluated();
+#elif Y_HAS_BUILTIN(__builtin_is_constant_evaluated)
+    return __builtin_is_constant_evaluated();
+#else
+    return true;
+#endif
+}
diff --git a/util/generic/utility_ut.cpp b/util/generic/utility_ut.cpp
index b4b8739794b..08d72b6eb6a 100644
--- a/util/generic/utility_ut.cpp
+++ b/util/generic/utility_ut.cpp
@@ -1,3 +1,4 @@
+#include "constant_evaluation.h"
 #include "utility.h"
 #include "ymath.h"
 
@@ -35,6 +36,15 @@ static bool operator>(const TUnorderedTag, const TUnorderedTag) = delete;
 
 Y_UNIT_TEST_SUITE(TUtilityTest) {
 
+    Y_UNIT_TEST(TestIsConstantEvaluated) {
+        static_assert(IsConstantEvaluated());
+#if defined(__cpp_lib_is_constant_evaluated) || Y_HAS_BUILTIN(__builtin_is_constant_evaluated)
+        UNIT_ASSERT(!IsConstantEvaluated());
+#else
+        UNIT_ASSERT(IsConstantEvaluated());
+#endif
+    }
+
     Y_UNIT_TEST(TestSwapPrimitive) {
         int i = 0;
         int j = 1;
diff --git a/util/random/common_ops.h b/util/random/common_ops.h
index 0bbb80f3d3e..d3fed6ef60c 100644
--- a/util/random/common_ops.h
+++ b/util/random/common_ops.h
@@ -33,7 +33,7 @@ namespace NPrivate {
     }
 
     template <class T>
-    static inline ui64 ToRand64(T&& rng, ui32 x) noexcept {
+    static constexpr ui64 ToRand64(T&& rng, ui32 x) noexcept {
         return ((ui64)x) | (((ui64)rng.GenRand()) << 32);
     }
 
@@ -65,7 +65,7 @@ struct TCommonRNG {
     using TResult = TRandType;
     using result_type = TRandType;
 
-    inline T& Engine() noexcept {
+    constexpr T& Engine() noexcept {
         return static_cast<T&>(*this);
     }
 
@@ -91,7 +91,7 @@ struct TCommonRNG {
     }
 
     /* generates 64-bit random number for current(may be 32 bit) rng */
-    inline ui64 GenRand64() noexcept {
+    constexpr ui64 GenRand64() noexcept {
         return ::NPrivate::ToRand64(Engine(), Engine().GenRand());
     }
 
@@ -116,7 +116,7 @@ struct TCommonRNG {
     }
 
     // compatibility stuff
-    inline TResult operator()() noexcept {
+    constexpr TResult operator()() noexcept {
         return Engine().GenRand();
     }
 
diff --git a/util/random/fast.cpp b/util/random/fast.cpp
index 2f98dfc5d35..027f71ac6df 100644
--- a/util/random/fast.cpp
+++ b/util/random/fast.cpp
@@ -2,31 +2,6 @@
 
 #include <util/stream/input.h>
 
-static inline ui32 FixSeq(ui32 seq1, ui32 seq2) noexcept {
-    const ui32 mask = (~(ui32)(0)) >> 1;
-
-    if ((seq1 & mask) == (seq2 & mask)) {
-        return ~seq2;
-    }
-
-    return seq2;
-}
-
-TFastRng64::TFastRng64(ui64 seed1, ui32 seq1, ui64 seed2, ui32 seq2) noexcept
-    : R1_(seed1, seq1)
-    , R2_(seed2, FixSeq(seq1, seq2))
-{
-}
-
-TFastRng64::TArgs::TArgs(ui64 seed) noexcept {
-    TReallyFastRng32 rng(seed);
-
-    Seed1 = rng.GenRand64();
-    Seq1 = rng.GenRand();
-    Seed2 = rng.GenRand64();
-    Seq2 = rng.GenRand();
-}
-
 TFastRng64::TArgs::TArgs(IInputStream& entropy) {
     static_assert(sizeof(*this) == 3 * sizeof(ui64), "please, fix me");
     entropy.LoadOrFail(this, sizeof(*this));
diff --git a/util/random/fast.h b/util/random/fast.h
index 70359135a24..fbd0c926290 100644
--- a/util/random/fast.h
+++ b/util/random/fast.h
@@ -9,7 +9,7 @@
 // based on http://www.pcg-random.org/. See T*FastRng* family below.
 
 struct TPCGMixer {
-    static inline ui32 Mix(ui64 x) noexcept {
+    static constexpr ui32 Mix(ui64 x) noexcept {
         const ui32 xorshifted = ((x >> 18u) ^ x) >> 27u;
         const ui32 rot = x >> 59u;
 
@@ -33,7 +33,7 @@ struct TFastRng32: public TCommonRNG<ui32, TFastRng32>, public TFastRng32Base {
 
 // faster than TFastRng32, but have only one possible stream sequence
 struct TReallyFastRng32: public TCommonRNG<ui32, TReallyFastRng32>, public TReallyFastRng32Base {
-    inline TReallyFastRng32(ui64 seed)
+    constexpr TReallyFastRng32(ui64 seed)
         : TReallyFastRng32Base(seed)
     {
     }
@@ -44,7 +44,15 @@ struct TReallyFastRng32: public TCommonRNG<ui32, TReallyFastRng32>, public TReal
 class TFastRng64: public TCommonRNG<ui64, TFastRng64> {
 public:
     struct TArgs {
-        TArgs(ui64 seed) noexcept;
+        constexpr TArgs(ui64 seed) noexcept {
+            TReallyFastRng32 rng(seed);
+
+            Seed1 = rng.GenRand64();
+            Seq1 = rng.GenRand();
+            Seed2 = rng.GenRand64();
+            Seq2 = rng.GenRand();
+        }
+
         TArgs(IInputStream& entropy);
 
         ui64 Seed1;
@@ -53,19 +61,23 @@ class TFastRng64: public TCommonRNG<ui64, TFastRng64> {
         ui32 Seq2;
     };
 
-    TFastRng64(ui64 seed1, ui32 seq1, ui64 seed2, ui32 seq2) noexcept;
+    constexpr TFastRng64(ui64 seed1, ui32 seq1, ui64 seed2, ui32 seq2) noexcept
+        : R1_(seed1, seq1)
+        , R2_(seed2, FixSeq(seq1, seq2))
+    {
+    }
 
     /*
      * simplify constructions like
      *     TFastRng64 rng(17);
      *     TFastRng64 rng(Seek()); //from any IInputStream
      */
-    inline TFastRng64(const TArgs& args) noexcept
+    constexpr TFastRng64(const TArgs& args) noexcept
         : TFastRng64(args.Seed1, args.Seq1, args.Seed2, args.Seq2)
     {
     }
 
-    inline ui64 GenRand() noexcept {
+    constexpr ui64 GenRand() noexcept {
         const ui64 x = R1_.GenRand();
         const ui64 y = R2_.GenRand();
 
@@ -78,6 +90,11 @@ class TFastRng64: public TCommonRNG<ui64, TFastRng64> {
     }
 
 private:
+    static constexpr ui32 FixSeq(ui32 seq1, ui32 seq2) noexcept {
+        constexpr ui32 mask = ~ui32{0} >> 1;
+        return (seq1 & mask) == (seq2 & mask) ? ~seq2 : seq2;
+    }
+
     TFastRng32Base R1_;
     TFastRng32Base R2_;
 };
diff --git a/util/random/fast_ut.cpp b/util/random/fast_ut.cpp
index 9254d0bbe12..014680d1160 100644
--- a/util/random/fast_ut.cpp
+++ b/util/random/fast_ut.cpp
@@ -2,6 +2,11 @@
 
 #include <library/cpp/testing/unittest/registar.h>
 
+static_assert([] {
+    TFastRng64 rng(17);
+    return rng.GenRand() == ULL(14895365814383052362);
+}());
+
 Y_UNIT_TEST_SUITE(TTestFastRng) {
     Y_UNIT_TEST(Test1) {
         TFastRng32 rng1(17, 0);
diff --git a/util/random/lcg_engine.h b/util/random/lcg_engine.h
index 88e79cabf66..7222867ce72 100644
--- a/util/random/lcg_engine.h
+++ b/util/random/lcg_engine.h
@@ -26,12 +26,12 @@ struct TFastLcgIterator {
 
 template <typename T, T A>
 struct TLcgIterator {
-    inline TLcgIterator(T seq) noexcept
+    constexpr TLcgIterator(T seq) noexcept
         : C((seq << 1u) | (T)1) // C must be odd
     {
     }
 
-    inline T Iterate(T x) const noexcept {
+    constexpr T Iterate(T x) const noexcept {
         return x * A + C;
     }
 
@@ -48,13 +48,13 @@ struct TLcgRngBase: public TIterator, public TMixer {
     using TResultType = decltype(std::declval<TMixer>().Mix(TStateType()));
 
     template <typename... Args>
-    inline TLcgRngBase(TStateType seed, Args&&... args)
+    constexpr TLcgRngBase(TStateType seed, Args&&... args)
         : TIterator(std::forward<Args>(args)...)
         , X(seed)
     {
     }
 
-    inline TResultType GenRand() noexcept {
+    constexpr TResultType GenRand() noexcept {
         return this->Mix(X = this->Iterate(X));
     }
 
diff --git a/util/string/cast.h b/util/string/cast.h
index 0c616738832..711c46a9e8b 100644
--- a/util/string/cast.h
+++ b/util/string/cast.h
@@ -2,6 +2,7 @@
 
 #include <util/system/defaults.h>
 #include <util/stream/str.h>
+#include <util/generic/constant_evaluation.h>
 #include <util/generic/maybe.h>
 #include <util/generic/string.h>
 #include <util/generic/strbuf.h>
@@ -416,15 +417,11 @@ class TIntStringBuf {
     template <std::enable_if_t<std::is_integral<T>::value, bool> = true>
     explicit constexpr TIntStringBuf(T t) {
         Size_ = Convert(t, Buf_, sizeof(Buf_));
-#if __cplusplus >= 202002L // is_constant_evaluated is not supported by CUDA yet
-        if (std::is_constant_evaluated()) {
-#endif
+        if (IsConstantEvaluated()) {
             // Init the rest of the array,
             // otherwise constexpr copy and move constructors don't work due to uninitialized data access
             std::fill(Buf_ + Size_, Buf_ + sizeof(Buf_), '\0');
-#if __cplusplus >= 202002L
         }
-#endif
     }
 
     constexpr operator TStringBuf() const noexcept {
diff --git a/util/system/compiler.h b/util/system/compiler.h
index e716438b97c..42f3f6e0d2b 100644
--- a/util/system/compiler.h
+++ b/util/system/compiler.h
@@ -694,6 +694,18 @@ Y_FORCE_INLINE void DoNotOptimizeAway(const T&) = delete;
 
 #endif
 
+/**
+ * @def Y_HAS_BUILTIN
+ *
+ * A wrapper around `__has_builtin` that evaluates to zero when the compiler
+ * does not provide the feature-checking macro.
+ */
+#ifdef __has_builtin
+    #define Y_HAS_BUILTIN(x) __has_builtin(x)
+#else
+    #define Y_HAS_BUILTIN(x) 0
+#endif
+
 /**
  * @def Y_HAS_ATTRIBUTE
  *