Merge 3f0db88 into 2ab0fd8

Author: fexolm

ydb/core/grpc_services/rpc_load_rows.cpp

@@ -75,9 +75,12 @@ bool ConvertArrowToYdbPrimitive(const arrow::DataType& type, Ydb::Type& toType)
         case arrow::Type::DURATION:
             toType.set_type_id(Ydb::Type::INTERVAL);
             return true;
-        case arrow::Type::DECIMAL:
-            // TODO
-            return false;
+        case arrow::Type::DECIMAL: {
+            Ydb::DecimalType* decimalType = toType.mutable_decimal_type();
+            decimalType->set_precision(22);
+            decimalType->set_scale(9);
+            return true;
+        }
         case arrow::Type::NA:
         case arrow::Type::HALF_FLOAT:
         case arrow::Type::FIXED_SIZE_BINARY:

ydb/core/kqp/ut/common/columnshard.cpp

@@ -246,6 +246,8 @@ namespace NKqp {
             return arrow::field(name, arrow::int64(), nullable);
         case NScheme::NTypeIds::JsonDocument:
             return arrow::field(name, arrow::binary(), nullable);
+        case NScheme::NTypeIds::Decimal:
+            return arrow::field(name, arrow::decimal(22, 9));
         case NScheme::NTypeIds::Pg:
             switch (NPg::PgTypeIdFromTypeDesc(typeDesc)) {
                 case INT2OID:

ydb/core/kqp/ut/olap/decimal_ut.cpp

@@ -0,0 +1,84 @@
+#include <ydb/core/formats/arrow/arrow_helpers.h>
+#include <ydb/core/kqp/ut/common/columnshard.h>
+#include <ydb/core/kqp/ut/common/kqp_ut_common.h>
+#include <ydb/core/testlib/common_helper.h>
+#include <ydb/core/testlib/cs_helper.h>
+#include <ydb/core/tx/columnshard/hooks/testing/controller.h>
+#include <ydb/core/tx/tx_proxy/proxy.h>
+
+#include <ydb/library/binary_json/write.h>
+#include <ydb/library/uuid/uuid.h>
+#include <ydb/public/sdk/cpp/client/draft/ydb_replication.h>
+#include <ydb/public/sdk/cpp/client/ydb_proto/accessor.h>
+#include <ydb/public/sdk/cpp/client/ydb_scheme/scheme.h>
+#include <ydb/public/sdk/cpp/client/ydb_topic/topic.h>
+
+#include <library/cpp/threading/local_executor/local_executor.h>
+#include <util/generic/serialized_enum.h>
+#include <util/string/printf.h>
+
+namespace NKikimr {
+namespace NKqp {
+
+using namespace NYdb;
+using namespace NYdb::NTable;
+
+Y_UNIT_TEST_SUITE(KqpDecimalColumnShard) {
+    class DecimalTestCase {
+    public:
+        DecimalTestCase()
+            : TestHelper(TKikimrSettings().SetWithSampleTables(false)) {
+            Schema = {
+                TTestHelper::TColumnSchema().SetName("id").SetType(NScheme::NTypeIds::Int32).SetNullable(false),
+                TTestHelper::TColumnSchema().SetName("int").SetType(NScheme::NTypeIds::Int64),
+                TTestHelper::TColumnSchema().SetName("dec").SetType(NScheme::NTypeIds::Decimal),
+            };
+
+            TestTable.SetName("/Root/ColumnTableTest").SetPrimaryKey({ "id" }).SetSharding({ "id" }).SetSchema(Schema);
+            TestHelper.CreateTable(TestTable);
+        }
+
+        TTestHelper::TUpdatesBuilder Inserter() {
+            return TTestHelper::TUpdatesBuilder(TestTable.GetArrowSchema(Schema));
+        }
+
+        void Upsert(TTestHelper::TUpdatesBuilder &inserter) {
+            TestHelper.BulkUpsert(TestTable, inserter);
+        }
+
+        void CheckQuery(const TString &query, const TString &expected) {
+            TestHelper.ReadData(query, expected);
+        }
+
+    private:
+        TTestHelper TestHelper;
+
+        TVector<TTestHelper::TColumnSchema> Schema;
+        TTestHelper::TColumnTable TestTable;
+    };
+
+    Y_UNIT_TEST(SimpleQueries) {
+        DecimalTestCase tester;
+
+        {
+            TTestHelper::TUpdatesBuilder inserter = tester.Inserter();
+            inserter.AddRow().Add(1).Add(4).Add(TDecimalValue("3.14"));
+            inserter.AddRow().Add(2).Add(3).Add(TDecimalValue("8.16"));
+            tester.Upsert(inserter);
+        }
+        tester.CheckQuery("SELECT * FROM `/Root/ColumnTableTest` WHERE id=1", "[[[\"3.14\"];1;[4]]]");
+
+        {
+            TTestHelper::TUpdatesBuilder inserter = tester.Inserter();
+            inserter.AddRow().Add(3).Add(2).Add(TDecimalValue("8.492"));
+            inserter.AddRow().Add(4).Add(1).Add(TDecimalValue("12.46"));
+            tester.Upsert(inserter);
+        }
+
+        tester.CheckQuery("SELECT * FROM `/Root/ColumnTableTest` order by id",
+            "[[[\"3.14\"];1;[4]];[[\"8.16\"];2;[3]];[[\"8.492\"];3;[2]];[[\"12.46\"];4;[1]]]");
+    }
+}
+
+}  // namespace NKqp
+}  // namespace NKikimr

ydb/core/kqp/ut/olap/ya.make

@@ -25,6 +25,7 @@ SRCS(
     write_ut.cpp
     sparsed_ut.cpp
     tiering_ut.cpp
+    decimal_ut.cpp
 )
 
 PEERDIR(

ydb/core/tx/columnshard/test_helper/columnshard_ut_common.h

@@ -476,11 +476,13 @@ namespace NKikimr::NColumnShard {
                 auto& builder = Owner.Builders[Index];
                 auto type = builder->type();
 
-                NArrow::SwitchType(type->id(), [&](const auto& t) {
+                Y_ABORT_UNLESS(NArrow::SwitchType(type->id(), [&](const auto& t) {
                     using TWrap = std::decay_t<decltype(t)>;
                     using T = typename TWrap::T;
                     using TBuilder = typename arrow::TypeTraits<typename TWrap::T>::BuilderType;
 
+                    AFL_NOTICE(NKikimrServices::TX_COLUMNSHARD)("T", typeid(T).name());
+
                     auto& typedBuilder = static_cast<TBuilder&>(*builder);
                     if constexpr (std::is_arithmetic<TData>::value) {
                         if constexpr (arrow::has_c_type<T>::value) {
@@ -495,9 +497,16 @@ namespace NKikimr::NColumnShard {
                             return true;
                         }
                     }
+
+                    if constexpr (std::is_same<TData, NYdb::TDecimalValue>::value) {
+                        if constexpr (arrow::is_decimal128_type<T>::value) {
+                            Y_ABORT_UNLESS(typedBuilder.Append(arrow::Decimal128(data.Hi_, data.Low_)).ok());
+                            return true;
+                        }
+                    }
                     Y_ABORT("Unknown type combination");
                     return false;
-                });
+                }));
                 return TRowBuilder(Index + 1, Owner);
             }