Embed short blobs in TCell, fix empty string (#13194)

Author: zverevgeny

ydb/core/kqp/runtime/kqp_write_table.cpp

@@ -278,10 +278,14 @@ class TRowBuilder {
             ? NYql::NUdf::TStringRef(cellInfo.PgBinaryValue)
             : cellInfo.Value.AsStringRef();
 
-        char* initialPtr = dataPtr;
-        std::memcpy(initialPtr, ref.Data(), ref.Size());
-        dataPtr += ref.Size();
-        return TCell(initialPtr, ref.Size());
+        if (TCell::CanInline(ref.Size())) {
+            return TCell(ref.Data(), ref.Size());
+        } else {
+            char* initialPtr = dataPtr;
+            std::memcpy(initialPtr, ref.Data(), ref.Size());
+            dataPtr += ref.Size();
+            return TCell(initialPtr, ref.Size());
+        }
     }
 
     size_t GetCellSize(const TCellInfo& cellInfo) const {
@@ -301,7 +305,8 @@ class TRowBuilder {
         if (cellInfo.Type.GetTypeId() == NScheme::NTypeIds::Pg) {
             return cellInfo.PgBinaryValue.size();
         }
-        return cellInfo.Value.AsStringRef().Size();
+        const auto s = cellInfo.Value.AsStringRef().Size();
+        return TCell::CanInline(s) ? 0 : s;
     }
 
     TCharVectorPtr Allocate(size_t size) {

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

@@ -3122,6 +3122,44 @@ Y_UNIT_TEST_SUITE(KqpOlap) {
             )", FormatResultSetYson(resultSet));
         }
     }
+
+    Y_UNIT_TEST(InsertEmptyString) {
+        NKikimrConfig::TAppConfig appConfig;
+        appConfig.MutableColumnShardConfig()->SetAllowNullableColumnsInPK(true);
+        auto settings = TKikimrSettings().SetAppConfig(appConfig).SetWithSampleTables(false);
+        TTestHelper testHelper(settings);
+
+        TVector<TTestHelper::TColumnSchema> schema = {
+            TTestHelper::TColumnSchema().SetName("id").SetType(NScheme::NTypeIds::Int64).SetNullable(false),
+            TTestHelper::TColumnSchema().SetName("value").SetType(NScheme::NTypeIds::String).SetNullable(false),
+        };
+        TTestHelper::TColumnTable testTable;
+        testTable.SetName("/Root/ttt").SetPrimaryKey({ "id", }).SetSharding({ "id" }).SetSchema(schema);
+        testHelper.CreateTable(testTable);
+        auto client = testHelper.GetKikimr().GetQueryClient();
+        const auto result = client
+            .ExecuteQuery(
+                R"(
+                 INSERT INTO `/Root/ttt` (id, value) VALUES
+                 (347, '')
+                )",
+                NYdb::NQuery::TTxControl::BeginTx().CommitTx())
+            .GetValueSync();
+        UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString());
+        {
+            const auto resultSelect = client
+                .ExecuteQuery(
+                    "SELECT * FROM `/Root/ttt`",
+                    NYdb::NQuery::TTxControl::BeginTx().CommitTx())
+                .GetValueSync();
+            UNIT_ASSERT_C(resultSelect.IsSuccess(), resultSelect.GetIssues().ToString());
+            const auto resultSets = resultSelect.GetResultSets();
+            UNIT_ASSERT_VALUES_EQUAL(resultSets.size(), 1);
+            const auto resultSet = resultSets[0];
+            UNIT_ASSERT_VALUES_EQUAL(resultSet.RowsCount(), 1);
+        }
+    }
+
 }
 
 }