TPC-DS Q64 — Cross-Sales Income Band

TPC-DS benchmark · all TPC-DS queries

Complex cross-sales analysis with income band filtering and year-over-year comparison

Representative SQL

Rendered for DataFusion with default parameters.

WITH cs_ui AS (
  SELECT
    cs_item_sk,
    SUM(cs_ext_list_price) AS sale,
    SUM(cr_refunded_cash + cr_reversed_charge + cr_store_credit) AS refund
  FROM catalog_sales, catalog_returns
  WHERE
    cs_item_sk = cr_item_sk AND cs_order_number = cr_order_number
  GROUP BY
    cs_item_sk
  HAVING
    SUM(cs_ext_list_price) > 2 * SUM(cr_refunded_cash + cr_reversed_charge + cr_store_credit)
), cross_sales AS (
  SELECT
    i_product_name AS product_name,
    i_item_sk AS item_sk,
    s_store_name AS store_name,
    s_zip AS store_zip,
    ad1.ca_street_number AS b_street_number,
    ad1.ca_street_name AS b_street_name,
    ad1.ca_city AS b_city,
    ad1.ca_zip AS b_zip,
    ad2.ca_street_number AS c_street_number,
    ad2.ca_street_name AS c_street_name,
    ad2.ca_city AS c_city,
    ad2.ca_zip AS c_zip,
    d1.d_year AS syear,
    d2.d_year AS fsyear,
    d3.d_year AS s2year,
    COUNT(*) AS cnt,
    SUM(ss_wholesale_cost) AS s1,
    SUM(ss_list_price) AS s2,
    SUM(ss_coupon_amt) AS s3
  FROM store_sales, store_returns, cs_ui, date_dim AS d1, date_dim AS d2, date_dim AS d3, store, customer, customer_demographics AS cd1, customer_demographics AS cd2, promotion, household_demographics AS hd1, household_demographics AS hd2, customer_address AS ad1, customer_address AS ad2, income_band AS ib1, income_band AS ib2, item
  WHERE
    ss_store_sk = s_store_sk
    AND ss_sold_date_sk = d1.d_date_sk
    AND ss_customer_sk = c_customer_sk
    AND ss_cdemo_sk = cd1.cd_demo_sk
    AND ss_hdemo_sk = hd1.hd_demo_sk
    AND ss_addr_sk = ad1.ca_address_sk
    AND ss_item_sk = i_item_sk
    AND ss_item_sk = sr_item_sk
    AND ss_ticket_number = sr_ticket_number
    AND ss_item_sk = cs_ui.cs_item_sk
    AND c_current_cdemo_sk = cd2.cd_demo_sk
    AND c_current_hdemo_sk = hd2.hd_demo_sk
    AND c_current_addr_sk = ad2.ca_address_sk
    AND c_first_sales_date_sk = d2.d_date_sk
    AND c_first_shipto_date_sk = d3.d_date_sk
    AND ss_promo_sk = p_promo_sk
    AND hd1.hd_income_band_sk = ib1.ib_income_band_sk
    AND hd2.hd_income_band_sk = ib2.ib_income_band_sk
    AND cd1.cd_marital_status <> cd2.cd_marital_status
    AND i_color IN ('maroon', 'burnished', 'dim', 'steel', 'navajo', 'chocolate')
    AND i_current_price BETWEEN 35 AND 35 + 10
    AND i_current_price BETWEEN 35 + 1 AND 35 + 15
  GROUP BY
    i_product_name,
    i_item_sk,
    s_store_name,
    s_zip,
    ad1.ca_street_number,
    ad1.ca_street_name,
    ad1.ca_city,
    ad1.ca_zip,
    ad2.ca_street_number,
    ad2.ca_street_name,
    ad2.ca_city,
    ad2.ca_zip,
    d1.d_year,
    d2.d_year,
    d3.d_year
)
SELECT
  cs1.product_name,
  cs1.store_name,
  cs1.store_zip,
  cs1.b_street_number,
  cs1.b_street_name,
  cs1.b_city,
  cs1.b_zip,
  cs1.c_street_number,
  cs1.c_street_name,
  cs1.c_city,
  cs1.c_zip,
  cs1.syear,
  cs1.cnt,
  cs1.s1 AS s11,
  cs1.s2 AS s21,
  cs1.s3 AS s31,
  cs2.s1 AS s12,
  cs2.s2 AS s22,
  cs2.s3 AS s32,
  cs2.syear,
  cs2.cnt
FROM cross_sales AS cs1, cross_sales AS cs2
WHERE
  cs1.item_sk = cs2.item_sk
  AND cs1.syear = 2000
  AND cs2.syear = 2000 + 1
  AND cs2.cnt <= cs1.cnt
  AND cs1.store_name = cs2.store_name
  AND cs1.store_zip = cs2.store_zip
ORDER BY
  cs1.product_name,
  cs1.store_name,
  cs2.cnt,
  cs1.s1,
  cs2.s1

Representative DataFrame

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

def q64_expression_impl(ctx: DataFrameContext) -> Any:
    params = get_parameters(64)
    year = params.get("year", 1999)
    colors = params.get("colors", ["slate", "blanched", "burnished", "chartreuse", "peru", "thistle"])
    price_min = params.get("price_min", 0)

    (
        store_sales,
        store_returns,
        catalog_sales,
        catalog_returns,
        date_dim,
        store,
        customer,
        customer_demographics,
        promotion,
        household_demographics,
        customer_address,
        income_band,
        item,
    ) = _tables(
        ctx,
        "store_sales",
        "store_returns",
        "catalog_sales",
        "catalog_returns",
        "date_dim",
        "store",
        "customer",
        "customer_demographics",
        "promotion",
        "household_demographics",
        "customer_address",
        "income_band",
        "item",
    )

    col = ctx.col
    lit = ctx.lit

    cs_with_returns = catalog_sales.join(
        catalog_returns,
        left_on=["cs_item_sk", "cs_order_number"],
        right_on=["cr_item_sk", "cr_order_number"],
    )
    cs_ui = (
        cs_with_returns.group_by("cs_item_sk")
        .agg(
            [
                _sum_or_null_expression(ctx, ctx.col("cs_ext_list_price")).alias("sale"),
                (
                    _sum_or_null_expression(ctx, ctx.col("cr_refunded_cash"))
                    + _sum_or_null_expression(ctx, ctx.col("cr_reversed_charge"))
                    + _sum_or_null_expression(ctx, ctx.col("cr_store_credit"))
                ).alias("refund"),
            ]
        )
        .filter(col("sale") > lit(2) * col("refund"))
        .select("cs_item_sk")
    )

    item_filtered = item.filter(
        col("i_color").is_in(colors)
        & (col("i_current_price") >= lit(price_min))
        & (col("i_current_price") <= lit(price_min + 10))
        & (col("i_current_price") >= lit(price_min + 1))
        & (col("i_current_price") <= lit(price_min + 15))
    )

    ss_sr = store_sales.join(
        store_returns,
        left_on=["ss_item_sk", "ss_ticket_number"],
        right_on=["sr_item_sk", "sr_ticket_number"],
    )

    ss_sr = ss_sr.join(cs_ui, left_on="ss_item_sk", right_on="cs_item_sk", how="semi")

    cross_sales = (
        ss_sr.join(item_filtered, left_on="ss_item_sk", right_on="i_item_sk")
        .join(store, left_on="ss_store_sk", right_on="s_store_sk")
        .join(customer, left_on="ss_customer_sk", right_on="c_customer_sk")
        .join(
            customer_demographics.select(["cd_demo_sk", "cd_marital_status"]).rename(
                {"cd_marital_status": "cd1_marital_status"}
            ),
            left_on="ss_cdemo_sk",
            right_on="cd_demo_sk",
        )
        .join(promotion, left_on="ss_promo_sk", right_on="p_promo_sk")
        .join(
            household_demographics.select(["hd_demo_sk", "hd_income_band_sk"]).rename(
                {"hd_income_band_sk": "hd1_ib_sk"}
            ),
            left_on="ss_hdemo_sk",
            right_on="hd_demo_sk",
        )
        .join(
            customer_address.select(
                ["ca_address_sk", "ca_street_number", "ca_street_name", "ca_city", "ca_zip"]
            ).rename(
                {
                    "ca_street_number": "b_street_number",
                    "ca_street_name": "b_street_name",
                    "ca_city": "b_city",
                    "ca_zip": "b_zip",
                }
            ),
            left_on="ss_addr_sk",
            right_on="ca_address_sk",
        )
        .join(
            date_dim.select(["d_date_sk", "d_year"]).rename({"d_year": "syear"}),
            left_on="ss_sold_date_sk",
            right_on="d_date_sk",
        )
        .join(
            date_dim.select(["d_date_sk", "d_year"]).rename({"d_date_sk": "d2_date_sk", "d_year": "fsyear"}),
            left_on="c_first_sales_date_sk",
            right_on="d2_date_sk",
        )
        .join(
            date_dim.select(["d_date_sk", "d_year"]).rename({"d_date_sk": "d3_date_sk", "d_year": "s2year"}),
            left_on="c_first_shipto_date_sk",
            right_on="d3_date_sk",
        )
        .join(
            customer_demographics.select(["cd_demo_sk", "cd_marital_status"]).rename(
                {"cd_marital_status": "cd2_marital_status", "cd_demo_sk": "cd2_demo_sk"}
            ),
            left_on="c_current_cdemo_sk",
            right_on="cd2_demo_sk",
        )
        .join(
            household_demographics.select(["hd_demo_sk", "hd_income_band_sk"]).rename(
                {"hd_income_band_sk": "hd2_ib_sk", "hd_demo_sk": "hd2_demo_sk"}
            ),
            left_on="c_current_hdemo_sk",
            right_on="hd2_demo_sk",
        )
        .join(
            customer_address.select(
                ["ca_address_sk", "ca_street_number", "ca_street_name", "ca_city", "ca_zip"]
            ).rename(
                {
                    "ca_address_sk": "c_addr_sk",
                    "ca_street_number": "c_street_number",
                    "ca_street_name": "c_street_name",
                    "ca_city": "c_city",
                    "ca_zip": "c_zip",
                }
            ),
            left_on="c_current_addr_sk",
            right_on="c_addr_sk",
        )
        .join(
            income_band.select(["ib_income_band_sk"]).rename({"ib_income_band_sk": "ib1_sk"}),
            left_on="hd1_ib_sk",
            right_on="ib1_sk",
        )
        .join(
            income_band.select(["ib_income_band_sk"]).rename({"ib_income_band_sk": "ib2_sk"}),
            left_on="hd2_ib_sk",
            right_on="ib2_sk",
        )
    )

    cross_sales = cross_sales.filter(col("cd1_marital_status") != col("cd2_marital_status"))

    grouped = cross_sales.group_by(
        [
            "i_product_name",
            "ss_item_sk",
            "s_store_name",
            "s_zip",
            "b_street_number",
            "b_street_name",
            "b_city",
            "b_zip",
            "c_street_number",
            "c_street_name",
            "c_city",
            "c_zip",
            "syear",
            "fsyear",
            "s2year",
        ]
    ).agg(
        [
            ctx.len().alias("cnt"),
            _sum_or_null_expression(ctx, ctx.col("ss_wholesale_cost")).alias("s1"),
            _sum_or_null_expression(ctx, ctx.col("ss_list_price")).alias("s2"),
            _sum_or_null_expression(ctx, ctx.col("ss_coupon_amt")).alias("s3"),
        ]
    )

    cs1 = grouped.filter(col("syear") == lit(year))
    cs2 = grouped.filter(col("syear") == lit(year + 1))

    return _sort_null_largest_expression(
        ctx,
        cs1.join(
            cs2,
            on=["ss_item_sk", "s_store_name", "s_zip"],
            suffix="_y2",
        )
        .filter(col("cnt_y2") <= col("cnt"))
        .select(
            [
                "i_product_name",
                "s_store_name",
                "s_zip",
                "b_street_number",
                "b_street_name",
                "b_city",
                "b_zip",
                "c_street_number",
                "c_street_name",
                "c_city",
                "c_zip",
                "syear",
                "cnt",
                "s1",
                "s2",
                "s3",
                col("s1_y2").alias("s12"),
                col("s2_y2").alias("s22"),
                col("s3_y2").alias("s32"),
                col("syear_y2").alias("syear2"),
                col("cnt_y2").alias("cnt2"),
            ]
        ),
        ["i_product_name", "s_store_name", "cnt2", "s1", "s12"],
    )

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

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

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

Source

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