feat(kqp): add UseSequentialReads pragma

Author: ulya-sidorina

ydb/core/kqp/opt/kqp_query_plan.cpp

@@ -1789,7 +1789,7 @@ class TxPlanSerializer {
             op.Properties["Reverse"] = true;
         }
 
-        if (settings.SequentialInFlight) {
+        if (settings.SequentialInFlight || SerializerCtx.Config->UseSequentialReadsEnabled()) {
             op.Properties["Scan"] = "Sequential";
         } else {
             op.Properties["Scan"] = "Parallel";

ydb/core/kqp/provider/yql_kikimr_settings.cpp

@@ -95,6 +95,7 @@ TKikimrConfiguration::TKikimrConfiguration() {
     REGISTER_SETTING(*this, MaxDPccpDPTableSize);
 
     REGISTER_SETTING(*this, MaxTasksPerStage);
+    REGISTER_SETTING(*this, UseSequentialReads);
 
     /* Runtime */
     REGISTER_SETTING(*this, ScanQuery);
@@ -164,6 +165,10 @@ NDq::EHashJoinMode TKikimrSettings::GetHashJoinMode() const {
     return maybeHashJoinMode ? *maybeHashJoinMode : NDq::EHashJoinMode::Off;
 }
 
+bool TKikimrSettings::UseSequentialReadsEnabled() const {
+    return GetOptionalFlagValue(UseSequentialReads.Get()) == EOptionalFlag::Enabled;
+}
+
 TKikimrSettings::TConstPtr TKikimrConfiguration::Snapshot() const {
     return std::make_shared<const TKikimrSettings>(*this);
 }

ydb/core/kqp/provider/yql_kikimr_settings.h

@@ -72,6 +72,7 @@ struct TKikimrSettings {
 
 
     NCommon::TConfSetting<ui32, false> MaxTasksPerStage;
+    NCommon::TConfSetting<bool, false> UseSequentialReads;
 
     /* Runtime */
     NCommon::TConfSetting<bool, true> ScanQuery;
@@ -82,6 +83,7 @@ struct TKikimrSettings {
     bool SystemColumnsEnabled() const;
     bool SpillingEnabled() const;
     bool DisableLlvmForUdfStages() const;
+    bool UseSequentialReadsEnabled() const;
 
     bool HasOptDisableTopSort() const;
     bool HasOptDisableSqlInToJoin() const;

ydb/core/kqp/query_compiler/kqp_query_compiler.cpp

@@ -1009,7 +1009,9 @@ class TKqpQueryCompiler : public IKqpQueryCompiler {
             readProto.SetSorted(readSettings.Sorted);
             YQL_ENSURE(readSettings.SkipNullKeys.empty());
 
-            if (readSettings.SequentialInFlight) {
+            if (Config->UseSequentialReadsEnabled()) {
+                readProto.SetSequentialInFlightShards(1);
+            } else if (readSettings.SequentialInFlight) {
                 readProto.SetSequentialInFlightShards(*readSettings.SequentialInFlight);
             }
 

ydb/core/kqp/ut/opt/kqp_ne_ut.cpp

@@ -4235,7 +4235,46 @@ Y_UNIT_TEST_SUITE(KqpNewEngine) {
         }
     }
 
+    Y_UNIT_TEST_TWIN(SequentialReadsPragma, Enabled) {
+        TKikimrRunner kikimr;
+        auto db = kikimr.GetTableClient();
+        auto session = db.CreateSession().GetValueSync().GetSession();
+
+        NYdb::NTable::TExecDataQuerySettings querySettings;
+        querySettings.CollectQueryStats(ECollectQueryStatsMode::Profile);
+
+        TString query = R"(
+            SELECT Key, Data FROM `/Root/EightShard` 
+                WHERE Text = "Value1" 
+                ORDER BY Key 
+                LIMIT 1;
+        )";
+
+        if (Enabled) {
+            TString pragma = TString(R"(
+                PRAGMA ydb.UseSequentialReads = "true";
+            )");
 
+            query = pragma + query;
+        }
+
+        auto result = session.ExecuteDataQuery(query, TTxControl::BeginTx().CommitTx(), querySettings).GetValueSync();
+        UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString());
+        CompareYson(R"([[[101u];[1]]])", FormatResultSetYson(result.GetResultSet(0)));
+
+        auto stats = NYdb::TProtoAccessor::GetProto(*result.GetStats());
+        for (const auto& phase : stats.query_phases()) {
+            for (const auto& access : phase.table_access()) {
+                if (access.name() == "/Root/EightShard") {
+                    if (Enabled) {
+                        UNIT_ASSERT_LT(access.partitions_count(), 8);
+                    } else {
+                        UNIT_ASSERT_EQUAL(access.partitions_count(), 8);
+                    }
+                }
+            }
+        }
+    }
 
 }