pushdown substring match functions (#18459)

8 files changed, 87 insertions, 47 deletions

diff --git a/ydb/core/kqp/compile_service/kqp_compile_actor.cpp b/ydb/core/kqp/compile_service/kqp_compile_actor.cpp
index cbbcf8eb2c7..b8b054d9b77 100644
--- a/ydb/core/kqp/compile_service/kqp_compile_actor.cpp
+++ b/ydb/core/kqp/compile_service/kqp_compile_actor.cpp
@@ -657,6 +657,7 @@ void ApplyServiceConfig(TKikimrConfiguration& kqpConfig, const TTableServiceConf
     kqpConfig.EnableNewRBO = serviceConfig.GetEnableNewRBO();
     kqpConfig.EnableSpillingInHashJoinShuffleConnections = serviceConfig.GetEnableSpillingInHashJoinShuffleConnections();
     kqpConfig.EnableOlapScalarApply = serviceConfig.GetEnableOlapScalarApply();
+    kqpConfig.EnableOlapSubstringPushdown = serviceConfig.GetEnableOlapSubstringPushdown();
 
     if (const auto limit = serviceConfig.GetResourceManager().GetMkqlHeavyProgramMemoryLimit()) {
         kqpConfig._KqpYqlCombinerMemoryLimit = std::max(1_GB, limit - (limit >> 2U));
diff --git a/ydb/core/kqp/compile_service/kqp_compile_service.cpp b/ydb/core/kqp/compile_service/kqp_compile_service.cpp
index 7fb74fbd6d4..fa05389185c 100644
--- a/ydb/core/kqp/compile_service/kqp_compile_service.cpp
+++ b/ydb/core/kqp/compile_service/kqp_compile_service.cpp
@@ -321,6 +321,7 @@ private:
         bool enableNewRBO = TableServiceConfig.GetEnableNewRBO();
         bool enableSpillingInHashJoinShuffleConnections = TableServiceConfig.GetEnableSpillingInHashJoinShuffleConnections();
         bool enableOlapScalarApply = TableServiceConfig.GetEnableOlapScalarApply();
+        bool enableOlapSubstringPushdown = TableServiceConfig.GetEnableOlapSubstringPushdown();
 
         TableServiceConfig.Swap(event.MutableConfig()->MutableTableServiceConfig());
         LOG_INFO(*TlsActivationContext, NKikimrServices::KQP_COMPILE_SERVICE, "Updated config");
@@ -357,7 +358,8 @@ private:
             TableServiceConfig.GetEnableQueryServiceSpilling() != enableSpilling ||
             TableServiceConfig.GetEnableNewRBO() != enableNewRBO ||
             TableServiceConfig.GetEnableSpillingInHashJoinShuffleConnections() != enableSpillingInHashJoinShuffleConnections ||
-            TableServiceConfig.GetEnableOlapScalarApply() != enableOlapScalarApply
+            TableServiceConfig.GetEnableOlapScalarApply() != enableOlapScalarApply ||
+            TableServiceConfig.GetEnableOlapSubstringPushdown() != enableOlapSubstringPushdown
         )
         {
 
diff --git a/ydb/core/kqp/opt/physical/kqp_opt_phy_olap_filter.cpp b/ydb/core/kqp/opt/physical/kqp_opt_phy_olap_filter.cpp
index d498c59c6bb..1ca3ed0c535 100644
--- a/ydb/core/kqp/opt/physical/kqp_opt_phy_olap_filter.cpp
+++ b/ydb/core/kqp/opt/physical/kqp_opt_phy_olap_filter.cpp
@@ -736,7 +736,10 @@ std::pair<TVector<TOLAPPredicateNode>, TVector<TOLAPPredicateNode>> SplitForPart
 TExprBase KqpPushOlapFilter(TExprBase node, TExprContext& ctx, const TKqpOptimizeContext& kqpCtx,
     TTypeAnnotationContext& typesCtx)
 {
-    const bool allowApply = kqpCtx.Config->EnableOlapScalarApply;
+    const auto pushdownOptions = TPushdownOptions{
+        kqpCtx.Config->EnableOlapScalarApply,
+        kqpCtx.Config->EnableOlapSubstringPushdown
+    };
     if (!kqpCtx.Config->HasOptEnableOlapPushdown()) {
         return node;
     }
@@ -768,16 +771,16 @@ TExprBase KqpPushOlapFilter(TExprBase node, TExprContext& ctx, const TKqpOptimiz
 
     TOLAPPredicateNode predicateTree;
     predicateTree.ExprNode = predicate.Ptr();
-    CollectPredicates(predicate, predicateTree, &lambdaArg, read.Process().Body(), false);
+    CollectPredicates(predicate, predicateTree, &lambdaArg, read.Process().Body(), pushdownOptions);
     YQL_ENSURE(predicateTree.IsValid(), "Collected OLAP predicates are invalid");
 
     auto [pushable, remaining] = SplitForPartialPushdown(predicateTree, false);
     TVector<TFilterOpsLevels> pushedPredicates;
     for (const auto& p: pushable) {
-        pushedPredicates.emplace_back(PredicatePushdown(TExprBase(p.ExprNode), lambdaArg, ctx, node.Pos(), allowApply));
+        pushedPredicates.emplace_back(PredicatePushdown(TExprBase(p.ExprNode), lambdaArg, ctx, node.Pos(), pushdownOptions.AllowOlapApply));
     }
 
-    if (allowApply) {
+    if (pushdownOptions.AllowOlapApply) {
         TVector<TOLAPPredicateNode> remainingAfterApply;
         for(const auto& p: remaining) {
             const auto recoveredOptinalIfForNonPushedDownPredicates = Build<TCoOptionalIf>(ctx, node.Pos())
@@ -797,13 +800,13 @@ TExprBase KqpPushOlapFilter(TExprBase node, TExprContext& ctx, const TKqpOptimiz
 
             TOLAPPredicateNode predicateTree;
             predicateTree.ExprNode = predicate.Ptr();
-            CollectPredicates(predicate, predicateTree, &lambdaArg, read.Process().Body(), true);
+            CollectPredicates(predicate, predicateTree, &lambdaArg, read.Process().Body(), {true, pushdownOptions.PushdownSubstring});
 
             YQL_ENSURE(predicateTree.IsValid(), "Collected OLAP predicates are invalid");
             auto [pushable, remaining] = SplitForPartialPushdown(predicateTree, true);
             for (const auto& p: pushable) {
                 if (p.CanBePushed) {
-                    auto pred = PredicatePushdown(TExprBase(p.ExprNode), lambdaArg, ctx, node.Pos(), allowApply);
+                    auto pred = PredicatePushdown(TExprBase(p.ExprNode), lambdaArg, ctx, node.Pos(), pushdownOptions.AllowOlapApply);
                     pushedPredicates.emplace_back(pred);
                 }
                 else {
diff --git a/ydb/core/kqp/opt/physical/predicate_collector.cpp b/ydb/core/kqp/opt/physical/predicate_collector.cpp
index 8d2c84f898a..a3fb9c8fa73 100644
--- a/ydb/core/kqp/opt/physical/predicate_collector.cpp
+++ b/ydb/core/kqp/opt/physical/predicate_collector.cpp
@@ -49,7 +49,7 @@ bool IsSupportedDataType(const TCoDataCtor& node, bool allowOlapApply) {
     return false;
 }
 
-bool IsSupportedCast(const TCoSafeCast& cast, bool allowOlapApply=false) {
+bool IsSupportedCast(const TCoSafeCast& cast, bool allowOlapApply) {
     auto maybeDataType = cast.Type().Maybe<TCoDataType>();
     if (!maybeDataType) {
         if (const auto maybeOptionalType = cast.Type().Maybe<TCoOptionalType>()) {
@@ -111,8 +111,26 @@ bool IsGoodTypeForComparsionPushdown(const TTypeAnnotationNode& type, bool allow
                                 NUdf::EDataTypeFeatures::TimeIntervalType) & features) && !(NUdf::EDataTypeFeatures::TzDateType & features)));
 }
 
+bool CanPushdownStringUdf(const TExprNode& udf, bool pushdownSubstring) {
+    if (!pushdownSubstring) {
+        return false;
+    }
+    const auto& name = udf.Head().Content();
+    static const THashSet<TString> substringMatchUdfs = {
+        "String.AsciiEqualsIgnoreCase",
+
+        "String.Contains",
+        "String.AsciiContainsIgnoreCase",
+        "String.StartsWith",
+        "String.AsciiStartsWithIgnoreCase",
+        "String.EndsWith",
+        "String.AsciiEndsWithIgnoreCase"
+    };
+    return substringMatchUdfs.contains(name);
+}
+
 [[maybe_unused]]
-bool AbstractTreeCanBePushed(const TExprBase& expr, const TExprNode* ) {
+bool AbstractTreeCanBePushed(const TExprBase& expr, const TExprNode*, bool pushdownSubstring) {
     if (!expr.Ref().IsCallable({"Apply", "Coalesce", "NamedApply", "IfPresent", "Visit"})) {
         return false;
     }
@@ -132,7 +150,7 @@ bool AbstractTreeCanBePushed(const TExprBase& expr, const TExprNode* ) {
     for (const auto& apply : applies) {
         const auto& udf = SkipCallables(apply->Head(), {"AssumeStrict"});
         const auto& udfName = udf.Head();
-        if (!(udfName.Content().starts_with("Json2.") || udfName.Content().starts_with("Re2."))) {
+        if (!(udfName.Content().starts_with("Json2.") || udfName.Content().starts_with("Re2.") || CanPushdownStringUdf(udf, pushdownSubstring))) {
             return false;
         }
 
@@ -159,8 +177,8 @@ bool IfPresentCanBePushed(const TCoIfPresent& ifPresent, const TExprNode* lambda
     return allowOlapApply;
 }
 
-bool CheckExpressionNodeForPushdown(const TExprBase& node, const TExprNode* lambdaArg, bool allowOlapApply) {
-    if (allowOlapApply) {
+bool CheckExpressionNodeForPushdown(const TExprBase& node, const TExprNode* lambdaArg, const TPushdownOptions& options) {
+    if (options.AllowOlapApply) {
         if (node.Maybe<TCoJust>() || node.Maybe<TCoCoalesce>()) {
             return true;
         }
@@ -174,9 +192,9 @@ bool CheckExpressionNodeForPushdown(const TExprBase& node, const TExprNode* lamb
     }
 
     if (const auto maybeSafeCast = node.Maybe<TCoSafeCast>()) {
-        return IsSupportedCast(maybeSafeCast.Cast(), allowOlapApply);
+        return IsSupportedCast(maybeSafeCast.Cast(), options.AllowOlapApply);
     } else if (const auto maybeData = node.Maybe<TCoDataCtor>()) {
-        return IsSupportedDataType(maybeData.Cast(), allowOlapApply);
+        return IsSupportedDataType(maybeData.Cast(), options.AllowOlapApply);
     } else if (const auto maybeMember = node.Maybe<TCoMember>()) {
         return IsMemberColumn(maybeMember.Cast(), lambdaArg);
     } else if (const auto maybeJsonValue = node.Maybe<TCoJsonValue>()) {
@@ -187,23 +205,23 @@ bool CheckExpressionNodeForPushdown(const TExprBase& node, const TExprNode* lamb
     }
 
     if (const auto op = node.Maybe<TCoUnaryArithmetic>()) {
-        return CheckExpressionNodeForPushdown(op.Cast().Arg(), lambdaArg, allowOlapApply) && IsGoodTypeForArithmeticPushdown(*op.Cast().Ref().GetTypeAnn(), allowOlapApply);
+        return CheckExpressionNodeForPushdown(op.Cast().Arg(), lambdaArg, options) && IsGoodTypeForArithmeticPushdown(*op.Cast().Ref().GetTypeAnn(), options.AllowOlapApply);
     } else if (const auto op = node.Maybe<TCoBinaryArithmetic>()) {
-        return CheckExpressionNodeForPushdown(op.Cast().Left(), lambdaArg, allowOlapApply) && CheckExpressionNodeForPushdown(op.Cast().Right(), lambdaArg, allowOlapApply)
-            && IsGoodTypeForArithmeticPushdown(*op.Cast().Ref().GetTypeAnn(), allowOlapApply) && !op.Cast().Maybe<TCoAggrAdd>();
+        return CheckExpressionNodeForPushdown(op.Cast().Left(), lambdaArg, options) && CheckExpressionNodeForPushdown(op.Cast().Right(), lambdaArg, options)
+            && IsGoodTypeForArithmeticPushdown(*op.Cast().Ref().GetTypeAnn(), options.AllowOlapApply) && !op.Cast().Maybe<TCoAggrAdd>();
     }
 
-    if (allowOlapApply) {
+    if (options.AllowOlapApply) {
         if (const auto maybeIfPresent = node.Maybe<TCoIfPresent>()) {
-            return IfPresentCanBePushed(maybeIfPresent.Cast(), lambdaArg, allowOlapApply);
+            return IfPresentCanBePushed(maybeIfPresent.Cast(), lambdaArg, options.AllowOlapApply);
         }
-        return AbstractTreeCanBePushed(node, lambdaArg);
+        return AbstractTreeCanBePushed(node, lambdaArg, options.PushdownSubstring);
     }
 
     return false;
 }
 
-bool IsGoodTypesForPushdownCompare(const TTypeAnnotationNode& typeOne, const TTypeAnnotationNode& typeTwo, bool allowOlapApply) {
+bool IsGoodTypesForPushdownCompare(const TTypeAnnotationNode& typeOne, const TTypeAnnotationNode& typeTwo, const TPushdownOptions& options) {
     const auto& rawOne = RemoveOptionality(typeOne);
     const auto& rawTwo = RemoveOptionality(typeTwo);
     if (IsSameAnnotation(rawOne, rawTwo))
@@ -225,14 +243,14 @@ bool IsGoodTypesForPushdownCompare(const TTypeAnnotationNode& typeOne, const TTy
             if (size != itemsTwo.size())
                 return false;
             for (auto i = 0U; i < size; ++i) {
-                if (!IsGoodTypesForPushdownCompare(*itemsOne[i], *itemsTwo[i], allowOlapApply)) {
+                if (!IsGoodTypesForPushdownCompare(*itemsOne[i], *itemsTwo[i], options)) {
                     return false;
                 }
             }
             return true;
         }
         case ETypeAnnotationKind::Data: {
-            return IsGoodTypeForComparsionPushdown(typeOne, allowOlapApply) && IsGoodTypeForComparsionPushdown(typeTwo, allowOlapApply);
+            return IsGoodTypeForComparsionPushdown(typeOne, options.AllowOlapApply) && IsGoodTypeForComparsionPushdown(typeTwo, options.AllowOlapApply);
         }
         default:
             break;
@@ -240,7 +258,7 @@ bool IsGoodTypesForPushdownCompare(const TTypeAnnotationNode& typeOne, const TTy
     return false;
 }
 
-bool CheckComparisonParametersForPushdown(const TCoCompare& compare, const TExprNode* lambdaArg, const TExprBase& input, bool allowOlapApply) {
+bool CheckComparisonParametersForPushdown(const TCoCompare& compare, const TExprNode* lambdaArg, const TExprBase& input, const TPushdownOptions& options) {
     const auto* inputType = input.Ref().GetTypeAnn();
     switch (inputType->GetKind()) {
         case ETypeAnnotationKind::Flow:
@@ -263,7 +281,7 @@ bool CheckComparisonParametersForPushdown(const TCoCompare& compare, const TExpr
         return false;
     }
 
-    if (!IsGoodTypesForPushdownCompare(*compare.Left().Ref().GetTypeAnn(), *compare.Right().Ref().GetTypeAnn(), allowOlapApply)) {
+    if (!IsGoodTypesForPushdownCompare(*compare.Left().Ref().GetTypeAnn(), *compare.Right().Ref().GetTypeAnn(), options)) {
         return false;
     }
 
@@ -272,7 +290,7 @@ bool CheckComparisonParametersForPushdown(const TCoCompare& compare, const TExpr
     YQL_ENSURE(leftList.size() == rightList.size(), "Different sizes of lists in comparison!");
 
     for (size_t i = 0; i < leftList.size(); ++i) {
-        if (!CheckExpressionNodeForPushdown(leftList[i], lambdaArg, allowOlapApply) || !CheckExpressionNodeForPushdown(rightList[i], lambdaArg, allowOlapApply)) {
+        if (!CheckExpressionNodeForPushdown(leftList[i], lambdaArg, options) || !CheckExpressionNodeForPushdown(rightList[i], lambdaArg, options)) {
             return false;
         }
     }
@@ -280,11 +298,11 @@ bool CheckComparisonParametersForPushdown(const TCoCompare& compare, const TExpr
     return true;
 }
 
-bool CompareCanBePushed(const TCoCompare& compare, const TExprNode* lambdaArg, const TExprBase& lambdaBody, bool allowOlapApply) {
-    return IsSupportedPredicate(compare) && CheckComparisonParametersForPushdown(compare, lambdaArg, lambdaBody, allowOlapApply);
+bool CompareCanBePushed(const TCoCompare& compare, const TExprNode* lambdaArg, const TExprBase& lambdaBody, const TPushdownOptions& options) {
+    return IsSupportedPredicate(compare) && CheckComparisonParametersForPushdown(compare, lambdaArg, lambdaBody, options);
 }
 
-bool SafeCastCanBePushed(const TCoFlatMap& flatmap, const TExprNode* lambdaArg, bool allowOlapApply) {
+bool SafeCastCanBePushed(const TCoFlatMap& flatmap, const TExprNode* lambdaArg, const TPushdownOptions& options) {
     /*
      * There are three ways of comparison in following format:
      *
@@ -305,7 +323,7 @@ bool SafeCastCanBePushed(const TCoFlatMap& flatmap, const TExprNode* lambdaArg,
     YQL_ENSURE(leftList.size() == rightList.size(), "Different sizes of lists in comparison!");
 
     for (size_t i = 0; i < leftList.size(); ++i) {
-        if (!CheckExpressionNodeForPushdown(leftList[i], lambdaArg, allowOlapApply) || !CheckExpressionNodeForPushdown(rightList[i], lambdaArg, allowOlapApply)) {
+        if (!CheckExpressionNodeForPushdown(leftList[i], lambdaArg, options) || !CheckExpressionNodeForPushdown(rightList[i], lambdaArg, options)) {
             return false;
         }
     }
@@ -339,20 +357,20 @@ bool JsonExistsCanBePushed(const TCoJsonExists& jsonExists, const TExprNode* lam
     return true;
 }
 
-bool CoalesceCanBePushed(const TCoCoalesce& coalesce, const TExprNode* lambdaArg, const TExprBase& lambdaBody, bool allowOlapApply) {
+bool CoalesceCanBePushed(const TCoCoalesce& coalesce, const TExprNode* lambdaArg, const TExprBase& lambdaBody, const TPushdownOptions& options) {
     if (!coalesce.Value().Maybe<TCoBool>()) {
         return false;
     }
 
     const auto predicate = coalesce.Predicate();
     if (const auto maybeCompare = predicate.Maybe<TCoCompare>()) {
-        return CompareCanBePushed(maybeCompare.Cast(), lambdaArg, lambdaBody, allowOlapApply);
+        return CompareCanBePushed(maybeCompare.Cast(), lambdaArg, lambdaBody, options);
     } else if (const auto maybeFlatmap = predicate.Maybe<TCoFlatMap>()) {
-        return SafeCastCanBePushed(maybeFlatmap.Cast(), lambdaArg, allowOlapApply);
+        return SafeCastCanBePushed(maybeFlatmap.Cast(), lambdaArg, options);
     } else if (const auto maybeJsonExists = predicate.Maybe<TCoJsonExists>()) {
         return JsonExistsCanBePushed(maybeJsonExists.Cast(), lambdaArg);
     } else if (const auto maybeIfPresent = predicate.Maybe<TCoIfPresent>()) {
-        return IfPresentCanBePushed(maybeIfPresent.Cast(), lambdaArg, allowOlapApply);
+        return IfPresentCanBePushed(maybeIfPresent.Cast(), lambdaArg, options.AllowOlapApply);
     }
 
     return false;
@@ -362,7 +380,7 @@ bool ExistsCanBePushed(const TCoExists& exists, const TExprNode* lambdaArg) {
     return IsMemberColumn(exists.Optional(), lambdaArg);
 }
 
-void CollectChildrenPredicates(const TExprNode& opNode, TOLAPPredicateNode& predicateTree, const TExprNode* lambdaArg, const TExprBase& lambdaBody, bool allowOlapApply) {
+void CollectChildrenPredicates(const TExprNode& opNode, TOLAPPredicateNode& predicateTree, const TExprNode* lambdaArg, const TExprBase& lambdaBody, const TPushdownOptions& options) {
     predicateTree.Children.reserve(opNode.ChildrenSize());
     predicateTree.CanBePushed = true;
     predicateTree.CanBePushedApply = true;
@@ -374,8 +392,9 @@ void CollectChildrenPredicates(const TExprNode& opNode, TOLAPPredicateNode& pred
             child.CanBePushed = IsSupportedDataType(maybeCtor.Cast(), false);
             child.CanBePushedApply = IsSupportedDataType(maybeCtor.Cast(), true);
         }
-        else
-            CollectPredicates(TExprBase(child.ExprNode), child, lambdaArg, lambdaBody, allowOlapApply);
+        else {
+            CollectPredicates(TExprBase(child.ExprNode), child, lambdaArg, lambdaBody, options);
+        }
         predicateTree.Children.emplace_back(child);
         predicateTree.CanBePushed &= child.CanBePushed;
         predicateTree.CanBePushedApply &= child.CanBePushedApply;
@@ -384,15 +403,15 @@ void CollectChildrenPredicates(const TExprNode& opNode, TOLAPPredicateNode& pred
 
 } // namespace
 
-void CollectPredicates(const TExprBase& predicate, TOLAPPredicateNode& predicateTree, const TExprNode* lambdaArg, const TExprBase& lambdaBody, bool allowOlapApply) {
+void CollectPredicates(const TExprBase& predicate, TOLAPPredicateNode& predicateTree, const TExprNode* lambdaArg, const TExprBase& lambdaBody, const TPushdownOptions& options) {
     if (predicate.Maybe<TCoNot>() || predicate.Maybe<TCoAnd>() || predicate.Maybe<TCoOr>() || predicate.Maybe<TCoXor>()) {
-        CollectChildrenPredicates(predicate.Ref(), predicateTree, lambdaArg, lambdaBody, allowOlapApply);
+        CollectChildrenPredicates(predicate.Ref(), predicateTree, lambdaArg, lambdaBody, options);
     } else if (const auto maybeCoalesce = predicate.Maybe<TCoCoalesce>()) {
-        predicateTree.CanBePushed = CoalesceCanBePushed(maybeCoalesce.Cast(), lambdaArg, lambdaBody, false);
-        predicateTree.CanBePushedApply = CoalesceCanBePushed(maybeCoalesce.Cast(), lambdaArg, lambdaBody, true);
+        predicateTree.CanBePushed = CoalesceCanBePushed(maybeCoalesce.Cast(), lambdaArg, lambdaBody, {false, options.PushdownSubstring});
+        predicateTree.CanBePushedApply = CoalesceCanBePushed(maybeCoalesce.Cast(), lambdaArg, lambdaBody, {true, options.PushdownSubstring});
     } else if (const auto maybeCompare = predicate.Maybe<TCoCompare>()) {
-        predicateTree.CanBePushed = CompareCanBePushed(maybeCompare.Cast(), lambdaArg, lambdaBody, false);
-        predicateTree.CanBePushedApply = CompareCanBePushed(maybeCompare.Cast(), lambdaArg, lambdaBody, true);
+        predicateTree.CanBePushed = CompareCanBePushed(maybeCompare.Cast(), lambdaArg, lambdaBody, {false, options.PushdownSubstring});
+        predicateTree.CanBePushedApply = CompareCanBePushed(maybeCompare.Cast(), lambdaArg, lambdaBody, {false, options.PushdownSubstring});
     } else if (const auto maybeExists = predicate.Maybe<TCoExists>()) {
         predicateTree.CanBePushed = ExistsCanBePushed(maybeExists.Cast(), lambdaArg);
         predicateTree.CanBePushedApply = predicateTree.CanBePushed;
@@ -401,12 +420,12 @@ void CollectPredicates(const TExprBase& predicate, TOLAPPredicateNode& predicate
         predicateTree.CanBePushedApply = predicateTree.CanBePushed;
     }
 
-    if (allowOlapApply && !predicateTree.CanBePushedApply){
+    if (options.AllowOlapApply && !predicateTree.CanBePushedApply){
         if (predicate.Maybe<TCoIf>() || predicate.Maybe<TCoJust>() || predicate.Maybe<TCoCoalesce>()) {
-            CollectChildrenPredicates(predicate.Ref(), predicateTree, lambdaArg, lambdaBody, true);    
+            CollectChildrenPredicates(predicate.Ref(), predicateTree, lambdaArg, lambdaBody, {true, options.PushdownSubstring});    
         }
         if (!predicateTree.CanBePushedApply) {
-            predicateTree.CanBePushedApply =  AbstractTreeCanBePushed(predicate, lambdaArg);
+            predicateTree.CanBePushedApply =  AbstractTreeCanBePushed(predicate, lambdaArg, options.PushdownSubstring);
         }
     }
 }
diff --git a/ydb/core/kqp/opt/physical/predicate_collector.h b/ydb/core/kqp/opt/physical/predicate_collector.h
index d85714c7d09..f4d48de2492 100644
--- a/ydb/core/kqp/opt/physical/predicate_collector.h
+++ b/ydb/core/kqp/opt/physical/predicate_collector.h
@@ -16,6 +16,11 @@ struct TOLAPPredicateNode {
     }
 };
 
-void CollectPredicates(const NNodes::TExprBase& predicate, TOLAPPredicateNode& predicateTree, const TExprNode* lambdaArg, const NNodes::TExprBase& lambdaBody, bool allowOlapApply);
+struct TPushdownOptions {
+    bool AllowOlapApply;
+    bool PushdownSubstring;
+};
+
+void CollectPredicates(const NNodes::TExprBase& predicate, TOLAPPredicateNode& predicateTree, const TExprNode* lambdaArg, const NNodes::TExprBase& lambdaBody, const TPushdownOptions& options);
 
 }
diff --git a/ydb/core/kqp/provider/yql_kikimr_settings.h b/ydb/core/kqp/provider/yql_kikimr_settings.h
index 417a5142200..edece0eedd7 100644
--- a/ydb/core/kqp/provider/yql_kikimr_settings.h
+++ b/ydb/core/kqp/provider/yql_kikimr_settings.h
@@ -188,6 +188,7 @@ struct TKikimrConfiguration : public TKikimrSettings, public NCommon::TSettingDi
     bool EnableNewRBO = false;
     bool EnableSpillingInHashJoinShuffleConnections = false;
     bool EnableOlapScalarApply = false;
+    bool EnableOlapSubstringPushdown = false;
 
     void SetDefaultEnabledSpillingNodes(const TString& node);
     ui64 GetEnabledSpillingNodes() const;
diff --git a/ydb/core/kqp/ut/olap/kqp_olap_ut.cpp b/ydb/core/kqp/ut/olap/kqp_olap_ut.cpp
index 31e0d9ea4cc..9768e24c060 100644
--- a/ydb/core/kqp/ut/olap/kqp_olap_ut.cpp
+++ b/ydb/core/kqp/ut/olap/kqp_olap_ut.cpp
@@ -1239,6 +1239,14 @@ Y_UNIT_TEST_SUITE(KqpOlap) {
             R"(ChooseMembers(TableRow(), ['level', 'uid', 'resource_id']) != <|level:1, uid:"uid_3000001", resource_id:"10001"|>)",
             R"(`uid` LIKE "_id%000_")",
             R"(`uid` ILIKE "UID%002")",
+
+            R"(Udf(String::AsciiEqualsIgnoreCase)(`uid`,  "UI"))",
+            R"(Udf(String::Contains)(`uid`,  "UI"))",
+            R"(Udf(String::AsciiContainsIgnoreCase)(`uid`,  "UI"))",
+            R"(Udf(String::StartsWith)(`uid`,  "UI"))",
+            R"(Udf(String::AsciiStartsWithIgnoreCase)(`uid`,  "UI"))",
+            R"(Udf(String::EndsWith)(`uid`,  "UI"))",
+            R"(Udf(String::AsciiEndsWithIgnoreCase)(`uid`,  "UI"))",
         };
 
         for (const auto& predicate: testData) {
diff --git a/ydb/core/protos/table_service_config.proto b/ydb/core/protos/table_service_config.proto
index 7e58d2bf6cf..73edf4aa402 100644
--- a/ydb/core/protos/table_service_config.proto
+++ b/ydb/core/protos/table_service_config.proto
@@ -389,4 +389,5 @@ message TTableServiceConfig {
     optional bool EnableNewRBO = 84 [default = false];
     optional bool EnableSpillingInHashJoinShuffleConnections = 85 [default = true];
     optional bool EnableOlapScalarApply = 86 [default = true];
+    optional bool EnableOlapSubstringPushdown = 87 [default = true];
 };