diff --git a/.claude/commands/run_cost_optimization.md b/.claude/commands/run_cost_optimization.md
new file mode 100644
index 0000000..3874d57
--- /dev/null
+++ b/.claude/commands/run_cost_optimization.md
@@ -0,0 +1 @@
+Read `cost-optimization/README.md` and follow its instructions exactly. Do not invoke it as a skill or slash command — just read the file and execute each step.
diff --git a/CLAUDE.md b/CLAUDE.md
index 966c344..ff78e7b 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -12,6 +12,7 @@ Available demos — run via slash command or natural language:
| Fraud Detection (Delta Lake) | `/run_fraud_delta_lake` | `run fraud delta lake demo` |
| Hopsworks + Feldera | `/run_hopsworks` | `run hopsworks demo` |
| TikTok Recommender System | `/run_tiktok` | `run tiktok recommender demo` |
+| Cost Optimization | `/run_cost_optimization` | `run cost optimization demo` |
When the user runs `/run_fraud_demo` or asks to run fraud detection, **read** `agentic-fraud-detection/feldera-analyze-fraud.md`
and follow its instructions exactly. Do not invoke it as a skill or slash command — just read the file.
@@ -36,3 +37,6 @@ and follow its instructions exactly. Do not invoke it as a skill or slash comman
When the user runs `/run_tiktok` or asks to run the TikTok recommender demo, **read** `tik-tok-recommender-system/README.md`
and follow its instructions exactly. Do not invoke it as a skill or slash command — just read the file.
+
+When the user runs `/run_cost_optimization` or asks to run the cost optimization demo, **read** `cost-optimization/README.md`
+and follow its instructions exactly. Do not invoke it as a skill or slash command — just read the file.
diff --git a/README.md b/README.md
index a19e404..1879e77 100644
--- a/README.md
+++ b/README.md
@@ -4,7 +4,7 @@
- 
+ 
@@ -45,6 +45,7 @@ Claude handles all steps automatically: Feldera setup, pipeline loading, SQL gen
| Debezium + JDBC | [debezium-jdbc/](debezium-jdbc/) | `/run_debezium_jdbc` | Postgres, Debezium, Redpanda/Kafka | CDC pipeline sinking Feldera views to Postgres via Redpanda/Kafka and JDBC sink connectors. |
| Hopsworks Integration | [hopsworks/](hopsworks/) | `/run_hopsworks` | Hopsworks, Kafka, XGBoost | Feature pipeline integration with Hopsworks feature store and Kafka, with XGBoost model training. |
| TikTok Recommender System | [tik-tok-recommender-system/](tik-tok-recommender-system/) | `/run_tiktok` | Redpanda/Kafka | TikTok-style recommendation system using Feldera and Redpanda/Kafka. |
+| Cost Optimization | [cost-optimization/](cost-optimization/) | `/run_cost_optimization` | TPC-H parquet | Shrink a pipeline's CPU/memory after backfill: start big, checkpoint, restart small at steady-state cost. |
## ⚙️ Pre-requisites
diff --git a/cost-optimization/README.md b/cost-optimization/README.md
new file mode 100644
index 0000000..2257d4a
--- /dev/null
+++ b/cost-optimization/README.md
@@ -0,0 +1,55 @@
+# Cost optimization
+
+Shrink a Feldera pipeline's CPU and memory allocation once the initial
+backfill is done — pay for backfill capacity only while you actually need it,
+then run the steady state on a fraction of the resources.
+
+## Flow
+
+1. Create the pipeline (`tpch.sql`) with generous resources for backfill
+ (4 GB / 4 cores by default).
+2. Start the pipeline. Wait until every input connector has finished its
+ initial snapshot — detected via `end_of_input` on each input endpoint
+ (the equivalent of `delta_phase` on CDC sources, which the script has a
+ commented-out branch for once we swap parquet → Delta Lake / S3).
+3. Stop the pipeline with `force=False` so Feldera writes a checkpoint.
+4. Patch the runtime config to the steady-state envelope (1 GB / 1 core).
+5. Start the pipeline again. It resumes from the checkpoint and runs at the
+ smaller cost.
+
+The orchestration lives in [`run.py`](run.py). Resource sizes, pipeline
+name, and worker count are constants at the top of the script.
+
+## Prerequisites
+
+- A reachable Feldera instance.
+- TPC-H parquet files served at `http://localhost:8000/{lineitem,orders,part,customer,supplier,partsupp,nation,region}.parquet`
+ — `tpch.sql` reads from those URLs. Generate them however you prefer
+ (DuckDB's `tpch` extension, `dbgen`, or a copy from S3) and run
+ `python -m http.server 8000` from the directory that holds them.
+- [`uv`](https://docs.astral.sh/uv/) for running the script.
+
+## Run
+
+```bash
+# point at your Feldera instance (defaults shown)
+export FELDERA_HOST=http://localhost:8080
+# export FELDERA_API_KEY=apikey:... # only for remote/cloud instances
+
+uv run cost-optimization/run.py
+```
+
+## What to look for
+
+While the script runs you'll see, in order:
+
+- `creating pipeline ... with backfill resources (4096 MB / 4 cores)`
+- `starting pipeline (backfill phase)`
+- progress lines like `3/8 done; waiting on [...]` until all input
+ connectors report `end_of_input`
+- `stopping pipeline gracefully (force=False) — Feldera will checkpoint`
+- `patching runtime config to steady-state resources (1024 MB / 1 cores)`
+- `restarting pipeline at steady-state cost — resumes from checkpoint`
+
+After the run, the pipeline `cost-optimization-tpch` keeps running with the
+smaller resource envelope.
diff --git a/cost-optimization/run.py b/cost-optimization/run.py
new file mode 100644
index 0000000..7b6f7d2
--- /dev/null
+++ b/cost-optimization/run.py
@@ -0,0 +1,163 @@
+# /// script
+# requires-python = ">=3.10"
+# dependencies = [
+# "feldera>=0.292.0",
+# ]
+# ///
+"""
+Cost-optimization demo: shrink a Feldera pipeline's resource footprint
+once backfill is complete.
+
+Flow (matches cost-optimization/README.md):
+ 1. Create the pipeline with generous resources for the backfill phase.
+ 2. Start it; wait until every input connector has fully ingested its
+ snapshot (end_of_input == True).
+ 3. Stop gracefully (force=False) so Feldera writes a checkpoint.
+ 4. Patch the runtime config to a smaller resource envelope.
+ 5. Start again — Feldera resumes from the checkpoint at steady-state cost.
+
+Env:
+ FELDERA_HOST default http://localhost:8080
+ FELDERA_API_KEY optional
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import time
+from pathlib import Path
+
+from feldera import FelderaClient, PipelineBuilder
+from feldera.runtime_config import RuntimeConfig, Resources
+
+PIPELINE_NAME = "cost-optimization-tpch"
+SQL_FILE = Path(__file__).parent / "tpch.sql"
+
+WORKERS = 4
+
+BACKFILL_RESOURCES = Resources(
+ cpu_cores_min=4,
+ cpu_cores_max=4,
+ memory_mb_min=4096,
+ memory_mb_max=4096,
+)
+
+STEADY_RESOURCES = Resources(
+ cpu_cores_min=1,
+ cpu_cores_max=1,
+ memory_mb_min=1024,
+ memory_mb_max=1024,
+)
+
+POLL_INTERVAL_S = 2.0
+BACKFILL_TIMEOUT_S = 60 * 30
+
+
+def log(msg: str) -> None:
+ print(f"[cost-opt] {msg}", flush=True)
+
+
+def make_client() -> FelderaClient:
+ host = os.environ.get("FELDERA_HOST", "http://localhost:8080")
+ api_key = os.environ.get("FELDERA_API_KEY")
+ log(f"connecting to {host}")
+ return FelderaClient(host, api_key=api_key)
+
+
+def backfill_complete(pipeline) -> tuple[bool, str]:
+ """All input connectors have finished their initial snapshot."""
+ stats = pipeline.stats()
+ inputs = stats.inputs or []
+ if not inputs:
+ return False, "no input connectors reporting yet"
+
+ pending = []
+ for ep in inputs:
+ m = ep.metrics
+ end_of_input = bool(getattr(m, "end_of_input", False)) if m else False
+
+ # Future CDC sources (e.g. Delta Lake / S3 with snapshot_and_follow)
+ # will expose `delta_phase` instead of `end_of_input`. When we move
+ # the connectors over, swap the check above for the block below:
+ #
+ # delta_phase = getattr(m, "delta_phase", None) if m else None
+ # done = delta_phase in ("follow", "replay") # past initial snapshot
+ # if not done:
+ # pending.append(ep.endpoint_name)
+
+ if not end_of_input:
+ pending.append(ep.endpoint_name or "")
+
+ if pending:
+ return False, f"{len(inputs) - len(pending)}/{len(inputs)} done; waiting on {pending}"
+ return True, f"all {len(inputs)} input connectors finished"
+
+
+def wait_for_backfill(pipeline) -> None:
+ log("waiting for backfill to complete...")
+ deadline = time.monotonic() + BACKFILL_TIMEOUT_S
+ last_msg = ""
+ while time.monotonic() < deadline:
+ done, msg = backfill_complete(pipeline)
+ if msg != last_msg:
+ log(msg)
+ last_msg = msg
+ if done:
+ return
+ time.sleep(POLL_INTERVAL_S)
+ raise TimeoutError(f"backfill did not complete within {BACKFILL_TIMEOUT_S}s")
+
+
+def main() -> int:
+ if not SQL_FILE.exists():
+ log(f"missing SQL file: {SQL_FILE}")
+ return 1
+ sql = SQL_FILE.read_text()
+
+ client = make_client()
+
+ log(f"creating pipeline '{PIPELINE_NAME}' with backfill resources "
+ f"({BACKFILL_RESOURCES.memory_mb_max} MB / {BACKFILL_RESOURCES.cpu_cores_max} cores)")
+ pipeline = PipelineBuilder(
+ client,
+ name=PIPELINE_NAME,
+ sql=sql,
+ runtime_config=RuntimeConfig(
+ workers=WORKERS,
+ storage=True,
+ resources=BACKFILL_RESOURCES,
+ ),
+ ).create_or_replace()
+
+ log("starting pipeline (backfill phase)")
+ pipeline.start()
+
+ wait_for_backfill(pipeline)
+
+ log("stopping pipeline gracefully (force=False) — Feldera will checkpoint")
+ pipeline.stop(force=False)
+
+ log(f"patching runtime config to steady-state resources "
+ f"({STEADY_RESOURCES.memory_mb_max} MB / {STEADY_RESOURCES.cpu_cores_max} cores)")
+ pipeline.set_runtime_config(
+ RuntimeConfig(
+ workers=WORKERS,
+ storage=True,
+ resources=STEADY_RESOURCES,
+ )
+ )
+
+ log("restarting pipeline at steady-state cost — resumes from checkpoint")
+ pipeline.start()
+
+ # sanity check; should be done already
+ status, msg = backfill_complete(pipeline)
+ log(f"final check: backfill complete={status}; {msg}")
+
+ log("done. pipeline is running with reduced resources.")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/cost-optimization/tpch.sql b/cost-optimization/tpch.sql
new file mode 100644
index 0000000..faec9c3
--- /dev/null
+++ b/cost-optimization/tpch.sql
@@ -0,0 +1,1024 @@
+CREATE TABLE LINEITEM (
+ L_ORDERKEY INTEGER NOT NULL,
+ L_PARTKEY INTEGER NOT NULL,
+ L_SUPPKEY INTEGER NOT NULL,
+ L_LINENUMBER INTEGER NOT NULL,
+ L_QUANTITY DECIMAL(15,2) NOT NULL,
+ L_EXTENDEDPRICE DECIMAL(15,2) NOT NULL,
+ L_DISCOUNT DECIMAL(15,2) NOT NULL,
+ L_TAX DECIMAL(15,2) NOT NULL,
+ L_RETURNFLAG CHAR(1) NOT NULL,
+ L_LINESTATUS CHAR(1) NOT NULL,
+ L_SHIPDATE DATE NOT NULL,
+ L_COMMITDATE DATE NOT NULL,
+ L_RECEIPTDATE DATE NOT NULL,
+ L_SHIPINSTRUCT CHAR(25) NOT NULL,
+ L_SHIPMODE STRING NOT NULL,
+ L_COMMENT VARCHAR(44) NOT NULL,
+ PRIMARY KEY (L_ORDERKEY, L_LINENUMBER)
+) WITH (
+ 'materialized' = 'true',
+ 'connectors' = '[{
+ "transport": {
+ "name": "url_input",
+ "config": {
+ "path": "http://localhost:8000/lineitem.parquet"
+ }
+ },
+ "format": {
+ "name": "parquet",
+ "config": {}
+ }
+ }]'
+);
+
+CREATE TABLE ORDERS (
+ O_ORDERKEY INTEGER NOT NULL PRIMARY KEY,
+ O_CUSTKEY INTEGER NOT NULL,
+ O_ORDERSTATUS CHAR(1) NOT NULL,
+ O_TOTALPRICE DECIMAL(15,2) NOT NULL,
+ O_ORDERDATE DATE NOT NULL,
+ O_ORDERPRIORITY CHAR(15) NOT NULL,
+ O_CLERK CHAR(15) NOT NULL,
+ O_SHIPPRIORITY INTEGER NOT NULL,
+ O_COMMENT VARCHAR(79) NOT NULL
+) WITH (
+ 'materialized' = 'true',
+ 'connectors' = '[{
+ "transport": { "name": "url_input", "config": { "path": "http://localhost:8000/orders.parquet" }},
+ "format": { "name": "parquet" }
+ }]'
+);
+
+CREATE TABLE PART (
+ P_PARTKEY INTEGER NOT NULL PRIMARY KEY,
+ P_NAME VARCHAR(55) NOT NULL,
+ P_MFGR CHAR(25) NOT NULL,
+ P_BRAND CHAR(10) NOT NULL,
+ P_TYPE VARCHAR(25) NOT NULL,
+ P_SIZE INTEGER NOT NULL,
+ P_CONTAINER STRING NOT NULL,
+ P_RETAILPRICE DECIMAL(15,2) NOT NULL,
+ P_COMMENT VARCHAR(23) NOT NULL
+) WITH (
+ 'materialized' = 'true',
+ 'connectors' = '[{
+ "transport": { "name": "url_input", "config": { "path": "http://localhost:8000/part.parquet" }},
+ "format": { "name": "parquet" }
+ }]'
+);
+
+CREATE TABLE CUSTOMER (
+ C_CUSTKEY INTEGER NOT NULL PRIMARY KEY,
+ C_NAME VARCHAR(25) NOT NULL,
+ C_ADDRESS VARCHAR(40) NOT NULL,
+ C_NATIONKEY INTEGER NOT NULL,
+ C_PHONE CHAR(15) NOT NULL,
+ C_ACCTBAL DECIMAL(15,2) NOT NULL,
+ C_MKTSEGMENT CHAR(10) NOT NULL,
+ C_COMMENT VARCHAR(117) NOT NULL
+) WITH (
+ 'materialized' = 'true',
+ 'connectors' = '[{
+ "transport": { "name": "url_input", "config": { "path": "http://localhost:8000/customer.parquet" }},
+ "format": { "name": "parquet" }
+ }]'
+);
+
+CREATE TABLE SUPPLIER (
+ S_SUPPKEY INTEGER NOT NULL PRIMARY KEY,
+ S_NAME CHAR(25) NOT NULL,
+ S_ADDRESS VARCHAR(40) NOT NULL,
+ S_NATIONKEY INTEGER NOT NULL,
+ S_PHONE CHAR(15) NOT NULL,
+ S_ACCTBAL DECIMAL(15,2) NOT NULL,
+ S_COMMENT VARCHAR(101) NOT NULL
+) WITH (
+ 'materialized' = 'true',
+ 'connectors' = '[{
+ "transport": { "name": "url_input", "config": { "path": "http://localhost:8000/supplier.parquet" }},
+ "format": { "name": "parquet" }
+ }]'
+);
+
+CREATE TABLE PARTSUPP (
+ PS_PARTKEY INTEGER NOT NULL,
+ PS_SUPPKEY INTEGER NOT NULL,
+ PS_AVAILQTY INTEGER NOT NULL,
+ PS_SUPPLYCOST DECIMAL(15,2) NOT NULL,
+ PS_COMMENT VARCHAR(199) NOT NULL,
+ PRIMARY KEY (PS_PARTKEY, PS_SUPPKEY)
+) WITH (
+ 'materialized' = 'true',
+ 'connectors' = '[{
+ "transport": { "name": "url_input", "config": { "path": "http://localhost:8000/partsupp.parquet" }},
+ "format": { "name": "parquet" }
+ }]'
+);
+
+CREATE TABLE NATION (
+ N_NATIONKEY INTEGER NOT NULL PRIMARY KEY,
+ N_NAME CHAR(25) NOT NULL,
+ N_REGIONKEY INTEGER NOT NULL,
+ N_COMMENT VARCHAR(152)
+) WITH (
+ 'materialized' = 'true',
+ 'connectors' = '[{
+ "transport": { "name": "url_input", "config": { "path": "http://localhost:8000/nation.parquet" }},
+ "format": { "name": "parquet" }
+ }]'
+);
+
+CREATE TABLE REGION (
+ R_REGIONKEY INTEGER NOT NULL PRIMARY KEY,
+ R_NAME CHAR(25) NOT NULL,
+ R_COMMENT VARCHAR(152)
+) WITH (
+ 'materialized' = 'true',
+ 'connectors' = '[{
+ "transport": { "name": "url_input", "config": { "path": "http://localhost:8000/region.parquet" }},
+ "format": { "name": "parquet" }
+ }]'
+);
+
+create materialized view q1
+with('connectors' = '[{
+ "index": "q1_idx",
+ "transport": {
+ "name": "delta_table_output",
+ "config": {
+ "uri": "file:///var/folders/jf/wkm7njns5g35b0r56nfmh_900000gn/T/tmp67l5teu4/q1",
+ "mode": "truncate"
+ }
+ },
+ "enable_output_buffer": true,
+ "max_output_buffer_time_millis": 2000
+ }]')
+ as
+
+ select
+ l_returnflag,
+ l_linestatus,
+ sum(l_quantity) as sum_qty,
+ sum(l_extendedprice) as sum_base_price,
+ sum(l_extendedprice * (1 - l_discount)) as sum_disc_price,
+ sum(l_extendedprice * (1 - l_discount) * (1 + l_tax)) as sum_charge,
+ CAST(ROUND(avg(l_quantity), 1) as DECIMAL(15, 1)) as avg_qty,
+ CAST(ROUND(avg(l_extendedprice), 1) as DECIMAL(15, 1)) as avg_price,
+ CAST(ROUND(avg(l_discount), 1) as DECIMAL(15, 1)) as avg_disc,
+ count(*) as count_order
+ from
+ lineitem
+ where
+ l_shipdate <= date '1998-12-01' - interval '71' DAY
+ group by
+ l_returnflag,
+ l_linestatus
+ order by
+ l_returnflag,
+ l_linestatus
+ ;
+
+create index q1_idx on q1(l_returnflag,l_linestatus);
+
+create materialized view q2 as
+
+ select
+ s_acctbal,
+ s_name,
+ n_name,
+ p_partkey,
+ p_mfgr,
+ s_address,
+ s_phone,
+ s_comment
+from
+ part,
+ supplier,
+ partsupp,
+ nation,
+ region
+where
+ p_partkey = ps_partkey
+ and s_suppkey = ps_suppkey
+ and p_size = 38
+ and p_type like '%TIN'
+ and s_nationkey = n_nationkey
+ and n_regionkey = r_regionkey
+ and r_name = 'MIDDLE EAST'
+ and ps_supplycost = (
+ select
+ min(ps_supplycost)
+ from
+ partsupp,
+ supplier,
+ nation,
+ region
+ where
+ p_partkey = ps_partkey
+ and s_suppkey = ps_suppkey
+ and s_nationkey = n_nationkey
+ and n_regionkey = r_regionkey
+ and r_name = 'MIDDLE EAST'
+ )
+order by
+ s_acctbal desc,
+ n_name,
+ s_name,
+ p_partkey
+LIMIT 100;
+
+
+create materialized view q3
+with('connectors' = '[{
+ "index": "q3_idx",
+ "transport": {
+ "name": "delta_table_output",
+ "config": {
+ "uri": "file:///var/folders/jf/wkm7njns5g35b0r56nfmh_900000gn/T/tmp67l5teu4/q3",
+ "mode": "truncate"
+ }
+ },
+ "enable_output_buffer": true,
+ "max_output_buffer_time_millis": 2000
+ }]')
+ as
+
+ select
+ l_orderkey,
+ sum(l_extendedprice * (1 - l_discount)) as revenue,
+ o_orderdate,
+ o_shippriority
+from
+ customer,
+ orders,
+ lineitem
+where
+ c_mktsegment = 'FURNITURE'
+ and c_custkey = o_custkey
+ and l_orderkey = o_orderkey
+ and o_orderdate < date '1995-03-29'
+ and l_shipdate > date '1995-03-29'
+group by
+ l_orderkey,
+ o_orderdate,
+ o_shippriority
+order by
+ revenue desc,
+ o_orderdate
+LIMIT 10;
+
+create index q3_idx on q3(l_orderkey,o_orderdate,o_shippriority);
+
+create materialized view q4
+with('connectors' = '[{
+ "index": "q4_idx",
+ "transport": {
+ "name": "delta_table_output",
+ "config": {
+ "uri": "file:///var/folders/jf/wkm7njns5g35b0r56nfmh_900000gn/T/tmp67l5teu4/q4",
+ "mode": "truncate"
+ }
+ },
+ "enable_output_buffer": true,
+ "max_output_buffer_time_millis": 2000
+ }]')
+ as
+
+select
+ o_orderpriority,
+ count(*) as order_count
+from
+ orders
+where
+ o_orderdate >= date '1997-07-01'
+ and o_orderdate < date '1997-07-01' + interval '3' month
+ and exists (
+ select
+ *
+ from
+ lineitem
+ where
+ l_orderkey = o_orderkey
+ and l_commitdate < l_receiptdate
+ )
+group by
+ o_orderpriority
+order by
+ o_orderpriority;
+
+create index q4_idx on q4(o_orderpriority);
+
+create materialized view q5
+with('connectors' = '[{
+ "index": "q5_idx",
+ "transport": {
+ "name": "delta_table_output",
+ "config": {
+ "uri": "file:///var/folders/jf/wkm7njns5g35b0r56nfmh_900000gn/T/tmp67l5teu4/q5",
+ "mode": "truncate"
+ }
+ },
+ "enable_output_buffer": true,
+ "max_output_buffer_time_millis": 2000
+ }]')
+ as
+
+ select
+ n_name,
+ sum(l_extendedprice * (1 - l_discount)) as revenue
+from
+ customer,
+ orders,
+ lineitem,
+ supplier,
+ nation,
+ region
+where
+ c_custkey = o_custkey
+ and l_orderkey = o_orderkey
+ and l_suppkey = s_suppkey
+ and c_nationkey = s_nationkey
+ and s_nationkey = n_nationkey
+ and n_regionkey = r_regionkey
+ and r_name = 'MIDDLE EAST'
+ and o_orderdate >= date '1994-01-01'
+ and o_orderdate < date '1994-01-01' + interval '1' year
+group by
+ n_name
+order by
+ revenue desc;
+
+create index q5_idx on q5(n_name);
+
+create materialized view q6
+with('connectors' = '[{
+ "transport": {
+ "name": "delta_table_output",
+ "config": {
+ "uri": "file:///var/folders/jf/wkm7njns5g35b0r56nfmh_900000gn/T/tmp67l5teu4/q6",
+ "mode": "truncate"
+ }
+ },
+ "enable_output_buffer": true,
+ "max_output_buffer_time_millis": 2000
+ }]')
+ as
+
+ select
+ sum(l_extendedprice * l_discount) as revenue
+from
+ lineitem
+where
+ l_shipdate >= date '1994-01-01'
+ and l_shipdate < date '1994-01-01' + interval '1' year
+ and l_discount between 0.08 - 0.01 and 0.08 + 0.01
+ and l_quantity < 24;
+
+
+create materialized view q8
+with('connectors' = '[{
+ "index": "q8_idx",
+ "transport": {
+ "name": "delta_table_output",
+ "config": {
+ "uri": "file:///var/folders/jf/wkm7njns5g35b0r56nfmh_900000gn/T/tmp67l5teu4/q8",
+ "mode": "truncate"
+ }
+ },
+ "enable_output_buffer": true,
+ "max_output_buffer_time_millis": 2000
+ }]')
+ as
+
+ select
+ o_year,
+ CAST(ROUND(sum(case
+ when nation = 'INDIA' then volume
+ else 0
+ end) / sum(volume)) as DECIMAL(15, 2)) as mkt_share
+from
+ (
+ select
+ extract(year from o_orderdate) as o_year,
+ l_extendedprice * (1 - l_discount) as volume,
+ n2.n_name as nation
+ from
+ part,
+ supplier,
+ lineitem,
+ orders,
+ customer,
+ nation n1,
+ nation n2,
+ region
+ where
+ p_partkey = l_partkey
+ and s_suppkey = l_suppkey
+ and l_orderkey = o_orderkey
+ and o_custkey = c_custkey
+ and c_nationkey = n1.n_nationkey
+ and n1.n_regionkey = r_regionkey
+ and r_name = 'ASIA'
+ and s_nationkey = n2.n_nationkey
+ and o_orderdate between date '1995-01-01' and date '1996-12-31'
+ and p_type = 'PROMO BRUSHED COPPER'
+ ) as all_nations
+group by
+ o_year
+order by
+ o_year;
+
+create index q8_idx on q8(o_year);
+
+create materialized view q9
+with('connectors' = '[{
+ "index": "q9_idx",
+ "transport": {
+ "name": "delta_table_output",
+ "config": {
+ "uri": "file:///var/folders/jf/wkm7njns5g35b0r56nfmh_900000gn/T/tmp67l5teu4/q9",
+ "mode": "truncate"
+ }
+ },
+ "enable_output_buffer": true,
+ "max_output_buffer_time_millis": 2000
+ }]')
+ as
+
+ select
+ nation,
+ o_year,
+ sum(amount) as sum_profit
+from
+ (
+ select
+ n_name as nation,
+ extract(year from o_orderdate) as o_year,
+ l_extendedprice * (1 - l_discount) - ps_supplycost * l_quantity as amount
+ from
+ part,
+ supplier,
+ lineitem,
+ partsupp,
+ orders,
+ nation
+ where
+ s_suppkey = l_suppkey
+ and ps_suppkey = l_suppkey
+ and ps_partkey = l_partkey
+ and p_partkey = l_partkey
+ and o_orderkey = l_orderkey
+ and s_nationkey = n_nationkey
+ and p_name like '%yellow%'
+ ) as profit
+group by
+ nation,
+ o_year
+order by
+nation,
+ o_year desc;
+
+create index q9_idx on q9(nation,o_year);
+
+create materialized view q10
+with('connectors' = '[{
+ "index": "q10_idx",
+ "transport": {
+ "name": "delta_table_output",
+ "config": {
+ "uri": "file:///var/folders/jf/wkm7njns5g35b0r56nfmh_900000gn/T/tmp67l5teu4/q10",
+ "mode": "truncate"
+ }
+ },
+ "enable_output_buffer": true,
+ "max_output_buffer_time_millis": 2000
+ }]')
+ as
+
+select
+ c_custkey,
+ c_name,
+ sum(l_extendedprice * (1 - l_discount)) as revenue,
+ c_acctbal,
+ n_name,
+ c_address,
+ c_phone,
+ c_comment
+from
+ customer,
+ orders,
+ lineitem,
+ nation
+where
+ c_custkey = o_custkey
+ and l_orderkey = o_orderkey
+ and o_orderdate >= date '1994-01-01'
+ and o_orderdate < date '1994-01-01' + interval '3' month
+ and l_returnflag = 'R'
+ and c_nationkey = n_nationkey
+group by
+ c_custkey,
+ c_name,
+ c_acctbal,
+ c_phone,
+ n_name,
+ c_address,
+ c_comment
+order by
+ revenue desc
+LIMIT 20;
+
+create index q10_idx on q10(c_custkey);
+
+create materialized view q11
+with('connectors' = '[{
+ "index": "q11_idx",
+ "transport": {
+ "name": "delta_table_output",
+ "config": {
+ "uri": "file:///var/folders/jf/wkm7njns5g35b0r56nfmh_900000gn/T/tmp67l5teu4/q11",
+ "mode": "truncate"
+ }
+ },
+ "enable_output_buffer": true,
+ "max_output_buffer_time_millis": 2000
+ }]')
+ as
+
+ select
+ ps_partkey,
+ sum(ps_supplycost * ps_availqty) as value
+from
+ partsupp,
+ supplier,
+ nation
+where
+ ps_suppkey = s_suppkey
+ and s_nationkey = n_nationkey
+ and n_name = 'ARGENTINA'
+group by
+ ps_partkey
+having
+ sum(ps_supplycost * ps_availqty) > (
+ select
+ sum(ps_supplycost * ps_availqty) * 0.0001000000
+ from
+ partsupp,
+ supplier,
+ nation
+ where
+ ps_suppkey = s_suppkey
+ and s_nationkey = n_nationkey
+ and n_name = 'ARGENTINA'
+ )
+order by
+ value desc;
+
+create index q11_idx on q11(ps_partkey);
+
+create materialized view q12
+with('connectors' = '[{
+ "index": "q12_idx",
+ "transport": {
+ "name": "delta_table_output",
+ "config": {
+ "uri": "file:///var/folders/jf/wkm7njns5g35b0r56nfmh_900000gn/T/tmp67l5teu4/q12",
+ "mode": "truncate"
+ }
+ },
+ "enable_output_buffer": true,
+ "max_output_buffer_time_millis": 2000
+ }]')
+ as
+
+ select
+ l_shipmode,
+ sum(case
+ when o_orderpriority = '1-URGENT'
+ or o_orderpriority = '2-HIGH'
+ then 1
+ else 0
+ end) as high_line_count,
+ sum(case
+ when o_orderpriority <> '1-URGENT'
+ and o_orderpriority <> '2-HIGH'
+ then 1
+ else 0
+ end) as low_line_count
+from
+ orders,
+ lineitem
+where
+ o_orderkey = l_orderkey
+ and l_shipmode in ('FOB', 'SHIP')
+ and l_commitdate < l_receiptdate
+ and l_shipdate < l_commitdate
+ and l_receiptdate >= date '1994-01-01'
+ and l_receiptdate < date '1994-01-01' + interval '1' year
+group by
+ l_shipmode
+order by
+ l_shipmode;
+
+create index q12_idx on q12(l_shipmode);
+
+create materialized view q13
+with('connectors' = '[{
+ "index": "q13_idx",
+ "transport": {
+ "name": "delta_table_output",
+ "config": {
+ "uri": "file:///var/folders/jf/wkm7njns5g35b0r56nfmh_900000gn/T/tmp67l5teu4/q13",
+ "mode": "truncate"
+ }
+ },
+ "enable_output_buffer": true,
+ "max_output_buffer_time_millis": 2000
+ }]')
+ as
+
+ select
+ c_count,
+ count(*) as custdist
+from
+ (
+ select
+ c_custkey,
+ count(o_orderkey)
+ from
+ customer left outer join orders on
+ c_custkey = o_custkey
+ and o_comment not like '%express%packages%'
+ group by
+ c_custkey
+ ) as c_orders (c_custkey, c_count)
+group by
+ c_count
+order by
+ custdist desc,
+ c_count desc;
+
+create index q13_idx on q13(c_count);
+
+create materialized view q14 as
+
+ select
+ CAST(ROUND(100.00 * sum(case
+ when p_type like 'PROMO%'
+ then l_extendedprice * (1 - l_discount)
+ else 0
+ end) / sum(l_extendedprice * (1 - l_discount))) as DECIMAL(15,2)) as promo_revenue
+from
+ lineitem,
+ part
+where
+ l_partkey = p_partkey
+ and l_shipdate >= date '1994-03-01'
+ and l_shipdate < date '1994-03-01' + interval '1' month;
+
+
+create materialized view q15 as
+
+ with revenue0 as (select
+ l_suppkey as supplier_no,
+ sum(l_extendedprice * (1 - l_discount)) as total_revenue
+ from
+ lineitem
+ where
+ l_shipdate >= date '1993-01-01'
+ and l_shipdate < date '1993-01-01' + interval '3' month
+ group by
+ l_suppkey)
+ select
+ s_suppkey,
+ s_name,
+ s_address,
+ s_phone,
+ total_revenue
+from
+ supplier,
+ revenue0
+where
+ s_suppkey = supplier_no
+ and total_revenue = (
+ select
+ max(total_revenue)
+ from
+ revenue0
+ )
+order by
+ s_suppkey;
+
+
+create materialized view q16
+with('connectors' = '[{
+ "transport": {
+ "name": "delta_table_output",
+ "config": {
+ "uri": "file:///var/folders/jf/wkm7njns5g35b0r56nfmh_900000gn/T/tmp67l5teu4/q16",
+ "mode": "truncate"
+ }
+ },
+ "enable_output_buffer": true,
+ "max_output_buffer_time_millis": 2000
+ }]')
+ as
+
+ select
+ p_brand,
+ p_type,
+ p_size,
+ count(distinct ps_suppkey) as supplier_cnt
+from
+ partsupp,
+ part
+where
+ p_partkey = ps_partkey
+ and p_brand <> 'Brand#45'
+ and p_type not like 'SMALL PLATED%'
+ and p_size in (19, 17, 16, 23, 10, 4, 38, 11)
+ and ps_suppkey not in (
+ select
+ s_suppkey
+ from
+ supplier
+ where
+ s_comment like '%Customer%Complaints%'
+ )
+group by
+ p_brand,
+ p_type,
+ p_size
+order by
+ supplier_cnt desc,
+ p_brand,
+ p_type,
+ p_size;
+
+
+create materialized view q17 as
+
+ select
+ CAST(ROUND(sum(l_extendedprice) / 7.0) as DECIMAL(15,2)) as avg_yearly
+from
+ lineitem,
+ part
+where
+ p_partkey = l_partkey
+ and p_brand = 'Brand#52'
+ and p_container = 'LG CAN'
+ and l_quantity < (
+ select
+ 0.2 * avg(l_quantity)
+ from
+ lineitem
+ where
+ l_partkey = p_partkey
+ );
+
+
+create materialized view q18
+with('connectors' = '[{
+ "index": "q18_idx",
+ "transport": {
+ "name": "delta_table_output",
+ "config": {
+ "uri": "file:///var/folders/jf/wkm7njns5g35b0r56nfmh_900000gn/T/tmp67l5teu4/q18",
+ "mode": "truncate"
+ }
+ },
+ "enable_output_buffer": true,
+ "max_output_buffer_time_millis": 2000
+ }]')
+ as
+
+ select
+ c_name,
+ c_custkey,
+ o_orderkey,
+ o_orderdate,
+ o_totalprice,
+ sum(l_quantity) as sum_quantity
+from
+ customer,
+ orders,
+ lineitem
+where
+ o_orderkey in (
+ select
+ l_orderkey
+ from
+ lineitem
+ group by
+ l_orderkey having
+ sum(l_quantity) > 313
+ )
+ and c_custkey = o_custkey
+ and o_orderkey = l_orderkey
+group by
+ c_name,
+ c_custkey,
+ o_orderkey,
+ o_orderdate,
+ o_totalprice
+order by
+ o_totalprice desc,
+ o_orderdate
+LIMIT 100;
+
+create index q18_idx on q18(c_custkey,o_orderkey);
+
+create materialized view q19 as
+
+ select
+ sum(l_extendedprice* (1 - l_discount)) as revenue
+from
+ lineitem,
+ part
+where
+ (
+ p_partkey = l_partkey
+ and p_brand = 'Brand#22'
+ and p_container in ('SM CASE', 'SM BOX', 'SM PACK', 'SM PKG')
+ and l_quantity >= 8 and l_quantity <= 8 + 10
+ and p_size between 1 and 5
+ and l_shipmode in ('AIR', 'AIR REG')
+ and l_shipinstruct = 'DELIVER IN PERSON'
+ )
+ or
+ (
+ p_partkey = l_partkey
+ and p_brand = 'Brand#23'
+ and p_container in ('MED BAG', 'MED BOX', 'MED PKG', 'MED PACK')
+ and l_quantity >= 10 and l_quantity <= 10 + 10
+ and p_size between 1 and 10
+ and l_shipmode in ('AIR', 'AIR REG')
+ and l_shipinstruct = 'DELIVER IN PERSON'
+ )
+ or
+ (
+ p_partkey = l_partkey
+ and p_brand = 'Brand#12'
+ and p_container in ('LG CASE', 'LG BOX', 'LG PACK', 'LG PKG')
+ and l_quantity >= 24 and l_quantity <= 24 + 10
+ and p_size between 1 and 15
+ and l_shipmode in ('AIR', 'AIR REG')
+ and l_shipinstruct = 'DELIVER IN PERSON'
+ );
+
+
+create materialized view q20 as
+
+ select
+ s_name,
+ s_address
+from
+ supplier,
+ nation
+where
+ s_suppkey in (
+ select
+ ps_suppkey
+ from
+ partsupp
+ where
+ ps_partkey in (
+ select
+ p_partkey
+ from
+ part
+ where
+ p_name like 'frosted%'
+ )
+ and ps_availqty > (
+ select
+ 0.5 * sum(l_quantity)
+ from
+ lineitem
+ where
+ l_partkey = ps_partkey
+ and l_suppkey = ps_suppkey
+ and l_shipdate >= date '1994-01-01'
+ and l_shipdate < date '1994-01-01' + interval '1' year
+ )
+ )
+ and s_nationkey = n_nationkey
+ and n_name = 'IRAN'
+order by
+ s_name;
+
+
+create materialized view q21
+with('connectors' = '[{
+ "index": "q21_idx",
+ "transport": {
+ "name": "delta_table_output",
+ "config": {
+ "uri": "file:///var/folders/jf/wkm7njns5g35b0r56nfmh_900000gn/T/tmp67l5teu4/q21",
+ "mode": "truncate"
+ }
+ },
+ "enable_output_buffer": true,
+ "max_output_buffer_time_millis": 2000
+ }]')
+ as
+
+ select
+ s_name,
+ count(*) as numwait
+from
+ supplier,
+ lineitem l1,
+ orders,
+ nation
+where
+ s_suppkey = l1.l_suppkey
+ and o_orderkey = l1.l_orderkey
+ and o_orderstatus = 'F'
+ and l1.l_receiptdate > l1.l_commitdate
+ and exists (
+ select
+ *
+ from
+ lineitem l2
+ where
+ l2.l_orderkey = l1.l_orderkey
+ and l2.l_suppkey <> l1.l_suppkey
+ )
+ and not exists (
+ select
+ *
+ from
+ lineitem l3
+ where
+ l3.l_orderkey = l1.l_orderkey
+ and l3.l_suppkey <> l1.l_suppkey
+ and l3.l_receiptdate > l3.l_commitdate
+ )
+ and s_nationkey = n_nationkey
+ and n_name = 'GERMANY'
+group by
+ s_name
+order by
+ numwait desc,
+ s_name
+LIMIT 100
+;
+
+create index q21_idx on q21(s_name);
+
+create materialized view q22
+with('connectors' = '[{
+ "index": "q22_idx",
+ "transport": {
+ "name": "delta_table_output",
+ "config": {
+ "uri": "file:///var/folders/jf/wkm7njns5g35b0r56nfmh_900000gn/T/tmp67l5teu4/q22",
+ "mode": "truncate"
+ }
+ },
+ "enable_output_buffer": true,
+ "max_output_buffer_time_millis": 2000
+ }]')
+ as
+
+ select
+ cntrycode,
+ count(*) as numcust,
+ sum(c_acctbal) as totacctbal
+from
+ (
+ select
+ substring(c_phone from 1 for 2) as cntrycode,
+ c_acctbal
+ from
+ customer
+ where
+ substring(c_phone from 1 for 2) in
+ ('30', '24', '31', '38', '25', '34', '37')
+ and c_acctbal > (
+ select
+ avg(c_acctbal)
+ from
+ customer
+ where
+ c_acctbal > 0.00
+ and substring(c_phone from 1 for 2) in
+ ('30', '24', '31', '38', '25', '34', '37')
+ )
+ and not exists (
+ select
+ *
+ from
+ orders
+ where
+ o_custkey = c_custkey
+ )
+ ) as custsale
+group by
+ cntrycode
+order by
+ cntrycode;
+
+create index q22_idx on q22(cntrycode);
\ No newline at end of file