Publish in KQP plan S3 Source info

6 files changed, 34 insertions, 2 deletions

diff --git a/ydb/core/kqp/opt/CMakeLists.darwin-x86_64.txt b/ydb/core/kqp/opt/CMakeLists.darwin-x86_64.txt
index 9c3d4e474f..6e1f5dcb11 100644
--- a/ydb/core/kqp/opt/CMakeLists.darwin-x86_64.txt
+++ b/ydb/core/kqp/opt/CMakeLists.darwin-x86_64.txt
@@ -29,6 +29,7 @@ target_link_libraries(core-kqp-opt PUBLIC
   kqp-opt-physical
   yql-dq-common
   yql-dq-opt
+  providers-s3-expr_nodes
   core-kqp-provider
   tools-enum_parser-enum_serialization_runtime
 )
diff --git a/ydb/core/kqp/opt/CMakeLists.linux-aarch64.txt b/ydb/core/kqp/opt/CMakeLists.linux-aarch64.txt
index 3cb516fc56..7b8adeb4e8 100644
--- a/ydb/core/kqp/opt/CMakeLists.linux-aarch64.txt
+++ b/ydb/core/kqp/opt/CMakeLists.linux-aarch64.txt
@@ -30,6 +30,7 @@ target_link_libraries(core-kqp-opt PUBLIC
   kqp-opt-physical
   yql-dq-common
   yql-dq-opt
+  providers-s3-expr_nodes
   core-kqp-provider
   tools-enum_parser-enum_serialization_runtime
 )
diff --git a/ydb/core/kqp/opt/CMakeLists.linux-x86_64.txt b/ydb/core/kqp/opt/CMakeLists.linux-x86_64.txt
index 3cb516fc56..7b8adeb4e8 100644
--- a/ydb/core/kqp/opt/CMakeLists.linux-x86_64.txt
+++ b/ydb/core/kqp/opt/CMakeLists.linux-x86_64.txt
@@ -30,6 +30,7 @@ target_link_libraries(core-kqp-opt PUBLIC
   kqp-opt-physical
   yql-dq-common
   yql-dq-opt
+  providers-s3-expr_nodes
   core-kqp-provider
   tools-enum_parser-enum_serialization_runtime
 )
diff --git a/ydb/core/kqp/opt/CMakeLists.windows-x86_64.txt b/ydb/core/kqp/opt/CMakeLists.windows-x86_64.txt
index 9c3d4e474f..6e1f5dcb11 100644
--- a/ydb/core/kqp/opt/CMakeLists.windows-x86_64.txt
+++ b/ydb/core/kqp/opt/CMakeLists.windows-x86_64.txt
@@ -29,6 +29,7 @@ target_link_libraries(core-kqp-opt PUBLIC
   kqp-opt-physical
   yql-dq-common
   yql-dq-opt
+  providers-s3-expr_nodes
   core-kqp-provider
   tools-enum_parser-enum_serialization_runtime
 )
diff --git a/ydb/core/kqp/opt/kqp_query_plan.cpp b/ydb/core/kqp/opt/kqp_query_plan.cpp
index 0529b4735c..2aa2a17de8 100644
--- a/ydb/core/kqp/opt/kqp_query_plan.cpp
+++ b/ydb/core/kqp/opt/kqp_query_plan.cpp
@@ -6,10 +6,12 @@
 #include <ydb/public/lib/value/value.h>
 
 #include <ydb/library/yql/ast/yql_ast_escaping.h>
+#include <ydb/library/yql/core/expr_nodes/yql_expr_nodes.h>
 #include <ydb/library/yql/core/yql_expr_optimize.h>
 #include <ydb/library/yql/core/yql_opt_utils.h>
 #include <ydb/library/yql/dq/opt/dq_opt.h>
 #include <ydb/library/yql/dq/tasks/dq_tasks_graph.h>
+#include <ydb/library/yql/providers/s3/expr_nodes/yql_s3_expr_nodes.h>
 
 #include <library/cpp/json/writer/json.h>
 #include <library/cpp/json/json_reader.h>
@@ -895,9 +897,34 @@ private:
 
             for (const auto& input : expr.Cast<TDqStageBase>().Inputs()) {
                 if (auto source = input.Maybe<TDqSource>()) {
-                    auto settings = source.Settings().Maybe<TKqpReadRangesSourceSettings>();
-                    if (settings.IsValid()) {
+                    if (auto settings = source.Settings().Maybe<TKqpReadRangesSourceSettings>(); settings.IsValid()) {
                         Visit(settings.Cast(), stagePlanNode);
+                    } else if (auto settings = source.Settings().Maybe<TS3SourceSettings>(); settings.IsValid()) {
+                        TOperator op;
+                        op.Properties["Name"] = S3ProviderName;
+                        op.Properties["Format"] = "raw";
+                        auto cluster = source.Cast().DataSource().Cast<TS3DataSource>().Cluster().StringValue();
+                        if (auto pos = cluster.rfind('/'); pos != TString::npos) {
+                            cluster = cluster.substr(pos + 1);
+                        }
+                        op.Properties["Cluster"] = cluster;
+                        AddOperator(stagePlanNode, "Source", op);
+                    } else if (auto settings = source.Settings().Maybe<TS3ParseSettingsBase>(); settings.IsValid()) {
+                        TOperator op;
+                        op.Properties["Name"] = S3ProviderName;
+                        op.Properties["Format"] = settings.Cast().Format().StringValue();
+                        auto cluster = source.Cast().DataSource().Cast<TS3DataSource>().Cluster().StringValue();
+                        if (auto pos = cluster.rfind('/'); pos != TString::npos) {
+                            cluster = cluster.substr(pos + 1);
+                        }
+                        op.Properties["Cluster"] = cluster;
+                        const TStructExprType* fullRowType = settings.Cast().RowType().Ref().GetTypeAnn()->Cast<TTypeExprType>()->GetType()->Cast<TStructExprType>();
+                        auto rowTypeItems = fullRowType->GetItems();
+                        auto& columns = op.Properties["ReadColumns"];
+                        for (auto& item : rowTypeItems) {
+                            columns.AppendValue(item->GetName());
+                        }
+                        AddOperator(stagePlanNode, "Source", op);
                     } else {
                         TOperator op;
                         op.Properties["Name"] = source.Cast().DataSource().Cast<TCoDataSource>().Category().StringValue();
diff --git a/ydb/core/kqp/opt/ya.make b/ydb/core/kqp/opt/ya.make
index 943fe63957..bc13e15d34 100644
--- a/ydb/core/kqp/opt/ya.make
+++ b/ydb/core/kqp/opt/ya.make
@@ -22,6 +22,7 @@ PEERDIR(
     ydb/core/kqp/opt/physical
     ydb/library/yql/dq/common
     ydb/library/yql/dq/opt
+    ydb/library/yql/providers/s3/expr_nodes
     ydb/core/kqp/provider
 )