"""
Benchmark-only Create Assessment dialog.

The current benchmark suite compares supported overlay/write strategies inside
modern QGIS environments. In particular, it avoids relying on
``native:identity`` because that algorithm is not available in current test
setups, and instead benchmarks an identity-style approximation built from
``intersection + difference + merge``.
"""

from __future__ import annotations

import sqlite3
import struct
import subprocess
import tempfile
import time
import traceback
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

from ..qt_compat import Qt, QApplication, QMessageBox, QProgressDialog, QWizard
from qgis.core import (
    QgsApplication,
    QgsCoordinateTransformContext,
    QgsDataSourceUri,
    QgsFeature,
    QgsField,
    QgsProcessing,
    QgsProcessingFeedback,
    QgsTransaction,
    QgsVectorFileWriter,
    QgsVectorLayer,
    QgsWkbTypes,
)

import processing

from .. import debug_logger
from ..gdal_capabilities import build_gdal_environment
from ..spatialite_manager import SpatialiteManager
from .create_assessment_dialog import CreateAssessmentDialog
from ..translator import tr


@dataclass
class BenchmarkMethodResult:
    method_name: str
    status: str
    target_layer_name: str
    include_layer_names: List[str]
    target_feature_count: int
    target_field_count: int
    include_layer_stats: List[Dict[str, Any]]
    output_feature_count: int = 0
    output_field_count: int = 0
    overlay_build_seconds: float = 0.0
    disk_write_seconds: float = 0.0
    reload_validation_seconds: float = 0.0
    total_seconds: float = 0.0
    output_table_name: str = ""
    error_message: str = ""
    traceback_text: str = ""


class CreateAssessmentBenchmarkDialog(CreateAssessmentDialog):
    """
    Benchmark-only dialog for assessment overlay and write strategies.
    
    Tier 1: Algorithm comparison (geometry math isolation)
    Tier 2: Write strategy comparison (same overlay, different persistence)
    Tier 3: Diagnostic controls (column-width thesis, format comparison)
    """

    RESULTS_PATH = Path(tempfile.gettempdir()) / "emds" / "performResultsOgr.txt"
    METHOD_OGR_IDENTITY_CHAIN = "ogr_identity_chain"

    # -------------------------------------------------------------------------
    # Boilerplate / Scaffolding
    # -------------------------------------------------------------------------

    def __init__(
        self,
        available_layers: Optional[List[str]] = None,
        parent=None,
        *,
        database_path: Optional[str] = None,
    ):
        super().__init__(available_layers, parent, database_path=database_path)
        self.setWindowTitle(tr("create_benchmark.window_title"))
        self._benchmark_results: List[BenchmarkMethodResult] = []

    def accept(self) -> None:
        if self.selected_features_layer is None and not self._validate_page_2():
            return
        if not self.database_path:
            QMessageBox.critical(self, tr("create_benchmark.err_title"), tr("create_benchmark.err_no_db"))
            return

        self._benchmark_results = []
        self._initialize_results_log()

        methods = [self.METHOD_OGR_IDENTITY_CHAIN]

        progress = QProgressDialog(
            tr("create_benchmark.progress_start"),
            tr("create_benchmark.btn_cancel"),
            0,
            len(methods),
            self,
        )
        progress.setWindowModality(Qt.WindowModality.WindowModal)
        progress.show()

        try:
            for index, method_name in enumerate(methods, start=1):
                if progress.wasCanceled():
                    break
                progress.setLabelText(
                    tr("create_benchmark.progress_running", method=method_name, index=index, total=len(methods))
                )
                progress.setValue(index - 1)
                QApplication.processEvents()

                result = self._run_benchmark_method(method_name)
                self._benchmark_results.append(result)
                self._write_method_result_to_log(result)

            progress.setValue(len(methods))
            progress.close()
            self._show_summary_dialog()
            self._cleanup_target_selection()
            QWizard.accept(self)

        except Exception as exc:
            progress.close()
            self._log_line(f"FATAL ERROR: {traceback.format_exc()}")
            QMessageBox.critical(self, tr("create_benchmark.err_benchmark_title"), str(exc))

    # -------------------------------------------------------------------------
    # Dispatch
    # -------------------------------------------------------------------------

    def _run_benchmark_method(self, method_name: str) -> BenchmarkMethodResult:
        target_name = self.target_layer.name() if self.target_layer else ""
        include_layers = self._get_layers_by_status(self.STATUS_INCLUDE)
        include_stats = [
            {
                "layer_name": l.name(),
                "feature_count": l.featureCount(),
                "field_count": l.fields().count(),
            }
            for l in include_layers
        ]

        result = BenchmarkMethodResult(
            method_name=method_name,
            status="FAILED",
            target_layer_name=target_name,
            include_layer_names=[l.name() for l in include_layers],
            target_feature_count=(
                self.selected_features_layer.featureCount()
                if self.selected_features_layer
                else 0
            ),
            target_field_count=(
                self.selected_features_layer.fields().count()
                if self.selected_features_layer
                else 0
            ),
            include_layer_stats=include_stats,
        )

        started_at = time.perf_counter()
        try:
            output_table_name = self._make_output_table_name(method_name)

            if method_name != self.METHOD_OGR_IDENTITY_CHAIN:
                raise RuntimeError(f"Unknown benchmark method: {method_name}")

            table, build_s = self._run_ogr_identity_chain(output_table_name)
            write_s = 0.0

            # ------------------------------------------------------------------
            # Validate output
            # ------------------------------------------------------------------
            val_s, feat_count, field_count = self._validate_output_table(table)

            result.status = "SUCCESS"
            result.output_table_name = table
            result.output_feature_count = feat_count
            result.output_field_count = field_count
            result.overlay_build_seconds = build_s
            result.disk_write_seconds = write_s
            result.reload_validation_seconds = val_s
            result.total_seconds = time.perf_counter() - started_at
            return result

        except Exception as exc:
            result.error_message = str(exc)
            result.traceback_text = traceback.format_exc()
            result.total_seconds = time.perf_counter() - started_at
            return result

    # -------------------------------------------------------------------------
    # OGR/GDAL Benchmark Path
    # -------------------------------------------------------------------------

    def _run_ogr_identity_chain(self, output_table_name: str) -> Tuple[str, float]:
        started = time.perf_counter()
        include_layers = self._get_layers_by_status(self.STATUS_INCLUDE)
        if not include_layers:
            raise RuntimeError("No assessment layers were selected for benchmarking.")
        if self.selected_features_layer is None:
            raise RuntimeError("No selected target features are available.")

        gdal_exe = self._find_gdal_executable()
        gdal_env = self._build_gdal_environment(gdal_exe)
        temp_root = Path(tempfile.mkdtemp(prefix="emds_ogr_benchmark_"))
        debug_logger.debug(
            "Starting OGR identity benchmark chain",
            {"temp_root": str(temp_root), "gdal_exe": gdal_exe},
        )
        self._log_layer_source_details(self.selected_features_layer, "selected_features")
        for index, include_layer in enumerate(include_layers, start=1):
            self._log_layer_source_details(include_layer, f"include_{index}")

        current_path = temp_root / "selected_features.gpkg"
        current_layer_name = "selected_features"
        self._export_layer_to_gpkg(
            self.selected_features_layer, current_path, current_layer_name
        )

        for index, include_layer in enumerate(include_layers, start=1):
            method_path = temp_root / f"include_{index}.gpkg"
            method_layer_name = f"include_{index}"
            output_path = temp_root / f"identity_step_{index}.gpkg"
            output_layer_name = (
                output_table_name if index == len(include_layers) else f"step_{index}"
            )
            method_prefix = f"{self._sanitize_name(include_layer.name())}_"

            self._export_layer_to_gpkg(include_layer, method_path, method_layer_name)
            command = [
                gdal_exe,
                "vector",
                "layer-algebra",
                "--operation",
                "identity",
                "--input",
                str(current_path),
                "--input-layer",
                current_layer_name,
                "--method",
                str(method_path),
                "--method-layer",
                method_layer_name,
                "--output",
                str(output_path),
                "--output-layer",
                output_layer_name,
                "--format",
                "GPKG",
                "--geometry-type",
                "GEOMETRY",
                "--overwrite",
                "--lco",
                "LAUNDER=NO",
                "--method-prefix",
                method_prefix,
            ]
            completed = subprocess.run(
                command,
                capture_output=True,
                text=True,
                check=False,
                env=gdal_env,
                **self._subprocess_window_kwargs(),
            )
            self._log_line(
                "OGR command: " + subprocess.list2cmdline(command)
            )
            if completed.stdout.strip():
                self._log_line("OGR stdout: " + completed.stdout.strip())
            if completed.stderr.strip():
                self._log_line("OGR stderr: " + completed.stderr.strip())
            if completed.returncode != 0:
                raise RuntimeError(
                    f"GDAL layer-algebra failed on include layer "
                    f"'{include_layer.name()}' with exit code {completed.returncode}."
                )

            current_path = output_path
            current_layer_name = output_layer_name

        final_uri = f"{current_path}|layername={current_layer_name}"
        return final_uri, time.perf_counter() - started

    @staticmethod
    def _find_gdal_executable() -> str:
        return CreateAssessmentDialog._find_gdal_executable()

    @staticmethod
    def _build_gdal_environment(gdal_exe: str) -> Dict[str, str]:
        return build_gdal_environment(gdal_exe)

    def _log_layer_source_details(self, layer: QgsVectorLayer, label: str) -> None:
        details = self._inspect_layer_storage(layer)
        parts = [f"{key}={value}" for key, value in details.items()]
        self._log_line(f"Layer source [{label}]: " + ", ".join(parts))

    def _inspect_layer_storage(self, layer: QgsVectorLayer) -> Dict[str, Any]:
        source = layer.source() or ""
        source_path = source.split("|", 1)[0]
        table_name = self._extract_layername_from_source(source) or layer.name()
        provider = layer.providerType()
        details: Dict[str, Any] = {
            "provider": provider,
            "source": source,
            "table": table_name,
        }

        path = Path(source_path)
        if not source_path or not path.exists():
            details["spatial_index"] = "unknown"
            return details

        suffix = path.suffix.lower()
        if suffix not in {".gpkg", ".sqlite", ".db"}:
            details["spatial_index"] = "not_checked"
            return details

        try:
            with sqlite3.connect(str(path)) as conn:
                geometry_column, enabled_flag = self._read_geometry_column_info(
                    conn, table_name
                )
                details["geometry_column"] = geometry_column or "unknown"
                if enabled_flag is not None:
                    details["spatial_index_enabled_flag"] = enabled_flag

                index_tables = []
                if geometry_column:
                    expected_names = [
                        f"rtree_{table_name}_{geometry_column}",
                        f"idx_{table_name}_{geometry_column}",
                    ]
                    for candidate in expected_names:
                        row = conn.execute(
                            "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?",
                            (candidate,),
                        ).fetchone()
                        if row:
                            index_tables.append(candidate)

                details["spatial_index"] = (
                    "present" if index_tables else "missing_or_unknown"
                )
                if index_tables:
                    details["index_tables"] = ",".join(index_tables)
        except Exception as exc:
            details["spatial_index"] = f"error:{exc}"

        return details

    @staticmethod
    def _read_geometry_column_info(
        conn: sqlite3.Connection, table_name: str
    ) -> Tuple[Optional[str], Optional[Any]]:
        if CreateAssessmentBenchmarkDialog._sqlite_table_exists(
            conn, "gpkg_geometry_columns"
        ):
            row = conn.execute(
                "SELECT column_name FROM gpkg_geometry_columns WHERE table_name = ?",
                (table_name,),
            ).fetchone()
            if row:
                geometry_column = row[0]
                extension_row = None
                if CreateAssessmentBenchmarkDialog._sqlite_table_exists(
                    conn, "gpkg_extensions"
                ):
                    extension_row = conn.execute(
                        """
                        SELECT 1
                        FROM gpkg_extensions
                        WHERE table_name = ? AND extension_name = 'gpkg_rtree_index'
                        """,
                        (table_name,),
                    ).fetchone()
                return geometry_column, 1 if extension_row else 0

        if CreateAssessmentBenchmarkDialog._sqlite_table_exists(conn, "geometry_columns"):
            row = conn.execute(
                """
                SELECT f_geometry_column, spatial_index_enabled
                FROM geometry_columns
                WHERE f_table_name = ?
                """,
                (table_name,),
            ).fetchone()
            if row:
                return row[0], row[1]

        return None, None

    @staticmethod
    def _sqlite_table_exists(conn: sqlite3.Connection, table_name: str) -> bool:
        row = conn.execute(
            "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?",
            (table_name,),
        ).fetchone()
        return row is not None

    def _export_layer_to_gpkg(
        self,
        layer: QgsVectorLayer,
        output_path: Path,
        layer_name: str,
    ) -> None:
        if output_path.exists():
            output_path.unlink()

        export_layer = self._prepare_layer_for_persist(layer, layer_name)
        options = QgsVectorFileWriter.SaveVectorOptions()
        options.driverName = "GPKG"
        options.fileEncoding = "UTF-8"
        options.layerName = layer_name

        error, msg, _, _ = QgsVectorFileWriter.writeAsVectorFormatV3(
            export_layer,
            str(output_path),
            QgsCoordinateTransformContext(),
            options,
        )
        if error != QgsVectorFileWriter.NoError:
            raise RuntimeError(
                f"Failed to export layer '{layer.name()}' to '{output_path}': {msg}"
            )

    def _build_overlay_loop_intersection(
        self,
    ) -> Tuple[QgsVectorLayer, float]:
        """
        Simulates the current production approach: loop native:intersection
        over each include layer, accumulating results.
        This is your baseline to beat.
        """
        started = time.perf_counter()
        include_layers = self._get_layers_by_status(self.STATUS_INCLUDE)
        current = self.selected_features_layer

        for inc_layer in include_layers:
            result = processing.run(
                "native:intersection",
                {
                    "INPUT": current,
                    "OVERLAY": inc_layer,
                    "INPUT_FIELDS": [],
                    "OVERLAY_FIELDS": [],
                    "OVERLAY_FIELDS_PREFIX": f"{inc_layer.name()[:4]}_",
                    "OUTPUT": "memory:",
                    "GRID_SIZE": None,
                },
            )
            current = result["OUTPUT"]

        return current, time.perf_counter() - started

    def _build_overlay_intersection_difference_merge(
        self,
    ) -> Tuple[QgsVectorLayer, float]:
        """
        Approximate identity semantics using only supported native algorithms:
        merge includes -> intersection(selected, merged_overlay)
        + difference(selected, merged_overlay)
        -> merge matched and unmatched pieces back together.

        This keeps all target features while still attaching overlay attributes
        to the intersecting pieces, which is close enough for benchmark
        comparison without depending on unavailable processing algorithms.
        """
        started = time.perf_counter()
        merged_overlay = self._merge_include_layers()

        intersection_result = processing.run(
            "native:intersection",
            {
                "INPUT": self.selected_features_layer,
                "OVERLAY": merged_overlay,
                "OVERLAY_FIELDS_PREFIX": "inc_",
                "OUTPUT": "memory:",
                "GRID_SIZE": None,
            },
        )
        matched_layer = intersection_result["OUTPUT"]

        difference_result = processing.run(
            "native:difference",
            {
                "INPUT": self.selected_features_layer,
                "OVERLAY": merged_overlay,
                "OUTPUT": "memory:",
                "GRID_SIZE": None,
            },
        )
        unmatched_layer = difference_result["OUTPUT"]

        merge_result = processing.run(
            "native:mergevectorlayers",
            {
                "LAYERS": [matched_layer, unmatched_layer],
                "CRS": self.selected_features_layer.crs(),
                "OUTPUT": "memory:",
            },
        )

        return merge_result["OUTPUT"], time.perf_counter() - started

    def _merge_include_layers(self) -> QgsVectorLayer:
        include_layers = self._get_layers_by_status(self.STATUS_INCLUDE)
        if not include_layers:
            raise RuntimeError("No assessment layers were selected for benchmarking.")

        if len(include_layers) == 1:
            return include_layers[0]

        merge_result = processing.run(
            "native:mergevectorlayers",
            {
                "LAYERS": include_layers,
                "CRS": include_layers[0].crs(),
                "OUTPUT": "memory:",
            },
        )
        merged_overlay = merge_result["OUTPUT"]
        if merged_overlay is None or not merged_overlay.isValid():
            raise RuntimeError("Failed to merge assessment layers for benchmark input.")
        return merged_overlay

    # -------------------------------------------------------------------------
    # Tier 2: Write Strategy Implementations
    # -------------------------------------------------------------------------

    def _prepare_layer_for_persist(
        self,
        layer: QgsVectorLayer,
        layer_name: str,
    ) -> QgsVectorLayer:
        """
        Copy features into a fresh memory layer so provider-assigned FIDs are
        regenerated and don't collide when persisted to GPKG/SQLite outputs.
        """
        uri = f"{QgsWkbTypes.displayString(layer.wkbType())}?crs={layer.crs().authid()}"
        prepared = QgsVectorLayer(uri, layer_name, "memory")
        provider = prepared.dataProvider()
        renamed_fields = []
        used_names = set()
        reserved_names = {"fid", "ogc_fid", "rowid"}
        renamed = {}

        for field in layer.fields():
            new_name = field.name()
            if new_name.lower() in reserved_names:
                new_name = f"{new_name}_attr"

            candidate = new_name
            suffix = 2
            while candidate.lower() in used_names:
                candidate = f"{new_name}_{suffix}"
                suffix += 1

            field_copy = QgsField(field)
            field_copy.setName(candidate)
            renamed_fields.append(field_copy)
            used_names.add(candidate.lower())
            if candidate != field.name():
                renamed[field.name()] = candidate

        provider.addAttributes(renamed_fields)
        prepared.updateFields()

        features: List[QgsFeature] = []
        for source_feature in layer.getFeatures():
            cloned_feature = QgsFeature(prepared.fields())
            cloned_feature.setId(-1)
            cloned_feature.setAttributes(source_feature.attributes())
            cloned_feature.setGeometry(source_feature.geometry())
            features.append(cloned_feature)

        provider.addFeatures(features)
        prepared.updateExtents()
        if renamed:
            debug_logger.debug(
                "Renamed reserved fields before benchmark persist",
                {"layer_name": layer_name, "renamed_fields": renamed},
            )
        return prepared

    @staticmethod
    def _is_algorithm_available(algorithm_id: str) -> bool:
        return QgsApplication.processingRegistry().algorithmById(algorithm_id) is not None

    def _persist_with_writer_gpkg(
        self,
        layer: QgsVectorLayer,
        table_name: str,
        deferred_index: bool = True,
    ) -> Tuple[float, str]:
        """
        QgsVectorFileWriter → GeoPackage.
        deferred_index=True skips spatial index build during write,
        rebuilds it afterward. Significant speedup for large wide layers.
        """
        started = time.perf_counter()
        output_path = str(
            Path(self.database_path).parent / f"{table_name}.gpkg"
        )
        write_layer = self._prepare_layer_for_persist(layer, table_name)

        options = QgsVectorFileWriter.SaveVectorOptions()
        options.driverName = "GPKG"
        options.fileEncoding = "UTF-8"
        options.layerName = table_name
        if deferred_index:
            # Write rows first, build spatial index in a second pass
            options.layerOptions = ["SPATIAL_INDEX=NO"]

        error, msg, _, _ = QgsVectorFileWriter.writeAsVectorFormatV3(
            write_layer,
            output_path,
            QgsCoordinateTransformContext(),
            options,
        )
        if error != QgsVectorFileWriter.NoError:
            raise RuntimeError(f"GPKG write failed: {msg}")

        if deferred_index:
            # Rebuild spatial index post-write (single sequential pass)
            conn = sqlite3.connect(output_path)
            try:
                conn.execute(
                    f"SELECT gpkg_ogr_contents_insert('{table_name}', NULL)"
                )
            except Exception:
                pass  # Not all GPKG builds expose this — harmless
            finally:
                conn.close()

        return time.perf_counter() - started, output_path

    def _persist_with_writer_spatialite(
        self,
        layer: QgsVectorLayer,
        table_name: str,
        deferred_index: bool = True,
    ) -> Tuple[float, str]:
        """
        QgsVectorFileWriter → SpatiaLite.
        deferred_index=True suppresses spatial index creation during write.
        This avoids per-feature R*Tree updates which are the main SL bottleneck.
        """
        started = time.perf_counter()
        output_path = str(
            Path(self.database_path).parent / f"{table_name}.sqlite"
        )
        write_layer = self._prepare_layer_for_persist(layer, table_name)

        options = QgsVectorFileWriter.SaveVectorOptions()
        options.driverName = "SQLite"
        options.fileEncoding = "UTF-8"
        options.layerName = table_name
        if deferred_index:
            options.layerOptions = [
                "SPATIALITE=YES",
                "SPATIAL_INDEX=NO",   # Skip R*Tree during bulk write
                "COMPRESS_GEOM=NO",   # Avoid per-row compression overhead
            ]
        else:
            options.layerOptions = ["SPATIALITE=YES"]

        error, msg, _, _ = QgsVectorFileWriter.writeAsVectorFormatV3(
            write_layer,
            output_path,
            QgsCoordinateTransformContext(),
            options,
        )
        if error != QgsVectorFileWriter.NoError:
            raise RuntimeError(f"SpatiaLite write failed: {msg}")

        if deferred_index:
            # Rebuild spatial index in one pass after all rows are committed
            conn = sqlite3.connect(output_path)
            conn.enable_load_extension(True)
            try:
                conn.execute("SELECT load_extension('mod_spatialite')")
                conn.execute(
                    f"SELECT CreateSpatialIndex('{table_name}', 'geometry')"
                )
                conn.commit()
            finally:
                conn.close()

        return time.perf_counter() - started, output_path

    def _persist_with_transaction(
        self,
        layer: QgsVectorLayer,
        table_name: str,
    ) -> Tuple[float, str]:
        """
        Creates the destination schema via a zero-feature write, then
        uses QgsTransaction to bulk-insert in a single atomic commit.
        Fixes the original double-write bug.
        """
        started = time.perf_counter()
        output_path = str(
            Path(self.database_path).parent / f"{table_name}.sqlite"
        )
        write_layer = self._prepare_layer_for_persist(layer, table_name)

        # --- Step 1: Schema-only write (zero features) ---
        options = QgsVectorFileWriter.SaveVectorOptions()
        options.driverName = "SQLite"
        options.fileEncoding = "UTF-8"
        options.layerName = table_name
        options.layerOptions = [
            "SPATIALITE=YES",
            "SPATIAL_INDEX=NO",
        ]
        options.onlySelectedFeatures = False

        # Write an empty clone to establish schema
        empty_clone = QgsVectorLayer(
            f"{QgsWkbTypes.displayString(write_layer.wkbType())}?crs={write_layer.crs().authid()}",
            "empty_schema",
            "memory",
        )
        empty_clone.dataProvider().addAttributes(write_layer.fields().toList())
        empty_clone.updateFields()

        error, msg, _, _ = QgsVectorFileWriter.writeAsVectorFormatV3(
            empty_clone,
            output_path,
            QgsCoordinateTransformContext(),
            options,
        )
        if error != QgsVectorFileWriter.NoError:
            raise RuntimeError(f"Schema write failed: {msg}")

        # --- Step 2: Load the empty destination layer ---
        uri = QgsDataSourceUri()
        uri.setDatabase(output_path)
        uri.setDataSource("", table_name, "geometry")
        dest_layer = QgsVectorLayer(uri.uri(), table_name, "spatialite")
        if not dest_layer.isValid():
            raise RuntimeError("Destination layer failed to load after schema write.")

        # --- Step 3: Bulk insert via QgsTransaction ---
        transaction = QgsTransaction.create(dest_layer.source())
        transaction.addLayer(dest_layer)

        if not transaction.begin():
            raise RuntimeError(
                f"Transaction failed to begin: {transaction.lastError()}"
            )
        try:
            features = list(write_layer.getFeatures())
            success, _ = dest_layer.dataProvider().addFeatures(features)
            if not success:
                raise RuntimeError("dataProvider().addFeatures() returned False.")
            transaction.commit()
        except Exception:
            transaction.rollback()
            raise

        # --- Step 4: Rebuild spatial index post-commit ---
        conn = sqlite3.connect(output_path)
        conn.enable_load_extension(True)
        try:
            conn.execute("SELECT load_extension('mod_spatialite')")
            conn.execute(
                f"SELECT CreateSpatialIndex('{table_name}', 'geometry')"
            )
            conn.commit()
        finally:
            conn.close()

        return time.perf_counter() - started, output_path

    def _persist_with_raw_sqlite3(
        self,
        layer: QgsVectorLayer,
        table_name: str,
    ) -> Tuple[float, str]:
        """
        Bypasses the QGIS write stack entirely.
        Uses raw sqlite3 + executemany with WKB geometry blobs.
        This is the theoretical minimum write time for SpatiaLite.
        
        If this is NOT significantly faster than _persist_with_writer_spatialite,
        the bottleneck is in the geometry math, not the write layer.
        """
        started = time.perf_counter()
        output_path = str(
            Path(self.database_path).parent / f"{table_name}.sqlite"
        )
        write_layer = self._prepare_layer_for_persist(layer, table_name)

        fields = write_layer.fields()
        field_names = [f.name() for f in fields]
        geom_type = QgsWkbTypes.displayString(write_layer.wkbType())
        srid = write_layer.crs().postgisSrid()

        conn = sqlite3.connect(output_path)
        conn.enable_load_extension(True)

        try:
            conn.execute("SELECT load_extension('mod_spatialite')")
            conn.execute("SELECT InitSpatialMetaData(1)")

            # Build CREATE TABLE from QGIS field types
            col_defs = []
            for f in fields:
                type_name = self._qgis_field_type_to_sqlite(f)
                col_defs.append(f'"{f.name()}" {type_name}')

            col_defs_str = ", ".join(col_defs)
            conn.execute(
                f'CREATE TABLE "{table_name}" (fid INTEGER PRIMARY KEY, {col_defs_str})'
            )
            conn.execute(
                f"SELECT AddGeometryColumn('{table_name}', 'geometry', {srid}, '{geom_type}', 'XY')"
            )

            # Build INSERT with placeholders
            placeholders = ", ".join(["?"] * (len(field_names) + 1))  # +1 for geom
            col_list = (
                ", ".join(f'"{n}"' for n in field_names) + ", geometry"
            )
            insert_sql = f'INSERT INTO "{table_name}" ({col_list}) VALUES ({placeholders})'

            # Batch rows — executemany is ~10x faster than individual execute
            BATCH_SIZE = 5000
            batch = []
            for feature in write_layer.getFeatures():
                attrs = feature.attributes()
                geom_wkb = (
                    feature.geometry().asWkb().data()
                    if feature.hasGeometry()
                    else None
                )
                batch.append(tuple(attrs) + (geom_wkb,))

                if len(batch) >= BATCH_SIZE:
                    conn.executemany(insert_sql, batch)
                    batch.clear()

            if batch:
                conn.executemany(insert_sql, batch)

            conn.commit()

            # Spatial index post-bulk-insert
            conn.execute(
                f"SELECT CreateSpatialIndex('{table_name}', 'geometry')"
            )
            conn.commit()

        finally:
            conn.close()

        return time.perf_counter() - started, output_path

    # -------------------------------------------------------------------------
    # Tier 3: Diagnostic Implementations
    # -------------------------------------------------------------------------

    def _persist_stripped_single_column(
        self,
        layer: QgsVectorLayer,
        table_name: str,
    ) -> Tuple[float, str]:
        """
        Strips all but the first attribute field before writing.
        If this is dramatically faster than the full write, the bottleneck
        is column-width serialization, not geometry or row count.
        """
        started = time.perf_counter()

        # Clone correctly via layer.clone() — avoids the source URI trap
        stripped = layer.clone()
        stripped.setName("stripped_control")

        if stripped.fields().count() > 1:
            indices_to_drop = list(range(1, stripped.fields().count()))
            stripped.dataProvider().deleteAttributes(indices_to_drop)
            stripped.updateFields()

        # Use the same GPKG writer so format is not a variable
        write_s, table = self._persist_with_writer_gpkg(
            stripped, table_name, deferred_index=True
        )
        return time.perf_counter() - started, table

    # -------------------------------------------------------------------------
    # Helpers
    # -------------------------------------------------------------------------

    @staticmethod
    def _qgis_field_type_to_sqlite(field) -> str:
        """Maps QgsField type to a SQLite column type string."""
        from qgis.core import QVariant
        type_map = {
            QVariant.Int:     "INTEGER",
            QVariant.LongLong: "INTEGER",
            QVariant.Double:  "REAL",
            QVariant.String:  "TEXT",
            QVariant.Bool:    "INTEGER",
            QVariant.Date:    "TEXT",
            QVariant.DateTime: "TEXT",
        }
        return type_map.get(field.type(), "TEXT")

    def _validate_output_table(
        self, path_or_table: str
    ) -> Tuple[float, int, int]:
        """Reloads the output layer and counts features + fields."""
        started = time.perf_counter()
        layer = QgsVectorLayer(path_or_table, "validation", "ogr")
        if not layer.isValid():
            # Try spatialite provider
            uri = QgsDataSourceUri()
            uri.setDatabase(path_or_table)
            layer = QgsVectorLayer(uri.uri(), "validation", "spatialite")

        feat_count = layer.featureCount() if layer.isValid() else -1
        field_count = layer.fields().count() if layer.isValid() else -1
        return time.perf_counter() - started, feat_count, field_count

    def _make_output_table_name(self, method_name: str) -> str:
        ts = datetime.now().strftime("%H%M%S")
        safe = method_name.replace(":", "_").replace("/", "_")
        return f"bm_{safe}_{ts}"

    def _initialize_results_log(self) -> None:
        self.RESULTS_PATH.parent.mkdir(parents=True, exist_ok=True)
        with open(self.RESULTS_PATH, "w") as f:
            f.write(f"Benchmark Run — {datetime.now().isoformat()}\n")
            f.write("=" * 80 + "\n\n")

    def _log_line(self, text: str) -> None:
        with open(self.RESULTS_PATH, "a") as f:
            f.write(text + "\n")

    def _write_method_result_to_log(self, result: BenchmarkMethodResult) -> None:
        lines = [
            f"\n--- {result.method_name} ---",
            f"  Status:        {result.status}",
            f"  Overlay build: {result.overlay_build_seconds:.3f}s",
            f"  Disk write:    {result.disk_write_seconds:.3f}s",
            f"  Validation:    {result.reload_validation_seconds:.3f}s",
            f"  Total:         {result.total_seconds:.3f}s",
            f"  Output feats:  {result.output_feature_count}",
            f"  Output fields: {result.output_field_count}",
        ]
        if result.error_message:
            lines.append(f"  ERROR: {result.error_message}")
            lines.append(f"  TRACEBACK:\n{result.traceback_text}")
        self._log_line("\n".join(lines))

    def _show_summary_dialog(self) -> None:
        lines = [tr("create_benchmark.results_header")]
        for r in self._benchmark_results:
            status_icon = "✓" if r.status == "SUCCESS" else "✗"
            lines.append(
                f"{status_icon} {r.method_name}\n"
                f"    overlay={r.overlay_build_seconds:.2f}s  "
                f"write={r.disk_write_seconds:.2f}s  "
                f"total={r.total_seconds:.2f}s"
            )
        QMessageBox.information(self, tr("create_benchmark.results_title"), "\n".join(lines))
