Merge pull request #10477 from zverevgeny/analytics_to_stable

Analytics to stable

Author: maximyurchuk

.github/config/muted_ya.txt

@@ -8,6 +8,7 @@ ydb/core/external_sources *
 ydb/core/quoter/ut QuoterWithKesusTest.PrefetchCoefficient
 ydb/core/keyvalue/ut_trace TKeyValueTracingTest.*
 ydb/core/kqp/provider/ut KikimrIcGateway.TestLoadBasicSecretValueFromExternalDataSourceMetadata
+ydb/core/kqp/ut/join KqpJoinOrder.Chain65Nodes
 ydb/core/kqp/ut/olap KqpOlapBlobsSharing.*
 ydb/core/kqp/ut/olap KqpOlapStatistics.StatsUsageWithTTL
 ydb/core/kqp/ut/pg KqpPg.CreateIndex
@@ -17,6 +18,10 @@ 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
@@ -26,6 +31,8 @@ ydb/core/kqp/ut/scheme [15/50]*
 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

.gitignore

@@ -26,6 +26,9 @@ __pycache__/
 *.pb.h
 *.pb.cc
 
+# Other generated
+*.fbs.h
+
 # MacOS specific
 .DS_Store
 

library/cpp/threading/future/core/future-inl.h

@@ -116,6 +116,9 @@ namespace NThreading {
             bool HasException() const {
                 return AtomicGet(State) == ExceptionSet;
             }
+            bool IsReady() const {
+                return AtomicGet(State) != NotReady;
+            }
 
             const T& GetValue(TDuration timeout = TDuration::Zero()) const {
                 AccessValue(timeout, ValueRead);
@@ -297,6 +300,9 @@ namespace NThreading {
             bool HasException() const {
                 return AtomicGet(State) == ExceptionSet;
             }
+            bool IsReady() const {
+                return AtomicGet(State) != NotReady;
+            }
 
             void GetValue(TDuration timeout = TDuration::Zero()) const {
                 TAtomicBase state = AtomicGet(State);
@@ -583,6 +589,10 @@ namespace NThreading {
     inline bool TFuture<T>::HasException() const {
         return State && State->HasException();
     }
+    template <typename T>
+    inline bool TFuture<T>::IsReady() const {
+        return State && State->IsReady();
+    }
 
     template <typename T>
     inline void TFuture<T>::Wait() const {
@@ -688,6 +698,9 @@ namespace NThreading {
     inline bool TFuture<void>::HasException() const {
         return State && State->HasException();
     }
+    inline bool TFuture<void>::IsReady() const {
+        return State && State->IsReady();
+    }
 
     inline void TFuture<void>::Wait() const {
         EnsureInitialized();
@@ -823,6 +836,11 @@ namespace NThreading {
         return State && State->HasException();
     }
 
+    template <typename T>
+    inline bool TPromise<T>::IsReady() const {
+        return State && State->IsReady();
+    }
+
     template <typename T>
     inline void TPromise<T>::SetException(const TString& e) {
         EnsureInitialized();
@@ -904,6 +922,10 @@ namespace NThreading {
         return State && State->HasException();
     }
 
+    inline bool TPromise<void>::IsReady() const {
+        return State && State->IsReady();
+    }
+
     inline void TPromise<void>::SetException(const TString& e) {
         EnsureInitialized();
         State->SetException(std::make_exception_ptr(yexception() << e));

library/cpp/threading/future/core/future.h

@@ -98,6 +98,12 @@ namespace NThreading {
         void TryRethrow() const;
         bool HasException() const;
 
+        // returns true if exception or value was set.
+        //   allows to check readiness without locking cheker-thread
+        //   NOTE: returns true even if value was extracted from promise
+        //   good replace for HasValue() || HasException()
+        bool IsReady() const;
+
         void Wait() const;
         bool Wait(TDuration timeout) const;
         bool Wait(TInstant deadline) const;
@@ -153,6 +159,11 @@ namespace NThreading {
         void TryRethrow() const;
         bool HasException() const;
 
+        // returns true if exception or value was set.
+        //   allows to check readiness without locking cheker-thread
+        //   good replace for HasValue() || HasException()
+        bool IsReady() const;
+
         void Wait() const;
         bool Wait(TDuration timeout) const;
         bool Wait(TInstant deadline) const;
@@ -216,6 +227,12 @@ namespace NThreading {
 
         void TryRethrow() const;
         bool HasException() const;
+
+        // returns true if exception or value was set.
+        //   allows to check readiness without locking cheker-thread
+        //   NOTE: returns true even if value was extracted from promise
+        //   good replace for HasValue() || HasException()
+        bool IsReady() const;
         void SetException(const TString& e);
         void SetException(std::exception_ptr e);
         bool TrySetException(std::exception_ptr e);
@@ -256,6 +273,11 @@ namespace NThreading {
 
         void TryRethrow() const;
         bool HasException() const;
+
+        // returns true if exception or value was set.
+        //   allows to check readiness without locking cheker-thread
+        //   good replace for HasValue() || HasException()
+        bool IsReady() const;
         void SetException(const TString& e);
         void SetException(std::exception_ptr e);
         bool TrySetException(std::exception_ptr e);

library/cpp/threading/future/future_ut.cpp

@@ -105,6 +105,7 @@ namespace {
 
             future = MakeFuture(345);
             UNIT_ASSERT(future.HasValue());
+            UNIT_ASSERT(future.IsReady());
             UNIT_ASSERT_EQUAL(future.GetValue(), 345);
         }
 
@@ -115,6 +116,7 @@ namespace {
 
             TFuture<void> future = promise.GetFuture();
             UNIT_ASSERT(future.HasValue());
+            UNIT_ASSERT(future.IsReady());
 
             future = MakeFuture();
             UNIT_ASSERT(future.HasValue());
@@ -523,6 +525,7 @@ namespace {
         {
             auto future1 = MakeErrorFuture<void>(std::make_exception_ptr(TFutureException()));
             UNIT_ASSERT(future1.HasException());
+            UNIT_ASSERT(future1.IsReady());
             UNIT_CHECK_GENERATED_EXCEPTION(future1.GetValue(), TFutureException);
 
             auto future2 = MakeErrorFuture<int>(std::make_exception_ptr(TFutureException()));
@@ -563,6 +566,7 @@ namespace {
             promise2.SetException("foo-exception");
             wait.Wait();
             UNIT_ASSERT(future2.HasException());
+            UNIT_ASSERT(!future1.IsReady());
             UNIT_ASSERT(!future1.HasValue() && !future1.HasException());
         }
 

ydb/core/driver_lib/run/kikimr_services_initializers.cpp

@@ -123,7 +123,7 @@
 
 #include <ydb/core/sys_view/processor/processor.h>
 #include <ydb/core/sys_view/service/sysview_service.h>
-#include <ydb/core/statistics/stat_service.h>
+#include <ydb/core/statistics/service/service.h>
 #include <ydb/core/statistics/aggregator/aggregator.h>
 
 #include <ydb/core/tablet/bootstrapper.h>

ydb/core/driver_lib/run/ya.make

@@ -108,8 +108,8 @@ PEERDIR(
     ydb/core/scheme_types
     ydb/core/security
     ydb/core/security/ldap_auth_provider
-    ydb/core/statistics
     ydb/core/statistics/aggregator
+    ydb/core/statistics/service    
     ydb/core/sys_view/processor
     ydb/core/sys_view/service
     ydb/core/tablet

ydb/core/kqp/common/events/events.h

@@ -111,6 +111,18 @@ struct TEvKqp {
     struct TEvScriptRequest : public TEventLocal<TEvScriptRequest, TKqpEvents::EvScriptRequest> {
         TEvScriptRequest() = default;
 
+        const TString& GetDatabase() const {
+            return Record.GetRequest().GetDatabase();
+        }
+
+        const TString& GetDatabaseId() const {
+            return Record.GetRequest().GetDatabaseId();
+        }
+
+        void SetDatabaseId(const TString& databaseId) {
+            Record.MutableRequest()->SetDatabaseId(databaseId);
+        }
+
         mutable NKikimrKqp::TEvQueryRequest Record;
         TDuration ForgetAfter;
         TDuration ResultsTtl;
@@ -164,6 +176,40 @@ struct TEvKqp {
             return issues;
         }
     };
+
+    struct TEvUpdateDatabaseInfo : public TEventLocal<TEvUpdateDatabaseInfo, TKqpEvents::EvUpdateDatabaseInfo> {
+        TEvUpdateDatabaseInfo(const TString& database, Ydb::StatusIds::StatusCode status, NYql::TIssues issues)
+            : Status(status)
+            , Database(database)
+            , Issues(std::move(issues))
+        {}
+
+        TEvUpdateDatabaseInfo(const TString& database, const TString& databaseId, bool serverless)
+            : Status(Ydb::StatusIds::SUCCESS)
+            , Database(database)
+            , DatabaseId(databaseId)
+            , Serverless(serverless)
+            , Issues({})
+        {}
+
+        Ydb::StatusIds::StatusCode Status;
+        TString Database;
+        TString DatabaseId;
+        bool Serverless = false;
+        NYql::TIssues Issues;
+    };
+
+    struct TEvDelayedRequestError : public TEventLocal<TEvDelayedRequestError, TKqpEvents::EvDelayedRequestError> {
+        TEvDelayedRequestError(THolder<IEventHandle> requestEvent, Ydb::StatusIds::StatusCode status, NYql::TIssues issues)
+            : RequestEvent(std::move(requestEvent))
+            , Status(status)
+            , Issues(std::move(issues))
+        {}
+
+        THolder<IEventHandle> RequestEvent;
+        Ydb::StatusIds::StatusCode Status;
+        NYql::TIssues Issues;
+    };
 };
 
 } // namespace NKikimr::NKqp

ydb/core/kqp/common/events/query.h

@@ -353,6 +353,14 @@ struct TEvQueryRequest: public NActors::TEventLocal<TEvQueryRequest, TKqpEvents:
         return PoolConfig;
     }
 
+    const TString& GetDatabaseId() const {
+        return DatabaseId ? DatabaseId : Record.GetRequest().GetDatabaseId();
+    }
+
+    void SetDatabaseId(const TString& databaseId) {
+        DatabaseId = databaseId;
+    }
+
     mutable NKikimrKqp::TEvQueryRequest Record;
 
 private:
@@ -365,6 +373,7 @@ struct TEvQueryRequest: public NActors::TEventLocal<TEvQueryRequest, TKqpEvents:
     mutable TIntrusiveConstPtr<NACLib::TUserToken> Token_;
     TActorId RequestActorId;
     TString Database;
+    TString DatabaseId;
     TString SessionId;
     TString YqlText;
     TString QueryId;

ydb/core/kqp/common/events/script_executions.h

@@ -22,13 +22,34 @@ enum EFinalizationStatus : i32 {
     FS_ROLLBACK,
 };
 
-struct TEvForgetScriptExecutionOperation : public NActors::TEventLocal<TEvForgetScriptExecutionOperation, TKqpScriptExecutionEvents::EvForgetScriptExecutionOperation> {
-    TEvForgetScriptExecutionOperation(const TString& database, const NOperationId::TOperationId& id)
+template <typename TEv, ui32 TEventType>
+struct TEventWithDatabaseId : public NActors::TEventLocal<TEv, TEventType> {
+    TEventWithDatabaseId(const TString& database)
         : Database(database)
-        , OperationId(id)
     {}
 
+    const TString& GetDatabase() const {
+        return Database;
+    }
+
+    const TString& GetDatabaseId() const {
+        return DatabaseId;
+    }
+
+    void SetDatabaseId(const TString& databaseId) {
+        DatabaseId = databaseId;
+    }
+
     const TString Database;
+    TString DatabaseId;
+};
+
+struct TEvForgetScriptExecutionOperation : public TEventWithDatabaseId<TEvForgetScriptExecutionOperation, TKqpScriptExecutionEvents::EvForgetScriptExecutionOperation> {
+    TEvForgetScriptExecutionOperation(const TString& database, const NOperationId::TOperationId& id)
+        : TEventWithDatabaseId(database)
+        , OperationId(id)
+    {}
+
     const NOperationId::TOperationId OperationId;
 };
 
@@ -43,14 +64,12 @@ struct TEvForgetScriptExecutionOperationResponse : public NActors::TEventLocal<T
     NYql::TIssues Issues;
 };
 
-struct TEvGetScriptExecutionOperation : public NActors::TEventLocal<TEvGetScriptExecutionOperation, TKqpScriptExecutionEvents::EvGetScriptExecutionOperation> {
-    explicit TEvGetScriptExecutionOperation(const TString& database, const NOperationId::TOperationId& id)
-        : Database(database)
+struct TEvGetScriptExecutionOperation : public TEventWithDatabaseId<TEvGetScriptExecutionOperation, TKqpScriptExecutionEvents::EvGetScriptExecutionOperation> {
+    TEvGetScriptExecutionOperation(const TString& database, const NOperationId::TOperationId& id)
+        : TEventWithDatabaseId(database)
         , OperationId(id)
-    {
-    }
+    {}
 
-    TString Database;
     NOperationId::TOperationId OperationId;
 };
 
@@ -97,14 +116,13 @@ struct TEvGetScriptExecutionOperationResponse : public NActors::TEventLocal<TEvG
     TMaybe<google::protobuf::Any> Metadata;
 };
 
-struct TEvListScriptExecutionOperations : public NActors::TEventLocal<TEvListScriptExecutionOperations, TKqpScriptExecutionEvents::EvListScriptExecutionOperations> {
+struct TEvListScriptExecutionOperations : public TEventWithDatabaseId<TEvListScriptExecutionOperations, TKqpScriptExecutionEvents::EvListScriptExecutionOperations> {
     TEvListScriptExecutionOperations(const TString& database, const ui64 pageSize, const TString& pageToken)
-        : Database(database)
+        : TEventWithDatabaseId(database)
         , PageSize(pageSize)
         , PageToken(pageToken)
     {}
 
-    TString Database;
     ui64 PageSize;
     TString PageToken;
 };
@@ -151,14 +169,12 @@ struct TEvCheckAliveRequest : public NActors::TEventPB<TEvCheckAliveRequest, NKi
 struct TEvCheckAliveResponse : public NActors::TEventPB<TEvCheckAliveResponse, NKikimrKqp::TEvCheckAliveResponse, TKqpScriptExecutionEvents::EvCheckAliveResponse> {
 };
 
-struct TEvCancelScriptExecutionOperation : public NActors::TEventLocal<TEvCancelScriptExecutionOperation, TKqpScriptExecutionEvents::EvCancelScriptExecutionOperation> {
-    explicit TEvCancelScriptExecutionOperation(const TString& database, const NOperationId::TOperationId& id)
-        : Database(database)
+struct TEvCancelScriptExecutionOperation : public TEventWithDatabaseId<TEvCancelScriptExecutionOperation, TKqpScriptExecutionEvents::EvCancelScriptExecutionOperation> {
+    TEvCancelScriptExecutionOperation(const TString& database, const NOperationId::TOperationId& id)
+        : TEventWithDatabaseId(database)
         , OperationId(id)
-    {
-    }
+    {}
 
-    TString Database;
     NOperationId::TOperationId OperationId;
 };