TPC-DS Q16 — Catalog Multi-Warehouse Orders Not Returned

TPC-DS benchmark · all TPC-DS queries

Catalog orders shipped from multiple warehouses not returned

Representative SQL

Rendered for DataFusion with default parameters.

SELECT
  COUNT(DISTINCT cs_order_number) AS "order count",
  SUM(cs_ext_ship_cost) AS "total shipping cost",
  SUM(cs_net_profit) AS "total net profit"
FROM catalog_sales AS cs1, date_dim, customer_address, call_center
WHERE
  d_date BETWEEN '1999-2-01' AND (
    CAST('1999-2-01' AS DATE) + INTERVAL '60 DAY'
  )
  AND cs1.cs_ship_date_sk = d_date_sk
  AND cs1.cs_ship_addr_sk = ca_address_sk
  AND ca_state = 'IL'
  AND cs1.cs_call_center_sk = cc_call_center_sk
  AND cc_county IN (
    'Williamson County',
    'Williamson County',
    'Williamson County',
    'Williamson County',
    'Williamson County'
  )
  AND EXISTS(
    SELECT
      *
    FROM catalog_sales AS cs2
    WHERE
      cs1.cs_order_number = cs2.cs_order_number
      AND cs1.cs_warehouse_sk <> cs2.cs_warehouse_sk
  )
  AND NOT EXISTS(
    SELECT
      *
    FROM catalog_returns AS cr1
    WHERE
      cs1.cs_order_number = cr1.cr_order_number
  )
ORDER BY
  COUNT(DISTINCT cs_order_number)
LIMIT 100

Representative DataFrame

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

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

    params = get_parameters(16)
    year = params.get("year", 1999)
    month = params.get("month", 2)
    state = params.get("state", "IL")
    counties = list(params.get("counties", ["Williamson County"] * 5))

    catalog_sales, catalog_returns, date_dim, customer_address, call_center = _tables(
        ctx, "catalog_sales", "catalog_returns", "date_dim", "customer_address", "call_center"
    )
    col = ctx.col
    lit = ctx.lit

    start_date = datetime(year, month, 1).date()
    end_date = start_date + timedelta(days=60)

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

    ca_filtered = customer_address.filter(col("ca_state") == lit(state))

    cc_filtered = call_center.filter(col("cc_county").is_in(counties))

    returned_orders = catalog_returns.select(["cr_order_number"]).unique()

    shipped = catalog_sales.filter(col("cs_warehouse_sk").is_not_null())
    platform = getattr(ctx, "platform", "polars")
    if platform == "datafusion":
        order_warehouses = shipped.select(["cs_order_number", "cs_warehouse_sk"]).unique()
        multi_warehouse_orders = (
            order_warehouses.join(
                order_warehouses.rename({"cs_warehouse_sk": "cs_warehouse_sk_2"}),
                on="cs_order_number",
            )
            .filter(col("cs_warehouse_sk") != col("cs_warehouse_sk_2"))
            .select(["cs_order_number"])
            .unique()
        )
    else:
        multi_warehouse_orders = (
            shipped.group_by("cs_order_number")
            .agg(col("cs_warehouse_sk").n_unique().alias("num_warehouses"))
            .filter(col("num_warehouses") > lit(1))
            .select("cs_order_number")
        )

    result = (
        shipped.join(date_filtered, left_on="cs_ship_date_sk", right_on="d_date_sk")
        .join(ca_filtered, left_on="cs_ship_addr_sk", right_on="ca_address_sk")
        .join(cc_filtered, left_on="cs_call_center_sk", right_on="cc_call_center_sk")
        .join(multi_warehouse_orders, on="cs_order_number", how="semi")
        .join(returned_orders, left_on="cs_order_number", right_on="cr_order_number", how="anti")
    )

    tallied = result.select(
        [
            col("cs_order_number").n_unique().alias("order count"),
            ctx.sum("cs_ext_ship_cost").alias("ship_cost"),
            ctx.sum("cs_net_profit").alias("net_profit"),
            col("cs_order_number").count().alias("n"),
        ]
    )
    return tallied.select(
        [
            col("order count"),
            ctx.when(col("n") > lit(0)).then(col("ship_cost")).otherwise(lit(None)).alias("total shipping cost"),
            ctx.when(col("n") > lit(0)).then(col("net_profit")).otherwise(lit(None)).alias("total net profit"),
        ]
    )

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

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

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

Source

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