TPC-DS Q33 — Three-Channel Sales by Manufacturer

TPC-DS benchmark · all TPC-DS queries

Union of store/catalog/web sales by manufacturer filtered by category

Representative SQL

Rendered for DataFusion with default parameters.

WITH ss AS (
  SELECT
    i_manufact_id,
    SUM(ss_ext_sales_price) AS total_sales
  FROM store_sales, date_dim, customer_address, item
  WHERE
    i_manufact_id IN (
      SELECT
        i_manufact_id
      FROM item
      WHERE
        i_category IN ('Books')
    )
    AND ss_item_sk = i_item_sk
    AND ss_sold_date_sk = d_date_sk
    AND d_year = 1999
    AND d_moy = 3
    AND ss_addr_sk = ca_address_sk
    AND ca_gmt_offset = -5
  GROUP BY
    i_manufact_id
), cs AS (
  SELECT
    i_manufact_id,
    SUM(cs_ext_sales_price) AS total_sales
  FROM catalog_sales, date_dim, customer_address, item
  WHERE
    i_manufact_id IN (
      SELECT
        i_manufact_id
      FROM item
      WHERE
        i_category IN ('Books')
    )
    AND cs_item_sk = i_item_sk
    AND cs_sold_date_sk = d_date_sk
    AND d_year = 1999
    AND d_moy = 3
    AND cs_bill_addr_sk = ca_address_sk
    AND ca_gmt_offset = -5
  GROUP BY
    i_manufact_id
), ws AS (
  SELECT
    i_manufact_id,
    SUM(ws_ext_sales_price) AS total_sales
  FROM web_sales, date_dim, customer_address, item
  WHERE
    i_manufact_id IN (
      SELECT
        i_manufact_id
      FROM item
      WHERE
        i_category IN ('Books')
    )
    AND ws_item_sk = i_item_sk
    AND ws_sold_date_sk = d_date_sk
    AND d_year = 1999
    AND d_moy = 3
    AND ws_bill_addr_sk = ca_address_sk
    AND ca_gmt_offset = -5
  GROUP BY
    i_manufact_id
)
SELECT
  i_manufact_id,
  SUM(total_sales) AS total_sales
FROM (
  SELECT
    *
  FROM ss
  UNION ALL
  SELECT
    *
  FROM cs
  UNION ALL
  SELECT
    *
  FROM ws
) AS tmp1
GROUP BY
  i_manufact_id
ORDER BY
  total_sales
LIMIT 100

Representative DataFrame

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

def q33_expression_impl(ctx: DataFrameContext) -> Any:
    col = ctx.col

    params = get_parameters(33)
    year = params.get("year", 1999)
    month = params.get("month", 3)
    gmt_offset = params.get("gmt_offset", -5)
    category = params.get("category", "Books")

    date_dim, customer_address, item = _tables(ctx, "date_dim", "customer_address", "item")

    mfg_ids = item.filter(col("i_category") == category).select("i_manufact_id").unique()
    date_filter = date_dim.filter((col("d_year") == year) & (col("d_moy") == month))
    addr_filter = customer_address.filter(col("ca_gmt_offset") == gmt_offset)

    def channel(table: str, item_key: str, date_key: str, addr_key: str, value_col: str) -> Any:
        return (
            ctx.get_table(table)
            .join(item, left_on=item_key, right_on="i_item_sk", how="inner")
            .join(date_filter, left_on=date_key, right_on="d_date_sk", how="inner")
            .join(addr_filter, left_on=addr_key, right_on="ca_address_sk", how="inner")
            .join(mfg_ids, on="i_manufact_id", how="semi")
            .group_by("i_manufact_id")
            .agg(_sum_or_null_expression(ctx, col(value_col)).alias("total_sales"))
        )

    combined = ctx.concat(
        [
            channel("store_sales", "ss_item_sk", "ss_sold_date_sk", "ss_addr_sk", "ss_ext_sales_price"),
            channel("catalog_sales", "cs_item_sk", "cs_sold_date_sk", "cs_bill_addr_sk", "cs_ext_sales_price"),
            channel("web_sales", "ws_item_sk", "ws_sold_date_sk", "ws_bill_addr_sk", "ws_ext_sales_price"),
        ]
    )
    return _sort_null_largest_expression(
        ctx,
        combined.group_by("i_manufact_id").agg(_sum_or_null_expression(ctx, col("total_sales")).alias("total_sales")),
        ["total_sales"],
    ).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['33'])

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

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

Source

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