YQL-13448: Introduce window RANGE logic

Introduce window RANGE logic.

Зеркальный пр с включение фичи путем инлайнинга в коде <https://nda.ya.ru/t/p2qAEoNq7SNwUJ>
Зеркальный пр где я провожу perf измерения <https://nda.ya.ru/t/7UdlI38n7SNwUL> - в нем я описал результаты
commit_hash:2626d7d6b77f1ccb31e395d974a2beaa60f27a97

6 files changed, 19 insertions, 6 deletions

diff --git a/yql/essentials/sql/v1/context.cpp b/yql/essentials/sql/v1/context.cpp
index 5ff1d3d8594..5e5359a598c 100644
--- a/yql/essentials/sql/v1/context.cpp
+++ b/yql/essentials/sql/v1/context.cpp
@@ -77,6 +77,7 @@ THashMap<TStringBuf, TPragmaField> CTX_PRAGMA_FIELDS = {
     {"OptimizeSimpleILIKE", &TContext::OptimizeSimpleIlike},
     {"DebugPositions", &TContext::DebugPositions},
     {"ExceptIntersectBefore202503", &TContext::ExceptIntersectBefore202503},
+    {"WindowNewPipeline", &TContext::WindowNewPipeline},
 };
 
 typedef TMaybe<bool> TContext::*TPragmaMaybeField;
diff --git a/yql/essentials/sql/v1/context.h b/yql/essentials/sql/v1/context.h
index 158d9e891eb..e360b369ac3 100644
--- a/yql/essentials/sql/v1/context.h
+++ b/yql/essentials/sql/v1/context.h
@@ -410,6 +410,7 @@ public:
     bool DisableLegacyNotNull = false;
     bool DebugPositions = false;
     bool StrictWarningAsError = false;
+    bool WindowNewPipeline = false;
     TMaybe<bool> DirectRowDependsOn;
     TVector<size_t> ForAllStatementsParts;
     TMaybe<TString> Engine;
diff --git a/yql/essentials/sql/v1/query.cpp b/yql/essentials/sql/v1/query.cpp
index 2896cac686c..d96d6b2a4bc 100644
--- a/yql/essentials/sql/v1/query.cpp
+++ b/yql/essentials/sql/v1/query.cpp
@@ -3496,6 +3496,11 @@ public:
                                             BuildQuotedAtom(Pos_, "DebugPositions"))));
                 }
 
+                if (ctx.WindowNewPipeline) {
+                    Add(Y("let", "world", Y(TString(ConfigureName), "world", configSource,
+                                            BuildQuotedAtom(Pos_, "WindowNewPipeline"))));
+                }
+
                 if (ctx.DirectRowDependsOn.Defined()) {
                     const TString pragmaName = *ctx.DirectRowDependsOn ? "DirectRowDependsOn" : "DisableDirectRowDependsOn";
                     currentWorlds->Add(Y("let", "world", Y(TString(ConfigureName), "world", configSource, BuildQuotedAtom(Pos_, pragmaName))));
diff --git a/yql/essentials/sql/v1/source.cpp b/yql/essentials/sql/v1/source.cpp
index cb3c5ea8dbf..5acc2069e35 100644
--- a/yql/essentials/sql/v1/source.cpp
+++ b/yql/essentials/sql/v1/source.cpp
@@ -745,7 +745,7 @@ TNodePtr BuildFrameNode(const TFrameBound& frame, EFrameType frameType) {
     return node;
 }
 
-TNodePtr ISource::BuildWindowFrame(const TFrameSpecification& spec, bool isCompact) {
+TNodePtr ISource::BuildWindowFrame(TContext& ctx, const TFrameSpecification& spec, bool isCompact, TNodePtr sortSpec) {
     YQL_ENSURE(spec.FrameExclusion == FrameExclNone);
     YQL_ENSURE(spec.FrameBegin);
     YQL_ENSURE(spec.FrameEnd);
@@ -755,8 +755,12 @@ TNodePtr ISource::BuildWindowFrame(const TFrameSpecification& spec, bool isCompa
 
     auto begin = Q(Y(Q("begin"), frameBeginNode));
     auto end = Q(Y(Q("end"), frameEndNode));
-
-    return isCompact ? Q(Y(begin, end, Q(Y(Q("compact"))))) : Q(Y(begin, end));
+    auto sortSpecNode = Q(Y(Q("sortSpec"), sortSpec));
+    if (ctx.WindowNewPipeline) {
+        return isCompact ? Q(Y(begin, end, Q(Y(Q("compact"))), sortSpecNode)) : Q(Y(begin, end, sortSpecNode));
+    } else {
+        return isCompact ? Q(Y(begin, end, Q(Y(Q("compact"))))) : Q(Y(begin, end));
+    }
 }
 
 class TSessionWindowTraits final: public TCallNode {
@@ -836,7 +840,9 @@ TNodePtr ISource::BuildCalcOverWindow(TContext& ctx, const TString& label) {
                 break;
         }
         YQL_ENSURE(frameType);
-        auto callOnFrame = Y(frameType, BuildWindowFrame(*spec->Frame, spec->IsCompact));
+        auto sortSpec = spec->OrderBy.empty() ? Y("Void") : BuildSortSpec(spec->OrderBy, useLabel, true, false);
+
+        auto callOnFrame = Y(frameType, BuildWindowFrame(ctx, *spec->Frame, spec->IsCompact, sortSpec));
         for (auto& agg : aggs) {
             auto winTraits = agg->WindowTraits(listType, ctx);
             callOnFrame = L(callOnFrame, winTraits);
@@ -854,7 +860,6 @@ TNodePtr ISource::BuildCalcOverWindow(TContext& ctx, const TString& label) {
             }
         }
 
-        auto sortSpec = spec->OrderBy.empty() ? Y("Void") : BuildSortSpec(spec->OrderBy, useLabel, true, false);
         if (spec->Session) {
             TString label = spec->Session->GetLabel();
             YQL_ENSURE(label);
diff --git a/yql/essentials/sql/v1/source.h b/yql/essentials/sql/v1/source.h
index 82386100e0e..36f62d86f70 100644
--- a/yql/essentials/sql/v1/source.h
+++ b/yql/essentials/sql/v1/source.h
@@ -126,7 +126,7 @@ protected:
     TVector<TNodePtr>& Expressions(EExprSeat exprSeat);
     TNodePtr AliasOrColumn(const TNodePtr& node, bool withSource);
 
-    TNodePtr BuildWindowFrame(const TFrameSpecification& spec, bool isCompact);
+    TNodePtr BuildWindowFrame(TContext& ctx, const TFrameSpecification& spec, bool isCompact, TNodePtr sortSpec);
 
     THashSet<TString> ExprAliases_;
     THashSet<TString> FlattenByAliases_;
diff --git a/yql/essentials/sql/v1/sql_query.cpp b/yql/essentials/sql/v1/sql_query.cpp
index 148a05dccd2..d3d236fdf46 100644
--- a/yql/essentials/sql/v1/sql_query.cpp
+++ b/yql/essentials/sql/v1/sql_query.cpp
@@ -3720,6 +3720,7 @@ THashMap<TString, TPragmaDescr> PragmaDescrs{
     TABLE_ELEM("UseTablePrefixForEach", PragmaUseTablePrefixForEach, true),
     PAIRED_TABLE_ELEM("SimpleColumns", SimpleColumns),
     PAIRED_TABLE_ELEM("DebugPositions", DebugPositions),
+    PAIRED_TABLE_ELEM("WindowNewPipeline", WindowNewPipeline),
     PAIRED_TABLE_ELEM("CoalesceJoinKeysOnQualifiedAll", CoalesceJoinKeysOnQualifiedAll),
     PAIRED_TABLE_ELEM("PullUpFlatMapOverJoin", PragmaPullUpFlatMapOverJoin),
     PAIRED_TABLE_ELEM("FilterPushdownOverJoinOptionalSide", FilterPushdownOverJoinOptionalSide),