diff --git a/CHANGELOG.md b/CHANGELOG.md index c98a971de9..2dc9c63ab2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -291,6 +291,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 mismatched keys are recorded as failed ledger statements and warned. Result metadata reports the live tuned values for shuffle partitions and adaptive execution. +- Tuned load-only runs (`--phases generate,load`) now go through the same + apply → load → reconcile → metadata → ledger/receipt → export lifecycle as + full runs, so a load-only bundle never carries a tuned mode without + `platform.tuning`. Adapters without the full lifecycle refuse tuned load-only + runs with an actionable error instead. - The Atom feed uses canonical `/blog/` self and entry links while retaining stable entry IDs for existing subscribers. - Deprecated option aliases now show their warning. The Polars `streaming` option, `--polars-streaming`, Dask `use_distributed`, pg-duckdb diff --git a/benchbox/core/runner/runner.py b/benchbox/core/runner/runner.py index 8be965e5cb..5e448ff04c 100644 --- a/benchbox/core/runner/runner.py +++ b/benchbox/core/runner/runner.py @@ -487,6 +487,102 @@ def _has_adapter_runtime_metadata(result: BenchmarkResults) -> bool: return has_runtime and has_deployment +def _resolve_load_only_tuning(adapter: object) -> tuple[bool, Any, bool]: + full_lifecycle = callable(getattr(adapter, "_setup_fresh_database_phases", None)) + get_effective_tuning = getattr(adapter, "get_effective_tuning_configuration", None) + effective_tuning_config = get_effective_tuning() if callable(get_effective_tuning) else None + tuned_run = bool(getattr(adapter, "tuning_enabled", False) and effective_tuning_config) + if tuned_run: + if not full_lifecycle: + raise RuntimeError( + f"Load-only mode cannot run a tuned configuration on platform " + f"'{getattr(adapter, 'platform_name', type(adapter).__name__)}': the adapter does not " + "implement the full apply → load → reconcile → export lifecycle. " + "Run the full lifecycle or use --tuning notuning." + ) + tuning_errors, tuning_warnings = effective_tuning_config.validate_for_platform_detailed( + getattr(adapter, "canonical_platform_type", getattr(adapter, "platform_name", "")) + ) + for tuning_warning in tuning_warnings: + logger.warning("Tuning configuration warning: %s", tuning_warning) + if tuning_errors: + raise ValueError(f"Invalid tuning configuration: {'; '.join(tuning_errors)}") + return full_lifecycle, effective_tuning_config, tuned_run + + +def _run_load_only_table_phase( + adapter: object, + benchmark: Any, + connection: Any, + data_dir: Any, + table_mode: str, + full_lifecycle: bool, + effective_tuning_config: Any, +) -> tuple[dict[str, int], float, Any, float, dict[str, Any], str, str | None, bool]: + table_stats: dict[str, int] = {} + load_time = 0.0 + schema_time = 0.0 + per_table_timings = None + tuning_metadata_saved = False + if table_mode == "external": + table_stats, load_time, per_table_timings = as_external_table_loader(adapter).create_external_tables( + benchmark, connection, data_dir + ) + schema_phase = { + "status": "SKIPPED", + "duration_ms": 0, + "reason": "External table mode bypasses native schema+load materialization", + } + return table_stats, load_time, per_table_timings, schema_time, schema_phase, "COMPLETED", None, False + if _adapter_reused_database(adapter): + logger.info("Database being reused - skipping schema creation and data loading (load-only mode)") + table_stats = _collect_reused_table_stats(adapter, benchmark, connection) + schema_phase = { + "status": "SKIPPED", + "duration_ms": 0, + "reason": "Reusing compatible existing database - schema creation skipped", + } + return ( + table_stats, + load_time, + per_table_timings, + schema_time, + schema_phase, + "SKIPPED", + "Reusing compatible existing database - data loading skipped", + False, + ) + if full_lifecycle: + ( + schema_time, + _schema_creation_phase, + load_time, + table_stats, + _data_loading_phase, + tuning_metadata_saved, + ) = adapter._setup_fresh_database_phases(benchmark, connection, effective_tuning_config) + per_table_timings = getattr(adapter, "_last_per_table_timings", None) + else: + native_loader = as_native_table_loader(adapter) + schema_time = native_loader.create_schema(benchmark, connection) + table_stats, load_time, per_table_timings = native_loader.load_data(benchmark, connection, data_dir) + require_loaded_tables(benchmark, table_stats) + schema_phase = { + "status": "COMPLETED", + "duration_ms": int(schema_time * 1000), + } + return ( + table_stats, + load_time, + per_table_timings, + schema_time, + schema_phase, + "COMPLETED", + None, + tuning_metadata_saved, + ) + + def _execute_load_only_mode( *, benchmark: Any, @@ -506,6 +602,7 @@ def _execute_load_only_mode( connection_lifecycle = as_connection_lifecycle(adapter) setattr(adapter, "_existing_db_decided", False) # noqa: B010 + full_lifecycle, effective_tuning_config, tuned_run = _resolve_load_only_tuning(adapter) try: connection = connection_lifecycle.create_connection(**(platform_config or {})) @@ -524,41 +621,20 @@ def _execute_load_only_mode( data_dir = _resolve_output_dir_handler(benchmark, None) if data_dir is None: raise RuntimeError("Benchmark output directory not configured; cannot perform load-only operations") - - if table_mode == "external": - table_stats, load_time, per_table_timings = as_external_table_loader(adapter).create_external_tables( - benchmark, connection, data_dir - ) - schema_phase = { - "status": "SKIPPED", - "duration_ms": 0, - "reason": "External table mode bypasses native schema+load materialization", - } - data_loading_phase_status = "COMPLETED" - data_loading_phase_reason: str | None = None - elif _adapter_reused_database(adapter): - logger.info("Database being reused - skipping schema creation and data loading (load-only mode)") - table_stats = _collect_reused_table_stats(adapter, benchmark, connection) - load_time = 0.0 - per_table_timings = None - schema_phase = { - "status": "SKIPPED", - "duration_ms": 0, - "reason": "Reusing compatible existing database - schema creation skipped", - } - data_loading_phase_status = "SKIPPED" - data_loading_phase_reason = "Reusing compatible existing database - data loading skipped" - else: - native_loader = as_native_table_loader(adapter) - schema_time = native_loader.create_schema(benchmark, connection) - table_stats, load_time, per_table_timings = native_loader.load_data(benchmark, connection, data_dir) - require_loaded_tables(benchmark, table_stats) - schema_phase = { - "status": "COMPLETED", - "duration_ms": int(schema_time * 1000), - } - data_loading_phase_status = "COMPLETED" - data_loading_phase_reason = None + tuning_metadata_saved = False + + ( + table_stats, + load_time, + per_table_timings, + schema_time, + schema_phase, + data_loading_phase_status, + data_loading_phase_reason, + tuning_metadata_saved, + ) = _run_load_only_table_phase( + adapter, benchmark, connection, data_dir, table_mode, full_lifecycle, effective_tuning_config + ) if validation_opts.enable_postload_validation: if hasattr(benchmark, "validate_loaded_data"): @@ -579,6 +655,25 @@ def _execute_load_only_mode( } if data_loading_phase_reason is not None: data_loading_phase["reason"] = data_loading_phase_reason + + tunings_applied_dict = None + requested_config_hash = None + tuning_validation_status = None + applied_ledger_payload = None + applied_ledger_hash = None + if full_lifecycle: + ( + tunings_applied_dict, + requested_config_hash, + tuning_validation_status, + applied_ledger_payload, + applied_ledger_hash, + ) = adapter._export_load_only_tuning( + connection, + getattr(benchmark_config, "benchmark_type", "olap"), + effective_tuning_config, + tuned_run, + ) phases = { "data_generation": {"status": "COMPLETED"}, "schema_creation": schema_phase, @@ -629,6 +724,14 @@ def _execute_load_only_mode( per_table_timings=per_table_timings, total_rows_loaded=total_rows, data_size_mb=data_size_mb, + tunings_applied=tunings_applied_dict, + tuning_validation_status=tuning_validation_status, + tuning_metadata_saved=tuning_metadata_saved, + tuning_config_hash=requested_config_hash, + applied_tuning_ledger=applied_ledger_payload, + applied_ledger_hash=applied_ledger_hash, + tuning_source_file=getattr(adapter, "tuning_source_file", None), + tuning_source=getattr(adapter, "tuning_source", None), ) return result_obj, postload_result finally: diff --git a/benchbox/platforms/base/adapter.py b/benchbox/platforms/base/adapter.py index c1d48e4de1..00ed510d2d 100644 --- a/benchbox/platforms/base/adapter.py +++ b/benchbox/platforms/base/adapter.py @@ -879,6 +879,35 @@ def _corroborate_applied_ledger(self, connection: Any, status: str) -> tuple[str def _attach_applied_ledger_payload(self, result: Any, status: str) -> None: tuning_trust.attach_applied_ledger_payload(self, result, status) + def _export_load_only_tuning( + self, + connection: Any, + benchmark_type: str, + effective_tuning_config: Any, + tuned_run: bool, + ) -> tuple[Any, Any, Any, Any, Any]: + tunings_applied_dict = None + requested_config_hash = None + if tuned_run: + tunings_applied_dict = effective_tuning_config.to_dict() + requested_config_hash = effective_tuning_config.get_configuration_hash() + self._fold_layout_operations_into_ledger() + self._reconcile_requested_tuning(effective_tuning_config) + self.configure_for_benchmark( + recording_connection( + connection, + self._applied_tuning_ledger, + PHASE_SESSION, + execute_verbs=self.ledger_execute_verbs, + ), + benchmark_type, + ) + apply_status = tuning_trust.apply_phase_status(self, bool(effective_tuning_config)) + final_status, ledger_payload, ledger_hash, _receipt_payload = tuning_trust.read_back_applied_ledger( + self, connection, apply_status, bool(effective_tuning_config) + ) + return tunings_applied_dict, requested_config_hash, final_status, ledger_payload, ledger_hash + def _build_drift_check_payload(self) -> dict[str, Any] | None: return tuning_trust.build_drift_check_payload(self) diff --git a/tests/unit/core/test_load_only_lifecycle_parity.py b/tests/unit/core/test_load_only_lifecycle_parity.py new file mode 100644 index 0000000000..d5036d8f51 --- /dev/null +++ b/tests/unit/core/test_load_only_lifecycle_parity.py @@ -0,0 +1,166 @@ +from __future__ import annotations + +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +from benchbox.core.runner.runner import ValidationOptions, _execute_load_only_mode +from benchbox.core.tuning.applied_ledger import AppliedTuningLedger +from benchbox.core.tuning.interface import TableTuning, TuningColumn, UnifiedTuningConfiguration + +pytestmark = [ + pytest.mark.unit, + pytest.mark.fast, +] + + +class _StubBenchmark: + name = "parity-probe" + + def __init__(self, output_dir: Path): + self.output_dir = output_dir + self.captured: dict = {} + + def get_create_tables_sql(self, dialect: str = "standard", tuning_config=None) -> str: + pk = "" + if tuning_config is not None and tuning_config.primary_keys.enabled: + pk = ", PRIMARY KEY (a)" + return f"CREATE TABLE t (a INTEGER, b VARCHAR{pk})" + + @property + def tables(self) -> dict: + return {"t": self.output_dir / "t.parquet"} + + def create_enhanced_benchmark_result(self, **kwargs): + self.captured = dict(kwargs) + return SimpleNamespace(execution_id="parity-probe", **kwargs) + + +def _benchmark_config() -> SimpleNamespace: + return SimpleNamespace(name="parity-probe", scale_factor=0.01) + + +def _tuned_sort_config() -> UnifiedTuningConfiguration: + config = UnifiedTuningConfiguration() + config.disable_all_constraints() + config.table_tunings["t"] = TableTuning( + table_name="t", + sorting=[TuningColumn(name="a", type="INTEGER", order=1)], + ) + return config + + +def _seed_parquet(data_dir: Path) -> None: + duckdb = pytest.importorskip("duckdb") + data_dir.mkdir(parents=True, exist_ok=True) + conn = duckdb.connect(":memory:") + conn.execute("CREATE TABLE seed (a INTEGER, b VARCHAR)") + conn.execute("INSERT INTO seed VALUES (2, 'y'), (1, 'x'), (3, 'z')") + conn.execute(f"COPY seed TO '{data_dir}/t.parquet' (FORMAT PARQUET)") + conn.close() + + +def _duckdb_adapter(database_path: Path, config: UnifiedTuningConfiguration | None): + from benchbox.platforms.duckdb import DuckDBAdapter + + adapter = DuckDBAdapter(database_path=str(database_path)) + adapter.tuning_enabled = config is not None + adapter.unified_tuning_configuration = config + adapter._applied_tuning_ledger = AppliedTuningLedger() + return adapter + + +def _ledger_statements(adapter) -> list[tuple[str, str | None, str]]: + return sorted( + (statement.phase, statement.mechanism, statement.statement) + for statement in adapter._applied_tuning_ledger.statements + if statement.status == "executed" + ) + + +class TestLoadOnlyLifecycleParity: + @pytest.fixture(autouse=True) + def _require_duckdb(self): + pytest.importorskip("duckdb") + + def test_load_only_tuned_run_matches_full_run_ledger(self, tmp_path: Path) -> None: + from benchbox.core.results.schema import build_tuning_payload + + _seed_parquet(tmp_path / "data") + config = _tuned_sort_config() + + full_adapter = _duckdb_adapter(tmp_path / "full.duckdb", config) + full_benchmark = _StubBenchmark(tmp_path / "data") + full_connection = full_adapter.create_connection() + try: + full_adapter._setup_fresh_database_phases(full_benchmark, full_connection, config) + full_adapter._fold_layout_operations_into_ledger() + full_adapter._reconcile_requested_tuning(config) + finally: + full_connection.close() + + load_adapter = _duckdb_adapter(tmp_path / "load.duckdb", config) + load_benchmark = _StubBenchmark(tmp_path / "data") + result, _postload = _execute_load_only_mode( + benchmark=load_benchmark, + benchmark_config=_benchmark_config(), + adapter=load_adapter, + platform_config={}, + validation_opts=ValidationOptions(), + table_mode="native", + ) + + assert _ledger_statements(load_adapter) == _ledger_statements(full_adapter) + assert result.tunings_applied is not None + assert result.applied_tuning_ledger is not None + payload = build_tuning_payload(result) + assert payload is not None + + def test_load_only_tuned_run_is_refused_without_full_lifecycle(self, tmp_path: Path) -> None: + config = _tuned_sort_config() + adapter = SimpleNamespace( + platform_name="legacy", + tuning_enabled=True, + unified_tuning_configuration=config, + get_effective_tuning_configuration=lambda: config, + create_connection=Mock(return_value="connection"), + close_connection=Mock(), + ) + benchmark = _StubBenchmark(tmp_path) + + with pytest.raises(RuntimeError, match="Load-only mode cannot run a tuned configuration"): + _execute_load_only_mode( + benchmark=benchmark, + benchmark_config=_benchmark_config(), + adapter=adapter, + platform_config={}, + validation_opts=ValidationOptions(), + table_mode="native", + ) + + def test_load_only_untuned_run_carries_no_tuning(self, tmp_path: Path) -> None: + adapter = SimpleNamespace( + platform_name="duckdb", + tuning_enabled=False, + create_connection=Mock(return_value="connection"), + close_connection=Mock(), + create_schema=Mock(return_value=0.25), + load_data=Mock(return_value=({"t": 3}, 0.5, None)), + _calculate_data_size=Mock(return_value=1.5), + ) + benchmark = _StubBenchmark(tmp_path) + + result, _postload = _execute_load_only_mode( + benchmark=benchmark, + benchmark_config=_benchmark_config(), + adapter=adapter, + platform_config={}, + validation_opts=ValidationOptions(), + table_mode="native", + ) + + assert result.tunings_applied is None + assert result.applied_tuning_ledger is None + assert result.tuning_validation_status is None