Copy arrow helpers for building ResultSet to KQP (ydb-platform#25531)

Author: dahbka-lis

ydb/core/formats/arrow/arrow_batch_builder.cpp

@@ -3,7 +3,7 @@
 #include <ydb/core/formats/arrow/arrow_helpers_minikql.h>
 #include <ydb/core/formats/arrow/switch/switch_type.h>
 #include <ydb/core/kqp/common/kqp_types.h>
-#include <ydb/library/yql/dq/runtime/dq_arrow_helpers.h>
+#include <ydb/core/kqp/common/result_set_format/kqp_result_set_arrow.h>
 
 #include <contrib/libs/apache/arrow/cpp/src/arrow/io/memory.h>
 #include <contrib/libs/apache/arrow/cpp/src/arrow/ipc/reader.h>
@@ -94,7 +94,7 @@ arrow::Status AppendCell(arrow::RecordBatchBuilder& builder, const TCell& cell,
 
 arrow::Status AppendValue(arrow::RecordBatchBuilder& builder, const NUdf::TUnboxedValue& value, ui32 colNum, const NKikimr::NMiniKQL::TType* type) {
     try {
-        NYql::NArrow::AppendElement(value, builder.GetField(colNum), type);
+        NKqp::NFormats::AppendElement(value, builder.GetField(colNum), type);
     } catch (const std::exception& e) {
         return arrow::Status::FromArgs(arrow::StatusCode::Invalid, e.what());
     }
@@ -248,7 +248,7 @@ arrow::Status TArrowBatchBuilder::Start(const std::vector<std::pair<TString, NSc
     return arrow::Status::OK();
 }
 
-arrow::Status TArrowBatchBuilder::Start(const std::vector<std::pair<TString, NKikimr::NMiniKQL::TType*>> yqlColumns) {
+arrow::Status TArrowBatchBuilder::Start(const std::vector<std::pair<TString, NKikimr::NMiniKQL::TType*>>& yqlColumns) {
     YqlSchema = yqlColumns;
     auto schema = MakeArrowSchema(yqlColumns, NotNullColumns);
     if (!schema.ok()) {

ydb/core/formats/arrow/arrow_batch_builder.h

@@ -193,7 +193,7 @@ class TArrowBatchBuilder : public NKikimr::IBlockBuilder {
     }
 
     arrow::Status Start(const std::vector<std::pair<TString, NScheme::TTypeInfo>>& columns);
-    arrow::Status Start(const std::vector<std::pair<TString, NKikimr::NMiniKQL::TType*>> columns);
+    arrow::Status Start(const std::vector<std::pair<TString, NKikimr::NMiniKQL::TType*>>& columns);
 
     std::shared_ptr<arrow::RecordBatch> FlushBatch(bool reinitialize, bool flushEmpty = false);
     std::shared_ptr<arrow::RecordBatch> GetBatch() const { return Batch; }
@@ -206,7 +206,7 @@ class TArrowBatchBuilder : public NKikimr::IBlockBuilder {
         return YdbSchema;
     }
 
-    const std::vector<std::pair<TString, NKikimr::NMiniKQL::TType*>> GetYqlSchema() const {
+    const std::vector<std::pair<TString, NKikimr::NMiniKQL::TType*>>& GetYqlSchema() const {
         return YqlSchema;
     }
 

ydb/core/formats/arrow/arrow_helpers_minikql.cpp

@@ -1,6 +1,6 @@
 #include "arrow_helpers_minikql.h"
 
-#include <ydb/library/yql/dq/runtime/dq_arrow_helpers.h>
+#include <ydb/core/kqp/common/result_set_format/kqp_result_set_arrow.h>
 #include <util/string/join.h>
 
 namespace NKikimr::NArrow {
@@ -15,7 +15,7 @@ arrow::Result<arrow::FieldVector> MakeArrowFields(
         std::shared_ptr<arrow::DataType> arrowType;
 
         try {
-            arrowType = NYql::NArrow::GetArrowType(mkqlType);
+            arrowType = NKqp::NFormats::GetArrowType(mkqlType);
         } catch (const yexception& e) {
             errors.emplace_back(colName + " error: " + e.what());
         }

ydb/core/formats/arrow/ya.make

@@ -24,7 +24,7 @@ PEERDIR(
     ydb/library/services
     yql/essentials/core/arrow_kernels/request
     yql/essentials/minikql
-    ydb/library/yql/dq/runtime
+    ydb/core/kqp/common/result_set_format
 )
 
 YQL_LAST_ABI_VERSION()

ydb/core/grpc_services/query/rpc_execute_query.cpp

@@ -275,9 +275,9 @@ class TExecuteQueryRPC : public TActorBootstrapped<TExecuteQueryRPC> {
         Ydb::Query::SchemaInclusionMode schemaInclusionMode = req->schema_inclusion_mode();
         Ydb::ResultSet::Format resultSetFormat = req->result_set_format();
 
-        std::optional<NKqp::TArrowFormatSettings> arrowFormatSettings;
+        std::optional<NKqp::NFormats::TArrowFormatSettings> arrowFormatSettings;
         if (req->has_arrow_format_settings()) {
-            arrowFormatSettings = NKqp::TArrowFormatSettings::ImportFromProto(req->arrow_format_settings());
+            arrowFormatSettings = NKqp::NFormats::TArrowFormatSettings::ImportFromProto(req->arrow_format_settings());
         }
 
         AuditContextAppend(Request_.get(), *req);

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

@@ -2,7 +2,7 @@
 
 #include <ydb/core/resource_pools/resource_pool_settings.h>
 #include <ydb/core/protos/kqp.pb.h>
-#include <ydb/core/kqp/common/kqp_result_set_format_settings.h>
+#include <ydb/core/kqp/common/result_set_format/kqp_result_set_format_settings.h>
 #include <ydb/core/kqp/common/kqp_user_request_context.h>
 #include <ydb/core/kqp/common/simple/kqp_event_ids.h>
 #include <ydb/core/grpc_services/base/iface.h>
@@ -83,7 +83,7 @@ struct TEvQueryRequest: public NActors::TEventLocal<TEvQueryRequest, TKqpEvents:
         const ::Ydb::Operations::OperationParams* operationParams,
         const TQueryRequestSettings& querySettings = TQueryRequestSettings(),
         const TString& poolId = "",
-        std::optional<NKqp::TArrowFormatSettings> arrowFormatSettings = std::nullopt);
+        std::optional<NFormats::TArrowFormatSettings> arrowFormatSettings = std::nullopt);
 
     TEvQueryRequest() {
         Record.MutableRequest()->SetUsePublicResponseDataFormat(true);
@@ -391,7 +391,7 @@ struct TEvQueryRequest: public NActors::TEventLocal<TEvQueryRequest, TKqpEvents:
         return ArrowFormatSettings.has_value();
     }
 
-    std::optional<NKqp::TArrowFormatSettings> GetArrowFormatSettings() const {
+    std::optional<NFormats::TArrowFormatSettings> GetArrowFormatSettings() const {
         return ArrowFormatSettings;
     }
 
@@ -457,7 +457,7 @@ struct TEvQueryRequest: public NActors::TEventLocal<TEvQueryRequest, TKqpEvents:
     TIntrusivePtr<TUserRequestContext> UserRequestContext;
     TDuration ProgressStatsPeriod;
     std::optional<NResourcePool::TPoolSettings> PoolConfig;
-    std::optional<NKqp::TArrowFormatSettings> ArrowFormatSettings;
+    std::optional<NFormats::TArrowFormatSettings> ArrowFormatSettings;
     bool SaveQueryPhysicalGraph = false;  // Used only in execute script queries
     std::shared_ptr<const NKikimrKqp::TQueryPhysicalGraph> QueryPhysicalGraph;
     i64 Generation = 0;

ydb/core/kqp/common/kqp_event_impl.cpp

@@ -20,7 +20,7 @@ TEvKqp::TEvQueryRequest::TEvQueryRequest(
     const ::Ydb::Operations::OperationParams* operationParams,
     const TQueryRequestSettings& querySettings,
     const TString& poolId,
-    std::optional<NKqp::TArrowFormatSettings> arrowFormatSettings)
+    std::optional<NKqp::NFormats::TArrowFormatSettings> arrowFormatSettings)
     : RequestCtx(ctx)
     , RequestActorId(requestActorId)
     , Database(CanonizePath(ctx->GetDatabaseName().GetOrElse("")))