Backport of ydb-ent fixes on top of stable-24-4

ydb/core/blobstorage/vdisk/protos/events.proto

@@ -108,13 +108,15 @@ message LogoBlobIdRange {
 message GetLogoBlobRequest {
     LogoBlobIdRange range = 1;
     bool show_internals = 5;
+    bool need_data = 6;
 };
 
 message GetLogoBlobResponse {
     message LogoBlob {
         string status = 1;
         optional string ingress = 2;
         string id = 3;
+        optional string data_base64 = 4;
     };
 
     repeated LogoBlob logoblobs = 1;

ydb/core/blobstorage/vdisk/skeleton/blobstorage_monactors.cpp

@@ -355,7 +355,7 @@ namespace NKikimr {
             TSkeletonFrontMonLogoBlobsQueryParams params{
                 .DbName = "",
                 .Form = "",
-                .IndexOnly = true,
+                .IndexOnly = !record.need_data(),
                 .ShowInternals = record.show_internals(),
                 .SubmitButton = true,
                 .AllButton = false,
@@ -368,6 +368,9 @@ namespace NKikimr {
             auto &range = record.range();
             params.From = convert(range.from());
             params.To = convert(range.to());
+            if (params.To == params.From) {
+                params.To.Clear();
+            }
 
             return params;
         }
@@ -479,14 +482,18 @@ namespace NKikimr {
             }
         }
 
-        void OutputOneQueryResultToProto(NKikimrVDisk::GetLogoBlobResponse::LogoBlob *blob, const NKikimrBlobStorage::TQueryResult &q) {
+        void OutputOneQueryResultToProto(NKikimrVDisk::GetLogoBlobResponse::LogoBlob *blob, const NKikimrBlobStorage::TQueryResult &q,
+                TEvBlobStorage::TEvVGetResult *ev) {
             TLogoBlobID id = LogoBlobIDFromLogoBlobID(q.GetBlobID());
             blob->set_id(id.ToString());
             blob->set_status(NKikimrProto::EReplyStatus_Name(q.GetStatus()));
             if (ShowInternals) {
                 TIngress ingress(q.GetIngress());
                 blob->set_ingress(ingress.ToString(Top.get(), TVDiskIdShort(SelfVDiskId), id));
             }
+            if (TRope data = ev->GetBlobData(q)) {
+                blob->set_data_base64(Base64Encode(data.ConvertToString()));
+            }
         }
 
 
@@ -521,7 +528,7 @@ namespace NKikimr {
             } else {
                 auto res = std::make_unique<TEvGetLogoBlobResponse>();
                 for (ui32 i = 0; i < size; i++) {
-                    OutputOneQueryResultToProto(res->Record.add_logoblobs(), rec.GetResult(i));
+                    OutputOneQueryResultToProto(res->Record.add_logoblobs(), rec.GetResult(i), ev->Get());
                 }
                 Finish(ctx, res.release());
             }

ydb/core/driver_lib/run/kikimr_services_initializers.cpp

@@ -155,6 +155,7 @@
 #include <ydb/core/tx/long_tx_service/public/events.h>
 #include <ydb/core/tx/long_tx_service/long_tx_service.h>
 
+#include <ydb/core/util/aws.h>
 #include <ydb/core/util/failure_injection.h>
 #include <ydb/core/util/memory_tracker.h>
 #include <ydb/core/util/sig.h>
@@ -244,32 +245,17 @@
 
 #include <util/system/hostname.h>
 
-#ifndef KIKIMR_DISABLE_S3_OPS
-#include <aws/core/Aws.h>
-#endif
-
-namespace {
+namespace NKikimr::NKikimrServicesInitializers {
 
-#ifndef KIKIMR_DISABLE_S3_OPS
 struct TAwsApiGuard {
     TAwsApiGuard() {
-        Aws::InitAPI(Options);
+        InitAwsAPI();
     }
 
     ~TAwsApiGuard() {
-        Aws::ShutdownAPI(Options);
+        ShutdownAwsAPI();
     }
-
-private:
-    Aws::SDKOptions Options;
 };
-#endif
-
-}
-
-namespace NKikimr {
-
-namespace NKikimrServicesInitializers {
 
 ui32 TFederatedQueryInitializer::IcPort = 0;
 
@@ -2884,7 +2870,6 @@ void TGraphServiceInitializer::InitializeServices(NActors::TActorSystemSetup* se
         TActorSetupCmd(NGraph::CreateGraphService(appData->TenantName), TMailboxType::HTSwap, appData->UserPoolId));
 }
 
-#ifndef KIKIMR_DISABLE_S3_OPS
 TAwsApiInitializer::TAwsApiInitializer(IGlobalObjectStorage& globalObjects)
     : GlobalObjects(globalObjects)
 {
@@ -2895,7 +2880,5 @@ void TAwsApiInitializer::InitializeServices(NActors::TActorSystemSetup* setup, c
     Y_UNUSED(appData);
     GlobalObjects.AddGlobalObject(std::make_shared<TAwsApiGuard>());
 }
-#endif
 
-} // namespace NKikimrServicesInitializers
-} // namespace NKikimr
+} // namespace NKikimr::NKikimrServicesInitializers

ydb/core/driver_lib/run/kikimr_services_initializers.h

@@ -618,7 +618,6 @@ class TGraphServiceInitializer : public IKikimrServicesInitializer {
     void InitializeServices(NActors::TActorSystemSetup* setup, const NKikimr::TAppData* appData) override;
 };
 
-#ifndef KIKIMR_DISABLE_S3_OPS
 class TAwsApiInitializer : public IServiceInitializer {
     IGlobalObjectStorage& GlobalObjects;
 
@@ -627,7 +626,6 @@ class TAwsApiInitializer : public IServiceInitializer {
 
     void InitializeServices(NActors::TActorSystemSetup* setup, const NKikimr::TAppData* appData) override;
 };
-#endif
 
 } // namespace NKikimrServicesInitializers
 } // namespace NKikimr

ydb/core/driver_lib/run/run.cpp

@@ -1685,11 +1685,9 @@ TIntrusivePtr<TServiceInitializersList> TKikimrRunner::CreateServiceInitializers
         sil->AddServiceInitializer(new TGraphServiceInitializer(runConfig));
     }
 
-#ifndef KIKIMR_DISABLE_S3_OPS
     if (serviceMask.EnableAwsService) {
         sil->AddServiceInitializer(new TAwsApiInitializer(*this));
     }
-#endif
 
     return sil;
 }

ydb/core/driver_lib/run/ya.make

@@ -1,15 +1,5 @@
 LIBRARY(run)
 
-IF (OS_WINDOWS)
-    CFLAGS(
-        -DKIKIMR_DISABLE_S3_OPS
-    )
-ELSE()
-    PEERDIR(
-        contrib/libs/aws-sdk-cpp/aws-cpp-sdk-core
-    )
-ENDIF()
-
 SRCS(
     auto_config_initializer.cpp
     config.cpp

ydb/core/engine/mkql_keys.cpp

@@ -251,8 +251,10 @@ THolder<TKeyDesc> ExtractEraseRow(TCallable& callable, const TTypeEnvironment& e
 #define MAKE_PRIMITIVE_TYPE_CELL(type, layout) \
     case NUdf::TDataType<type>::Id: return MakeCell<layout>(value);
 
-TCell MakeCell(NScheme::TTypeInfo type, const NUdf::TUnboxedValuePod& value,
-    const TTypeEnvironment& env, bool copy,
+
+template<typename TStringBackend>
+TCell MakeCellImpl(NScheme::TTypeInfo type, const NUdf::TUnboxedValuePod& value,
+    const TStringBackend& env, bool copy,
     i32 typmod, TMaybe<TString>* error)
 {
     if (!value)
@@ -300,6 +302,21 @@ TCell MakeCell(NScheme::TTypeInfo type, const NUdf::TUnboxedValuePod& value,
     return TCell(val.Data(), val.Size());
 }
 
+TCell MakeCell(NScheme::TTypeInfo type, const NUdf::TUnboxedValuePod& value,
+    const TTypeEnvironment& env, bool copy,
+    i32 typmod, TMaybe<TString>* error)
+{
+    return MakeCellImpl(type, value, env, copy, typmod, error);
+}
+
+TCell MakeCell(NScheme::TTypeInfo type, const NUdf::TUnboxedValuePod& value,
+    const TStringProviderBackend& env, bool copy,
+    i32 typmod, TMaybe<TString>* error)
+{
+    return MakeCellImpl(type, value, env, copy, typmod, error);
+}
+
+
 #undef MAKE_PRIMITIVE_TYPE_CELL
 
 TReadTarget ExtractFlatReadTarget(TRuntimeNode modeInput) {

ydb/core/engine/mkql_keys.h

@@ -52,10 +52,48 @@ TCell MakeCell(const NUdf::TUnboxedValuePod& value) {
     return TCell(reinterpret_cast<const char*>(&v), sizeof(v));
 }
 
+struct TStringProviderBackend {
+    mutable TMemoryPool MemoryPool;
+
+    TStringProviderBackend()
+        : MemoryPool(256)
+    {}
+
+    class TMutableStringData {
+        friend struct TStringProviderBackend;
+
+    private:
+        char* Data_;
+        size_t Size_ = 0;
+
+        explicit TMutableStringData(char* data, size_t size)
+            : Data_(data)
+            , Size_(size)
+        {}
+
+    public:
+        char *Data() const noexcept {
+            return Data_;
+        }
+
+        size_t Size() const noexcept {
+            return Size_;
+        }
+    };
+
+    TMutableStringData NewString(ui32 size) const {
+        return TMutableStringData(reinterpret_cast<char*>(MemoryPool.Allocate(size)), size);
+    }
+};
+
 TCell MakeCell(NScheme::TTypeInfo type, const NUdf::TUnboxedValuePod& value,
     const TTypeEnvironment& env, bool copy = true,
     i32 typmod = -1, TMaybe<TString>* error = {});
 
+TCell MakeCell(NScheme::TTypeInfo type, const NUdf::TUnboxedValuePod& value,
+    const TStringProviderBackend& env, bool copy = true,
+    i32 typmod = -1, TMaybe<TString>* error = {});
+
 void FillKeyTupleValue(const NUdf::TUnboxedValue& row, const TVector<ui32>& rowIndices,
     const TVector<NScheme::TTypeInfo>& rowTypes, TVector<TCell>& cells, const TTypeEnvironment& env);
 

ydb/core/grpc_services/query/rpc_execute_query.cpp

@@ -320,13 +320,13 @@ class TExecuteQueryRPC : public TActorBootstrapped<TExecuteQueryRPC> {
     }
 
     void Handle(NKqp::TEvKqpExecuter::TEvStreamData::TPtr& ev, const TActorContext& ctx) {
-        Ydb::Query::ExecuteQueryResponsePart response;
-        response.set_status(Ydb::StatusIds::SUCCESS);
-        response.set_result_set_index(ev->Get()->Record.GetQueryResultIndex());
-        response.mutable_result_set()->Swap(ev->Get()->Record.MutableResultSet());
+        Ydb::Query::ExecuteQueryResponsePart *response = ev->Get()->Arena->Allocate<Ydb::Query::ExecuteQueryResponsePart>();
+        response->set_status(Ydb::StatusIds::SUCCESS);
+        response->set_result_set_index(ev->Get()->Record.GetQueryResultIndex());
+        response->mutable_result_set()->Swap(ev->Get()->Record.MutableResultSet());
 
         TString out;
-        Y_PROTOBUF_SUPPRESS_NODISCARD response.SerializeToString(&out);
+        Y_PROTOBUF_SUPPRESS_NODISCARD response->SerializeToString(&out);
 
         FlowControl_.PushResponse(out.size());
         const i64 freeSpaceBytes = FlowControl_.FreeSpaceBytes();

ydb/core/kqp/compute_actor/kqp_scan_compute_actor.h

@@ -26,7 +26,7 @@ class TKqpScanComputeActor: public TSchedulableComputeActorBase<TKqpScanComputeA
     const ui32 LockNodeId;
 
     struct TLockHash {
-        bool operator()(const NKikimrDataEvents::TLock& lock) {
+        size_t operator()(const NKikimrDataEvents::TLock& lock) {
             return MultiHash(
                 lock.GetLockId(),
                 lock.GetDataShard(),

ydb/core/kqp/executer_actor/kqp_data_executer.cpp

@@ -2676,8 +2676,9 @@ class TKqpDataExecuter: public TKqpExecuterBase<TKqpDataExecuter, EExecType::Dat
             hFunc(TEvDqCompute::TEvState, HandleShutdown);
             hFunc(TEvInterconnect::TEvNodeDisconnected, HandleShutdown);
             hFunc(TEvents::TEvPoison, HandleShutdown);
+            hFunc(TEvDq::TEvAbortExecution, HandleShutdown);
             default:
-                LOG_E("Unexpected event: " << ev->GetTypeName()); // ignore all other events
+                LOG_E("Unexpected event while waiting for shutdown: " << ev->GetTypeName()); // ignore all other events
         }
     }
 
@@ -2719,6 +2720,16 @@ class TKqpDataExecuter: public TKqpExecuterBase<TKqpDataExecuter, EExecType::Dat
         }
     }
 
+    void HandleShutdown(TEvDq::TEvAbortExecution::TPtr& ev) {
+        auto statusCode = NYql::NDq::DqStatusToYdbStatus(ev->Get()->Record.GetStatusCode());
+
+        // In case of external timeout the response is already sent to the client - no need to wait for stats.
+        if (statusCode == Ydb::StatusIds::TIMEOUT) {
+            LOG_I("External timeout while waiting for Compute Actors to finish - forcing shutdown. Sender: " << ev->Sender);
+            PassAway();
+        }
+    }
+
 private:
     void ReplyTxStateUnknown(ui64 shardId) {
         auto message = TStringBuilder() << "Tx state unknown for shard " << shardId << ", txid " << TxId;

ydb/core/kqp/executer_actor/kqp_executer.h

@@ -66,8 +66,15 @@ struct TEvKqpExecuter {
         }
     };
 
-    struct TEvStreamData : public TEventPB<TEvStreamData, NKikimrKqp::TEvExecuterStreamData,
-        TKqpExecuterEvents::EvStreamData> {};
+    struct TEvStreamData : public TEventPBWithArena<TEvStreamData, NKikimrKqp::TEvExecuterStreamData, TKqpExecuterEvents::EvStreamData> {
+        using TBaseEv = TEventPBWithArena<TEvStreamData, NKikimrKqp::TEvExecuterStreamData, TKqpExecuterEvents::EvStreamData>;
+        using TBaseEv::TEventPBBase;
+
+        TEvStreamData() = default;
+        explicit TEvStreamData(TIntrusivePtr<NActors::TProtoArenaHolder> arena)
+            : TEventPBBase(std::move(arena))
+        {}
+    };
 
     struct TEvStreamDataAck : public TEventPB<TEvStreamDataAck, NKikimrKqp::TEvExecuterStreamDataAck,
         TKqpExecuterEvents::EvStreamDataAck>

ydb/core/kqp/executer_actor/kqp_executer_impl.h

@@ -294,14 +294,15 @@ class TKqpExecuterBase : public TActorBootstrapped<TDerived> {
             batch.Payload = std::move(computeData.Payload);
 
             TKqpProtoBuilder protoBuilder{*AppData()->FunctionRegistry};
-            auto resultSet = protoBuilder.BuildYdbResultSet(std::move(batches), txResult.MkqlItemType, txResult.ColumnOrder);
 
             if (!trailingResults) {
                 auto streamEv = MakeHolder<TEvKqpExecuter::TEvStreamData>();
                 streamEv->Record.SetSeqNo(computeData.Proto.GetSeqNo());
                 streamEv->Record.SetQueryResultIndex(*txResult.QueryResultIndex + StatementResultIndex);
                 streamEv->Record.SetChannelId(channel.Id);
-                streamEv->Record.MutableResultSet()->Swap(&resultSet);
+
+                protoBuilder.BuildYdbResultSet(*streamEv->Record.MutableResultSet(), std::move(batches),
+                    txResult.MkqlItemType, txResult.ColumnOrder);
 
                 LOG_D("Send TEvStreamData to " << Target << ", seqNo: " << streamEv->Record.GetSeqNo()
                     << ", nRows: " << streamEv->Record.GetResultSet().rows().size());
@@ -314,8 +315,11 @@ class TKqpExecuterBase : public TActorBootstrapped<TDerived> {
                 ackEv->Record.SetChannelId(channel.Id);
                 ackEv->Record.SetFreeSpace(50_MB);
                 this->Send(channelComputeActorId, ackEv.Release(), /* TODO: undelivery */ 0, /* cookie */ channel.Id);
-                txResult.TrailingResult.Swap(&resultSet);
+
+                protoBuilder.BuildYdbResultSet(txResult.TrailingResult, std::move(batches),
+                    txResult.MkqlItemType, txResult.ColumnOrder);
                 txResult.HasTrailingResult = true;
+
                 LOG_D("staging TEvStreamData to " << Target << ", seqNo: " << computeData.Proto.GetSeqNo()
                     << ", nRows: " << txResult.TrailingResult.rows().size());
             }
@@ -1629,9 +1633,13 @@ class TKqpExecuterBase : public TActorBootstrapped<TDerived> {
 
 protected:
     void UnexpectedEvent(const TString& state, ui32 eventType) {
-        LOG_C("TKqpExecuter, unexpected event: " << eventType << ", at state:" << state << ", selfID: " << this->SelfId());
-        InternalError(TStringBuilder() << "Unexpected event at TKqpExecuter, state: " << state
-            << ", event: " << eventType);
+        if (eventType == TEvents::TEvPoison::EventType) {
+            LOG_D("TKqpExecuter, TEvPoison event at state:" << state << ", selfID: " << this->SelfId());
+            InternalError(TStringBuilder() << "TKqpExecuter got poisoned, state: " << state);
+        } else {
+            LOG_E("TKqpExecuter, unexpected event: " << eventType << ", at state:" << state << ", selfID: " << this->SelfId());
+            InternalError(TStringBuilder() << "Unexpected event at TKqpExecuter, state: " << state << ", event: " << eventType);
+        }
     }
 
     void InternalError(const NYql::TIssues& issues) {

ydb/core/kqp/executer_actor/kqp_result_channel.cpp

@@ -156,13 +156,12 @@ class TResultStreamChannelProxy : public TResultCommonChannelProxy {
         batch.Proto = std::move(*computeData.Proto.MutableChannelData()->MutableData());
         batch.Payload = std::move(computeData.Payload);
 
-        TKqpProtoBuilder protoBuilder{*AppData()->FunctionRegistry};
-        auto resultSet = protoBuilder.BuildYdbResultSet(std::move(batches), ItemType, ColumnOrder);
-
         auto streamEv = MakeHolder<TEvKqpExecuter::TEvStreamData>();
         streamEv->Record.SetSeqNo(computeData.Proto.GetSeqNo());
         streamEv->Record.SetQueryResultIndex(QueryResultIndex + StatementResultIndex);
-        streamEv->Record.MutableResultSet()->Swap(&resultSet);
+
+        TKqpProtoBuilder protoBuilder{*AppData()->FunctionRegistry};
+        protoBuilder.BuildYdbResultSet(*streamEv->Record.MutableResultSet(), std::move(batches), ItemType, ColumnOrder);
 
         LOG_DEBUG_S(*NActors::TlsActivationContext, NKikimrServices::KQP_EXECUTER,
             "Send TEvStreamData to " << Target << ", seqNo: " << streamEv->Record.GetSeqNo()