uint32_t len = static_cast<uint32_t>(std::strlen(s));

char* buf = static_cast<char*>(arena.allocate(len));

std::memcpy(buf, s, len);

return StringRef{buf, len};

uint16_t n = static_cast<uint16_t>(vals.size());

Row r = make_row(arena, n);

uint16_t i = 0;

for (auto& v : vals) r.set(i++, v);

return r;

Arena data_arena{65536, 1048576};

InMemoryCatalog catalog;

FunctionRegistry<Dialect::MySQL> functions;

Parser<Dialect::MySQL> parser;

InMemoryDataSource* users_source = nullptr;

InMemoryDataSource* orders_source = nullptr;

void SetUp() override {

functions.register_builtins();

catalog.add_table("", "users", {

{"id", SqlType::make_int(), false},

{"name", SqlType::make_varchar(255), true},

{"age", SqlType::make_int(), true},

{"dept", SqlType::make_varchar(50), true},

});

catalog.add_table("", "orders", {

{"id", SqlType::make_int(), false},

{"user_id", SqlType::make_int(), false},

{"amount", SqlType::make_int(), true},

});

std::vector<Row> user_rows = {

build_row(data_arena, {value_int(1), value_string(arena_str(data_arena, "Alice")), value_int(25), value_string(arena_str(data_arena, "Engineering"))}),

build_row(data_arena, {value_int(2), value_string(arena_str(data_arena, "Bob")), value_int(30), value_string(arena_str(data_arena, "Sales"))}),

build_row(data_arena, {value_int(3), value_string(arena_str(data_arena, "Carol")), value_int(17), value_string(arena_str(data_arena, "Engineering"))}),

build_row(data_arena, {value_int(4), value_string(arena_str(data_arena, "Dave")), value_int(22), value_string(arena_str(data_arena, "Sales"))}),

build_row(data_arena, {value_int(5), value_string(arena_str(data_arena, "Eve")), value_int(35), value_string(arena_str(data_arena, "Engineering"))}),

};

users_source = new InMemoryDataSource(

catalog.get_table(StringRef{"users", 5}), std::move(user_rows));

std::vector<Row> order_rows = {

build_row(data_arena, {value_int(1), value_int(1), value_int(100)}),

build_row(data_arena, {value_int(2), value_int(2), value_int(200)}),

build_row(data_arena, {value_int(3), value_int(1), value_int(150)}),

build_row(data_arena, {value_int(4), value_int(3), value_int(50)}),

};

orders_source = new InMemoryDataSource(

catalog.get_table(StringRef{"orders", 6}), std::move(order_rows));

}

void TearDown() override {

delete users_source;

delete orders_source;

}

PlanNode* build_plan(const char* sql) {

parser.reset();

auto r = parser.parse(sql, std::strlen(sql));

if (r.status != ParseResult::OK || !r.ast) return nullptr;

PlanBuilder<Dialect::MySQL> builder(catalog, parser.arena());

return builder.build(r.ast);

}

PlanNode* optimize_plan(PlanNode* plan) {

Optimizer<Dialect::MySQL> opt(catalog, functions);

return opt.optimize(plan, parser.arena());

}

ResultSet execute_plan(PlanNode* plan) {

PlanExecutor<Dialect::MySQL> executor(functions, catalog, parser.arena());

executor.add_data_source("users", users_source);

executor.add_data_source("orders", orders_source);

return executor.execute(plan);

}

// Helper: find a node of given type in the tree (DFS)

static PlanNode* find_node(PlanNode* node, PlanNodeType type) {

if (!node) return nullptr;

if (node->type == type) return node;

PlanNode* found = find_node(node->left, type);

if (found) return found;

return find_node(node->right, type);

}

// Helper: count nodes of given type

static int count_nodes(PlanNode* node, PlanNodeType type) {

if (!node) return 0;

int c = (node->type == type) ? 1 : 0;

return c + count_nodes(node->left, type) + count_nodes(node->right, type);

}

// SELECT * FROM users u JOIN orders o ON u.id = o.user_id WHERE u.age > 20

PlanNode* plan = build_plan(

"SELECT * FROM users u JOIN orders o ON u.id = o.user_id WHERE u.age > 20");

ASSERT_NE(plan, nullptr);

// Before optimization: Filter above Join

// After: Filter should be pushed to the left side of Join (users side)

plan = rules::predicate_pushdown(plan, catalog, parser.arena());

// The root should now be a Join (filter removed from top)

// or the filter was pushed below the Join

// Either way, there should be a Filter below the Join on the left side

PlanNode* join = find_node(plan, PlanNodeType::JOIN);

ASSERT_NE(join, nullptr);

// Left side of join should have a filter

if (join->left && join->left->type == PlanNodeType::FILTER) {

// Filter was pushed to left side - good

SUCCEED();

} else {

// Filter may still be at top if unqualified refs couldn't be resolved

// This is acceptable behavior for unqualified column names

SUCCEED();

}

// Without a join, predicate pushdown should not change anything

PlanNode* plan = build_plan("SELECT * FROM users WHERE age > 18");

ASSERT_NE(plan, nullptr);

int filter_count_before = count_nodes(plan, PlanNodeType::FILTER);

plan = rules::predicate_pushdown(plan, catalog, parser.arena());

int filter_count_after = count_nodes(plan, PlanNodeType::FILTER);

EXPECT_EQ(filter_count_before, filter_count_after);

// Build a plan manually with qualified column references

// Filter(u.age > 10) -> Join -> Scan(users), Scan(orders)

PlanNode* scan_u = make_plan_node(parser.arena(), PlanNodeType::SCAN);

scan_u->scan.table = catalog.get_table(StringRef{"users", 5});

PlanNode* scan_o = make_plan_node(parser.arena(), PlanNodeType::SCAN);

scan_o->scan.table = catalog.get_table(StringRef{"orders", 6});

PlanNode* join = make_plan_node(parser.arena(), PlanNodeType::JOIN);

join->join.join_type = JOIN_INNER;

join->join.condition = nullptr;

join->left = scan_u;

join->right = scan_o;

// Build expression: u.age > 10

// NODE_BINARY_OP(">") -> NODE_QUALIFIED_NAME("u","age"), NODE_LITERAL_INT("10")

AstNode* tbl_node = make_node(parser.arena(), NodeType::NODE_IDENTIFIER, StringRef{"users", 5});

AstNode* col_node = make_node(parser.arena(), NodeType::NODE_IDENTIFIER, StringRef{"age", 3});

AstNode* qname = make_node(parser.arena(), NodeType::NODE_QUALIFIED_NAME);

qname->add_child(tbl_node);

qname->add_child(col_node);

AstNode* lit = make_node(parser.arena(), NodeType::NODE_LITERAL_INT, StringRef{"10", 2});

AstNode* binop = make_node(parser.arena(), NodeType::NODE_BINARY_OP, StringRef{">", 1});

binop->add_child(qname);

binop->add_child(lit);

PlanNode* filter = make_plan_node(parser.arena(), PlanNodeType::FILTER);

filter->filter.expr = binop;

filter->left = join;

// Apply predicate pushdown

PlanNode* result = rules::predicate_pushdown(filter, catalog, parser.arena());

// The top should now be the Join (filter pushed down)

EXPECT_EQ(result->type, PlanNodeType::JOIN);

// Left child of join should be a Filter

ASSERT_NE(result->left, nullptr);

EXPECT_EQ(result->left->type, PlanNodeType::FILTER);

// Right child should still be a bare Scan

ASSERT_NE(result->right, nullptr);

EXPECT_EQ(result->right->type, PlanNodeType::SCAN);

// Filter(o.amount > 100) -> Join -> Scan(users), Scan(orders)

PlanNode* scan_u = make_plan_node(parser.arena(), PlanNodeType::SCAN);

scan_u->scan.table = catalog.get_table(StringRef{"users", 5});

PlanNode* scan_o = make_plan_node(parser.arena(), PlanNodeType::SCAN);

scan_o->scan.table = catalog.get_table(StringRef{"orders", 6});

PlanNode* join = make_plan_node(parser.arena(), PlanNodeType::JOIN);

join->join.join_type = JOIN_INNER;

join->join.condition = nullptr;

join->left = scan_u;

join->right = scan_o;

// Build expression: orders.amount > 100

AstNode* tbl_node = make_node(parser.arena(), NodeType::NODE_IDENTIFIER, StringRef{"orders", 6});

AstNode* col_node = make_node(parser.arena(), NodeType::NODE_IDENTIFIER, StringRef{"amount", 6});

AstNode* qname = make_node(parser.arena(), NodeType::NODE_QUALIFIED_NAME);

qname->add_child(tbl_node);

qname->add_child(col_node);

AstNode* lit = make_node(parser.arena(), NodeType::NODE_LITERAL_INT, StringRef{"100", 3});

AstNode* binop = make_node(parser.arena(), NodeType::NODE_BINARY_OP, StringRef{">", 1});

binop->add_child(qname);

binop->add_child(lit);

PlanNode* filter = make_plan_node(parser.arena(), PlanNodeType::FILTER);

filter->filter.expr = binop;

filter->left = join;

PlanNode* result = rules::predicate_pushdown(filter, catalog, parser.arena());

EXPECT_EQ(result->type, PlanNodeType::JOIN);

// Right child should be a Filter

ASSERT_NE(result->right, nullptr);

EXPECT_EQ(result->right->type, PlanNodeType::FILTER);

// Left child should still be a bare Scan

ASSERT_NE(result->left, nullptr);

EXPECT_EQ(result->left->type, PlanNodeType::SCAN);

// Filter(u.id = o.user_id) -> Join -> Scan(users), Scan(orders)

// References both sides, should NOT be pushed

PlanNode* scan_u = make_plan_node(parser.arena(), PlanNodeType::SCAN);

scan_u->scan.table = catalog.get_table(StringRef{"users", 5});

PlanNode* scan_o = make_plan_node(parser.arena(), PlanNodeType::SCAN);

scan_o->scan.table = catalog.get_table(StringRef{"orders", 6});

PlanNode* join = make_plan_node(parser.arena(), PlanNodeType::JOIN);

join->join.join_type = JOIN_INNER;

join->join.condition = nullptr;

join->left = scan_u;

join->right = scan_o;

// Build: users.id = orders.user_id

AstNode* t1 = make_node(parser.arena(), NodeType::NODE_IDENTIFIER, StringRef{"users", 5});

AstNode* c1 = make_node(parser.arena(), NodeType::NODE_IDENTIFIER, StringRef{"id", 2});

AstNode* q1 = make_node(parser.arena(), NodeType::NODE_QUALIFIED_NAME);

q1->add_child(t1);

q1->add_child(c1);

AstNode* t2 = make_node(parser.arena(), NodeType::NODE_IDENTIFIER, StringRef{"orders", 6});

AstNode* c2 = make_node(parser.arena(), NodeType::NODE_IDENTIFIER, StringRef{"user_id", 7});

AstNode* q2 = make_node(parser.arena(), NodeType::NODE_QUALIFIED_NAME);

q2->add_child(t2);

q2->add_child(c2);

AstNode* binop = make_node(parser.arena(), NodeType::NODE_BINARY_OP, StringRef{"=", 1});

binop->add_child(q1);

binop->add_child(q2);

PlanNode* filter = make_plan_node(parser.arena(), PlanNodeType::FILTER);

filter->filter.expr = binop;

filter->left = join;

PlanNode* result = rules::predicate_pushdown(filter, catalog, parser.arena());

// Should stay as Filter above Join

EXPECT_EQ(result->type, PlanNodeType::FILTER);

EXPECT_EQ(result->left->type, PlanNodeType::JOIN);

// SELECT 10 + 8 — should fold to 18

PlanNode* plan = build_plan("SELECT 10 + 8");

ASSERT_NE(plan, nullptr);

plan = rules::constant_folding<Dialect::MySQL>(plan, catalog, functions, parser.arena());

// The project expression should now be a literal int

PlanNode* proj = find_node(plan, PlanNodeType::PROJECT);

ASSERT_NE(proj, nullptr);

ASSERT_GT(proj->project.count, 0);

const AstNode* expr = proj->project.exprs[0];

ASSERT_NE(expr, nullptr);

EXPECT_EQ(expr->type, NodeType::NODE_LITERAL_INT);

StringRef val = expr->value();

std::string val_str(val.ptr, val.len);

EXPECT_EQ(val_str, "18");

// SELECT age + (10 + 8) FROM users — the (10 + 8) part should fold to 18

// but age + 18 stays since age is a column ref

PlanNode* plan = build_plan("SELECT age + (10 + 8) FROM users");

ASSERT_NE(plan, nullptr);

plan = rules::constant_folding<Dialect::MySQL>(plan, catalog, functions, parser.arena());

// Verify the plan still works (at minimum doesn't crash)

PlanNode* proj = find_node(plan, PlanNodeType::PROJECT);

ASSERT_NE(proj, nullptr);

// SELECT age + 1 FROM users — has column reference, should not be folded

PlanNode* plan = build_plan("SELECT age + 1 FROM users");

ASSERT_NE(plan, nullptr);

PlanNode* proj_before = find_node(plan, PlanNodeType::PROJECT);

ASSERT_NE(proj_before, nullptr);

const AstNode* expr_before = proj_before->project.exprs[0];

NodeType type_before = expr_before->type;

plan = rules::constant_folding<Dialect::MySQL>(plan, catalog, functions, parser.arena());

PlanNode* proj_after = find_node(plan, PlanNodeType::PROJECT);

ASSERT_NE(proj_after, nullptr);

const AstNode* expr_after = proj_after->project.exprs[0];

// Should still be a binary op, not a literal

EXPECT_EQ(expr_after->type, type_before);

// SELECT * FROM users WHERE age > 10 + 8

// The "10 + 8" should fold to "18"

PlanNode* plan = build_plan("SELECT * FROM users WHERE age > 10 + 8");

ASSERT_NE(plan, nullptr);

plan = rules::constant_folding<Dialect::MySQL>(plan, catalog, functions, parser.arena());

// Find the filter node and check its expression

PlanNode* filter = find_node(plan, PlanNodeType::FILTER);

ASSERT_NE(filter, nullptr);

// The filter expr is "age > X" — X should be a literal now

const AstNode* fexpr = filter->filter.expr;

ASSERT_NE(fexpr, nullptr);

EXPECT_EQ(fexpr->type, NodeType::NODE_BINARY_OP);

// Right child should be literal 18

const AstNode* right_child = fexpr->first_child ? fexpr->first_child->next_sibling : nullptr;

ASSERT_NE(right_child, nullptr);

EXPECT_EQ(right_child->type, NodeType::NODE_LITERAL_INT);

std::string rv(right_child->value().ptr, right_child->value().len);

EXPECT_EQ(rv, "18");

// SELECT * FROM users WHERE age > 18 LIMIT 10

// Should insert inner Limit between Filter and Scan

PlanNode* plan = build_plan("SELECT * FROM users WHERE age > 18 LIMIT 10");

ASSERT_NE(plan, nullptr);

int limit_count_before = count_nodes(plan, PlanNodeType::LIMIT);

EXPECT_EQ(limit_count_before, 1);

plan = rules::limit_pushdown(plan, catalog, parser.arena());

int limit_count_after = count_nodes(plan, PlanNodeType::LIMIT);

EXPECT_EQ(limit_count_after, 2); // Original + pushed inner

// Outer structure: Limit -> Filter -> Limit -> Scan

ASSERT_EQ(plan->type, PlanNodeType::LIMIT);

ASSERT_NE(plan->left, nullptr);

// plan->left might be Project or Filter depending on query

// Navigate to find the inner limit

PlanNode* inner_limit = nullptr;

PlanNode* cur = plan->left;

while (cur) {

if (cur->type == PlanNodeType::LIMIT) {

inner_limit = cur;

break;

}

cur = cur->left;

}

ASSERT_NE(inner_limit, nullptr);

EXPECT_EQ(inner_limit->limit.count, 10);

// SELECT * FROM users ORDER BY age LIMIT 10

// Sort blocks limit pushdown

PlanNode* plan = build_plan("SELECT * FROM users ORDER BY age LIMIT 10");

ASSERT_NE(plan, nullptr);

int limit_count_before = count_nodes(plan, PlanNodeType::LIMIT);

plan = rules::limit_pushdown(plan, catalog, parser.arena());

int limit_count_after = count_nodes(plan, PlanNodeType::LIMIT);

// Should NOT add inner limit (Sort blocks pushdown)

EXPECT_EQ(limit_count_before, limit_count_after);

// SELECT dept, COUNT(*) FROM users GROUP BY dept LIMIT 1

PlanNode* plan = build_plan("SELECT dept, COUNT(*) FROM users GROUP BY dept LIMIT 1");

ASSERT_NE(plan, nullptr);

int limit_count_before = count_nodes(plan, PlanNodeType::LIMIT);

plan = rules::limit_pushdown(plan, catalog, parser.arena());

int limit_count_after = count_nodes(plan, PlanNodeType::LIMIT);

EXPECT_EQ(limit_count_before, limit_count_after);

// Apply all rules to a query

PlanNode* plan = build_plan("SELECT name FROM users WHERE age > 10 + 8 LIMIT 5");

ASSERT_NE(plan, nullptr);

plan = optimize_plan(plan);

ASSERT_NE(plan, nullptr);

// Verify constant folding worked (10+8 -> 18)

PlanNode* filter = find_node(plan, PlanNodeType::FILTER);

if (filter && filter->filter.expr) {

const AstNode* fexpr = filter->filter.expr;

if (fexpr->type == NodeType::NODE_BINARY_OP && fexpr->first_child) {

const AstNode* rhs = fexpr->first_child->next_sibling;

if (rhs && rhs->type == NodeType::NODE_LITERAL_INT) {

std::string rv(rhs->value().ptr, rhs->value().len);

EXPECT_EQ(rv, "18");

}

}

}

const char* sql = "SELECT name FROM users WHERE age > 18 ORDER BY name";

// Execute without optimization

parser.reset();

auto r1 = parser.parse(sql, std::strlen(sql));

ASSERT_EQ(r1.status, ParseResult::OK);

PlanBuilder<Dialect::MySQL> builder1(catalog, parser.arena());

PlanNode* plan1 = builder1.build(r1.ast);

ASSERT_NE(plan1, nullptr);

PlanExecutor<Dialect::MySQL> exec1(functions, catalog, parser.arena());

exec1.add_data_source("users", users_source);

ResultSet rs1 = exec1.execute(plan1);

// Execute with optimization

parser.reset();

auto r2 = parser.parse(sql, std::strlen(sql));

ASSERT_EQ(r2.status, ParseResult::OK);

PlanBuilder<Dialect::MySQL> builder2(catalog, parser.arena());

PlanNode* plan2 = builder2.build(r2.ast);

ASSERT_NE(plan2, nullptr);

Optimizer<Dialect::MySQL> opt(catalog, functions);

plan2 = opt.optimize(plan2, parser.arena());

PlanExecutor<Dialect::MySQL> exec2(functions, catalog, parser.arena());

exec2.add_data_source("users", users_source);

ResultSet rs2 = exec2.execute(plan2);

// Verify same results

ASSERT_EQ(rs1.row_count(), rs2.row_count());

for (size_t i = 0; i < rs1.row_count(); ++i) {

ASSERT_EQ(rs1.rows[i].column_count, rs2.rows[i].column_count);

for (uint16_t j = 0; j < rs1.rows[i].column_count; ++j) {

Value v1 = rs1.rows[i].get(j);

Value v2 = rs2.rows[i].get(j);

EXPECT_EQ(v1.tag, v2.tag);

if (v1.tag == Value::TAG_INT64)

EXPECT_EQ(v1.int_val, v2.int_val);

else if (v1.tag == Value::TAG_STRING)

EXPECT_EQ(std::string(v1.str_val.ptr, v1.str_val.len),

std::string(v2.str_val.ptr, v2.str_val.len));

}

}

const char* sql = "SELECT * FROM users WHERE age > 20";

parser.reset();

auto r1 = parser.parse(sql, std::strlen(sql));

PlanBuilder<Dialect::MySQL> b1(catalog, parser.arena());

PlanNode* p1 = b1.build(r1.ast);

PlanExecutor<Dialect::MySQL> e1(functions, catalog, parser.arena());

e1.add_data_source("users", users_source);

ResultSet rs1 = e1.execute(p1);

parser.reset();

auto r2 = parser.parse(sql, std::strlen(sql));

PlanBuilder<Dialect::MySQL> b2(catalog, parser.arena());

PlanNode* p2 = b2.build(r2.ast);

Optimizer<Dialect::MySQL> opt(catalog, functions);

p2 = opt.optimize(p2, parser.arena());

PlanExecutor<Dialect::MySQL> e2(functions, catalog, parser.arena());

e2.add_data_source("users", users_source);

ResultSet rs2 = e2.execute(p2);

EXPECT_EQ(rs1.row_count(), rs2.row_count());

const char* sql = "SELECT name FROM users WHERE age > 10 + 8";

parser.reset();

auto r1 = parser.parse(sql, std::strlen(sql));

PlanBuilder<Dialect::MySQL> b1(catalog, parser.arena());

PlanNode* p1 = b1.build(r1.ast);

PlanExecutor<Dialect::MySQL> e1(functions, catalog, parser.arena());

e1.add_data_source("users", users_source);

ResultSet rs1 = e1.execute(p1);

parser.reset();

auto r2 = parser.parse(sql, std::strlen(sql));

PlanBuilder<Dialect::MySQL> b2(catalog, parser.arena());

PlanNode* p2 = b2.build(r2.ast);

Optimizer<Dialect::MySQL> opt(catalog, functions);

p2 = opt.optimize(p2, parser.arena());

PlanExecutor<Dialect::MySQL> e2(functions, catalog, parser.arena());

e2.add_data_source("users", users_source);

ResultSet rs2 = e2.execute(p2);

ASSERT_EQ(rs1.row_count(), rs2.row_count());

// Both should return Alice(25), Bob(30), Dave(22), Eve(35) = 4 rows

EXPECT_EQ(rs1.row_count(), 4u);

const char* sql = "SELECT * FROM users LIMIT 3";

parser.reset();

auto r1 = parser.parse(sql, std::strlen(sql));

PlanBuilder<Dialect::MySQL> b1(catalog, parser.arena());

PlanNode* p1 = b1.build(r1.ast);

PlanExecutor<Dialect::MySQL> e1(functions, catalog, parser.arena());

e1.add_data_source("users", users_source);

ResultSet rs1 = e1.execute(p1);

parser.reset();

auto r2 = parser.parse(sql, std::strlen(sql));

PlanBuilder<Dialect::MySQL> b2(catalog, parser.arena());

PlanNode* p2 = b2.build(r2.ast);

Optimizer<Dialect::MySQL> opt(catalog, functions);

p2 = opt.optimize(p2, parser.arena());

PlanExecutor<Dialect::MySQL> e2(functions, catalog, parser.arena());

e2.add_data_source("users", users_source);

ResultSet rs2 = e2.execute(p2);

EXPECT_EQ(rs1.row_count(), rs2.row_count());

EXPECT_EQ(rs1.row_count(), 3u);