TPC-DS Q58 — Three-Channel Item Sales Balance Check

TPC-DS benchmark · all TPC-DS queries

Items with balanced sales across all channels within 10% range

Representative SQL

Rendered for DataFusion with default parameters.

WITH ss_items AS (
  SELECT
    i_item_id AS item_id,
    SUM(ss_ext_sales_price) AS ss_item_rev
  FROM store_sales, item, date_dim
  WHERE
    ss_item_sk = i_item_sk
    AND d_date IN (
      SELECT
        d_date
      FROM date_dim
      WHERE
        d_week_seq = (
          SELECT
            d_week_seq
          FROM date_dim
          WHERE
            d_date = '1998-02-19'
        )
    )
    AND ss_sold_date_sk = d_date_sk
  GROUP BY
    i_item_id
), cs_items AS (
  SELECT
    i_item_id AS item_id,
    SUM(cs_ext_sales_price) AS cs_item_rev
  FROM catalog_sales, item, date_dim
  WHERE
    cs_item_sk = i_item_sk
    AND d_date IN (
      SELECT
        d_date
      FROM date_dim
      WHERE
        d_week_seq = (
          SELECT
            d_week_seq
          FROM date_dim
          WHERE
            d_date = '1998-02-19'
        )
    )
    AND cs_sold_date_sk = d_date_sk
  GROUP BY
    i_item_id
), ws_items AS (
  SELECT
    i_item_id AS item_id,
    SUM(ws_ext_sales_price) AS ws_item_rev
  FROM web_sales, item, date_dim
  WHERE
    ws_item_sk = i_item_sk
    AND d_date IN (
      SELECT
        d_date
      FROM date_dim
      WHERE
        d_week_seq = (
          SELECT
            d_week_seq
          FROM date_dim
          WHERE
            d_date = '1998-02-19'
        )
    )
    AND ws_sold_date_sk = d_date_sk
  GROUP BY
    i_item_id
)
SELECT
  ss_items.item_id,
  ss_item_rev,
  ss_item_rev / (
    (
      ss_item_rev + cs_item_rev + ws_item_rev
    ) / 3
  ) * 100 AS ss_dev,
  cs_item_rev,
  cs_item_rev / (
    (
      ss_item_rev + cs_item_rev + ws_item_rev
    ) / 3
  ) * 100 AS cs_dev,
  ws_item_rev,
  ws_item_rev / (
    (
      ss_item_rev + cs_item_rev + ws_item_rev
    ) / 3
  ) * 100 AS ws_dev,
  (
    ss_item_rev + cs_item_rev + ws_item_rev
  ) / 3 AS average
FROM ss_items, cs_items, ws_items
WHERE
  ss_items.item_id = cs_items.item_id
  AND ss_items.item_id = ws_items.item_id
  AND ss_item_rev BETWEEN 0.9 * cs_item_rev AND 1.1 * cs_item_rev
  AND ss_item_rev BETWEEN 0.9 * ws_item_rev AND 1.1 * ws_item_rev
  AND cs_item_rev BETWEEN 0.9 * ss_item_rev AND 1.1 * ss_item_rev
  AND cs_item_rev BETWEEN 0.9 * ws_item_rev AND 1.1 * ws_item_rev
  AND ws_item_rev BETWEEN 0.9 * ss_item_rev AND 1.1 * ss_item_rev
  AND ws_item_rev BETWEEN 0.9 * cs_item_rev AND 1.1 * cs_item_rev
ORDER BY
  ss_items.item_id,
  ss_item_rev
LIMIT 100

Representative DataFrame

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

def q58_expression_impl(ctx: DataFrameContext) -> Any:
    params = get_parameters(58)
    sales_date = datetime.strptime(params.get("sales_date", "2000-01-03"), "%Y-%m-%d").date()
    item, date_dim = _tables(ctx, "item", "date_dim")
    col = ctx.col
    lit = ctx.lit

    week_seq_df = date_dim.filter(col("d_date") == lit(sales_date)).select("d_week_seq")
    dates_in_week = date_dim.join(week_seq_df, on="d_week_seq").select("d_date_sk")

    def channel(table: str, item_key: str, date_key: str, value_col: str, alias: str) -> Any:
        return (
            ctx.get_table(table)
            .join(item, left_on=item_key, right_on="i_item_sk")
            .join(dates_in_week, left_on=date_key, right_on="d_date_sk")
            .group_by("i_item_id")
            .agg(col(value_col).sum().alias(alias))
        )

    result = (
        channel("store_sales", "ss_item_sk", "ss_sold_date_sk", "ss_ext_sales_price", "ss_item_rev")
        .join(
            channel("catalog_sales", "cs_item_sk", "cs_sold_date_sk", "cs_ext_sales_price", "cs_item_rev"),
            on="i_item_id",
        )
        .join(
            channel("web_sales", "ws_item_sk", "ws_sold_date_sk", "ws_ext_sales_price", "ws_item_rev"), on="i_item_id"
        )
    )
    return (
        _q58_balance_expression(ctx, result)
        .with_columns(((col("ss_item_rev") + col("cs_item_rev") + col("ws_item_rev")) / lit(3)).alias("average"))
        .with_columns(
            [
                (col("ss_item_rev") / col("average") * lit(100)).alias("ss_dev"),
                (col("cs_item_rev") / col("average") * lit(100)).alias("cs_dev"),
                (col("ws_item_rev") / col("average") * lit(100)).alias("ws_dev"),
            ]
        )
        .select(
            [
                col("i_item_id").alias("item_id"),
                "ss_item_rev",
                "ss_dev",
                "cs_item_rev",
                "cs_dev",
                "ws_item_rev",
                "ws_dev",
                "average",
            ]
        )
        .sort(["item_id", "ss_item_rev"])
        .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['58'])

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

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

Source

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