use std::collections::HashMap;
use std::collections::HashSet;
use std::fs::File;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Instant;
use arrow_array::{Array, RecordBatch, UInt32Array};
use arrow_schema::{DataType, Field, Schema, SchemaRef};
use arrow_select::concat::concat_batches;
use arrow_select::take::take;
use geo::{Area, BoundingRect, Geometry};
use geoarrow::array::{from_arrow_array, GeometryBuilder};
use geoarrow::datatypes::GeometryType;
use geoarrow_array::GeoArrowArray;
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
use crate::input::InputSource;
use crate::input_set::ConvertSource;
use serde::Serialize;
use crate::batch_processor::extract_geometries_opt_from_array;
use super::accumulate::{is_carrier, level_accumulates, tiny_polygon_carriers, AccumulateLevel};
use super::assign::{
apply_density_budget, assign_levels_bounded, AssignConfig, AssignFeature, Assignment,
DensityBudgetConfig, FeatureKind, SUPERCELL_GSD_FACTOR,
};
use super::cluster::{
build_cluster_tables, verify_sum_invariant, AccumulateSpec, ClusterEntry, ClusterTables,
POINT_COUNT_COLUMN,
};
use super::coalesce::{
coalesce_level_lines, CoalesceInput, CoalesceParams, COALESCED_COUNT_COLUMN,
DEFAULT_COALESCE_MAX_LEVEL_ROWS, DEFAULT_JUNCTION_ANGLE_DEG, DEFAULT_SNAP_GSD_FACTOR,
};
use super::ladder::{build_ladder, entry_levels, EntryZoomSpec};
use super::level::{
gsd_with_base, AccumulatedColumn, ClusteringProvenance, CoalescingProvenance, Crs,
DensityProvenance, Generalization, GeneralizationLevel, MemoryProfile, Mode, RankingProvenance,
RepresentationBandProvenance, GSD_TILE_BASE, METERS_PER_DEGREE,
};
use super::properties::{PropertySelection, PropertySelectionError};
use super::simplify::{
carrier_square, simplify_cascade, simplify_for_level, simplify_step, CascadeStep, CollapseMode,
Representation, Simplified, SimplifyOptions,
};
use super::writer::{
LevelSpec, LevelWriteOutcome, OverviewWriter, RowGroupSizePolicy, WriterError, LEVEL_COLUMN,
};
#[derive(Debug, Clone, PartialEq)]
pub enum LevelPlan {
ZoomRange {
min_zoom: u8,
max_zoom: u8,
},
Gsds(Vec<f64>),
}
pub(super) const MAX_LEVELS: usize = 255;
impl LevelPlan {
pub(super) fn resolve(&self, gsd_base: f64) -> Result<Vec<(f64, Option<u8>)>, ConvertError> {
let check_len = |n: usize| {
if n > MAX_LEVELS {
return Err(ConvertError::InvalidLevels(format!(
"{n} levels requested; at most {MAX_LEVELS} levels are supported"
)));
}
Ok(())
};
match self {
LevelPlan::ZoomRange { min_zoom, max_zoom } => {
if min_zoom > max_zoom {
return Err(ConvertError::InvalidLevels(format!(
"min_zoom {min_zoom} must be <= max_zoom {max_zoom}"
)));
}
check_len(*max_zoom as usize - *min_zoom as usize + 1)?;
Ok((*min_zoom..=*max_zoom)
.map(|z| (gsd_with_base(z, gsd_base), Some(z)))
.collect())
}
LevelPlan::Gsds(gsds) => {
if gsds.is_empty() {
return Err(ConvertError::InvalidLevels(
"explicit gsd list must be non-empty".to_string(),
));
}
check_len(gsds.len())?;
let mut prev: Option<f64> = None;
for (i, &g) in gsds.iter().enumerate() {
if g <= 0.0 || g.is_nan() {
return Err(ConvertError::InvalidLevels(format!(
"gsd[{i}] = {g} must be > 0"
)));
}
if let Some(p) = prev {
if g >= p {
return Err(ConvertError::InvalidLevels(format!(
"gsd list must be strictly decreasing coarse→fine (gsd[{i}] = {g} >= previous {p})"
)));
}
}
prev = Some(g);
}
Ok(gsds.iter().map(|&g| (g, None)).collect())
}
}
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct ClassRanking {
pub column: String,
pub ranks: Vec<(String, f64)>,
pub unknown_rank: f64,
}
const MAX_PROVENANCE_RANKS: usize = 64;
pub fn overture_road_ranking(column: String) -> ClassRanking {
let ordered = [
"motorway", "trunk",
"primary",
"secondary",
"tertiary",
"residential",
"unclassified",
"service",
"living_street", "pedestrian",
"track",
"cycleway",
"bridleway",
"footway",
"steps",
"path",
"driveway",
"parking_aisle",
];
let n = ordered.len();
let ranks = ordered
.iter()
.enumerate()
.map(|(i, &c)| (c.to_string(), (n - i) as f64))
.collect();
ClassRanking {
column,
ranks,
unknown_rank: 0.0,
}
}
pub(super) const KNOWN_ROAD_CLASSES: &[&str] = &[
"motorway",
"trunk",
"primary",
"secondary",
"tertiary",
"residential",
"unclassified",
"service",
"living_street",
"pedestrian",
"track",
"cycleway",
"bridleway",
"footway",
"steps",
"path",
"driveway",
"parking_aisle",
"unknown",
"standard_gauge",
"light_rail",
"tram",
"subway",
"monorail",
"funicular",
];
pub(super) const ROAD_VOCAB_MIN_DISTINCT: usize = 3;
#[derive(Debug, Clone)]
pub struct ConvertOptions {
pub mode: Mode,
pub levels: LevelPlan,
pub assign: AssignConfig,
pub entry_zoom: Option<EntryZoomSpec>,
pub sort_key: Option<String>,
pub class_ranking: Option<ClassRanking>,
pub no_auto_rank: bool,
pub simplify: SimplifyOptions,
pub representation: Vec<RepresentationBand>,
pub density: DensityBudgetConfig,
pub gsd_base: f64,
pub cogp_compat_key: bool,
pub max_row_group_size: usize,
pub row_group_size_policy: RowGroupSizePolicy,
pub full_column_stats: bool,
pub streaming: bool,
pub read_batch_size: usize,
pub profile: MemoryProfile,
pub in_flight_batches: usize,
pub cluster: bool,
pub accumulate: Vec<AccumulateSpec>,
pub coalesce_lines: bool,
pub coalesce_snap: f64,
pub coalesce_max_level_rows: usize,
pub coalesce_junction_angle: f64,
pub bbox: Option<[f64; 4]>,
pub filter: Option<String>,
pub properties: PropertySelection,
pub spill_dir: Option<PathBuf>,
}
pub const DEFAULT_READ_BATCH_SIZE: usize = 8192;
pub const IN_FLIGHT_BATCHES_AUTO: usize = 0;
pub const IN_FLIGHT_BATCHES_MIN: usize = 4;
pub const IN_FLIGHT_BATCHES_MAX: usize = 16;
pub fn resolve_in_flight_batches(requested: usize) -> usize {
if requested == IN_FLIGHT_BATCHES_AUTO {
std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(IN_FLIGHT_BATCHES_MIN)
.clamp(IN_FLIGHT_BATCHES_MIN, IN_FLIGHT_BATCHES_MAX)
} else {
requested
}
}
impl ConvertOptions {
#[must_use]
pub fn verbatim(mut self) -> Self {
self.assign.point_thinning = 0.0;
self.assign.line_thinning = 0.0;
self.assign.polygon_thinning = 0.0;
self.assign.line_visibility = 0.0;
self.assign.polygon_visibility = 0.0;
self.simplify.factor = 0.0;
self.density.enabled = false;
self.coalesce_lines = false;
self
}
pub fn is_verbatim(&self) -> bool {
self.generalization_is_off() && self.entry_zoom.is_none()
}
pub(crate) fn generalization_is_off(&self) -> bool {
self.assign.point_thinning == 0.0
&& self.assign.line_thinning == 0.0
&& self.assign.polygon_thinning == 0.0
&& self.assign.line_visibility == 0.0
&& self.assign.polygon_visibility == 0.0
&& self.simplify.factor == 0.0
&& !self.density.enabled
&& !self.coalesce_lines
}
}
impl Default for ConvertOptions {
fn default() -> Self {
Self {
mode: Mode::Duplicating,
levels: LevelPlan::ZoomRange {
min_zoom: 0,
max_zoom: 6,
},
assign: AssignConfig::default(),
entry_zoom: None,
sort_key: None,
class_ranking: None,
no_auto_rank: false,
simplify: SimplifyOptions::default(),
representation: Vec::new(),
density: DensityBudgetConfig::default(),
gsd_base: GSD_TILE_BASE,
cogp_compat_key: false,
max_row_group_size: super::writer::DEFAULT_MAX_ROW_GROUP_SIZE,
row_group_size_policy: RowGroupSizePolicy::default(),
full_column_stats: false,
streaming: true,
read_batch_size: DEFAULT_READ_BATCH_SIZE,
profile: MemoryProfile::Auto,
in_flight_batches: IN_FLIGHT_BATCHES_AUTO,
cluster: false,
accumulate: Vec::new(),
coalesce_lines: true,
coalesce_snap: DEFAULT_SNAP_GSD_FACTOR,
coalesce_max_level_rows: DEFAULT_COALESCE_MAX_LEVEL_ROWS,
coalesce_junction_angle: DEFAULT_JUNCTION_ANGLE_DEG,
bbox: None,
filter: None,
properties: PropertySelection::default(),
spill_dir: None,
}
}
}
#[derive(Debug, Clone, PartialEq, Serialize)]
pub struct LevelReport {
pub level: usize,
pub gsd: f64,
pub zoom: Option<u8>,
pub feature_count: usize,
pub vertex_count: usize,
pub uncompressed_bytes: i64,
pub compressed_bytes: i64,
}
#[derive(Debug, Clone, PartialEq, Serialize)]
pub struct SkippedLevelReport {
pub planned_level: usize,
pub gsd: f64,
pub zoom: Option<u8>,
}
pub(super) fn warn_plan_skipped_levels(
skipped: &[SkippedLevelReport],
input_features: usize,
first_written_gsd: f64,
first_written_zoom: Option<u8>,
) {
if skipped.is_empty() {
return;
}
let ids: Vec<String> = skipped
.iter()
.map(|s| s.planned_level.to_string())
.collect();
let gsd_max = skipped.iter().map(|s| s.gsd).fold(f64::MIN, f64::max);
let gsd_min = skipped.iter().map(|s| s.gsd).fold(f64::MAX, f64::min);
let zoom_note = first_written_zoom.map_or_else(String::new, |z| format!(" (zoom {z})"));
log::warn!(
"omitting {} empty level(s) [{}] spanning GSD {:.2}–{:.2} m: none of the {} input \
feature(s) are visible at those scales (visibility gates / density budget); the \
output pyramid starts at GSD {:.2} m{}. To populate coarse levels, lower \
--polygon-visibility/--line-visibility, or pass --collapse to keep sub-GSD \
polygons as representative points (see docs/OVERVIEW_TUNING.md)",
skipped.len(),
ids.join(", "),
gsd_max,
gsd_min,
input_features,
first_written_gsd,
zoom_note,
);
}
pub(super) fn record_level_outcome(
outcome: LevelWriteOutcome,
planned: SkippedLevelReport,
candidates: usize,
rows: usize,
vertices: usize,
level_reports: &mut Vec<LevelReport>,
skipped: &mut Vec<SkippedLevelReport>,
) {
match outcome {
LevelWriteOutcome::SkippedEmpty => {
log::warn!(
"level planned at GSD {:.2} m{} became empty after simplification \
(all {} candidate feature(s) collapsed); omitted from the output pyramid. \
Pass --collapse to keep sub-GSD polygons as representative points at \
coarse levels (see docs/OVERVIEW_TUNING.md)",
planned.gsd,
planned
.zoom
.map_or_else(String::new, |z| format!(" (zoom {z})")),
candidates,
);
skipped.push(planned);
}
LevelWriteOutcome::Written => level_reports.push(LevelReport {
level: level_reports.len(),
gsd: planned.gsd,
zoom: planned.zoom,
feature_count: rows,
vertex_count: vertices,
uncompressed_bytes: 0,
compressed_bytes: 0,
}),
}
}
#[derive(Debug, Clone, PartialEq, Serialize)]
pub struct ConvertReport {
pub mode: Mode,
pub levels: Vec<LevelReport>,
pub skipped_empty_levels: Vec<SkippedLevelReport>,
pub input_features: usize,
pub total_rows: usize,
pub total_vertices: usize,
pub total_compressed_bytes: i64,
pub row_groups_total: usize,
pub row_groups_read: usize,
pub antimeridian_suspect_features: usize,
pub duration_secs: f64,
pub remote_fetch: Option<crate::input::FetchStats>,
}
#[derive(Debug, thiserror::Error)]
pub enum ConvertError {
#[error("io error: {0}")]
Io(#[from] std::io::Error),
#[error("input error: {0}")]
Input(#[from] crate::input::InputError),
#[error("property selection: {0}")]
Properties(#[from] PropertySelectionError),
#[error("parquet error: {0}")]
Parquet(#[from] parquet::errors::ParquetError),
#[error("arrow error: {0}")]
Arrow(#[from] arrow_schema::ArrowError),
#[error("{0}")]
Core(#[from] crate::Error),
#[error("writer error: {0}")]
Writer(#[from] WriterError),
#[error("unsupported input CRS {crs:?}: overviews require EPSG:4326 or EPSG:3857")]
UnsupportedCrs {
crs: String,
},
#[error("input has no geometry column")]
NoGeometryColumn,
#[error("sort-key column {name:?} not found in input schema")]
SortKeyColumnMissing {
name: String,
},
#[error("--sort-key and --class-rank are mutually exclusive; supply at most one")]
RankingConflict,
#[error("class-rank column {name:?} not found in input schema")]
ClassRankColumnMissing {
name: String,
},
#[error("class-rank column {name:?} is {data_type} but must be a string column")]
ClassRankColumnNotString {
name: String,
data_type: String,
},
#[error("invalid level specification: {0}")]
InvalidLevels(String),
#[error("invalid option: {0}")]
InvalidConfig(String),
#[error(transparent)]
Filter(#[from] super::filter::FilterError),
#[error(
"--cluster requires duplicating mode: a partitioning-mode feature has one \
row read across many zoom prefixes, so a per-level point_count cannot be \
represented without double counting"
)]
ClusterPartitioningUnsupported,
#[error(
"--verbatim requires duplicating mode: partitioning places each feature at \
exactly one level, so with thinning off every feature lands in the coarsest \
level and every finer level is empty"
)]
VerbatimPartitioningUnsupported,
#[error("--accumulate-attribute requires --cluster")]
AccumulateWithoutCluster,
#[error(
"multi-partition input requires the streaming pipeline; \
remove --no-streaming"
)]
MultiPartitionRequiresStreaming,
#[error("accumulate-attribute column {name:?} not found in input schema")]
AccumulateColumnMissing {
name: String,
},
#[error(
"accumulate-attribute column {name:?} is {data_type} but must be numeric \
(int/uint/float)"
)]
AccumulateColumnNotNumeric {
name: String,
data_type: String,
},
#[error(
"input already contains a '{POINT_COUNT_COLUMN}' column; rename it before \
converting with --cluster"
)]
PointCountColumnPresent,
#[error(
"input already contains a '{COALESCED_COUNT_COLUMN}' column; rename it \
before converting with --coalesce-lines"
)]
CoalescedCountColumnPresent,
#[error("no output rows produced (empty input or all features dropped)")]
NoData,
#[error("cluster invariant violated (spec §12.1): {0}")]
ClusterInvariant(String),
}
fn validate_options(options: &ConvertOptions) -> Result<(), ConvertError> {
let positive = |name: &str, v: f64| {
if !v.is_finite() || v <= 0.0 {
return Err(ConvertError::InvalidConfig(format!(
"{name} = {v} must be a finite value > 0"
)));
}
Ok(())
};
let non_negative = |name: &str, v: f64| {
if !v.is_finite() || v < 0.0 {
return Err(ConvertError::InvalidConfig(format!(
"{name} = {v} must be a finite value >= 0"
)));
}
Ok(())
};
positive("gsd-base", options.gsd_base)?;
non_negative("point-thinning", options.assign.point_thinning)?;
non_negative("line-thinning", options.assign.line_thinning)?;
non_negative("polygon-thinning", options.assign.polygon_thinning)?;
non_negative("line-visibility", options.assign.line_visibility)?;
non_negative("polygon-visibility", options.assign.polygon_visibility)?;
if options.coalesce_snap.is_nan() {
return Err(ConvertError::InvalidConfig(
"coalesce-snap must not be NaN (use <= 0 to disable snapping)".to_string(),
));
}
if options.coalesce_junction_angle.is_nan() {
return Err(ConvertError::InvalidConfig(
"coalesce-junction-angle must not be NaN (use 0 to disable)".to_string(),
));
}
if let Some(bb) = &options.bbox {
if bb.iter().any(|v| !v.is_finite()) {
return Err(ConvertError::InvalidConfig(format!(
"bbox {bb:?} must contain only finite values"
)));
}
if bb[0] > bb[2] || bb[1] > bb[3] {
return Err(ConvertError::InvalidConfig(format!(
"bbox {bb:?} must satisfy xmin <= xmax and ymin <= ymax"
)));
}
}
if let Some(f) = &options.filter {
super::filter::parse_filter(f)?;
}
if !options.representation.is_empty() {
if matches!(options.mode, Mode::Partitioning)
&& options
.representation
.iter()
.any(|b| b.repr != Representation::Geometry)
{
return Err(ConvertError::InvalidConfig(
"representation bands require duplicating mode: partitioning places \
each feature exactly once with geometry verbatim (spec §2.3), which \
a point or square representation cannot satisfy"
.to_string(),
));
}
let (plan_min, plan_max) = match &options.levels {
LevelPlan::ZoomRange { min_zoom, max_zoom } => (*min_zoom, *max_zoom),
LevelPlan::Gsds(_) => {
return Err(ConvertError::InvalidConfig(
"representation bands require a zoom-range level plan \
(--min-zoom/--max-zoom); an explicit --gsd plan carries no \
per-level zooms to band on"
.to_string(),
));
}
};
for band in &options.representation {
let (lo, hi, repr) = (band.min_zoom, band.max_zoom, band.repr);
let kw = repr.as_str();
if lo > hi {
return Err(ConvertError::InvalidConfig(format!(
"representation band {lo}-{hi}:{kw} must satisfy LO <= HI"
)));
}
if lo < plan_min || hi > plan_max {
return Err(ConvertError::InvalidConfig(format!(
"representation band {lo}-{hi}:{kw} lies outside the level plan \
({plan_min}-{plan_max})"
)));
}
if repr != Representation::Geometry && hi >= plan_max {
return Err(ConvertError::InvalidConfig(format!(
"representation band {lo}-{hi}:{kw} must end before the plan's \
max zoom ({plan_max}): the canonical (finest) level reproduces \
source geometry verbatim (spec §2.4)"
)));
}
}
let mut claimed: Vec<Option<Representation>> =
vec![None; plan_max as usize - plan_min as usize + 1];
for band in &options.representation {
for z in band.min_zoom..=band.max_zoom {
let slot = &mut claimed[(z - plan_min) as usize];
if slot.is_some() {
return Err(ConvertError::InvalidConfig(format!(
"representation bands overlap at zoom {z}"
)));
}
*slot = Some(band.repr);
}
}
let mut seen_non_point_coarser = false;
for slot in &claimed {
match slot {
Some(Representation::Point) => {
if seen_non_point_coarser {
return Err(ConvertError::InvalidConfig(
"point bands must start at the plan's min zoom and be \
contiguous from the coarsest level: a coarser non-point \
level would still receive the cascaded point"
.to_string(),
));
}
}
_ => seen_non_point_coarser = true,
}
}
}
if let Some(dir) = &options.spill_dir {
if !dir.is_dir() {
return Err(ConvertError::InvalidConfig(format!(
"spill-dir {} is not an existing directory",
dir.display()
)));
}
}
Ok(())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RepresentationBand {
pub min_zoom: u8,
pub max_zoom: u8,
pub repr: Representation,
}
pub(super) fn representation_for_zoom(
bands: &[RepresentationBand],
zoom: Option<u8>,
) -> Representation {
let Some(z) = zoom else {
return Representation::Geometry;
};
bands
.iter()
.find(|b| b.min_zoom <= z && z <= b.max_zoom)
.map(|b| b.repr)
.unwrap_or_default()
}
pub fn parse_representation_spec(spec: &str) -> Result<Vec<RepresentationBand>, String> {
let mut bands = Vec::new();
for part in spec.split(',') {
let part = part.trim();
if part.is_empty() {
continue;
}
let (range, kind) = part.split_once(':').ok_or_else(|| {
format!("representation entry {part:?} must be LO-HI:KIND (e.g. 0-7:point)")
})?;
let repr = match kind.trim() {
"geom" | "geometry" => Representation::Geometry,
"point" => Representation::Point,
"square" => Representation::Square,
other => {
return Err(format!(
"unknown representation {other:?} (expected geom, point, or square)"
))
}
};
let range = range.trim();
let (lo, hi) = match range.split_once('-') {
Some((lo, hi)) => (lo.trim(), hi.trim()),
None => (range, range),
};
let parse_zoom = |s: &str| {
s.parse::<u8>()
.map_err(|_| format!("invalid zoom {s:?} in representation entry {part:?}"))
};
bands.push(RepresentationBand {
min_zoom: parse_zoom(lo)?,
max_zoom: parse_zoom(hi)?,
repr,
});
}
if bands.is_empty() {
return Err("representation spec is empty".to_string());
}
Ok(bands)
}
pub(super) fn level_representations(
level_specs: &[(f64, Option<u8>)],
bands: &[RepresentationBand],
) -> Vec<Representation> {
level_specs
.iter()
.map(|&(_, zoom)| representation_for_zoom(bands, zoom))
.collect()
}
pub fn convert_to_overviews(
input_path: impl AsRef<Path>,
output_path: impl AsRef<Path>,
options: &ConvertOptions,
) -> Result<ConvertReport, ConvertError> {
let source = ConvertSource::resolve_path(input_path.as_ref())?;
convert_to_overviews_source_strategy(
&source,
output_path.as_ref(),
options,
super::stream::Pass2Strategy::Pipelined,
)
}
pub fn convert_to_overviews_sources(
source: &ConvertSource,
output_path: &Path,
options: &ConvertOptions,
) -> Result<ConvertReport, ConvertError> {
convert_to_overviews_source_strategy(
source,
output_path,
options,
super::stream::Pass2Strategy::Pipelined,
)
}
pub fn convert_to_overviews_source(
source: &InputSource,
output_path: &Path,
options: &ConvertOptions,
) -> Result<ConvertReport, ConvertError> {
convert_to_overviews_source_strategy(
&ConvertSource::single(source.clone()),
output_path,
options,
super::stream::Pass2Strategy::Pipelined,
)
}
#[cfg(test)]
pub(crate) fn convert_to_overviews_strategy(
input_path: impl AsRef<Path>,
output_path: impl AsRef<Path>,
options: &ConvertOptions,
strategy: super::stream::Pass2Strategy,
) -> Result<ConvertReport, ConvertError> {
let source = ConvertSource::resolve_path(input_path.as_ref())?;
convert_to_overviews_source_strategy(&source, output_path.as_ref(), options, strategy)
}
fn decode_and_filter_geometries(
full: RecordBatch,
geom_idx: usize,
geom_field: &Field,
bbox_units: Option<&[f64; 4]>,
filter_mask: Option<&[Option<bool>]>,
) -> Result<(RecordBatch, Vec<Geometry<f64>>), ConvertError> {
let geom_array: Arc<dyn GeoArrowArray> =
from_arrow_array(full.column(geom_idx).as_ref(), geom_field)
.map_err(|e| crate::Error::GeoParquetRead(format!("geometry decode: {e}")))?;
let mut geom_opts: Vec<Option<Geometry<f64>>> = Vec::with_capacity(full.num_rows());
extract_geometries_opt_from_array(geom_array.as_ref(), &mut geom_opts)?;
let mut geom_skipped = 0usize;
let keep: Vec<bool> = geom_opts
.iter()
.enumerate()
.map(|(i, g)| {
if let Some(mask) = filter_mask {
if mask[i] != Some(true) {
return false;
}
}
match g.as_ref().filter(|g| usable_geometry(g)) {
Some(g) => bbox_units.is_none_or(|bb| bboxes_intersect(&geometry_bbox(g), bb)),
None => {
geom_skipped += 1;
false
}
}
})
.collect();
let dropped = keep.iter().filter(|k| !**k).count();
if dropped == 0 {
return Ok((full, geom_opts.into_iter().flatten().collect()));
}
if geom_skipped > 0 {
log::warn!(
"skipping {geom_skipped} of {} input rows with a null, empty, or \
non-finite geometry",
full.num_rows()
);
}
let mask = arrow_array::BooleanArray::from(keep.clone());
let filtered = arrow_select::filter::filter_record_batch(&full, &mask)?;
let geoms = geom_opts
.into_iter()
.zip(&keep)
.filter(|(_, k)| **k)
.map(|(g, _)| g.expect("kept rows are Some"))
.collect();
Ok((filtered, geoms))
}
pub(crate) fn adjusted_for_ladder_and_mode(options: &ConvertOptions) -> Option<ConvertOptions> {
let coalescing_off = options.coalesce_lines
&& match options.mode {
Mode::Partitioning => Some(
"line coalescing is inert in partitioning mode (feature-once / \
geometry-verbatim contract); converting without it",
),
_ if options.entry_zoom.is_some() => Some(
"line coalescing is inert with an entry-zoom ladder (a merged chain \
has no entry zoom to inherit, and coalescing would re-gate the \
lines the ladder promoted); converting without it",
),
_ => None,
}
.inspect(|why| log::info!("{why}"))
.is_some();
let collapse_to_point =
options.entry_zoom.is_some() && matches!(options.simplify.collapse, CollapseMode::Drop);
if collapse_to_point {
log::info!(
"an entry-zoom ladder implies collapse-to-point: a promoted feature is \
usually below its level's simplification tolerance and would be deleted \
there instead of drawn. Coarse levels therefore carry representative \
POINTS for those features — style them with a circle layer, or ask for \
--collapse-square to keep polygons."
);
}
if matches!(options.mode, Mode::Partitioning)
&& matches!(options.simplify.collapse, CollapseMode::Square)
{
log::info!(
"collapse-square has no effect in partitioning mode (levels are verbatim: \
no polygon is dropped or collapsed, so there is nothing to stand in for)"
);
}
if !coalescing_off && !collapse_to_point {
return None;
}
Some(ConvertOptions {
coalesce_lines: options.coalesce_lines && !coalescing_off,
simplify: SimplifyOptions {
collapse: if collapse_to_point {
CollapseMode::Point
} else {
options.simplify.collapse
},
..options.simplify
},
..options.clone()
})
}
#[cfg(test)]
mod ladder_adjustment_tests {
use super::*;
fn with_ladder() -> ConvertOptions {
ConvertOptions {
entry_zoom: Some(crate::overview::ladder::EntryZoomSpec {
column: "level".to_string(),
kind: crate::overview::ladder::EntryZoomKind::DenseRank { step: 1 },
}),
..Default::default()
}
}
#[test]
fn ladder_implies_collapse_to_point() {
let adjusted = adjusted_for_ladder_and_mode(&with_ladder())
.expect("a ladder must trigger an adjustment");
assert_eq!(adjusted.simplify.collapse, CollapseMode::Point);
}
#[test]
fn an_explicit_collapse_square_survives_the_implication() {
let mut o = with_ladder();
o.simplify.collapse = CollapseMode::Square;
let adjusted = adjusted_for_ladder_and_mode(&o);
assert!(adjusted.is_none_or(|a| a.simplify.collapse == CollapseMode::Square));
}
#[test]
fn ladder_turns_line_coalescing_off() {
let mut o = with_ladder();
o.coalesce_lines = true;
let adjusted = adjusted_for_ladder_and_mode(&o).expect("adjusted");
assert!(!adjusted.coalesce_lines);
}
#[test]
fn no_ladder_means_no_adjustment() {
assert!(adjusted_for_ladder_and_mode(&ConvertOptions::default()).is_none());
}
}
fn check_mode_combinations(options: &ConvertOptions) -> Result<(), ConvertError> {
if options.cluster && matches!(options.mode, Mode::Partitioning) {
return Err(ConvertError::ClusterPartitioningUnsupported);
}
if options.generalization_is_off() && matches!(options.mode, Mode::Partitioning) {
return Err(ConvertError::VerbatimPartitioningUnsupported);
}
if !options.accumulate.is_empty() && !options.cluster {
return Err(ConvertError::AccumulateWithoutCluster);
}
Ok(())
}
fn knob_columns(options: &ConvertOptions) -> Vec<(String, String)> {
let mut out = Vec::new();
if let Some(c) = &options.sort_key {
out.push((c.clone(), "--sort-key".to_string()));
}
if let Some(r) = &options.class_ranking {
out.push((r.column.clone(), "--class-rank".to_string()));
}
if let Some(e) = &options.entry_zoom {
out.push((
e.column.clone(),
"--magnitude-ladder / --entry-zoom".to_string(),
));
}
for a in &options.accumulate {
out.push((a.column.clone(), "--accumulate-attribute".to_string()));
}
if let Some(f) = &options.filter {
if let Ok(expr) = super::filter::parse_filter(f) {
for c in expr.column_names() {
out.push((c, "--filter".to_string()));
}
}
}
out
}
fn apply_property_selection(
source: &ConvertSource,
options: &ConvertOptions,
) -> Result<(), ConvertError> {
if source.column_projection().is_some() {
return Err(
crate::input::InputError::Arrow(arrow_schema::ArrowError::SchemaError(
"column restriction already applied to this source; a ConvertSource is \
single-use once a property selection has been applied"
.to_string(),
))
.into(),
);
}
if options.properties.is_identity() {
return Ok(());
}
let schema = source.file_schema()?;
let geom_idx = find_geometry_column(&schema).ok_or(ConvertError::NoGeometryColumn)?;
let mut warn = |msg: String| log::warn!("[convert] {msg}");
let keep = options
.properties
.resolve(&schema, geom_idx, &knob_columns(options), &mut warn)?;
let kept_names: Vec<&str> = keep
.iter()
.filter(|&&i| i != geom_idx)
.map(|&i| schema.field(i).name().as_str())
.collect();
let dropped = schema.fields().len() - keep.len();
log::info!(
"[convert] property selection: keeping {} of {} property column(s) ({}), dropping {dropped}",
kept_names.len(),
schema.fields().len() - 1,
if kept_names.is_empty() {
"none — geometry only".to_string()
} else {
kept_names
.iter()
.map(|n| format!("{n:?}"))
.collect::<Vec<_>>()
.join(", ")
}
);
source.restrict_columns(keep)?;
Ok(())
}
fn project_builder_to_selection(
source: &ConvertSource,
builder: parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder<
crate::input::InputReader,
>,
) -> Result<
(
parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder<crate::input::InputReader>,
SchemaRef,
),
ConvertError,
> {
match source.column_projection() {
Some(keep) => {
let mask = parquet::arrow::ProjectionMask::roots(
builder.parquet_schema(),
keep.iter().copied(),
);
let projected = Arc::new(builder.schema().project(keep)?);
Ok((builder.with_projection(mask), projected))
}
None => {
let schema = builder.schema().clone();
Ok((builder, schema))
}
}
}
fn intern_coalesce_groups(
input_schema: &Schema,
full: &RecordBatch,
ranking_provenance: &RankingProvenance,
) -> Option<Vec<u32>> {
let col = coalesce_group_column(ranking_provenance)?;
let idx = input_schema.index_of(col).expect("ranking column exists");
let mut interner = GroupInterner::default();
let mut groups = Vec::with_capacity(full.num_rows());
interner.extend(full.column(idx).as_ref(), &mut groups);
Some(groups)
}
pub(super) fn build_verified_cluster_tables(
features: &[AssignFeature],
min_levels: &[u8],
level_gsds: &[f64],
acc_values: &[Vec<Option<f64>>],
crs: Crs,
options: &ConvertOptions,
) -> Result<ClusterTables, ConvertError> {
let ops: Vec<_> = options.accumulate.iter().map(|s| s.op).collect();
let tables = build_cluster_tables(
features,
min_levels,
level_gsds,
&options.assign,
crs,
acc_values,
&ops,
);
verify_sum_invariant(features, min_levels, &tables).map_err(ConvertError::ClusterInvariant)?;
Ok(tables)
}
struct LoadedInput {
options: ConvertOptions,
input_schema: SchemaRef,
crs: Crs,
renames: Vec<(String, String)>,
geom_idx: usize,
geom_field: Field,
acc_cols: Vec<usize>,
full: RecordBatch,
geometries: Vec<Geometry<f64>>,
row_groups_total: usize,
row_groups_read: usize,
}
fn load_input_table(
source: &ConvertSource,
source_single: &InputSource,
options: &ConvertOptions,
) -> Result<LoadedInput, ConvertError> {
let (builder, read_schema) = project_builder_to_selection(source, source_single.open()?)?;
let crs = detect_crs_from_kv(builder.metadata().file_metadata().key_value_metadata())?;
let mut resolved = options.clone();
let (input_schema, renames) = resolve_reserved_column_collisions(&read_schema, &mut resolved);
let options = &resolved;
let bound_filter = bind_attribute_filter(options, &input_schema, &renames)?;
let geom_idx = find_geometry_column(&input_schema).ok_or(ConvertError::NoGeometryColumn)?;
let geom_field = input_schema.field(geom_idx).clone();
let acc_cols = validate_cluster_schema(&input_schema, options)?;
validate_coalesce_schema(&input_schema, options)?;
let row_groups_total = builder.metadata().num_row_groups();
let bbox_units = options.bbox.map(|b| bbox_to_crs_units(&b, crs));
let combined_sel = select_input_row_groups_combined(
builder.metadata(),
bbox_units.as_ref(),
bound_filter.as_ref(),
);
let (builder, row_groups_read) = match combined_sel {
Some(sel) => {
let n = sel.len();
let what = pruning_label(bbox_units.is_some(), bound_filter.is_some());
log::info!("{what} filter: reading {n}/{row_groups_total} input row groups");
(builder.with_row_groups(sel), n)
}
None => (builder, row_groups_total),
};
let reader = builder.build()?;
let mut batches: Vec<RecordBatch> = Vec::new();
for batch in reader {
batches.push(batch?);
}
let full = concat_batches(&read_schema, &batches)?;
let filter_mask: Option<Vec<Option<bool>>> =
bound_filter.as_ref().map(|f| f.eval_mask(&full, &|i| i));
let (full, geometries) = decode_and_filter_geometries(
full,
geom_idx,
&geom_field,
bbox_units.as_ref(),
filter_mask.as_deref(),
)?;
let full = if Arc::ptr_eq(&input_schema, &read_schema) {
full
} else {
RecordBatch::try_new(input_schema.clone(), full.columns().to_vec())?
};
Ok(LoadedInput {
options: resolved.clone(),
input_schema,
crs,
renames,
geom_idx,
geom_field,
acc_cols,
full,
geometries,
row_groups_total,
row_groups_read,
})
}
struct EmittedLevel {
orig: usize,
gsd: f64,
zoom: Option<u8>,
indices: Vec<usize>,
geoms: Vec<Geometry<f64>>,
vertex_count: usize,
coalesce: Option<CoalesceTable>,
}
#[allow(clippy::too_many_arguments)]
fn build_emitted_levels(
assignment: &Assignment,
features: &[AssignFeature],
geometries: &[Geometry<f64>],
level_specs: &[(f64, Option<u8>)],
level_reprs: &[Representation],
line_groups: Option<&Vec<u32>>,
coalesce_on: bool,
finest: usize,
crs: Crs,
row_min_levels: &[u8],
carriers: &[Vec<usize>],
options: &ConvertOptions,
) -> (Vec<EmittedLevel>, Vec<SkippedLevelReport>) {
let mut emitted: Vec<EmittedLevel> = Vec::new();
let mut skipped: Vec<SkippedLevelReport> = Vec::new();
for (level, &(gsd_m, zoom)) in level_specs.iter().enumerate() {
let member_indices: Vec<usize> = match options.mode {
Mode::Duplicating => {
let mut v = assignment.duplicating_at_level(level as u8);
if !carriers[level].is_empty() {
v.extend_from_slice(&carriers[level]);
v.sort_unstable();
}
v
}
Mode::Partitioning => assignment.partitioning_at_level(level as u8),
};
let verbatim = matches!(options.mode, Mode::Partitioning) || level == finest;
let coalesce: Option<CoalesceTable> = if coalesce_on && !verbatim {
let inputs: Vec<CoalesceInput<'_>> = features
.iter()
.filter(|f| f.kind == FeatureKind::Line)
.map(|f| CoalesceInput {
index: f.index,
geom: &geometries[f.index],
sort_key: f.sort_key,
group: line_groups.as_ref().map_or(0, |g| g[f.index]),
})
.collect();
Some(build_level_coalesce_table(
&inputs, level, finest, gsd_m, crs, options,
))
} else {
None
};
let member_indices: Vec<usize> = if let Some(table) = &coalesce {
let mut v: Vec<usize> = member_indices
.into_iter()
.filter(|&i| features[i].kind != FeatureKind::Line)
.collect();
v.extend(table.keys().copied());
v.sort_unstable();
v
} else {
member_indices
};
let mut indices = Vec::with_capacity(member_indices.len());
let mut geoms = Vec::with_capacity(member_indices.len());
let mut vertex_count = 0usize;
let repr = level_reprs[level];
let cascade_chain: Vec<CascadeStep> = if options.simplify.cascade && !verbatim {
(level..finest)
.rev()
.map(|li| CascadeStep {
gsd_meters: level_specs[li].0,
repr: level_reprs[li],
})
.collect()
} else {
Vec::new()
};
if verbatim {
for i in member_indices {
let g = &geometries[i];
vertex_count += count_vertices(g);
indices.push(i);
geoms.push(g.clone());
}
} else {
for i in member_indices {
if let Some((g, _)) = coalesce.as_ref().and_then(|t| t.get(&i)) {
vertex_count += count_vertices(g);
indices.push(i);
geoms.push(g.clone());
continue;
}
if usize::from(row_min_levels[i]) > level && is_carrier(&carriers[level], i) {
if let Some(sq) = carrier_square(&geometries[i], gsd_m, crs, &options.simplify)
{
vertex_count += count_vertices(&sq);
indices.push(i);
geoms.push(sq);
}
continue;
}
let simplified = if cascade_chain.is_empty() {
simplify_step(&geometries[i], gsd_m, crs, &options.simplify, repr)
} else {
simplify_cascade(&geometries[i], &cascade_chain, crs, &options.simplify)
};
match simplified {
Simplified::Keep(g) => {
vertex_count += count_vertices(&g);
indices.push(i);
geoms.push(g);
}
Simplified::Dropped => {}
}
}
}
if indices.is_empty() {
skipped.push(SkippedLevelReport {
planned_level: level,
gsd: gsd_m,
zoom,
});
continue;
}
emitted.push(EmittedLevel {
orig: level,
gsd: gsd_m,
zoom,
indices,
geoms,
vertex_count,
coalesce,
});
}
(emitted, skipped)
}
pub(crate) fn convert_to_overviews_source_strategy(
source: &ConvertSource,
output_path: &Path,
options: &ConvertOptions,
strategy: super::stream::Pass2Strategy,
) -> Result<ConvertReport, ConvertError> {
validate_options(options)?;
apply_property_selection(source, options)?;
source.set_spill_dir(options.spill_dir.as_deref());
check_mode_combinations(options)?;
let inert_options: ConvertOptions;
let options: &ConvertOptions = match adjusted_for_ladder_and_mode(options) {
Some(adjusted) => {
inert_options = adjusted;
&inert_options
}
None => options,
};
if options.streaming {
return super::stream::convert_streaming_strategy(source, output_path, options, strategy);
}
let source_single: &InputSource = match source {
ConvertSource::Single(s) => s.input(),
ConvertSource::Multi(_) => return Err(ConvertError::MultiPartitionRequiresStreaming),
};
let start = Instant::now();
if options.sort_key.is_some() && options.class_ranking.is_some() {
return Err(ConvertError::RankingConflict);
}
let LoadedInput {
options: resolved_options,
input_schema,
crs,
renames,
geom_idx,
geom_field,
acc_cols,
full,
geometries,
row_groups_total,
row_groups_read,
} = load_input_table(source, source_single, options)?;
let options = &resolved_options;
let num_features = full.num_rows();
let (sort_keys, ranking_provenance) =
resolve_ranking(&input_schema, &full, &geometries, options)?;
let num_lines = geometries
.iter()
.filter(|g| feature_kind(g) == FeatureKind::Line)
.count();
let coalesce_on = coalesce_effective(options, num_lines);
let line_groups: Option<Vec<u32>> = coalesce_on
.then(|| intern_coalesce_groups(&input_schema, &full, &ranking_provenance))
.flatten();
let level_specs = options.levels.resolve(options.gsd_base)?;
let level_gsds: Vec<f64> = level_specs.iter().map(|(g, _)| *g).collect();
let ladder_values = entry_zoom_column_values(options, &input_schema, &full)?;
let entry = resolve_entry_levels(options, &ladder_values, &level_specs)?;
let features: Vec<AssignFeature> = geometries
.iter()
.enumerate()
.map(|(i, g)| AssignFeature {
index: i,
bbox: geometry_bbox(g),
kind: feature_kind(g),
sort_key: sort_keys[i],
entry_level: entry.as_ref().and_then(|e| e[i]),
})
.collect();
let antimeridian_suspect_features = features
.iter()
.filter(|f| bbox_antimeridian_suspect(&f.bbox, crs))
.count();
warn_antimeridian_suspects(antimeridian_suspect_features);
let level_reprs = 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 assignment = if options.density.enabled {
apply_density_budget(
&assignment,
&features,
&level_gsds,
&options.assign,
&options.density,
crs,
)
} else {
assignment
};
let num_levels = level_gsds.len();
let finest = num_levels.saturating_sub(1);
let row_min_levels: Vec<u8> = assignment.assignments.iter().map(|a| a.min_level).collect();
let carriers = in_memory_carriers(
options,
&features,
&row_min_levels,
&geometries,
&level_gsds,
&level_reprs,
crs,
);
let cluster_tables = if options.cluster {
let min_levels: Vec<u8> = assignment.assignments.iter().map(|a| a.min_level).collect();
let acc_values = extract_accumulate_values(&full, &acc_cols);
Some(build_verified_cluster_tables(
&features,
&min_levels,
&level_gsds,
&acc_values,
crs,
options,
)?)
} else {
None
};
let (emitted, mut skipped) = build_emitted_levels(
&assignment,
&features,
&geometries,
&level_specs,
&level_reprs,
line_groups.as_ref(),
coalesce_on,
finest,
crs,
&row_min_levels,
&carriers,
options,
);
if emitted.is_empty() {
return Err(ConvertError::NoData);
}
warn_plan_skipped_levels(&skipped, num_features, emitted[0].gsd, emitted[0].zoom);
let geom_name = geom_field.name().clone();
let (source_schema, cluster_schema, out_schema) =
super::stream::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 = super::stream::build_writer_options(
writer_levels,
&emitted_gsds,
crs,
ranking_provenance,
&renames,
options,
);
let mut 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();
let mut level_reports = write_emitted_levels(
&mut writer,
&emitted,
&LevelWriteInputs {
full: &full,
source_schema: &source_schema,
cluster_schema: &cluster_schema,
out_schema: &out_schema,
non_geom_cols: &non_geom_cols,
geom_idx,
cluster_tables: cluster_tables.as_ref(),
acc_cols: &acc_cols,
finest,
},
options,
&mut skipped,
)?;
skipped.sort_by_key(|s| s.planned_level);
let meta = writer.finish()?;
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: log_remote_fetch(source),
})
}
struct LevelWriteInputs<'a> {
full: &'a RecordBatch,
source_schema: &'a Schema,
cluster_schema: &'a Schema,
out_schema: &'a Schema,
non_geom_cols: &'a [usize],
geom_idx: usize,
cluster_tables: Option<&'a ClusterTables>,
acc_cols: &'a [usize],
finest: usize,
}
fn write_emitted_levels(
writer: &mut OverviewWriter<File>,
emitted: &[EmittedLevel],
inputs: &LevelWriteInputs<'_>,
options: &ConvertOptions,
skipped: &mut Vec<SkippedLevelReport>,
) -> Result<Vec<LevelReport>, ConvertError> {
let mut level_reports = Vec::with_capacity(emitted.len());
for (level_idx, e) in emitted.iter().enumerate() {
let mut batch = build_level_batch(
inputs.source_schema,
inputs.full,
inputs.non_geom_cols,
inputs.geom_idx,
&e.indices,
&e.geoms,
)?;
if let Some(tables) = inputs.cluster_tables {
let table = (e.orig != inputs.finest).then(|| &tables[e.orig]);
batch = apply_cluster_columns(
batch,
inputs.cluster_schema,
&e.indices,
table,
inputs.acc_cols,
)?;
}
if options.coalesce_lines {
batch =
apply_coalesced_count(batch, inputs.out_schema, &e.indices, e.coalesce.as_ref())?;
}
let outcome =
writer.write_level(level_idx, Some(e.indices.len()), std::iter::once(batch))?;
record_level_outcome(
outcome,
SkippedLevelReport {
planned_level: e.orig,
gsd: e.gsd,
zoom: e.zoom,
},
e.indices.len(),
e.indices.len(),
e.vertex_count,
&mut level_reports,
skipped,
);
}
Ok(level_reports)
}
fn in_memory_carriers(
options: &ConvertOptions,
features: &[AssignFeature],
row_min_levels: &[u8],
geometries: &[Geometry<f64>],
level_gsds: &[f64],
level_reprs: &[Representation],
crs: Crs,
) -> Vec<Vec<usize>> {
let num_levels = level_gsds.len();
let finest = num_levels.saturating_sub(1);
let enabled = matches!(options.mode, Mode::Duplicating)
&& (options.simplify.collapse == CollapseMode::Square
|| level_reprs.contains(&Representation::Square));
let acc_levels: Vec<AccumulateLevel> = level_gsds
.iter()
.enumerate()
.map(|(l, &gsd)| AccumulateLevel {
gsd_meters: gsd,
enabled: enabled
&& l != finest
&& level_accumulates(options.simplify.collapse, level_reprs[l]),
})
.collect();
if !acc_levels.iter().any(|l| l.enabled) {
return vec![Vec::new(); num_levels];
}
let areas: Vec<f32> = geometries
.iter()
.map(|g| match g {
Geometry::Polygon(p) => p.unsigned_area() as f32,
Geometry::MultiPolygon(mp) => mp.unsigned_area() as f32,
_ => 0.0,
})
.collect();
tiny_polygon_carriers(
features,
row_min_levels,
&areas,
&acc_levels,
crs,
options.simplify.factor,
)
}
pub(super) fn log_remote_fetch(source: &ConvertSource) -> Option<crate::input::FetchStats> {
let stats = source.fetch_stats()?;
let pct = if stats.object_size > 0 {
100.0 * stats.bytes_fetched as f64 / stats.object_size as f64
} else {
0.0
};
log::info!(
"remote input: {} range requests, {:.2} MiB fetched of a {:.2} MiB object ({:.1}%)",
stats.requests,
stats.bytes_fetched as f64 / (1024.0 * 1024.0),
stats.object_size as f64 / (1024.0 * 1024.0),
pct
);
Some(stats)
}
pub(crate) fn detect_crs_from_kv(
kv: Option<&Vec<parquet::file::metadata::KeyValue>>,
) -> Result<Crs, ConvertError> {
let info = crate::quality::crs_info_from_kv_metadata(kv)?;
if info.is_wgs84 {
return Ok(Crs::Epsg4326);
}
if let Some(id) = &info.identifier {
let up = id.to_uppercase();
if up.contains("3857") || up.contains("900913") {
return Ok(Crs::Epsg3857);
}
}
Err(ConvertError::UnsupportedCrs {
crs: info
.identifier
.clone()
.or_else(|| info.name.clone())
.unwrap_or_else(|| "unknown".to_string()),
})
}
const WEBMERC_HALF_M: f64 = 20_037_508.342_789_244;
const WEBMERC_MAX_LAT: f64 = 85.051_128_779_806_59;
#[inline]
fn lnglat_to_webmerc(lng: f64, lat: f64) -> (f64, f64) {
use std::f64::consts::{FRAC_PI_4, PI};
let x = lng / 180.0 * WEBMERC_HALF_M;
let lat = lat.clamp(-WEBMERC_MAX_LAT, WEBMERC_MAX_LAT);
let y = (FRAC_PI_4 + lat.to_radians() / 2.0).tan().ln() / PI * WEBMERC_HALF_M;
(x, y)
}
pub(super) fn bbox_to_crs_units(bbox: &[f64; 4], crs: Crs) -> [f64; 4] {
match crs {
Crs::Epsg4326 => *bbox,
Crs::Epsg3857 => {
let (xmin, ymin) = lnglat_to_webmerc(bbox[0], bbox[1]);
let (xmax, ymax) = lnglat_to_webmerc(bbox[2], bbox[3]);
[xmin, ymin, xmax, ymax]
}
}
}
pub(super) fn bboxes_intersect(a: &[f64; 4], b: &[f64; 4]) -> bool {
a[0] <= b[2] && a[2] >= b[0] && a[1] <= b[3] && a[3] >= b[1]
}
pub(super) fn encode_concurrency_for(profile: MemoryProfile) -> usize {
const BOUNDED_ENCODE_CONCURRENCY_CAP: usize = 4;
let threads = rayon::current_num_threads().max(1);
match profile {
MemoryProfile::Bounded => threads.min(BOUNDED_ENCODE_CONCURRENCY_CAP),
MemoryProfile::Speed | MemoryProfile::Auto => threads,
}
}
pub(crate) fn select_input_row_groups(
metadata: &parquet::file::metadata::ParquetMetaData,
bbox_units: &[f64; 4],
) -> Vec<usize> {
let bounds = crate::covering::extract_row_group_bounds_from_metadata(metadata)
.unwrap_or_else(|_| vec![None; metadata.num_row_groups()]);
let filter = crate::tile::TileBounds {
lng_min: bbox_units[0],
lat_min: bbox_units[1],
lng_max: bbox_units[2],
lat_max: bbox_units[3],
};
(0..metadata.num_row_groups())
.filter(|&i| match bounds.get(i).and_then(|b| b.as_ref()) {
Some(b) => b.intersects(&filter),
None => true, })
.collect()
}
pub(super) fn bind_attribute_filter(
options: &ConvertOptions,
input_schema: &Schema,
renames: &[(String, String)],
) -> Result<Option<super::filter::BoundFilter>, ConvertError> {
let Some(src) = options.filter.as_deref() else {
return Ok(None);
};
let expr = super::filter::parse_filter(src)?;
Ok(Some(super::filter::BoundFilter::bind(
&expr,
input_schema,
renames,
)?))
}
pub(super) fn pruning_label(bbox: bool, filter: bool) -> &'static str {
match (bbox, filter) {
(true, true) => "bbox+attribute",
(true, false) => "bbox",
_ => "attribute",
}
}
fn select_input_row_groups_combined(
metadata: &parquet::file::metadata::ParquetMetaData,
bbox_units: Option<&[f64; 4]>,
filter: Option<&super::filter::BoundFilter>,
) -> Option<Vec<usize>> {
let bbox_sel: Option<Vec<usize>> = bbox_units.map(|bb| select_input_row_groups(metadata, bb));
let filter_sel: Option<Vec<usize>> = filter.map(|f| f.select_row_groups(metadata));
match (bbox_sel, filter_sel) {
(Some(a), Some(b)) => Some(a.into_iter().filter(|i| b.contains(i)).collect()),
(Some(a), None) => Some(a),
(None, Some(b)) => Some(b),
(None, None) => None,
}
}
pub(super) fn find_geometry_column(schema: &Schema) -> Option<usize> {
schema
.fields()
.iter()
.position(|f| f.name() == "geometry")
.or_else(|| {
schema
.fields()
.iter()
.position(|f| f.name().contains("geom"))
})
}
pub(super) fn geometry_bbox(g: &Geometry<f64>) -> [f64; 4] {
match g.bounding_rect() {
Some(r) => [r.min().x, r.min().y, r.max().x, r.max().y],
None => [0.0, 0.0, 0.0, 0.0],
}
}
pub(super) fn bbox_antimeridian_suspect(bbox: &[f64; 4], crs: Crs) -> bool {
bbox[2] - bbox[0] > crs.meters_to_units(180.0 * METERS_PER_DEGREE)
}
pub(super) fn warn_antimeridian_suspects(count: usize) {
if count > 0 {
log::warn!(
"{count} feature(s) have bounding boxes wider than 180° of longitude — \
likely antimeridian-crossing geometry. These will be assigned to \
overly coarse levels and defeat bbox pruning; pre-split them at \
±180° before converting (see docs/advanced-usage.md, \
\"Antimeridian-Crossing Geometry\")."
);
}
}
pub(super) const FULL_FILE_REMOTE_WARN_BYTES: u64 = 1 << 30;
pub(super) fn full_file_remote_warning(
remote_parts: usize,
row_groups_read: usize,
row_groups_total: usize,
object_size: u64,
) -> Option<String> {
if remote_parts == 0
|| row_groups_read < row_groups_total
|| object_size < FULL_FILE_REMOTE_WARN_BYTES
{
return None;
}
let gib = object_size as f64 / (1024.0 * 1024.0 * 1024.0);
let what = if remote_parts > 1 {
format!("{remote_parts} remote partitions totalling {gib:.1} GiB")
} else {
format!("a {gib:.1} GiB object")
};
Some(format!(
"full-file remote convert of {what}: the input is fetched once over \
the network (≈1× — the local spill keeps later passes off the network, \
#219) and staged under $TMPDIR. For a region of interest pass --bbox to \
fetch only the covering row groups (and skip the spill); otherwise \
downloading first (e.g. `aws s3 cp`) and converting locally avoids the \
second-pass disk read. Point --spill-dir (or $TMPDIR) at fast local disk \
with room for it, not a small tmpfs.",
))
}
pub(super) fn warn_full_file_remote(
source: &ConvertSource,
row_groups_read: usize,
row_groups_total: usize,
) {
let object_size = source.fetch_stats().map_or(0, |s| s.object_size);
let remote_parts = source.parts().iter().filter(|p| p.is_remote()).count();
if let Some(msg) =
full_file_remote_warning(remote_parts, row_groups_read, row_groups_total, object_size)
{
log::warn!("{msg}");
}
}
const SPILL_MARGIN_DENOM: u64 = 20;
pub(super) fn spill_space_warning(
estimated_spill_bytes: u64,
available_bytes: u64,
spill_dir: &Path,
) -> Option<String> {
let need = estimated_spill_bytes + estimated_spill_bytes / SPILL_MARGIN_DENOM;
if available_bytes >= need {
return None;
}
let gib = |b: u64| b as f64 / (1024.0 * 1024.0 * 1024.0);
Some(format!(
"projected input spill (≈{:.1} GiB — the selected input bytes are \
staged on local disk so later passes stay off the network, #219) may \
not fit: {} has {:.1} GiB free ({:.1} GiB short, including a 5% \
margin). If the volume fills mid-convert the spill degrades to \
network re-fetch; pass --spill-dir (spill_dir) to place it on a \
roomier volume, or free up space first.",
gib(estimated_spill_bytes),
spill_dir.display(),
gib(available_bytes),
gib(need - available_bytes),
))
}
pub(super) fn spill_space_check(
is_remote: bool,
estimated_spill_bytes: u64,
spill_dir: &Path,
probe: impl FnOnce(&Path) -> Option<u64>,
) -> Option<String> {
if !is_remote || estimated_spill_bytes == 0 {
return None;
}
let available = probe(spill_dir)?;
spill_space_warning(estimated_spill_bytes, available, spill_dir)
}
fn probe_available_space(dir: &Path) -> Option<u64> {
#[cfg(feature = "remote")]
{
fs4::available_space(dir).ok()
}
#[cfg(not(feature = "remote"))]
{
let _ = dir;
None
}
}
pub(super) fn warn_spill_space(
source: &ConvertSource,
estimated_spill_bytes: u64,
spill_dir: Option<&Path>,
) {
let dir = spill_dir.map_or_else(std::env::temp_dir, Path::to_path_buf);
if let Some(msg) = spill_space_check(
source.is_remote(),
estimated_spill_bytes,
&dir,
probe_available_space,
) {
log::warn!("{msg}");
}
}
pub(super) fn usable_geometry(g: &Geometry<f64>) -> bool {
use geo::coords_iter::CoordsIter;
let mut any = false;
for c in g.coords_iter() {
if !c.x.is_finite() || !c.y.is_finite() {
return false;
}
any = true;
}
any
}
pub(super) fn feature_kind(g: &Geometry<f64>) -> FeatureKind {
match g {
Geometry::Point(_) | Geometry::MultiPoint(_) => FeatureKind::Point,
Geometry::LineString(_) | Geometry::MultiLineString(_) | Geometry::Line(_) => {
FeatureKind::Line
}
_ => FeatureKind::Polygon,
}
}
#[derive(Debug)]
struct FeatureScan {
min_x: f64,
min_y: f64,
max_x: f64,
max_y: f64,
bbox_seen: bool,
any_coord: bool,
finite: bool,
}
impl FeatureScan {
fn new() -> Self {
FeatureScan {
min_x: f64::INFINITY,
min_y: f64::INFINITY,
max_x: f64::NEG_INFINITY,
max_y: f64::NEG_INFINITY,
bbox_seen: false,
any_coord: false,
finite: true,
}
}
#[inline]
fn note_bbox(&mut self, c: geo::Coord<f64>) {
self.any_coord = true;
if !c.x.is_finite() || !c.y.is_finite() {
self.finite = false;
return;
}
self.bbox_seen = true;
if c.x < self.min_x {
self.min_x = c.x;
}
if c.y < self.min_y {
self.min_y = c.y;
}
if c.x > self.max_x {
self.max_x = c.x;
}
if c.y > self.max_y {
self.max_y = c.y;
}
}
#[inline]
fn note_finite_only(&mut self, c: geo::Coord<f64>) {
self.any_coord = true;
if !c.x.is_finite() || !c.y.is_finite() {
self.finite = false;
}
}
fn bbox(&self) -> [f64; 4] {
if self.bbox_seen {
[self.min_x, self.min_y, self.max_x, self.max_y]
} else {
[0.0, 0.0, 0.0, 0.0]
}
}
}
fn scan_geometry_into(g: &Geometry<f64>, scan: &mut FeatureScan) {
use geo::coords_iter::CoordsIter;
let scan_polygon = |poly: &geo::Polygon<f64>, scan: &mut FeatureScan| {
for c in poly.exterior().coords_iter() {
scan.note_bbox(c);
}
for ring in poly.interiors() {
for c in ring.coords_iter() {
scan.note_finite_only(c);
}
}
};
match g {
Geometry::Polygon(poly) => scan_polygon(poly, scan),
Geometry::MultiPolygon(mp) => {
for poly in &mp.0 {
scan_polygon(poly, scan);
}
}
Geometry::GeometryCollection(gc) => {
for child in &gc.0 {
scan_geometry_into(child, scan);
}
}
other => {
for c in other.coords_iter() {
scan.note_bbox(c);
}
}
}
}
pub(super) fn scan_feature(g: &Geometry<f64>) -> Option<(FeatureKind, [f64; 4])> {
let mut scan = FeatureScan::new();
scan_geometry_into(g, &mut scan);
if !scan.finite || !scan.any_coord {
return None;
}
Some((feature_kind(g), scan.bbox()))
}
pub(super) fn count_vertices(g: &Geometry<f64>) -> usize {
use geo::coords_iter::CoordsIter;
g.coords_count()
}
pub(super) fn extract_sort_keys(col: &dyn Array) -> Vec<Option<f64>> {
use arrow_array::cast::AsArray;
use arrow_array::types::{
Float32Type, Float64Type, Int16Type, Int32Type, Int64Type, Int8Type, UInt16Type,
UInt32Type, UInt64Type, UInt8Type,
};
use arrow_schema::DataType;
let n = col.len();
macro_rules! collect_prim {
($ty:ty) => {{
let a = col.as_primitive::<$ty>();
(0..n)
.map(|i| {
if a.is_null(i) {
None
} else {
Some(a.value(i) as f64)
}
})
.collect()
}};
}
match col.data_type() {
DataType::Int8 => collect_prim!(Int8Type),
DataType::Int16 => collect_prim!(Int16Type),
DataType::Int32 => collect_prim!(Int32Type),
DataType::Int64 => collect_prim!(Int64Type),
DataType::UInt8 => collect_prim!(UInt8Type),
DataType::UInt16 => collect_prim!(UInt16Type),
DataType::UInt32 => collect_prim!(UInt32Type),
DataType::UInt64 => collect_prim!(UInt64Type),
DataType::Float32 => collect_prim!(Float32Type),
DataType::Float64 => collect_prim!(Float64Type),
_ => vec![None; n],
}
}
pub(super) fn mixed_geometry_field(name: &str) -> Arc<Field> {
use geoarrow_array::GeoArrowArray;
let typ = GeometryType::new(Default::default());
let empty = GeometryBuilder::new(typ).with_prefer_multi(false).finish();
Arc::new(empty.data_type().to_field(name, true))
}
pub(super) fn build_source_schema(
input_schema: &Schema,
geom_idx: usize,
geom_out_field: Arc<Field>,
) -> Schema {
let fields: Vec<Arc<Field>> = input_schema
.fields()
.iter()
.enumerate()
.map(|(i, f)| {
if i == geom_idx {
geom_out_field.clone()
} else {
f.clone()
}
})
.collect();
Schema::new(fields)
}
pub(super) fn build_level_batch(
source_schema: &Schema,
full: &RecordBatch,
non_geom_cols: &[usize],
geom_idx: usize,
indices: &[usize],
geoms: &[Geometry<f64>],
) -> Result<RecordBatch, ConvertError> {
let take_idx = UInt32Array::from(indices.iter().map(|&i| i as u32).collect::<Vec<_>>());
let mut columns: Vec<Arc<dyn Array>> = Vec::with_capacity(source_schema.fields().len());
let mut non_geom_iter = non_geom_cols.iter();
for i in 0..source_schema.fields().len() {
if i == geom_idx {
let typ = GeometryType::new(Default::default());
let mut b = GeometryBuilder::new(typ).with_prefer_multi(false);
b.extend_from_iter(geoms.iter().map(Some));
columns.push(b.finish().to_array_ref());
} else {
let src_col = *non_geom_iter.next().expect("non-geom column index");
let taken = take(full.column(src_col).as_ref(), &take_idx, None)?;
columns.push(taken);
}
}
Ok(RecordBatch::try_new(
Arc::new(source_schema.clone()),
columns,
)?)
}
pub(super) fn validate_cluster_schema(
schema: &Schema,
options: &ConvertOptions,
) -> Result<Vec<usize>, ConvertError> {
if !options.cluster {
return Ok(Vec::new());
}
if schema
.fields()
.iter()
.any(|f| f.name().eq_ignore_ascii_case(POINT_COUNT_COLUMN))
{
return Err(ConvertError::PointCountColumnPresent);
}
let mut indices = Vec::with_capacity(options.accumulate.len());
for spec in &options.accumulate {
let idx =
schema
.index_of(&spec.column)
.map_err(|_| ConvertError::AccumulateColumnMissing {
name: spec.column.clone(),
})?;
let dt = schema.field(idx).data_type();
if !is_numeric_type(dt) {
return Err(ConvertError::AccumulateColumnNotNumeric {
name: spec.column.clone(),
data_type: format!("{dt:?}"),
});
}
indices.push(idx);
}
Ok(indices)
}
fn is_numeric_type(dt: &DataType) -> bool {
matches!(
dt,
DataType::Int8
| DataType::Int16
| DataType::Int32
| DataType::Int64
| DataType::UInt8
| DataType::UInt16
| DataType::UInt32
| DataType::UInt64
| DataType::Float32
| DataType::Float64
)
}
pub(super) fn append_point_count_field(schema: &Schema) -> Schema {
let mut fields: Vec<Arc<Field>> = schema.fields().iter().cloned().collect();
fields.push(Arc::new(Field::new(
POINT_COUNT_COLUMN,
DataType::Int64,
false,
)));
Schema::new(fields)
}
pub(super) fn extract_accumulate_values(
batch: &RecordBatch,
acc_col_indices: &[usize],
) -> Vec<Vec<Option<f64>>> {
acc_col_indices
.iter()
.map(|&idx| extract_sort_keys(batch.column(idx).as_ref()))
.collect()
}
pub(super) fn apply_cluster_columns(
batch: RecordBatch,
out_schema: &Schema,
global_indices: &[usize],
table: Option<&HashMap<usize, ClusterEntry>>,
acc_cols: &[usize],
) -> Result<RecordBatch, ConvertError> {
use arrow_array::Int64Array;
debug_assert_eq!(batch.num_rows(), global_indices.len());
let mut columns: Vec<Arc<dyn Array>> = batch.columns().to_vec();
if let Some(table) = table {
for (s, &col_idx) in acc_cols.iter().enumerate() {
let overrides: Vec<Option<f64>> = global_indices
.iter()
.map(|g| table.get(g).and_then(|e| e.aggregates[s]))
.collect();
if overrides.iter().any(|o| o.is_some()) {
columns[col_idx] = overwrite_numeric_column(&columns[col_idx], &overrides)?;
}
}
}
let counts: Vec<i64> = global_indices
.iter()
.map(|g| table.and_then(|t| t.get(g)).map_or(1, |e| e.point_count))
.collect();
columns.push(Arc::new(Int64Array::from(counts)));
Ok(RecordBatch::try_new(Arc::new(out_schema.clone()), columns)?)
}
fn overwrite_numeric_column(
col: &Arc<dyn Array>,
overrides: &[Option<f64>],
) -> Result<Arc<dyn Array>, ConvertError> {
use arrow_array::cast::AsArray;
use arrow_array::types::{
Float32Type, Float64Type, Int16Type, Int32Type, Int64Type, Int8Type, UInt16Type,
UInt32Type, UInt64Type, UInt8Type,
};
use arrow_array::PrimitiveArray;
macro_rules! rebuild {
($ty:ty, $cast:expr) => {{
let a = col.as_primitive::<$ty>();
let rebuilt: PrimitiveArray<$ty> = (0..a.len())
.map(|i| match overrides[i] {
Some(v) => Some($cast(v)),
None => {
if a.is_null(i) {
None
} else {
Some(a.value(i))
}
}
})
.collect();
Ok(Arc::new(rebuilt))
}};
}
match col.data_type() {
DataType::Int8 => rebuild!(Int8Type, |v: f64| v.round() as i8),
DataType::Int16 => rebuild!(Int16Type, |v: f64| v.round() as i16),
DataType::Int32 => rebuild!(Int32Type, |v: f64| v.round() as i32),
DataType::Int64 => rebuild!(Int64Type, |v: f64| v.round() as i64),
DataType::UInt8 => rebuild!(UInt8Type, |v: f64| v.round() as u8),
DataType::UInt16 => rebuild!(UInt16Type, |v: f64| v.round() as u16),
DataType::UInt32 => rebuild!(UInt32Type, |v: f64| v.round() as u32),
DataType::UInt64 => rebuild!(UInt64Type, |v: f64| v.round() as u64),
DataType::Float32 => rebuild!(Float32Type, |v: f64| v as f32),
DataType::Float64 => rebuild!(Float64Type, |v: f64| v),
other => Err(ConvertError::AccumulateColumnNotNumeric {
name: "<accumulate column>".to_string(),
data_type: format!("{other:?}"),
}),
}
}
pub(super) type CoalesceTable = HashMap<usize, (Geometry<f64>, i32)>;
pub(super) fn validate_coalesce_schema(
schema: &Schema,
options: &ConvertOptions,
) -> Result<(), ConvertError> {
if !options.coalesce_lines {
return Ok(());
}
if schema
.fields()
.iter()
.any(|f| f.name().eq_ignore_ascii_case(COALESCED_COUNT_COLUMN))
{
return Err(ConvertError::CoalescedCountColumnPresent);
}
Ok(())
}
fn reserved_output_columns(options: &ConvertOptions) -> Vec<&'static str> {
let mut names = vec![LEVEL_COLUMN];
if options.cluster {
names.push(POINT_COUNT_COLUMN);
}
if options.coalesce_lines {
names.push(COALESCED_COUNT_COLUMN);
}
names
}
pub(super) fn resolve_reserved_column_collisions(
input_schema: &SchemaRef,
options: &mut ConvertOptions,
) -> (SchemaRef, Vec<(String, String)>) {
let reserved = reserved_output_columns(options);
let match_reserved = |name: &str| -> Option<&'static str> {
reserved
.iter()
.copied()
.find(|r| name.eq_ignore_ascii_case(r))
};
let mut taken: HashSet<String> = input_schema
.fields()
.iter()
.map(|f| f.name().to_ascii_lowercase())
.collect();
let mut renames: Vec<(String, String)> = Vec::new();
let mut new_fields: Vec<Arc<Field>> = Vec::with_capacity(input_schema.fields().len());
for field in input_schema.fields() {
let name = field.name();
let Some(reserved_name) = match_reserved(name) else {
new_fields.push(field.clone());
continue;
};
let mut candidate = format!("{name}_");
while taken.contains(&candidate.to_ascii_lowercase())
|| match_reserved(&candidate).is_some()
{
candidate.push('_');
}
taken.insert(candidate.to_ascii_lowercase());
log::warn!(
"input column {name:?} collides with the reserved overview column \
{reserved_name:?}; renaming the input column to {candidate:?} in \
the output (the reserved {reserved_name:?} column is authoritative)"
);
renames.push((name.to_string(), candidate.clone()));
new_fields.push(Arc::new(
Field::new(candidate, field.data_type().clone(), field.is_nullable())
.with_metadata(field.metadata().clone()),
));
}
if renames.is_empty() {
return (input_schema.clone(), renames);
}
let remap = |col: &str| -> Option<String> {
renames
.iter()
.find(|(old, _)| col.eq_ignore_ascii_case(old))
.map(|(_, new)| new.clone())
};
if let Some(new) = options.sort_key.as_deref().and_then(remap) {
options.sort_key = Some(new);
}
if let Some(cr) = options.class_ranking.as_mut() {
if let Some(new) = remap(&cr.column) {
cr.column = new;
}
}
for spec in &mut options.accumulate {
if let Some(new) = remap(&spec.column) {
spec.column = new;
}
}
if let Some(spec) = options.entry_zoom.as_mut() {
if let Some(new) = remap(&spec.column) {
spec.column = new;
}
}
let schema = Arc::new(Schema::new_with_metadata(
new_fields,
input_schema.metadata().clone(),
));
(schema, renames)
}
pub(super) fn append_coalesced_count_field(schema: &Schema) -> Schema {
let mut fields: Vec<Arc<Field>> = schema.fields().iter().cloned().collect();
fields.push(Arc::new(Field::new(
COALESCED_COUNT_COLUMN,
DataType::Int32,
false,
)));
Schema::new(fields)
}
pub(super) fn apply_coalesced_count(
batch: RecordBatch,
out_schema: &Schema,
global_indices: &[usize],
table: Option<&CoalesceTable>,
) -> Result<RecordBatch, ConvertError> {
use arrow_array::Int32Array;
debug_assert_eq!(batch.num_rows(), global_indices.len());
let mut columns: Vec<Arc<dyn Array>> = batch.columns().to_vec();
let counts: Vec<i32> = global_indices
.iter()
.map(|g| table.and_then(|t| t.get(g)).map_or(1, |(_, c)| *c))
.collect();
columns.push(Arc::new(Int32Array::from(counts)));
Ok(RecordBatch::try_new(Arc::new(out_schema.clone()), columns)?)
}
pub(super) fn coalesce_group_column(ranking: &RankingProvenance) -> Option<&str> {
match ranking.mode.as_str() {
"class-ranking" | "auto-overture-roads" => ranking.column.as_deref(),
_ => None,
}
}
#[derive(Debug, Default)]
pub(super) struct GroupInterner {
map: HashMap<String, u32>,
}
impl GroupInterner {
pub(super) const NULL_GROUP: u32 = u32::MAX;
pub(super) fn extend(&mut self, col: &dyn Array, out: &mut Vec<u32>) {
use arrow_array::cast::AsArray;
macro_rules! intern {
($arr:expr) => {{
let a = $arr;
for i in 0..a.len() {
if a.is_null(i) {
out.push(Self::NULL_GROUP);
} else {
let next = self.map.len() as u32;
let id = *self.map.entry(a.value(i).to_string()).or_insert(next);
out.push(id);
}
}
}};
}
match col.data_type() {
DataType::Utf8 => intern!(col.as_string::<i32>()),
DataType::LargeUtf8 => intern!(col.as_string::<i64>()),
_ => out.extend(std::iter::repeat_n(Self::NULL_GROUP, col.len())),
}
}
}
pub(super) fn coalesce_level_chains(
inputs: &[CoalesceInput<'_>],
level: usize,
finest: usize,
gsd_m: f64,
crs: Crs,
options: &ConvertOptions,
) -> Vec<super::coalesce::CoalescedLine> {
let budget = if options.density.enabled
&& options.density.drop_rate > 1.0
&& !options.density.drop_rate.is_nan()
&& level < finest
{
let keep = 1.0 / options.density.drop_rate;
let raw = inputs.len() as f64 * keep.powi((finest - level) as i32);
let max_chains = (raw.round() as usize).max(super::assign::MIN_DENSITY_LEVEL_FEATURES);
Some((max_chains, options.density.gamma))
} else {
None
};
coalesce_level_lines(
inputs,
gsd_m,
crs,
&options.assign,
&CoalesceParams {
snap_gsd_factor: options.coalesce_snap,
junction_angle_deg: options.coalesce_junction_angle,
budget,
},
)
}
pub(super) fn build_level_coalesce_table(
inputs: &[CoalesceInput<'_>],
level: usize,
finest: usize,
gsd_m: f64,
crs: Crs,
options: &ConvertOptions,
) -> CoalesceTable {
let chains = coalesce_level_chains(inputs, level, finest, gsd_m, crs, options);
let mut table = CoalesceTable::with_capacity(chains.len());
for chain in chains {
match simplify_for_level(&chain.geom, gsd_m, crs, &options.simplify) {
Simplified::Keep(g) => {
table.insert(chain.rep, (g, chain.count));
}
Simplified::Dropped => {}
}
}
table
}
pub(super) fn coalesce_effective(options: &ConvertOptions, num_lines: usize) -> bool {
if !options.coalesce_lines {
return false;
}
if num_lines > options.coalesce_max_level_rows {
log::warn!(
"coalescing skipped: {num_lines} candidate lines exceed \
--coalesce-max-level-rows {} (chaining holds a level's line \
geometries in memory; near-canonical levels this large need \
coalescing least). Output keeps the coalesced_count column \
(all 1).",
options.coalesce_max_level_rows
);
return false;
}
true
}
pub(super) fn resolve_entry_levels(
options: &ConvertOptions,
column_values: &[Option<f64>],
level_specs: &[(f64, Option<u8>)],
) -> Result<Option<Vec<Option<u8>>>, ConvertError> {
let Some(spec) = &options.entry_zoom else {
return Ok(None);
};
let level_zooms: Vec<Option<u8>> = level_specs.iter().map(|(_, z)| *z).collect();
let ladder = build_ladder(spec, column_values, &level_zooms)
.map_err(|e| ConvertError::InvalidConfig(format!("entry-zoom: {e}")))?;
let levels = entry_levels(&ladder, column_values, &level_zooms);
let placed = levels.iter().filter(|l| l.is_some()).count();
log::info!(
"[assign] entry-zoom ladder on {:?}: {} rung(s) {:?}; {placed} of {} feature(s) placed",
spec.column,
ladder.len(),
ladder.to_map(),
column_values.len(),
);
if matches!(options.simplify.collapse, CollapseMode::Drop) && options.simplify.factor > 0.0 {
log::warn!(
"[assign] entry-zoom ladder with --collapse off: a laddered feature \
admitted to a coarse level is still DROPPED there if its geometry \
simplifies below that level's tolerance, which is the usual case for \
the small-but-strong features a ladder exists to promote. Pass \
--collapse (representative point) or --collapse-square to keep them, \
or --simplify-factor 0 to disable simplification."
);
}
if options.density.enabled && options.sort_key.is_none() {
log::warn!(
"[assign] entry-zoom ladder with the density budget on and no \
--sort-key: the budget caps each level and sheds its lowest-priority \
survivors by SIZE, which can drop the small-but-strong features the \
ladder just promoted. Pass --no-density-drop, or --sort-key on the \
same column so the budget ranks the way the ladder does."
);
}
Ok(Some(levels))
}
pub(super) fn entry_zoom_column_values(
options: &ConvertOptions,
schema: &Schema,
table: &RecordBatch,
) -> Result<Vec<Option<f64>>, ConvertError> {
let Some(spec) = &options.entry_zoom else {
return Ok(Vec::new());
};
let idx = schema.index_of(&spec.column).map_err(|_| {
ConvertError::InvalidConfig(format!(
"entry-zoom column {:?} not found in the input schema",
spec.column
))
})?;
Ok(extract_sort_keys(table.column(idx).as_ref()))
}
pub(super) fn build_generalization(
gsds: &[f64],
_crs: Crs,
options: &ConvertOptions,
ranking: RankingProvenance,
renames: &[(String, String)],
) -> Generalization {
let levels = gsds
.iter()
.map(|&gsd_m| GeneralizationLevel {
simplify_tolerance_m: match options.mode {
Mode::Duplicating => options.simplify.factor * gsd_m,
Mode::Partitioning => 0.0,
},
thinning_factor: options.assign.polygon_thinning,
visibility_gate_m: options.assign.polygon_visibility * gsd_m,
geometry_types: Vec::new(),
})
.collect();
Generalization {
engine: format!("tylertoo {}", env!("CARGO_PKG_VERSION")),
gsd_base: if options.gsd_base == GSD_TILE_BASE {
None
} else {
Some(options.gsd_base)
},
levels,
cascade: if matches!(options.mode, Mode::Duplicating) && options.simplify.cascade {
Some(true)
} else {
None
},
collapse: match options.simplify.collapse {
CollapseMode::Drop => None,
mode => Some(
match mode {
CollapseMode::Point => "point",
CollapseMode::Square => "square",
CollapseMode::Drop => unreachable!(),
}
.to_string(),
),
},
representation: if options.representation.is_empty() {
None
} else {
Some(
options
.representation
.iter()
.map(|b| RepresentationBandProvenance {
zooms: [b.min_zoom, b.max_zoom],
repr: b.repr.as_str().to_string(),
})
.collect(),
)
},
ranking: Some(ranking),
density_drop: if options.density.enabled {
Some(DensityProvenance {
drop_rate: options.density.drop_rate,
gamma: options.density.gamma,
supercell_gsd_factor: SUPERCELL_GSD_FACTOR,
})
} else {
None
},
coalescing: if options.coalesce_lines {
Some(CoalescingProvenance {
enabled: true,
snap_tolerance_gsd_factor: options.coalesce_snap,
junction_angle: Some(options.coalesce_junction_angle),
max_level_rows: Some(options.coalesce_max_level_rows as u64),
coalesced_count_column: COALESCED_COUNT_COLUMN.to_string(),
})
} else {
None
},
clustering: if options.cluster {
Some(ClusteringProvenance {
enabled: true,
point_count_column: POINT_COUNT_COLUMN.to_string(),
accumulated: options
.accumulate
.iter()
.map(|s| AccumulatedColumn {
column: s.column.clone(),
op: s.op.as_str().to_string(),
})
.collect(),
})
} else {
None
},
renamed_columns: if renames.is_empty() {
None
} else {
Some(
renames
.iter()
.map(|(old, new)| (new.clone(), old.clone()))
.collect(),
)
},
}
}
fn resolve_ranking(
input_schema: &Schema,
full: &RecordBatch,
geometries: &[Geometry<f64>],
options: &ConvertOptions,
) -> Result<(Vec<Option<f64>>, RankingProvenance), ConvertError> {
let n = full.num_rows();
if let Some(name) = &options.sort_key {
let idx = input_schema
.index_of(name)
.map_err(|_| ConvertError::SortKeyColumnMissing { name: name.clone() })?;
let keys = extract_sort_keys(full.column(idx));
log::info!("overview ranking: explicit numeric sort-key column {name:?}");
return Ok((
keys,
RankingProvenance {
mode: "explicit-sort-key".to_string(),
column: Some(name.clone()),
ranks: None,
unknown_rank: None,
},
));
}
if let Some(cr) = &options.class_ranking {
let idx = input_schema.index_of(&cr.column).map_err(|_| {
ConvertError::ClassRankColumnMissing {
name: cr.column.clone(),
}
})?;
let keys = extract_class_ranks(full.column(idx), cr)?;
log::info!(
"overview ranking: explicit class-ranking on column {:?} ({} named classes, unknown_rank={})",
cr.column,
cr.ranks.len(),
cr.unknown_rank
);
return Ok((keys, class_ranking_provenance("class-ranking", cr)));
}
if !options.no_auto_rank {
if let Some((idx, col_name)) = find_road_class_column(input_schema, full) {
let cr = overture_road_ranking(col_name.clone());
let keys = extract_class_ranks(full.column(idx), &cr)?;
log::info!(
"overview ranking: auto-detected Overture road classes in column {col_name:?}; \
applying built-in ranking (motorway > … > service > tail)"
);
return Ok((keys, class_ranking_provenance("auto-overture-roads", &cr)));
}
if let Some((idx, col_name)) = find_confidence_column(input_schema, geometries) {
let keys = extract_sort_keys(full.column(idx));
log::info!(
"overview ranking: auto-detected Overture places confidence column {col_name:?} \
(numeric point ranking)"
);
return Ok((
keys,
RankingProvenance {
mode: "auto-confidence".to_string(),
column: Some(col_name),
ranks: None,
unknown_rank: None,
},
));
}
}
log::info!(
"overview ranking: no sort key specified or auto-detected; using size + \
deterministic-hash fallback"
);
Ok((
vec![None; n],
RankingProvenance {
mode: "size-fallback".to_string(),
column: None,
ranks: None,
unknown_rank: None,
},
))
}
pub(super) fn class_ranking_provenance(mode: &str, cr: &ClassRanking) -> RankingProvenance {
let ranks = if cr.ranks.len() <= MAX_PROVENANCE_RANKS {
Some(cr.ranks.iter().cloned().collect())
} else {
None
};
RankingProvenance {
mode: mode.to_string(),
column: Some(cr.column.clone()),
ranks,
unknown_rank: Some(cr.unknown_rank),
}
}
pub(super) fn extract_class_ranks(
col: &dyn Array,
ranking: &ClassRanking,
) -> Result<Vec<Option<f64>>, ConvertError> {
use arrow_array::cast::AsArray;
let map: HashMap<&str, f64> = ranking
.ranks
.iter()
.map(|(k, v)| (k.as_str(), *v))
.collect();
let n = col.len();
macro_rules! collect_str {
($arr:expr) => {{
let a = $arr;
(0..n)
.map(|i| {
if a.is_null(i) {
None
} else {
Some(*map.get(a.value(i)).unwrap_or(&ranking.unknown_rank))
}
})
.collect()
}};
}
match col.data_type() {
DataType::Utf8 => Ok(collect_str!(col.as_string::<i32>())),
DataType::LargeUtf8 => Ok(collect_str!(col.as_string::<i64>())),
other => Err(ConvertError::ClassRankColumnNotString {
name: ranking.column.clone(),
data_type: format!("{other:?}"),
}),
}
}
fn find_road_class_column(schema: &Schema, full: &RecordBatch) -> Option<(usize, String)> {
for (idx, f) in schema.fields().iter().enumerate() {
let lname = f.name().to_ascii_lowercase();
if lname != "road_class" && lname != "class" {
continue;
}
if !matches!(f.data_type(), DataType::Utf8 | DataType::LargeUtf8) {
continue;
}
if column_overlaps_road_vocab(full.column(idx)) {
return Some((idx, f.name().clone()));
}
}
None
}
fn column_overlaps_road_vocab(col: &dyn Array) -> bool {
use arrow_array::cast::AsArray;
let vocab: HashSet<&str> = KNOWN_ROAD_CLASSES.iter().copied().collect();
let mut found: HashSet<&str> = HashSet::new();
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 true;
}
}
}
}};
}
match col.data_type() {
DataType::Utf8 => scan!(col.as_string::<i32>()),
DataType::LargeUtf8 => scan!(col.as_string::<i64>()),
_ => return false,
}
found.len() >= ROAD_VOCAB_MIN_DISTINCT
}
fn find_confidence_column(
schema: &Schema,
geometries: &[Geometry<f64>],
) -> Option<(usize, String)> {
if geometries.is_empty() {
return None;
}
let points = geometries
.iter()
.filter(|g| matches!(feature_kind(g), FeatureKind::Point))
.count();
if points * 2 < geometries.len() {
return None;
}
for (idx, f) in schema.fields().iter().enumerate() {
if f.name().eq_ignore_ascii_case("confidence")
&& matches!(f.data_type(), DataType::Float32 | DataType::Float64)
{
return Some((idx, f.name().clone()));
}
}
None
}
pub(super) fn fill_level_bytes(
output_path: &Path,
meta: &super::level::OverviewsMeta,
reports: &mut [LevelReport],
) -> Result<(), ConvertError> {
let file = std::fs::File::open(output_path)?;
let pq = ParquetRecordBatchReaderBuilder::try_new(file)?;
let pmeta = pq.metadata();
let mut start = 0usize;
for (level, report) in meta.levels.iter().zip(reports.iter_mut()) {
let end = level.row_group_end as usize;
let mut uncompressed = 0i64;
let mut compressed = 0i64;
for rg in start..=end {
let rgm = pmeta.row_group(rg);
uncompressed += rgm.total_byte_size();
compressed += rgm.compressed_size();
}
report.uncompressed_bytes = uncompressed;
report.compressed_bytes = compressed;
start = end + 1;
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
mod scan_feature_props {
use super::super::{feature_kind, geometry_bbox, scan_feature, usable_geometry};
use geo::{Geometry, LineString, MultiLineString, MultiPolygon, Point, Polygon};
use proptest::prelude::*;
fn any_f64() -> impl Strategy<Value = f64> {
prop_oneof![
8 => -1.0e6f64..1.0e6f64,
1 => Just(f64::NAN),
1 => prop_oneof![Just(f64::INFINITY), Just(f64::NEG_INFINITY)],
]
}
fn any_coord() -> impl Strategy<Value = geo::Coord<f64>> {
(any_f64(), any_f64()).prop_map(|(x, y)| geo::Coord { x, y })
}
fn any_ring() -> impl Strategy<Value = LineString<f64>> {
proptest::collection::vec(any_coord(), 0..6).prop_map(LineString::new)
}
fn any_polygon() -> impl Strategy<Value = Polygon<f64>> {
(any_ring(), proptest::collection::vec(any_ring(), 0..3))
.prop_map(|(ext, ints)| Polygon::new(ext, ints))
}
fn any_geometry() -> impl Strategy<Value = Geometry<f64>> {
prop_oneof![
any_coord().prop_map(|c| Geometry::Point(Point::from(c))),
proptest::collection::vec(any_coord(), 0..5)
.prop_map(|cs| Geometry::MultiPoint(cs.into_iter().map(Point::from).collect())),
any_ring().prop_map(Geometry::LineString),
proptest::collection::vec(any_ring(), 0..3)
.prop_map(|ls| Geometry::MultiLineString(MultiLineString::new(ls))),
any_polygon().prop_map(Geometry::Polygon),
proptest::collection::vec(any_polygon(), 0..3)
.prop_map(|ps| Geometry::MultiPolygon(MultiPolygon::new(ps))),
]
}
proptest! {
#[test]
fn scan_feature_matches_components(g in any_geometry()) {
let expected = if usable_geometry(&g) {
Some((feature_kind(&g), geometry_bbox(&g)))
} else {
None
};
prop_assert_eq!(scan_feature(&g), expected);
}
}
}
use crate::overview::check::validate_file;
use crate::overview::level::gsd;
use crate::overview::reader::OverviewReader;
use arrow_array::{Float64Array, Int64Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema};
use geo::{Geometry, LineString, Point, Polygon};
use geoarrow::array::GeometryBuilder;
use geoarrow::datatypes::GeometryType;
use geoarrow_array::GeoArrowArray;
use geoparquet::writer::{
GeoParquetRecordBatchEncoder, GeoParquetWriterEncoding, GeoParquetWriterOptionsBuilder,
};
use parquet::arrow::ArrowWriter;
use crate::batch_processor::extract_geometries_from_array;
#[test]
fn full_file_remote_warning_gated_on_large_unpruned_remote() {
const BIG: u64 = 4 << 30; assert!(full_file_remote_warning(0, 8, 8, BIG).is_none());
assert!(full_file_remote_warning(1, 2, 8, BIG).is_none());
assert!(full_file_remote_warning(1, 8, 8, 100 << 20).is_none());
assert!(full_file_remote_warning(1, 8, 8, FULL_FILE_REMOTE_WARN_BYTES - 1).is_none());
assert!(full_file_remote_warning(1, 8, 8, FULL_FILE_REMOTE_WARN_BYTES).is_some());
let msg = full_file_remote_warning(1, 8, 8, BIG)
.expect("large full-file remote convert should warn");
assert!(msg.contains("--bbox"), "nudge should mention --bbox: {msg}");
assert!(
msg.contains("4.0 GiB"),
"nudge should state the object size: {msg}"
);
assert!(
!msg.contains("partitions"),
"single object: no part count in the message: {msg}"
);
}
#[test]
fn full_file_remote_warning_names_partition_count() {
const PART: u64 = 600 << 20; let msg = full_file_remote_warning(20, 40, 40, 20 * PART)
.expect("20 x 600 MB unpruned remote parts should warn");
assert!(
msg.contains("20 remote partitions"),
"message names the part count: {msg}"
);
assert!(
msg.contains("11.7 GiB"),
"message states the summed size: {msg}"
);
assert!(full_file_remote_warning(20, 40, 40, 500 << 20).is_none());
}
#[test]
fn spill_space_warning_names_dir_and_shortfall() {
let dir = Path::new("/mnt/scratch");
let msg = spill_space_warning(10 << 30, 1 << 30, dir).expect("shortfall should warn");
assert!(
msg.contains("/mnt/scratch"),
"warning names the spill dir: {msg}"
);
assert!(
msg.contains("--spill-dir"),
"warning suggests --spill-dir: {msg}"
);
assert!(
msg.contains("9.5 GiB"),
"warning states the shortfall: {msg}"
);
}
#[test]
fn spill_space_warning_quiet_with_ample_space() {
let dir = Path::new("/tmp");
assert!(spill_space_warning(1 << 30, 20 << 30, dir).is_none());
let est: u64 = 20 << 20;
let need = est + est / 20;
assert!(spill_space_warning(est, need, dir).is_none());
assert!(spill_space_warning(est, need - 1, dir).is_some());
}
#[test]
fn spill_space_check_gating() {
let dir = Path::new("/tmp");
assert!(spill_space_check(false, 10 << 30, dir, |_| panic!(
"free-space probe must not run for local inputs"
))
.is_none());
assert!(spill_space_check(true, 10 << 30, dir, |_| None).is_none());
assert!(spill_space_check(true, 0, dir, |_| Some(0)).is_none());
assert!(spill_space_check(true, 10 << 30, dir, |_| Some(1)).is_some());
}
#[test]
fn validate_options_rejects_missing_spill_dir() {
let opts = ConvertOptions {
spill_dir: Some(std::path::PathBuf::from(
"/nonexistent/tylertoo-spill-dir-272",
)),
..Default::default()
};
let err = validate_options(&opts).expect_err("missing spill dir must be rejected");
let msg = err.to_string();
assert!(msg.contains("spill-dir"), "error names the option: {msg}");
assert!(
msg.contains("/nonexistent/tylertoo-spill-dir-272"),
"error names the path: {msg}"
);
}
fn synthetic_geometries() -> Vec<Geometry<f64>> {
let mut geoms = Vec::new();
for i in 0..6 {
let x = i as f64 * 5.0;
let y = i as f64 * 3.0;
geoms.push(Geometry::Point(Point::new(x, y)));
}
for i in 0..4 {
let base = 40.0 + i as f64 * 10.0;
let ls = LineString::from(
(0..12)
.map(|k| {
(
base + k as f64 * 0.5,
(k as f64 * 0.6).sin() + i as f64 * 8.0,
)
})
.collect::<Vec<_>>(),
);
geoms.push(Geometry::LineString(ls));
}
for i in 0..4 {
let cx = -60.0 + i as f64 * 12.0;
let cy = -40.0 - i as f64 * 5.0;
let half = 2.0 + i as f64 * 1.5;
let ext = LineString::from(vec![
(cx - half, cy - half),
(cx + half, cy - half),
(cx + half, cy + half),
(cx - half, cy + half),
(cx - half, cy - half),
]);
geoms.push(Geometry::Polygon(Polygon::new(ext, vec![])));
}
geoms
}
fn build_geometry_array(geoms: &[Geometry<f64>]) -> geoarrow::array::GeometryArray {
let typ = GeometryType::new(Default::default());
let mut b = GeometryBuilder::new(typ).with_prefer_multi(false);
b.extend_from_iter(geoms.iter().map(Some));
b.finish()
}
fn output_column_names(path: &Path) -> Vec<String> {
let file = std::fs::File::open(path).unwrap();
let builder = ParquetRecordBatchReaderBuilder::try_new(file).unwrap();
builder
.schema()
.fields()
.iter()
.map(|f| f.name().clone())
.collect()
}
fn write_input(
path: &Path,
geoms: &[Geometry<f64>],
extra_level_col: bool,
crs_metadata: Option<geoarrow::datatypes::Metadata>,
) {
let n = geoms.len();
let id = Int64Array::from((0..n as i64).collect::<Vec<_>>());
let name = StringArray::from((0..n).map(|i| format!("f{i}")).collect::<Vec<_>>());
let rank = Float64Array::from((0..n).map(|i| (n - i) as f64).collect::<Vec<_>>());
let geom_arr = if let Some(md) = crs_metadata {
let typ = GeometryType::new(Arc::new(md));
let mut b = GeometryBuilder::new(typ).with_prefer_multi(false);
b.extend_from_iter(geoms.iter().map(Some));
b.finish()
} else {
build_geometry_array(geoms)
};
let geom_field = geom_arr.data_type().to_field("geometry", true);
let mut fields = vec![
Arc::new(Field::new("id", DataType::Int64, false)),
Arc::new(Field::new("name", DataType::Utf8, false)),
Arc::new(Field::new("rank", DataType::Float64, false)),
];
let mut columns: Vec<Arc<dyn Array>> = vec![Arc::new(id), Arc::new(name), Arc::new(rank)];
if extra_level_col {
fields.push(Arc::new(Field::new("level", DataType::Int32, false)));
columns.push(Arc::new(arrow_array::Int32Array::from(vec![0i32; n])));
}
fields.push(Arc::new(geom_field));
columns.push(geom_arr.to_array_ref());
let schema = Arc::new(Schema::new(fields));
let batch = RecordBatch::try_new(schema.clone(), columns).unwrap();
let gpq_options = GeoParquetWriterOptionsBuilder::default()
.set_encoding(GeoParquetWriterEncoding::WKB)
.set_generate_covering(true)
.build();
let encoder = GeoParquetRecordBatchEncoder::try_new(&schema, &gpq_options).unwrap();
let target_schema = encoder.target_schema();
let file = std::fs::File::create(path).unwrap();
let mut writer = ArrowWriter::try_new(file, target_schema, None).unwrap();
let mut encoder = encoder;
let encoded = encoder.encode_record_batch(&batch).unwrap();
writer.write(&encoded).unwrap();
writer.append_key_value_metadata(encoder.into_keyvalue().unwrap());
writer.close().unwrap();
}
fn read_level_rows(
reader: &OverviewReader,
level: usize,
) -> Vec<(i64, String, f64, Geometry<f64>)> {
use arrow_array::cast::AsArray;
use arrow_array::types::Float64Type;
let rdr = reader.read_level(level, None).unwrap();
let mut out = Vec::new();
for batch in rdr {
let batch = batch.unwrap();
let ids = batch
.column(batch.schema().index_of("id").unwrap())
.as_primitive::<arrow_array::types::Int64Type>()
.clone();
let names = batch
.column(batch.schema().index_of("name").unwrap())
.as_string::<i32>()
.clone();
let ranks = batch
.column(batch.schema().index_of("rank").unwrap())
.as_primitive::<Float64Type>()
.clone();
let gcol = batch.column(batch.schema().index_of("geometry").unwrap());
let garr: Arc<dyn GeoArrowArray> = from_arrow_array(
gcol.as_ref(),
batch
.schema()
.field(batch.schema().index_of("geometry").unwrap()),
)
.unwrap();
let mut gvec = Vec::new();
extract_geometries_from_array(garr.as_ref(), &mut gvec).unwrap();
for (i, g) in gvec.iter().enumerate() {
out.push((
ids.value(i),
names.value(i).to_string(),
ranks.value(i),
g.clone(),
));
}
}
out
}
fn write_input_partition(
path: &Path,
geoms: &[Geometry<f64>],
range: std::ops::Range<usize>,
row_group_rows: Option<usize>,
) {
use parquet::file::properties::WriterProperties;
let total = geoms.len();
let idx: Vec<usize> = range.collect();
let id = Int64Array::from(idx.iter().map(|&i| i as i64).collect::<Vec<_>>());
let name = StringArray::from(idx.iter().map(|&i| format!("f{i}")).collect::<Vec<_>>());
let rank = Float64Array::from(idx.iter().map(|&i| (total - i) as f64).collect::<Vec<_>>());
let part_geoms: Vec<Geometry<f64>> = idx.iter().map(|&i| geoms[i].clone()).collect();
let geom_arr = build_geometry_array(&part_geoms);
let geom_field = geom_arr.data_type().to_field("geometry", true);
let fields = vec![
Arc::new(Field::new("id", DataType::Int64, false)),
Arc::new(Field::new("name", DataType::Utf8, false)),
Arc::new(Field::new("rank", DataType::Float64, false)),
Arc::new(geom_field),
];
let columns: Vec<Arc<dyn Array>> = vec![
Arc::new(id),
Arc::new(name),
Arc::new(rank),
geom_arr.to_array_ref(),
];
let schema = Arc::new(Schema::new(fields));
let batch = RecordBatch::try_new(schema.clone(), columns).unwrap();
let gpq_options = GeoParquetWriterOptionsBuilder::default()
.set_encoding(GeoParquetWriterEncoding::WKB)
.set_generate_covering(true)
.build();
let mut encoder = GeoParquetRecordBatchEncoder::try_new(&schema, &gpq_options).unwrap();
let target_schema = encoder.target_schema();
let props = row_group_rows.map(|n| {
WriterProperties::builder()
.set_max_row_group_row_count(Some(n))
.build()
});
let file = std::fs::File::create(path).unwrap();
let mut writer = ArrowWriter::try_new(file, target_schema, props).unwrap();
let encoded = encoder.encode_record_batch(&batch).unwrap();
writer.write(&encoded).unwrap();
writer.append_key_value_metadata(encoder.into_keyvalue().unwrap());
writer.close().unwrap();
}
fn convert_and_export(input: &Path, workdir: &Path, opts: &ConvertOptions) -> Vec<u8> {
use crate::overview::export::{export_pmtiles, ExportOptions};
let stem = input
.file_name()
.unwrap_or_default()
.to_string_lossy()
.replace('.', "_");
let overview = workdir.join(format!("{stem}-overview.parquet"));
let pmtiles = workdir.join(format!("{stem}.pmtiles"));
convert_to_overviews(input, &overview, opts).unwrap();
export_pmtiles(&overview, &pmtiles, &ExportOptions::default()).unwrap();
std::fs::read(&pmtiles).unwrap()
}
fn multi_test_options() -> ConvertOptions {
ConvertOptions {
mode: Mode::Duplicating,
levels: LevelPlan::ZoomRange {
min_zoom: 2,
max_zoom: 8,
},
..Default::default()
}
}
#[test]
fn multi_partition_output_matches_single_file() {
let geoms = synthetic_geometries();
let n = geoms.len();
let dir = tempfile::tempdir().unwrap();
let single = dir.path().join("single.parquet");
write_input_partition(&single, &geoms, 0..n, None);
let parts = dir.path().join("parts");
std::fs::create_dir(&parts).unwrap();
write_input_partition(&parts.join("part-000.parquet"), &geoms, 0..5, None);
write_input_partition(&parts.join("part-001.parquet"), &geoms, 5..9, None);
write_input_partition(&parts.join("part-002.parquet"), &geoms, 9..n, None);
let opts = multi_test_options();
let report =
convert_to_overviews(&parts, dir.path().join("probe-overview.parquet"), &opts).unwrap();
assert_eq!(report.input_features, n);
assert_eq!(report.row_groups_total, 3, "one row group per partition");
let pm_single = convert_and_export(&single, dir.path(), &opts);
let pm_multi = convert_and_export(&parts, dir.path(), &opts);
assert!(
pm_single == pm_multi,
"multi-partition output must be byte-identical to single-file \
({} vs {} bytes)",
pm_single.len(),
pm_multi.len()
);
}
#[test]
fn multi_partition_zero_row_part_matches_single_file() {
let geoms = synthetic_geometries();
let n = geoms.len();
let dir = tempfile::tempdir().unwrap();
let single = dir.path().join("single.parquet");
write_input_partition(&single, &geoms, 0..n, None);
let parts = dir.path().join("parts");
std::fs::create_dir(&parts).unwrap();
write_input_partition(&parts.join("part-000.parquet"), &geoms, 0..5, None);
write_input_partition(&parts.join("part-001.parquet"), &geoms, 5..5, None); write_input_partition(&parts.join("part-002.parquet"), &geoms, 5..n, None);
let opts = multi_test_options();
let pm_single = convert_and_export(&single, dir.path(), &opts);
let pm_multi = convert_and_export(&parts, dir.path(), &opts);
assert!(
pm_single == pm_multi,
"0-row partition must not shift rows or offsets"
);
}
#[test]
fn multi_partition_bbox_selection_matches_single_file() {
let geoms = synthetic_geometries();
let n = geoms.len();
let dir = tempfile::tempdir().unwrap();
let single = dir.path().join("single.parquet");
write_input_partition(&single, &geoms, 0..n, Some(2));
let parts = dir.path().join("parts");
std::fs::create_dir(&parts).unwrap();
write_input_partition(&parts.join("part-000.parquet"), &geoms, 0..6, Some(2));
write_input_partition(&parts.join("part-001.parquet"), &geoms, 6..10, Some(2));
write_input_partition(&parts.join("part-002.parquet"), &geoms, 10..n, Some(2));
let bbox = Some([-100.0, -70.0, 30.0, 20.0]);
let opts = ConvertOptions {
bbox,
..multi_test_options()
};
let report =
convert_to_overviews(&parts, dir.path().join("probe-overview.parquet"), &opts).unwrap();
assert!(
report.row_groups_read < report.row_groups_total,
"bbox must prune row groups across parts: {}/{}",
report.row_groups_read,
report.row_groups_total
);
let pm_single = convert_and_export(&single, dir.path(), &opts);
let pm_multi = convert_and_export(&parts, dir.path(), &opts);
assert!(
pm_single == pm_multi,
"per-part bbox selection must keep offsets aligned"
);
}
#[test]
fn multi_partition_requires_streaming_pipeline() {
let geoms = synthetic_geometries();
let dir = tempfile::tempdir().unwrap();
let parts = dir.path().join("parts");
std::fs::create_dir(&parts).unwrap();
write_input_partition(&parts.join("part-000.parquet"), &geoms, 0..5, None);
write_input_partition(
&parts.join("part-001.parquet"),
&geoms,
5..geoms.len(),
None,
);
let opts = ConvertOptions {
streaming: false,
..multi_test_options()
};
let err = convert_to_overviews(&parts, dir.path().join("out.parquet"), &opts).unwrap_err();
assert!(
matches!(err, ConvertError::MultiPartitionRequiresStreaming),
"expected MultiPartitionRequiresStreaming, got {err:?}"
);
}
#[test]
fn multi_partition_crs_mismatch_rejected() {
let geoms = synthetic_geometries();
let dir = tempfile::tempdir().unwrap();
let parts = dir.path().join("parts");
std::fs::create_dir(&parts).unwrap();
write_input(&parts.join("a.parquet"), &geoms, false, None);
let projjson = serde_json::json!({
"type": "ProjectedCRS",
"name": "UTM zone 33N",
"id": { "authority": "EPSG", "code": 32633 }
});
let md = geoarrow::datatypes::Metadata::new(
geoarrow::datatypes::Crs::from_projjson(projjson),
None,
);
write_input(&parts.join("b.parquet"), &geoms, false, Some(md));
let err = convert_to_overviews(
&parts,
dir.path().join("out.parquet"),
&multi_test_options(),
)
.unwrap_err();
match err {
ConvertError::Input(crate::input::InputError::IncompatiblePartition {
offender,
..
}) => {
assert!(offender.ends_with("b.parquet"), "offender: {offender}");
}
other => panic!("expected IncompatiblePartition, got {other:?}"),
}
}
#[test]
fn duplicating_canonical_matches_input() {
let geoms = synthetic_geometries();
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let opts = ConvertOptions {
mode: Mode::Duplicating,
levels: LevelPlan::ZoomRange {
min_zoom: 2,
max_zoom: 8,
},
..Default::default()
};
let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
let vr = validate_file(tout.path()).unwrap();
assert!(
vr.is_valid(),
"failures: {:?}",
vr.failures().collect::<Vec<_>>()
);
let reader = OverviewReader::open(tout.path()).unwrap();
assert_eq!(reader.mode(), Mode::Duplicating);
let canonical = reader.num_levels() - 1;
let rows = read_level_rows(&reader, canonical);
assert_eq!(rows.len(), geoms.len());
for (i, (id, name, rank, geom)) in rows.iter().enumerate() {
assert_eq!(*id, i as i64);
assert_eq!(name, &format!("f{i}"));
assert_eq!(*rank, (geoms.len() - i) as f64);
assert_eq!(geom, &geoms[i], "canonical geometry must be verbatim");
}
assert_eq!(report.input_features, geoms.len());
assert_eq!(report.levels[canonical].feature_count, geoms.len());
}
#[test]
fn duplicating_coarse_levels_monotone() {
let geoms = synthetic_geometries();
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let opts = ConvertOptions {
mode: Mode::Duplicating,
levels: LevelPlan::ZoomRange {
min_zoom: 1,
max_zoom: 10,
},
..Default::default()
};
let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
for w in report.levels.windows(2) {
assert!(
w[0].feature_count <= w[1].feature_count,
"feature counts not monotone: {:?}",
report.levels
);
assert!(
w[0].vertex_count <= w[1].vertex_count,
"vertex counts not monotone: {:?}",
report.levels
);
}
assert_eq!(report.levels.last().unwrap().feature_count, geoms.len());
assert!(validate_file(tout.path()).unwrap().is_valid());
}
fn polygon_fixture() -> Vec<Geometry<f64>> {
let mut geoms = Vec::new();
for i in 0..12 {
let cx = -150.0 + i as f64 * 25.0;
let cy = -60.0 + (i % 5) as f64 * 27.0;
let half = 20.0 / (3f64).powi(i % 9);
let ext = LineString::from(vec![
(cx - half, cy - half),
(cx + half, cy - half),
(cx + half, cy + half),
(cx - half, cy + half),
(cx - half, cy - half),
]);
geoms.push(Geometry::Polygon(Polygon::new(ext, vec![])));
}
geoms
}
fn band(min_zoom: u8, max_zoom: u8, repr: Representation) -> RepresentationBand {
RepresentationBand {
min_zoom,
max_zoom,
repr,
}
}
#[test]
fn representation_point_band_points_coarse_polygons_fine() {
let geoms = polygon_fixture();
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let opts = ConvertOptions {
mode: Mode::Duplicating,
levels: LevelPlan::ZoomRange {
min_zoom: 2,
max_zoom: 8,
},
representation: vec![band(2, 5, Representation::Point)],
..Default::default()
};
let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
assert!(validate_file(tout.path()).unwrap().is_valid());
let reader = OverviewReader::open(tout.path()).unwrap();
for (idx, lvl) in report.levels.iter().enumerate() {
let rows = read_level_rows(&reader, idx);
assert!(!rows.is_empty());
let zoom = lvl.zoom.expect("zoom-range plan records zooms");
if zoom <= 5 {
for (id, _, _, g) in &rows {
assert!(
matches!(g, Geometry::Point(_)),
"level z{zoom} feature {id} must be a Point, got {g:?}"
);
}
} else {
assert!(
rows.iter().any(|(_, _, _, g)| matches!(
g,
Geometry::Polygon(_) | Geometry::MultiPolygon(_)
)),
"level z{zoom} must keep polygons"
);
}
}
let canonical = read_level_rows(&reader, report.levels.len() - 1);
assert_eq!(canonical.len(), geoms.len());
for (i, (_, _, _, g)) in canonical.iter().enumerate() {
assert_eq!(g, &geoms[i]);
}
let plain = ConvertOptions {
representation: Vec::new(),
..opts.clone()
};
let tout2 = tempfile::NamedTempFile::new().unwrap();
let plain_report = convert_to_overviews(tin.path(), tout2.path(), &plain).unwrap();
assert!(
report.levels[0].feature_count >= plain_report.levels[0].feature_count,
"point band must not lose coarse coverage vs the gated polygon run"
);
let gen = reader
.meta()
.generalization
.as_ref()
.expect("generalization block present");
let bands = gen.representation.as_ref().expect("bands recorded");
assert_eq!(bands.len(), 1);
assert_eq!(bands[0].zooms, [2, 5]);
assert_eq!(bands[0].repr, "point");
assert!(gen.collapse.is_none(), "no global collapse requested");
}
#[test]
fn representation_point_band_streaming_matches_in_memory() {
let geoms = polygon_fixture();
let tin = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let base = ConvertOptions {
mode: Mode::Duplicating,
levels: LevelPlan::ZoomRange {
min_zoom: 2,
max_zoom: 8,
},
representation: vec![band(2, 4, Representation::Point)],
..Default::default()
};
let t_stream = tempfile::NamedTempFile::new().unwrap();
let t_mem = tempfile::NamedTempFile::new().unwrap();
let r1 = convert_to_overviews(tin.path(), t_stream.path(), &base).unwrap();
let mem = ConvertOptions {
streaming: false,
..base
};
let r2 = convert_to_overviews(tin.path(), t_mem.path(), &mem).unwrap();
assert_eq!(r1.levels.len(), r2.levels.len());
let rd1 = OverviewReader::open(t_stream.path()).unwrap();
let rd2 = OverviewReader::open(t_mem.path()).unwrap();
for lvl in 0..r1.levels.len() {
assert_eq!(
read_level_rows(&rd1, lvl),
read_level_rows(&rd2, lvl),
"level {lvl} rows must match across engines"
);
}
}
#[test]
fn representation_square_band_type_preserving() {
let mut geoms = polygon_fixture();
for i in 0..30 {
let cx = 10.0 + (i % 6) as f64 * 0.02;
let cy = 20.0 + (i / 6) as f64 * 0.02;
let half = 0.004;
let ext = LineString::from(vec![
(cx - half, cy - half),
(cx + half, cy - half),
(cx + half, cy + half),
(cx - half, cy + half),
(cx - half, cy - half),
]);
geoms.push(Geometry::Polygon(Polygon::new(ext, vec![])));
}
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let opts = ConvertOptions {
mode: Mode::Duplicating,
levels: LevelPlan::ZoomRange {
min_zoom: 4,
max_zoom: 10,
},
representation: vec![band(4, 7, Representation::Square)],
..Default::default()
};
let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
assert!(validate_file(tout.path()).unwrap().is_valid());
let reader = OverviewReader::open(tout.path()).unwrap();
for (idx, lvl) in report.levels.iter().enumerate() {
let zoom = lvl.zoom.unwrap();
let rows = read_level_rows(&reader, idx);
for (id, _, _, g) in &rows {
assert!(
matches!(g, Geometry::Polygon(_) | Geometry::MultiPolygon(_)),
"square band is type-preserving; level z{zoom} feature {id} \
is {g:?}"
);
}
}
let z4_rows = read_level_rows(&reader, 0);
let tol_deg = report.levels[0].gsd / METERS_PER_DEGREE;
let squares = z4_rows
.iter()
.filter(|(_, _, _, g)| match g {
Geometry::Polygon(p) => {
let r = geo::BoundingRect::bounding_rect(p).unwrap();
p.exterior().0.len() == 5
&& (r.width() - tol_deg).abs() < 1e-9
&& (r.height() - tol_deg).abs() < 1e-9
}
_ => false,
})
.count();
assert!(
squares > 0,
"band level must contain dithered placeholder squares"
);
let gen = reader.meta().generalization.as_ref().unwrap();
let bands = gen.representation.as_ref().unwrap();
assert_eq!(bands[0].repr, "square");
}
#[test]
fn collapse_square_global_disposition() {
let geoms = polygon_fixture();
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let opts = ConvertOptions {
mode: Mode::Duplicating,
levels: LevelPlan::ZoomRange {
min_zoom: 2,
max_zoom: 8,
},
simplify: SimplifyOptions {
collapse: CollapseMode::Square,
..Default::default()
},
..Default::default()
};
let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
assert!(validate_file(tout.path()).unwrap().is_valid());
let reader = OverviewReader::open(tout.path()).unwrap();
for idx in 0..report.levels.len() {
for (_, _, _, g) in read_level_rows(&reader, idx) {
assert!(matches!(
g,
Geometry::Polygon(_) | Geometry::MultiPolygon(_)
));
}
}
let gen = reader.meta().generalization.as_ref().unwrap();
assert_eq!(gen.collapse.as_deref(), Some("square"));
assert!(gen.representation.is_none());
}
#[test]
fn representation_validation_rejects_bad_bands() {
let mk = |mode: Mode, levels: LevelPlan, bands: Vec<RepresentationBand>| ConvertOptions {
mode,
levels,
representation: bands,
..Default::default()
};
let zr = |lo: u8, hi: u8| LevelPlan::ZoomRange {
min_zoom: lo,
max_zoom: hi,
};
let err = validate_options(&mk(
Mode::Partitioning,
zr(0, 6),
vec![band(0, 3, Representation::Point)],
))
.unwrap_err();
assert!(matches!(err, ConvertError::InvalidConfig(_)), "{err}");
let err = validate_options(&mk(
Mode::Duplicating,
LevelPlan::Gsds(vec![1000.0, 100.0]),
vec![band(0, 3, Representation::Point)],
))
.unwrap_err();
assert!(matches!(err, ConvertError::InvalidConfig(_)), "{err}");
let err = validate_options(&mk(
Mode::Duplicating,
zr(0, 6),
vec![band(0, 6, Representation::Point)],
))
.unwrap_err();
assert!(matches!(err, ConvertError::InvalidConfig(_)), "{err}");
let err = validate_options(&mk(
Mode::Duplicating,
zr(0, 8),
vec![
band(0, 3, Representation::Point),
band(3, 5, Representation::Square),
],
))
.unwrap_err();
assert!(matches!(err, ConvertError::InvalidConfig(_)), "{err}");
let err = validate_options(&mk(
Mode::Duplicating,
zr(0, 8),
vec![band(3, 5, Representation::Point)],
))
.unwrap_err();
assert!(matches!(err, ConvertError::InvalidConfig(_)), "{err}");
validate_options(&mk(
Mode::Duplicating,
zr(0, 8),
vec![band(3, 5, Representation::Square)],
))
.expect("mid-plan square band is valid");
validate_options(&mk(
Mode::Duplicating,
zr(0, 8),
vec![
band(0, 3, Representation::Point),
band(4, 6, Representation::Square),
],
))
.expect("point prefix + square band is valid");
}
#[test]
fn parse_representation_spec_grammar() {
let bands = parse_representation_spec("0-7:point,8-14:geom").unwrap();
assert_eq!(bands.len(), 2);
assert_eq!(bands[0], band(0, 7, Representation::Point));
assert_eq!(bands[1], band(8, 14, Representation::Geometry));
let bands = parse_representation_spec(" 0-5 : square ").unwrap();
assert_eq!(bands, vec![band(0, 5, Representation::Square)]);
let bands = parse_representation_spec("3:geometry").unwrap();
assert_eq!(bands, vec![band(3, 3, Representation::Geometry)]);
assert!(parse_representation_spec("").is_err());
assert!(parse_representation_spec("0-7").is_err());
assert!(parse_representation_spec("0-7:blob").is_err());
assert!(parse_representation_spec("x-7:point").is_err());
}
#[test]
fn partitioning_total_equals_input() {
let geoms = synthetic_geometries();
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let opts = ConvertOptions {
mode: Mode::Partitioning,
levels: LevelPlan::ZoomRange {
min_zoom: 2,
max_zoom: 8,
},
..Default::default()
};
let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
let vr = validate_file(tout.path()).unwrap();
assert!(
vr.is_valid(),
"failures: {:?}",
vr.failures().collect::<Vec<_>>()
);
let reader = OverviewReader::open(tout.path()).unwrap();
assert_eq!(reader.mode(), Mode::Partitioning);
assert_eq!(report.total_rows, geoms.len());
let mut all_ids = Vec::new();
for level in 0..reader.num_levels() {
for (id, _, _, _) in read_level_rows(&reader, level) {
all_ids.push(id);
}
}
all_ids.sort();
assert_eq!(all_ids, (0..geoms.len() as i64).collect::<Vec<_>>());
for level in 0..reader.num_levels() {
for (id, _, _, geom) in read_level_rows(&reader, level) {
assert_eq!(geom, geoms[id as usize], "partitioning geometry verbatim");
}
}
}
#[test]
fn explicit_gsd_list_works() {
let geoms = synthetic_geometries();
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let opts = ConvertOptions {
mode: Mode::Duplicating,
levels: LevelPlan::Gsds(vec![gsd(3), gsd(6), gsd(9)]),
..Default::default()
};
let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
assert!(validate_file(tout.path()).unwrap().is_valid());
for w in report.levels.windows(2) {
assert!(w[0].gsd > w[1].gsd);
}
assert_eq!(report.levels.last().unwrap().feature_count, geoms.len());
}
#[test]
fn default_gsd_base_footer_gsds_match_const() {
use crate::overview::level::gsd;
let geoms = synthetic_geometries();
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let opts = ConvertOptions {
mode: Mode::Duplicating,
levels: LevelPlan::ZoomRange {
min_zoom: 2,
max_zoom: 8,
},
..Default::default()
};
assert_eq!(
opts.gsd_base, GSD_TILE_BASE,
"default must be the const base"
);
convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
let reader = OverviewReader::open(tout.path()).unwrap();
for level in &reader.meta().levels {
let z = level.zoom.expect("zoom-range plan records zooms");
assert_eq!(level.gsd, gsd(z), "footer gsd must match const gsd(z={z})");
}
assert_eq!(
reader.meta().generalization.as_ref().unwrap().gsd_base,
None,
"default gsd_base must be absent from provenance"
);
}
#[test]
fn nondefault_gsd_base_scales_footer_gsds() {
use crate::overview::level::{gsd_with_base, GSD_TILE_BASE};
let geoms = synthetic_geometries();
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let base = GSD_TILE_BASE * 2.0;
let opts = ConvertOptions {
mode: Mode::Duplicating,
levels: LevelPlan::ZoomRange {
min_zoom: 2,
max_zoom: 8,
},
gsd_base: base,
..Default::default()
};
convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
let reader = OverviewReader::open(tout.path()).unwrap();
assert!(validate_file(tout.path()).unwrap().is_valid());
for level in &reader.meta().levels {
let z = level.zoom.unwrap();
assert_eq!(
level.gsd,
gsd_with_base(z, base),
"scaled footer gsd (z={z})"
);
}
assert_eq!(
reader.meta().generalization.as_ref().unwrap().gsd_base,
Some(base),
"non-default gsd_base must be recorded"
);
}
#[test]
fn rejects_unsupported_crs() {
let geoms = synthetic_geometries();
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
let projjson = serde_json::json!({
"type": "ProjectedCRS",
"name": "UTM zone 33N",
"id": { "authority": "EPSG", "code": 32633 }
});
let md = geoarrow::datatypes::Metadata::new(
geoarrow::datatypes::Crs::from_projjson(projjson),
None,
);
write_input(tin.path(), &geoms, false, Some(md));
let opts = ConvertOptions::default();
let err = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap_err();
assert!(
matches!(err, ConvertError::UnsupportedCrs { .. }),
"expected UnsupportedCrs, got {err:?}"
);
}
#[test]
fn renames_existing_level_column() {
let geoms = synthetic_geometries();
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, true, None);
let opts = ConvertOptions::default();
convert_to_overviews(tin.path(), tout.path(), &opts)
.expect("colliding `level` column must be auto-renamed, not rejected");
let report = crate::overview::check::validate_file(tout.path()).unwrap();
assert!(
report.is_valid(),
"failures: {:?}",
report.failures().collect::<Vec<_>>()
);
let names = output_column_names(tout.path());
assert_eq!(
names
.iter()
.filter(|n| n.eq_ignore_ascii_case("level"))
.count(),
1,
"exactly one authoritative `level` column, names={names:?}"
);
assert!(
names.iter().any(|n| n == "level_"),
"renamed source column `level_` present, names={names:?}"
);
}
#[test]
fn resolver_renames_and_loops_suffix() {
let schema = Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("level", DataType::Int32, false),
Field::new("level_", DataType::Int32, false),
Field::new("geometry", DataType::Binary, false),
]));
let mut opts = ConvertOptions::default();
let (out, renames) = resolve_reserved_column_collisions(&schema, &mut opts);
assert_eq!(renames, vec![("level".to_string(), "level__".to_string())]);
let names: Vec<_> = out.fields().iter().map(|f| f.name().clone()).collect();
assert_eq!(names, vec!["id", "level__", "level_", "geometry"]);
}
#[test]
fn resolver_rewrites_option_column_references() {
let schema = Arc::new(Schema::new(vec![
Field::new("level", DataType::Int32, false),
Field::new("geometry", DataType::Binary, false),
]));
let mut opts = ConvertOptions {
sort_key: Some("LEVEL".to_string()),
..Default::default()
};
let (_out, renames) = resolve_reserved_column_collisions(&schema, &mut opts);
assert_eq!(renames.len(), 1);
assert_eq!(opts.sort_key.as_deref(), Some("level_"));
}
#[test]
fn resolver_rewrites_the_entry_zoom_column() {
use super::super::ladder::{EntryZoomKind, EntryZoomSpec};
let schema = Arc::new(Schema::new(vec![
Field::new("level", DataType::Float64, false),
Field::new("geometry", DataType::Binary, false),
]));
let mut opts = ConvertOptions {
entry_zoom: Some(EntryZoomSpec {
column: "level".to_string(),
kind: EntryZoomKind::DenseRank { step: 1 },
}),
..Default::default()
};
let (_out, renames) = resolve_reserved_column_collisions(&schema, &mut opts);
assert_eq!(renames.len(), 1);
assert_eq!(
opts.entry_zoom.as_ref().unwrap().column,
"level_",
"the ladder column must follow the reserved-column rename"
);
}
#[test]
fn resolver_noop_when_no_collision() {
let schema = Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("geometry", DataType::Binary, false),
]));
let mut opts = ConvertOptions::default();
let (out, renames) = resolve_reserved_column_collisions(&schema, &mut opts);
assert!(renames.is_empty());
assert!(Arc::ptr_eq(&out, &schema));
}
#[test]
fn resolver_reserves_count_columns_only_when_enabled() {
let schema = Arc::new(Schema::new(vec![
Field::new("point_count", DataType::Int64, false),
Field::new("geometry", DataType::Binary, false),
]));
let mut off = ConvertOptions {
cluster: false,
coalesce_lines: false,
..Default::default()
};
let (_out, renames) = resolve_reserved_column_collisions(&schema, &mut off);
assert!(
renames.is_empty(),
"point_count is a passthrough without --cluster"
);
let mut on = ConvertOptions {
cluster: true,
..Default::default()
};
let (_out, renames) = resolve_reserved_column_collisions(&schema, &mut on);
assert_eq!(
renames,
vec![("point_count".to_string(), "point_count_".to_string())]
);
}
fn write_class_input(path: &Path, geoms: &[Geometry<f64>], classes: &[Option<&str>]) {
let n = geoms.len();
assert_eq!(n, classes.len());
let id = Int64Array::from((0..n as i64).collect::<Vec<_>>());
let class = StringArray::from(classes.to_vec());
let geom_arr = build_geometry_array(geoms);
let geom_field = geom_arr.data_type().to_field("geometry", true);
let fields = vec![
Arc::new(Field::new("id", DataType::Int64, false)),
Arc::new(Field::new("road_class", DataType::Utf8, true)),
Arc::new(geom_field),
];
let columns: Vec<Arc<dyn Array>> =
vec![Arc::new(id), Arc::new(class), geom_arr.to_array_ref()];
let schema = Arc::new(Schema::new(fields));
let batch = RecordBatch::try_new(schema.clone(), columns).unwrap();
let gpq_options = GeoParquetWriterOptionsBuilder::default()
.set_encoding(GeoParquetWriterEncoding::WKB)
.set_generate_covering(true)
.build();
let encoder = GeoParquetRecordBatchEncoder::try_new(&schema, &gpq_options).unwrap();
let target_schema = encoder.target_schema();
let file = std::fs::File::create(path).unwrap();
let mut writer = ArrowWriter::try_new(file, target_schema, None).unwrap();
let mut encoder = encoder;
let encoded = encoder.encode_record_batch(&batch).unwrap();
writer.write(&encoded).unwrap();
writer.append_key_value_metadata(encoder.into_keyvalue().unwrap());
writer.close().unwrap();
}
fn min_level_by_id(reader: &OverviewReader) -> std::collections::HashMap<i64, usize> {
use arrow_array::cast::AsArray;
use arrow_array::types::Int64Type;
let mut out: std::collections::HashMap<i64, usize> = std::collections::HashMap::new();
for level in 0..reader.num_levels() {
let rdr = reader.read_level(level, None).unwrap();
for batch in rdr {
let batch = batch.unwrap();
let ids = batch
.column(batch.schema().index_of("id").unwrap())
.as_primitive::<Int64Type>()
.clone();
for i in 0..batch.num_rows() {
let id = ids.value(i);
out.entry(id)
.and_modify(|l| *l = (*l).min(level))
.or_insert(level);
}
}
}
out
}
#[test]
fn class_rank_maps_named_unknown_null() {
let col = StringArray::from(vec![
Some("motorway"),
Some("driveway"), None, ]);
let ranking = ClassRanking {
column: "road_class".to_string(),
ranks: vec![("motorway".to_string(), 5.0)],
unknown_rank: 0.0,
};
let keys = extract_class_ranks(&col, &ranking).unwrap();
assert_eq!(keys, vec![Some(5.0), Some(0.0), None]);
}
#[test]
fn class_rank_rejects_non_string_column() {
let col = Float64Array::from(vec![1.0, 2.0]);
let ranking = ClassRanking {
column: "road_class".to_string(),
ranks: vec![("x".to_string(), 1.0)],
unknown_rank: 0.0,
};
let err = extract_class_ranks(&col, &ranking).unwrap_err();
assert!(matches!(err, ConvertError::ClassRankColumnNotString { .. }));
}
#[test]
fn overture_road_ranking_spine_is_ordered() {
let cr = overture_road_ranking("road_class".to_string());
let rank = |v: &str| -> f64 {
cr.ranks
.iter()
.find(|(k, _)| k == v)
.map(|(_, r)| *r)
.unwrap()
};
let spine = [
"motorway",
"trunk",
"primary",
"secondary",
"tertiary",
"residential",
"unclassified",
"service",
];
for w in spine.windows(2) {
assert!(rank(w[0]) > rank(w[1]), "{} !> {}", w[0], w[1]);
}
for tail in ["living_street", "footway", "path", "cycleway", "track"] {
assert!(rank(tail) < rank("service"), "{tail} !< service");
assert!(rank(tail) > cr.unknown_rank, "{tail} !> unknown");
}
assert!(cr.ranks.iter().all(|(k, _)| k != "standard_gauge"));
}
#[test]
fn sort_key_and_class_ranking_conflict() {
let geoms = synthetic_geometries();
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let opts = ConvertOptions {
sort_key: Some("rank".to_string()),
class_ranking: Some(ClassRanking {
column: "road_class".to_string(),
ranks: vec![("motorway".to_string(), 1.0)],
unknown_rank: 0.0,
}),
..Default::default()
};
let err = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap_err();
assert!(matches!(err, ConvertError::RankingConflict));
}
fn overture_line_geoms() -> (Vec<Geometry<f64>>, Vec<Option<&'static str>>) {
let classes = [
"motorway",
"trunk",
"primary",
"residential",
"footway",
"service",
];
let mut geoms = Vec::new();
let mut cls = Vec::new();
for (i, c) in classes.iter().enumerate() {
let base = i as f64 * 2.0; let ls = LineString::from(vec![(base, 0.0), (base + 0.5, 0.3)]);
geoms.push(Geometry::LineString(ls));
cls.push(Some(*c));
}
(geoms, cls)
}
fn ranking_mode_of(path: &Path) -> String {
let reader = OverviewReader::open(path).unwrap();
reader
.meta()
.generalization
.as_ref()
.unwrap()
.ranking
.as_ref()
.unwrap()
.mode
.clone()
}
#[test]
fn auto_detect_overture_roads_triggers() {
let (geoms, classes) = overture_line_geoms();
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_class_input(tin.path(), &geoms, &classes);
let opts = ConvertOptions {
levels: LevelPlan::ZoomRange {
min_zoom: 4,
max_zoom: 10,
},
..Default::default()
};
convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
assert_eq!(ranking_mode_of(tout.path()), "auto-overture-roads");
}
#[test]
fn auto_detect_disabled_falls_back_to_size() {
let (geoms, classes) = overture_line_geoms();
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_class_input(tin.path(), &geoms, &classes);
let opts = ConvertOptions {
levels: LevelPlan::ZoomRange {
min_zoom: 4,
max_zoom: 10,
},
no_auto_rank: true,
..Default::default()
};
convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
assert_eq!(ranking_mode_of(tout.path()), "size-fallback");
}
#[test]
fn auto_detect_non_trigger_without_class_column() {
let geoms = synthetic_geometries();
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let opts = ConvertOptions::default();
convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
assert_eq!(ranking_mode_of(tout.path()), "size-fallback");
}
#[test]
fn class_rank_provenance_recorded() {
let (geoms, classes) = overture_line_geoms();
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_class_input(tin.path(), &geoms, &classes);
let opts = ConvertOptions {
levels: LevelPlan::ZoomRange {
min_zoom: 4,
max_zoom: 10,
},
class_ranking: Some(ClassRanking {
column: "road_class".to_string(),
ranks: vec![("motorway".to_string(), 5.0), ("footway".to_string(), 1.0)],
unknown_rank: 0.0,
}),
..Default::default()
};
convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
let reader = OverviewReader::open(tout.path()).unwrap();
let r = reader
.meta()
.generalization
.as_ref()
.unwrap()
.ranking
.clone()
.unwrap();
assert_eq!(r.mode, "class-ranking");
assert_eq!(r.column.as_deref(), Some("road_class"));
let ranks = r.ranks.unwrap();
assert_eq!(ranks.len(), 2);
assert_eq!(ranks.get("motorway"), Some(&5.0));
assert_eq!(ranks.get("footway"), Some(&1.0));
assert_eq!(r.unknown_rank, Some(0.0));
let json = overviews_footer_json(tout.path());
assert!(
json.contains(r#""ranks":{"footway":1.0,"motorway":5.0}"#),
"footer must serialize ranks as an object map, got {json}"
);
}
#[test]
fn high_class_small_feature_wins_coarse_cell() {
let big_low = Geometry::LineString(LineString::from(vec![
(-0.03, -0.04),
(0.07, 0.06), ]));
let small_high = Geometry::LineString(LineString::from(vec![
(0.0, 0.0),
(0.04, 0.02), ]));
let geoms = vec![big_low, small_high];
let classes = vec![Some("footway"), Some("motorway")];
let tin = tempfile::NamedTempFile::new().unwrap();
{
let tout = tempfile::NamedTempFile::new().unwrap();
write_class_input(tin.path(), &geoms, &classes);
let opts = ConvertOptions {
levels: LevelPlan::ZoomRange {
min_zoom: 4,
max_zoom: 10,
},
no_auto_rank: true, ..Default::default()
};
convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
let reader = OverviewReader::open(tout.path()).unwrap();
let ml = min_level_by_id(&reader);
assert!(
ml[&0] < ml[&1],
"size fallback: big low-class (id0) should win coarse, got {ml:?}"
);
}
{
let tout = tempfile::NamedTempFile::new().unwrap();
write_class_input(tin.path(), &geoms, &classes);
let opts = ConvertOptions {
levels: LevelPlan::ZoomRange {
min_zoom: 4,
max_zoom: 10,
},
class_ranking: Some(ClassRanking {
column: "road_class".to_string(),
ranks: vec![("motorway".to_string(), 5.0), ("footway".to_string(), 1.0)],
unknown_rank: 0.0,
}),
..Default::default()
};
convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
let reader = OverviewReader::open(tout.path()).unwrap();
let ml = min_level_by_id(&reader);
assert!(
ml[&1] < ml[&0],
"class ranking: small high-class (id1) must win coarse, got {ml:?}"
);
assert_eq!(ml[&1], 0, "high-class line should reach the coarsest level");
}
}
fn density_provenance_of(path: &Path) -> Option<crate::overview::level::DensityProvenance> {
let reader = OverviewReader::open(path).unwrap();
reader
.meta()
.generalization
.as_ref()
.unwrap()
.density_drop
.clone()
}
fn grid_points(n: usize) -> Vec<Geometry<f64>> {
(0..n)
.map(|i| {
let x = (i % 40) as f64 * 0.3 - 6.0;
let y = (i / 40) as f64 * 0.3 - 6.0;
Geometry::Point(Point::new(x, y))
})
.collect()
}
#[test]
fn density_provenance_recorded_by_default() {
let geoms = synthetic_geometries();
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let opts = ConvertOptions {
levels: LevelPlan::ZoomRange {
min_zoom: 2,
max_zoom: 8,
},
..Default::default()
};
convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
let d = density_provenance_of(tout.path()).expect("default run records density_drop");
assert_eq!(d.drop_rate, DensityBudgetConfig::default().drop_rate);
assert_eq!(d.gamma, DensityBudgetConfig::default().gamma);
assert_eq!(d.supercell_gsd_factor, SUPERCELL_GSD_FACTOR);
}
#[test]
fn density_disabled_omits_provenance_and_keeps_canonical() {
let geoms = grid_points(600);
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let opts = ConvertOptions {
levels: LevelPlan::ZoomRange {
min_zoom: 0,
max_zoom: 8,
},
density: DensityBudgetConfig {
enabled: false,
..DensityBudgetConfig::default()
},
..Default::default()
};
let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
assert!(
density_provenance_of(tout.path()).is_none(),
"disabled budget must not emit provenance"
);
assert!(validate_file(tout.path()).unwrap().is_valid());
assert_eq!(report.levels.last().unwrap().feature_count, geoms.len());
}
#[test]
fn density_thins_midlevels_keeps_canonical() {
let geoms = grid_points(600);
let tin = tempfile::NamedTempFile::new().unwrap();
let tout_on = tempfile::NamedTempFile::new().unwrap();
let tout_off = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let base = ConvertOptions {
levels: LevelPlan::ZoomRange {
min_zoom: 0,
max_zoom: 8,
},
..Default::default()
};
let on = convert_to_overviews(tin.path(), tout_on.path(), &base).unwrap();
let off = convert_to_overviews(
tin.path(),
tout_off.path(),
&ConvertOptions {
density: DensityBudgetConfig {
enabled: false,
..DensityBudgetConfig::default()
},
..base.clone()
},
)
.unwrap();
assert_eq!(on.levels.last().unwrap().feature_count, geoms.len());
assert_eq!(off.levels.last().unwrap().feature_count, geoms.len());
assert!(validate_file(tout_on.path()).unwrap().is_valid());
assert!(
on.total_rows < off.total_rows,
"density budget should thin mid levels: on={} off={}",
on.total_rows,
off.total_rows
);
for w in on.levels.windows(2) {
assert!(w[0].feature_count <= w[1].feature_count);
}
}
fn overviews_footer_json(path: &Path) -> String {
let file = std::fs::File::open(path).unwrap();
let b = ParquetRecordBatchReaderBuilder::try_new(file).unwrap();
b.metadata()
.file_metadata()
.key_value_metadata()
.unwrap()
.iter()
.find(|kv| kv.key == crate::overview::level::OVERVIEWS_KEY)
.expect("geo:overviews key")
.value
.clone()
.unwrap()
}
fn assert_streaming_equivalent(input: &Path, base: &ConvertOptions) {
let mem_out = tempfile::NamedTempFile::new().unwrap();
let stream_out = tempfile::NamedTempFile::new().unwrap();
let mem_opts = ConvertOptions {
streaming: false,
..base.clone()
};
let stream_opts = ConvertOptions {
streaming: true,
..base.clone()
};
let mem = convert_to_overviews(input, mem_out.path(), &mem_opts).unwrap();
let strm = convert_to_overviews(input, stream_out.path(), &stream_opts).unwrap();
assert_eq!(mem.mode, strm.mode);
assert_eq!(mem.input_features, strm.input_features);
assert_eq!(mem.total_rows, strm.total_rows);
assert_eq!(mem.total_vertices, strm.total_vertices);
assert_eq!(mem.levels.len(), strm.levels.len());
for (a, b) in mem.levels.iter().zip(&strm.levels) {
assert_eq!(a.level, b.level);
assert_eq!(a.gsd, b.gsd, "level {} gsd", a.level);
assert_eq!(a.zoom, b.zoom, "level {} zoom", a.level);
assert_eq!(
a.feature_count, b.feature_count,
"level {} feature count",
a.level
);
assert_eq!(
a.vertex_count, b.vertex_count,
"level {} vertex count",
a.level
);
}
assert_eq!(
overviews_footer_json(mem_out.path()),
overviews_footer_json(stream_out.path()),
"geo:overviews footers differ"
);
assert!(validate_file(mem_out.path()).unwrap().is_valid());
assert!(validate_file(stream_out.path()).unwrap().is_valid());
let mr = OverviewReader::open(mem_out.path()).unwrap();
let sr = OverviewReader::open(stream_out.path()).unwrap();
assert_eq!(mr.num_levels(), sr.num_levels());
for level in 0..mr.num_levels() {
assert_eq!(
read_level_rows(&mr, level),
read_level_rows(&sr, level),
"level {level} rows differ"
);
}
}
fn assert_outputs_equivalent(a: &Path, b: &Path, ctx: &str) {
use parquet::file::reader::{FileReader, SerializedFileReader};
let layout = |p: &Path| -> Vec<(i64, usize)> {
let r = SerializedFileReader::new(std::fs::File::open(p).unwrap()).unwrap();
let md = r.metadata();
(0..md.num_row_groups())
.map(|i| (md.row_group(i).num_rows(), md.row_group(i).columns().len()))
.collect()
};
assert_eq!(layout(a), layout(b), "{ctx}: row-group layout differs");
assert_eq!(
overviews_footer_json(a),
overviews_footer_json(b),
"{ctx}: geo:overviews footer differs"
);
let ra = OverviewReader::open(a).unwrap();
let rb = OverviewReader::open(b).unwrap();
assert_eq!(
ra.num_levels(),
rb.num_levels(),
"{ctx}: level count differs"
);
for level in 0..ra.num_levels() {
assert_eq!(
read_level_rows(&ra, level),
read_level_rows(&rb, level),
"{ctx}: level {level} rows differ"
);
}
}
#[test]
fn pipelined_matches_serial() {
use super::super::level::MemoryProfile;
use super::super::stream::Pass2Strategy;
let poly_in = tempfile::NamedTempFile::new().unwrap();
write_input(poly_in.path(), &synthetic_geometries(), false, None);
let grid_in = tempfile::NamedTempFile::new().unwrap();
write_input(grid_in.path(), &grid_points(600), false, None);
let cases: Vec<(&str, &Path, ConvertOptions)> = vec![
(
"duplicating",
poly_in.path(),
ConvertOptions {
mode: Mode::Duplicating,
levels: LevelPlan::ZoomRange {
min_zoom: 1,
max_zoom: 10,
},
..Default::default()
},
),
(
"partitioning",
grid_in.path(),
ConvertOptions {
mode: Mode::Partitioning,
levels: LevelPlan::ZoomRange {
min_zoom: 1,
max_zoom: 9,
},
..Default::default()
},
),
(
"clustering",
grid_in.path(),
ConvertOptions {
mode: Mode::Duplicating,
cluster: true,
levels: LevelPlan::ZoomRange {
min_zoom: 1,
max_zoom: 9,
},
..Default::default()
},
),
];
for (name, input, base) in &cases {
let serial_out = tempfile::NamedTempFile::new().unwrap();
convert_to_overviews_strategy(input, serial_out.path(), base, Pass2Strategy::Serial)
.unwrap();
for profile in [
MemoryProfile::Speed,
MemoryProfile::Bounded,
MemoryProfile::Auto,
] {
let opts = ConvertOptions {
profile,
..base.clone()
};
let piped_out = tempfile::NamedTempFile::new().unwrap();
convert_to_overviews_strategy(
input,
piped_out.path(),
&opts,
Pass2Strategy::Pipelined,
)
.unwrap();
assert_outputs_equivalent(
serial_out.path(),
piped_out.path(),
&format!("{name}/{profile:?}"),
);
}
}
}
#[test]
fn pipelined_invariant_to_batching_knobs() {
use super::super::stream::Pass2Strategy;
let geoms = grid_points(600);
let tin = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let base = ConvertOptions {
mode: Mode::Duplicating,
levels: LevelPlan::ZoomRange {
min_zoom: 1,
max_zoom: 9,
},
..Default::default()
};
let convert = |read_batch_size: usize, in_flight_batches: usize| {
let opts = ConvertOptions {
read_batch_size,
in_flight_batches,
..base.clone()
};
let out = tempfile::NamedTempFile::new().unwrap();
convert_to_overviews_strategy(tin.path(), out.path(), &opts, Pass2Strategy::Pipelined)
.unwrap();
out
};
let reference = convert(7, 1);
for (rbs, ifb) in [(64usize, 4usize), (4096, 8), (8192, 0)] {
let candidate = convert(rbs, ifb);
assert_outputs_equivalent(
reference.path(),
candidate.path(),
&format!("read_batch_size={rbs} in_flight_batches={ifb}"),
);
}
}
#[test]
fn resolve_in_flight_batches_auto_and_explicit() {
let auto = resolve_in_flight_batches(IN_FLIGHT_BATCHES_AUTO);
assert!(
(IN_FLIGHT_BATCHES_MIN..=IN_FLIGHT_BATCHES_MAX).contains(&auto),
"auto in-flight {auto} outside [{IN_FLIGHT_BATCHES_MIN}, {IN_FLIGHT_BATCHES_MAX}]"
);
let cores = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(IN_FLIGHT_BATCHES_MIN);
assert_eq!(
auto,
cores.clamp(IN_FLIGHT_BATCHES_MIN, IN_FLIGHT_BATCHES_MAX),
"auto must equal core count clamped to the configured range"
);
assert_eq!(resolve_in_flight_batches(1), 1);
assert_eq!(resolve_in_flight_batches(7), 7);
assert_eq!(
resolve_in_flight_batches(IN_FLIGHT_BATCHES_MAX + 100),
IN_FLIGHT_BATCHES_MAX + 100
);
}
#[test]
fn streaming_matches_in_memory_duplicating() {
let geoms = synthetic_geometries();
let tin = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let base = ConvertOptions {
mode: Mode::Duplicating,
levels: LevelPlan::ZoomRange {
min_zoom: 1,
max_zoom: 10,
},
..Default::default()
};
assert_streaming_equivalent(tin.path(), &base);
}
#[test]
fn streaming_matches_in_memory_partitioning() {
let geoms = synthetic_geometries();
let tin = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let base = ConvertOptions {
mode: Mode::Partitioning,
levels: LevelPlan::ZoomRange {
min_zoom: 2,
max_zoom: 8,
},
..Default::default()
};
assert_streaming_equivalent(tin.path(), &base);
}
#[test]
fn streaming_matches_with_small_read_batches_and_density_budget() {
let geoms = grid_points(600);
let tin = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let base = ConvertOptions {
levels: LevelPlan::ZoomRange {
min_zoom: 0,
max_zoom: 8,
},
read_batch_size: 7,
..Default::default()
};
assert_streaming_equivalent(tin.path(), &base);
}
#[test]
fn streaming_matches_with_explicit_sort_key() {
let geoms = synthetic_geometries();
let tin = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let base = ConvertOptions {
levels: LevelPlan::ZoomRange {
min_zoom: 2,
max_zoom: 8,
},
sort_key: Some("rank".to_string()),
..Default::default()
};
assert_streaming_equivalent(tin.path(), &base);
}
#[test]
fn streaming_auto_rank_matches_in_memory() {
let (geoms, classes) = overture_line_geoms();
let tin = tempfile::NamedTempFile::new().unwrap();
write_class_input(tin.path(), &geoms, &classes);
let base = ConvertOptions {
levels: LevelPlan::ZoomRange {
min_zoom: 4,
max_zoom: 10,
},
read_batch_size: 2,
..Default::default()
};
let mem_out = tempfile::NamedTempFile::new().unwrap();
let stream_out = tempfile::NamedTempFile::new().unwrap();
convert_to_overviews(
tin.path(),
mem_out.path(),
&ConvertOptions {
streaming: false,
..base.clone()
},
)
.unwrap();
convert_to_overviews(tin.path(), stream_out.path(), &base).unwrap();
assert_eq!(ranking_mode_of(mem_out.path()), "auto-overture-roads");
assert_eq!(ranking_mode_of(stream_out.path()), "auto-overture-roads");
assert_eq!(
overviews_footer_json(mem_out.path()),
overviews_footer_json(stream_out.path())
);
let mr = OverviewReader::open(mem_out.path()).unwrap();
let sr = OverviewReader::open(stream_out.path()).unwrap();
assert_eq!(min_level_by_id(&mr), min_level_by_id(&sr));
}
fn read_point_counts(reader: &OverviewReader, level: usize) -> Vec<i64> {
use arrow_array::cast::AsArray;
use arrow_array::types::Int64Type;
let rdr = reader.read_level(level, None).unwrap();
let mut out = Vec::new();
for batch in rdr {
let batch = batch.unwrap();
let idx = batch.schema().index_of("point_count").unwrap();
let col = batch.column(idx).as_primitive::<Int64Type>().clone();
assert_eq!(col.null_count(), 0, "point_count must be NOT NULL");
out.extend(col.values().iter().copied());
}
out
}
fn read_f64_column(reader: &OverviewReader, level: usize, name: &str) -> Vec<Option<f64>> {
use arrow_array::cast::AsArray;
use arrow_array::types::Float64Type;
let rdr = reader.read_level(level, None).unwrap();
let mut out = Vec::new();
for batch in rdr {
let batch = batch.unwrap();
let idx = batch.schema().index_of(name).unwrap();
let col = batch.column(idx).as_primitive::<Float64Type>().clone();
for i in 0..col.len() {
out.push(if col.is_null(i) {
None
} else {
Some(col.value(i))
});
}
}
out
}
#[test]
fn cluster_duplicating_end_to_end_counts_partition_every_level() {
let geoms = grid_points(600);
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let opts = ConvertOptions {
levels: LevelPlan::ZoomRange {
min_zoom: 0,
max_zoom: 8,
},
cluster: true,
..Default::default()
};
let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
let vr = validate_file(tout.path()).unwrap();
assert!(
vr.is_valid(),
"failures: {:?}",
vr.failures().collect::<Vec<_>>()
);
let reader = OverviewReader::open(tout.path()).unwrap();
let canonical = reader.num_levels() - 1;
for level in 0..reader.num_levels() {
let counts = read_point_counts(&reader, level);
assert!(counts.iter().all(|&c| c >= 1), "level {level} counts >= 1");
assert_eq!(
counts.iter().sum::<i64>(),
geoms.len() as i64,
"level {level}: sum(point_count) must equal source count"
);
}
assert!(read_point_counts(&reader, canonical)
.iter()
.all(|&c| c == 1));
let coarse = read_point_counts(&reader, 0);
assert!(coarse.iter().any(|&c| c > 1), "no clustering happened");
assert!(coarse.len() < geoms.len());
assert_eq!(report.levels.last().unwrap().feature_count, geoms.len());
}
#[test]
fn cluster_sum_invariant_across_knob_combinations() {
let geoms = grid_points(300);
let tin = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
for streaming in [true, false] {
for density in [true, false] {
for thinning in [2.0, 8.0] {
let tout = tempfile::NamedTempFile::new().unwrap();
let mut opts = ConvertOptions {
levels: LevelPlan::ZoomRange {
min_zoom: 0,
max_zoom: 6,
},
cluster: true,
streaming,
..Default::default()
};
opts.density.enabled = density;
opts.assign.point_thinning = thinning;
let label =
format!("streaming={streaming} density={density} thinning={thinning}");
convert_to_overviews(tin.path(), tout.path(), &opts)
.unwrap_or_else(|e| panic!("{label}: {e}"));
let vr = validate_file(tout.path()).unwrap();
assert!(
vr.is_valid(),
"{label}: failures: {:?}",
vr.failures().collect::<Vec<_>>()
);
assert_eq!(
vr.check_passed("cluster_sum_invariant"),
Some(true),
"{label}"
);
let reader = OverviewReader::open(tout.path()).unwrap();
let canonical = reader.num_levels() - 1;
for level in 0..reader.num_levels() {
let counts = read_point_counts(&reader, level);
assert!(
!counts.is_empty(),
"{label}: level {level} thinned points to zero"
);
assert_eq!(
counts.iter().sum::<i64>(),
geoms.len() as i64,
"{label}: level {level} must partition the source set"
);
}
assert!(
read_point_counts(&reader, canonical)
.iter()
.all(|&c| c == 1),
"{label}: canonical band must be singleton-only"
);
}
}
}
}
#[test]
fn cluster_accumulate_sum_and_mean_consistent() {
let geoms = grid_points(400);
let n = geoms.len();
let tin = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let source_sum: f64 = (1..=n).map(|v| v as f64).sum();
let tout = tempfile::NamedTempFile::new().unwrap();
let opts = ConvertOptions {
levels: LevelPlan::ZoomRange {
min_zoom: 0,
max_zoom: 8,
},
cluster: true,
accumulate: vec![AccumulateSpec {
column: "rank".to_string(),
op: super::super::cluster::AccumulateOp::Sum,
}],
..Default::default()
};
convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
let reader = OverviewReader::open(tout.path()).unwrap();
for level in 0..reader.num_levels() {
let ranks = read_f64_column(&reader, level, "rank");
let total: f64 = ranks.iter().flatten().sum();
assert!(
(total - source_sum).abs() < 1e-6,
"level {level}: rank sum {total} != source {source_sum}"
);
}
let canonical = reader.num_levels() - 1;
let rows = read_level_rows(&reader, canonical);
for (id, _, rank, _) in rows {
assert_eq!(rank, (n as i64 - id) as f64, "canonical rank verbatim");
}
let tout2 = tempfile::NamedTempFile::new().unwrap();
let opts_mean = ConvertOptions {
accumulate: vec![AccumulateSpec {
column: "rank".to_string(),
op: super::super::cluster::AccumulateOp::Mean,
}],
..opts.clone()
};
convert_to_overviews(tin.path(), tout2.path(), &opts_mean).unwrap();
let reader2 = OverviewReader::open(tout2.path()).unwrap();
for level in 0..reader2.num_levels() {
let means = read_f64_column(&reader2, level, "rank");
let counts = read_point_counts(&reader2, level);
let total: f64 = means
.iter()
.zip(&counts)
.map(|(m, &c)| m.unwrap() * c as f64)
.sum();
assert!(
(total - source_sum).abs() < 1e-6,
"level {level}: Σ mean×count {total} != source {source_sum}"
);
}
}
#[test]
fn cluster_footer_provenance_recorded() {
let geoms = grid_points(100);
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let opts = ConvertOptions {
cluster: true,
accumulate: vec![AccumulateSpec {
column: "rank".to_string(),
op: super::super::cluster::AccumulateOp::Mean,
}],
..Default::default()
};
convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
let reader = OverviewReader::open(tout.path()).unwrap();
let c = reader
.meta()
.generalization
.as_ref()
.unwrap()
.clustering
.clone()
.expect("clustering provenance recorded");
assert!(c.enabled);
assert_eq!(c.point_count_column, "point_count");
assert_eq!(c.accumulated.len(), 1);
assert_eq!(c.accumulated[0].column, "rank");
assert_eq!(c.accumulated[0].op, "mean");
let tout_off = tempfile::NamedTempFile::new().unwrap();
convert_to_overviews(tin.path(), tout_off.path(), &ConvertOptions::default()).unwrap();
let r_off = OverviewReader::open(tout_off.path()).unwrap();
assert!(r_off
.meta()
.generalization
.as_ref()
.unwrap()
.clustering
.is_none());
}
#[test]
fn cluster_option_errors() {
let geoms = grid_points(20);
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let err = convert_to_overviews(
tin.path(),
tout.path(),
&ConvertOptions {
mode: Mode::Partitioning,
cluster: true,
..Default::default()
},
)
.unwrap_err();
assert!(matches!(err, ConvertError::ClusterPartitioningUnsupported));
let err = convert_to_overviews(
tin.path(),
tout.path(),
&ConvertOptions {
accumulate: vec![AccumulateSpec {
column: "rank".to_string(),
op: super::super::cluster::AccumulateOp::Sum,
}],
..Default::default()
},
)
.unwrap_err();
assert!(matches!(err, ConvertError::AccumulateWithoutCluster));
for streaming in [true, false] {
let err = convert_to_overviews(
tin.path(),
tout.path(),
&ConvertOptions {
cluster: true,
streaming,
accumulate: vec![AccumulateSpec {
column: "nonexistent".to_string(),
op: super::super::cluster::AccumulateOp::Sum,
}],
..Default::default()
},
)
.unwrap_err();
assert!(
matches!(err, ConvertError::AccumulateColumnMissing { .. }),
"streaming={streaming}: got {err:?}"
);
}
let err = convert_to_overviews(
tin.path(),
tout.path(),
&ConvertOptions {
cluster: true,
accumulate: vec![AccumulateSpec {
column: "name".to_string(),
op: super::super::cluster::AccumulateOp::Max,
}],
..Default::default()
},
)
.unwrap_err();
assert!(matches!(
err,
ConvertError::AccumulateColumnNotNumeric { .. }
));
}
#[test]
fn cluster_renames_existing_point_count_column() {
let geoms = grid_points(10);
let n = geoms.len();
let tin = tempfile::NamedTempFile::new().unwrap();
{
let id = Int64Array::from((0..n as i64).collect::<Vec<_>>());
let pc = Int64Array::from(vec![7i64; n]);
let geom_arr = build_geometry_array(&geoms);
let geom_field = geom_arr.data_type().to_field("geometry", true);
let fields = vec![
Arc::new(Field::new("id", DataType::Int64, false)),
Arc::new(Field::new("Point_Count", DataType::Int64, false)),
Arc::new(geom_field),
];
let columns: Vec<Arc<dyn Array>> =
vec![Arc::new(id), Arc::new(pc), geom_arr.to_array_ref()];
let schema = Arc::new(Schema::new(fields));
let batch = RecordBatch::try_new(schema.clone(), columns).unwrap();
let gpq_options = GeoParquetWriterOptionsBuilder::default()
.set_encoding(GeoParquetWriterEncoding::WKB)
.set_generate_covering(true)
.build();
let mut encoder = GeoParquetRecordBatchEncoder::try_new(&schema, &gpq_options).unwrap();
let target_schema = encoder.target_schema();
let file = std::fs::File::create(tin.path()).unwrap();
let mut writer = ArrowWriter::try_new(file, target_schema, None).unwrap();
writer
.write(&encoder.encode_record_batch(&batch).unwrap())
.unwrap();
writer.append_key_value_metadata(encoder.into_keyvalue().unwrap());
writer.close().unwrap();
}
let tout = tempfile::NamedTempFile::new().unwrap();
convert_to_overviews(
tin.path(),
tout.path(),
&ConvertOptions {
cluster: true,
..Default::default()
},
)
.expect("colliding `Point_Count` column must be auto-renamed, not rejected");
let names = output_column_names(tout.path());
assert!(
names.iter().any(|n| n == "Point_Count_"),
"renamed source column present, names={names:?}"
);
assert_eq!(
names
.iter()
.filter(|n| n.eq_ignore_ascii_case("point_count"))
.count(),
1,
"one authoritative `point_count`, names={names:?}"
);
convert_to_overviews(tin.path(), tout.path(), &ConvertOptions::default()).unwrap();
let names = output_column_names(tout.path());
assert!(
names.iter().any(|n| n == "Point_Count"),
"passthrough column kept verbatim, names={names:?}"
);
}
#[test]
fn streaming_matches_in_memory_clustering() {
let geoms = grid_points(600);
let tin = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let base = ConvertOptions {
levels: LevelPlan::ZoomRange {
min_zoom: 0,
max_zoom: 8,
},
read_batch_size: 7,
cluster: true,
accumulate: vec![
AccumulateSpec {
column: "rank".to_string(),
op: super::super::cluster::AccumulateOp::Sum,
},
AccumulateSpec {
column: "rank".to_string(),
op: super::super::cluster::AccumulateOp::Mean,
},
],
..Default::default()
};
assert_streaming_equivalent(tin.path(), &base);
let mem_out = tempfile::NamedTempFile::new().unwrap();
let stream_out = tempfile::NamedTempFile::new().unwrap();
convert_to_overviews(
tin.path(),
mem_out.path(),
&ConvertOptions {
streaming: false,
..base.clone()
},
)
.unwrap();
convert_to_overviews(tin.path(), stream_out.path(), &base).unwrap();
let mr = OverviewReader::open(mem_out.path()).unwrap();
let sr = OverviewReader::open(stream_out.path()).unwrap();
for level in 0..mr.num_levels() {
assert_eq!(
read_point_counts(&mr, level),
read_point_counts(&sr, level),
"level {level} point_count differs"
);
}
}
fn read_coalesced_counts(reader: &OverviewReader, level: usize) -> Vec<i32> {
use arrow_array::cast::AsArray;
use arrow_array::types::Int32Type;
let rdr = reader.read_level(level, None).unwrap();
let mut out = Vec::new();
for batch in rdr {
let batch = batch.unwrap();
let idx = batch.schema().index_of("coalesced_count").unwrap();
let col = batch.column(idx).as_primitive::<Int32Type>().clone();
assert_eq!(col.null_count(), 0, "coalesced_count must be NOT NULL");
out.extend(col.values().iter().copied());
}
out
}
fn fragment_chain_geoms(n: usize) -> Vec<Geometry<f64>> {
let mut geoms: Vec<Geometry<f64>> = (0..n)
.map(|i| {
let x0 = i as f64 * 0.01;
Geometry::LineString(LineString::from(vec![(x0, 0.0), (x0 + 0.01, 0.0)]))
})
.collect();
geoms.push(Geometry::Point(Point::new(5.0, 5.0)));
geoms
}
#[test]
fn coalesce_reclaims_sub_visibility_fragments_and_keeps_canonical() {
let geoms = fragment_chain_geoms(6);
let tin = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let opts = ConvertOptions {
levels: LevelPlan::ZoomRange {
min_zoom: 4,
max_zoom: 10,
},
no_auto_rank: true,
..Default::default() };
let tout_off = tempfile::NamedTempFile::new().unwrap();
let off = convert_to_overviews(
tin.path(),
tout_off.path(),
&ConvertOptions {
coalesce_lines: false,
..opts.clone()
},
)
.unwrap();
assert_eq!(
off.levels[0].feature_count, 1,
"without coalescing only the point survives level 0: {:?}",
off.levels
);
let tout = tempfile::NamedTempFile::new().unwrap();
let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
assert_eq!(
report.levels[0].feature_count, 2,
"chain + point at level 0: {:?}",
report.levels
);
let vr = validate_file(tout.path()).unwrap();
assert!(
vr.is_valid(),
"failures: {:?}",
vr.failures().collect::<Vec<_>>()
);
let reader = OverviewReader::open(tout.path()).unwrap();
let counts0 = read_coalesced_counts(&reader, 0);
let mut sorted = counts0.clone();
sorted.sort_unstable();
assert_eq!(sorted, vec![1, 6], "point=1, merged chain=6: {counts0:?}");
let canonical = reader.num_levels() - 1;
let rows = read_level_rows(&reader, canonical);
assert_eq!(rows.len(), geoms.len());
for (id, _, _, geom) in &rows {
assert_eq!(
geom, &geoms[*id as usize],
"canonical geometry verbatim (never coalesced)"
);
}
assert!(read_coalesced_counts(&reader, canonical)
.iter()
.all(|&c| c == 1));
}
#[test]
fn coalesce_groups_by_auto_detected_class() {
let mut geoms = vec![
Geometry::LineString(LineString::from(vec![(0.0, 0.0), (0.1, 0.0)])),
Geometry::LineString(LineString::from(vec![(0.1, 0.0), (0.2, 0.0)])),
Geometry::LineString(LineString::from(vec![(0.2, 0.0), (0.2, 0.1)])),
];
let mut classes = vec![Some("motorway"), Some("motorway"), Some("footway")];
for (i, c) in ["primary", "service", "residential"].iter().enumerate() {
geoms.push(Geometry::LineString(LineString::from(vec![
(3.0 + i as f64, 3.0),
(3.1 + i as f64, 3.05),
])));
classes.push(Some(*c));
}
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_class_input(tin.path(), &geoms, &classes);
let opts = ConvertOptions {
levels: LevelPlan::ZoomRange {
min_zoom: 4,
max_zoom: 10,
},
coalesce_lines: true,
..Default::default()
};
convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
assert_eq!(ranking_mode_of(tout.path()), "auto-overture-roads");
let reader = OverviewReader::open(tout.path()).unwrap();
let counts0 = read_coalesced_counts(&reader, 0);
assert_eq!(
counts0.iter().filter(|&&c| c == 2).count(),
1,
"exactly one 2-segment motorway chain: {counts0:?}"
);
assert!(
counts0.iter().all(|&c| c <= 2),
"footway never merges into the motorway chain: {counts0:?}"
);
}
#[test]
fn coalesce_inert_in_partitioning() {
let geoms = fragment_chain_geoms(3);
let tin = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
for streaming in [true, false] {
let tout = tempfile::NamedTempFile::new().unwrap();
let report = convert_to_overviews(
tin.path(),
tout.path(),
&ConvertOptions {
mode: Mode::Partitioning,
coalesce_lines: true, streaming,
..Default::default()
},
)
.unwrap();
assert_eq!(report.total_rows, geoms.len(), "streaming={streaming}");
let reader = OverviewReader::open(tout.path()).unwrap();
assert!(
reader
.meta()
.generalization
.as_ref()
.unwrap()
.coalescing
.is_none(),
"no coalescing provenance in partitioning mode"
);
let batch_schema = reader.read_level(0, None).unwrap().next().unwrap().unwrap();
assert!(
batch_schema.schema().index_of("coalesced_count").is_err(),
"no coalesced_count column in partitioning mode"
);
assert!(validate_file(tout.path()).unwrap().is_valid());
}
}
#[test]
fn coalesce_renames_existing_coalesced_count_column() {
let geoms = fragment_chain_geoms(2);
let n = geoms.len();
let tin = tempfile::NamedTempFile::new().unwrap();
{
let id = Int64Array::from((0..n as i64).collect::<Vec<_>>());
let cc = arrow_array::Int32Array::from(vec![7i32; n]);
let geom_arr = build_geometry_array(&geoms);
let geom_field = geom_arr.data_type().to_field("geometry", true);
let fields = vec![
Arc::new(Field::new("id", DataType::Int64, false)),
Arc::new(Field::new("Coalesced_Count", DataType::Int32, false)),
Arc::new(geom_field),
];
let columns: Vec<Arc<dyn Array>> =
vec![Arc::new(id), Arc::new(cc), geom_arr.to_array_ref()];
let schema = Arc::new(Schema::new(fields));
let batch = RecordBatch::try_new(schema.clone(), columns).unwrap();
let gpq_options = GeoParquetWriterOptionsBuilder::default()
.set_encoding(GeoParquetWriterEncoding::WKB)
.set_generate_covering(true)
.build();
let mut encoder = GeoParquetRecordBatchEncoder::try_new(&schema, &gpq_options).unwrap();
let target_schema = encoder.target_schema();
let file = std::fs::File::create(tin.path()).unwrap();
let mut writer = ArrowWriter::try_new(file, target_schema, None).unwrap();
writer
.write(&encoder.encode_record_batch(&batch).unwrap())
.unwrap();
writer.append_key_value_metadata(encoder.into_keyvalue().unwrap());
writer.close().unwrap();
}
let tout = tempfile::NamedTempFile::new().unwrap();
for streaming in [true, false] {
convert_to_overviews(
tin.path(),
tout.path(),
&ConvertOptions {
coalesce_lines: true,
streaming,
..Default::default()
},
)
.unwrap_or_else(|e| {
panic!("streaming={streaming}: `Coalesced_Count` must be auto-renamed, got {e}")
});
let names = output_column_names(tout.path());
assert!(
names.iter().any(|n| n == "Coalesced_Count_"),
"streaming={streaming}: renamed source column present, names={names:?}"
);
}
convert_to_overviews(
tin.path(),
tout.path(),
&ConvertOptions {
coalesce_lines: false,
..Default::default()
},
)
.unwrap();
let names = output_column_names(tout.path());
assert!(
names.iter().any(|n| n == "Coalesced_Count"),
"passthrough column kept verbatim, names={names:?}"
);
}
#[test]
fn coalesce_footer_provenance_recorded() {
let geoms = fragment_chain_geoms(3);
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let opts = ConvertOptions {
coalesce_lines: true,
coalesce_snap: 1.5,
coalesce_junction_angle: 30.0,
coalesce_max_level_rows: 123_456,
..Default::default()
};
convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
let reader = OverviewReader::open(tout.path()).unwrap();
let c = reader
.meta()
.generalization
.as_ref()
.unwrap()
.coalescing
.clone()
.expect("coalescing provenance recorded");
assert!(c.enabled);
assert_eq!(c.snap_tolerance_gsd_factor, 1.5);
assert_eq!(c.junction_angle, Some(30.0));
assert_eq!(c.max_level_rows, Some(123_456));
assert_eq!(c.coalesced_count_column, "coalesced_count");
let tout_off = tempfile::NamedTempFile::new().unwrap();
convert_to_overviews(
tin.path(),
tout_off.path(),
&ConvertOptions {
coalesce_lines: false,
..Default::default()
},
)
.unwrap();
let r_off = OverviewReader::open(tout_off.path()).unwrap();
assert!(r_off
.meta()
.generalization
.as_ref()
.unwrap()
.coalescing
.is_none());
}
#[test]
fn coalesce_guard_skips_chaining_but_keeps_column() {
let geoms = fragment_chain_geoms(6);
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let opts = ConvertOptions {
levels: LevelPlan::ZoomRange {
min_zoom: 4,
max_zoom: 10,
},
no_auto_rank: true,
coalesce_lines: true,
coalesce_max_level_rows: 2, ..Default::default()
};
let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
assert_eq!(
report.levels[0].feature_count, 1,
"guard-skipped run behaves like non-coalesced: {:?}",
report.levels
);
let reader = OverviewReader::open(tout.path()).unwrap();
for level in 0..reader.num_levels() {
assert!(read_coalesced_counts(&reader, level)
.iter()
.all(|&c| c == 1));
}
assert!(validate_file(tout.path()).unwrap().is_valid());
}
#[test]
fn streaming_matches_in_memory_coalescing() {
let geoms = fragment_chain_geoms(6);
let tin = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let base = ConvertOptions {
levels: LevelPlan::ZoomRange {
min_zoom: 4,
max_zoom: 10,
},
no_auto_rank: true,
coalesce_lines: true,
read_batch_size: 2,
..Default::default()
};
assert_streaming_equivalent(tin.path(), &base);
let mem_out = tempfile::NamedTempFile::new().unwrap();
let stream_out = tempfile::NamedTempFile::new().unwrap();
convert_to_overviews(
tin.path(),
mem_out.path(),
&ConvertOptions {
streaming: false,
..base.clone()
},
)
.unwrap();
convert_to_overviews(tin.path(), stream_out.path(), &base).unwrap();
let mr = OverviewReader::open(mem_out.path()).unwrap();
let sr = OverviewReader::open(stream_out.path()).unwrap();
for level in 0..mr.num_levels() {
assert_eq!(
read_coalesced_counts(&mr, level),
read_coalesced_counts(&sr, level),
"level {level} coalesced_count differs"
);
}
}
#[test]
fn streaming_matches_in_memory_coalescing_with_class_groups() {
let mut geoms = vec![
Geometry::LineString(LineString::from(vec![(0.0, 0.0), (0.1, 0.0)])),
Geometry::LineString(LineString::from(vec![(0.1, 0.0), (0.2, 0.0)])),
Geometry::LineString(LineString::from(vec![(0.2, 0.0), (0.2, 0.1)])),
];
let mut classes = vec![Some("motorway"), Some("motorway"), Some("footway")];
for (i, c) in ["primary", "service", "residential", "trunk"]
.iter()
.enumerate()
{
geoms.push(Geometry::LineString(LineString::from(vec![
(3.0 + i as f64, 3.0),
(3.1 + i as f64, 3.05),
])));
classes.push(Some(*c));
}
let tin = tempfile::NamedTempFile::new().unwrap();
write_class_input(tin.path(), &geoms, &classes);
let base = ConvertOptions {
levels: LevelPlan::ZoomRange {
min_zoom: 4,
max_zoom: 10,
},
coalesce_lines: true,
read_batch_size: 2,
..Default::default()
};
let mem_out = tempfile::NamedTempFile::new().unwrap();
let stream_out = tempfile::NamedTempFile::new().unwrap();
convert_to_overviews(
tin.path(),
mem_out.path(),
&ConvertOptions {
streaming: false,
..base.clone()
},
)
.unwrap();
convert_to_overviews(tin.path(), stream_out.path(), &base).unwrap();
assert_eq!(
overviews_footer_json(mem_out.path()),
overviews_footer_json(stream_out.path())
);
let mr = OverviewReader::open(mem_out.path()).unwrap();
let sr = OverviewReader::open(stream_out.path()).unwrap();
assert_eq!(mr.num_levels(), sr.num_levels());
for level in 0..mr.num_levels() {
assert_eq!(
read_coalesced_counts(&mr, level),
read_coalesced_counts(&sr, level),
"level {level} coalesced_count differs"
);
}
}
#[test]
fn sort_key_column_missing_errors() {
let geoms = synthetic_geometries();
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let opts = ConvertOptions {
sort_key: Some("nonexistent".to_string()),
..Default::default()
};
let err = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap_err();
assert!(matches!(err, ConvertError::SortKeyColumnMissing { .. }));
}
fn write_multi_rg_input(path: &Path, coords: &[(f64, f64)], with_covering: bool) {
use parquet::file::properties::WriterProperties;
let geoms: Vec<Geometry<f64>> = coords
.iter()
.map(|&(x, y)| Geometry::Point(Point::new(x, y)))
.collect();
let n = geoms.len();
let id = Int64Array::from((0..n as i64).collect::<Vec<_>>());
let geom_arr = build_geometry_array(&geoms);
let geom_field = geom_arr.data_type().to_field("geometry", true);
let fields = vec![
Arc::new(Field::new("id", DataType::Int64, false)),
Arc::new(geom_field),
];
let columns: Vec<Arc<dyn Array>> = vec![Arc::new(id), geom_arr.to_array_ref()];
let schema = Arc::new(Schema::new(fields));
let batch = RecordBatch::try_new(schema.clone(), columns).unwrap();
let gpq_options = GeoParquetWriterOptionsBuilder::default()
.set_encoding(GeoParquetWriterEncoding::WKB)
.set_generate_covering(with_covering)
.build();
let encoder = GeoParquetRecordBatchEncoder::try_new(&schema, &gpq_options).unwrap();
let target_schema = encoder.target_schema();
let props = WriterProperties::builder()
.set_max_row_group_row_count(Some(1))
.build();
let file = std::fs::File::create(path).unwrap();
let mut writer = ArrowWriter::try_new(file, target_schema, Some(props)).unwrap();
let mut encoder = encoder;
let encoded = encoder.encode_record_batch(&batch).unwrap();
writer.write(&encoded).unwrap();
writer.append_key_value_metadata(encoder.into_keyvalue().unwrap());
writer.close().unwrap();
}
fn read_all_ids(reader: &OverviewReader) -> Vec<i64> {
use arrow_array::cast::AsArray;
let mut ids = Vec::new();
for level in 0..reader.num_levels() {
let rdr = reader.read_level(level, None).unwrap();
for batch in rdr {
let batch = batch.unwrap();
let col = batch
.column(batch.schema().index_of("id").unwrap())
.as_primitive::<arrow_array::types::Int64Type>();
ids.extend(col.iter().flatten());
}
}
ids.sort_unstable();
ids.dedup();
ids
}
#[test]
fn bbox_filter_matches_posthoc_filter() {
let coords = vec![(0.0, 0.0), (10.0, 10.0), (20.0, 20.0), (30.0, 30.0)];
let tin = tempfile::NamedTempFile::new().unwrap();
write_multi_rg_input(tin.path(), &coords, true);
let tout_full = tempfile::NamedTempFile::new().unwrap();
let opts_full = ConvertOptions {
mode: Mode::Duplicating,
levels: LevelPlan::ZoomRange {
min_zoom: 6,
max_zoom: 6,
},
..Default::default()
};
let report_full = convert_to_overviews(tin.path(), tout_full.path(), &opts_full).unwrap();
assert_eq!(report_full.row_groups_total, 4);
assert_eq!(report_full.row_groups_read, 4);
let reader_full = OverviewReader::open(tout_full.path()).unwrap();
let ids_full = read_all_ids(&reader_full);
let tout_bbox = tempfile::NamedTempFile::new().unwrap();
let opts_bbox = ConvertOptions {
bbox: Some([9.0, 9.0, 11.0, 11.0]),
..opts_full.clone()
};
let report_bbox = convert_to_overviews(tin.path(), tout_bbox.path(), &opts_bbox).unwrap();
assert_eq!(report_bbox.row_groups_total, 4);
assert_eq!(
report_bbox.row_groups_read, 1,
"bbox pruning did not fire: read {} row groups",
report_bbox.row_groups_read
);
let reader_bbox = OverviewReader::open(tout_bbox.path()).unwrap();
let ids_bbox = read_all_ids(&reader_bbox);
assert_eq!(ids_bbox, vec![1], "bbox filter kept wrong ids");
let ids_posthoc: Vec<i64> = ids_full
.into_iter()
.filter(|&id| {
let (x, y) = coords[id as usize];
(9.0..=11.0).contains(&x) && (9.0..=11.0).contains(&y)
})
.collect();
assert_eq!(ids_bbox, ids_posthoc);
}
#[test]
fn bbox_filter_stats_free_degradation() {
let coords = vec![(0.0, 0.0), (10.0, 10.0), (20.0, 20.0), (30.0, 30.0)];
let tin = tempfile::NamedTempFile::new().unwrap();
write_multi_rg_input(tin.path(), &coords, false);
let tout = tempfile::NamedTempFile::new().unwrap();
let opts = ConvertOptions {
mode: Mode::Duplicating,
levels: LevelPlan::ZoomRange {
min_zoom: 6,
max_zoom: 6,
},
bbox: Some([9.0, 9.0, 11.0, 11.0]),
..Default::default()
};
let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
assert_eq!(report.row_groups_total, 4);
assert_eq!(
report.row_groups_read, 4,
"stats-free should read all row groups"
);
let reader = OverviewReader::open(tout.path()).unwrap();
let ids = read_all_ids(&reader);
assert_eq!(ids, vec![1], "exact filter did not apply");
}
#[test]
fn bbox_filter_nothing_intersects() {
let coords = vec![(0.0, 0.0), (10.0, 10.0)];
let tin = tempfile::NamedTempFile::new().unwrap();
write_multi_rg_input(tin.path(), &coords, true);
let tout = tempfile::NamedTempFile::new().unwrap();
let opts = ConvertOptions {
mode: Mode::Duplicating,
levels: LevelPlan::ZoomRange {
min_zoom: 6,
max_zoom: 6,
},
bbox: Some([100.0, 100.0, 110.0, 110.0]), ..Default::default()
};
let err = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap_err();
assert!(
matches!(err, ConvertError::NoData),
"expected NoData, got {err:?}"
);
}
#[test]
fn bbox_filter_everything_intersects() {
let coords = vec![(0.0, 0.0), (10.0, 10.0)];
let tin = tempfile::NamedTempFile::new().unwrap();
write_multi_rg_input(tin.path(), &coords, true);
let tout_full = tempfile::NamedTempFile::new().unwrap();
let opts_full = ConvertOptions {
mode: Mode::Duplicating,
levels: LevelPlan::ZoomRange {
min_zoom: 6,
max_zoom: 6,
},
..Default::default()
};
let _report_full = convert_to_overviews(tin.path(), tout_full.path(), &opts_full).unwrap();
let reader_full = OverviewReader::open(tout_full.path()).unwrap();
let ids_full = read_all_ids(&reader_full);
let tout_bbox = tempfile::NamedTempFile::new().unwrap();
let opts_bbox = ConvertOptions {
bbox: Some([-1.0, -1.0, 11.0, 11.0]),
..opts_full.clone()
};
let report_bbox = convert_to_overviews(tin.path(), tout_bbox.path(), &opts_bbox).unwrap();
assert_eq!(report_bbox.row_groups_read, report_bbox.row_groups_total);
let reader_bbox = OverviewReader::open(tout_bbox.path()).unwrap();
let ids_bbox = read_all_ids(&reader_bbox);
assert_eq!(ids_bbox, ids_full);
}
type AttrRow = ((f64, f64), Option<f64>, Option<&'static str>);
fn write_multi_rg_attr_input(path: &Path, rows: &[AttrRow]) {
use parquet::file::properties::WriterProperties;
let geoms: Vec<Geometry<f64>> = rows
.iter()
.map(|&((x, y), _, _)| Geometry::Point(Point::new(x, y)))
.collect();
let n = geoms.len();
let id = Int64Array::from((0..n as i64).collect::<Vec<_>>());
let confidence =
arrow_array::Float64Array::from(rows.iter().map(|(_, c, _)| *c).collect::<Vec<_>>());
let crop =
arrow_array::StringArray::from(rows.iter().map(|(_, _, s)| *s).collect::<Vec<_>>());
let geom_arr = build_geometry_array(&geoms);
let geom_field = geom_arr.data_type().to_field("geometry", true);
let fields = vec![
Arc::new(Field::new("id", DataType::Int64, false)),
Arc::new(Field::new("confidence", DataType::Float64, true)),
Arc::new(Field::new("crop", DataType::Utf8, true)),
Arc::new(geom_field),
];
let columns: Vec<Arc<dyn Array>> = vec![
Arc::new(id),
Arc::new(confidence),
Arc::new(crop),
geom_arr.to_array_ref(),
];
let schema = Arc::new(Schema::new(fields));
let batch = RecordBatch::try_new(schema.clone(), columns).unwrap();
let gpq_options = GeoParquetWriterOptionsBuilder::default()
.set_encoding(GeoParquetWriterEncoding::WKB)
.set_generate_covering(true)
.build();
let encoder = GeoParquetRecordBatchEncoder::try_new(&schema, &gpq_options).unwrap();
let target_schema = encoder.target_schema();
let props = WriterProperties::builder()
.set_max_row_group_row_count(Some(1))
.build();
let file = std::fs::File::create(path).unwrap();
let mut writer = ArrowWriter::try_new(file, target_schema, Some(props)).unwrap();
let mut encoder = encoder;
let encoded = encoder.encode_record_batch(&batch).unwrap();
writer.write(&encoded).unwrap();
writer.append_key_value_metadata(encoder.into_keyvalue().unwrap());
writer.close().unwrap();
}
fn attr_rows() -> Vec<AttrRow> {
vec![
((0.0, 0.0), Some(0.1), Some("soy")),
((10.0, 10.0), Some(0.9), Some("corn")),
((20.0, 20.0), Some(0.85), Some("soy")),
((30.0, 30.0), None, Some("rice")),
]
}
fn attr_opts() -> ConvertOptions {
ConvertOptions {
mode: Mode::Duplicating,
levels: LevelPlan::ZoomRange {
min_zoom: 6,
max_zoom: 6,
},
..Default::default()
}
}
#[test]
fn attribute_filter_matches_posthoc_and_prunes_row_groups() {
let tin = tempfile::NamedTempFile::new().unwrap();
write_multi_rg_attr_input(tin.path(), &attr_rows());
for streaming in [true, false] {
let tout = tempfile::NamedTempFile::new().unwrap();
let opts = ConvertOptions {
filter: Some("confidence > 0.8".to_string()),
streaming,
..attr_opts()
};
let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
assert_eq!(report.row_groups_total, 4, "streaming={streaming}");
assert_eq!(
report.row_groups_read, 2,
"stats pushdown did not fire (streaming={streaming})"
);
let ids = read_all_ids(&OverviewReader::open(tout.path()).unwrap());
assert_eq!(ids, vec![1, 2], "streaming={streaming}");
}
}
#[test]
fn attribute_filter_composes_with_bbox() {
let tin = tempfile::NamedTempFile::new().unwrap();
write_multi_rg_attr_input(tin.path(), &attr_rows());
for streaming in [true, false] {
let tout = tempfile::NamedTempFile::new().unwrap();
let opts = ConvertOptions {
bbox: Some([9.0, 9.0, 11.0, 11.0]),
filter: Some("confidence > 0.8".to_string()),
streaming,
..attr_opts()
};
let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
assert_eq!(
report.row_groups_read, 1,
"bbox+filter selection must intersect (streaming={streaming})"
);
let ids = read_all_ids(&OverviewReader::open(tout.path()).unwrap());
assert_eq!(ids, vec![1], "streaming={streaming}");
}
}
#[test]
fn attribute_filter_string_in_and_null_semantics() {
let tin = tempfile::NamedTempFile::new().unwrap();
write_multi_rg_attr_input(tin.path(), &attr_rows());
for streaming in [true, false] {
let tout = tempfile::NamedTempFile::new().unwrap();
let opts = ConvertOptions {
filter: Some("crop IN ('corn', 'rice')".to_string()),
streaming,
..attr_opts()
};
convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
let ids = read_all_ids(&OverviewReader::open(tout.path()).unwrap());
assert_eq!(ids, vec![1, 3], "IN (streaming={streaming})");
let tout = tempfile::NamedTempFile::new().unwrap();
let opts = ConvertOptions {
filter: Some("confidence IS NULL".to_string()),
streaming,
..attr_opts()
};
convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
let ids = read_all_ids(&OverviewReader::open(tout.path()).unwrap());
assert_eq!(ids, vec![3], "IS NULL (streaming={streaming})");
let tout = tempfile::NamedTempFile::new().unwrap();
let opts = ConvertOptions {
filter: Some("confidence > 0.8 OR crop = 'rice'".to_string()),
streaming,
..attr_opts()
};
convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
let ids = read_all_ids(&OverviewReader::open(tout.path()).unwrap());
assert_eq!(ids, vec![1, 2, 3], "OR (streaming={streaming})");
}
}
#[test]
fn attribute_filter_error_paths() {
let tin = tempfile::NamedTempFile::new().unwrap();
write_multi_rg_attr_input(tin.path(), &attr_rows());
let tout = tempfile::NamedTempFile::new().unwrap();
let bad_syntax = ConvertOptions {
filter: Some("confidence >".to_string()),
..attr_opts()
};
let err = convert_to_overviews(tin.path(), tout.path(), &bad_syntax).unwrap_err();
assert!(matches!(err, ConvertError::Filter(_)), "got {err:?}");
let unknown = ConvertOptions {
filter: Some("nope = 1".to_string()),
..attr_opts()
};
let err = convert_to_overviews(tin.path(), tout.path(), &unknown).unwrap_err();
assert!(
matches!(err, ConvertError::Filter(_)) && err.to_string().contains("unknown column"),
"got {err:?}"
);
let mismatch = ConvertOptions {
filter: Some("crop > 3".to_string()),
..attr_opts()
};
let err = convert_to_overviews(tin.path(), tout.path(), &mismatch).unwrap_err();
assert!(matches!(err, ConvertError::Filter(_)), "got {err:?}");
let none = ConvertOptions {
filter: Some("confidence > 99".to_string()),
..attr_opts()
};
let err = convert_to_overviews(tin.path(), tout.path(), &none).unwrap_err();
assert!(matches!(err, ConvertError::NoData), "got {err:?}");
}
#[cfg(feature = "remote")]
mod remote_input {
use super::*;
use crate::input::{test_memory_source, InputSource};
fn row_group_spans(bytes: &[u8]) -> Vec<std::ops::Range<u64>> {
let builder =
ParquetRecordBatchReaderBuilder::try_new(bytes::Bytes::from(bytes.to_vec()))
.unwrap();
builder
.metadata()
.row_groups()
.iter()
.map(|rg| {
let mut start = u64::MAX;
let mut end = 0u64;
for col in rg.columns() {
let (s, len) = col.byte_range();
start = start.min(s);
end = end.max(s + len);
}
start..end
})
.collect()
}
fn assert_bbox_extract_fetches_only_selected(streaming: bool) {
let coords = vec![(0.0, 0.0), (10.0, 10.0), (20.0, 20.0), (30.0, 30.0)];
let tin = tempfile::NamedTempFile::new().unwrap();
write_multi_rg_input(tin.path(), &coords, true);
let bytes = std::fs::read(tin.path()).unwrap();
let spans = row_group_spans(&bytes);
assert_eq!(spans.len(), 4);
let source = test_memory_source(bytes, "multi.parquet");
let tout = tempfile::NamedTempFile::new().unwrap();
let opts = ConvertOptions {
mode: Mode::Duplicating,
levels: LevelPlan::ZoomRange {
min_zoom: 6,
max_zoom: 6,
},
bbox: Some([9.0, 9.0, 11.0, 11.0]),
streaming,
..Default::default()
};
let report = convert_to_overviews_source(&source, tout.path(), &opts).unwrap();
assert_eq!(report.row_groups_total, 4);
assert_eq!(report.row_groups_read, 1, "bbox pruning must fire");
let ids = read_all_ids(&OverviewReader::open(tout.path()).unwrap());
assert_eq!(ids, vec![1], "only the (10,10) feature survives");
let fetched = source.fetched_ranges().unwrap();
assert!(!fetched.is_empty());
for (i, span) in spans.iter().enumerate() {
if i == 1 {
continue;
}
for r in &fetched {
assert!(
r.end <= span.start || r.start >= span.end,
"fetched range {r:?} overlaps PRUNED row group {i} ({span:?})"
);
}
}
assert!(
fetched
.iter()
.any(|r| r.start >= spans[1].start && r.end <= spans[1].end),
"selected row group 1 ({:?}) never fetched: {fetched:?}",
spans[1]
);
let stats = report.remote_fetch.expect("remote stats in report");
assert!(stats.requests as usize >= fetched.len());
assert!(
stats.bytes_fetched < stats.object_size,
"bbox extract must move fewer bytes than the object: {stats:?}"
);
}
#[test]
fn bbox_extract_streaming_fetches_only_selected_row_groups() {
assert_bbox_extract_fetches_only_selected(true);
}
#[test]
fn bbox_extract_in_memory_fetches_only_selected_row_groups() {
assert_bbox_extract_fetches_only_selected(false);
}
#[test]
fn attribute_filter_remote_fetches_only_matching_row_groups() {
let tin = tempfile::NamedTempFile::new().unwrap();
write_multi_rg_attr_input(tin.path(), &attr_rows());
let bytes = std::fs::read(tin.path()).unwrap();
let spans = row_group_spans(&bytes);
assert_eq!(spans.len(), 4);
let source = test_memory_source(bytes, "attrs.parquet");
let tout = tempfile::NamedTempFile::new().unwrap();
let opts = ConvertOptions {
filter: Some("confidence > 0.8".to_string()),
..attr_opts()
};
let report = convert_to_overviews_source(&source, tout.path(), &opts).unwrap();
assert_eq!(report.row_groups_total, 4);
assert_eq!(report.row_groups_read, 2, "filter pushdown must fire");
let ids = read_all_ids(&OverviewReader::open(tout.path()).unwrap());
assert_eq!(ids, vec![1, 2]);
let fetched = source.fetched_ranges().unwrap();
assert!(!fetched.is_empty());
for (i, span) in spans.iter().enumerate() {
if i == 1 || i == 2 {
continue;
}
for r in &fetched {
assert!(
r.end <= span.start || r.start >= span.end,
"fetched range {r:?} overlaps PRUNED row group {i} ({span:?})"
);
}
}
let stats = report.remote_fetch.expect("remote stats in report");
assert!(
stats.bytes_fetched < stats.object_size,
"filter extract must move fewer bytes than the object: {stats:?}"
);
}
#[test]
fn remote_convert_matches_local_convert() {
let geoms = synthetic_geometries();
let tin = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let opts = ConvertOptions::default();
let tout_local = tempfile::NamedTempFile::new().unwrap();
let report_local = convert_to_overviews(tin.path(), tout_local.path(), &opts).unwrap();
assert!(report_local.remote_fetch.is_none(), "local input: no stats");
let source = test_memory_source(std::fs::read(tin.path()).unwrap(), "in.parquet");
let tout_remote = tempfile::NamedTempFile::new().unwrap();
let report_remote =
convert_to_overviews_source(&source, tout_remote.path(), &opts).unwrap();
assert_eq!(report_remote.input_features, report_local.input_features);
assert_eq!(report_remote.total_rows, report_local.total_rows);
assert_eq!(
read_all_ids(&OverviewReader::open(tout_remote.path()).unwrap()),
read_all_ids(&OverviewReader::open(tout_local.path()).unwrap()),
);
assert!(report_remote.remote_fetch.is_some());
}
#[test]
fn remote_convert_coalesces_fetches_to_one_request_per_row_group() {
let geoms = synthetic_geometries();
let n = geoms.len();
let tin = tempfile::NamedTempFile::new().unwrap();
write_input_partition(tin.path(), &geoms, 0..n, Some(2));
let bytes = std::fs::read(tin.path()).unwrap();
let rg = row_group_spans(&bytes).len();
assert!(rg >= 3, "test needs several row groups, got {rg}");
let opts = ConvertOptions::default();
let tout_local = tempfile::NamedTempFile::new().unwrap();
convert_to_overviews(tin.path(), tout_local.path(), &opts).unwrap();
let local_ids = read_all_ids(&OverviewReader::open(tout_local.path()).unwrap());
let source = test_memory_source(bytes, "staged.parquet");
let tout = tempfile::NamedTempFile::new().unwrap();
let report = convert_to_overviews_source(&source, tout.path(), &opts).unwrap();
assert_eq!(
read_all_ids(&OverviewReader::open(tout.path()).unwrap()),
local_ids,
"staged remote convert must match the local convert row-for-row"
);
let stats = report.remote_fetch.expect("remote stats in report");
const FOOTER_SLACK: u64 = 4;
assert!(
stats.requests <= rg as u64 + FOOTER_SLACK,
"expected ~1 request per row group (<= {} for {rg} row groups); \
got {} — fetches not coalesced (#287) or properties re-fetched (#286)",
rg as u64 + FOOTER_SLACK,
stats.requests,
);
assert!(
stats.bytes_fetched <= stats.object_size + stats.object_size / 2,
"staging must stay ≈1× the object: {} of {} bytes",
stats.bytes_fetched,
stats.object_size,
);
}
fn partition_bytes(
geoms: &[Geometry<f64>],
range: std::ops::Range<usize>,
row_group_rows: Option<usize>,
) -> Vec<u8> {
let tmp = tempfile::NamedTempFile::new().unwrap();
write_input_partition(tmp.path(), geoms, range, row_group_rows);
std::fs::read(tmp.path()).unwrap()
}
fn convert_sources_and_export(
source: &crate::input_set::ConvertSource,
workdir: &Path,
tag: &str,
opts: &ConvertOptions,
) -> Vec<u8> {
use crate::overview::export::{export_pmtiles, ExportOptions};
let overview = workdir.join(format!("{tag}-overview.parquet"));
let pmtiles = workdir.join(format!("{tag}.pmtiles"));
convert_to_overviews_sources(source, &overview, opts).unwrap();
export_pmtiles(&overview, &pmtiles, &ExportOptions::default()).unwrap();
std::fs::read(&pmtiles).unwrap()
}
#[test]
fn multi_part_three_pass_moves_each_part_once() {
let geoms = synthetic_geometries();
let n = geoms.len();
let (source, parts) = crate::input::test_memory_multi_source(vec![
("p0.parquet", partition_bytes(&geoms, 0..5, None)),
("p1.parquet", partition_bytes(&geoms, 5..9, None)),
("p2.parquet", partition_bytes(&geoms, 9..n, None)),
]);
assert_eq!(parts.len(), 3);
let tout = tempfile::NamedTempFile::new().unwrap();
let report =
convert_to_overviews_sources(&source, tout.path(), &multi_test_options()).unwrap();
assert_eq!(report.input_features, n);
let summed = report.remote_fetch.expect("multi remote reports stats");
let mut object_total = 0;
for part in &parts {
let stats = part.fetch_stats().expect("remote part has stats");
object_total += stats.object_size;
assert!(
stats.bytes_fetched <= stats.object_size + stats.object_size / 2,
"part {} moved {} bytes for a {}-byte object (>1.5x, #219 \
must hold per part)",
part.display_name(),
stats.bytes_fetched,
stats.object_size,
);
let mut seen = std::collections::HashSet::new();
for r in part.fetched_ranges().expect("remote part logs ranges") {
assert!(
seen.insert((r.start, r.end)),
"part {}: range {r:?} fetched more than once (#219)",
part.display_name(),
);
}
}
assert_eq!(
summed.object_size, object_total,
"ConvertReport.remote_fetch.object_size sums the parts"
);
}
#[test]
fn multi_part_remote_coalesces_per_part_row_groups() {
let geoms = synthetic_geometries();
let n = geoms.len();
let p0 = partition_bytes(&geoms, 0..5, Some(2));
let p1 = partition_bytes(&geoms, 5..9, Some(2));
let p2 = partition_bytes(&geoms, 9..n, Some(2));
let rg = [&p0, &p1, &p2].map(|b| row_group_spans(b).len());
assert!(
rg.iter().all(|&r| r >= 2),
"each part needs several row groups: {rg:?}"
);
let (source, parts) = crate::input::test_memory_multi_source(vec![
("p0.parquet", p0),
("p1.parquet", p1),
("p2.parquet", p2),
]);
let tout = tempfile::NamedTempFile::new().unwrap();
convert_to_overviews_sources(&source, tout.path(), &multi_test_options()).unwrap();
const FOOTER_SLACK: u64 = 4;
for (i, part) in parts.iter().enumerate() {
let stats = part.fetch_stats().expect("remote part has stats");
assert!(
stats.requests <= rg[i] as u64 + FOOTER_SLACK,
"part {i}: expected ~1 request per row group (<= {} for {} \
row groups); got {} — not coalesced (#287) or properties \
re-fetched (#286)",
rg[i] as u64 + FOOTER_SLACK,
rg[i],
stats.requests,
);
}
}
#[test]
fn multi_part_bbox_prunes_part_to_footer_only() {
let geoms = synthetic_geometries();
let n = geoms.len();
let p1_bytes = partition_bytes(&geoms, 6..10, None);
let p1_spans = row_group_spans(&p1_bytes);
let (source, parts) = crate::input::test_memory_multi_source(vec![
("p0.parquet", partition_bytes(&geoms, 0..6, None)),
("p1.parquet", p1_bytes),
("p2.parquet", partition_bytes(&geoms, 10..n, None)),
]);
let opts = ConvertOptions {
bbox: Some([-100.0, -70.0, 30.0, 20.0]),
..multi_test_options()
};
let tout = tempfile::NamedTempFile::new().unwrap();
let report = convert_to_overviews_sources(&source, tout.path(), &opts).unwrap();
assert!(
report.row_groups_read < report.row_groups_total,
"bbox must prune the lines part: {}/{}",
report.row_groups_read,
report.row_groups_total
);
let pruned = &parts[1];
let fetched = pruned.fetched_ranges().expect("remote part logs ranges");
assert!(
!fetched.is_empty(),
"the footer itself is fetched at set construction"
);
for r in &fetched {
for span in &p1_spans {
assert!(
r.end <= span.start || r.start >= span.end,
"pruned part fetched data-page range {r:?} \
(row-group span {span:?}) — must be footer-only"
);
}
}
}
#[test]
fn multi_part_remote_output_matches_single_remote() {
let geoms = synthetic_geometries();
let n = geoms.len();
let dir = tempfile::tempdir().unwrap();
let opts = multi_test_options();
let (single, _) = crate::input::test_memory_multi_source(vec![(
"single.parquet",
partition_bytes(&geoms, 0..n, None),
)]);
let (multi, parts) = crate::input::test_memory_multi_source(vec![
("part-000.parquet", partition_bytes(&geoms, 0..5, None)),
("part-001.parquet", partition_bytes(&geoms, 5..9, None)),
("part-002.parquet", partition_bytes(&geoms, 9..n, None)),
]);
assert_eq!(parts.len(), 3);
let pm_single = convert_sources_and_export(&single, dir.path(), "single", &opts);
let pm_multi = convert_sources_and_export(&multi, dir.path(), "multi", &opts);
assert!(
pm_single == pm_multi,
"remote multi-partition output must be byte-identical to the \
single remote object ({} vs {} bytes)",
pm_single.len(),
pm_multi.len()
);
}
fn serve_bytes_over_http(body: Vec<u8>) -> String {
use std::io::{BufRead, BufReader, Write};
use std::net::TcpListener;
use std::sync::Arc;
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap();
let body = Arc::new(body);
std::thread::spawn(move || {
for stream in listener.incoming() {
let Ok(mut stream) = stream else { continue };
let body = Arc::clone(&body);
std::thread::spawn(move || {
let peer = stream.try_clone().unwrap();
let mut reader = BufReader::new(peer);
let size = body.len() as u64;
loop {
let mut request_line = String::new();
match reader.read_line(&mut request_line) {
Ok(0) | Err(_) => break, Ok(_) => {}
}
let mut parts = request_line.split_whitespace();
let method = parts.next().unwrap_or("").to_string();
if method.is_empty() {
break;
}
let mut range: Option<(u64, u64)> = None;
loop {
let mut header = String::new();
if reader.read_line(&mut header).unwrap_or(0) == 0 {
break;
}
if header == "\r\n" || header == "\n" {
break;
}
let lower = header.to_ascii_lowercase();
let Some(spec) = lower
.strip_prefix("range:")
.and_then(|v| v.trim().strip_prefix("bytes="))
else {
continue;
};
let spec = spec.split(',').next().unwrap_or("").trim();
let (a, b) = spec.split_once('-').unwrap_or((spec, ""));
let (start, end) = if a.is_empty() {
let n: u64 = b.trim().parse().unwrap_or(0);
(size.saturating_sub(n), size.saturating_sub(1))
} else {
let start = a.trim().parse().unwrap_or(0);
let end = if b.trim().is_empty() {
size.saturating_sub(1)
} else {
b.trim().parse().unwrap_or(size - 1)
};
(start, end.min(size.saturating_sub(1)))
};
range = Some((start, end));
}
let response: Vec<u8> = match (method.as_str(), range) {
("HEAD", _) => format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/octet-stream\r\n\
Accept-Ranges: bytes\r\nContent-Length: {size}\r\n\r\n"
)
.into_bytes(),
("GET", Some((start, end))) => {
let slice = &body[start as usize..=end as usize];
let mut resp = format!(
"HTTP/1.1 206 Partial Content\r\n\
Content-Type: application/octet-stream\r\n\
Accept-Ranges: bytes\r\n\
Content-Range: bytes {start}-{end}/{size}\r\n\
Content-Length: {}\r\n\r\n",
slice.len()
)
.into_bytes();
resp.extend_from_slice(slice);
resp
}
("GET", None) => {
let mut resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/octet-stream\r\n\
Accept-Ranges: bytes\r\nContent-Length: {size}\r\n\r\n"
)
.into_bytes();
resp.extend_from_slice(&body);
resp
}
_ => {
b"HTTP/1.1 405 Method Not Allowed\r\nContent-Length: 0\r\n\r\n"
.to_vec()
}
};
if stream.write_all(&response).is_err() {
break;
}
let _ = stream.flush();
}
});
}
});
format!("http://{addr}")
}
#[test]
fn remote_http_convert_matches_local() {
let geoms = synthetic_geometries();
let tin = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let opts = ConvertOptions::default();
let tout_local = tempfile::NamedTempFile::new().unwrap();
let report_local = convert_to_overviews(tin.path(), tout_local.path(), &opts).unwrap();
let base = serve_bytes_over_http(std::fs::read(tin.path()).unwrap());
let url = format!("{base}/in.parquet");
let source = InputSource::from_str_input(&url).unwrap();
assert!(source.is_remote(), "http:// input must be remote");
let tout_remote = tempfile::NamedTempFile::new().unwrap();
let report_remote =
convert_to_overviews_source(&source, tout_remote.path(), &opts).unwrap();
assert_eq!(report_remote.input_features, report_local.input_features);
assert_eq!(
read_all_ids(&OverviewReader::open(tout_remote.path()).unwrap()),
read_all_ids(&OverviewReader::open(tout_local.path()).unwrap()),
);
let stats = report_remote
.remote_fetch
.expect("http input reports fetch stats");
assert!(
stats.bytes_fetched > 0 && stats.requests > 0,
"http conversion moved bytes: {stats:?}"
);
}
#[test]
fn unsupported_scheme_errors_through_convert() {
let tout = tempfile::NamedTempFile::new().unwrap();
let err = convert_to_overviews(
Path::new("ftp://example.com/x.parquet"),
tout.path(),
&ConvertOptions::default(),
)
.unwrap_err();
assert!(matches!(err, ConvertError::Input(_)), "got: {err}");
assert!(err.to_string().contains("s3://"), "helpful message: {err}");
}
#[test]
fn remote_s3_city_extract_integration() {
const URL: &str = "s3://tylertoo-bench/corpus/points-nyc-medium.rg20k.parquet";
let source = match InputSource::from_str_input(URL) {
Ok(s) => s,
Err(e) => {
eprintln!(
"SKIP remote_s3_city_extract_integration (no credentials/network): {e}"
);
return;
}
};
let tout = tempfile::NamedTempFile::new().unwrap();
let opts = ConvertOptions {
bbox: Some([-73.99, 40.72, -73.98, 40.73]),
..Default::default()
};
let report = match convert_to_overviews_source(&source, tout.path(), &opts) {
Ok(r) => r,
Err(e) => {
eprintln!("SKIP remote_s3_city_extract_integration (network flake?): {e}");
return;
}
};
assert!(report.input_features > 0, "bbox should select features");
assert!(
report.row_groups_read < report.row_groups_total,
"row-group pruning should fire on the Hilbert-sorted input \
({}/{} read)",
report.row_groups_read,
report.row_groups_total
);
let stats = report.remote_fetch.expect("remote stats");
assert!(
stats.bytes_fetched * 4 < stats.object_size,
"city extract should move <25% of the remote object even \
across streaming passes: {stats:?}"
);
eprintln!(
"remote_s3_city_extract_integration: {} requests, {} of {} bytes ({:.2}%)",
stats.requests,
stats.bytes_fetched,
stats.object_size,
100.0 * stats.bytes_fetched as f64 / stats.object_size as f64
);
}
}
fn tiny_polygons(n: usize) -> Vec<Geometry<f64>> {
(0..n)
.map(|i| {
let cx = -150.0 + (i % 10) as f64 * 3.0;
let cy = -60.0 + (i / 10) as f64 * 1.5;
let h = 5e-5;
let ext = LineString::from(vec![
(cx - h, cy - h),
(cx + h, cy - h),
(cx + h, cy + h),
(cx - h, cy + h),
(cx - h, cy - h),
]);
Geometry::Polygon(Polygon::new(ext, vec![]))
})
.collect()
}
fn assert_clamped_pyramid(mode: Mode, streaming: bool) {
let geoms = tiny_polygons(20);
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let opts = ConvertOptions {
mode,
levels: LevelPlan::ZoomRange {
min_zoom: 0,
max_zoom: 4,
},
streaming,
..Default::default()
};
let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
assert_eq!(report.levels.len(), 1, "expected a single written level");
assert_eq!(report.levels[0].level, 0);
assert_eq!(report.levels[0].zoom, Some(4));
assert_eq!(report.levels[0].feature_count, geoms.len());
assert_eq!(report.total_rows, geoms.len());
let skipped: Vec<(usize, Option<u8>)> = report
.skipped_empty_levels
.iter()
.map(|s| (s.planned_level, s.zoom))
.collect();
assert_eq!(
skipped,
vec![(0, Some(0)), (1, Some(1)), (2, Some(2)), (3, Some(3))]
);
assert!(report.skipped_empty_levels.iter().all(|s| s.gsd > 0.0));
let vr = validate_file(tout.path()).unwrap();
assert!(
vr.is_valid(),
"failures: {:?}",
vr.failures().collect::<Vec<_>>()
);
let reader = OverviewReader::open(tout.path()).unwrap();
assert_eq!(reader.num_levels(), 1);
let tpm = tempfile::NamedTempFile::new().unwrap();
let export = crate::overview::export::export_pmtiles(
tout.path(),
tpm.path(),
&crate::overview::export::ExportOptions::default(),
)
.unwrap();
assert_eq!(export.min_zoom, 4);
assert_eq!(export.max_zoom, 4);
}
#[test]
fn empty_coarse_levels_clamped_duplicating_memory() {
assert_clamped_pyramid(Mode::Duplicating, false);
}
#[test]
fn empty_coarse_levels_clamped_duplicating_streaming() {
assert_clamped_pyramid(Mode::Duplicating, true);
}
#[test]
fn empty_coarse_levels_clamped_partitioning_memory() {
assert_clamped_pyramid(Mode::Partitioning, false);
}
#[test]
fn empty_coarse_levels_clamped_partitioning_streaming() {
assert_clamped_pyramid(Mode::Partitioning, true);
}
#[test]
fn all_levels_empty_is_hard_error() {
for streaming in [false, true] {
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &[], false, None);
let opts = ConvertOptions {
mode: Mode::Duplicating,
levels: LevelPlan::ZoomRange {
min_zoom: 0,
max_zoom: 3,
},
streaming,
..Default::default()
};
let err = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap_err();
assert!(
matches!(err, ConvertError::NoData),
"streaming={streaming}: expected NoData, got {err:?}"
);
}
}
#[test]
fn write_time_empty_level_skipped_streaming() {
let mut geoms = tiny_polygons(8);
let square = |cx: f64, cy: f64| {
let h = 5e-5;
Polygon::new(
LineString::from(vec![
(cx - h, cy - h),
(cx + h, cy - h),
(cx + h, cy + h),
(cx - h, cy + h),
(cx - h, cy - h),
]),
vec![],
)
};
geoms.push(Geometry::MultiPolygon(geo::MultiPolygon::new(vec![
square(-80.0, 10.0),
square(80.0, 30.0),
])));
let tin = tempfile::NamedTempFile::new().unwrap();
let tout = tempfile::NamedTempFile::new().unwrap();
write_input(tin.path(), &geoms, false, None);
let opts = ConvertOptions {
mode: Mode::Duplicating,
levels: LevelPlan::ZoomRange {
min_zoom: 0,
max_zoom: 3,
},
streaming: true,
..Default::default()
};
let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
assert_eq!(report.levels.len(), 1);
assert_eq!(report.levels[0].level, 0);
assert_eq!(report.levels[0].zoom, Some(3));
assert_eq!(report.levels[0].feature_count, geoms.len());
assert_eq!(
report
.skipped_empty_levels
.iter()
.map(|s| s.planned_level)
.collect::<Vec<_>>(),
vec![0, 1, 2]
);
let vr = validate_file(tout.path()).unwrap();
assert!(
vr.is_valid(),
"failures: {:?}",
vr.failures().collect::<Vec<_>>()
);
let reader = OverviewReader::open(tout.path()).unwrap();
assert_eq!(reader.num_levels(), 1);
}
}