TPC-DS Q44 — Store Sales Item Ranking

TPC-DS benchmark · all TPC-DS queries

Ranks items by net profit for a store, finding best and worst performers

Representative SQL

Rendered for DataFusion with default parameters.

SELECT
  asceding.rnk,
  i1.i_product_name AS best_performing,
  i2.i_product_name AS worst_performing
FROM (
  SELECT
    *
  FROM (
    SELECT
      item_sk,
      RANK() OVER (ORDER BY rank_col ASC) AS rnk
    FROM (
      SELECT
        ss_item_sk AS item_sk,
        AVG(ss_net_profit) AS rank_col
      FROM store_sales AS ss1
      WHERE
        ss_store_sk = 2
      GROUP BY
        ss_item_sk
      HAVING
        AVG(ss_net_profit) > 0.9 * (
          SELECT
            AVG(ss_net_profit) AS rank_col
          FROM store_sales
          WHERE
            ss_store_sk = 2 AND ss_hdemo_sk IS NULL
          GROUP BY
            ss_store_sk
        )
    ) AS V1
  ) AS V11
  WHERE
    rnk < 11
) AS asceding, (
  SELECT
    *
  FROM (
    SELECT
      item_sk,
      RANK() OVER (ORDER BY rank_col DESC) AS rnk
    FROM (
      SELECT
        ss_item_sk AS item_sk,
        AVG(ss_net_profit) AS rank_col
      FROM store_sales AS ss1
      WHERE
        ss_store_sk = 2
      GROUP BY
        ss_item_sk
      HAVING
        AVG(ss_net_profit) > 0.9 * (
          SELECT
            AVG(ss_net_profit) AS rank_col
          FROM store_sales
          WHERE
            ss_store_sk = 2 AND ss_hdemo_sk IS NULL
          GROUP BY
            ss_store_sk
        )
    ) AS V2
  ) AS V21
  WHERE
    rnk < 11
) AS descending, item AS i1, item AS i2
WHERE
  asceding.rnk = descending.rnk
  AND i1.i_item_sk = asceding.item_sk
  AND i2.i_item_sk = descending.item_sk
ORDER BY
  asceding.rnk
LIMIT 100

Representative DataFrame

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

def q44_expression_impl(ctx: DataFrameContext) -> Any:
    params = get_parameters(44)
    store_sk = params.get("store_sk", 4)
    null_col = params.get("null_col", "ss_addr_sk")

    col = ctx.col
    lit = ctx.lit

    store_sales, item = _tables(ctx, "store_sales", "item")

    ss_store = store_sales.filter(col("ss_store_sk") == lit(store_sk))

    threshold = 0.9

    item_avg = ss_store.group_by("ss_item_sk").agg(col("ss_net_profit").mean().alias("rank_col"))

    ss_null = store_sales.filter((col("ss_store_sk") == lit(store_sk)) & col(null_col).is_null())
    store_baseline = ss_null.select(col("ss_net_profit").mean().alias("baseline"))

    item_with_baseline = item_avg.join(store_baseline, how="cross")
    qualified = item_with_baseline.filter(col("rank_col") > (lit(threshold) * col("baseline")))

    ascending = (
        qualified.with_columns(col("rank_col").rank(method="min", descending=False).alias("rnk"))
        .filter(col("rnk") <= 10)
        .select(["ss_item_sk", "rnk"])
        .rename({"ss_item_sk": "item_sk_asc", "rnk": "rnk_asc"})
    )

    descending = (
        qualified.with_columns(col("rank_col").rank(method="min", descending=True).alias("rnk"))
        .filter(col("rnk") <= 10)
        .select(["ss_item_sk", "rnk"])
        .rename({"ss_item_sk": "item_sk_desc", "rnk": "rnk_desc"})
    )

    result = ascending.join(descending, left_on="rnk_asc", right_on="rnk_desc")

    result = (
        result.join(item, left_on="item_sk_asc", right_on="i_item_sk")
        .rename({"i_product_name": "best_performing"})
        .select(["rnk_asc", "best_performing", "item_sk_desc"])
    )

    return (
        result.join(item, left_on="item_sk_desc", right_on="i_item_sk")
        .rename({"i_product_name": "worst_performing"})
        .select(["rnk_asc", "best_performing", "worst_performing"])
        .rename({"rnk_asc": "rnk"})
        .sort("rnk")
    )

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

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

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

Source

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