TPC-DS Q5 — Three-Channel Sales-Returns Union with ROLLUP

TPC-DS benchmark · all TPC-DS queries

Unions sales and returns per channel then aggregates with ROLLUP

Representative SQL

Rendered for DataFusion with default parameters.

WITH ssr AS (
  SELECT
    s_store_id,
    SUM(sales_price) AS sales,
    SUM(profit) AS profit,
    SUM(return_amt) AS returns,
    SUM(net_loss) AS profit_loss
  FROM (
    SELECT
      ss_store_sk AS store_sk,
      ss_sold_date_sk AS date_sk,
      ss_ext_sales_price AS sales_price,
      ss_net_profit AS profit,
      CAST(0 AS DECIMAL(7, 2)) AS return_amt,
      CAST(0 AS DECIMAL(7, 2)) AS net_loss
    FROM store_sales
    UNION ALL
    SELECT
      sr_store_sk AS store_sk,
      sr_returned_date_sk AS date_sk,
      CAST(0 AS DECIMAL(7, 2)) AS sales_price,
      CAST(0 AS DECIMAL(7, 2)) AS profit,
      sr_return_amt AS return_amt,
      sr_net_loss AS net_loss
    FROM store_returns
  ) AS salesreturns, date_dim, store
  WHERE
    date_sk = d_date_sk
    AND d_date BETWEEN CAST('1998-08-04' AS DATE) AND (
      CAST('1998-08-04' AS DATE) + INTERVAL '14 DAY'
    )
    AND store_sk = s_store_sk
  GROUP BY
    s_store_id
), csr AS (
  SELECT
    cp_catalog_page_id,
    SUM(sales_price) AS sales,
    SUM(profit) AS profit,
    SUM(return_amt) AS returns,
    SUM(net_loss) AS profit_loss
  FROM (
    SELECT
      cs_catalog_page_sk AS page_sk,
      cs_sold_date_sk AS date_sk,
      cs_ext_sales_price AS sales_price,
      cs_net_profit AS profit,
      CAST(0 AS DECIMAL(7, 2)) AS return_amt,
      CAST(0 AS DECIMAL(7, 2)) AS net_loss
    FROM catalog_sales
    UNION ALL
    SELECT
      cr_catalog_page_sk AS page_sk,
      cr_returned_date_sk AS date_sk,
      CAST(0 AS DECIMAL(7, 2)) AS sales_price,
      CAST(0 AS DECIMAL(7, 2)) AS profit,
      cr_return_amount AS return_amt,
      cr_net_loss AS net_loss
    FROM catalog_returns
  ) AS salesreturns, date_dim, catalog_page
  WHERE
    date_sk = d_date_sk
    AND d_date BETWEEN CAST('1998-08-04' AS DATE) AND (
      CAST('1998-08-04' AS DATE) + INTERVAL '14 DAY'
    )
    AND page_sk = cp_catalog_page_sk
  GROUP BY
    cp_catalog_page_id
), wsr AS (
  SELECT
    web_site_id,
    SUM(sales_price) AS sales,
    SUM(profit) AS profit,
    SUM(return_amt) AS returns,
    SUM(net_loss) AS profit_loss
  FROM (
    SELECT
      ws_web_site_sk AS wsr_web_site_sk,
      ws_sold_date_sk AS date_sk,
      ws_ext_sales_price AS sales_price,
      ws_net_profit AS profit,
      CAST(0 AS DECIMAL(7, 2)) AS return_amt,
      CAST(0 AS DECIMAL(7, 2)) AS net_loss
    FROM web_sales
    UNION ALL
    SELECT
      ws_web_site_sk AS wsr_web_site_sk,
      wr_returned_date_sk AS date_sk,
      CAST(0 AS DECIMAL(7, 2)) AS sales_price,
      CAST(0 AS DECIMAL(7, 2)) AS profit,
      wr_return_amt AS return_amt,
      wr_net_loss AS net_loss
    FROM web_returns
    LEFT OUTER JOIN web_sales
      ON (
        wr_item_sk = ws_item_sk AND wr_order_number = ws_order_number
      )
  ) AS salesreturns, date_dim, web_site
  WHERE
    date_sk = d_date_sk
    AND d_date BETWEEN CAST('1998-08-04' AS DATE) AND (
      CAST('1998-08-04' AS DATE) + INTERVAL '14 DAY'
    )
    AND wsr_web_site_sk = web_site_sk
  GROUP BY
    web_site_id
)
SELECT
  channel,
  id,
  SUM(sales) AS sales,
  SUM(returns) AS returns,
  SUM(profit) AS profit
FROM (
  SELECT
    'store channel' AS channel,
    'store' || s_store_id AS id,
    sales,
    returns,
    (
      profit - profit_loss
    ) AS profit
  FROM ssr
  UNION ALL
  SELECT
    'catalog channel' AS channel,
    'catalog_page' || cp_catalog_page_id AS id,
    sales,
    returns,
    (
      profit - profit_loss
    ) AS profit
  FROM csr
  UNION ALL
  SELECT
    'web channel' AS channel,
    'web_site' || web_site_id AS id,
    sales,
    returns,
    (
      profit - profit_loss
    ) AS profit
  FROM wsr
) AS x
GROUP BY
  ROLLUP (
    channel,
    id
  )
ORDER BY
  channel,
  id
LIMIT 100

Representative DataFrame

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

def q5_expression_impl(ctx: DataFrameContext) -> Any:
    col = ctx.col
    lit = ctx.lit
    date_filtered = _date_window_expression(ctx, 5, "1998-08-04", days=14)

    def channel(
        channel_name: str,
        id_prefix: str,
        dimension_table: str,
        dimension_key: str,
        dimension_id: str,
        sales_spec: tuple[str, str, str, str, str],
        returns_spec: tuple[str, str, str, str, str],
        returns_join: tuple[str, tuple[str, str], tuple[str, str]] | None = None,
    ) -> Any:
        sales_table, sales_dim_col, sales_date_col, sales_price_col, sales_profit_col = sales_spec
        returns_table, returns_dim_col, returns_date_col, returns_amt_col, returns_loss_col = returns_spec
        sales_part = ctx.get_table(sales_table).select(
            [
                col(sales_dim_col).alias("dim_sk"),
                col(sales_date_col).alias("date_sk"),
                col(sales_price_col).alias("sales_price"),
                col(sales_profit_col).alias("profit"),
                lit(0.0).alias("return_amt"),
                lit(0.0).alias("net_loss"),
            ]
        )
        returns = ctx.get_table(returns_table)
        if returns_join is not None:
            join_table, left_on, right_on = returns_join
            returns = returns.join(
                ctx.get_table(join_table), left_on=list(left_on), right_on=list(right_on), how="left"
            )
        returns_part = returns.select(
            [
                col(returns_dim_col).alias("dim_sk"),
                col(returns_date_col).alias("date_sk"),
                lit(0.0).alias("sales_price"),
                lit(0.0).alias("profit"),
                col(returns_amt_col).alias("return_amt"),
                col(returns_loss_col).alias("net_loss"),
            ]
        )
        return (
            ctx.concat([sales_part, returns_part])
            .join(date_filtered, left_on="date_sk", right_on="d_date_sk")
            .join(ctx.get_table(dimension_table), left_on="dim_sk", right_on=dimension_key)
            .group_by(dimension_id)
            .agg(
                [
                    col("sales_price").sum().alias("sales"),
                    col("profit").sum().alias("profit_sum"),
                    col("return_amt").sum().alias("returns"),
                    col("net_loss").sum().alias("profit_loss"),
                    col("sales_price").count().alias("sales_n"),
                    col("profit").count().alias("profit_sum_n"),
                    col("return_amt").count().alias("returns_n"),
                    col("net_loss").count().alias("profit_loss_n"),
                ]
            )
            .with_columns(
                [
                    ctx.when(col(f"{name}_n") > lit(0)).then(col(name)).otherwise(lit(None)).alias(name)
                    for name in ("sales", "profit_sum", "returns", "profit_loss")
                ]
            )
            .with_columns(
                [
                    lit(channel_name).alias("channel"),
                    (lit(id_prefix) + col(dimension_id)).alias("id"),
                    (col("profit_sum") - col("profit_loss")).alias("profit"),
                ]
            )
            .select(["channel", "id", "sales", "returns", "profit"])
        )

    # fmt: off
    specs = (
        ("store channel", "store", "store", "s_store_sk", "s_store_id", ("store_sales", "ss_store_sk", "ss_sold_date_sk", "ss_ext_sales_price", "ss_net_profit"), ("store_returns", "sr_store_sk", "sr_returned_date_sk", "sr_return_amt", "sr_net_loss")),
        ("catalog channel", "catalog_page", "catalog_page", "cp_catalog_page_sk", "cp_catalog_page_id", ("catalog_sales", "cs_catalog_page_sk", "cs_sold_date_sk", "cs_ext_sales_price", "cs_net_profit"), ("catalog_returns", "cr_catalog_page_sk", "cr_returned_date_sk", "cr_return_amount", "cr_net_loss")),
        ("web channel", "web_site", "web_site", "web_site_sk", "web_site_id", ("web_sales", "ws_web_site_sk", "ws_sold_date_sk", "ws_ext_sales_price", "ws_net_profit"), ("web_returns", "ws_web_site_sk", "wr_returned_date_sk", "wr_return_amt", "wr_net_loss"), ("web_sales", ("wr_item_sk", "wr_order_number"), ("ws_item_sk", "ws_order_number"))),
    )
    # fmt: on
    combined = ctx.concat([channel(*spec) for spec in specs])
    return _sales_returns_rollup_expression(ctx, combined)

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['5'])

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

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

Source

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