YDB FQ: pushdown Decimal type (#26593)

Author: vitalyisaev2

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

@@ -1,7 +1,6 @@
 #include "collection.h"
 
 #include <yql/essentials/core/yql_expr_type_annotation.h>
-#include <yql/essentials/utils/log/log.h>
 
 #include <vector>
 
@@ -211,6 +210,9 @@ class TPredicateMarkup {
         if (Settings.IsEnabled(EFlag::StringTypes) && (node.Maybe<TCoUtf8>() || node.Maybe<TCoString>())) {
             return true;
         }
+        if (Settings.IsEnabled(EFlag::DecimalCtor) && node.Maybe<TCoDecimal>()) {
+            return true;
+        }
         return false;
     }
 

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

@@ -45,6 +45,7 @@ struct TSettings {
         IntervalCtor = 1 << 26,
         MinMax = 1 << 27,
         NonDeterministic = 1 << 28,
+        DecimalCtor = 1 << 29, 
     };
 
     explicit TSettings(NLog::EComponent logComponent)

ydb/library/yql/providers/generic/provider/ut/pushdown/pushdown_ut.cpp

@@ -143,6 +143,24 @@ struct TFakeGenericClient: public NConnector::IClient {
         PRIMITIVE_TYPE_COL("json_document", JSON_DOCUMENT);
         PRIMITIVE_TYPE_COL("dynumber", DYNUMBER);
 
+        // Add decimal columns
+        {
+            auto* col = schema.add_columns();
+            col->set_name("col_decimal_precision10_scale0");
+            auto* t = col->mutable_type();
+            auto* decimalType = t->mutable_decimal_type();
+            decimalType->set_precision(10);
+            decimalType->set_scale(0);
+        }
+        {
+            auto* col = schema.add_columns();
+            col->set_name("col_decimal_precision4_scale2");
+            auto* t = col->mutable_type();
+            auto* decimalType = t->mutable_decimal_type();
+            decimalType->set_precision(4);
+            decimalType->set_scale(2);
+        }
+
         return NThreading::MakeFuture<NConnector::TDescribeTableAsyncResult::value_type>(std::move(result));
     }
 
@@ -765,4 +783,69 @@ Y_UNIT_TEST_SUITE_F(PushdownTest, TPushdownFixture) {
             )proto"
         );
     }
+
+    Y_UNIT_TEST(DecimalPushdownPrecision10Scale0) {
+        AssertFilter(
+            R"ast(
+                (==
+                    (Member $row '"col_decimal_precision10_scale0")
+                    (Decimal '"1" '"10" '"0")
+                )
+                )ast",
+            R"proto(
+                comparison {
+                    operation: EQ
+                    left_value {
+                        column: "col_decimal_precision10_scale0"
+                    }
+                    right_value {
+                        typed_value {
+                            type {
+                                decimal_type {
+                                    precision: 10
+                                    scale: 0
+                                }
+                            }
+                            value {
+                                bytes_value: "\001\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000"
+                            }
+                        }
+                    }
+                }
+            )proto"
+        );
+    }
+
+    Y_UNIT_TEST(DecimalPushdownPrecesion4Scale2) {
+        // Test with negative decimal value with fractional part
+        AssertFilter(
+            R"ast(
+                (==
+                    (Member $row '"col_decimal_precision4_scale2")
+                    (Decimal '"-22.22" '"4" '"2")
+                )
+                )ast",
+            R"proto(
+                comparison {
+                    operation: EQ
+                    left_value {
+                        column: "col_decimal_precision4_scale2"
+                    }
+                    right_value {
+                        typed_value {
+                            type {
+                                decimal_type {
+                                    precision: 4
+                                    scale: 2
+                                }
+                            }
+                            value {
+                                bytes_value: "R\367\377\377\377\377\377\377\377\377\377\377\377\377\377\377"
+                            }
+                        }
+                    }
+                }
+            )proto"
+        );
+    }
 }

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

@@ -37,7 +37,8 @@ namespace NYql {
                     EFlag::LikeOperator |
                     EFlag::JustPassthroughOperators | // To pushdown REGEXP over String column
                     EFlag::FlatMapOverOptionals | // To pushdown REGEXP over Utf8 column
-                    EFlag::ToStringFromStringExpressions // To pushdown REGEXP over Utf8 column
+                    EFlag::ToStringFromStringExpressions | // To pushdown REGEXP over Utf8 column
+                    EFlag::DecimalType | EFlag::DecimalCtor
                 );
                 EnableFunction("Re2.Grep");  // For REGEXP pushdown
             }

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

@@ -227,6 +227,27 @@ namespace NYql {
             return SerializeExpression(lambda.Body(), dstProto->mutable_then_expression(), ctx, depth + 1);
         }
 
+        bool SerializeDecimal(const TCoDecimal& coDecimal, TExpression* proto, TSerializationContext& /*ctx*/, ui64 /*depth*/) {
+            auto* protoTypedValue = proto->mutable_typed_value();
+            auto* protoDecimalType = protoTypedValue->mutable_type()->mutable_decimal_type();
+
+            // extract precision and scale
+            auto precision = FromString<ui32>(coDecimal.Precision().StringValue());
+            auto scale = FromString<ui32>(coDecimal.Scale().StringValue());
+            protoDecimalType->set_precision(precision);
+            protoDecimalType->set_scale(scale);
+
+            // extract decimal value itself 
+            auto decimal = NDecimal::FromString(coDecimal.Cast<TCoDecimal>().Literal().Value(), precision, scale);
+            static_assert(sizeof(decimal) == 16, "wrong TInt128 size");
+            
+            // Set the bytes value with the 16 buffer containing the decimal value bytes
+            protoTypedValue->mutable_value()->set_bytes_value(reinterpret_cast<char*>(&decimal), sizeof(decimal));
+            
+            return true;
+        }
+
+
 #define MATCH_ATOM(AtomType, ATOM_ENUM, proto_name, cpp_type)                             \
     if (auto atom = expression.Maybe<Y_CAT(TCo, AtomType)>()) {                           \
         auto* value = proto->mutable_typed_value();                                       \
@@ -300,6 +321,9 @@ namespace NYql {
             if (auto dependsOn = expression.Maybe<TCoDependsOn>()) {
                 return SerializeExpression(dependsOn.Cast().Input(), proto, ctx, depth + 1);
             }
+            if (auto decimal = expression.Maybe<TCoDecimal>()) {
+                return SerializeDecimal(decimal.Cast(), proto, ctx, depth);
+            }
 
             // data
             MATCH_ATOM(Bool, BOOL, bool, bool);