Read Primitives Benchmark API

Python API reference for the Read Primitives benchmark.

Overview

The Read Primitives benchmark tests 157 fundamental database read operations using the TPC-H schema. It provides focused testing of specific database capabilities without business logic complexity, ideal for performance regression detection and optimization validation.

Key Features:

157 primitive operation queries - Comprehensive database operation coverage
31 operation categories - Aggregation, joins, filters, window functions, optimizer tests, and more
DataFrame support - 152 of the queries have DataFrame implementations
TPC-H schema - 8 TPC-H tables; the data is generated by the TPC-H generator and shared with TPC-H
Category-based execution - Test specific operation types
Performance regression detection - Isolated operation testing
Optimizer sniff tests - 14 queries in the optimizer category
Extensible architecture - Custom primitives support

Quick Start

from benchbox import ReadPrimitives
from benchbox.platforms.duckdb import DuckDBAdapter

benchmark = ReadPrimitives(scale_factor=1.0)

benchmark.generate_data()

adapter = DuckDBAdapter()
results = benchmark.run_with_platform(adapter)

print(f"Completed {results.total_queries} queries in {results.total_execution_time:.2f}s")

The DuckDB adapter needs the duckdb package. Scale factor 1.0 generates about 6 million lineitem rows (about 1 GB).

Shared query and category contracts are maintained in Shared Benchmark Facade Mixins.

API Reference

ReadPrimitives Class

`benchbox.ReadPrimitives`

Creates a Read Primitives benchmark that generates the TPC-H tables and serves 157 single-purpose read queries in 31 categories. It offers catalog-driven read operations over the TPC-H schema. The query and category methods that it inherits from QueryFacadeMixin and QueryCategoryFacadeMixin remain part of the supported surface. Their maintained contracts are in Shared Benchmark Facade Mixins, and shared lifecycle behavior is in Base Benchmark API.

Import: from benchbox import ReadPrimitives · Extras: none

Parameters

Name	Type	Default	Meaning
`scale_factor`	`float`	`1.0`	Data size multiplier, as for TPC-H. Must be positive. Values of 1 or more must be whole numbers.
`output_dir`	`str`, `Path` or `None`	`None`	Directory for generated data files. When `None`, data goes to the TPC-H directory `benchmark_runs/datagen/tpch_<sf token>` under the current directory (for example `tpch_sf001` for 0.01), or under `$BENCHBOX_OUTPUT_DIR/datagen`, so it is shared with a `TPCH` benchmark of the same scale factor. The instance’s own `output_dir` attribute then names `read_primitives_<sf token>`, which is not where the files are written.
`**kwargs`	keyword arguments	none	`verbose` (`bool` or `int`) and `quiet` (`bool`) set the log level. Other keywords are stored as attributes and passed to the data generator.

The constructor creates no files. Data is written by generate_data().

Returns

A ReadPrimitives instance. It subclasses BaseBenchmark; see Base Benchmark API for the shared interface.

Raises

ValueError when scale_factor is zero or negative, or is 1 or more and not a whole number.

Example

from benchbox import ReadPrimitives

benchmark = ReadPrimitives(scale_factor=0.01, output_dir="primitives_data")
print(len(benchmark.get_queries()), len(benchmark.get_query_categories()))
print(benchmark.get_data_source_benchmark())

Output on 0.4.1:

157 31
tpch

Compatibility

benchbox.read_primitives.ReadPrimitives is the same class. The class attribute DATA_SOURCE_BENCHMARK is None on the wrapper; get_data_source_benchmark() returns "tpch".

Constructor

ReadPrimitives(scale_factor=1.0, output_dir=None, **kwargs). The arguments are in the Parameters table above.

Methods

generate_data(tables=None)

generate_data(tables=None) -> dict[str, str] generates the TPC-H tables and returns a mapping from table name to file path (str). benchmark.tables is the same mapping.

Name	Type	Default	Meaning
`tables`	`list[str]` or `None`	`None`	Table names to include in the result: `customer`, `lineitem`, `nation`, `orders`, `part`, `partsupp`, `region`, `supplier`. `None` returns all eight. Unknown names raise `ValueError` (`Invalid table names: {'bogus'}`).

The files are pipe-delimited .tbl files (lineitem.tbl), written by the TPC-H dbgen tool, so the directory also holds dists.dss and _datagen_manifest.json. At scale factor 0.01 all eight files are written even when tables names two of them; the result holds only the named ones.

table_files = benchmark.generate_data()

table_files = benchmark.generate_data(tables=["lineitem", "orders"])

for table_name, file_path in table_files.items():
    print(f"{table_name}: {file_path}")

The first call generates all tables. The second names two tables.

get_query(query_id, *, params=None)

get_query(query_id, *, params=None) -> str returns the SQL text of one primitive query. The text starts with a -- comment line that describes the operation.

Name	Type	Default	Meaning
`query_id`	`str`	required	A query name such as `"aggregation_simple"`. There are 157 names, listed by `get_queries()`. Integers are not valid ids.
`params`	`None`	`None`	The queries are static. Any non-`None` value, even `{}`, raises `ValueError`.

Raises ValueError for an unknown id (Invalid query ID: zz. Available: ...) and for non-None params (Read Primitives queries are static and don't accept parameters).

q1 = benchmark.get_query("aggregation_simple")

q2 = benchmark.get_query("window_rank")

q3 = benchmark.get_query("broadcast_join_two_tables")

get_queries(dialect=None)

get_queries(dialect=None) -> dict[str, str] returns all queries keyed by name.

With a dialect, each query is translated with SQLGlot, or replaced by a hand-written variant for that dialect where one exists (45 queries have variants). Queries that cannot run on the dialect are left out, so the result can be smaller than 157: 153 for duckdb, 157 for postgres, 147 for clickhouse, 154 for snowflake and 144 for bigquery. An unknown dialect name returns all 157 queries unchanged. Translation problems are logged as warnings.

queries = benchmark.get_queries()
print(f"Total queries: {len(queries)}")

queries_sf = benchmark.get_queries(dialect="snowflake")
print(len(queries_sf))

The first call gets all queries and prints Total queries: 157. The second gets them with dialect translation and prints 154.

get_queries_by_category(category)

get_queries_by_category(category) -> dict[str, str] returns the queries in one category, keyed by name. An unknown category returns {} without an error.

agg_queries = benchmark.get_queries_by_category("aggregation")

window_queries = benchmark.get_queries_by_category("window")

print(len(agg_queries), len(window_queries))

This prints 15 9.

get_query_categories()

get_query_categories() -> list[str] returns the 31 category names in alphabetical order.

categories = benchmark.get_query_categories()

print("Available categories:")
for category in categories:
    queries = benchmark.get_queries_by_category(category)
    print(f"  {category}: {len(queries)} queries")

Available Categories (queries per category): aggregation (15), array (9), broadcast (3), count (1), decimal (1), empty (1), exchange (3), filter (13), fulltext (3), groupby (5), intrinsic (1), json (3), lambda (3), limit (2), long (1), map (3), max (3), min (3), olap (2), optimizer (14), orderby (11), pivot (2), predicate (4), qualify (6), shuffle (6), statistical (5), string (16), struct (3), timeseries (2), topn (4) and window (9). Every query is in exactly one category (157 in total).

get_schema()

get_schema() -> dict[str, dict] returns the TPC-H schema as a mapping from table name to a definition with name and columns.

schema = benchmark.get_schema()

for table_name, table_def in schema.items():
    print(f"{table_name}: {len(table_def['columns'])} columns")

get_create_tables_sql(dialect=”standard”, tuning_config=None)

get_create_tables_sql(dialect="standard", tuning_config=None) -> str returns the CREATE TABLE statements for the eight TPC-H tables.

Name	Type	Default	Meaning
`dialect`	`str`	`"standard"`	Accepted but has no effect: `standard`, `duckdb`, `postgres`, `bigquery` and `clickhouse` return the same script.
`tuning_config`	`UnifiedTuningConfiguration` or `None`	`None`	With `None`, no constraints are emitted. A new `UnifiedTuningConfiguration()` enables primary and foreign keys, adding 8 `PRIMARY KEY` and 9 `FOREIGN KEY` clauses.

Raises RuntimeError when tuning_config lacks primary_keys or foreign_keys attributes.

create_sql = benchmark.get_create_tables_sql()

create_sql_pg = benchmark.get_create_tables_sql(dialect="postgres")

from benchbox.core.tuning.interface import UnifiedTuningConfiguration
tuning = UnifiedTuningConfiguration()
create_sql_tuned = benchmark.get_create_tables_sql(tuning_config=tuning)

The three calls return standard SQL, SQL for a specific dialect, and SQL with a tuning configuration.

run_benchmark(connection, queries=None, iterations=1, categories=None)

run_benchmark(connection, queries=None, iterations=1, categories=None) -> dict runs the queries on an open connection that already holds the TPC-H tables. It replaces the base class method of the same name with a different signature; use run_with_platform(adapter) for a full platform run.

Name	Type	Default	Meaning
`connection`	database connection	required	Connection with `execute()` or `cursor()`, such as `adapter.create_connection()`.
`queries`	`list[str]` or `None`	`None`	Query names to run. `None` runs all 157.
`iterations`	`int`	`1`	Times each query runs.
`categories`	`list[str]` or `None`	`None`	Categories to run; when given, it overrides `queries`. Unknown categories contribute no queries.

The result dict has benchmark ("Read Primitives"), scale_factor, iterations, categories, variant_comparability and queries. Each queries[name] entry has query_id, category, comparability, iterations (one dict per run with iteration, time, rows and success, or error when it failed), avg_time, min_time, max_time, sql_text and, when a run succeeded, rows_returned. An unknown query name is recorded as a failed iteration with its error message and category "unknown"; nothing is raised. The result has no total_time key.

from benchbox.platforms.duckdb import DuckDBAdapter

adapter = DuckDBAdapter()
conn = adapter.create_connection()
adapter.create_schema(benchmark, conn)
adapter.load_data(benchmark, conn, benchmark.output_dir)

results = benchmark.run_benchmark(conn)

results = benchmark.run_benchmark(
    conn,
    queries=["aggregation_simple", "window_rank"],
    iterations=3
)

results = benchmark.run_benchmark(
    conn,
    categories=["aggregation", "window"],
    iterations=3
)
print(len(results["queries"]))

The first call runs all queries. The second runs two named queries three times each. The third runs the aggregation and window categories, so results["queries"] holds 15 plus 9 entries and the example prints 24.

run_category_benchmark(connection, category, iterations=1)

run_category_benchmark(connection, category, iterations=1) -> dict runs every query of one category and returns the same structure as run_benchmark(..., categories=[category]). An unknown category returns a result with an empty queries dict.

agg_results = benchmark.run_category_benchmark(conn, "aggregation", iterations=3)

window_results = benchmark.run_category_benchmark(conn, "window", iterations=3)

print(len(agg_results["queries"]), len(window_results["queries"]))

This prints 15 9.

load_data_to_database(connection, tables=None)

load_data_to_database(connection, tables=None) -> None creates the TPC-H tables on the connection and loads the files from generate_data() into them, row by row. It is very slow: at scale factor 0.01 (60,175 lineitem rows) loading all tables took about 85 seconds. Prefer adapter.create_schema(benchmark, conn) and adapter.load_data(benchmark, conn, benchmark.output_dir).

Name	Type	Default	Meaning
`connection`	database connection	required	Target connection.
`tables`	`list[str]` or `None`	`None`	Tables to load. `None` loads all eight. All eight tables are created either way.

Raises ValueError (No data generated. Call generate_data() first.) before generate_data() has run. It fails with the database’s own error if the tables already exist, so do not call adapter.create_schema first.

execute_query(query_id, connection, params=None)

execute_query(query_id, connection, params=None) runs one query on the connection and returns the rows from fetchall(), a list of tuples. params is ignored. Raises ValueError for an unknown id.

result = benchmark.execute_query("aggregation_simple", conn)
print(len(result))

This prints 1, because the query returns one row.

get_benchmark_info()

get_benchmark_info() -> dict returns name ("Read Primitives Benchmark"), version ("1.0"), description, scale_factor, total_queries (157), categories (the 31 names), tables (the eight table names), schema ("TPC-H") and variant_comparability (total_queries, queries_with_variants, issue_count, comparable, per_query).

info = benchmark.get_benchmark_info()

print(f"Benchmark: {info['name']}")
print(f"Total queries: {info['total_queries']}")
print(f"Categories: {len(info['categories'])}")
print(f"Scale factor: {info['scale_factor']}")

Inherited members

Every other member comes from BaseBenchmark. See Base Benchmark API for what they do. The ids on this page for the inherited members are kept in the table.

Group	Member	Kind	Notes
Run and results	`cleanup`	method
Run and results	`create_enhanced_benchmark_result`	method
Run and results	`create_minimal_benchmark_result`	method
Run and results	`format_results`	method
Run and results	`run_query`	method
Run and results	`run_with_platform`	method
Run and results	`setup_database`	method
Run and results	`translate_query`	method
Validation	`validate_loaded_data`	method
Validation	`validate_manifest`	method
Validation	`validate_preflight`	method
Verbosity and logging	`apply_verbosity`	method
Verbosity and logging	`log_debug_info`	method
Verbosity and logging	`log_error_with_debug_info`	method
Verbosity and logging	`log_notice`	method
Verbosity and logging	`log_operation_complete`	method
Verbosity and logging	`log_operation_start`	method
Verbosity and logging	`log_verbose`	method
Verbosity and logging	`log_version_warning`	method
Verbosity and logging	`log_very_verbose`	method
Verbosity and logging	`logger`	property
Verbosity and logging	`quiet`	class attribute
Verbosity and logging	`verbose`	class attribute
Verbosity and logging	`verbose_enabled`	class attribute
Verbosity and logging	`verbose_level`	class attribute
Verbosity and logging	`verbosity_settings`	property
Verbosity and logging	`very_verbose`	class attribute
Data and configuration	`api_surface`	class attribute
Data and configuration	`benchmark_name`	property
Data and configuration	`csv_delimiter`	property	`None`.
Data and configuration	`csv_null_marker`	property	`None`.
Data and configuration	`DATA_SOURCE_BENCHMARK`	class attribute	`None` on this class; `get_data_source_benchmark()` returns `"tpch"`.
Data and configuration	`get_csv_loading_config`	method	`None`.
Data and configuration	`get_data_source_benchmark`	method
Data and configuration	`output_dir`	property	The resolved directory from the constructor argument, but see the `output_dir` row in the Parameters table for the default.
Data and configuration	`run_with_platform_api_surface`	class attribute
Data and configuration	`scale_factor`	instance attribute	The constructor argument.
Data and configuration	`SKIP_DATA_LOADING`	class attribute	Defined on `BaseBenchmark` from 0.4.2, default `False`. Set it to `True` for a benchmark that needs schema objects but no data files.
Data and configuration	`tables`	property	Empty until `generate_data()` has run, then the table-to-path mapping.

Usage Examples

Basic Benchmark Run

from benchbox import ReadPrimitives
from benchbox.platforms.duckdb import DuckDBAdapter

benchmark = ReadPrimitives(scale_factor=0.1)

benchmark.generate_data()

adapter = DuckDBAdapter()
results = benchmark.run_with_platform(adapter)

print(f"Benchmark: {results.benchmark_name}")
print(f"Total time: {results.total_execution_time:.2f}s")
print(f"Queries: {results.successful_queries}/{results.total_queries}")

Scale factor 0.1 generates about 100 MB of data.

Category-Based Execution

from benchbox import ReadPrimitives
from benchbox.platforms.duckdb import DuckDBAdapter

benchmark = ReadPrimitives(scale_factor=0.1)
benchmark.generate_data()

adapter = DuckDBAdapter()
conn = adapter.create_connection()

adapter.create_schema(benchmark, conn)
adapter.load_data(benchmark, conn, benchmark.output_dir)

categories = benchmark.get_query_categories()

category_results = {}

for category in categories:
    print(f"\nRunning {category} category...")
    results = benchmark.run_category_benchmark(conn, category, iterations=3)

    queries = results["queries"].values()
    successful = [q for q in queries if q["avg_time"] > 0]
    total_time = sum(q["avg_time"] for q in successful)

    category_results[category] = {
        "total_time": total_time,
        "avg_time": total_time / len(successful) if successful else 0,
        "query_count": len(results["queries"]),
        "successful": len(successful),
    }

    print(f"  Total time: {total_time:.2f}s")
    print(f"  Successful: {len(successful)}/{len(results['queries'])}")

print("\n" + "="*60)
print("Category Performance Summary")
print("="*60)
for category, stats in sorted(category_results.items()):
    print(f"{category:15s}: {stats['avg_time']:6.3f}s avg ({stats['query_count']} queries)")

Selective Query Execution

from benchbox import ReadPrimitives
from benchbox.platforms.duckdb import DuckDBAdapter
import time

benchmark = ReadPrimitives(scale_factor=0.1)
adapter = DuckDBAdapter()

benchmark.generate_data()
conn = adapter.create_connection()
adapter.create_schema(benchmark, conn)
adapter.load_data(benchmark, conn, benchmark.output_dir)

selected_queries = [
    "aggregation_simple",
    "aggregation_groupby_small",
    "window_rank",
    "broadcast_join_two_tables",
    "filter_string_selective"
]

print("Running selected primitive queries:")
query_times = {}

for query_id in selected_queries:
    query = benchmark.get_query(query_id)

    start = time.time()
    result = benchmark.execute_query(query_id, conn)
    duration = time.time() - start

    query_times[query_id] = duration
    print(f"  {query_id:25s}: {duration:.3f}s")

total_time = sum(query_times.values())
avg_time = total_time / len(query_times)
print(f"\nTotal: {total_time:.2f}s, Average: {avg_time:.3f}s")

Performance Regression Testing

from benchbox import ReadPrimitives
from benchbox.platforms.duckdb import DuckDBAdapter
import json

benchmark = ReadPrimitives(scale_factor=0.1)
adapter = DuckDBAdapter()

benchmark.generate_data()
conn = adapter.create_connection()
adapter.create_schema(benchmark, conn)
adapter.load_data(benchmark, conn, benchmark.output_dir)

print("Running baseline benchmark...")
baseline_results = benchmark.run_benchmark(conn, categories=["aggregation", "window"], iterations=3)

with open("baseline_results.json", "w") as f:
    json.dump(baseline_results, f, indent=2)

print("\nRunning comparison benchmark...")
comparison_results = benchmark.run_benchmark(conn, categories=["aggregation", "window"], iterations=3)

print("\nPerformance Comparison:")
print(f"{'Query':<30s} {'Baseline':<12s} {'Current':<12s} {'Change':<10s}")
print("-" * 70)

for query_id, baseline in baseline_results["queries"].items():
    baseline_time = baseline["avg_time"]
    current_time = comparison_results["queries"][query_id]["avg_time"]
    if baseline_time == 0:
        continue
    change_pct = ((current_time - baseline_time) / baseline_time) * 100

    status = "REGRESSION" if change_pct > 10 else "OK"
    print(f"{query_id:<30s} {baseline_time:>10.3f}s  {current_time:>10.3f}s  "
          f"{change_pct:>+7.1f}% {status}")

Between the baseline run and the comparison run, change the database configuration that you want to test. The example as written runs both benchmarks back to back without a change, so it shows only run-to-run variation. A query counts as a regression when its average time rises by more than 10%.

Multi-Category Analysis

from benchbox import ReadPrimitives
from benchbox.platforms.duckdb import DuckDBAdapter
import pandas as pd

benchmark = ReadPrimitives(scale_factor=0.1)
adapter = DuckDBAdapter()

benchmark.generate_data()
conn = adapter.create_connection()
adapter.create_schema(benchmark, conn)
adapter.load_data(benchmark, conn, benchmark.output_dir)

categories_to_test = ["aggregation", "window", "filter", "string"]

results_data = []

for category in categories_to_test:
    print(f"\nTesting {category} category...")
    results = benchmark.run_category_benchmark(conn, category, iterations=3)

    queries = list(results["queries"].values())
    successful = [q for q in queries if q["avg_time"] > 0]
    total_time = sum(q["avg_time"] for q in successful)

    results_data.append({
        "category": category,
        "total_time": total_time,
        "avg_time": total_time / len(successful) if successful else 0,
        "query_count": len(queries),
        "successful": len(successful),
        "failed": len(queries) - len(successful),
        "success_rate": (len(successful) / len(queries)) * 100
    })

df = pd.DataFrame(results_data)

print("\nCategory Analysis:")
print(df.to_string(index=False))

df_sorted = df.sort_values("avg_time", ascending=False)
print("\nSlowest categories:")
for idx, row in df_sorted.head(3).iterrows():
    print(f"  {row['category']}: {row['avg_time']:.3f}s average")

Multi-Platform Comparison

Run the same generated data through several adapters and compare the result objects. This example uses two DuckDB configurations; add other adapters to the dictionary the same way.

from benchbox import ReadPrimitives
from benchbox.platforms.duckdb import DuckDBAdapter
import pandas as pd

benchmark = ReadPrimitives(scale_factor=0.1, output_dir="./data/primitives")
benchmark.generate_data()

platforms = {
    "DuckDB 1GB": DuckDBAdapter(memory_limit="1GB"),
    "DuckDB 4GB": DuckDBAdapter(memory_limit="4GB"),
}

results_data = []

for name, adapter in platforms.items():
    print(f"\nBenchmarking {name}...")
    results = benchmark.run_with_platform(adapter)

    results_data.append({
        "platform": name,
        "total_time": results.total_execution_time,
        "avg_query_time": results.average_query_time,
        "successful": results.successful_queries,
        "failed": results.failed_queries,
    })

df = pd.DataFrame(results_data)
print("\nPlatform Comparison:")
print(df)

Custom Primitive Development

from benchbox import ReadPrimitives

class CustomPrimitives(ReadPrimitives):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self._custom_queries = {
            "custom_complex_join": """
                SELECT l.l_orderkey, o.o_custkey, c.c_name, p.p_name
                FROM lineitem l
                JOIN orders o ON l.l_orderkey = o.o_orderkey
                JOIN customer c ON o.o_custkey = c.c_custkey
                JOIN part p ON l.l_partkey = p.p_partkey
                WHERE l.l_shipdate > '1995-01-01'
                LIMIT 100
            """,
            "custom_window_aggregation": """
                SELECT
                    l_orderkey,
                    l_partkey,
                    l_quantity,
                    SUM(l_quantity) OVER (PARTITION BY l_partkey) as total_qty,
                    AVG(l_quantity) OVER (PARTITION BY l_partkey) as avg_qty
                FROM lineitem
                WHERE l_shipdate > '1995-01-01'
                LIMIT 1000
            """
        }

    def get_query(self, query_id, *, params=None):
        if query_id in self._custom_queries:
            return self._custom_queries[query_id]

        return super().get_query(query_id, params=params)

custom_bench = CustomPrimitives(scale_factor=0.1)
custom_bench.generate_data()

custom_query = custom_bench.get_query("custom_complex_join")
print(f"Custom query: {custom_query}")

The subclass adds custom queries and checks them first in get_query. Any other query ID falls back to the standard primitives. Only get_query() sees the custom queries. execute_query(), get_queries() and run_benchmark() use the implementation’s own query set, so they do not know the custom names.

DataFrame Support

Read Primitives has DataFrame implementations for 152 of its 157 queries, defined in benchbox.core.read_primitives.dataframe_queries (not one of the listed public symbols, so its layout may change).

Query Support:

Skipped queries: 5 queries have no DataFrame equivalent. The module lists them in SKIP_FOR_DATAFRAME.

from benchbox.core.read_primitives.dataframe_queries import SKIP_FOR_DATAFRAME

print(SKIP_FOR_DATAFRAME)