Merge pull request #10871 from uzhastik/24_3_merge_9

24 3 merge 9

Author: maximyurchuk

ydb/core/base/counters.cpp

@@ -40,6 +40,7 @@ static const THashSet<TString> DATABASE_SERVICES
          TString("pqproxy|readSession"),
          TString("pqproxy|schemecache"),
          TString("pqproxy|mirrorWriteTimeLag"),
+         TString("pqproxy|userAgents"),
          TString("datastreams"),
     }};
 

ydb/core/blobstorage/vdisk/repl/query_donor.h

@@ -68,10 +68,9 @@ namespace NKikimr {
             }
 
             if (action) {
-                const TActorId temp(actorId);
                 LOG_DEBUG_S(*TlsActivationContext, NKikimrServices::BS_VDISK_GET, SelfId() << " sending " << query->ToString()
-                    << " to " << temp);
-                Send(actorId, query.release());
+                    << " to " << actorId);
+                Send(actorId, query.release(), IEventHandle::FlagTrackDelivery);
             } else {
                 PassAway();
             }
@@ -116,6 +115,7 @@ namespace NKikimr {
 
         STRICT_STFUNC(StateFunc,
             hFunc(TEvBlobStorage::TEvVGetResult, Handle);
+            cFunc(TEvents::TSystem::Undelivered, Step);
             cFunc(TEvents::TSystem::Poison, PassAway);
         )
     };

ydb/core/driver_lib/run/run.cpp

@@ -122,6 +122,7 @@
 #include <ydb/services/ydb/ydb_scripting.h>
 #include <ydb/services/ydb/ydb_table.h>
 #include <ydb/services/ydb/ydb_object_storage.h>
+#include <ydb/services/view/grpc_service.h>
 
 #include <ydb/core/fq/libs/init/init.h>
 
@@ -600,6 +601,8 @@ void TKikimrRunner::InitializeGRpc(const TKikimrRunConfig& runConfig) {
         names["keyvalue"] = &hasKeyValue;
         TServiceCfg hasReplication = services.empty();
         names["replication"] = &hasReplication;
+        TServiceCfg hasView = services.empty();
+        names["view"] = &hasView;
 
         std::unordered_set<TString> enabled;
         for (const auto& name : services) {
@@ -875,6 +878,11 @@ void TKikimrRunner::InitializeGRpc(const TKikimrRunConfig& runConfig) {
                 grpcRequestProxies[0], hasReplication.IsRlAllowed()));
         }
 
+        if (hasView) {
+            server.AddService(new NGRpcService::TGRpcViewService(ActorSystem.Get(), Counters,
+                grpcRequestProxies[0], hasView.IsRlAllowed()));
+        }
+
         if (ModuleFactories) {
             for (const auto& service : ModuleFactories->GrpcServiceFactory.Create(enabled, disabled, ActorSystem.Get(), Counters, grpcRequestProxies[0])) {
                 server.AddService(service);
@@ -1190,6 +1198,8 @@ void TKikimrRunner::InitializeLogSettings(const TKikimrRunConfig& runConfig)
     );
 
     LogSettings->ClusterName = logConfig.HasClusterName() ? logConfig.GetClusterName() : "";
+    LogSettings->TenantName = runConfig.TenantName;
+    LogSettings->NodeId = runConfig.NodeId;
 
     if (logConfig.GetFormat() == "full") {
         LogSettings->Format = NLog::TSettings::PLAIN_FULL_FORMAT;

ydb/core/driver_lib/run/ya.make

@@ -172,6 +172,7 @@ PEERDIR(
     ydb/services/persqueue_v1
     ydb/services/rate_limiter
     ydb/services/replication
+    ydb/services/view
     ydb/services/ydb
 )
 

ydb/core/engine/mkql_proto.cpp

@@ -223,6 +223,19 @@ bool CellsFromTuple(const NKikimrMiniKQL::TType* tupleType,
             }
             break;
         }
+        case NScheme::NTypeIds::Decimal:
+        {
+            if (v.HasLow128() && v.HasHi128()) {
+                NYql::NDecimal::TInt128 int128 = NYql::NDecimal::FromProto(v);
+                auto &data = memoryOwner.emplace_back();
+                data.resize(sizeof(NYql::NDecimal::TInt128));
+                std::memcpy(data.Detach(), &int128, sizeof(NYql::NDecimal::TInt128));
+                c = TCell(data);                
+            } else {
+                CHECK_OR_RETURN_ERROR(false, Sprintf("Cannot parse value of type Decimal in tuple at position %" PRIu32, i));
+            }
+            break;
+        }        
         default:
             CHECK_OR_RETURN_ERROR(false, Sprintf("Unsupported typeId %" PRIu16 " at index %" PRIu32, typeId, i));
             break;
@@ -328,6 +341,13 @@ bool CellToValue(NScheme::TTypeInfo type, const TCell& c, NKikimrMiniKQL::TValue
         val.MutableOptional()->SetText(c.Data(), c.Size());
         break;
 
+    case NScheme::NTypeIds::Decimal: {
+        const auto loHi = c.AsValue<std::pair<ui64, i64>>();
+        val.MutableOptional()->SetLow128(loHi.first);
+        val.MutableOptional()->SetHi128(loHi.second);
+        break;
+    }
+
     case NScheme::NTypeIds::Pg: {
         auto convert = NPg::PgNativeTextFromNativeBinary(c.AsBuf(), type.GetTypeDesc());
         if (convert.Error) {

ydb/core/grpc_services/rpc_object_storage.cpp

@@ -144,6 +144,19 @@ bool CellFromTuple(NScheme::TTypeInfo type,
         }
         break;
     }
+    case NScheme::NTypeIds::Decimal:
+    {
+        if (tupleValue.Haslow_128()) {
+            NYql::NDecimal::TInt128 int128 = NYql::NDecimal::FromHalfs(tupleValue.Getlow_128(), tupleValue.Gethigh_128());
+            auto &data = memoryOwner.emplace_back();
+            data.resize(sizeof(NYql::NDecimal::TInt128));
+            std::memcpy(data.Detach(), &int128, sizeof(NYql::NDecimal::TInt128));
+            c = TCell(data);                
+        } else {
+            CHECK_OR_RETURN_ERROR(false, Sprintf("Cannot parse value of type Decimal in tuple at position %" PRIu32, position));
+        }
+        break;
+    }    
     default:
         CHECK_OR_RETURN_ERROR(false, Sprintf("Unsupported typeId %" PRIu16 " at index %" PRIu32, typeId, position));
         break;

ydb/core/grpc_services/rpc_view.cpp

@@ -0,0 +1,94 @@
+#include "rpc_scheme_base.h"
+#include "service_view.h"
+
+#include <ydb/core/grpc_services/base/base.h>
+#include <ydb/core/tx/schemeshard/schemeshard.h>
+#include <ydb/core/ydb_convert/ydb_convert.h>
+#include <ydb/library/actors/core/actor.h>
+#include <ydb/library/actors/core/hfunc.h>
+#include <ydb/public/api/protos/draft/ydb_view.pb.h>
+
+namespace NKikimr::NGRpcService {
+
+using namespace Ydb;
+
+using TEvDescribeView = TGrpcRequestOperationCall<View::DescribeViewRequest, View::DescribeViewResponse>;
+
+class TDescribeViewRPC : public TRpcSchemeRequestActor<TDescribeViewRPC, TEvDescribeView> {
+    using TBase = TRpcSchemeRequestActor<TDescribeViewRPC, TEvDescribeView>;
+
+public:
+    using TBase::TBase;
+
+    void Bootstrap() {
+        DescribeScheme();
+    }
+
+    void PassAway() override {
+        TBase::PassAway();
+    }
+
+private:
+    void DescribeScheme() {
+        auto ev = std::make_unique<TEvTxUserProxy::TEvNavigate>();
+        SetAuthToken(ev, *Request_);
+        SetDatabase(ev.get(), *Request_);
+        ev->Record.MutableDescribePath()->SetPath(GetProtoRequest()->path());
+
+        Send(MakeTxProxyID(), ev.release());
+        Become(&TDescribeViewRPC::StateDescribeScheme);
+    }
+
+    STATEFN(StateDescribeScheme) {
+        switch (ev->GetTypeRewrite()) {
+            HFunc(NSchemeShard::TEvSchemeShard::TEvDescribeSchemeResult, Handle);
+        default:
+            return TBase::StateWork(ev);
+        }
+    }
+
+    void Handle(NSchemeShard::TEvSchemeShard::TEvDescribeSchemeResult::TPtr& ev, const TActorContext& ctx) {
+        const auto& record = ev->Get()->GetRecord();
+        const auto& desc = record.GetPathDescription();
+
+        if (record.HasReason()) {
+            Request_->RaiseIssue(NYql::TIssue(record.GetReason()));
+        }
+
+        switch (record.GetStatus()) {
+            case NKikimrScheme::StatusSuccess:
+                if (desc.GetSelf().GetPathType() != NKikimrSchemeOp::EPathTypeView) {
+                    auto message = TStringBuilder() << "Expected a view, but got: " << desc.GetSelf().GetPathType();
+                    Request_->RaiseIssue(NYql::TIssue(message));
+                    return Reply(StatusIds::SCHEME_ERROR, ctx);
+                }
+
+                ConvertDirectoryEntry(desc.GetSelf(), Result_.mutable_self(), true);
+                Result_.set_query_text(desc.GetViewDescription().GetQueryText());
+
+                return ReplyWithResult(StatusIds::SUCCESS, Result_, ctx);
+
+            case NKikimrScheme::StatusPathDoesNotExist:
+            case NKikimrScheme::StatusSchemeError:
+                return Reply(StatusIds::SCHEME_ERROR, ctx);
+
+            case NKikimrScheme::StatusAccessDenied:
+                return Reply(StatusIds::UNAUTHORIZED, ctx);
+
+            case NKikimrScheme::StatusNotAvailable:
+                return Reply(StatusIds::UNAVAILABLE, ctx);
+
+            default:
+                return Reply(StatusIds::GENERIC_ERROR, ctx);
+        }
+    }
+
+private:
+    View::DescribeViewResult Result_;
+};
+
+void DoDescribeView(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f) {
+    f.RegisterActor(new TDescribeViewRPC(p.release()));
+}
+
+}

ydb/core/grpc_services/service_view.h

@@ -0,0 +1,12 @@
+#pragma once
+
+#include <memory>
+
+namespace NKikimr::NGRpcService {
+
+class IRequestOpCtx;
+class IFacilityProvider;
+
+void DoDescribeView(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f);
+
+}

ydb/core/grpc_services/ya.make

@@ -74,6 +74,7 @@ SRCS(
     rpc_stream_execute_yql_script.cpp
     rpc_whoami.cpp
     rpc_object_storage.cpp
+    rpc_view.cpp
     table_settings.cpp
 
     rpc_common/rpc_common_kqp_session.cpp

ydb/core/health_check/health_check.cpp

@@ -189,21 +189,34 @@ class TSelfCheckRequest : public TActorBootstrapped<TSelfCheckRequest> {
             int Count = 1;
             TStackVec<TString> Identifiers;
 
-            TNodeTabletStateCount(const NKikimrHive::TTabletInfo& info, const TTabletStateSettings& settings) {
-                Type = info.tablettype();
-                Leader = info.followerid() == 0;
+            static ETabletState GetState(const NKikimrHive::TTabletInfo& info, const TTabletStateSettings& settings) {
                 if (info.volatilestate() == NKikimrHive::TABLET_VOLATILE_STATE_STOPPED) {
-                    State = ETabletState::Stopped;
-                } else if (!settings.IsHiveSynchronizationPeriod
-                            && info.volatilestate() != NKikimrHive::TABLET_VOLATILE_STATE_RUNNING
-                            && TInstant::MilliSeconds(info.lastalivetimestamp()) < settings.AliveBarrier
-                            && info.tabletbootmode() == NKikimrHive::TABLET_BOOT_MODE_DEFAULT) {
-                    State = ETabletState::Dead;
-                } else if (info.restartsperperiod() >= settings.MaxRestartsPerPeriod) {
-                    State = ETabletState::RestartsTooOften;
-                } else {
-                    State = ETabletState::Good;
+                    return ETabletState::Stopped;
+                }
+                ETabletState state = (info.restartsperperiod() >= settings.MaxRestartsPerPeriod) ? ETabletState::RestartsTooOften : ETabletState::Good;
+                if (info.volatilestate() == NKikimrHive::TABLET_VOLATILE_STATE_RUNNING) {
+                    return state;
+                }
+                if (info.tabletbootmode() != NKikimrHive::TABLET_BOOT_MODE_DEFAULT) {
+                    return state;
+                }
+                if (info.lastalivetimestamp() != 0 && TInstant::MilliSeconds(info.lastalivetimestamp()) < settings.AliveBarrier) {
+                    // Tablet is not alive for a long time
+                    // We should report it as dead unless it's just waiting to be created
+                    if (info.generation() == 0 && info.volatilestate() == NKikimrHive::TABLET_VOLATILE_STATE_BOOTING && !info.inwaitqueue()) {
+                        return state;
+                    }
+                    return ETabletState::Dead;
                 }
+                return state;
+
+            }
+
+            TNodeTabletStateCount(const NKikimrHive::TTabletInfo& info, const TTabletStateSettings& settings)
+                : Type(info.tablettype())
+                , State(GetState(info, settings))
+                , Leader(info.followerid() == 0)
+            {
             }
 
             bool operator ==(const TNodeTabletStateCount& o) const {
@@ -1983,6 +1996,8 @@ class TSelfCheckRequest : public TActorBootstrapped<TSelfCheckRequest> {
             }
         }
 
+        // do not propagate RED status to vdisk - so that vdisk is not considered down when computing group status
+        context.OverallStatus = MinStatus(context.OverallStatus, Ydb::Monitoring::StatusFlag::ORANGE);
         storagePDiskStatus.set_overall(context.GetOverallStatus());
     }