[pg] initialize tx timestamps

влияет на функции now, statement_timestamp - они будут брать кешированное значение с момента инициализации comp graph.
clock_timestamp сейчас всегда возвращает некешированное системное время - это потом надо будет увести в TimeProvider

4 files changed, 48 insertions, 21 deletions

diff --git a/library/cpp/yt/memory/serialize-inl.h b/library/cpp/yt/memory/serialize-inl.h
new file mode 100644
index 0000000000..493ba9101e
--- /dev/null
+++ b/library/cpp/yt/memory/serialize-inl.h
@@ -0,0 +1,35 @@
+#ifndef SERIALIZE_PTR_INL_H_
+#error "Direct inclusion of this file is not allowed, include serialize.h"
+// For the sake of sane code completion.
+#include "serialize.h"
+#endif
+
+#include "new.h"
+
+////////////////////////////////////////////////////////////////////////////////
+
+template <class T>
+void TSerializer<NYT::TIntrusivePtr<T>>::Save(IOutputStream* output, const NYT::TIntrusivePtr<T>& ptr)
+{
+    bool hasValue = ptr.operator bool();
+    ::Save(output, hasValue);
+    if (hasValue) {
+        ::Save(output, *ptr);
+    }
+}
+
+template <class T>
+void TSerializer<NYT::TIntrusivePtr<T>>::Load(IInputStream* input, NYT::TIntrusivePtr<T>& ptr)
+{
+    bool hasValue;
+    ::Load(input, hasValue);
+    if (hasValue) {
+        auto tmp = NYT::New<T>();
+        ::Load(input, *tmp);
+        ptr = std::move(tmp);
+    } else {
+        ptr.Reset();
+    }
+}
+
+////////////////////////////////////////////////////////////////////////////////
diff --git a/library/cpp/yt/memory/serialize.h b/library/cpp/yt/memory/serialize.h
index 03fc286515..c113e04961 100644
--- a/library/cpp/yt/memory/serialize.h
+++ b/library/cpp/yt/memory/serialize.h
@@ -1,6 +1,6 @@
 #pragma once
 
-#include "new.h"
+#include "intrusive_ptr.h"
 
 ////////////////////////////////////////////////////////////////////////////////
 
@@ -8,27 +8,13 @@ template <class T>
 class TSerializer<NYT::TIntrusivePtr<T>>
 {
 public:
-    static inline void Save(IOutputStream* out, const NYT::TIntrusivePtr<T>& ptr)
-    {
-        bool hasValue = ptr.operator bool();
-        ::Save(out, hasValue);
-        if (hasValue) {
-            ::Save(out, *ptr);
-        }
-    }
+    static inline void Save(IOutputStream* output, const NYT::TIntrusivePtr<T>& ptr);
 
-    static inline void Load(IInputStream* in, NYT::TIntrusivePtr<T>& ptr)
-    {
-        bool hasValue;
-        ::Load(in, hasValue);
-        if (hasValue) {
-            auto tmp = NYT::New<T>();
-            ::Load(in, *tmp);
-            ptr = std::move(tmp);
-        } else {
-            ptr.Reset();
-        }
-    }
+    static inline void Load(IInputStream* input, NYT::TIntrusivePtr<T>& ptr);
 };
 
 ////////////////////////////////////////////////////////////////////////////////
+
+#define SERIALIZE_PTR_INL_H_
+#include "serialize-inl.h"
+#undef SERIALIZE_PTR_INL_H_
diff --git a/library/cpp/yt/memory/unittests/intrusive_ptr_ut.cpp b/library/cpp/yt/memory/unittests/intrusive_ptr_ut.cpp
index 9a94c82411..6171dc7aee 100644
--- a/library/cpp/yt/memory/unittests/intrusive_ptr_ut.cpp
+++ b/library/cpp/yt/memory/unittests/intrusive_ptr_ut.cpp
@@ -5,7 +5,9 @@
 #include <library/cpp/yt/memory/serialize.h>
 
 #include <util/generic/buffer.h>
+
 #include <util/stream/buffer.h>
+
 #include <util/ysaveload.h>
 
 namespace NYT {
diff --git a/ydb/library/yql/parser/pg_wrapper/comp_factory.cpp b/ydb/library/yql/parser/pg_wrapper/comp_factory.cpp
index 68d87ae532..5d7b386667 100644
--- a/ydb/library/yql/parser/pg_wrapper/comp_factory.cpp
+++ b/ydb/library/yql/parser/pg_wrapper/comp_factory.cpp
@@ -23,6 +23,7 @@
 #undef SIZEOF_SIZE_T
 extern "C" {
 #include "postgres.h"
+#include "access/xact.h"
 #include "catalog/pg_type_d.h"
 #include "catalog/pg_collation_d.h"
 #include "utils/builtins.h"
@@ -67,6 +68,7 @@ struct TMainContext {
     MemoryContextData Data;
     MemoryContext PrevCurrentMemoryContext = nullptr;
     MemoryContext PrevErrorContext = nullptr;
+    TimestampTz StartTimestamp;
 };
 
 ui32 GetFullVarSize(const text* s) {
@@ -2787,6 +2789,7 @@ void* PgInitializeMainContext() {
         &MkqlMethods,
         nullptr,
         "mkql");
+    ctx->StartTimestamp = GetCurrentTimestamp();
     return ctx;
 }
 
@@ -2801,6 +2804,7 @@ void PgAcquireThreadContext(void* ctx) {
         main->PrevCurrentMemoryContext = CurrentMemoryContext;
         main->PrevErrorContext = ErrorContext;
         CurrentMemoryContext = ErrorContext = (MemoryContext)&main->Data;
+        SetParallelStartTimestamps(main->StartTimestamp, main->StartTimestamp);
     }
 }