Merge 68fe409 into 407fc46

ydb-platform · Mar 2, 2024 · ff95897 · ff95897
2 parents 407fc46 + 68fe409
commit ff95897
Show file tree

Hide file tree

Showing 58 changed files with 1,511 additions and 248 deletions.
diff --git a/ydb/core/kqp/gateway/behaviour/tablestore/operations/drop_stat.cpp b/ydb/core/kqp/gateway/behaviour/tablestore/operations/drop_stat.cpp
@@ -0,0 +1,21 @@
+#include "drop_stat.h"
+#include <util/string/type.h>
+
+namespace NKikimr::NKqp {
+
+TConclusionStatus TDropStatOperation::DoDeserialize(NYql::TObjectSettingsImpl::TFeaturesExtractor& features) {
+    {
+        auto fValue = features.Extract("NAME");
+        if (!fValue) {
+            return TConclusionStatus::Fail("can't find parameter NAME");
+        }
+        Name = *fValue;
+    }
+    return TConclusionStatus::Success();
+}
+
+void TDropStatOperation::DoSerializeScheme(NKikimrSchemeOp::TAlterColumnTableSchema& schemaData) const {
+    *schemaData.AddDropStatistics() = Name;
+}
+
+}
diff --git a/ydb/core/kqp/gateway/behaviour/tablestore/operations/drop_stat.h b/ydb/core/kqp/gateway/behaviour/tablestore/operations/drop_stat.h
@@ -0,0 +1,19 @@
+#include "abstract.h"
+
+namespace NKikimr::NKqp {
+
+class TDropStatOperation : public ITableStoreOperation {
+    static TString GetTypeName() {
+        return "DROP_STAT";
+    }
+
+    static inline auto Registrator = TFactory::TRegistrator<TDropStatOperation>(GetTypeName());
+private:
+    TString Name;
+public:
+    TConclusionStatus DoDeserialize(NYql::TObjectSettingsImpl::TFeaturesExtractor& features) override;
+    void DoSerializeScheme(NKikimrSchemeOp::TAlterColumnTableSchema& schemaData) const override;
+};
+
+}
+
diff --git a/ydb/core/kqp/gateway/behaviour/tablestore/operations/upsert_stat.cpp b/ydb/core/kqp/gateway/behaviour/tablestore/operations/upsert_stat.cpp
@@ -0,0 +1,49 @@
+#include "upsert_stat.h"
+#include <util/string/type.h>
+#include <library/cpp/json/json_reader.h>
+
+namespace NKikimr::NKqp {
+
+TConclusionStatus TUpsertStatOperation::DoDeserialize(NYql::TObjectSettingsImpl::TFeaturesExtractor& features) {
+    {
+        auto fValue = features.Extract("NAME");
+        if (!fValue) {
+            return TConclusionStatus::Fail("can't find  alter parameter NAME");
+        }
+        Name = *fValue;
+    }
+    TString type;
+    {
+        auto fValue = features.Extract("TYPE");
+        if (!fValue) {
+            return TConclusionStatus::Fail("can't find alter parameter TYPE");
+        }
+        type = *fValue;
+    }
+    {
+        auto fValue = features.Extract("FEATURES");
+        if (!fValue) {
+            return TConclusionStatus::Fail("can't find alter parameter FEATURES");
+        }
+        if (!Constructor.Initialize(type)) {
+            return TConclusionStatus::Fail("can't initialize stat constructor object for type \"" + type + "\"");
+        }
+        NJson::TJsonValue jsonData;
+        if (!NJson::ReadJsonFastTree(*fValue, &jsonData)) {
+            return TConclusionStatus::Fail("incorrect json in request FEATURES parameter");
+        }
+        auto result = Constructor->DeserializeFromJson(jsonData);
+        if (result.IsFail()) {
+            return result;
+        }
+    }
+    return TConclusionStatus::Success();
+}
+
+void TUpsertStatOperation::DoSerializeScheme(NKikimrSchemeOp::TAlterColumnTableSchema& schemaData) const {
+    auto* proto = schemaData.AddUpsertStatistics();
+    proto->SetName(Name);
+    Constructor.SerializeToProto(*proto);
+}
+
+}
diff --git a/ydb/core/kqp/gateway/behaviour/tablestore/operations/upsert_stat.h b/ydb/core/kqp/gateway/behaviour/tablestore/operations/upsert_stat.h
@@ -0,0 +1,23 @@
+#include "abstract.h"
+#include <ydb/core/tx/columnshard/engines/scheme/statistics/abstract/constructor.h>
+
+namespace NKikimr::NKqp {
+
+class TUpsertStatOperation : public ITableStoreOperation {
+private:
+    static TString GetTypeName() {
+        return "UPSERT_STAT";
+    }
+
+    static inline const auto Registrator = TFactory::TRegistrator<TUpsertStatOperation>(GetTypeName());
+private:
+    TString Name;
+    NOlap::NStatistics::TConstructorContainer Constructor;
+public:
+    TConclusionStatus DoDeserialize(NYql::TObjectSettingsImpl::TFeaturesExtractor& features) override;
+
+    void DoSerializeScheme(NKikimrSchemeOp::TAlterColumnTableSchema& schemaData) const override;
+};
+
+}
+
diff --git a/ydb/core/kqp/gateway/behaviour/tablestore/operations/ya.make b/ydb/core/kqp/gateway/behaviour/tablestore/operations/ya.make
@@ -7,6 +7,8 @@ SRCS(
     GLOBAL drop_column.cpp
     GLOBAL upsert_index.cpp
     GLOBAL drop_index.cpp
+    GLOBAL upsert_stat.cpp
+    GLOBAL drop_stat.cpp
 )
 
 PEERDIR(

diff --git a/ydb/core/kqp/ut/olap/kqp_olap_ut.cpp b/ydb/core/kqp/ut/olap/kqp_olap_ut.cpp
@@ -3836,6 +3836,36 @@ Y_UNIT_TEST_SUITE(KqpOlap) {
         Y_ABORT_UNLESS(bytesPack / bytesUnpack < 0.1);
     }
 
+    Y_UNIT_TEST(StatsUsage) {
+        auto csController = NYDBTest::TControllers::RegisterCSControllerGuard<NYDBTest::NColumnShard::TController>();
+        {
+            auto settings = TKikimrSettings().SetWithSampleTables(false);
+            TKikimrRunner kikimr(settings);
+            Tests::NCommon::TLoggerInit(kikimr).Initialize();
+            TTypedLocalHelper helper("Utf8", kikimr);
+            helper.CreateTestOlapTable();
+            auto tableClient = kikimr.GetTableClient();
+            {
+                auto alterQuery = TStringBuilder() << "ALTER OBJECT `/Root/olapStore` (TYPE TABLESTORE) SET (ACTION=UPSERT_STAT, TYPE=max, NAME=max_pk_int, FEATURES=`{\"column_name\": \"pk_int\"}`);";
+                auto session = tableClient.CreateSession().GetValueSync().GetSession();
+                auto alterResult = session.ExecuteSchemeQuery(alterQuery).GetValueSync();
+                UNIT_ASSERT_VALUES_EQUAL_C(alterResult.GetStatus(), EStatus::SUCCESS, alterResult.GetIssues().ToString());
+            }
+            {
+                auto alterQuery = TStringBuilder() << "ALTER OBJECT `/Root/olapStore` (TYPE TABLESTORE) SET (ACTION=UPSERT_STAT, TYPE=max, NAME=max_field, FEATURES=`{\"column_name\": \"field\"}`);";
+                auto session = tableClient.CreateSession().GetValueSync().GetSession();
+                auto alterResult = session.ExecuteSchemeQuery(alterQuery).GetValueSync();
+                UNIT_ASSERT_VALUES_UNEQUAL_C(alterResult.GetStatus(), EStatus::SUCCESS, alterResult.GetIssues().ToString());
+            }
+            {
+                auto alterQuery = TStringBuilder() << "ALTER OBJECT `/Root/olapStore` (TYPE TABLESTORE) SET (ACTION=DROP_STAT, TYPE=max, NAME=max_pk_int);";
+                auto session = tableClient.CreateSession().GetValueSync().GetSession();
+                auto alterResult = session.ExecuteSchemeQuery(alterQuery).GetValueSync();
+                UNIT_ASSERT_VALUES_EQUAL_C(alterResult.GetStatus(), EStatus::SUCCESS, alterResult.GetIssues().ToString());
+            }
+        }
+    }
+
     namespace {
     class TTransferStatus {
     private:

diff --git a/ydb/core/protos/flat_scheme_op.proto b/ydb/core/protos/flat_scheme_op.proto
@@ -17,6 +17,7 @@ import "ydb/public/api/protos/ydb_value.proto";
 import "ydb/library/actors/protos/actors.proto";
 import "ydb/library/mkql_proto/protos/minikql.proto";
 import "ydb/core/protos/index_builder.proto";
+import "ydb/core/tx/columnshard/engines/scheme/statistics/protos/data.proto";
 
 import "google/protobuf/empty.proto";
 
@@ -512,6 +513,7 @@ message TColumnTableSchema {
 
     optional bool CompositeMarks = 9 [ default = false ];
     repeated TOlapIndexDescription Indexes = 10;
+    repeated NKikimrColumnShardStatisticsProto.TOperatorContainer Statistics = 11;
 }
 
 message TAlterColumnTableSchema {
@@ -521,6 +523,8 @@ message TAlterColumnTableSchema {
     repeated TOlapColumnDiff AlterColumns = 7;
     repeated TOlapIndexRequested UpsertIndexes = 8;
     repeated string DropIndexes = 9;
+    repeated NKikimrColumnShardStatisticsProto.TConstructorContainer UpsertStatistics = 10;
+    repeated string DropStatistics = 11;
 }
 
 // Schema presets are used to manage multiple tables with the same schema

diff --git a/ydb/core/protos/tx_columnshard.proto b/ydb/core/protos/tx_columnshard.proto
@@ -1,4 +1,5 @@
 import "ydb/library/actors/protos/actors.proto";
+import "ydb/core/tx/columnshard/engines/scheme/statistics/protos/data.proto";
 import "ydb/core/protos/flat_scheme_op.proto";
 import "ydb/core/protos/long_tx_service.proto";
 import "ydb/core/protos/subdomains.proto";
@@ -281,6 +282,7 @@ message TIndexPortionMeta {
     optional bytes PrimaryKeyBorders = 6; // arrow::RecordBatch with first and last ReplaceKey rows
     optional TSnapshot RecordSnapshotMin = 7;
     optional TSnapshot RecordSnapshotMax = 8;
+    optional NKikimrColumnShardStatisticsProto.TPortionStorage StatisticsStorage = 9;
 }
 
 message TIndexColumnMeta {

diff --git a/ydb/core/protos/ya.make b/ydb/core/protos/ya.make
@@ -163,6 +163,7 @@ PEERDIR(
     ydb/library/yql/public/types
     ydb/library/services
     ydb/library/ydb_issue/proto
+    ydb/core/tx/columnshard/engines/scheme/statistics/protos
 )
 
 CPP_PROTO_PLUGIN0(config_proto_plugin ydb/core/config/tools/protobuf_plugin)

diff --git a/ydb/core/tx/columnshard/columnshard_ut_common.h b/ydb/core/tx/columnshard/columnshard_ut_common.h
@@ -3,6 +3,7 @@
 #include "columnshard.h"
 #include "columnshard_impl.h"
 #include "blob_cache.h"
+#include "engines/scheme/statistics/max/operator.h"
 
 #include <ydb/core/formats/arrow/arrow_batch_builder.h>
 #include <ydb/core/tx/columnshard/test_helper/helper.h>
@@ -217,6 +218,9 @@ struct TTestSchema {
 
         for (ui32 i = 0; i < columns.size(); ++i) {
             *schema->MutableColumns()->Add() = columns[i].CreateColumn(i + 1);
+//            if (NOlap::NStatistics::NMax::TOperator::IsAvailableType(columns[i].GetType())) {
+//                *schema->AddStatistics() = NOlap::NStatistics::TOperatorContainer(std::make_shared<NOlap::NStatistics::NMax::TOperator>(i + 1)).SerializeToProto();
+//            }
         }
 
         Y_ABORT_UNLESS(pk.size() > 0);

diff --git a/ydb/core/tx/columnshard/counters/engine_logs.cpp b/ydb/core/tx/columnshard/counters/engine_logs.cpp
@@ -47,6 +47,9 @@ TEngineLogsCounters::TEngineLogsCounters()
 
     PortionNoBorderCount = TBase::GetDeriviative("Ttl/PortionNoBorder/Count");
     PortionNoBorderBytes = TBase::GetDeriviative("Ttl/PortionNoBorder/Bytes");
+
+    StatUsageForTTLCount = TBase::GetDeriviative("Ttl/StatUsageForTTLCount/Count");
+    ChunkUsageForTTLCount = TBase::GetDeriviative("Ttl/ChunkUsageForTTLCount/Count");
 }
 
 void TEngineLogsCounters::TPortionsInfoGuard::OnNewPortion(const std::shared_ptr<NOlap::TPortionInfo>& portion) const {

diff --git a/ydb/core/tx/columnshard/counters/engine_logs.h b/ydb/core/tx/columnshard/counters/engine_logs.h
@@ -243,6 +243,9 @@ class TEngineLogsCounters: public TCommonCountersOwner {
     NMonitoring::TDynamicCounters::TCounterPtr PortionNoTtlColumnCount;
     NMonitoring::TDynamicCounters::TCounterPtr PortionNoTtlColumnBytes;
 
+    NMonitoring::TDynamicCounters::TCounterPtr StatUsageForTTLCount;
+    NMonitoring::TDynamicCounters::TCounterPtr ChunkUsageForTTLCount;
+
     NMonitoring::TDynamicCounters::TCounterPtr PortionNoBorderCount;
     NMonitoring::TDynamicCounters::TCounterPtr PortionNoBorderBytes;
 
@@ -302,6 +305,14 @@ class TEngineLogsCounters: public TCommonCountersOwner {
         PortionNoTtlColumnBytes->Add(size);
     }
 
+    void OnChunkUsageForTTL() const {
+        ChunkUsageForTTLCount->Add(1);
+    }
+
+    void OnStatUsageForTTL() const {
+        StatUsageForTTLCount->Add(1);
+    }
+
     void OnPortionNoBorder(const ui64 size) const {
         PortionNoBorderCount->Add(1);
         PortionNoBorderBytes->Add(size);

diff --git a/ydb/core/tx/columnshard/engines/changes/general_compaction.cpp b/ydb/core/tx/columnshard/engines/changes/general_compaction.cpp
@@ -195,7 +195,8 @@ void TGeneralCompactColumnEngineChanges::BuildAppendedPortionsByChunks(TConstruc
         for (auto&& i : packs) {
             TGeneralSerializedSlice slice(std::move(i), GetSplitSettings());
             auto b = batchResult->Slice(recordIdx, slice.GetRecordsCount());
-            AppendedPortions.emplace_back(TPortionInfoWithBlobs::BuildByBlobs(slice.GroupChunksByBlobs(groups), nullptr, GranuleMeta->GetPathId(), resultSchema->GetSnapshot(), SaverContext.GetStoragesManager()));
+            AppendedPortions.emplace_back(TPortionInfoWithBlobs::BuildByBlobs(slice.GroupChunksByBlobs(groups), nullptr, GranuleMeta->GetPathId(), 
+                resultSchema->GetSnapshot(), SaverContext.GetStoragesManager(), resultSchema));
             NArrow::TFirstLastSpecialKeys primaryKeys(slice.GetFirstLastPKBatch(resultSchema->GetIndexInfo().GetReplaceKey()));
             NArrow::TMinMaxSpecialKeys snapshotKeys(b, TIndexInfo::ArrowSchemaSnapshot());
             AppendedPortions.back().GetPortionInfo().AddMetadata(*resultSchema, primaryKeys, snapshotKeys, IStoragesManager::DefaultStorageId);

diff --git a/ydb/core/tx/columnshard/engines/changes/with_appended.cpp b/ydb/core/tx/columnshard/engines/changes/with_appended.cpp
@@ -124,7 +124,7 @@ std::vector<TPortionInfoWithBlobs> TChangesWithAppend::MakeAppendedPortions(cons
         for (auto&& i : packs) {
             TGeneralSerializedSlice slice(std::move(i), GetSplitSettings());
             auto b = batch->Slice(recordIdx, slice.GetRecordsCount());
-            out.emplace_back(TPortionInfoWithBlobs::BuildByBlobs(slice.GroupChunksByBlobs(groups), nullptr, pathId, snapshot, SaverContext.GetStoragesManager()));
+            out.emplace_back(TPortionInfoWithBlobs::BuildByBlobs(slice.GroupChunksByBlobs(groups), nullptr, pathId, snapshot, SaverContext.GetStoragesManager(), resultSchema));
             NArrow::TFirstLastSpecialKeys primaryKeys(slice.GetFirstLastPKBatch(resultSchema->GetIndexInfo().GetReplaceKey()));
             NArrow::TMinMaxSpecialKeys snapshotKeys(b, TIndexInfo::ArrowSchemaSnapshot());
             out.back().GetPortionInfo().AddMetadata(*resultSchema, primaryKeys, snapshotKeys, IStoragesManager::DefaultStorageId);