# Copyright Materialize, Inc. and contributors. All rights reserved.
#
# Use of this software is governed by the Business Source License
# included in the LICENSE file at the root of this repository.
#
# As of the Change Date specified in that file, in accordance with
# the Business Source License, use of this software will be governed
# by the Apache License, Version 2.0.

statement ok
CREATE TABLE materialized_view_refresh_log (
	view_name TEXT,
	last_refresh TIMESTAMP DEFAULT now(),
	refresh_duration DOUBLE PRECISION DEFAULT 0
);

statement ok
CREATE TABLE products (
	product_id INTEGER NOT NULL,
	product_name VARCHAR(255) NOT NULL,
	base_price NUMERIC(10, 2) NOT NULL,
	category_id INTEGER NOT NULL,
	supplier_id INTEGER NOT NULL,
	available BOOLEAN NOT NULL,
	last_update_time TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);

statement ok
CREATE TABLE categories (
	category_id INTEGER NOT NULL,
	category_name VARCHAR(255) NOT NULL,
	parent_id INT
);

statement ok
CREATE TABLE suppliers (
	supplier_id INTEGER NOT NULL,
	supplier_name VARCHAR(255) NOT NULL
);

statement ok
CREATE TABLE sales (
	sale_id INTEGER NOT NULL,
	product_id INTEGER NOT NULL,
	sale_price NUMERIC(10, 2) NOT NULL,
	sale_date TIMESTAMP NOT NULL,
	price NUMERIC(10, 2) NOT NULL
);

statement ok
CREATE TABLE inventory (
	inventory_id INTEGER NOT NULL,
	product_id INTEGER NOT NULL,
	stock INTEGER NOT NULL,
	warehouse_id INTEGER NOT NULL,
	restock_date TIMESTAMP NOT NULL
);

statement ok
CREATE TABLE promotions (
	promotion_id INTEGER NOT NULL,
	product_id INTEGER NOT NULL,
	promotion_discount NUMERIC(10, 2) NOT NULL,
	start_date TIMESTAMP NOT NULL,
	end_date TIMESTAMP NOT NULL,
	active BOOLEAN NOT NULL,
	updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);

statement ok
CREATE TABLE shopping_cart (
    product_id INT NOT NULL,
	product_name TEXT NOT NULL,
	category_id INT NOT NULL,
	price NUMERIC(10, 2) NOT NULL,
    ts TIMESTAMP NOT NULL DEFAULT NOW()
);

statement ok
CREATE INDEX idx_products_product_name ON products(product_name);

statement ok
CREATE INDEX idx_sales_product_id ON sales(product_id);

statement ok
CREATE INDEX idx_sales_sale_date ON sales(sale_date);

statement ok
CREATE INDEX idx_sales_product_id_sale_date ON sales(product_id, sale_date);

statement ok
CREATE INDEX idx_promotions_product_id ON promotions(product_id);

statement ok
CREATE INDEX idx_promotions_active ON promotions(active);

statement ok
CREATE INDEX idx_promotions_product_id_active ON promotions(product_id, active);

statement ok
CREATE INDEX idx_inventory_product_id ON inventory(product_id);

query T multiline
EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR
WITH recent_prices AS (
	SELECT grp.product_id, AVG(price) AS avg_price
	FROM (SELECT DISTINCT product_id FROM sales) grp,
	LATERAL (
		SELECT product_id, price
		FROM sales
		WHERE sales.product_id = grp.product_id
		ORDER BY sale_date DESC LIMIT 10
	) sub
	GROUP BY grp.product_id
),
promotion_effect AS (
	SELECT
		p.product_id,
		MIN(pr.promotion_discount) AS promotion_discount
	FROM promotions pr
	JOIN products p ON pr.product_id = p.product_id
	WHERE pr.active = TRUE
	GROUP BY p.product_id
),
popularity_score AS (
	SELECT
		s.product_id,
		RANK() OVER (PARTITION BY p.category_id ORDER BY COUNT(s.sale_id) DESC) AS popularity_rank,
		COUNT(s.sale_id) AS sale_count
	FROM sales s
	JOIN products p ON s.product_id = p.product_id
	GROUP BY s.product_id, p.category_id
),
inventory_status AS (
	SELECT
		i.product_id,
		SUM(i.stock) AS total_stock,
		RANK() OVER (ORDER BY SUM(i.stock) DESC) AS stock_rank
	FROM inventory i
	GROUP BY i.product_id
),
high_demand_products AS (
	SELECT
		p.product_id,
		AVG(s.sale_price) AS avg_sale_price,
		COUNT(s.sale_id) AS total_sales
	FROM products p
	JOIN sales s ON p.product_id = s.product_id
	GROUP BY p.product_id
	HAVING COUNT(s.sale_id) > (SELECT AVG(total_sales) FROM (SELECT COUNT(*) AS total_sales FROM sales GROUP BY product_id) subquery)
),
dynamic_pricing AS (
	SELECT
		p.product_id,
		p.base_price,
		CASE
			WHEN pop.popularity_rank <= 3 THEN 1.2
			WHEN pop.popularity_rank BETWEEN 4 AND 10 THEN 1.1
			ELSE 0.9
		END AS popularity_adjustment,
		rp.avg_price,
		COALESCE(1.0 - (pe.promotion_discount / 100), 1) AS promotion_discount,
		CASE
			WHEN inv.stock_rank <= 3 THEN 1.1
			WHEN inv.stock_rank BETWEEN 4 AND 10 THEN 1.05
			ELSE 1
		END AS stock_adjustment,
		CASE
			WHEN p.base_price > rp.avg_price THEN 1 + (p.base_price - rp.avg_price) / rp.avg_price
			ELSE 1 - (rp.avg_price - p.base_price) / rp.avg_price
		END AS demand_multiplier,
		hd.avg_sale_price,
		CASE
			WHEN p.product_name ilike '%cheap%' THEN 0.8
			ELSE 1.0
		END AS additional_discount
	FROM products p
	LEFT JOIN recent_prices rp ON p.product_id = rp.product_id
	LEFT JOIN promotion_effect pe ON p.product_id = pe.product_id
	JOIN popularity_score pop ON p.product_id = pop.product_id
	LEFT JOIN inventory_status inv ON p.product_id = inv.product_id
	LEFT JOIN high_demand_products hd ON p.product_id = hd.product_id
)
SELECT
	dp.product_id,
	dp.base_price * dp.popularity_adjustment * dp.promotion_discount * dp.stock_adjustment * dp.demand_multiplier * dp.additional_discount AS adjusted_price,
	p.last_update_time
FROM dynamic_pricing dp
JOIN products p ON dp.product_id = p.product_id;
----
Explained Query:
  With
    cte l0 =
      Project (#0{product_id}, #3)
        Map ((#1{sum_price} / bigint_to_numeric(case when (#2{count} = 0) then null else #2{count} end)))
          Reduce group_by=[#0{product_id}] aggregates=[sum(#1{price}), count(*)]
            Project (#0{product_id}, #2{price})
              TopK group_by=[#0{product_id}] order_by=[#1{sale_date} desc nulls_first] limit=10
                Project (#1{product_id}, #3{sale_date}, #4{price})
                  ReadIndex on=sales idx_sales_product_id=[*** full scan ***]
    cte l1 =
      Project (#0{product_id})
        ReadIndex on=products idx_products_product_name=[*** full scan ***]
    cte l2 =
      ArrangeBy keys=[[#0{product_id}]]
        Get l1
    cte l3 =
      Reduce group_by=[#0{product_id}] aggregates=[min(#1{promotion_discount})]
        Project (#1{product_id}, #2{promotion_discount})
          Join on=(#1{product_id} = #8{product_id}) type=differential
            ArrangeBy keys=[[#1{product_id}]]
              ReadIndex on=materialize.public.promotions idx_promotions_active=[lookup value=(true)]
            Get l2
    cte l4 =
      Distinct project=[#0{product_id}]
        Get l1
    cte l5 =
      ArrangeBy keys=[[#1{product_id}]]
        ReadIndex on=sales idx_sales_product_id=[differential join]
    cte l6 =
      Project (#0{product_id}..=#2{base_price}, #10..=#12)
        Map (case when (#5) IS NULL then null else #4 end, case when (#8) IS NULL then null else #7{min_promotion_discount} end)
          Join on=(#0{product_id} = #3{product_id} = #6{product_id} = #9{product_id}) type=delta
            ArrangeBy keys=[[#0{product_id}]]
              Project (#0{product_id}..=#2{base_price})
                ReadIndex on=products idx_products_product_name=[*** full scan ***]
            ArrangeBy keys=[[#0{product_id}]]
              Union
                Map (true)
                  Get l0
                Map (null, null)
                  Threshold
                    Union
                      Negate
                        Project (#0{product_id})
                          Get l0
                      Get l4
            ArrangeBy keys=[[#0{product_id}]]
              Union
                Map (true)
                  Get l3
                Map (null, null)
                  Threshold
                    Union
                      Negate
                        Project (#0{product_id})
                          Get l3
                      Get l4
            ArrangeBy keys=[[#0{product_id}]]
              Project (#2, #3)
                Filter (#2) IS NOT NULL
                  Map (record_get[0](record_get[1](#1)), record_get[0](#1))
                    FlatMap unnest_list(#0{rank})
                      Project (#1{rank})
                        Reduce group_by=[#1{category_id}] aggregates=[rank[order_by=[#0{product_id} desc nulls_first]](row(list[row(#0{product_id}, #1{category_id}, #2{count})], #2{count}))]
                          Reduce group_by=[#0{product_id}, #1{category_id}] aggregates=[count(*)]
                            Project (#1{product_id}, #6{category_id})
                              Join on=(#1{product_id} = #5{product_id}) type=differential
                                Get l5
                                ArrangeBy keys=[[#0{product_id}]]
                                  Project (#0{product_id}, #3{category_id})
                                    ReadIndex on=products idx_products_product_name=[*** full scan ***]
    cte l7 =
      Project (#2, #3)
        Filter (#2) IS NOT NULL
          Map (record_get[0](record_get[1](#1)), record_get[0](#1))
            FlatMap unnest_list(#0{rank})
              Reduce aggregates=[rank[order_by=[#0{product_id} desc nulls_first]](row(list[row(#0{product_id}, #1{sum_stock})], #1{sum_stock}))]
                Reduce group_by=[#0{product_id}] aggregates=[sum(#1{stock})]
                  Project (#1{product_id}, #2{stock})
                    ReadIndex on=inventory idx_inventory_product_id=[*** full scan ***]
    cte l8 =
      Project (#0{product_id})
        Filter (bigint_to_numeric(#1{count}) > (#2{sum_count} / bigint_to_numeric(case when (#3{count} = 0) then null else #3{count} end)))
          CrossJoin type=differential
            ArrangeBy keys=[[]]
              Reduce group_by=[#0{product_id}] aggregates=[count(*)]
                Project (#0{product_id})
                  Join on=(#0{product_id} = #2{product_id}) type=differential
                    Get l2
                    Get l5
            ArrangeBy keys=[[]]
              Reduce aggregates=[sum(#0{count}), count(*)]
                Project (#1{count})
                  Reduce group_by=[#0{product_id}] aggregates=[count(*)]
                    Project (#1{product_id})
                      ReadIndex on=sales idx_sales_product_id=[*** full scan ***]
    cte l9 =
      Distinct project=[#0{product_id}]
        Project (#0{product_id})
          Get l6
  Return
    Project (#0{product_id}, #13, #11{last_update_time})
      Map (case when (#8) IS NULL then null else #7 end, (((((#2{base_price} * case when (#3{popularity_rank} <= 3) then 1.2 else case when ((#3{popularity_rank} <= 10) AND (#3{popularity_rank} >= 4)) then 1.1 else 0.9 end end) * coalesce((1 - (#5{promotion_discount} / 100)), 1)) * case when (#12{stock_rank} <= 3) then 1.1 else case when ((#12 <= 10) AND (#12 >= 4)) then 1.05 else 1 end end) * case when (#2{base_price} > #4{avg_price}) then (1 + ((#2{base_price} - #4{avg_price}) / #4{avg_price})) else (1 - ((#4{avg_price} - #2{base_price}) / #4{avg_price})) end) * case when ilike["%cheap%"](varchar_to_text(#1{product_name})) then 0.8 else 1 end))
        Join on=(#0{product_id} = #6{product_id} = #9{product_id} = #10{product_id}) type=delta
          ArrangeBy keys=[[#0{product_id}]]
            Get l6
          ArrangeBy keys=[[#0{product_id}]]
            Union
              Map (true)
                Get l7
              Map (null, null)
                Threshold
                  Union
                    Negate
                      Project (#0)
                        Get l7
                    Get l9
          ArrangeBy keys=[[#0{product_id}]]
            Union
              Get l8
              Threshold
                Union
                  Negate
                    Get l8
                  Get l9
          ArrangeBy keys=[[#0{product_id}]]
            Project (#0{product_id}, #6{last_update_time})
              ReadIndex on=products idx_products_product_name=[*** full scan ***]

Used Indexes:
  - materialize.public.idx_products_product_name (*** full scan ***)
  - materialize.public.idx_sales_product_id (*** full scan ***, differential join)
  - materialize.public.idx_promotions_active (lookup)
  - materialize.public.idx_inventory_product_id (*** full scan ***)

Target cluster: quickstart

EOF

statement ok
CREATE VIEW dynamic_pricing AS
WITH recent_prices AS (
	SELECT grp.product_id, AVG(price) AS avg_price
	FROM (SELECT DISTINCT product_id FROM sales) grp,
	LATERAL (
		SELECT product_id, price
		FROM sales
		WHERE sales.product_id = grp.product_id
		ORDER BY sale_date DESC LIMIT 10
	) sub
	GROUP BY grp.product_id
),
promotion_effect AS (
	SELECT
		p.product_id,
		MIN(pr.promotion_discount) AS promotion_discount
	FROM promotions pr
	JOIN products p ON pr.product_id = p.product_id
	WHERE pr.active = TRUE
	GROUP BY p.product_id
),
popularity_score AS (
	SELECT
		s.product_id,
		RANK() OVER (PARTITION BY p.category_id ORDER BY COUNT(s.sale_id) DESC) AS popularity_rank,
		COUNT(s.sale_id) AS sale_count
	FROM sales s
	JOIN products p ON s.product_id = p.product_id
	GROUP BY s.product_id, p.category_id
),
inventory_status AS (
	SELECT
		i.product_id,
		SUM(i.stock) AS total_stock,
		RANK() OVER (ORDER BY SUM(i.stock) DESC) AS stock_rank
	FROM inventory i
	GROUP BY i.product_id
),
high_demand_products AS (
	SELECT
		p.product_id,
		AVG(s.sale_price) AS avg_sale_price,
		COUNT(s.sale_id) AS total_sales
	FROM products p
	JOIN sales s ON p.product_id = s.product_id
	GROUP BY p.product_id
	HAVING COUNT(s.sale_id) > (SELECT AVG(total_sales) FROM (SELECT COUNT(*) AS total_sales FROM sales GROUP BY product_id) subquery)
),
dynamic_pricing AS (
	SELECT
		p.product_id,
		p.base_price,
		CASE
			WHEN pop.popularity_rank <= 3 THEN 1.2
			WHEN pop.popularity_rank BETWEEN 4 AND 10 THEN 1.1
			ELSE 0.9
		END AS popularity_adjustment,
		rp.avg_price,
		COALESCE(1.0 - (pe.promotion_discount / 100), 1) AS promotion_discount,
		CASE
			WHEN inv.stock_rank <= 3 THEN 1.1
			WHEN inv.stock_rank BETWEEN 4 AND 10 THEN 1.05
			ELSE 1
		END AS stock_adjustment,
		CASE
			WHEN p.base_price > rp.avg_price THEN 1 + (p.base_price - rp.avg_price) / rp.avg_price
			ELSE 1 - (rp.avg_price - p.base_price) / rp.avg_price
		END AS demand_multiplier,
		hd.avg_sale_price,
		CASE
			WHEN p.product_name ilike '%cheap%' THEN 0.8
			ELSE 1.0
		END AS additional_discount
	FROM products p
	LEFT JOIN recent_prices rp ON p.product_id = rp.product_id
	LEFT JOIN promotion_effect pe ON p.product_id = pe.product_id
	JOIN popularity_score pop ON p.product_id = pop.product_id
	LEFT JOIN inventory_status inv ON p.product_id = inv.product_id
	LEFT JOIN high_demand_products hd ON p.product_id = hd.product_id
)
SELECT
	dp.product_id,
	dp.base_price * dp.popularity_adjustment * dp.promotion_discount * dp.stock_adjustment * dp.demand_multiplier * dp.additional_discount AS adjusted_price,
	p.last_update_time
FROM dynamic_pricing dp
JOIN products p ON dp.product_id = p.product_id;

statement ok
CREATE INDEX IF NOT EXISTS dynamic_pricing_product_id_idx ON dynamic_pricing (product_id);

statement ok
CREATE MATERIALIZED VIEW mv_dynamic_pricing AS
WITH recent_prices AS (
	SELECT grp.product_id, AVG(price) AS avg_price
	FROM (SELECT DISTINCT product_id FROM sales) grp,
	LATERAL (
		SELECT product_id, price
		FROM sales
		WHERE sales.product_id = grp.product_id
		ORDER BY sale_date DESC LIMIT 10
	) sub
	GROUP BY grp.product_id
),
promotion_effect AS (
	SELECT
		p.product_id,
		MIN(pr.promotion_discount) AS promotion_discount
	FROM promotions pr
	JOIN products p ON pr.product_id = p.product_id
	WHERE pr.active = TRUE
	GROUP BY p.product_id
),
popularity_score AS (
	SELECT
		s.product_id,
		RANK() OVER (PARTITION BY p.category_id ORDER BY COUNT(s.sale_id) DESC) AS popularity_rank,
		COUNT(s.sale_id) AS sale_count
	FROM sales s
	JOIN products p ON s.product_id = p.product_id
	GROUP BY s.product_id, p.category_id
),
inventory_status AS (
	SELECT
		i.product_id,
		SUM(i.stock) AS total_stock,
		RANK() OVER (ORDER BY SUM(i.stock) DESC) AS stock_rank
	FROM inventory i
	GROUP BY i.product_id
),
high_demand_products AS (
	SELECT
		p.product_id,
		AVG(s.sale_price) AS avg_sale_price,
		COUNT(s.sale_id) AS total_sales
	FROM products p
	JOIN sales s ON p.product_id = s.product_id
	GROUP BY p.product_id
	HAVING COUNT(s.sale_id) > (SELECT AVG(total_sales) FROM (SELECT COUNT(*) AS total_sales FROM sales GROUP BY product_id) subquery)
),
dynamic_pricing AS (
	SELECT
		p.product_id,
		p.base_price,
		CASE
			WHEN pop.popularity_rank <= 3 THEN 1.2
			WHEN pop.popularity_rank BETWEEN 4 AND 10 THEN 1.1
			ELSE 0.9
		END AS popularity_adjustment,
		rp.avg_price,
		COALESCE(1.0 - (pe.promotion_discount / 100), 1) AS promotion_discount,
		CASE
			WHEN inv.stock_rank <= 3 THEN 1.1
			WHEN inv.stock_rank BETWEEN 4 AND 10 THEN 1.05
			ELSE 1
		END AS stock_adjustment,
		CASE
			WHEN p.base_price > rp.avg_price THEN 1 + (p.base_price - rp.avg_price) / rp.avg_price
			ELSE 1 - (rp.avg_price - p.base_price) / rp.avg_price
		END AS demand_multiplier,
		hd.avg_sale_price,
		CASE
			WHEN p.product_name ilike '%cheap%' THEN 0.8
			ELSE 1.0
		END AS additional_discount
	FROM products p
	LEFT JOIN recent_prices rp ON p.product_id = rp.product_id
	LEFT JOIN promotion_effect pe ON p.product_id = pe.product_id
	JOIN popularity_score pop ON p.product_id = pop.product_id
	LEFT JOIN inventory_status inv ON p.product_id = inv.product_id
	LEFT JOIN high_demand_products hd ON p.product_id = hd.product_id
)
SELECT
	dp.product_id,
	dp.base_price * dp.popularity_adjustment * dp.promotion_discount * dp.stock_adjustment * dp.demand_multiplier * dp.additional_discount AS adjusted_price,
	p.last_update_time
FROM dynamic_pricing dp
JOIN products p ON dp.product_id = p.product_id;

statement ok
CREATE INDEX idx_product_id ON mv_dynamic_pricing(product_id);

query T multiline
EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR
SELECT
  sc.product_id,
  sc.product_name,
  c.category_id,
  c.category_name,
  dp.adjusted_price AS price,
  COALESCE(SUM(i.stock), 0) as available_stock
FROM
    shopping_cart sc
JOIN
    products p ON sc.product_id = p.product_id
JOIN
    categories c ON p.category_id = c.category_id
JOIN
    dynamic_pricing dp ON p.product_id = dp.product_id
LEFT JOIN
    inventory i ON p.product_id = i.product_id
GROUP BY
    sc.product_id,
    sc.product_name,
    c.category_id,
    c.category_name,
    dp.adjusted_price;
----
Explained Query:
  With
    cte l0 =
      Project (#0{product_id}, #1{product_name}, #3{category_id}, #5{category_name}, #7{adjusted_price})
        Join on=(#0{product_id} = #2{product_id} = #6{product_id} AND #3{category_id} = #4{category_id}) type=delta
          ArrangeBy keys=[[#0{product_id}]]
            Project (#0{product_id}, #1{product_name})
              ReadStorage materialize.public.shopping_cart
          ArrangeBy keys=[[#0{product_id}], [#1{category_id}]]
            Project (#0{product_id}, #3{category_id})
              ReadIndex on=products idx_products_product_name=[*** full scan ***]
          ArrangeBy keys=[[#0{category_id}]]
            Project (#0{category_id}, #1{category_name})
              ReadStorage materialize.public.categories
          ArrangeBy keys=[[#0{product_id}]]
            ReadIndex on=dynamic_pricing dynamic_pricing_product_id_idx=[delta join lookup]
    cte l1 =
      ArrangeBy keys=[[#0{product_id}]]
        Get l0
    cte l2 =
      Project (#0{product_id}..=#4{adjusted_price}, #7{stock})
        Join on=(#0{product_id} = #6{product_id}) type=differential
          Get l1
          ArrangeBy keys=[[#1{product_id}]]
            ReadIndex on=inventory idx_inventory_product_id=[differential join]
  Return
    Project (#0{product_id}..=#4{adjusted_price}, #6)
      Map (coalesce(#5{sum_stock}, 0))
        Reduce group_by=[#0{product_id}..=#4{adjusted_price}] aggregates=[sum(#5{stock})]
          Union
            Map (null)
              Union
                Negate
                  Project (#0{product_id}..=#4{adjusted_price})
                    Join on=(#0{product_id} = #5{product_id}) type=differential
                      Get l1
                      ArrangeBy keys=[[#0{product_id}]]
                        Distinct project=[#0{product_id}]
                          Project (#0{product_id})
                            Get l2
                Get l0
            Get l2

Source materialize.public.categories
Source materialize.public.shopping_cart

Used Indexes:
  - materialize.public.idx_products_product_name (*** full scan ***)
  - materialize.public.idx_inventory_product_id (differential join)
  - materialize.public.dynamic_pricing_product_id_idx (delta join lookup)

Target cluster: quickstart

EOF

statement ok
CREATE VIEW dynamic_price_shopping_cart AS SELECT
  sc.product_id,
  sc.product_name,
  c.category_id,
  c.category_name,
  dp.adjusted_price AS price,
  COALESCE(SUM(i.stock), 0) as available_stock
FROM
    shopping_cart sc
JOIN
    products p ON sc.product_id = p.product_id
JOIN
    categories c ON p.category_id = c.category_id
JOIN
    dynamic_pricing dp ON p.product_id = dp.product_id
LEFT JOIN
    inventory i ON p.product_id = i.product_id
GROUP BY
    sc.product_id,
    sc.product_name,
    c.category_id,
    c.category_name,
    dp.adjusted_price;

statement ok
CREATE DEFAULT INDEX IF NOT EXISTS dynamic_price_shopping_cart_idx ON dynamic_price_shopping_cart;

query T multiline
EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR
WITH MUTUALLY RECURSIVE
  rollup(category_id int, total numeric(20,10), item_count int) AS (
    -- Base: calculate each category's direct total and item count
    SELECT
      c.category_id,
      COALESCE(SUM(d.price), 0)::numeric(20,10),
      COUNT(d.price)
    FROM categories c
    LEFT JOIN dynamic_price_shopping_cart d
      ON c.category_id = d.category_id
    GROUP BY c.category_id
    UNION ALL
    -- Recursive: bubble each category's totals upward to its parent
    SELECT
      c.parent_id,
      r.total,
      r.item_count
    FROM rollup r
    JOIN categories c
      ON r.category_id = c.category_id
    WHERE c.parent_id IS NOT NULL
  ),
  totals(category_id int, total numeric(20,10), item_count int) AS (
    SELECT
      c.category_id,
      SUM(r.total)::numeric(20,10) AS total,
      SUM(r.item_count) AS item_count
    FROM categories c
    JOIN rollup r
      ON c.category_id = r.category_id
    GROUP BY c.category_id
    HAVING SUM(r.item_count) > 0  -- only include categories with items
  ),
  has_subcategories(category_id int, has_subcategory boolean) AS (
    SELECT
      a.category_id,
      count(*) FILTER (WHERE b.parent_id IS NOT NULL) > 0 AS has_subcategory
    FROM categories a
    LEFT JOIN categories b ON a.category_id = b.parent_id
    GROUP BY a.category_id
  )
SELECT
  t.category_id,
  c.parent_id,
  s.has_subcategory,
  c.category_name,
  t.total,
  t.item_count
FROM totals t
JOIN categories c USING (category_id)
JOIN has_subcategories s USING (category_id);
----
Explained Query:
  With
    cte l0 =
      Project (#0{category_id})
        ReadStorage materialize.public.categories
    cte l1 =
      ArrangeBy keys=[[#0{category_id}]]
        Get l0
    cte l2 =
      Project (#0{category_id}, #2{price})
        Join on=(#0{category_id} = #1{category_id}) type=differential
          Get l1
          ArrangeBy keys=[[#0{category_id}]]
            Project (#2{category_id}, #4{price})
              ReadIndex on=dynamic_price_shopping_cart dynamic_price_shopping_cart_idx=[*** full scan ***]
  Return
    With Mutually Recursive
      cte l3 =
        Project (#0{category_id}, #1, #3)
          Map (bigint_to_integer(#2{count_price}))
            Union
              Project (#0{category_id}, #3, #2{count_price})
                Map (adjust_numeric_scale(coalesce(#1{sum_price}, 0)))
                  Reduce group_by=[#0{category_id}] aggregates=[sum(#1{price}), count(#1{price})]
                    Union
                      Map (null)
                        Union
                          Negate
                            Project (#0{category_id})
                              Join on=(#0{category_id} = #1{category_id}) type=differential
                                Get l1
                                ArrangeBy keys=[[#0{category_id}]]
                                  Distinct project=[#0{category_id}]
                                    Project (#0{category_id})
                                      Get l2
                          Get l0
                      Get l2
              Project (#4{parent_id}, #1, #5)
                Map (integer_to_bigint(#2))
                  Join on=(#0{category_id} = #3{category_id}) type=differential
                    ArrangeBy keys=[[#0{category_id}]]
                      Get l3
                    ArrangeBy keys=[[#0{category_id}]]
                      Project (#0{category_id}, #2{parent_id})
                        Filter (#2{parent_id}) IS NOT NULL
                          ReadStorage materialize.public.categories
    Return
      With
        cte l4 =
          Project (#0{category_id})
            ReadStorage materialize.public.categories
        cte l5 =
          ArrangeBy keys=[[#0{category_id}]]
            Get l4
        cte l6 =
          Project (#0{category_id})
            Join on=(#0{category_id} = #1{parent_id}) type=differential
              Get l5
              ArrangeBy keys=[[#0{parent_id}]]
                Project (#2{parent_id})
                  Filter (#2{parent_id}) IS NOT NULL
                    ReadStorage materialize.public.categories
      Return
        Project (#0{category_id}, #5{parent_id}, #9, #4{category_name}, #1{sum}, #8)
          Filter (#2{sum} > 0)
            Map (bigint_to_integer(#2{sum}), (#7{count} > 0))
              Join on=(#0{category_id} = #3{category_id} = #6{category_id}) type=delta
                ArrangeBy keys=[[#0{category_id}]]
                  Reduce group_by=[#0{category_id}] aggregates=[sum(#1{total}), sum(#2{item_count})]
                    Project (#0{category_id}, #2, #3)
                      Join on=(#0{category_id} = #1{category_id}) type=differential
                        Get l5
                        ArrangeBy keys=[[#0{category_id}]]
                          Get l3
                ArrangeBy keys=[[#0{category_id}]]
                  ReadStorage materialize.public.categories
                ArrangeBy keys=[[#0{category_id}]]
                  Reduce group_by=[#0{category_id}] aggregates=[count((null OR (#1{category_id}) IS NOT NULL))]
                    Union
                      Map (null)
                        Union
                          Negate
                            Project (#0{category_id})
                              Join on=(#0{category_id} = #1{category_id}) type=differential
                                Get l5
                                ArrangeBy keys=[[#0{category_id}]]
                                  Distinct project=[#0{category_id}]
                                    Get l6
                          Get l4
                      Project (#0{category_id}, #0{category_id})
                        Get l6

Source materialize.public.categories

Used Indexes:
  - materialize.public.dynamic_price_shopping_cart_idx (*** full scan ***)

Target cluster: quickstart

EOF

statement ok
CREATE VIEW category_totals AS
WITH MUTUALLY RECURSIVE
  rollup(category_id int, total numeric(20,10), item_count int) AS (
    -- Base: calculate each category's direct total and item count
    SELECT
      c.category_id,
      COALESCE(SUM(d.price), 0)::numeric(20,10),
      COUNT(d.price)
    FROM categories c
    LEFT JOIN dynamic_price_shopping_cart d
      ON c.category_id = d.category_id
    GROUP BY c.category_id
    UNION ALL
    -- Recursive: bubble each category's totals upward to its parent
    SELECT
      c.parent_id,
      r.total,
      r.item_count
    FROM rollup r
    JOIN categories c
      ON r.category_id = c.category_id
    WHERE c.parent_id IS NOT NULL
  ),
  totals(category_id int, total numeric(20,10), item_count int) AS (
    SELECT
      c.category_id,
      SUM(r.total)::numeric(20,10) AS total,
      SUM(r.item_count) AS item_count
    FROM categories c
    JOIN rollup r
      ON c.category_id = r.category_id
    GROUP BY c.category_id
    HAVING SUM(r.item_count) > 0  -- only include categories with items
  ),
  has_subcategories(category_id int, has_subcategory boolean) AS (
    SELECT
      a.category_id,
      count(*) FILTER (WHERE b.parent_id IS NOT NULL) > 0 AS has_subcategory
    FROM categories a
    LEFT JOIN categories b ON a.category_id = b.parent_id
    GROUP BY a.category_id
  )
SELECT
  t.category_id,
  c.parent_id,
  s.has_subcategory,
  c.category_name,
  t.total,
  t.item_count
FROM totals t
JOIN categories c USING (category_id)
JOIN has_subcategories s USING (category_id);

statement ok
CREATE DEFAULT INDEX IF NOT EXISTS category_totals_category_id_idx ON category_totals;

statement ok
CREATE INDEX IF NOT EXISTS category_totals_parent_id_idx ON category_totals (parent_id);

query T multiline
EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR SELECT EXISTS( SELECT 1 FROM inventory WHERE product_id = 1 );
----
Explained Query:
  With
    cte l0 =
      Distinct project=[]
        Project ()
          ReadIndex on=materialize.public.inventory idx_inventory_product_id=[lookup value=(1)]
  Return
    Union
      Map (true)
        Get l0
      Map (false)
        Union
          Negate
            Get l0
          Constant
            - ()

Used Indexes:
  - materialize.public.idx_inventory_product_id (lookup)

Target cluster: quickstart

EOF

query T multiline
EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR WITH selected_product AS (
                        SELECT product_id, product_name, category_id, base_price
                        FROM products
                        WHERE product_id != 1
                        LIMIT 1
                    )
SELECT product_id, product_name, category_id, base_price
                    FROM selected_product;
----
Explained Query:
  TopK limit=1
    Project (#0{product_id}, #1{product_name}, #3{category_id}, #2{base_price})
      Filter (#0{product_id} != 1)
        ReadIndex on=products idx_products_product_name=[*** full scan ***]

Used Indexes:
  - materialize.public.idx_products_product_name (*** full scan ***)

Target cluster: quickstart

EOF
query T multiline
EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR
SELECT product_id, adjusted_price, last_update_time
FROM dynamic_pricing
WHERE product_id = 5;
----
Explained Query (fast path):
  Project (#0{product_id}..=#2{last_update_time})
    ReadIndex on=materialize.public.dynamic_pricing dynamic_pricing_product_id_idx=[lookup value=(5)]

Used Indexes:
  - materialize.public.dynamic_pricing_product_id_idx (lookup)

Target cluster: quickstart

EOF

query T multiline
EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR
SELECT product_id, adjusted_price, last_update_time
FROM mv_dynamic_pricing
WHERE product_id = 5;
----
Explained Query (fast path):
  Project (#0{product_id}..=#2{last_update_time})
    ReadIndex on=materialize.public.mv_dynamic_pricing idx_product_id=[lookup value=(5)]

Used Indexes:
  - materialize.public.idx_product_id (lookup)

Target cluster: quickstart

EOF

query T multiline
EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR
SELECT DISTINCT category_id, category_name
FROM categories
ORDER BY category_name;
----
Explained Query:
  Finish order_by=[#1{category_name} asc nulls_last] output=[#0, #1]
    Distinct project=[#0{category_id}, #1{category_name}]
      Project (#0{category_id}, #1{category_name})
        ReadStorage materialize.public.categories

Source materialize.public.categories

Target cluster: quickstart

EOF

query T multiline
EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR
SELECT
    category_name,
    item_count,
    total as subtotal
FROM category_totals
ORDER BY category_name ASC;
----
Explained Query (fast path):
  Finish order_by=[#0 asc nulls_last] output=[#0..=#2]
    Project (#3{category_name}, #5{item_count}, #4{total})
      ReadIndex on=materialize.public.category_totals category_totals_category_id_idx=[*** full scan ***]

Used Indexes:
  - materialize.public.category_totals_category_id_idx (*** full scan ***)

Target cluster: quickstart

EOF