rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-27 22:26:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(query): backport region scan hash partitioning fix to release/5e02efa9 (#9769)

Browse Source 
fix(query): stop reporting region scans as hash partitioned (#9491)

* fix(query): stop reporting region scans as hash partitioned



* test(query): use current DataFusion plan downcasts



* test(query): match null rendering in join regression



* test(query): refresh union plan for unknown scan partitioning



* test(query): refresh shared plans after distribution rule removal



* test(query): cover semi joins with remote dynamic filtering



* test(query): strengthen partitioned merge scan regressions



* test(query): remove unused merge scan test helpers



* test(perf): measure merge scan hash repartition cost



* fix(query): adapt merge scan deserialization to two-arg constructor

Adapt the production and test serializer callers added by #9206 and remove the obsolete partition_cols assertion.



* test(query): adapt nested merge scan integration constructor



---------


(cherry picked from commit 32e8467cda)

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>
Co-authored-by: discord9 <discord9@163.com>
This commit is contained in:
Lei, HUANG
2026-10-20 16:33:19 +08:00
committed by GitHub
co-authored by discord9
parent 8282f72c74
commit ef3c9d7ccc
19 changed files with 846 additions and 671 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests-integration/src/tests/instance_test.rs
+284
View File
@@ -29,6 +29,7 @@ use frontend::error::Error;
use frontend::instance::Instance;
use operator::error::Error as OperatorError;
use query::datafusion::QUERY_PARALLELISM_HINT;
use query::options::QUERY_ENABLE_REMOTE_DYNAMIC_FILTER_PUSHDOWN;
use rstest::rstest;
use rstest_reuse::apply;
use servers::error as server_error;
@@ -257,6 +258,289 @@ WHERE greptime_timestamp = (
);
}
#[apply(both_instances_cases)]
async fn test_partitioned_merge_scan_left_join_repartitions(instance: Arc<dyn MockInstance>) {
let frontend = instance.frontend();
execute_sql(
&frontend,
r#"
CREATE TABLE join_repro (
k STRING,
kind STRING,
v DOUBLE,
ts TIMESTAMP NOT NULL,
TIME INDEX (ts)
)
PARTITION ON COLUMNS (k) (
k < '1',
k >= '1' AND k < '2',
k >= '2' AND k < '3',
k >= '3' AND k < '4',
k >= '4' AND k < '5',
k >= '5' AND k < '6',
k >= '6' AND k < '7',
k >= '7' AND k < '8',
k >= '8' AND k < '9',
k >= '9' AND k < 'a',
k >= 'a' AND k < 'b',
k >= 'b' AND k < 'c',
k >= 'c' AND k < 'd',
k >= 'd' AND k < 'e',
k >= 'e' AND k < 'f',
k >= 'f'
)
ENGINE = mito
WITH (append_mode = 'true')
"#,
)
.await;
let mut rows = Vec::new();
for key in [
'0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f', 'z',
] {
rows.push(format!("('{key}', 'root', 1.0, '2024-01-01 00:00:00')"));
if key != 'z' {
for _ in 0..3 {
rows.push(format!("('{key}', 'child', 1.0, '2024-01-01 00:00:00')"));
}
}
}
execute_sql(
&frontend,
&format!("INSERT INTO join_repro VALUES {}", rows.join(",")),
)
.await;
let aggregate_sql = r#"SELECT count(*) roots, count(c.n) roots_with_children, sum(c.n) children
FROM (SELECT k FROM join_repro WHERE kind = 'root') r
LEFT JOIN (SELECT k, count(*) n FROM join_repro WHERE kind = 'child' GROUP BY k) c ON c.k = r.k"#;
for parallelism in [16, 8] {
let output = execute_sql_with_query_parallelism(&frontend, aggregate_sql, parallelism)
.await
.data
.pretty_print()
.await;
assert_eq!(
output,
"+-------+---------------------+----------+\n| roots | roots_with_children | children |\n+-------+---------------------+----------+\n| 17 | 16 | 48 |\n+-------+---------------------+----------+"
);
let explain = execute_sql_with_query_parallelism(
&frontend,
&format!("EXPLAIN {aggregate_sql}"),
parallelism,
)
.await
.data
.pretty_print()
.await;
assert!(
explain.contains("HashJoinExec: mode=Partitioned"),
"query_parallelism={parallelism} should use a partitioned hash join; explain:\n{explain}"
);
assert!(
explain.contains("RepartitionExec: partitioning=Hash([k@0]"),
"query_parallelism={parallelism} should repartition on the join key; explain:\n{explain}"
);
let ordered_sql = r#"SELECT r.k, c.n
FROM (SELECT k FROM join_repro WHERE kind = 'root') r
LEFT JOIN (SELECT k, count(*) n FROM join_repro WHERE kind = 'child' GROUP BY k) c ON c.k = r.k
ORDER BY r.k"#;
let ordered = execute_sql_with_query_parallelism(&frontend, ordered_sql, parallelism)
.await
.data
.pretty_print()
.await;
assert_eq!(
ordered,
r#"+---+---+
| k | n |
+---+---+
| 0 | 3 |
| 1 | 3 |
| 2 | 3 |
| 3 | 3 |
| 4 | 3 |
| 5 | 3 |
| 6 | 3 |
| 7 | 3 |
| 8 | 3 |
| 9 | 3 |
| a | 3 |
| b | 3 |
| c | 3 |
| d | 3 |
| e | 3 |
| f | 3 |
| z | |
+---+---+"#
);
let pruned_sql = r#"SELECT count(*) roots, count(c.n) roots_with_children, sum(c.n) children
FROM (SELECT k FROM join_repro WHERE kind = 'root' AND k >= '1' AND k < '9') r
LEFT JOIN (SELECT k, count(*) n FROM join_repro WHERE kind = 'child' AND k < '8' GROUP BY k) c ON c.k = r.k"#;
let pruned = execute_sql_with_query_parallelism(&frontend, pruned_sql, parallelism)
.await
.data
.pretty_print()
.await;
assert_eq!(
pruned,
"+-------+---------------------+----------+\n| roots | roots_with_children | children |\n+-------+---------------------+----------+\n| 8 | 7 | 21 |\n+-------+---------------------+----------+"
);
let pruned_explain = execute_sql_with_query_parallelism(
&frontend,
&format!("EXPLAIN {pruned_sql}"),
parallelism,
)
.await
.data
.pretty_print()
.await;
assert!(
pruned_explain.contains("HashJoinExec: mode=Partitioned"),
"query_parallelism={parallelism} pruned join should remain partitioned; explain:\n{pruned_explain}"
);
let hash_repartition_lines = pruned_explain
.lines()
.filter(|line| line.contains("RepartitionExec: partitioning=Hash([k@0]"))
.count();
assert_eq!(
hash_repartition_lines, 2,
"query_parallelism={parallelism} pruned join should repartition both inputs on k; explain:\n{pruned_explain}"
);
}
}
#[apply(both_instances_cases)]
async fn test_partitioned_merge_scan_semi_join_repartitions(instance: Arc<dyn MockInstance>) {
let frontend = instance.frontend();
execute_sql(
&frontend,
r#"
CREATE TABLE semi_join_repro (
ts TIMESTAMP TIME INDEX,
key_id INT,
is_match BOOLEAN,
PRIMARY KEY (key_id)
)
PARTITION ON COLUMNS (key_id) (
key_id < 10,
key_id >= 10 AND key_id < 20,
key_id >= 20 AND key_id < 30,
key_id >= 30
)
ENGINE = mito
"#,
)
.await;
execute_sql(
&frontend,
r#"INSERT INTO semi_join_repro VALUES
('2026-10-09 00:00:00', 1, false),
('2026-10-09 00:00:01', 1, true),
('2026-10-09 00:00:00', 11, false),
('2026-10-09 00:00:01', 11, true),
('2026-10-09 00:00:00', 21, false),
('2026-10-09 00:00:01', 21, true),
('2026-10-09 00:00:00', 31, false),
('2026-10-09 00:00:01', 31, true)"#,
)
.await;
let query = r#"SELECT t.key_id, MIN(t.ts) AS first_ts
FROM semi_join_repro t
WHERE t.key_id IN (
SELECT s.key_id FROM semi_join_repro s WHERE s.is_match = true
)
GROUP BY t.key_id
ORDER BY t.key_id"#;
let mut query_ctx = QueryContext::with_db_name(None);
query_ctx.set_extension(QUERY_PARALLELISM_HINT, "2");
query_ctx.set_extension(QUERY_ENABLE_REMOTE_DYNAMIC_FILTER_PUSHDOWN, "true");
let query_ctx = Arc::new(query_ctx);
let output = execute_sql_with(&frontend, query, query_ctx.clone())
.await
.data
.pretty_print()
.await;
let expected = r#"+--------+---------------------+
| key_id | first_ts |
+--------+---------------------+
| 1 | 2026-10-09T00:00:00 |
| 11 | 2026-10-09T00:00:00 |
| 21 | 2026-10-09T00:00:00 |
| 31 | 2026-10-09T00:00:00 |
+--------+---------------------+"#;
assert_eq!(output, expected);
let explain = execute_sql_with(&frontend, &format!("EXPLAIN {query}"), query_ctx.clone())
.await
.data
.pretty_print()
.await;
assert!(
explain.contains("HashJoinExec: mode=Partitioned, join_type=LeftSemi"),
"parallel IN query should use a partitioned left-semi join; explain:\n{explain}"
);
assert!(
explain.contains("RepartitionExec: partitioning=Hash([key_id@1], 2)")
&& explain.contains("RepartitionExec: partitioning=Hash([key_id@0], 2)"),
"parallel left-semi join should repartition both inputs on key_id; explain:\n{explain}"
);
let analyze_explain = execute_sql_with(
&frontend,
&format!("EXPLAIN ANALYZE VERBOSE {query}"),
query_ctx.clone(),
)
.await
.data
.pretty_print()
.await;
let has_remote_dyn_filter = |explain: &str| {
explain.lines().any(|line| {
line.contains("SeqScan: region=")
&& line.contains("\"dyn_filters\"")
&& line.contains("DynamicFilter")
})
};
assert!(
has_remote_dyn_filter(&analyze_explain),
"expected remote region SeqScan dyn_filters to include DynamicFilter; explain:\n{analyze_explain}"
);
let mut query_ctx_without_rdf = QueryContext::with_db_name(None);
query_ctx_without_rdf.set_extension(QUERY_PARALLELISM_HINT, "2");
query_ctx_without_rdf.set_extension(QUERY_ENABLE_REMOTE_DYNAMIC_FILTER_PUSHDOWN, "false");
let query_ctx_without_rdf = Arc::new(query_ctx_without_rdf);
let output_without_rdf = execute_sql_with(&frontend, query, query_ctx_without_rdf.clone())
.await
.data
.pretty_print()
.await;
assert_eq!(output_without_rdf, expected);
let analyze_explain_without_rdf = execute_sql_with(
&frontend,
&format!("EXPLAIN ANALYZE VERBOSE {query}"),
query_ctx_without_rdf,
)
.await
.data
.pretty_print()
.await;
assert!(
!analyze_explain_without_rdf
.lines()
.any(|line| line.contains("SeqScan: region=") && line.contains("\"dyn_filters\"")),
"expected no region SeqScan dyn_filters line when remote dynamic filter pushdown is disabled; explain:\n{analyze_explain_without_rdf}"
);
}
#[apply(both_instances_cases)]
async fn test_show_create_table(instance: Arc<dyn MockInstance>) {
let frontend = instance.frontend();