-
Notifications
You must be signed in to change notification settings - Fork 19
Plain merge tree partition exports #2290
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: antalya-26.6
Are you sure you want to change the base?
Changes from 6 commits
69f071a
3190603
d94ea53
b2af06c
f2ee03f
240da81
9ff4a4a
6c86d2f
a622f44
23d90fa
fe7e61a
7c959fb
f06a7dd
7482d6b
c8d038d
875cc47
9496d85
3af519a
31b04b1
01b5d38
a0badb4
2eab452
cd502a6
3fc5e5f
cae460a
d4a3859
b50cee3
b89bfa1
00b1540
f93a925
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -265,65 +265,119 @@ BlockIO InterpreterKillQueryQuery::execute() | |
| "Exporting merge tree partition is experimental. Set the server setting `allow_experimental_export_merge_tree_partition` to enable it"); | ||
| } | ||
|
|
||
| Block exports_block = getSelectResult( | ||
| "source_database, source_table, transaction_id, destination_database, destination_table, partition_id", | ||
| "system.replicated_partition_exports"); | ||
| if (exports_block.empty()) | ||
| return res_io; | ||
| const String export_columns = "source_database, source_table, transaction_id, destination_database, destination_table, partition_id"; | ||
|
|
||
| /// Partition exports live in two system tables: `replicated_partition_exports` for | ||
| /// ReplicatedMergeTree and `partition_exports` for plain MergeTree. Query both so a single | ||
| /// KILL EXPORT PARTITION targets either engine. Each query applies the user WHERE to its own | ||
| /// table (which exposes all of its columns); a WHERE that references columns specific to the | ||
| /// other engine will fail against the table that lacks them. We tolerate that per-table so a | ||
| /// predicate on engine-specific columns still reaches the matching engine — but if BOTH reads | ||
| /// fail (e.g. a genuinely bad predicate / unknown column), we rethrow so the error surfaces. | ||
| Block replicated_exports_block; | ||
| Block plain_exports_block; | ||
| std::exception_ptr replicated_error; | ||
| std::exception_ptr plain_error; | ||
|
|
||
| try | ||
| { | ||
| replicated_exports_block = getSelectResult(export_columns, "system.replicated_partition_exports"); | ||
| } | ||
| catch (...) | ||
| { | ||
| replicated_error = std::current_exception(); | ||
| tryLogCurrentException(getLogger("InterpreterKillQueryQuery"), | ||
| "KILL EXPORT PARTITION: could not read system.replicated_partition_exports (the WHERE may " | ||
| "reference columns that only exist for plain MergeTree); ignoring ReplicatedMergeTree tables"); | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
If either system-table query fails for an operational reason such as a memory limit, timeout, or internal exception, this broad catch treats it like an engine-specific unknown column and continues with the other table. The command can therefore report successful results while silently leaving matching exports from the failed table running. Only suppress the specific predicate-resolution error that proves the filter belongs exclusively to the other table; propagate all other failures. AGENTS.md reference: AGENTS.md:L153-L153 Useful? React with 👍 / 👎. |
||
| } | ||
|
|
||
| try | ||
| { | ||
| plain_exports_block = getSelectResult(export_columns, "system.partition_exports"); | ||
| } | ||
| catch (...) | ||
| { | ||
| plain_error = std::current_exception(); | ||
| tryLogCurrentException(getLogger("InterpreterKillQueryQuery"), | ||
| "KILL EXPORT PARTITION: could not read system.partition_exports (the WHERE may reference " | ||
| "columns that only exist for ReplicatedMergeTree); ignoring plain MergeTree tables"); | ||
| } | ||
|
|
||
| const ColumnString & src_db_col = typeid_cast<const ColumnString &>(*exports_block.getByName("source_database").column); | ||
| const ColumnString & src_table_col = typeid_cast<const ColumnString &>(*exports_block.getByName("source_table").column); | ||
| const ColumnString & dst_db_col = typeid_cast<const ColumnString &>(*exports_block.getByName("destination_database").column); | ||
| const ColumnString & dst_table_col = typeid_cast<const ColumnString &>(*exports_block.getByName("destination_table").column); | ||
| const ColumnString & tx_col = typeid_cast<const ColumnString &>(*exports_block.getByName("transaction_id").column); | ||
| /// Neither table could be read: the predicate is invalid for both, so surface the error | ||
| /// instead of silently matching nothing. | ||
| if (replicated_error && plain_error) | ||
| std::rethrow_exception(plain_error); | ||
|
|
||
| if (replicated_exports_block.empty() && plain_exports_block.empty()) | ||
| return res_io; | ||
|
|
||
| auto header = exports_block.cloneEmpty(); | ||
| /// Build the result header explicitly from the fixed projection so it does not depend on | ||
| /// whether either source block ended up with rows. | ||
| Block header; | ||
| for (const auto & column_name : {"source_database", "source_table", "transaction_id", | ||
| "destination_database", "destination_table", "partition_id"}) | ||
| header.insert({ColumnString::create(), std::make_shared<DataTypeString>(), column_name}); | ||
| header.insert(0, {ColumnString::create(), std::make_shared<DataTypeString>(), "kill_status"}); | ||
|
|
||
| MutableColumns res_columns = header.cloneEmptyColumns(); | ||
| AccessRightsElements required_access_rights; | ||
| auto access = getContext()->getAccess(); | ||
| bool access_denied = false; | ||
|
|
||
| for (size_t i = 0; i < exports_block.rows(); ++i) | ||
| auto process_block = [&](const Block & exports_block) | ||
| { | ||
| const auto src_database = src_db_col.getDataAt(i); | ||
| const auto src_table = src_table_col.getDataAt(i); | ||
| const auto dst_database = dst_db_col.getDataAt(i); | ||
| const auto dst_table = dst_table_col.getDataAt(i); | ||
| if (exports_block.empty()) | ||
| return; | ||
|
|
||
| const auto table_id = StorageID{std::string{src_database}, std::string{src_table}}; | ||
| const auto transaction_id = tx_col.getDataAt(i); | ||
| const ColumnString & src_db_col = typeid_cast<const ColumnString &>(*exports_block.getByName("source_database").column); | ||
| const ColumnString & src_table_col = typeid_cast<const ColumnString &>(*exports_block.getByName("source_table").column); | ||
| const ColumnString & dst_db_col = typeid_cast<const ColumnString &>(*exports_block.getByName("destination_database").column); | ||
| const ColumnString & dst_table_col = typeid_cast<const ColumnString &>(*exports_block.getByName("destination_table").column); | ||
| const ColumnString & tx_col = typeid_cast<const ColumnString &>(*exports_block.getByName("transaction_id").column); | ||
|
|
||
| CancellationCode code = CancellationCode::Unknown; | ||
| if (!query.test) | ||
| for (size_t i = 0; i < exports_block.rows(); ++i) | ||
| { | ||
| auto storage = DatabaseCatalog::instance().tryGetTable(table_id, getContext()); | ||
| if (!storage) | ||
| code = CancellationCode::NotFound; | ||
| else | ||
| { | ||
| ASTAlterCommand alter_command{}; | ||
| alter_command.type = ASTAlterCommand::EXPORT_PARTITION; | ||
| alter_command.move_destination_type = DataDestinationType::TABLE; | ||
| alter_command.from_database = src_database; | ||
| alter_command.from_table = src_table; | ||
| alter_command.to_database = dst_database; | ||
| alter_command.to_table = dst_table; | ||
| const auto src_database = src_db_col.getDataAt(i); | ||
| const auto src_table = src_table_col.getDataAt(i); | ||
| const auto dst_database = dst_db_col.getDataAt(i); | ||
| const auto dst_table = dst_table_col.getDataAt(i); | ||
|
|
||
| required_access_rights = InterpreterAlterQuery::getRequiredAccessForCommand( | ||
| alter_command, table_id.database_name, table_id.table_name); | ||
| if (!access->isGranted(required_access_rights)) | ||
| const auto table_id = StorageID{std::string{src_database}, std::string{src_table}}; | ||
| const auto transaction_id = tx_col.getDataAt(i); | ||
|
|
||
| CancellationCode code = CancellationCode::Unknown; | ||
| if (!query.test) | ||
| { | ||
| auto storage = DatabaseCatalog::instance().tryGetTable(table_id, getContext()); | ||
| if (!storage) | ||
| code = CancellationCode::NotFound; | ||
| else | ||
| { | ||
| access_denied = true; | ||
| continue; | ||
| ASTAlterCommand alter_command{}; | ||
| alter_command.type = ASTAlterCommand::EXPORT_PARTITION; | ||
| alter_command.move_destination_type = DataDestinationType::TABLE; | ||
| alter_command.from_database = src_database; | ||
| alter_command.from_table = src_table; | ||
| alter_command.to_database = dst_database; | ||
| alter_command.to_table = dst_table; | ||
|
|
||
| required_access_rights = InterpreterAlterQuery::getRequiredAccessForCommand( | ||
| alter_command, table_id.database_name, table_id.table_name); | ||
| if (!access->isGranted(required_access_rights)) | ||
| { | ||
| access_denied = true; | ||
| continue; | ||
| } | ||
| code = storage->killExportPartition(std::string{transaction_id}); | ||
| } | ||
| code = storage->killExportPartition(std::string{transaction_id}); | ||
| } | ||
|
|
||
| insertResultRow(i, code, exports_block, header, res_columns); | ||
| } | ||
| }; | ||
|
|
||
| insertResultRow(i, code, exports_block, header, res_columns); | ||
| } | ||
| process_block(replicated_exports_block); | ||
| process_block(plain_exports_block); | ||
|
|
||
| if (res_columns[0]->empty() && access_denied) | ||
| throw Exception(ErrorCodes::ACCESS_DENIED, "Not allowed to kill export partition. " | ||
|
|
||
Uh oh!
There was an error while loading. Please reload this page.