Line data Source code
1 : #include "backend/engine/duckdb/transpiler/transpiler_test_fixture.h"
2 :
3 : // R17: multi-CTE attribution SELECT / INSERT...SELECT / CTAS with
4 : // COALESCE(SUM) + ROW_NUMBER over joins must bind in DuckDB (no stale
5 : // `__bq_j_<id>` across CTE / LEFT JOIN / analytic boundaries).
6 :
7 : namespace bigquery_emulator {
8 : namespace backend {
9 : namespace engine {
10 : namespace duckdb {
11 : namespace transpiler {
12 :
13 : class TranspilerAttributionCompositionTest : public TranspilerBindFixture {
14 : protected:
15 4 : void SetUp() override {
16 4 : TranspilerBindFixture::SetUp();
17 :
18 : // R17 attribution shape (bench/cases/attribution_insert_10k.yaml).
19 4 : auto activity_events = std::make_unique<::googlesql::SimpleTable>(
20 4 : "activity_events",
21 4 : std::vector<::googlesql::SimpleTable::NameAndType>{
22 4 : {"tenant_id", type_factory_->get_string()},
23 4 : {"activity_id", type_factory_->get_string()},
24 4 : {"user_public_id", type_factory_->get_string()},
25 4 : {"action_type", type_factory_->get_string()},
26 4 : {"channel", type_factory_->get_string()},
27 4 : {"occurred_at", type_factory_->get_timestamp()},
28 4 : {"spend", type_factory_->get_double()},
29 4 : });
30 4 : catalog_->AddOwnedTable(std::move(activity_events));
31 :
32 4 : auto attr_users = std::make_unique<::googlesql::SimpleTable>(
33 4 : "attr_users",
34 4 : std::vector<::googlesql::SimpleTable::NameAndType>{
35 4 : {"id", type_factory_->get_int64()},
36 4 : {"public_id", type_factory_->get_string()},
37 4 : });
38 4 : catalog_->AddOwnedTable(std::move(attr_users));
39 :
40 4 : auto attr_transactions = std::make_unique<::googlesql::SimpleTable>(
41 4 : "attr_transactions",
42 4 : std::vector<::googlesql::SimpleTable::NameAndType>{
43 4 : {"id", type_factory_->get_int64()},
44 4 : {"user_id", type_factory_->get_int64()},
45 4 : {"total_amount", type_factory_->get_double()},
46 4 : {"total_reversed", type_factory_->get_double()},
47 4 : {"is_deleted", type_factory_->get_bool()},
48 4 : {"status", type_factory_->get_string()},
49 4 : {"source_created_at", type_factory_->get_timestamp()},
50 4 : });
51 4 : catalog_->AddOwnedTable(std::move(attr_transactions));
52 :
53 4 : auto attr_summary = std::make_unique<::googlesql::SimpleTable>(
54 4 : "attr_summary",
55 4 : std::vector<::googlesql::SimpleTable::NameAndType>{
56 4 : {"activity_id", type_factory_->get_string()},
57 4 : {"total_transactions", type_factory_->get_int64()},
58 4 : {"gross_value", type_factory_->get_double()},
59 4 : });
60 4 : catalog_->AddOwnedTable(std::move(attr_summary));
61 :
62 4 : ExecDdl(
63 4 : "CREATE TABLE activity_events (tenant_id VARCHAR, activity_id "
64 4 : "VARCHAR, user_public_id VARCHAR, action_type VARCHAR, channel "
65 4 : "VARCHAR, occurred_at TIMESTAMPTZ, spend DOUBLE)");
66 4 : ExecDdl("CREATE TABLE attr_users (id BIGINT, public_id VARCHAR)");
67 4 : ExecDdl(
68 4 : "CREATE TABLE attr_transactions (id BIGINT, user_id BIGINT, "
69 4 : "total_amount DOUBLE, total_reversed DOUBLE, is_deleted BOOLEAN, "
70 4 : "status VARCHAR, source_created_at TIMESTAMPTZ)");
71 4 : ExecDdl(
72 4 : "CREATE TABLE attr_summary (activity_id VARCHAR, "
73 4 : "total_transactions BIGINT, gross_value DOUBLE)");
74 4 : }
75 :
76 : // Compact R17 attribution body: multi-CTE + COALESCE(SUM) + JOIN +
77 : // ROW_NUMBER + LEFT JOIN + final COALESCE(SUM). Matches the bench
78 : // case shape that previously stayed on semantic_executor.
79 : static constexpr const char kAttributionSelectBody[] = R"sql(
80 : WITH ActivityLogs AS (
81 : SELECT tenant_id, activity_id, user_public_id, action_type, channel,
82 : occurred_at, spend
83 : FROM activity_events
84 : ),
85 : MetricStats AS (
86 : SELECT
87 : tenant_id,
88 : activity_id,
89 : COUNT(DISTINCT IF(action_type = 'action_1', user_public_id, NULL))
90 : AS metric_a,
91 : COUNTIF(action_type = 'action_3') AS metric_c_total,
92 : COALESCE(SUM(spend), 0) AS total_spend
93 : FROM ActivityLogs
94 : GROUP BY tenant_id, activity_id
95 : ),
96 : ValidActivityDispatches AS (
97 : SELECT user_public_id, activity_id, channel, occurred_at AS dispatched_at
98 : FROM ActivityLogs
99 : WHERE action_type = 'action_3'
100 : ),
101 : LatestTransactions AS (
102 : SELECT * FROM attr_transactions
103 : ),
104 : AttributedTransactionsRaw AS (
105 : SELECT
106 : logs.activity_id,
107 : logs.channel,
108 : txn.id AS txn_id,
109 : txn.total_amount,
110 : COALESCE(txn.total_reversed, 0) AS reversed_amount,
111 : ROW_NUMBER() OVER (
112 : PARTITION BY txn.id ORDER BY logs.dispatched_at DESC
113 : ) AS attribution_rank
114 : FROM ValidActivityDispatches logs
115 : JOIN attr_users users ON logs.user_public_id = users.public_id
116 : JOIN LatestTransactions txn ON users.id = txn.user_id
117 : WHERE txn.source_created_at > logs.dispatched_at
118 : AND txn.source_created_at <= TIMESTAMP_ADD(
119 : logs.dispatched_at, INTERVAL 7 DAY
120 : )
121 : AND COALESCE(txn.is_deleted, FALSE) = FALSE
122 : AND txn.status NOT IN ('status_x', 'status_y')
123 : ),
124 : ValueStats AS (
125 : SELECT
126 : activity_id,
127 : COUNT(txn_id) AS total_transactions,
128 : COALESCE(SUM(total_amount), 0) AS gross_value
129 : FROM AttributedTransactionsRaw
130 : WHERE attribution_rank = 1
131 : GROUP BY activity_id
132 : ),
133 : Attribution AS (
134 : SELECT
135 : p.activity_id AS id,
136 : COALESCE(r.total_transactions, 0) AS total_transactions,
137 : COALESCE(r.gross_value, 0) AS gross_value
138 : FROM MetricStats p
139 : LEFT JOIN ValueStats r ON p.activity_id = r.activity_id
140 : )
141 : )sql";
142 : };
143 :
144 : // R17 follow-up: full multi-CTE attribution SELECT with COALESCE(SUM)
145 : // must bind (no stale `__bq_j_<id>` across CTE / LEFT JOIN boundaries).
146 1 : TEST_F(TranspilerAttributionCompositionTest, AttributionMultiCteSelectBinds) {
147 1 : const std::string sql =
148 1 : absl::StrCat(kAttributionSelectBody,
149 1 : "SELECT COUNT(*) AS activity_rows, "
150 1 : "COALESCE(SUM(total_transactions), 0) AS txn_sum, "
151 1 : "COALESCE(SUM(gross_value), 0) AS gross_sum "
152 1 : "FROM Attribution");
153 1 : AssertSqlTranspileBinds(sql);
154 1 : }
155 :
156 : // R17 follow-up: compact INSERT...SELECT (single CTE + ROW_NUMBER over
157 : // joins + COALESCE(SUM)) must bind — this is the conformance fixture
158 : // shape that previously stayed on semantic_executor.
159 : TEST_F(TranspilerAttributionCompositionTest,
160 1 : AttributionCompactInsertSelectBinds) {
161 1 : static constexpr const char kCompactSql[] = R"sql(
162 1 : INSERT INTO attr_summary (activity_id, total_transactions, gross_value)
163 1 : WITH Attributed AS (
164 1 : SELECT
165 1 : logs.activity_id,
166 1 : txn.id AS txn_id,
167 1 : txn.total_amount,
168 1 : ROW_NUMBER() OVER (
169 1 : PARTITION BY txn.id ORDER BY logs.occurred_at DESC
170 1 : ) AS attribution_rank
171 1 : FROM activity_events logs
172 1 : JOIN attr_users users ON logs.user_public_id = users.public_id
173 1 : JOIN attr_transactions txn ON users.id = txn.user_id
174 1 : WHERE txn.source_created_at > logs.occurred_at
175 1 : AND txn.source_created_at <= TIMESTAMP_ADD(
176 1 : logs.occurred_at, INTERVAL 7 DAY
177 1 : )
178 1 : AND txn.status NOT IN ('status_x', 'status_y')
179 1 : )
180 1 : SELECT
181 1 : activity_id,
182 1 : COUNT(txn_id) AS total_transactions,
183 1 : COALESCE(SUM(total_amount), 0) AS gross_value
184 1 : FROM Attributed
185 1 : WHERE attribution_rank = 1
186 1 : GROUP BY activity_id
187 1 : )sql";
188 1 : const ::googlesql::ResolvedStatement* stmt = Analyze(kCompactSql);
189 1 : ASSERT_NE(stmt, nullptr);
190 1 : ASSERT_EQ(stmt->node_kind(), ::googlesql::RESOLVED_INSERT_STMT);
191 1 : TestTranspiler t;
192 1 : const std::string emitted =
193 1 : t.EmitInsertSelect(stmt->GetAs<::googlesql::ResolvedInsertStmt>());
194 2 : ASSERT_FALSE(emitted.empty()) << "EmitInsertSelect returned empty for:\n"
195 2 : << kCompactSql;
196 1 : SCOPED_TRACE(emitted);
197 1 : ::duckdb_result result{};
198 1 : const auto rc = ::duckdb_query(conn_, emitted.c_str(), &result);
199 1 : if (rc != ::DuckDBSuccess) {
200 0 : const char* err = ::duckdb_result_error(&result);
201 0 : FAIL() << "DuckDB rejected EmitInsertSelect SQL\n"
202 0 : << "source_sql:\n"
203 0 : << kCompactSql << "\n"
204 0 : << "emitted_sql:\n"
205 0 : << emitted << "\n"
206 0 : << "duckdb_error:\n"
207 0 : << (err == nullptr ? "(null)" : err);
208 0 : }
209 1 : ::duckdb_destroy_result(&result);
210 1 : }
211 :
212 : // R17 follow-up: INSERT...SELECT materialization of the attribution
213 : // body must bind (EmitInsertSelect clears join-alias flags).
214 1 : TEST_F(TranspilerAttributionCompositionTest, AttributionInsertSelectBinds) {
215 1 : const std::string sql = absl::StrCat(
216 1 : "INSERT INTO attr_summary "
217 1 : "(activity_id, total_transactions, gross_value) ",
218 1 : kAttributionSelectBody,
219 1 : "SELECT id, total_transactions, gross_value FROM Attribution");
220 1 : const ::googlesql::ResolvedStatement* stmt = Analyze(sql);
221 1 : ASSERT_NE(stmt, nullptr);
222 1 : ASSERT_EQ(stmt->node_kind(), ::googlesql::RESOLVED_INSERT_STMT);
223 1 : TestTranspiler t;
224 1 : const std::string emitted =
225 1 : t.EmitInsertSelect(stmt->GetAs<::googlesql::ResolvedInsertStmt>());
226 2 : ASSERT_FALSE(emitted.empty()) << "EmitInsertSelect returned empty for:\n"
227 2 : << sql;
228 1 : SCOPED_TRACE(emitted);
229 1 : ::duckdb_result result{};
230 1 : const auto rc = ::duckdb_query(conn_, emitted.c_str(), &result);
231 1 : if (rc != ::DuckDBSuccess) {
232 0 : const char* err = ::duckdb_result_error(&result);
233 0 : FAIL() << "DuckDB rejected EmitInsertSelect SQL\n"
234 0 : << "source_sql:\n"
235 0 : << sql << "\n"
236 0 : << "emitted_sql:\n"
237 0 : << emitted << "\n"
238 0 : << "duckdb_error:\n"
239 0 : << (err == nullptr ? "(null)" : err);
240 0 : }
241 1 : ::duckdb_destroy_result(&result);
242 1 : }
243 :
244 : // R17 follow-up: CTAS materialization of the attribution body must bind.
245 1 : TEST_F(TranspilerAttributionCompositionTest, AttributionCtasSelectBinds) {
246 1 : const std::string sql =
247 1 : absl::StrCat("CREATE TABLE attr_summary_ctas AS ",
248 1 : kAttributionSelectBody,
249 1 : "SELECT id AS activity_id, total_transactions, gross_value "
250 1 : "FROM Attribution");
251 1 : const ::googlesql::ResolvedStatement* stmt = Analyze(sql);
252 1 : ASSERT_NE(stmt, nullptr);
253 1 : ASSERT_EQ(stmt->node_kind(),
254 1 : ::googlesql::RESOLVED_CREATE_TABLE_AS_SELECT_STMT);
255 1 : TestTranspiler t;
256 1 : const std::string emitted = t.EmitCtasSelect(
257 1 : stmt->GetAs<::googlesql::ResolvedCreateTableAsSelectStmt>());
258 2 : ASSERT_FALSE(emitted.empty()) << "EmitCtasSelect returned empty for:\n"
259 2 : << sql;
260 1 : SCOPED_TRACE(emitted);
261 1 : ::duckdb_result result{};
262 1 : const auto rc = ::duckdb_query(conn_, emitted.c_str(), &result);
263 1 : if (rc != ::DuckDBSuccess) {
264 0 : const char* err = ::duckdb_result_error(&result);
265 0 : FAIL() << "DuckDB rejected EmitCtasSelect SQL\n"
266 0 : << "source_sql:\n"
267 0 : << sql << "\n"
268 0 : << "emitted_sql:\n"
269 0 : << emitted << "\n"
270 0 : << "duckdb_error:\n"
271 0 : << (err == nullptr ? "(null)" : err);
272 0 : }
273 1 : ::duckdb_destroy_result(&result);
274 1 : }
275 :
276 : } // namespace transpiler
277 : } // namespace duckdb
278 : } // namespace engine
279 : } // namespace backend
280 : } // namespace bigquery_emulator
|
