HLL in YT statistics (#8184)

ydb/library/yql/providers/yt/common/yql_yt_settings.cpp

@@ -484,6 +484,7 @@ TYtConfiguration::TYtConfiguration()
         });
     REGISTER_SETTING(*this, MinColumnGroupSize).Lower(2);
     REGISTER_SETTING(*this, MaxColumnGroups);
+    REGISTER_SETTING(*this, ExtendedStatsMaxChunkCount);
 }
 
 EReleaseTempDataMode GetReleaseTempDataMode(const TYtSettings& settings) {

ydb/library/yql/providers/yt/common/yql_yt_settings.h

@@ -280,6 +280,7 @@ struct TYtSettings {
     NCommon::TConfSetting<EColumnGroupMode, false> ColumnGroupMode;
     NCommon::TConfSetting<ui16, false> MinColumnGroupSize;
     NCommon::TConfSetting<ui16, false> MaxColumnGroups;
+    NCommon::TConfSetting<ui64, false> ExtendedStatsMaxChunkCount;
 };
 
 EReleaseTempDataMode GetReleaseTempDataMode(const TYtSettings& settings);

ydb/library/yql/providers/yt/gateway/lib/transaction_cache.cpp

@@ -101,9 +101,9 @@ TMaybe<ui64> TTransactionCache::TEntry::GetColumnarStat(NYT::TRichYPath ytPath)
 
     auto guard = Guard(Lock_);
     if (auto p = StatisticsCache.FindPtr(NYT::NodeToCanonicalYsonString(NYT::PathToNode(ytPath), NYT::NYson::EYsonFormat::Text))) {
-        ui64 sum = p->LegacyChunksDataWeight;
+        ui64 sum = p->ColumnarStat.LegacyChunksDataWeight;
         for (auto& column: columns) {
-            if (auto c = p->ColumnDataWeight.FindPtr(column)) {
+            if (auto c = p->ColumnarStat.ColumnDataWeight.FindPtr(column)) {
                 sum += *c;
             } else {
                 return Nothing();
@@ -114,30 +114,66 @@ TMaybe<ui64> TTransactionCache::TEntry::GetColumnarStat(NYT::TRichYPath ytPath)
     return Nothing();
 }
 
+TMaybe<NYT::TTableColumnarStatistics> TTransactionCache::TEntry::GetExtendedColumnarStat(NYT::TRichYPath ytPath) const {
+    TVector<TString> columns(std::move(ytPath.Columns_->Parts_));
+    ytPath.Columns_.Clear();
+    auto cacheKey = NYT::NodeToCanonicalYsonString(NYT::PathToNode(ytPath), NYT::NYson::EYsonFormat::Text);
+
+    auto guard = Guard(Lock_);
+    auto p = StatisticsCache.FindPtr(cacheKey);
+    if (!p) {
+        return Nothing();
+    }
+
+    NYT::TTableColumnarStatistics res;
+    for (auto& column: columns) {
+        if (p->ExtendedStatColumns.count(column) == 0) {
+            return Nothing();
+        }
+        if (auto c = p->ColumnarStat.ColumnDataWeight.FindPtr(column)) {
+            res.ColumnDataWeight[column] = *c;
+        }
+        if (auto c = p->ColumnarStat.ColumnEstimatedUniqueCounts.FindPtr(column)) {
+            res.ColumnEstimatedUniqueCounts[column] = *c;
+        }
+    }
+    return res;
+}
+
 void TTransactionCache::TEntry::UpdateColumnarStat(NYT::TRichYPath ytPath, ui64 size) {
     YQL_ENSURE(ytPath.Columns_.Defined());
     TVector<TString> columns(std::move(ytPath.Columns_->Parts_));
     ytPath.Columns_.Clear();
+    auto cacheKey = NYT::NodeToCanonicalYsonString(NYT::PathToNode(ytPath), NYT::NYson::EYsonFormat::Text);
 
     auto guard = Guard(Lock_);
-    NYT::TTableColumnarStatistics& cacheColumnStat = StatisticsCache[NYT::NodeToCanonicalYsonString(NYT::PathToNode(ytPath), NYT::NYson::EYsonFormat::Text)];
-    cacheColumnStat.LegacyChunksDataWeight = size;
-    for (auto& c: cacheColumnStat.ColumnDataWeight) {
+    auto& cacheEntry = StatisticsCache[cacheKey];
+    cacheEntry.ColumnarStat.LegacyChunksDataWeight = size;
+    for (auto& c: cacheEntry.ColumnarStat.ColumnDataWeight) {
         c.second = 0;
     }
     for (auto& c: columns) {
-        cacheColumnStat.ColumnDataWeight[c] = 0;
+        cacheEntry.ColumnarStat.ColumnDataWeight[c] = 0;
     }
 }
 
-void TTransactionCache::TEntry::UpdateColumnarStat(NYT::TRichYPath ytPath, const NYT::TTableColumnarStatistics& columnStat) {
+void TTransactionCache::TEntry::UpdateColumnarStat(NYT::TRichYPath ytPath, const NYT::TTableColumnarStatistics& columnStat, bool extended) {
+    TVector<TString> columns(std::move(ytPath.Columns_->Parts_));
     ytPath.Columns_.Clear();
     auto guard = Guard(Lock_);
-    NYT::TTableColumnarStatistics& cacheColumnStat = StatisticsCache[NYT::NodeToCanonicalYsonString(NYT::PathToNode(ytPath), NYT::NYson::EYsonFormat::Text)];
-    cacheColumnStat.LegacyChunksDataWeight = columnStat.LegacyChunksDataWeight;
-    cacheColumnStat.TimestampTotalWeight = columnStat.TimestampTotalWeight;
+    auto& cacheEntry = StatisticsCache[NYT::NodeToCanonicalYsonString(NYT::PathToNode(ytPath), NYT::NYson::EYsonFormat::Text)];
+    if (extended) {
+        std::copy(columns.begin(), columns.end(), std::inserter(cacheEntry.ExtendedStatColumns, cacheEntry.ExtendedStatColumns.end()));
+    }
+    cacheEntry.ColumnarStat.LegacyChunksDataWeight = columnStat.LegacyChunksDataWeight;
+    cacheEntry.ColumnarStat.TimestampTotalWeight = columnStat.TimestampTotalWeight;
     for (auto& c: columnStat.ColumnDataWeight) {
-        cacheColumnStat.ColumnDataWeight[c.first] = c.second;
+        cacheEntry.ColumnarStat.ColumnDataWeight[c.first] = c.second;
+    }
+    if (extended) {
+        for (auto& c : columnStat.ColumnEstimatedUniqueCounts) {
+            cacheEntry.ColumnarStat.ColumnEstimatedUniqueCounts[c.first] = c.second;
+        }
     }
 }
 

ydb/library/yql/providers/yt/gateway/lib/transaction_cache.h

@@ -47,7 +47,6 @@ public:
         bool KeepTables = false;
         THashMap<std::pair<TString, ui32>, std::tuple<TString, NYT::TTransactionId, ui64>> Snapshots; // {tablepath, epoch} -> {table_id, transaction_id, revision}
         NYT::TNode TransactionSpec;
-        THashMap<TString, NYT::TTableColumnarStatistics> StatisticsCache;
         THashMap<TString, TString> BinarySnapshots; // remote path -> snapshot path
         NYT::ITransactionPtr BinarySnapshotTx;
         THashMap<TString, NYT::ITransactionPtr> CheckpointTxs;
@@ -114,8 +113,10 @@ public:
         void CompleteWriteTx(const NYT::TTransactionId& id, bool abort);
 
         TMaybe<ui64> GetColumnarStat(NYT::TRichYPath ytPath) const;
+        TMaybe<NYT::TTableColumnarStatistics> GetExtendedColumnarStat(NYT::TRichYPath ytPath) const;
+
         void UpdateColumnarStat(NYT::TRichYPath ytPath, ui64 size);
-        void UpdateColumnarStat(NYT::TRichYPath ytPath, const NYT::TTableColumnarStatistics& columnStat);
+        void UpdateColumnarStat(NYT::TRichYPath ytPath, const NYT::TTableColumnarStatistics& columnStat, bool extended = false);
 
         std::pair<TString, NYT::TTransactionId> GetBinarySnapshot(TString remoteTmpFolder, const TString& md5, const TString& localPath, TDuration expirationInterval);
 
@@ -124,6 +125,13 @@ public:
         using TPtr = TIntrusivePtr<TEntry>;
 
     private:
+        struct TStatisticsCacheEntry {
+            std::unordered_set<TString> ExtendedStatColumns;
+            NYT::TTableColumnarStatistics ColumnarStat;
+        };
+
+        THashMap<TString, TStatisticsCacheEntry> StatisticsCache;
+
         void DeleteAtFinalizeUnlocked(const TString& table, bool isInternal);
         bool CancelDeleteAtFinalizeUnlocked(const TString& table, bool isInternal);
         void DoRemove(const TString& table);

ydb/library/yql/providers/yt/gateway/native/yql_yt_native.cpp

@@ -195,6 +195,27 @@ inline TType OptionFromNode(const NYT::TNode& value) {
     }
 }
 
+void PopulatePathStatResult(IYtGateway::TPathStatResult& out, int index, NYT::TTableColumnarStatistics& extendedStat) {
+    for (const auto& entry : extendedStat.ColumnDataWeight) {
+        out.DataSize[index] += entry.second;
+    }
+    out.Extended[index] = IYtGateway::TPathStatResult::TExtendedResult{
+        .DataWeight = extendedStat.ColumnDataWeight,
+        .EstimatedUniqueCounts = extendedStat.ColumnEstimatedUniqueCounts
+    };
+}
+
+TString DebugPath(NYT::TRichYPath path) {
+    constexpr int maxDebugColumns = 20;
+    if (!path.Columns_ || std::ssize(path.Columns_->Parts_) <= maxDebugColumns) {
+        return NYT::NodeToCanonicalYsonString(NYT::PathToNode(path), NYT::NYson::EYsonFormat::Text);
+    }
+    int numColumns = std::ssize(path.Columns_->Parts_);
+    path.Columns_->Parts_.erase(path.Columns_->Parts_.begin() + maxDebugColumns, path.Columns_->Parts_.end());
+    path.Columns_->Parts_.push_back("...");
+    return NYT::NodeToCanonicalYsonString(NYT::PathToNode(path), NYT::NYson::EYsonFormat::Text) + " (" + std::to_string(numColumns) + " columns)";
+}
+
 } // unnamed
 
 ///////////////////////////////////////////////////////////////////////////////////////////////////////
@@ -4495,6 +4516,7 @@ private:
         try {
             TPathStatResult res;
             res.DataSize.resize(execCtx->Options_.Paths().size(), 0);
+            res.Extended.resize(execCtx->Options_.Paths().size());
 
             auto entry = execCtx->GetOrCreateEntry();
             auto tx = entry->Tx;
@@ -4502,6 +4524,7 @@ private:
             const NYT::EOptimizeForAttr tmpOptimizeFor = execCtx->Options_.Config()->OptimizeFor.Get(execCtx->Cluster_).GetOrElse(NYT::EOptimizeForAttr::OF_LOOKUP_ATTR);
             TVector<NYT::TRichYPath> ytPaths(Reserve(execCtx->Options_.Paths().size()));
             TVector<size_t> pathMap;
+            bool extended = execCtx->Options_.Extended();
 
             auto extractSysColumns = [] (NYT::TRichYPath& ytPath) -> TVector<TString> {
                 TVector<TString> res;
@@ -4545,16 +4568,19 @@ private:
                             YQL_CLOG(INFO, ProviderYt) << "Adding stat for " << col << ": " << size << " (virtual)";
                         }
                     }
-                    if (auto val = entry->GetColumnarStat(ytPath)) {
-                        res.DataSize[i] += *val;
-                        YQL_CLOG(INFO, ProviderYt) << "Stat for " << req.Path().Path_ << ": " << res.DataSize[i] << " (from cache)";
+                    TMaybe<ui64> cachedStat;
+                    TMaybe<NYT::TTableColumnarStatistics> cachedExtendedStat;
+                    if (!extended && (cachedStat = entry->GetColumnarStat(ytPath))) {
+                        res.DataSize[i] += *cachedStat;
+                        YQL_CLOG(INFO, ProviderYt) << "Stat for " << DebugPath(req.Path()) << ": " << res.DataSize[i] << " (from cache, extended: false)";
+                    } else if (extended && (cachedExtendedStat = entry->GetExtendedColumnarStat(ytPath))) {
+                        PopulatePathStatResult(res, i, *cachedExtendedStat);
+                        YQL_CLOG(INFO, ProviderYt) << "Stat for " << DebugPath(req.Path()) << " (from cache, extended: true)";
                     } else if (onlyCached) {
-                        YQL_CLOG(INFO, ProviderYt) << "Stat for " << req.Path().Path_ << " is missing in cache - sync path stat failed";
+                        YQL_CLOG(INFO, ProviderYt) << "Stat for " << DebugPath(req.Path()) << " is missing in cache - sync path stat failed (extended: " << extended << ")";
                         return res;
-                    } else if (NYT::EOptimizeForAttr::OF_SCAN_ATTR == tmpOptimizeFor) {
-                        pathMap.push_back(i);
-                        ytPaths.push_back(ytPath);
-                    } else {
+                    } else if (NYT::EOptimizeForAttr::OF_SCAN_ATTR != tmpOptimizeFor && !extended) {
+
                         // Use entire table size for lookup tables (YQL-7257)
                         if (attrs.IsUndefined()) {
                             attrs = tx->Get(ytPath.Path_ + "/@", NYT::TGetOptions().AttributeFilter(
@@ -4566,7 +4592,10 @@ private:
                         auto size = CalcDataSize(ytPath, attrs);
                         res.DataSize[i] += size;
                         entry->UpdateColumnarStat(ytPath, size);
-                        YQL_CLOG(INFO, ProviderYt) << "Stat for " << req.Path().Path_ << ": " << res.DataSize[i] << " (uncompressed_data_size for lookup)";
+                        YQL_CLOG(INFO, ProviderYt) << "Stat for " << DebugPath(req.Path()) << ": " << res.DataSize[i] << " (uncompressed_data_size for lookup, extended: false)";
+                    } else {
+                        ytPaths.push_back(ytPath);
+                        pathMap.push_back(i);
                     }
                 } else {
                     auto p = entry->Snapshots.FindPtr(std::make_pair(tablePath, req.Epoch()));
@@ -4597,11 +4626,19 @@ private:
                             YQL_CLOG(INFO, ProviderYt) << "Adding stat for " << col << ": " << size << " (virtual)";
                         }
                     }
-                    if (auto val = entry->GetColumnarStat(ytPath)) {
-                        res.DataSize[i] += *val;
-                        YQL_CLOG(INFO, ProviderYt) << "Stat for " << req.Path().Path_ << " (epoch=" << req.Epoch() << "): " << res.DataSize[i] << " (from cache)";
+                    TMaybe<ui64> cachedStat;
+                    TMaybe<NYT::TTableColumnarStatistics> cachedExtendedStat;
+                    if (!extended && (cachedStat = entry->GetColumnarStat(ytPath))) {
+                        res.DataSize[i] += *cachedStat;
+                        YQL_CLOG(INFO, ProviderYt) << "Stat for " << DebugPath(req.Path()) << " (epoch=" << req.Epoch() << "): " << res.DataSize[i] << " (from cache, extended: false)";
+                    } else if (extended && (cachedExtendedStat = entry->GetExtendedColumnarStat(ytPath))) {
+                        PopulatePathStatResult(res, i, *cachedExtendedStat);
+                        YQL_CLOG(INFO, ProviderYt) << "Stat for " << DebugPath(req.Path()) << " (from cache, extended: true)";
                     } else if (onlyCached) {
-                        YQL_CLOG(INFO, ProviderYt) << "Stat for " << req.Path().Path_ << " (epoch=" << req.Epoch() << ") is missing in cache - sync path stat failed";
+                        YQL_CLOG(INFO, ProviderYt)
+                            << "Stat for " << DebugPath(req.Path())
+                            << " (epoch=" << req.Epoch() << ", extended: " << extended
+                            << ") is missing in cache - sync path stat failed";
                         return res;
                     } else {
                         if (attrs.IsUndefined()) {
@@ -4613,18 +4650,19 @@ private:
                                     .AddAttribute(TString("schema"))
                             ));
                         }
-                        if (attrs.HasKey("optimize_for") && attrs["optimize_for"] == "scan" &&
-                            AllPathColumnsAreInSchema(req.Path(), attrs))
+                        if (extended ||
+                            (attrs.HasKey("optimize_for") && attrs["optimize_for"] == "scan" &&
+                            AllPathColumnsAreInSchema(req.Path(), attrs)))
                         {
                             pathMap.push_back(i);
                             ytPaths.push_back(ytPath);
-                            YQL_CLOG(INFO, ProviderYt) << "Stat for " << req.Path().Path_ << " (epoch=" << req.Epoch() << ") add for request with path " << ytPath.Path_;
+                            YQL_CLOG(INFO, ProviderYt) << "Stat for " << DebugPath(req.Path()) << " (epoch=" << req.Epoch() << ") add for request with path " << ytPath.Path_ << " (extended: " << extended << ")";
                         } else {
                             // Use entire table size for lookup tables (YQL-7257)
                             auto size = CalcDataSize(ytPath, attrs);
                             res.DataSize[i] += size;
                             entry->UpdateColumnarStat(ytPath, size);
-                            YQL_CLOG(INFO, ProviderYt) << "Stat for " << req.Path().Path_ << " (epoch=" << req.Epoch() << "): " << res.DataSize[i] << " (uncompressed_data_size for lookup)";
+                            YQL_CLOG(INFO, ProviderYt) << "Stat for " << DebugPath(req.Path()) << " (epoch=" << req.Epoch() << "): " << res.DataSize[i] << " (uncompressed_data_size for lookup)";
                         }
                     }
                 }
@@ -4632,17 +4670,23 @@ private:
 
             if (ytPaths) {
                 YQL_ENSURE(!onlyCached);
-                auto fetchMode = execCtx->Options_.Config()->JoinColumnarStatisticsFetcherMode.Get().GetOrElse(NYT::EColumnarStatisticsFetcherMode::Fallback);
+                auto fetchMode = extended ?
+                    NYT::EColumnarStatisticsFetcherMode::FromNodes :
+                    execCtx->Options_.Config()->JoinColumnarStatisticsFetcherMode.Get().GetOrElse(NYT::EColumnarStatisticsFetcherMode::Fallback);
                 auto columnStats = tx->GetTableColumnarStatistics(ytPaths, NYT::TGetTableColumnarStatisticsOptions().FetcherMode(fetchMode));
                 YQL_ENSURE(pathMap.size() == columnStats.size());
-                    for (size_t i: xrange(columnStats.size())) {
+                for (size_t i: xrange(columnStats.size())) {
                     auto& columnStat = columnStats[i];
                     const ui64 weight = columnStat.LegacyChunksDataWeight +
                         Accumulate(columnStat.ColumnDataWeight.begin(), columnStat.ColumnDataWeight.end(), 0ull,
                             [](ui64 sum, decltype(*columnStat.ColumnDataWeight.begin())& v) { return sum + v.second; });
 
+                    if (extended) {
+                        PopulatePathStatResult(res, pathMap[i], columnStat);
+                    }
+
                     res.DataSize[pathMap[i]] += weight;
-                    entry->UpdateColumnarStat(ytPaths[i], columnStat);
+                    entry->UpdateColumnarStat(ytPaths[i], columnStat, extended);
                     YQL_CLOG(INFO, ProviderYt) << "Stat for " << execCtx->Options_.Paths()[pathMap[i]].Path().Path_ << ": " << weight << " (fetched)";
                 }
             }

ydb/library/yql/providers/yt/provider/phy_opt/yql_yt_phy_opt_join.cpp

@@ -7,6 +7,7 @@
 
 #include <ydb/library/yql/core/yql_type_helpers.h>
 #include <ydb/library/yql/core/yql_opt_utils.h>
+#include <ydb/library/yql/providers/common/provider/yql_provider.h>
 
 #include <ydb/library/yql/utils/log/log.h>
 
@@ -321,6 +322,7 @@ TMaybeNode<TExprBase> TYtPhysicalOptProposalTransformer::EarlyMergeJoin(TExprBas
 
 TMaybeNode<TExprBase> TYtPhysicalOptProposalTransformer::RuntimeEquiJoin(TExprBase node, TExprContext& ctx) const {
     auto equiJoin = node.Cast<TYtEquiJoin>();
+    auto cluster = equiJoin.DataSink().Cluster().StringValue();
 
     const bool tryReorder = State_->Types->CostBasedOptimizer != ECostBasedOptimizerType::Disable
         && equiJoin.Input().Size() > 2
@@ -338,10 +340,19 @@ TMaybeNode<TExprBase> TYtPhysicalOptProposalTransformer::RuntimeEquiJoin(TExprBa
             }
         }
     }
-
     const auto tree = ImportYtEquiJoin(equiJoin, ctx);
+
+    const TMaybe<ui64> maxChunkCountExtendedStats = State_->Configuration->ExtendedStatsMaxChunkCount.Get();
+
+    if (tryReorder && waitAllInputs && maxChunkCountExtendedStats) {
+        YQL_CLOG(INFO, ProviderYt) << "Collecting cbo stats for equiJoin";
+        auto collectStatus = CollectCboStats(cluster, *tree, State_, ctx);
+        if (collectStatus == TStatus::Repeat) {
+            return ExportYtEquiJoin(equiJoin, *tree, ctx, State_);
+        }
+    }
     if (tryReorder) {
-        const auto optimizedTree = OrderJoins(tree, State_, ctx);
+        const auto optimizedTree = OrderJoins(tree, State_, cluster, ctx);
         if (optimizedTree != tree) {
             return ExportYtEquiJoin(equiJoin, *optimizedTree, ctx, State_);
         }

ydb/library/yql/providers/yt/provider/ut/yql_yt_cbo_ut.cpp

@@ -60,6 +60,7 @@ TYtJoinNodeLeaf::TPtr MakeLeaf(const std::vector<TString>& label, TVector<TStrin
 Y_UNIT_TEST_SUITE(TYqlCBO) {
 
 Y_UNIT_TEST(OrderJoinsDoesNothingWhenCBODisabled) {
+    const TString cluster("ut_cluster");
     TYtState::TPtr state = MakeIntrusive<TYtState>();
     TYtJoinNodeOp::TPtr tree = nullptr;
     TYtJoinNodeOp::TPtr optimizedTree;
@@ -68,7 +69,7 @@ Y_UNIT_TEST(OrderJoinsDoesNothingWhenCBODisabled) {
 
     TExprContext ctx;
 
-    optimizedTree = OrderJoins(tree, state, ctx);
+    optimizedTree = OrderJoins(tree, state, cluster, ctx);
     UNIT_ASSERT_VALUES_EQUAL(tree, optimizedTree);
 }
 
@@ -98,6 +99,8 @@ Y_UNIT_TEST(NonReordable) {
 }
 
 Y_UNIT_TEST(BuildOptimizerTree2Tables) {
+    const TString cluster("ut_cluster");
+    TYtState::TPtr state = MakeIntrusive<TYtState>();
     TExprContext exprCtx;
     auto tree = MakeOp({"c", "c_nationkey"}, {"n", "n_nationkey"}, {"c", "n"}, exprCtx);
     tree->Left = MakeLeaf({"c"}, {"c"},  100000, 12333, exprCtx);
@@ -105,7 +108,7 @@ Y_UNIT_TEST(BuildOptimizerTree2Tables) {
 
     std::shared_ptr<IBaseOptimizerNode> resultTree;
     std::shared_ptr<IProviderContext> resultCtx;
-    BuildOptimizerJoinTree(resultTree, resultCtx, tree);
+    BuildOptimizerJoinTree(state, cluster, resultTree, resultCtx, tree, exprCtx);
 
     UNIT_ASSERT(resultTree->Kind == JoinNodeType);
     auto root = std::static_pointer_cast<TJoinOptimizerNode>(resultTree);
@@ -122,6 +125,8 @@ Y_UNIT_TEST(BuildOptimizerTree2Tables) {
 }
 
 Y_UNIT_TEST(BuildOptimizerTree2TablesComplexLabel) {
+    const TString cluster("ut_cluster");
+    TYtState::TPtr state = MakeIntrusive<TYtState>();
     TExprContext exprCtx;
     auto tree = MakeOp({"c", "c_nationkey"}, {"n", "n_nationkey"}, {"c", "n", "e"}, exprCtx);
     tree->Left = MakeLeaf({"c"}, {"c"}, 1000000, 1233333, exprCtx);
@@ -129,7 +134,7 @@ Y_UNIT_TEST(BuildOptimizerTree2TablesComplexLabel) {
 
     std::shared_ptr<IBaseOptimizerNode> resultTree;
     std::shared_ptr<IProviderContext> resultCtx;
-    BuildOptimizerJoinTree(resultTree, resultCtx, tree);
+    BuildOptimizerJoinTree(state, cluster, resultTree, resultCtx, tree, exprCtx);
 
     UNIT_ASSERT(resultTree->Kind == JoinNodeType);
     auto root = std::static_pointer_cast<TJoinOptimizerNode>(resultTree);
@@ -146,6 +151,8 @@ Y_UNIT_TEST(BuildOptimizerTree2TablesComplexLabel) {
 }
 
 Y_UNIT_TEST(BuildYtJoinTree2Tables) {
+    const TString cluster("ut_cluster");
+    TYtState::TPtr state = MakeIntrusive<TYtState>();
     TExprContext exprCtx;
     auto tree = MakeOp({"c", "c_nationkey"}, {"n", "n_nationkey"}, {"c", "n"}, exprCtx);
     tree->Left = MakeLeaf({"c"}, {"c"},  100000, 12333, exprCtx);
@@ -153,7 +160,7 @@ Y_UNIT_TEST(BuildYtJoinTree2Tables) {
 
     std::shared_ptr<IBaseOptimizerNode> resultTree;
     std::shared_ptr<IProviderContext> resultCtx;
-    BuildOptimizerJoinTree(resultTree, resultCtx, tree);
+    BuildOptimizerJoinTree(state, cluster, resultTree, resultCtx, tree, exprCtx);
 
     auto joinTree = BuildYtJoinTree(resultTree, exprCtx, {});
 
@@ -161,6 +168,8 @@ Y_UNIT_TEST(BuildYtJoinTree2Tables) {
 }
 
 Y_UNIT_TEST(BuildYtJoinTree2TablesForceMergeJoib) {
+    const TString cluster("ut_cluster");
+    TYtState::TPtr state = MakeIntrusive<TYtState>();
     TExprContext exprCtx;
     auto tree = MakeOp({"c", "c_nationkey"}, {"n", "n_nationkey"}, {"c", "n"}, exprCtx);
     tree->Left = MakeLeaf({"c"}, {"c"},  100000, 12333, exprCtx);
@@ -169,7 +178,7 @@ Y_UNIT_TEST(BuildYtJoinTree2TablesForceMergeJoib) {
 
     std::shared_ptr<IBaseOptimizerNode> resultTree;
     std::shared_ptr<IProviderContext> resultCtx;
-    BuildOptimizerJoinTree(resultTree, resultCtx, tree);
+    BuildOptimizerJoinTree(state, cluster, resultTree, resultCtx, tree, exprCtx);
 
     auto joinTree = BuildYtJoinTree(resultTree, exprCtx, {});
 
@@ -177,6 +186,8 @@ Y_UNIT_TEST(BuildYtJoinTree2TablesForceMergeJoib) {
 }
 
 Y_UNIT_TEST(BuildYtJoinTree2TablesComplexLabel) {
+    const TString cluster("ut_cluster");
+    TYtState::TPtr state = MakeIntrusive<TYtState>();
     TExprContext exprCtx;
     auto tree = MakeOp({"c", "c_nationkey"}, {"n", "n_nationkey"}, {"c", "n", "e"}, exprCtx);
     tree->Left = MakeLeaf({"c"}, {"c"}, 1000000, 1233333, exprCtx);
@@ -184,7 +195,7 @@ Y_UNIT_TEST(BuildYtJoinTree2TablesComplexLabel) {
 
     std::shared_ptr<IBaseOptimizerNode> resultTree;
     std::shared_ptr<IProviderContext> resultCtx;
-    BuildOptimizerJoinTree(resultTree, resultCtx, tree);
+    BuildOptimizerJoinTree(state, cluster, resultTree, resultCtx, tree, exprCtx);
     auto joinTree = BuildYtJoinTree(resultTree, exprCtx, {});
 
     UNIT_ASSERT(AreSimilarTrees(joinTree, tree));
@@ -192,6 +203,8 @@ Y_UNIT_TEST(BuildYtJoinTree2TablesComplexLabel) {
 
 Y_UNIT_TEST(BuildYtJoinTree2TablesTableIn2Rels)
 {
+    const TString cluster("ut_cluster");
+    TYtState::TPtr state = MakeIntrusive<TYtState>();
     TExprContext exprCtx;
     auto tree = MakeOp({"c", "c_nationkey"}, {"n", "n_nationkey"}, {"c", "n", "c"}, exprCtx);
     tree->Left = MakeLeaf({"c"}, {"c"}, 1000000, 1233333, exprCtx);
@@ -199,7 +212,7 @@ Y_UNIT_TEST(BuildYtJoinTree2TablesTableIn2Rels)
 
     std::shared_ptr<IBaseOptimizerNode> resultTree;
     std::shared_ptr<IProviderContext> resultCtx;
-    BuildOptimizerJoinTree(resultTree, resultCtx, tree);
+    BuildOptimizerJoinTree(state, cluster, resultTree, resultCtx, tree, exprCtx);
     auto joinTree = BuildYtJoinTree(resultTree, exprCtx, {});
 
     UNIT_ASSERT(AreSimilarTrees(joinTree, tree));
@@ -214,6 +227,7 @@ Y_UNIT_TEST(BuildYtJoinTree2TablesTableIn2Rels)
     }
 
 void OrderJoins2Tables(auto optimizerType) {
+    const TString cluster("ut_cluster");
     TExprContext exprCtx;
     auto tree = MakeOp({"c", "c_nationkey"}, {"n", "n_nationkey"}, {"c", "n"}, exprCtx);
     tree->Left = MakeLeaf({"c"}, {"c"},  100000, 12333, exprCtx);
@@ -223,7 +237,7 @@ void OrderJoins2Tables(auto optimizerType) {
     TTypeAnnotationContext typeCtx;
     typeCtx.CostBasedOptimizer = optimizerType;
     state->Types = &typeCtx;
-    auto optimizedTree = OrderJoins(tree, state, exprCtx, true);
+    auto optimizedTree = OrderJoins(tree, state, cluster, exprCtx, true);
     UNIT_ASSERT(optimizedTree != tree);
     UNIT_ASSERT(optimizedTree->Left);
     UNIT_ASSERT(optimizedTree->Right);
@@ -242,6 +256,7 @@ ADD_TEST(OrderJoins2Tables)
 
 void OrderJoins2TablesComplexLabel(auto optimizerType)
 {
+    const TString cluster("ut_cluster");
     TExprContext exprCtx;
     auto tree = MakeOp({"c", "c_nationkey"}, {"n", "n_nationkey"}, {"c", "n", "e"}, exprCtx);
     tree->Left = MakeLeaf({"c"}, {"c"}, 1000000, 1233333, exprCtx);
@@ -251,7 +266,7 @@ void OrderJoins2TablesComplexLabel(auto optimizerType)
     TYtState::TPtr state = MakeIntrusive<TYtState>();
     typeCtx.CostBasedOptimizer = optimizerType;
     state->Types = &typeCtx;
-    auto optimizedTree = OrderJoins(tree, state, exprCtx, true);
+    auto optimizedTree = OrderJoins(tree, state, cluster, exprCtx, true);
     UNIT_ASSERT(optimizedTree != tree);
 }
 
@@ -259,6 +274,7 @@ ADD_TEST(OrderJoins2TablesComplexLabel)
 
 void OrderJoins2TablesTableIn2Rels(auto optimizerType)
 {
+    const TString cluster("ut_cluster");
     TExprContext exprCtx;
     auto tree = MakeOp({"c", "c_nationkey"}, {"n", "n_nationkey"}, {"c", "n", "e"}, exprCtx);
     tree->Left = MakeLeaf({"c"}, {"c"}, 1000000, 1233333, exprCtx);
@@ -268,7 +284,7 @@ void OrderJoins2TablesTableIn2Rels(auto optimizerType)
     TYtState::TPtr state = MakeIntrusive<TYtState>();
     typeCtx.CostBasedOptimizer = optimizerType;
     state->Types = &typeCtx;
-    auto optimizedTree = OrderJoins(tree, state, exprCtx, true);
+    auto optimizedTree = OrderJoins(tree, state, cluster, exprCtx, true);
     UNIT_ASSERT(optimizedTree != tree);
 }
 
@@ -276,6 +292,7 @@ ADD_TEST(OrderJoins2TablesTableIn2Rels)
 
 Y_UNIT_TEST(OrderLeftJoin)
 {
+    const TString cluster("ut_cluster");
     TExprContext exprCtx;
     auto tree = MakeOp({"c", "c_nationkey"}, {"n", "n_nationkey"}, {"c", "n"}, exprCtx);
     tree->Left = MakeLeaf({"c"}, {"c"}, 1000000, 1233333, exprCtx);
@@ -286,13 +303,14 @@ Y_UNIT_TEST(OrderLeftJoin)
     TYtState::TPtr state = MakeIntrusive<TYtState>();
     typeCtx.CostBasedOptimizer = ECostBasedOptimizerType::PG;
     state->Types = &typeCtx;
-    auto optimizedTree = OrderJoins(tree, state, exprCtx, true);
+    auto optimizedTree = OrderJoins(tree, state, cluster, exprCtx, true);
     UNIT_ASSERT(optimizedTree != tree);
     UNIT_ASSERT_STRINGS_EQUAL("Left", optimizedTree->JoinKind->Content());
 }
 
 Y_UNIT_TEST(UnsupportedJoin)
 {
+    const TString cluster("ut_cluster");
     TExprContext exprCtx;
     auto tree = MakeOp({"c", "c_nationkey"}, {"n", "n_nationkey"}, {"c", "n"}, exprCtx);
     tree->Left = MakeLeaf({"c"}, {"c"}, 1000000, 1233333, exprCtx);
@@ -303,11 +321,12 @@ Y_UNIT_TEST(UnsupportedJoin)
     TYtState::TPtr state = MakeIntrusive<TYtState>();
     typeCtx.CostBasedOptimizer = ECostBasedOptimizerType::PG;
     state->Types = &typeCtx;
-    auto optimizedTree = OrderJoins(tree, state, exprCtx, true);
+    auto optimizedTree = OrderJoins(tree, state, cluster, exprCtx, true);
     UNIT_ASSERT(optimizedTree == tree);
 }
 
 Y_UNIT_TEST(OrderJoinSinglePass) {
+    const TString cluster("ut_cluster");
     TExprContext exprCtx;
     auto tree = MakeOp({"c", "c_nationkey"}, {"n", "n_nationkey"}, {"c", "n"}, exprCtx);
     tree->Left = MakeLeaf({"c"}, {"c"}, 1000000, 1233333, exprCtx);
@@ -318,12 +337,13 @@ Y_UNIT_TEST(OrderJoinSinglePass) {
     TYtState::TPtr state = MakeIntrusive<TYtState>();
     typeCtx.CostBasedOptimizer = ECostBasedOptimizerType::PG;
     state->Types = &typeCtx;
-    auto optimizedTree = OrderJoins(tree, state, exprCtx, true);
+    auto optimizedTree = OrderJoins(tree, state, cluster, exprCtx, true);
     UNIT_ASSERT(optimizedTree != tree);
     UNIT_ASSERT(optimizedTree->CostBasedOptPassed);
 }
 
 Y_UNIT_TEST(OrderJoinsDoesNothingWhenCBOAlreadyPassed) {
+    const TString cluster("ut_cluster");
     TExprContext exprCtx;
     auto tree = MakeOp({"c", "c_nationkey"}, {"n", "n_nationkey"}, {"c", "n"}, exprCtx);
     tree->Left = MakeLeaf({"c"}, {"c"}, 1000000, 1233333, exprCtx);
@@ -335,7 +355,7 @@ Y_UNIT_TEST(OrderJoinsDoesNothingWhenCBOAlreadyPassed) {
     TYtState::TPtr state = MakeIntrusive<TYtState>();
     typeCtx.CostBasedOptimizer = ECostBasedOptimizerType::PG;
     state->Types = &typeCtx;
-    auto optimizedTree = OrderJoins(tree, state, exprCtx, true);
+    auto optimizedTree = OrderJoins(tree, state, cluster, exprCtx, true);
     UNIT_ASSERT(optimizedTree == tree);
 }
 

ydb/library/yql/providers/yt/provider/yql_yt_gateway.h

@@ -515,10 +515,16 @@ public:
         OPTION_FIELD(TString, Cluster)
         OPTION_FIELD(TVector<TPathStatReq>, Paths)
         OPTION_FIELD(TYtSettings::TConstPtr, Config)
+        OPTION_FIELD_DEFAULT(bool, Extended, false)
     };
 
     struct TPathStatResult: public NCommon::TOperationResult {
+        struct TExtendedResult {
+            THashMap<TString, i64> DataWeight;
+            THashMap<TString, ui64> EstimatedUniqueCounts;
+        };
         TVector<ui64> DataSize;
+        TVector<TMaybe<TExtendedResult>> Extended;
     };
 
     struct TFullResultTableOptions : public TCommonOptions {
@@ -623,4 +629,4 @@ public:
     virtual void AddCluster(const TYtClusterConfig& cluster) = 0;
 };
 
-}
+}

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

@@ -240,10 +240,11 @@ IGraphTransformer::TStatus TryEstimateDataSizeChecked(TVector<ui64>& result, TYt
 
     TSet<TString> requestedColumns;
     auto status = TryEstimateDataSize(result, requestedColumns, cluster, paths, columns, state, ctx);
+    auto settings = inputSection.Settings().Ptr();
     if (status == TStatus::Repeat) {
         bool hasStatColumns = NYql::HasSetting(inputSection.Settings().Ref(), EYtSettingType::StatColumns);
         if (hasStatColumns) {
-            auto oldColumns = NYql::GetSettingAsColumnList(inputSection.Settings().Ref(), EYtSettingType::StatColumns);
+            auto oldColumns = NYql::GetSettingAsColumnList(*settings, EYtSettingType::StatColumns);
             TSet<TString> oldColumnSet(oldColumns.begin(), oldColumns.end());
 
             bool alreadyRequested = AllOf(requestedColumns, [&](const auto& c) {
@@ -251,6 +252,8 @@ IGraphTransformer::TStatus TryEstimateDataSizeChecked(TVector<ui64>& result, TYt
             });
 
             YQL_ENSURE(!alreadyRequested);
+
+            settings = NYql::RemoveSetting(*settings, EYtSettingType::StatColumns, ctx);
         }
 
         YQL_CLOG(INFO, ProviderYt) << "Stat missing for columns: " << JoinSeq(", ", requestedColumns) << ", rebuilding section";
@@ -258,7 +261,7 @@ IGraphTransformer::TStatus TryEstimateDataSizeChecked(TVector<ui64>& result, TYt
 
         inputSection = Build<TYtSection>(ctx, inputSection.Ref().Pos())
             .InitFrom(inputSection)
-            .Settings(NYql::AddSettingAsColumnList(inputSection.Settings().Ref(), EYtSettingType::StatColumns, requestedColumnList, ctx))
+            .Settings(NYql::AddSettingAsColumnList(*settings, EYtSettingType::StatColumns, requestedColumnList, ctx))
             .Done();
     }
     return status;
@@ -2906,7 +2909,8 @@ TStatus CollectPathsAndLabels(TVector<TYtPathInfo::TPtr>& tables, TJoinLabels& l
     return TStatus::Ok;
 }
 
-TStatus CollectPathsAndLabelsReady(bool& ready, TVector<TYtPathInfo::TPtr>& tables, TJoinLabels& labels,
+IGraphTransformer::TStatus CollectPathsAndLabelsReady(
+    bool& ready, TVector<TYtPathInfo::TPtr>& tables, TJoinLabels& labels,
     const TStructExprType*& itemType, const TStructExprType*& itemTypeBeforePremap,
     const TYtJoinNodeLeaf& leaf, TExprContext& ctx)
 {
@@ -4682,6 +4686,111 @@ TYtJoinNodeOp::TPtr ImportYtEquiJoin(TYtEquiJoin equiJoin, TExprContext& ctx) {
     return root;
 }
 
+IGraphTransformer::TStatus CollectCboStatsLeaf(
+    const THashMap<TString, THashSet<TString>>& relJoinColumns,
+    const TString& cluster,
+    TYtJoinNodeLeaf& leaf,
+    const TYtState::TPtr& state,
+    TExprContext& ctx) {
+
+    const TMaybe<ui64> maxChunkCountExtendedStats = state->Configuration->ExtendedStatsMaxChunkCount.Get();
+    if (!maxChunkCountExtendedStats) {
+        return TStatus::Ok;
+    }
+
+    TVector<TString> keyList;
+    auto columnsPos = relJoinColumns.find(JoinLeafLabel(leaf.Label));
+    if (columnsPos != relJoinColumns.end()) {
+        keyList.assign(columnsPos->second.begin(), columnsPos->second.end());
+    }
+    TVector<IYtGateway::TPathStatReq> pathStatReqs;
+
+    ui64 sectionChunkCount = 0;
+    TVector<TString> requestedColumnList;
+    for (auto path: leaf.Section.Paths()) {
+        auto pathInfo = MakeIntrusive<TYtPathInfo>(path);
+        sectionChunkCount += pathInfo->Table->Stat->ChunkCount;
+
+        auto ytPath = BuildYtPathForStatRequest(cluster, *pathInfo, keyList, *state, ctx);
+
+        if (!ytPath) {
+            return IGraphTransformer::TStatus::Error;
+        }
+
+        pathStatReqs.push_back(
+            IYtGateway::TPathStatReq()
+                .Path(*ytPath)
+                .IsTemp(pathInfo->Table->IsTemp)
+                .IsAnonymous(pathInfo->Table->IsAnonymous)
+                .Epoch(pathInfo->Table->Epoch.GetOrElse(0)));
+
+        std::copy(ytPath->Columns_->Parts_.begin(), ytPath->Columns_->Parts_.end(), std::back_inserter(requestedColumnList));
+    }
+
+    if ((*maxChunkCountExtendedStats != 0 && sectionChunkCount > *maxChunkCountExtendedStats)
+        || !pathStatReqs) {
+        return TStatus::Ok;
+    }
+
+    IYtGateway::TPathStatOptions pathStatOptions =
+        IYtGateway::TPathStatOptions(state->SessionId)
+            .Cluster(cluster)
+            .Paths(pathStatReqs)
+            .Config(state->Configuration->Snapshot())
+            .Extended(true);
+
+    IYtGateway::TPathStatResult pathStats = state->Gateway->TryPathStat(std::move(pathStatOptions));
+
+    if (pathStats.Success()) {
+        return TStatus::Ok;
+    }
+    std::sort(requestedColumnList.begin(), requestedColumnList.end());
+    requestedColumnList.erase(std::unique(requestedColumnList.begin(), requestedColumnList.end()),
+        requestedColumnList.end());
+    leaf.Section = Build<TYtSection>(ctx, leaf.Section.Ref().Pos())
+        .InitFrom(leaf.Section)
+        .Settings(NYql::AddSettingAsColumnList(leaf.Section.Settings().Ref(), EYtSettingType::StatColumns, requestedColumnList, ctx))
+        .Done();
+    return TStatus::Repeat;
+}
+
+void AddJoinColumns(THashMap<TString, THashSet<TString>>& relJoinColumns, const TYtJoinNodeOp& op) {
+    for (ui32 i = 0; i < op.LeftLabel->ChildrenSize(); i += 2) {
+        auto ltable = op.LeftLabel->Child(i)->Content();
+        auto lcolumn = op.LeftLabel->Child(i + 1)->Content();
+        auto rtable = op.RightLabel->Child(i)->Content();
+        auto rcolumn = op.RightLabel->Child(i + 1)->Content();
+
+        relJoinColumns[TString(ltable)].insert(TString(lcolumn));
+        relJoinColumns[TString(rtable)].insert(TString(rcolumn));
+    }
+}
+
+IGraphTransformer::TStatus CollectCboStatsNode(THashMap<TString, THashSet<TString>>& relJoinColumns, const TString& cluster, TYtJoinNodeOp& op, const TYtState::TPtr& state, TExprContext& ctx) {
+    IGraphTransformer::TStatus result = TStatus::Ok;
+    TYtJoinNodeLeaf* leftLeaf = dynamic_cast<TYtJoinNodeLeaf*>(op.Left.Get());
+    TYtJoinNodeLeaf* rightLeaf = dynamic_cast<TYtJoinNodeLeaf*>(op.Right.Get());
+    AddJoinColumns(relJoinColumns, op);
+    if (leftLeaf) {
+        result = result.Combine(CollectCboStatsLeaf(relJoinColumns, cluster, *leftLeaf, state, ctx));
+    } else {
+        auto& leftOp = *dynamic_cast<TYtJoinNodeOp*>(op.Left.Get());
+        result = result.Combine(CollectCboStatsNode(relJoinColumns, cluster, leftOp, state, ctx));
+    }
+    if (rightLeaf) {
+        result = result.Combine(CollectCboStatsLeaf(relJoinColumns, cluster, *rightLeaf, state, ctx));
+    } else {
+        auto& rightOp = *dynamic_cast<TYtJoinNodeOp*>(op.Right.Get());
+        result = result.Combine(CollectCboStatsNode(relJoinColumns, cluster, rightOp, state, ctx));
+    }
+    return result;
+}
+
+IGraphTransformer::TStatus CollectCboStats(const TString& cluster, TYtJoinNodeOp& op, const TYtState::TPtr& state, TExprContext& ctx) {
+    THashMap<TString, THashSet<TString>> relJoinColumns;
+    return CollectCboStatsNode(relJoinColumns, cluster, op, state, ctx);
+}
+
 IGraphTransformer::TStatus RewriteYtEquiJoin(TYtEquiJoin equiJoin, TYtJoinNodeOp& op, const TYtState::TPtr& state, TExprContext& ctx) {
     switch (RewriteYtEquiJoinStar(equiJoin, op, state, ctx)) {
     case EStarRewriteStatus::Error:
@@ -4790,4 +4899,19 @@ TMaybeNode<TExprBase> ExportYtEquiJoin(TYtEquiJoin equiJoin, const TYtJoinNodeOp
     return TExprBase(ctx.ChangeChildren(join.Ref(), std::move(children)));
 }
 
+TString JoinLeafLabel(TExprNode::TPtr label) {
+    if (label->ChildrenSize() == 0) {
+        return TString(label->Content());
+    }
+    TString result;
+    for (ui32 i = 0; i < label->ChildrenSize(); ++i) {
+        result += label->Child(i)->Content();
+        if (i+1 != label->ChildrenSize()) {
+            result += ",";
+        }
+    }
+
+    return result;
+}
+
 }

ydb/library/yql/providers/yt/provider/yql_yt_join_impl.h

@@ -62,15 +62,19 @@ struct TYtJoinNodeOp : TYtJoinNode {
 };
 
 TYtJoinNodeOp::TPtr ImportYtEquiJoin(TYtEquiJoin equiJoin, TExprContext& ctx);
+
+IGraphTransformer::TStatus CollectCboStats(const TString& cluster, TYtJoinNodeOp& op, const TYtState::TPtr& state, TExprContext& ctx);
+
 IGraphTransformer::TStatus RewriteYtEquiJoinLeaves(TYtEquiJoin equiJoin, TYtJoinNodeOp& op, const TYtState::TPtr& state, TExprContext& ctx);
 IGraphTransformer::TStatus RewriteYtEquiJoin(TYtEquiJoin equiJoin, TYtJoinNodeOp& op, const TYtState::TPtr& state, TExprContext& ctx);
 TMaybeNode<TExprBase> ExportYtEquiJoin(TYtEquiJoin equiJoin, const TYtJoinNodeOp& op, TExprContext& ctx, const TYtState::TPtr& state);
-TYtJoinNodeOp::TPtr OrderJoins(TYtJoinNodeOp::TPtr op, const TYtState::TPtr& state, TExprContext& ctx, bool debug = false);
+TYtJoinNodeOp::TPtr OrderJoins(TYtJoinNodeOp::TPtr op, const TYtState::TPtr& state, const TString& cluster,  TExprContext& ctx, bool debug = false);
+TString JoinLeafLabel(TExprNode::TPtr label);
 
 struct IBaseOptimizerNode;
 struct IProviderContext;
 
-void BuildOptimizerJoinTree(std::shared_ptr<IBaseOptimizerNode>& tree, std::shared_ptr<IProviderContext>& ctx, TYtJoinNodeOp::TPtr op);
+void BuildOptimizerJoinTree(TYtState::TPtr state, const TString& cluster, std::shared_ptr<IBaseOptimizerNode>& tree, std::shared_ptr<IProviderContext>& providerCtx, TYtJoinNodeOp::TPtr op, TExprContext& ctx);
 TYtJoinNode::TPtr BuildYtJoinTree(std::shared_ptr<IBaseOptimizerNode> node, TExprContext& ctx, TPositionHandle pos);
 bool AreSimilarTrees(TYtJoinNode::TPtr node1, TYtJoinNode::TPtr node2);