Merge f5a8b54 into 8a37438

Author: GrigoriyPA

ydb/library/yql/providers/common/pushdown/collection.cpp

@@ -39,7 +39,7 @@ TPredicateNode SplitForPartialPushdown(const NPushdown::TPredicateNode& predicat
 
 }
 
-TMaybeNode<TCoLambda> MakePushdownPredicate(const TCoLambda& lambda, TExprContext& ctx, const TPositionHandle& pos, const TSettings& settings) {
+NPushdown::TPredicateNode MakePushdownNode(const NNodes::TCoLambda& lambda, TExprContext& ctx, const TPositionHandle& pos, const TSettings& settings) {
  auto lambdaArg = lambda.Args().Arg(0).Ptr();
 
  YQL_LOG(TRACE) << "Push filter. Initial filter lambda: " << NCommon::ExprToPrettyString(ctx, lambda.Ref());
@@ -54,7 +54,11 @@ TMaybeNode<TCoLambda> MakePushdownPredicate(const TCoLambda& lambda, TExprContex
  NPushdown::CollectPredicates(optionalIf.Predicate(), predicateTree, TExprBase(lambdaArg), TExprBase(lambdaArg), settings);
  YQL_ENSURE(predicateTree.IsValid(), "Collected filter predicates are invalid");
 
- NPushdown::TPredicateNode predicateToPush = SplitForPartialPushdown(predicateTree, ctx, pos, settings);
+ return SplitForPartialPushdown(predicateTree, ctx, pos, settings);
+}
+
+TMaybeNode<TCoLambda> MakePushdownPredicate(const TCoLambda& lambda, TExprContext& ctx, const TPositionHandle& pos, const TSettings& settings) {
+ NPushdown::TPredicateNode predicateToPush = MakePushdownNode(lambda, ctx, pos, settings);
  if (!predicateToPush.IsValid()) {
  return {};
  }
@@ -64,7 +68,7 @@ TMaybeNode<TCoLambda> MakePushdownPredicate(const TCoLambda& lambda, TExprContex
  .Args({"filter_row"})
  .Body<TExprApplier>()
  .Apply(predicateToPush.ExprNode.Cast())
- .With(TExprBase(lambdaArg), "filter_row")
+ .With(lambda.Args().Arg(0), "filter_row")
  .Build()
  .Done();
  // clang-format on

ydb/library/yql/providers/common/pushdown/physical_opt.cpp

@@ -1,11 +1,15 @@
 #pragma once
 
+#include "predicate_node.h"
+
 #include <yql/essentials/ast/yql_expr.h>
 #include <yql/essentials/ast/yql_pos_handle.h>
+#include <yql/essentials/core/expr_nodes/yql_expr_nodes.h>
 #include <yql/essentials/core/expr_nodes_gen/yql_expr_nodes_gen.h>
 
 namespace NYql::NPushdown {
 
+NPushdown::TPredicateNode MakePushdownNode(const NNodes::TCoLambda& lambda, TExprContext& ctx, const TPositionHandle& pos, const TSettings& settings);
 NNodes::TMaybeNode<NNodes::TCoLambda> MakePushdownPredicate(const NNodes::TCoLambda& lambda, TExprContext& ctx, const TPositionHandle& pos, const TSettings& settings);
 
 } // namespace NYql::NPushdown

ydb/library/yql/providers/common/pushdown/physical_opt.h

@@ -30,6 +30,16 @@ bool TPredicateNode::IsValid() const {
  return res && ExprNode.IsValid();
 }
 
+bool TPredicateNode::IsEmpty() const {
+ if (!ExprNode || !IsValid()) {
+ return true;
+ }
+ if (const auto maybeBool = ExprNode.Maybe<NNodes::TCoBool>()) {
+ return TStringBuf(maybeBool.Cast().Literal()) == "true"sv;
+ }
+ return false;
+}
+
 void TPredicateNode::SetPredicates(const std::vector<TPredicateNode>& predicates, TExprContext& ctx, TPositionHandle pos, EBoolOp op) {
  auto predicatesSize = predicates.size();
  if (predicatesSize == 0) {

ydb/library/yql/providers/common/pushdown/predicate_node.cpp

@@ -23,6 +23,7 @@ struct TPredicateNode {
  ~TPredicateNode();
 
  bool IsValid() const;
+ bool IsEmpty() const;
  void SetPredicates(const std::vector<TPredicateNode>& predicates, TExprContext& ctx, TPositionHandle pos, EBoolOp op);
 
  NNodes::TMaybeNode<NNodes::TExprBase> ExprNode;

ydb/library/yql/providers/common/pushdown/predicate_node.h

@@ -38,7 +38,9 @@ struct TSettings {
  // May be partially pushdowned as:
  // $A OR $C
  // In case of unsupported / complicated expressions $B and $D
- SplitOrOperator = 1 << 22
+ SplitOrOperator = 1 << 22,
+ ToBytesFromStringExpressions = 1 << 23, // ToBytes(string like)
+ FlatMapOverOptionals = 1 << 24 // FlatMap(Optional<T>, Lmabda (T) -> Optional<U>)
  };
 
  explicit TSettings(NLog::EComponent logComponent)

ydb/library/yql/providers/common/pushdown/settings.h

@@ -352,6 +352,12 @@ message TExpression {
  TExpression else_expression = 3;
  }
 
+ // CAST($value AS $type)
+ message TCast {
+ TExpression value = 1;
+ string type = 2;
+ }
+
  message TNull {
  }
 
@@ -368,6 +374,8 @@ message TExpression {
  TCoalesce coalesce = 5;
 
  TIf if = 6;
+
+ TCast cast = 7;
  }
 }
 

ydb/library/yql/providers/generic/connector/api/service/protos/connector.proto

@@ -9,6 +9,7 @@ namespace NYql::NConnector::NApi {
 namespace NYql {
 
  bool IsEmptyFilterPredicate(const NNodes::TCoLambda& lambda);
+ bool SerializeFilterPredicate(const NNodes::TExprBase& predicateBody, const NNodes::TCoArgument& predicateArgument, NConnector::NApi::TPredicate* proto, TStringBuilder& err);
  bool SerializeFilterPredicate(const NNodes::TCoLambda& predicate, NConnector::NApi::TPredicate* proto, TStringBuilder& err);
  TString FormatWhere(const NConnector::NApi::TPredicate& predicate);
 } // namespace NYql

ydb/library/yql/providers/generic/provider/yql_generic_predicate_pushdown.cpp

@@ -55,6 +55,7 @@ namespace NYql::NGenericPushDown {
  case NYql::NConnector::NApi::TExpression::kNull:
  case NYql::NConnector::NApi::TExpression::kCoalesce:
  case NYql::NConnector::NApi::TExpression::kIf:
+ case NYql::NConnector::NApi::TExpression::kCast:
  case NYql::NConnector::NApi::TExpression::PAYLOAD_NOT_SET:
  return false;
  }
@@ -70,6 +71,7 @@ namespace NYql::NGenericPushDown {
  case NYql::NConnector::NApi::TExpression::kNull:
  case NYql::NConnector::NApi::TExpression::kCoalesce:
  case NYql::NConnector::NApi::TExpression::kIf:
+ case NYql::NConnector::NApi::TExpression::kCast:
  case NYql::NConnector::NApi::TExpression::PAYLOAD_NOT_SET:
  return false;
  }
@@ -281,6 +283,7 @@ namespace NYql::NGenericPushDown {
  case NYql::NConnector::NApi::TExpression::kNull:
  case NYql::NConnector::NApi::TExpression::kCoalesce:
  case NYql::NConnector::NApi::TExpression::kIf:
+ case NYql::NConnector::NApi::TExpression::kCast:
  case NYql::NConnector::NApi::TExpression::PAYLOAD_NOT_SET:
  return Triple::Unknown;
  }