Skip to content

Search docs and blog

Adding New Platform Adapters

This guide provides step-by-step instructions for implementing new platform adapters in BenchBox. Platform adapters enable BenchBox to run benchmarks on different database systems.

Before opening a PR, run through the New Platform Acceptance Checklist. It covers support_status, registry metadata, optional dependencies, SQL compatibility governance, docs, tests, and UAT scope.

Overview

Platform adapters in BenchBox follow a consistent architecture based on the PlatformAdapter base class. Each adapter provides:

  • Database connection management
  • Schema creation and data loading
  • Query execution with dialect translation
  • Performance tuning and optimization
  • Platform-specific features integration

Platform Manifest Authority

benchbox/core/platform_manifest.py is the only static authority for platform identity and metadata. Each typed manifest entry owns its canonical key, every scoped alias and its implied execution mode, support status, execution/deployment capabilities, and the optional adapter module/class/order coordinates. The module uses only the Python standard library; reading catalog metadata must not import an optional SDK or instantiate an adapter.

The remaining platform-shaped maps have narrower runtime ownership:

  • PlatformRegistry derives metadata, only registry-scoped compatibility aliases, and lazy adapter imports from the manifest, then owns loaded adapter classes and availability state. Its public adapter list preserves the manifest’s explicit registration_order.
  • CLI platform normalization derives only CLI-scoped aliases. A -df spelling must also declare implied_mode: dataframe; it is not automatically a core registry alias.
  • platform hooks, local DataFrame factories, tuning aliases, and file-format capabilities describe behavior rather than platform identity. They remain local to their subsystem, but make platform-manifest-check rejects keys or aliases that do not resolve through the manifest.
  • benchbox.platforms lazy exports are API exposure. They are checked against manifest runtime classes or the DataFrame factory, not treated as another registry.
  • Every DataFrame factory row is checked for a CLI-scoped, DataFrame-implying spelling, DataFrame-capable canonical target, real lazy adapter export, real lazy availability constant, and nonempty installation guidance.

Do not add a second canonical-key, alias, capability, support-status, or adapter-coordinate registry. In particular, decorator-based discovery was not introduced: it would require importing adapter modules to discover static metadata and would weaken the lazy optional-dependency boundary.

benchbox.platforms.manifest remains a compatibility import path that only re-exports the core-owned authority. New internal code should import benchbox.core.platform_manifest directly so the enforced utils < core < platforms < cli dependency direction remains intact.

The following inventory is generated from the manifest:

Canonical keyScoped aliasesModesSupportRuntime adapter order and coordinate
athena—SQLbeta#32 benchbox.platforms.athena:AthenaAdapter
athena-spark—SQL + DataFrameexperimental#35 benchbox.platforms.aws:AthenaSparkAdapter
bigquerybq (cli), gbq (cli)SQLbeta#12 benchbox.platforms.bigquery:BigQueryAdapter
cedardb—SQLexperimental#23 benchbox.platforms.cedardb:CedarDBAdapter
citus—SQLexperimental#48 benchbox.platforms.citus:CitusAdapter
clickhouse—SQLdeprecated#6 benchbox.platforms.clickhouse:ClickHouseAdapter
clickhouse-cloud—SQLbeta#9 benchbox.platforms.clickhouse_cloud:ClickHouseCloudAdapter
clickhouse-localch (cli)SQLbeta#7 benchbox.platforms.clickhouse_local:ClickHouseLocalAdapter
clickhouse-server—SQLbeta#8 benchbox.platforms.clickhouse_server:ClickHouseServerAdapter
cudfcudf-df (cli, mode=dataframe)DataFrameexperimentalDataFrame factory only
daskdask-df (cli, mode=dataframe)DataFramebetaDataFrame factory only
databend—SQLbeta#27 benchbox.platforms.databend:DatabendAdapter
databricksdbx (cli)SQL + DataFramebeta#4 benchbox.platforms.databricks:DatabricksAdapter
databricks-df—SQL + DataFrameexperimental#5 benchbox.platforms.databricks:DatabricksDataFrameAdapter
datafusiondatafusion-df (cli, mode=dataframe), fusion (cli)SQL + DataFramestable#3 benchbox.platforms.datafusion:DataFusionAdapter
dataproc—SQL + DataFrameexperimental#36 benchbox.platforms.gcp:DataprocAdapter
dataproc-serverless—SQL + DataFrameexperimental#37 benchbox.platforms.gcp:DataprocServerlessAdapter
doris—SQLbeta#28 benchbox.platforms.doris:DorisAdapter
duckdbduck (cli)SQLstable#0 benchbox.platforms.duckdb:DuckDBAdapter
ducklake—SQLbeta#2 benchbox.platforms.ducklake:DuckLakeAdapter
emr-serverless—SQL + DataFrameexperimental#34 benchbox.platforms.aws:EMRServerlessAdapter
fabric-lakehousefabric_lakehouse (registry)SQLbeta#39 benchbox.platforms.fabric_lakehouse:FabricLakehouseAdapter
fabric-spark—SQL + DataFrameexperimental#38 benchbox.platforms.azure:FabricSparkAdapter
fabric_dwfabric-dw (cli/registry)SQLbeta#31 benchbox.platforms.fabric_warehouse:FabricWarehouseAdapter
firebolt—SQLbeta#26 benchbox.platforms.firebolt:FireboltAdapter
glue—SQL + DataFrameexperimental#33 benchbox.platforms.aws:AWSGlueAdapter
influxdb—SQLbeta#30 benchbox.platforms.influxdb:InfluxDBAdapter
lakesaillakesail-df (cli, mode=dataframe)SQL + DataFrameexperimental#42 benchbox.platforms.lakesail:LakeSailAdapter
motherduck—SQLbeta#1 benchbox.platforms.motherduck:MotherDuckAdapter
pandaspandas-df (cli, mode=dataframe)DataFramestableDataFrame factory only
paradedb—SQLexperimental#47 benchbox.platforms.paradedb:ParadeDBAdapter
pg-duckdbpg_duckdb (registry)SQLexperimental#20 benchbox.platforms.pg_duckdb:PgDuckDBAdapter
pg-mooncake—SQLexperimental#21 benchbox.platforms.pg_mooncake:PgMooncakeAdapter
polarspolars-df (cli, mode=dataframe)DataFramestable#44 benchbox.platforms.polars_platform:PolarsAdapter
postgresqlpg (cli), pgsql (cli), postgres (cli)SQLbeta#18 benchbox.platforms.postgresql:PostgreSQLAdapter
prestoprestodb (cli)SQLbeta#17 benchbox.platforms.presto:PrestoAdapter
pysparkpyspark-df (cli, mode=dataframe)SQL + DataFramebeta#25 benchbox.platforms.pyspark:PySparkSQLAdapter
quanton—SQL + DataFrameexperimental#46 benchbox.platforms.onehouse:QuantonAdapter
questdb—SQLbeta#22 benchbox.platforms.questdb:QuestDBAdapter
redshiftrs (cli)SQLbeta#13 benchbox.platforms.redshift:RedshiftAdapter
singlestore—SQLbeta#29 benchbox.platforms.singlestore:SingleStoreAdapter
snowflakesnow (cli)SQLbeta#14 benchbox.platforms.snowflake:SnowflakeAdapter
snowpark-connect—SQL + DataFrameexperimental#45 benchbox.platforms.snowpark_connect:SnowparkConnectAdapter
spark—SQLbeta#41 benchbox.platforms.spark:SparkAdapter
sqlitesqlite3 (registry)SQLstable#11 benchbox.platforms.sqlite:SQLiteAdapter
starburst—SQLbeta#16 benchbox.platforms.starburst:StarburstAdapter
starrocks—SQLbeta#10 benchbox.platforms.starrocks:StarRocksAdapter
synapseazure-synapse (cli), azure_synapse (registry), azuresynapse (cli)SQLbeta#24 benchbox.platforms.azure_synapse:AzureSynapseAdapter
synapse-spark—SQL + DataFrameexperimental#40 benchbox.platforms.azure:SynapseSparkAdapter
timescaledb—SQLbeta#19 benchbox.platforms.timescaledb:TimescaleDBAdapter
trinotrinodb (cli)SQLbeta#15 benchbox.platforms.trino:TrinoAdapter
velox—SQLexperimental#43 benchbox.platforms.velox:VeloxAdapter

Prerequisites

Before implementing a new platform adapter, ensure you have:

  • Understanding of the target database platform
  • Access to the platform for testing
  • Python client library implementing DB API 2.0 specification (see DB API 2.0 documentation)
  • Familiarity with BenchBox architecture
  • Knowledge of SQL dialect differences

DB API 2.0 Requirement

BenchBox platform selection is based on the availability of a robust Python library using DB API 2.0 (PEP 249). This standardized interface enables universal database access and simplifies platform integration. See the DB API 2.0 documentation for detailed information on:

  • Why DB API 2.0 is fundamental to BenchBox
  • How BenchBox uses DB API 2.0 protocols and patterns
  • Platform compliance levels and requirements
  • Testing and validation procedures

Architecture Overview

Base Class Structure

All platform adapters inherit from PlatformAdapter in benchbox/platforms/base/:

from benchbox.platforms.base import PlatformAdapter


class NewDatabaseAdapter(PlatformAdapter):


    @property
    def platform_name(self) -> str:
        return "NewDatabase"


    def get_target_dialect(self) -> str:
        return "newdatabase"

get_target_dialect() returns the SQL dialect identifier. This snippet is abridged: a real adapter implements all required abstract methods.

Lifecycle and Run-Scoped State

BenchBox treats PlatformAdapter instances as serial execution objects. Reusing one adapter instance for multiple benchmark runs is supported only sequentially; concurrent run_benchmark() calls on the same adapter instance are not supported.

The base adapter resets its run-scoped caches at the start of each run, including database-reuse decisions, cached power/throughput phase results, sorted-ingestion metadata, plan-capture counters, and captured SQL buffers when explicit dry-run mode is active. Run-config plan-capture overrides (capture_plans, strict_plan_capture, and plan_capture_timeout_seconds) are restored after the run.

Adapter authors should keep connection handles, phase-result caches, plan-capture counters, skipped-query tracking, and dry-run SQL capture out of class-level state. Store them on the adapter only when they are truly adapter-scoped and can be reset or overwritten before each run. If an adapter needs concurrent execution, design a per-run context first and update the public contract map before claiming support.

Required Dependencies

Add platform dependencies to pyproject.toml via the [project.optional-dependencies] table:

[project.optional-dependencies]
newdatabase = ["newdatabase-python>=2.0.0"]
all = ["newdatabase-python>=2.0.0", "benchbox[cloud]"]

This allows contributors to install the adapter with uv pip install "benchbox[newdatabase]".

Step-by-Step Implementation

Step 1: Create Adapter File

Create a new file: benchbox/platforms/newdatabase.py

import time
import logging
from pathlib import Path
from typing import Any, Dict, Tuple, Optional


from .base import DriverIsolationCapability, PlatformAdapter, ConnectionConfig


try:
    import newdatabase
    from newdatabase import Connection
    from newdatabase.errors import DatabaseError
except ImportError:
    newdatabase = None
    Connection = None
    DatabaseError = Exception


logger = logging.getLogger(__name__)


class NewDatabaseAdapter(PlatformAdapter):


    driver_isolation_capability = DriverIsolationCapability.FEASIBLE_CLIENT_ONLY


    def __init__(self, **config):
        super().__init__(**config)
        if newdatabase is None:
            raise ImportError("NewDatabase client not installed. Install with: uv add newdatabase-python")


        self.host = config.get('host', 'localhost')
        self.port = config.get('port', 5432)
        self.database = config.get('database', 'benchbox')
        self.username = config.get('username', 'user')
        self.password = config.get('password', '')


        self.connection_pool_size = config.get('connection_pool_size', 5)
        self.query_timeout = config.get('query_timeout', 300)


    @property
    def platform_name(self) -> str:
        return "NewDatabase"


    def get_target_dialect(self) -> str:
        return "newdatabase"

Declaring driver_isolation_capability is required, and CI tests validate it. Choose one of:

  • SUPPORTED: full driver isolation (DuckDB, DataFusion).
  • FEASIBLE_CLIENT_ONLY: the client can be isolated, but the engine version is external.
  • NOT_FEASIBLE: technical constraints prevent isolation (JVM, C libraries).
  • NOT_APPLICABLE: no versioned driver package (for example SQLite and DataFrames).

The remaining required methods are implemented in the following steps.

Step 2: Implement Connection Management

Important: Ensure the returned connection object is DB API 2.0 compliant. It must support either:

  1. Standard cursor pattern: connection.cursor() returning a cursor with execute() method
  2. Direct execute pattern: connection.execute() method available directly

See DB API 2.0 documentation for details on both patterns.

The example below creates the connection, checks existing databases, and tests the connection with the standard cursor pattern. check_server_database_exists uses an admin connection that does not specify a database.

def create_connection(self, **connection_config) -> Any:
    self.handle_existing_database(**connection_config)


    host = connection_config.get('host', self.host)
    port = connection_config.get('port', self.port)
    database = connection_config.get('database', self.database)
    username = connection_config.get('username', self.username)
    password = connection_config.get('password', self.password)


    try:
        connection = newdatabase.connect(
            host=host,
            port=port,
            database=database,
            username=username,
            password=password,
            timeout=self.query_timeout
        )


        cursor = connection.cursor()
        cursor.execute("SELECT 1")
        cursor.fetchall()
        cursor.close()


        logger.info(f"Connected to NewDatabase at {host}:{port}")
        return connection


    except Exception as e:
        logger.error(f"Failed to connect to NewDatabase: {e}")
        raise


def check_server_database_exists(self, **connection_config) -> bool:
    try:
        admin_connection = newdatabase.connect(
            host=connection_config.get('host', self.host),
            port=connection_config.get('port', self.port),
            username=connection_config.get('username', self.username),
            password=connection_config.get('password', self.password)
        )


        database = connection_config.get('database', self.database)


        cursor = admin_connection.cursor()
        cursor.execute("SHOW DATABASES")
        databases = [row[0] for row in cursor.fetchall()]


        return database in databases


    except Exception:
        return False
    finally:
        if 'admin_connection' in locals() and admin_connection:
            admin_connection.close()


def drop_database(self, **connection_config) -> None:
    try:
        admin_connection = newdatabase.connect(
            host=connection_config.get('host', self.host),
            port=connection_config.get('port', self.port),
            username=connection_config.get('username', self.username),
            password=connection_config.get('password', self.password)
        )


        database = connection_config.get('database', self.database)


        cursor = admin_connection.cursor()
        cursor.execute(f"DROP DATABASE IF EXISTS {database}")
        cursor.close()


    except Exception as e:
        raise RuntimeError(f"Failed to drop NewDatabase database: {e}")
    finally:
        if 'admin_connection' in locals() and admin_connection:
            admin_connection.close()

Step 3: Implement Schema Creation

def create_schema(
    self,
    benchmark,
    connection: Any,
    enable_primary_keys: bool = True,
    enable_foreign_keys: bool = True
) -> float:
    start_time = time.time()


    try:
        schema_sql = benchmark.get_create_tables_sql(
            enable_primary_keys=enable_primary_keys,
            enable_foreign_keys=enable_foreign_keys
        )


        if hasattr(self, 'translate_sql'):
            schema_sql = self.translate_sql(schema_sql, "duckdb")


        statements = [stmt.strip() for stmt in schema_sql.split(';') if stmt.strip()]


        cursor = connection.cursor()
        for statement in statements:
            statement = self._optimize_table_definition(statement)
            cursor.execute(statement)
            logger.debug(f"Executed schema statement: {statement[:100]}...")


        cursor.close()
        connection.commit()


        logger.info(f"Schema created (PKs: {enable_primary_keys}, FKs: {enable_foreign_keys})")


    except Exception as e:
        logger.error(f"Schema creation failed: {e}")
        raise


    return time.time() - start_time


def _optimize_table_definition(self, statement: str) -> str:
    if not statement.upper().startswith('CREATE TABLE'):
        return statement


    if 'ENGINE' not in statement.upper():
        statement += " ENGINE=InnoDB"


    return statement

The translate_sql(schema_sql, "duckdb") call translates from the DuckDB dialect. _optimize_table_definition is an example hook for adding a storage engine or other platform-specific options. The ENGINE=InnoDB option shown is an example for MySQL-like databases.

Step 4: Implement Data Loading

def load_data(self, benchmark, connection: Any, data_dir: Path) -> Tuple[Dict[str, int], float]:
    start_time = time.time()
    table_stats = {}


    if hasattr(benchmark, 'tables') and benchmark.tables:
        data_files = benchmark.tables
    else:
        raise ValueError("No data files found. Ensure benchmark.generate_data() was called first.")


    cursor = connection.cursor()


    for table_name, file_path in data_files.items():
        file_path = Path(file_path)
        if not file_path.exists() or file_path.stat().st_size == 0:
            logger.warning(f"Skipping {table_name} - no data file or empty file")
            table_stats[table_name] = 0
            continue


        try:
            load_start = time.time()
            table_name_upper = table_name.upper()


            if self._supports_bulk_copy():
                copy_command = self._build_copy_command(table_name_upper, file_path)
                cursor.execute(copy_command)
            else:
                self._load_via_inserts(cursor, table_name_upper, file_path)


            cursor.execute(f"SELECT COUNT(*) FROM {table_name_upper}")
            row_count = cursor.fetchone()[0]
            table_stats[table_name_upper] = row_count


            load_time = time.time() - load_start
            logger.info(f"✅ Loaded {row_count:,} rows into {table_name_upper} in {load_time:.2f}s")


        except Exception as e:
            logger.error(f"Failed to load {table_name}: {str(e)[:100]}...")
            table_stats[table_name.upper()] = 0


    cursor.close()
    connection.commit()


    total_time = time.time() - start_time
    total_rows = sum(table_stats.values())
    logger.info(f"✅ Loaded {total_rows:,} total rows in {total_time:.2f}s")


    return table_stats, total_time


def _supports_bulk_copy(self) -> bool:
    return True


def _build_copy_command(self, table_name: str, file_path: Path) -> str:
    delimiter = '|' if file_path.suffix == '.tbl' else ','
    return f"""
        COPY {table_name} FROM '{file_path}'
        WITH (FORMAT CSV, DELIMITER '{delimiter}', HEADER FALSE)
    """


def _load_via_inserts(self, cursor, table_name: str, file_path: Path):
    delimiter = '|' if file_path.suffix == '.tbl' else ','


    with open(file_path, 'r') as f:
        batch = []
        batch_size = 1000


        for line in f:
            line = line.strip()
            if line.endswith(delimiter):
                line = line[:-1]


            values = line.split(delimiter)
            placeholders = ','.join(['?' for _ in values])


            if not batch:
                insert_sql = f"INSERT INTO {table_name} VALUES ({placeholders})"


            batch.append(tuple(values))


            if len(batch) >= batch_size:
                cursor.executemany(insert_sql, batch)
                batch = []


        if batch:
            cursor.executemany(insert_sql, batch)

load_data uses the platform’s bulk loading method when _supports_bulk_copy() is true (a COPY command or equivalent), and falls back to INSERT statements otherwise. The return True in _supports_bulk_copy is a placeholder: implement it based on platform capabilities.

Step 5: Implement Query Execution

Important: Use standard DB API 2.0 cursor methods for query execution. BenchBox’s DatabaseConnection wrapper handles both cursor patterns automatically, but adapter implementations should follow DB API 2.0 conventions.

execute_query Variant Families (2026-02)

execute_query is not one universal flow across SQL platforms. Use the family model below before deciding to consolidate logic.

FamilyRepresentative adaptersCore flowConsolidation status
DBAPI cursor + validation helperfirebolt, presto, trinocursor.execute() -> fetchall() -> optional QueryValidator row-count check -> standardized result payloadConsolidated in CursorValidationQueryExecutionMixin (benchbox/core/benchmark_mixins.py)
Spark SQL collectspark, lakesailspark.sql(query) -> .collect() -> optional row-count validation -> standardized result payloadSimilar enough for shared helpers; keep session/config differences adapter-local
Cloud Spark submit/wait/fetch jobemr_serverless, dataproc, dataproc_serverless, athena_spark, glue, quantonSubmit remote job/batch -> poll async state -> fetch JSON results from object storage or APIIntentionally distinct from DBAPI cursor flow; async control plane and artifact retrieval dominate behavior
Livy statement APIsynapse_spark, fabric_sparkSubmit statement to Livy session -> poll statement state -> parse data.values payloadKeep as a separate family; may share Livy polling/parsing utilities

Engine-specific outliers should remain separate unless their full contracts are aligned:

  • bigquery: query jobs with bytes billed/slot metrics and table qualification.
  • datafusion: SessionContext.sql().collect() plus SQL compatibility rewrite pipeline.
  • databend: explicit USE <database> and query_iter result streaming.
  • motherduck: legacy tuple-return API over DuckDB (rows, execution_time).
  • influxdb: time-series-specific tuple return (execution_time, row_count, query_plan).

Rule of thumb:

  • Consolidate only within the same family and same output contract.
  • Keep different transport/protocol families distinct, even if control flow looks superficially similar.
def execute_query(self, connection: Any, query: str, query_id: str) -> Dict[str, Any]:
    start_time = time.time()


    try:
        cursor = connection.cursor()


        self._apply_query_optimizations(cursor)


        cursor.execute(query)


        results = cursor.fetchall()


        execution_time = time.time() - start_time


        query_metrics = self._get_query_metrics(cursor)


        cursor.close()


        return {
            'query_id': query_id,
            'status': 'SUCCESS',
            'execution_time': execution_time,
            'rows_returned': len(results),
            'first_row': results[0] if results else None,
            'platform_metrics': query_metrics
        }


    except Exception as e:
        execution_time = time.time() - start_time


        return {
            'query_id': query_id,
            'status': 'FAILED',
            'execution_time': execution_time,
            'rows_returned': 0,
            'error': str(e),
            'error_type': type(e).__name__
        }


def _apply_query_optimizations(self, cursor):
    cursor.execute("SET query_cache = ON")
    cursor.execute("SET optimizer_mode = 'performance'")


def _get_query_metrics(self, cursor) -> Dict[str, Any]:
    try:
        cursor.execute("SHOW QUERY STATS")
        stats = cursor.fetchall()
        return {'query_stats': stats}
    except:
        return {}


def configure_for_benchmark(self, connection: Any, benchmark_type: str) -> None:
    cursor = connection.cursor()


    if benchmark_type.lower() in ['olap', 'analytics', 'tpch', 'tpcds']:
        cursor.execute("SET join_algorithm = 'hash'")
        cursor.execute("SET parallel_workers = 8")
        cursor.execute("SET work_mem = '256MB'")
    elif benchmark_type.lower() in ['oltp', 'transactional']:
        cursor.execute("SET synchronous_commit = ON")
        cursor.execute("SET random_page_cost = 1.1")


    cursor.close()

The first branch applies OLAP optimizations. The second applies OLTP optimizations.

Step 6: Implement Platform Metadata

def _get_platform_metadata(self, connection: Any) -> Dict[str, Any]:
    metadata = {
        "platform": self.platform_name,
        "host": self.host,
        "port": self.port,
        "database": self.database
    }


    try:
        cursor = connection.cursor()


        cursor.execute("SELECT VERSION()")
        version_result = cursor.fetchone()
        metadata["version"] = version_result[0] if version_result else "unknown"


        cursor.execute("SHOW VARIABLES LIKE 'max_connections'")
        settings = cursor.fetchall()
        metadata["settings"] = {name: value for name, value in settings}


        cursor.execute("""
            SELECT
                table_name,
                table_rows,
                data_length,
                index_length
            FROM information_schema.tables
            WHERE table_schema = DATABASE()
        """)


        tables = cursor.fetchall()
        metadata["tables"] = [
            {
                "name": table[0],
                "rows": table[1],
                "data_size": table[2],
                "index_size": table[3]
            } for table in tables
        ]


        cursor.close()


    except Exception as e:
        metadata["metadata_error"] = str(e)


    return metadata

Step 7: Implement Performance Tuning (Optional)

In the example, SORTING is supported through indexes, CLUSTERING through clustered indexes, and PARTITIONING through table partitioning.

def supports_tuning_type(self, tuning_type) -> bool:
    try:
        from benchbox.core.tuning.interface import TuningType
        supported_types = {
            TuningType.SORTING,
            TuningType.CLUSTERING,
            TuningType.PARTITIONING
        }
        return tuning_type in supported_types
    except ImportError:
        return False


def generate_tuning_clause(self, table_tuning) -> str:
    if not table_tuning or not table_tuning.has_any_tuning():
        return ""


    clauses = []


    try:
        from benchbox.core.tuning.interface import TuningType


        partition_columns = table_tuning.get_columns_by_type(TuningType.PARTITIONING)
        if partition_columns:
            sorted_cols = sorted(partition_columns, key=lambda col: col.order)
            partition_col = sorted_cols[0].name
            clauses.append(f"PARTITION BY HASH({partition_col})")


        cluster_columns = table_tuning.get_columns_by_type(TuningType.CLUSTERING)
        if cluster_columns:
            sorted_cols = sorted(cluster_columns, key=lambda col: col.order)
            column_names = [col.name for col in sorted_cols]
            clauses.append(f"CLUSTERED INDEX ({', '.join(column_names)})")


    except ImportError:
        pass


    return " ".join(clauses)


def apply_table_tunings(self, table_tuning, connection: Any) -> None:
    if not table_tuning or not table_tuning.has_any_tuning():
        return


    table_name = table_tuning.table_name
    cursor = connection.cursor()


    try:
        from benchbox.core.tuning.interface import TuningType


        sort_columns = table_tuning.get_columns_by_type(TuningType.SORTING)
        if sort_columns:
            sorted_cols = sorted(sort_columns, key=lambda col: col.order)
            column_names = [col.name for col in sorted_cols]


            index_name = f"idx_{table_name.lower()}_sort"
            index_sql = f"CREATE INDEX {index_name} ON {table_name} ({', '.join(column_names)})"


            cursor.execute(index_sql)
            logger.info(f"Created sort index on {table_name}: {', '.join(column_names)}")


        cursor.close()
        connection.commit()


    except Exception as e:
        logger.error(f"Failed to apply tunings to {table_name}: {e}")
        raise

Step 8: Add Power/Throughput Tests (Optional)

def run_power_test(self, benchmark, **kwargs) -> Dict[str, Any]:
    return self.run_benchmark(benchmark, **kwargs).__dict__


def run_throughput_test(self, benchmark, **kwargs) -> Dict[str, Any]:
    return self.run_power_test(benchmark, **kwargs)


def run_maintenance_test(self, benchmark, **kwargs) -> Dict[str, Any]:
    return {"status": "NOT_IMPLEMENTED", "message": "Maintenance test not implemented"}

The throughput test runs as a single stream for now. Extend it for true multi-stream execution later.

Integration Steps

Step 1: Register the Adapter

Add one typed entry to the manifest source of truth, benchbox/core/platform_manifest.py. The entry below is abridged and JSON-shaped. A real entry also carries display metadata, libraries, requirements, and adoption fields. Set registration_order to the next contiguous value, and do not reorder existing adapters:

{
    "key": "newdatabase",
    "aliases": [
        {"name": "newdb", "scopes": ["cli"]},
    ],
    "adapter": {
        "module": "benchbox.platforms.newdatabase",
        "class_name": "NewDatabaseAdapter",
        "registration_order": 47,
    },
    "support_status": "experimental",
    "capabilities": {
        "supports_sql": True,
        "supports_dataframe": False,
        "default_mode": "sql",
    },
}

Do not add parallel entries to PlatformRegistry or the CLI. They derive their metadata, scoped aliases, and runtime import rows from the manifest. Use registry scope only for core API compatibility spellings; use cli for CLI shorthands. If an alias implies DataFrame execution, declare "implied_mode": "dataframe". Preserve existing registration order and assign the next contiguous integer to a new runtime adapter. support_status is a product support classification, not local dependency availability.

If the adapter should be importable from benchbox.platforms, add a lazy entry and __all__ export in benchbox/platforms/__init__.py; keep optional SDK imports guarded so import benchbox does not load the platform dependency.

If the platform has subsystem behavior, add only the relevant local entries: platform option specs/config builders, DataFrame factory routing, tuning capabilities, or format preferences. Those maps do not define platform identity and their keys must resolve through the manifest drift gate.

Before implementation, maintainers can print the expected file plan with the platform scaffold script (--kind sql).

Declare compute and gateway support with the compute and gateway items of the SQL platform checklist.

Regenerate the inventory and run the drift gate after editing the manifest:

Terminal window
make platform-manifest
make platform-manifest-check

Declare execution engines when the platform offers more than one way to execute a query. Add a manifest capabilities.execution_engines block where each value carries class, display_name, description, dependencies and selectable. The class comes from the closed core list (adaptive, in-memory, streaming, gpu, native-vectorized, standard, distributed, jit, delegated); the vocabulary parity test rejects unknown classes and enforces that one engine name maps to one class on every platform. Name each value with the platform’s own documented term when one exists, otherwise the class name. default is a reserved sentinel and is never declared. The worked example is polars-df:

{
    "key": "polars",
    "capabilities": {
        "execution_engines": {
            "auto": {
                "class": "adaptive",
                "display_name": "Auto",
                "description": "Polars chooses the engine for each query",
                "dependencies": ["polars"],
                "selectable": True,
            },
            "in-memory": {
                "class": "in-memory",
                "display_name": "In-memory",
                "description": "Materializes each query in memory",
                "dependencies": ["polars"],
                "selectable": True,
            },
            "streaming": {
                "class": "streaming",
                "display_name": "Streaming",
                "description": "Processes each query in batches to bound memory",
                "dependencies": ["polars"],
                "selectable": True,
            },
        },
    },
}

Adapters inherit resolve_execution_engine from the adapter base and map the resolved engine into their query path; unsupported values fail closed. Never name a canonical option engine, mode, backend, runtime, executor or warehouse; a reserved-word spelling survives only as a deprecated alias registered in the core DEPRECATED_OPTION_ALIASES table with its removal release.

Step 2: Add SQL Dialect Support

If your platform has a unique SQL dialect, add SQL compatibility rules or an explicit exemption under the phase-aware SQL compatibility system. Do not hide CREATE TABLE rewrites inside the adapter without registering or exempting them in the DDL governance inventory. Put the rules in benchbox/sql_compat/rules/ddl_optimize/newdatabase_ddl_rewrites.py:

from benchbox.sql_compat.rules._registration import register_ddl_rewrite


register_ddl_rewrite(
    platform="newdatabase",
    rule_name="convert_to_newdatabase_table",
    transformer_id="newdatabase_convert_to_newdatabase_table",
    description="What the adapter rewrites in CREATE TABLE for this platform.",
    reason="Why this platform needs the rewrite.",
)

Step 3: Create Tests

Create comprehensive tests in tests/unit/platforms/test_newdatabase_adapter.py. The TestNewDatabaseIntegration class at the end needs an actual NewDatabase instance and is skipped when the client library is not installed:

import pytest
from unittest.mock import Mock, patch, MagicMock
from benchbox.platforms.newdatabase import NewDatabaseAdapter


class TestNewDatabaseAdapter:


    @pytest.fixture
    def adapter(self):
        return NewDatabaseAdapter(
            host='localhost',
            port=5432,
            database='test',
            username='test',
            password='test'
        )


    def test_platform_name(self, adapter):
        assert adapter.platform_name == "NewDatabase"


    def test_target_dialect(self, adapter):
        assert adapter.get_target_dialect() == "newdatabase"


    @patch('newdatabase.connect')
    def test_create_connection_success(self, mock_connect, adapter):
        mock_connection = Mock()
        mock_cursor = Mock()
        mock_connection.cursor.return_value = mock_cursor
        mock_connect.return_value = mock_connection


        connection = adapter.create_connection()


        assert connection is mock_connection
        mock_connect.assert_called_once()
        mock_cursor.execute.assert_called_with("SELECT 1")


    @patch('newdatabase.connect')
    def test_create_connection_failure(self, mock_connect, adapter):
        mock_connect.side_effect = Exception("Connection failed")


        with pytest.raises(Exception, match="Connection failed"):
            adapter.create_connection()


    def test_schema_creation(self, adapter):
        mock_benchmark = Mock()
        mock_benchmark.get_create_tables_sql.return_value = """
            CREATE TABLE test_table (id INTEGER, name VARCHAR(50));
            CREATE TABLE test_table2 (id INTEGER, value DECIMAL(10,2));
        """


        mock_connection = Mock()
        mock_cursor = Mock()
        mock_connection.cursor.return_value = mock_cursor


        duration = adapter.create_schema(mock_benchmark, mock_connection)


        assert duration > 0
        assert mock_cursor.execute.call_count == 2
        mock_connection.commit.assert_called_once()


    def test_data_loading(self, adapter, tmp_path):
        test_data_file = tmp_path / "test_table.tbl"
        test_data_file.write_text("1|John Doe\n2|Jane Smith\n")


        mock_benchmark = Mock()
        mock_benchmark.tables = {"test_table": str(test_data_file)}


        mock_connection = Mock()
        mock_cursor = Mock()
        mock_cursor.fetchone.return_value = (2,)
        mock_connection.cursor.return_value = mock_cursor


        with patch.object(adapter, '_supports_bulk_copy', return_value=True):
            with patch.object(adapter, '_build_copy_command', return_value="COPY test_table FROM 'file'"):
                table_stats, load_time = adapter.load_data(mock_benchmark, mock_connection, tmp_path)


        assert table_stats["TEST_TABLE"] == 2
        assert load_time > 0


    def test_query_execution_success(self, adapter):
        mock_connection = Mock()
        mock_cursor = Mock()
        mock_cursor.fetchall.return_value = [(1, 'test'), (2, 'data')]
        mock_connection.cursor.return_value = mock_cursor


        result = adapter.execute_query(mock_connection, "SELECT * FROM test", "Q1")


        assert result['status'] == 'SUCCESS'
        assert result['query_id'] == 'Q1'
        assert result['rows_returned'] == 2
        assert result['execution_time'] > 0


    def test_query_execution_failure(self, adapter):
        mock_connection = Mock()
        mock_cursor = Mock()
        mock_cursor.execute.side_effect = Exception("SQL error")
        mock_connection.cursor.return_value = mock_cursor


        result = adapter.execute_query(mock_connection, "INVALID SQL", "Q1")


        assert result['status'] == 'FAILED'
        assert result['query_id'] == 'Q1'
        assert result['rows_returned'] == 0
        assert 'SQL error' in result['error']


    def test_platform_metadata(self, adapter):
        mock_connection = Mock()
        mock_cursor = Mock()
        mock_cursor.fetchone.return_value = ("NewDatabase 1.0",)
        mock_cursor.fetchall.return_value = [("max_connections", "100")]
        mock_connection.cursor.return_value = mock_cursor


        metadata = adapter._get_platform_metadata(mock_connection)


        assert metadata['platform'] == 'NewDatabase'
        assert metadata['version'] == 'NewDatabase 1.0'
        assert 'settings' in metadata


class TestNewDatabaseIntegration:


    @pytest.fixture
    def integration_adapter(self):
        pytest.importorskip("newdatabase")


        return NewDatabaseAdapter(
            host=os.getenv('NEWDATABASE_HOST', 'localhost'),
            port=int(os.getenv('NEWDATABASE_PORT', '5432')),
            database=os.getenv('NEWDATABASE_DATABASE', 'test'),
            username=os.getenv('NEWDATABASE_USERNAME', 'test'),
            password=os.getenv('NEWDATABASE_PASSWORD', 'test')
        )


    @pytest.mark.integration
    def test_end_to_end_benchmark(self, integration_adapter):
        from benchbox import ReadPrimitives


        benchmark = ReadPrimitives(scale_factor=0.001)
        results = integration_adapter.run_benchmark(benchmark)


        assert results.status == 'SUCCESS'
        assert results.total_time > 0
        assert len(results.query_results) > 0

Step 4: Add Documentation

Create platform documentation following the existing pattern in docs/platforms/newdatabase.md.

Step 5: Update Configuration

Add the new platform to the example configuration files:

platforms:
  newdatabase:
    host: "localhost"
    port: 5432
    database: "benchbox"
    username: "benchbox"
    password: "secure_password"

Testing Your Adapter

Unit Tests

Run unit tests to verify basic functionality:

Terminal window
uv run -- python -m pytest tests/unit/platforms/test_newdatabase_adapter.py -v


uv run -- python -m pytest tests/unit/platforms/ -v

Integration Tests

Test with actual database instance:

Terminal window
export NEWDATABASE_HOST=localhost
export NEWDATABASE_PORT=5432
export NEWDATABASE_DATABASE=test
export NEWDATABASE_USERNAME=test
export NEWDATABASE_PASSWORD=test


uv run -- python -m pytest tests/integration/ -k newdatabase -v

Benchmark Tests

Test with small benchmark:

from benchbox import ReadPrimitives
from benchbox.platforms.newdatabase import NewDatabaseAdapter


benchmark = ReadPrimitives(scale_factor=0.001)
adapter = NewDatabaseAdapter(
    host="localhost",
    database="test",
    username="test",
    password="test"
)


results = adapter.run_benchmark(benchmark)
print(f"Benchmark completed in {results.total_time:.2f}s")

Best Practices

Error Handling

  • Implement comprehensive error handling for connection failures
  • Provide clear error messages with troubleshooting hints
  • Handle platform-specific exceptions gracefully
  • Log errors at appropriate levels

Performance Optimization

  • Use platform-native bulk loading methods when available
  • Implement connection pooling for concurrent workloads
  • Apply platform-specific query optimizations
  • Monitor and report platform-specific metrics

Configuration Management

  • Support environment variable configuration
  • Provide sensible defaults for all parameters
  • Validate configuration parameters early
  • Document all configuration options

Testing Strategy

  • Write comprehensive unit tests with mocking
  • Create integration tests for real database connections
  • Test error conditions and edge cases
  • Validate performance characteristics

Documentation

  • Document all configuration parameters
  • Provide installation and setup instructions
  • Include troubleshooting guides
  • Add usage examples and best practices

Common Patterns

Connection String Parsing

The example parses connection strings of the form newdatabase://user:pass@host:port/database:

def _parse_connection_string(self, conn_str: str) -> Dict[str, str]:
    import re


    pattern = r"newdatabase://(?:([^:]*):([^@]*)@)?([^:]*):(\d+)/(.+)"
    match = re.match(pattern, conn_str)


    if not match:
        raise ValueError(f"Invalid connection string: {conn_str}")


    return {
        'username': match.group(1) or '',
        'password': match.group(2) or '',
        'host': match.group(3),
        'port': int(match.group(4)),
        'database': match.group(5)
    }

Retry Logic

Retry transient failures with exponential backoff (the wait doubles after each attempt):

def _execute_with_retry(self, cursor, query: str, max_retries: int = 3):
    import time


    for attempt in range(max_retries):
        try:
            return cursor.execute(query)
        except TransientError as e:
            if attempt == max_retries - 1:
                raise


            wait_time = 2 ** attempt
            logger.warning(f"Query failed (attempt {attempt + 1}), retrying in {wait_time}s: {e}")
            time.sleep(wait_time)

Resource Management

from contextlib import contextmanager


@contextmanager
def managed_connection(self, **config):
    connection = None
    try:
        connection = self.create_connection(**config)
        yield connection
    finally:
        if connection:
            connection.close()


@contextmanager
def managed_cursor(self, connection):
    cursor = None
    try:
        cursor = connection.cursor()
        yield cursor
    finally:
        if cursor:
            cursor.close()

Advanced Features

Connection Pooling

class PooledNewDatabaseAdapter(NewDatabaseAdapter):


    def __init__(self, **config):
        super().__init__(**config)
        self._connection_pool = None
        self._pool_size = config.get('pool_size', 10)


    def _get_connection_pool(self):
        if self._connection_pool is None:
            from newdatabase.pool import ConnectionPool


            self._connection_pool = ConnectionPool(
                host=self.host,
                port=self.port,
                database=self.database,
                username=self.username,
                password=self.password,
                max_connections=self._pool_size
            )


        return self._connection_pool


    def create_connection(self, **connection_config) -> Any:
        pool = self._get_connection_pool()
        return pool.get_connection()

Async Support

import asyncio
from typing import AsyncGenerator


class AsyncNewDatabaseAdapter(NewDatabaseAdapter):


    async def create_async_connection(self, **config):
        import newdatabase.asyncio as async_newdb


        return await async_newdb.connect(
            host=self.host,
            port=self.port,
            database=self.database,
            username=self.username,
            password=self.password
        )


    async def execute_query_async(self, connection, query: str, query_id: str):
        start_time = time.time()


        try:
            cursor = await connection.cursor()
            await cursor.execute(query)
            results = await cursor.fetchall()


            return {
                'query_id': query_id,
                'status': 'SUCCESS',
                'execution_time': time.time() - start_time,
                'rows_returned': len(results),
                'first_row': results[0] if results else None
            }


        except Exception as e:
            return {
                'query_id': query_id,
                'status': 'FAILED',
                'execution_time': time.time() - start_time,
                'error': str(e)
            }

Troubleshooting

Common Issues

  1. Import Errors: Ensure client library is installed and accessible
  2. Connection Failures: Check network connectivity, credentials, and firewall settings
  3. SQL Compatibility: Verify dialect translation is working correctly
  4. Performance Issues: Check for proper indexing and query optimization
  5. Memory Usage: Monitor memory consumption during large data loads

Debugging Tips

  1. Enable debug logging to trace execution flow
  2. Use database-specific profiling tools
  3. Test with small datasets first
  4. Validate schema creation independently
  5. Check platform-specific documentation for optimization hints

Contributing Your Adapter

When your adapter is complete and tested:

  1. Create a pull request with your implementation
  2. Include comprehensive tests and documentation
  3. Add the platform to the CI/CD pipeline
  4. Update the README with the new platform support
  5. Consider adding examples and tutorials

Your contribution helps make BenchBox more comprehensive and valuable for the analytics community!

Resources

  • Base Platform Adapter: benchbox/platforms/base/ package
  • Existing Adapters: benchbox/platforms/ directory
  • SQL Compatibility: benchbox/sql_compat/ and docs/compat/
  • Test Examples: tests/unit/platforms/
  • Documentation Examples: docs/platforms/
  • Acceptance Checklist: new-platform-acceptance-checklist.md
  • BenchBox Architecture: architecture.md

BenchBox is open source under the MIT License.