forked from apache/datafusion
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathrepartition.slt
More file actions
148 lines (123 loc) · 5.01 KB
/
Copy pathrepartition.slt
File metadata and controls
148 lines (123 loc) · 5.01 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
# http://www.apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
##########
# Tests for repartitioning
##########
# Set 4 partitions for deterministic output plans
statement ok
set datafusion.execution.target_partitions = 4;
statement ok
COPY (VALUES (1, 2), (2, 5), (3, 2), (4, 5), (5, 0)) TO 'test_files/scratch/repartition/parquet_table/2.parquet'
STORED AS PARQUET;
statement ok
CREATE EXTERNAL TABLE parquet_table(column1 int, column2 int)
STORED AS PARQUET
LOCATION 'test_files/scratch/repartition/parquet_table/';
# enable round robin repartitioning
statement ok
set datafusion.optimizer.enable_round_robin_repartition = true;
query TT
EXPLAIN SELECT column1, SUM(column2) FROM parquet_table GROUP BY column1;
----
logical_plan
01)Aggregate: groupBy=[[parquet_table.column1]], aggr=[[sum(CAST(parquet_table.column2 AS Int64))]]
02)--TableScan: parquet_table projection=[column1, column2]
physical_plan
01)AggregateExec: mode=FinalPartitioned, gby=[column1@0 as column1], aggr=[sum(parquet_table.column2)]
02)--CoalesceBatchesExec: target_batch_size=8192
03)----RepartitionExec: partitioning=Hash([column1@0], 4), input_partitions=4
04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
05)--------AggregateExec: mode=Partial, gby=[column1@0 as column1], aggr=[sum(parquet_table.column2)]
06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition/parquet_table/2.parquet]]}, projection=[column1, column2], file_type=parquet
# disable round robin repartitioning
statement ok
set datafusion.optimizer.enable_round_robin_repartition = false;
query TT
EXPLAIN SELECT column1, SUM(column2) FROM parquet_table GROUP BY column1;
----
logical_plan
01)Aggregate: groupBy=[[parquet_table.column1]], aggr=[[sum(CAST(parquet_table.column2 AS Int64))]]
02)--TableScan: parquet_table projection=[column1, column2]
physical_plan
01)AggregateExec: mode=FinalPartitioned, gby=[column1@0 as column1], aggr=[sum(parquet_table.column2)]
02)--CoalesceBatchesExec: target_batch_size=8192
03)----RepartitionExec: partitioning=Hash([column1@0], 4), input_partitions=1
04)------AggregateExec: mode=Partial, gby=[column1@0 as column1], aggr=[sum(parquet_table.column2)]
05)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition/parquet_table/2.parquet]]}, projection=[column1, column2], file_type=parquet
# Cleanup
statement ok
DROP TABLE parquet_table;
# Unbounded repartition
# See https://github.com/apache/datafusion/issues/5278
# Set up unbounded table and run a query - the query plan should display a `RepartitionExec`
# and a `CoalescePartitionsExec`
statement ok
CREATE UNBOUNDED EXTERNAL TABLE sink_table (
c1 VARCHAR NOT NULL,
c2 TINYINT NOT NULL,
c3 SMALLINT NOT NULL,
c4 SMALLINT NOT NULL,
c5 INTEGER NOT NULL,
c6 BIGINT NOT NULL,
c7 SMALLINT NOT NULL,
c8 INT NOT NULL,
c9 INT UNSIGNED NOT NULL,
c10 BIGINT UNSIGNED NOT NULL,
c11 FLOAT NOT NULL,
c12 DOUBLE NOT NULL,
c13 VARCHAR NOT NULL
)
STORED AS CSV
LOCATION '../../testing/data/csv/aggregate_test_100.csv'
OPTIONS ('format.has_header' 'true');
query TII
SELECT c1, c2, c3 FROM sink_table WHERE c3 > 0 LIMIT 5;
----
c 2 1
b 1 29
e 3 104
a 3 13
d 1 38
statement ok
set datafusion.execution.target_partitions = 3;
statement ok
set datafusion.optimizer.enable_round_robin_repartition = true;
query TT
EXPLAIN SELECT c1, c2, c3 FROM sink_table WHERE c3 > 0 LIMIT 5;
----
logical_plan
01)Limit: skip=0, fetch=5
02)--Filter: sink_table.c3 > Int16(0)
03)----TableScan: sink_table projection=[c1, c2, c3]
physical_plan
01)CoalescePartitionsExec: fetch=5
02)--CoalesceBatchesExec: target_batch_size=8192, fetch=5
03)----FilterExec: c3@2 > 0
04)------RepartitionExec: partitioning=RoundRobinBatch(3), input_partitions=1
05)--------StreamingTableExec: partition_sizes=1, projection=[c1, c2, c3], infinite_source=true
# Start repratition on empty column test.
# See https://github.com/apache/datafusion/issues/12057
statement ok
CREATE TABLE t1(v1 int);
statement ok
INSERT INTO t1 values(42);
query I
SELECT sum(1) OVER (PARTITION BY false=false)
FROM t1 WHERE ((false > (v1 = v1)) IS DISTINCT FROM true);
----
1
statement ok
DROP TABLE t1;
# End repartition on empty columns test