Skip to content

Commit

Permalink
[yt provider] Bypass YtCopy before YtLength (#11436)
Browse files Browse the repository at this point in the history
  • Loading branch information
rvu1024 authored Nov 10, 2024
1 parent a7178b2 commit a27ab2e
Showing 1 changed file with 26 additions and 24 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -182,7 +182,7 @@ class TYtPhysicalFinalizingTransformer : public TSyncTransformerBase {
}

if (!disableOptimizers.contains("BypassMergeBeforeLength")) {
status = BypassMergeBeforeLength(input, output, opDeps, lefts, ctx);
status = BypassMergeOrCopyBeforeLength(input, output, opDeps, lefts, ctx);
if (status.Level != TStatus::Ok) {
return status;
}
Expand Down Expand Up @@ -1610,33 +1610,35 @@ class TYtPhysicalFinalizingTransformer : public TSyncTransformerBase {
return lambda;
}

TStatus BypassMergeBeforeLength(TExprNode::TPtr input, TExprNode::TPtr& output, const TOpDeps& opDeps, const TNodeSet& lefts, TExprContext& ctx) {
TStatus BypassMergeOrCopyBeforeLength(TExprNode::TPtr input, TExprNode::TPtr& output, const TOpDeps& opDeps, const TNodeSet& lefts, TExprContext& ctx) {
TNodeOnNodeOwnedMap replaces;
for (auto& x: opDeps) {
if (TYtMerge::Match(x.first) && x.second.size() > 0 && AllOf(x.second, [](const auto& item) { return TYtLength::Match(std::get<0>(item)); } )) {
auto merge = TYtMerge(x.first);
if (merge.Ref().HasResult()) {
if ((TYtMerge::Match(x.first) || TYtCopy::Match(x.first)) && x.second.size() > 0 && AllOf(x.second, [](const auto& item) { return TYtLength::Match(std::get<0>(item)); } )) {
auto op = TYtTransientOpBase(x.first);
if (op.Ref().HasResult()) {
continue;
}

if (NYql::HasSetting(merge.Settings().Ref(), EYtSettingType::Limit)) {
continue;
}
auto section = op.Input().Item(0);
if (op.Maybe<TYtMerge>()) {
if (NYql::HasSetting(op.Settings().Ref(), EYtSettingType::Limit)) {
continue;
}

auto section = merge.Input().Item(0);
if (NYql::HasAnySetting(section.Settings().Ref(), EYtSettingType::Take | EYtSettingType::Skip | EYtSettingType::Sample)) {
continue;
}
if (NYql::HasNonEmptyKeyFilter(section)) {
continue;
}
if (NYql::HasAnySetting(section.Settings().Ref(), EYtSettingType::Take | EYtSettingType::Skip | EYtSettingType::Sample)) {
continue;
}
if (NYql::HasNonEmptyKeyFilter(section)) {
continue;
}

if (AnyOf(section.Paths(), [](const TYtPath& path) { return !path.Ranges().Maybe<TCoVoid>() || TYtTableBaseInfo::GetMeta(path.Table())->IsDynamic; })) {
continue;
}
// Dependency on more than 1 operation
if (1 < Accumulate(section.Paths(), 0ull, [](ui64 val, const TYtPath& path) { return val + path.Table().Maybe<TYtOutput>().IsValid(); })) {
continue;
if (AnyOf(section.Paths(), [](const TYtPath& path) { return !path.Ranges().Maybe<TCoVoid>() || TYtTableBaseInfo::GetMeta(path.Table())->IsDynamic; })) {
continue;
}
// Dependency on more than 1 operation
if (1 < Accumulate(section.Paths(), 0ull, [](ui64 val, const TYtPath& path) { return val + path.Table().Maybe<TYtOutput>().IsValid(); })) {
continue;
}
}

TSyncMap syncList;
Expand All @@ -1645,10 +1647,10 @@ class TYtPhysicalFinalizingTransformer : public TSyncTransformerBase {
syncList.emplace(GetOutputOp(out.Cast()).Ptr(), syncList.size());
}
}
auto newWorld = ApplySyncListToWorld(merge.World().Ptr(), syncList, ctx);
auto newWorld = ApplySyncListToWorld(op.World().Ptr(), syncList, ctx);
for (auto node: lefts) {
TCoLeft left(node);
if (left.Input().Raw() == merge.Raw()) {
if (left.Input().Raw() == op.Raw()) {
replaces[node] = newWorld;
}
}
Expand Down Expand Up @@ -1695,7 +1697,7 @@ class TYtPhysicalFinalizingTransformer : public TSyncTransformerBase {
}

if (!replaces.empty()) {
YQL_CLOG(INFO, ProviderYt) << "PhysicalFinalizing-BypassMergeBeforeLength";
YQL_CLOG(INFO, ProviderYt) << "PhysicalFinalizing-BypassMergeOrCopyBeforeLength";
return RemapExpr(input, output, replaces, ctx, TOptimizeExprSettings(State_->Types));
}

Expand Down

0 comments on commit a27ab2e

Please sign in to comment.