Merge 1b8c280 into bf5f3d7

Author: zverevgeny

.github/config/muted_ya.txt

@@ -8,7 +8,6 @@ ydb/core/external_sources *
 ydb/core/quoter/ut QuoterWithKesusTest.PrefetchCoefficient
 ydb/core/keyvalue/ut_trace TKeyValueTracingTest.*
 ydb/core/kqp/provider/ut KikimrIcGateway.TestLoadBasicSecretValueFromExternalDataSourceMetadata
-ydb/core/kqp/ut/olap KqpOlapIndexes.IndexesActualization
 ydb/core/kqp/ut/olap KqpOlapBlobsSharing.*
 ydb/core/kqp/ut/olap KqpOlap.ScanQueryOltpAndOlap
 ydb/core/kqp/ut/olap KqpOlapStatistics.StatsUsageWithTTL

ydb/core/formats/arrow/arrow_helpers.cpp

@@ -892,24 +892,33 @@ std::shared_ptr<arrow::RecordBatch> MergeColumns(const std::vector<std::shared_p
 }
 
 std::vector<std::shared_ptr<arrow::RecordBatch>> SliceToRecordBatches(const std::shared_ptr<arrow::Table>& t) {
-    std::set<ui32> splitPositions;
-    const ui32 numRows = t->num_rows();
-    for (auto&& i : t->columns()) {
-        ui32 pos = 0;
-        for (auto&& arr : i->chunks()) {
-            splitPositions.emplace(pos);
-            pos += arr->length();
+    if (!t->num_rows()) {
+        return {};
+    }
+    std::vector<ui32> positions;
+    {
+        for (auto&& i : t->columns()) {
+            ui32 pos = 0;
+            for (auto&& arr : i->chunks()) {
+                positions.emplace_back(pos);
+                pos += arr->length();
+            }
+            AFL_VERIFY(pos == t->num_rows());
         }
-        AFL_VERIFY(pos == t->num_rows());
+        positions.emplace_back(t->num_rows());
     }
+    std::sort(positions.begin(), positions.end());
+    positions.erase(std::unique(positions.begin(), positions.end()), positions.end());
+
     std::vector<std::vector<std::shared_ptr<arrow::Array>>> slicedData;
-    slicedData.resize(splitPositions.size());
-    std::vector<ui32> positions(splitPositions.begin(), splitPositions.end());
-    for (auto&& i : t->columns()) {
-        for (ui32 idx = 0; idx < positions.size(); ++idx) {
-            auto slice = i->Slice(positions[idx], ((idx + 1 == positions.size()) ? numRows : positions[idx + 1]) - positions[idx]);
-            AFL_VERIFY(slice->num_chunks() == 1);
-            slicedData[idx].emplace_back(slice->chunks().front());
+    slicedData.resize(positions.size() - 1);
+    {
+        for (auto&& i : t->columns()) {
+            for (ui32 idx = 0; idx + 1 < positions.size(); ++idx) {
+                auto slice = i->Slice(positions[idx], positions[idx + 1] - positions[idx]);
+                AFL_VERIFY(slice->num_chunks() == 1);
+                slicedData[idx].emplace_back(slice->chunks().front());
+            }
         }
     }
     std::vector<std::shared_ptr<arrow::RecordBatch>> result;

ydb/core/formats/arrow/program.cpp

@@ -946,18 +946,18 @@ arrow::Result<std::shared_ptr<NArrow::TColumnFilter>> TProgramStep::BuildFilter(
 }
 
 const std::set<ui32>& TProgramStep::GetFilterOriginalColumnIds() const {
-    AFL_VERIFY(IsFilterOnly());
+//    AFL_VERIFY(IsFilterOnly());
     return FilterOriginalColumnIds;
 }
 
 std::set<std::string> TProgram::GetEarlyFilterColumns() const {
     std::set<std::string> result;
     for (ui32 i = 0; i < Steps.size(); ++i) {
+        auto stepFields = Steps[i]->GetColumnsInUsage(true);
+        result.insert(stepFields.begin(), stepFields.end());
         if (!Steps[i]->IsFilterOnly()) {
             break;
         }
-        auto stepFields = Steps[i]->GetColumnsInUsage();
-        result.insert(stepFields.begin(), stepFields.end());
     }
     return result;
 }

ydb/core/formats/arrow/scalar/serialization.cpp

@@ -0,0 +1,47 @@
+#include "serialization.h"
+#include <ydb/core/formats/arrow/switch/switch_type.h>
+#include <ydb/library/actors/core/log.h>
+
+namespace NKikimr::NArrow::NScalar {
+
+TConclusion<TString> TSerializer::SerializePayloadToString(const std::shared_ptr<arrow::Scalar>& scalar) {
+    TString resultString;
+    const bool resultFlag = NArrow::SwitchType(scalar->type->id(), [&](const auto& type) {
+        using TWrap = std::decay_t<decltype(type)>;
+        if constexpr (arrow::has_c_type<typename TWrap::T>()) {
+            using CType = typename TWrap::T::c_type;
+            using ScalarType = typename arrow::TypeTraits<typename TWrap::T>::ScalarType;
+            const ScalarType* scalarTyped = static_cast<const ScalarType*>(scalar.get());
+            resultString = TString(sizeof(CType), '\0');
+            memcpy(&resultString[0], scalarTyped->data(), sizeof(CType));
+            return true;
+        }
+        return false;
+    });
+    if (!resultFlag) {
+        return TConclusionStatus::Fail("incorrect scalar type for payload serialization: " + scalar->type->ToString());
+    }
+    return resultString;
+}
+
+TConclusion<std::shared_ptr<arrow::Scalar>> TSerializer::DeserializeFromStringWithPayload(const TString& data, const std::shared_ptr<arrow::DataType>& dataType) {
+    AFL_VERIFY(dataType);
+    std::shared_ptr<arrow::Scalar> result;
+    const bool resultFlag = NArrow::SwitchType(dataType->id(), [&](const auto& type) {
+        using TWrap = std::decay_t<decltype(type)>;
+        if constexpr (arrow::has_c_type<typename TWrap::T>()) {
+            using CType = typename TWrap::T::c_type;
+            AFL_VERIFY(data.size() == sizeof(CType));
+            using ScalarType = typename arrow::TypeTraits<typename TWrap::T>::ScalarType;
+            result = std::make_shared<ScalarType>(*(CType*)&data[0], dataType);
+            return true;
+        }
+        return false;
+    });
+    if (!resultFlag) {
+        return TConclusionStatus::Fail("incorrect scalar type for payload deserialization: " + dataType->ToString());
+    }
+    return result;
+}
+
+}

ydb/core/formats/arrow/scalar/serialization.h

@@ -0,0 +1,15 @@
+#pragma once
+#include <ydb/library/conclusion/result.h>
+
+#include <contrib/libs/apache/arrow/cpp/src/arrow/scalar.h>
+#include <contrib/libs/apache/arrow/cpp/src/arrow/type.h>
+
+#include <util/generic/string.h>
+
+namespace NKikimr::NArrow::NScalar {
+class TSerializer {
+public:
+    static TConclusion<TString> SerializePayloadToString(const std::shared_ptr<arrow::Scalar>& scalar);
+    static TConclusion<std::shared_ptr<arrow::Scalar>> DeserializeFromStringWithPayload(const TString& data, const std::shared_ptr<arrow::DataType>& dataType);
+};
+}

ydb/core/formats/arrow/scalar/ya.make

@@ -0,0 +1,14 @@
+LIBRARY()
+
+PEERDIR(
+    contrib/libs/apache/arrow
+    ydb/library/conclusion
+    ydb/core/formats/arrow/switch
+    ydb/library/actors/core
+)
+
+SRCS(
+    serialization.cpp
+)
+
+END()

ydb/core/formats/arrow/size_calcer.cpp

@@ -242,11 +242,12 @@ ui64 GetArrayDataSize(const std::shared_ptr<arrow::Array>& column) {
 }
 
 NKikimr::NArrow::TSerializedBatch TSerializedBatch::Build(std::shared_ptr<arrow::RecordBatch> batch, const TBatchSplitttingContext& context) {
-    std::optional<TFirstLastSpecialKeys> specialKeys;
+    std::optional<TString> specialKeys;
     if (context.GetFieldsForSpecialKeys().size()) {
-        specialKeys = TFirstLastSpecialKeys(batch, context.GetFieldsForSpecialKeys());
+        specialKeys = TFirstLastSpecialKeys(batch, context.GetFieldsForSpecialKeys()).SerializeToString();
     }
-    return TSerializedBatch(NArrow::SerializeSchema(*batch->schema()), NArrow::SerializeBatchNoCompression(batch), batch->num_rows(), NArrow::GetBatchDataSize(batch), specialKeys);
+    return TSerializedBatch(NArrow::SerializeSchema(*batch->schema()), NArrow::SerializeBatchNoCompression(batch), batch->num_rows(), 
+        NArrow::GetBatchDataSize(batch), specialKeys);
 }
 
 TConclusionStatus TSerializedBatch::BuildWithLimit(std::shared_ptr<arrow::RecordBatch> batch, const TBatchSplitttingContext& context, std::optional<TSerializedBatch>& sbL, std::optional<TSerializedBatch>& sbR) {

ydb/core/formats/arrow/size_calcer.h

@@ -74,13 +74,13 @@ class TSerializedBatch {
     YDB_READONLY_DEF(TString, Data);
     YDB_READONLY(ui32, RowsCount, 0);
     YDB_READONLY(ui32, RawBytes, 0);
-    std::optional<TFirstLastSpecialKeys> SpecialKeys;
+    std::optional<TString> SpecialKeys;
 public:
     size_t GetSize() const {
         return Data.size();
     }
 
-    const TFirstLastSpecialKeys& GetSpecialKeysSafe() const {
+    const TString& GetSpecialKeysSafe() const {
         AFL_VERIFY(SpecialKeys);
         return *SpecialKeys;
     }
@@ -95,7 +95,7 @@ class TSerializedBatch {
     static TConclusionStatus BuildWithLimit(std::shared_ptr<arrow::RecordBatch> batch, const TBatchSplitttingContext& context, std::optional<TSerializedBatch>& sbL, std::optional<TSerializedBatch>& sbR);
     static TSerializedBatch Build(std::shared_ptr<arrow::RecordBatch> batch, const TBatchSplitttingContext& context);
 
-    TSerializedBatch(TString&& schemaData, TString&& data, const ui32 rowsCount, const ui32 rawBytes, const std::optional<TFirstLastSpecialKeys>& specialKeys)
+    TSerializedBatch(TString&& schemaData, TString&& data, const ui32 rowsCount, const ui32 rawBytes, const std::optional<TString>& specialKeys)
         : SchemaData(schemaData)
         , Data(data)
         , RowsCount(rowsCount)

ydb/core/formats/arrow/ya.make

@@ -12,6 +12,7 @@ PEERDIR(
     ydb/core/formats/arrow/dictionary
     ydb/core/formats/arrow/transformer
     ydb/core/formats/arrow/reader
+    ydb/core/formats/arrow/scalar
     ydb/core/formats/arrow/hash
     ydb/library/actors/core
     ydb/library/arrow_kernels

ydb/core/kqp/gateway/behaviour/tablestore/operations/alter_sharding.cpp

@@ -1,4 +1,5 @@
 #include "alter_sharding.h"
+#include <ydb/library/actors/core/log.h>
 #include <util/string/type.h>
 
 namespace NKikimr::NKqp {
@@ -26,4 +27,8 @@ void TAlterShardingOperation::DoSerializeScheme(NKikimrSchemeOp::TModifyScheme&
     scheme.MutableAlterColumnTable()->MutableReshardColumnTable()->SetIncrease(*Increase);
 }
 
+void TAlterShardingOperation::DoSerializeScheme(NKikimrSchemeOp::TAlterColumnTableSchema& /*scheme*/) const {
+    AFL_VERIFY(false);
+}
+
 }