1use std::collections::HashMap;
33use std::collections::HashSet;
34use std::fs::File;
35use std::path::{Path, PathBuf};
36use std::sync::Arc;
37use std::time::Instant;
38
39use arrow_array::{Array, RecordBatch, UInt32Array};
40use arrow_schema::{DataType, Field, Schema, SchemaRef};
41use arrow_select::concat::concat_batches;
42use arrow_select::take::take;
43use geo::{Area, BoundingRect, Geometry};
44use geoarrow::array::{from_arrow_array, GeometryBuilder};
45use geoarrow::datatypes::GeometryType;
46use geoarrow_array::GeoArrowArray;
47use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
48
49use crate::input::InputSource;
50use crate::input_set::ConvertSource;
51use serde::Serialize;
52
53use crate::batch_processor::extract_geometries_opt_from_array;
54
55use super::accumulate::{is_carrier, level_accumulates, tiny_polygon_carriers, AccumulateLevel};
56use super::assign::{
57 apply_density_budget, assign_levels_bounded, AssignConfig, AssignFeature, Assignment,
58 DensityBudgetConfig, FeatureKind, SUPERCELL_GSD_FACTOR,
59};
60use super::cluster::{
61 build_cluster_tables, verify_sum_invariant, AccumulateSpec, ClusterEntry, ClusterTables,
62 POINT_COUNT_COLUMN,
63};
64use super::coalesce::{
65 coalesce_level_lines, CoalesceInput, CoalesceParams, COALESCED_COUNT_COLUMN,
66 DEFAULT_COALESCE_MAX_LEVEL_ROWS, DEFAULT_JUNCTION_ANGLE_DEG, DEFAULT_SNAP_GSD_FACTOR,
67};
68use super::ladder::{build_ladder, entry_levels, EntryZoomSpec};
69use super::level::{
70 gsd_with_base, AccumulatedColumn, ClusteringProvenance, CoalescingProvenance, Crs,
71 DensityProvenance, Generalization, GeneralizationLevel, MemoryProfile, Mode, RankingProvenance,
72 RepresentationBandProvenance, GSD_TILE_BASE, METERS_PER_DEGREE,
73};
74use super::properties::{PropertySelection, PropertySelectionError};
75use super::simplify::{
76 carrier_square, simplify_cascade, simplify_for_level, simplify_step, CascadeStep, CollapseMode,
77 Representation, Simplified, SimplifyOptions,
78};
79use super::writer::{
80 LevelSpec, LevelWriteOutcome, OverviewWriter, RowGroupSizePolicy, WriterError, LEVEL_COLUMN,
81};
82
83#[derive(Debug, Clone, PartialEq)]
85pub enum LevelPlan {
86 ZoomRange {
89 min_zoom: u8,
91 max_zoom: u8,
93 },
94 Gsds(Vec<f64>),
97}
98
99pub(super) const MAX_LEVELS: usize = 255;
104
105impl LevelPlan {
106 pub(super) fn resolve(&self, gsd_base: f64) -> Result<Vec<(f64, Option<u8>)>, ConvertError> {
113 let check_len = |n: usize| {
114 if n > MAX_LEVELS {
115 return Err(ConvertError::InvalidLevels(format!(
116 "{n} levels requested; at most {MAX_LEVELS} levels are supported"
117 )));
118 }
119 Ok(())
120 };
121 match self {
122 LevelPlan::ZoomRange { min_zoom, max_zoom } => {
123 if min_zoom > max_zoom {
124 return Err(ConvertError::InvalidLevels(format!(
125 "min_zoom {min_zoom} must be <= max_zoom {max_zoom}"
126 )));
127 }
128 check_len(*max_zoom as usize - *min_zoom as usize + 1)?;
129 Ok((*min_zoom..=*max_zoom)
130 .map(|z| (gsd_with_base(z, gsd_base), Some(z)))
131 .collect())
132 }
133 LevelPlan::Gsds(gsds) => {
134 if gsds.is_empty() {
135 return Err(ConvertError::InvalidLevels(
136 "explicit gsd list must be non-empty".to_string(),
137 ));
138 }
139 check_len(gsds.len())?;
140 let mut prev: Option<f64> = None;
141 for (i, &g) in gsds.iter().enumerate() {
142 if g <= 0.0 || g.is_nan() {
143 return Err(ConvertError::InvalidLevels(format!(
144 "gsd[{i}] = {g} must be > 0"
145 )));
146 }
147 if let Some(p) = prev {
148 if g >= p {
149 return Err(ConvertError::InvalidLevels(format!(
150 "gsd list must be strictly decreasing coarse→fine (gsd[{i}] = {g} >= previous {p})"
151 )));
152 }
153 }
154 prev = Some(g);
155 }
156 Ok(gsds.iter().map(|&g| (g, None)).collect())
157 }
158 }
159 }
160}
161
162#[derive(Debug, Clone, PartialEq)]
173pub struct ClassRanking {
174 pub column: String,
176 pub ranks: Vec<(String, f64)>,
179 pub unknown_rank: f64,
182}
183
184const MAX_PROVENANCE_RANKS: usize = 64;
187
188pub fn overture_road_ranking(column: String) -> ClassRanking {
196 let ordered = [
198 "motorway", "trunk",
200 "primary",
201 "secondary",
202 "tertiary",
203 "residential",
204 "unclassified",
205 "service",
206 "living_street", "pedestrian",
208 "track",
209 "cycleway",
210 "bridleway",
211 "footway",
212 "steps",
213 "path",
214 "driveway",
215 "parking_aisle",
216 ];
217 let n = ordered.len();
218 let ranks = ordered
219 .iter()
220 .enumerate()
221 .map(|(i, &c)| (c.to_string(), (n - i) as f64))
224 .collect();
225 ClassRanking {
226 column,
227 ranks,
228 unknown_rank: 0.0,
229 }
230}
231
232pub(super) const KNOWN_ROAD_CLASSES: &[&str] = &[
237 "motorway",
238 "trunk",
239 "primary",
240 "secondary",
241 "tertiary",
242 "residential",
243 "unclassified",
244 "service",
245 "living_street",
246 "pedestrian",
247 "track",
248 "cycleway",
249 "bridleway",
250 "footway",
251 "steps",
252 "path",
253 "driveway",
254 "parking_aisle",
255 "unknown",
256 "standard_gauge",
258 "light_rail",
259 "tram",
260 "subway",
261 "monorail",
262 "funicular",
263];
264
265pub(super) const ROAD_VOCAB_MIN_DISTINCT: usize = 3;
268
269#[derive(Debug, Clone)]
271pub struct ConvertOptions {
272 pub mode: Mode,
274 pub levels: LevelPlan,
276 pub assign: AssignConfig,
278 pub entry_zoom: Option<EntryZoomSpec>,
288 pub sort_key: Option<String>,
291 pub class_ranking: Option<ClassRanking>,
294 pub no_auto_rank: bool,
297 pub simplify: SimplifyOptions,
299 pub representation: Vec<RepresentationBand>,
316 pub density: DensityBudgetConfig,
319 pub gsd_base: f64,
324 pub cogp_compat_key: bool,
326 pub max_row_group_size: usize,
328 pub row_group_size_policy: RowGroupSizePolicy,
333 pub full_column_stats: bool,
339 pub streaming: bool,
348 pub read_batch_size: usize,
353 pub profile: MemoryProfile,
358 pub in_flight_batches: usize,
368 pub cluster: bool,
375 pub accumulate: Vec<AccumulateSpec>,
380 pub coalesce_lines: bool,
397 pub coalesce_snap: f64,
402 pub coalesce_max_level_rows: usize,
408 pub coalesce_junction_angle: f64,
416 pub bbox: Option<[f64; 4]>,
428 pub filter: Option<String>,
440 pub properties: PropertySelection,
449 pub spill_dir: Option<PathBuf>,
457}
458
459pub const DEFAULT_READ_BATCH_SIZE: usize = 8192;
461
462pub const IN_FLIGHT_BATCHES_AUTO: usize = 0;
467
468pub const IN_FLIGHT_BATCHES_MIN: usize = 4;
471
472pub const IN_FLIGHT_BATCHES_MAX: usize = 16;
478
479pub fn resolve_in_flight_batches(requested: usize) -> usize {
489 if requested == IN_FLIGHT_BATCHES_AUTO {
490 std::thread::available_parallelism()
491 .map(|n| n.get())
492 .unwrap_or(IN_FLIGHT_BATCHES_MIN)
493 .clamp(IN_FLIGHT_BATCHES_MIN, IN_FLIGHT_BATCHES_MAX)
494 } else {
495 requested
496 }
497}
498
499impl ConvertOptions {
500 #[must_use]
533 pub fn verbatim(mut self) -> Self {
534 self.assign.point_thinning = 0.0;
535 self.assign.line_thinning = 0.0;
536 self.assign.polygon_thinning = 0.0;
537 self.assign.line_visibility = 0.0;
538 self.assign.polygon_visibility = 0.0;
539 self.simplify.factor = 0.0;
540 self.density.enabled = false;
541 self.coalesce_lines = false;
542 self
543 }
544
545 pub fn is_verbatim(&self) -> bool {
550 self.generalization_is_off() && self.entry_zoom.is_none()
556 }
557
558 pub(crate) fn generalization_is_off(&self) -> bool {
566 self.assign.point_thinning == 0.0
567 && self.assign.line_thinning == 0.0
568 && self.assign.polygon_thinning == 0.0
569 && self.assign.line_visibility == 0.0
570 && self.assign.polygon_visibility == 0.0
571 && self.simplify.factor == 0.0
572 && !self.density.enabled
573 && !self.coalesce_lines
574 }
575}
576
577impl Default for ConvertOptions {
578 fn default() -> Self {
579 Self {
580 mode: Mode::Duplicating,
581 levels: LevelPlan::ZoomRange {
582 min_zoom: 0,
583 max_zoom: 6,
584 },
585 assign: AssignConfig::default(),
586 entry_zoom: None,
587 sort_key: None,
588 class_ranking: None,
589 no_auto_rank: false,
590 simplify: SimplifyOptions::default(),
591 representation: Vec::new(),
592 density: DensityBudgetConfig::default(),
593 gsd_base: GSD_TILE_BASE,
594 cogp_compat_key: false,
595 max_row_group_size: super::writer::DEFAULT_MAX_ROW_GROUP_SIZE,
596 row_group_size_policy: RowGroupSizePolicy::default(),
597 full_column_stats: false,
598 streaming: true,
599 read_batch_size: DEFAULT_READ_BATCH_SIZE,
600 profile: MemoryProfile::Auto,
601 in_flight_batches: IN_FLIGHT_BATCHES_AUTO,
602 cluster: false,
603 accumulate: Vec::new(),
604 coalesce_lines: true,
605 coalesce_snap: DEFAULT_SNAP_GSD_FACTOR,
606 coalesce_max_level_rows: DEFAULT_COALESCE_MAX_LEVEL_ROWS,
607 coalesce_junction_angle: DEFAULT_JUNCTION_ANGLE_DEG,
608 bbox: None,
609 filter: None,
610 properties: PropertySelection::default(),
611 spill_dir: None,
612 }
613 }
614}
615
616#[derive(Debug, Clone, PartialEq, Serialize)]
618pub struct LevelReport {
619 pub level: usize,
621 pub gsd: f64,
623 pub zoom: Option<u8>,
625 pub feature_count: usize,
627 pub vertex_count: usize,
629 pub uncompressed_bytes: i64,
631 pub compressed_bytes: i64,
633}
634
635#[derive(Debug, Clone, PartialEq, Serialize)]
643pub struct SkippedLevelReport {
644 pub planned_level: usize,
647 pub gsd: f64,
649 pub zoom: Option<u8>,
651}
652
653pub(super) fn warn_plan_skipped_levels(
656 skipped: &[SkippedLevelReport],
657 input_features: usize,
658 first_written_gsd: f64,
659 first_written_zoom: Option<u8>,
660) {
661 if skipped.is_empty() {
662 return;
663 }
664 let ids: Vec<String> = skipped
665 .iter()
666 .map(|s| s.planned_level.to_string())
667 .collect();
668 let gsd_max = skipped.iter().map(|s| s.gsd).fold(f64::MIN, f64::max);
669 let gsd_min = skipped.iter().map(|s| s.gsd).fold(f64::MAX, f64::min);
670 let zoom_note = first_written_zoom.map_or_else(String::new, |z| format!(" (zoom {z})"));
671 log::warn!(
672 "omitting {} empty level(s) [{}] spanning GSD {:.2}–{:.2} m: none of the {} input \
673 feature(s) are visible at those scales (visibility gates / density budget); the \
674 output pyramid starts at GSD {:.2} m{}. To populate coarse levels, lower \
675 --polygon-visibility/--line-visibility, or pass --collapse to keep sub-GSD \
676 polygons as representative points (see docs/OVERVIEW_TUNING.md)",
677 skipped.len(),
678 ids.join(", "),
679 gsd_max,
680 gsd_min,
681 input_features,
682 first_written_gsd,
683 zoom_note,
684 );
685}
686
687pub(super) fn record_level_outcome(
693 outcome: LevelWriteOutcome,
694 planned: SkippedLevelReport,
695 candidates: usize,
696 rows: usize,
697 vertices: usize,
698 level_reports: &mut Vec<LevelReport>,
699 skipped: &mut Vec<SkippedLevelReport>,
700) {
701 match outcome {
702 LevelWriteOutcome::SkippedEmpty => {
703 log::warn!(
704 "level planned at GSD {:.2} m{} became empty after simplification \
705 (all {} candidate feature(s) collapsed); omitted from the output pyramid. \
706 Pass --collapse to keep sub-GSD polygons as representative points at \
707 coarse levels (see docs/OVERVIEW_TUNING.md)",
708 planned.gsd,
709 planned
710 .zoom
711 .map_or_else(String::new, |z| format!(" (zoom {z})")),
712 candidates,
713 );
714 skipped.push(planned);
715 }
716 LevelWriteOutcome::Written => level_reports.push(LevelReport {
717 level: level_reports.len(),
718 gsd: planned.gsd,
719 zoom: planned.zoom,
720 feature_count: rows,
721 vertex_count: vertices,
722 uncompressed_bytes: 0,
723 compressed_bytes: 0,
724 }),
725 }
726}
727
728#[derive(Debug, Clone, PartialEq, Serialize)]
730pub struct ConvertReport {
731 pub mode: Mode,
733 pub levels: Vec<LevelReport>,
738 pub skipped_empty_levels: Vec<SkippedLevelReport>,
743 pub input_features: usize,
745 pub total_rows: usize,
747 pub total_vertices: usize,
749 pub total_compressed_bytes: i64,
751 pub row_groups_total: usize,
753 pub row_groups_read: usize,
758 pub antimeridian_suspect_features: usize,
763 pub duration_secs: f64,
765 pub remote_fetch: Option<crate::input::FetchStats>,
770}
771
772#[derive(Debug, thiserror::Error)]
774pub enum ConvertError {
775 #[error("io error: {0}")]
777 Io(#[from] std::io::Error),
778 #[error("input error: {0}")]
780 Input(#[from] crate::input::InputError),
781 #[error("property selection: {0}")]
783 Properties(#[from] PropertySelectionError),
784 #[error("parquet error: {0}")]
786 Parquet(#[from] parquet::errors::ParquetError),
787 #[error("arrow error: {0}")]
789 Arrow(#[from] arrow_schema::ArrowError),
790 #[error("{0}")]
792 Core(#[from] crate::Error),
793 #[error("writer error: {0}")]
795 Writer(#[from] WriterError),
796 #[error("unsupported input CRS {crs:?}: overviews require EPSG:4326 or EPSG:3857")]
798 UnsupportedCrs {
799 crs: String,
801 },
802 #[error("input has no geometry column")]
804 NoGeometryColumn,
805 #[error("sort-key column {name:?} not found in input schema")]
807 SortKeyColumnMissing {
808 name: String,
810 },
811 #[error("--sort-key and --class-rank are mutually exclusive; supply at most one")]
813 RankingConflict,
814 #[error("class-rank column {name:?} not found in input schema")]
816 ClassRankColumnMissing {
817 name: String,
819 },
820 #[error("class-rank column {name:?} is {data_type} but must be a string column")]
822 ClassRankColumnNotString {
823 name: String,
825 data_type: String,
827 },
828 #[error("invalid level specification: {0}")]
830 InvalidLevels(String),
831 #[error("invalid option: {0}")]
834 InvalidConfig(String),
835 #[error(transparent)]
838 Filter(#[from] super::filter::FilterError),
839 #[error(
846 "--cluster requires duplicating mode: a partitioning-mode feature has one \
847 row read across many zoom prefixes, so a per-level point_count cannot be \
848 represented without double counting"
849 )]
850 ClusterPartitioningUnsupported,
851 #[error(
861 "--verbatim requires duplicating mode: partitioning places each feature at \
862 exactly one level, so with thinning off every feature lands in the coarsest \
863 level and every finer level is empty"
864 )]
865 VerbatimPartitioningUnsupported,
866 #[error("--accumulate-attribute requires --cluster")]
868 AccumulateWithoutCluster,
869 #[error(
872 "multi-partition input requires the streaming pipeline; \
873 remove --no-streaming"
874 )]
875 MultiPartitionRequiresStreaming,
876 #[error("accumulate-attribute column {name:?} not found in input schema")]
878 AccumulateColumnMissing {
879 name: String,
881 },
882 #[error(
884 "accumulate-attribute column {name:?} is {data_type} but must be numeric \
885 (int/uint/float)"
886 )]
887 AccumulateColumnNotNumeric {
888 name: String,
890 data_type: String,
892 },
893 #[error(
895 "input already contains a '{POINT_COUNT_COLUMN}' column; rename it before \
896 converting with --cluster"
897 )]
898 PointCountColumnPresent,
899 #[error(
902 "input already contains a '{COALESCED_COUNT_COLUMN}' column; rename it \
903 before converting with --coalesce-lines"
904 )]
905 CoalescedCountColumnPresent,
906 #[error("no output rows produced (empty input or all features dropped)")]
908 NoData,
909 #[error("cluster invariant violated (spec §12.1): {0}")]
916 ClusterInvariant(String),
917}
918
919fn validate_options(options: &ConvertOptions) -> Result<(), ConvertError> {
930 let positive = |name: &str, v: f64| {
931 if !v.is_finite() || v <= 0.0 {
932 return Err(ConvertError::InvalidConfig(format!(
933 "{name} = {v} must be a finite value > 0"
934 )));
935 }
936 Ok(())
937 };
938 let non_negative = |name: &str, v: f64| {
939 if !v.is_finite() || v < 0.0 {
940 return Err(ConvertError::InvalidConfig(format!(
941 "{name} = {v} must be a finite value >= 0"
942 )));
943 }
944 Ok(())
945 };
946 positive("gsd-base", options.gsd_base)?;
947 non_negative("point-thinning", options.assign.point_thinning)?;
951 non_negative("line-thinning", options.assign.line_thinning)?;
952 non_negative("polygon-thinning", options.assign.polygon_thinning)?;
953 non_negative("line-visibility", options.assign.line_visibility)?;
954 non_negative("polygon-visibility", options.assign.polygon_visibility)?;
955 if options.coalesce_snap.is_nan() {
958 return Err(ConvertError::InvalidConfig(
959 "coalesce-snap must not be NaN (use <= 0 to disable snapping)".to_string(),
960 ));
961 }
962 if options.coalesce_junction_angle.is_nan() {
963 return Err(ConvertError::InvalidConfig(
964 "coalesce-junction-angle must not be NaN (use 0 to disable)".to_string(),
965 ));
966 }
967 if let Some(bb) = &options.bbox {
968 if bb.iter().any(|v| !v.is_finite()) {
969 return Err(ConvertError::InvalidConfig(format!(
970 "bbox {bb:?} must contain only finite values"
971 )));
972 }
973 if bb[0] > bb[2] || bb[1] > bb[3] {
974 return Err(ConvertError::InvalidConfig(format!(
975 "bbox {bb:?} must satisfy xmin <= xmax and ymin <= ymax"
976 )));
977 }
978 }
979 if let Some(f) = &options.filter {
983 super::filter::parse_filter(f)?;
984 }
985 if !options.representation.is_empty() {
987 if matches!(options.mode, Mode::Partitioning)
988 && options
989 .representation
990 .iter()
991 .any(|b| b.repr != Representation::Geometry)
992 {
993 return Err(ConvertError::InvalidConfig(
994 "representation bands require duplicating mode: partitioning places \
995 each feature exactly once with geometry verbatim (spec §2.3), which \
996 a point or square representation cannot satisfy"
997 .to_string(),
998 ));
999 }
1000 let (plan_min, plan_max) = match &options.levels {
1001 LevelPlan::ZoomRange { min_zoom, max_zoom } => (*min_zoom, *max_zoom),
1002 LevelPlan::Gsds(_) => {
1003 return Err(ConvertError::InvalidConfig(
1004 "representation bands require a zoom-range level plan \
1005 (--min-zoom/--max-zoom); an explicit --gsd plan carries no \
1006 per-level zooms to band on"
1007 .to_string(),
1008 ));
1009 }
1010 };
1011 for band in &options.representation {
1012 let (lo, hi, repr) = (band.min_zoom, band.max_zoom, band.repr);
1013 let kw = repr.as_str();
1014 if lo > hi {
1015 return Err(ConvertError::InvalidConfig(format!(
1016 "representation band {lo}-{hi}:{kw} must satisfy LO <= HI"
1017 )));
1018 }
1019 if lo < plan_min || hi > plan_max {
1020 return Err(ConvertError::InvalidConfig(format!(
1021 "representation band {lo}-{hi}:{kw} lies outside the level plan \
1022 ({plan_min}-{plan_max})"
1023 )));
1024 }
1025 if repr != Representation::Geometry && hi >= plan_max {
1026 return Err(ConvertError::InvalidConfig(format!(
1027 "representation band {lo}-{hi}:{kw} must end before the plan's \
1028 max zoom ({plan_max}): the canonical (finest) level reproduces \
1029 source geometry verbatim (spec §2.4)"
1030 )));
1031 }
1032 }
1033 let mut claimed: Vec<Option<Representation>> =
1035 vec![None; plan_max as usize - plan_min as usize + 1];
1036 for band in &options.representation {
1037 for z in band.min_zoom..=band.max_zoom {
1038 let slot = &mut claimed[(z - plan_min) as usize];
1039 if slot.is_some() {
1040 return Err(ConvertError::InvalidConfig(format!(
1041 "representation bands overlap at zoom {z}"
1042 )));
1043 }
1044 *slot = Some(band.repr);
1045 }
1046 }
1047 let mut seen_non_point_coarser = false;
1052 for slot in &claimed {
1053 match slot {
1054 Some(Representation::Point) => {
1055 if seen_non_point_coarser {
1056 return Err(ConvertError::InvalidConfig(
1057 "point bands must start at the plan's min zoom and be \
1058 contiguous from the coarsest level: a coarser non-point \
1059 level would still receive the cascaded point"
1060 .to_string(),
1061 ));
1062 }
1063 }
1064 _ => seen_non_point_coarser = true,
1065 }
1066 }
1067 }
1068 if let Some(dir) = &options.spill_dir {
1075 if !dir.is_dir() {
1076 return Err(ConvertError::InvalidConfig(format!(
1077 "spill-dir {} is not an existing directory",
1078 dir.display()
1079 )));
1080 }
1081 }
1082 Ok(())
1083}
1084
1085#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1088pub struct RepresentationBand {
1089 pub min_zoom: u8,
1091 pub max_zoom: u8,
1093 pub repr: Representation,
1095}
1096
1097pub(super) fn representation_for_zoom(
1100 bands: &[RepresentationBand],
1101 zoom: Option<u8>,
1102) -> Representation {
1103 let Some(z) = zoom else {
1104 return Representation::Geometry;
1105 };
1106 bands
1107 .iter()
1108 .find(|b| b.min_zoom <= z && z <= b.max_zoom)
1109 .map(|b| b.repr)
1110 .unwrap_or_default()
1111}
1112
1113pub fn parse_representation_spec(spec: &str) -> Result<Vec<RepresentationBand>, String> {
1120 let mut bands = Vec::new();
1121 for part in spec.split(',') {
1122 let part = part.trim();
1123 if part.is_empty() {
1124 continue;
1125 }
1126 let (range, kind) = part.split_once(':').ok_or_else(|| {
1127 format!("representation entry {part:?} must be LO-HI:KIND (e.g. 0-7:point)")
1128 })?;
1129 let repr = match kind.trim() {
1130 "geom" | "geometry" => Representation::Geometry,
1131 "point" => Representation::Point,
1132 "square" => Representation::Square,
1133 other => {
1134 return Err(format!(
1135 "unknown representation {other:?} (expected geom, point, or square)"
1136 ))
1137 }
1138 };
1139 let range = range.trim();
1140 let (lo, hi) = match range.split_once('-') {
1141 Some((lo, hi)) => (lo.trim(), hi.trim()),
1142 None => (range, range),
1143 };
1144 let parse_zoom = |s: &str| {
1145 s.parse::<u8>()
1146 .map_err(|_| format!("invalid zoom {s:?} in representation entry {part:?}"))
1147 };
1148 bands.push(RepresentationBand {
1149 min_zoom: parse_zoom(lo)?,
1150 max_zoom: parse_zoom(hi)?,
1151 repr,
1152 });
1153 }
1154 if bands.is_empty() {
1155 return Err("representation spec is empty".to_string());
1156 }
1157 Ok(bands)
1158}
1159
1160pub(super) fn level_representations(
1164 level_specs: &[(f64, Option<u8>)],
1165 bands: &[RepresentationBand],
1166) -> Vec<Representation> {
1167 level_specs
1168 .iter()
1169 .map(|&(_, zoom)| representation_for_zoom(bands, zoom))
1170 .collect()
1171}
1172
1173pub fn convert_to_overviews(
1174 input_path: impl AsRef<Path>,
1175 output_path: impl AsRef<Path>,
1176 options: &ConvertOptions,
1177) -> Result<ConvertReport, ConvertError> {
1178 let source = ConvertSource::resolve_path(input_path.as_ref())?;
1186 convert_to_overviews_source_strategy(
1187 &source,
1188 output_path.as_ref(),
1189 options,
1190 super::stream::Pass2Strategy::Pipelined,
1191 )
1192}
1193
1194pub fn convert_to_overviews_sources(
1205 source: &ConvertSource,
1206 output_path: &Path,
1207 options: &ConvertOptions,
1208) -> Result<ConvertReport, ConvertError> {
1209 convert_to_overviews_source_strategy(
1210 source,
1211 output_path,
1212 options,
1213 super::stream::Pass2Strategy::Pipelined,
1214 )
1215}
1216
1217pub fn convert_to_overviews_source(
1221 source: &InputSource,
1222 output_path: &Path,
1223 options: &ConvertOptions,
1224) -> Result<ConvertReport, ConvertError> {
1225 convert_to_overviews_source_strategy(
1227 &ConvertSource::single(source.clone()),
1228 output_path,
1229 options,
1230 super::stream::Pass2Strategy::Pipelined,
1231 )
1232}
1233
1234#[cfg(test)]
1240pub(crate) fn convert_to_overviews_strategy(
1241 input_path: impl AsRef<Path>,
1242 output_path: impl AsRef<Path>,
1243 options: &ConvertOptions,
1244 strategy: super::stream::Pass2Strategy,
1245) -> Result<ConvertReport, ConvertError> {
1246 let source = ConvertSource::resolve_path(input_path.as_ref())?;
1247 convert_to_overviews_source_strategy(&source, output_path.as_ref(), options, strategy)
1248}
1249
1250fn decode_and_filter_geometries(
1257 full: RecordBatch,
1258 geom_idx: usize,
1259 geom_field: &Field,
1260 bbox_units: Option<&[f64; 4]>,
1261 filter_mask: Option<&[Option<bool>]>,
1262) -> Result<(RecordBatch, Vec<Geometry<f64>>), ConvertError> {
1263 let geom_array: Arc<dyn GeoArrowArray> =
1264 from_arrow_array(full.column(geom_idx).as_ref(), geom_field)
1265 .map_err(|e| crate::Error::GeoParquetRead(format!("geometry decode: {e}")))?;
1266 let mut geom_opts: Vec<Option<Geometry<f64>>> = Vec::with_capacity(full.num_rows());
1267 extract_geometries_opt_from_array(geom_array.as_ref(), &mut geom_opts)?;
1268 let mut geom_skipped = 0usize;
1269 let keep: Vec<bool> = geom_opts
1270 .iter()
1271 .enumerate()
1272 .map(|(i, g)| {
1273 if let Some(mask) = filter_mask {
1276 if mask[i] != Some(true) {
1277 return false;
1278 }
1279 }
1280 match g.as_ref().filter(|g| usable_geometry(g)) {
1281 Some(g) => bbox_units.is_none_or(|bb| bboxes_intersect(&geometry_bbox(g), bb)),
1284 None => {
1285 geom_skipped += 1;
1286 false
1287 }
1288 }
1289 })
1290 .collect();
1291 let dropped = keep.iter().filter(|k| !**k).count();
1292 if dropped == 0 {
1293 return Ok((full, geom_opts.into_iter().flatten().collect()));
1294 }
1295 if geom_skipped > 0 {
1296 log::warn!(
1297 "skipping {geom_skipped} of {} input rows with a null, empty, or \
1298 non-finite geometry",
1299 full.num_rows()
1300 );
1301 }
1302 let mask = arrow_array::BooleanArray::from(keep.clone());
1303 let filtered = arrow_select::filter::filter_record_batch(&full, &mask)?;
1304 let geoms = geom_opts
1305 .into_iter()
1306 .zip(&keep)
1307 .filter(|(_, k)| **k)
1308 .map(|(g, _)| g.expect("kept rows are Some"))
1309 .collect();
1310 Ok((filtered, geoms))
1311}
1312
1313pub(crate) fn adjusted_for_ladder_and_mode(options: &ConvertOptions) -> Option<ConvertOptions> {
1321 let coalescing_off = options.coalesce_lines
1326 && match options.mode {
1327 Mode::Partitioning => Some(
1328 "line coalescing is inert in partitioning mode (feature-once / \
1329 geometry-verbatim contract); converting without it",
1330 ),
1331 _ if options.entry_zoom.is_some() => Some(
1332 "line coalescing is inert with an entry-zoom ladder (a merged chain \
1333 has no entry zoom to inherit, and coalescing would re-gate the \
1334 lines the ladder promoted); converting without it",
1335 ),
1336 _ => None,
1337 }
1338 .inspect(|why| log::info!("{why}"))
1339 .is_some();
1340
1341 let collapse_to_point =
1346 options.entry_zoom.is_some() && matches!(options.simplify.collapse, CollapseMode::Drop);
1347 if collapse_to_point {
1348 log::info!(
1349 "an entry-zoom ladder implies collapse-to-point: a promoted feature is \
1350 usually below its level's simplification tolerance and would be deleted \
1351 there instead of drawn. Coarse levels therefore carry representative \
1352 POINTS for those features — style them with a circle layer, or ask for \
1353 --collapse-square to keep polygons."
1354 );
1355 }
1356
1357 if matches!(options.mode, Mode::Partitioning)
1362 && matches!(options.simplify.collapse, CollapseMode::Square)
1363 {
1364 log::info!(
1365 "collapse-square has no effect in partitioning mode (levels are verbatim: \
1366 no polygon is dropped or collapsed, so there is nothing to stand in for)"
1367 );
1368 }
1369
1370 if !coalescing_off && !collapse_to_point {
1371 return None;
1372 }
1373 Some(ConvertOptions {
1374 coalesce_lines: options.coalesce_lines && !coalescing_off,
1375 simplify: SimplifyOptions {
1376 collapse: if collapse_to_point {
1377 CollapseMode::Point
1378 } else {
1379 options.simplify.collapse
1380 },
1381 ..options.simplify
1382 },
1383 ..options.clone()
1384 })
1385}
1386
1387#[cfg(test)]
1388mod ladder_adjustment_tests {
1389 use super::*;
1390
1391 fn with_ladder() -> ConvertOptions {
1392 ConvertOptions {
1393 entry_zoom: Some(crate::overview::ladder::EntryZoomSpec {
1394 column: "level".to_string(),
1395 kind: crate::overview::ladder::EntryZoomKind::DenseRank { step: 1 },
1396 }),
1397 ..Default::default()
1398 }
1399 }
1400
1401 #[test]
1406 fn ladder_implies_collapse_to_point() {
1407 let adjusted = adjusted_for_ladder_and_mode(&with_ladder())
1408 .expect("a ladder must trigger an adjustment");
1409 assert_eq!(adjusted.simplify.collapse, CollapseMode::Point);
1410 }
1411
1412 #[test]
1414 fn an_explicit_collapse_square_survives_the_implication() {
1415 let mut o = with_ladder();
1416 o.simplify.collapse = CollapseMode::Square;
1417 let adjusted = adjusted_for_ladder_and_mode(&o);
1418 assert!(adjusted.is_none_or(|a| a.simplify.collapse == CollapseMode::Square));
1419 }
1420
1421 #[test]
1424 fn ladder_turns_line_coalescing_off() {
1425 let mut o = with_ladder();
1426 o.coalesce_lines = true;
1427 let adjusted = adjusted_for_ladder_and_mode(&o).expect("adjusted");
1428 assert!(!adjusted.coalesce_lines);
1429 }
1430
1431 #[test]
1433 fn no_ladder_means_no_adjustment() {
1434 assert!(adjusted_for_ladder_and_mode(&ConvertOptions::default()).is_none());
1435 }
1436}
1437
1438fn check_mode_combinations(options: &ConvertOptions) -> Result<(), ConvertError> {
1444 if options.cluster && matches!(options.mode, Mode::Partitioning) {
1447 return Err(ConvertError::ClusterPartitioningUnsupported);
1448 }
1449 if options.generalization_is_off() && matches!(options.mode, Mode::Partitioning) {
1454 return Err(ConvertError::VerbatimPartitioningUnsupported);
1455 }
1456 if !options.accumulate.is_empty() && !options.cluster {
1458 return Err(ConvertError::AccumulateWithoutCluster);
1459 }
1460 Ok(())
1461}
1462
1463fn knob_columns(options: &ConvertOptions) -> Vec<(String, String)> {
1466 let mut out = Vec::new();
1467 if let Some(c) = &options.sort_key {
1468 out.push((c.clone(), "--sort-key".to_string()));
1469 }
1470 if let Some(r) = &options.class_ranking {
1471 out.push((r.column.clone(), "--class-rank".to_string()));
1472 }
1473 if let Some(e) = &options.entry_zoom {
1474 out.push((
1475 e.column.clone(),
1476 "--magnitude-ladder / --entry-zoom".to_string(),
1477 ));
1478 }
1479 for a in &options.accumulate {
1480 out.push((a.column.clone(), "--accumulate-attribute".to_string()));
1481 }
1482 if let Some(f) = &options.filter {
1483 if let Ok(expr) = super::filter::parse_filter(f) {
1487 for c in expr.column_names() {
1488 out.push((c, "--filter".to_string()));
1489 }
1490 }
1491 }
1492 out
1493}
1494
1495fn apply_property_selection(
1505 source: &ConvertSource,
1506 options: &ConvertOptions,
1507) -> Result<(), ConvertError> {
1508 if source.column_projection().is_some() {
1509 return Err(
1510 crate::input::InputError::Arrow(arrow_schema::ArrowError::SchemaError(
1511 "column restriction already applied to this source; a ConvertSource is \
1512 single-use once a property selection has been applied"
1513 .to_string(),
1514 ))
1515 .into(),
1516 );
1517 }
1518 if options.properties.is_identity() {
1519 return Ok(());
1520 }
1521 let schema = source.file_schema()?;
1522 let geom_idx = find_geometry_column(&schema).ok_or(ConvertError::NoGeometryColumn)?;
1523 let mut warn = |msg: String| log::warn!("[convert] {msg}");
1524 let keep = options
1525 .properties
1526 .resolve(&schema, geom_idx, &knob_columns(options), &mut warn)?;
1527 let kept_names: Vec<&str> = keep
1528 .iter()
1529 .filter(|&&i| i != geom_idx)
1530 .map(|&i| schema.field(i).name().as_str())
1531 .collect();
1532 let dropped = schema.fields().len() - keep.len();
1533 log::info!(
1534 "[convert] property selection: keeping {} of {} property column(s) ({}), dropping {dropped}",
1535 kept_names.len(),
1536 schema.fields().len() - 1,
1537 if kept_names.is_empty() {
1538 "none — geometry only".to_string()
1539 } else {
1540 kept_names
1541 .iter()
1542 .map(|n| format!("{n:?}"))
1543 .collect::<Vec<_>>()
1544 .join(", ")
1545 }
1546 );
1547 source.restrict_columns(keep)?;
1548 Ok(())
1549}
1550
1551fn project_builder_to_selection(
1556 source: &ConvertSource,
1557 builder: parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder<
1558 crate::input::InputReader,
1559 >,
1560) -> Result<
1561 (
1562 parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder<crate::input::InputReader>,
1563 SchemaRef,
1564 ),
1565 ConvertError,
1566> {
1567 match source.column_projection() {
1568 Some(keep) => {
1569 let mask = parquet::arrow::ProjectionMask::roots(
1570 builder.parquet_schema(),
1571 keep.iter().copied(),
1572 );
1573 let projected = Arc::new(builder.schema().project(keep)?);
1574 Ok((builder.with_projection(mask), projected))
1575 }
1576 None => {
1577 let schema = builder.schema().clone();
1578 Ok((builder, schema))
1579 }
1580 }
1581}
1582
1583fn intern_coalesce_groups(
1588 input_schema: &Schema,
1589 full: &RecordBatch,
1590 ranking_provenance: &RankingProvenance,
1591) -> Option<Vec<u32>> {
1592 let col = coalesce_group_column(ranking_provenance)?;
1593 let idx = input_schema.index_of(col).expect("ranking column exists");
1594 let mut interner = GroupInterner::default();
1595 let mut groups = Vec::with_capacity(full.num_rows());
1596 interner.extend(full.column(idx).as_ref(), &mut groups);
1597 Some(groups)
1598}
1599
1600pub(super) fn build_verified_cluster_tables(
1605 features: &[AssignFeature],
1606 min_levels: &[u8],
1607 level_gsds: &[f64],
1608 acc_values: &[Vec<Option<f64>>],
1609 crs: Crs,
1610 options: &ConvertOptions,
1611) -> Result<ClusterTables, ConvertError> {
1612 let ops: Vec<_> = options.accumulate.iter().map(|s| s.op).collect();
1613 let tables = build_cluster_tables(
1614 features,
1615 min_levels,
1616 level_gsds,
1617 &options.assign,
1618 crs,
1619 acc_values,
1620 &ops,
1621 );
1622 verify_sum_invariant(features, min_levels, &tables).map_err(ConvertError::ClusterInvariant)?;
1625 Ok(tables)
1626}
1627
1628struct LoadedInput {
1633 options: ConvertOptions,
1636 input_schema: SchemaRef,
1637 crs: Crs,
1638 renames: Vec<(String, String)>,
1639 geom_idx: usize,
1640 geom_field: Field,
1641 acc_cols: Vec<usize>,
1643 full: RecordBatch,
1645 geometries: Vec<Geometry<f64>>,
1647 row_groups_total: usize,
1648 row_groups_read: usize,
1649}
1650
1651fn load_input_table(
1652 source: &ConvertSource,
1653 source_single: &InputSource,
1654 options: &ConvertOptions,
1655) -> Result<LoadedInput, ConvertError> {
1656 let (builder, read_schema) = project_builder_to_selection(source, source_single.open()?)?;
1661
1662 let crs = detect_crs_from_kv(builder.metadata().file_metadata().key_value_metadata())?;
1664
1665 let mut resolved = options.clone();
1672 let (input_schema, renames) = resolve_reserved_column_collisions(&read_schema, &mut resolved);
1673 let options = &resolved;
1674
1675 let bound_filter = bind_attribute_filter(options, &input_schema, &renames)?;
1678
1679 let geom_idx = find_geometry_column(&input_schema).ok_or(ConvertError::NoGeometryColumn)?;
1680 let geom_field = input_schema.field(geom_idx).clone();
1681
1682 let acc_cols = validate_cluster_schema(&input_schema, options)?;
1684 validate_coalesce_schema(&input_schema, options)?;
1686
1687 let row_groups_total = builder.metadata().num_row_groups();
1693 let bbox_units = options.bbox.map(|b| bbox_to_crs_units(&b, crs));
1694 let combined_sel = select_input_row_groups_combined(
1695 builder.metadata(),
1696 bbox_units.as_ref(),
1697 bound_filter.as_ref(),
1698 );
1699 let (builder, row_groups_read) = match combined_sel {
1700 Some(sel) => {
1701 let n = sel.len();
1702 let what = pruning_label(bbox_units.is_some(), bound_filter.is_some());
1703 log::info!("{what} filter: reading {n}/{row_groups_total} input row groups");
1704 (builder.with_row_groups(sel), n)
1705 }
1706 None => (builder, row_groups_total),
1707 };
1708
1709 let reader = builder.build()?;
1710 let mut batches: Vec<RecordBatch> = Vec::new();
1711 for batch in reader {
1712 batches.push(batch?);
1713 }
1714 let full = concat_batches(&read_schema, &batches)?;
1716
1717 let filter_mask: Option<Vec<Option<bool>>> =
1726 bound_filter.as_ref().map(|f| f.eval_mask(&full, &|i| i));
1727 let (full, geometries) = decode_and_filter_geometries(
1728 full,
1729 geom_idx,
1730 &geom_field,
1731 bbox_units.as_ref(),
1732 filter_mask.as_deref(),
1733 )?;
1734
1735 let full = if Arc::ptr_eq(&input_schema, &read_schema) {
1739 full
1740 } else {
1741 RecordBatch::try_new(input_schema.clone(), full.columns().to_vec())?
1742 };
1743
1744 Ok(LoadedInput {
1745 options: resolved.clone(),
1746 input_schema,
1747 crs,
1748 renames,
1749 geom_idx,
1750 geom_field,
1751 acc_cols,
1752 full,
1753 geometries,
1754 row_groups_total,
1755 row_groups_read,
1756 })
1757}
1758
1759struct EmittedLevel {
1760 orig: usize,
1763 gsd: f64,
1764 zoom: Option<u8>,
1765 indices: Vec<usize>,
1766 geoms: Vec<Geometry<f64>>,
1767 vertex_count: usize,
1768 coalesce: Option<CoalesceTable>,
1771}
1772
1773#[allow(clippy::too_many_arguments)]
1778fn build_emitted_levels(
1779 assignment: &Assignment,
1780 features: &[AssignFeature],
1781 geometries: &[Geometry<f64>],
1782 level_specs: &[(f64, Option<u8>)],
1783 level_reprs: &[Representation],
1784 line_groups: Option<&Vec<u32>>,
1785 coalesce_on: bool,
1786 finest: usize,
1787 crs: Crs,
1788 row_min_levels: &[u8],
1791 carriers: &[Vec<usize>],
1792 options: &ConvertOptions,
1793) -> (Vec<EmittedLevel>, Vec<SkippedLevelReport>) {
1794 let mut emitted: Vec<EmittedLevel> = Vec::new();
1797 let mut skipped: Vec<SkippedLevelReport> = Vec::new();
1798
1799 for (level, &(gsd_m, zoom)) in level_specs.iter().enumerate() {
1800 let member_indices: Vec<usize> = match options.mode {
1801 Mode::Duplicating => {
1802 let mut v = assignment.duplicating_at_level(level as u8);
1803 if !carriers[level].is_empty() {
1806 v.extend_from_slice(&carriers[level]);
1807 v.sort_unstable();
1808 }
1809 v
1810 }
1811 Mode::Partitioning => assignment.partitioning_at_level(level as u8),
1812 };
1813
1814 let verbatim = matches!(options.mode, Mode::Partitioning) || level == finest;
1817
1818 let coalesce: Option<CoalesceTable> = if coalesce_on && !verbatim {
1824 let inputs: Vec<CoalesceInput<'_>> = features
1825 .iter()
1826 .filter(|f| f.kind == FeatureKind::Line)
1827 .map(|f| CoalesceInput {
1828 index: f.index,
1829 geom: &geometries[f.index],
1830 sort_key: f.sort_key,
1831 group: line_groups.as_ref().map_or(0, |g| g[f.index]),
1832 })
1833 .collect();
1834 Some(build_level_coalesce_table(
1835 &inputs, level, finest, gsd_m, crs, options,
1836 ))
1837 } else {
1838 None
1839 };
1840 let member_indices: Vec<usize> = if let Some(table) = &coalesce {
1841 let mut v: Vec<usize> = member_indices
1842 .into_iter()
1843 .filter(|&i| features[i].kind != FeatureKind::Line)
1844 .collect();
1845 v.extend(table.keys().copied());
1846 v.sort_unstable();
1847 v
1848 } else {
1849 member_indices
1850 };
1851
1852 let mut indices = Vec::with_capacity(member_indices.len());
1853 let mut geoms = Vec::with_capacity(member_indices.len());
1854 let mut vertex_count = 0usize;
1855
1856 let repr = level_reprs[level];
1864 let cascade_chain: Vec<CascadeStep> = if options.simplify.cascade && !verbatim {
1865 (level..finest)
1866 .rev()
1867 .map(|li| CascadeStep {
1868 gsd_meters: level_specs[li].0,
1869 repr: level_reprs[li],
1870 })
1871 .collect()
1872 } else {
1873 Vec::new()
1874 };
1875
1876 if verbatim {
1877 for i in member_indices {
1878 let g = &geometries[i];
1879 vertex_count += count_vertices(g);
1880 indices.push(i);
1881 geoms.push(g.clone());
1882 }
1883 } else {
1884 for i in member_indices {
1885 if let Some((g, _)) = coalesce.as_ref().and_then(|t| t.get(&i)) {
1887 vertex_count += count_vertices(g);
1888 indices.push(i);
1889 geoms.push(g.clone());
1890 continue;
1891 }
1892 if usize::from(row_min_levels[i]) > level && is_carrier(&carriers[level], i) {
1895 if let Some(sq) = carrier_square(&geometries[i], gsd_m, crs, &options.simplify)
1896 {
1897 vertex_count += count_vertices(&sq);
1898 indices.push(i);
1899 geoms.push(sq);
1900 }
1901 continue;
1902 }
1903 let simplified = if cascade_chain.is_empty() {
1904 simplify_step(&geometries[i], gsd_m, crs, &options.simplify, repr)
1905 } else {
1906 simplify_cascade(&geometries[i], &cascade_chain, crs, &options.simplify)
1907 };
1908 match simplified {
1909 Simplified::Keep(g) => {
1910 vertex_count += count_vertices(&g);
1911 indices.push(i);
1912 geoms.push(g);
1913 }
1914 Simplified::Dropped => {}
1915 }
1916 }
1917 }
1918
1919 if indices.is_empty() {
1922 skipped.push(SkippedLevelReport {
1923 planned_level: level,
1924 gsd: gsd_m,
1925 zoom,
1926 });
1927 continue;
1928 }
1929 emitted.push(EmittedLevel {
1930 orig: level,
1931 gsd: gsd_m,
1932 zoom,
1933 indices,
1934 geoms,
1935 vertex_count,
1936 coalesce,
1937 });
1938 }
1939
1940 (emitted, skipped)
1941}
1942
1943pub(crate) fn convert_to_overviews_source_strategy(
1947 source: &ConvertSource,
1948 output_path: &Path,
1949 options: &ConvertOptions,
1950 strategy: super::stream::Pass2Strategy,
1951) -> Result<ConvertReport, ConvertError> {
1952 validate_options(options)?;
1954 apply_property_selection(source, options)?;
1957 source.set_spill_dir(options.spill_dir.as_deref());
1960 check_mode_combinations(options)?;
1961 let inert_options: ConvertOptions;
1969 let options: &ConvertOptions = match adjusted_for_ladder_and_mode(options) {
1970 Some(adjusted) => {
1971 inert_options = adjusted;
1972 &inert_options
1973 }
1974 None => options,
1975 };
1976
1977 if options.streaming {
1980 return super::stream::convert_streaming_strategy(source, output_path, options, strategy);
1981 }
1982
1983 let source_single: &InputSource = match source {
1987 ConvertSource::Single(s) => s.input(),
1988 ConvertSource::Multi(_) => return Err(ConvertError::MultiPartitionRequiresStreaming),
1989 };
1990
1991 let start = Instant::now();
1992
1993 if options.sort_key.is_some() && options.class_ranking.is_some() {
1996 return Err(ConvertError::RankingConflict);
1997 }
1998
1999 let LoadedInput {
2000 options: resolved_options,
2001 input_schema,
2002 crs,
2003 renames,
2004 geom_idx,
2005 geom_field,
2006 acc_cols,
2007 full,
2008 geometries,
2009 row_groups_total,
2010 row_groups_read,
2011 } = load_input_table(source, source_single, options)?;
2012 let options = &resolved_options;
2013 let num_features = full.num_rows();
2014
2015 let (sort_keys, ranking_provenance) =
2019 resolve_ranking(&input_schema, &full, &geometries, options)?;
2020
2021 let num_lines = geometries
2023 .iter()
2024 .filter(|g| feature_kind(g) == FeatureKind::Line)
2025 .count();
2026 let coalesce_on = coalesce_effective(options, num_lines);
2027 let line_groups: Option<Vec<u32>> = coalesce_on
2028 .then(|| intern_coalesce_groups(&input_schema, &full, &ranking_provenance))
2029 .flatten();
2030
2031 let level_specs = options.levels.resolve(options.gsd_base)?;
2033 let level_gsds: Vec<f64> = level_specs.iter().map(|(g, _)| *g).collect();
2034
2035 let ladder_values = entry_zoom_column_values(options, &input_schema, &full)?;
2038 let entry = resolve_entry_levels(options, &ladder_values, &level_specs)?;
2039
2040 let features: Vec<AssignFeature> = geometries
2041 .iter()
2042 .enumerate()
2043 .map(|(i, g)| AssignFeature {
2044 index: i,
2045 bbox: geometry_bbox(g),
2046 kind: feature_kind(g),
2047 sort_key: sort_keys[i],
2048 entry_level: entry.as_ref().and_then(|e| e[i]),
2049 })
2050 .collect();
2051
2052 let antimeridian_suspect_features = features
2054 .iter()
2055 .filter(|f| bbox_antimeridian_suspect(&f.bbox, crs))
2056 .count();
2057 warn_antimeridian_suspects(antimeridian_suspect_features);
2058
2059 let level_reprs = level_representations(&level_specs, &options.representation);
2064 let assignment = assign_levels_bounded(
2065 &features,
2066 &level_gsds,
2067 &options.assign,
2068 crs,
2069 super::pipeline::pass1_grid_budget_bytes(options.profile),
2070 &level_reprs,
2071 );
2072 let assignment = if options.density.enabled {
2077 apply_density_budget(
2078 &assignment,
2079 &features,
2080 &level_gsds,
2081 &options.assign,
2082 &options.density,
2083 crs,
2084 )
2085 } else {
2086 assignment
2087 };
2088 let num_levels = level_gsds.len();
2089 let finest = num_levels.saturating_sub(1);
2090
2091 let row_min_levels: Vec<u8> = assignment.assignments.iter().map(|a| a.min_level).collect();
2095 let carriers = in_memory_carriers(
2096 options,
2097 &features,
2098 &row_min_levels,
2099 &geometries,
2100 &level_gsds,
2101 &level_reprs,
2102 crs,
2103 );
2104
2105 let cluster_tables = if options.cluster {
2107 let min_levels: Vec<u8> = assignment.assignments.iter().map(|a| a.min_level).collect();
2108 let acc_values = extract_accumulate_values(&full, &acc_cols);
2109 Some(build_verified_cluster_tables(
2110 &features,
2111 &min_levels,
2112 &level_gsds,
2113 &acc_values,
2114 crs,
2115 options,
2116 )?)
2117 } else {
2118 None
2119 };
2120
2121 let (emitted, mut skipped) = build_emitted_levels(
2122 &assignment,
2123 &features,
2124 &geometries,
2125 &level_specs,
2126 &level_reprs,
2127 line_groups.as_ref(),
2128 coalesce_on,
2129 finest,
2130 crs,
2131 &row_min_levels,
2132 &carriers,
2133 options,
2134 );
2135
2136 if emitted.is_empty() {
2137 return Err(ConvertError::NoData);
2138 }
2139 warn_plan_skipped_levels(&skipped, num_features, emitted[0].gsd, emitted[0].zoom);
2140
2141 let geom_name = geom_field.name().clone();
2145 let (source_schema, cluster_schema, out_schema) =
2146 super::stream::build_level_schemas(&input_schema, geom_idx, &geom_name, options);
2147
2148 let writer_levels: Vec<LevelSpec> = emitted
2149 .iter()
2150 .map(|e| LevelSpec::new(e.gsd, e.zoom))
2151 .collect();
2152 let emitted_gsds: Vec<f64> = emitted.iter().map(|e| e.gsd).collect();
2153 let writer_opts = super::stream::build_writer_options(
2154 writer_levels,
2155 &emitted_gsds,
2156 crs,
2157 ranking_provenance,
2158 &renames,
2159 options,
2160 );
2161
2162 let mut writer = OverviewWriter::create(output_path, &out_schema, writer_opts)?;
2163
2164 let non_geom_cols: Vec<usize> = (0..input_schema.fields().len())
2166 .filter(|&c| c != geom_idx)
2167 .collect();
2168
2169 let mut level_reports = write_emitted_levels(
2170 &mut writer,
2171 &emitted,
2172 &LevelWriteInputs {
2173 full: &full,
2174 source_schema: &source_schema,
2175 cluster_schema: &cluster_schema,
2176 out_schema: &out_schema,
2177 non_geom_cols: &non_geom_cols,
2178 geom_idx,
2179 cluster_tables: cluster_tables.as_ref(),
2180 acc_cols: &acc_cols,
2181 finest,
2182 },
2183 options,
2184 &mut skipped,
2185 )?;
2186 skipped.sort_by_key(|s| s.planned_level);
2187
2188 let meta = writer.finish()?;
2189
2190 fill_level_bytes(output_path, &meta, &mut level_reports)?;
2192
2193 let total_rows: usize = level_reports.iter().map(|l| l.feature_count).sum();
2194 let total_vertices: usize = level_reports.iter().map(|l| l.vertex_count).sum();
2195 let total_compressed_bytes: i64 = level_reports.iter().map(|l| l.compressed_bytes).sum();
2196
2197 Ok(ConvertReport {
2198 mode: options.mode,
2199 levels: level_reports,
2200 skipped_empty_levels: skipped,
2201 input_features: num_features,
2202 total_rows,
2203 total_vertices,
2204 total_compressed_bytes,
2205 row_groups_total,
2206 row_groups_read,
2207 antimeridian_suspect_features,
2208 duration_secs: start.elapsed().as_secs_f64(),
2209 remote_fetch: log_remote_fetch(source),
2210 })
2211}
2212
2213struct LevelWriteInputs<'a> {
2219 full: &'a RecordBatch,
2221 source_schema: &'a Schema,
2222 cluster_schema: &'a Schema,
2223 out_schema: &'a Schema,
2224 non_geom_cols: &'a [usize],
2225 geom_idx: usize,
2226 cluster_tables: Option<&'a ClusterTables>,
2228 acc_cols: &'a [usize],
2230 finest: usize,
2232}
2233
2234fn write_emitted_levels(
2241 writer: &mut OverviewWriter<File>,
2242 emitted: &[EmittedLevel],
2243 inputs: &LevelWriteInputs<'_>,
2244 options: &ConvertOptions,
2245 skipped: &mut Vec<SkippedLevelReport>,
2246) -> Result<Vec<LevelReport>, ConvertError> {
2247 let mut level_reports = Vec::with_capacity(emitted.len());
2248 for (level_idx, e) in emitted.iter().enumerate() {
2249 let mut batch = build_level_batch(
2250 inputs.source_schema,
2251 inputs.full,
2252 inputs.non_geom_cols,
2253 inputs.geom_idx,
2254 &e.indices,
2255 &e.geoms,
2256 )?;
2257 if let Some(tables) = inputs.cluster_tables {
2258 let table = (e.orig != inputs.finest).then(|| &tables[e.orig]);
2260 batch = apply_cluster_columns(
2261 batch,
2262 inputs.cluster_schema,
2263 &e.indices,
2264 table,
2265 inputs.acc_cols,
2266 )?;
2267 }
2268 if options.coalesce_lines {
2269 batch =
2271 apply_coalesced_count(batch, inputs.out_schema, &e.indices, e.coalesce.as_ref())?;
2272 }
2273 let outcome =
2276 writer.write_level(level_idx, Some(e.indices.len()), std::iter::once(batch))?;
2277 record_level_outcome(
2278 outcome,
2279 SkippedLevelReport {
2280 planned_level: e.orig,
2281 gsd: e.gsd,
2282 zoom: e.zoom,
2283 },
2284 e.indices.len(),
2285 e.indices.len(),
2286 e.vertex_count,
2287 &mut level_reports,
2288 skipped,
2289 );
2290 }
2291 Ok(level_reports)
2292}
2293
2294fn in_memory_carriers(
2298 options: &ConvertOptions,
2299 features: &[AssignFeature],
2300 row_min_levels: &[u8],
2301 geometries: &[Geometry<f64>],
2302 level_gsds: &[f64],
2303 level_reprs: &[Representation],
2304 crs: Crs,
2305) -> Vec<Vec<usize>> {
2306 let num_levels = level_gsds.len();
2307 let finest = num_levels.saturating_sub(1);
2308 let enabled = matches!(options.mode, Mode::Duplicating)
2309 && (options.simplify.collapse == CollapseMode::Square
2310 || level_reprs.contains(&Representation::Square));
2311 let acc_levels: Vec<AccumulateLevel> = level_gsds
2312 .iter()
2313 .enumerate()
2314 .map(|(l, &gsd)| AccumulateLevel {
2315 gsd_meters: gsd,
2316 enabled: enabled
2317 && l != finest
2318 && level_accumulates(options.simplify.collapse, level_reprs[l]),
2319 })
2320 .collect();
2321 if !acc_levels.iter().any(|l| l.enabled) {
2322 return vec![Vec::new(); num_levels];
2323 }
2324 let areas: Vec<f32> = geometries
2325 .iter()
2326 .map(|g| match g {
2327 Geometry::Polygon(p) => p.unsigned_area() as f32,
2328 Geometry::MultiPolygon(mp) => mp.unsigned_area() as f32,
2329 _ => 0.0,
2330 })
2331 .collect();
2332 tiny_polygon_carriers(
2333 features,
2334 row_min_levels,
2335 &areas,
2336 &acc_levels,
2337 crs,
2338 options.simplify.factor,
2339 )
2340}
2341
2342pub(super) fn log_remote_fetch(source: &ConvertSource) -> Option<crate::input::FetchStats> {
2346 let stats = source.fetch_stats()?;
2347 let pct = if stats.object_size > 0 {
2348 100.0 * stats.bytes_fetched as f64 / stats.object_size as f64
2349 } else {
2350 0.0
2351 };
2352 log::info!(
2353 "remote input: {} range requests, {:.2} MiB fetched of a {:.2} MiB object ({:.1}%)",
2354 stats.requests,
2355 stats.bytes_fetched as f64 / (1024.0 * 1024.0),
2356 stats.object_size as f64 / (1024.0 * 1024.0),
2357 pct
2358 );
2359 Some(stats)
2360}
2361
2362pub(crate) fn detect_crs_from_kv(
2370 kv: Option<&Vec<parquet::file::metadata::KeyValue>>,
2371) -> Result<Crs, ConvertError> {
2372 let info = crate::quality::crs_info_from_kv_metadata(kv)?;
2373 if info.is_wgs84 {
2374 return Ok(Crs::Epsg4326);
2375 }
2376 if let Some(id) = &info.identifier {
2377 let up = id.to_uppercase();
2378 if up.contains("3857") || up.contains("900913") {
2379 return Ok(Crs::Epsg3857);
2380 }
2381 }
2382 Err(ConvertError::UnsupportedCrs {
2383 crs: info
2384 .identifier
2385 .clone()
2386 .or_else(|| info.name.clone())
2387 .unwrap_or_else(|| "unknown".to_string()),
2388 })
2389}
2390
2391const WEBMERC_HALF_M: f64 = 20_037_508.342_789_244;
2394
2395const WEBMERC_MAX_LAT: f64 = 85.051_128_779_806_59;
2397
2398#[inline]
2402fn lnglat_to_webmerc(lng: f64, lat: f64) -> (f64, f64) {
2403 use std::f64::consts::{FRAC_PI_4, PI};
2404 let x = lng / 180.0 * WEBMERC_HALF_M;
2405 let lat = lat.clamp(-WEBMERC_MAX_LAT, WEBMERC_MAX_LAT);
2406 let y = (FRAC_PI_4 + lat.to_radians() / 2.0).tan().ln() / PI * WEBMERC_HALF_M;
2407 (x, y)
2408}
2409
2410pub(super) fn bbox_to_crs_units(bbox: &[f64; 4], crs: Crs) -> [f64; 4] {
2414 match crs {
2415 Crs::Epsg4326 => *bbox,
2416 Crs::Epsg3857 => {
2417 let (xmin, ymin) = lnglat_to_webmerc(bbox[0], bbox[1]);
2418 let (xmax, ymax) = lnglat_to_webmerc(bbox[2], bbox[3]);
2419 [xmin, ymin, xmax, ymax]
2420 }
2421 }
2422}
2423
2424pub(super) fn bboxes_intersect(a: &[f64; 4], b: &[f64; 4]) -> bool {
2428 a[0] <= b[2] && a[2] >= b[0] && a[1] <= b[3] && a[3] >= b[1]
2429}
2430
2431pub(super) fn encode_concurrency_for(profile: MemoryProfile) -> usize {
2437 const BOUNDED_ENCODE_CONCURRENCY_CAP: usize = 4;
2441
2442 let threads = rayon::current_num_threads().max(1);
2443 match profile {
2444 MemoryProfile::Bounded => threads.min(BOUNDED_ENCODE_CONCURRENCY_CAP),
2445 MemoryProfile::Speed | MemoryProfile::Auto => threads,
2446 }
2447}
2448
2449pub(crate) fn select_input_row_groups(
2458 metadata: &parquet::file::metadata::ParquetMetaData,
2459 bbox_units: &[f64; 4],
2460) -> Vec<usize> {
2461 let bounds = crate::covering::extract_row_group_bounds_from_metadata(metadata)
2462 .unwrap_or_else(|_| vec![None; metadata.num_row_groups()]);
2463 let filter = crate::tile::TileBounds {
2464 lng_min: bbox_units[0],
2465 lat_min: bbox_units[1],
2466 lng_max: bbox_units[2],
2467 lat_max: bbox_units[3],
2468 };
2469 (0..metadata.num_row_groups())
2470 .filter(|&i| match bounds.get(i).and_then(|b| b.as_ref()) {
2471 Some(b) => b.intersects(&filter),
2472 None => true, })
2474 .collect()
2475}
2476
2477pub(super) fn bind_attribute_filter(
2480 options: &ConvertOptions,
2481 input_schema: &Schema,
2482 renames: &[(String, String)],
2483) -> Result<Option<super::filter::BoundFilter>, ConvertError> {
2484 let Some(src) = options.filter.as_deref() else {
2485 return Ok(None);
2486 };
2487 let expr = super::filter::parse_filter(src)?;
2488 Ok(Some(super::filter::BoundFilter::bind(
2489 &expr,
2490 input_schema,
2491 renames,
2492 )?))
2493}
2494
2495pub(super) fn pruning_label(bbox: bool, filter: bool) -> &'static str {
2497 match (bbox, filter) {
2498 (true, true) => "bbox+attribute",
2499 (true, false) => "bbox",
2500 _ => "attribute",
2501 }
2502}
2503
2504fn select_input_row_groups_combined(
2508 metadata: &parquet::file::metadata::ParquetMetaData,
2509 bbox_units: Option<&[f64; 4]>,
2510 filter: Option<&super::filter::BoundFilter>,
2511) -> Option<Vec<usize>> {
2512 let bbox_sel: Option<Vec<usize>> = bbox_units.map(|bb| select_input_row_groups(metadata, bb));
2513 let filter_sel: Option<Vec<usize>> = filter.map(|f| f.select_row_groups(metadata));
2514 match (bbox_sel, filter_sel) {
2515 (Some(a), Some(b)) => Some(a.into_iter().filter(|i| b.contains(i)).collect()),
2516 (Some(a), None) => Some(a),
2517 (None, Some(b)) => Some(b),
2518 (None, None) => None,
2519 }
2520}
2521
2522pub(super) fn find_geometry_column(schema: &Schema) -> Option<usize> {
2524 schema
2525 .fields()
2526 .iter()
2527 .position(|f| f.name() == "geometry")
2528 .or_else(|| {
2529 schema
2530 .fields()
2531 .iter()
2532 .position(|f| f.name().contains("geom"))
2533 })
2534}
2535
2536pub(super) fn geometry_bbox(g: &Geometry<f64>) -> [f64; 4] {
2538 match g.bounding_rect() {
2539 Some(r) => [r.min().x, r.min().y, r.max().x, r.max().y],
2540 None => [0.0, 0.0, 0.0, 0.0],
2541 }
2542}
2543
2544pub(super) fn bbox_antimeridian_suspect(bbox: &[f64; 4], crs: Crs) -> bool {
2550 bbox[2] - bbox[0] > crs.meters_to_units(180.0 * METERS_PER_DEGREE)
2551}
2552
2553pub(super) fn warn_antimeridian_suspects(count: usize) {
2556 if count > 0 {
2557 log::warn!(
2558 "{count} feature(s) have bounding boxes wider than 180° of longitude — \
2559 likely antimeridian-crossing geometry. These will be assigned to \
2560 overly coarse levels and defeat bbox pruning; pre-split them at \
2561 ±180° before converting (see docs/advanced-usage.md, \
2562 \"Antimeridian-Crossing Geometry\")."
2563 );
2564 }
2565}
2566
2567pub(super) const FULL_FILE_REMOTE_WARN_BYTES: u64 = 1 << 30;
2571
2572pub(super) fn full_file_remote_warning(
2584 remote_parts: usize,
2585 row_groups_read: usize,
2586 row_groups_total: usize,
2587 object_size: u64,
2588) -> Option<String> {
2589 if remote_parts == 0
2590 || row_groups_read < row_groups_total
2591 || object_size < FULL_FILE_REMOTE_WARN_BYTES
2592 {
2593 return None;
2594 }
2595 let gib = object_size as f64 / (1024.0 * 1024.0 * 1024.0);
2596 let what = if remote_parts > 1 {
2600 format!("{remote_parts} remote partitions totalling {gib:.1} GiB")
2601 } else {
2602 format!("a {gib:.1} GiB object")
2603 };
2604 Some(format!(
2605 "full-file remote convert of {what}: the input is fetched once over \
2606 the network (≈1× — the local spill keeps later passes off the network, \
2607 #219) and staged under $TMPDIR. For a region of interest pass --bbox to \
2608 fetch only the covering row groups (and skip the spill); otherwise \
2609 downloading first (e.g. `aws s3 cp`) and converting locally avoids the \
2610 second-pass disk read. Point --spill-dir (or $TMPDIR) at fast local disk \
2611 with room for it, not a small tmpfs.",
2612 ))
2613}
2614
2615pub(super) fn warn_full_file_remote(
2620 source: &ConvertSource,
2621 row_groups_read: usize,
2622 row_groups_total: usize,
2623) {
2624 let object_size = source.fetch_stats().map_or(0, |s| s.object_size);
2625 let remote_parts = source.parts().iter().filter(|p| p.is_remote()).count();
2626 if let Some(msg) =
2627 full_file_remote_warning(remote_parts, row_groups_read, row_groups_total, object_size)
2628 {
2629 log::warn!("{msg}");
2630 }
2631}
2632
2633const SPILL_MARGIN_DENOM: u64 = 20;
2638
2639pub(super) fn spill_space_warning(
2645 estimated_spill_bytes: u64,
2646 available_bytes: u64,
2647 spill_dir: &Path,
2648) -> Option<String> {
2649 let need = estimated_spill_bytes + estimated_spill_bytes / SPILL_MARGIN_DENOM;
2650 if available_bytes >= need {
2651 return None;
2652 }
2653 let gib = |b: u64| b as f64 / (1024.0 * 1024.0 * 1024.0);
2654 Some(format!(
2655 "projected input spill (≈{:.1} GiB — the selected input bytes are \
2656 staged on local disk so later passes stay off the network, #219) may \
2657 not fit: {} has {:.1} GiB free ({:.1} GiB short, including a 5% \
2658 margin). If the volume fills mid-convert the spill degrades to \
2659 network re-fetch; pass --spill-dir (spill_dir) to place it on a \
2660 roomier volume, or free up space first.",
2661 gib(estimated_spill_bytes),
2662 spill_dir.display(),
2663 gib(available_bytes),
2664 gib(need - available_bytes),
2665 ))
2666}
2667
2668pub(super) fn spill_space_check(
2674 is_remote: bool,
2675 estimated_spill_bytes: u64,
2676 spill_dir: &Path,
2677 probe: impl FnOnce(&Path) -> Option<u64>,
2678) -> Option<String> {
2679 if !is_remote || estimated_spill_bytes == 0 {
2680 return None;
2681 }
2682 let available = probe(spill_dir)?;
2683 spill_space_warning(estimated_spill_bytes, available, spill_dir)
2684}
2685
2686fn probe_available_space(dir: &Path) -> Option<u64> {
2691 #[cfg(feature = "remote")]
2692 {
2693 fs4::available_space(dir).ok()
2694 }
2695 #[cfg(not(feature = "remote"))]
2696 {
2697 let _ = dir;
2698 None
2699 }
2700}
2701
2702pub(super) fn warn_spill_space(
2707 source: &ConvertSource,
2708 estimated_spill_bytes: u64,
2709 spill_dir: Option<&Path>,
2710) {
2711 let dir = spill_dir.map_or_else(std::env::temp_dir, Path::to_path_buf);
2712 if let Some(msg) = spill_space_check(
2713 source.is_remote(),
2714 estimated_spill_bytes,
2715 &dir,
2716 probe_available_space,
2717 ) {
2718 log::warn!("{msg}");
2719 }
2720}
2721
2722pub(super) fn usable_geometry(g: &Geometry<f64>) -> bool {
2730 use geo::coords_iter::CoordsIter;
2731 let mut any = false;
2732 for c in g.coords_iter() {
2733 if !c.x.is_finite() || !c.y.is_finite() {
2734 return false;
2735 }
2736 any = true;
2737 }
2738 any
2739}
2740
2741pub(super) fn feature_kind(g: &Geometry<f64>) -> FeatureKind {
2743 match g {
2744 Geometry::Point(_) | Geometry::MultiPoint(_) => FeatureKind::Point,
2745 Geometry::LineString(_) | Geometry::MultiLineString(_) | Geometry::Line(_) => {
2746 FeatureKind::Line
2747 }
2748 _ => FeatureKind::Polygon,
2749 }
2750}
2751
2752#[derive(Debug)]
2755struct FeatureScan {
2756 min_x: f64,
2757 min_y: f64,
2758 max_x: f64,
2759 max_y: f64,
2760 bbox_seen: bool,
2763 any_coord: bool,
2765 finite: bool,
2767}
2768
2769impl FeatureScan {
2770 fn new() -> Self {
2771 FeatureScan {
2772 min_x: f64::INFINITY,
2773 min_y: f64::INFINITY,
2774 max_x: f64::NEG_INFINITY,
2775 max_y: f64::NEG_INFINITY,
2776 bbox_seen: false,
2777 any_coord: false,
2778 finite: true,
2779 }
2780 }
2781
2782 #[inline]
2786 fn note_bbox(&mut self, c: geo::Coord<f64>) {
2787 self.any_coord = true;
2788 if !c.x.is_finite() || !c.y.is_finite() {
2789 self.finite = false;
2790 return;
2791 }
2792 self.bbox_seen = true;
2793 if c.x < self.min_x {
2794 self.min_x = c.x;
2795 }
2796 if c.y < self.min_y {
2797 self.min_y = c.y;
2798 }
2799 if c.x > self.max_x {
2800 self.max_x = c.x;
2801 }
2802 if c.y > self.max_y {
2803 self.max_y = c.y;
2804 }
2805 }
2806
2807 #[inline]
2811 fn note_finite_only(&mut self, c: geo::Coord<f64>) {
2812 self.any_coord = true;
2813 if !c.x.is_finite() || !c.y.is_finite() {
2814 self.finite = false;
2815 }
2816 }
2817
2818 fn bbox(&self) -> [f64; 4] {
2819 if self.bbox_seen {
2820 [self.min_x, self.min_y, self.max_x, self.max_y]
2821 } else {
2822 [0.0, 0.0, 0.0, 0.0]
2824 }
2825 }
2826}
2827
2828fn scan_geometry_into(g: &Geometry<f64>, scan: &mut FeatureScan) {
2833 use geo::coords_iter::CoordsIter;
2834 let scan_polygon = |poly: &geo::Polygon<f64>, scan: &mut FeatureScan| {
2835 for c in poly.exterior().coords_iter() {
2836 scan.note_bbox(c);
2837 }
2838 for ring in poly.interiors() {
2839 for c in ring.coords_iter() {
2840 scan.note_finite_only(c);
2841 }
2842 }
2843 };
2844 match g {
2845 Geometry::Polygon(poly) => scan_polygon(poly, scan),
2846 Geometry::MultiPolygon(mp) => {
2847 for poly in &mp.0 {
2848 scan_polygon(poly, scan);
2849 }
2850 }
2851 Geometry::GeometryCollection(gc) => {
2852 for child in &gc.0 {
2853 scan_geometry_into(child, scan);
2854 }
2855 }
2856 other => {
2858 for c in other.coords_iter() {
2859 scan.note_bbox(c);
2860 }
2861 }
2862 }
2863}
2864
2865pub(super) fn scan_feature(g: &Geometry<f64>) -> Option<(FeatureKind, [f64; 4])> {
2876 let mut scan = FeatureScan::new();
2877 scan_geometry_into(g, &mut scan);
2878 if !scan.finite || !scan.any_coord {
2879 return None;
2880 }
2881 Some((feature_kind(g), scan.bbox()))
2882}
2883
2884pub(super) fn count_vertices(g: &Geometry<f64>) -> usize {
2886 use geo::coords_iter::CoordsIter;
2887 g.coords_count()
2888}
2889
2890pub(super) fn extract_sort_keys(col: &dyn Array) -> Vec<Option<f64>> {
2893 use arrow_array::cast::AsArray;
2894 use arrow_array::types::{
2895 Float32Type, Float64Type, Int16Type, Int32Type, Int64Type, Int8Type, UInt16Type,
2896 UInt32Type, UInt64Type, UInt8Type,
2897 };
2898 use arrow_schema::DataType;
2899
2900 let n = col.len();
2901 macro_rules! collect_prim {
2902 ($ty:ty) => {{
2903 let a = col.as_primitive::<$ty>();
2904 (0..n)
2905 .map(|i| {
2906 if a.is_null(i) {
2907 None
2908 } else {
2909 Some(a.value(i) as f64)
2910 }
2911 })
2912 .collect()
2913 }};
2914 }
2915 match col.data_type() {
2916 DataType::Int8 => collect_prim!(Int8Type),
2917 DataType::Int16 => collect_prim!(Int16Type),
2918 DataType::Int32 => collect_prim!(Int32Type),
2919 DataType::Int64 => collect_prim!(Int64Type),
2920 DataType::UInt8 => collect_prim!(UInt8Type),
2921 DataType::UInt16 => collect_prim!(UInt16Type),
2922 DataType::UInt32 => collect_prim!(UInt32Type),
2923 DataType::UInt64 => collect_prim!(UInt64Type),
2924 DataType::Float32 => collect_prim!(Float32Type),
2925 DataType::Float64 => collect_prim!(Float64Type),
2926 _ => vec![None; n],
2927 }
2928}
2929
2930pub(super) fn mixed_geometry_field(name: &str) -> Arc<Field> {
2932 use geoarrow_array::GeoArrowArray;
2933 let typ = GeometryType::new(Default::default());
2934 let empty = GeometryBuilder::new(typ).with_prefer_multi(false).finish();
2935 Arc::new(empty.data_type().to_field(name, true))
2936}
2937
2938pub(super) fn build_source_schema(
2941 input_schema: &Schema,
2942 geom_idx: usize,
2943 geom_out_field: Arc<Field>,
2944) -> Schema {
2945 let fields: Vec<Arc<Field>> = input_schema
2946 .fields()
2947 .iter()
2948 .enumerate()
2949 .map(|(i, f)| {
2950 if i == geom_idx {
2951 geom_out_field.clone()
2952 } else {
2953 f.clone()
2954 }
2955 })
2956 .collect();
2957 Schema::new(fields)
2958}
2959
2960pub(super) fn build_level_batch(
2963 source_schema: &Schema,
2964 full: &RecordBatch,
2965 non_geom_cols: &[usize],
2966 geom_idx: usize,
2967 indices: &[usize],
2968 geoms: &[Geometry<f64>],
2969) -> Result<RecordBatch, ConvertError> {
2970 let take_idx = UInt32Array::from(indices.iter().map(|&i| i as u32).collect::<Vec<_>>());
2971
2972 let mut columns: Vec<Arc<dyn Array>> = Vec::with_capacity(source_schema.fields().len());
2974 let mut non_geom_iter = non_geom_cols.iter();
2975 for i in 0..source_schema.fields().len() {
2976 if i == geom_idx {
2977 let typ = GeometryType::new(Default::default());
2978 let mut b = GeometryBuilder::new(typ).with_prefer_multi(false);
2979 b.extend_from_iter(geoms.iter().map(Some));
2980 columns.push(b.finish().to_array_ref());
2981 } else {
2982 let src_col = *non_geom_iter.next().expect("non-geom column index");
2983 let taken = take(full.column(src_col).as_ref(), &take_idx, None)?;
2984 columns.push(taken);
2985 }
2986 }
2987 Ok(RecordBatch::try_new(
2988 Arc::new(source_schema.clone()),
2989 columns,
2990 )?)
2991}
2992
2993pub(super) fn validate_cluster_schema(
3005 schema: &Schema,
3006 options: &ConvertOptions,
3007) -> Result<Vec<usize>, ConvertError> {
3008 if !options.cluster {
3009 return Ok(Vec::new());
3010 }
3011 if schema
3015 .fields()
3016 .iter()
3017 .any(|f| f.name().eq_ignore_ascii_case(POINT_COUNT_COLUMN))
3018 {
3019 return Err(ConvertError::PointCountColumnPresent);
3020 }
3021 let mut indices = Vec::with_capacity(options.accumulate.len());
3022 for spec in &options.accumulate {
3023 let idx =
3024 schema
3025 .index_of(&spec.column)
3026 .map_err(|_| ConvertError::AccumulateColumnMissing {
3027 name: spec.column.clone(),
3028 })?;
3029 let dt = schema.field(idx).data_type();
3030 if !is_numeric_type(dt) {
3031 return Err(ConvertError::AccumulateColumnNotNumeric {
3032 name: spec.column.clone(),
3033 data_type: format!("{dt:?}"),
3034 });
3035 }
3036 indices.push(idx);
3037 }
3038 Ok(indices)
3039}
3040
3041fn is_numeric_type(dt: &DataType) -> bool {
3043 matches!(
3044 dt,
3045 DataType::Int8
3046 | DataType::Int16
3047 | DataType::Int32
3048 | DataType::Int64
3049 | DataType::UInt8
3050 | DataType::UInt16
3051 | DataType::UInt32
3052 | DataType::UInt64
3053 | DataType::Float32
3054 | DataType::Float64
3055 )
3056}
3057
3058pub(super) fn append_point_count_field(schema: &Schema) -> Schema {
3061 let mut fields: Vec<Arc<Field>> = schema.fields().iter().cloned().collect();
3062 fields.push(Arc::new(Field::new(
3063 POINT_COUNT_COLUMN,
3064 DataType::Int64,
3065 false,
3066 )));
3067 Schema::new(fields)
3068}
3069
3070pub(super) fn extract_accumulate_values(
3073 batch: &RecordBatch,
3074 acc_col_indices: &[usize],
3075) -> Vec<Vec<Option<f64>>> {
3076 acc_col_indices
3077 .iter()
3078 .map(|&idx| extract_sort_keys(batch.column(idx).as_ref()))
3079 .collect()
3080}
3081
3082pub(super) fn apply_cluster_columns(
3099 batch: RecordBatch,
3100 out_schema: &Schema,
3101 global_indices: &[usize],
3102 table: Option<&HashMap<usize, ClusterEntry>>,
3103 acc_cols: &[usize],
3104) -> Result<RecordBatch, ConvertError> {
3105 use arrow_array::Int64Array;
3106
3107 debug_assert_eq!(batch.num_rows(), global_indices.len());
3108 let mut columns: Vec<Arc<dyn Array>> = batch.columns().to_vec();
3109
3110 if let Some(table) = table {
3112 for (s, &col_idx) in acc_cols.iter().enumerate() {
3113 let overrides: Vec<Option<f64>> = global_indices
3114 .iter()
3115 .map(|g| table.get(g).and_then(|e| e.aggregates[s]))
3116 .collect();
3117 if overrides.iter().any(|o| o.is_some()) {
3118 columns[col_idx] = overwrite_numeric_column(&columns[col_idx], &overrides)?;
3119 }
3120 }
3121 }
3122
3123 let counts: Vec<i64> = global_indices
3125 .iter()
3126 .map(|g| table.and_then(|t| t.get(g)).map_or(1, |e| e.point_count))
3127 .collect();
3128 columns.push(Arc::new(Int64Array::from(counts)));
3129
3130 Ok(RecordBatch::try_new(Arc::new(out_schema.clone()), columns)?)
3131}
3132
3133fn overwrite_numeric_column(
3137 col: &Arc<dyn Array>,
3138 overrides: &[Option<f64>],
3139) -> Result<Arc<dyn Array>, ConvertError> {
3140 use arrow_array::cast::AsArray;
3141 use arrow_array::types::{
3142 Float32Type, Float64Type, Int16Type, Int32Type, Int64Type, Int8Type, UInt16Type,
3143 UInt32Type, UInt64Type, UInt8Type,
3144 };
3145 use arrow_array::PrimitiveArray;
3146
3147 macro_rules! rebuild {
3148 ($ty:ty, $cast:expr) => {{
3149 let a = col.as_primitive::<$ty>();
3150 let rebuilt: PrimitiveArray<$ty> = (0..a.len())
3151 .map(|i| match overrides[i] {
3152 Some(v) => Some($cast(v)),
3153 None => {
3154 if a.is_null(i) {
3155 None
3156 } else {
3157 Some(a.value(i))
3158 }
3159 }
3160 })
3161 .collect();
3162 Ok(Arc::new(rebuilt))
3163 }};
3164 }
3165 match col.data_type() {
3166 DataType::Int8 => rebuild!(Int8Type, |v: f64| v.round() as i8),
3167 DataType::Int16 => rebuild!(Int16Type, |v: f64| v.round() as i16),
3168 DataType::Int32 => rebuild!(Int32Type, |v: f64| v.round() as i32),
3169 DataType::Int64 => rebuild!(Int64Type, |v: f64| v.round() as i64),
3170 DataType::UInt8 => rebuild!(UInt8Type, |v: f64| v.round() as u8),
3171 DataType::UInt16 => rebuild!(UInt16Type, |v: f64| v.round() as u16),
3172 DataType::UInt32 => rebuild!(UInt32Type, |v: f64| v.round() as u32),
3173 DataType::UInt64 => rebuild!(UInt64Type, |v: f64| v.round() as u64),
3174 DataType::Float32 => rebuild!(Float32Type, |v: f64| v as f32),
3175 DataType::Float64 => rebuild!(Float64Type, |v: f64| v),
3176 other => Err(ConvertError::AccumulateColumnNotNumeric {
3177 name: "<accumulate column>".to_string(),
3178 data_type: format!("{other:?}"),
3179 }),
3180 }
3181}
3182
3183pub(super) type CoalesceTable = HashMap<usize, (Geometry<f64>, i32)>;
3190
3191pub(super) fn validate_coalesce_schema(
3195 schema: &Schema,
3196 options: &ConvertOptions,
3197) -> Result<(), ConvertError> {
3198 if !options.coalesce_lines {
3199 return Ok(());
3200 }
3201 if schema
3206 .fields()
3207 .iter()
3208 .any(|f| f.name().eq_ignore_ascii_case(COALESCED_COUNT_COLUMN))
3209 {
3210 return Err(ConvertError::CoalescedCountColumnPresent);
3211 }
3212 Ok(())
3213}
3214
3215fn reserved_output_columns(options: &ConvertOptions) -> Vec<&'static str> {
3225 let mut names = vec![LEVEL_COLUMN];
3226 if options.cluster {
3227 names.push(POINT_COUNT_COLUMN);
3228 }
3229 if options.coalesce_lines {
3230 names.push(COALESCED_COUNT_COLUMN);
3231 }
3232 names
3233}
3234
3235pub(super) fn resolve_reserved_column_collisions(
3255 input_schema: &SchemaRef,
3256 options: &mut ConvertOptions,
3257) -> (SchemaRef, Vec<(String, String)>) {
3258 let reserved = reserved_output_columns(options);
3259 let match_reserved = |name: &str| -> Option<&'static str> {
3260 reserved
3261 .iter()
3262 .copied()
3263 .find(|r| name.eq_ignore_ascii_case(r))
3264 };
3265
3266 let mut taken: HashSet<String> = input_schema
3267 .fields()
3268 .iter()
3269 .map(|f| f.name().to_ascii_lowercase())
3270 .collect();
3271 let mut renames: Vec<(String, String)> = Vec::new();
3272 let mut new_fields: Vec<Arc<Field>> = Vec::with_capacity(input_schema.fields().len());
3273
3274 for field in input_schema.fields() {
3275 let name = field.name();
3276 let Some(reserved_name) = match_reserved(name) else {
3277 new_fields.push(field.clone());
3278 continue;
3279 };
3280 let mut candidate = format!("{name}_");
3284 while taken.contains(&candidate.to_ascii_lowercase())
3285 || match_reserved(&candidate).is_some()
3286 {
3287 candidate.push('_');
3288 }
3289 taken.insert(candidate.to_ascii_lowercase());
3290 log::warn!(
3291 "input column {name:?} collides with the reserved overview column \
3292 {reserved_name:?}; renaming the input column to {candidate:?} in \
3293 the output (the reserved {reserved_name:?} column is authoritative)"
3294 );
3295 renames.push((name.to_string(), candidate.clone()));
3296 new_fields.push(Arc::new(
3297 Field::new(candidate, field.data_type().clone(), field.is_nullable())
3298 .with_metadata(field.metadata().clone()),
3299 ));
3300 }
3301
3302 if renames.is_empty() {
3303 return (input_schema.clone(), renames);
3304 }
3305
3306 let remap = |col: &str| -> Option<String> {
3310 renames
3311 .iter()
3312 .find(|(old, _)| col.eq_ignore_ascii_case(old))
3313 .map(|(_, new)| new.clone())
3314 };
3315 if let Some(new) = options.sort_key.as_deref().and_then(remap) {
3316 options.sort_key = Some(new);
3317 }
3318 if let Some(cr) = options.class_ranking.as_mut() {
3319 if let Some(new) = remap(&cr.column) {
3320 cr.column = new;
3321 }
3322 }
3323 for spec in &mut options.accumulate {
3324 if let Some(new) = remap(&spec.column) {
3325 spec.column = new;
3326 }
3327 }
3328 if let Some(spec) = options.entry_zoom.as_mut() {
3332 if let Some(new) = remap(&spec.column) {
3333 spec.column = new;
3334 }
3335 }
3336
3337 let schema = Arc::new(Schema::new_with_metadata(
3338 new_fields,
3339 input_schema.metadata().clone(),
3340 ));
3341 (schema, renames)
3342}
3343
3344pub(super) fn append_coalesced_count_field(schema: &Schema) -> Schema {
3347 let mut fields: Vec<Arc<Field>> = schema.fields().iter().cloned().collect();
3348 fields.push(Arc::new(Field::new(
3349 COALESCED_COUNT_COLUMN,
3350 DataType::Int32,
3351 false,
3352 )));
3353 Schema::new(fields)
3354}
3355
3356pub(super) fn apply_coalesced_count(
3360 batch: RecordBatch,
3361 out_schema: &Schema,
3362 global_indices: &[usize],
3363 table: Option<&CoalesceTable>,
3364) -> Result<RecordBatch, ConvertError> {
3365 use arrow_array::Int32Array;
3366
3367 debug_assert_eq!(batch.num_rows(), global_indices.len());
3368 let mut columns: Vec<Arc<dyn Array>> = batch.columns().to_vec();
3369 let counts: Vec<i32> = global_indices
3370 .iter()
3371 .map(|g| table.and_then(|t| t.get(g)).map_or(1, |(_, c)| *c))
3372 .collect();
3373 columns.push(Arc::new(Int32Array::from(counts)));
3374 Ok(RecordBatch::try_new(Arc::new(out_schema.clone()), columns)?)
3375}
3376
3377pub(super) fn coalesce_group_column(ranking: &RankingProvenance) -> Option<&str> {
3382 match ranking.mode.as_str() {
3383 "class-ranking" | "auto-overture-roads" => ranking.column.as_deref(),
3384 _ => None,
3385 }
3386}
3387
3388#[derive(Debug, Default)]
3392pub(super) struct GroupInterner {
3393 map: HashMap<String, u32>,
3394}
3395
3396impl GroupInterner {
3397 pub(super) const NULL_GROUP: u32 = u32::MAX;
3399
3400 pub(super) fn extend(&mut self, col: &dyn Array, out: &mut Vec<u32>) {
3404 use arrow_array::cast::AsArray;
3405
3406 macro_rules! intern {
3407 ($arr:expr) => {{
3408 let a = $arr;
3409 for i in 0..a.len() {
3410 if a.is_null(i) {
3411 out.push(Self::NULL_GROUP);
3412 } else {
3413 let next = self.map.len() as u32;
3414 let id = *self.map.entry(a.value(i).to_string()).or_insert(next);
3415 out.push(id);
3416 }
3417 }
3418 }};
3419 }
3420 match col.data_type() {
3421 DataType::Utf8 => intern!(col.as_string::<i32>()),
3422 DataType::LargeUtf8 => intern!(col.as_string::<i64>()),
3423 _ => out.extend(std::iter::repeat_n(Self::NULL_GROUP, col.len())),
3424 }
3425 }
3426}
3427
3428pub(super) fn coalesce_level_chains(
3437 inputs: &[CoalesceInput<'_>],
3438 level: usize,
3439 finest: usize,
3440 gsd_m: f64,
3441 crs: Crs,
3442 options: &ConvertOptions,
3443) -> Vec<super::coalesce::CoalescedLine> {
3444 let budget = if options.density.enabled
3445 && options.density.drop_rate > 1.0
3446 && !options.density.drop_rate.is_nan()
3447 && level < finest
3448 {
3449 let keep = 1.0 / options.density.drop_rate;
3450 let raw = inputs.len() as f64 * keep.powi((finest - level) as i32);
3451 let max_chains = (raw.round() as usize).max(super::assign::MIN_DENSITY_LEVEL_FEATURES);
3452 Some((max_chains, options.density.gamma))
3453 } else {
3454 None
3455 };
3456 coalesce_level_lines(
3457 inputs,
3458 gsd_m,
3459 crs,
3460 &options.assign,
3461 &CoalesceParams {
3462 snap_gsd_factor: options.coalesce_snap,
3463 junction_angle_deg: options.coalesce_junction_angle,
3464 budget,
3465 },
3466 )
3467}
3468
3469pub(super) fn build_level_coalesce_table(
3475 inputs: &[CoalesceInput<'_>],
3476 level: usize,
3477 finest: usize,
3478 gsd_m: f64,
3479 crs: Crs,
3480 options: &ConvertOptions,
3481) -> CoalesceTable {
3482 let chains = coalesce_level_chains(inputs, level, finest, gsd_m, crs, options);
3483 let mut table = CoalesceTable::with_capacity(chains.len());
3484 for chain in chains {
3485 match simplify_for_level(&chain.geom, gsd_m, crs, &options.simplify) {
3486 Simplified::Keep(g) => {
3487 table.insert(chain.rep, (g, chain.count));
3488 }
3489 Simplified::Dropped => {}
3490 }
3491 }
3492 table
3493}
3494
3495pub(super) fn coalesce_effective(options: &ConvertOptions, num_lines: usize) -> bool {
3500 if !options.coalesce_lines {
3501 return false;
3502 }
3503 if num_lines > options.coalesce_max_level_rows {
3504 log::warn!(
3505 "coalescing skipped: {num_lines} candidate lines exceed \
3506 --coalesce-max-level-rows {} (chaining holds a level's line \
3507 geometries in memory; near-canonical levels this large need \
3508 coalescing least). Output keeps the coalesced_count column \
3509 (all 1).",
3510 options.coalesce_max_level_rows
3511 );
3512 return false;
3513 }
3514 true
3515}
3516
3517pub(super) fn resolve_entry_levels(
3525 options: &ConvertOptions,
3526 column_values: &[Option<f64>],
3527 level_specs: &[(f64, Option<u8>)],
3528) -> Result<Option<Vec<Option<u8>>>, ConvertError> {
3529 let Some(spec) = &options.entry_zoom else {
3530 return Ok(None);
3531 };
3532 let level_zooms: Vec<Option<u8>> = level_specs.iter().map(|(_, z)| *z).collect();
3533 let ladder = build_ladder(spec, column_values, &level_zooms)
3534 .map_err(|e| ConvertError::InvalidConfig(format!("entry-zoom: {e}")))?;
3535 let levels = entry_levels(&ladder, column_values, &level_zooms);
3536 let placed = levels.iter().filter(|l| l.is_some()).count();
3537 log::info!(
3538 "[assign] entry-zoom ladder on {:?}: {} rung(s) {:?}; {placed} of {} feature(s) placed",
3539 spec.column,
3540 ladder.len(),
3541 ladder.to_map(),
3542 column_values.len(),
3543 );
3544 if matches!(options.simplify.collapse, CollapseMode::Drop) && options.simplify.factor > 0.0 {
3552 log::warn!(
3553 "[assign] entry-zoom ladder with --collapse off: a laddered feature \
3554 admitted to a coarse level is still DROPPED there if its geometry \
3555 simplifies below that level's tolerance, which is the usual case for \
3556 the small-but-strong features a ladder exists to promote. Pass \
3557 --collapse (representative point) or --collapse-square to keep them, \
3558 or --simplify-factor 0 to disable simplification."
3559 );
3560 }
3561 if options.density.enabled && options.sort_key.is_none() {
3566 log::warn!(
3567 "[assign] entry-zoom ladder with the density budget on and no \
3568 --sort-key: the budget caps each level and sheds its lowest-priority \
3569 survivors by SIZE, which can drop the small-but-strong features the \
3570 ladder just promoted. Pass --no-density-drop, or --sort-key on the \
3571 same column so the budget ranks the way the ladder does."
3572 );
3573 }
3574 Ok(Some(levels))
3575}
3576
3577pub(super) fn entry_zoom_column_values(
3579 options: &ConvertOptions,
3580 schema: &Schema,
3581 table: &RecordBatch,
3582) -> Result<Vec<Option<f64>>, ConvertError> {
3583 let Some(spec) = &options.entry_zoom else {
3584 return Ok(Vec::new());
3585 };
3586 let idx = schema.index_of(&spec.column).map_err(|_| {
3587 ConvertError::InvalidConfig(format!(
3588 "entry-zoom column {:?} not found in the input schema",
3589 spec.column
3590 ))
3591 })?;
3592 Ok(extract_sort_keys(table.column(idx).as_ref()))
3593}
3594
3595pub(super) fn build_generalization(
3597 gsds: &[f64],
3598 _crs: Crs,
3599 options: &ConvertOptions,
3600 ranking: RankingProvenance,
3601 renames: &[(String, String)],
3602) -> Generalization {
3603 let levels = gsds
3604 .iter()
3605 .map(|&gsd_m| GeneralizationLevel {
3606 simplify_tolerance_m: match options.mode {
3607 Mode::Duplicating => options.simplify.factor * gsd_m,
3608 Mode::Partitioning => 0.0,
3609 },
3610 thinning_factor: options.assign.polygon_thinning,
3611 visibility_gate_m: options.assign.polygon_visibility * gsd_m,
3612 geometry_types: Vec::new(),
3613 })
3614 .collect();
3615 Generalization {
3616 engine: format!("tylertoo {}", env!("CARGO_PKG_VERSION")),
3617 gsd_base: if options.gsd_base == GSD_TILE_BASE {
3621 None
3622 } else {
3623 Some(options.gsd_base)
3624 },
3625 levels,
3626 cascade: if matches!(options.mode, Mode::Duplicating) && options.simplify.cascade {
3631 Some(true)
3632 } else {
3633 None
3634 },
3635 collapse: match options.simplify.collapse {
3640 CollapseMode::Drop => None,
3641 mode => Some(
3642 match mode {
3643 CollapseMode::Point => "point",
3644 CollapseMode::Square => "square",
3645 CollapseMode::Drop => unreachable!(),
3646 }
3647 .to_string(),
3648 ),
3649 },
3650 representation: if options.representation.is_empty() {
3654 None
3655 } else {
3656 Some(
3657 options
3658 .representation
3659 .iter()
3660 .map(|b| RepresentationBandProvenance {
3661 zooms: [b.min_zoom, b.max_zoom],
3662 repr: b.repr.as_str().to_string(),
3663 })
3664 .collect(),
3665 )
3666 },
3667 ranking: Some(ranking),
3668 density_drop: if options.density.enabled {
3671 Some(DensityProvenance {
3672 drop_rate: options.density.drop_rate,
3673 gamma: options.density.gamma,
3674 supercell_gsd_factor: SUPERCELL_GSD_FACTOR,
3675 })
3676 } else {
3677 None
3678 },
3679 coalescing: if options.coalesce_lines {
3682 Some(CoalescingProvenance {
3683 enabled: true,
3684 snap_tolerance_gsd_factor: options.coalesce_snap,
3685 junction_angle: Some(options.coalesce_junction_angle),
3689 max_level_rows: Some(options.coalesce_max_level_rows as u64),
3690 coalesced_count_column: COALESCED_COUNT_COLUMN.to_string(),
3691 })
3692 } else {
3693 None
3694 },
3695 clustering: if options.cluster {
3698 Some(ClusteringProvenance {
3699 enabled: true,
3700 point_count_column: POINT_COUNT_COLUMN.to_string(),
3701 accumulated: options
3702 .accumulate
3703 .iter()
3704 .map(|s| AccumulatedColumn {
3705 column: s.column.clone(),
3706 op: s.op.as_str().to_string(),
3707 })
3708 .collect(),
3709 })
3710 } else {
3711 None
3712 },
3713 renamed_columns: if renames.is_empty() {
3718 None
3719 } else {
3720 Some(
3721 renames
3722 .iter()
3723 .map(|(old, new)| (new.clone(), old.clone()))
3724 .collect(),
3725 )
3726 },
3727 }
3728}
3729
3730fn resolve_ranking(
3742 input_schema: &Schema,
3743 full: &RecordBatch,
3744 geometries: &[Geometry<f64>],
3745 options: &ConvertOptions,
3746) -> Result<(Vec<Option<f64>>, RankingProvenance), ConvertError> {
3747 let n = full.num_rows();
3748
3749 if let Some(name) = &options.sort_key {
3751 let idx = input_schema
3752 .index_of(name)
3753 .map_err(|_| ConvertError::SortKeyColumnMissing { name: name.clone() })?;
3754 let keys = extract_sort_keys(full.column(idx));
3755 log::info!("overview ranking: explicit numeric sort-key column {name:?}");
3756 return Ok((
3757 keys,
3758 RankingProvenance {
3759 mode: "explicit-sort-key".to_string(),
3760 column: Some(name.clone()),
3761 ranks: None,
3762 unknown_rank: None,
3763 },
3764 ));
3765 }
3766
3767 if let Some(cr) = &options.class_ranking {
3769 let idx = input_schema.index_of(&cr.column).map_err(|_| {
3770 ConvertError::ClassRankColumnMissing {
3771 name: cr.column.clone(),
3772 }
3773 })?;
3774 let keys = extract_class_ranks(full.column(idx), cr)?;
3775 log::info!(
3776 "overview ranking: explicit class-ranking on column {:?} ({} named classes, unknown_rank={})",
3777 cr.column,
3778 cr.ranks.len(),
3779 cr.unknown_rank
3780 );
3781 return Ok((keys, class_ranking_provenance("class-ranking", cr)));
3782 }
3783
3784 if !options.no_auto_rank {
3786 if let Some((idx, col_name)) = find_road_class_column(input_schema, full) {
3788 let cr = overture_road_ranking(col_name.clone());
3789 let keys = extract_class_ranks(full.column(idx), &cr)?;
3790 log::info!(
3791 "overview ranking: auto-detected Overture road classes in column {col_name:?}; \
3792 applying built-in ranking (motorway > … > service > tail)"
3793 );
3794 return Ok((keys, class_ranking_provenance("auto-overture-roads", &cr)));
3795 }
3796 if let Some((idx, col_name)) = find_confidence_column(input_schema, geometries) {
3798 let keys = extract_sort_keys(full.column(idx));
3799 log::info!(
3800 "overview ranking: auto-detected Overture places confidence column {col_name:?} \
3801 (numeric point ranking)"
3802 );
3803 return Ok((
3804 keys,
3805 RankingProvenance {
3806 mode: "auto-confidence".to_string(),
3807 column: Some(col_name),
3808 ranks: None,
3809 unknown_rank: None,
3810 },
3811 ));
3812 }
3813 }
3814
3815 log::info!(
3817 "overview ranking: no sort key specified or auto-detected; using size + \
3818 deterministic-hash fallback"
3819 );
3820 Ok((
3821 vec![None; n],
3822 RankingProvenance {
3823 mode: "size-fallback".to_string(),
3824 column: None,
3825 ranks: None,
3826 unknown_rank: None,
3827 },
3828 ))
3829}
3830
3831pub(super) fn class_ranking_provenance(mode: &str, cr: &ClassRanking) -> RankingProvenance {
3835 let ranks = if cr.ranks.len() <= MAX_PROVENANCE_RANKS {
3836 Some(cr.ranks.iter().cloned().collect())
3837 } else {
3838 None
3839 };
3840 RankingProvenance {
3841 mode: mode.to_string(),
3842 column: Some(cr.column.clone()),
3843 ranks,
3844 unknown_rank: Some(cr.unknown_rank),
3845 }
3846}
3847
3848pub(super) fn extract_class_ranks(
3852 col: &dyn Array,
3853 ranking: &ClassRanking,
3854) -> Result<Vec<Option<f64>>, ConvertError> {
3855 use arrow_array::cast::AsArray;
3856
3857 let map: HashMap<&str, f64> = ranking
3858 .ranks
3859 .iter()
3860 .map(|(k, v)| (k.as_str(), *v))
3861 .collect();
3862 let n = col.len();
3863
3864 macro_rules! collect_str {
3865 ($arr:expr) => {{
3866 let a = $arr;
3867 (0..n)
3868 .map(|i| {
3869 if a.is_null(i) {
3870 None
3871 } else {
3872 Some(*map.get(a.value(i)).unwrap_or(&ranking.unknown_rank))
3873 }
3874 })
3875 .collect()
3876 }};
3877 }
3878
3879 match col.data_type() {
3880 DataType::Utf8 => Ok(collect_str!(col.as_string::<i32>())),
3881 DataType::LargeUtf8 => Ok(collect_str!(col.as_string::<i64>())),
3882 other => Err(ConvertError::ClassRankColumnNotString {
3883 name: ranking.column.clone(),
3884 data_type: format!("{other:?}"),
3885 }),
3886 }
3887}
3888
3889fn find_road_class_column(schema: &Schema, full: &RecordBatch) -> Option<(usize, String)> {
3894 for (idx, f) in schema.fields().iter().enumerate() {
3895 let lname = f.name().to_ascii_lowercase();
3896 if lname != "road_class" && lname != "class" {
3897 continue;
3898 }
3899 if !matches!(f.data_type(), DataType::Utf8 | DataType::LargeUtf8) {
3900 continue;
3901 }
3902 if column_overlaps_road_vocab(full.column(idx)) {
3903 return Some((idx, f.name().clone()));
3904 }
3905 }
3906 None
3907}
3908
3909fn column_overlaps_road_vocab(col: &dyn Array) -> bool {
3912 use arrow_array::cast::AsArray;
3913
3914 let vocab: HashSet<&str> = KNOWN_ROAD_CLASSES.iter().copied().collect();
3915 let mut found: HashSet<&str> = HashSet::new();
3916
3917 macro_rules! scan {
3918 ($arr:expr) => {{
3919 let a = $arr;
3920 for i in 0..a.len() {
3921 if a.is_null(i) {
3922 continue;
3923 }
3924 if let Some(&hit) = vocab.get(a.value(i)) {
3925 found.insert(hit);
3926 if found.len() >= ROAD_VOCAB_MIN_DISTINCT {
3927 return true;
3928 }
3929 }
3930 }
3931 }};
3932 }
3933
3934 match col.data_type() {
3935 DataType::Utf8 => scan!(col.as_string::<i32>()),
3936 DataType::LargeUtf8 => scan!(col.as_string::<i64>()),
3937 _ => return false,
3938 }
3939 found.len() >= ROAD_VOCAB_MIN_DISTINCT
3940}
3941
3942fn find_confidence_column(
3946 schema: &Schema,
3947 geometries: &[Geometry<f64>],
3948) -> Option<(usize, String)> {
3949 if geometries.is_empty() {
3950 return None;
3951 }
3952 let points = geometries
3953 .iter()
3954 .filter(|g| matches!(feature_kind(g), FeatureKind::Point))
3955 .count();
3956 if points * 2 < geometries.len() {
3958 return None;
3959 }
3960 for (idx, f) in schema.fields().iter().enumerate() {
3961 if f.name().eq_ignore_ascii_case("confidence")
3962 && matches!(f.data_type(), DataType::Float32 | DataType::Float64)
3963 {
3964 return Some((idx, f.name().clone()));
3965 }
3966 }
3967 None
3968}
3969
3970pub(super) fn fill_level_bytes(
3973 output_path: &Path,
3974 meta: &super::level::OverviewsMeta,
3975 reports: &mut [LevelReport],
3976) -> Result<(), ConvertError> {
3977 let file = std::fs::File::open(output_path)?;
3978 let pq = ParquetRecordBatchReaderBuilder::try_new(file)?;
3979 let pmeta = pq.metadata();
3980
3981 let mut start = 0usize;
3982 for (level, report) in meta.levels.iter().zip(reports.iter_mut()) {
3983 let end = level.row_group_end as usize;
3984 let mut uncompressed = 0i64;
3985 let mut compressed = 0i64;
3986 for rg in start..=end {
3987 let rgm = pmeta.row_group(rg);
3988 uncompressed += rgm.total_byte_size();
3989 compressed += rgm.compressed_size();
3990 }
3991 report.uncompressed_bytes = uncompressed;
3992 report.compressed_bytes = compressed;
3993 start = end + 1;
3994 }
3995 Ok(())
3996}
3997
3998#[cfg(test)]
3999mod tests {
4000 use super::*;
4001
4002 mod scan_feature_props {
4008 use super::super::{feature_kind, geometry_bbox, scan_feature, usable_geometry};
4009 use geo::{Geometry, LineString, MultiLineString, MultiPolygon, Point, Polygon};
4010 use proptest::prelude::*;
4011
4012 fn any_f64() -> impl Strategy<Value = f64> {
4013 prop_oneof![
4014 8 => -1.0e6f64..1.0e6f64,
4015 1 => Just(f64::NAN),
4016 1 => prop_oneof![Just(f64::INFINITY), Just(f64::NEG_INFINITY)],
4017 ]
4018 }
4019 fn any_coord() -> impl Strategy<Value = geo::Coord<f64>> {
4020 (any_f64(), any_f64()).prop_map(|(x, y)| geo::Coord { x, y })
4021 }
4022 fn any_ring() -> impl Strategy<Value = LineString<f64>> {
4023 proptest::collection::vec(any_coord(), 0..6).prop_map(LineString::new)
4024 }
4025 fn any_polygon() -> impl Strategy<Value = Polygon<f64>> {
4026 (any_ring(), proptest::collection::vec(any_ring(), 0..3))
4027 .prop_map(|(ext, ints)| Polygon::new(ext, ints))
4028 }
4029 fn any_geometry() -> impl Strategy<Value = Geometry<f64>> {
4030 prop_oneof![
4031 any_coord().prop_map(|c| Geometry::Point(Point::from(c))),
4032 proptest::collection::vec(any_coord(), 0..5)
4033 .prop_map(|cs| Geometry::MultiPoint(cs.into_iter().map(Point::from).collect())),
4034 any_ring().prop_map(Geometry::LineString),
4035 proptest::collection::vec(any_ring(), 0..3)
4036 .prop_map(|ls| Geometry::MultiLineString(MultiLineString::new(ls))),
4037 any_polygon().prop_map(Geometry::Polygon),
4038 proptest::collection::vec(any_polygon(), 0..3)
4039 .prop_map(|ps| Geometry::MultiPolygon(MultiPolygon::new(ps))),
4040 ]
4041 }
4042
4043 proptest! {
4044 #[test]
4045 fn scan_feature_matches_components(g in any_geometry()) {
4046 let expected = if usable_geometry(&g) {
4047 Some((feature_kind(&g), geometry_bbox(&g)))
4048 } else {
4049 None
4050 };
4051 prop_assert_eq!(scan_feature(&g), expected);
4052 }
4053 }
4054 }
4055
4056 use crate::overview::check::validate_file;
4057 use crate::overview::level::gsd;
4058 use crate::overview::reader::OverviewReader;
4059 use arrow_array::{Float64Array, Int64Array, RecordBatch, StringArray};
4060 use arrow_schema::{DataType, Field, Schema};
4061 use geo::{Geometry, LineString, Point, Polygon};
4062 use geoarrow::array::GeometryBuilder;
4063 use geoarrow::datatypes::GeometryType;
4064 use geoarrow_array::GeoArrowArray;
4065 use geoparquet::writer::{
4066 GeoParquetRecordBatchEncoder, GeoParquetWriterEncoding, GeoParquetWriterOptionsBuilder,
4067 };
4068 use parquet::arrow::ArrowWriter;
4069
4070 use crate::batch_processor::extract_geometries_from_array;
4071
4072 #[test]
4076 fn full_file_remote_warning_gated_on_large_unpruned_remote() {
4077 const BIG: u64 = 4 << 30; assert!(full_file_remote_warning(0, 8, 8, BIG).is_none());
4081 assert!(full_file_remote_warning(1, 2, 8, BIG).is_none());
4083 assert!(full_file_remote_warning(1, 8, 8, 100 << 20).is_none());
4085 assert!(full_file_remote_warning(1, 8, 8, FULL_FILE_REMOTE_WARN_BYTES - 1).is_none());
4087 assert!(full_file_remote_warning(1, 8, 8, FULL_FILE_REMOTE_WARN_BYTES).is_some());
4089 let msg = full_file_remote_warning(1, 8, 8, BIG)
4091 .expect("large full-file remote convert should warn");
4092 assert!(msg.contains("--bbox"), "nudge should mention --bbox: {msg}");
4093 assert!(
4094 msg.contains("4.0 GiB"),
4095 "nudge should state the object size: {msg}"
4096 );
4097 assert!(
4098 !msg.contains("partitions"),
4099 "single object: no part count in the message: {msg}"
4100 );
4101 }
4102
4103 #[test]
4108 fn full_file_remote_warning_names_partition_count() {
4109 const PART: u64 = 600 << 20; let msg = full_file_remote_warning(20, 40, 40, 20 * PART)
4111 .expect("20 x 600 MB unpruned remote parts should warn");
4112 assert!(
4113 msg.contains("20 remote partitions"),
4114 "message names the part count: {msg}"
4115 );
4116 assert!(
4117 msg.contains("11.7 GiB"),
4118 "message states the summed size: {msg}"
4119 );
4120 assert!(full_file_remote_warning(20, 40, 40, 500 << 20).is_none());
4122 }
4123
4124 #[test]
4129 fn spill_space_warning_names_dir_and_shortfall() {
4130 let dir = Path::new("/mnt/scratch");
4131 let msg = spill_space_warning(10 << 30, 1 << 30, dir).expect("shortfall should warn");
4132 assert!(
4133 msg.contains("/mnt/scratch"),
4134 "warning names the spill dir: {msg}"
4135 );
4136 assert!(
4137 msg.contains("--spill-dir"),
4138 "warning suggests --spill-dir: {msg}"
4139 );
4140 assert!(
4142 msg.contains("9.5 GiB"),
4143 "warning states the shortfall: {msg}"
4144 );
4145 }
4146
4147 #[test]
4150 fn spill_space_warning_quiet_with_ample_space() {
4151 let dir = Path::new("/tmp");
4152 assert!(spill_space_warning(1 << 30, 20 << 30, dir).is_none());
4153 let est: u64 = 20 << 20;
4154 let need = est + est / 20;
4155 assert!(spill_space_warning(est, need, dir).is_none());
4156 assert!(spill_space_warning(est, need - 1, dir).is_some());
4157 }
4158
4159 #[test]
4163 fn spill_space_check_gating() {
4164 let dir = Path::new("/tmp");
4165 assert!(spill_space_check(false, 10 << 30, dir, |_| panic!(
4166 "free-space probe must not run for local inputs"
4167 ))
4168 .is_none());
4169 assert!(spill_space_check(true, 10 << 30, dir, |_| None).is_none());
4170 assert!(spill_space_check(true, 0, dir, |_| Some(0)).is_none());
4171 assert!(spill_space_check(true, 10 << 30, dir, |_| Some(1)).is_some());
4172 }
4173
4174 #[test]
4178 fn validate_options_rejects_missing_spill_dir() {
4179 let opts = ConvertOptions {
4180 spill_dir: Some(std::path::PathBuf::from(
4181 "/nonexistent/tylertoo-spill-dir-272",
4182 )),
4183 ..Default::default()
4184 };
4185 let err = validate_options(&opts).expect_err("missing spill dir must be rejected");
4186 let msg = err.to_string();
4187 assert!(msg.contains("spill-dir"), "error names the option: {msg}");
4188 assert!(
4189 msg.contains("/nonexistent/tylertoo-spill-dir-272"),
4190 "error names the path: {msg}"
4191 );
4192 }
4193
4194 fn synthetic_geometries() -> Vec<Geometry<f64>> {
4199 let mut geoms = Vec::new();
4200 for i in 0..6 {
4203 let x = i as f64 * 5.0;
4204 let y = i as f64 * 3.0;
4205 geoms.push(Geometry::Point(Point::new(x, y)));
4206 }
4207 for i in 0..4 {
4209 let base = 40.0 + i as f64 * 10.0;
4210 let ls = LineString::from(
4211 (0..12)
4212 .map(|k| {
4213 (
4214 base + k as f64 * 0.5,
4215 (k as f64 * 0.6).sin() + i as f64 * 8.0,
4216 )
4217 })
4218 .collect::<Vec<_>>(),
4219 );
4220 geoms.push(Geometry::LineString(ls));
4221 }
4222 for i in 0..4 {
4224 let cx = -60.0 + i as f64 * 12.0;
4225 let cy = -40.0 - i as f64 * 5.0;
4226 let half = 2.0 + i as f64 * 1.5;
4227 let ext = LineString::from(vec![
4228 (cx - half, cy - half),
4229 (cx + half, cy - half),
4230 (cx + half, cy + half),
4231 (cx - half, cy + half),
4232 (cx - half, cy - half),
4233 ]);
4234 geoms.push(Geometry::Polygon(Polygon::new(ext, vec![])));
4235 }
4236 geoms
4237 }
4238
4239 fn build_geometry_array(geoms: &[Geometry<f64>]) -> geoarrow::array::GeometryArray {
4240 let typ = GeometryType::new(Default::default());
4241 let mut b = GeometryBuilder::new(typ).with_prefer_multi(false);
4242 b.extend_from_iter(geoms.iter().map(Some));
4243 b.finish()
4244 }
4245
4246 fn output_column_names(path: &Path) -> Vec<String> {
4248 let file = std::fs::File::open(path).unwrap();
4249 let builder = ParquetRecordBatchReaderBuilder::try_new(file).unwrap();
4250 builder
4251 .schema()
4252 .fields()
4253 .iter()
4254 .map(|f| f.name().clone())
4255 .collect()
4256 }
4257
4258 fn write_input(
4263 path: &Path,
4264 geoms: &[Geometry<f64>],
4265 extra_level_col: bool,
4266 crs_metadata: Option<geoarrow::datatypes::Metadata>,
4267 ) {
4268 let n = geoms.len();
4269 let id = Int64Array::from((0..n as i64).collect::<Vec<_>>());
4270 let name = StringArray::from((0..n).map(|i| format!("f{i}")).collect::<Vec<_>>());
4271 let rank = Float64Array::from((0..n).map(|i| (n - i) as f64).collect::<Vec<_>>());
4272
4273 let geom_arr = if let Some(md) = crs_metadata {
4274 let typ = GeometryType::new(Arc::new(md));
4275 let mut b = GeometryBuilder::new(typ).with_prefer_multi(false);
4276 b.extend_from_iter(geoms.iter().map(Some));
4277 b.finish()
4278 } else {
4279 build_geometry_array(geoms)
4280 };
4281 let geom_field = geom_arr.data_type().to_field("geometry", true);
4282
4283 let mut fields = vec![
4284 Arc::new(Field::new("id", DataType::Int64, false)),
4285 Arc::new(Field::new("name", DataType::Utf8, false)),
4286 Arc::new(Field::new("rank", DataType::Float64, false)),
4287 ];
4288 let mut columns: Vec<Arc<dyn Array>> = vec![Arc::new(id), Arc::new(name), Arc::new(rank)];
4289 if extra_level_col {
4290 fields.push(Arc::new(Field::new("level", DataType::Int32, false)));
4291 columns.push(Arc::new(arrow_array::Int32Array::from(vec![0i32; n])));
4292 }
4293 fields.push(Arc::new(geom_field));
4294 columns.push(geom_arr.to_array_ref());
4295
4296 let schema = Arc::new(Schema::new(fields));
4297 let batch = RecordBatch::try_new(schema.clone(), columns).unwrap();
4298
4299 let gpq_options = GeoParquetWriterOptionsBuilder::default()
4300 .set_encoding(GeoParquetWriterEncoding::WKB)
4301 .set_generate_covering(true)
4302 .build();
4303 let encoder = GeoParquetRecordBatchEncoder::try_new(&schema, &gpq_options).unwrap();
4304 let target_schema = encoder.target_schema();
4305
4306 let file = std::fs::File::create(path).unwrap();
4307 let mut writer = ArrowWriter::try_new(file, target_schema, None).unwrap();
4308 let mut encoder = encoder;
4310 let encoded = encoder.encode_record_batch(&batch).unwrap();
4311 writer.write(&encoded).unwrap();
4312 writer.append_key_value_metadata(encoder.into_keyvalue().unwrap());
4313 writer.close().unwrap();
4314 }
4315
4316 fn read_level_rows(
4318 reader: &OverviewReader,
4319 level: usize,
4320 ) -> Vec<(i64, String, f64, Geometry<f64>)> {
4321 use arrow_array::cast::AsArray;
4322 use arrow_array::types::Float64Type;
4323 let rdr = reader.read_level(level, None).unwrap();
4324 let mut out = Vec::new();
4325 for batch in rdr {
4326 let batch = batch.unwrap();
4327 let ids = batch
4328 .column(batch.schema().index_of("id").unwrap())
4329 .as_primitive::<arrow_array::types::Int64Type>()
4330 .clone();
4331 let names = batch
4332 .column(batch.schema().index_of("name").unwrap())
4333 .as_string::<i32>()
4334 .clone();
4335 let ranks = batch
4336 .column(batch.schema().index_of("rank").unwrap())
4337 .as_primitive::<Float64Type>()
4338 .clone();
4339 let gcol = batch.column(batch.schema().index_of("geometry").unwrap());
4340 let garr: Arc<dyn GeoArrowArray> = from_arrow_array(
4341 gcol.as_ref(),
4342 batch
4343 .schema()
4344 .field(batch.schema().index_of("geometry").unwrap()),
4345 )
4346 .unwrap();
4347 let mut gvec = Vec::new();
4348 extract_geometries_from_array(garr.as_ref(), &mut gvec).unwrap();
4349 for (i, g) in gvec.iter().enumerate() {
4350 out.push((
4351 ids.value(i),
4352 names.value(i).to_string(),
4353 ranks.value(i),
4354 g.clone(),
4355 ));
4356 }
4357 }
4358 out
4359 }
4360
4361 fn write_input_partition(
4368 path: &Path,
4369 geoms: &[Geometry<f64>],
4370 range: std::ops::Range<usize>,
4371 row_group_rows: Option<usize>,
4372 ) {
4373 use parquet::file::properties::WriterProperties;
4374 let total = geoms.len();
4375 let idx: Vec<usize> = range.collect();
4376 let id = Int64Array::from(idx.iter().map(|&i| i as i64).collect::<Vec<_>>());
4377 let name = StringArray::from(idx.iter().map(|&i| format!("f{i}")).collect::<Vec<_>>());
4378 let rank = Float64Array::from(idx.iter().map(|&i| (total - i) as f64).collect::<Vec<_>>());
4379 let part_geoms: Vec<Geometry<f64>> = idx.iter().map(|&i| geoms[i].clone()).collect();
4380 let geom_arr = build_geometry_array(&part_geoms);
4381 let geom_field = geom_arr.data_type().to_field("geometry", true);
4382
4383 let fields = vec![
4384 Arc::new(Field::new("id", DataType::Int64, false)),
4385 Arc::new(Field::new("name", DataType::Utf8, false)),
4386 Arc::new(Field::new("rank", DataType::Float64, false)),
4387 Arc::new(geom_field),
4388 ];
4389 let columns: Vec<Arc<dyn Array>> = vec![
4390 Arc::new(id),
4391 Arc::new(name),
4392 Arc::new(rank),
4393 geom_arr.to_array_ref(),
4394 ];
4395 let schema = Arc::new(Schema::new(fields));
4396 let batch = RecordBatch::try_new(schema.clone(), columns).unwrap();
4397
4398 let gpq_options = GeoParquetWriterOptionsBuilder::default()
4399 .set_encoding(GeoParquetWriterEncoding::WKB)
4400 .set_generate_covering(true)
4401 .build();
4402 let mut encoder = GeoParquetRecordBatchEncoder::try_new(&schema, &gpq_options).unwrap();
4403 let target_schema = encoder.target_schema();
4404 let props = row_group_rows.map(|n| {
4405 WriterProperties::builder()
4406 .set_max_row_group_row_count(Some(n))
4407 .build()
4408 });
4409 let file = std::fs::File::create(path).unwrap();
4410 let mut writer = ArrowWriter::try_new(file, target_schema, props).unwrap();
4411 let encoded = encoder.encode_record_batch(&batch).unwrap();
4412 writer.write(&encoded).unwrap();
4413 writer.append_key_value_metadata(encoder.into_keyvalue().unwrap());
4414 writer.close().unwrap();
4415 }
4416
4417 fn convert_and_export(input: &Path, workdir: &Path, opts: &ConvertOptions) -> Vec<u8> {
4421 use crate::overview::export::{export_pmtiles, ExportOptions};
4422 let stem = input
4423 .file_name()
4424 .unwrap_or_default()
4425 .to_string_lossy()
4426 .replace('.', "_");
4427 let overview = workdir.join(format!("{stem}-overview.parquet"));
4428 let pmtiles = workdir.join(format!("{stem}.pmtiles"));
4429 convert_to_overviews(input, &overview, opts).unwrap();
4430 export_pmtiles(&overview, &pmtiles, &ExportOptions::default()).unwrap();
4431 std::fs::read(&pmtiles).unwrap()
4432 }
4433
4434 fn multi_test_options() -> ConvertOptions {
4435 ConvertOptions {
4436 mode: Mode::Duplicating,
4437 levels: LevelPlan::ZoomRange {
4438 min_zoom: 2,
4439 max_zoom: 8,
4440 },
4441 ..Default::default()
4442 }
4443 }
4444
4445 #[test]
4450 fn multi_partition_output_matches_single_file() {
4451 let geoms = synthetic_geometries();
4452 let n = geoms.len();
4453 let dir = tempfile::tempdir().unwrap();
4454 let single = dir.path().join("single.parquet");
4455 write_input_partition(&single, &geoms, 0..n, None);
4456 let parts = dir.path().join("parts");
4457 std::fs::create_dir(&parts).unwrap();
4458 write_input_partition(&parts.join("part-000.parquet"), &geoms, 0..5, None);
4459 write_input_partition(&parts.join("part-001.parquet"), &geoms, 5..9, None);
4460 write_input_partition(&parts.join("part-002.parquet"), &geoms, 9..n, None);
4461
4462 let opts = multi_test_options();
4463 let report =
4464 convert_to_overviews(&parts, dir.path().join("probe-overview.parquet"), &opts).unwrap();
4465 assert_eq!(report.input_features, n);
4466 assert_eq!(report.row_groups_total, 3, "one row group per partition");
4467
4468 let pm_single = convert_and_export(&single, dir.path(), &opts);
4469 let pm_multi = convert_and_export(&parts, dir.path(), &opts);
4470 assert!(
4471 pm_single == pm_multi,
4472 "multi-partition output must be byte-identical to single-file \
4473 ({} vs {} bytes)",
4474 pm_single.len(),
4475 pm_multi.len()
4476 );
4477 }
4478
4479 #[test]
4482 fn multi_partition_zero_row_part_matches_single_file() {
4483 let geoms = synthetic_geometries();
4484 let n = geoms.len();
4485 let dir = tempfile::tempdir().unwrap();
4486 let single = dir.path().join("single.parquet");
4487 write_input_partition(&single, &geoms, 0..n, None);
4488 let parts = dir.path().join("parts");
4489 std::fs::create_dir(&parts).unwrap();
4490 write_input_partition(&parts.join("part-000.parquet"), &geoms, 0..5, None);
4491 write_input_partition(&parts.join("part-001.parquet"), &geoms, 5..5, None); write_input_partition(&parts.join("part-002.parquet"), &geoms, 5..n, None);
4493
4494 let opts = multi_test_options();
4495 let pm_single = convert_and_export(&single, dir.path(), &opts);
4496 let pm_multi = convert_and_export(&parts, dir.path(), &opts);
4497 assert!(
4498 pm_single == pm_multi,
4499 "0-row partition must not shift rows or offsets"
4500 );
4501 }
4502
4503 #[test]
4507 fn multi_partition_bbox_selection_matches_single_file() {
4508 let geoms = synthetic_geometries();
4509 let n = geoms.len();
4510 let dir = tempfile::tempdir().unwrap();
4511 let single = dir.path().join("single.parquet");
4512 write_input_partition(&single, &geoms, 0..n, Some(2));
4513 let parts = dir.path().join("parts");
4514 std::fs::create_dir(&parts).unwrap();
4515 write_input_partition(&parts.join("part-000.parquet"), &geoms, 0..6, Some(2));
4516 write_input_partition(&parts.join("part-001.parquet"), &geoms, 6..10, Some(2));
4517 write_input_partition(&parts.join("part-002.parquet"), &geoms, 10..n, Some(2));
4518
4519 let bbox = Some([-100.0, -70.0, 30.0, 20.0]);
4522 let opts = ConvertOptions {
4523 bbox,
4524 ..multi_test_options()
4525 };
4526 let report =
4527 convert_to_overviews(&parts, dir.path().join("probe-overview.parquet"), &opts).unwrap();
4528 assert!(
4529 report.row_groups_read < report.row_groups_total,
4530 "bbox must prune row groups across parts: {}/{}",
4531 report.row_groups_read,
4532 report.row_groups_total
4533 );
4534
4535 let pm_single = convert_and_export(&single, dir.path(), &opts);
4536 let pm_multi = convert_and_export(&parts, dir.path(), &opts);
4537 assert!(
4538 pm_single == pm_multi,
4539 "per-part bbox selection must keep offsets aligned"
4540 );
4541 }
4542
4543 #[test]
4546 fn multi_partition_requires_streaming_pipeline() {
4547 let geoms = synthetic_geometries();
4548 let dir = tempfile::tempdir().unwrap();
4549 let parts = dir.path().join("parts");
4550 std::fs::create_dir(&parts).unwrap();
4551 write_input_partition(&parts.join("part-000.parquet"), &geoms, 0..5, None);
4552 write_input_partition(
4553 &parts.join("part-001.parquet"),
4554 &geoms,
4555 5..geoms.len(),
4556 None,
4557 );
4558
4559 let opts = ConvertOptions {
4560 streaming: false,
4561 ..multi_test_options()
4562 };
4563 let err = convert_to_overviews(&parts, dir.path().join("out.parquet"), &opts).unwrap_err();
4564 assert!(
4565 matches!(err, ConvertError::MultiPartitionRequiresStreaming),
4566 "expected MultiPartitionRequiresStreaming, got {err:?}"
4567 );
4568 }
4569
4570 #[test]
4573 fn multi_partition_crs_mismatch_rejected() {
4574 let geoms = synthetic_geometries();
4575 let dir = tempfile::tempdir().unwrap();
4576 let parts = dir.path().join("parts");
4577 std::fs::create_dir(&parts).unwrap();
4578 write_input(&parts.join("a.parquet"), &geoms, false, None);
4579 let projjson = serde_json::json!({
4582 "type": "ProjectedCRS",
4583 "name": "UTM zone 33N",
4584 "id": { "authority": "EPSG", "code": 32633 }
4585 });
4586 let md = geoarrow::datatypes::Metadata::new(
4587 geoarrow::datatypes::Crs::from_projjson(projjson),
4588 None,
4589 );
4590 write_input(&parts.join("b.parquet"), &geoms, false, Some(md));
4591
4592 let err = convert_to_overviews(
4593 &parts,
4594 dir.path().join("out.parquet"),
4595 &multi_test_options(),
4596 )
4597 .unwrap_err();
4598 match err {
4599 ConvertError::Input(crate::input::InputError::IncompatiblePartition {
4600 offender,
4601 ..
4602 }) => {
4603 assert!(offender.ends_with("b.parquet"), "offender: {offender}");
4604 }
4605 other => panic!("expected IncompatiblePartition, got {other:?}"),
4606 }
4607 }
4608
4609 #[test]
4612 fn duplicating_canonical_matches_input() {
4613 let geoms = synthetic_geometries();
4614 let tin = tempfile::NamedTempFile::new().unwrap();
4615 let tout = tempfile::NamedTempFile::new().unwrap();
4616 write_input(tin.path(), &geoms, false, None);
4617
4618 let opts = ConvertOptions {
4619 mode: Mode::Duplicating,
4620 levels: LevelPlan::ZoomRange {
4621 min_zoom: 2,
4622 max_zoom: 8,
4623 },
4624 ..Default::default()
4625 };
4626 let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
4627
4628 let vr = validate_file(tout.path()).unwrap();
4630 assert!(
4631 vr.is_valid(),
4632 "failures: {:?}",
4633 vr.failures().collect::<Vec<_>>()
4634 );
4635
4636 let reader = OverviewReader::open(tout.path()).unwrap();
4637 assert_eq!(reader.mode(), Mode::Duplicating);
4638 let canonical = reader.num_levels() - 1;
4639
4640 let rows = read_level_rows(&reader, canonical);
4642 assert_eq!(rows.len(), geoms.len());
4643 for (i, (id, name, rank, geom)) in rows.iter().enumerate() {
4644 assert_eq!(*id, i as i64);
4645 assert_eq!(name, &format!("f{i}"));
4646 assert_eq!(*rank, (geoms.len() - i) as f64);
4647 assert_eq!(geom, &geoms[i], "canonical geometry must be verbatim");
4648 }
4649
4650 assert_eq!(report.input_features, geoms.len());
4652 assert_eq!(report.levels[canonical].feature_count, geoms.len());
4653 }
4654
4655 #[test]
4656 fn duplicating_coarse_levels_monotone() {
4657 let geoms = synthetic_geometries();
4658 let tin = tempfile::NamedTempFile::new().unwrap();
4659 let tout = tempfile::NamedTempFile::new().unwrap();
4660 write_input(tin.path(), &geoms, false, None);
4661
4662 let opts = ConvertOptions {
4663 mode: Mode::Duplicating,
4664 levels: LevelPlan::ZoomRange {
4665 min_zoom: 1,
4666 max_zoom: 10,
4667 },
4668 ..Default::default()
4669 };
4670 let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
4671
4672 for w in report.levels.windows(2) {
4674 assert!(
4675 w[0].feature_count <= w[1].feature_count,
4676 "feature counts not monotone: {:?}",
4677 report.levels
4678 );
4679 assert!(
4680 w[0].vertex_count <= w[1].vertex_count,
4681 "vertex counts not monotone: {:?}",
4682 report.levels
4683 );
4684 }
4685 assert_eq!(report.levels.last().unwrap().feature_count, geoms.len());
4687
4688 assert!(validate_file(tout.path()).unwrap().is_valid());
4690 }
4691
4692 fn polygon_fixture() -> Vec<Geometry<f64>> {
4697 let mut geoms = Vec::new();
4698 for i in 0..12 {
4699 let cx = -150.0 + i as f64 * 25.0;
4700 let cy = -60.0 + (i % 5) as f64 * 27.0;
4701 let half = 20.0 / (3f64).powi(i % 9);
4703 let ext = LineString::from(vec![
4704 (cx - half, cy - half),
4705 (cx + half, cy - half),
4706 (cx + half, cy + half),
4707 (cx - half, cy + half),
4708 (cx - half, cy - half),
4709 ]);
4710 geoms.push(Geometry::Polygon(Polygon::new(ext, vec![])));
4711 }
4712 geoms
4713 }
4714
4715 fn band(min_zoom: u8, max_zoom: u8, repr: Representation) -> RepresentationBand {
4716 RepresentationBand {
4717 min_zoom,
4718 max_zoom,
4719 repr,
4720 }
4721 }
4722
4723 #[test]
4727 fn representation_point_band_points_coarse_polygons_fine() {
4728 let geoms = polygon_fixture();
4729 let tin = tempfile::NamedTempFile::new().unwrap();
4730 let tout = tempfile::NamedTempFile::new().unwrap();
4731 write_input(tin.path(), &geoms, false, None);
4732
4733 let opts = ConvertOptions {
4734 mode: Mode::Duplicating,
4735 levels: LevelPlan::ZoomRange {
4736 min_zoom: 2,
4737 max_zoom: 8,
4738 },
4739 representation: vec![band(2, 5, Representation::Point)],
4740 ..Default::default()
4741 };
4742 let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
4743 assert!(validate_file(tout.path()).unwrap().is_valid());
4744
4745 let reader = OverviewReader::open(tout.path()).unwrap();
4746 for (idx, lvl) in report.levels.iter().enumerate() {
4747 let rows = read_level_rows(&reader, idx);
4748 assert!(!rows.is_empty());
4749 let zoom = lvl.zoom.expect("zoom-range plan records zooms");
4750 if zoom <= 5 {
4751 for (id, _, _, g) in &rows {
4752 assert!(
4753 matches!(g, Geometry::Point(_)),
4754 "level z{zoom} feature {id} must be a Point, got {g:?}"
4755 );
4756 }
4757 } else {
4758 assert!(
4759 rows.iter().any(|(_, _, _, g)| matches!(
4760 g,
4761 Geometry::Polygon(_) | Geometry::MultiPolygon(_)
4762 )),
4763 "level z{zoom} must keep polygons"
4764 );
4765 }
4766 }
4767 let canonical = read_level_rows(&reader, report.levels.len() - 1);
4769 assert_eq!(canonical.len(), geoms.len());
4770 for (i, (_, _, _, g)) in canonical.iter().enumerate() {
4771 assert_eq!(g, &geoms[i]);
4772 }
4773
4774 let plain = ConvertOptions {
4778 representation: Vec::new(),
4779 ..opts.clone()
4780 };
4781 let tout2 = tempfile::NamedTempFile::new().unwrap();
4782 let plain_report = convert_to_overviews(tin.path(), tout2.path(), &plain).unwrap();
4783 assert!(
4784 report.levels[0].feature_count >= plain_report.levels[0].feature_count,
4785 "point band must not lose coarse coverage vs the gated polygon run"
4786 );
4787
4788 let gen = reader
4790 .meta()
4791 .generalization
4792 .as_ref()
4793 .expect("generalization block present");
4794 let bands = gen.representation.as_ref().expect("bands recorded");
4795 assert_eq!(bands.len(), 1);
4796 assert_eq!(bands[0].zooms, [2, 5]);
4797 assert_eq!(bands[0].repr, "point");
4798 assert!(gen.collapse.is_none(), "no global collapse requested");
4799 }
4800
4801 #[test]
4804 fn representation_point_band_streaming_matches_in_memory() {
4805 let geoms = polygon_fixture();
4806 let tin = tempfile::NamedTempFile::new().unwrap();
4807 write_input(tin.path(), &geoms, false, None);
4808
4809 let base = ConvertOptions {
4810 mode: Mode::Duplicating,
4811 levels: LevelPlan::ZoomRange {
4812 min_zoom: 2,
4813 max_zoom: 8,
4814 },
4815 representation: vec![band(2, 4, Representation::Point)],
4816 ..Default::default()
4817 };
4818 let t_stream = tempfile::NamedTempFile::new().unwrap();
4819 let t_mem = tempfile::NamedTempFile::new().unwrap();
4820 let r1 = convert_to_overviews(tin.path(), t_stream.path(), &base).unwrap();
4821 let mem = ConvertOptions {
4822 streaming: false,
4823 ..base
4824 };
4825 let r2 = convert_to_overviews(tin.path(), t_mem.path(), &mem).unwrap();
4826 assert_eq!(r1.levels.len(), r2.levels.len());
4827
4828 let rd1 = OverviewReader::open(t_stream.path()).unwrap();
4829 let rd2 = OverviewReader::open(t_mem.path()).unwrap();
4830 for lvl in 0..r1.levels.len() {
4831 assert_eq!(
4832 read_level_rows(&rd1, lvl),
4833 read_level_rows(&rd2, lvl),
4834 "level {lvl} rows must match across engines"
4835 );
4836 }
4837 }
4838
4839 #[test]
4843 fn representation_square_band_type_preserving() {
4844 let mut geoms = polygon_fixture();
4846 for i in 0..30 {
4847 let cx = 10.0 + (i % 6) as f64 * 0.02;
4848 let cy = 20.0 + (i / 6) as f64 * 0.02;
4849 let half = 0.004;
4850 let ext = LineString::from(vec![
4851 (cx - half, cy - half),
4852 (cx + half, cy - half),
4853 (cx + half, cy + half),
4854 (cx - half, cy + half),
4855 (cx - half, cy - half),
4856 ]);
4857 geoms.push(Geometry::Polygon(Polygon::new(ext, vec![])));
4858 }
4859 let tin = tempfile::NamedTempFile::new().unwrap();
4860 let tout = tempfile::NamedTempFile::new().unwrap();
4861 write_input(tin.path(), &geoms, false, None);
4862
4863 let opts = ConvertOptions {
4864 mode: Mode::Duplicating,
4865 levels: LevelPlan::ZoomRange {
4866 min_zoom: 4,
4867 max_zoom: 10,
4868 },
4869 representation: vec![band(4, 7, Representation::Square)],
4870 ..Default::default()
4871 };
4872 let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
4873 assert!(validate_file(tout.path()).unwrap().is_valid());
4874
4875 let reader = OverviewReader::open(tout.path()).unwrap();
4876 for (idx, lvl) in report.levels.iter().enumerate() {
4877 let zoom = lvl.zoom.unwrap();
4878 let rows = read_level_rows(&reader, idx);
4879 for (id, _, _, g) in &rows {
4880 assert!(
4881 matches!(g, Geometry::Polygon(_) | Geometry::MultiPolygon(_)),
4882 "square band is type-preserving; level z{zoom} feature {id} \
4883 is {g:?}"
4884 );
4885 }
4886 }
4887
4888 let z4_rows = read_level_rows(&reader, 0);
4892 let tol_deg = report.levels[0].gsd / METERS_PER_DEGREE;
4893 let squares = z4_rows
4894 .iter()
4895 .filter(|(_, _, _, g)| match g {
4896 Geometry::Polygon(p) => {
4897 let r = geo::BoundingRect::bounding_rect(p).unwrap();
4898 p.exterior().0.len() == 5
4899 && (r.width() - tol_deg).abs() < 1e-9
4900 && (r.height() - tol_deg).abs() < 1e-9
4901 }
4902 _ => false,
4903 })
4904 .count();
4905 assert!(
4906 squares > 0,
4907 "band level must contain dithered placeholder squares"
4908 );
4909
4910 let gen = reader.meta().generalization.as_ref().unwrap();
4911 let bands = gen.representation.as_ref().unwrap();
4912 assert_eq!(bands[0].repr, "square");
4913 }
4914
4915 #[test]
4919 fn collapse_square_global_disposition() {
4920 let geoms = polygon_fixture();
4921 let tin = tempfile::NamedTempFile::new().unwrap();
4922 let tout = tempfile::NamedTempFile::new().unwrap();
4923 write_input(tin.path(), &geoms, false, None);
4924
4925 let opts = ConvertOptions {
4926 mode: Mode::Duplicating,
4927 levels: LevelPlan::ZoomRange {
4928 min_zoom: 2,
4929 max_zoom: 8,
4930 },
4931 simplify: SimplifyOptions {
4932 collapse: CollapseMode::Square,
4933 ..Default::default()
4934 },
4935 ..Default::default()
4936 };
4937 let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
4938 assert!(validate_file(tout.path()).unwrap().is_valid());
4939
4940 let reader = OverviewReader::open(tout.path()).unwrap();
4941 for idx in 0..report.levels.len() {
4943 for (_, _, _, g) in read_level_rows(&reader, idx) {
4944 assert!(matches!(
4945 g,
4946 Geometry::Polygon(_) | Geometry::MultiPolygon(_)
4947 ));
4948 }
4949 }
4950 let gen = reader.meta().generalization.as_ref().unwrap();
4951 assert_eq!(gen.collapse.as_deref(), Some("square"));
4952 assert!(gen.representation.is_none());
4953 }
4954
4955 #[test]
4956 fn representation_validation_rejects_bad_bands() {
4957 let mk = |mode: Mode, levels: LevelPlan, bands: Vec<RepresentationBand>| ConvertOptions {
4958 mode,
4959 levels,
4960 representation: bands,
4961 ..Default::default()
4962 };
4963 let zr = |lo: u8, hi: u8| LevelPlan::ZoomRange {
4964 min_zoom: lo,
4965 max_zoom: hi,
4966 };
4967
4968 let err = validate_options(&mk(
4970 Mode::Partitioning,
4971 zr(0, 6),
4972 vec![band(0, 3, Representation::Point)],
4973 ))
4974 .unwrap_err();
4975 assert!(matches!(err, ConvertError::InvalidConfig(_)), "{err}");
4976
4977 let err = validate_options(&mk(
4979 Mode::Duplicating,
4980 LevelPlan::Gsds(vec![1000.0, 100.0]),
4981 vec![band(0, 3, Representation::Point)],
4982 ))
4983 .unwrap_err();
4984 assert!(matches!(err, ConvertError::InvalidConfig(_)), "{err}");
4985
4986 let err = validate_options(&mk(
4988 Mode::Duplicating,
4989 zr(0, 6),
4990 vec![band(0, 6, Representation::Point)],
4991 ))
4992 .unwrap_err();
4993 assert!(matches!(err, ConvertError::InvalidConfig(_)), "{err}");
4994
4995 let err = validate_options(&mk(
4997 Mode::Duplicating,
4998 zr(0, 8),
4999 vec![
5000 band(0, 3, Representation::Point),
5001 band(3, 5, Representation::Square),
5002 ],
5003 ))
5004 .unwrap_err();
5005 assert!(matches!(err, ConvertError::InvalidConfig(_)), "{err}");
5006
5007 let err = validate_options(&mk(
5010 Mode::Duplicating,
5011 zr(0, 8),
5012 vec![band(3, 5, Representation::Point)],
5013 ))
5014 .unwrap_err();
5015 assert!(matches!(err, ConvertError::InvalidConfig(_)), "{err}");
5016
5017 validate_options(&mk(
5019 Mode::Duplicating,
5020 zr(0, 8),
5021 vec![band(3, 5, Representation::Square)],
5022 ))
5023 .expect("mid-plan square band is valid");
5024
5025 validate_options(&mk(
5027 Mode::Duplicating,
5028 zr(0, 8),
5029 vec![
5030 band(0, 3, Representation::Point),
5031 band(4, 6, Representation::Square),
5032 ],
5033 ))
5034 .expect("point prefix + square band is valid");
5035 }
5036
5037 #[test]
5038 fn parse_representation_spec_grammar() {
5039 let bands = parse_representation_spec("0-7:point,8-14:geom").unwrap();
5040 assert_eq!(bands.len(), 2);
5041 assert_eq!(bands[0], band(0, 7, Representation::Point));
5042 assert_eq!(bands[1], band(8, 14, Representation::Geometry));
5043
5044 let bands = parse_representation_spec(" 0-5 : square ").unwrap();
5045 assert_eq!(bands, vec![band(0, 5, Representation::Square)]);
5046
5047 let bands = parse_representation_spec("3:geometry").unwrap();
5049 assert_eq!(bands, vec![band(3, 3, Representation::Geometry)]);
5050
5051 assert!(parse_representation_spec("").is_err());
5052 assert!(parse_representation_spec("0-7").is_err());
5053 assert!(parse_representation_spec("0-7:blob").is_err());
5054 assert!(parse_representation_spec("x-7:point").is_err());
5055 }
5056
5057 #[test]
5058 fn partitioning_total_equals_input() {
5059 let geoms = synthetic_geometries();
5060 let tin = tempfile::NamedTempFile::new().unwrap();
5061 let tout = tempfile::NamedTempFile::new().unwrap();
5062 write_input(tin.path(), &geoms, false, None);
5063
5064 let opts = ConvertOptions {
5065 mode: Mode::Partitioning,
5066 levels: LevelPlan::ZoomRange {
5067 min_zoom: 2,
5068 max_zoom: 8,
5069 },
5070 ..Default::default()
5071 };
5072 let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
5073
5074 let vr = validate_file(tout.path()).unwrap();
5075 assert!(
5076 vr.is_valid(),
5077 "failures: {:?}",
5078 vr.failures().collect::<Vec<_>>()
5079 );
5080
5081 let reader = OverviewReader::open(tout.path()).unwrap();
5082 assert_eq!(reader.mode(), Mode::Partitioning);
5083
5084 assert_eq!(report.total_rows, geoms.len());
5086
5087 let mut all_ids = Vec::new();
5089 for level in 0..reader.num_levels() {
5090 for (id, _, _, _) in read_level_rows(&reader, level) {
5091 all_ids.push(id);
5092 }
5093 }
5094 all_ids.sort();
5095 assert_eq!(all_ids, (0..geoms.len() as i64).collect::<Vec<_>>());
5096
5097 for level in 0..reader.num_levels() {
5099 for (id, _, _, geom) in read_level_rows(&reader, level) {
5100 assert_eq!(geom, geoms[id as usize], "partitioning geometry verbatim");
5101 }
5102 }
5103 }
5104
5105 #[test]
5106 fn explicit_gsd_list_works() {
5107 let geoms = synthetic_geometries();
5108 let tin = tempfile::NamedTempFile::new().unwrap();
5109 let tout = tempfile::NamedTempFile::new().unwrap();
5110 write_input(tin.path(), &geoms, false, None);
5111
5112 let opts = ConvertOptions {
5113 mode: Mode::Duplicating,
5114 levels: LevelPlan::Gsds(vec![gsd(3), gsd(6), gsd(9)]),
5115 ..Default::default()
5116 };
5117 let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
5118 assert!(validate_file(tout.path()).unwrap().is_valid());
5119 for w in report.levels.windows(2) {
5121 assert!(w[0].gsd > w[1].gsd);
5122 }
5123 assert_eq!(report.levels.last().unwrap().feature_count, geoms.len());
5124 }
5125
5126 #[test]
5127 fn default_gsd_base_footer_gsds_match_const() {
5128 use crate::overview::level::gsd;
5132 let geoms = synthetic_geometries();
5133 let tin = tempfile::NamedTempFile::new().unwrap();
5134 let tout = tempfile::NamedTempFile::new().unwrap();
5135 write_input(tin.path(), &geoms, false, None);
5136
5137 let opts = ConvertOptions {
5138 mode: Mode::Duplicating,
5139 levels: LevelPlan::ZoomRange {
5140 min_zoom: 2,
5141 max_zoom: 8,
5142 },
5143 ..Default::default()
5144 };
5145 assert_eq!(
5146 opts.gsd_base, GSD_TILE_BASE,
5147 "default must be the const base"
5148 );
5149 convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
5150
5151 let reader = OverviewReader::open(tout.path()).unwrap();
5152 for level in &reader.meta().levels {
5154 let z = level.zoom.expect("zoom-range plan records zooms");
5155 assert_eq!(level.gsd, gsd(z), "footer gsd must match const gsd(z={z})");
5156 }
5157 assert_eq!(
5159 reader.meta().generalization.as_ref().unwrap().gsd_base,
5160 None,
5161 "default gsd_base must be absent from provenance"
5162 );
5163 }
5164
5165 #[test]
5166 fn nondefault_gsd_base_scales_footer_gsds() {
5167 use crate::overview::level::{gsd_with_base, GSD_TILE_BASE};
5170 let geoms = synthetic_geometries();
5171 let tin = tempfile::NamedTempFile::new().unwrap();
5172 let tout = tempfile::NamedTempFile::new().unwrap();
5173 write_input(tin.path(), &geoms, false, None);
5174
5175 let base = GSD_TILE_BASE * 2.0;
5176 let opts = ConvertOptions {
5177 mode: Mode::Duplicating,
5178 levels: LevelPlan::ZoomRange {
5179 min_zoom: 2,
5180 max_zoom: 8,
5181 },
5182 gsd_base: base,
5183 ..Default::default()
5184 };
5185 convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
5186
5187 let reader = OverviewReader::open(tout.path()).unwrap();
5188 assert!(validate_file(tout.path()).unwrap().is_valid());
5189 for level in &reader.meta().levels {
5190 let z = level.zoom.unwrap();
5191 assert_eq!(
5192 level.gsd,
5193 gsd_with_base(z, base),
5194 "scaled footer gsd (z={z})"
5195 );
5196 }
5197 assert_eq!(
5199 reader.meta().generalization.as_ref().unwrap().gsd_base,
5200 Some(base),
5201 "non-default gsd_base must be recorded"
5202 );
5203 }
5204
5205 #[test]
5206 fn rejects_unsupported_crs() {
5207 let geoms = synthetic_geometries();
5208 let tin = tempfile::NamedTempFile::new().unwrap();
5209 let tout = tempfile::NamedTempFile::new().unwrap();
5210
5211 let projjson = serde_json::json!({
5214 "type": "ProjectedCRS",
5215 "name": "UTM zone 33N",
5216 "id": { "authority": "EPSG", "code": 32633 }
5217 });
5218 let md = geoarrow::datatypes::Metadata::new(
5219 geoarrow::datatypes::Crs::from_projjson(projjson),
5220 None,
5221 );
5222 write_input(tin.path(), &geoms, false, Some(md));
5223
5224 let opts = ConvertOptions::default();
5225 let err = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap_err();
5226 assert!(
5227 matches!(err, ConvertError::UnsupportedCrs { .. }),
5228 "expected UnsupportedCrs, got {err:?}"
5229 );
5230 }
5231
5232 #[test]
5233 fn renames_existing_level_column() {
5234 let geoms = synthetic_geometries();
5239 let tin = tempfile::NamedTempFile::new().unwrap();
5240 let tout = tempfile::NamedTempFile::new().unwrap();
5241 write_input(tin.path(), &geoms, true, None);
5242
5243 let opts = ConvertOptions::default();
5244 convert_to_overviews(tin.path(), tout.path(), &opts)
5245 .expect("colliding `level` column must be auto-renamed, not rejected");
5246
5247 let report = crate::overview::check::validate_file(tout.path()).unwrap();
5250 assert!(
5251 report.is_valid(),
5252 "failures: {:?}",
5253 report.failures().collect::<Vec<_>>()
5254 );
5255 let names = output_column_names(tout.path());
5256 assert_eq!(
5257 names
5258 .iter()
5259 .filter(|n| n.eq_ignore_ascii_case("level"))
5260 .count(),
5261 1,
5262 "exactly one authoritative `level` column, names={names:?}"
5263 );
5264 assert!(
5265 names.iter().any(|n| n == "level_"),
5266 "renamed source column `level_` present, names={names:?}"
5267 );
5268 }
5269
5270 #[test]
5271 fn resolver_renames_and_loops_suffix() {
5272 let schema = Arc::new(Schema::new(vec![
5275 Field::new("id", DataType::Int64, false),
5276 Field::new("level", DataType::Int32, false),
5277 Field::new("level_", DataType::Int32, false),
5278 Field::new("geometry", DataType::Binary, false),
5279 ]));
5280 let mut opts = ConvertOptions::default();
5281 let (out, renames) = resolve_reserved_column_collisions(&schema, &mut opts);
5282 assert_eq!(renames, vec![("level".to_string(), "level__".to_string())]);
5283 let names: Vec<_> = out.fields().iter().map(|f| f.name().clone()).collect();
5284 assert_eq!(names, vec!["id", "level__", "level_", "geometry"]);
5285 }
5286
5287 #[test]
5288 fn resolver_rewrites_option_column_references() {
5289 let schema = Arc::new(Schema::new(vec![
5292 Field::new("level", DataType::Int32, false),
5293 Field::new("geometry", DataType::Binary, false),
5294 ]));
5295 let mut opts = ConvertOptions {
5296 sort_key: Some("LEVEL".to_string()),
5297 ..Default::default()
5298 };
5299 let (_out, renames) = resolve_reserved_column_collisions(&schema, &mut opts);
5300 assert_eq!(renames.len(), 1);
5301 assert_eq!(opts.sort_key.as_deref(), Some("level_"));
5302 }
5303
5304 #[test]
5309 fn resolver_rewrites_the_entry_zoom_column() {
5310 use super::super::ladder::{EntryZoomKind, EntryZoomSpec};
5311
5312 let schema = Arc::new(Schema::new(vec![
5313 Field::new("level", DataType::Float64, false),
5314 Field::new("geometry", DataType::Binary, false),
5315 ]));
5316 let mut opts = ConvertOptions {
5317 entry_zoom: Some(EntryZoomSpec {
5318 column: "level".to_string(),
5319 kind: EntryZoomKind::DenseRank { step: 1 },
5320 }),
5321 ..Default::default()
5322 };
5323 let (_out, renames) = resolve_reserved_column_collisions(&schema, &mut opts);
5324 assert_eq!(renames.len(), 1);
5325 assert_eq!(
5326 opts.entry_zoom.as_ref().unwrap().column,
5327 "level_",
5328 "the ladder column must follow the reserved-column rename"
5329 );
5330 }
5331
5332 #[test]
5333 fn resolver_noop_when_no_collision() {
5334 let schema = Arc::new(Schema::new(vec![
5337 Field::new("id", DataType::Int64, false),
5338 Field::new("geometry", DataType::Binary, false),
5339 ]));
5340 let mut opts = ConvertOptions::default();
5341 let (out, renames) = resolve_reserved_column_collisions(&schema, &mut opts);
5342 assert!(renames.is_empty());
5343 assert!(Arc::ptr_eq(&out, &schema));
5344 }
5345
5346 #[test]
5347 fn resolver_reserves_count_columns_only_when_enabled() {
5348 let schema = Arc::new(Schema::new(vec![
5351 Field::new("point_count", DataType::Int64, false),
5352 Field::new("geometry", DataType::Binary, false),
5353 ]));
5354 let mut off = ConvertOptions {
5356 cluster: false,
5357 coalesce_lines: false,
5358 ..Default::default()
5359 };
5360 let (_out, renames) = resolve_reserved_column_collisions(&schema, &mut off);
5361 assert!(
5362 renames.is_empty(),
5363 "point_count is a passthrough without --cluster"
5364 );
5365 let mut on = ConvertOptions {
5367 cluster: true,
5368 ..Default::default()
5369 };
5370 let (_out, renames) = resolve_reserved_column_collisions(&schema, &mut on);
5371 assert_eq!(
5372 renames,
5373 vec![("point_count".to_string(), "point_count_".to_string())]
5374 );
5375 }
5376
5377 fn write_class_input(path: &Path, geoms: &[Geometry<f64>], classes: &[Option<&str>]) {
5380 let n = geoms.len();
5381 assert_eq!(n, classes.len());
5382 let id = Int64Array::from((0..n as i64).collect::<Vec<_>>());
5383 let class = StringArray::from(classes.to_vec());
5384 let geom_arr = build_geometry_array(geoms);
5385 let geom_field = geom_arr.data_type().to_field("geometry", true);
5386
5387 let fields = vec![
5388 Arc::new(Field::new("id", DataType::Int64, false)),
5389 Arc::new(Field::new("road_class", DataType::Utf8, true)),
5390 Arc::new(geom_field),
5391 ];
5392 let columns: Vec<Arc<dyn Array>> =
5393 vec![Arc::new(id), Arc::new(class), geom_arr.to_array_ref()];
5394 let schema = Arc::new(Schema::new(fields));
5395 let batch = RecordBatch::try_new(schema.clone(), columns).unwrap();
5396
5397 let gpq_options = GeoParquetWriterOptionsBuilder::default()
5398 .set_encoding(GeoParquetWriterEncoding::WKB)
5399 .set_generate_covering(true)
5400 .build();
5401 let encoder = GeoParquetRecordBatchEncoder::try_new(&schema, &gpq_options).unwrap();
5402 let target_schema = encoder.target_schema();
5403 let file = std::fs::File::create(path).unwrap();
5404 let mut writer = ArrowWriter::try_new(file, target_schema, None).unwrap();
5405 let mut encoder = encoder;
5406 let encoded = encoder.encode_record_batch(&batch).unwrap();
5407 writer.write(&encoded).unwrap();
5408 writer.append_key_value_metadata(encoder.into_keyvalue().unwrap());
5409 writer.close().unwrap();
5410 }
5411
5412 fn min_level_by_id(reader: &OverviewReader) -> std::collections::HashMap<i64, usize> {
5414 use arrow_array::cast::AsArray;
5415 use arrow_array::types::Int64Type;
5416 let mut out: std::collections::HashMap<i64, usize> = std::collections::HashMap::new();
5417 for level in 0..reader.num_levels() {
5418 let rdr = reader.read_level(level, None).unwrap();
5419 for batch in rdr {
5420 let batch = batch.unwrap();
5421 let ids = batch
5422 .column(batch.schema().index_of("id").unwrap())
5423 .as_primitive::<Int64Type>()
5424 .clone();
5425 for i in 0..batch.num_rows() {
5426 let id = ids.value(i);
5427 out.entry(id)
5428 .and_modify(|l| *l = (*l).min(level))
5429 .or_insert(level);
5430 }
5431 }
5432 }
5433 out
5434 }
5435
5436 #[test]
5439 fn class_rank_maps_named_unknown_null() {
5440 let col = StringArray::from(vec![
5442 Some("motorway"),
5443 Some("driveway"), None, ]);
5446 let ranking = ClassRanking {
5447 column: "road_class".to_string(),
5448 ranks: vec![("motorway".to_string(), 5.0)],
5449 unknown_rank: 0.0,
5450 };
5451 let keys = extract_class_ranks(&col, &ranking).unwrap();
5452 assert_eq!(keys, vec![Some(5.0), Some(0.0), None]);
5453 }
5454
5455 #[test]
5456 fn class_rank_rejects_non_string_column() {
5457 let col = Float64Array::from(vec![1.0, 2.0]);
5458 let ranking = ClassRanking {
5459 column: "road_class".to_string(),
5460 ranks: vec![("x".to_string(), 1.0)],
5461 unknown_rank: 0.0,
5462 };
5463 let err = extract_class_ranks(&col, &ranking).unwrap_err();
5464 assert!(matches!(err, ConvertError::ClassRankColumnNotString { .. }));
5465 }
5466
5467 #[test]
5468 fn overture_road_ranking_spine_is_ordered() {
5469 let cr = overture_road_ranking("road_class".to_string());
5470 let rank = |v: &str| -> f64 {
5471 cr.ranks
5472 .iter()
5473 .find(|(k, _)| k == v)
5474 .map(|(_, r)| *r)
5475 .unwrap()
5476 };
5477 let spine = [
5479 "motorway",
5480 "trunk",
5481 "primary",
5482 "secondary",
5483 "tertiary",
5484 "residential",
5485 "unclassified",
5486 "service",
5487 ];
5488 for w in spine.windows(2) {
5489 assert!(rank(w[0]) > rank(w[1]), "{} !> {}", w[0], w[1]);
5490 }
5491 for tail in ["living_street", "footway", "path", "cycleway", "track"] {
5493 assert!(rank(tail) < rank("service"), "{tail} !< service");
5494 assert!(rank(tail) > cr.unknown_rank, "{tail} !> unknown");
5495 }
5496 assert!(cr.ranks.iter().all(|(k, _)| k != "standard_gauge"));
5498 }
5499
5500 #[test]
5503 fn sort_key_and_class_ranking_conflict() {
5504 let geoms = synthetic_geometries();
5505 let tin = tempfile::NamedTempFile::new().unwrap();
5506 let tout = tempfile::NamedTempFile::new().unwrap();
5507 write_input(tin.path(), &geoms, false, None);
5508
5509 let opts = ConvertOptions {
5510 sort_key: Some("rank".to_string()),
5511 class_ranking: Some(ClassRanking {
5512 column: "road_class".to_string(),
5513 ranks: vec![("motorway".to_string(), 1.0)],
5514 unknown_rank: 0.0,
5515 }),
5516 ..Default::default()
5517 };
5518 let err = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap_err();
5519 assert!(matches!(err, ConvertError::RankingConflict));
5520 }
5521
5522 fn overture_line_geoms() -> (Vec<Geometry<f64>>, Vec<Option<&'static str>>) {
5525 let classes = [
5526 "motorway",
5527 "trunk",
5528 "primary",
5529 "residential",
5530 "footway",
5531 "service",
5532 ];
5533 let mut geoms = Vec::new();
5534 let mut cls = Vec::new();
5535 for (i, c) in classes.iter().enumerate() {
5536 let base = i as f64 * 2.0; let ls = LineString::from(vec![(base, 0.0), (base + 0.5, 0.3)]);
5538 geoms.push(Geometry::LineString(ls));
5539 cls.push(Some(*c));
5540 }
5541 (geoms, cls)
5542 }
5543
5544 fn ranking_mode_of(path: &Path) -> String {
5545 let reader = OverviewReader::open(path).unwrap();
5546 reader
5547 .meta()
5548 .generalization
5549 .as_ref()
5550 .unwrap()
5551 .ranking
5552 .as_ref()
5553 .unwrap()
5554 .mode
5555 .clone()
5556 }
5557
5558 #[test]
5559 fn auto_detect_overture_roads_triggers() {
5560 let (geoms, classes) = overture_line_geoms();
5561 let tin = tempfile::NamedTempFile::new().unwrap();
5562 let tout = tempfile::NamedTempFile::new().unwrap();
5563 write_class_input(tin.path(), &geoms, &classes);
5564
5565 let opts = ConvertOptions {
5566 levels: LevelPlan::ZoomRange {
5567 min_zoom: 4,
5568 max_zoom: 10,
5569 },
5570 ..Default::default()
5571 };
5572 convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
5573 assert_eq!(ranking_mode_of(tout.path()), "auto-overture-roads");
5574 }
5575
5576 #[test]
5577 fn auto_detect_disabled_falls_back_to_size() {
5578 let (geoms, classes) = overture_line_geoms();
5579 let tin = tempfile::NamedTempFile::new().unwrap();
5580 let tout = tempfile::NamedTempFile::new().unwrap();
5581 write_class_input(tin.path(), &geoms, &classes);
5582
5583 let opts = ConvertOptions {
5584 levels: LevelPlan::ZoomRange {
5585 min_zoom: 4,
5586 max_zoom: 10,
5587 },
5588 no_auto_rank: true,
5589 ..Default::default()
5590 };
5591 convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
5592 assert_eq!(ranking_mode_of(tout.path()), "size-fallback");
5593 }
5594
5595 #[test]
5596 fn auto_detect_non_trigger_without_class_column() {
5597 let geoms = synthetic_geometries();
5599 let tin = tempfile::NamedTempFile::new().unwrap();
5600 let tout = tempfile::NamedTempFile::new().unwrap();
5601 write_input(tin.path(), &geoms, false, None);
5602
5603 let opts = ConvertOptions::default();
5604 convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
5605 assert_eq!(ranking_mode_of(tout.path()), "size-fallback");
5606 }
5607
5608 #[test]
5609 fn class_rank_provenance_recorded() {
5610 let (geoms, classes) = overture_line_geoms();
5611 let tin = tempfile::NamedTempFile::new().unwrap();
5612 let tout = tempfile::NamedTempFile::new().unwrap();
5613 write_class_input(tin.path(), &geoms, &classes);
5614
5615 let opts = ConvertOptions {
5616 levels: LevelPlan::ZoomRange {
5617 min_zoom: 4,
5618 max_zoom: 10,
5619 },
5620 class_ranking: Some(ClassRanking {
5621 column: "road_class".to_string(),
5622 ranks: vec![("motorway".to_string(), 5.0), ("footway".to_string(), 1.0)],
5623 unknown_rank: 0.0,
5624 }),
5625 ..Default::default()
5626 };
5627 convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
5628 let reader = OverviewReader::open(tout.path()).unwrap();
5629 let r = reader
5630 .meta()
5631 .generalization
5632 .as_ref()
5633 .unwrap()
5634 .ranking
5635 .clone()
5636 .unwrap();
5637 assert_eq!(r.mode, "class-ranking");
5638 assert_eq!(r.column.as_deref(), Some("road_class"));
5639 let ranks = r.ranks.unwrap();
5640 assert_eq!(ranks.len(), 2);
5641 assert_eq!(ranks.get("motorway"), Some(&5.0));
5642 assert_eq!(ranks.get("footway"), Some(&1.0));
5643 assert_eq!(r.unknown_rank, Some(0.0));
5644
5645 let json = overviews_footer_json(tout.path());
5648 assert!(
5649 json.contains(r#""ranks":{"footway":1.0,"motorway":5.0}"#),
5650 "footer must serialize ranks as an object map, got {json}"
5651 );
5652 }
5653
5654 #[test]
5657 fn high_class_small_feature_wins_coarse_cell() {
5658 let big_low = Geometry::LineString(LineString::from(vec![
5668 (-0.03, -0.04),
5669 (0.07, 0.06), ]));
5671 let small_high = Geometry::LineString(LineString::from(vec![
5672 (0.0, 0.0),
5673 (0.04, 0.02), ]));
5675 let geoms = vec![big_low, small_high];
5676 let classes = vec![Some("footway"), Some("motorway")];
5677
5678 let tin = tempfile::NamedTempFile::new().unwrap();
5679
5680 {
5682 let tout = tempfile::NamedTempFile::new().unwrap();
5683 write_class_input(tin.path(), &geoms, &classes);
5684 let opts = ConvertOptions {
5685 levels: LevelPlan::ZoomRange {
5686 min_zoom: 4,
5687 max_zoom: 10,
5688 },
5689 no_auto_rank: true, ..Default::default()
5691 };
5692 convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
5693 let reader = OverviewReader::open(tout.path()).unwrap();
5694 let ml = min_level_by_id(&reader);
5695 assert!(
5696 ml[&0] < ml[&1],
5697 "size fallback: big low-class (id0) should win coarse, got {ml:?}"
5698 );
5699 }
5700
5701 {
5703 let tout = tempfile::NamedTempFile::new().unwrap();
5704 write_class_input(tin.path(), &geoms, &classes);
5705 let opts = ConvertOptions {
5706 levels: LevelPlan::ZoomRange {
5707 min_zoom: 4,
5708 max_zoom: 10,
5709 },
5710 class_ranking: Some(ClassRanking {
5711 column: "road_class".to_string(),
5712 ranks: vec![("motorway".to_string(), 5.0), ("footway".to_string(), 1.0)],
5713 unknown_rank: 0.0,
5714 }),
5715 ..Default::default()
5716 };
5717 convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
5718 let reader = OverviewReader::open(tout.path()).unwrap();
5719 let ml = min_level_by_id(&reader);
5720 assert!(
5721 ml[&1] < ml[&0],
5722 "class ranking: small high-class (id1) must win coarse, got {ml:?}"
5723 );
5724 assert_eq!(ml[&1], 0, "high-class line should reach the coarsest level");
5725 }
5726 }
5727
5728 fn density_provenance_of(path: &Path) -> Option<crate::overview::level::DensityProvenance> {
5731 let reader = OverviewReader::open(path).unwrap();
5732 reader
5733 .meta()
5734 .generalization
5735 .as_ref()
5736 .unwrap()
5737 .density_drop
5738 .clone()
5739 }
5740
5741 fn grid_points(n: usize) -> Vec<Geometry<f64>> {
5744 (0..n)
5745 .map(|i| {
5746 let x = (i % 40) as f64 * 0.3 - 6.0;
5747 let y = (i / 40) as f64 * 0.3 - 6.0;
5748 Geometry::Point(Point::new(x, y))
5749 })
5750 .collect()
5751 }
5752
5753 #[test]
5754 fn density_provenance_recorded_by_default() {
5755 let geoms = synthetic_geometries();
5756 let tin = tempfile::NamedTempFile::new().unwrap();
5757 let tout = tempfile::NamedTempFile::new().unwrap();
5758 write_input(tin.path(), &geoms, false, None);
5759
5760 let opts = ConvertOptions {
5761 levels: LevelPlan::ZoomRange {
5762 min_zoom: 2,
5763 max_zoom: 8,
5764 },
5765 ..Default::default()
5766 };
5767 convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
5768 let d = density_provenance_of(tout.path()).expect("default run records density_drop");
5769 assert_eq!(d.drop_rate, DensityBudgetConfig::default().drop_rate);
5770 assert_eq!(d.gamma, DensityBudgetConfig::default().gamma);
5771 assert_eq!(d.supercell_gsd_factor, SUPERCELL_GSD_FACTOR);
5772 }
5773
5774 #[test]
5775 fn density_disabled_omits_provenance_and_keeps_canonical() {
5776 let geoms = grid_points(600);
5777 let tin = tempfile::NamedTempFile::new().unwrap();
5778 let tout = tempfile::NamedTempFile::new().unwrap();
5779 write_input(tin.path(), &geoms, false, None);
5780
5781 let opts = ConvertOptions {
5782 levels: LevelPlan::ZoomRange {
5783 min_zoom: 0,
5784 max_zoom: 8,
5785 },
5786 density: DensityBudgetConfig {
5787 enabled: false,
5788 ..DensityBudgetConfig::default()
5789 },
5790 ..Default::default()
5791 };
5792 let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
5793 assert!(
5795 density_provenance_of(tout.path()).is_none(),
5796 "disabled budget must not emit provenance"
5797 );
5798 assert!(validate_file(tout.path()).unwrap().is_valid());
5799 assert_eq!(report.levels.last().unwrap().feature_count, geoms.len());
5800 }
5801
5802 #[test]
5803 fn density_thins_midlevels_keeps_canonical() {
5804 let geoms = grid_points(600);
5805 let tin = tempfile::NamedTempFile::new().unwrap();
5806 let tout_on = tempfile::NamedTempFile::new().unwrap();
5807 let tout_off = tempfile::NamedTempFile::new().unwrap();
5808 write_input(tin.path(), &geoms, false, None);
5809
5810 let base = ConvertOptions {
5811 levels: LevelPlan::ZoomRange {
5812 min_zoom: 0,
5813 max_zoom: 8,
5814 },
5815 ..Default::default()
5816 };
5817 let on = convert_to_overviews(tin.path(), tout_on.path(), &base).unwrap();
5818 let off = convert_to_overviews(
5819 tin.path(),
5820 tout_off.path(),
5821 &ConvertOptions {
5822 density: DensityBudgetConfig {
5823 enabled: false,
5824 ..DensityBudgetConfig::default()
5825 },
5826 ..base.clone()
5827 },
5828 )
5829 .unwrap();
5830
5831 assert_eq!(on.levels.last().unwrap().feature_count, geoms.len());
5833 assert_eq!(off.levels.last().unwrap().feature_count, geoms.len());
5834 assert!(validate_file(tout_on.path()).unwrap().is_valid());
5835
5836 assert!(
5839 on.total_rows < off.total_rows,
5840 "density budget should thin mid levels: on={} off={}",
5841 on.total_rows,
5842 off.total_rows
5843 );
5844 for w in on.levels.windows(2) {
5846 assert!(w[0].feature_count <= w[1].feature_count);
5847 }
5848 }
5849
5850 fn overviews_footer_json(path: &Path) -> String {
5854 let file = std::fs::File::open(path).unwrap();
5855 let b = ParquetRecordBatchReaderBuilder::try_new(file).unwrap();
5856 b.metadata()
5857 .file_metadata()
5858 .key_value_metadata()
5859 .unwrap()
5860 .iter()
5861 .find(|kv| kv.key == crate::overview::level::OVERVIEWS_KEY)
5862 .expect("geo:overviews key")
5863 .value
5864 .clone()
5865 .unwrap()
5866 }
5867
5868 fn assert_streaming_equivalent(input: &Path, base: &ConvertOptions) {
5873 let mem_out = tempfile::NamedTempFile::new().unwrap();
5874 let stream_out = tempfile::NamedTempFile::new().unwrap();
5875
5876 let mem_opts = ConvertOptions {
5877 streaming: false,
5878 ..base.clone()
5879 };
5880 let stream_opts = ConvertOptions {
5881 streaming: true,
5882 ..base.clone()
5883 };
5884 let mem = convert_to_overviews(input, mem_out.path(), &mem_opts).unwrap();
5885 let strm = convert_to_overviews(input, stream_out.path(), &stream_opts).unwrap();
5886
5887 assert_eq!(mem.mode, strm.mode);
5889 assert_eq!(mem.input_features, strm.input_features);
5890 assert_eq!(mem.total_rows, strm.total_rows);
5891 assert_eq!(mem.total_vertices, strm.total_vertices);
5892 assert_eq!(mem.levels.len(), strm.levels.len());
5893 for (a, b) in mem.levels.iter().zip(&strm.levels) {
5894 assert_eq!(a.level, b.level);
5895 assert_eq!(a.gsd, b.gsd, "level {} gsd", a.level);
5896 assert_eq!(a.zoom, b.zoom, "level {} zoom", a.level);
5897 assert_eq!(
5898 a.feature_count, b.feature_count,
5899 "level {} feature count",
5900 a.level
5901 );
5902 assert_eq!(
5903 a.vertex_count, b.vertex_count,
5904 "level {} vertex count",
5905 a.level
5906 );
5907 }
5908
5909 assert_eq!(
5911 overviews_footer_json(mem_out.path()),
5912 overviews_footer_json(stream_out.path()),
5913 "geo:overviews footers differ"
5914 );
5915 assert!(validate_file(mem_out.path()).unwrap().is_valid());
5916 assert!(validate_file(stream_out.path()).unwrap().is_valid());
5917
5918 let mr = OverviewReader::open(mem_out.path()).unwrap();
5920 let sr = OverviewReader::open(stream_out.path()).unwrap();
5921 assert_eq!(mr.num_levels(), sr.num_levels());
5922 for level in 0..mr.num_levels() {
5923 assert_eq!(
5924 read_level_rows(&mr, level),
5925 read_level_rows(&sr, level),
5926 "level {level} rows differ"
5927 );
5928 }
5929 }
5930
5931 fn assert_outputs_equivalent(a: &Path, b: &Path, ctx: &str) {
5939 use parquet::file::reader::{FileReader, SerializedFileReader};
5940
5941 let layout = |p: &Path| -> Vec<(i64, usize)> {
5943 let r = SerializedFileReader::new(std::fs::File::open(p).unwrap()).unwrap();
5944 let md = r.metadata();
5945 (0..md.num_row_groups())
5946 .map(|i| (md.row_group(i).num_rows(), md.row_group(i).columns().len()))
5947 .collect()
5948 };
5949 assert_eq!(layout(a), layout(b), "{ctx}: row-group layout differs");
5950
5951 assert_eq!(
5953 overviews_footer_json(a),
5954 overviews_footer_json(b),
5955 "{ctx}: geo:overviews footer differs"
5956 );
5957
5958 let ra = OverviewReader::open(a).unwrap();
5960 let rb = OverviewReader::open(b).unwrap();
5961 assert_eq!(
5962 ra.num_levels(),
5963 rb.num_levels(),
5964 "{ctx}: level count differs"
5965 );
5966 for level in 0..ra.num_levels() {
5967 assert_eq!(
5968 read_level_rows(&ra, level),
5969 read_level_rows(&rb, level),
5970 "{ctx}: level {level} rows differ"
5971 );
5972 }
5973 }
5974
5975 #[test]
5981 fn pipelined_matches_serial() {
5982 use super::super::level::MemoryProfile;
5983 use super::super::stream::Pass2Strategy;
5984
5985 let poly_in = tempfile::NamedTempFile::new().unwrap();
5989 write_input(poly_in.path(), &synthetic_geometries(), false, None);
5990 let grid_in = tempfile::NamedTempFile::new().unwrap();
5991 write_input(grid_in.path(), &grid_points(600), false, None);
5992
5993 let cases: Vec<(&str, &Path, ConvertOptions)> = vec![
5994 (
5995 "duplicating",
5996 poly_in.path(),
5997 ConvertOptions {
5998 mode: Mode::Duplicating,
5999 levels: LevelPlan::ZoomRange {
6000 min_zoom: 1,
6001 max_zoom: 10,
6002 },
6003 ..Default::default()
6004 },
6005 ),
6006 (
6007 "partitioning",
6008 grid_in.path(),
6009 ConvertOptions {
6010 mode: Mode::Partitioning,
6011 levels: LevelPlan::ZoomRange {
6012 min_zoom: 1,
6013 max_zoom: 9,
6014 },
6015 ..Default::default()
6016 },
6017 ),
6018 (
6019 "clustering",
6020 grid_in.path(),
6021 ConvertOptions {
6022 mode: Mode::Duplicating,
6023 cluster: true,
6024 levels: LevelPlan::ZoomRange {
6025 min_zoom: 1,
6026 max_zoom: 9,
6027 },
6028 ..Default::default()
6029 },
6030 ),
6031 ];
6032
6033 for (name, input, base) in &cases {
6034 let serial_out = tempfile::NamedTempFile::new().unwrap();
6035 convert_to_overviews_strategy(input, serial_out.path(), base, Pass2Strategy::Serial)
6036 .unwrap();
6037
6038 for profile in [
6043 MemoryProfile::Speed,
6044 MemoryProfile::Bounded,
6045 MemoryProfile::Auto,
6046 ] {
6047 let opts = ConvertOptions {
6048 profile,
6049 ..base.clone()
6050 };
6051 let piped_out = tempfile::NamedTempFile::new().unwrap();
6052 convert_to_overviews_strategy(
6053 input,
6054 piped_out.path(),
6055 &opts,
6056 Pass2Strategy::Pipelined,
6057 )
6058 .unwrap();
6059 assert_outputs_equivalent(
6060 serial_out.path(),
6061 piped_out.path(),
6062 &format!("{name}/{profile:?}"),
6063 );
6064 }
6065 }
6066 }
6067
6068 #[test]
6073 fn pipelined_invariant_to_batching_knobs() {
6074 use super::super::stream::Pass2Strategy;
6075
6076 let geoms = grid_points(600);
6077 let tin = tempfile::NamedTempFile::new().unwrap();
6078 write_input(tin.path(), &geoms, false, None);
6079
6080 let base = ConvertOptions {
6081 mode: Mode::Duplicating,
6082 levels: LevelPlan::ZoomRange {
6083 min_zoom: 1,
6084 max_zoom: 9,
6085 },
6086 ..Default::default()
6087 };
6088
6089 let convert = |read_batch_size: usize, in_flight_batches: usize| {
6090 let opts = ConvertOptions {
6091 read_batch_size,
6092 in_flight_batches,
6093 ..base.clone()
6094 };
6095 let out = tempfile::NamedTempFile::new().unwrap();
6096 convert_to_overviews_strategy(tin.path(), out.path(), &opts, Pass2Strategy::Pipelined)
6097 .unwrap();
6098 out
6099 };
6100
6101 let reference = convert(7, 1);
6102 for (rbs, ifb) in [(64usize, 4usize), (4096, 8), (8192, 0)] {
6106 let candidate = convert(rbs, ifb);
6107 assert_outputs_equivalent(
6108 reference.path(),
6109 candidate.path(),
6110 &format!("read_batch_size={rbs} in_flight_batches={ifb}"),
6111 );
6112 }
6113 }
6114
6115 #[test]
6116 fn resolve_in_flight_batches_auto_and_explicit() {
6117 let auto = resolve_in_flight_batches(IN_FLIGHT_BATCHES_AUTO);
6119 assert!(
6120 (IN_FLIGHT_BATCHES_MIN..=IN_FLIGHT_BATCHES_MAX).contains(&auto),
6121 "auto in-flight {auto} outside [{IN_FLIGHT_BATCHES_MIN}, {IN_FLIGHT_BATCHES_MAX}]"
6122 );
6123 let cores = std::thread::available_parallelism()
6124 .map(|n| n.get())
6125 .unwrap_or(IN_FLIGHT_BATCHES_MIN);
6126 assert_eq!(
6127 auto,
6128 cores.clamp(IN_FLIGHT_BATCHES_MIN, IN_FLIGHT_BATCHES_MAX),
6129 "auto must equal core count clamped to the configured range"
6130 );
6131 assert_eq!(resolve_in_flight_batches(1), 1);
6133 assert_eq!(resolve_in_flight_batches(7), 7);
6134 assert_eq!(
6135 resolve_in_flight_batches(IN_FLIGHT_BATCHES_MAX + 100),
6136 IN_FLIGHT_BATCHES_MAX + 100
6137 );
6138 }
6139
6140 #[test]
6141 fn streaming_matches_in_memory_duplicating() {
6142 let geoms = synthetic_geometries();
6143 let tin = tempfile::NamedTempFile::new().unwrap();
6144 write_input(tin.path(), &geoms, false, None);
6145
6146 let base = ConvertOptions {
6147 mode: Mode::Duplicating,
6148 levels: LevelPlan::ZoomRange {
6149 min_zoom: 1,
6150 max_zoom: 10,
6151 },
6152 ..Default::default()
6153 };
6154 assert_streaming_equivalent(tin.path(), &base);
6155 }
6156
6157 #[test]
6158 fn streaming_matches_in_memory_partitioning() {
6159 let geoms = synthetic_geometries();
6160 let tin = tempfile::NamedTempFile::new().unwrap();
6161 write_input(tin.path(), &geoms, false, None);
6162
6163 let base = ConvertOptions {
6164 mode: Mode::Partitioning,
6165 levels: LevelPlan::ZoomRange {
6166 min_zoom: 2,
6167 max_zoom: 8,
6168 },
6169 ..Default::default()
6170 };
6171 assert_streaming_equivalent(tin.path(), &base);
6172 }
6173
6174 #[test]
6175 fn streaming_matches_with_small_read_batches_and_density_budget() {
6176 let geoms = grid_points(600);
6179 let tin = tempfile::NamedTempFile::new().unwrap();
6180 write_input(tin.path(), &geoms, false, None);
6181
6182 let base = ConvertOptions {
6183 levels: LevelPlan::ZoomRange {
6184 min_zoom: 0,
6185 max_zoom: 8,
6186 },
6187 read_batch_size: 7,
6188 ..Default::default()
6189 };
6190 assert_streaming_equivalent(tin.path(), &base);
6191 }
6192
6193 #[test]
6194 fn streaming_matches_with_explicit_sort_key() {
6195 let geoms = synthetic_geometries();
6196 let tin = tempfile::NamedTempFile::new().unwrap();
6197 write_input(tin.path(), &geoms, false, None);
6198
6199 let base = ConvertOptions {
6200 levels: LevelPlan::ZoomRange {
6201 min_zoom: 2,
6202 max_zoom: 8,
6203 },
6204 sort_key: Some("rank".to_string()),
6205 ..Default::default()
6206 };
6207 assert_streaming_equivalent(tin.path(), &base);
6208 }
6209
6210 #[test]
6211 fn streaming_auto_rank_matches_in_memory() {
6212 let (geoms, classes) = overture_line_geoms();
6215 let tin = tempfile::NamedTempFile::new().unwrap();
6216 write_class_input(tin.path(), &geoms, &classes);
6217
6218 let base = ConvertOptions {
6219 levels: LevelPlan::ZoomRange {
6220 min_zoom: 4,
6221 max_zoom: 10,
6222 },
6223 read_batch_size: 2,
6224 ..Default::default()
6225 };
6226 let mem_out = tempfile::NamedTempFile::new().unwrap();
6227 let stream_out = tempfile::NamedTempFile::new().unwrap();
6228 convert_to_overviews(
6229 tin.path(),
6230 mem_out.path(),
6231 &ConvertOptions {
6232 streaming: false,
6233 ..base.clone()
6234 },
6235 )
6236 .unwrap();
6237 convert_to_overviews(tin.path(), stream_out.path(), &base).unwrap();
6238
6239 assert_eq!(ranking_mode_of(mem_out.path()), "auto-overture-roads");
6240 assert_eq!(ranking_mode_of(stream_out.path()), "auto-overture-roads");
6241 assert_eq!(
6242 overviews_footer_json(mem_out.path()),
6243 overviews_footer_json(stream_out.path())
6244 );
6245 let mr = OverviewReader::open(mem_out.path()).unwrap();
6246 let sr = OverviewReader::open(stream_out.path()).unwrap();
6247 assert_eq!(min_level_by_id(&mr), min_level_by_id(&sr));
6248 }
6249
6250 fn read_point_counts(reader: &OverviewReader, level: usize) -> Vec<i64> {
6254 use arrow_array::cast::AsArray;
6255 use arrow_array::types::Int64Type;
6256 let rdr = reader.read_level(level, None).unwrap();
6257 let mut out = Vec::new();
6258 for batch in rdr {
6259 let batch = batch.unwrap();
6260 let idx = batch.schema().index_of("point_count").unwrap();
6261 let col = batch.column(idx).as_primitive::<Int64Type>().clone();
6262 assert_eq!(col.null_count(), 0, "point_count must be NOT NULL");
6263 out.extend(col.values().iter().copied());
6264 }
6265 out
6266 }
6267
6268 fn read_f64_column(reader: &OverviewReader, level: usize, name: &str) -> Vec<Option<f64>> {
6270 use arrow_array::cast::AsArray;
6271 use arrow_array::types::Float64Type;
6272 let rdr = reader.read_level(level, None).unwrap();
6273 let mut out = Vec::new();
6274 for batch in rdr {
6275 let batch = batch.unwrap();
6276 let idx = batch.schema().index_of(name).unwrap();
6277 let col = batch.column(idx).as_primitive::<Float64Type>().clone();
6278 for i in 0..col.len() {
6279 out.push(if col.is_null(i) {
6280 None
6281 } else {
6282 Some(col.value(i))
6283 });
6284 }
6285 }
6286 out
6287 }
6288
6289 #[test]
6290 fn cluster_duplicating_end_to_end_counts_partition_every_level() {
6291 let geoms = grid_points(600);
6294 let tin = tempfile::NamedTempFile::new().unwrap();
6295 let tout = tempfile::NamedTempFile::new().unwrap();
6296 write_input(tin.path(), &geoms, false, None);
6297
6298 let opts = ConvertOptions {
6299 levels: LevelPlan::ZoomRange {
6300 min_zoom: 0,
6301 max_zoom: 8,
6302 },
6303 cluster: true,
6304 ..Default::default()
6305 };
6306 let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
6307 let vr = validate_file(tout.path()).unwrap();
6308 assert!(
6309 vr.is_valid(),
6310 "failures: {:?}",
6311 vr.failures().collect::<Vec<_>>()
6312 );
6313
6314 let reader = OverviewReader::open(tout.path()).unwrap();
6315 let canonical = reader.num_levels() - 1;
6316 for level in 0..reader.num_levels() {
6317 let counts = read_point_counts(&reader, level);
6318 assert!(counts.iter().all(|&c| c >= 1), "level {level} counts >= 1");
6319 assert_eq!(
6320 counts.iter().sum::<i64>(),
6321 geoms.len() as i64,
6322 "level {level}: sum(point_count) must equal source count"
6323 );
6324 }
6325 assert!(read_point_counts(&reader, canonical)
6327 .iter()
6328 .all(|&c| c == 1));
6329 let coarse = read_point_counts(&reader, 0);
6332 assert!(coarse.iter().any(|&c| c > 1), "no clustering happened");
6333 assert!(coarse.len() < geoms.len());
6334 assert_eq!(report.levels.last().unwrap().feature_count, geoms.len());
6335 }
6336
6337 #[test]
6345 fn cluster_sum_invariant_across_knob_combinations() {
6346 let geoms = grid_points(300);
6347 let tin = tempfile::NamedTempFile::new().unwrap();
6348 write_input(tin.path(), &geoms, false, None);
6349
6350 for streaming in [true, false] {
6351 for density in [true, false] {
6352 for thinning in [2.0, 8.0] {
6353 let tout = tempfile::NamedTempFile::new().unwrap();
6354 let mut opts = ConvertOptions {
6355 levels: LevelPlan::ZoomRange {
6356 min_zoom: 0,
6357 max_zoom: 6,
6358 },
6359 cluster: true,
6360 streaming,
6361 ..Default::default()
6362 };
6363 opts.density.enabled = density;
6364 opts.assign.point_thinning = thinning;
6365 let label =
6366 format!("streaming={streaming} density={density} thinning={thinning}");
6367 convert_to_overviews(tin.path(), tout.path(), &opts)
6368 .unwrap_or_else(|e| panic!("{label}: {e}"));
6369
6370 let vr = validate_file(tout.path()).unwrap();
6371 assert!(
6372 vr.is_valid(),
6373 "{label}: failures: {:?}",
6374 vr.failures().collect::<Vec<_>>()
6375 );
6376 assert_eq!(
6377 vr.check_passed("cluster_sum_invariant"),
6378 Some(true),
6379 "{label}"
6380 );
6381
6382 let reader = OverviewReader::open(tout.path()).unwrap();
6383 let canonical = reader.num_levels() - 1;
6384 for level in 0..reader.num_levels() {
6385 let counts = read_point_counts(&reader, level);
6386 assert!(
6387 !counts.is_empty(),
6388 "{label}: level {level} thinned points to zero"
6389 );
6390 assert_eq!(
6391 counts.iter().sum::<i64>(),
6392 geoms.len() as i64,
6393 "{label}: level {level} must partition the source set"
6394 );
6395 }
6396 assert!(
6397 read_point_counts(&reader, canonical)
6398 .iter()
6399 .all(|&c| c == 1),
6400 "{label}: canonical band must be singleton-only"
6401 );
6402 }
6403 }
6404 }
6405 }
6406
6407 #[test]
6408 fn cluster_accumulate_sum_and_mean_consistent() {
6409 let geoms = grid_points(400);
6414 let n = geoms.len();
6415 let tin = tempfile::NamedTempFile::new().unwrap();
6416 write_input(tin.path(), &geoms, false, None);
6417 let source_sum: f64 = (1..=n).map(|v| v as f64).sum();
6418
6419 let tout = tempfile::NamedTempFile::new().unwrap();
6421 let opts = ConvertOptions {
6422 levels: LevelPlan::ZoomRange {
6423 min_zoom: 0,
6424 max_zoom: 8,
6425 },
6426 cluster: true,
6427 accumulate: vec![AccumulateSpec {
6428 column: "rank".to_string(),
6429 op: super::super::cluster::AccumulateOp::Sum,
6430 }],
6431 ..Default::default()
6432 };
6433 convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
6434 let reader = OverviewReader::open(tout.path()).unwrap();
6435 for level in 0..reader.num_levels() {
6436 let ranks = read_f64_column(&reader, level, "rank");
6437 let total: f64 = ranks.iter().flatten().sum();
6438 assert!(
6439 (total - source_sum).abs() < 1e-6,
6440 "level {level}: rank sum {total} != source {source_sum}"
6441 );
6442 }
6443 let canonical = reader.num_levels() - 1;
6445 let rows = read_level_rows(&reader, canonical);
6446 for (id, _, rank, _) in rows {
6447 assert_eq!(rank, (n as i64 - id) as f64, "canonical rank verbatim");
6448 }
6449
6450 let tout2 = tempfile::NamedTempFile::new().unwrap();
6453 let opts_mean = ConvertOptions {
6454 accumulate: vec![AccumulateSpec {
6455 column: "rank".to_string(),
6456 op: super::super::cluster::AccumulateOp::Mean,
6457 }],
6458 ..opts.clone()
6459 };
6460 convert_to_overviews(tin.path(), tout2.path(), &opts_mean).unwrap();
6461 let reader2 = OverviewReader::open(tout2.path()).unwrap();
6462 for level in 0..reader2.num_levels() {
6463 let means = read_f64_column(&reader2, level, "rank");
6464 let counts = read_point_counts(&reader2, level);
6465 let total: f64 = means
6466 .iter()
6467 .zip(&counts)
6468 .map(|(m, &c)| m.unwrap() * c as f64)
6469 .sum();
6470 assert!(
6471 (total - source_sum).abs() < 1e-6,
6472 "level {level}: Σ mean×count {total} != source {source_sum}"
6473 );
6474 }
6475 }
6476
6477 #[test]
6478 fn cluster_footer_provenance_recorded() {
6479 let geoms = grid_points(100);
6480 let tin = tempfile::NamedTempFile::new().unwrap();
6481 let tout = tempfile::NamedTempFile::new().unwrap();
6482 write_input(tin.path(), &geoms, false, None);
6483
6484 let opts = ConvertOptions {
6485 cluster: true,
6486 accumulate: vec![AccumulateSpec {
6487 column: "rank".to_string(),
6488 op: super::super::cluster::AccumulateOp::Mean,
6489 }],
6490 ..Default::default()
6491 };
6492 convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
6493 let reader = OverviewReader::open(tout.path()).unwrap();
6494 let c = reader
6495 .meta()
6496 .generalization
6497 .as_ref()
6498 .unwrap()
6499 .clustering
6500 .clone()
6501 .expect("clustering provenance recorded");
6502 assert!(c.enabled);
6503 assert_eq!(c.point_count_column, "point_count");
6504 assert_eq!(c.accumulated.len(), 1);
6505 assert_eq!(c.accumulated[0].column, "rank");
6506 assert_eq!(c.accumulated[0].op, "mean");
6507
6508 let tout_off = tempfile::NamedTempFile::new().unwrap();
6510 convert_to_overviews(tin.path(), tout_off.path(), &ConvertOptions::default()).unwrap();
6511 let r_off = OverviewReader::open(tout_off.path()).unwrap();
6512 assert!(r_off
6513 .meta()
6514 .generalization
6515 .as_ref()
6516 .unwrap()
6517 .clustering
6518 .is_none());
6519 }
6520
6521 #[test]
6522 fn cluster_option_errors() {
6523 let geoms = grid_points(20);
6524 let tin = tempfile::NamedTempFile::new().unwrap();
6525 let tout = tempfile::NamedTempFile::new().unwrap();
6526 write_input(tin.path(), &geoms, false, None);
6527
6528 let err = convert_to_overviews(
6530 tin.path(),
6531 tout.path(),
6532 &ConvertOptions {
6533 mode: Mode::Partitioning,
6534 cluster: true,
6535 ..Default::default()
6536 },
6537 )
6538 .unwrap_err();
6539 assert!(matches!(err, ConvertError::ClusterPartitioningUnsupported));
6540
6541 let err = convert_to_overviews(
6543 tin.path(),
6544 tout.path(),
6545 &ConvertOptions {
6546 accumulate: vec![AccumulateSpec {
6547 column: "rank".to_string(),
6548 op: super::super::cluster::AccumulateOp::Sum,
6549 }],
6550 ..Default::default()
6551 },
6552 )
6553 .unwrap_err();
6554 assert!(matches!(err, ConvertError::AccumulateWithoutCluster));
6555
6556 for streaming in [true, false] {
6558 let err = convert_to_overviews(
6559 tin.path(),
6560 tout.path(),
6561 &ConvertOptions {
6562 cluster: true,
6563 streaming,
6564 accumulate: vec![AccumulateSpec {
6565 column: "nonexistent".to_string(),
6566 op: super::super::cluster::AccumulateOp::Sum,
6567 }],
6568 ..Default::default()
6569 },
6570 )
6571 .unwrap_err();
6572 assert!(
6573 matches!(err, ConvertError::AccumulateColumnMissing { .. }),
6574 "streaming={streaming}: got {err:?}"
6575 );
6576 }
6577
6578 let err = convert_to_overviews(
6580 tin.path(),
6581 tout.path(),
6582 &ConvertOptions {
6583 cluster: true,
6584 accumulate: vec![AccumulateSpec {
6585 column: "name".to_string(),
6586 op: super::super::cluster::AccumulateOp::Max,
6587 }],
6588 ..Default::default()
6589 },
6590 )
6591 .unwrap_err();
6592 assert!(matches!(
6593 err,
6594 ConvertError::AccumulateColumnNotNumeric { .. }
6595 ));
6596 }
6597
6598 #[test]
6599 fn cluster_renames_existing_point_count_column() {
6600 let geoms = grid_points(10);
6604 let n = geoms.len();
6605 let tin = tempfile::NamedTempFile::new().unwrap();
6606 {
6607 let id = Int64Array::from((0..n as i64).collect::<Vec<_>>());
6608 let pc = Int64Array::from(vec![7i64; n]);
6609 let geom_arr = build_geometry_array(&geoms);
6610 let geom_field = geom_arr.data_type().to_field("geometry", true);
6611 let fields = vec![
6612 Arc::new(Field::new("id", DataType::Int64, false)),
6613 Arc::new(Field::new("Point_Count", DataType::Int64, false)),
6614 Arc::new(geom_field),
6615 ];
6616 let columns: Vec<Arc<dyn Array>> =
6617 vec![Arc::new(id), Arc::new(pc), geom_arr.to_array_ref()];
6618 let schema = Arc::new(Schema::new(fields));
6619 let batch = RecordBatch::try_new(schema.clone(), columns).unwrap();
6620 let gpq_options = GeoParquetWriterOptionsBuilder::default()
6621 .set_encoding(GeoParquetWriterEncoding::WKB)
6622 .set_generate_covering(true)
6623 .build();
6624 let mut encoder = GeoParquetRecordBatchEncoder::try_new(&schema, &gpq_options).unwrap();
6625 let target_schema = encoder.target_schema();
6626 let file = std::fs::File::create(tin.path()).unwrap();
6627 let mut writer = ArrowWriter::try_new(file, target_schema, None).unwrap();
6628 writer
6629 .write(&encoder.encode_record_batch(&batch).unwrap())
6630 .unwrap();
6631 writer.append_key_value_metadata(encoder.into_keyvalue().unwrap());
6632 writer.close().unwrap();
6633 }
6634
6635 let tout = tempfile::NamedTempFile::new().unwrap();
6636 convert_to_overviews(
6637 tin.path(),
6638 tout.path(),
6639 &ConvertOptions {
6640 cluster: true,
6641 ..Default::default()
6642 },
6643 )
6644 .expect("colliding `Point_Count` column must be auto-renamed, not rejected");
6645 let names = output_column_names(tout.path());
6646 assert!(
6647 names.iter().any(|n| n == "Point_Count_"),
6648 "renamed source column present, names={names:?}"
6649 );
6650 assert_eq!(
6651 names
6652 .iter()
6653 .filter(|n| n.eq_ignore_ascii_case("point_count"))
6654 .count(),
6655 1,
6656 "one authoritative `point_count`, names={names:?}"
6657 );
6658
6659 convert_to_overviews(tin.path(), tout.path(), &ConvertOptions::default()).unwrap();
6662 let names = output_column_names(tout.path());
6663 assert!(
6664 names.iter().any(|n| n == "Point_Count"),
6665 "passthrough column kept verbatim, names={names:?}"
6666 );
6667 }
6668
6669 #[test]
6670 fn streaming_matches_in_memory_clustering() {
6671 let geoms = grid_points(600);
6675 let tin = tempfile::NamedTempFile::new().unwrap();
6676 write_input(tin.path(), &geoms, false, None);
6677
6678 let base = ConvertOptions {
6679 levels: LevelPlan::ZoomRange {
6680 min_zoom: 0,
6681 max_zoom: 8,
6682 },
6683 read_batch_size: 7,
6684 cluster: true,
6685 accumulate: vec![
6686 AccumulateSpec {
6687 column: "rank".to_string(),
6688 op: super::super::cluster::AccumulateOp::Sum,
6689 },
6690 AccumulateSpec {
6691 column: "rank".to_string(),
6692 op: super::super::cluster::AccumulateOp::Mean,
6693 },
6694 ],
6695 ..Default::default()
6696 };
6697 assert_streaming_equivalent(tin.path(), &base);
6700
6701 let mem_out = tempfile::NamedTempFile::new().unwrap();
6704 let stream_out = tempfile::NamedTempFile::new().unwrap();
6705 convert_to_overviews(
6706 tin.path(),
6707 mem_out.path(),
6708 &ConvertOptions {
6709 streaming: false,
6710 ..base.clone()
6711 },
6712 )
6713 .unwrap();
6714 convert_to_overviews(tin.path(), stream_out.path(), &base).unwrap();
6715 let mr = OverviewReader::open(mem_out.path()).unwrap();
6716 let sr = OverviewReader::open(stream_out.path()).unwrap();
6717 for level in 0..mr.num_levels() {
6718 assert_eq!(
6719 read_point_counts(&mr, level),
6720 read_point_counts(&sr, level),
6721 "level {level} point_count differs"
6722 );
6723 }
6724 }
6725
6726 fn read_coalesced_counts(reader: &OverviewReader, level: usize) -> Vec<i32> {
6730 use arrow_array::cast::AsArray;
6731 use arrow_array::types::Int32Type;
6732 let rdr = reader.read_level(level, None).unwrap();
6733 let mut out = Vec::new();
6734 for batch in rdr {
6735 let batch = batch.unwrap();
6736 let idx = batch.schema().index_of("coalesced_count").unwrap();
6737 let col = batch.column(idx).as_primitive::<Int32Type>().clone();
6738 assert_eq!(col.null_count(), 0, "coalesced_count must be NOT NULL");
6739 out.extend(col.values().iter().copied());
6740 }
6741 out
6742 }
6743
6744 fn fragment_chain_geoms(n: usize) -> Vec<Geometry<f64>> {
6748 let mut geoms: Vec<Geometry<f64>> = (0..n)
6749 .map(|i| {
6750 let x0 = i as f64 * 0.01;
6751 Geometry::LineString(LineString::from(vec![(x0, 0.0), (x0 + 0.01, 0.0)]))
6752 })
6753 .collect();
6754 geoms.push(Geometry::Point(Point::new(5.0, 5.0)));
6755 geoms
6756 }
6757
6758 #[test]
6759 fn coalesce_reclaims_sub_visibility_fragments_and_keeps_canonical() {
6760 let geoms = fragment_chain_geoms(6);
6764 let tin = tempfile::NamedTempFile::new().unwrap();
6765 write_input(tin.path(), &geoms, false, None);
6766
6767 let opts = ConvertOptions {
6768 levels: LevelPlan::ZoomRange {
6769 min_zoom: 4,
6770 max_zoom: 10,
6771 },
6772 no_auto_rank: true,
6773 ..Default::default() };
6775
6776 let tout_off = tempfile::NamedTempFile::new().unwrap();
6778 let off = convert_to_overviews(
6779 tin.path(),
6780 tout_off.path(),
6781 &ConvertOptions {
6782 coalesce_lines: false,
6783 ..opts.clone()
6784 },
6785 )
6786 .unwrap();
6787 assert_eq!(
6788 off.levels[0].feature_count, 1,
6789 "without coalescing only the point survives level 0: {:?}",
6790 off.levels
6791 );
6792
6793 let tout = tempfile::NamedTempFile::new().unwrap();
6795 let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
6796 assert_eq!(
6797 report.levels[0].feature_count, 2,
6798 "chain + point at level 0: {:?}",
6799 report.levels
6800 );
6801
6802 let vr = validate_file(tout.path()).unwrap();
6803 assert!(
6804 vr.is_valid(),
6805 "failures: {:?}",
6806 vr.failures().collect::<Vec<_>>()
6807 );
6808
6809 let reader = OverviewReader::open(tout.path()).unwrap();
6810 let counts0 = read_coalesced_counts(&reader, 0);
6811 let mut sorted = counts0.clone();
6812 sorted.sort_unstable();
6813 assert_eq!(sorted, vec![1, 6], "point=1, merged chain=6: {counts0:?}");
6814
6815 let canonical = reader.num_levels() - 1;
6817 let rows = read_level_rows(&reader, canonical);
6818 assert_eq!(rows.len(), geoms.len());
6819 for (id, _, _, geom) in &rows {
6820 assert_eq!(
6821 geom, &geoms[*id as usize],
6822 "canonical geometry verbatim (never coalesced)"
6823 );
6824 }
6825 assert!(read_coalesced_counts(&reader, canonical)
6826 .iter()
6827 .all(|&c| c == 1));
6828 }
6829
6830 #[test]
6831 fn coalesce_groups_by_auto_detected_class() {
6832 let mut geoms = vec![
6836 Geometry::LineString(LineString::from(vec![(0.0, 0.0), (0.1, 0.0)])),
6837 Geometry::LineString(LineString::from(vec![(0.1, 0.0), (0.2, 0.0)])),
6838 Geometry::LineString(LineString::from(vec![(0.2, 0.0), (0.2, 0.1)])),
6839 ];
6840 let mut classes = vec![Some("motorway"), Some("motorway"), Some("footway")];
6841 for (i, c) in ["primary", "service", "residential"].iter().enumerate() {
6842 geoms.push(Geometry::LineString(LineString::from(vec![
6843 (3.0 + i as f64, 3.0),
6844 (3.1 + i as f64, 3.05),
6845 ])));
6846 classes.push(Some(*c));
6847 }
6848 let tin = tempfile::NamedTempFile::new().unwrap();
6849 let tout = tempfile::NamedTempFile::new().unwrap();
6850 write_class_input(tin.path(), &geoms, &classes);
6851
6852 let opts = ConvertOptions {
6853 levels: LevelPlan::ZoomRange {
6854 min_zoom: 4,
6855 max_zoom: 10,
6856 },
6857 coalesce_lines: true,
6858 ..Default::default()
6859 };
6860 convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
6861 assert_eq!(ranking_mode_of(tout.path()), "auto-overture-roads");
6862
6863 let reader = OverviewReader::open(tout.path()).unwrap();
6864 let counts0 = read_coalesced_counts(&reader, 0);
6865 assert_eq!(
6866 counts0.iter().filter(|&&c| c == 2).count(),
6867 1,
6868 "exactly one 2-segment motorway chain: {counts0:?}"
6869 );
6870 assert!(
6871 counts0.iter().all(|&c| c <= 2),
6872 "footway never merges into the motorway chain: {counts0:?}"
6873 );
6874 }
6875
6876 #[test]
6877 fn coalesce_inert_in_partitioning() {
6878 let geoms = fragment_chain_geoms(3);
6882 let tin = tempfile::NamedTempFile::new().unwrap();
6883 write_input(tin.path(), &geoms, false, None);
6884
6885 for streaming in [true, false] {
6886 let tout = tempfile::NamedTempFile::new().unwrap();
6887 let report = convert_to_overviews(
6888 tin.path(),
6889 tout.path(),
6890 &ConvertOptions {
6891 mode: Mode::Partitioning,
6892 coalesce_lines: true, streaming,
6894 ..Default::default()
6895 },
6896 )
6897 .unwrap();
6898 assert_eq!(report.total_rows, geoms.len(), "streaming={streaming}");
6900 let reader = OverviewReader::open(tout.path()).unwrap();
6901 assert!(
6902 reader
6903 .meta()
6904 .generalization
6905 .as_ref()
6906 .unwrap()
6907 .coalescing
6908 .is_none(),
6909 "no coalescing provenance in partitioning mode"
6910 );
6911 let batch_schema = reader.read_level(0, None).unwrap().next().unwrap().unwrap();
6912 assert!(
6913 batch_schema.schema().index_of("coalesced_count").is_err(),
6914 "no coalesced_count column in partitioning mode"
6915 );
6916 assert!(validate_file(tout.path()).unwrap().is_valid());
6917 }
6918 }
6919
6920 #[test]
6921 fn coalesce_renames_existing_coalesced_count_column() {
6922 let geoms = fragment_chain_geoms(2);
6926 let n = geoms.len();
6927 let tin = tempfile::NamedTempFile::new().unwrap();
6928 {
6929 let id = Int64Array::from((0..n as i64).collect::<Vec<_>>());
6930 let cc = arrow_array::Int32Array::from(vec![7i32; n]);
6931 let geom_arr = build_geometry_array(&geoms);
6932 let geom_field = geom_arr.data_type().to_field("geometry", true);
6933 let fields = vec![
6934 Arc::new(Field::new("id", DataType::Int64, false)),
6935 Arc::new(Field::new("Coalesced_Count", DataType::Int32, false)),
6936 Arc::new(geom_field),
6937 ];
6938 let columns: Vec<Arc<dyn Array>> =
6939 vec![Arc::new(id), Arc::new(cc), geom_arr.to_array_ref()];
6940 let schema = Arc::new(Schema::new(fields));
6941 let batch = RecordBatch::try_new(schema.clone(), columns).unwrap();
6942 let gpq_options = GeoParquetWriterOptionsBuilder::default()
6943 .set_encoding(GeoParquetWriterEncoding::WKB)
6944 .set_generate_covering(true)
6945 .build();
6946 let mut encoder = GeoParquetRecordBatchEncoder::try_new(&schema, &gpq_options).unwrap();
6947 let target_schema = encoder.target_schema();
6948 let file = std::fs::File::create(tin.path()).unwrap();
6949 let mut writer = ArrowWriter::try_new(file, target_schema, None).unwrap();
6950 writer
6951 .write(&encoder.encode_record_batch(&batch).unwrap())
6952 .unwrap();
6953 writer.append_key_value_metadata(encoder.into_keyvalue().unwrap());
6954 writer.close().unwrap();
6955 }
6956
6957 let tout = tempfile::NamedTempFile::new().unwrap();
6958 for streaming in [true, false] {
6959 convert_to_overviews(
6960 tin.path(),
6961 tout.path(),
6962 &ConvertOptions {
6963 coalesce_lines: true,
6964 streaming,
6965 ..Default::default()
6966 },
6967 )
6968 .unwrap_or_else(|e| {
6969 panic!("streaming={streaming}: `Coalesced_Count` must be auto-renamed, got {e}")
6970 });
6971 let names = output_column_names(tout.path());
6972 assert!(
6973 names.iter().any(|n| n == "Coalesced_Count_"),
6974 "streaming={streaming}: renamed source column present, names={names:?}"
6975 );
6976 }
6977 convert_to_overviews(
6980 tin.path(),
6981 tout.path(),
6982 &ConvertOptions {
6983 coalesce_lines: false,
6984 ..Default::default()
6985 },
6986 )
6987 .unwrap();
6988 let names = output_column_names(tout.path());
6989 assert!(
6990 names.iter().any(|n| n == "Coalesced_Count"),
6991 "passthrough column kept verbatim, names={names:?}"
6992 );
6993 }
6994
6995 #[test]
6996 fn coalesce_footer_provenance_recorded() {
6997 let geoms = fragment_chain_geoms(3);
6998 let tin = tempfile::NamedTempFile::new().unwrap();
6999 let tout = tempfile::NamedTempFile::new().unwrap();
7000 write_input(tin.path(), &geoms, false, None);
7001
7002 let opts = ConvertOptions {
7003 coalesce_lines: true,
7004 coalesce_snap: 1.5,
7005 coalesce_junction_angle: 30.0,
7006 coalesce_max_level_rows: 123_456,
7007 ..Default::default()
7008 };
7009 convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
7010 let reader = OverviewReader::open(tout.path()).unwrap();
7011 let c = reader
7012 .meta()
7013 .generalization
7014 .as_ref()
7015 .unwrap()
7016 .coalescing
7017 .clone()
7018 .expect("coalescing provenance recorded");
7019 assert!(c.enabled);
7020 assert_eq!(c.snap_tolerance_gsd_factor, 1.5);
7021 assert_eq!(c.junction_angle, Some(30.0));
7024 assert_eq!(c.max_level_rows, Some(123_456));
7025 assert_eq!(c.coalesced_count_column, "coalesced_count");
7026
7027 let tout_off = tempfile::NamedTempFile::new().unwrap();
7029 convert_to_overviews(
7030 tin.path(),
7031 tout_off.path(),
7032 &ConvertOptions {
7033 coalesce_lines: false,
7034 ..Default::default()
7035 },
7036 )
7037 .unwrap();
7038 let r_off = OverviewReader::open(tout_off.path()).unwrap();
7039 assert!(r_off
7040 .meta()
7041 .generalization
7042 .as_ref()
7043 .unwrap()
7044 .coalescing
7045 .is_none());
7046 }
7047
7048 #[test]
7049 fn coalesce_guard_skips_chaining_but_keeps_column() {
7050 let geoms = fragment_chain_geoms(6);
7054 let tin = tempfile::NamedTempFile::new().unwrap();
7055 let tout = tempfile::NamedTempFile::new().unwrap();
7056 write_input(tin.path(), &geoms, false, None);
7057
7058 let opts = ConvertOptions {
7059 levels: LevelPlan::ZoomRange {
7060 min_zoom: 4,
7061 max_zoom: 10,
7062 },
7063 no_auto_rank: true,
7064 coalesce_lines: true,
7065 coalesce_max_level_rows: 2, ..Default::default()
7067 };
7068 let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
7069 assert_eq!(
7070 report.levels[0].feature_count, 1,
7071 "guard-skipped run behaves like non-coalesced: {:?}",
7072 report.levels
7073 );
7074 let reader = OverviewReader::open(tout.path()).unwrap();
7075 for level in 0..reader.num_levels() {
7076 assert!(read_coalesced_counts(&reader, level)
7077 .iter()
7078 .all(|&c| c == 1));
7079 }
7080 assert!(validate_file(tout.path()).unwrap().is_valid());
7081 }
7082
7083 #[test]
7084 fn streaming_matches_in_memory_coalescing() {
7085 let geoms = fragment_chain_geoms(6);
7089 let tin = tempfile::NamedTempFile::new().unwrap();
7090 write_input(tin.path(), &geoms, false, None);
7091
7092 let base = ConvertOptions {
7093 levels: LevelPlan::ZoomRange {
7094 min_zoom: 4,
7095 max_zoom: 10,
7096 },
7097 no_auto_rank: true,
7098 coalesce_lines: true,
7099 read_batch_size: 2,
7100 ..Default::default()
7101 };
7102 assert_streaming_equivalent(tin.path(), &base);
7103
7104 let mem_out = tempfile::NamedTempFile::new().unwrap();
7106 let stream_out = tempfile::NamedTempFile::new().unwrap();
7107 convert_to_overviews(
7108 tin.path(),
7109 mem_out.path(),
7110 &ConvertOptions {
7111 streaming: false,
7112 ..base.clone()
7113 },
7114 )
7115 .unwrap();
7116 convert_to_overviews(tin.path(), stream_out.path(), &base).unwrap();
7117 let mr = OverviewReader::open(mem_out.path()).unwrap();
7118 let sr = OverviewReader::open(stream_out.path()).unwrap();
7119 for level in 0..mr.num_levels() {
7120 assert_eq!(
7121 read_coalesced_counts(&mr, level),
7122 read_coalesced_counts(&sr, level),
7123 "level {level} coalesced_count differs"
7124 );
7125 }
7126 }
7127
7128 #[test]
7129 fn streaming_matches_in_memory_coalescing_with_class_groups() {
7130 let mut geoms = vec![
7133 Geometry::LineString(LineString::from(vec![(0.0, 0.0), (0.1, 0.0)])),
7134 Geometry::LineString(LineString::from(vec![(0.1, 0.0), (0.2, 0.0)])),
7135 Geometry::LineString(LineString::from(vec![(0.2, 0.0), (0.2, 0.1)])),
7136 ];
7137 let mut classes = vec![Some("motorway"), Some("motorway"), Some("footway")];
7138 for (i, c) in ["primary", "service", "residential", "trunk"]
7139 .iter()
7140 .enumerate()
7141 {
7142 geoms.push(Geometry::LineString(LineString::from(vec![
7143 (3.0 + i as f64, 3.0),
7144 (3.1 + i as f64, 3.05),
7145 ])));
7146 classes.push(Some(*c));
7147 }
7148 let tin = tempfile::NamedTempFile::new().unwrap();
7149 write_class_input(tin.path(), &geoms, &classes);
7150
7151 let base = ConvertOptions {
7152 levels: LevelPlan::ZoomRange {
7153 min_zoom: 4,
7154 max_zoom: 10,
7155 },
7156 coalesce_lines: true,
7157 read_batch_size: 2,
7158 ..Default::default()
7159 };
7160 let mem_out = tempfile::NamedTempFile::new().unwrap();
7161 let stream_out = tempfile::NamedTempFile::new().unwrap();
7162 convert_to_overviews(
7163 tin.path(),
7164 mem_out.path(),
7165 &ConvertOptions {
7166 streaming: false,
7167 ..base.clone()
7168 },
7169 )
7170 .unwrap();
7171 convert_to_overviews(tin.path(), stream_out.path(), &base).unwrap();
7172 assert_eq!(
7173 overviews_footer_json(mem_out.path()),
7174 overviews_footer_json(stream_out.path())
7175 );
7176 let mr = OverviewReader::open(mem_out.path()).unwrap();
7177 let sr = OverviewReader::open(stream_out.path()).unwrap();
7178 assert_eq!(mr.num_levels(), sr.num_levels());
7179 for level in 0..mr.num_levels() {
7180 assert_eq!(
7181 read_coalesced_counts(&mr, level),
7182 read_coalesced_counts(&sr, level),
7183 "level {level} coalesced_count differs"
7184 );
7185 }
7186 }
7187
7188 #[test]
7189 fn sort_key_column_missing_errors() {
7190 let geoms = synthetic_geometries();
7191 let tin = tempfile::NamedTempFile::new().unwrap();
7192 let tout = tempfile::NamedTempFile::new().unwrap();
7193 write_input(tin.path(), &geoms, false, None);
7194
7195 let opts = ConvertOptions {
7196 sort_key: Some("nonexistent".to_string()),
7197 ..Default::default()
7198 };
7199 let err = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap_err();
7200 assert!(matches!(err, ConvertError::SortKeyColumnMissing { .. }));
7201 }
7202
7203 fn write_multi_rg_input(path: &Path, coords: &[(f64, f64)], with_covering: bool) {
7211 use parquet::file::properties::WriterProperties;
7212
7213 let geoms: Vec<Geometry<f64>> = coords
7214 .iter()
7215 .map(|&(x, y)| Geometry::Point(Point::new(x, y)))
7216 .collect();
7217 let n = geoms.len();
7218 let id = Int64Array::from((0..n as i64).collect::<Vec<_>>());
7219 let geom_arr = build_geometry_array(&geoms);
7220 let geom_field = geom_arr.data_type().to_field("geometry", true);
7221 let fields = vec![
7222 Arc::new(Field::new("id", DataType::Int64, false)),
7223 Arc::new(geom_field),
7224 ];
7225 let columns: Vec<Arc<dyn Array>> = vec![Arc::new(id), geom_arr.to_array_ref()];
7226 let schema = Arc::new(Schema::new(fields));
7227 let batch = RecordBatch::try_new(schema.clone(), columns).unwrap();
7228
7229 let gpq_options = GeoParquetWriterOptionsBuilder::default()
7230 .set_encoding(GeoParquetWriterEncoding::WKB)
7231 .set_generate_covering(with_covering)
7232 .build();
7233 let encoder = GeoParquetRecordBatchEncoder::try_new(&schema, &gpq_options).unwrap();
7234 let target_schema = encoder.target_schema();
7235 let props = WriterProperties::builder()
7237 .set_max_row_group_row_count(Some(1))
7238 .build();
7239 let file = std::fs::File::create(path).unwrap();
7240 let mut writer = ArrowWriter::try_new(file, target_schema, Some(props)).unwrap();
7241 let mut encoder = encoder;
7242 let encoded = encoder.encode_record_batch(&batch).unwrap();
7243 writer.write(&encoded).unwrap();
7244 writer.append_key_value_metadata(encoder.into_keyvalue().unwrap());
7245 writer.close().unwrap();
7246 }
7247
7248 fn read_all_ids(reader: &OverviewReader) -> Vec<i64> {
7250 use arrow_array::cast::AsArray;
7251 let mut ids = Vec::new();
7252 for level in 0..reader.num_levels() {
7253 let rdr = reader.read_level(level, None).unwrap();
7254 for batch in rdr {
7255 let batch = batch.unwrap();
7256 let col = batch
7257 .column(batch.schema().index_of("id").unwrap())
7258 .as_primitive::<arrow_array::types::Int64Type>();
7259 ids.extend(col.iter().flatten());
7260 }
7261 }
7262 ids.sort_unstable();
7263 ids.dedup();
7264 ids
7265 }
7266
7267 #[test]
7268 fn bbox_filter_matches_posthoc_filter() {
7269 let coords = vec![(0.0, 0.0), (10.0, 10.0), (20.0, 20.0), (30.0, 30.0)];
7272 let tin = tempfile::NamedTempFile::new().unwrap();
7273 write_multi_rg_input(tin.path(), &coords, true);
7274
7275 let tout_full = tempfile::NamedTempFile::new().unwrap();
7277 let opts_full = ConvertOptions {
7278 mode: Mode::Duplicating,
7279 levels: LevelPlan::ZoomRange {
7280 min_zoom: 6,
7281 max_zoom: 6,
7282 },
7283 ..Default::default()
7284 };
7285 let report_full = convert_to_overviews(tin.path(), tout_full.path(), &opts_full).unwrap();
7286 assert_eq!(report_full.row_groups_total, 4);
7287 assert_eq!(report_full.row_groups_read, 4);
7288 let reader_full = OverviewReader::open(tout_full.path()).unwrap();
7289 let ids_full = read_all_ids(&reader_full);
7290
7291 let tout_bbox = tempfile::NamedTempFile::new().unwrap();
7293 let opts_bbox = ConvertOptions {
7294 bbox: Some([9.0, 9.0, 11.0, 11.0]),
7295 ..opts_full.clone()
7296 };
7297 let report_bbox = convert_to_overviews(tin.path(), tout_bbox.path(), &opts_bbox).unwrap();
7298 assert_eq!(report_bbox.row_groups_total, 4);
7300 assert_eq!(
7301 report_bbox.row_groups_read, 1,
7302 "bbox pruning did not fire: read {} row groups",
7303 report_bbox.row_groups_read
7304 );
7305 let reader_bbox = OverviewReader::open(tout_bbox.path()).unwrap();
7306 let ids_bbox = read_all_ids(&reader_bbox);
7307 assert_eq!(ids_bbox, vec![1], "bbox filter kept wrong ids");
7308
7309 let ids_posthoc: Vec<i64> = ids_full
7311 .into_iter()
7312 .filter(|&id| {
7313 let (x, y) = coords[id as usize];
7314 (9.0..=11.0).contains(&x) && (9.0..=11.0).contains(&y)
7315 })
7316 .collect();
7317 assert_eq!(ids_bbox, ids_posthoc);
7318 }
7319
7320 #[test]
7321 fn bbox_filter_stats_free_degradation() {
7322 let coords = vec![(0.0, 0.0), (10.0, 10.0), (20.0, 20.0), (30.0, 30.0)];
7324 let tin = tempfile::NamedTempFile::new().unwrap();
7325 write_multi_rg_input(tin.path(), &coords, false);
7326
7327 let tout = tempfile::NamedTempFile::new().unwrap();
7328 let opts = ConvertOptions {
7329 mode: Mode::Duplicating,
7330 levels: LevelPlan::ZoomRange {
7331 min_zoom: 6,
7332 max_zoom: 6,
7333 },
7334 bbox: Some([9.0, 9.0, 11.0, 11.0]),
7335 ..Default::default()
7336 };
7337 let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
7338 assert_eq!(report.row_groups_total, 4);
7340 assert_eq!(
7341 report.row_groups_read, 4,
7342 "stats-free should read all row groups"
7343 );
7344 let reader = OverviewReader::open(tout.path()).unwrap();
7346 let ids = read_all_ids(&reader);
7347 assert_eq!(ids, vec![1], "exact filter did not apply");
7348 }
7349
7350 #[test]
7351 fn bbox_filter_nothing_intersects() {
7352 let coords = vec![(0.0, 0.0), (10.0, 10.0)];
7353 let tin = tempfile::NamedTempFile::new().unwrap();
7354 write_multi_rg_input(tin.path(), &coords, true);
7355
7356 let tout = tempfile::NamedTempFile::new().unwrap();
7357 let opts = ConvertOptions {
7358 mode: Mode::Duplicating,
7359 levels: LevelPlan::ZoomRange {
7360 min_zoom: 6,
7361 max_zoom: 6,
7362 },
7363 bbox: Some([100.0, 100.0, 110.0, 110.0]), ..Default::default()
7365 };
7366 let err = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap_err();
7367 assert!(
7369 matches!(err, ConvertError::NoData),
7370 "expected NoData, got {err:?}"
7371 );
7372 }
7373
7374 #[test]
7375 fn bbox_filter_everything_intersects() {
7376 let coords = vec![(0.0, 0.0), (10.0, 10.0)];
7377 let tin = tempfile::NamedTempFile::new().unwrap();
7378 write_multi_rg_input(tin.path(), &coords, true);
7379
7380 let tout_full = tempfile::NamedTempFile::new().unwrap();
7382 let opts_full = ConvertOptions {
7383 mode: Mode::Duplicating,
7384 levels: LevelPlan::ZoomRange {
7385 min_zoom: 6,
7386 max_zoom: 6,
7387 },
7388 ..Default::default()
7389 };
7390 let _report_full = convert_to_overviews(tin.path(), tout_full.path(), &opts_full).unwrap();
7391 let reader_full = OverviewReader::open(tout_full.path()).unwrap();
7392 let ids_full = read_all_ids(&reader_full);
7393
7394 let tout_bbox = tempfile::NamedTempFile::new().unwrap();
7396 let opts_bbox = ConvertOptions {
7397 bbox: Some([-1.0, -1.0, 11.0, 11.0]),
7398 ..opts_full.clone()
7399 };
7400 let report_bbox = convert_to_overviews(tin.path(), tout_bbox.path(), &opts_bbox).unwrap();
7401 assert_eq!(report_bbox.row_groups_read, report_bbox.row_groups_total);
7403 let reader_bbox = OverviewReader::open(tout_bbox.path()).unwrap();
7404 let ids_bbox = read_all_ids(&reader_bbox);
7405 assert_eq!(ids_bbox, ids_full);
7406 }
7407
7408 type AttrRow = ((f64, f64), Option<f64>, Option<&'static str>);
7414
7415 fn write_multi_rg_attr_input(path: &Path, rows: &[AttrRow]) {
7420 use parquet::file::properties::WriterProperties;
7421
7422 let geoms: Vec<Geometry<f64>> = rows
7423 .iter()
7424 .map(|&((x, y), _, _)| Geometry::Point(Point::new(x, y)))
7425 .collect();
7426 let n = geoms.len();
7427 let id = Int64Array::from((0..n as i64).collect::<Vec<_>>());
7428 let confidence =
7429 arrow_array::Float64Array::from(rows.iter().map(|(_, c, _)| *c).collect::<Vec<_>>());
7430 let crop =
7431 arrow_array::StringArray::from(rows.iter().map(|(_, _, s)| *s).collect::<Vec<_>>());
7432 let geom_arr = build_geometry_array(&geoms);
7433 let geom_field = geom_arr.data_type().to_field("geometry", true);
7434 let fields = vec![
7435 Arc::new(Field::new("id", DataType::Int64, false)),
7436 Arc::new(Field::new("confidence", DataType::Float64, true)),
7437 Arc::new(Field::new("crop", DataType::Utf8, true)),
7438 Arc::new(geom_field),
7439 ];
7440 let columns: Vec<Arc<dyn Array>> = vec![
7441 Arc::new(id),
7442 Arc::new(confidence),
7443 Arc::new(crop),
7444 geom_arr.to_array_ref(),
7445 ];
7446 let schema = Arc::new(Schema::new(fields));
7447 let batch = RecordBatch::try_new(schema.clone(), columns).unwrap();
7448
7449 let gpq_options = GeoParquetWriterOptionsBuilder::default()
7450 .set_encoding(GeoParquetWriterEncoding::WKB)
7451 .set_generate_covering(true)
7452 .build();
7453 let encoder = GeoParquetRecordBatchEncoder::try_new(&schema, &gpq_options).unwrap();
7454 let target_schema = encoder.target_schema();
7455 let props = WriterProperties::builder()
7456 .set_max_row_group_row_count(Some(1))
7457 .build();
7458 let file = std::fs::File::create(path).unwrap();
7459 let mut writer = ArrowWriter::try_new(file, target_schema, Some(props)).unwrap();
7460 let mut encoder = encoder;
7461 let encoded = encoder.encode_record_batch(&batch).unwrap();
7462 writer.write(&encoded).unwrap();
7463 writer.append_key_value_metadata(encoder.into_keyvalue().unwrap());
7464 writer.close().unwrap();
7465 }
7466
7467 fn attr_rows() -> Vec<AttrRow> {
7469 vec![
7470 ((0.0, 0.0), Some(0.1), Some("soy")),
7471 ((10.0, 10.0), Some(0.9), Some("corn")),
7472 ((20.0, 20.0), Some(0.85), Some("soy")),
7473 ((30.0, 30.0), None, Some("rice")),
7474 ]
7475 }
7476
7477 fn attr_opts() -> ConvertOptions {
7478 ConvertOptions {
7479 mode: Mode::Duplicating,
7480 levels: LevelPlan::ZoomRange {
7481 min_zoom: 6,
7482 max_zoom: 6,
7483 },
7484 ..Default::default()
7485 }
7486 }
7487
7488 #[test]
7493 fn attribute_filter_matches_posthoc_and_prunes_row_groups() {
7494 let tin = tempfile::NamedTempFile::new().unwrap();
7495 write_multi_rg_attr_input(tin.path(), &attr_rows());
7496
7497 for streaming in [true, false] {
7498 let tout = tempfile::NamedTempFile::new().unwrap();
7499 let opts = ConvertOptions {
7500 filter: Some("confidence > 0.8".to_string()),
7501 streaming,
7502 ..attr_opts()
7503 };
7504 let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
7505 assert_eq!(report.row_groups_total, 4, "streaming={streaming}");
7506 assert_eq!(
7508 report.row_groups_read, 2,
7509 "stats pushdown did not fire (streaming={streaming})"
7510 );
7511 let ids = read_all_ids(&OverviewReader::open(tout.path()).unwrap());
7512 assert_eq!(ids, vec![1, 2], "streaming={streaming}");
7513 }
7514 }
7515
7516 #[test]
7519 fn attribute_filter_composes_with_bbox() {
7520 let tin = tempfile::NamedTempFile::new().unwrap();
7521 write_multi_rg_attr_input(tin.path(), &attr_rows());
7522
7523 for streaming in [true, false] {
7524 let tout = tempfile::NamedTempFile::new().unwrap();
7525 let opts = ConvertOptions {
7526 bbox: Some([9.0, 9.0, 11.0, 11.0]),
7529 filter: Some("confidence > 0.8".to_string()),
7530 streaming,
7531 ..attr_opts()
7532 };
7533 let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
7534 assert_eq!(
7535 report.row_groups_read, 1,
7536 "bbox+filter selection must intersect (streaming={streaming})"
7537 );
7538 let ids = read_all_ids(&OverviewReader::open(tout.path()).unwrap());
7539 assert_eq!(ids, vec![1], "streaming={streaming}");
7540 }
7541 }
7542
7543 #[test]
7545 fn attribute_filter_string_in_and_null_semantics() {
7546 let tin = tempfile::NamedTempFile::new().unwrap();
7547 write_multi_rg_attr_input(tin.path(), &attr_rows());
7548
7549 for streaming in [true, false] {
7550 let tout = tempfile::NamedTempFile::new().unwrap();
7552 let opts = ConvertOptions {
7553 filter: Some("crop IN ('corn', 'rice')".to_string()),
7554 streaming,
7555 ..attr_opts()
7556 };
7557 convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
7558 let ids = read_all_ids(&OverviewReader::open(tout.path()).unwrap());
7559 assert_eq!(ids, vec![1, 3], "IN (streaming={streaming})");
7560
7561 let tout = tempfile::NamedTempFile::new().unwrap();
7563 let opts = ConvertOptions {
7564 filter: Some("confidence IS NULL".to_string()),
7565 streaming,
7566 ..attr_opts()
7567 };
7568 convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
7569 let ids = read_all_ids(&OverviewReader::open(tout.path()).unwrap());
7570 assert_eq!(ids, vec![3], "IS NULL (streaming={streaming})");
7571
7572 let tout = tempfile::NamedTempFile::new().unwrap();
7574 let opts = ConvertOptions {
7575 filter: Some("confidence > 0.8 OR crop = 'rice'".to_string()),
7576 streaming,
7577 ..attr_opts()
7578 };
7579 convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
7580 let ids = read_all_ids(&OverviewReader::open(tout.path()).unwrap());
7581 assert_eq!(ids, vec![1, 2, 3], "OR (streaming={streaming})");
7582 }
7583 }
7584
7585 #[test]
7588 fn attribute_filter_error_paths() {
7589 let tin = tempfile::NamedTempFile::new().unwrap();
7590 write_multi_rg_attr_input(tin.path(), &attr_rows());
7591
7592 let tout = tempfile::NamedTempFile::new().unwrap();
7593 let bad_syntax = ConvertOptions {
7594 filter: Some("confidence >".to_string()),
7595 ..attr_opts()
7596 };
7597 let err = convert_to_overviews(tin.path(), tout.path(), &bad_syntax).unwrap_err();
7598 assert!(matches!(err, ConvertError::Filter(_)), "got {err:?}");
7599
7600 let unknown = ConvertOptions {
7601 filter: Some("nope = 1".to_string()),
7602 ..attr_opts()
7603 };
7604 let err = convert_to_overviews(tin.path(), tout.path(), &unknown).unwrap_err();
7605 assert!(
7606 matches!(err, ConvertError::Filter(_)) && err.to_string().contains("unknown column"),
7607 "got {err:?}"
7608 );
7609
7610 let mismatch = ConvertOptions {
7611 filter: Some("crop > 3".to_string()),
7612 ..attr_opts()
7613 };
7614 let err = convert_to_overviews(tin.path(), tout.path(), &mismatch).unwrap_err();
7615 assert!(matches!(err, ConvertError::Filter(_)), "got {err:?}");
7616
7617 let none = ConvertOptions {
7619 filter: Some("confidence > 99".to_string()),
7620 ..attr_opts()
7621 };
7622 let err = convert_to_overviews(tin.path(), tout.path(), &none).unwrap_err();
7623 assert!(matches!(err, ConvertError::NoData), "got {err:?}");
7624 }
7625
7626 #[cfg(feature = "remote")]
7631 mod remote_input {
7632 use super::*;
7633 use crate::input::{test_memory_source, InputSource};
7634
7635 fn row_group_spans(bytes: &[u8]) -> Vec<std::ops::Range<u64>> {
7637 let builder =
7638 ParquetRecordBatchReaderBuilder::try_new(bytes::Bytes::from(bytes.to_vec()))
7639 .unwrap();
7640 builder
7641 .metadata()
7642 .row_groups()
7643 .iter()
7644 .map(|rg| {
7645 let mut start = u64::MAX;
7646 let mut end = 0u64;
7647 for col in rg.columns() {
7648 let (s, len) = col.byte_range();
7649 start = start.min(s);
7650 end = end.max(s + len);
7651 }
7652 start..end
7653 })
7654 .collect()
7655 }
7656
7657 fn assert_bbox_extract_fetches_only_selected(streaming: bool) {
7661 let coords = vec![(0.0, 0.0), (10.0, 10.0), (20.0, 20.0), (30.0, 30.0)];
7664 let tin = tempfile::NamedTempFile::new().unwrap();
7665 write_multi_rg_input(tin.path(), &coords, true);
7666 let bytes = std::fs::read(tin.path()).unwrap();
7667 let spans = row_group_spans(&bytes);
7668 assert_eq!(spans.len(), 4);
7669
7670 let source = test_memory_source(bytes, "multi.parquet");
7671 let tout = tempfile::NamedTempFile::new().unwrap();
7672 let opts = ConvertOptions {
7673 mode: Mode::Duplicating,
7674 levels: LevelPlan::ZoomRange {
7675 min_zoom: 6,
7676 max_zoom: 6,
7677 },
7678 bbox: Some([9.0, 9.0, 11.0, 11.0]),
7679 streaming,
7680 ..Default::default()
7681 };
7682 let report = convert_to_overviews_source(&source, tout.path(), &opts).unwrap();
7683
7684 assert_eq!(report.row_groups_total, 4);
7685 assert_eq!(report.row_groups_read, 1, "bbox pruning must fire");
7686 let ids = read_all_ids(&OverviewReader::open(tout.path()).unwrap());
7687 assert_eq!(ids, vec![1], "only the (10,10) feature survives");
7688
7689 let fetched = source.fetched_ranges().unwrap();
7691 assert!(!fetched.is_empty());
7692 for (i, span) in spans.iter().enumerate() {
7693 if i == 1 {
7694 continue;
7695 }
7696 for r in &fetched {
7697 assert!(
7698 r.end <= span.start || r.start >= span.end,
7699 "fetched range {r:?} overlaps PRUNED row group {i} ({span:?})"
7700 );
7701 }
7702 }
7703 assert!(
7705 fetched
7706 .iter()
7707 .any(|r| r.start >= spans[1].start && r.end <= spans[1].end),
7708 "selected row group 1 ({:?}) never fetched: {fetched:?}",
7709 spans[1]
7710 );
7711
7712 let stats = report.remote_fetch.expect("remote stats in report");
7714 assert!(stats.requests as usize >= fetched.len());
7715 assert!(
7716 stats.bytes_fetched < stats.object_size,
7717 "bbox extract must move fewer bytes than the object: {stats:?}"
7718 );
7719 }
7720
7721 #[test]
7722 fn bbox_extract_streaming_fetches_only_selected_row_groups() {
7723 assert_bbox_extract_fetches_only_selected(true);
7724 }
7725
7726 #[test]
7727 fn bbox_extract_in_memory_fetches_only_selected_row_groups() {
7728 assert_bbox_extract_fetches_only_selected(false);
7729 }
7730
7731 #[test]
7736 fn attribute_filter_remote_fetches_only_matching_row_groups() {
7737 let tin = tempfile::NamedTempFile::new().unwrap();
7740 write_multi_rg_attr_input(tin.path(), &attr_rows());
7741 let bytes = std::fs::read(tin.path()).unwrap();
7742 let spans = row_group_spans(&bytes);
7743 assert_eq!(spans.len(), 4);
7744
7745 let source = test_memory_source(bytes, "attrs.parquet");
7746 let tout = tempfile::NamedTempFile::new().unwrap();
7747 let opts = ConvertOptions {
7748 filter: Some("confidence > 0.8".to_string()),
7749 ..attr_opts()
7750 };
7751 let report = convert_to_overviews_source(&source, tout.path(), &opts).unwrap();
7752
7753 assert_eq!(report.row_groups_total, 4);
7754 assert_eq!(report.row_groups_read, 2, "filter pushdown must fire");
7755 let ids = read_all_ids(&OverviewReader::open(tout.path()).unwrap());
7756 assert_eq!(ids, vec![1, 2]);
7757
7758 let fetched = source.fetched_ranges().unwrap();
7760 assert!(!fetched.is_empty());
7761 for (i, span) in spans.iter().enumerate() {
7762 if i == 1 || i == 2 {
7763 continue;
7764 }
7765 for r in &fetched {
7766 assert!(
7767 r.end <= span.start || r.start >= span.end,
7768 "fetched range {r:?} overlaps PRUNED row group {i} ({span:?})"
7769 );
7770 }
7771 }
7772 let stats = report.remote_fetch.expect("remote stats in report");
7773 assert!(
7774 stats.bytes_fetched < stats.object_size,
7775 "filter extract must move fewer bytes than the object: {stats:?}"
7776 );
7777 }
7778
7779 #[test]
7782 fn remote_convert_matches_local_convert() {
7783 let geoms = synthetic_geometries();
7784 let tin = tempfile::NamedTempFile::new().unwrap();
7785 write_input(tin.path(), &geoms, false, None);
7786 let opts = ConvertOptions::default();
7787
7788 let tout_local = tempfile::NamedTempFile::new().unwrap();
7789 let report_local = convert_to_overviews(tin.path(), tout_local.path(), &opts).unwrap();
7790 assert!(report_local.remote_fetch.is_none(), "local input: no stats");
7791
7792 let source = test_memory_source(std::fs::read(tin.path()).unwrap(), "in.parquet");
7793 let tout_remote = tempfile::NamedTempFile::new().unwrap();
7794 let report_remote =
7795 convert_to_overviews_source(&source, tout_remote.path(), &opts).unwrap();
7796
7797 assert_eq!(report_remote.input_features, report_local.input_features);
7798 assert_eq!(report_remote.total_rows, report_local.total_rows);
7799 assert_eq!(
7800 read_all_ids(&OverviewReader::open(tout_remote.path()).unwrap()),
7801 read_all_ids(&OverviewReader::open(tout_local.path()).unwrap()),
7802 );
7803 assert!(report_remote.remote_fetch.is_some());
7804 }
7805
7806 #[test]
7815 fn remote_convert_coalesces_fetches_to_one_request_per_row_group() {
7816 let geoms = synthetic_geometries();
7817 let n = geoms.len();
7818 let tin = tempfile::NamedTempFile::new().unwrap();
7822 write_input_partition(tin.path(), &geoms, 0..n, Some(2));
7823 let bytes = std::fs::read(tin.path()).unwrap();
7824 let rg = row_group_spans(&bytes).len();
7825 assert!(rg >= 3, "test needs several row groups, got {rg}");
7826
7827 let opts = ConvertOptions::default();
7828
7829 let tout_local = tempfile::NamedTempFile::new().unwrap();
7831 convert_to_overviews(tin.path(), tout_local.path(), &opts).unwrap();
7832 let local_ids = read_all_ids(&OverviewReader::open(tout_local.path()).unwrap());
7833
7834 let source = test_memory_source(bytes, "staged.parquet");
7835 let tout = tempfile::NamedTempFile::new().unwrap();
7836 let report = convert_to_overviews_source(&source, tout.path(), &opts).unwrap();
7837
7838 assert_eq!(
7840 read_all_ids(&OverviewReader::open(tout.path()).unwrap()),
7841 local_ids,
7842 "staged remote convert must match the local convert row-for-row"
7843 );
7844
7845 let stats = report.remote_fetch.expect("remote stats in report");
7846 const FOOTER_SLACK: u64 = 4;
7850 assert!(
7851 stats.requests <= rg as u64 + FOOTER_SLACK,
7852 "expected ~1 request per row group (<= {} for {rg} row groups); \
7853 got {} — fetches not coalesced (#287) or properties re-fetched (#286)",
7854 rg as u64 + FOOTER_SLACK,
7855 stats.requests,
7856 );
7857 assert!(
7859 stats.bytes_fetched <= stats.object_size + stats.object_size / 2,
7860 "staging must stay ≈1× the object: {} of {} bytes",
7861 stats.bytes_fetched,
7862 stats.object_size,
7863 );
7864 }
7865
7866 fn partition_bytes(
7870 geoms: &[Geometry<f64>],
7871 range: std::ops::Range<usize>,
7872 row_group_rows: Option<usize>,
7873 ) -> Vec<u8> {
7874 let tmp = tempfile::NamedTempFile::new().unwrap();
7875 write_input_partition(tmp.path(), geoms, range, row_group_rows);
7876 std::fs::read(tmp.path()).unwrap()
7877 }
7878
7879 fn convert_sources_and_export(
7882 source: &crate::input_set::ConvertSource,
7883 workdir: &Path,
7884 tag: &str,
7885 opts: &ConvertOptions,
7886 ) -> Vec<u8> {
7887 use crate::overview::export::{export_pmtiles, ExportOptions};
7888 let overview = workdir.join(format!("{tag}-overview.parquet"));
7889 let pmtiles = workdir.join(format!("{tag}.pmtiles"));
7890 convert_to_overviews_sources(source, &overview, opts).unwrap();
7891 export_pmtiles(&overview, &pmtiles, &ExportOptions::default()).unwrap();
7892 std::fs::read(&pmtiles).unwrap()
7893 }
7894
7895 #[test]
7900 fn multi_part_three_pass_moves_each_part_once() {
7901 let geoms = synthetic_geometries();
7902 let n = geoms.len();
7903 let (source, parts) = crate::input::test_memory_multi_source(vec![
7904 ("p0.parquet", partition_bytes(&geoms, 0..5, None)),
7905 ("p1.parquet", partition_bytes(&geoms, 5..9, None)),
7906 ("p2.parquet", partition_bytes(&geoms, 9..n, None)),
7907 ]);
7908 assert_eq!(parts.len(), 3);
7909
7910 let tout = tempfile::NamedTempFile::new().unwrap();
7911 let report =
7912 convert_to_overviews_sources(&source, tout.path(), &multi_test_options()).unwrap();
7913 assert_eq!(report.input_features, n);
7914
7915 let summed = report.remote_fetch.expect("multi remote reports stats");
7916 let mut object_total = 0;
7917 for part in &parts {
7918 let stats = part.fetch_stats().expect("remote part has stats");
7919 object_total += stats.object_size;
7920 assert!(
7921 stats.bytes_fetched <= stats.object_size + stats.object_size / 2,
7922 "part {} moved {} bytes for a {}-byte object (>1.5x, #219 \
7923 must hold per part)",
7924 part.display_name(),
7925 stats.bytes_fetched,
7926 stats.object_size,
7927 );
7928 let mut seen = std::collections::HashSet::new();
7930 for r in part.fetched_ranges().expect("remote part logs ranges") {
7931 assert!(
7932 seen.insert((r.start, r.end)),
7933 "part {}: range {r:?} fetched more than once (#219)",
7934 part.display_name(),
7935 );
7936 }
7937 }
7938 assert_eq!(
7939 summed.object_size, object_total,
7940 "ConvertReport.remote_fetch.object_size sums the parts"
7941 );
7942 }
7943
7944 #[test]
7949 fn multi_part_remote_coalesces_per_part_row_groups() {
7950 let geoms = synthetic_geometries();
7951 let n = geoms.len();
7952 let p0 = partition_bytes(&geoms, 0..5, Some(2));
7956 let p1 = partition_bytes(&geoms, 5..9, Some(2));
7957 let p2 = partition_bytes(&geoms, 9..n, Some(2));
7958 let rg = [&p0, &p1, &p2].map(|b| row_group_spans(b).len());
7959 assert!(
7960 rg.iter().all(|&r| r >= 2),
7961 "each part needs several row groups: {rg:?}"
7962 );
7963
7964 let (source, parts) = crate::input::test_memory_multi_source(vec![
7965 ("p0.parquet", p0),
7966 ("p1.parquet", p1),
7967 ("p2.parquet", p2),
7968 ]);
7969 let tout = tempfile::NamedTempFile::new().unwrap();
7970 convert_to_overviews_sources(&source, tout.path(), &multi_test_options()).unwrap();
7971
7972 const FOOTER_SLACK: u64 = 4;
7973 for (i, part) in parts.iter().enumerate() {
7974 let stats = part.fetch_stats().expect("remote part has stats");
7975 assert!(
7976 stats.requests <= rg[i] as u64 + FOOTER_SLACK,
7977 "part {i}: expected ~1 request per row group (<= {} for {} \
7978 row groups); got {} — not coalesced (#287) or properties \
7979 re-fetched (#286)",
7980 rg[i] as u64 + FOOTER_SLACK,
7981 rg[i],
7982 stats.requests,
7983 );
7984 }
7985 }
7986
7987 #[test]
7991 fn multi_part_bbox_prunes_part_to_footer_only() {
7992 let geoms = synthetic_geometries();
7993 let n = geoms.len();
7994 let p1_bytes = partition_bytes(&geoms, 6..10, None);
7996 let p1_spans = row_group_spans(&p1_bytes);
7997 let (source, parts) = crate::input::test_memory_multi_source(vec![
7998 ("p0.parquet", partition_bytes(&geoms, 0..6, None)),
7999 ("p1.parquet", p1_bytes),
8000 ("p2.parquet", partition_bytes(&geoms, 10..n, None)),
8001 ]);
8002
8003 let opts = ConvertOptions {
8004 bbox: Some([-100.0, -70.0, 30.0, 20.0]),
8005 ..multi_test_options()
8006 };
8007 let tout = tempfile::NamedTempFile::new().unwrap();
8008 let report = convert_to_overviews_sources(&source, tout.path(), &opts).unwrap();
8009 assert!(
8010 report.row_groups_read < report.row_groups_total,
8011 "bbox must prune the lines part: {}/{}",
8012 report.row_groups_read,
8013 report.row_groups_total
8014 );
8015
8016 let pruned = &parts[1];
8017 let fetched = pruned.fetched_ranges().expect("remote part logs ranges");
8018 assert!(
8019 !fetched.is_empty(),
8020 "the footer itself is fetched at set construction"
8021 );
8022 for r in &fetched {
8023 for span in &p1_spans {
8024 assert!(
8025 r.end <= span.start || r.start >= span.end,
8026 "pruned part fetched data-page range {r:?} \
8027 (row-group span {span:?}) — must be footer-only"
8028 );
8029 }
8030 }
8031 }
8032
8033 #[test]
8037 fn multi_part_remote_output_matches_single_remote() {
8038 let geoms = synthetic_geometries();
8039 let n = geoms.len();
8040 let dir = tempfile::tempdir().unwrap();
8041 let opts = multi_test_options();
8042
8043 let (single, _) = crate::input::test_memory_multi_source(vec![(
8044 "single.parquet",
8045 partition_bytes(&geoms, 0..n, None),
8046 )]);
8047 let (multi, parts) = crate::input::test_memory_multi_source(vec![
8048 ("part-000.parquet", partition_bytes(&geoms, 0..5, None)),
8049 ("part-001.parquet", partition_bytes(&geoms, 5..9, None)),
8050 ("part-002.parquet", partition_bytes(&geoms, 9..n, None)),
8051 ]);
8052 assert_eq!(parts.len(), 3);
8053
8054 let pm_single = convert_sources_and_export(&single, dir.path(), "single", &opts);
8055 let pm_multi = convert_sources_and_export(&multi, dir.path(), "multi", &opts);
8056 assert!(
8057 pm_single == pm_multi,
8058 "remote multi-partition output must be byte-identical to the \
8059 single remote object ({} vs {} bytes)",
8060 pm_single.len(),
8061 pm_multi.len()
8062 );
8063 }
8064
8065 fn serve_bytes_over_http(body: Vec<u8>) -> String {
8073 use std::io::{BufRead, BufReader, Write};
8074 use std::net::TcpListener;
8075 use std::sync::Arc;
8076
8077 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
8078 let addr = listener.local_addr().unwrap();
8079 let body = Arc::new(body);
8080 std::thread::spawn(move || {
8081 for stream in listener.incoming() {
8082 let Ok(mut stream) = stream else { continue };
8083 let body = Arc::clone(&body);
8084 std::thread::spawn(move || {
8085 let peer = stream.try_clone().unwrap();
8086 let mut reader = BufReader::new(peer);
8087 let size = body.len() as u64;
8088 loop {
8090 let mut request_line = String::new();
8091 match reader.read_line(&mut request_line) {
8092 Ok(0) | Err(_) => break, Ok(_) => {}
8094 }
8095 let mut parts = request_line.split_whitespace();
8096 let method = parts.next().unwrap_or("").to_string();
8097 if method.is_empty() {
8098 break;
8099 }
8100 let mut range: Option<(u64, u64)> = None;
8102 loop {
8103 let mut header = String::new();
8104 if reader.read_line(&mut header).unwrap_or(0) == 0 {
8105 break;
8106 }
8107 if header == "\r\n" || header == "\n" {
8108 break;
8109 }
8110 let lower = header.to_ascii_lowercase();
8111 let Some(spec) = lower
8112 .strip_prefix("range:")
8113 .and_then(|v| v.trim().strip_prefix("bytes="))
8114 else {
8115 continue;
8116 };
8117 let spec = spec.split(',').next().unwrap_or("").trim();
8118 let (a, b) = spec.split_once('-').unwrap_or((spec, ""));
8119 let (start, end) = if a.is_empty() {
8120 let n: u64 = b.trim().parse().unwrap_or(0);
8122 (size.saturating_sub(n), size.saturating_sub(1))
8123 } else {
8124 let start = a.trim().parse().unwrap_or(0);
8125 let end = if b.trim().is_empty() {
8126 size.saturating_sub(1)
8127 } else {
8128 b.trim().parse().unwrap_or(size - 1)
8129 };
8130 (start, end.min(size.saturating_sub(1)))
8131 };
8132 range = Some((start, end));
8133 }
8134 let response: Vec<u8> = match (method.as_str(), range) {
8135 ("HEAD", _) => format!(
8136 "HTTP/1.1 200 OK\r\nContent-Type: application/octet-stream\r\n\
8137 Accept-Ranges: bytes\r\nContent-Length: {size}\r\n\r\n"
8138 )
8139 .into_bytes(),
8140 ("GET", Some((start, end))) => {
8141 let slice = &body[start as usize..=end as usize];
8142 let mut resp = format!(
8143 "HTTP/1.1 206 Partial Content\r\n\
8144 Content-Type: application/octet-stream\r\n\
8145 Accept-Ranges: bytes\r\n\
8146 Content-Range: bytes {start}-{end}/{size}\r\n\
8147 Content-Length: {}\r\n\r\n",
8148 slice.len()
8149 )
8150 .into_bytes();
8151 resp.extend_from_slice(slice);
8152 resp
8153 }
8154 ("GET", None) => {
8155 let mut resp = format!(
8156 "HTTP/1.1 200 OK\r\nContent-Type: application/octet-stream\r\n\
8157 Accept-Ranges: bytes\r\nContent-Length: {size}\r\n\r\n"
8158 )
8159 .into_bytes();
8160 resp.extend_from_slice(&body);
8161 resp
8162 }
8163 _ => {
8164 b"HTTP/1.1 405 Method Not Allowed\r\nContent-Length: 0\r\n\r\n"
8165 .to_vec()
8166 }
8167 };
8168 if stream.write_all(&response).is_err() {
8169 break;
8170 }
8171 let _ = stream.flush();
8172 }
8173 });
8174 }
8175 });
8176 format!("http://{addr}")
8177 }
8178
8179 #[test]
8184 fn remote_http_convert_matches_local() {
8185 let geoms = synthetic_geometries();
8186 let tin = tempfile::NamedTempFile::new().unwrap();
8187 write_input(tin.path(), &geoms, false, None);
8188 let opts = ConvertOptions::default();
8189
8190 let tout_local = tempfile::NamedTempFile::new().unwrap();
8191 let report_local = convert_to_overviews(tin.path(), tout_local.path(), &opts).unwrap();
8192
8193 let base = serve_bytes_over_http(std::fs::read(tin.path()).unwrap());
8194 let url = format!("{base}/in.parquet");
8195 let source = InputSource::from_str_input(&url).unwrap();
8196 assert!(source.is_remote(), "http:// input must be remote");
8197
8198 let tout_remote = tempfile::NamedTempFile::new().unwrap();
8199 let report_remote =
8200 convert_to_overviews_source(&source, tout_remote.path(), &opts).unwrap();
8201
8202 assert_eq!(report_remote.input_features, report_local.input_features);
8203 assert_eq!(
8204 read_all_ids(&OverviewReader::open(tout_remote.path()).unwrap()),
8205 read_all_ids(&OverviewReader::open(tout_local.path()).unwrap()),
8206 );
8207 let stats = report_remote
8208 .remote_fetch
8209 .expect("http input reports fetch stats");
8210 assert!(
8211 stats.bytes_fetched > 0 && stats.requests > 0,
8212 "http conversion moved bytes: {stats:?}"
8213 );
8214 }
8215
8216 #[test]
8219 fn unsupported_scheme_errors_through_convert() {
8220 let tout = tempfile::NamedTempFile::new().unwrap();
8221 let err = convert_to_overviews(
8222 Path::new("ftp://example.com/x.parquet"),
8223 tout.path(),
8224 &ConvertOptions::default(),
8225 )
8226 .unwrap_err();
8227 assert!(matches!(err, ConvertError::Input(_)), "got: {err}");
8228 assert!(err.to_string().contains("s3://"), "helpful message: {err}");
8229 }
8230
8231 #[test]
8239 fn remote_s3_city_extract_integration() {
8240 const URL: &str = "s3://tylertoo-bench/corpus/points-nyc-medium.rg20k.parquet";
8244 let source = match InputSource::from_str_input(URL) {
8245 Ok(s) => s,
8246 Err(e) => {
8247 eprintln!(
8248 "SKIP remote_s3_city_extract_integration (no credentials/network): {e}"
8249 );
8250 return;
8251 }
8252 };
8253 let tout = tempfile::NamedTempFile::new().unwrap();
8254 let opts = ConvertOptions {
8260 bbox: Some([-73.99, 40.72, -73.98, 40.73]),
8261 ..Default::default()
8262 };
8263 let report = match convert_to_overviews_source(&source, tout.path(), &opts) {
8264 Ok(r) => r,
8265 Err(e) => {
8266 eprintln!("SKIP remote_s3_city_extract_integration (network flake?): {e}");
8267 return;
8268 }
8269 };
8270 assert!(report.input_features > 0, "bbox should select features");
8271 assert!(
8272 report.row_groups_read < report.row_groups_total,
8273 "row-group pruning should fire on the Hilbert-sorted input \
8274 ({}/{} read)",
8275 report.row_groups_read,
8276 report.row_groups_total
8277 );
8278 let stats = report.remote_fetch.expect("remote stats");
8279 assert!(
8280 stats.bytes_fetched * 4 < stats.object_size,
8281 "city extract should move <25% of the remote object even \
8282 across streaming passes: {stats:?}"
8283 );
8284 eprintln!(
8285 "remote_s3_city_extract_integration: {} requests, {} of {} bytes ({:.2}%)",
8286 stats.requests,
8287 stats.bytes_fetched,
8288 stats.object_size,
8289 100.0 * stats.bytes_fetched as f64 / stats.object_size as f64
8290 );
8291 }
8292 }
8293
8294 fn tiny_polygons(n: usize) -> Vec<Geometry<f64>> {
8300 (0..n)
8301 .map(|i| {
8302 let cx = -150.0 + (i % 10) as f64 * 3.0;
8303 let cy = -60.0 + (i / 10) as f64 * 1.5;
8304 let h = 5e-5;
8305 let ext = LineString::from(vec![
8306 (cx - h, cy - h),
8307 (cx + h, cy - h),
8308 (cx + h, cy + h),
8309 (cx - h, cy + h),
8310 (cx - h, cy - h),
8311 ]);
8312 Geometry::Polygon(Polygon::new(ext, vec![]))
8313 })
8314 .collect()
8315 }
8316
8317 fn assert_clamped_pyramid(mode: Mode, streaming: bool) {
8320 let geoms = tiny_polygons(20);
8321 let tin = tempfile::NamedTempFile::new().unwrap();
8322 let tout = tempfile::NamedTempFile::new().unwrap();
8323 write_input(tin.path(), &geoms, false, None);
8324
8325 let opts = ConvertOptions {
8326 mode,
8327 levels: LevelPlan::ZoomRange {
8328 min_zoom: 0,
8329 max_zoom: 4,
8330 },
8331 streaming,
8332 ..Default::default()
8333 };
8334 let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
8335
8336 assert_eq!(report.levels.len(), 1, "expected a single written level");
8338 assert_eq!(report.levels[0].level, 0);
8339 assert_eq!(report.levels[0].zoom, Some(4));
8340 assert_eq!(report.levels[0].feature_count, geoms.len());
8341 assert_eq!(report.total_rows, geoms.len());
8342
8343 let skipped: Vec<(usize, Option<u8>)> = report
8346 .skipped_empty_levels
8347 .iter()
8348 .map(|s| (s.planned_level, s.zoom))
8349 .collect();
8350 assert_eq!(
8351 skipped,
8352 vec![(0, Some(0)), (1, Some(1)), (2, Some(2)), (3, Some(3))]
8353 );
8354 assert!(report.skipped_empty_levels.iter().all(|s| s.gsd > 0.0));
8355
8356 let vr = validate_file(tout.path()).unwrap();
8359 assert!(
8360 vr.is_valid(),
8361 "failures: {:?}",
8362 vr.failures().collect::<Vec<_>>()
8363 );
8364 let reader = OverviewReader::open(tout.path()).unwrap();
8365 assert_eq!(reader.num_levels(), 1);
8366 let tpm = tempfile::NamedTempFile::new().unwrap();
8367 let export = crate::overview::export::export_pmtiles(
8368 tout.path(),
8369 tpm.path(),
8370 &crate::overview::export::ExportOptions::default(),
8371 )
8372 .unwrap();
8373 assert_eq!(export.min_zoom, 4);
8374 assert_eq!(export.max_zoom, 4);
8375 }
8376
8377 #[test]
8378 fn empty_coarse_levels_clamped_duplicating_memory() {
8379 assert_clamped_pyramid(Mode::Duplicating, false);
8380 }
8381
8382 #[test]
8383 fn empty_coarse_levels_clamped_duplicating_streaming() {
8384 assert_clamped_pyramid(Mode::Duplicating, true);
8385 }
8386
8387 #[test]
8388 fn empty_coarse_levels_clamped_partitioning_memory() {
8389 assert_clamped_pyramid(Mode::Partitioning, false);
8390 }
8391
8392 #[test]
8393 fn empty_coarse_levels_clamped_partitioning_streaming() {
8394 assert_clamped_pyramid(Mode::Partitioning, true);
8395 }
8396
8397 #[test]
8400 fn all_levels_empty_is_hard_error() {
8401 for streaming in [false, true] {
8402 let tin = tempfile::NamedTempFile::new().unwrap();
8403 let tout = tempfile::NamedTempFile::new().unwrap();
8404 write_input(tin.path(), &[], false, None);
8405 let opts = ConvertOptions {
8406 mode: Mode::Duplicating,
8407 levels: LevelPlan::ZoomRange {
8408 min_zoom: 0,
8409 max_zoom: 3,
8410 },
8411 streaming,
8412 ..Default::default()
8413 };
8414 let err = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap_err();
8415 assert!(
8416 matches!(err, ConvertError::NoData),
8417 "streaming={streaming}: expected NoData, got {err:?}"
8418 );
8419 }
8420 }
8421
8422 #[test]
8427 fn write_time_empty_level_skipped_streaming() {
8428 let mut geoms = tiny_polygons(8);
8429 let square = |cx: f64, cy: f64| {
8435 let h = 5e-5;
8436 Polygon::new(
8437 LineString::from(vec![
8438 (cx - h, cy - h),
8439 (cx + h, cy - h),
8440 (cx + h, cy + h),
8441 (cx - h, cy + h),
8442 (cx - h, cy - h),
8443 ]),
8444 vec![],
8445 )
8446 };
8447 geoms.push(Geometry::MultiPolygon(geo::MultiPolygon::new(vec![
8448 square(-80.0, 10.0),
8449 square(80.0, 30.0),
8450 ])));
8451
8452 let tin = tempfile::NamedTempFile::new().unwrap();
8453 let tout = tempfile::NamedTempFile::new().unwrap();
8454 write_input(tin.path(), &geoms, false, None);
8455
8456 let opts = ConvertOptions {
8457 mode: Mode::Duplicating,
8458 levels: LevelPlan::ZoomRange {
8459 min_zoom: 0,
8460 max_zoom: 3,
8461 },
8462 streaming: true,
8463 ..Default::default()
8464 };
8465 let report = convert_to_overviews(tin.path(), tout.path(), &opts).unwrap();
8466
8467 assert_eq!(report.levels.len(), 1);
8471 assert_eq!(report.levels[0].level, 0);
8472 assert_eq!(report.levels[0].zoom, Some(3));
8473 assert_eq!(report.levels[0].feature_count, geoms.len());
8474 assert_eq!(
8475 report
8476 .skipped_empty_levels
8477 .iter()
8478 .map(|s| s.planned_level)
8479 .collect::<Vec<_>>(),
8480 vec![0, 1, 2]
8481 );
8482
8483 let vr = validate_file(tout.path()).unwrap();
8484 assert!(
8485 vr.is_valid(),
8486 "failures: {:?}",
8487 vr.failures().collect::<Vec<_>>()
8488 );
8489 let reader = OverviewReader::open(tout.path()).unwrap();
8490 assert_eq!(reader.num_levels(), 1);
8491 }
8492}