TPC-DS Q78 — Three-Channel Sales without Returns Comparison

TPC-DS benchmark · all TPC-DS queries

Compares store sales to combined web+catalog sales excluding returned items

Representative SQL

Rendered for DataFusion with default parameters.

WITH ws AS (
  SELECT
    d_year AS ws_sold_year,
    ws_item_sk,
    ws_bill_customer_sk AS ws_customer_sk,
    SUM(ws_quantity) AS ws_qty,
    SUM(ws_wholesale_cost) AS ws_wc,
    SUM(ws_sales_price) AS ws_sp
  FROM web_sales
  LEFT JOIN web_returns
    ON wr_order_number = ws_order_number AND ws_item_sk = wr_item_sk
  JOIN date_dim
    ON ws_sold_date_sk = d_date_sk
  WHERE
    wr_order_number IS NULL
  GROUP BY
    d_year,
    ws_item_sk,
    ws_bill_customer_sk
), cs AS (
  SELECT
    d_year AS cs_sold_year,
    cs_item_sk,
    cs_bill_customer_sk AS cs_customer_sk,
    SUM(cs_quantity) AS cs_qty,
    SUM(cs_wholesale_cost) AS cs_wc,
    SUM(cs_sales_price) AS cs_sp
  FROM catalog_sales
  LEFT JOIN catalog_returns
    ON cr_order_number = cs_order_number AND cs_item_sk = cr_item_sk
  JOIN date_dim
    ON cs_sold_date_sk = d_date_sk
  WHERE
    cr_order_number IS NULL
  GROUP BY
    d_year,
    cs_item_sk,
    cs_bill_customer_sk
), ss AS (
  SELECT
    d_year AS ss_sold_year,
    ss_item_sk,
    ss_customer_sk,
    SUM(ss_quantity) AS ss_qty,
    SUM(ss_wholesale_cost) AS ss_wc,
    SUM(ss_sales_price) AS ss_sp
  FROM store_sales
  LEFT JOIN store_returns
    ON sr_ticket_number = ss_ticket_number AND ss_item_sk = sr_item_sk
  JOIN date_dim
    ON ss_sold_date_sk = d_date_sk
  WHERE
    sr_ticket_number IS NULL
  GROUP BY
    d_year,
    ss_item_sk,
    ss_customer_sk
)
SELECT
  ss_sold_year,
  ss_item_sk,
  ss_customer_sk,
  ROUND(ss_qty / (
    COALESCE(ws_qty, 0) + COALESCE(cs_qty, 0)
  ), 2) AS ratio,
  ss_qty AS store_qty,
  ss_wc AS store_wholesale_cost,
  ss_sp AS store_sales_price,
  COALESCE(ws_qty, 0) + COALESCE(cs_qty, 0) AS other_chan_qty,
  COALESCE(ws_wc, 0) + COALESCE(cs_wc, 0) AS other_chan_wholesale_cost,
  COALESCE(ws_sp, 0) + COALESCE(cs_sp, 0) AS other_chan_sales_price
FROM ss
LEFT JOIN ws
  ON (
    ws_sold_year = ss_sold_year
    AND ws_item_sk = ss_item_sk
    AND ws_customer_sk = ss_customer_sk
  )
LEFT JOIN cs
  ON (
    cs_sold_year = ss_sold_year
    AND cs_item_sk = ss_item_sk
    AND cs_customer_sk = ss_customer_sk
  )
WHERE
  (
    COALESCE(ws_qty, 0) > 0 OR COALESCE(cs_qty, 0) > 0
  ) AND ss_sold_year = 2000
ORDER BY
  ss_sold_year,
  ss_item_sk,
  ss_customer_sk,
  ss_qty DESC,
  ss_wc DESC,
  ss_sp DESC,
  other_chan_qty,
  other_chan_wholesale_cost,
  other_chan_sales_price,
  ratio
LIMIT 100

Representative DataFrame

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

def q78_expression_impl(ctx: DataFrameContext) -> Any:
    params = get_parameters(78)
    year = params.get("year", 2000)
    select_columns = _q78_select_columns(params)

    date_dim = ctx.get_table("date_dim")
    col = ctx.col
    lit = ctx.lit

    def channel(
        sales_table: str,
        returns_table: str,
        left_on: list[str],
        right_on: list[str],
        date_key: str,
        return_null_col: str,
        group_cols: list[str],
        quantity_col: str,
        wholesale_col: str,
        sales_price_col: str,
        aliases: dict[str, str],
    ) -> Any:
        return (
            ctx.get_table(sales_table)
            .join(
                ctx.get_table(returns_table).with_columns(col(return_null_col).alias("_returned")),
                left_on=left_on,
                right_on=right_on,
                how="left",
            )
            .join(date_dim, left_on=date_key, right_on="d_date_sk")
            .filter(col("_returned").is_null())
            .group_by(group_cols)
            .agg(
                [
                    ctx.when(col(quantity_col).count() > lit(0))
                    .then(col(quantity_col).sum())
                    .otherwise(lit(None))
                    .alias(aliases["qty"]),
                    ctx.when(col(wholesale_col).count() > lit(0))
                    .then(col(wholesale_col).sum())
                    .otherwise(lit(None))
                    .alias(aliases["wc"]),
                    ctx.when(col(sales_price_col).count() > lit(0))
                    .then(col(sales_price_col).sum())
                    .otherwise(lit(None))
                    .alias(aliases["sp"]),
                ]
            )
            .rename({key: value for key, value in aliases.items() if key not in {"qty", "wc", "sp"}})
        )

    ss = channel(
        "store_sales",
        "store_returns",
        ["ss_ticket_number", "ss_item_sk"],
        ["sr_ticket_number", "sr_item_sk"],
        "ss_sold_date_sk",
        "sr_ticket_number",
        ["d_year", "ss_item_sk", "ss_customer_sk"],
        "ss_quantity",
        "ss_wholesale_cost",
        "ss_sales_price",
        {"d_year": "ss_sold_year", "qty": "ss_qty", "wc": "ss_wc", "sp": "ss_sp"},
    )
    cs = channel(
        "catalog_sales",
        "catalog_returns",
        ["cs_order_number", "cs_item_sk"],
        ["cr_order_number", "cr_item_sk"],
        "cs_sold_date_sk",
        "cr_order_number",
        ["d_year", "cs_item_sk", "cs_bill_customer_sk"],
        "cs_quantity",
        "cs_wholesale_cost",
        "cs_sales_price",
        {
            "d_year": "cs_sold_year",
            "cs_bill_customer_sk": "cs_customer_sk",
            "qty": "cs_qty",
            "wc": "cs_wc",
            "sp": "cs_sp",
        },
    )
    ws = channel(
        "web_sales",
        "web_returns",
        ["ws_order_number", "ws_item_sk"],
        ["wr_order_number", "wr_item_sk"],
        "ws_sold_date_sk",
        "wr_order_number",
        ["d_year", "ws_item_sk", "ws_bill_customer_sk"],
        "ws_quantity",
        "ws_wholesale_cost",
        "ws_sales_price",
        {
            "d_year": "ws_sold_year",
            "ws_bill_customer_sk": "ws_customer_sk",
            "qty": "ws_qty",
            "wc": "ws_wc",
            "sp": "ws_sp",
        },
    )

    selected = (
        ss.join(
            ws,
            left_on=["ss_sold_year", "ss_item_sk", "ss_customer_sk"],
            right_on=["ws_sold_year", "ws_item_sk", "ws_customer_sk"],
            how="left",
        )
        .join(
            cs,
            left_on=["ss_sold_year", "ss_item_sk", "ss_customer_sk"],
            right_on=["cs_sold_year", "cs_item_sk", "cs_customer_sk"],
            how="left",
        )
        .filter(
            (col("ss_sold_year") == lit(year))
            & ((col("ws_qty").fill_null(0) > lit(0)) | (col("cs_qty").fill_null(0) > lit(0)))
        )
        .with_columns(
            [
                _round_half_up_expression(
                    col("ss_qty").cast_float64()
                    / (col("ws_qty").fill_null(0) + col("cs_qty").fill_null(0)).cast_float64(),
                    2,
                ).alias("ratio"),
                (col("ws_qty").fill_null(0) + col("cs_qty").fill_null(0)).alias("other_chan_qty"),
                (col("ws_wc").fill_null(0) + col("cs_wc").fill_null(0)).alias("other_chan_wholesale_cost"),
                (col("ws_sp").fill_null(0) + col("cs_sp").fill_null(0)).alias("other_chan_sales_price"),
            ]
        )
        .select(
            [
                *select_columns,
                "ratio",
                col("ss_qty").alias("store_qty"),
                col("ss_wc").alias("store_wholesale_cost"),
                col("ss_sp").alias("store_sales_price"),
                "other_chan_qty",
                "other_chan_wholesale_cost",
                "other_chan_sales_price",
            ]
        )
    )
    order_by, descending = _q78_order(select_columns)
    return _sort_null_largest_expression(ctx, selected, order_by, descending).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['78'])

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

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

Source

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