Merge stable-24-1-16-analytics into stable-24-1 (#5596)

Co-authored-by: Vitalii Gridnev <gridnevvvit@gmail.com> Co-authored-by: Andrey Neporada <aneporada@ydb.tech> Co-authored-by: niksaveliev <nik@saveliev.me> Co-authored-by: Sergey Veselov <sergeyveselov@ydb.tech> Co-authored-by: alexnick88 <alexnick@ydb.tech> Co-authored-by: Ilnaz Nizametdinov <ilnaz@ydb.tech> Co-authored-by: Iuliia Sidorina <yulia@ydb.tech> Co-authored-by: kungurtsev <kungasc@ydb.tech> Co-authored-by: Nikolay Shestakov <tesseract@ydb.tech> Co-authored-by: azevaykin <145343289+azevaykin@users.noreply.github.com> Co-authored-by: DimasKovas <34828390+DimasKovas@users.noreply.github.com> Co-authored-by: ijon <ijon@ydb.tech> Co-authored-by: vporyadke <zalyalov@ydb.tech> Co-authored-by: kruall <kruall@ydb.tech> Co-authored-by: ivanmorozov333 <ivanmorozov@ydb.tech> Co-authored-by: Alexander Rutkovsky <alexvru@ydb.tech> Co-authored-by: ildar-khisambeev <ikhis@ydb.tech> Co-authored-by: nsofya <nsofya@ydb.tech> Co-authored-by: nsofya <nsofya@yandex.ru> Co-authored-by: Sofya Novozhilova <nsofya@qavm-4bd1829b.qemu> Co-authored-by: Олег <150132506+iddqdex@users.noreply.github.com> Co-authored-by: Ivan Morozov <ivanmorozov@localhost.localdomain> Co-authored-by: Andrei Rykov <arykov@ydb.tech> Co-authored-by: qyryq <qyryq@ydb.tech> Co-authored-by: Daniil Cherednik <dcherednik@ydb.tech> Co-authored-by: Aleksei Borzenkov <snaury@ydb.tech>

Author: 27 people

.github/config/muted_test.txt

@@ -3,8 +3,6 @@ ydb-core-blobstorage-ut_blobstorage/SpaceCheckForDiskReassign::*
 ydb-services-ydb-sdk_sessions_pool_ut/YdbSdkSessionsPool::StressTestSync10
 ydb-tests-functional-kqp-kqp_query_session/KqpQuerySession::NoLocalAttach
 ydb-core-blobstorage-ut_blobstorage/VDiskAssimilation::Test
-ydb-core-tx-columnshard-ut_schema/TColumnShardTestSchema::ForgetAfterFail
-ydb-core-tx-columnshard-ut_schema/TColumnShardTestSchema::RebootForgetAfterFail
 ydb-library-yql-sql-pg-ut/PgSqlParsingAutoparam::AutoParamValues_DifferentTypes
 ydb-core-blobstorage-ut_blobstorage/[6/10]*
 ydb/core/blobstorage/ut_blobstorage/Defragmentation::DoesItWork

.github/config/muted_ya.txt

@@ -19,7 +19,13 @@ ydb/core/kafka_proxy/ut KafkaProtocol.CreatePartitionsScenario
 ydb/core/kafka_proxy/ut KafkaProtocol.ProduceScenario
 ydb/core/kqp/provider/ut KikimrIcGateway.TestLoadBasicSecretValueFromExternalDataSourceMetadata
 ydb/core/kqp/ut/federated_query/generic *
-ydb/core/kqp/ut/olap *
+ydb/core/kqp/ut/olap KqpOlapAggregations.Json_Exists
+ydb/core/kqp/ut/olap KqpOlapIndexes.Indexes
+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/kqp/ut/olap KqpOlap.YqlScriptOltpAndOlap
 ydb/core/kqp/ut/pg KqpPg.CreateIndex
 ydb/core/kqp/ut/query KqpLimits.QueryReplySize
 ydb/core/kqp/ut/query KqpQuery.QueryTimeout
@@ -32,9 +38,6 @@ ydb/core/kqp/ut/service KqpQueryService.QueryOnClosedSession
 ydb/core/kqp/ut/service KqpQueryServiceScripts.ForgetScriptExecutionRace
 ydb/core/kqp/ut/service KqpService.CloseSessionsWithLoad
 ydb/core/kqp/ut/service [38/50]*
-ydb/core/tx/columnshard/ut_schema TColumnShardTestSchema.ForgetAfterFail
-ydb/core/tx/columnshard/ut_schema TColumnShardTestSchema.RebootForgetAfterFail
-ydb/core/tx/columnshard/engines/ut *
 ydb/core/tx/coordinator/ut Coordinator.RestoreTenantConfiguration
 ydb/core/tx/datashard/ut_change_exchange Cdc.InitialScanDebezium
 ydb/core/tx/replication/ydb_proxy/ut YdbProxyTests.ReadTopic

ydb/core/base/blobstorage.h

@@ -981,14 +981,16 @@ struct TEvBlobStorage {
  bool WrittenBeyondBarrier = false; // was this blob written beyond the barrier?
  mutable NLWTrace::TOrbit Orbit;
  std::shared_ptr<TExecutionRelay> ExecutionRelay;
+ const TString StorageId;
 
  TEvPutResult(NKikimrProto::EReplyStatus status, const TLogoBlobID &id, const TStorageStatusFlags statusFlags,
- ui32 groupId, float approximateFreeSpaceShare)
+ ui32 groupId, float approximateFreeSpaceShare, const TString& storageId = Default<TString>())
  : Status(status)
  , Id(id)
  , StatusFlags(statusFlags)
  , GroupId(groupId)
  , ApproximateFreeSpaceShare(approximateFreeSpaceShare)
+ , StorageId(storageId)
  {}
 
  TString Print(bool isFull) const {

ydb/core/formats/arrow/arrow_batch_builder.cpp

@@ -1,7 +1,7 @@
 #include "arrow_batch_builder.h"
+#include "switch/switch_type.h"
 #include <contrib/libs/apache/arrow/cpp/src/arrow/io/memory.h>
 #include <contrib/libs/apache/arrow/cpp/src/arrow/ipc/reader.h>
-
 namespace NKikimr::NArrow {
 
 namespace {
@@ -195,12 +195,18 @@ TArrowBatchBuilder::TArrowBatchBuilder(arrow::Compression::type codec, const std
  WriteOptions.use_threads = false;
 }
 
-bool TArrowBatchBuilder::Start(const std::vector<std::pair<TString, NScheme::TTypeInfo>>& ydbColumns) {
+arrow::Status TArrowBatchBuilder::Start(const std::vector<std::pair<TString, NScheme::TTypeInfo>>& ydbColumns) {
  YdbSchema = ydbColumns;
  auto schema = MakeArrowSchema(ydbColumns, NotNullColumns);
- auto status = arrow::RecordBatchBuilder::Make(schema, arrow::default_memory_pool(), RowsToReserve, &BatchBuilder);
+ if (!schema.ok()) {
+ return arrow::Status::FromArgs(schema.status().code(), "Cannot make arrow schema: ", schema.status().ToString());
+ }
+ auto status = arrow::RecordBatchBuilder::Make(*schema, arrow::default_memory_pool(), RowsToReserve, &BatchBuilder);
  NumRows = NumBytes = 0;
- return status.ok();
+ if (!status.ok()) {
+ return arrow::Status::FromArgs(schema.status().code(), "Cannot make arrow builder: ", status.ToString());
+ }
+ return arrow::Status::OK();
 }
 
 void TArrowBatchBuilder::AppendCell(const TCell& cell, ui32 colNum) {
@@ -259,7 +265,7 @@ void TArrowBatchBuilder::ReserveData(ui32 columnNo, size_t size) {
  Y_ABORT_UNLESS(columnNo < YdbSchema.size());
  auto type = YdbSchema[columnNo].second;
 
- SwitchYqlTypeToArrowType(type, [&](const auto& type) {
+ Y_ABORT_UNLESS(SwitchYqlTypeToArrowType(type, [&](const auto& type) {
  using TWrap = std::decay_t<decltype(type)>;
  using TBuilder = typename arrow::TypeTraits<typename TWrap::T>::BuilderType;
 
@@ -270,7 +276,7 @@ void TArrowBatchBuilder::ReserveData(ui32 columnNo, size_t size) {
  Y_ABORT_UNLESS(status.ok());
  }
  return true;
- });
+ }));
 }
 
 std::shared_ptr<arrow::RecordBatch> TArrowBatchBuilder::FlushBatch(bool reinitialize) {

ydb/core/formats/arrow/arrow_batch_builder.h

@@ -2,6 +2,7 @@
 #include "arrow_helpers.h"
 #include <ydb/core/formats/factory.h>
 #include <ydb/core/scheme/scheme_tablecell.h>
+#include <ydb/library/conclusion/status.h>
 
 namespace NKikimr::NArrow {
 
@@ -155,8 +156,11 @@ class TArrowBatchBuilder : public NKikimr::IBlockBuilder {
  ui64 maxRowsInBlock, ui64 maxBytesInBlock, TString& err) override {
  Y_UNUSED(maxRowsInBlock);
  Y_UNUSED(maxBytesInBlock);
- Y_UNUSED(err);
- return Start(columns);
+ const auto result = Start(columns);
+ if (!result.ok()) {
+ err = result.ToString();
+ }
+ return result.ok();
  }
 
  void AddRow(const NKikimr::TDbTupleRef& key, const NKikimr::TDbTupleRef& value) override;
@@ -175,7 +179,7 @@ class TArrowBatchBuilder : public NKikimr::IBlockBuilder {
  return NumBytes;
  }
 
- bool Start(const std::vector<std::pair<TString, NScheme::TTypeInfo>>& columns);
+ arrow::Status Start(const std::vector<std::pair<TString, NScheme::TTypeInfo>>& columns);
  std::shared_ptr<arrow::RecordBatch> FlushBatch(bool reinitialize);
  std::shared_ptr<arrow::RecordBatch> GetBatch() const { return Batch; }
 

ydb/core/formats/arrow/arrow_filter.cpp

@@ -1,5 +1,8 @@
 #include "arrow_filter.h"
 #include "switch_type.h"
+#include "common/container.h"
+#include "common/adapter.h"
+
 #include <contrib/libs/apache/arrow/cpp/src/arrow/array/builder_primitive.h>
 #include <contrib/libs/apache/arrow/cpp/src/arrow/chunked_array.h>
 #include <contrib/libs/apache/arrow/cpp/src/arrow/compute/api_vector.h>
@@ -307,7 +310,7 @@ NKikimr::NArrow::TColumnFilter TColumnFilter::MakePredicateFilter(const arrow::D
  return NArrow::TColumnFilter(std::move(bits));
 }
 
-template <arrow::Datum::Kind kindExpected, class TData>
+template <class TData>
 bool ApplyImpl(const TColumnFilter& filter, std::shared_ptr<TData>& batch, const std::optional<ui32> startPos, const std::optional<ui32> count) {
  if (!batch || !batch->num_rows()) {
  return false;
@@ -322,33 +325,26 @@ bool ApplyImpl(const TColumnFilter& filter, std::shared_ptr<TData>& batch, const
  }
  }
  if (filter.IsTotalDenyFilter()) {
- batch = batch->Slice(0, 0);
+ batch = NAdapter::TDataBuilderPolicy<TData>::GetEmptySame(batch);
  return true;
  }
  if (filter.IsTotalAllowFilter()) {
  return true;
  }
- auto res = arrow::compute::Filter(batch, filter.BuildArrowFilter(batch->num_rows(), startPos, count));
- Y_VERIFY_S(res.ok(), res.status().message());
- Y_ABORT_UNLESS((*res).kind() == kindExpected);
- if constexpr (kindExpected == arrow::Datum::TABLE) {
- batch = (*res).table();
- return batch->num_rows();
- }
- if constexpr (kindExpected == arrow::Datum::RECORD_BATCH) {
- batch = (*res).record_batch();
- return batch->num_rows();
- }
- AFL_VERIFY(false);
- return false;
+ batch = NAdapter::TDataBuilderPolicy<TData>::ApplyArrowFilter(batch, filter.BuildArrowFilter(batch->num_rows(), startPos, count));
+ return batch->num_rows();
+}
+
+bool TColumnFilter::Apply(std::shared_ptr<TGeneralContainer>& batch, const std::optional<ui32> startPos, const std::optional<ui32> count) const {
+ return ApplyImpl(*this, batch, startPos, count);
 }
 
 bool TColumnFilter::Apply(std::shared_ptr<arrow::Table>& batch, const std::optional<ui32> startPos, const std::optional<ui32> count) const {
- return ApplyImpl<arrow::Datum::TABLE>(*this, batch, startPos, count);
+ return ApplyImpl(*this, batch, startPos, count);
 }
 
 bool TColumnFilter::Apply(std::shared_ptr<arrow::RecordBatch>& batch, const std::optional<ui32> startPos, const std::optional<ui32> count) const {
- return ApplyImpl<arrow::Datum::RECORD_BATCH>(*this, batch, startPos, count);
+ return ApplyImpl(*this, batch, startPos, count);
 }
 
 void TColumnFilter::Apply(const ui32 expectedRecordsCount, std::vector<arrow::Datum*>& datums) const {

ydb/core/formats/arrow/arrow_filter.h

@@ -8,6 +8,8 @@
 
 namespace NKikimr::NArrow {
 
+class TGeneralContainer;
+
 enum class ECompareType {
  LESS = 1,
  LESS_OR_EQUAL,
@@ -62,6 +64,10 @@ class TColumnFilter {
  return Filter.capacity() * sizeof(ui32) + Count * sizeof(bool);
  }
 
+ static ui64 GetPredictedMemorySize(const ui32 recordsCount) {
+ return 2 /* capacity */ * recordsCount * (sizeof(ui32) + sizeof(bool));
+ }
+
  class TIterator {
  private:
  i64 InternalPosition = 0;
@@ -172,6 +178,7 @@ class TColumnFilter {
  // It makes a filter using composite predicate
  static TColumnFilter MakePredicateFilter(const arrow::Datum& datum, const arrow::Datum& border, ECompareType compareType);
 
+ bool Apply(std::shared_ptr<TGeneralContainer>& batch, const std::optional<ui32> startPos = {}, const std::optional<ui32> count = {}) const;
  bool Apply(std::shared_ptr<arrow::Table>& batch, const std::optional<ui32> startPos = {}, const std::optional<ui32> count = {}) const;
  bool Apply(std::shared_ptr<arrow::RecordBatch>& batch, const std::optional<ui32> startPos = {}, const std::optional<ui32> count = {}) const;
  void Apply(const ui32 expectedRecordsCount, std::vector<arrow::Datum*>& datums) const;