TPC-DS Q40 — Catalog Sales Before/After Returns

TPC-DS benchmark · all TPC-DS queries

Compares catalog sales before/after date accounting for returns

Representative SQL

Rendered for DataFusion with default parameters.

SELECT
  w_state,
  i_item_id,
  SUM(
    CASE
      WHEN (
        CAST(d_date AS DATE) < CAST('1998-04-08' AS DATE)
      )
      THEN cs_sales_price - COALESCE(cr_refunded_cash, 0)
      ELSE 0
    END
  ) AS sales_before,
  SUM(
    CASE
      WHEN (
        CAST(d_date AS DATE) >= CAST('1998-04-08' AS DATE)
      )
      THEN cs_sales_price - COALESCE(cr_refunded_cash, 0)
      ELSE 0
    END
  ) AS sales_after
FROM catalog_sales
LEFT OUTER JOIN catalog_returns
  ON (
    cs_order_number = cr_order_number AND cs_item_sk = cr_item_sk
  ), warehouse, item, date_dim
WHERE
  i_current_price BETWEEN 0.99 AND 1.49
  AND i_item_sk = cs_item_sk
  AND cs_warehouse_sk = w_warehouse_sk
  AND cs_sold_date_sk = d_date_sk
  AND d_date BETWEEN (
    CAST('1998-04-08' AS DATE) - INTERVAL '30 DAY'
  ) AND (
    CAST('1998-04-08' AS DATE) + INTERVAL '30 DAY'
  )
GROUP BY
  w_state,
  i_item_id
ORDER BY
  w_state,
  i_item_id
LIMIT 100

Representative DataFrame

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

def q40_expression_impl(ctx: DataFrameContext) -> Any:
    from datetime import datetime, timedelta

    params = get_parameters(40)
    sales_date = datetime.strptime(params.get("sales_date", "1998-04-08"), "%Y-%m-%d").date()

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

    start_date = sales_date - timedelta(days=30)
    end_date = sales_date + timedelta(days=30)

    date_filtered = date_dim.filter((col("d_date") >= lit(start_date)) & (col("d_date") <= lit(end_date)))

    item_filtered = item.filter((col("i_current_price") >= lit(0.99)) & (col("i_current_price") <= lit(1.49)))

    cs_with_cr = catalog_sales.join(
        catalog_returns,
        left_on=["cs_order_number", "cs_item_sk"],
        right_on=["cr_order_number", "cr_item_sk"],
        how="left",
    )

    result = (
        cs_with_cr.join(date_filtered, left_on="cs_sold_date_sk", right_on="d_date_sk")
        .join(warehouse, left_on="cs_warehouse_sk", right_on="w_warehouse_sk")
        .join(item_filtered, left_on="cs_item_sk", right_on="i_item_sk")
    )

    result = result.with_columns(
        [
            ctx.when(col("d_date") < lit(sales_date))
            .then(col("cs_sales_price") - col("cr_refunded_cash").fill_null(0))
            .otherwise(lit(0))
            .alias("sales_before_val"),
            ctx.when(col("d_date") >= lit(sales_date))
            .then(col("cs_sales_price") - col("cr_refunded_cash").fill_null(0))
            .otherwise(lit(0))
            .alias("sales_after_val"),
        ]
    )

    return _sort_null_largest_expression(
        ctx,
        result.group_by(["w_state", "i_item_id"]).agg(
            [
                ctx.sum("sales_before_val").alias("sales_before"),
                ctx.sum("sales_after_val").alias("sales_after"),
            ]
        ),
        ["w_state", "i_item_id"],
    ).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['40'])

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

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

Source

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