TPC-DS Q51 — Cumulative Web/Store Sales

TPC-DS benchmark · all TPC-DS queries

Compares cumulative web and store sales where web exceeds store

Representative SQL

Rendered for DataFusion with default parameters.

WITH web_v1 AS (
  SELECT
    ws_item_sk AS item_sk,
    d_date,
    SUM(SUM(ws_sales_price)) OVER (
      PARTITION BY ws_item_sk
      ORDER BY d_date
      rows BETWEEN UNBOUNDED preceding AND CURRENT ROW
    ) AS cume_sales
  FROM web_sales, date_dim
  WHERE
    ws_sold_date_sk = d_date_sk
    AND d_month_seq BETWEEN 1212 AND 1212 + 11
    AND ws_item_sk IS NOT NULL
  GROUP BY
    ws_item_sk,
    d_date
), store_v1 AS (
  SELECT
    ss_item_sk AS item_sk,
    d_date,
    SUM(SUM(ss_sales_price)) OVER (
      PARTITION BY ss_item_sk
      ORDER BY d_date
      rows BETWEEN UNBOUNDED preceding AND CURRENT ROW
    ) AS cume_sales
  FROM store_sales, date_dim
  WHERE
    ss_sold_date_sk = d_date_sk
    AND d_month_seq BETWEEN 1212 AND 1212 + 11
    AND ss_item_sk IS NOT NULL
  GROUP BY
    ss_item_sk,
    d_date
)
SELECT
  *
FROM (
  SELECT
    item_sk,
    d_date,
    web_sales,
    store_sales,
    MAX(web_sales) OVER (
      PARTITION BY item_sk
      ORDER BY d_date
      rows BETWEEN UNBOUNDED preceding AND CURRENT ROW
    ) AS web_cumulative,
    MAX(store_sales) OVER (
      PARTITION BY item_sk
      ORDER BY d_date
      rows BETWEEN UNBOUNDED preceding AND CURRENT ROW
    ) AS store_cumulative
  FROM (
    SELECT
      CASE WHEN web.item_sk IS NOT NULL THEN web.item_sk ELSE store.item_sk END AS item_sk,
      CASE WHEN web.d_date IS NOT NULL THEN web.d_date ELSE store.d_date END AS d_date,
      web.cume_sales AS web_sales,
      store.cume_sales AS store_sales
    FROM web_v1 AS web
    FULL OUTER JOIN store_v1 AS store
      ON (
        web.item_sk = store.item_sk AND web.d_date = store.d_date
      )
  ) AS x
) AS y
WHERE
  web_cumulative > store_cumulative
ORDER BY
  item_sk,
  d_date
LIMIT 100

Representative DataFrame

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

def q51_expression_impl(ctx: DataFrameContext) -> Any:
    params = get_parameters(51)
    dms = params.get("dms", 1212)

    col = ctx.col
    lit = ctx.lit

    web_sales, store_sales, date_dim = _tables(ctx, "web_sales", "store_sales", "date_dim")

    dates = date_dim.filter(col("d_month_seq").is_between(dms, dms + 11))

    web_base = (
        web_sales.filter(col("ws_item_sk").is_not_null())
        .join(dates, left_on="ws_sold_date_sk", right_on="d_date_sk")
        .group_by(["ws_item_sk", "d_date"])
        .agg(
            col("ws_sales_price").count().alias("daily_count"),
            col("ws_sales_price").sum().alias("daily_sales"),
        )
        .with_columns(
            ctx.when(col("daily_count") > lit(0)).then(col("daily_sales")).otherwise(lit(None)).alias("daily_sales")
        )
    )

    web_v1 = (
        web_base.sort(["ws_item_sk", "d_date"])
        .with_columns(_q51_running_total(ctx, "ws_item_sk").alias("cume_sales"))
        .select(
            col("ws_item_sk").alias("item_sk"),
            col("d_date"),
            col("cume_sales"),
        )
    )

    store_base = (
        store_sales.filter(col("ss_item_sk").is_not_null())
        .join(dates, left_on="ss_sold_date_sk", right_on="d_date_sk")
        .group_by(["ss_item_sk", "d_date"])
        .agg(
            col("ss_sales_price").count().alias("daily_count"),
            col("ss_sales_price").sum().alias("daily_sales"),
        )
        .with_columns(
            ctx.when(col("daily_count") > lit(0)).then(col("daily_sales")).otherwise(lit(None)).alias("daily_sales")
        )
    )

    store_v1 = (
        store_base.sort(["ss_item_sk", "d_date"])
        .with_columns(_q51_running_total(ctx, "ss_item_sk").alias("cume_sales"))
        .select(
            col("ss_item_sk").alias("item_sk"),
            col("d_date"),
            col("cume_sales"),
        )
    )

    joined = _q51_merge_channels(ctx, web_v1, store_v1).sort(["item_sk_final", "d_date_final"])

    web_cumulative = (
        col("web_sales").fill_null(float("-inf")).cum_max().over(["item_sk_final"], order_by="d_date_final")
    )
    store_cumulative = (
        col("store_sales").fill_null(float("-inf")).cum_max().over(["item_sk_final"], order_by="d_date_final")
    )
    result = joined.with_columns(
        ctx.when(web_cumulative == lit(float("-inf"))).then(None).otherwise(web_cumulative).alias("web_cumulative"),
        ctx.when(store_cumulative == lit(float("-inf")))
        .then(None)
        .otherwise(store_cumulative)
        .alias("store_cumulative"),
    )

    return (
        result.filter(col("web_cumulative") > col("store_cumulative"))
        .select(
            col("item_sk_final").alias("item_sk"),
            col("d_date_final").alias("d_date"),
            col("web_sales"),
            col("store_sales"),
            col("web_cumulative"),
            col("store_cumulative"),
        )
        .sort(["item_sk", "d_date"])
        .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['51'])

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

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

Source

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