Merge branch 'antalya-25.8' into bugfix/antalya-25.8/fix_in_for_cluster_request

Author: ianton-ru

docs/en/engines/table-engines/mergetree-family/partition_export.md

@@ -0,0 +1,170 @@
+# ALTER TABLE EXPORT PARTITION
+
+## Overview
+
+The `ALTER TABLE EXPORT PARTITION` command exports entire partitions from Replicated*MergeTree tables to object storage (S3, Azure Blob Storage, etc.), typically in Parquet format. This feature coordinates export part operations across all replicas using ZooKeeper.
+
+Each MergeTree part will become a separate file with the following name convention: `<table_directory>/<partitioning>/<data_part_name>_<merge_tree_part_checksum>.<format>`. To ensure atomicity, a commit file containing the relative paths of all exported parts is also shipped. A data file should only be considered part of the dataset if a commit file references it. The commit file will be named using the following convention: `<table_directory>/commit_<partition_id>_<transaction_id>`.
+
+The set of parts that are exported is based on the list of parts the replica that received the export command sees. The other replicas will assist in the export process if they have those parts locally. Otherwise they will ignore it.
+
+The partition export tasks can be observed through `system.replicated_partition_exports`. Querying this table results in a query to ZooKeeper, so it must be used with care. Individual part export progress can be observed as usual through `system.exports`.
+
+The same partition can not be exported to the same destination more than once. There are two ways to override this behavior: either by setting the `export_merge_tree_partition_force_export` setting or waiting for the task to expire.
+
+The export task can be killed by issuing the kill command: `KILL EXPORT PARTITION <where predicate for system.replicated_partition_exports>`.
+
+The task is persistent - it should be resumed after crashes, failures and etc.
+
+## Syntax
+
+```sql
+ALTER TABLE [database.]table_name 
+EXPORT PARTITION ID 'partition_id' 
+TO TABLE [destination_database.]destination_table 
+[SETTINGS setting_name = value, ...]
+```
+
+### Parameters
+
+- **`table_name`**: The source Replicated*MergeTree table containing the partition to export
+- **`partition_id`**: The partition identifier to export (e.g., `'2020'`, `'2021'`)
+- **`destination_table`**: The target table for the export (typically an S3, Azure, or other object storage table)
+
+## Settings
+
+### Server Settings
+
+#### `enable_experimental_export_merge_tree_partition_feature` (Required)
+
+- **Type**: `Bool`
+- **Default**: `false`
+- **Description**: Enable export replicated merge tree partition feature. It is experimental and not yet ready for production use.
+
+### Query Settings
+
+#### `export_merge_tree_partition_force_export` (Optional)
+
+- **Type**: `Bool`
+- **Default**: `false`
+- **Description**: Ignore existing partition export and overwrite the ZooKeeper entry. Allows re-exporting a partition to the same destination before the manifest expires.
+
+#### `export_merge_tree_partition_max_retries` (Optional)
+
+- **Type**: `UInt64`
+- **Default**: `3`
+- **Description**: Maximum number of retries for exporting a merge tree part in an export partition task. If it exceeds, the entire task fails.
+
+#### `export_merge_tree_partition_manifest_ttl` (Optional)
+
+- **Type**: `UInt64`
+- **Default**: `180` (seconds)
+- **Description**: Determines how long the manifest will live in ZooKeeper. It prevents the same partition from being exported twice to the same destination. This setting does not affect or delete in-progress tasks; it only cleans up completed ones.
+
+#### `export_merge_tree_part_file_already_exists_policy` (Optional)
+
+- **Type**: `MergeTreePartExportFileAlreadyExistsPolicy`
+- **Default**: `skip`
+- **Description**: Policy for handling files that already exist during export. Possible values:
+  - `skip` - Skip the file if it already exists
+  - `error` - Throw an error if the file already exists
+  - `overwrite` - Overwrite the file
+
+## Examples
+
+### Basic Export to S3
+
+```sql
+CREATE TABLE rmt_table (id UInt64, year UInt16) 
+ENGINE = ReplicatedMergeTree('/clickhouse/tables/{database}/rmt_table', 'replica1') 
+PARTITION BY year ORDER BY tuple();
+
+CREATE TABLE s3_table (id UInt64, year UInt16) 
+ENGINE = S3(s3_conn, filename='data', format=Parquet, partition_strategy='hive') 
+PARTITION BY year;
+
+INSERT INTO rmt_table VALUES (1, 2020), (2, 2020), (3, 2020), (4, 2021);
+
+ALTER TABLE rmt_table EXPORT PARTITION ID '2020' TO TABLE s3_table;
+
+## Killing Exports
+
+You can cancel in-progress partition exports using the `KILL EXPORT PARTITION` command:
+
+```sql
+KILL EXPORT PARTITION 
+WHERE partition_id = '2020' 
+  AND source_table = 'rmt_table' 
+  AND destination_table = 's3_table'
+```
+
+The `WHERE` clause filters exports from the `system.replicated_partition_exports` table. You can use any columns from that table in the filter.
+
+## Monitoring
+
+### Active and Completed Exports
+
+Monitor partition exports using the `system.replicated_partition_exports` table:
+
+```sql
+arthur :) select * from system.replicated_partition_exports Format Vertical;
+
+SELECT *
+FROM system.replicated_partition_exports
+FORMAT Vertical
+
+Query id: 9efc271a-a501-44d1-834f-bc4d20156164
+
+Row 1:
+──────
+source_database:      default
+source_table:         replicated_source
+destination_database: default
+destination_table:    replicated_destination
+create_time:          2025-11-21 18:21:51
+partition_id:         2022
+transaction_id:       7397746091717128192
+source_replica:       r1
+parts:                ['2022_0_0_0','2022_1_1_0','2022_2_2_0']
+parts_count:          3
+parts_to_do:          0
+status:               COMPLETED
+exception_replica:    
+last_exception:       
+exception_part:       
+exception_count:      0
+
+Row 2:
+──────
+source_database:      default
+source_table:         replicated_source
+destination_database: default
+destination_table:    replicated_destination
+create_time:          2025-11-21 18:20:35
+partition_id:         2021
+transaction_id:       7397745772618674176
+source_replica:       r1
+parts:                ['2021_0_0_0']
+parts_count:          1
+parts_to_do:          0
+status:               COMPLETED
+exception_replica:    
+last_exception:       
+exception_part:       
+exception_count:      0
+
+2 rows in set. Elapsed: 0.019 sec. 
+
+arthur :) 
+```
+
+Status values include:
+- `PENDING` - Export is queued / in progress
+- `COMPLETED` - Export finished successfully
+- `FAILED` - Export failed
+- `KILLED` - Export was cancelled
+
+## Related Features
+
+- [ALTER TABLE EXPORT PART](/docs/en/engines/table-engines/mergetree-family/part_export.md) - Export individual parts (non-replicated)
+

…le-engines/special/tiered-distributed.md …/engines/table-engines/special/hybrid.mddocs/en/engines/table-engines/special/tiered-distributed.md renamed to docs/en/engines/table-engines/special/hybrid.md docs/en/engines/table-engines/special/tiered-distributed.md renamed to docs/en/engines/table-engines/special/hybrid.md

@@ -1,6 +1,6 @@
 ---
-description: 'Hybrid unions multiple data sources behind per-layer predicates so queries behave like a single table while data is migrated or tiered.'
-slug: /engines/table-engines/special/tiered-distributed
+description: 'Hybrid unions multiple data sources behind per-segment predicates so queries behave like a single table while data is migrated or tiered.'
+slug: /engines/table-engines/special/hybrid
 title: 'Hybrid Table Engine'
 sidebar_label: 'Hybrid'
 sidebar_position: 11
@@ -9,7 +9,7 @@ sidebar_position: 11
 # Hybrid table engine
 
 `Hybrid` builds on top of the [Distributed](./distributed.md) table engine. It lets you expose several data sources as one logical table and assign every source its own predicate.
-The engine rewrites incoming queries so that each layer receives the original query plus its predicate. This keeps all of the Distributed optimisations (remote aggregation, `skip_unused_shards`,
+The engine rewrites incoming queries so that each segment receives the original query plus its predicate. This keeps all of the Distributed optimisations (remote aggregation, `skip_unused_shards`,
 global JOIN pushdown, and so on) while you duplicate or migrate data across clusters, storage types, or formats.
 
 It keeps the same execution pipeline as `engine=Distributed` but can read from multiple underlying sources simultaneously—similar to `engine=Merge`—while still pushing logic down to each source.
@@ -20,7 +20,21 @@ Typical use cases include:
 - Tiered storage, for example fresh data on a local cluster and historical data in S3.
 - Gradual roll-outs where only a subset of rows should be served from a new backend.
 
-By giving mutually exclusive predicates to the layers (for example, `date < watermark` and `date >= watermark`), you ensure that each row is read from exactly one source.
+By giving mutually exclusive predicates to the segments (for example, `date < watermark` and `date >= watermark`), you ensure that each row is read from exactly one source.
+
+## Enable the engine
+
+The Hybrid engine is experimental. Enable it per session (or in the user profile) before creating tables:
+
+```sql
+SET allow_experimental_hybrid_table = 1;
+```
+
+### Automatic Type Alignment
+
+Hybrid segments can evolve independently, so the same logical column may use different physical types. With the experimental `hybrid_table_auto_cast_columns = 1` **(enabled by default and requires `allow_experimental_analyzer = 1`)**, the engine inserts the necessary `CAST` operations into each rewritten query so every shard receives the schema defined by the Hybrid table. You can opt out by setting the flag to `0` if it causes issues.
+
+Segment schemas are cached when you create or attach a Hybrid table. If you alter a segment later (for example change a column type), refresh the Hybrid table (detach/attach or recreate it) so the cached headers stay in sync with the new schema; otherwise the auto-cast feature may miss the change and queries can still fail with header/type errors.
 
 ## Engine definition
 
@@ -39,14 +53,14 @@ You must pass at least two arguments – the first table function and its predic
 ### Arguments and behaviour
 
 - `table_function_n` must be a valid table function (for example `remote`, `remoteSecure`, `cluster`, `clusterAllReplicas`, `s3Cluster`) or a fully qualified table name (`database.table`). The first argument must be a table function—such as `remote` or `cluster`—because it instantiates the underlying `Distributed` storage.
-- `predicate_n` must be an expression that can be evaluated on the table columns. The engine adds it to the layer's query with an additional `AND`, so expressions like `event_date >= '2025-09-01'` or `id BETWEEN 10 AND 15` are typical.
-- The query planner picks the same processing stage for every layer as it does for the base `Distributed` plan, so remote aggregation, ORDER BY pushdown, `skip_unused_shards`, and the legacy/analyzer execution modes behave the same way.
+- `predicate_n` must be an expression that can be evaluated on the table columns. The engine adds it to the segment's query with an additional `AND`, so expressions like `event_date >= '2025-09-01'` or `id BETWEEN 10 AND 15` are typical.
+- The query planner picks the same processing stage for every segment as it does for the base `Distributed` plan, so remote aggregation, ORDER BY pushdown, `skip_unused_shards`, and the legacy/analyzer execution modes behave the same way.
 - `INSERT` statements are forwarded to the first table function only. If you need multi-destination writes, use explicit `INSERT` statements into the respective sources.
-- Align schemas across the layers. ClickHouse builds a common header; if the physical types differ you may need to add casts on one side or in the query, just as you would when reading from heterogeneous replicas.
+- Align schemas across the segments. ClickHouse builds a common header and rejects creation if any segment misses a column defined in the Hybrid schema. If the physical types differ you may need to add casts on one side or in the query, just as you would when reading from heterogeneous replicas.
 
 ## Example: local cluster plus S3 historical tier
 
-The following commands illustrate a two-layer layout. Hot data stays on a local ClickHouse cluster, while historical rows come from public S3 Parquet files.
+The following commands illustrate a two-segment layout. Hot data stays on a local ClickHouse cluster, while historical rows come from public S3 Parquet files.
 
 ```sql
 -- Local MergeTree table that keeps current data
@@ -79,7 +93,7 @@ CREATE OR REPLACE TABLE btc_blocks ENGINE = Hybrid(
     s3('s3://aws-public-blockchain/v1.0/btc/blocks/**.parquet', NOSIGN), date < '2025-09-01'
 ) AS btc_blocks_local;
 
--- Writes target the first (remote) layer
+-- Writes target the first (remote) segment
 INSERT INTO btc_blocks
 SELECT *
 FROM s3('s3://aws-public-blockchain/v1.0/btc/blocks/**.parquet', NOSIGN)
@@ -103,4 +117,4 @@ GROUP BY date
 ORDER BY date;
 ```
 
-Because the predicates are applied inside every layer, queries such as `ORDER BY`, `GROUP BY`, `LIMIT`, `JOIN`, and `EXPLAIN` behave as if you were reading from a single `Distributed` table. When sources expose different physical types (for example `FixedString(64)` versus `String` in Parquet), add explicit casts during ingestion or in the query, as shown above.
+Because the predicates are applied inside every segment, queries such as `ORDER BY`, `GROUP BY`, `LIMIT`, `JOIN`, and `EXPLAIN` behave as if you were reading from a single `Distributed` table. When sources expose different physical types (for example `FixedString(64)` versus `String` in Parquet), add explicit casts during ingestion or in the query, as shown above.

src/Analyzer/Passes/HybridCastsPass.cpp

@@ -0,0 +1,150 @@
+#include <Analyzer/Passes/HybridCastsPass.h>
+
+#include <Analyzer/QueryTreeBuilder.h>
+#include <Analyzer/QueryTreePassManager.h>
+#include <Analyzer/Passes/QueryAnalysisPass.h>
+#include <Analyzer/Utils.h>
+#include <Analyzer/Resolve/IdentifierResolver.h>
+#include <Analyzer/QueryNode.h>
+#include <Analyzer/TableNode.h>
+#include <Analyzer/UnionNode.h>
+#include <Analyzer/FunctionNode.h>
+#include <Analyzer/ColumnNode.h>
+#include <Analyzer/InDepthQueryTreeVisitor.h>
+
+#include <Storages/IStorage.h>
+#include <Storages/StorageDistributed.h>
+
+#include <Core/Settings.h>
+#include <Core/SettingsEnums.h>
+#include <Common/Exception.h>
+
+namespace DB
+{
+
+namespace Setting
+{
+    extern const SettingsBool hybrid_table_auto_cast_columns;
+}
+
+namespace ErrorCodes
+{
+    extern const int LOGICAL_ERROR;
+}
+
+namespace
+{
+
+/// Collect Hybrid table expressions that require casts to normalize headers across segments.
+///
+/// Hybrid is currently exposed only as an engine (TableNode). If it ever gets a table function
+/// wrapper, this visitor must also look at TableFunctionNode and unwrap to the underlying
+/// StorageDistributed so cached casts can be picked up there as well.
+class HybridCastTablesCollector : public InDepthQueryTreeVisitor<HybridCastTablesCollector>
+{
+public:
+    explicit HybridCastTablesCollector(std::unordered_map<const IQueryTreeNode *, ColumnsDescription> & cast_map_)
+        : cast_map(cast_map_)
+    {}
+
+    static bool needChildVisit(QueryTreeNodePtr &, QueryTreeNodePtr &) { return true; }
+
+    void visitImpl(QueryTreeNodePtr & node)
+    {
+        const auto * table = node->as<TableNode>();
+        if (!table)
+            return;
+
+        const auto * storage = table->getStorage().get();
+        if (const auto * distributed = typeid_cast<const StorageDistributed *>(storage))
+        {
+            ColumnsDescription to_cast = distributed->getColumnsToCast();
+            if (!to_cast.empty())
+                cast_map.emplace(node.get(), std::move(to_cast)); // repeated table_expression can overwrite
+        }
+    }
+
+private:
+    std::unordered_map<const IQueryTreeNode *, ColumnsDescription> & cast_map;
+};
+
+// Visitor replaces all usages of the column with CAST(column, type) in the query tree.
+class HybridCastVisitor : public InDepthQueryTreeVisitor<HybridCastVisitor>
+{
+public:
+    HybridCastVisitor(
+        const std::unordered_map<const IQueryTreeNode *, ColumnsDescription> & cast_map_,
+        ContextPtr context_)
+        : cast_map(cast_map_)
+        , context(std::move(context_))
+    {}
+
+    bool shouldTraverseTopToBottom() const { return false; }
+
+    static bool needChildVisit(QueryTreeNodePtr &, QueryTreeNodePtr & child)
+    {
+        /// Traverse all child nodes so casts also apply inside subqueries and UNION branches.
+        (void)child;
+        return true;
+    }
+
+    void visitImpl(QueryTreeNodePtr & node)
+    {
+        auto * column_node = node->as<ColumnNode>();
+        if (!column_node)
+            return;
+
+        auto column_source = column_node->getColumnSourceOrNull();
+        if (!column_source)
+            return;
+
+        auto it = cast_map.find(column_source.get());
+        if (it == cast_map.end())
+            return;
+
+        const auto & column_name = column_node->getColumnName();
+        auto expected_column_opt = it->second.tryGetPhysical(column_name);
+        if (!expected_column_opt)
+            return;
+
+        auto column_clone = std::static_pointer_cast<ColumnNode>(column_node->clone());
+
+        auto cast_node = buildCastFunction(column_clone, expected_column_opt->type, context);
+        const auto & alias = node->getAlias();
+        if (!alias.empty())
+            cast_node->setAlias(alias);
+        else
+            cast_node->setAlias(expected_column_opt->name);
+
+        node = cast_node;
+    }
+
+private:
+    const std::unordered_map<const IQueryTreeNode *, ColumnsDescription> & cast_map;
+    ContextPtr context;
+};
+
+
+} // namespace
+
+void HybridCastsPass::run(QueryTreeNodePtr & query_tree_node, ContextPtr context)
+{
+    const auto & settings = context->getSettingsRef();
+    if (!settings[Setting::hybrid_table_auto_cast_columns])
+        return;
+
+    auto * query = query_tree_node->as<QueryNode>();
+    if (!query)
+        return;
+
+    std::unordered_map<const IQueryTreeNode *, ColumnsDescription> cast_map;
+    HybridCastTablesCollector collector(cast_map);
+    collector.visit(query_tree_node);
+    if (cast_map.empty())
+        return;
+
+    HybridCastVisitor visitor(cast_map, context);
+    visitor.visit(query_tree_node);
+}
+
+}