tylertoo-core 0.7.0

//! Two-pass bounded-memory streaming overview conversion (task V4 / H3).
//!
//! The in-memory pipeline in [`super::convert`] materializes the entire input
//! table, every decoded geometry, and a cloned geometry set per level — `O(N)`
//! memory (Moldova, 632k polygons: 5.44 GB peak RSS). This module implements
//! the same conversion in bounded memory:
//!
//! - **Pass 1** streams the input once (row batches of
//!   [`ConvertOptions::read_batch_size`] rows, geometry + ranking columns
//!   only): per feature it keeps only the bbox, [`FeatureKind`], and sort key
//!   — a small [`AssignFeature`] — plus the incremental state for Q1 ranking
//!   auto-detection. [`assign_levels_bounded`] and [`apply_density_budget`] then run
//!   over those (the assign engine's own state is `O(occupied cells)` per
//!   level), producing the **winner table**: one `min_level` byte per feature.
//! - **Pass 2** re-reads the (seekable) input once **per level**, coarse →
//!   fine: each batch is filtered against the winner table, simplified for the
//!   level (non-canonical duplicating levels only), and handed straight to the
//!   [`OverviewWriter`]. Nothing is retained across batches.
//!
//! Peak memory is `O(read batch + winner tables)`: the winner table is 1 byte
//! per feature; pass 1 additionally holds the `AssignFeature` vector (~48
//! bytes/feature) and per-candidate ranking keys (16 bytes/feature) while the
//! assignment runs, all freed before pass 2. Residual `O(N)` state is
//! therefore ~50–80 bytes per input feature — for 632k features, a few tens of
//! MB — far below the geometry payload the in-memory path holds.
//!
//! Hilbert order: input order is preserved within each level (the documented
//! gpio-sorted input contract, spec §4.3), exactly as in the in-memory path —
//! no in-memory per-level sort exists in either path.
//!
//! # Behavior parity with the in-memory path
//!
//! Level assignments, density-budget cuts, ranking resolution (explicit /
//! auto-detected / fallback), footer metadata, and per-level row values are
//! identical (tested in `convert::tests`). One documented divergence: the
//! in-memory path omits (and renumbers past) a level whose *simplified*
//! output is empty, whereas this path decides level omission from the winner
//! table before simplification. The two only differ when **every** winner of
//! a level degenerates during simplification — rare under the default knobs
//! (the assign visibility gates, 2 × GSD, are stricter than the simplify
//! drop gate at 1 × GSD) but real on dirty data (#211: a sliver with a huge
//! bbox passes the gate, then collapses). When it happens the writer skips
//! the level ([`LevelWriteOutcome::SkippedEmpty`]) and this driver records it
//! in [`ConvertReport::skipped_empty_levels`], exactly like a plan-time
//! omission — the two pipelines converge on the same output pyramid.
//!
//! [`LevelWriteOutcome::SkippedEmpty`]: super::writer::LevelWriteOutcome::SkippedEmpty
//! [`ConvertReport::skipped_empty_levels`]: super::convert::ConvertReport::skipped_empty_levels

use std::cell::Cell;
use std::collections::HashSet;
use std::fs::File;
use std::path::Path;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};

use crossbeam_channel::{Receiver, Sender};

use arrow_array::{Array, RecordBatch, UInt32Array};
use arrow_schema::{DataType, Field, Schema, SchemaRef};
use arrow_select::take::take;
use geo::{Area, Geometry};
use geoarrow::array::from_arrow_array;
use rayon::prelude::*;

use crate::batch_processor::{extract_geometries_from_array, extract_geometries_opt_from_array};
use crate::input_set::{ConvertSource, ReadPlan, RowGroupSelection};

use super::accumulate::{is_carrier, level_accumulates, tiny_polygon_carriers, AccumulateLevel};
use super::assign::{apply_density_budget, assign_levels_bounded, AssignFeature, FeatureKind};
use super::cluster::{ClusterEntry, ClusterTables};
use super::coalesce::CoalesceInput;
use super::convert::{
    append_coalesced_count_field, append_point_count_field, apply_cluster_columns,
    apply_coalesced_count, build_generalization, build_level_batch, build_level_coalesce_table,
    build_source_schema, class_ranking_provenance, coalesce_effective, coalesce_level_chains,
    count_vertices, encode_concurrency_for, extract_class_ranks, extract_sort_keys,
    fill_level_bytes, find_geometry_column, mixed_geometry_field, overture_road_ranking,
    record_level_outcome, resolve_reserved_column_collisions, scan_feature,
    validate_cluster_schema, validate_coalesce_schema, warn_plan_skipped_levels, ClassRanking,
    CoalesceTable, ConvertError, ConvertOptions, ConvertReport, GroupInterner, SkippedLevelReport,
    KNOWN_ROAD_CLASSES, ROAD_VOCAB_MIN_DISTINCT,
};
use super::level::{Crs, Mode, RankingProvenance};
use super::pipe::scoped_pipe;
use super::pipeline;
use super::simplify::{
    carrier_square, full_resolution_fallback_count, simplify_cascade, simplify_step,
    validation_skip_count, CascadeStep, CollapseMode, Representation, Simplified, SimplifyOptions,
};
use super::writer::{LevelSpec, LevelWriteOutcome, OverviewWriter, OverviewWriterOptions};

/// Row-indexed winner-table sentinel for rows with no feature (null, empty,
/// or non-finite geometry — skipped in pass 1). It matches no level in either
/// mode: [`super::convert::MAX_LEVELS`] caps the plan at 255 levels, so the
/// finest level index is at most 254.
const UNASSIGNED_LEVEL: u8 = u8::MAX;

/// A level actually emitted to the output (levels with zero winners are
/// omitted and renumbered, spec §7.3, matching the in-memory path).
struct EmitLevel {
    /// Index in the *resolved* level plan (drives winner-table membership).
    orig: u8,
    gsd: f64,
    zoom: Option<u8>,
    /// Winner count — the writer's `level_row_hint` for row-group sizing.
    hint: usize,
}

/// Pass-2 execution strategy. Both produce byte-identical output; `Serial` is
/// the pre-#213 per-level-re-read reference, retained for differential testing.
#[derive(Clone, Copy)]
pub(crate) enum Pass2Strategy {
    /// One in-order re-read per level (the reference path).
    #[cfg_attr(not(test), allow(dead_code))]
    Serial,
    /// Single-read pipelined engine ([`super::pipeline`]); the production path.
    Pipelined,
}

/// Streaming counterpart of [`super::convert::convert_to_overviews`], with an
/// explicit pass-2 [`Pass2Strategy`] (production uses `Pipelined`; tests pin
/// `Serial` to assert the pipelined engine is equivalent).
/// Info-level summary of RDP candidates whose validity check was skipped by
/// the vertex cap during this conversion (#242). `skips_before` is the
/// process-wide counter snapshot taken before pass 2.
fn log_validation_skips(skips_before: u64) {
    let skips = validation_skip_count() - skips_before;
    if skips > 0 {
        log::info!(
            "[convert] {skips} oversized RDP candidate(s) skipped exact \
             validity checking and were assumed valid (#242; geometry \
             validity is not an overviews conformance requirement)"
        );
    }
}

/// Pass 0 (#286/#287): stage the selected row groups to local disk up front so
/// the two passes below read from the spill, not the network.
///
/// A row group's column chunks are a contiguous span, so each is fetched as ONE
/// coalesced range request (several in flight per part) and spilled. This
/// removes the per-column-chunk serial re-fetch (#287) and the cold pass-2
/// re-fetch of the property columns pass 1's projection skips (#286).
/// Best-effort and no-op for local input: a staging error is logged and the
/// passes fall back to the reader's lazy network path (the same bytes, just
/// uncoalesced), so staging never regresses correctness.
fn stage_input_pass0(
    source: &ConvertSource,
    selected_row_groups: Option<&RowGroupSelection>,
    row_groups_read: usize,
) {
    if !source.is_remote() {
        return;
    }
    let t_stage = Instant::now();
    match source.stage_selected(selected_row_groups) {
        Ok(()) => log::info!(
            "[convert] staged {row_groups_read} selected row group(s) to local \
             disk in {:.1}s",
            t_stage.elapsed().as_secs_f64()
        ),
        Err(e) => log::warn!(
            "[convert] input staging failed ({e}); passes will read over the \
             network (uncoalesced)"
        ),
    }
}

/// Resolve the pass-2 in-flight depth (auto-sizing from available cores when
/// the caller left it at [`super::convert::IN_FLIGHT_BATCHES_AUTO`]) and log
/// the chosen depth alongside the detected core count, so pass-2 core
/// utilization is observable rather than a mystery (#264).
fn resolve_and_log_in_flight_batches(requested: usize) -> usize {
    let in_flight = super::convert::resolve_in_flight_batches(requested);
    log::info!(
        "[convert] pass 2 parallelism: {in_flight} read batch(es) in flight ({} core(s) detected)",
        std::thread::available_parallelism()
            .map(|n| n.get())
            .unwrap_or(0)
    );
    in_flight
}

/// Current process resident-set size (RSS) in MiB, if the platform exposes it.
fn current_rss_mib() -> Option<f64> {
    memory_stats::memory_stats().map(|m| m.physical_mem as f64 / (1024.0 * 1024.0))
}

/// Log the process RSS at a convert-phase boundary (#295 instrumentation) and
/// fold it into the running peak. Silent when the platform can't report RSS.
///
/// These `[rss] <phase>` lines pinpoint which phase dominates peak memory —
/// pass-1 winner tables (O(dataset)) vs the pass-2 output sink (bounded by the
/// #294 auto profile) — and validate the auto backing choice on real runs.
fn log_phase_rss(phase: &str, peak_mib: &mut f64) {
    if let Some(rss) = current_rss_mib() {
        if rss > *peak_mib {
            *peak_mib = rss;
        }
        log::info!("[rss] {phase}: {rss:.0} MiB");
    }
}

/// Split the resolved level plan into emitted levels (have winners) and skipped
/// levels (no winners → omitted per §7.3 / #211 auto-clamp). `counts[l]` is the
/// winner count for planned level `l`.
fn partition_emitted_levels(
    level_specs: &[(f64, Option<u8>)],
    counts: &[usize],
) -> (Vec<EmitLevel>, Vec<SkippedLevelReport>) {
    let skipped = level_specs
        .iter()
        .enumerate()
        .filter(|&(l, _)| counts[l] == 0)
        .map(|(l, &(gsd, zoom))| SkippedLevelReport {
            planned_level: l,
            gsd,
            zoom,
        })
        .collect();
    let emitted = level_specs
        .iter()
        .enumerate()
        .filter(|&(l, _)| counts[l] > 0)
        .map(|(l, &(gsd, zoom))| EmitLevel {
            orig: l as u8,
            gsd,
            zoom,
            hint: counts[l],
        })
        .collect();
    (emitted, skipped)
}

/// Build the three writer schemas (identical to the in-memory path):
/// `source` (base), `cluster` (+ `point_count` when clustering, Q4), and `out`
/// (+ `coalesced_count` when coalescing, Q3). All three are needed downstream,
/// so they are returned together.
pub(super) fn build_level_schemas(
    input_schema: &Schema,
    geom_idx: usize,
    geom_name: &str,
    options: &ConvertOptions,
) -> (Schema, Schema, Schema) {
    let geom_out_field = mixed_geometry_field(geom_name);
    let source_schema = build_source_schema(input_schema, geom_idx, geom_out_field);
    let cluster_schema = if options.cluster {
        append_point_count_field(&source_schema)
    } else {
        source_schema.clone()
    };
    let out_schema = if options.coalesce_lines {
        append_coalesced_count_field(&cluster_schema)
    } else {
        cluster_schema.clone()
    };
    (source_schema, cluster_schema, out_schema)
}

/// Build the writer options both convert paths use: the shared knobs plus the
/// generalization provenance recorded in the footer (§3.5).
pub(super) fn build_writer_options(
    writer_levels: Vec<LevelSpec>,
    emitted_gsds: &[f64],
    crs: Crs,
    ranking_provenance: RankingProvenance,
    renames: &[(String, String)],
    options: &ConvertOptions,
) -> OverviewWriterOptions {
    let mut writer_opts = OverviewWriterOptions::new(options.mode, writer_levels);
    writer_opts.max_row_group_size = options.max_row_group_size;
    writer_opts.row_group_size_policy = options.row_group_size_policy;
    writer_opts.full_column_stats = options.full_column_stats;
    writer_opts.cogp_compat_key = options.cogp_compat_key;
    writer_opts.encode_concurrency = encode_concurrency_for(options.profile);
    writer_opts.generalization = Some(build_generalization(
        emitted_gsds,
        crs,
        options,
        ranking_provenance,
        renames,
    ));
    writer_opts
}

/// Combined per-part footer-statistics row-group selection for the streaming
/// path: bbox covering pruning (#102) intersected with attribute-filter
/// statistics pushdown (#315). `None` when neither pruning is active.
fn select_row_groups_streaming(
    source: &ConvertSource,
    bbox_units: Option<&[f64; 4]>,
    filter: Option<&super::filter::BoundFilter>,
) -> Result<Option<RowGroupSelection>, ConvertError> {
    let bbox_selection: Option<RowGroupSelection> = match bbox_units {
        Some(bb) => Some(source.select_row_groups(bb)?),
        None => None,
    };
    let filter_selection: Option<RowGroupSelection> = match filter {
        Some(f) => Some(source.select_row_groups_matching(f)?),
        None => None,
    };
    Ok(match (bbox_selection, filter_selection) {
        (Some(a), Some(b)) => Some(a.intersect(&b)),
        (Some(a), None) => Some(a),
        (None, Some(b)) => Some(b),
        (None, None) => None,
    })
}

/// Unsigned area of a polygonal geometry in CRS units², 0 for anything else
/// (the accumulator's per-feature input, #384).
fn polygon_area_f32(g: &Geometry<f64>) -> f32 {
    match g {
        Geometry::Polygon(p) => p.unsigned_area() as f32,
        Geometry::MultiPolygon(mp) => mp.unsigned_area() as f32,
        _ => 0.0,
    }
}

/// Run the tiny-polygon accumulator (#384) for the streaming path: per
/// planned level, the sorted row indices of its carriers; empty per level
/// unless the accumulator applies there.
fn streaming_carriers(
    options: &ConvertOptions,
    features: &[AssignFeature],
    feat_min_levels: &[u8],
    areas: Vec<f32>,
    level_gsds: &[f64],
    level_reprs: &[Representation],
    crs: Crs,
) -> Vec<Vec<usize>> {
    let finest_planned = level_gsds.len().saturating_sub(1);
    let acc_levels: Vec<AccumulateLevel> = level_gsds
        .iter()
        .enumerate()
        .map(|(l, &gsd)| AccumulateLevel {
            gsd_meters: gsd,
            enabled: l != finest_planned
                && accumulator_enabled(options)
                && level_accumulates(options.simplify.collapse, level_reprs[l]),
        })
        .collect();
    if !acc_levels.iter().any(|l| l.enabled) {
        return vec![Vec::new(); level_gsds.len()];
    }
    let t = Instant::now();
    let carriers = tiny_polygon_carriers(
        features,
        feat_min_levels,
        &areas,
        &acc_levels,
        crs,
        options.simplify.factor,
    );
    let total: usize = carriers.iter().map(Vec::len).sum();
    log::info!(
        "[convert] tiny-polygon accumulator: {total} placeholder square(s) across {} \
         level(s) stand in for the polygons those levels dropped ({:.2}s)",
        acc_levels.iter().filter(|l| l.enabled).count(),
        t.elapsed().as_secs_f64()
    );
    carriers
}

/// Whether the tiny-polygon accumulator (#384) is in play for this run:
/// duplicating mode (a carrier is a second appearance of a feature, which
/// partitioning's feature-once contract cannot represent) with the square
/// disposition somewhere — globally via `--collapse-square`, or in a
/// `--representation` square band.
fn accumulator_enabled(options: &ConvertOptions) -> bool {
    matches!(options.mode, Mode::Duplicating)
        && (options.simplify.collapse == CollapseMode::Square
            || options
                .representation
                .iter()
                .any(|b| b.repr == Representation::Square))
}

/// Prebuild every non-verbatim level's coalesce chain table.
///
/// The single read fans each batch out to all levels at once, so every level's
/// table has to exist before the read starts. Deterministic and keyed by rep
/// row, so the result is byte-identical to the former per-level build.
fn build_pass2_coalesce_tables(
    coalesce_scratch: Option<&CoalesceScratch>,
    emitted: &[EmitLevel],
    finest: usize,
    crs: Crs,
    options: &ConvertOptions,
) -> Vec<Option<CoalesceTable>> {
    match coalesce_scratch {
        Some(scratch) => {
            log::info!(
                "[convert] building coalesce chain tables for {} level(s)",
                emitted.len()
            );
            let inputs = scratch.inputs();
            emitted
                .par_iter()
                .map(|e| {
                    let verbatim =
                        matches!(options.mode, Mode::Partitioning) || e.orig as usize == finest;
                    (!verbatim).then(|| {
                        build_level_coalesce_table(
                            &inputs,
                            e.orig as usize,
                            finest,
                            e.gsd,
                            crs,
                            options,
                        )
                    })
                })
                .collect()
        }
        None => std::iter::repeat_with(|| None)
            .take(emitted.len())
            .collect(),
    }
}

/// Cascading (#218): per level, the fine→coarse GSD chain from the finest
/// non-canonical level down to (and including) that level.
///
/// Chains are built from the emitted plan, so the Serial fold, the pipelined
/// incremental fold, and the in-memory path all step through the same GSD
/// sequence. The chain is empty when cascading does not apply to the level.
/// Zoom-band representation selector (#317 / #279): steps carry each
/// contributing level's representation so the fold pointifies / squarifies at
/// the right band level (see `simplify_cascade`).
fn build_cascade_chains(
    emitted: &[EmitLevel],
    finest: usize,
    duplicating: bool,
    options: &ConvertOptions,
) -> Vec<Vec<CascadeStep>> {
    let repr_of =
        |zoom: Option<u8>| super::convert::representation_for_zoom(&options.representation, zoom);
    emitted
        .iter()
        .map(|e| {
            let verbatim = matches!(options.mode, Mode::Partitioning) || e.orig as usize == finest;
            if !duplicating || verbatim || !options.simplify.cascade {
                return Vec::new();
            }
            let mut chain: Vec<CascadeStep> = emitted
                .iter()
                .filter(|f| (f.orig as usize) < finest && f.orig >= e.orig)
                .map(|f| CascadeStep {
                    gsd_meters: f.gsd,
                    repr: repr_of(f.zoom),
                })
                .collect();
            chain.reverse();
            chain
        })
        .collect()
}

/// Everything the per-level pass-2 contexts borrow from the convert driver.
///
/// `LevelStreamCtx` holds a dozen borrows into pass-1 state. Threading them
/// through as separate parameters made the builder's signature longer than its
/// body, so they travel together.
struct LevelCtxInputs<'a> {
    source_schema: &'a Schema,
    cluster_schema: &'a Schema,
    out_schema: &'a Schema,
    non_geom_cols: &'a [usize],
    geom_idx: usize,
    min_levels: &'a [u8],
    acc_cols: &'a [usize],
    kinds: Option<&'a [FeatureKind]>,
    cluster_tables: Option<&'a ClusterTables>,
    coalesce_tables: &'a [Option<CoalesceTable>],
    cascade_chains: &'a [Vec<CascadeStep>],
    /// Per planned level, the tiny-polygon accumulator's carrier rows (#384).
    carriers: &'a [Vec<usize>],
    crs: Crs,
    finest: usize,
    duplicating: bool,
}

/// Build one `LevelStreamCtx` per emitted level, in level order.
fn build_level_ctxs<'a>(
    emitted: &[EmitLevel],
    options: &'a ConvertOptions,
    inputs: &LevelCtxInputs<'a>,
) -> Vec<LevelStreamCtx<'a>> {
    let repr_of =
        |zoom: Option<u8>| super::convert::representation_for_zoom(&options.representation, zoom);
    emitted
        .iter()
        .enumerate()
        .map(|(i, e)| {
            let verbatim =
                matches!(options.mode, Mode::Partitioning) || e.orig as usize == inputs.finest;
            LevelStreamCtx {
                source_schema: inputs.source_schema,
                cluster_schema: inputs.cluster_schema,
                out_schema: inputs.out_schema,
                non_geom_cols: inputs.non_geom_cols,
                geom_idx: inputs.geom_idx,
                min_levels: inputs.min_levels,
                orig_level: e.orig,
                duplicating: inputs.duplicating,
                verbatim,
                gsd_m: e.gsd,
                repr: repr_of(e.zoom),
                crs: inputs.crs,
                simplify: &options.simplify,
                cluster_enabled: options.cluster,
                // Canonical level: singleton clusters, columns verbatim (§2.4).
                cluster_table: inputs
                    .cluster_tables
                    .filter(|_| e.orig as usize != inputs.finest)
                    .map(|t| &t[e.orig as usize]),
                acc_cols: inputs.acc_cols,
                coalesce_enabled: options.coalesce_lines,
                kinds: inputs.kinds,
                coalesce_table: inputs.coalesce_tables[i].as_ref(),
                cascade_chain: &inputs.cascade_chains[i],
                carriers: &inputs.carriers[e.orig as usize],
            }
        })
        .collect()
}

/// Run pass 2 over every emitted level and return `(outcome, rows, vertices)`
/// per level, in level order.
///
/// The outcome distinguishes a written level from one the writer skipped
/// because every candidate collapsed during simplification (#211).
#[allow(clippy::too_many_arguments)]
fn run_pass2_levels(
    writer: &mut OverviewWriter<File>,
    ctxs: &[LevelStreamCtx<'_>],
    hints: &[usize],
    source: &ConvertSource,
    options: &ConvertOptions,
    selected_row_groups: Option<&RowGroupSelection>,
    in_flight_batches: usize,
    out_schema: &Schema,
    num_rows: usize,
    geom_bytes: u64,
    strategy: Pass2Strategy,
) -> Result<Vec<(LevelWriteOutcome, usize, usize)>, ConvertError> {
    let n = ctxs.len();
    let level_stats: Vec<(LevelWriteOutcome, usize, usize)> = match strategy {
        // Reference: one in-order re-read per level (pre-#213 behavior).
        Pass2Strategy::Serial => ctxs
            .iter()
            .enumerate()
            .map(|(i, ctx)| {
                write_level_streaming(
                    writer,
                    i,
                    hints[i],
                    source,
                    options.read_batch_size,
                    in_flight_batches,
                    selected_row_groups,
                    ctx,
                )
            })
            .collect::<Result<_, _>>()?,
        // Production: buffer levels 0..n-1 from a single read, then stream the
        // finest (verbatim, largest) level last straight into the writer.
        Pass2Strategy::Pipelined => {
            let buffered_rows: usize = hints[..n - 1].iter().sum();
            // #305: pass 1's measured average encoded-geometry size per input
            // row sizes the RAM-vs-spill estimate (falls back to calibrated
            // constants on an empty scan). Same decision timing as before —
            // pass 1 has always completed by this point (the assign barrier).
            let avg_geom_bytes = (num_rows > 0).then(|| geom_bytes / num_rows as u64);
            let backing = pipeline::resolve_backing(
                options.profile,
                options.mode,
                buffered_rows,
                avg_geom_bytes,
            );
            log::info!(
                "[convert] pass 2: building {n} overview level(s) from a \
                 single read (finest level streamed last)"
            );
            let mut stats = if n > 1 {
                pipeline::run_pass2_buffered(
                    writer,
                    &ctxs[..n - 1],
                    &hints[..n - 1],
                    source,
                    options.read_batch_size,
                    selected_row_groups,
                    in_flight_batches,
                    backing,
                    out_schema,
                )?
            } else {
                Vec::new()
            };
            stats.push(write_level_streaming(
                writer,
                n - 1,
                hints[n - 1],
                source,
                options.read_batch_size,
                in_flight_batches,
                selected_row_groups,
                &ctxs[n - 1],
            )?);
            stats
        }
    };
    Ok(level_stats)
}

/// The resolved ranking tier: the per-row sort keys (absent for the size
/// fallback), the provenance record, and the per-row class groups that line
/// coalescing needs (present only for the class-based tiers).
type ResolvedRanking = (
    Option<Vec<Option<f64>>>,
    RankingProvenance,
    Option<Vec<u32>>,
);

/// Pick the ranking tier, in the same order and with the same logging as the
/// in-memory path.
fn resolve_ranking_tier(
    plan: RankPlan,
    explicit_keys: Vec<Option<f64>>,
    confidence_keys: Vec<Option<f64>>,
    explicit_groups: Vec<u32>,
    collect_lines: bool,
    feature_count: usize,
    point_count: usize,
) -> ResolvedRanking {
    let n = feature_count;
    let size_fallback = || {
        log::info!(
            "overview ranking: no sort key specified or auto-detected; using size + \
             deterministic-hash fallback"
        );
        RankingProvenance {
            mode: "size-fallback".to_string(),
            column: None,
            ranks: None,
            unknown_rank: None,
        }
    };

    // Resolve the tier (same order + logging as the in-memory path). The
    // third element is the all-row class-group vector for coalescing, present
    // only for the class-based tiers (matches `coalesce_group_column`).
    match plan {
        RankPlan::ExplicitSort { name, .. } => {
            log::info!("overview ranking: explicit numeric sort-key column {name:?}");
            (
                Some(explicit_keys),
                RankingProvenance {
                    mode: "explicit-sort-key".to_string(),
                    column: Some(name),
                    ranks: None,
                    unknown_rank: None,
                },
                None,
            )
        }
        RankPlan::ExplicitClass { ranking, .. } => {
            log::info!(
                "overview ranking: explicit class-ranking on column {:?} ({} named classes, unknown_rank={})",
                ranking.column,
                ranking.ranks.len(),
                ranking.unknown_rank
            );
            (
                Some(explicit_keys),
                class_ranking_provenance("class-ranking", &ranking),
                collect_lines.then_some(explicit_groups),
            )
        }
        RankPlan::Auto { roads, confidence } => {
            if let Some(cand) = roads
                .into_iter()
                .find(|c| c.found.len() >= ROAD_VOCAB_MIN_DISTINCT)
            {
                log::info!(
                    "overview ranking: auto-detected Overture road classes in column {:?}; \
                     applying built-in ranking (motorway > … > service > tail)",
                    cand.ranking.column
                );
                let prov = class_ranking_provenance("auto-overture-roads", &cand.ranking);
                (Some(cand.keys), prov, collect_lines.then_some(cand.groups))
            } else if let Some((_, col_name)) = confidence.filter(|_| n > 0 && point_count * 2 >= n)
            {
                log::info!(
                    "overview ranking: auto-detected Overture places confidence column {col_name:?} \
                     (numeric point ranking)"
                );
                (
                    Some(confidence_keys),
                    RankingProvenance {
                        mode: "auto-confidence".to_string(),
                        column: Some(col_name),
                        ranks: None,
                        unknown_rank: None,
                    },
                    None,
                )
            } else {
                (None, size_fallback(), None)
            }
        }
        RankPlan::SizeFallback => (None, size_fallback(), None),
    }
}

/// The writer and the three schemas pass 2 encodes against.
struct LevelWriter {
    writer: OverviewWriter<File>,
    source_schema: Schema,
    cluster_schema: Schema,
    out_schema: Schema,
    /// Input-schema indices of every column except the geometry column.
    non_geom_cols: Vec<usize>,
}

#[allow(clippy::too_many_arguments)]
fn create_level_writer(
    output_path: &Path,
    input_schema: &Schema,
    geom_idx: usize,
    geom_field: &Field,
    emitted: &[EmitLevel],
    crs: Crs,
    ranking_provenance: RankingProvenance,
    renames: &[(String, String)],
    options: &ConvertOptions,
) -> Result<LevelWriter, ConvertError> {
    // --- Writer setup (identical to the in-memory path). ---------------------
    // Writer schemas: base + point_count when clustering (Q4) + coalesced_count
    // when coalescing (Q3).
    let geom_name = geom_field.name().clone();
    let (source_schema, cluster_schema, out_schema) =
        build_level_schemas(input_schema, geom_idx, &geom_name, options);

    let writer_levels: Vec<LevelSpec> = emitted
        .iter()
        .map(|e| LevelSpec::new(e.gsd, e.zoom))
        .collect();
    let emitted_gsds: Vec<f64> = emitted.iter().map(|e| e.gsd).collect();
    let writer_opts = build_writer_options(
        writer_levels,
        &emitted_gsds,
        crs,
        ranking_provenance,
        renames,
        options,
    );

    let writer = OverviewWriter::create(output_path, &out_schema, writer_opts)?;

    let non_geom_cols: Vec<usize> = (0..input_schema.fields().len())
        .filter(|&c| c != geom_idx)
        .collect();

    Ok(LevelWriter {
        writer,
        source_schema,
        cluster_schema,
        out_schema,
        non_geom_cols,
    })
}

/// The winner tables: which level each row belongs to, and how many rows each
/// level gets.
///
/// Built from the pass-1 feature scratch, which this stage frees before it
/// returns. Everything here is O(dataset); pass 2 only carries the row-indexed
/// `min_levels` byte table plus the cluster and coalesce tables.
struct WinnerTables {
    /// The resolved level plan: `(gsd, zoom)` per planned level.
    level_specs: Vec<(f64, Option<u8>)>,
    /// Cluster tables (Q4), or `None` when clustering is off.
    cluster_tables: Option<ClusterTables>,
    /// Per-row geometry kinds (Q3), or `None` when line coalescing is off.
    kinds: Option<Vec<FeatureKind>>,
    /// The pass-1 line scratch, kept only when coalescing survives the memory
    /// guard.
    coalesce_scratch: Option<CoalesceScratch>,
    /// Coarsest level per INPUT ROW; [`UNASSIGNED_LEVEL`] for skipped rows.
    min_levels: Vec<u8>,
    /// Per-level winner counts, cumulative in duplicating mode.
    counts: Vec<usize>,
    /// Per planned level, the sorted row indices of the tiny-polygon
    /// accumulator's carriers (#384); empty per level unless it applies.
    carriers: Vec<Vec<usize>>,
    finest: usize,
}

#[allow(clippy::too_many_arguments)]
fn resolve_winner_tables(
    features: &mut Vec<AssignFeature>,
    acc_values: Vec<Vec<Option<f64>>>,
    areas: Vec<f32>,
    coalesce_scratch: Option<CoalesceScratch>,
    num_rows: usize,
    crs: Crs,
    options: &ConvertOptions,
    peak_rss_mib: &mut f64,
) -> Result<WinnerTables, ConvertError> {
    // --- Winner tables (assignment + Q2 density budget). ---------------------
    let level_specs = options.levels.resolve(options.gsd_base)?;
    let level_gsds: Vec<f64> = level_specs.iter().map(|(g, _)| *g).collect();

    let t_assign = Instant::now();
    // #306: cap the transient winner-grid memory (the pass-1 peak #300's [rss]
    // logs pinned) at the profile-derived RAM budget; `speed` stays unbounded.
    // Zoom-band representation selector (#317 / #279): per-level
    // representations, parallel to the plan.
    let level_reprs = super::convert::level_representations(&level_specs, &options.representation);
    let assignment = assign_levels_bounded(
        features,
        &level_gsds,
        &options.assign,
        crs,
        super::pipeline::pass1_grid_budget_bytes(options.profile),
        &level_reprs,
    );
    let assign_secs = t_assign.elapsed().as_secs_f64();
    let t_budget = Instant::now();
    let assignment = if options.density.enabled {
        apply_density_budget(
            &assignment,
            features,
            &level_gsds,
            &options.assign,
            &options.density,
            crs,
        )
    } else {
        assignment
    };
    log::debug!(
        "[profile] assignment+budget: {:.2}s (assign {:.2}s + budget {:.2}s)",
        t_assign.elapsed().as_secs_f64(),
        assign_secs,
        t_budget.elapsed().as_secs_f64()
    );
    log::info!(
        "[convert] level assignment complete: {} level(s) in {:.1}s",
        level_gsds.len(),
        t_assign.elapsed().as_secs_f64()
    );
    // Pass-1 winner tables (bboxes/kinds/sort-keys for every feature across all
    // levels) are the O(dataset) peak candidate flagged in #295.
    log_phase_rss("assignment+budget (winner tables)", peak_rss_mib);

    // The feature-parallel winner table (coarsest level per FEATURE, in
    // `features` order) feeds the cluster stage and the per-level counts.
    let feat_min_levels: Vec<u8> = assignment.assignments.iter().map(|a| a.min_level).collect();
    drop(assignment);

    // #384: tiny-polygon accumulator — per level, the carriers that stand in
    // for the sub-visible polygons the level dropped. Row-indexed like the
    // winner table; empty per level unless the accumulator applies there.
    let carriers = streaming_carriers(
        options,
        features,
        &feat_min_levels,
        areas,
        &level_gsds,
        &level_reprs,
        crs,
    );

    // Cluster tables (Q4): built from the pass-1 features + final winner
    // table, before the O(N) scratch is freed. Memory afterwards is
    // O(non-singleton clusters), carried into pass 2 alongside `min_levels`.
    // Accumulate values are extracted per ROW; the cluster stage indexes them
    // by feature position, so remap through each feature's row index.
    let cluster_tables: Option<ClusterTables> = if options.cluster {
        let acc_feat: Vec<Vec<Option<f64>>> = acc_values
            .iter()
            .map(|vals| features.iter().map(|f| vals[f.index]).collect())
            .collect();
        Some(super::convert::build_verified_cluster_tables(
            features,
            &feat_min_levels,
            &level_gsds,
            &acc_feat,
            crs,
            options,
        )?)
    } else {
        None
    };
    drop(acc_values);

    // Coalescing (Q3): keep the per-row kinds (1 byte/row — line rows bypass
    // the winner table at coalesced levels; skipped-geometry rows default to
    // Point, which never matches the Line bypass) and the pass-1 line
    // scratch; apply the memory guard.
    let coalesce_on = coalesce_effective(
        options,
        coalesce_scratch.as_ref().map_or(0, |s| s.rows.len()),
    );
    let kinds: Option<Vec<FeatureKind>> = options.coalesce_lines.then(|| {
        let mut k = vec![FeatureKind::Point; num_rows];
        for f in features.iter() {
            k[f.index] = f.kind;
        }
        k
    });
    let coalesce_scratch = coalesce_scratch.filter(|_| coalesce_on);

    let num_levels = level_gsds.len();
    let finest = num_levels.saturating_sub(1);

    // The ROW-indexed winner table pass 2 addresses (`row_offset + i`), one
    // byte per input row. Skipped-geometry rows keep the UNASSIGNED sentinel,
    // which matches no level in either mode (the level plan is capped at
    // [`super::convert::MAX_LEVELS`] levels, so `finest < u8::MAX`).
    let mut min_levels = vec![UNASSIGNED_LEVEL; num_rows];
    for (f, &ml) in features.iter().zip(&feat_min_levels) {
        min_levels[f.index] = ml;
    }

    // Per-level winner counts (exact row counts in partitioning mode; in
    // duplicating mode exact up to simplification drops — used as the writer's
    // row-group sizing hint and for empty-level omission). With coalescing,
    // line rows leave the winner table at non-canonical levels: their count
    // is the level's surviving chain count instead (computed by running the
    // chain stage per level — cheap relative to decode; the tables are
    // rebuilt, with simplification, per level in pass 2 rather than held for
    // every level at once).
    let mut hist = vec![0usize; num_levels];
    for (f, &ml) in features.iter().zip(&feat_min_levels) {
        if coalesce_scratch.is_some() && f.kind == FeatureKind::Line {
            continue; // counted via the per-level chain stage below
        }
        hist[(ml as usize).min(finest)] += 1;
    }
    drop(feat_min_levels);
    features.clear();
    features.shrink_to_fit(); // free the pass-1 O(N)·48B scratch before pass 2
    let mut counts: Vec<usize> = match options.mode {
        Mode::Duplicating => hist
            .iter()
            .scan(0usize, |acc, &c| {
                *acc += c;
                Some(*acc)
            })
            .collect(),
        Mode::Partitioning => hist,
    };
    // #384: carriers are members of their level only (not of finer ones —
    // there the feature is either a real member already or absent).
    for (count, level_carriers) in counts.iter_mut().zip(&carriers) {
        *count += level_carriers.len();
    }
    if let Some(scratch) = &coalesce_scratch {
        // Duplicating only (partitioning + coalescing is rejected upstream).
        let inputs = scratch.inputs();
        #[allow(clippy::needless_range_loop)]
        for level in 0..num_levels {
            if level == finest {
                counts[level] += scratch.rows.len(); // canonical: verbatim
            } else {
                counts[level] +=
                    coalesce_level_chains(&inputs, level, finest, level_gsds[level], crs, options)
                        .len();
            }
        }
    }

    Ok(WinnerTables {
        level_specs,
        cluster_tables,
        kinds,
        coalesce_scratch,
        min_levels,
        counts,
        carriers,
        finest,
    })
}

/// Footer-only preparation: everything the convert driver settles before the
/// first data page is read.
///
/// Schema and CRS checks, reserved-column renames (#288), attribute-filter
/// binding (#315), row-group pruning (#102 / #315), and the pass-0 staging of
/// the selected groups to local disk (#286/#287).
struct Preflight {
    /// `options` with reserved-column renames applied; the driver borrows this
    /// for the rest of the conversion.
    options: ConvertOptions,
    input_schema: SchemaRef,
    crs: Crs,
    renames: Vec<(String, String)>,
    geom_idx: usize,
    geom_field: Field,
    /// Schema indices of the accumulate columns (Q4).
    acc_cols: Vec<usize>,
    bbox_units: Option<[f64; 4]>,
    bound_filter: Option<super::filter::BoundFilter>,
    selected_row_groups: Option<RowGroupSelection>,
    row_groups_total: usize,
    row_groups_read: usize,
}

fn convert_preflight(
    source: &ConvertSource,
    options: &ConvertOptions,
) -> Result<Preflight, ConvertError> {
    // Schema checks (level column, geometry column) — footer-only reads.
    // (For a remote source, #210, the footer is range-fetched once here and
    // cached across the passes below. For a multi-partition source the
    // schema is the validated union schema and the key-value metadata is
    // partition 0's — construction proved all parts agree.)
    let input_schema: SchemaRef = source.schema()?;

    // CRS detection + rejection (spec Q3) — footer metadata only.
    let kv = source.key_value_metadata()?;
    let crs = super::convert::detect_crs_from_kv(kv.as_ref())?;

    // Reserved-column collisions (#288) are resolved BEFORE row-group
    // selection so the attribute filter (#315) can bind against the final
    // (possibly renamed) schema; the rename is metadata-only and never
    // affects the footer statistics either pruning path reads. See the
    // full #288 rationale on the block below.
    let mut resolved = options.clone();
    let (input_schema, renames) = resolve_reserved_column_collisions(&input_schema, &mut resolved);
    let options = &resolved;

    // Attribute filter (#315): parse + bind against the input schema.
    // Syntax was already validated in `validate_options`; binding resolves
    // column names to indices and type-checks literals.
    let bound_filter = super::convert::bind_attribute_filter(options, &input_schema, &renames)?;

    // Regional extract (#102) + attribute filter (#315): prune input row
    // groups by footer statistics — bbox covering stats for `--bbox`,
    // per-column min/max/null-count stats for `--filter` — before any data
    // pages are read. The two prunings compose by intersection. The
    // selection is PER PART and every pass reads the same selection in the
    // same part order, so the global row indices addressing the winner
    // tables stay aligned. Groups without stats are kept; the exact
    // per-feature filters in pass 1 guarantee identical output either way.
    let row_groups_total = source.num_row_groups_total()?;
    let bbox_units = options
        .bbox
        .map(|b| super::convert::bbox_to_crs_units(&b, crs));
    let selected_row_groups =
        select_row_groups_streaming(source, bbox_units.as_ref(), bound_filter.as_ref())?;
    let row_groups_read = selected_row_groups
        .as_ref()
        .map_or(row_groups_total, RowGroupSelection::total_selected);
    if selected_row_groups.is_some() {
        let what = super::convert::pruning_label(options.bbox.is_some(), bound_filter.is_some());
        log::info!("{what} filter: reading {row_groups_read}/{row_groups_total} input row groups");
    }
    // #267: nudge toward --bbox / download-first for a large whole-file remote
    // convert (quiet for local inputs and effective bbox extracts).
    super::convert::warn_full_file_remote(source, row_groups_read, row_groups_total);
    // #272: preflight the spill volume. The disk spill (#219) grows to ≈ the
    // selected input bytes — known exactly here, the first moment after
    // row-group selection (summed per part for a multi source) — so compare
    // it against the free space where the spill will live and warn up front
    // (naming the dir and the shortfall) instead of silently degrading to
    // network re-fetch mid-convert.
    super::convert::warn_spill_space(
        source,
        source.selected_input_bytes(selected_row_groups.as_ref())?,
        options.spill_dir.as_deref(),
    );

    // Pass 0 (#286/#287): stage the selected row groups to local disk so both
    // passes below read from the spill, not the network.
    stage_input_pass0(source, selected_row_groups.as_ref(), row_groups_read);

    // (Reserved-column collisions, #288, were resolved above, before the
    // row-group selection: any input column named `level` / `point_count` /
    // `coalesced_count` (case-insensitive) is renamed instead of rejecting
    // the file, keeping the reserved output columns authoritative. The
    // rename preserves column order, so the projection indices pass 1
    // computes against `input_schema` stay valid against the raw file, and
    // pass 2 relabels non-geometry columns positionally into the renamed
    // source schema (`build_source_schema`). `options` was cloned so
    // by-name ranking/accumulate options could be rewritten to the renamed
    // columns.)

    let geom_idx = find_geometry_column(&input_schema).ok_or(ConvertError::NoGeometryColumn)?;
    let geom_field = input_schema.field(geom_idx).clone();

    // Clustering schema checks + accumulate column resolution (Q4).
    let acc_cols = validate_cluster_schema(&input_schema, options)?;
    // Coalescing schema check (Q3).
    validate_coalesce_schema(&input_schema, options)?;

    Ok(Preflight {
        options: resolved.clone(),
        input_schema,
        crs,
        renames,
        geom_idx,
        geom_field,
        acc_cols,
        bbox_units,
        bound_filter,
        selected_row_groups,
        row_groups_total,
        row_groups_read,
    })
}

pub(crate) fn convert_streaming_strategy(
    source: &ConvertSource,
    output_path: &Path,
    options: &ConvertOptions,
    strategy: Pass2Strategy,
) -> Result<ConvertReport, ConvertError> {
    let start = Instant::now();
    // #295: peak-RSS-by-phase instrumentation. Each phase boundary logs process
    // RSS; the max is reported at the end so a single run shows both the peak
    // and which phase produced it.
    let mut peak_rss_mib = 0.0f64;

    if options.sort_key.is_some() && options.class_ranking.is_some() {
        return Err(ConvertError::RankingConflict);
    }

    let Preflight {
        options: resolved_options,
        input_schema,
        crs,
        renames,
        geom_idx,
        geom_field,
        acc_cols,
        bbox_units,
        bound_filter,
        selected_row_groups,
        row_groups_total,
        row_groups_read,
    } = convert_preflight(source, options)?;
    let options = &resolved_options;

    // --- Pass 1: stream → AssignFeatures + resolved ranking. -----------------
    let t_pass1 = Instant::now();
    let Pass1Output {
        mut features,
        areas,
        provenance: ranking_provenance,
        acc_values,
        coalesce: coalesce_scratch,
        num_rows,
        skipped_rows,
        geom_bytes,
    } = run_pass1(
        source,
        &input_schema,
        geom_idx,
        options,
        &acc_cols,
        selected_row_groups.as_ref(),
        bbox_units.as_ref(),
        bound_filter.as_ref(),
    )?;
    if skipped_rows > 0 {
        log::warn!(
            "skipping {skipped_rows} of {num_rows} input rows with a null, \
             empty, or non-finite geometry"
        );
    }
    let num_features = features.len();

    // #188 follow-up: count antimeridian-suspect bboxes and warn once.
    let antimeridian_suspect_features = features
        .iter()
        .filter(|f| super::convert::bbox_antimeridian_suspect(&f.bbox, crs))
        .count();
    super::convert::warn_antimeridian_suspects(antimeridian_suspect_features);

    // Stage markers (#242): everything between pass 1 and the writer used to
    // run in total info-level silence — on planet-scale inputs that was tens
    // of minutes with no output.
    log::info!("[convert] scan complete: {num_features} feature(s) from {num_rows} row(s)");
    log::debug!(
        "[profile] pass1 stream+scan: {:.2}s",
        t_pass1.elapsed().as_secs_f64()
    );
    log_phase_rss("pass1 scan", &mut peak_rss_mib);

    let WinnerTables {
        level_specs,
        cluster_tables,
        kinds,
        coalesce_scratch,
        min_levels,
        counts,
        carriers,
        finest,
    } = resolve_winner_tables(
        &mut features,
        acc_values,
        areas,
        coalesce_scratch,
        num_rows,
        crs,
        options,
        &mut peak_rss_mib,
    )?;

    // Planned levels with no winners are omitted (§7.3, #211 auto-clamp);
    // record them for the report + warning.
    let (emitted, mut skipped) = partition_emitted_levels(&level_specs, &counts);
    if emitted.is_empty() {
        return Err(ConvertError::NoData);
    }
    warn_plan_skipped_levels(&skipped, num_features, emitted[0].gsd, emitted[0].zoom);

    let LevelWriter {
        mut writer,
        source_schema,
        cluster_schema,
        out_schema,
        non_geom_cols,
    } = create_level_writer(
        output_path,
        &input_schema,
        geom_idx,
        &geom_field,
        &emitted,
        crs,
        ranking_provenance,
        &renames,
        options,
    )?;

    // --- Pass 2: single-read pipelined engine + canonical streamed last. -----
    // Pass-1 O(N) scratch has been freed by here; this marks the memory floor
    // the pass-2 output sink builds on (its ceiling is the #294 auto choice).
    log_phase_rss("pre-pass2 (winner tables freed)", &mut peak_rss_mib);
    let t_pass2 = Instant::now();

    let coalesce_tables =
        build_pass2_coalesce_tables(coalesce_scratch.as_ref(), &emitted, finest, crs, options);

    let duplicating = matches!(options.mode, Mode::Duplicating);
    let cascade_chains = build_cascade_chains(&emitted, finest, duplicating, options);
    let ctxs = build_level_ctxs(
        &emitted,
        options,
        &LevelCtxInputs {
            source_schema: &source_schema,
            cluster_schema: &cluster_schema,
            out_schema: &out_schema,
            non_geom_cols: &non_geom_cols,
            geom_idx,
            min_levels: &min_levels,
            acc_cols: &acc_cols,
            kinds: kinds.as_deref(),
            cluster_tables: cluster_tables.as_ref(),
            coalesce_tables: &coalesce_tables,
            cascade_chains: &cascade_chains,
            carriers: &carriers,
            crs,
            finest,
            duplicating,
        },
    );

    let hints: Vec<usize> = emitted.iter().map(|e| e.hint).collect();

    // Snapshot for the end-of-pass-2 summary: the counter is process-wide,
    // so report the delta from this conversion only (#242).
    let validation_skips_before = validation_skip_count();

    // `(outcome, rows, vertices)` per emitted level, in level order. The
    // outcome distinguishes a written level from one the writer skipped because
    // every candidate collapsed during simplification (#211).
    // Resolve the in-flight depth once (auto-sizes from available cores when
    // the caller left it at IN_FLIGHT_BATCHES_AUTO) and surface it (#264).
    let in_flight_batches = resolve_and_log_in_flight_batches(options.in_flight_batches);

    let level_stats = run_pass2_levels(
        &mut writer,
        &ctxs,
        &hints,
        source,
        options,
        selected_row_groups.as_ref(),
        in_flight_batches,
        &out_schema,
        num_rows,
        geom_bytes,
        strategy,
    )?;
    log_validation_skips(validation_skips_before);

    // Fold each emitted level's write outcome into the shared bookkeeping
    // (#211): `record_level_outcome` appends a renumbered `LevelReport` for a
    // written level, or — for a level the writer omitted because every
    // candidate collapsed during simplification — warns and records the plan in
    // `skipped`, exactly like a plan-time omission.
    let mut level_reports = Vec::with_capacity(emitted.len());
    for (e, (outcome, rows, vertices)) in emitted.iter().zip(level_stats) {
        record_level_outcome(
            outcome,
            SkippedLevelReport {
                planned_level: e.orig as usize,
                gsd: e.gsd,
                zoom: e.zoom,
            },
            e.hint,
            rows,
            vertices,
            &mut level_reports,
            &mut skipped,
        );
    }
    skipped.sort_by_key(|s| s.planned_level);
    if level_reports.is_empty() {
        // Every emitted level collapsed at write time: no valid overview file
        // can be produced (`levels` MUST be non-empty, §3.3).
        return Err(ConvertError::NoData);
    }

    log::debug!(
        "[profile] pass2 total: {:.2}s",
        t_pass2.elapsed().as_secs_f64()
    );
    log_phase_rss("pass2 (output sink)", &mut peak_rss_mib);

    let t_finish = Instant::now();
    let meta = writer.finish()?;
    log::debug!(
        "[profile] writer.finish: {:.2}s",
        t_finish.elapsed().as_secs_f64()
    );
    log_phase_rss("writer.finish", &mut peak_rss_mib);
    log::info!("[rss] convert peak: {peak_rss_mib:.0} MiB");
    fill_level_bytes(output_path, &meta, &mut level_reports)?;

    let total_rows: usize = level_reports.iter().map(|l| l.feature_count).sum();
    let total_vertices: usize = level_reports.iter().map(|l| l.vertex_count).sum();
    let total_compressed_bytes: i64 = level_reports.iter().map(|l| l.compressed_bytes).sum();

    Ok(ConvertReport {
        mode: options.mode,
        levels: level_reports,
        skipped_empty_levels: skipped,
        input_features: num_features,
        total_rows,
        total_vertices,
        total_compressed_bytes,
        row_groups_total,
        row_groups_read,
        antimeridian_suspect_features,
        duration_secs: start.elapsed().as_secs_f64(),
        remote_fetch: super::convert::log_remote_fetch(source),
    })
}

// ============================================================================
// Pass 1: streaming feature scan + ranking resolution
// ============================================================================

/// A candidate Overture road-class column tracked incrementally during pass 1.
struct RoadCandidate {
    idx: usize,
    ranking: ClassRanking,
    /// Distinct known-vocabulary classes seen so far (detection gate).
    found: HashSet<&'static str>,
    /// Per-row class-rank keys, extracted as we stream.
    keys: Vec<Option<f64>>,
    /// Per-row interned class values (coalescing groups, Q3). Populated
    /// only when coalescing is enabled.
    groups: Vec<u32>,
    interner: GroupInterner,
}

/// The ranking tier resolved from the options + schema *before* reading data
/// (Q1). Mirrors `convert::resolve_ranking`'s tier order; the auto tier needs
/// data (vocab overlap, point majority) so its decision lands after pass 1.
enum RankPlan {
    ExplicitSort {
        idx: usize,
        name: String,
    },
    ExplicitClass {
        idx: usize,
        ranking: ClassRanking,
    },
    Auto {
        roads: Vec<RoadCandidate>,
        confidence: Option<(usize, String)>,
    },
    SizeFallback,
}

/// Build the [`RankPlan`] from the schema, validating explicit columns eagerly
/// (same error variants as the in-memory path).
fn build_rank_plan(schema: &Schema, options: &ConvertOptions) -> Result<RankPlan, ConvertError> {
    if let Some(name) = &options.sort_key {
        let idx = schema
            .index_of(name)
            .map_err(|_| ConvertError::SortKeyColumnMissing { name: name.clone() })?;
        return Ok(RankPlan::ExplicitSort {
            idx,
            name: name.clone(),
        });
    }
    if let Some(cr) = &options.class_ranking {
        let idx =
            schema
                .index_of(&cr.column)
                .map_err(|_| ConvertError::ClassRankColumnMissing {
                    name: cr.column.clone(),
                })?;
        let dt = schema.field(idx).data_type();
        if !matches!(dt, DataType::Utf8 | DataType::LargeUtf8) {
            return Err(ConvertError::ClassRankColumnNotString {
                name: cr.column.clone(),
                data_type: format!("{dt:?}"),
            });
        }
        return Ok(RankPlan::ExplicitClass {
            idx,
            ranking: cr.clone(),
        });
    }
    if !options.no_auto_rank {
        // Candidate Overture road-class columns, in schema order (the first
        // one passing the vocab-overlap gate wins, as in the in-memory path).
        let roads: Vec<RoadCandidate> = schema
            .fields()
            .iter()
            .enumerate()
            .filter(|(_, f)| {
                let lname = f.name().to_ascii_lowercase();
                (lname == "road_class" || lname == "class")
                    && matches!(f.data_type(), DataType::Utf8 | DataType::LargeUtf8)
            })
            .map(|(idx, f)| RoadCandidate {
                idx,
                ranking: overture_road_ranking(f.name().clone()),
                found: HashSet::new(),
                keys: Vec::new(),
                groups: Vec::new(),
                interner: GroupInterner::default(),
            })
            .collect();
        // Candidate Overture places confidence column (point-majority gate is
        // decided after pass 1, once kinds are known).
        let confidence = schema
            .fields()
            .iter()
            .enumerate()
            .find(|(_, f)| {
                f.name().eq_ignore_ascii_case("confidence")
                    && matches!(f.data_type(), DataType::Float32 | DataType::Float64)
            })
            .map(|(idx, f)| (idx, f.name().clone()));
        if !roads.is_empty() || confidence.is_some() {
            return Ok(RankPlan::Auto { roads, confidence });
        }
    }
    Ok(RankPlan::SizeFallback)
}

/// Incrementally scan a string column for known road classes, growing `found`
/// until it reaches [`ROAD_VOCAB_MIN_DISTINCT`] (then stops scanning).
fn scan_road_vocab(col: &dyn Array, found: &mut HashSet<&'static str>) {
    use arrow_array::cast::AsArray;

    if found.len() >= ROAD_VOCAB_MIN_DISTINCT {
        return;
    }
    let vocab: HashSet<&'static str> = KNOWN_ROAD_CLASSES.iter().copied().collect();

    macro_rules! scan {
        ($arr:expr) => {{
            let a = $arr;
            for i in 0..a.len() {
                if a.is_null(i) {
                    continue;
                }
                if let Some(&hit) = vocab.get(a.value(i)) {
                    found.insert(hit);
                    if found.len() >= ROAD_VOCAB_MIN_DISTINCT {
                        return;
                    }
                }
            }
        }};
    }
    match col.data_type() {
        DataType::Utf8 => scan!(col.as_string::<i32>()),
        DataType::LargeUtf8 => scan!(col.as_string::<i64>()),
        _ => {}
    }
}

/// Line geometries (+ compatibility groups) collected during pass 1 for the
/// coalescing stage (Q3). This is the streaming pipeline's one deliberate
/// residual `O(lines)` allocation: chaining needs a level's candidate line
/// geometries together, and the candidate set at every non-canonical
/// duplicating level is ALL lines (chains of sub-visibility fragments must
/// be reclaimable, so no winner-table pre-filter applies). Bounded by
/// [`ConvertOptions::coalesce_max_level_rows`]; beyond it coalescing is
/// skipped and this scratch is never built.
struct CoalesceScratch {
    /// Source row index per collected line, ascending input order.
    rows: Vec<usize>,
    /// The lines' decoded geometries, parallel to `rows`.
    geoms: Vec<Geometry<f64>>,
    /// Sort key per line (Q1 ranking), parallel to `rows`; filled after the
    /// ranking tier resolves.
    sort_keys: Vec<Option<f64>>,
    /// Interned class group per line, parallel to `rows`; `None` = no class
    /// ranking active (all lines compatible).
    groups: Option<Vec<u32>>,
}

impl CoalesceScratch {
    /// The per-level chaining inputs (borrowing the collected geometries).
    fn inputs(&self) -> Vec<CoalesceInput<'_>> {
        (0..self.rows.len())
            .map(|i| CoalesceInput {
                index: self.rows[i],
                geom: &self.geoms[i],
                sort_key: self.sort_keys[i],
                group: self.groups.as_ref().map_or(0, |g| g[i]),
            })
            .collect()
    }
}

/// Result of [`run_pass1`].
struct Pass1Output {
    /// Per-feature assignment inputs (bbox, kind, resolved sort key).
    features: Vec<AssignFeature>,
    /// Per-feature unsigned polygon area in CRS units² (0 for other kinds),
    /// parallel to `features`; empty unless the tiny-polygon accumulator is
    /// on (#384), since it is the one consumer.
    areas: Vec<f32>,
    /// Resolved ranking provenance (§3.5).
    provenance: RankingProvenance,
    /// Per-accumulate-spec source values (Q4), parallel to `acc_cols`.
    acc_values: Vec<Vec<Option<f64>>>,
    /// Line geometries + groups for coalescing (Q3); `None` unless enabled.
    coalesce: Option<CoalesceScratch>,
    /// Total input rows streamed (INCLUDING skipped-geometry rows): the
    /// domain of every row-indexed table pass 2 addresses.
    num_rows: usize,
    /// Rows skipped for a null, empty, or non-finite geometry (H4).
    skipped_rows: usize,
    /// Total in-memory Arrow byte size of the encoded geometry column across
    /// every scanned batch (#305). `geom_bytes / num_rows` is the measured
    /// average encoded-geometry size per input row that sizes the pass-2
    /// RAM-vs-spill decision; near-free to collect (one buffer-size sum per
    /// batch — no re-encode).
    geom_bytes: u64,
}

/// Pass 1: stream the input (geometry + ranking/accumulate columns only) and
/// produce the per-feature [`AssignFeature`]s (with resolved sort keys), the
/// ranking provenance block (§3.5), and — when clustering with aggregation —
/// the per-spec source values (parallel to `acc_cols`). Memory: `O(read
/// batch)` transient + `O(N)` small per-feature records.
/// The sorted, deduplicated column projection pass 1 reads: geometry +
/// ranking candidates + accumulate columns (Q4) + attribute-filter columns
/// (#315).
fn pass1_projection(
    geom_idx: usize,
    plan: &RankPlan,
    acc_cols: &[usize],
    filter: Option<&super::filter::BoundFilter>,
    ladder_col: Option<usize>,
) -> Vec<usize> {
    let mut cols: Vec<usize> = vec![geom_idx];
    // Entry-zoom ladder (#364): pass 1 decides the level, so its column has
    // to be read here even though nothing else in pass 1 looks at it.
    cols.extend(ladder_col);
    if let Some(f) = filter {
        cols.extend(f.columns().iter().copied());
    }
    match plan {
        RankPlan::ExplicitSort { idx, .. } | RankPlan::ExplicitClass { idx, .. } => cols.push(*idx),
        RankPlan::Auto { roads, confidence } => {
            cols.extend(roads.iter().map(|r| r.idx));
            if let Some((idx, _)) = confidence {
                cols.push(*idx);
            }
        }
        RankPlan::SizeFallback => {}
    }
    cols.extend(acc_cols.iter().copied());
    cols.sort_unstable();
    cols.dedup();
    cols
}

/// Stamp each feature's entry level from the ladder column (#364).
///
/// Resolved against the same level plan the buffered pipeline uses, so both
/// engines place a feature identically. A spec that yields no ladder — an
/// unusable column, a GSD-only plan — leaves every `entry_level` as `None`,
/// which is the "no ladder opinion" case the assignment already handles.
///
/// Lifted out of [`run_pass1`] rather than inlined: pass 1 is already at the
/// cognitive-complexity ceiling the workspace lints enforce, and this is a
/// self-contained step with no other reader in that function.
fn apply_entry_levels(
    options: &ConvertOptions,
    ladder_values: &[Option<f64>],
    num_rows: usize,
    features: &mut [AssignFeature],
) -> Result<(), ConvertError> {
    if options.entry_zoom.is_none() {
        return Ok(());
    }
    debug_assert_eq!(ladder_values.len(), num_rows);
    let level_specs = options.levels.resolve(options.gsd_base)?;
    if let Some(entry) = super::convert::resolve_entry_levels(options, ladder_values, &level_specs)?
    {
        for f in features.iter_mut() {
            f.entry_level = entry.get(f.index).copied().flatten();
        }
    }
    Ok(())
}

#[allow(clippy::too_many_arguments)]
fn run_pass1(
    source: &ConvertSource,
    input_schema: &Schema,
    geom_idx: usize,
    options: &ConvertOptions,
    acc_cols: &[usize],
    row_groups: Option<&RowGroupSelection>,
    bbox_units: Option<&[f64; 4]>,
    filter: Option<&super::filter::BoundFilter>,
) -> Result<Pass1Output, ConvertError> {
    let mut plan = build_rank_plan(input_schema, options)?;

    // Entry-zoom ladder column (#364), resolved by name against the (already
    // #288-renamed) schema so a `--magnitude-ladder level` on a source that
    // also has a reserved `level` still finds the caller's column.
    let ladder_col = options
        .entry_zoom
        .as_ref()
        .map(|spec| {
            input_schema.index_of(&spec.column).map_err(|_| {
                ConvertError::InvalidConfig(format!(
                    "entry-zoom column {:?} not found in the input schema",
                    spec.column
                ))
            })
        })
        .transpose()?;

    let cols = pass1_projection(geom_idx, &plan, acc_cols, filter, ladder_col);
    // Original schema index → projected batch column index.
    let proj = |orig: usize| cols.binary_search(&orig).expect("projected column");

    // Regional extract (#102): read only the bbox-selected row groups
    // (identical per-part selection in pass 2, keeping row indices aligned).
    let reader = source.open_stream(&ReadPlan {
        batch_size: options.read_batch_size.max(1),
        projection: Some(&cols),
        row_groups,
    })?;

    let mut features: Vec<AssignFeature> = Vec::new();
    // #384: polygon areas for the tiny-polygon accumulator, when it is on.
    let want_areas = accumulator_enabled(options);
    let mut areas: Vec<f32> = Vec::new();
    let mut num_rows = 0usize;
    let mut geom_bytes = 0u64;
    let mut skipped_rows = 0usize;
    let mut point_count = 0usize;
    let mut explicit_keys: Vec<Option<f64>> = Vec::new();
    let mut confidence_keys: Vec<Option<f64>> = Vec::new();
    let mut acc_values: Vec<Vec<Option<f64>>> = vec![Vec::new(); acc_cols.len()];
    // Entry-zoom ladder column values (#364), row-indexed.
    let mut ladder_values: Vec<Option<f64>> = Vec::new();
    let mut geoms_buf: Vec<Option<Geometry<f64>>> = Vec::new();
    // Coalescing (Q3): line rows + geometries, and — for an explicit class
    // ranking — the interned per-row class groups. `line_feat_pos` holds each
    // line's position in `features` (NOT its row index: skipped-geometry rows
    // make the two diverge).
    let collect_lines = options.coalesce_lines;
    let mut line_rows: Vec<usize> = Vec::new();
    let mut line_feat_pos: Vec<usize> = Vec::new();
    let mut line_geoms: Vec<Geometry<f64>> = Vec::new();
    let mut explicit_groups: Vec<u32> = Vec::new();
    let mut explicit_interner = GroupInterner::default();

    for batch in reader {
        let batch = batch?;
        let gcol_idx = proj(geom_idx);
        let schema = batch.schema();
        let gfield = schema.field(gcol_idx);
        let garr = from_arrow_array(batch.column(gcol_idx).as_ref(), gfield)
            .map_err(|e| crate::Error::GeoParquetRead(format!("geometry decode: {e}")))?;
        geoms_buf.clear();
        extract_geometries_opt_from_array(garr.as_ref(), &mut geoms_buf)?;

        // Attribute filter (#315): evaluate the predicate over the projected
        // batch once. A row whose result is not TRUE (FALSE or SQL-UNKNOWN)
        // produces no AssignFeature — exactly like a bbox miss below — while
        // the row index still advances, keeping row-keyed tables aligned.
        let filter_mask: Option<Vec<Option<bool>>> = filter.map(|f| f.eval_mask(&batch, &proj));

        // `AssignFeature::index` is the GLOBAL ROW index: pass 2 addresses the
        // winner tables by raw row position. Rows with a null, empty, or
        // non-finite geometry produce no feature but still advance the row
        // index, so every row-keyed table stays aligned (H4 hardening; a
        // skipped row must never shift attributes onto a neighbor's geometry).
        let base = num_rows;
        // #364: which rows of this batch became features. The ladder ranks
        // DISTINCT values, so it must see the same multiset the buffered engine
        // sees — that one reads the column off the table AFTER filtering, so a
        // value carried only by a rejected row must not create a rung here.
        let mut kept_row = vec![false; geoms_buf.len()];
        for (i, gopt) in geoms_buf.iter().enumerate() {
            // Attribute filter (#315): keep only rows where the predicate is
            // TRUE. The row index still advances (row-keyed tables stay
            // aligned); the slot stays UNASSIGNED so pass 2 drops it too.
            if let Some(mask) = &filter_mask {
                if mask[i] != Some(true) {
                    continue;
                }
            }
            let Some(g) = gopt.as_ref() else {
                skipped_rows += 1;
                continue;
            };
            // #274: a single geometry walk yields the usable filter, bbox, and
            // kind (was `usable_geometry` + `geometry_bbox` + `feature_kind`,
            // which traversed the coords twice). `None` == unusable (empty or
            // non-finite), identical to the old `usable_geometry` reject.
            let Some((kind, fbbox)) = scan_feature(g) else {
                skipped_rows += 1;
                continue;
            };
            // Regional extract (#102): a feature whose bbox misses the region
            // produces no AssignFeature — its winner-table slot stays at the
            // UNASSIGNED sentinel, so pass 2 drops the row too. The row index
            // still advances (row-keyed tables stay aligned).
            if let Some(bb) = bbox_units {
                if !super::convert::bboxes_intersect(&fbbox, bb) {
                    continue;
                }
            }
            if matches!(kind, FeatureKind::Point) {
                point_count += 1;
            }
            if collect_lines && matches!(kind, FeatureKind::Line) {
                line_rows.push(base + i);
                line_feat_pos.push(features.len());
                line_geoms.push(g.clone());
            }
            kept_row[i] = true;
            if want_areas {
                areas.push(polygon_area_f32(g));
            }
            features.push(AssignFeature {
                index: base + i,
                bbox: fbbox,
                kind,
                sort_key: None, // filled below once the ranking tier resolves
                entry_level: None,
            });
        }
        num_rows += geoms_buf.len();
        // #305: measure the encoded geometry column's in-memory size so the
        // pass-2 RAM-vs-spill estimate can use this input's actual average
        // geometry weight instead of a one-size-fits-all constant. O(#buffers)
        // per batch — no per-row work, no re-encode.
        geom_bytes += batch.column(gcol_idx).get_array_memory_size() as u64;

        match &mut plan {
            RankPlan::ExplicitSort { idx, .. } => {
                explicit_keys.extend(extract_sort_keys(batch.column(proj(*idx)).as_ref()));
            }
            RankPlan::ExplicitClass { idx, ranking } => {
                let col = batch.column(proj(*idx));
                explicit_keys.extend(extract_class_ranks(col.as_ref(), ranking)?);
                if collect_lines {
                    explicit_interner.extend(col.as_ref(), &mut explicit_groups);
                }
            }
            RankPlan::Auto { roads, confidence } => {
                for cand in roads.iter_mut() {
                    let col = batch.column(proj(cand.idx));
                    scan_road_vocab(col.as_ref(), &mut cand.found);
                    cand.keys
                        .extend(extract_class_ranks(col.as_ref(), &cand.ranking)?);
                    if collect_lines {
                        cand.interner.extend(col.as_ref(), &mut cand.groups);
                    }
                }
                if let Some((idx, _)) = confidence {
                    confidence_keys.extend(extract_sort_keys(batch.column(proj(*idx)).as_ref()));
                }
            }
            RankPlan::SizeFallback => {}
        }

        // Accumulate columns (Q4): per-spec source values, in row order.
        for (s, &idx) in acc_cols.iter().enumerate() {
            acc_values[s].extend(extract_sort_keys(batch.column(proj(idx)).as_ref()));
        }

        // Entry-zoom ladder (#364): row-indexed, like the ranking keys above,
        // but blanked for rows this pass rejected (null/unusable geometry, a
        // false `--filter` predicate, a `--bbox` miss). Those rows produce no
        // feature, so letting their values into the ladder would add rungs the
        // buffered engine never sees and shift every weaker feature by `step`.
        if let Some(idx) = ladder_col {
            let keys = extract_sort_keys(batch.column(proj(idx)).as_ref());
            ladder_values.extend(
                keys.into_iter()
                    .zip(&kept_row)
                    .map(|(k, keep)| if *keep { k } else { None }),
            );
        }
    }

    let (keys, provenance, all_groups) = resolve_ranking_tier(
        plan,
        explicit_keys,
        confidence_keys,
        explicit_groups,
        collect_lines,
        features.len(),
        point_count,
    );

    if let Some(keys) = keys {
        // Keys are extracted per ROW (including skipped-geometry rows), so
        // they are looked up by each feature's row index, not zipped
        // positionally.
        debug_assert_eq!(keys.len(), num_rows);
        for f in features.iter_mut() {
            f.sort_key = keys[f.index];
        }
    }

    apply_entry_levels(options, &ladder_values, num_rows, &mut features)?;

    // Coalescing scratch (Q3): line sort keys + per-line groups. `rows` and
    // `groups` are row-indexed; sort keys live on the features.
    let coalesce = collect_lines.then(|| CoalesceScratch {
        sort_keys: line_feat_pos
            .iter()
            .map(|&p| features[p].sort_key)
            .collect(),
        groups: all_groups.map(|g| line_rows.iter().map(|&r| g[r]).collect()),
        rows: line_rows,
        geoms: line_geoms,
    });

    Ok(Pass1Output {
        features,
        areas,
        provenance,
        acc_values,
        coalesce,
        num_rows,
        skipped_rows,
        geom_bytes,
    })
}

// ============================================================================
// Pass 2: per-level streaming filter → simplify → write
// ============================================================================

/// Wall-time accumulators for pass-2 stages ([profile] logging), stored as
/// nanoseconds. Atomic so the pipelined engine ([`super::pipeline`]) can share
/// one set across the parallel per-level processing of a batch; the serial
/// [`write_level_streaming`] path uses it single-threaded.
#[derive(Default)]
pub(super) struct Pass2Timers {
    /// Parquet read + Arrow decode of the raw batch (`reader.next()`).
    read: AtomicU64,
    /// Winner selection + geometry take/decode to `geo::Geometry`.
    decode: AtomicU64,
    /// Simplification (or verbatim vertex counting at the canonical level).
    simplify: AtomicU64,
    /// Output batch assembly (`build_level_batch`).
    build: AtomicU64,
}

impl Pass2Timers {
    fn add(cell: &AtomicU64, start: Instant) {
        cell.fetch_add(start.elapsed().as_nanos() as u64, Ordering::Relaxed);
    }
    /// Add a pre-measured duration (used by the reader thread for read time).
    pub(super) fn add_dur(cell: &AtomicU64, dur: Duration) {
        cell.fetch_add(dur.as_nanos() as u64, Ordering::Relaxed);
    }
    fn secs(cell: &AtomicU64) -> f64 {
        Duration::from_nanos(cell.load(Ordering::Relaxed)).as_secs_f64()
    }
    pub(super) fn read_cell(&self) -> &AtomicU64 {
        &self.read
    }
    /// Emit the aggregated per-stage breakdown ([profile] logging) for the
    /// pipelined engine, where stages interleave across levels so a per-level
    /// split is not meaningful.
    pub(super) fn log_engine_summary(&self, total_secs: f64, rows: usize) {
        let read_s = Self::secs(&self.read);
        let decode_s = Self::secs(&self.decode);
        let simplify_s = Self::secs(&self.simplify);
        let build_s = Self::secs(&self.build);
        log::debug!(
            "[profile] pass2 engine ({rows} rows): wall={total_secs:.2}s \
             read={read_s:.2}s decode={decode_s:.2}s simplify={simplify_s:.2}s \
             build={build_s:.2}s (stage sums are core-seconds, overlap wall)"
        );
    }
}

/// Immutable context for one level's pass-2 stream.
pub(super) struct LevelStreamCtx<'a> {
    source_schema: &'a Schema,
    /// `source_schema` + trailing `point_count` when clustering, otherwise
    /// identical (the schema [`apply_cluster_columns`] produces).
    cluster_schema: &'a Schema,
    /// Final writer schema: `cluster_schema` + trailing `coalesced_count`
    /// when coalescing, otherwise identical.
    out_schema: &'a Schema,
    non_geom_cols: &'a [usize],
    geom_idx: usize,
    /// Winner table: per input row, its coarsest level.
    min_levels: &'a [u8],
    /// Level index in the *resolved* plan (membership is tested against this,
    /// not the emitted/renumbered index).
    orig_level: u8,
    duplicating: bool,
    verbatim: bool,
    gsd_m: f64,
    /// Zoom-band representation (#317 / #279): how this level renders
    /// polygonal features (full geometry, representative points, or
    /// dithered placeholder squares for the below-tolerance ones).
    repr: Representation,
    crs: Crs,
    simplify: &'a SimplifyOptions,
    /// Clustering (Q4): append `point_count` + rewrite accumulate columns.
    cluster_enabled: bool,
    /// This level's cluster table; `None` at the canonical level (singletons)
    /// or when clustering is off.
    cluster_table: Option<&'a std::collections::HashMap<usize, ClusterEntry>>,
    /// Schema indices of the accumulate columns.
    acc_cols: &'a [usize],
    /// Coalescing (Q3): append `coalesced_count` at every level.
    coalesce_enabled: bool,
    /// Per-row geometry kinds (line rows bypass the winner table at
    /// coalesced levels); `Some` iff coalescing is enabled.
    kinds: Option<&'a [FeatureKind]>,
    /// This level's chain table (rep row → merged simplified geometry +
    /// member count); `None` at verbatim levels or when coalescing is
    /// off/guard-skipped.
    coalesce_table: Option<&'a CoalesceTable>,
    /// Cascading simplification (#218): fine→coarse step chain ending at
    /// this level (`[step_finest-1, …, step_this]`), fed to
    /// [`simplify_cascade`]. Empty when cascading does not apply (cascade
    /// off, partitioning, or verbatim level) — the level then simplifies
    /// canonical geometry directly with `gsd_m` / `point_repr`.
    cascade_chain: &'a [CascadeStep],
    /// Tiny-polygon accumulator carriers at this level (#384): sorted row
    /// indices of polygons that are NOT members (`min_level > orig_level`)
    /// but are emitted as a placeholder square standing in for the dropped
    /// area around them. Empty unless the accumulator applies.
    carriers: &'a [usize],
}

impl LevelStreamCtx<'_> {
    /// Is row `g` emitted at this level: a winner-table member, or a
    /// tiny-polygon carrier (#384)?
    #[inline]
    fn is_member(&self, g: usize) -> bool {
        let ml = self.min_levels[g];
        if self.duplicating {
            ml <= self.orig_level || is_carrier(self.carriers, g)
        } else {
            ml == self.orig_level
        }
    }

    /// Is row `g` a carrier here (emitted as a square, not as itself)?
    #[inline]
    fn is_carrier_row(&self, g: usize) -> bool {
        self.duplicating && self.min_levels[g] > self.orig_level && is_carrier(self.carriers, g)
    }
}

impl LevelStreamCtx<'_> {
    /// Whether the pipelined engine should process batches through the
    /// cascade fan-out ([`process_batch_cascade`], #218): duplicating mode
    /// with cascading enabled. Uniform across a conversion's level set.
    pub(super) fn is_cascading_duplicating(&self) -> bool {
        self.duplicating && self.simplify.cascade
    }
}

/// Stream one level from the input file into the writer. Returns the writer
/// outcome (a level whose every candidate collapses during simplification is
/// skipped, #211) plus `(rows_written, vertex_count)`.
#[allow(clippy::too_many_arguments)]
fn write_level_streaming(
    writer: &mut OverviewWriter<File>,
    level_idx: usize,
    hint: usize,
    source: &ConvertSource,
    read_batch_size: usize,
    in_flight: usize,
    row_groups: Option<&RowGroupSelection>,
    ctx: &LevelStreamCtx<'_>,
) -> Result<(LevelWriteOutcome, usize, usize), ConvertError> {
    let rows = Cell::new(0usize);
    let vertices = Cell::new(0usize);
    // Writer-thread time spent blocked waiting on the producer; the writer's
    // own busy time is `total - recv_wait` ([profile] logging).
    let recv_wait_ns = Cell::new(0u64);
    let timers = Pass2Timers::default();
    let fallbacks_before = full_resolution_fallback_count();
    let t_level = Instant::now();

    // One processed output batch handed from the producer to the writer.
    struct Processed {
        batch: RecordBatch,
        verts: usize,
    }

    // Overlap decode→process with the single-threaded parquet writer (#264,
    // extending the #213 pipeline discipline to the streamed finest level): a
    // producer thread reads input batches and runs `process_level_batch`
    // (read + geometry decode + simplify + assemble), pushing finished output
    // batches over a bounded channel; the writer drains it on this thread.
    // Batches stay in read order (FIFO channel, single producer), so output —
    // and therefore row-group boundaries — are byte-identical to a serial
    // build. Channel depth bounds read/compute run-ahead the same way the
    // buffered engine's reader channel does.
    // Shared by reference into the producer thread (a `&Pass2Timers` is `Copy`,
    // so the producer closure copies the borrow and leaves `timers` owned here
    // for the post-scope read).
    let timers = &timers;
    let outcome = scoped_pipe(
        in_flight,
        // Producer: read + process, in order, until EOF or the writer
        // hangs up. Returns the first stream/processing error, if any.
        // Everything the closure touches (`source`, `ctx`, `&timers`,
        // `row_groups`, the `usize`s) is `Copy`, so the outer bindings —
        // notably `timers`, read back afterwards — stay valid.
        |tx: &Sender<Processed>| -> Result<(), ConvertError> {
            // Regional extract (#102): read the same per-part bbox-selected
            // row groups as pass 1, so the winner tables' global row indices
            // line up.
            let mut reader = source.open_stream(&ReadPlan {
                batch_size: read_batch_size.max(1),
                projection: None,
                row_groups,
            })?;
            let mut row_offset = 0usize;
            // Heartbeat (#242): the finest level re-streams the whole
            // input; keep the operator informed on planet-scale files
            // (quiet on small ones).
            let mut last_progress = Instant::now();
            loop {
                if last_progress.elapsed().as_secs() >= 10 {
                    last_progress = Instant::now();
                    log::info!(
                        "[convert] level {level_idx}: {row_offset} input \
                             row(s) scanned",
                    );
                }
                let t_read = Instant::now();
                let batch = match reader.next() {
                    None => return Ok(()),
                    Some(Err(e)) => return Err(e.into()),
                    Some(Ok(b)) => b,
                };
                Pass2Timers::add(&timers.read, t_read);
                let offset = row_offset;
                row_offset += batch.num_rows();
                match process_level_batch(&batch, offset, ctx, timers)? {
                    None => continue, // no members of this level in the batch
                    Some((out, verts)) => {
                        // Writer gone (it errored and dropped the receiver):
                        // stop; the writer's error is reported by the caller.
                        if tx.send(Processed { batch: out, verts }).is_err() {
                            return Ok(());
                        }
                    }
                }
            }
        },
        // Writer (this thread): drain processed batches in order. Dropping
        // the producer's sender (EOF, error, or writer-gone) fuses `recv`;
        // `scoped_pipe` owns the mirror-image guarantee that this receiver is
        // dropped before the producer is joined (#362).
        |rx: Receiver<Processed>| -> Result<LevelWriteOutcome, ConvertError> {
            let batches = std::iter::from_fn(|| {
                let t_wait = Instant::now();
                match rx.recv() {
                    Ok(msg) => {
                        recv_wait_ns.set(recv_wait_ns.get() + t_wait.elapsed().as_nanos() as u64);
                        rows.set(rows.get() + msg.batch.num_rows());
                        vertices.set(vertices.get() + msg.verts);
                        Some(msg.batch)
                    }
                    Err(_) => {
                        recv_wait_ns.set(recv_wait_ns.get() + t_wait.elapsed().as_nanos() as u64);
                        None
                    }
                }
            });
            Ok(writer.write_level(level_idx, Some(hint), batches)?)
        },
    )?;
    let total = t_level.elapsed().as_secs_f64();
    let read_s = Pass2Timers::secs(&timers.read);
    let decode_s = Pass2Timers::secs(&timers.decode);
    let simplify_s = Pass2Timers::secs(&timers.simplify);
    let build_s = Pass2Timers::secs(&timers.build);
    // Read/decode/simplify/build run on the producer thread and overlap the
    // writer (#264), so these stage sums are core-seconds that overlap the
    // `total` wall time — the writer's own cost is roughly
    // `total - max(producer stages)`, not `total - sum`.
    let writer_busy = total - Duration::from_nanos(recv_wait_ns.get()).as_secs_f64();
    log::debug!(
        "[profile] level {} ({}, {} rows): total={:.2}s read={:.2}s decode={:.2}s \
         simplify={:.2}s build={:.2}s writer_busy={:.2}s (read/decode/simplify/build \
         overlap the writer)",
        level_idx,
        if ctx.verbatim { "verbatim" } else { "simplify" },
        rows.get(),
        total,
        read_s,
        decode_s,
        simplify_s,
        build_s,
        writer_busy,
    );
    let fallbacks = full_resolution_fallback_count() - fallbacks_before;
    if fallbacks > 0 {
        log::debug!(
            "[profile] level {level_idx}: {fallbacks} feature(s) kept at full \
             resolution (invalid RDP candidate after all epsilon retries)"
        );
    }
    Ok((outcome, rows.get(), vertices.get()))
}

/// Process one input batch for one level: select the level's members from the
/// winner table, decode only their geometries, simplify (unless verbatim), and
/// assemble the output batch. Returns `None` when no member row survives.
pub(super) fn process_level_batch(
    batch: &RecordBatch,
    row_offset: usize,
    ctx: &LevelStreamCtx<'_>,
    timers: &Pass2Timers,
) -> Result<Option<(RecordBatch, usize)>, ConvertError> {
    let n = batch.num_rows();
    let t_decode = Instant::now();
    let selected: Vec<usize> = (0..n)
        .filter(|&i| {
            let g = row_offset + i;
            // Coalesced level: line rows bypass the winner table entirely —
            // only surviving chain reps are emitted (with merged geometry).
            if let Some(table) = ctx.coalesce_table {
                if ctx.kinds.expect("kinds present when coalescing")[g] == FeatureKind::Line {
                    return table.contains_key(&g);
                }
            }
            ctx.is_member(g)
        })
        .collect();
    if selected.is_empty() {
        return Ok(None);
    }

    // Decode only the selected rows' geometries (take → decode, not
    // decode-all → filter).
    let take_idx = UInt32Array::from(selected.iter().map(|&i| i as u32).collect::<Vec<_>>());
    let geom_taken = take(batch.column(ctx.geom_idx).as_ref(), &take_idx, None)?;
    let schema = batch.schema();
    let gfield = schema.field(ctx.geom_idx);
    let garr = from_arrow_array(geom_taken.as_ref(), gfield)
        .map_err(|e| crate::Error::GeoParquetRead(format!("geometry decode: {e}")))?;
    let mut geoms: Vec<Geometry<f64>> = Vec::with_capacity(selected.len());
    extract_geometries_from_array(garr.as_ref(), &mut geoms)?;
    Pass2Timers::add(&timers.decode, t_decode);

    let t_simplify = Instant::now();
    let mut kept_idx: Vec<usize> = Vec::with_capacity(selected.len());
    let mut verts = 0usize;

    let kept_geoms: Vec<Geometry<f64>> = if ctx.verbatim {
        for (g, &i) in geoms.iter().zip(&selected) {
            verts += count_vertices(g);
            kept_idx.push(i);
        }
        geoms
    } else {
        // Simplification is >95% of pass-2 wall time (H3(c) profile) and
        // embarrassingly parallel per feature. `par_iter().map().collect()`
        // preserves within-batch order, so the output stays byte-identical to
        // the serial path; the writer (our single caller) remains
        // single-threaded, and memory stays bounded by one read batch.
        // Chain reps substitute their merged, already-simplified geometry
        // (simplified once in `build_level_coalesce_table`, identically to
        // the in-memory path).
        //
        // Cascading (#218): a non-empty `cascade_chain` folds canonical
        // geometry fine→coarse down to this level. This per-level recompute
        // is O(levels) per feature — it exists for the Serial reference
        // engine; the pipelined engine shares fold prefixes across levels
        // via `process_batch_cascade` and computes identical results.
        let simplified: Vec<Simplified> = geoms
            .par_iter()
            .zip(&selected)
            .map(|(g, &i)| {
                if let Some((merged, _)) = ctx.coalesce_table.and_then(|t| t.get(&(row_offset + i)))
                {
                    Simplified::Keep(merged.clone())
                } else if ctx.is_carrier_row(row_offset + i) {
                    // #384: a carrier stands in for its neighbourhood's
                    // dropped area as one placeholder square.
                    carrier_square(g, ctx.gsd_m, ctx.crs, ctx.simplify)
                        .map_or(Simplified::Dropped, Simplified::Keep)
                } else if !ctx.cascade_chain.is_empty() {
                    simplify_cascade(g, ctx.cascade_chain, ctx.crs, ctx.simplify)
                } else {
                    simplify_step(g, ctx.gsd_m, ctx.crs, ctx.simplify, ctx.repr)
                }
            })
            .collect();
        let mut out = Vec::with_capacity(selected.len());
        for (s, &i) in simplified.into_iter().zip(&selected) {
            match s {
                Simplified::Keep(s) => {
                    verts += count_vertices(&s);
                    kept_idx.push(i);
                    out.push(s);
                }
                Simplified::Dropped => {}
            }
        }
        if out.is_empty() {
            Pass2Timers::add(&timers.simplify, t_simplify);
            return Ok(None);
        }
        out
    };
    Pass2Timers::add(&timers.simplify, t_simplify);

    let t_build = Instant::now();
    let out_batch = assemble_level_batch(batch, row_offset, ctx, &kept_idx, &kept_geoms)?;
    Pass2Timers::add(&timers.build, t_build);
    Ok(Some((out_batch, verts)))
}

/// Assemble one level's output batch from kept row indices + geometries:
/// project source columns, splice the geometry column, then append
/// cluster / coalesced-count columns. Shared by [`process_level_batch`] and
/// [`process_batch_cascade`].
fn assemble_level_batch(
    batch: &RecordBatch,
    row_offset: usize,
    ctx: &LevelStreamCtx<'_>,
    kept_idx: &[usize],
    kept_geoms: &[Geometry<f64>],
) -> Result<RecordBatch, ConvertError> {
    let mut out_batch = build_level_batch(
        ctx.source_schema,
        batch,
        ctx.non_geom_cols,
        ctx.geom_idx,
        kept_idx,
        kept_geoms,
    )?;
    if ctx.cluster_enabled || ctx.coalesce_enabled {
        // Cluster/coalesce-table keys are global row indices; kept_idx is
        // batch-local.
        let globals: Vec<usize> = kept_idx.iter().map(|&i| row_offset + i).collect();
        if ctx.cluster_enabled {
            out_batch = apply_cluster_columns(
                out_batch,
                ctx.cluster_schema,
                &globals,
                ctx.cluster_table,
                ctx.acc_cols,
            )?;
        }
        if ctx.coalesce_enabled {
            out_batch =
                apply_coalesced_count(out_batch, ctx.out_schema, &globals, ctx.coalesce_table)?;
        }
    }
    Ok(out_batch)
}

/// Pipelined-engine batch processor for cascading simplification (#218).
///
/// Instead of every level independently decoding canonical geometry and
/// simplifying it from full resolution ([`process_level_batch`] per level),
/// this decodes each batch's member geometries **once**, computes each
/// feature's fine→coarse simplification fold **once** (level *k* consumes
/// level *k+1*'s output — the shared prefix is what the per-level path
/// recomputes), then assembles every level's output batch.
///
/// Bit-identical to running [`process_level_batch`] per level with the same
/// ctxs (the Serial reference): the incremental fold steps through exactly
/// the per-level `cascade_chain` GSD sequence, and each level's rows are
/// gathered in the same ascending batch order the per-level selection uses.
///
/// `ctxs` must be the pipelined engine's buffered slice: all non-verbatim
/// duplicating levels, coarse→fine.
pub(super) fn process_batch_cascade(
    batch: &RecordBatch,
    row_offset: usize,
    ctxs: &[LevelStreamCtx<'_>],
    timers: &Pass2Timers,
) -> Result<Vec<Option<(RecordBatch, usize)>>, ConvertError> {
    let Some(finest) = ctxs.last() else {
        return Ok(Vec::new());
    };
    debug_assert!(ctxs.iter().all(|c| c.duplicating && !c.verbatim));
    // The incremental fold steps ctx-by-ctx; each level's cascade_chain must
    // be exactly the GSD suffix from the finest buffered level down to it,
    // or Serial and Pipelined would diverge.
    debug_assert!(ctxs
        .iter()
        .enumerate()
        .all(|(li, c)| c.cascade_chain.len() == ctxs.len() - li
            && c.cascade_chain.last()
                == Some(&CascadeStep {
                    gsd_meters: c.gsd_m,
                    repr: c.repr,
                })));
    // Coalesce-table presence is uniform across buffered levels (tables are
    // built for every non-verbatim level or none); the superset selection
    // below relies on it.
    debug_assert!(ctxs
        .iter()
        .all(|c| c.coalesce_table.is_some() == finest.coalesce_table.is_some()));

    let n = batch.num_rows();

    // --- Select the cascade superset: members of the finest buffered level.
    // Coalesced line rows never cascade — each level emits its own chain
    // reps with merged, per-level-simplified geometry instead.
    let t_decode = Instant::now();
    let mut pos_of_row: Vec<u32> = vec![u32::MAX; n];
    let mut selected: Vec<usize> = Vec::with_capacity(n);
    for (i, pos) in pos_of_row.iter_mut().enumerate() {
        let g = row_offset + i;
        if finest.coalesce_table.is_some()
            && finest.kinds.expect("kinds present when coalescing")[g] == FeatureKind::Line
        {
            continue;
        }
        if finest.min_levels[g] <= finest.orig_level
            || ctxs.iter().any(|c| is_carrier(c.carriers, g))
        {
            *pos = u32::try_from(selected.len()).expect("batch rows fit in u32");
            selected.push(i);
        }
    }

    // Decode only the selected rows' geometries, once for all levels.
    let mut geoms: Vec<Geometry<f64>> = Vec::with_capacity(selected.len());
    if !selected.is_empty() {
        let take_idx = UInt32Array::from(selected.iter().map(|&i| i as u32).collect::<Vec<_>>());
        let geom_taken = take(batch.column(finest.geom_idx).as_ref(), &take_idx, None)?;
        let schema = batch.schema();
        let gfield = schema.field(finest.geom_idx);
        let garr = from_arrow_array(geom_taken.as_ref(), gfield)
            .map_err(|e| crate::Error::GeoParquetRead(format!("geometry decode: {e}")))?;
        extract_geometries_from_array(garr.as_ref(), &mut geoms)?;
    }
    Pass2Timers::add(&timers.decode, t_decode);

    // --- Per-feature incremental fold, fine→coarse, parallel over features.
    // folds[pos][d] is the result at ctxs[len-1-d]; entries stop at the
    // feature's coarsest member level, or earlier once Dropped with only
    // Geometry ctxs remaining (drops are monotone along geometry steps, so
    // a missing depth reads as dropped). Point / Square ctxs (#317 / #279)
    // REVIVE from canonical geometry — see `simplify_cascade`, whose fold
    // this mirrors step-for-step so Serial and Pipelined stay identical.
    //
    // has_band_upto[li]: whether any ctx at index <= li (i.e. this level or
    // a coarser one) carries a non-Geometry representation — the condition
    // under which a dropped fold must keep walking instead of breaking.
    let has_band_upto: Vec<bool> = {
        let mut v = Vec::with_capacity(ctxs.len());
        let mut any = false;
        for c in ctxs.iter() {
            any = any || c.repr != Representation::Geometry;
            v.push(any);
        }
        v
    };
    let t_simplify = Instant::now();
    let folds: Vec<Vec<Simplified>> = geoms
        .par_iter()
        .zip(&selected)
        .map(|(g, &i)| {
            let ml = finest.min_levels[row_offset + i];
            let mut out: Vec<Simplified> = Vec::with_capacity(ctxs.len());
            let mut current: Option<Geometry<f64>> = None;
            let mut alive = true;
            for (li, ctx) in ctxs.iter().enumerate().rev() {
                if ml > ctx.orig_level {
                    break; // duplicating membership is a contiguous fine suffix
                }
                if !alive && !has_band_upto[li] {
                    break; // only geometry ctxs remain: dropped stays dropped
                }
                let step = if !alive && ctx.repr == Representation::Geometry {
                    Simplified::Dropped
                } else {
                    let input = if alive {
                        current.as_ref().unwrap_or(g)
                    } else {
                        g // revive from canonical (Point / Square band)
                    };
                    simplify_step(input, ctx.gsd_m, ctx.crs, ctx.simplify, ctx.repr)
                };
                match step {
                    Simplified::Keep(s) => {
                        out.push(Simplified::Keep(s.clone()));
                        current = Some(s);
                        alive = true;
                    }
                    Simplified::Dropped => {
                        out.push(Simplified::Dropped);
                        alive = false;
                    }
                }
            }
            out
        })
        .collect();
    Pass2Timers::add(&timers.simplify, t_simplify);

    // --- Assemble every level's batch, in the per-level selection's
    // ascending row order (chain reps interleaved by global row index).
    let t_build = Instant::now();
    let results: Vec<Result<Option<(RecordBatch, usize)>, ConvertError>> = ctxs
        .par_iter()
        .enumerate()
        .map(|(li, ctx)| {
            let depth = ctxs.len() - 1 - li;
            let mut kept_idx: Vec<usize> = Vec::new();
            let mut kept_geoms: Vec<Geometry<f64>> = Vec::new();
            let mut verts = 0usize;
            for (i, &pos) in pos_of_row.iter().enumerate() {
                let g = row_offset + i;
                if let Some(table) = ctx.coalesce_table {
                    if ctx.kinds.expect("kinds present when coalescing")[g] == FeatureKind::Line {
                        if let Some((merged, _)) = table.get(&g) {
                            verts += count_vertices(merged);
                            kept_idx.push(i);
                            kept_geoms.push(merged.clone());
                        }
                        continue;
                    }
                }
                if ctx.min_levels[g] <= ctx.orig_level {
                    debug_assert_ne!(pos, u32::MAX, "member row missing from cascade superset");
                    if let Some(Simplified::Keep(s)) = folds[pos as usize].get(depth) {
                        verts += count_vertices(s);
                        kept_idx.push(i);
                        kept_geoms.push(s.clone());
                    }
                } else if ctx.is_carrier_row(g) {
                    // #384: not a member, but the carrier of its cell's
                    // dropped area — one placeholder square.
                    debug_assert_ne!(pos, u32::MAX, "carrier row missing from cascade superset");
                    if let Some(sq) =
                        carrier_square(&geoms[pos as usize], ctx.gsd_m, ctx.crs, ctx.simplify)
                    {
                        verts += count_vertices(&sq);
                        kept_idx.push(i);
                        kept_geoms.push(sq);
                    }
                }
            }
            if kept_idx.is_empty() {
                return Ok(None);
            }
            let out_batch = assemble_level_batch(batch, row_offset, ctx, &kept_idx, &kept_geoms)?;
            Ok(Some((out_batch, verts)))
        })
        .collect();
    Pass2Timers::add(&timers.build, t_build);

    let mut per_level = Vec::with_capacity(results.len());
    for res in results {
        per_level.push(res?);
    }
    Ok(per_level)
}