Fix linking re templates

Author: alephonea

ydb/library/yql/providers/yt/provider/transformer/yql_yt_transformer_merge.cpp

@@ -482,9 +482,4 @@ TMaybeNode<TExprBase> TYtPhysicalOptProposalTransformer::MergeToCopy(TExprBase n
         .Done();
 }
 
-template <class TExpr>
-TMaybeNode<TExpr> TYtPhysicalOptProposalTransformer::CleanupWorld(TExpr node, TExprContext& ctx) const {
-    return TMaybeNode<TExpr>(YtCleanupWorld(node.Ptr(), ctx, State_));
-}
-
 }  // namespace NYql 

ydb/library/yql/providers/yt/provider/yql_yt_transformer.cpp

@@ -85,387 +85,4 @@ TYtPhysicalOptProposalTransformer::TYtPhysicalOptProposalTransformer(TYtState::T
 #undef HNDL
 }
 
-TMaybe<bool> TYtPhysicalOptProposalTransformer::CanFuseLambdas(const TCoLambda& innerLambda, const TCoLambda& outerLambda, TExprContext& ctx) const {
-    auto maxJobMemoryLimit = State_->Configuration->MaxExtraJobMemoryToFuseOperations.Get();
-    auto maxOperationFiles = State_->Configuration->MaxOperationFiles.Get().GetOrElse(DEFAULT_MAX_OPERATION_FILES);
-    TMap<TStringBuf, ui64> memUsage;
-
-    TExprNode::TPtr updatedBody = innerLambda.Body().Ptr();
-    if (maxJobMemoryLimit) {
-        auto status = UpdateTableContentMemoryUsage(innerLambda.Body().Ptr(), updatedBody, State_, ctx);
-        if (status.Level != TStatus::Ok) {
-            return {};
-        }
-    }
-    size_t innerFiles = 1; // jobstate. Take into account only once
-    ScanResourceUsage(*updatedBody, *State_->Configuration, State_->Types, maxJobMemoryLimit ? &memUsage : nullptr, nullptr, &innerFiles);
-
-    auto prevMemory = Accumulate(memUsage.begin(), memUsage.end(), 0ul,
-        [](ui64 sum, const std::pair<const TStringBuf, ui64>& val) { return sum + val.second; });
-
-    updatedBody = outerLambda.Body().Ptr();
-    if (maxJobMemoryLimit) {
-        auto status = UpdateTableContentMemoryUsage(outerLambda.Body().Ptr(), updatedBody, State_, ctx);
-        if (status.Level != TStatus::Ok) {
-            return {};
-        }
-    }
-    size_t outerFiles = 0;
-    ScanResourceUsage(*updatedBody, *State_->Configuration, State_->Types, maxJobMemoryLimit ? &memUsage : nullptr, nullptr, &outerFiles);
-
-    auto currMemory = Accumulate(memUsage.begin(), memUsage.end(), 0ul,
-        [](ui64 sum, const std::pair<const TStringBuf, ui64>& val) { return sum + val.second; });
-
-    if (maxJobMemoryLimit && currMemory != prevMemory && currMemory > *maxJobMemoryLimit) {
-        YQL_CLOG(DEBUG, ProviderYt) << "Memory usage: innerLambda=" << prevMemory
-            << ", joinedLambda=" << currMemory << ", MaxJobMemoryLimit=" << *maxJobMemoryLimit;
-        return false;
-    }
-    if (innerFiles + outerFiles > maxOperationFiles) {
-        YQL_CLOG(DEBUG, ProviderYt) << "Files usage: innerLambda=" << innerFiles
-            << ", outerLambda=" << outerFiles << ", MaxOperationFiles=" << maxOperationFiles;
-        return false;
-    }
-
-    if (auto maxReplcationFactor = State_->Configuration->MaxReplicationFactorToFuseOperations.Get()) {
-        double replicationFactor1 = NCommon::GetDataReplicationFactor(innerLambda.Ref(), ctx);
-        double replicationFactor2 = NCommon::GetDataReplicationFactor(outerLambda.Ref(), ctx);
-        YQL_CLOG(DEBUG, ProviderYt) << "Replication factors: innerLambda=" << replicationFactor1
-            << ", outerLambda=" << replicationFactor2 << ", MaxReplicationFactorToFuseOperations=" << *maxReplcationFactor;
-
-        if (replicationFactor1 > 1.0 && replicationFactor2 > 1.0 && replicationFactor1 * replicationFactor2 > *maxReplcationFactor) {
-            return false;
-        }
-    }
-    return true;
-}
-
-template<bool WithList>
-TCoLambda TYtPhysicalOptProposalTransformer::MakeJobLambda(TCoLambda lambda, bool useFlow, TExprContext& ctx) const {
-    if (useFlow) {
-        if constexpr (WithList) {
-            return Build<TCoLambda>(ctx, lambda.Pos())
-                .Args({"flow"})
-                .Body<TCoToFlow>()
-                    .Input<TExprApplier>()
-                        .Apply(lambda)
-                        .With<TCoForwardList>(0)
-                            .Stream("flow")
-                        .Build()
-                    .Build()
-                .Build()
-            .Done();
-        } else {
-            return Build<TCoLambda>(ctx, lambda.Pos())
-                .Args({"flow"})
-                .Body<TCoToFlow>()
-                    .Input<TExprApplier>()
-                        .Apply(lambda)
-                        .With<TCoFromFlow>(0)
-                            .Input("flow")
-                        .Build()
-                    .Build()
-                .Build()
-            .Done();
-        }
-    } else {
-        if constexpr (WithList) {
-            return Build<TCoLambda>(ctx, lambda.Pos())
-                .Args({"stream"})
-                .Body<TCoToStream>()
-                    .Input<TExprApplier>()
-                        .Apply(lambda)
-                        .With<TCoForwardList>(0)
-                            .Stream("stream")
-                        .Build()
-                    .Build()
-                .Build()
-            .Done();
-        } else {
-            return Build<TCoLambda>(ctx, lambda.Pos())
-                .Args({"stream"})
-                .Body<TCoToStream>()
-                    .Input<TExprApplier>()
-                        .Apply(lambda)
-                        .With(0, "stream")
-                    .Build()
-                .Build()
-            .Done();
-        }
-    }
-}
-
-template<bool IsTop>
-TMaybeNode<TExprBase> TYtPhysicalOptProposalTransformer::Sort(TExprBase node, TExprContext& ctx) const {
-    if (State_->Types->EvaluationInProgress || State_->PassiveExecution) {
-        return node;
-    }
-
-    const auto sort = node.Cast<std::conditional_t<IsTop, TCoTopBase, TCoSort>>();
-    if (!IsYtProviderInput(sort.Input())) {
-        return node;
-    }
-
-    auto sortDirections = sort.SortDirections();
-    if (!IsConstExpSortDirections(sortDirections)) {
-        return node;
-    }
-
-    auto keySelectorLambda = sort.KeySelectorLambda();
-    auto cluster = TString{GetClusterName(sort.Input())};
-    TSyncMap syncList;
-    if (!IsYtCompleteIsolatedLambda(keySelectorLambda.Ref(), syncList, cluster, true, false)) {
-        return node;
-    }
-
-    const TStructExprType* outType = nullptr;
-    if (auto type = GetSequenceItemType(node, false, ctx)) {
-        outType = type->Cast<TStructExprType>();
-    } else {
-        return {};
-    }
-
-    TVector<TYtPathInfo::TPtr> inputInfos = GetInputPaths(sort.Input());
-
-    TMaybe<NYT::TNode> firstNativeType;
-    if (!inputInfos.empty()) {
-        firstNativeType = inputInfos.front()->GetNativeYtType();
-    }
-    auto maybeReadSettings = sort.Input().template Maybe<TCoRight>().Input().template Maybe<TYtReadTable>().Input().Item(0).Settings();
-    const ui64 nativeTypeFlags = State_->Configuration->UseNativeYtTypes.Get().GetOrElse(DEFAULT_USE_NATIVE_YT_TYPES)
-         ? GetNativeYtTypeFlags(*outType)
-         : 0ul;
-    const bool needMap = (maybeReadSettings && NYql::HasSetting(maybeReadSettings.Ref(), EYtSettingType::SysColumns))
-        || AnyOf(inputInfos, [nativeTypeFlags, firstNativeType] (const TYtPathInfo::TPtr& path) {
-            return path->RequiresRemap()
-                || nativeTypeFlags != path->GetNativeYtTypeFlags()
-                || firstNativeType != path->GetNativeYtType();
-        });
-
-    bool useExplicitColumns = AnyOf(inputInfos, [] (const TYtPathInfo::TPtr& path) {
-        return !path->Table->IsTemp || (path->Table->RowSpec && path->Table->RowSpec->HasAuxColumns());
-    });
-
-    const bool needMerge = maybeReadSettings && NYql::HasSetting(maybeReadSettings.Ref(), EYtSettingType::Sample);
-
-    const bool useNativeDescSort = State_->Configuration->UseNativeDescSort.Get().GetOrElse(DEFAULT_USE_NATIVE_DESC_SORT);
-
-    TKeySelectorBuilder builder(node.Pos(), ctx, useNativeDescSort, outType);
-    builder.ProcessKeySelector(keySelectorLambda.Ptr(), sortDirections.Ptr());
-
-    TYtOutTableInfo sortOut(outType, nativeTypeFlags);
-    builder.FillRowSpecSort(*sortOut.RowSpec);
-    sortOut.SetUnique(sort.Ref().template GetConstraint<TDistinctConstraintNode>(), node.Pos(), ctx);
-
-    TExprBase sortInput = sort.Input();
-    TExprBase world = TExprBase(ApplySyncListToWorld(GetWorld(sortInput, {}, ctx).Ptr(), syncList, ctx));
-    bool unordered = ctx.IsConstraintEnabled<TSortedConstraintNode>();
-    if (needMap || builder.NeedMap()) {
-        auto mapper = builder.MakeRemapLambda();
-
-        auto mapperClean = CleanupWorld(TCoLambda(mapper), ctx);
-        if (!mapperClean) {
-            return {};
-        }
-
-        TYtOutTableInfo mapOut(builder.MakeRemapType(), nativeTypeFlags);
-        mapOut.SetUnique(sort.Ref().template GetConstraint<TDistinctConstraintNode>(), node.Pos(), ctx);
-
-        sortInput = Build<TYtOutput>(ctx, node.Pos())
-            .Operation<TYtMap>()
-                .World(world)
-                .DataSink(GetDataSink(sort.Input(), ctx))
-                .Input(ConvertInputTable(sort.Input(), ctx, TConvertInputOpts().MakeUnordered(unordered)))
-                .Output()
-                    .Add(mapOut.ToExprNode(ctx, node.Pos()).Cast<TYtOutTable>())
-                .Build()
-                .Settings(GetFlowSettings(node.Pos(), *State_, ctx))
-                .Mapper(mapperClean.Cast())
-            .Build()
-            .OutIndex()
-                .Value(0U)
-            .Build()
-            .Done();
-        world = TExprBase(ctx.NewWorld(node.Pos()));
-        unordered = false;
-    }
-    else if (needMerge) {
-        TYtOutTableInfo mergeOut(outType, nativeTypeFlags);
-        mergeOut.SetUnique(sort.Ref().template GetConstraint<TDistinctConstraintNode>(), node.Pos(), ctx);
-        if (firstNativeType) {
-            mergeOut.RowSpec->CopyTypeOrders(*firstNativeType);
-            sortOut.RowSpec->CopyTypeOrders(*firstNativeType);
-        }
-
-        TConvertInputOpts opts;
-        if (useExplicitColumns) {
-            opts.ExplicitFields(*mergeOut.RowSpec, node.Pos(), ctx);
-            useExplicitColumns = false;
-        }
-
-        sortInput = Build<TYtOutput>(ctx, node.Pos())
-            .Operation<TYtMerge>()
-                .World(world)
-                .DataSink(GetDataSink(sort.Input(), ctx))
-                .Input(ConvertInputTable(sort.Input(), ctx, opts.MakeUnordered(unordered)))
-                .Output()
-                    .Add(mergeOut.ToExprNode(ctx, node.Pos()).Cast<TYtOutTable>())
-                .Build()
-                .Settings()
-                    .Add()
-                        .Name()
-                            .Value(ToString(EYtSettingType::ForceTransform), TNodeFlags::Default)
-                        .Build()
-                    .Build()
-                .Build()
-            .Build()
-            .OutIndex()
-                .Value(0U)
-            .Build()
-            .Done();
-        world = TExprBase(ctx.NewWorld(node.Pos()));
-        unordered = false;
-    } else if (firstNativeType) {
-        sortOut.RowSpec->CopyTypeOrders(*firstNativeType);
-    }
-
-    bool canUseMerge = !needMap && !needMerge;
-    if (auto maxTablesForSortedMerge = State_->Configuration->MaxInputTablesForSortedMerge.Get()) {
-        if (inputInfos.size() > *maxTablesForSortedMerge) {
-            canUseMerge = false;
-        }
-    }
-
-    if (canUseMerge) {
-        TYqlRowSpecInfo commonSorted = *sortOut.RowSpec;
-        for (auto& pathInfo: inputInfos) {
-            commonSorted.MakeCommonSortness(*pathInfo->Table->RowSpec);
-        }
-        // input is sorted at least as strictly as output
-        if (!sortOut.RowSpec->CompareSortness(commonSorted)) {
-            canUseMerge = false;
-        }
-    }
-
-    sortOut.RowSpec->SetConstraints(sort.Ref().GetConstraintSet());
-
-    TConvertInputOpts opts;
-    if (useExplicitColumns) {
-        opts.ExplicitFields(*sortOut.RowSpec, node.Pos(), ctx);
-    }
-
-    auto res = canUseMerge ?
-        TExprBase(Build<TYtMerge>(ctx, node.Pos())
-            .World(world)
-            .DataSink(GetDataSink(sortInput, ctx))
-            .Input(ConvertInputTable(sortInput, ctx, opts.ClearUnordered()))
-            .Output()
-                .Add(sortOut.ToExprNode(ctx, node.Pos()).Cast<TYtOutTable>())
-            .Build()
-            .Settings()
-                .Add()
-                    .Name()
-                        .Value(ToString(EYtSettingType::KeepSorted), TNodeFlags::Default)
-                    .Build()
-                .Build()
-            .Build()
-        .Done()):
-        TExprBase(Build<TYtSort>(ctx, node.Pos())
-            .World(world)
-            .DataSink(GetDataSink(sortInput, ctx))
-            .Input(ConvertInputTable(sortInput, ctx, opts.MakeUnordered(unordered)))
-            .Output()
-                .Add(sortOut.ToExprNode(ctx, node.Pos()).Cast<TYtOutTable>())
-            .Build()
-            .Settings().Build()
-        .Done());
-
-    res = Build<TYtOutput>(ctx, node.Pos())
-        .Operation(res)
-        .OutIndex().Value(0U).Build()
-        .Done();
-
-
-    if constexpr (IsTop) {
-        res = Build<TCoTake>(ctx, node.Pos())
-            .Input(res)
-            .Count(sort.Count())
-            .Done();
-    }
-
-    return res;
-}
-
-template <typename TLMapType>
-TMaybeNode<TExprBase> TYtPhysicalOptProposalTransformer::LMap(TExprBase node, TExprContext& ctx) const {
-    if (State_->Types->EvaluationInProgress || State_->PassiveExecution) {
-        return node;
-    }
-
-    auto lmap = node.Cast<TLMapType>();
-
-    if (!IsYtProviderInput(lmap.Input(), true)) {
-        return node;
-    }
-
-    const auto inItemType = GetSequenceItemType(lmap.Input(), true, ctx);
-    if (!inItemType) {
-        return {};
-    }
-    const auto outItemType = SilentGetSequenceItemType(lmap.Lambda().Body().Ref(), true);
-    if (!outItemType || !outItemType->IsPersistable()) {
-        return node;
-    }
-
-    auto cluster = TString{GetClusterName(lmap.Input())};
-    TSyncMap syncList;
-    if (!IsYtCompleteIsolatedLambda(lmap.Lambda().Ref(), syncList, cluster, true, false)) {
-        return node;
-    }
-
-    auto cleanup = CleanupWorld(lmap.Lambda(), ctx);
-    if (!cleanup) {
-        return {};
-    }
-
-    auto mapper = cleanup.Cast().Ptr();
-    bool sortedOutput = false;
-    TVector<TYtOutTable> outTables = ConvertOutTablesWithSortAware(mapper, sortedOutput, lmap.Pos(),
-        outItemType, ctx, State_, lmap.Ref().GetConstraintSet());
-
-    const bool useFlow = State_->Configuration->UseFlow.Get().GetOrElse(DEFAULT_USE_FLOW);
-
-    auto settingsBuilder = Build<TCoNameValueTupleList>(ctx, lmap.Pos());
-    if (std::is_same<TLMapType, TCoOrderedLMap>::value) {
-        settingsBuilder
-            .Add()
-                .Name()
-                    .Value(ToString(EYtSettingType::Ordered))
-                .Build()
-            .Build();
-    }
-
-    if (useFlow) {
-        settingsBuilder
-            .Add()
-                .Name()
-                    .Value(ToString(EYtSettingType::Flow))
-                .Build()
-            .Build();
-    }
-
-    auto map = Build<TYtMap>(ctx, lmap.Pos())
-        .World(ApplySyncListToWorld(GetWorld(lmap.Input(), {}, ctx).Ptr(), syncList, ctx))
-        .DataSink(GetDataSink(lmap.Input(), ctx))
-        .Input(ConvertInputTable(lmap.Input(), ctx))
-        .Output()
-            .Add(outTables)
-        .Build()
-        .Settings(settingsBuilder.Done())
-        .Mapper(MakeJobLambda<false>(TCoLambda(mapper), useFlow, ctx))
-        .Done();
-
-    return WrapOp(map, ctx);
-}
-
 }  // namespace NYql