Merge branch 'antalya-25.8' into backports/antalya-25.8/87600

Enmk · web-flow · commit d13eab3764c1 · 2025-11-21T17:19:45.000+01:00
diff --git a/.github/actions/create_workflow_report/create_workflow_report.py b/.github/actions/create_workflow_report/create_workflow_report.py
@@ -229,10 +229,9 @@ def get_checks_known_fails(
     if len(known_fails) == 0:
         return pd.DataFrame()
 
-    query = f"""SELECT job_status, job_name, status as test_status, test_name, results_link
+    query = f"""SELECT job_name, status as test_status, test_name, results_link
         FROM (
             SELECT
-                argMax(check_status, check_start_time) as job_status,
                 check_name as job_name,
                 argMax(test_status, check_start_time) as status,
                 test_name,
diff --git a/ci/jobs/scripts/functional_tests_results.py b/ci/jobs/scripts/functional_tests_results.py
@@ -356,7 +356,7 @@ def run(self):
             pass
 
         if not info:
-            info = f"Failed: {s.failed}, Passed: {s.success}, Skipped: {s.skipped}"
+            info = f"Failed: {s.failed}, Passed: {s.success}, Skipped: {s.skipped}, Broken: {s.broken}"
 
         result = Result.create_from(
             name="Tests",
diff --git a/ci/jobs/scripts/integration_tests_runner.py b/ci/jobs/scripts/integration_tests_runner.py
@@ -1380,7 +1380,7 @@ def run_normal_check(self):
                 for c in counters[state]
             ]
         failed_sum = len(counters["FAILED"]) + len(counters["ERROR"])
-        status_text = f"fail: {failed_sum}, passed: {len(counters['PASSED'])}"
+        status_text = f"fail: {failed_sum}, passed: {len(counters['PASSED'])}, broken: {len(counters['BROKEN'])}"
 
         if not counters or sum(len(counter) for counter in counters.values()) == 0:
             status_text = "No tests found for some reason! It's a bug"
diff --git a/docs/en/engines/table-engines/mergetree-family/part_export.md b/docs/en/engines/table-engines/mergetree-family/part_export.md
@@ -0,0 +1,160 @@
+# ALTER TABLE EXPORT PART
+
+## Overview
+
+The `ALTER TABLE EXPORT PART` command exports individual MergeTree data parts to object storage (S3, Azure Blob Storage, etc.), typically in Parquet format.
+
+**Key Characteristics:**
+- **Experimental feature** - must be enabled via `allow_experimental_export_merge_tree_part` setting
+- **Asynchronous** - executes in the background, returns immediately
+- **Ephemeral** - no automatic retry mechanism; manual retry required on failure
+- **Idempotent** - safe to re-export the same part (skips by default if file exists)
+- **Preserves sort order** from the source table
+
+## Syntax
+
+```sql
+ALTER TABLE [database.]table_name 
+EXPORT PART 'part_name' 
+TO TABLE [destination_database.]destination_table 
+SETTINGS allow_experimental_export_merge_tree_part = 1 
+         [, setting_name = value, ...]
+```
+
+### Parameters
+
+- **`table_name`**: The source MergeTree table containing the part to export
+- **`part_name`**: The exact name of the data part to export (e.g., `'2020_1_1_0'`, `'all_1_1_0'`)
+- **`destination_table`**: The target table for the export (typically an S3, Azure, or other object storage table)
+
+## Requirements
+
+Source and destination tables must be 100% compatible:
+
+1. **Identical schemas** - same columns, types, and order
+2. **Matching partition keys** - partition expressions must be identical
+
+## Settings
+
+### `allow_experimental_export_merge_tree_part` (Required)
+
+- **Type**: `Bool`
+- **Default**: `false`
+- **Description**: Must be set to `true` to enable the experimental feature.
+
+### `export_merge_tree_part_overwrite_file_if_exists` (Optional)
+
+- **Type**: `Bool`
+- **Default**: `false`
+- **Description**: If set to `true`, it will overwrite the file. Otherwise, fails with exception.
+
+## Examples
+
+### Basic Export to S3
+
+```sql
+-- Create source and destination tables
+CREATE TABLE mt_table (id UInt64, year UInt16) 
+ENGINE = MergeTree() 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 and export
+INSERT INTO mt_table VALUES (1, 2020), (2, 2020), (3, 2021);
+
+ALTER TABLE mt_table EXPORT PART '2020_1_1_0' TO TABLE s3_table 
+SETTINGS allow_experimental_export_merge_tree_part = 1;
+
+ALTER TABLE mt_table EXPORT PART '2021_2_2_0' TO TABLE s3_table 
+SETTINGS allow_experimental_export_merge_tree_part = 1;
+```
+
+## Monitoring
+
+### Active Exports
+
+Active exports can be found in the `system.exports` table. As of now, it only shows currently executing exports. It will not show pending or finished exports.
+
+```sql
+arthur :) select * from system.exports;
+
+SELECT *
+FROM system.exports
+
+Query id: 2026718c-d249-4208-891b-a271f1f93407
+
+Row 1:
+──────
+source_database:               default
+source_table:                  source_mt_table
+destination_database:          default
+destination_table:             destination_table
+create_time:                   2025-11-19 09:09:11
+part_name:                     20251016-365_1_1_0
+destination_file_path:         table_root/eventDate=2025-10-16/retention=365/20251016-365_1_1_0_17B2F6CD5D3C18E787C07AE3DAF16EB1.parquet
+elapsed:                       2.04845441
+rows_read:                     1138688 -- 1.14 million
+total_rows_to_read:            550961374 -- 550.96 million
+total_size_bytes_compressed:   37619147120 -- 37.62 billion
+total_size_bytes_uncompressed: 138166213721 -- 138.17 billion
+bytes_read_uncompressed:       316892925 -- 316.89 million
+memory_usage:                  596006095 -- 596.01 million
+peak_memory_usage:             601239033 -- 601.24 million
+```
+
+### Export History
+
+You can query succeeded or failed exports in `system.part_log`. For now, it only keeps track of completion events (either success or fails).
+
+```sql
+arthur :) select * from system.part_log where event_type='ExportPart' and table = 'replicated_source' order by event_time desc limit 1;
+
+SELECT *
+FROM system.part_log
+WHERE (event_type = 'ExportPart') AND (`table` = 'replicated_source')
+ORDER BY event_time DESC
+LIMIT 1
+
+Query id: ae1c1cd3-c20e-4f20-8b82-ed1f6af0237f
+
+Row 1:
+──────
+hostname:                arthur
+query_id:                
+event_type:              ExportPart
+merge_reason:            NotAMerge
+merge_algorithm:         Undecided
+event_date:              2025-11-19
+event_time:              2025-11-19 09:08:31
+event_time_microseconds: 2025-11-19 09:08:31.974701
+duration_ms:             4
+database:                default
+table:                   replicated_source
+table_uuid:              78471c67-24f4-4398-9df5-ad0a6c3daf41
+part_name:               2021_0_0_0
+partition_id:            2021
+partition:               2021
+part_type:               Compact
+disk_name:               default
+path_on_disk:            year=2021/2021_0_0_0_78C704B133D41CB0EF64DD2A9ED3B6BA.parquet
+rows:                    1
+size_in_bytes:           272
+merged_from:             ['2021_0_0_0']
+bytes_uncompressed:      86
+read_rows:               1
+read_bytes:              6
+peak_memory_usage:       22
+error:                   0
+exception:               
+ProfileEvents:           {}
+```
+
+### Profile Events
+
+- `PartsExports` - Successful exports
+- `PartsExportFailures` - Failed exports
+- `PartsExportDuplicated` - Number of part exports that failed because target already exists.
+- `PartsExportTotalMilliseconds` - Total time
+
diff --git a/docs/en/operations/system-tables/exports.md b/docs/en/operations/system-tables/exports.md
@@ -0,0 +1,56 @@
+---
+description: 'System table containing information about in progress merge tree part exports'
+keywords: ['system table', 'exports', 'merge tree', 'part']
+slug: /operations/system-tables/exports
+title: 'system.exports'
+---
+
+Contains information about in progress merge tree part exports
+
+Columns:
+
+- `source_database` ([String](/docs/en/sql-reference/data-types/string.md)) — Name of the source database.
+- `source_table` ([String](/docs/en/sql-reference/data-types/string.md)) — Name of the source table.
+- `destination_database` ([String](/docs/en/sql-reference/data-types/string.md)) — Name of the destination database.
+- `destination_table` ([String](/docs/en/sql-reference/data-types/string.md)) — Name of the destination table.
+- `create_time` ([DateTime](/docs/en/sql-reference/data-types/datetime.md)) — Date and time when the export command was received in the server.
+- `part_name` ([String](/docs/en/sql-reference/data-types/string.md)) — Name of the part.
+- `destination_file_path` ([String](/docs/en/sql-reference/data-types/string.md)) — File path relative to where the part is being exported to.
+- `elapsed` ([Float64](/docs/en/sql-reference/data-types/float.md)) — The time elapsed (in seconds) since the export started.
+- `rows_read` ([UInt64](/docs/en/sql-reference/data-types/int-uint.md)) — The number of rows read from the exported part.
+- `total_rows_to_read` ([UInt64](/docs/en/sql-reference/data-types/int-uint.md)) — The total number of rows to read from the exported part.
+- `total_size_bytes_compressed` ([UInt64](/docs/en/sql-reference/data-types/int-uint.md)) — The total size of the compressed data in the exported part.
+- `total_size_bytes_uncompressed` ([UInt64](/docs/en/sql-reference/data-types/int-uint.md)) — The total size of the uncompressed data in the exported part.
+- `bytes_read_uncompressed` ([UInt64](/docs/en/sql-reference/data-types/int-uint.md)) — The number of uncompressed bytes read from the exported part.
+- `memory_usage` ([UInt64](/docs/en/sql-reference/data-types/int-uint.md)) — Current memory usage in bytes for the export operation.
+- `peak_memory_usage` ([UInt64](/docs/en/sql-reference/data-types/int-uint.md)) — Peak memory usage in bytes during the export operation.
+
+**Example**
+
+```sql
+arthur :) select * from system.exports;
+
+SELECT *
+FROM system.exports
+
+Query id: 2026718c-d249-4208-891b-a271f1f93407
+
+Row 1:
+──────
+source_database:               default
+source_table:                  source_mt_table
+destination_database:          default
+destination_table:             destination_table
+create_time:                   2025-11-19 09:09:11
+part_name:                     20251016-365_1_1_0
+destination_file_path:         table_root/eventDate=2025-10-16/retention=365/20251016-365_1_1_0_17B2F6CD5D3C18E787C07AE3DAF16EB1.parquet
+elapsed:                       2.04845441
+rows_read:                     1138688 -- 1.14 million
+total_rows_to_read:            550961374 -- 550.96 million
+total_size_bytes_compressed:   37619147120 -- 37.62 billion
+total_size_bytes_uncompressed: 138166213721 -- 138.17 billion
+bytes_read_uncompressed:       316892925 -- 316.89 million
+memory_usage:                  596006095 -- 596.01 million
+peak_memory_usage:             601239033 -- 601.24 million
+```
+
diff --git a/src/Databases/DataLake/GlueCatalog.cpp b/src/Databases/DataLake/GlueCatalog.cpp
@@ -485,7 +485,7 @@ bool GlueCatalog::classifyTimestampTZ(const String & column_name, const TableMet
         DB::StoredObject metadata_stored_object(metadata_path);
         auto read_buf = object_storage->readObject(metadata_stored_object, read_settings);
         String metadata_file;
-        readString(metadata_file, *read_buf);
+        readStringUntilEOF(metadata_file, *read_buf);
 
         Poco::JSON::Parser parser;
         Poco::Dynamic::Var result = parser.parse(metadata_file);
diff --git a/src/Storages/MergeTree/ExportPartTask.cpp b/src/Storages/MergeTree/ExportPartTask.cpp
@@ -47,9 +47,9 @@ ExportPartTask::ExportPartTask(MergeTreeData & storage_, const MergeTreePartExpo
 
 bool ExportPartTask::executeStep()
 {
-    auto metadata_snapshot = storage.getInMemoryMetadataPtr();
+    const auto & metadata_snapshot = manifest.storage_snapshot->metadata;
+
     Names columns_to_read = metadata_snapshot->getColumns().getNamesOfPhysical();
-    StorageSnapshotPtr storage_snapshot = storage.getStorageSnapshot(metadata_snapshot, local_context);
 
     MergeTreeSequentialSourceType read_type = MergeTreeSequentialSourceType::Export;
 
@@ -142,13 +142,8 @@ bool ExportPartTask::executeStep()
     bool read_with_direct_io = local_context->getSettingsRef()[Setting::min_bytes_to_use_direct_io] > manifest.data_part->getBytesOnDisk();
     bool prefetch = false;
 
-    MergeTreeData::IMutationsSnapshot::Params params
-    {
-        .metadata_version = metadata_snapshot->getMetadataVersion(),
-        .min_part_metadata_version = manifest.data_part->getMetadataVersion(),
-    };
-
-    auto mutations_snapshot = storage.getMutationsSnapshot(params);
+    const auto & snapshot_data = assert_cast<const MergeTreeData::SnapshotData &>(*manifest.storage_snapshot->data);
+    auto mutations_snapshot = snapshot_data.mutations_snapshot;
 
     auto alter_conversions = MergeTreeData::getAlterConversionsForPart(
         manifest.data_part,
@@ -161,7 +156,7 @@ bool ExportPartTask::executeStep()
         read_type,
         plan_for_part,
         storage,
-        storage_snapshot,
+        manifest.storage_snapshot,
         RangesInDataPart(manifest.data_part),
         alter_conversions,
         nullptr,
diff --git a/src/Storages/MergeTree/MergeTreeData.cpp b/src/Storages/MergeTree/MergeTreeData.cpp
@@ -6239,13 +6239,13 @@ void MergeTreeData::exportPartToTable(
         return ast ? ast->formatWithSecretsOneLine() : "";
     };
 
-    auto src_snapshot = getInMemoryMetadataPtr();
-    auto destination_snapshot = dest_storage->getInMemoryMetadataPtr();
+    auto source_metadata_ptr = getInMemoryMetadataPtr();
+    auto destination_metadata_ptr = dest_storage->getInMemoryMetadataPtr();
 
-    if (destination_snapshot->getColumns().getAllPhysical().sizeOfDifference(src_snapshot->getColumns().getAllPhysical()))
+    if (destination_metadata_ptr->getColumns().getAllPhysical().sizeOfDifference(source_metadata_ptr->getColumns().getAllPhysical()))
         throw Exception(ErrorCodes::INCOMPATIBLE_COLUMNS, "Tables have different structure");
 
-    if (query_to_string(src_snapshot->getPartitionKeyAST()) != query_to_string(destination_snapshot->getPartitionKeyAST()))
+    if (query_to_string(source_metadata_ptr->getPartitionKeyAST()) != query_to_string(destination_metadata_ptr->getPartitionKeyAST()))
         throw Exception(ErrorCodes::BAD_ARGUMENTS, "Tables have different partition key");
 
     auto part = getPartIfExists(part_name, {MergeTreeDataPartState::Active, MergeTreeDataPartState::Outdated});
@@ -6262,6 +6262,7 @@ void MergeTreeData::exportPartToTable(
             transaction_id,
             query_context->getSettingsRef()[Setting::export_merge_tree_part_file_already_exists_policy].value,
             format_settings,
+            getStorageSnapshot(source_metadata_ptr, query_context),
             completion_callback);
 
         std::lock_guard lock(export_manifests_mutex);
diff --git a/src/Storages/MergeTree/MergeTreePartExportManifest.h b/src/Storages/MergeTree/MergeTreePartExportManifest.h
@@ -2,6 +2,7 @@
 
 #include <Interpreters/StorageID.h>
 #include <Storages/MergeTree/IMergeTreeDataPart.h>
+#include <Storages/StorageSnapshot.h>
 #include <QueryPipeline/QueryPipeline.h>
 #include <optional>
 
@@ -46,12 +47,14 @@ struct MergeTreePartExportManifest
         const String & transaction_id_,
         FileAlreadyExistsPolicy file_already_exists_policy_,
         const FormatSettings & format_settings_,
+        const StorageSnapshotPtr & storage_snapshot_,
         std::function<void(CompletionCallbackResult)> completion_callback_ = {})
         : destination_storage_id(destination_storage_id_),
           data_part(data_part_),
           transaction_id(transaction_id_),
           file_already_exists_policy(file_already_exists_policy_),
           format_settings(format_settings_),
+          storage_snapshot(storage_snapshot_),
           completion_callback(completion_callback_),
           create_time(time(nullptr)) {}
 
@@ -62,6 +65,10 @@ struct MergeTreePartExportManifest
     FileAlreadyExistsPolicy file_already_exists_policy;
     FormatSettings format_settings;
 
+    /// Storage snapshot captured at the time of query validation to prevent race conditions with mutations
+    /// Otherwise the export could fail if the schema changes between validation and execution
+    StorageSnapshotPtr storage_snapshot;
+
     std::function<void(CompletionCallbackResult)> completion_callback;
 
     time_t create_time;
diff --git a/src/Storages/MergeTree/tests/gtest_export_partition_ordering.cpp b/src/Storages/MergeTree/tests/gtest_export_partition_ordering.cpp
@@ -26,18 +26,21 @@ TEST_F(ExportPartitionOrderingTest, IterationOrderMatchesCreateTime)
     manifest1.partition_id = "2020";
     manifest1.destination_database = "db1";
     manifest1.destination_table = "table1";
+    manifest1.transaction_id = "tx1";
     manifest1.create_time = base_time + 300; // Latest
     
     ExportReplicatedMergeTreePartitionManifest manifest2;
     manifest2.partition_id = "2021";
     manifest2.destination_database = "db1";
     manifest2.destination_table = "table1";
+    manifest2.transaction_id = "tx2";
     manifest2.create_time = base_time + 100; // Middle
     
     ExportReplicatedMergeTreePartitionManifest manifest3;
     manifest3.partition_id = "2022";
     manifest3.destination_database = "db1";
     manifest3.destination_table = "table1";
+    manifest3.transaction_id = "tx3";
     manifest3.create_time = base_time; // Oldest
 
     ExportReplicatedMergeTreePartitionTaskEntry entry1{manifest1, ExportReplicatedMergeTreePartitionTaskEntry::Status::PENDING, {}};
diff --git a/src/Storages/System/StorageSystemExports.cpp b/src/Storages/System/StorageSystemExports.cpp
@@ -20,7 +20,7 @@ ColumnsDescription StorageSystemExports::getColumnsDescription()
         {"source_table", std::make_shared<DataTypeString>(), "Name of the source table."},
         {"destination_database", std::make_shared<DataTypeString>(), "Name of the destination database."},
         {"destination_table", std::make_shared<DataTypeString>(), "Name of the destination table."},
-        {"create_time", std::make_shared<DataTypeDateTime>(), "Date and time when the export command was submitted for execution."},
+        {"create_time", std::make_shared<DataTypeDateTime>(), "Date and time when the export command was received in the server."},
         {"part_name", std::make_shared<DataTypeString>(), "Name of the part"},
         {"destination_file_path", std::make_shared<DataTypeString>(), "File path where the part is being exported."},
         {"elapsed", std::make_shared<DataTypeFloat64>(), "The time elapsed (in seconds) since the export started."},
diff --git a/tests/integration/test_export_merge_tree_part_to_object_storage/__init__.py b/tests/integration/test_export_merge_tree_part_to_object_storage/__init__.py
diff --git a/tests/integration/test_export_merge_tree_part_to_object_storage/configs/named_collections.xml b/tests/integration/test_export_merge_tree_part_to_object_storage/configs/named_collections.xml
@@ -0,0 +1,9 @@
+<clickhouse>
+    <named_collections>
+        <s3_conn>
+            <url>http://minio1:9001/root/data</url>
+            <access_key_id>minio</access_key_id>
+            <secret_access_key>ClickHouse_Minio_P@ssw0rd</secret_access_key>
+        </s3_conn>
+    </named_collections>
+</clickhouse>
diff --git a/tests/integration/test_export_merge_tree_part_to_object_storage/test.py b/tests/integration/test_export_merge_tree_part_to_object_storage/test.py
diff --git a/tests/integration/test_export_replicated_mt_partition_to_object_storage/test.py b/tests/integration/test_export_replicated_mt_partition_to_object_storage/test.py

Original file line number	Diff line number	Diff line change
`@@ -1380,7 +1380,7 @@ def run_normal_check(self):`
`1380`	`1380`	`for c in counters[state]`
`1381`	`1381`	`]`
`1382`	`1382`	`failed_sum = len(counters["FAILED"]) + len(counters["ERROR"])`
`1383`		`- status_text = f"fail: {failed_sum}, passed: {len(counters['PASSED'])}"`
	`1383`	`+ status_text = f"fail: {failed_sum}, passed: {len(counters['PASSED'])}, broken: {len(counters['BROKEN'])}"`
`1384`	`1384`
`1385`	`1385`	`if not counters or sum(len(counter) for counter in counters.values()) == 0:`
`1386`	`1386`	`status_text = "No tests found for some reason! It's a bug"`