TPC-DS Q54 — Catalog+Web Customer Store Revenue Segments

TPC-DS benchmark · all TPC-DS queries

Segments catalog/web customers by their store sales revenue

Representative SQL

Rendered for DataFusion with default parameters.

WITH my_customers AS (
  SELECT DISTINCT
    c_customer_sk,
    c_current_addr_sk
  FROM (
    SELECT
      cs_sold_date_sk AS sold_date_sk,
      cs_bill_customer_sk AS customer_sk,
      cs_item_sk AS item_sk
    FROM catalog_sales
    UNION ALL
    SELECT
      ws_sold_date_sk AS sold_date_sk,
      ws_bill_customer_sk AS customer_sk,
      ws_item_sk AS item_sk
    FROM web_sales
  ) AS cs_or_ws_sales, item, date_dim, customer
  WHERE
    sold_date_sk = d_date_sk
    AND item_sk = i_item_sk
    AND i_category = 'Jewelry'
    AND i_class = 'consignment'
    AND c_customer_sk = cs_or_ws_sales.customer_sk
    AND d_moy = 3
    AND d_year = 1999
), my_revenue AS (
  SELECT
    c_customer_sk,
    SUM(ss_ext_sales_price) AS revenue
  FROM my_customers, store_sales, customer_address, store, date_dim
  WHERE
    c_current_addr_sk = ca_address_sk
    AND ca_county = s_county
    AND ca_state = s_state
    AND ss_sold_date_sk = d_date_sk
    AND c_customer_sk = ss_customer_sk
    AND d_month_seq BETWEEN (
      SELECT DISTINCT
        d_month_seq + 1
      FROM date_dim
      WHERE
        d_year = 1999 AND d_moy = 3
    ) AND (
      SELECT DISTINCT
        d_month_seq + 3
      FROM date_dim
      WHERE
        d_year = 1999 AND d_moy = 3
    )
  GROUP BY
    c_customer_sk
), segments AS (
  SELECT
    CAST((
      revenue / 50
    ) AS INT) AS segment
  FROM my_revenue
)
SELECT
  segment,
  COUNT(*) AS num_customers,
  segment * 50 AS segment_base
FROM segments
GROUP BY
  segment
ORDER BY
  segment,
  num_customers
LIMIT 100

Representative DataFrame

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

def q54_expression_impl(ctx: DataFrameContext) -> Any:
    params = get_parameters(54)
    year = params.get("year", 1998)
    month = params.get("month", 12)
    category = params.get("category", "Women")
    item_class = params.get("class", "maternity")

    catalog_sales, web_sales, store_sales, customer, customer_address, store, item, date_dim = _tables(
        ctx, "catalog_sales", "web_sales", "store_sales", "customer", "customer_address", "store", "item", "date_dim"
    )
    col = ctx.col
    lit = ctx.lit

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

    date_filtered = date_dim.filter((col("d_year") == lit(year)) & (col("d_moy") == lit(month)))

    cs_sales = catalog_sales.select(
        [
            col("cs_sold_date_sk").alias("sold_date_sk"),
            col("cs_bill_customer_sk").alias("customer_sk"),
            col("cs_item_sk").alias("item_sk"),
        ]
    )

    ws_sales = web_sales.select(
        [
            col("ws_sold_date_sk").alias("sold_date_sk"),
            col("ws_bill_customer_sk").alias("customer_sk"),
            col("ws_item_sk").alias("item_sk"),
        ]
    )

    cs_or_ws = ctx.concat([cs_sales, ws_sales])

    my_customers = (
        cs_or_ws.join(item_filtered, left_on="item_sk", right_on="i_item_sk")
        .join(date_filtered, left_on="sold_date_sk", right_on="d_date_sk")
        .join(customer, left_on="customer_sk", right_on="c_customer_sk")
        .select([col("customer_sk").alias("c_customer_sk"), "c_current_addr_sk"])
        .unique()
    )

    month_seq_df = (
        date_dim.filter((col("d_year") == lit(year)) & (col("d_moy") == lit(month))).select("d_month_seq").unique()
    )

    following_dates = (
        date_dim.join(month_seq_df, how="cross")
        .filter(
            (col("d_month_seq") >= col("d_month_seq_right") + lit(1))
            & (col("d_month_seq") <= col("d_month_seq_right") + lit(3))
        )
        .select("d_date_sk")
    )

    my_revenue = (
        my_customers.join(customer_address, left_on="c_current_addr_sk", right_on="ca_address_sk")
        .join(store, left_on=["ca_county", "ca_state"], right_on=["s_county", "s_state"])
        .join(store_sales, left_on="c_customer_sk", right_on="ss_customer_sk")
        .join(following_dates, left_on="ss_sold_date_sk", right_on="d_date_sk")
        .group_by("c_customer_sk")
        .agg(_sum_or_null_expression(ctx, col("ss_ext_sales_price")).alias("revenue"))
    )

    segments = my_revenue.with_columns((col("revenue") / lit(50)).round(0).cast_int64().alias("segment"))

    return _sort_null_largest_expression(
        ctx,
        segments.group_by("segment")
        .agg(ctx.len().alias("num_customers"))
        .with_columns((col("segment") * lit(50)).alias("segment_base"))
        .select(["segment", "num_customers", "segment_base"]),
        ["segment", "num_customers"],
    ).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['54'])

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

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

Source

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