Merge branch 'stable-24-3' into fix-11186

Author: Vladislav Gogov

.github/config/muted_ya.txt

@@ -18,10 +18,7 @@ ydb/core/kqp/ut/tx KqpSnapshotRead.ReadOnlyTxWithIndexCommitsOnConcurrentWrite+w
 ydb/core/kqp/ut/tx KqpSinkTx.InvalidateOnError
 ydb/core/kqp/ut/query KqpLimits.QueryReplySize
 ydb/core/kqp/ut/query KqpQuery.QueryTimeout
-ydb/core/kqp/ut/service KqpQueryService.TableSink_OlapRWQueries
 ydb/core/kqp/ut/service KqpQueryService.TableSink_OltpReplace+HasSecondaryIndex
-ydb/core/kqp/ut/query KqpQuery.OlapCreateAsSelect_Complex
-ydb/core/kqp/ut/query KqpQuery.OlapCreateAsSelect_Simple
 ydb/core/kqp/ut/scan KqpRequestContext.TraceIdInErrorMessage
 ydb/core/kqp/ut/scheme KqpOlapScheme.TenThousandColumns
 ydb/core/kqp/ut/scheme KqpScheme.AlterAsyncReplication
@@ -32,10 +29,8 @@ ydb/core/kqp/ut/scheme [44/50]*
 ydb/core/kqp/ut/service KqpQueryService.ExecuteQueryPgTableSelect
 ydb/core/kqp/ut/service KqpQueryService.QueryOnClosedSession
 ydb/core/kqp/ut/service KqpQueryService.TableSink_OltpUpdate
-ydb/core/kqp/ut/service KqpQueryService.TableSink_Htap*
 ydb/core/kqp/ut/service KqpService.CloseSessionsWithLoad
 ydb/core/kqp/ut/service [38/50]*
-ydb/core/kqp/ut/service KqpQueryService.TableSink_OltpUpdate
 ydb/core/kqp/ut/service KqpQueryService.TableSink_OltpReplace+HasSecondaryIndex
 ydb/core/persqueue/ut [37/40] chunk chunk
 ydb/core/persqueue/ut [38/40] chunk chunk

ydb/core/base/ticket_parser.h

@@ -154,14 +154,24 @@ namespace NKikimr {
 
         struct TError {
             TString Message;
+            TString LogMessage;
             bool Retryable = true;
 
             bool empty() const {
-                return Message.empty();
+                return Message.empty() && LogMessage.empty();
+            }
+
+            bool HasMessage() const {
+                return !Message.empty();
+            }
+
+            bool HasLogMessage() const {
+                return !LogMessage.empty();
             }
 
             void clear() {
                 Message.clear();
+                LogMessage.clear();
                 Retryable = true;
             }
 

ydb/core/cms/console/configs_dispatcher.cpp

@@ -300,7 +300,13 @@ void TConfigsDispatcher::Bootstrap()
         CurrentConfig,
         0,
         true,
-        1);
+        1,
+        {},
+        {},
+        TNodeInfo{
+            .Tenant = Labels.contains("tenant") ? Labels.at("tenant") : TString(""),
+            .NodeType = Labels.contains("node_type") ? Labels.at("node_type") : TString(""),
+        });
     CommonSubscriptionClient = RegisterWithSameMailbox(commonClient);
 
     Become(&TThis::StateInit);

ydb/core/cms/console/console_configs_subscriber.cpp

@@ -48,7 +48,8 @@ class TConfigsSubscriber : public TActorBootstrapped<TConfigsSubscriber> {
             bool processYaml,
             ui64 version,
             const TString &yamlConfig,
-            const TMap<ui64, TString> &volatileYamlConfigs)
+            const TMap<ui64, TString> &volatileYamlConfigs,
+            const std::optional<TNodeInfo> explicitNodeInfo)
         : OwnerId(ownerId)
         , Cookie(cookie)
         , Kinds(kinds)
@@ -67,6 +68,15 @@ class TConfigsSubscriber : public TActorBootstrapped<TConfigsSubscriber> {
                 VolatileYamlConfigHashes[id] = THash<TString>()(config);
             }
         }
+
+        if (explicitNodeInfo) {
+            if (explicitNodeInfo->Tenant) {
+                Tenant = explicitNodeInfo->Tenant;
+            } else {
+                Tenant = "<none>";
+            }
+            NodeType = explicitNodeInfo->NodeType;
+        }
     }
 
     static constexpr NKikimrServices::TActivity::EType ActorActivityType() {
@@ -82,7 +92,11 @@ class TConfigsSubscriber : public TActorBootstrapped<TConfigsSubscriber> {
             return;
         }
 
-        SendPoolStatusRequest(ctx);
+        if (!Tenant) {
+            SendPoolStatusRequest(ctx);
+        } else {
+            Subscribe(ctx);
+        }
         Become(&TThis::StateWork);
     }
 
@@ -417,9 +431,19 @@ IActor *CreateConfigsSubscriber(
     bool processYaml,
     ui64 version,
     const TString &yamlConfig,
-    const TMap<ui64, TString> &volatileYamlConfigs)
+    const TMap<ui64, TString> &volatileYamlConfigs,
+    const std::optional<TNodeInfo> explicitNodeInfo)
 {
-    return new TConfigsSubscriber(ownerId, cookie, kinds, currentConfig, processYaml, version, yamlConfig, volatileYamlConfigs);
+    return new TConfigsSubscriber(
+        ownerId,
+        cookie,
+        kinds,
+        currentConfig,
+        processYaml,
+        version,
+        yamlConfig,
+        volatileYamlConfigs,
+        explicitNodeInfo);
 }
 
 } // namespace NKikimr::NConsole

ydb/core/cms/console/console_configs_subscriber.h

@@ -6,8 +6,15 @@
 
 #include <ydb/library/actors/core/actor.h>
 
+#include <optional>
+
 namespace NKikimr::NConsole {
 
+struct TNodeInfo {
+    TString Tenant;
+    TString NodeType;
+};
+
 IActor *CreateConfigsSubscriber(
     const TActorId &ownerId,
     const TVector<ui32> &kinds,
@@ -16,6 +23,7 @@ IActor *CreateConfigsSubscriber(
     bool processYaml = false,
     ui64 version = 0,
     const TString &yamlConfig = {},
-    const TMap<ui64, TString> &volatileYamlConfigs = {});
+    const TMap<ui64, TString> &volatileYamlConfigs = {},
+    const std::optional<TNodeInfo> explicitNodeInfo = std::nullopt);
 
 } // namespace NKikimr::NConsole

ydb/core/cms/console/log_settings_configurator_ut.cpp

@@ -263,7 +263,11 @@ Y_UNIT_TEST_SUITE(TLogSettingsConfiguratorTests)
 
     Y_UNIT_TEST(TestRemoveComponentEntries)
     {
-        TTenantTestRuntime runtime(DefaultConsoleTestConfig());
+        NKikimrConfig::TAppConfig ext;
+        auto& label = *ext.AddLabels();
+        label.SetName("tenant");
+        label.SetValue(TENANT1_1_NAME);
+        TTenantTestRuntime runtime(DefaultConsoleTestConfig(), ext);
         auto settings = InitLogSettingsConfigurator(runtime);
         WaitForUpdate(runtime); // initial update
 
@@ -292,7 +296,11 @@ Y_UNIT_TEST_SUITE(TLogSettingsConfiguratorTests)
 
     Y_UNIT_TEST(TestChangeDefaults)
     {
-        TTenantTestRuntime runtime(DefaultConsoleTestConfig());
+        NKikimrConfig::TAppConfig ext;
+        auto& label = *ext.AddLabels();
+        label.SetName("tenant");
+        label.SetValue(TENANT1_1_NAME);
+        TTenantTestRuntime runtime(DefaultConsoleTestConfig(), ext);
         auto settings = InitLogSettingsConfigurator(runtime);
         WaitForUpdate(runtime); // initial update
 

ydb/core/formats/arrow/arrow_filter.cpp

@@ -575,8 +575,10 @@ TColumnFilter TColumnFilter::CombineSequentialAnd(const TColumnFilter& extFilter
 }
 
 TColumnFilter::TIterator TColumnFilter::GetIterator(const bool reverse, const ui32 expectedSize) const {
-    if ((IsTotalAllowFilter() || IsTotalDenyFilter()) && !Filter.size()) {
-        return TIterator(reverse, expectedSize, LastValue);
+    if (IsTotalAllowFilter()) {
+        return TIterator(reverse, expectedSize, true);
+    } else if (IsTotalDenyFilter()) {
+        return TIterator(reverse, expectedSize, false);
     } else {
         AFL_VERIFY(expectedSize == Size())("expected", expectedSize)("size", Size())("reverse", reverse);
         return TIterator(reverse, Filter, GetStartValue(reverse));

ydb/core/formats/arrow/arrow_helpers.cpp

@@ -196,8 +196,18 @@ bool IsSortedAndUnique(const std::shared_ptr<arrow::RecordBatch>& batch,
     }
 }
 
-std::shared_ptr<arrow::RecordBatch> SortBatch(const std::shared_ptr<arrow::RecordBatch>& batch,
-                                              const std::shared_ptr<arrow::Schema>& sortingKey, const bool andUnique) {
+std::shared_ptr<arrow::RecordBatch> SortBatch(
+    const std::shared_ptr<arrow::RecordBatch>& batch, const std::vector<std::shared_ptr<arrow::Array>>& sortingKey, const bool andUnique) {
+    auto sortPermutation = MakeSortPermutation(sortingKey, andUnique);
+    if (sortPermutation) {
+        return Reorder(batch, sortPermutation, andUnique);
+    } else {
+        return batch;
+    }
+}
+
+std::shared_ptr<arrow::RecordBatch> SortBatch(const std::shared_ptr<arrow::RecordBatch>& batch, const std::shared_ptr<arrow::Schema>& sortingKey,
+    const bool andUnique) {
     auto sortPermutation = MakeSortPermutation(batch, sortingKey, andUnique);
     if (sortPermutation) {
         return Reorder(batch, sortPermutation, andUnique);

ydb/core/formats/arrow/arrow_helpers.h

@@ -26,8 +26,10 @@ TString SerializeBatchNoCompression(const std::shared_ptr<arrow::RecordBatch>& b
 std::shared_ptr<arrow::RecordBatch> DeserializeBatch(const TString& blob,
                                                      const std::shared_ptr<arrow::Schema>& schema);
 
-std::shared_ptr<arrow::RecordBatch> SortBatch(const std::shared_ptr<arrow::RecordBatch>& batch,
-                                              const std::shared_ptr<arrow::Schema>& sortingKey, const bool andUnique);
+std::shared_ptr<arrow::RecordBatch> SortBatch(
+    const std::shared_ptr<arrow::RecordBatch>& batch, const std::shared_ptr<arrow::Schema>& sortingKey, const bool andUnique);
+std::shared_ptr<arrow::RecordBatch> SortBatch(
+    const std::shared_ptr<arrow::RecordBatch>& batch, const std::vector<std::shared_ptr<arrow::Array>>& sortingKey, const bool andUnique);
 bool IsSorted(const std::shared_ptr<arrow::RecordBatch>& batch,
     const std::shared_ptr<arrow::Schema>& sortingKey,
     bool desc = false);

ydb/core/formats/arrow/permutations.cpp

@@ -15,18 +15,25 @@
 
 namespace NKikimr::NArrow {
 
-std::shared_ptr<arrow::UInt64Array> MakeSortPermutation(const std::shared_ptr<arrow::RecordBatch>& batch, const std::shared_ptr<arrow::Schema>& sortingKey, const bool andUnique) {
-    auto keyBatch = TColumnOperator().VerifyIfAbsent().Adapt(batch, sortingKey).DetachResult();
-    auto keyColumns = std::make_shared<TArrayVec>(keyBatch->columns());
+std::shared_ptr<arrow::UInt64Array> MakeSortPermutation(const std::vector<std::shared_ptr<arrow::Array>>& keyColumns, const bool andUnique) {
+    std::optional<i64> count;
+    for (auto&& i : keyColumns) {
+        AFL_VERIFY(i);
+        if (!count) {
+            count = i->length();
+        } else {
+            AFL_VERIFY(*count == i->length());
+        }
+    }
+    AFL_VERIFY(count);
     std::vector<TRawReplaceKey> points;
-    points.reserve(keyBatch->num_rows());
-
-    for (int i = 0; i < keyBatch->num_rows(); ++i) {
-        points.push_back(TRawReplaceKey(keyColumns.get(), i));
+    points.reserve(*count);
+    for (int i = 0; i < *count; ++i) {
+        points.push_back(TRawReplaceKey(&keyColumns, i));
     }
 
     bool haveNulls = false;
-    for (auto& column : *keyColumns) {
+    for (auto& column : keyColumns) {
         if (HasNulls(column)) {
             haveNulls = true;
             break;
@@ -36,11 +43,9 @@ std::shared_ptr<arrow::UInt64Array> MakeSortPermutation(const std::shared_ptr<ar
     if (haveNulls) {
         std::sort(points.begin(), points.end());
     } else {
-        std::sort(points.begin(), points.end(),
-            [](const TRawReplaceKey& a, const TRawReplaceKey& b) {
-                return a.CompareNotNull(b) == std::partial_ordering::less;
-            }
-        );
+        std::sort(points.begin(), points.end(), [](const TRawReplaceKey& a, const TRawReplaceKey& b) {
+            return a.CompareNotNull(b) == std::partial_ordering::less;
+        });
     }
 
     arrow::UInt64Builder builder;
@@ -78,6 +83,12 @@ std::shared_ptr<arrow::UInt64Array> MakeSortPermutation(const std::shared_ptr<ar
     return out;
 }
 
+std::shared_ptr<arrow::UInt64Array> MakeSortPermutation(const std::shared_ptr<arrow::RecordBatch>& batch,
+    const std::shared_ptr<arrow::Schema>& sortingKey, const bool andUnique) {
+    auto keyBatch = TColumnOperator().VerifyIfAbsent().Adapt(batch, sortingKey).DetachResult();
+    return MakeSortPermutation(keyBatch->columns(), andUnique);
+}
+
 namespace {
 
 template <class TDataContainer>