TPC-DS Q75 — Three-Channel Sales with Returns Year-over-Year

TPC-DS benchmark · all TPC-DS queries

Year-over-year comparison of net sales (sales - returns) by item attributes

Representative SQL

Rendered for DataFusion with default parameters.

WITH all_sales AS (
  SELECT
    d_year,
    i_brand_id,
    i_class_id,
    i_category_id,
    i_manufact_id,
    SUM(sales_cnt) AS sales_cnt,
    SUM(sales_amt) AS sales_amt
  FROM (
    SELECT
      d_year,
      i_brand_id,
      i_class_id,
      i_category_id,
      i_manufact_id,
      cs_quantity - COALESCE(cr_return_quantity, 0) AS sales_cnt,
      cs_ext_sales_price - COALESCE(cr_return_amount, 0.0) AS sales_amt
    FROM catalog_sales
    JOIN item
      ON i_item_sk = cs_item_sk
    JOIN date_dim
      ON d_date_sk = cs_sold_date_sk
    LEFT JOIN catalog_returns
      ON (
        cs_order_number = cr_order_number AND cs_item_sk = cr_item_sk
      )
    WHERE
      i_category = 'Sports'
    UNION
    SELECT
      d_year,
      i_brand_id,
      i_class_id,
      i_category_id,
      i_manufact_id,
      ss_quantity - COALESCE(sr_return_quantity, 0) AS sales_cnt,
      ss_ext_sales_price - COALESCE(sr_return_amt, 0.0) AS sales_amt
    FROM store_sales
    JOIN item
      ON i_item_sk = ss_item_sk
    JOIN date_dim
      ON d_date_sk = ss_sold_date_sk
    LEFT JOIN store_returns
      ON (
        ss_ticket_number = sr_ticket_number AND ss_item_sk = sr_item_sk
      )
    WHERE
      i_category = 'Sports'
    UNION
    SELECT
      d_year,
      i_brand_id,
      i_class_id,
      i_category_id,
      i_manufact_id,
      ws_quantity - COALESCE(wr_return_quantity, 0) AS sales_cnt,
      ws_ext_sales_price - COALESCE(wr_return_amt, 0.0) AS sales_amt
    FROM web_sales
    JOIN item
      ON i_item_sk = ws_item_sk
    JOIN date_dim
      ON d_date_sk = ws_sold_date_sk
    LEFT JOIN web_returns
      ON (
        ws_order_number = wr_order_number AND ws_item_sk = wr_item_sk
      )
    WHERE
      i_category = 'Sports'
  ) AS sales_detail
  GROUP BY
    d_year,
    i_brand_id,
    i_class_id,
    i_category_id,
    i_manufact_id
)
SELECT
  prev_yr.d_year AS prev_year,
  curr_yr.d_year AS year,
  curr_yr.i_brand_id,
  curr_yr.i_class_id,
  curr_yr.i_category_id,
  curr_yr.i_manufact_id,
  prev_yr.sales_cnt AS prev_yr_cnt,
  curr_yr.sales_cnt AS curr_yr_cnt,
  curr_yr.sales_cnt - prev_yr.sales_cnt AS sales_cnt_diff,
  curr_yr.sales_amt - prev_yr.sales_amt AS sales_amt_diff
FROM all_sales AS curr_yr, all_sales AS prev_yr
WHERE
  curr_yr.i_brand_id = prev_yr.i_brand_id
  AND curr_yr.i_class_id = prev_yr.i_class_id
  AND curr_yr.i_category_id = prev_yr.i_category_id
  AND curr_yr.i_manufact_id = prev_yr.i_manufact_id
  AND curr_yr.d_year = 2002
  AND prev_yr.d_year = 2002 - 1
  AND CAST(curr_yr.sales_cnt AS DECIMAL(17, 2)) / CAST(prev_yr.sales_cnt AS DECIMAL(17, 2)) < 0.9
ORDER BY
  sales_cnt_diff,
  sales_amt_diff
LIMIT 100

Representative DataFrame

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

def q75_expression_impl(ctx: DataFrameContext) -> Any:
    params = get_parameters(75)
    year = params.get("year", 2001)
    category = params.get("category", "Books")

    item, date_dim = _tables(ctx, "item", "date_dim")
    col = ctx.col
    lit = ctx.lit

    item_filtered = item.filter(col("i_category") == lit(category))

    def channel(
        sales_table: str,
        returns_table: str,
        item_key: str,
        date_key: str,
        order_key: str,
        returns_order_key: str,
        returns_item_key: str,
        quantity_col: str,
        return_qty_col: str,
        amount_col: str,
        return_amount_col: str,
    ) -> Any:
        return (
            ctx.get_table(sales_table)
            .join(item_filtered, left_on=item_key, right_on="i_item_sk")
            .join(date_dim, left_on=date_key, right_on="d_date_sk")
            .join(
                ctx.get_table(returns_table),
                left_on=[order_key, item_key],
                right_on=[returns_order_key, returns_item_key],
                how="left",
            )
            .select(
                [
                    col("d_year"),
                    col("i_brand_id"),
                    col("i_class_id"),
                    col("i_category_id"),
                    col("i_manufact_id"),
                    (col(quantity_col) - col(return_qty_col).fill_null(0)).alias("sales_cnt"),
                    (
                        (col(amount_col) * lit(100)).round(0).cast_int64()
                        - (col(return_amount_col).fill_null(0.0) * lit(100)).round(0).cast_int64()
                    ).alias("sales_amt_cents"),
                ]
            )
        )

    # fmt: off
    channel_specs = (
        ("catalog_sales", "catalog_returns", "cs_item_sk", "cs_sold_date_sk", "cs_order_number", "cr_order_number", "cr_item_sk", "cs_quantity", "cr_return_quantity", "cs_ext_sales_price", "cr_return_amount"),
        ("store_sales", "store_returns", "ss_item_sk", "ss_sold_date_sk", "ss_ticket_number", "sr_ticket_number", "sr_item_sk", "ss_quantity", "sr_return_quantity", "ss_ext_sales_price", "sr_return_amt"),
        ("web_sales", "web_returns", "ws_item_sk", "ws_sold_date_sk", "ws_order_number", "wr_order_number", "wr_item_sk", "ws_quantity", "wr_return_quantity", "ws_ext_sales_price", "wr_return_amt"),
    )
    # fmt: on
    combined = ctx.concat([channel(*spec) for spec in channel_specs]).unique()

    all_sales = combined.group_by(["d_year", "i_brand_id", "i_class_id", "i_category_id", "i_manufact_id"]).agg(
        [
            _sum_or_null_expression(ctx, col("sales_cnt")).alias("sales_cnt"),
            _sum_or_null_expression(ctx, col("sales_amt_cents")).alias("sales_amt_cents"),
        ]
    )

    curr_yr = all_sales.filter(col("d_year") == lit(year))
    prev_yr = all_sales.filter(col("d_year") == lit(year - 1))

    return _sort_null_largest_expression(
        ctx,
        curr_yr.join(
            prev_yr,
            on=["i_brand_id", "i_class_id", "i_category_id", "i_manufact_id"],
            suffix="_prev",
        )
        .filter((col("sales_cnt").cast_float64() / col("sales_cnt_prev").cast_float64()) < lit(0.9))
        .select(
            [
                col("d_year_prev").alias("prev_year"),
                col("d_year").alias("year"),
                col("i_brand_id"),
                col("i_class_id"),
                col("i_category_id"),
                col("i_manufact_id"),
                col("sales_cnt_prev").alias("prev_yr_cnt"),
                col("sales_cnt").alias("curr_yr_cnt"),
                (col("sales_cnt") - col("sales_cnt_prev")).alias("sales_cnt_diff"),
                ((col("sales_amt_cents") - col("sales_amt_cents_prev")).cast_float64() / lit(100)).alias(
                    "sales_amt_diff"
                ),
            ]
        ),
        ["sales_cnt_diff", "sales_amt_diff"],
    ).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['75'])

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

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

Source

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