TPC-DS Q29 — Store/Catalog Sales-Returns Aggregation

TPC-DS benchmark · all TPC-DS queries

Store sales/returns joined with catalog sales aggregated by item/store

Representative SQL

Rendered for DataFusion with default parameters.

SELECT
  i_item_id,
  i_item_desc,
  s_store_id,
  s_store_name,
  SUM(ss_quantity) AS store_sales_quantity,
  SUM(sr_return_quantity) AS store_returns_quantity,
  SUM(cs_quantity) AS catalog_sales_quantity
FROM store_sales, store_returns, catalog_sales, date_dim AS d1, date_dim AS d2, date_dim AS d3, store, item
WHERE
  d1.d_moy = 4
  AND d1.d_year = 1999
  AND d1.d_date_sk = ss_sold_date_sk
  AND i_item_sk = ss_item_sk
  AND s_store_sk = ss_store_sk
  AND ss_customer_sk = sr_customer_sk
  AND ss_item_sk = sr_item_sk
  AND ss_ticket_number = sr_ticket_number
  AND sr_returned_date_sk = d2.d_date_sk
  AND d2.d_moy BETWEEN 4 AND 4 + 3
  AND d2.d_year = 1999
  AND sr_customer_sk = cs_bill_customer_sk
  AND sr_item_sk = cs_item_sk
  AND cs_sold_date_sk = d3.d_date_sk
  AND d3.d_year IN (1999, 1999 + 1, 1999 + 2)
GROUP BY
  i_item_id,
  i_item_desc,
  s_store_id,
  s_store_name
ORDER BY
  i_item_id,
  i_item_desc,
  s_store_id,
  s_store_name
LIMIT 100

Representative DataFrame

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

def q29_expression_impl(ctx: DataFrameContext) -> Any:
    params = get_parameters(29)
    year = params.get("year", 1999)
    month = params.get("month", 4)
    agg = _q29_aggregate(params.get("agg", "sum"))

    store_sales, store_returns, catalog_sales, date_dim, store, item = _tables(
        ctx, "store_sales", "store_returns", "catalog_sales", "date_dim", "store", "item"
    )
    col = ctx.col
    lit = ctx.lit

    d1 = date_dim.filter((col("d_moy") == lit(month)) & (col("d_year") == lit(year)))

    d2 = date_dim.filter((col("d_moy") >= lit(month)) & (col("d_moy") <= lit(month + 3)) & (col("d_year") == lit(year)))

    d3 = date_dim.filter(col("d_year").is_in([year, year + 1, year + 2]))

    ss_joined = (
        store_sales.join(d1.select("d_date_sk"), left_on="ss_sold_date_sk", right_on="d_date_sk")
        .join(item, left_on="ss_item_sk", right_on="i_item_sk")
        .join(store, left_on="ss_store_sk", right_on="s_store_sk")
    )

    sr_joined = store_returns.join(d2.select("d_date_sk"), left_on="sr_returned_date_sk", right_on="d_date_sk")

    ss_sr = ss_joined.join(
        sr_joined,
        left_on=["ss_customer_sk", "ss_item_sk", "ss_ticket_number"],
        right_on=["sr_customer_sk", "sr_item_sk", "sr_ticket_number"],
    )

    cs_joined = catalog_sales.join(d3.select("d_date_sk"), left_on="cs_sold_date_sk", right_on="d_date_sk")

    result = ss_sr.join(
        cs_joined,
        left_on=["ss_customer_sk", "ss_item_sk"],
        right_on=["cs_bill_customer_sk", "cs_item_sk"],
    )

    return _sort_null_largest_expression(
        ctx,
        result.group_by(["i_item_id", "i_item_desc", "s_store_id", "s_store_name"]).agg(
            [
                (
                    ctx.when(col("ss_quantity").count() > lit(0)).then(col("ss_quantity").sum()).otherwise(lit(None))
                    if agg == "sum"
                    else getattr(col("ss_quantity"), agg)()
                ).alias("store_sales_quantity"),
                (
                    ctx.when(col("sr_return_quantity").count() > lit(0))
                    .then(col("sr_return_quantity").sum())
                    .otherwise(lit(None))
                    if agg == "sum"
                    else getattr(col("sr_return_quantity"), agg)()
                ).alias("store_returns_quantity"),
                (
                    ctx.when(col("cs_quantity").count() > lit(0)).then(col("cs_quantity").sum()).otherwise(lit(None))
                    if agg == "sum"
                    else getattr(col("cs_quantity"), agg)()
                ).alias("catalog_sales_quantity"),
            ]
        ),
        ["i_item_id", "i_item_desc", "s_store_id", "s_store_name"],
    ).head(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['29'])

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

print(bench.get_query(29, 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', '29', family="expression").source)  # or family="pandas"

Source

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