TPC-DS Q1 — Customer Returns Analysis

TPC-DS benchmark · all TPC-DS queries

Finds customers with above-average returns for their state

Representative SQL

Rendered for DataFusion with default parameters.

WITH customer_total_return AS (
  SELECT
    sr_customer_sk AS ctr_customer_sk,
    sr_store_sk AS ctr_store_sk,
    SUM(SR_FEE) AS ctr_total_return
  FROM store_returns, date_dim
  WHERE
    sr_returned_date_sk = d_date_sk AND d_year = 2000
  GROUP BY
    sr_customer_sk,
    sr_store_sk
)
SELECT
  c_customer_id
FROM customer_total_return AS ctr1, store, customer
WHERE
  ctr1.ctr_total_return > (
    SELECT
      AVG(ctr_total_return) * 1.2
    FROM customer_total_return AS ctr2
    WHERE
      ctr1.ctr_store_sk = ctr2.ctr_store_sk
  )
  AND s_store_sk = ctr1.ctr_store_sk
  AND s_state = 'TN'
  AND ctr1.ctr_customer_sk = c_customer_sk
ORDER BY
  c_customer_id
LIMIT 100

Representative DataFrame

Expression family (Polars, PySpark, DataFusion), default parameters.

def q1_expression_impl(ctx: DataFrameContext) -> Any:
    params = get_parameters(1)
    year = params.get("year", 2000)
    state = params.get("state", "TN")
    agg_field = params.get("agg_field")

    store_returns, date_dim, store, customer = _tables(ctx, "store_returns", "date_dim", "store", "customer")
    col = ctx.col
    lit = ctx.lit

    customer_total = (
        store_returns.join(date_dim, left_on="sr_returned_date_sk", right_on="d_date_sk")
        .filter(col("d_year") == lit(year))
        .group_by(col("sr_customer_sk").alias("ctr_customer_sk"), col("sr_store_sk").alias("ctr_store_sk"))
        .agg(
            ctx.when(col(agg_field).count() > lit(0))
            .then(col(agg_field).sum())
            .otherwise(lit(None))
            .alias("ctr_total_return")
        )
    )

    store_avg = customer_total.group_by("ctr_store_sk").agg(col("ctr_total_return").mean().alias("store_avg"))

    return (
        customer_total.join(store_avg, on="ctr_store_sk")
        .filter(col("ctr_total_return") > col("store_avg") * lit(1.2))
        .join(store.filter(col("s_state") == lit(state)), left_on="ctr_store_sk", right_on="s_store_sk")
        .join(customer, left_on="ctr_customer_sk", right_on="c_customer_sk")
        .select("c_customer_id")
        .sort("c_customer_id")
        .limit(100)
    )

Get the exact query for your run

The blocks above are representative. To get the precise statement a specific platform runs at a specific scale, substitute your own values:

from benchbox import TPCDS

bench = TPCDS(scale_factor=1.0)            # your scale factor
queries = bench.get_queries(dialect="datafusion")   # your platform's dialect; drop dialect= for native SQL
print(queries['1'])

Parameter substitution for this benchmark is seed-driven; pass a seed for a specific stream’s parameters:

print(bench.get_query(1, scale_factor=1.0, seed=42, dialect="datafusion"))

For the DataFrame form:

from benchbox.core.query_catalog import get_dataframe_render

print(get_dataframe_render('tpcds', '1', family="expression").source)  # or family="pandas"

Source

Query text: _sources/tpc-ds/query_templates/query1.tpl