Merge 833eaca into 8d2cc65

Author: uzhastik

ydb/core/fq/libs/init/init.cpp

@@ -218,7 +218,7 @@ void Init(
                 protoConfig.GetGateways().GetS3().GetBlockFileSizeLimit();
         }
 
-        RegisterDqPqReadActorFactory(*asyncIoFactory, yqSharedResources->UserSpaceYdbDriver, credentialsFactory, !protoConfig.GetReadActorsFactoryConfig().GetPqReadActorFactoryConfig().GetCookieCommitMode());
+        RegisterDqPqReadActorFactory(*asyncIoFactory, yqSharedResources->UserSpaceYdbDriver, credentialsFactory);
         RegisterYdbReadActorFactory(*asyncIoFactory, yqSharedResources->UserSpaceYdbDriver, credentialsFactory);
         RegisterS3ReadActorFactory(*asyncIoFactory, credentialsFactory, httpGateway, s3HttpRetryPolicy, readActorFactoryCfg,
             yqCounters->GetSubgroup("subsystem", "S3ReadActor"));

ydb/library/yql/providers/pq/async_io/dq_pq_meta_extractor.cpp

@@ -1,20 +1,12 @@
 #include "dq_pq_meta_extractor.h"
 
-#include <optional>
-
 #include <ydb/library/yql/minikql/mkql_string_util.h>
 #include <ydb/library/yql/providers/pq/common/pq_meta_fields.h>
-#include <ydb/library/yql/public/udf/udf_data_type.h>
-#include <ydb/library/yql/public/udf/udf_value.h>
-
-#include <ydb/public/sdk/cpp/client/ydb_persqueue_core/persqueue.h>
-
-#include <util/generic/string.h>
 
 namespace {
     const std::unordered_map<TString, NYql::NDq::TPqMetaExtractor::TPqMetaExtractorLambda> ExtractorsMap = {
         {
-            "_yql_sys_create_time", [](const NYdb::NPersQueue::TReadSessionEvent::TDataReceivedEvent::TMessage& message){
+            "_yql_sys_create_time", [](const NYdb::NTopic::TReadSessionEvent::TDataReceivedEvent::TMessage& message){
                 using TDataType = NYql::NUdf::TDataType<NYql::NUdf::TTimestamp>;
                 return std::make_pair(
                     NYql::NUdf::TUnboxedValuePod(static_cast<TDataType::TLayout>(message.GetCreateTime().MicroSeconds())),
@@ -23,7 +15,7 @@ namespace {
             }
         },
         {
-            "_yql_sys_tsp_write_time", [](const NYdb::NPersQueue::TReadSessionEvent::TDataReceivedEvent::TMessage& message){
+            "_yql_sys_tsp_write_time", [](const NYdb::NTopic::TReadSessionEvent::TDataReceivedEvent::TMessage& message){
                 using TDataType = NYql::NUdf::TDataType<NYql::NUdf::TTimestamp>;
                 return std::make_pair(
                     NYql::NUdf::TUnboxedValuePod(static_cast<TDataType::TLayout>(message.GetWriteTime().MicroSeconds())),
@@ -32,24 +24,24 @@ namespace {
             }
         },
         {
-            "_yql_sys_partition_id", [](const NYdb::NPersQueue::TReadSessionEvent::TDataReceivedEvent::TMessage& message){
+            "_yql_sys_partition_id", [](const NYdb::NTopic::TReadSessionEvent::TDataReceivedEvent::TMessage& message){
                 using TDataType = NYql::NUdf::TDataType<ui64>;
                 return std::make_pair(
-                    NYql::NUdf::TUnboxedValuePod(message.GetPartitionStream()->GetPartitionId()),
+                    NYql::NUdf::TUnboxedValuePod(message.GetPartitionSession()->GetPartitionId()),
                     NYql::NUdf::GetDataTypeInfo(TDataType::Slot).FixedSize
                 );
             }
         },
         {
-            "_yql_sys_offset", [](const NYdb::NPersQueue::TReadSessionEvent::TDataReceivedEvent::TMessage& message){
+            "_yql_sys_offset", [](const NYdb::NTopic::TReadSessionEvent::TDataReceivedEvent::TMessage& message){
                 using TDataType = NYql::NUdf::TDataType<ui64>;
                 return std::make_pair(
                     NYql::NUdf::TUnboxedValuePod(message.GetOffset()),
                     NYql::NUdf::GetDataTypeInfo(TDataType::Slot).FixedSize);
             }
         },
         {
-            "_yql_sys_message_group_id", [](const NYdb::NPersQueue::TReadSessionEvent::TDataReceivedEvent::TMessage& message){
+            "_yql_sys_message_group_id", [](const NYdb::NTopic::TReadSessionEvent::TDataReceivedEvent::TMessage& message){
                 const auto& data = message.GetMessageGroupId();
                 return std::make_pair(
                     NKikimr::NMiniKQL::MakeString(NYql::NUdf::TStringRef(data.Data(), data.Size())),
@@ -58,7 +50,7 @@ namespace {
             }
         },
         {
-            "_yql_sys_seq_no", [](const NYdb::NPersQueue::TReadSessionEvent::TDataReceivedEvent::TMessage& message){
+            "_yql_sys_seq_no", [](const NYdb::NTopic::TReadSessionEvent::TDataReceivedEvent::TMessage& message){
                 using TDataType = NYql::NUdf::TDataType<ui64>;
                 return std::make_pair(
                     NYql::NUdf::TUnboxedValuePod(message.GetSeqNo()),
@@ -72,14 +64,14 @@ namespace {
 namespace NYql::NDq {
 
 TPqMetaExtractor::TPqMetaExtractor() {
-    for (auto key : AllowedPqMetaSysColumns()) {
+    for (const auto& key : AllowedPqMetaSysColumns()) {
         Y_ENSURE(
             ExtractorsMap.contains(key),
             "Pq metadata field " << key << " hasn't valid runtime extractor. You should add it.");
     }
 }
 
-TPqMetaExtractor::TPqMetaExtractorLambda TPqMetaExtractor::FindExtractorLambda(TString sysColumn) const {
+TPqMetaExtractor::TPqMetaExtractorLambda TPqMetaExtractor::FindExtractorLambda(const TString& sysColumn) const {
     auto iter = ExtractorsMap.find(sysColumn);
     Y_ENSURE(iter != ExtractorsMap.end(), sysColumn);
 

ydb/library/yql/providers/pq/async_io/dq_pq_meta_extractor.h

@@ -1,21 +1,20 @@
 #pragma once
 
-#include "ydb/library/yql/minikql/mkql_string_util.h"
-#include <optional>
+#include <functional>
 
 #include <ydb/library/yql/public/udf/udf_data_type.h>
 #include <ydb/library/yql/public/udf/udf_value.h>
 
-#include <ydb/public/sdk/cpp/client/ydb_persqueue_core/persqueue.h>
+#include <ydb/public/sdk/cpp/client/ydb_topic/topic.h>
 
 #include <util/generic/string.h>
 
 namespace NYql::NDq {
     struct TPqMetaExtractor {
-        using TPqMetaExtractorLambda = std::function<std::pair<NYql::NUdf::TUnboxedValuePod, i64>(const NYdb::NPersQueue::TReadSessionEvent::TDataReceivedEvent::TMessage&)>;
+        using TPqMetaExtractorLambda = std::function<std::pair<NYql::NUdf::TUnboxedValuePod, i64>(const NYdb::NTopic::TReadSessionEvent::TDataReceivedEvent::TMessage&)>;
 
     public:
         TPqMetaExtractor();
-        TPqMetaExtractorLambda FindExtractorLambda(TString sysColumn) const;
+        TPqMetaExtractorLambda FindExtractorLambda(const TString& sysColumn) const;
     };
 }