From 3657c83a5733588bb1ed118806f596482888b101 Mon Sep 17 00:00:00 2001 From: Azat Khuzhin Date: Wed, 20 May 2026 12:44:00 +0000 Subject: [PATCH 1/3] Merge pull request #104296 from azat/pr-additional-filters-fix Disable additional_table_filters with parallel replicas and analyzer and w/o query plan serialization --- src/Planner/Planner.cpp | 22 +++++ .../04207_pr_additional_filters.reference | 54 +++++++++++++ .../04207_pr_additional_filters.sql | 80 +++++++++++++++++++ 3 files changed, 156 insertions(+) create mode 100644 tests/queries/0_stateless/04207_pr_additional_filters.reference create mode 100644 tests/queries/0_stateless/04207_pr_additional_filters.sql diff --git a/src/Planner/Planner.cpp b/src/Planner/Planner.cpp index f5be70fa40e1..c2ece667ac57 100644 --- a/src/Planner/Planner.cpp +++ b/src/Planner/Planner.cpp @@ -101,6 +101,7 @@ namespace DB { namespace Setting { + extern const SettingsMap additional_table_filters; extern const SettingsUInt64 aggregation_in_order_max_block_bytes; extern const SettingsUInt64 aggregation_memory_efficient_merge_threads; extern const SettingsUInt64 allow_experimental_parallel_reading_from_replicas; @@ -126,6 +127,7 @@ namespace Setting extern const SettingsString parallel_replicas_custom_key; extern const SettingsUInt64 parallel_replicas_min_number_of_rows_per_replica; extern const SettingsBool query_plan_enable_multithreading_after_window_functions; + extern const SettingsBool serialize_query_plan; extern const SettingsBool throw_on_unsupported_query_inside_transaction; extern const SettingsFloat totals_auto_threshold; extern const SettingsTotalsMode totals_mode; @@ -1571,6 +1573,26 @@ void Planner::buildPlanForQueryNode() } } + /// `additional_table_filters` keys are resolved against the initiator's session current database, + /// but on followers the rewritten `SELECT` uses fully qualified names and the follower's current + /// database is the initiator's user-default DB, so the filter match is unreliable. Rather than + /// patch the match (which differs case by case), disable the combination on the analyzer path. + /// With `serialize_query_plan` the initiator lowers `additional_table_filters` into an explicit + /// `FilterStep` and ships the serialized plan, so the follower never re-resolves the setting — + /// the combination works there and the check is skipped. + if (query_context->canUseParallelReplicasOnInitiator() + && !settings[Setting::serialize_query_plan] + && !settings[Setting::additional_table_filters].value.empty()) + { + if (settings[Setting::allow_experimental_parallel_reading_from_replicas] >= 2) + throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, + "additional_table_filters is not supported with parallel and without serialize_query_plan=1"); + + auto & mutable_context = planner_context->getMutableQueryContext(); + mutable_context->setSetting("allow_experimental_parallel_reading_from_replicas", Field(0)); + LOG_DEBUG(log, "Disabling parallel replicas to execute a query with additional_table_filters"); + } + collectTableExpressionData(query_tree, planner_context); checkStoragesSupportTransactions(planner_context); diff --git a/tests/queries/0_stateless/04207_pr_additional_filters.reference b/tests/queries/0_stateless/04207_pr_additional_filters.reference new file mode 100644 index 000000000000..c1033d1bdeb5 --- /dev/null +++ b/tests/queries/0_stateless/04207_pr_additional_filters.reference @@ -0,0 +1,54 @@ +3 +3 +3 +Expression ((Project names + Projection)) +Actions: INPUT :: 0 -> count() UInt64 : 0 +Positions: 0 + MergingAggregated + Keys: + Aggregates: + count() + Function: count() → UInt64 + Arguments: none + Union + Aggregating + Keys: + Aggregates: + count() + Function: count() → UInt64 + Arguments: none + Skip merging: 0 + Expression ((Before GROUP BY + Change column names to column identifiers)) + Positions: + Filter (additional filter) + Filter column: lessOrEquals(x, 2_UInt8) (removed) + Actions: INPUT : 0 -> x UInt64 : 0 + COLUMN Const(UInt8) -> 2_UInt8 UInt8 : 1 + FUNCTION lessOrEquals(x :: 0, 2_UInt8 :: 1) -> lessOrEquals(x, 2_UInt8) UInt8 : 2 + Positions: 2 + ReadFromMergeTree (default.atf_p) + ReadType: Default + Parts: 1 + Granules: 1 + ReadFromRemoteParallelReplicas (Query: SELECT count() AS `count()` FROM `default`.`atf_p` AS `__table1` Replicas: 127.0.0.2:9000, 127.0.0.3:9000) + Aggregating + Keys: + Aggregates: + count() + Function: count() → UInt64 + Arguments: none + Skip merging: 0 + Expression (Before GROUP BY) + Actions: INPUT :: 0 -> __table1.x UInt64 : 0 + Positions: + Expression (Change column names to column identifiers) + Actions: INPUT : 0 -> x UInt64 : 0 + ALIAS x :: 0 -> __table1.x UInt64 : 1 + Positions: 1 + Filter (additional filter) + Filter column: lessOrEquals(x, 2_UInt8) (removed) + Actions: INPUT : 0 -> x UInt64 : 0 + COLUMN Const(UInt8) -> 2_UInt8 UInt8 : 1 + FUNCTION lessOrEquals(x : 0, 2_UInt8 :: 1) -> lessOrEquals(x, 2_UInt8) UInt8 : 2 + Positions: 2 0 + ReadFromTable diff --git a/tests/queries/0_stateless/04207_pr_additional_filters.sql b/tests/queries/0_stateless/04207_pr_additional_filters.sql new file mode 100644 index 000000000000..3d3e24b24a89 --- /dev/null +++ b/tests/queries/0_stateless/04207_pr_additional_filters.sql @@ -0,0 +1,80 @@ +-- Tags: no-parallel +-- ^ failpoint + +DROP TABLE IF EXISTS atf_p; +CREATE TABLE atf_p (x UInt64) ENGINE = MergeTree ORDER BY tuple(); +INSERT INTO atf_p SELECT number FROM numbers(10); + +-- The failpoint disables cancellation of unused replicas after all ranges +-- are assigned, so every replica's contribution lands on the initiator and any +-- missing-filter regression is deterministic regardless of timing. + +-- `additional_table_filters` keys are resolved against the initiator's session current +-- database, so they cannot be reliably matched on parallel-replica followers (the +-- rewritten query qualifies the table and the follower's current database differs). +-- On the legacy AST-forwarding path the analyzer rejects the combination instead of +-- silently dropping the filter. +SYSTEM ENABLE FAILPOINT parallel_replicas_wait_for_unused_replicas; +SELECT count() FROM atf_p SETTINGS additional_table_filters = {'atf_p': 'x <= 2'}, + enable_analyzer = 1, + enable_parallel_replicas = 2, + automatic_parallel_replicas_mode = 0, + max_parallel_replicas = 3, + cluster_for_parallel_replicas = 'test_cluster_one_shard_three_replicas_localhost', + parallel_replicas_for_non_replicated_merge_tree = 1, + parallel_replicas_local_plan = 1, + serialize_query_plan = 0; -- { serverError SUPPORT_IS_DISABLED } + +SELECT count() FROM atf_p SETTINGS additional_table_filters = {'atf_p': 'x <= 2'}, + enable_analyzer = 0, + enable_parallel_replicas = 2, + automatic_parallel_replicas_mode = 0, + max_parallel_replicas = 3, + cluster_for_parallel_replicas = 'test_cluster_one_shard_three_replicas_localhost', + parallel_replicas_for_non_replicated_merge_tree = 1, + parallel_replicas_local_plan = 1, + serialize_query_plan = 0; + +-- With `enable_parallel_replicas = 1` (best-effort) parallel replicas is silently +-- disabled and the query runs locally with the filter applied. +SYSTEM ENABLE FAILPOINT parallel_replicas_wait_for_unused_replicas; +SELECT count() FROM atf_p SETTINGS additional_table_filters = {'atf_p': 'x <= 2'}, + enable_analyzer = 1, + enable_parallel_replicas = 1, + automatic_parallel_replicas_mode = 0, + max_parallel_replicas = 3, + cluster_for_parallel_replicas = 'test_cluster_one_shard_three_replicas_localhost', + parallel_replicas_for_non_replicated_merge_tree = 1, + parallel_replicas_local_plan = 1, + serialize_query_plan = 0; + +-- With `serialize_query_plan = 1` the initiator lowers the additional filter into an +-- explicit `FilterStep` and ships the serialized plan, so the follower never +-- re-resolves the setting and the combination works. +SYSTEM ENABLE FAILPOINT parallel_replicas_wait_for_unused_replicas; +SELECT count() FROM atf_p SETTINGS additional_table_filters = {'atf_p': 'x <= 2'}, + enable_analyzer = 1, + enable_parallel_replicas = 2, + automatic_parallel_replicas_mode = 0, + max_parallel_replicas = 3, + cluster_for_parallel_replicas = 'test_cluster_one_shard_three_replicas_localhost', + parallel_replicas_for_non_replicated_merge_tree = 1, + parallel_replicas_local_plan = 1, + serialize_query_plan = 1; + +-- Verify the additional filter is present in the serialized plan shipped to followers. +SYSTEM ENABLE FAILPOINT parallel_replicas_wait_for_unused_replicas; +EXPLAIN PLAN actions = 1, distributed = 1 +SELECT count() FROM atf_p SETTINGS additional_table_filters = {'atf_p': 'x <= 2'}, + enable_analyzer = 1, + enable_parallel_replicas = 2, + automatic_parallel_replicas_mode = 0, + max_parallel_replicas = 3, + cluster_for_parallel_replicas = 'test_cluster_one_shard_three_replicas_localhost', + parallel_replicas_for_non_replicated_merge_tree = 1, + query_plan_remove_unused_columns = 1, + parallel_replicas_local_plan = 1, + serialize_query_plan = 1; + +SYSTEM DISABLE FAILPOINT parallel_replicas_wait_for_unused_replicas; +DROP TABLE atf_p; From ce5273708ac7296dafc5dc7acacb2ecf299b952d Mon Sep 17 00:00:00 2001 From: Mikhail Koviazin Date: Fri, 17 Jul 2026 10:26:28 +0200 Subject: [PATCH 2/3] tests: properly backport 04207 to 25.8 25.8 is missing some of the settings that are used by the original PR. 25.8 is also missing the EXPLAIN-visualisation for parallel-replicas remote step, so the last query in the test doesn't actually check anything and is therefore dropped. --- .../04207_pr_additional_filters.reference | 51 ------------------- .../04207_pr_additional_filters.sql | 18 ------- 2 files changed, 69 deletions(-) diff --git a/tests/queries/0_stateless/04207_pr_additional_filters.reference b/tests/queries/0_stateless/04207_pr_additional_filters.reference index c1033d1bdeb5..1f242fa6f000 100644 --- a/tests/queries/0_stateless/04207_pr_additional_filters.reference +++ b/tests/queries/0_stateless/04207_pr_additional_filters.reference @@ -1,54 +1,3 @@ 3 3 3 -Expression ((Project names + Projection)) -Actions: INPUT :: 0 -> count() UInt64 : 0 -Positions: 0 - MergingAggregated - Keys: - Aggregates: - count() - Function: count() → UInt64 - Arguments: none - Union - Aggregating - Keys: - Aggregates: - count() - Function: count() → UInt64 - Arguments: none - Skip merging: 0 - Expression ((Before GROUP BY + Change column names to column identifiers)) - Positions: - Filter (additional filter) - Filter column: lessOrEquals(x, 2_UInt8) (removed) - Actions: INPUT : 0 -> x UInt64 : 0 - COLUMN Const(UInt8) -> 2_UInt8 UInt8 : 1 - FUNCTION lessOrEquals(x :: 0, 2_UInt8 :: 1) -> lessOrEquals(x, 2_UInt8) UInt8 : 2 - Positions: 2 - ReadFromMergeTree (default.atf_p) - ReadType: Default - Parts: 1 - Granules: 1 - ReadFromRemoteParallelReplicas (Query: SELECT count() AS `count()` FROM `default`.`atf_p` AS `__table1` Replicas: 127.0.0.2:9000, 127.0.0.3:9000) - Aggregating - Keys: - Aggregates: - count() - Function: count() → UInt64 - Arguments: none - Skip merging: 0 - Expression (Before GROUP BY) - Actions: INPUT :: 0 -> __table1.x UInt64 : 0 - Positions: - Expression (Change column names to column identifiers) - Actions: INPUT : 0 -> x UInt64 : 0 - ALIAS x :: 0 -> __table1.x UInt64 : 1 - Positions: 1 - Filter (additional filter) - Filter column: lessOrEquals(x, 2_UInt8) (removed) - Actions: INPUT : 0 -> x UInt64 : 0 - COLUMN Const(UInt8) -> 2_UInt8 UInt8 : 1 - FUNCTION lessOrEquals(x : 0, 2_UInt8 :: 1) -> lessOrEquals(x, 2_UInt8) UInt8 : 2 - Positions: 2 0 - ReadFromTable diff --git a/tests/queries/0_stateless/04207_pr_additional_filters.sql b/tests/queries/0_stateless/04207_pr_additional_filters.sql index 3d3e24b24a89..2fd94023cbfb 100644 --- a/tests/queries/0_stateless/04207_pr_additional_filters.sql +++ b/tests/queries/0_stateless/04207_pr_additional_filters.sql @@ -18,7 +18,6 @@ SYSTEM ENABLE FAILPOINT parallel_replicas_wait_for_unused_replicas; SELECT count() FROM atf_p SETTINGS additional_table_filters = {'atf_p': 'x <= 2'}, enable_analyzer = 1, enable_parallel_replicas = 2, - automatic_parallel_replicas_mode = 0, max_parallel_replicas = 3, cluster_for_parallel_replicas = 'test_cluster_one_shard_three_replicas_localhost', parallel_replicas_for_non_replicated_merge_tree = 1, @@ -28,7 +27,6 @@ SELECT count() FROM atf_p SETTINGS additional_table_filters = {'atf_p': 'x <= 2' SELECT count() FROM atf_p SETTINGS additional_table_filters = {'atf_p': 'x <= 2'}, enable_analyzer = 0, enable_parallel_replicas = 2, - automatic_parallel_replicas_mode = 0, max_parallel_replicas = 3, cluster_for_parallel_replicas = 'test_cluster_one_shard_three_replicas_localhost', parallel_replicas_for_non_replicated_merge_tree = 1, @@ -41,7 +39,6 @@ SYSTEM ENABLE FAILPOINT parallel_replicas_wait_for_unused_replicas; SELECT count() FROM atf_p SETTINGS additional_table_filters = {'atf_p': 'x <= 2'}, enable_analyzer = 1, enable_parallel_replicas = 1, - automatic_parallel_replicas_mode = 0, max_parallel_replicas = 3, cluster_for_parallel_replicas = 'test_cluster_one_shard_three_replicas_localhost', parallel_replicas_for_non_replicated_merge_tree = 1, @@ -55,26 +52,11 @@ SYSTEM ENABLE FAILPOINT parallel_replicas_wait_for_unused_replicas; SELECT count() FROM atf_p SETTINGS additional_table_filters = {'atf_p': 'x <= 2'}, enable_analyzer = 1, enable_parallel_replicas = 2, - automatic_parallel_replicas_mode = 0, max_parallel_replicas = 3, cluster_for_parallel_replicas = 'test_cluster_one_shard_three_replicas_localhost', parallel_replicas_for_non_replicated_merge_tree = 1, parallel_replicas_local_plan = 1, serialize_query_plan = 1; --- Verify the additional filter is present in the serialized plan shipped to followers. -SYSTEM ENABLE FAILPOINT parallel_replicas_wait_for_unused_replicas; -EXPLAIN PLAN actions = 1, distributed = 1 -SELECT count() FROM atf_p SETTINGS additional_table_filters = {'atf_p': 'x <= 2'}, - enable_analyzer = 1, - enable_parallel_replicas = 2, - automatic_parallel_replicas_mode = 0, - max_parallel_replicas = 3, - cluster_for_parallel_replicas = 'test_cluster_one_shard_three_replicas_localhost', - parallel_replicas_for_non_replicated_merge_tree = 1, - query_plan_remove_unused_columns = 1, - parallel_replicas_local_plan = 1, - serialize_query_plan = 1; - SYSTEM DISABLE FAILPOINT parallel_replicas_wait_for_unused_replicas; DROP TABLE atf_p; From 6877ef37c38a34cf5bf41d5b1d26538ed0ac7076 Mon Sep 17 00:00:00 2001 From: Andrey Zvonov <32552679+zvonand@users.noreply.github.com> Date: Mon, 20 Jul 2026 19:25:30 +0200 Subject: [PATCH 3/3] adjust test for 25.8 --- src/Planner/Planner.cpp | 11 +++++------ .../0_stateless/04207_pr_additional_filters.reference | 1 - .../0_stateless/04207_pr_additional_filters.sql | 9 +++++---- 3 files changed, 10 insertions(+), 11 deletions(-) diff --git a/src/Planner/Planner.cpp b/src/Planner/Planner.cpp index c2ece667ac57..d316651608f9 100644 --- a/src/Planner/Planner.cpp +++ b/src/Planner/Planner.cpp @@ -127,7 +127,6 @@ namespace Setting extern const SettingsString parallel_replicas_custom_key; extern const SettingsUInt64 parallel_replicas_min_number_of_rows_per_replica; extern const SettingsBool query_plan_enable_multithreading_after_window_functions; - extern const SettingsBool serialize_query_plan; extern const SettingsBool throw_on_unsupported_query_inside_transaction; extern const SettingsFloat totals_auto_threshold; extern const SettingsTotalsMode totals_mode; @@ -1577,16 +1576,16 @@ void Planner::buildPlanForQueryNode() /// but on followers the rewritten `SELECT` uses fully qualified names and the follower's current /// database is the initiator's user-default DB, so the filter match is unreliable. Rather than /// patch the match (which differs case by case), disable the combination on the analyzer path. - /// With `serialize_query_plan` the initiator lowers `additional_table_filters` into an explicit - /// `FilterStep` and ships the serialized plan, so the follower never re-resolves the setting — - /// the combination works there and the check is skipped. + /// Upstream `serialize_query_plan=1` is exempt because the initiator lowers the filter into an + /// explicit `FilterStep` and ships the serialized plan, but in 25.8 parallel-replicas followers + /// always receive the rewritten AST (see ReadFromParallelRemoteReplicasStep), so the setting + /// does not help here and the check applies unconditionally. if (query_context->canUseParallelReplicasOnInitiator() - && !settings[Setting::serialize_query_plan] && !settings[Setting::additional_table_filters].value.empty()) { if (settings[Setting::allow_experimental_parallel_reading_from_replicas] >= 2) throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, - "additional_table_filters is not supported with parallel and without serialize_query_plan=1"); + "additional_table_filters is not supported with parallel replicas"); auto & mutable_context = planner_context->getMutableQueryContext(); mutable_context->setSetting("allow_experimental_parallel_reading_from_replicas", Field(0)); diff --git a/tests/queries/0_stateless/04207_pr_additional_filters.reference b/tests/queries/0_stateless/04207_pr_additional_filters.reference index 1f242fa6f000..a5c8806279fa 100644 --- a/tests/queries/0_stateless/04207_pr_additional_filters.reference +++ b/tests/queries/0_stateless/04207_pr_additional_filters.reference @@ -1,3 +1,2 @@ 3 3 -3 diff --git a/tests/queries/0_stateless/04207_pr_additional_filters.sql b/tests/queries/0_stateless/04207_pr_additional_filters.sql index 2fd94023cbfb..031665dbaf0f 100644 --- a/tests/queries/0_stateless/04207_pr_additional_filters.sql +++ b/tests/queries/0_stateless/04207_pr_additional_filters.sql @@ -45,9 +45,10 @@ SELECT count() FROM atf_p SETTINGS additional_table_filters = {'atf_p': 'x <= 2' parallel_replicas_local_plan = 1, serialize_query_plan = 0; --- With `serialize_query_plan = 1` the initiator lowers the additional filter into an --- explicit `FilterStep` and ships the serialized plan, so the follower never --- re-resolves the setting and the combination works. +-- Upstream `serialize_query_plan = 1` makes the combination work (the initiator lowers the +-- additional filter into an explicit `FilterStep` and ships the serialized plan), but in 25.8 +-- parallel-replicas followers always receive the rewritten AST, so the setting does not help +-- and the combination is rejected as well. SYSTEM ENABLE FAILPOINT parallel_replicas_wait_for_unused_replicas; SELECT count() FROM atf_p SETTINGS additional_table_filters = {'atf_p': 'x <= 2'}, enable_analyzer = 1, @@ -56,7 +57,7 @@ SELECT count() FROM atf_p SETTINGS additional_table_filters = {'atf_p': 'x <= 2' cluster_for_parallel_replicas = 'test_cluster_one_shard_three_replicas_localhost', parallel_replicas_for_non_replicated_merge_tree = 1, parallel_replicas_local_plan = 1, - serialize_query_plan = 1; + serialize_query_plan = 1; -- { serverError SUPPORT_IS_DISABLED } SYSTEM DISABLE FAILPOINT parallel_replicas_wait_for_unused_replicas; DROP TABLE atf_p;