merge to stable-25-3: AvailabilityPeriod for topic consumers (#24752) (#26308)

Introduce the availability_period option to extend the period for which unprocessed messages can be retained in the topic.

KIKIMR-24054

Author: ubyte

ydb/core/kqp/provider/yql_kikimr_exec.cpp

@@ -469,6 +469,10 @@ namespace {
                 protoConsumer->set_important(FromString<bool>(
                         setting.Value().Cast<TCoDataCtor>().Literal().Cast<TCoAtom>().Value()
                 ));
+            } else if (name == "setAvailabilityPeriod"sv) {
+                auto period = TDuration::MicroSeconds(FromString<ui64>(setting.Value().Cast<TCoInterval>().Literal().Value()));
+                protoConsumer->mutable_availability_period()->set_seconds(period.Seconds());
+                protoConsumer->mutable_availability_period()->set_nanos(period.NanoSecondsOfSecond());
             } else if (name == "setReadFromTs") {
                 ui64 tsValue = 0;
                 if(setting.Value().Maybe<TCoDatetime>()) {
@@ -514,6 +518,12 @@ namespace {
                 protoConsumer->set_set_important(FromString<bool>(
                         setting.Value().Cast<TCoDataCtor>().Literal().Cast<TCoAtom>().Value()
                 ));
+            } else if (name == "setAvailabilityPeriod"sv) {
+                auto period = TDuration::MicroSeconds(FromString<ui64>(setting.Value().Cast<TCoInterval>().Literal().Value()));
+                protoConsumer->mutable_set_availability_period()->set_seconds(period.Seconds());
+                protoConsumer->mutable_set_availability_period()->set_nanos(period.NanoSecondsOfSecond());
+            } else if (name == "resetAvailabilityPeriod"sv) {
+                protoConsumer->mutable_reset_availability_period();
             } else if (name == "setReadFromTs") {
                 ui64 tsValue = 0;
                 if(setting.Value().Maybe<TCoDatetime>()) {
@@ -1716,7 +1726,7 @@ class TKiSinkCallableExecutionTransformer : public TAsyncCallbackTransformer<TKi
                                     return SyncError();
                                 } else if (constraint.Name().Value() == "default") {
                                     if (table.Metadata->Kind == EKikimrTableKind::Olap) {
-                                        ctx.AddError(TIssue(ctx.GetPosition(constraint.Pos()), 
+                                        ctx.AddError(TIssue(ctx.GetPosition(constraint.Pos()),
                                             "Default values are not supported in column tables"));
                                         return SyncError();
                                     }
@@ -2046,10 +2056,10 @@ class TKiSinkCallableExecutionTransformer : public TAsyncCallbackTransformer<TKi
                             }
                         } else if (name == "indexSettings") {
                             YQL_ENSURE(add_index->type_case() == Ydb::Table::TableIndex::kGlobalVectorKmeansTreeIndex);
-                            
+
                             Ydb::Table::KMeansTreeSettings& settings = *add_index->mutable_global_vector_kmeans_tree_index()->mutable_vector_settings();
                             TString error;
-                            
+
                             auto indexSettings = columnTuple.Item(1).Cast<TCoAtomList>();
                             for (const auto& indexSetting : indexSettings.Cast<TCoNameValueTupleList>()) {
                                 YQL_ENSURE(indexSetting.Value().Maybe<TCoAtom>());

ydb/core/kqp/provider/yql_kikimr_type_ann.cpp

@@ -1650,9 +1650,9 @@ virtual TStatus HandleCreateTable(TKiCreateTable create, TExprContext& ctx) over
     }
     static bool CheckConsumerSettings(const TCoNameValueTupleList& settings, TExprContext& ctx) {
         for (const auto& setting : settings) {
-            auto name = setting.Name().Value();
-            auto val = TString(setting.Value().Cast<TCoDataCtor>().Literal().template Cast<TCoAtom>().Value());
+            const auto name = setting.Name().Value();
             if (name == "setSupportedCodecs") {
+                auto val = TString(setting.Value().Cast<TCoDataCtor>().Literal().template Cast<TCoAtom>().Value());
                 auto codecsList = GetTopicCodecsFromString(val);
                 if (codecsList.empty()) {
                     ctx.AddError(TIssue(ctx.GetPosition(setting.Value().Ref().Pos()),

ydb/core/kqp/ut/scheme/kqp_scheme_ut.cpp

@@ -4137,7 +4137,7 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
             } else {
                 query = Sprintf(R"(
                     --!syntax_v1
-                    ALTER TABLE `/Root/TestTable` ADD INDEX vector_idx%d 
+                    ALTER TABLE `/Root/TestTable` ADD INDEX vector_idx%d
                         GLOBAL USING vector_kmeans_tree
                         ON (Embedding)
                         WITH (%s);
@@ -4161,18 +4161,18 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
         // valid settings:
         check("similarity=inner_product, vector_type=float, vector_dimension=1024, levels=3, clusters=10", "");
 
-        // unknown index setting: 
+        // unknown index setting:
         check("XxX=YyY, similarity=inner_product, vector_type=float, vector_dimension=1024, levels=3, clusters=10",
             "Unknown index setting: xxx");
         check("XxX=42, similarity=inner_product, vector_type=float, vector_dimension=1024, levels=3, clusters=10",
             "Unknown index setting: xxx");
-        
+
         // distance:
         check("distance=XxX, vector_type=float, vector_dimension=1024, levels=3, clusters=10",
             "Invalid distance: xxx");
         check("distance=42, vector_type=float, vector_dimension=1024, levels=3, clusters=10",
             "Invalid distance: 42");
-        
+
         // similarity
         check("similarity=XxX, vector_type=float, vector_dimension=1024, levels=3, clusters=10",
             "Invalid similarity: xxx");
@@ -4194,7 +4194,7 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
             "Invalid vector_type: 42");
         check("similarity=inner_product, vector_dimension=1024, levels=3, clusters=10",
             "vector_type should be set");
-        
+
         // vector_dimension
         check("similarity=inner_product, vector_type=float, vector_dimension=XxX, levels=3, clusters=10",
             "Invalid vector_dimension: xxx");
@@ -4211,7 +4211,7 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
             "Invalid vector_dimension: 99999999999999999999");
         check("similarity=inner_product, vector_type=float, levels=3, clusters=10",
             "vector_dimension should be set");
-            
+
         // levels
         check("similarity=inner_product, vector_type=float, vector_dimension=1024, levels=XxX, clusters=2",
             "Invalid levels: xxx");
@@ -4247,13 +4247,13 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
             "Invalid clusters: 99999999999999999999");
         check("similarity=inner_product, vector_type=float, vector_dimension=1024, levels=1",
             "clusters should be set");
-        
+
         // clusters^levels
         check("similarity=inner_product, vector_type=float, vector_dimension=1024, levels=10, clusters=10",
             "Invalid clusters^levels: 10^10 should be less than 1073741824");
         check("similarity=inner_product, vector_type=float, vector_dimension=1024, levels=16, clusters=1024",
             "Invalid clusters^levels: 1024^16 should be less than 1073741824");
-        
+
         // vector_dimension*clusters
         check("similarity=inner_product, vector_type=float, vector_dimension=2048, levels=1, clusters=2048", "");
         check("similarity=inner_product, vector_type=float, vector_dimension=2049, levels=1, clusters=2048",
@@ -11452,6 +11452,77 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
         }
     }
 
+     Y_UNIT_TEST_TWIN(CreateAndAlterTopicAvailabilityPeriod, UseQueryService) {
+        TKikimrRunner kikimr;
+        auto queryClient = kikimr.GetQueryClient();
+        auto db = kikimr.GetTableClient();
+        auto session = db.CreateSession().GetValueSync().GetSession();
+
+        auto executeQuery = [&queryClient, &session](const TString& query) {
+            return ExecuteGeneric<UseQueryService>(queryClient, session, query);
+        };
+
+        // ok
+        {
+            const auto query = R"(
+                --!syntax_v1
+                CREATE TOPIC `/Root/topic` (
+                    CONSUMER cons1 WITH (availability_period = Interval('PT1H'))
+                )
+            )";
+            const auto result = executeQuery(query);
+            UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+        }
+        {
+            const auto query = R"(
+                --!syntax_v1
+                ALTER TOPIC `/Root/topic`
+                    ALTER CONSUMER cons1 SET (availability_period = Interval('PT9H'))
+            )";
+            const auto result = executeQuery(query);
+            UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+        }
+        {
+            const auto query = R"(
+                --!syntax_v1
+                ALTER TOPIC `/Root/topic`
+                    DROP CONSUMER cons1,
+                    ADD CONSUMER cons2 WITH (availability_period = Interval('PT8H'))
+            )";
+            const auto result = executeQuery(query);
+            UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+        }
+        {
+            const auto query = R"(
+                --!syntax_v1
+                ALTER TOPIC `/Root/topic`
+                    ALTER CONSUMER cons2 RESET (availability_period)
+            )";
+            const auto result = executeQuery(query);
+            UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+        }
+        // bad
+        {
+            const auto query = R"(
+                --!syntax_v1
+                ALTER TOPIC `/Root/topic`
+                    ALTER CONSUMER cons2 SET (availability_period = 0)
+            )";
+            const auto result = executeQuery(query);
+            UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::GENERIC_ERROR, result.GetIssues().ToString());
+            UNIT_ASSERT_STRING_CONTAINS_C(result.GetIssues().ToString(), "Interval type is expected", result.GetIssues().ToString());
+        }
+        {
+            const auto query = R"(
+                --!syntax_v1
+                ALTER TOPIC `/Root/topic`
+                     ADD CONSUMER cons_neg WITH (availability_period = Interval('-PT8H'))
+            )";
+            const auto result = executeQuery(query);
+            UNIT_ASSERT_VALUES_UNEQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+        }
+    }
+
     Y_UNIT_TEST(DisableResourcePools) {
         NKikimrConfig::TAppConfig config;
         config.MutableFeatureFlags()->SetEnableResourcePools(false);

ydb/core/persqueue/pqtablet/partition/consumer_offset_tracker.cpp

@@ -0,0 +1,37 @@
+#include "consumer_offset_tracker.h"
+
+namespace NKikimr::NPQ {
+
+bool ImportantConsumerNeedToKeepCurrentKey(const TDuration availabilityPeriod, const ui64 offset, const TDataKey& currentKey, const TDataKey& nextKey, const TInstant now) {
+    const TInstant endOfLife = currentKey.Timestamp + availabilityPeriod; // note: sum with saturation
+    if (endOfLife < now) {
+        // The current key is too old. It doesn't matter whether the consumer has read it or not. It can be retired.
+        return false;
+    }
+    if (offset < nextKey.Key.GetOffset()) {
+        // The first message in the next blob was not read by an important consumer.
+        // We also save the current blob, since not all messages from it could be read.
+        return true;
+    }
+    if (offset == nextKey.Key.GetOffset() && nextKey.Key.GetPartNo() != 0) {
+        // We save all the blobs that contain parts of the last message read by an important consumer.
+        return true;
+    }
+    return false;
+}
+
+TImportantConsumerOffsetTracker::TImportantConsumerOffsetTracker(std::vector<TImportantConsumerOffsetTracker::TConsumerOffset> consumersToCheck)
+    : Consumers_(std::move(consumersToCheck))
+{
+}
+
+bool TImportantConsumerOffsetTracker::ShouldKeepCurrentKey(const TDataKey& currentKey, const TDataKey& nextKey, const TInstant now) const {
+    for (const auto& consumer : Consumers_) {
+        if (ImportantConsumerNeedToKeepCurrentKey(consumer.AvailabilityPeriod, consumer.Offset, currentKey, nextKey, now)) {
+            return true;
+        }
+    }
+    return false;
+}
+
+} // namespace NKikimr::NPQ

ydb/core/persqueue/pqtablet/partition/consumer_offset_tracker.h

@@ -0,0 +1,26 @@
+#pragma once
+
+#include <ydb/core/persqueue/events/internal.h>
+#include <util/datetime/base.h>
+#include <vector>
+
+namespace NKikimr::NPQ {
+
+    class TImportantConsumerOffsetTracker {
+    public:
+        struct TConsumerOffset {
+            TDuration AvailabilityPeriod = TDuration::Max();
+            ui64 Offset = 0;
+        };
+
+        explicit TImportantConsumerOffsetTracker(std::vector<TConsumerOffset> consumers);
+
+        bool ShouldKeepCurrentKey(const TDataKey& currentKey, const TDataKey& nextKey, const TInstant now) const;
+
+    private:
+        std::vector<TConsumerOffset> Consumers_;
+    };
+
+    bool ImportantConsumerNeedToKeepCurrentKey(const TDuration availabilityPeriod, const ui64 offset, const TDataKey& currentKey, const TDataKey& nextKey, const TInstant now);
+
+} // namespace NKikimr::NPQ