mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-10-15 00:12:29 +00:00
discord9
4ffd93d8dc
* fix(query): preserve global fetch through physical-optimizer repartitioning Aggregate soft-limit pushdown inserts a global CoalescePartitionsExec(fetch) below an AggregateExec whose input is HashPartitioned/KeyPartitioned. The DataFusion physical optimizer's EnforceDistribution/EnforceSorting passes then dropped or rewrote that fetched coalesce, losing the global limit. Root-caused and fixed in the DataFusion fork (GreptimeTeam/datafusion PR #35); this repo pins to that fix commit and adds regression coverage. - pin datafusion to GreptimeTeam/datafusion 48510c7f5 (fetch preservation fix) - global_limit.rs: adapt to new DF API (input_distribution_requirements, child_distribution, replace_children+Recompute); accept HashPartitioned and KeyPartitioned as partitioning to restore; add regression unit tests - sqlness: extend standalone common ssts and refresh distributed ssts_limit result for the preserved fetch plan Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * refactor: drop dead HashPartitioned compat arms and test wrapper - remove HashPartitioned compatibility matches (pinned DF only emits KeyPartitioned); drop their #[expect(deprecated)] attributes - remove single-use agg_with_limit helper whose seed limit is always overwritten by the soft-limit transform - restore Cargo.lock dependency edges unrelated to the DataFusion pin bump (cargo update --precise had downgraded 26 unrelated packages) Addresses oracle ablation review. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * chore(deps): pin DataFusion to merged fetch preservation fix Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --------- Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
102 lines
6.7 KiB
Plaintext
102 lines
6.7 KiB
Plaintext
CREATE TABLE ssts_limit_case (
|
|
a INT PRIMARY KEY INVERTED INDEX,
|
|
b STRING SKIPPING INDEX,
|
|
c STRING FULLTEXT INDEX,
|
|
ts TIMESTAMP TIME INDEX,
|
|
)
|
|
PARTITION ON COLUMNS (a) (
|
|
a < 1000,
|
|
a >= 1000 AND a < 2000,
|
|
a >= 2000
|
|
);
|
|
|
|
Affected Rows: 0
|
|
|
|
INSERT INTO ssts_limit_case VALUES
|
|
(500, 'a', 'a', 1),
|
|
(1500, 'b', 'b', 2),
|
|
(2500, 'c', 'c', 3);
|
|
|
|
Affected Rows: 3
|
|
|
|
ADMIN FLUSH_TABLE('ssts_limit_case');
|
|
|
|
+--------------------------------------+
|
|
| ADMIN FLUSH_TABLE('ssts_limit_case') |
|
|
+--------------------------------------+
|
|
| 0 |
|
|
+--------------------------------------+
|
|
|
|
SELECT COUNT(DISTINCT node_id) > 1 AS has_multi_datanodes
|
|
FROM information_schema.ssts_manifest;
|
|
|
|
+---------------------+
|
|
| has_multi_datanodes |
|
|
+---------------------+
|
|
| true |
|
|
+---------------------+
|
|
|
|
SELECT COUNT(*) AS limited_rows
|
|
FROM (
|
|
SELECT region_id
|
|
FROM information_schema.ssts_manifest
|
|
LIMIT 1
|
|
);
|
|
|
|
+--------------+
|
|
| limited_rows |
|
|
+--------------+
|
|
| 1 |
|
|
+--------------+
|
|
|
|
SELECT COUNT(*) AS filtered_limited_rows
|
|
FROM (
|
|
SELECT region_id
|
|
FROM information_schema.ssts_manifest
|
|
WHERE region_id > 0
|
|
LIMIT 1
|
|
);
|
|
|
|
+-----------------------+
|
|
| filtered_limited_rows |
|
|
+-----------------------+
|
|
| 1 |
|
|
+-----------------------+
|
|
|
|
-- SQLNESS REPLACE (RepartitionExec:.*) RepartitionExec: REDACTED
|
|
EXPLAIN SELECT COUNT(*) AS filtered_limited_rows
|
|
FROM (
|
|
SELECT region_id
|
|
FROM information_schema.ssts_manifest
|
|
WHERE table_id > 0
|
|
LIMIT 1
|
|
);
|
|
|
|
+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|
|
| plan_type | plan |
|
|
+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|
|
| logical_plan | MergeScan [is_placeholder=false, remote_input=[ |
|
|
| | Projection: count(Int64(1)) AS count(*) AS filtered_limited_rows |
|
|
| | Aggregate: groupBy=[[]], aggr=[[count(Int64(1))]] |
|
|
| | Limit: skip=0, fetch=1 |
|
|
| | Projection: information_schema.ssts_manifest.region_id |
|
|
| | Filter: information_schema.ssts_manifest.table_id > UInt32(0) |
|
|
| | TableScan: information_schema.ssts_manifest, partial_filters=[information_schema.ssts_manifest.table_id > UInt32(0)] |
|
|
| | ]] |
|
|
| physical_plan | ProjectionExec: expr=[count(Int64(1))@0 as filtered_limited_rows] |
|
|
| | AggregateExec: mode=Final, gby=[], aggr=[count(Int64(1))] |
|
|
| | CoalescePartitionsExec |
|
|
| | AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1))] |
|
|
| | RepartitionExec: REDACTED
|
|
| | CoalescePartitionsExec: fetch=1 |
|
|
| | FilterExec: table_id@0 > 0, projection=[], fetch=1 |
|
|
| | RepartitionExec: REDACTED
|
|
| | DistributedInspectExec: kind=SstManifest, scan=ScanRequest { projection: [2], filters: [table_id > UInt32(0)] }, schema=Schema { fields: [Field { name: "table_id", data_type: UInt32 }], metadata: {"greptime:version": "0"} } |
|
|
| | |
|
|
+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|
|
|
|
DROP TABLE ssts_limit_case;
|
|
|
|
Affected Rows: 0
|
|
|