special json accessors (#13933)

write json as subcolumns accessor for fast way select columns data

Author: ivanmorozov333

.github/config/muted_ya.txt

@@ -28,6 +28,8 @@ ydb/core/kqp/ut/data_integrity KqpDataIntegrityTrails.Select
 ydb/core/kqp/ut/olap KqpDecimalColumnShard.TestAggregation
 ydb/core/kqp/ut/olap KqpDecimalColumnShard.TestFilterCompare
 ydb/core/kqp/ut/olap KqpOlap.TableSinkWithOlapStore
+ydb/core/kqp/ut/olap KqpOlapSysView.StatsSysViewBytesDictActualization
+ydb/core/kqp/ut/olap KqpOlapSysView.StatsSysViewBytesDictStatActualization
 ydb/core/kqp/ut/olap KqpOlapBlobsSharing.BlobsSharingSplit1_1
 ydb/core/kqp/ut/olap KqpOlapBlobsSharing.BlobsSharingSplit1_1_clean
 ydb/core/kqp/ut/olap KqpOlapBlobsSharing.BlobsSharingSplit1_1_clean_with_restarts

ydb/core/formats/arrow/accessor/abstract/constructor.h

@@ -1,8 +1,8 @@
 #pragma once
 
-#include <ydb/library/formats/arrow/protos/accessor.pb.h>
 #include <ydb/library/formats/arrow/accessor/abstract/accessor.h>
 #include <ydb/library/formats/arrow/accessor/common/chunk_data.h>
+#include <ydb/library/formats/arrow/protos/accessor.pb.h>
 #include <ydb/services/bg_tasks/abstract/interface.h>
 
 #include <library/cpp/object_factory/object_factory.h>
@@ -15,27 +15,33 @@ class IConstructor {
     using TProto = NKikimrArrowAccessorProto::TConstructor;
 
 private:
-    virtual TConclusion<std::shared_ptr<NArrow::NAccessor::IChunkedArray>> DoConstruct(
-        const std::shared_ptr<arrow::RecordBatch>& originalData, const TChunkConstructionData& externalInfo) const = 0;
-    virtual TConclusion<std::shared_ptr<NArrow::NAccessor::IChunkedArray>> DoConstructDefault(
-        const TChunkConstructionData& externalInfo) const = 0;
+    YDB_READONLY(IChunkedArray::EType, Type, IChunkedArray::EType::Undefined);
+
+    virtual TConclusion<std::shared_ptr<IChunkedArray>> DoDeserializeFromString(
+        const TString& originalData, const TChunkConstructionData& externalInfo) const = 0;
+    virtual TConclusion<std::shared_ptr<IChunkedArray>> DoConstructDefault(const TChunkConstructionData& externalInfo) const = 0;
     virtual NKikimrArrowAccessorProto::TConstructor DoSerializeToProto() const = 0;
     virtual bool DoDeserializeFromProto(const NKikimrArrowAccessorProto::TConstructor& proto) = 0;
-    virtual std::shared_ptr<arrow::Schema> DoGetExpectedSchema(const std::shared_ptr<arrow::Field>& resultColumn) const = 0;
     virtual TString DoDebugString() const {
         return "";
     }
     virtual bool DoIsEqualWithSameTypeTo(const IConstructor& item) const = 0;
-    virtual std::shared_ptr<arrow::RecordBatch> DoConstruct(
-        const std::shared_ptr<IChunkedArray>& columnData, const TChunkConstructionData& externalInfo) const = 0;
+    virtual TString DoSerializeToString(const std::shared_ptr<IChunkedArray>& columnData, const TChunkConstructionData& externalInfo) const = 0;
+
+    virtual TConclusion<std::shared_ptr<IChunkedArray>> DoConstruct(
+        const std::shared_ptr<NArrow::NAccessor::IChunkedArray>& originalArray, const TChunkConstructionData& externalInfo) const = 0;
 
 public:
+    IConstructor(const IChunkedArray::EType type)
+        : Type(type) {
+    }
+
     virtual ~IConstructor() = default;
 
-    std::shared_ptr<arrow::RecordBatch> Construct(
-        const std::shared_ptr<IChunkedArray>& columnData, const TChunkConstructionData& externalInfo) const {
+    TString SerializeToString(const std::shared_ptr<IChunkedArray>& columnData, const TChunkConstructionData& externalInfo) const {
         AFL_VERIFY(columnData);
-        return DoConstruct(columnData, externalInfo);
+        AFL_VERIFY(columnData->GetType() == Type)("column", columnData->GetType())("current", Type);
+        return DoSerializeToString(columnData, externalInfo);
     }
 
     bool IsEqualWithSameTypeTo(const IConstructor& item) const {
@@ -46,15 +52,25 @@ class IConstructor {
         return TStringBuilder() << GetClassName() << ":" << DoDebugString();
     }
 
-    TConclusion<std::shared_ptr<NArrow::NAccessor::IChunkedArray>> Construct(
-        const std::shared_ptr<arrow::RecordBatch>& originalData, const TChunkConstructionData& externalInfo) const {
-        return DoConstruct(originalData, externalInfo);
+    TConclusion<std::shared_ptr<IChunkedArray>> DeserializeFromString(
+        const TString& originalData, const TChunkConstructionData& externalInfo) const {
+        return DoDeserializeFromString(originalData, externalInfo);
     }
 
-    TConclusion<std::shared_ptr<NArrow::NAccessor::IChunkedArray>> ConstructDefault(const TChunkConstructionData& externalInfo) const {
+    TConclusion<std::shared_ptr<IChunkedArray>> ConstructDefault(const TChunkConstructionData& externalInfo) const {
         return DoConstructDefault(externalInfo);
     }
 
+    TConclusion<std::shared_ptr<IChunkedArray>> Construct(
+        const std::shared_ptr<IChunkedArray>& originalArray, const TChunkConstructionData& externalInfo) const {
+        AFL_VERIFY(originalArray);
+        if (originalArray->GetType() == GetType()) {
+            return originalArray;
+        } else {
+            return DoConstruct(originalArray, externalInfo);
+        }
+    }
+
     bool DeserializeFromProto(const NKikimrArrowAccessorProto::TConstructor& proto) {
         return DoDeserializeFromProto(proto);
     }
@@ -67,17 +83,13 @@ class IConstructor {
         proto = DoSerializeToProto();
     }
 
-    std::shared_ptr<arrow::Schema> GetExpectedSchema(const std::shared_ptr<arrow::Field>& resultColumn) const {
-        AFL_VERIFY(resultColumn);
-        return DoGetExpectedSchema(resultColumn);
-    }
-
     virtual TString GetClassName() const = 0;
 };
 
 class TConstructorContainer: public NBackgroundTasks::TInterfaceProtoContainer<IConstructor> {
 private:
     using TBase = NBackgroundTasks::TInterfaceProtoContainer<IConstructor>;
+
 public:
     using TBase::TBase;
 
@@ -94,9 +106,15 @@ class TConstructorContainer: public NBackgroundTasks::TInterfaceProtoContainer<I
         }
     }
 
-    std::shared_ptr<arrow::RecordBatch> Construct(const std::shared_ptr<IChunkedArray>& batch, const TChunkConstructionData& externalInfo) const {
+    TString SerializeToString(const std::shared_ptr<IChunkedArray>& batch, const TChunkConstructionData& externalInfo) const {
+        AFL_VERIFY(!!GetObjectPtr());
+        return GetObjectPtr()->SerializeToString(batch, externalInfo);
+    }
+
+    TConclusion<std::shared_ptr<IChunkedArray>> DeserializeFromString(
+        const TString& originalData, const TChunkConstructionData& externalInfo) const {
         AFL_VERIFY(!!GetObjectPtr());
-        return GetObjectPtr()->Construct(batch, externalInfo);
+        return GetObjectPtr()->DeserializeFromString(originalData, externalInfo);
     }
 
     static TConstructorContainer GetDefaultConstructor();

ydb/core/formats/arrow/accessor/composite_serial/accessor.h

@@ -1,12 +1,14 @@
 #pragma once
 #include <ydb/core/formats/arrow/save_load/loader.h>
+
 #include <ydb/library/formats/arrow/accessor/abstract/accessor.h>
+#include <ydb/library/formats/arrow/accessor/composite/accessor.h>
 
 namespace NKikimr::NArrow::NAccessor {
 
-class TDeserializeChunkedArray: public NArrow::NAccessor::IChunkedArray {
+class TDeserializeChunkedArray: public ICompositeChunkedArray {
 private:
-    using TBase = NArrow::NAccessor::IChunkedArray;
+    using TBase = ICompositeChunkedArray;
 
 public:
     class TChunk {
@@ -40,16 +42,19 @@ class TDeserializeChunkedArray: public NArrow::NAccessor::IChunkedArray {
     std::vector<TChunk> Chunks;
 
 protected:
+    virtual ui32 DoGetNullsCount() const override {
+        AFL_VERIFY(false);
+        return 0;
+    }
+    virtual ui32 DoGetValueRawBytes() const override {
+        AFL_VERIFY(false);
+        return 0;
+    }
+
     virtual TLocalChunkedArrayAddress DoGetLocalChunkedArray(
         const std::optional<TCommonChunkAddress>& chunkCurrent, const ui64 position) const override;
     virtual TLocalDataAddress DoGetLocalData(const std::optional<TCommonChunkAddress>& chunkCurrent, const ui64 position) const override;
 
-    virtual std::vector<TChunkedArraySerialized> DoSplitBySizes(
-        const TColumnSaver& /*saver*/, const TString& /*fullSerializedData*/, const std::vector<ui64>& /*splitSizes*/) override {
-        AFL_VERIFY(false);
-        return {};
-    }
-
     virtual std::shared_ptr<arrow::Scalar> DoGetScalar(const ui32 /*index*/) const override {
         AFL_VERIFY(false)("problem", "cannot use method");
         return nullptr;
@@ -61,7 +66,7 @@ class TDeserializeChunkedArray: public NArrow::NAccessor::IChunkedArray {
         AFL_VERIFY(false);
         return nullptr;
     }
-    virtual std::shared_ptr<arrow::ChunkedArray> DoGetChunkedArray() const override {
+    virtual std::shared_ptr<arrow::ChunkedArray> GetChunkedArray() const override {
         AFL_VERIFY(false);
         return nullptr;
     }

ydb/core/formats/arrow/accessor/plain/accessor.cpp

@@ -1,6 +1,7 @@
 #include "accessor.h"
 
 #include <ydb/core/formats/arrow/arrow_helpers.h>
+#include <ydb/core/formats/arrow/save_load/loader.h>
 #include <ydb/core/formats/arrow/size_calcer.h>
 #include <ydb/core/formats/arrow/splitter/simple.h>
 
@@ -10,24 +11,15 @@ std::optional<ui64> TTrivialArray::DoGetRawSize() const {
     return NArrow::GetArrayDataSize(Array);
 }
 
-std::vector<NKikimr::NArrow::NAccessor::TChunkedArraySerialized> TTrivialArray::DoSplitBySizes(
-    const TColumnSaver& saver, const TString& fullSerializedData, const std::vector<ui64>& splitSizes) {
-    auto schema = std::make_shared<arrow::Schema>(arrow::FieldVector({ std::make_shared<arrow::Field>("f", GetDataType()) }));
-    auto chunks = NArrow::NSplitter::TSimpleSplitter(saver).SplitBySizes(
-        arrow::RecordBatch::Make(schema, GetRecordsCount(), { Array }), fullSerializedData, splitSizes);
-    std::vector<TChunkedArraySerialized> result;
-    for (auto&& i : chunks) {
-        AFL_VERIFY(i.GetSlicedBatch()->num_columns() == 1);
-        result.emplace_back(std::make_shared<TTrivialArray>(i.GetSlicedBatch()->column(0)), i.GetSerializedChunk());
-    }
-    return result;
-}
-
 std::shared_ptr<arrow::Scalar> TTrivialArray::DoGetMaxScalar() const {
     auto minMaxPos = NArrow::FindMinMaxPosition(Array);
     return NArrow::TStatusValidator::GetValid(Array->GetScalar(minMaxPos.second));
 }
 
+ui32 TTrivialArray::DoGetValueRawBytes() const {
+    return NArrow::GetArrayDataSize(Array);
+}
+
 namespace {
 class TChunkAccessor {
 private:
@@ -85,4 +77,12 @@ std::shared_ptr<arrow::Scalar> TTrivialChunkedArray::DoGetMaxScalar() const {
     return result;
 }
 
+ui32 TTrivialChunkedArray::DoGetValueRawBytes() const {
+    ui32 result = 0;
+    for (auto&& i : Array->chunks()) {
+        result += NArrow::GetArrayDataSize(i);
+    }
+    return result;
+}
+
 }   // namespace NKikimr::NArrow::NAccessor

ydb/core/formats/arrow/accessor/plain/accessor.h

@@ -1,5 +1,6 @@
 #pragma once
 #include <ydb/library/formats/arrow/accessor/abstract/accessor.h>
+#include <ydb/library/formats/arrow/arrow_helpers.h>
 #include <ydb/library/formats/arrow/validation/validation.h>
 
 namespace NKikimr::NArrow::NAccessor {
@@ -12,24 +13,21 @@ class TTrivialArray: public IChunkedArray {
 protected:
     virtual std::optional<ui64> DoGetRawSize() const override;
 
-    virtual TLocalDataAddress DoGetLocalData(const std::optional<TCommonChunkAddress>& /*chunkCurrent*/, const ui64 /*position*/) const override {
+    virtual TLocalDataAddress DoGetLocalData(
+        const std::optional<TCommonChunkAddress>& /*chunkCurrent*/, const ui64 /*position*/) const override {
         return TLocalDataAddress(Array, 0, 0);
     }
-    virtual std::shared_ptr<arrow::ChunkedArray> DoGetChunkedArray() const override {
-        return std::make_shared<arrow::ChunkedArray>(Array);
-    }
     virtual std::shared_ptr<arrow::Scalar> DoGetScalar(const ui32 index) const override {
         return NArrow::TStatusValidator::GetValid(Array->GetScalar(index));
     }
     virtual std::shared_ptr<arrow::Scalar> DoGetMaxScalar() const override;
-    virtual std::vector<TChunkedArraySerialized> DoSplitBySizes(
-        const TColumnSaver& saver, const TString& fullSerializedData, const std::vector<ui64>& splitSizes) override;
-
-    virtual TLocalChunkedArrayAddress DoGetLocalChunkedArray(
-        const std::optional<TCommonChunkAddress>& /*chunkCurrent*/, const ui64 /*position*/) const override {
-        AFL_VERIFY(false);
-        return TLocalChunkedArrayAddress(nullptr, TCommonChunkAddress(0, GetRecordsCount(), 0));
+    virtual std::shared_ptr<IChunkedArray> DoISlice(const ui32 offset, const ui32 count) const override {
+        return std::make_shared<TTrivialArray>(Array->Slice(offset, count));
+    }
+    virtual ui32 DoGetNullsCount() const override {
+        return Array->null_count();
     }
+    virtual ui32 DoGetValueRawBytes() const override;
 
 public:
     const std::shared_ptr<arrow::Array>& GetArray() const {
@@ -40,6 +38,43 @@ class TTrivialArray: public IChunkedArray {
         : TBase(data->length(), EType::Array, data->type())
         , Array(data) {
     }
+
+    template <class TArrowDataType = arrow::StringType>
+    class TPlainBuilder {
+    private:
+        std::unique_ptr<arrow::ArrayBuilder> Builder;
+        std::optional<ui32> LastRecordIndex;
+
+    public:
+        TPlainBuilder(const ui32 reserveItems = 0, const ui32 reserveSize = 0) {
+            Builder = NArrow::MakeBuilder(arrow::TypeTraits<TArrowDataType>::type_singleton(), reserveItems, reserveSize);
+        }
+
+        void AddRecord(const ui32 recordIndex, const std::string_view value) {
+            if (LastRecordIndex) {
+                AFL_VERIFY(*LastRecordIndex < recordIndex)("last", LastRecordIndex)("index", recordIndex);
+                TStatusValidator::Validate(Builder->AppendNulls(recordIndex - *LastRecordIndex - 1));
+            } else {
+                TStatusValidator::Validate(Builder->AppendNulls(recordIndex));
+            }
+            LastRecordIndex = recordIndex;
+            AFL_VERIFY(NArrow::Append<TArrowDataType>(*Builder, arrow::util::string_view(value.data(), value.size())));
+        }
+
+        std::shared_ptr<IChunkedArray> Finish(const ui32 recordsCount) {
+            if (LastRecordIndex) {
+                AFL_VERIFY(*LastRecordIndex < recordsCount)("last", LastRecordIndex)("count", recordsCount);
+                TStatusValidator::Validate(Builder->AppendNulls(recordsCount - *LastRecordIndex - 1));
+            } else {
+                TStatusValidator::Validate(Builder->AppendNulls(recordsCount));
+            }
+            return std::make_shared<TTrivialArray>(NArrow::FinishBuilder(std::move(Builder)));
+        }
+    };
+
+    static TPlainBuilder<arrow::StringType> MakeBuilderUtf8(const ui32 reserveItems = 0, const ui32 reserveSize = 0) {
+        return TPlainBuilder<arrow::StringType>(reserveItems, reserveSize);
+    }
 };
 
 class TTrivialChunkedArray: public IChunkedArray {
@@ -48,28 +83,21 @@ class TTrivialChunkedArray: public IChunkedArray {
     const std::shared_ptr<arrow::ChunkedArray> Array;
 
 protected:
-    virtual TLocalDataAddress DoGetLocalData(const std::optional<TCommonChunkAddress>& chunkCurrent, const ui64 position) const override;
-    virtual std::shared_ptr<arrow::ChunkedArray> DoGetChunkedArray() const override {
-        return Array;
+    virtual ui32 DoGetValueRawBytes() const override;
+    virtual ui32 DoGetNullsCount() const override {
+        return Array->null_count();
     }
+    virtual TLocalDataAddress DoGetLocalData(const std::optional<TCommonChunkAddress>& chunkCurrent, const ui64 position) const override;
     virtual std::optional<ui64> DoGetRawSize() const override;
     virtual std::shared_ptr<arrow::Scalar> DoGetScalar(const ui32 index) const override {
         auto chunk = GetChunkSlow(index);
         return NArrow::TStatusValidator::GetValid(chunk.GetArray()->GetScalar(chunk.GetAddress().GetLocalIndex(index)));
     }
-    virtual std::vector<TChunkedArraySerialized> DoSplitBySizes(
-        const TColumnSaver& /*saver*/, const TString& /*fullSerializedData*/, const std::vector<ui64>& /*splitSizes*/) override {
-        AFL_VERIFY(false);
-        return {};
-    }
 
-    virtual std::shared_ptr<arrow::Scalar> DoGetMaxScalar() const override;
-
-    virtual TLocalChunkedArrayAddress DoGetLocalChunkedArray(
-        const std::optional<TCommonChunkAddress>& /*chunkCurrent*/, const ui64 /*position*/) const override {
-        AFL_VERIFY(false);
-        return TLocalChunkedArrayAddress(nullptr, TCommonChunkAddress(0, 0, 0));
+    virtual std::shared_ptr<IChunkedArray> DoISlice(const ui32 offset, const ui32 count) const override {
+        return std::make_shared<TTrivialChunkedArray>(Array->Slice(offset, count));
     }
+    virtual std::shared_ptr<arrow::Scalar> DoGetMaxScalar() const override;
 
 public:
     TTrivialChunkedArray(const std::shared_ptr<arrow::ChunkedArray>& data)