Line data Source code
1 : // Analytic numbering / navigation / aggregate-window tests for
2 : // `SemanticExecutor` (R14 NTILE, R15 RANK/DENSE_RANK/LAG/LEAD,
3 : // frame-aware SUM/AVG/MIN/MAX/COUNT). Fixture shared via
4 : // `executor_test_fixture.h`.
5 :
6 : #include <cstdint>
7 : #include <string>
8 : #include <utility>
9 : #include <vector>
10 :
11 : #include "absl/status/status.h"
12 : #include "backend/engine/semantic/executor.h"
13 : #include "backend/engine/semantic/executor_test_fixture.h"
14 : #include "backend/storage/storage.h"
15 : #include "googlesql/public/analyzer.h"
16 : #include "googlesql/public/analyzer_options.h"
17 : #include "gtest/gtest.h"
18 :
19 : namespace bigquery_emulator {
20 : namespace backend {
21 : namespace engine {
22 : namespace semantic {
23 : namespace {
24 :
25 : // R14: NTILE over aggregate input (RFM shape) — semantic executor path.
26 1 : TEST_F(SemanticExecutorTest, NtileOverAggregateInputUnevenBuckets) {
27 1 : const std::string sql =
28 1 : "WITH rfm_raw AS ("
29 1 : " SELECT customer_id, SUM(amount) AS monetary FROM ("
30 1 : " SELECT 1 AS customer_id, 100 AS amount UNION ALL"
31 1 : " SELECT 2, 200 UNION ALL SELECT 3, 300 UNION ALL"
32 1 : " SELECT 4, 400 UNION ALL SELECT 5, 500 UNION ALL"
33 1 : " SELECT 6, 600 UNION ALL SELECT 7, 700"
34 1 : " ) GROUP BY customer_id"
35 1 : "), rfm_scored AS ("
36 1 : " SELECT customer_id, monetary,"
37 1 : " NTILE(5) OVER (ORDER BY monetary ASC) AS m_score"
38 1 : " FROM rfm_raw"
39 1 : ") "
40 1 : "SELECT customer_id, m_score FROM rfm_scored ORDER BY customer_id";
41 1 : const auto* stmt = Analyze(sql, MakeAnalyzerOptions());
42 1 : ASSERT_NE(stmt, nullptr);
43 1 : SemanticExecutor exec;
44 1 : auto source = exec.ExecuteQuery(MakeRequest(sql), *stmt, catalog_.get());
45 2 : ASSERT_TRUE(source.ok()) << source.status();
46 : // 7 rows into 5 buckets → sizes 2,2,1,1,1 for customers 1..7.
47 1 : const std::vector<int64_t> want_scores = {1, 1, 2, 2, 3, 4, 5};
48 8 : for (size_t i = 0; i < want_scores.size(); ++i) {
49 7 : storage::Row row;
50 7 : auto has = (*source)->Next(&row);
51 14 : ASSERT_TRUE(has.ok()) << has.status();
52 14 : ASSERT_TRUE(*has) << "missing row " << i;
53 7 : ASSERT_EQ(row.cells.size(), 2u);
54 7 : EXPECT_EQ(row.cells[0].int64_value(), static_cast<int64_t>(i + 1));
55 14 : EXPECT_EQ(row.cells[1].int64_value(), want_scores[i]) << "row " << i;
56 7 : }
57 1 : storage::Row extra;
58 1 : auto has_extra = (*source)->Next(&extra);
59 1 : ASSERT_TRUE(has_extra.ok());
60 1 : EXPECT_FALSE(*has_extra);
61 1 : }
62 :
63 1 : TEST_F(SemanticExecutorTest, NtileNonPositiveBucketsRejectedAtAnalyze) {
64 : // GoogleSQL rejects a constant non-positive NTILE argument before
65 : // execution; ApplyAnalyticNtile also guards the runtime path for
66 : // non-constant expressions that slip through.
67 1 : const std::string sql =
68 1 : "WITH rfm_raw AS ("
69 1 : " SELECT customer_id, SUM(amount) AS monetary FROM ("
70 1 : " SELECT 1 AS customer_id, 100 AS amount UNION ALL SELECT 2, 200"
71 1 : " ) GROUP BY customer_id"
72 1 : ") "
73 1 : "SELECT NTILE(0) OVER (ORDER BY monetary ASC) AS m_score FROM rfm_raw";
74 1 : last_output_.reset();
75 1 : absl::Status s = ::googlesql::AnalyzeStatement(sql,
76 1 : MakeAnalyzerOptions(),
77 1 : catalog_.get(),
78 1 : type_factory_.get(),
79 1 : &last_output_);
80 1 : EXPECT_FALSE(s.ok());
81 2 : EXPECT_NE(std::string(s.message()).find("NTILE"), std::string::npos) << s;
82 1 : }
83 :
84 : // R15: RANK / DENSE_RANK over nested aggregate with ties.
85 1 : TEST_F(SemanticExecutorTest, RankDenseRankOverNestedAggregateTies) {
86 1 : const std::string sql =
87 1 : "SELECT customer_id,"
88 1 : " RANK() OVER (ORDER BY SUM(amount) DESC) AS rnk,"
89 1 : " DENSE_RANK() OVER (ORDER BY SUM(amount) DESC) AS drnk "
90 1 : "FROM ("
91 1 : " SELECT 1 AS customer_id, 100 AS amount UNION ALL"
92 1 : " SELECT 1, 50 UNION ALL"
93 1 : " SELECT 2, 100 UNION ALL"
94 1 : " SELECT 2, 50 UNION ALL"
95 1 : " SELECT 3, 200 UNION ALL"
96 1 : " SELECT 4, 50"
97 1 : ") GROUP BY customer_id "
98 1 : "ORDER BY customer_id";
99 1 : const auto* stmt = Analyze(sql, MakeAnalyzerOptions());
100 1 : ASSERT_NE(stmt, nullptr);
101 1 : SemanticExecutor exec;
102 1 : auto source = exec.ExecuteQuery(MakeRequest(sql), *stmt, catalog_.get());
103 2 : ASSERT_TRUE(source.ok()) << source.status();
104 : // Sums DESC: 200,150,150,50 → RANK 1,2,2,4 and DENSE_RANK 1,2,2,3.
105 : // Ordered by customer_id: c1=150→2/2, c2=150→2/2, c3=200→1/1, c4=50→4/3.
106 1 : const std::vector<std::pair<int64_t, int64_t>> want = {
107 1 : {2, 2}, {2, 2}, {1, 1}, {4, 3}};
108 5 : for (size_t i = 0; i < want.size(); ++i) {
109 4 : storage::Row row;
110 4 : auto has = (*source)->Next(&row);
111 8 : ASSERT_TRUE(has.ok()) << has.status();
112 8 : ASSERT_TRUE(*has) << "missing row " << i;
113 4 : ASSERT_EQ(row.cells.size(), 3u);
114 4 : EXPECT_EQ(row.cells[0].int64_value(), static_cast<int64_t>(i + 1));
115 8 : EXPECT_EQ(row.cells[1].int64_value(), want[i].first) << "rank row " << i;
116 8 : EXPECT_EQ(row.cells[2].int64_value(), want[i].second)
117 8 : << "dense_rank row " << i;
118 4 : }
119 1 : storage::Row extra;
120 1 : auto has_extra = (*source)->Next(&extra);
121 1 : ASSERT_TRUE(has_extra.ok());
122 1 : EXPECT_FALSE(*has_extra);
123 1 : }
124 :
125 : // Analytic output is stabilized on PARTITION BY + ORDER BY (no outer
126 : // ORDER BY), matching DuckDB CaptureAnalyticOutputOrder. Pins the
127 : // gateway e2e LEAD/NTILE finishers shape when FORMAT_TIMESTAMP promotes
128 : // the query onto the semantic route.
129 1 : TEST_F(SemanticExecutorTest, LeadOutputOrderFollowsPartitionAndOrderBy) {
130 1 : const std::string sql =
131 1 : "WITH finishers AS ("
132 1 : " SELECT 'Sophia' AS name, TIMESTAMP '2016-10-18 02:51:45+00' AS "
133 1 : "finish_time, 'F30-34' AS division UNION ALL"
134 1 : " SELECT 'Lisa', TIMESTAMP '2016-10-18 02:54:11+00', 'F35-39' UNION ALL"
135 1 : " SELECT 'Nikki', TIMESTAMP '2016-10-18 02:59:01+00', 'F30-34' UNION ALL"
136 1 : " SELECT 'Carly', TIMESTAMP '2016-10-18 03:08:58+00', 'F25-29'"
137 1 : ") "
138 1 : "SELECT name, division,"
139 1 : " LEAD(name) OVER (PARTITION BY division ORDER BY finish_time ASC) "
140 1 : "AS followed_by "
141 1 : "FROM finishers";
142 1 : const auto* stmt = Analyze(sql, MakeAnalyzerOptions());
143 1 : ASSERT_NE(stmt, nullptr);
144 1 : SemanticExecutor exec;
145 1 : auto source = exec.ExecuteQuery(MakeRequest(sql), *stmt, catalog_.get());
146 2 : ASSERT_TRUE(source.ok()) << source.status();
147 1 : struct Want {
148 1 : const char* name;
149 1 : const char* division;
150 1 : bool followed_null;
151 1 : const char* followed;
152 1 : };
153 1 : const std::vector<Want> want = {
154 1 : {"Carly", "F25-29", true, nullptr},
155 1 : {"Sophia", "F30-34", false, "Nikki"},
156 1 : {"Nikki", "F30-34", true, nullptr},
157 1 : {"Lisa", "F35-39", true, nullptr},
158 1 : };
159 5 : for (size_t i = 0; i < want.size(); ++i) {
160 4 : storage::Row row;
161 4 : auto has = (*source)->Next(&row);
162 8 : ASSERT_TRUE(has.ok()) << has.status();
163 8 : ASSERT_TRUE(*has) << "missing row " << i;
164 4 : ASSERT_EQ(row.cells.size(), 3u);
165 8 : EXPECT_EQ(row.cells[0].string_value(), want[i].name) << "row " << i;
166 8 : EXPECT_EQ(row.cells[1].string_value(), want[i].division) << "row " << i;
167 4 : if (want[i].followed_null) {
168 6 : EXPECT_TRUE(row.cells[2].is_null()) << "lead row " << i;
169 3 : } else {
170 2 : ASSERT_FALSE(row.cells[2].is_null()) << "lead row " << i;
171 2 : EXPECT_EQ(row.cells[2].string_value(), want[i].followed) << "row " << i;
172 1 : }
173 4 : }
174 1 : storage::Row extra;
175 1 : auto has_extra = (*source)->Next(&extra);
176 1 : ASSERT_TRUE(has_extra.ok());
177 1 : EXPECT_FALSE(*has_extra);
178 1 : }
179 :
180 : // R15: LAG / LEAD over nested aggregate with offset and default.
181 1 : TEST_F(SemanticExecutorTest, LagLeadOverNestedAggregate) {
182 1 : const std::string sql =
183 1 : "SELECT customer_id,"
184 1 : " LAG(SUM(amount), 1) OVER (ORDER BY customer_id) AS prev_m,"
185 1 : " LEAD(SUM(amount), 2, 0) OVER (ORDER BY customer_id) AS next2_m "
186 1 : "FROM ("
187 1 : " SELECT 1 AS customer_id, 100 AS amount UNION ALL"
188 1 : " SELECT 2, 200 UNION ALL"
189 1 : " SELECT 3, 300 UNION ALL"
190 1 : " SELECT 4, 400"
191 1 : ") GROUP BY customer_id "
192 1 : "ORDER BY customer_id";
193 1 : const auto* stmt = Analyze(sql, MakeAnalyzerOptions());
194 1 : ASSERT_NE(stmt, nullptr);
195 1 : SemanticExecutor exec;
196 1 : auto source = exec.ExecuteQuery(MakeRequest(sql), *stmt, catalog_.get());
197 2 : ASSERT_TRUE(source.ok()) << source.status();
198 : // monetary: 100,200,300,400 → LAG1: NULL,100,200,300; LEAD2 default0:
199 : // 300,400,0,0
200 1 : struct Want {
201 1 : bool prev_null;
202 1 : int64_t prev;
203 1 : int64_t next2;
204 1 : };
205 1 : const std::vector<Want> want = {
206 1 : {true, 0, 300}, {false, 100, 400}, {false, 200, 0}, {false, 300, 0}};
207 5 : for (size_t i = 0; i < want.size(); ++i) {
208 4 : storage::Row row;
209 4 : auto has = (*source)->Next(&row);
210 8 : ASSERT_TRUE(has.ok()) << has.status();
211 8 : ASSERT_TRUE(*has) << "missing row " << i;
212 4 : ASSERT_EQ(row.cells.size(), 3u);
213 4 : EXPECT_EQ(row.cells[0].int64_value(), static_cast<int64_t>(i + 1));
214 4 : if (want[i].prev_null) {
215 2 : EXPECT_TRUE(row.cells[1].is_null()) << "lag row " << i;
216 3 : } else {
217 6 : ASSERT_FALSE(row.cells[1].is_null()) << "lag row " << i;
218 6 : EXPECT_EQ(row.cells[1].int64_value(), want[i].prev) << "lag row " << i;
219 3 : }
220 8 : ASSERT_FALSE(row.cells[2].is_null()) << "lead row " << i;
221 8 : EXPECT_EQ(row.cells[2].int64_value(), want[i].next2) << "lead row " << i;
222 4 : }
223 1 : storage::Row extra;
224 1 : auto has_extra = (*source)->Next(&extra);
225 1 : ASSERT_TRUE(has_extra.ok());
226 1 : EXPECT_FALSE(*has_extra);
227 1 : }
228 :
229 : // Whole-partition MAX/MIN/AVG/COUNT(*) over aggregate input.
230 1 : TEST_F(SemanticExecutorTest, AggregateWindowsWholePartitionOverGroups) {
231 1 : const std::string sql =
232 1 : "SELECT customer_id,"
233 1 : " SUM(amount) AS monetary,"
234 1 : " MAX(SUM(amount)) OVER () AS max_m,"
235 1 : " MIN(SUM(amount)) OVER () AS min_m,"
236 1 : " AVG(SUM(amount)) OVER () AS avg_m,"
237 1 : " COUNT(*) OVER () AS n_groups "
238 1 : "FROM ("
239 1 : " SELECT 1 AS customer_id, 100 AS amount UNION ALL"
240 1 : " SELECT 2, 200 UNION ALL"
241 1 : " SELECT 3, 300"
242 1 : ") GROUP BY customer_id "
243 1 : "ORDER BY customer_id";
244 1 : const auto* stmt = Analyze(sql, MakeAnalyzerOptions());
245 1 : ASSERT_NE(stmt, nullptr);
246 1 : SemanticExecutor exec;
247 1 : auto source = exec.ExecuteQuery(MakeRequest(sql), *stmt, catalog_.get());
248 2 : ASSERT_TRUE(source.ok()) << source.status();
249 : // Group sums: 100,200,300 → max=300, min=100, avg=200, count=3.
250 4 : for (int64_t id = 1; id <= 3; ++id) {
251 3 : storage::Row row;
252 3 : auto has = (*source)->Next(&row);
253 6 : ASSERT_TRUE(has.ok()) << has.status();
254 6 : ASSERT_TRUE(*has) << "missing row " << id;
255 3 : ASSERT_EQ(row.cells.size(), 6u);
256 3 : EXPECT_EQ(row.cells[0].int64_value(), id);
257 3 : EXPECT_EQ(row.cells[1].int64_value(), id * 100);
258 3 : EXPECT_EQ(row.cells[2].int64_value(), 300);
259 3 : EXPECT_EQ(row.cells[3].int64_value(), 100);
260 3 : EXPECT_DOUBLE_EQ(row.cells[4].float64_value(), 200.0);
261 3 : EXPECT_EQ(row.cells[5].int64_value(), 3);
262 3 : }
263 1 : storage::Row extra;
264 1 : auto has_extra = (*source)->Next(&extra);
265 1 : ASSERT_TRUE(has_extra.ok());
266 1 : EXPECT_FALSE(*has_extra);
267 1 : }
268 :
269 : // Default RANGE frame running SUM with a peer tie.
270 1 : TEST_F(SemanticExecutorTest, RunningSumOverNestedAggregateWithTies) {
271 1 : const std::string sql =
272 1 : "SELECT customer_id,"
273 1 : " SUM(amount) AS monetary,"
274 1 : " SUM(SUM(amount)) OVER (ORDER BY SUM(amount)) AS running "
275 1 : "FROM ("
276 1 : " SELECT 1 AS customer_id, 100 AS amount UNION ALL"
277 1 : " SELECT 2, 100 UNION ALL"
278 1 : " SELECT 3, 200 UNION ALL"
279 1 : " SELECT 4, 50"
280 1 : ") GROUP BY customer_id "
281 1 : "ORDER BY customer_id";
282 1 : const auto* stmt = Analyze(sql, MakeAnalyzerOptions());
283 1 : ASSERT_NE(stmt, nullptr);
284 1 : SemanticExecutor exec;
285 1 : auto source = exec.ExecuteQuery(MakeRequest(sql), *stmt, catalog_.get());
286 2 : ASSERT_TRUE(source.ok()) << source.status();
287 : // Ordered by monetary ASC: 50(c4), 100(c1), 100(c2), 200(c3).
288 : // Default RANGE UNBOUNDED PRECEDING .. CURRENT ROW includes peers, so
289 : // both 100-rows see 50+100+100=250 and the 200-row sees 450.
290 : // Ordered by customer_id: c1→250, c2→250, c3→450, c4→50.
291 1 : const std::vector<int64_t> want_running = {250, 250, 450, 50};
292 5 : for (size_t i = 0; i < want_running.size(); ++i) {
293 4 : storage::Row row;
294 4 : auto has = (*source)->Next(&row);
295 8 : ASSERT_TRUE(has.ok()) << has.status();
296 8 : ASSERT_TRUE(*has) << "missing row " << i;
297 4 : ASSERT_EQ(row.cells.size(), 3u);
298 4 : EXPECT_EQ(row.cells[0].int64_value(), static_cast<int64_t>(i + 1));
299 8 : EXPECT_EQ(row.cells[2].int64_value(), want_running[i])
300 8 : << "running row " << i;
301 4 : }
302 1 : }
303 :
304 : // Explicit ROWS frame: preceding + current only (no peer expansion).
305 1 : TEST_F(SemanticExecutorTest, RowsFrameSumOverNestedAggregate) {
306 1 : const std::string sql =
307 1 : "SELECT customer_id,"
308 1 : " SUM(SUM(amount)) OVER ("
309 1 : " ORDER BY customer_id"
310 1 : " ROWS BETWEEN 1 PRECEDING AND CURRENT ROW"
311 1 : " ) AS win "
312 1 : "FROM ("
313 1 : " SELECT 1 AS customer_id, 100 AS amount UNION ALL"
314 1 : " SELECT 2, 200 UNION ALL"
315 1 : " SELECT 3, 300 UNION ALL"
316 1 : " SELECT 4, 400"
317 1 : ") GROUP BY customer_id "
318 1 : "ORDER BY customer_id";
319 1 : const auto* stmt = Analyze(sql, MakeAnalyzerOptions());
320 1 : ASSERT_NE(stmt, nullptr);
321 1 : SemanticExecutor exec;
322 1 : auto source = exec.ExecuteQuery(MakeRequest(sql), *stmt, catalog_.get());
323 2 : ASSERT_TRUE(source.ok()) << source.status();
324 : // Row sums 100,200,300,400 → windows: 100, 300, 500, 700.
325 1 : const std::vector<int64_t> want = {100, 300, 500, 700};
326 5 : for (size_t i = 0; i < want.size(); ++i) {
327 4 : storage::Row row;
328 4 : auto has = (*source)->Next(&row);
329 8 : ASSERT_TRUE(has.ok()) << has.status();
330 8 : ASSERT_TRUE(*has) << "missing row " << i;
331 4 : ASSERT_EQ(row.cells.size(), 2u);
332 4 : EXPECT_EQ(row.cells[0].int64_value(), static_cast<int64_t>(i + 1));
333 8 : EXPECT_EQ(row.cells[1].int64_value(), want[i]) << "rows frame row " << i;
334 4 : }
335 1 : }
336 :
337 : // Numeric RANGE value-offset COUNT (replaces the old CountRange lane;
338 : // BigQuery requires a numeric ORDER BY for RANGE offsets).
339 1 : TEST_F(SemanticExecutorTest, NumericRangeCountOverAggregateInput) {
340 1 : const std::string sql =
341 1 : "SELECT customer_id,"
342 1 : " COUNT(*) OVER ("
343 1 : " ORDER BY SUM(amount)"
344 1 : " RANGE BETWEEN 1 PRECEDING AND CURRENT ROW"
345 1 : " ) AS cnt "
346 1 : "FROM ("
347 1 : " SELECT 1 AS customer_id, 10 AS amount UNION ALL"
348 1 : " SELECT 2, 11 UNION ALL"
349 1 : " SELECT 3, 12 UNION ALL"
350 1 : " SELECT 4, 20"
351 1 : ") GROUP BY customer_id "
352 1 : "ORDER BY customer_id";
353 1 : const auto* stmt = Analyze(sql, MakeAnalyzerOptions());
354 1 : ASSERT_NE(stmt, nullptr);
355 1 : SemanticExecutor exec;
356 1 : auto source = exec.ExecuteQuery(MakeRequest(sql), *stmt, catalog_.get());
357 2 : ASSERT_TRUE(source.ok()) << source.status();
358 : // Sums 10,11,12,20 → RANGE ±1 preceding..current:
359 : // 10→{10}, 11→{10,11}, 12→{11,12}, 20→{20}.
360 1 : const std::vector<int64_t> want = {1, 2, 2, 1};
361 5 : for (size_t i = 0; i < want.size(); ++i) {
362 4 : storage::Row row;
363 4 : auto has = (*source)->Next(&row);
364 8 : ASSERT_TRUE(has.ok()) << has.status();
365 8 : ASSERT_TRUE(*has) << "missing row " << i;
366 4 : ASSERT_EQ(row.cells.size(), 2u);
367 4 : EXPECT_EQ(row.cells[0].int64_value(), static_cast<int64_t>(i + 1));
368 8 : EXPECT_EQ(row.cells[1].int64_value(), want[i]) << "range count row " << i;
369 4 : }
370 1 : }
371 :
372 : } // namespace
373 : } // namespace semantic
374 : } // namespace engine
375 : } // namespace backend
376 : } // namespace bigquery_emulator
|
