Merge pull request #17452 from dorooleg/stable-25-1-analytics-cs-main-nikvas0

merge main to stable-25-1-analytics

Author: maximyurchuk

.github/config/muted_ya.txt

@@ -20,25 +20,6 @@ ydb/core/keyvalue/ut_trace TKeyValueTracingTest.WriteSmall
 ydb/core/kqp/ut/cost KqpCost.OlapWriteRow
 ydb/core/kqp/ut/olap KqpDecimalColumnShard.TestAggregation
 ydb/core/kqp/ut/olap KqpDecimalColumnShard.TestFilterCompare
-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/kqp/ut/olap KqpOlapBlobsSharing.BlobsSharingSplit1_3_1
-ydb/core/kqp/ut/olap KqpOlapBlobsSharing.BlobsSharingSplit1_3_2_1_clean
-ydb/core/kqp/ut/olap KqpOlapBlobsSharing.BlobsSharingSplit3_1
-ydb/core/kqp/ut/olap KqpOlapBlobsSharing.ChangeSchemaAndSplit
-ydb/core/kqp/ut/olap KqpOlapBlobsSharing.HugeSchemeHistory
-ydb/core/kqp/ut/olap KqpOlapBlobsSharing.MultipleMerge
-ydb/core/kqp/ut/olap KqpOlapBlobsSharing.MultipleMergesWithRestartsAfterWait
-ydb/core/kqp/ut/olap KqpOlapBlobsSharing.MultipleMergesWithRestartsWhenWait
-ydb/core/kqp/ut/olap KqpOlapBlobsSharing.MultipleSchemaVersions
-ydb/core/kqp/ut/olap KqpOlapBlobsSharing.MultipleSplits
-ydb/core/kqp/ut/olap KqpOlapBlobsSharing.MultipleSplitsThenMerges
-ydb/core/kqp/ut/olap KqpOlapBlobsSharing.MultipleSplitsWithRestartsAfterWait
-ydb/core/kqp/ut/olap KqpOlapBlobsSharing.MultipleSplitsWithRestartsWhenWait
-ydb/core/kqp/ut/olap KqpOlapBlobsSharing.TableReshardingConsistency64
-ydb/core/kqp/ut/olap KqpOlapBlobsSharing.TableReshardingModuloN
-ydb/core/kqp/ut/olap KqpOlapBlobsSharing.UpsertWhileSplitTest
 ydb/core/kqp/ut/olap KqpOlapJson.BloomIndexesVariants
 ydb/core/kqp/ut/olap KqpOlapSysView.StatsSysViewBytesDictActualization
 ydb/core/kqp/ut/olap KqpOlapSysView.StatsSysViewBytesDictStatActualization
@@ -47,6 +28,9 @@ ydb/core/kqp/ut/olap [*/*] chunk chunk
 ydb/core/kqp/ut/query KqpAnalyze.AnalyzeTable+ColumnStore
 ydb/core/kqp/ut/query KqpAnalyze.AnalyzeTable-ColumnStore
 ydb/core/kqp/ut/query KqpStats.SysViewClientLost
+ydb/core/kqp/ut/query KqpLimits.OutOfSpaceYQLUpsertFail+useSink
+ydb/core/kqp/ut/query KqpLimits.QSReplySizeEnsureMemoryLimits+useSink
+ydb/core/kqp/ut/query KqpStats.SysViewClientLost
 ydb/core/kqp/ut/scheme KqpOlapScheme.TenThousandColumns
 ydb/core/kqp/ut/scheme KqpScheme.AlterAsyncReplication
 ydb/core/kqp/ut/scheme [*/*] chunk chunk
@@ -163,6 +147,13 @@ ydb/tests/functional/tpc/large test_tpcds.py.TestTpcdsS1.test_tpcds[36]
 ydb/tests/functional/tpc/large test_tpcds.py.TestTpcdsS1.test_tpcds[67]
 ydb/tests/functional/tpc/large test_tpcds.py.TestTpcdsS1.test_tpcds[86]
 ydb/tests/functional/tpc/large test_tpcds.py.TestTpcdsS1.test_tpcds[9]
+ydb/tests/functional/tpc/large test_tpch_spilling.py.TestTpchSpillingS10.test_tpch[7]
+ydb/tests/olap sole chunk chunk
+ydb/tests/olap test_quota_exhaustion.py.TestYdbWorkload.test_delete
+ydb/tests/olap/data_quotas test_quota_exhaustion.py.TestYdbWorkload.test_duplicates
+ydb/tests/olap/column_family/compression alter_compression.py.TestAlterCompression.test_all_supported_compression
+ydb/tests/olap/column_family/compression sole chunk chunk
+ydb/tests/olap/oom overlapping_portions.py.TestOverlappingPortions.test
 ydb/tests/olap/scenario sole chunk chunk
 ydb/tests/olap/scenario test_alter_compression.py.TestAlterCompression.test[alter_compression]
 ydb/tests/olap/scenario test_alter_tiering.py.TestAlterTiering.test[many_tables]
@@ -172,6 +163,7 @@ ydb/tests/olap/ttl_tiering data_migration_when_alter_ttl.py.TestDataMigrationWhe
 ydb/tests/olap/ttl_tiering sole chunk chunk
 ydb/tests/olap/ttl_tiering ttl_delete_s3.py.TestDeleteS3Ttl.test_data_unchanged_after_ttl_change
 ydb/tests/olap/ttl_tiering ttl_delete_s3.py.TestDeleteS3Ttl.test_ttl_delete
+ydb/tests/olap/ttl_tiering ttl_delete_s3.py.TestDeleteS3Ttl.test_delete_s3_tiering
 ydb/tests/olap/ttl_tiering ttl_unavailable_s3.py.TestUnavailableS3.test
 ydb/tests/olap/ttl_tiering ttl_delete_s3.py.TestDeleteS3Ttl.test_delete_s3_tiering
 ydb/tests/olap/ttl_tiering unstable_connection.py.TestUnstableConnection.test

ydb/core/driver_lib/run/run.cpp

@@ -149,6 +149,7 @@
 #include <ydb/library/actors/util/memory_track.h>
 #include <ydb/library/actors/prof/tag.h>
 #include <ydb/library/security/ydb_credentials_provider_factory.h>
+#include <ydb/library/signal_backtrace/signal_backtrace.h>
 #include <yql/essentials/minikql/invoke_builtins/mkql_builtins.h>
 
 #include <util/charset/wide.h>
@@ -1939,6 +1940,10 @@ void TKikimrRunner::SetSignalHandlers() {
     signal(SIGINT, &TKikimrRunner::OnTerminate);
     signal(SIGTERM, &TKikimrRunner::OnTerminate);
 
+    if (IsTrue(GetEnv("YDB_ENABLE_SIGNAL_BACKTRACE"))) {
+        Singleton<TTraceCollector>(TTraceCollector::DEFAULT_SIGNALS);
+    }
+
 #if !defined(_win_)
     SetAsyncSignalHandler(SIGHUP, [](int) {
         TLogBackend::ReopenAllBackends();

ydb/core/driver_lib/run/ya.make

@@ -138,6 +138,7 @@ PEERDIR(
     ydb/library/grpc/server/actors
     ydb/library/pdisk_io
     ydb/library/security
+    ydb/library/signal_backtrace
     ydb/library/yql/providers/pq/cm_client
     ydb/library/yql/providers/s3/actors
     ydb/public/lib/base

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

@@ -24,26 +24,7 @@ std::shared_ptr<arrow::Array> IChunkedArray::TReader::CopyRecord(const ui64 reco
 }
 
 std::shared_ptr<arrow::ChunkedArray> IChunkedArray::Slice(const ui32 offset, const ui32 count) const {
-    AFL_VERIFY(offset + count <= (ui64)GetRecordsCount())("offset", offset)("count", count)("length", GetRecordsCount());
-    ui32 currentOffset = offset;
-    ui32 countLeast = count;
-    std::vector<std::shared_ptr<arrow::Array>> chunks;
-    auto address = GetChunkSlow(offset);
-    while (countLeast) {
-        address = GetChunk(address.GetAddress(), currentOffset);
-        const ui64 internalPos = address.GetAddress().GetLocalIndex(currentOffset);
-        if (internalPos + countLeast <= (ui64)address.GetArray()->length()) {
-            chunks.emplace_back(address.GetArray()->Slice(internalPos, countLeast));
-            break;
-        } else {
-            const ui32 deltaCount = address.GetArray()->length() - internalPos;
-            chunks.emplace_back(address.GetArray()->Slice(internalPos, deltaCount));
-            AFL_VERIFY(countLeast >= deltaCount);
-            countLeast -= deltaCount;
-            currentOffset += deltaCount;
-        }
-    }
-    return std::make_shared<arrow::ChunkedArray>(chunks, DataType);
+    return GetChunkedArray(TColumnConstructionContext().SetStartIndex(offset).SetRecordsCount(count));
 }
 
 IChunkedArray::TFullDataAddress IChunkedArray::GetChunk(const std::optional<TAddressChain>& chunkCurrent, const ui64 position) const {
@@ -62,10 +43,10 @@ IChunkedArray::TFullDataAddress IChunkedArray::GetChunk(const std::optional<TAdd
         return TFullDataAddress(localAddress.GetArray(), std::move(addressChain));
     } else {
         auto chunkedArrayAddress = GetArray(chunkCurrent, position, nullptr);
-        if (chunkCurrent) {
-            AFL_VERIFY(chunkCurrent->GetSize() == 1 + chunkedArrayAddress.GetAddress().GetSize())("current", chunkCurrent->GetSize())(
-                                                      "chunked", chunkedArrayAddress.GetAddress().GetSize());
-        }
+        //        if (chunkCurrent) {
+        //            AFL_VERIFY(chunkCurrent->GetSize() == chunkedArrayAddress.GetAddress().GetSize())("current", chunkCurrent->GetSize())(
+        //                                                      "chunked", chunkedArrayAddress.GetAddress().GetSize());
+        //        }
         auto localAddress = chunkedArrayAddress.GetArray()->GetLocalData(address, chunkedArrayAddress.GetAddress().GetLocalIndex(position));
         auto fullAddress = std::move(chunkedArrayAddress.MutableAddress());
         fullAddress.Add(localAddress.GetAddress());
@@ -112,7 +93,7 @@ std::shared_ptr<IChunkedArray> IChunkedArray::DoApplyFilter(const TColumnFilter&
     auto schema = std::make_shared<arrow::Schema>(fields);
     auto table = arrow::Table::Make(schema, { arr }, GetRecordsCount());
     AFL_VERIFY(table->num_columns() == 1);
-    AFL_VERIFY(filter.Apply(table));
+    filter.Apply(table);
     if (table->column(0)->num_chunks() == 1) {
         return std::make_shared<TTrivialArray>(table->column(0)->chunk(0));
     } else {
@@ -121,8 +102,8 @@ std::shared_ptr<IChunkedArray> IChunkedArray::DoApplyFilter(const TColumnFilter&
 }
 
 std::shared_ptr<IChunkedArray> IChunkedArray::ApplyFilter(const TColumnFilter& filter, const std::shared_ptr<IChunkedArray>& selfPtr) const {
-    AFL_VERIFY(selfPtr);
     if (filter.IsTotalAllowFilter()) {
+        AFL_VERIFY(selfPtr);
         return selfPtr;
     }
     if (filter.IsTotalDenyFilter()) {
@@ -134,7 +115,7 @@ std::shared_ptr<IChunkedArray> IChunkedArray::ApplyFilter(const TColumnFilter& f
     return result;
 }
 
-std::shared_ptr<arrow::ChunkedArray> IChunkedArray::GetChunkedArray() const {
+std::shared_ptr<arrow::ChunkedArray> IChunkedArray::GetChunkedArrayTrivial() const {
     std::vector<std::shared_ptr<arrow::Array>> chunks;
     std::optional<TFullDataAddress> address;
     for (ui32 position = 0; position < GetRecordsCount();) {
@@ -145,6 +126,23 @@ std::shared_ptr<arrow::ChunkedArray> IChunkedArray::GetChunkedArray() const {
     return std::make_shared<arrow::ChunkedArray>(chunks, GetDataType());
 }
 
+std::shared_ptr<arrow::ChunkedArray> IChunkedArray::GetChunkedArray(const TColumnConstructionContext& context) const {
+    if (context.GetStartIndex() || context.GetRecordsCount()) {
+        const ui32 start = context.GetStartIndex().value_or(0);
+        const ui32 count = context.GetRecordsCount().value_or(GetRecordsCount() - start);
+        auto slice = ISlice(start, count);
+        if (context.GetFilter() && !context.GetFilter()->IsTotalAllowFilter()) {
+            return slice->ApplyFilter(context.GetFilter()->Slice(start, count), slice)->GetChunkedArrayTrivial();
+        } else {
+            return slice->GetChunkedArrayTrivial();
+        }
+    } else if (context.GetFilter() && !context.GetFilter()->IsTotalAllowFilter()) {
+        return ApplyFilter(*context.GetFilter(), nullptr)->GetChunkedArrayTrivial();
+    } else {
+        return GetChunkedArrayTrivial();
+    }
+}
+
 TString IChunkedArray::TReader::DebugString(const ui32 position) const {
     auto address = GetReadChunk(position);
     return NArrow::DebugString(address.GetArray(), address.GetPosition());

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

@@ -1,5 +1,7 @@
 #pragma once
 
+#include "common.h"
+
 #include <ydb/library/accessor/accessor.h>
 #include <ydb/library/accessor/validator.h>
 #include <ydb/library/formats/arrow/splitter/similar_packer.h>
@@ -264,6 +266,8 @@ class IChunkedArray {
     virtual void DoVisitValues(const TValuesSimpleVisitor& visitor) const = 0;
 
 protected:
+    virtual std::shared_ptr<arrow::ChunkedArray> GetChunkedArrayTrivial() const;
+
     std::shared_ptr<arrow::Schema> GetArraySchema() const {
         const arrow::FieldVector fields = { std::make_shared<arrow::Field>("val", GetDataType()) };
         return std::make_shared<arrow::Schema>(fields);
@@ -344,7 +348,6 @@ class IChunkedArray {
     }
 
     virtual void Reallocate() {
-
     }
 
     void VisitValues(const TValuesSimpleVisitor& visitor) const {
@@ -444,7 +447,8 @@ class IChunkedArray {
         return *result;
     }
 
-    virtual std::shared_ptr<arrow::ChunkedArray> GetChunkedArray() const;
+    virtual std::shared_ptr<arrow::ChunkedArray> GetChunkedArray(
+        const TColumnConstructionContext& context = Default<TColumnConstructionContext>()) const;
     virtual ~IChunkedArray() = default;
 
     std::shared_ptr<arrow::ChunkedArray> Slice(const ui32 offset, const ui32 count) const;

ydb/core/formats/arrow/accessor/abstract/common.cpp

@@ -0,0 +1,31 @@
+#include "common.h"
+
+#include <ydb/core/formats/arrow/arrow_filter.h>
+
+namespace NKikimr::NArrow::NAccessor {
+
+TColumnConstructionContext& TColumnConstructionContext::SetFilter(const std::shared_ptr<TColumnFilter>& val) {
+    if (!val || val->IsTotalAllowFilter()) {
+        Filter = nullptr;
+    } else {
+        Filter = val;
+    }
+    return *this;
+}
+
+std::optional<TColumnConstructionContext> TColumnConstructionContext::Slice(const ui32 offset, const ui32 count) const {
+    std::optional<TColumnConstructionContext> result;
+    const ui32 start = std::max<ui32>(offset, StartIndex.value_or(0));
+    const ui32 finish = std::min<ui32>(offset + count, StartIndex.value_or(0) + RecordsCount.value_or(offset + count));
+    if (finish <= start) {
+        result = std::nullopt;
+    } else {
+        result = TColumnConstructionContext().SetStartIndex(start - offset).SetRecordsCount(finish - start, count);
+    }
+    if (result && Filter) {
+        result->SetFilter(std::make_shared<TColumnFilter>(Filter->Slice(offset, count)));
+    }
+    return result;
+}
+
+}   // namespace NKikimr::NArrow::NAccessor

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

@@ -0,0 +1,47 @@
+#pragma once
+#include <ydb/library/accessor/accessor.h>
+
+#include <util/system/types.h>
+
+#include <memory>
+#include <optional>
+
+namespace NKikimr::NArrow {
+class TColumnFilter;
+}
+
+namespace NKikimr::NArrow::NAccessor {
+
+class TColumnConstructionContext {
+private:
+    YDB_READONLY_DEF(std::optional<ui32>, StartIndex);
+    YDB_ACCESSOR_DEF(std::optional<ui32>, RecordsCount);
+    YDB_READONLY_DEF(std::shared_ptr<TColumnFilter>, Filter);
+
+public:
+    TColumnConstructionContext& SetRecordsCount(const ui32 recordsCount, const ui32 defValue) {
+        if (recordsCount == defValue) {
+            RecordsCount.reset();
+        } else {
+            RecordsCount = recordsCount;
+        }
+        return *this;
+    }
+
+    TColumnConstructionContext& SetStartIndex(const ui32 startIndex) {
+        if (startIndex) {
+            StartIndex = startIndex;
+        } else {
+            StartIndex.reset();
+        }
+        return *this;
+    }
+
+    TColumnConstructionContext& SetFilter(const std::shared_ptr<TColumnFilter>& val);
+
+    std::optional<TColumnConstructionContext> Slice(const ui32 offset, const ui32 count) const;
+
+    TColumnConstructionContext() = default;
+};
+
+}   // namespace NKikimr::NArrow::NAccessor

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

@@ -33,6 +33,10 @@ class IConstructor {
         const std::shared_ptr<NArrow::NAccessor::IChunkedArray>& originalArray, const TChunkConstructionData& externalInfo) const = 0;
 
 public:
+    virtual bool HasInternalConversion() const {
+        return false;
+    }
+
     IConstructor(const IChunkedArray::EType type)
         : Type(type) {
     }

ydb/core/formats/arrow/accessor/abstract/ya.make

@@ -10,6 +10,7 @@ PEERDIR(
 )
 
 SRCS(
+    common.cpp
     constructor.cpp
     request.cpp
     accessor.cpp

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

@@ -110,4 +110,24 @@ std::optional<bool> TCompositeChunkedArray::DoCheckOneValueAccessor(std::shared_
     return true;
 }
 
+std::shared_ptr<arrow::ChunkedArray> TCompositeChunkedArray::GetChunkedArray(const TColumnConstructionContext& context) const {
+    ui32 pos = 0;
+    std::vector<std::shared_ptr<arrow::Array>> chunks;
+    for (auto&& i : Chunks) {
+        auto sliceCtx = context.Slice(pos, i->GetRecordsCount());
+        if (!sliceCtx) {
+            if (chunks.size()) {
+                break;
+            } else {
+                pos += i->GetRecordsCount();
+                continue;
+            }
+        }
+        std::shared_ptr<arrow::ChunkedArray> arr = i->GetChunkedArray(*sliceCtx);
+        chunks.insert(chunks.end(), arr->chunks().begin(), arr->chunks().end());
+        pos += i->GetRecordsCount();
+    }
+    return std::make_shared<arrow::ChunkedArray>(std::move(chunks));
+}
+
 }   // namespace NKikimr::NArrow::NAccessor

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

@@ -22,6 +22,8 @@ class TCompositeChunkedArray: public ICompositeChunkedArray {
 private:
     YDB_READONLY_DEF(std::vector<std::shared_ptr<IChunkedArray>>, Chunks);
 
+    virtual std::shared_ptr<arrow::ChunkedArray> GetChunkedArray(const TColumnConstructionContext& context) const override;
+
     virtual void DoVisitValues(const TValuesSimpleVisitor& visitor) const override {
         for (auto&& i : Chunks) {
             i->VisitValues(visitor);