Merge branch 'ydb-platform:main' into fuse_shuffle_by_keys

Author: Tony-Romanov

ydb/core/fq/libs/actors/database_resolver.cpp

@@ -478,10 +478,12 @@ class TDatabaseResolver: public TActor<TDatabaseResolver>
                     continue;
                 }
 
+                const auto& servicesArray = hostMap.at("services").GetArraySafe();
+
                 // check if all services of a particular host are alive
                 const bool alive = std::all_of(
-                    hostMap.at("services").GetArraySafe().begin(),
-                    hostMap.at("services").GetArraySafe().end(),
+                    servicesArray.begin(),
+                    servicesArray.end(),
                     [](const auto& service) {
                         return service["health"].GetString() == "ALIVE";
                     }

ydb/core/fq/libs/config/protos/control_plane_storage.proto

@@ -24,11 +24,15 @@ message TQueryMapping {
 
 // 1. StatusCode(s) are handled with defined policies, non-unique StatusCode(s) across all policies is UB
 // 2. RetryCount and RetryPeriodMs are used to calculate actual RetryRate, if it exceeds RetryCount, query is aborted
+//    - Number of retries during RetryPeriod time less than 2 * RetryCount due to RetryRate
 // 3. BackoffPeriodMs is factor of RetryRate to delay query execution before next retry
-// 4. There are no default retry policy, all unhandled statuses are fatal
+// 4. RetryLimit is hard limit for amount query retry count, after that query is aborted
+//    - If RetryLimit = 0, query can be abborted only by RetryRate
+// 5. There are no default retry policy, all unhandled statuses are fatal
 
 message TRetryPolicy {
     uint64 RetryCount = 1;
+    uint64 RetryLimit = 4;
     string RetryPeriod = 2;
     string BackoffPeriod = 3;
 }

ydb/core/fq/libs/control_plane_storage/config.cpp

@@ -50,10 +50,11 @@ TControlPlaneStorageConfig::TControlPlaneStorageConfig(const NConfig::TControlPl
     for (const auto& mapping : Proto.GetRetryPolicyMapping()) {
         auto& retryPolicy = mapping.GetPolicy();
         auto retryCount = retryPolicy.GetRetryCount();
+        auto retryLimit = retryPolicy.GetRetryLimit();
         auto retryPeriod = GetDuration(retryPolicy.GetRetryPeriod(), TDuration::Hours(1));
         auto backoffPeriod = GetDuration(retryPolicy.GetBackoffPeriod(), TDuration::Zero());
         for (const auto statusCode: mapping.GetStatusCode()) {
-            RetryPolicies.emplace(statusCode, TRetryPolicyItem(retryCount, retryPeriod, backoffPeriod));
+            RetryPolicies.emplace(statusCode, TRetryPolicyItem(retryCount, retryLimit, retryPeriod, backoffPeriod));
         }
     }
 

ydb/core/fq/libs/control_plane_storage/internal/task_ping.cpp

@@ -173,7 +173,7 @@ TPingTaskParams ConstructHardPingTask(
                 internal.clear_operation_id();
             }
 
-            TRetryPolicyItem policy(0, TDuration::Seconds(1), TDuration::Zero());
+            TRetryPolicyItem policy(0, 0, TDuration::Seconds(1), TDuration::Zero());
             auto it = retryPolicies.find(request.status_code());
             auto policyFound = it != retryPolicies.end();
             if (policyFound) {
@@ -200,7 +200,7 @@ TPingTaskParams ConstructHardPingTask(
                 TStringBuilder builder;
                 builder << "Query failed with code " << NYql::NDqProto::StatusIds_StatusCode_Name(request.status_code());
                 if (policy.RetryCount) {
-                    builder << " (failure rate " << retryLimiter.RetryRate << " exceeds limit of "  << policy.RetryCount << ")";
+                    builder << " (" << retryLimiter.LastError << ")";
                 }
                 builder << " at " << Now();
 

ydb/core/fq/libs/control_plane_storage/util.cpp

@@ -28,7 +28,16 @@ bool TRetryLimiter::UpdateOnRetry(const TInstant& lastSeenAt, const TRetryPolicy
             RetryRate = 0.0;
         }
     }
-    bool shouldRetry = RetryRate < policy.RetryCount;
+
+    bool shouldRetry = true;
+    if (RetryRate >= policy.RetryCount) {
+        shouldRetry = false;
+        LastError = TStringBuilder() << "failure rate " << RetryRate << " exceeds limit of "  << policy.RetryCount;
+    } else if (policy.RetryLimit && RetryCount >= policy.RetryLimit) {
+        shouldRetry = false;
+        LastError = TStringBuilder() << "retry count reached limit of "  << policy.RetryLimit;
+    }
+
     if (shouldRetry) {
         RetryCount++;
         RetryCounterUpdatedAt = now;
@@ -145,6 +154,7 @@ NConfig::TControlPlaneStorageConfig FillDefaultParameters(NConfig::TControlPlane
             policyMapping.AddStatusCode(NYql::NDqProto::StatusIds::EXTERNAL_ERROR);
             auto& policy = *policyMapping.MutablePolicy();
             policy.SetRetryCount(10);
+            policy.SetRetryLimit(40);
             policy.SetRetryPeriod("1m");
             policy.SetBackoffPeriod("1s");
         }

ydb/core/fq/libs/control_plane_storage/util.h

@@ -15,10 +15,11 @@ namespace NFq {
 class TRetryPolicyItem {
 public:
     TRetryPolicyItem() = default;
-    TRetryPolicyItem(ui64 retryCount, const TDuration& retryPeriod, const TDuration& backoffPeriod)
-    : RetryCount(retryCount), RetryPeriod(retryPeriod), BackoffPeriod(backoffPeriod)
+    TRetryPolicyItem(ui64 retryCount, ui64 retryLimit, const TDuration& retryPeriod, const TDuration& backoffPeriod)
+    : RetryCount(retryCount), RetryLimit(retryLimit), RetryPeriod(retryPeriod), BackoffPeriod(backoffPeriod)
     { }
     ui64 RetryCount = 0;
+    ui64 RetryLimit = 0;
     TDuration RetryPeriod = TDuration::Zero();
     TDuration BackoffPeriod = TDuration::Zero();
 };
@@ -32,6 +33,7 @@ class TRetryLimiter {
     ui64 RetryCount = 0;
     TInstant RetryCounterUpdatedAt = TInstant::Zero();
     double RetryRate = 0.0;
+    TString LastError;
 };
 
 bool IsTerminalStatus(FederatedQuery::QueryMeta::ComputeStatus status);

ydb/core/protos/tx_datashard.proto

@@ -1829,6 +1829,7 @@ message TEvApplyReplicationChangesResult {
         REASON_SCHEME_ERROR = 2;
         REASON_BAD_REQUEST = 3;
         REASON_UNEXPECTED_ROW_OPERATION = 4;
+        REASON_OUTDATED_SCHEME = 6;
     }
 
     optional EStatus Status = 1;

ydb/core/tx/datashard/datashard_repl_apply.cpp

@@ -27,14 +27,16 @@ class TDataShard::TTxApplyReplicationChanges : public TTransactionBase<TDataShar
         if (Self->State != TShardState::Ready) {
             Result = MakeHolder<TEvDataShard::TEvApplyReplicationChangesResult>(
                 NKikimrTxDataShard::TEvApplyReplicationChangesResult::STATUS_REJECTED,
-                NKikimrTxDataShard::TEvApplyReplicationChangesResult::REASON_WRONG_STATE);
+                NKikimrTxDataShard::TEvApplyReplicationChangesResult::REASON_WRONG_STATE,
+                TStringBuilder() << "DataShard is not ready");
             return true;
         }
 
         if (!Self->IsReplicated()) {
             Result = MakeHolder<TEvDataShard::TEvApplyReplicationChangesResult>(
                 NKikimrTxDataShard::TEvApplyReplicationChangesResult::STATUS_REJECTED,
-                NKikimrTxDataShard::TEvApplyReplicationChangesResult::REASON_BAD_REQUEST);
+                NKikimrTxDataShard::TEvApplyReplicationChangesResult::REASON_BAD_REQUEST,
+                TStringBuilder() << "Table is not replicated");
             return true;
         }
 
@@ -65,7 +67,9 @@ class TDataShard::TTxApplyReplicationChanges : public TTransactionBase<TDataShar
                 << " and cannot apply changes for schema version " << tableId.GetSchemaVersion();
             Result = MakeHolder<TEvDataShard::TEvApplyReplicationChangesResult>(
                 NKikimrTxDataShard::TEvApplyReplicationChangesResult::STATUS_REJECTED,
-                NKikimrTxDataShard::TEvApplyReplicationChangesResult::REASON_SCHEME_ERROR,
+                tableId.GetSchemaVersion() < userTable.GetTableSchemaVersion()
+                    ? NKikimrTxDataShard::TEvApplyReplicationChangesResult::REASON_OUTDATED_SCHEME
+                    : NKikimrTxDataShard::TEvApplyReplicationChangesResult::REASON_SCHEME_ERROR,
                 std::move(error));
             return true;
         }

ydb/docs/en/core/_includes/not_allow_for_olap_note.md

@@ -0,0 +1,9 @@
+{% if backend_name == "YDB" %}
+
+{% note warning %}
+
+{% include [OLAP_not_allow_text](not_allow_for_olap_text.md) %}
+
+{% endnote %}
+
+{% endif %}

ydb/docs/en/core/_includes/not_allow_for_olap_text.md

@@ -0,0 +1 @@
+Supported only for [row-oriented](../concepts/datamodel/table.md#row-oriented-tables) tables. Support for [column-oriented](../concepts/datamodel/table.md#column-oriented-tables) tables is currently under development.

ydb/docs/en/core/dev/example-app/_includes/steps/04_query_processing.md

@@ -1,4 +1,4 @@
 ## Retrieving data with a Select {#query-processing}
 
-Retrieving data using a [`SELECT`](../../../../yql/reference/syntax/select.md) statement in [YQL](../../../../yql/reference/index.md). Handling the retrieved data selection in the app.
+Retrieving data using a [`SELECT`](../../../../yql/reference/syntax/select/index.md) statement in [YQL](../../../../yql/reference/index.md). Handling the retrieved data selection in the app.
 

ydb/docs/en/core/dev/yql-tutorial/select_all_columns.md

@@ -1,6 +1,6 @@
 # Selecting data from all columns
 
-Select all columns from the table using [SELECT](../../yql/reference/syntax/select.md):
+Select all columns from the table using [SELECT](../../yql/reference/syntax/select/index.md):
 
 {% include [yql-reference-prerequisites](_includes/yql_tutorial_prerequisites.md) %}
 

ydb/docs/en/core/reference/ydb-sdk/topic.md

@@ -200,7 +200,14 @@ When you update a topic, you must specify the topic path and the parameters to b
 
 - Python
 
-   This feature is under development.
+  Example of updating a topic's list of supported codecs and minimum number of partitions:
+
+  ```python
+  driver.topic_client.create_topic(topic_path,
+      set_supported_codecs=[ydb.TopicCodec.RAW, ydb.TopicCodec.GZIP], # optional
+      set_min_active_partitions=3,                                    # optional
+  )
+  ```
 
 - Java
 

ydb/docs/en/core/yql/reference/yql-core/builtins/_includes/basic/evaluate_expr_atom.md

@@ -1,6 +1,6 @@
 ## EvaluateExpr, EvaluateAtom {#evaluate_expr_atom}
 
-Evaluate an expression before the start of the main calculation and input its result to the query as a literal (constant). In many contexts, where only a constant would be expected in standard SQL (for example, in table names, in the number of rows in [LIMIT](../../../syntax/select.md#limit), and so on), this functionality is implicitly enabled automatically.
+Evaluate an expression before the start of the main calculation and input its result to the query as a literal (constant). In many contexts, where only a constant would be expected in standard SQL (for example, in table names, in the number of rows in [LIMIT](../../../syntax/select/limit_offset.md), and so on), this functionality is implicitly enabled automatically.
 
 EvaluateExpr can be used where the grammar already expects an expression. For example, you can use it to:
 

ydb/docs/en/core/yql/reference/yql-core/syntax/_includes/commit.md

@@ -0,0 +1 @@
+{% include [x](_includes/select/commit.md) %}

ydb/docs/en/core/yql/reference/yql-core/syntax/_includes/discard.md

@@ -1,6 +1,6 @@
 # DISCARD
 
-Calculates {% if select_command == "SELECT STREAM" %}[`SELECT STREAM`](../select_stream.md){% else %}[`SELECT`](../select.md){% endif %}{% if feature_mapreduce %}{% if reduce_command %}, [`{{ reduce_command }}`](../reduce.md){% endif %}, or [`{{ process_command }}`](../process.md){% endif %} without returning the result neither to the client or table. {% if feature_mapreduce %}You can't use it along with [INTO RESULT](../into_result.md).{% endif %}
+Calculates {% if select_command == "SELECT STREAM" %}[`SELECT STREAM`](../select_stream.md){% else %}[`SELECT`](../select/index.md){% endif %}{% if feature_mapreduce %}{% if reduce_command %}, [`{{ reduce_command }}`](../reduce.md){% endif %}, or [`{{ process_command }}`](../process.md){% endif %} without returning the result neither to the client or table. {% if feature_mapreduce %}You can't use it along with [INTO RESULT](../into_result.md).{% endif %}
 
 It's good to combine it with [`Ensure`](../../builtins/basic.md#ensure) to check the final calculation result against the user's criteria.
 

ydb/docs/en/core/yql/reference/yql-core/syntax/_includes/expressions/lambda.md

@@ -6,7 +6,7 @@ List arguments in round brackets, following them by the arrow and lambda functio
 
 The scope for the lambda body: first the local named expressions, then arguments, then named expressions defined above by the lambda function at the top level of the query.
 
-Only use pure expressions inside the lambda body (those might also be other lambdas, possibly passed through arguments). However, you can't use [SELECT](../../select.md), [INSERT INTO](../../insert_into.md), or other top-level expressions.
+Only use pure expressions inside the lambda body (those might also be other lambdas, possibly passed through arguments). However, you can't use [SELECT](../../select/index.md), [INSERT INTO](../../insert_into.md), or other top-level expressions.
 
 One or more of the last lambda parameters can be marked with a question mark as optional: if they haven't been specified when calling lambda, they are assigned the `NULL` value.
 

ydb/docs/en/core/yql/reference/yql-core/syntax/_includes/expressions/tables.md

@@ -10,9 +10,9 @@ A table expression is an expression that returns a table. Table expressions in Y
 
 Semantics of a table expression depends on the context where it is used. In YQL, table expressions can be used in the following contexts:
 
-* Table context: after [FROM](../../select.md#from).
+* Table context: after [FROM](../../select/from.md).
 In this case, table expressions work as expected: for example, `$input = SELECT a, b, c FROM T; SELECT * FROM $input` returns a table with three columns.
-The table context also occurs after [UNION ALL](../../select.md#unionall){% if feature_join %}, [JOIN](../../join.md#join){% endif %}{% if feature_mapreduce and process_command == "PROCESS" %}, [PROCESS](../../process.md#process), [REDUCE](../../reduce.md#reduce){% endif %};
+The table context also occurs after [UNION ALL](../../select/union.md#unionall){% if feature_join %}, [JOIN](../../join.md#join){% endif %}{% if feature_mapreduce and process_command == "PROCESS" %}, [PROCESS](../../process.md#process), [REDUCE](../../reduce.md#reduce){% endif %};
 * Vector context: after [IN](#in). In this context, the table expression must contain exactly one column (the name of this column doesn't affect the expression result in any way).
 A table expression in a vector context is typed as a list (the type of the list element is the same as the column type in this case). Example: `SELECT * FROM T WHERE key IN (SELECT k FROM T1)`;
 * A scalar context arises _in all the other cases_. As in a vector context, a table expression must contain exactly one column, but the value of the table expression is a scalar, that is, an arbitrarily selected value of this column (if no rows are returned, the result is `NULL`). Example: `$count = SELECT COUNT(*) FROM T; SELECT * FROM T ORDER BY key LIMIT $count / 2`;

ydb/docs/en/core/yql/reference/yql-core/syntax/_includes/group_by/distinct.md

@@ -4,7 +4,7 @@ Applying [aggregate functions](../../../builtins/aggregation.md) only to distinc
 
 {% note info %}
 
-Applying `DISTINCT` to calculated values is not currently implemented. For this purpose, you can use a [subquery](../../select.md#from) or the expression `GROUP BY ... AS ...`.
+Applying `DISTINCT` to calculated values is not currently implemented. For this purpose, you can use a [subquery](../../select/from.md) or the expression `GROUP BY ... AS ...`.
 
 {% endnote %}
 
@@ -20,5 +20,5 @@ ORDER BY count DESC
 LIMIT 3;
 ```
 
-You can also use `DISTINCT` to fetch distinct rows using [`SELECT DISTINCT`](../../select.md#distinct).
+You can also use `DISTINCT` to fetch distinct rows using [`SELECT DISTINCT`](../../select/distinct.md).
 

ydb/docs/en/core/yql/reference/yql-core/syntax/_includes/group_by/general.md

@@ -60,7 +60,7 @@ GROUP BY
 
 {% note warning %}
 
-Specifying a name for a column or expression in `GROUP BY .. AS foo` it is an extension on top of YQL. Such a name becomes visible in `WHERE` despite the fact that filtering by `WHERE` is executed [before](../../select.md#selectexec) the grouping. For example, if the `T` table includes two columns, `foo` and `bar`, then the query `SELECT foo FROM T WHERE foo > 0 GROUP BY bar AS foo` would actually filter data by the `bar` column from the source table.
+Specifying a name for a column or expression in `GROUP BY .. AS foo` it is an extension on top of YQL. Such a name becomes visible in `WHERE` despite the fact that filtering by `WHERE` is executed [before](../../select/where.md) the grouping. For example, if the `T` table includes two columns, `foo` and `bar`, then the query `SELECT foo FROM T WHERE foo > 0 GROUP BY bar AS foo` would actually filter data by the `bar` column from the source table.
 
 {% endnote %}
 

ydb/docs/en/core/yql/reference/yql-core/syntax/_includes/group_by/having.md

@@ -1,6 +1,6 @@
 ## HAVING {#having}
 
-Filtering a `SELECT` based on the calculation results of [aggregate functions](../../../builtins/aggregation.md). The syntax is similar to [WHERE](../../select.md#where).
+Filtering a `SELECT` based on the calculation results of [aggregate functions](../../../builtins/aggregation.md). The syntax is similar to [WHERE](../../select/where.md).
 
 **Example**
 

ydb/docs/en/core/yql/reference/yql-core/syntax/_includes/into_result.md

@@ -1,6 +1,6 @@
 # INTO RESULT
 
-Lets you set a custom label for [SELECT](../select.md){% if feature_mapreduce and process_command == "PROCESS" %}, [PROCESS](../process.md), or [REDUCE](../reduce.md){% endif %}. It can't be used along with [DISCARD](../discard.md).
+Lets you set a custom label for [SELECT](../select/index.md){% if feature_mapreduce and process_command == "PROCESS" %}, [PROCESS](../process.md), or [REDUCE](../reduce.md){% endif %}. It can't be used along with [DISCARD](../discard.md).
 
 **Examples:**
 

ydb/docs/en/core/yql/reference/yql-core/syntax/_includes/pragma/global.md

@@ -6,7 +6,7 @@
 | --- | --- |
 | Flag | false |
 
-Automatically run [COMMIT](../../select.md#commit) after every statement.
+Automatically run [COMMIT](../../commit.md) after every statement.
 
 ### TablePathPrefix {#table-path-prefix}
 
@@ -75,7 +75,7 @@ When set to "auto", it enables a new compute engine. Computing is made, whenever
 
 When you use `SELECT foo.* FROM ... AS foo`, remove the `foo.` prefix from the names of the result columns.
 
-It can be also used with a [JOIN](../../join.md), but in this case it may fail in the case of a name conflict (that can be resolved by using [WITHOUT](../../select.md#without) and renaming columns). For JOIN in SimpleColumns mode, an implicit Coalesce is made for key columns: the query `SELECT * FROM T1 AS a JOIN T2 AS b USING(key)` in the SimpleColumns mode works same as `SELECT a.key ?? b.key AS key, ... FROM T1 AS a JOIN T2 AS b USING(key)`.
+It can be also used with a [JOIN](../../join.md), but in this case it may fail in the case of a name conflict (that can be resolved by using [WITHOUT](../../select/without.md) and renaming columns). For JOIN in SimpleColumns mode, an implicit Coalesce is made for key columns: the query `SELECT * FROM T1 AS a JOIN T2 AS b USING(key)` in the SimpleColumns mode works same as `SELECT a.key ?? b.key AS key, ... FROM T1 AS a JOIN T2 AS b USING(key)`.
 
 ### CoalesceJoinKeysOnQualifiedAll
 
@@ -158,11 +158,11 @@ You can explicitly select the old behavior by using the `DisableAnsiOrderByLimit
 
 `OrderedColumns`/`DisableOrderedColumns`
 
-Output the [column order](../../select.md#orderedcolumns) in SELECT/JOIN/UNION ALL and preserve it when writing the results. The order of columns is undefined by default.
+Output the [column order](../../select/order_by.md) in SELECT/JOIN/UNION ALL and preserve it when writing the results. The order of columns is undefined by default.
 
 ### PositionalUnionAll {#positionalunionall}
 
-Enable the standard column-by-column execution for [UNION ALL](../../select.md#unionall). This automatically enables
+Enable the standard column-by-column execution for [UNION ALL](../../select/union.md#unionall). This automatically enables
 [ordered columns](#orderedcolumns).
 
 ### RegexUseRe2

ydb/docs/en/core/yql/reference/yql-core/syntax/_includes/select/assume_order_by.md

@@ -1,4 +1,4 @@
-## ASSUME ORDER BY
+# ASSUME ORDER BY
 
 Checking that the `SELECT` result is sorted by the value in the specified column or multiple columns. The result of such a `SELECT` statement is treated as sorted, but without actually running a sort. Sort check is performed at the query execution stage.
 

ydb/docs/en/core/yql/reference/yql-core/syntax/_includes/select/calc.md

@@ -1,4 +1,4 @@
-## SELECT
+# SELECT
 
 Returns the result of evaluating the expressions specified after `SELECT`.