TPC-DS Q83 — Cross-Channel Returns Analysis

TPC-DS benchmark · all TPC-DS queries

Compares return quantities across store, catalog, and web channels

Representative SQL

Rendered for DataFusion with default parameters.

WITH sr_items AS (
  SELECT
    i_item_id AS item_id,
    SUM(sr_return_quantity) AS sr_item_qty
  FROM store_returns, item, date_dim
  WHERE
    sr_item_sk = i_item_sk
    AND d_date IN (
      SELECT
        d_date
      FROM date_dim
      WHERE
        d_week_seq IN (
          SELECT
            d_week_seq
          FROM date_dim
          WHERE
            d_date IN ('1998-01-02', '1998-10-15', '1998-11-10')
        )
    )
    AND sr_returned_date_sk = d_date_sk
  GROUP BY
    i_item_id
), cr_items AS (
  SELECT
    i_item_id AS item_id,
    SUM(cr_return_quantity) AS cr_item_qty
  FROM catalog_returns, item, date_dim
  WHERE
    cr_item_sk = i_item_sk
    AND d_date IN (
      SELECT
        d_date
      FROM date_dim
      WHERE
        d_week_seq IN (
          SELECT
            d_week_seq
          FROM date_dim
          WHERE
            d_date IN ('1998-01-02', '1998-10-15', '1998-11-10')
        )
    )
    AND cr_returned_date_sk = d_date_sk
  GROUP BY
    i_item_id
), wr_items AS (
  SELECT
    i_item_id AS item_id,
    SUM(wr_return_quantity) AS wr_item_qty
  FROM web_returns, item, date_dim
  WHERE
    wr_item_sk = i_item_sk
    AND d_date IN (
      SELECT
        d_date
      FROM date_dim
      WHERE
        d_week_seq IN (
          SELECT
            d_week_seq
          FROM date_dim
          WHERE
            d_date IN ('1998-01-02', '1998-10-15', '1998-11-10')
        )
    )
    AND wr_returned_date_sk = d_date_sk
  GROUP BY
    i_item_id
)
SELECT
  sr_items.item_id,
  sr_item_qty,
  sr_item_qty / (
    sr_item_qty + cr_item_qty + wr_item_qty
  ) / 3.0 * 100 AS sr_dev,
  cr_item_qty,
  cr_item_qty / (
    sr_item_qty + cr_item_qty + wr_item_qty
  ) / 3.0 * 100 AS cr_dev,
  wr_item_qty,
  wr_item_qty / (
    sr_item_qty + cr_item_qty + wr_item_qty
  ) / 3.0 * 100 AS wr_dev,
  (
    sr_item_qty + cr_item_qty + wr_item_qty
  ) / 3.0 AS average
FROM sr_items, cr_items, wr_items
WHERE
  sr_items.item_id = cr_items.item_id AND sr_items.item_id = wr_items.item_id
ORDER BY
  sr_items.item_id,
  sr_item_qty
LIMIT 100

Representative DataFrame

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

def q83_expression_impl(ctx: DataFrameContext) -> Any:
    params = get_parameters(83)
    return_dates = params.get("dates", ["1998-01-02", "1998-10-15", "1998-11-10"])

    col = ctx.col
    lit = ctx.lit

    store_returns, catalog_returns, web_returns, item, date_dim = _tables(
        ctx, "store_returns", "catalog_returns", "web_returns", "item", "date_dim"
    )

    lit = ctx.lit

    date_filter = lit(False)
    for dt in return_dates:
        date_filter = date_filter | (col("d_date").cast_string().str.starts_with(dt))

    target_weeks = date_dim.filter(date_filter).select(col("d_week_seq")).unique()

    valid_dates = date_dim.join(target_weeks, on="d_week_seq").select(col("d_date_sk")).unique()

    sr_items = (
        store_returns.join(item, left_on="sr_item_sk", right_on="i_item_sk")
        .join(valid_dates, left_on="sr_returned_date_sk", right_on="d_date_sk")
        .group_by(col("i_item_id").alias("item_id"))
        .agg(col("sr_return_quantity").sum().alias("sr_item_qty"))
    )

    cr_items = (
        catalog_returns.join(item, left_on="cr_item_sk", right_on="i_item_sk")
        .join(valid_dates, left_on="cr_returned_date_sk", right_on="d_date_sk")
        .group_by(col("i_item_id").alias("item_id"))
        .agg(col("cr_return_quantity").sum().alias("cr_item_qty"))
    )

    wr_items = (
        web_returns.join(item, left_on="wr_item_sk", right_on="i_item_sk")
        .join(valid_dates, left_on="wr_returned_date_sk", right_on="d_date_sk")
        .group_by(col("i_item_id").alias("item_id"))
        .agg(col("wr_return_quantity").sum().alias("wr_item_qty"))
    )

    return (
        sr_items.join(cr_items, on="item_id")
        .join(wr_items, on="item_id")
        .with_columns((col("sr_item_qty") + col("cr_item_qty") + col("wr_item_qty")).alias("total_qty"))
        .with_columns(
            ((col("sr_item_qty") / col("total_qty") / lit(3.0)) * lit(100.0)).alias("sr_dev"),
            ((col("cr_item_qty") / col("total_qty") / lit(3.0)) * lit(100.0)).alias("cr_dev"),
            ((col("wr_item_qty") / col("total_qty") / lit(3.0)) * lit(100.0)).alias("wr_dev"),
            (col("total_qty") / lit(3.0)).alias("average"),
        )
        .select(
            col("item_id"),
            col("sr_item_qty"),
            col("sr_dev"),
            col("cr_item_qty"),
            col("cr_dev"),
            col("wr_item_qty"),
            col("wr_dev"),
            col("average"),
        )
        .sort(["item_id", "sr_item_qty"])
        .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['83'])

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

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

Source

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