TPC-DS Q14 — Cross-channel Item Sales

TPC-DS benchmark · all TPC-DS queries

Finds items sold across all channels with ROLLUP aggregation

Representative SQL

Rendered for DataFusion with default parameters.

WITH cross_items AS (
  SELECT
    i_item_sk AS ss_item_sk
  FROM item, (
    SELECT
      iss.i_brand_id AS brand_id,
      iss.i_class_id AS class_id,
      iss.i_category_id AS category_id
    FROM store_sales, item AS iss, date_dim AS d1
    WHERE
      ss_item_sk = iss.i_item_sk
      AND ss_sold_date_sk = d1.d_date_sk
      AND d1.d_year BETWEEN 1998 AND 1998 + 2
    INTERSECT
    SELECT
      ics.i_brand_id,
      ics.i_class_id,
      ics.i_category_id
    FROM catalog_sales, item AS ics, date_dim AS d2
    WHERE
      cs_item_sk = ics.i_item_sk
      AND cs_sold_date_sk = d2.d_date_sk
      AND d2.d_year BETWEEN 1998 AND 1998 + 2
    INTERSECT
    SELECT
      iws.i_brand_id,
      iws.i_class_id,
      iws.i_category_id
    FROM web_sales, item AS iws, date_dim AS d3
    WHERE
      ws_item_sk = iws.i_item_sk
      AND ws_sold_date_sk = d3.d_date_sk
      AND d3.d_year BETWEEN 1998 AND 1998 + 2
  )
  WHERE
    i_brand_id = brand_id AND i_class_id = class_id AND i_category_id = category_id
), avg_sales AS (
  SELECT
    AVG(quantity * list_price) AS average_sales
  FROM (
    SELECT
      ss_quantity AS quantity,
      ss_list_price AS list_price
    FROM store_sales, date_dim
    WHERE
      ss_sold_date_sk = d_date_sk AND d_year BETWEEN 1998 AND 1998 + 2
    UNION ALL
    SELECT
      cs_quantity AS quantity,
      cs_list_price AS list_price
    FROM catalog_sales, date_dim
    WHERE
      cs_sold_date_sk = d_date_sk AND d_year BETWEEN 1998 AND 1998 + 2
    UNION ALL
    SELECT
      ws_quantity AS quantity,
      ws_list_price AS list_price
    FROM web_sales, date_dim
    WHERE
      ws_sold_date_sk = d_date_sk AND d_year BETWEEN 1998 AND 1998 + 2
  ) AS x
)
SELECT
  channel,
  i_brand_id,
  i_class_id,
  i_category_id,
  SUM(sales),
  SUM(number_sales)
FROM (
  SELECT
    'store' AS channel,
    i_brand_id,
    i_class_id,
    i_category_id,
    SUM(ss_quantity * ss_list_price) AS sales,
    COUNT(*) AS number_sales
  FROM store_sales, item, date_dim
  WHERE
    ss_item_sk IN (
      SELECT
        ss_item_sk
      FROM cross_items
    )
    AND ss_item_sk = i_item_sk
    AND ss_sold_date_sk = d_date_sk
    AND d_year = 1998 + 2
    AND d_moy = 11
  GROUP BY
    i_brand_id,
    i_class_id,
    i_category_id
  HAVING
    SUM(ss_quantity * ss_list_price) > (
      SELECT
        average_sales
      FROM avg_sales
    )
  UNION ALL
  SELECT
    'catalog' AS channel,
    i_brand_id,
    i_class_id,
    i_category_id,
    SUM(cs_quantity * cs_list_price) AS sales,
    COUNT(*) AS number_sales
  FROM catalog_sales, item, date_dim
  WHERE
    cs_item_sk IN (
      SELECT
        ss_item_sk
      FROM cross_items
    )
    AND cs_item_sk = i_item_sk
    AND cs_sold_date_sk = d_date_sk
    AND d_year = 1998 + 2
    AND d_moy = 11
  GROUP BY
    i_brand_id,
    i_class_id,
    i_category_id
  HAVING
    SUM(cs_quantity * cs_list_price) > (
      SELECT
        average_sales
      FROM avg_sales
    )
  UNION ALL
  SELECT
    'web' AS channel,
    i_brand_id,
    i_class_id,
    i_category_id,
    SUM(ws_quantity * ws_list_price) AS sales,
    COUNT(*) AS number_sales
  FROM web_sales, item, date_dim
  WHERE
    ws_item_sk IN (
      SELECT
        ss_item_sk
      FROM cross_items
    )
    AND ws_item_sk = i_item_sk
    AND ws_sold_date_sk = d_date_sk
    AND d_year = 1998 + 2
    AND d_moy = 11
  GROUP BY
    i_brand_id,
    i_class_id,
    i_category_id
  HAVING
    SUM(ws_quantity * ws_list_price) > (
      SELECT
        average_sales
      FROM avg_sales
    )
) AS y
GROUP BY
  ROLLUP (
    channel,
    i_brand_id,
    i_class_id,
    i_category_id
  )
ORDER BY
  channel,
  i_brand_id,
  i_class_id,
  i_category_id
LIMIT 100

Representative DataFrame

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

def q14_expression_impl(ctx: DataFrameContext) -> Any:
    from .rollup_helper import expand_rollup_expression

    params = get_parameters(14)
    year = params.get("year", 1998)

    store_sales, catalog_sales, web_sales, item, date_dim = _tables(
        ctx, "store_sales", "catalog_sales", "web_sales", "item", "date_dim"
    )

    col = ctx.col
    lit = ctx.lit

    cross_item_sks, avg_sales_val = _q14_cross_items_expression(
        ctx, year, store_sales, catalog_sales, web_sales, item, date_dim
    )

    date_target = date_dim.filter((col("d_year") == lit(year + 2)) & (col("d_moy") == lit(11)))

    def channel_sales(sales_df, date_col, item_col, qty_col, price_col, channel_name):
        df = (
            sales_df.join(date_target, left_on=date_col, right_on="d_date_sk")
            .join(cross_item_sks, left_on=item_col, right_on="i_item_sk", how="semi")
            .join(item, left_on=item_col, right_on="i_item_sk")
            .with_columns((col(qty_col) * col(price_col)).alias("sales"))
        )
        grouped = df.group_by(["i_brand_id", "i_class_id", "i_category_id"]).agg(
            col("sales").sum().alias("sales"),
            ctx.len().alias("number_sales"),
        )
        grouped = grouped.filter(col("sales") > lit(avg_sales_val))
        grouped = grouped.with_columns(lit(channel_name).alias("channel"))
        return grouped

    ss_result = channel_sales(store_sales, "ss_sold_date_sk", "ss_item_sk", "ss_quantity", "ss_list_price", "store")
    cs_result = channel_sales(catalog_sales, "cs_sold_date_sk", "cs_item_sk", "cs_quantity", "cs_list_price", "catalog")
    ws_result = channel_sales(web_sales, "ws_sold_date_sk", "ws_item_sk", "ws_quantity", "ws_list_price", "web")

    combined = ctx.concat([ss_result, cs_result, ws_result])

    group_cols = ["channel", "i_brand_id", "i_class_id", "i_category_id"]
    agg_exprs = [
        ctx.sum("sales").alias("sum_sales"),
        ctx.sum("number_sales").alias("sum_number_sales"),
    ]

    result = expand_rollup_expression(combined, group_cols, agg_exprs, ctx)
    return result.select([*group_cols, "sum_sales", "sum_number_sales"]).sort(group_cols, nulls_last=True).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['14'])

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

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

Source

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