1use std::collections::{BTreeMap, BTreeSet};
18use std::sync::Arc;
19
20use anyhow::{Context, Result};
21use arrow::array::RecordBatch;
22use arrow::datatypes::{DataType, Field, Schema};
23use arrow::ipc::writer::StreamWriter;
24use serde::{Deserialize, Serialize};
25use stt_core::arrow_tile::{DecodedLayer, STT_QUANT_ATTR_META_KEY, STT_QUANT_META_KEY};
26use stt_core::compression::compress_zstd_with_dict_level;
27
28use crate::packed::PackedTileset;
29
30const COLUMN_ZSTD_LEVEL: i32 = 19;
34
35#[derive(Debug, Clone, Serialize, Deserialize)]
37pub struct ZoomStats {
38 pub zoom: u8,
40 pub entries: u64,
42 pub distinct_blobs: u64,
45 pub blob_bytes_total: u64,
48 pub blob_bytes_max: u64,
50 pub avg_blob_bytes: f64,
52 pub t_buckets: u64,
54}
55
56#[derive(Debug, Clone, Serialize, Deserialize)]
58pub struct DedupStats {
59 pub entries: u64,
61 pub distinct_blobs: u64,
63 pub dedup_ratio: f64,
66}
67
68#[derive(Debug, Clone, Serialize, Deserialize)]
70pub struct DecodeStats {
71 pub tiles_decoded: u64,
73 pub tiles_total: u64,
75 pub sampled: bool,
78 pub features_decoded: u64,
80 pub distinct_layer_schemas: u64,
83}
84
85#[derive(Debug, Clone, Serialize, Deserialize)]
88pub struct ColumnCost {
89 pub name: String,
91 pub dtype: String,
93 pub compressed_bytes: u64,
95 pub share: f64,
97 pub bytes_per_feature: f64,
99 pub encoding_note: String,
103}
104
105#[derive(Debug, Clone, Serialize, Deserialize)]
107pub struct InspectReport {
108 pub name: String,
110 pub min_zoom: u8,
112 pub max_zoom: u8,
114 pub time_start_ms: u64,
116 pub time_end_ms: u64,
118 pub temporal_bucket_ms: u64,
120 pub tile_count: u64,
122 pub feature_count: u64,
124 pub pack_count: u64,
126 pub paged_directory: bool,
128 pub compressed_bytes: u64,
130 pub uncompressed_bytes: u64,
132 pub compression_ratio: f64,
134 pub per_zoom: Vec<ZoomStats>,
136 pub dedup: DedupStats,
138 pub decode: DecodeStats,
140 pub per_column: Vec<ColumnCost>,
143}
144
145fn sample_stride(total: usize, n: usize) -> usize {
150 total.div_ceil(n).max(1)
151}
152
153fn schema_signature(layers: &[DecodedLayer]) -> String {
156 let mut parts: Vec<String> = layers
157 .iter()
158 .map(|layer| {
159 let cols: Vec<String> = layer
160 .batch
161 .schema()
162 .fields()
163 .iter()
164 .map(|f| format!("{}:{:?}", f.name(), f.data_type()))
165 .collect();
166 format!("{}{{{}}}", layer.name, cols.join(","))
167 })
168 .collect();
169 parts.sort();
170 parts.join("|")
171}
172
173fn contains_f64(dt: &DataType) -> bool {
175 match dt {
176 DataType::Float64 => true,
177 DataType::List(f) | DataType::LargeList(f) | DataType::FixedSizeList(f, _) => {
178 contains_f64(f.data_type())
179 }
180 DataType::Dictionary(_, v) => contains_f64(v),
181 _ => false,
182 }
183}
184
185fn contains_leaf(dt: &DataType, needle: &DataType) -> bool {
187 if dt == needle {
188 return true;
189 }
190 match dt {
191 DataType::List(f) | DataType::LargeList(f) | DataType::FixedSizeList(f, _) => {
192 contains_leaf(f.data_type(), needle)
193 }
194 _ => false,
195 }
196}
197
198fn encoding_note(field: &Field) -> String {
201 if field.metadata().contains_key(STT_QUANT_META_KEY) {
202 return "quantized coords (stt:quant)".to_string();
203 }
204 if field.metadata().contains_key(STT_QUANT_ATTR_META_KEY) {
205 return "quantized attr (stt:qa)".to_string();
206 }
207 if matches!(field.data_type(), DataType::Dictionary(_, _)) {
208 return "dictionary-encoded".to_string();
209 }
210 if field.name() == "vertex_time" {
211 if contains_leaf(field.data_type(), &DataType::UInt16) {
212 return "u16 vertex-time deltas".to_string();
213 }
214 if contains_leaf(field.data_type(), &DataType::Int64) {
215 return "i64 absolute vertex-time".to_string();
216 }
217 }
218 if contains_f64(field.data_type()) {
219 return "plain f64 (unquantized)".to_string();
220 }
221 String::new()
222}
223
224fn ipc_zstd_len(batch: &RecordBatch) -> Result<u64> {
227 let mut buf = Vec::new();
228 {
229 let mut w =
230 StreamWriter::try_new(&mut buf, &batch.schema()).context("column IPC writer init")?;
231 w.write(batch).context("column IPC write")?;
232 w.finish().context("column IPC finish")?;
233 }
234 Ok(compress_zstd_with_dict_level(&buf, None, COLUMN_ZSTD_LEVEL)?.len() as u64)
235}
236
237pub fn inspect(tileset: &PackedTileset, sample: Option<usize>) -> Result<InspectReport> {
246 let entries = tileset.entries();
247 let meta = tileset.metadata();
248
249 #[derive(Default)]
251 struct ZoomAcc {
252 entries: u64,
253 blobs: BTreeSet<(u32, u64)>,
254 bytes_total: u64,
255 bytes_max: u64,
256 t_starts: BTreeSet<i64>,
257 }
258 let mut per_zoom: BTreeMap<u8, ZoomAcc> = BTreeMap::new();
259 let mut all_blobs: BTreeSet<(u32, u64)> = BTreeSet::new();
260 let mut compressed_bytes = 0u64;
261 let mut uncompressed_bytes = 0u64;
262 let mut feature_count = 0u64;
263 for e in entries {
264 let z = per_zoom.entry(e.zoom).or_default();
265 z.entries += 1;
266 z.blobs.insert((e.pack_id, e.offset));
267 z.bytes_total += e.length as u64;
268 z.bytes_max = z.bytes_max.max(e.length as u64);
269 z.t_starts.insert(e.time_start);
270 all_blobs.insert((e.pack_id, e.offset));
271 compressed_bytes += e.length as u64;
272 uncompressed_bytes += e.uncompressed_size as u64;
273 feature_count += e.feature_count as u64;
274 }
275 let per_zoom: Vec<ZoomStats> = per_zoom
276 .into_iter()
277 .map(|(zoom, z)| ZoomStats {
278 zoom,
279 entries: z.entries,
280 distinct_blobs: z.blobs.len() as u64,
281 blob_bytes_total: z.bytes_total,
282 blob_bytes_max: z.bytes_max,
283 avg_blob_bytes: z.bytes_total as f64 / z.entries.max(1) as f64,
284 t_buckets: z.t_starts.len() as u64,
285 })
286 .collect();
287 let dedup = DedupStats {
288 entries: entries.len() as u64,
289 distinct_blobs: all_blobs.len() as u64,
290 dedup_ratio: all_blobs.len() as f64 / entries.len().max(1) as f64,
291 };
292
293 #[derive(Default)]
295 struct ColAcc {
296 dtype: String,
297 note: String,
298 compressed: u64,
299 rows: u64,
300 }
301 let mut cols: BTreeMap<String, ColAcc> = BTreeMap::new();
302 let mut schemas: BTreeSet<String> = BTreeSet::new();
303 let mut tiles_decoded = 0u64;
304 let mut features_decoded = 0u64;
305 let stride = sample.map(|n| {
306 if n == 0 {
307 usize::MAX
308 } else {
309 sample_stride(entries.len(), n)
310 }
311 });
312 for (idx, e) in entries.iter().enumerate() {
313 let decode_this = match stride {
314 None => true,
315 Some(usize::MAX) => false,
316 Some(s) => idx % s == 0,
317 };
318 if !decode_this {
319 continue;
320 }
321 let layers = tileset.read_layers(e).with_context(|| {
322 format!(
323 "decoding tile z{}/{}/{} t{}",
324 e.zoom, e.x, e.y, e.time_start
325 )
326 })?;
327 tiles_decoded += 1;
328 schemas.insert(schema_signature(&layers));
329 for layer in &layers {
330 let batch = &layer.batch;
331 let rows = batch.num_rows() as u64;
332 features_decoded += rows;
333 let schema = batch.schema();
334 for (i, field) in schema.fields().iter().enumerate() {
335 let clean = field.as_ref().clone().with_metadata(Default::default());
341 let one = RecordBatch::try_new(
342 Arc::new(Schema::new(vec![clean])),
343 vec![batch.column(i).clone()],
344 )
345 .context("single-column batch")?;
346 let c = cols.entry(field.name().clone()).or_default();
347 c.compressed += ipc_zstd_len(&one)?;
348 c.rows += rows;
349 c.dtype = format!("{:?}", field.data_type());
350 c.note = encoding_note(field);
351 }
352 }
353 }
354 let col_total: u64 = cols.values().map(|c| c.compressed).sum();
355 let mut per_column: Vec<ColumnCost> = cols
356 .into_iter()
357 .map(|(name, c)| ColumnCost {
358 name,
359 dtype: c.dtype,
360 compressed_bytes: c.compressed,
361 share: c.compressed as f64 / col_total.max(1) as f64,
362 bytes_per_feature: c.compressed as f64 / c.rows.max(1) as f64,
363 encoding_note: c.note,
364 })
365 .collect();
366 per_column.sort_by(|a, b| b.compressed_bytes.cmp(&a.compressed_bytes));
367
368 let time_range = tileset.time_range();
369 Ok(InspectReport {
370 name: tileset.name().to_string(),
371 min_zoom: meta.min_zoom,
372 max_zoom: meta.max_zoom,
373 time_start_ms: time_range.start,
374 time_end_ms: time_range.end,
375 temporal_bucket_ms: meta.temporal_bucket_ms,
376 tile_count: entries.len() as u64,
377 feature_count,
378 pack_count: tileset.pack_count() as u64,
379 paged_directory: tileset.is_paged(),
380 compressed_bytes,
381 uncompressed_bytes,
382 compression_ratio: uncompressed_bytes as f64 / compressed_bytes.max(1) as f64,
383 per_zoom,
384 dedup,
385 decode: DecodeStats {
386 tiles_decoded,
387 tiles_total: entries.len() as u64,
388 sampled: stride.is_some(),
389 features_decoded,
390 distinct_layer_schemas: schemas.len() as u64,
391 },
392 per_column,
393 })
394}
395
396pub fn format_text(report: &InspectReport) -> String {
398 let mut out = String::new();
399 out.push_str("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n");
400 out.push_str(&format!(" STT Inspect - {}\n", report.name));
401 out.push_str("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n\n");
402
403 out.push_str("📊 Dataset\n");
404 out.push_str(&format!(
405 " Tiles: {} Features (index): {} Zoom: {}-{}\n",
406 report.tile_count, report.feature_count, report.min_zoom, report.max_zoom
407 ));
408 out.push_str(&format!(
409 " Time: {}..{} ms Base bucket: {} ms\n",
410 report.time_start_ms, report.time_end_ms, report.temporal_bucket_ms
411 ));
412 out.push_str(&format!(
413 " Packs: {} Directory: {}\n",
414 report.pack_count,
415 if report.paged_directory {
416 "paged"
417 } else {
418 "single"
419 }
420 ));
421 out.push_str(&format!(
422 " Wire: {:.2} MB compressed -> {:.2} MB decoded ({:.2}x)\n\n",
423 report.compressed_bytes as f64 / 1e6,
424 report.uncompressed_bytes as f64 / 1e6,
425 report.compression_ratio
426 ));
427
428 out.push_str("🗂 Per-zoom directory\n");
429 out.push_str(" zoom | entries | distinct | total MB | max KB | avg KB | t-buckets\n");
430 for z in &report.per_zoom {
431 out.push_str(&format!(
432 " {:2} | {:8} | {:8} | {:9.2} | {:7.1} | {:7.1} | {:9}\n",
433 z.zoom,
434 z.entries,
435 z.distinct_blobs,
436 z.blob_bytes_total as f64 / 1e6,
437 z.blob_bytes_max as f64 / 1e3,
438 z.avg_blob_bytes / 1e3,
439 z.t_buckets
440 ));
441 }
442 out.push_str(&format!(
443 " dedup: {} entries -> {} distinct blobs (ratio {:.3})\n\n",
444 report.dedup.entries, report.dedup.distinct_blobs, report.dedup.dedup_ratio
445 ));
446
447 out.push_str(&format!(
448 "🔬 Decode ({} of {} tiles{})\n",
449 report.decode.tiles_decoded,
450 report.decode.tiles_total,
451 if report.decode.sampled {
452 ", sampled"
453 } else {
454 ""
455 }
456 ));
457 out.push_str(&format!(
458 " features decoded: {} distinct layer schemas: {}\n\n",
459 report.decode.features_decoded, report.decode.distinct_layer_schemas
460 ));
461
462 if !report.per_column.is_empty() {
463 out.push_str("💾 Per-column cost (standalone IPC+zstd-19; shares, not absolute wire)\n");
464 out.push_str(&format!(
465 " {:<22} {:<28} {:>10} {:>9} {:>7} note\n",
466 "column", "dtype", "comp KB", "B/feat", "share%"
467 ));
468 for c in &report.per_column {
469 let dt = if c.dtype.len() > 27 {
470 format!("{}…", &c.dtype[..26])
471 } else {
472 c.dtype.clone()
473 };
474 out.push_str(&format!(
475 " {:<22} {:<28} {:>10.1} {:>9.2} {:>6.1}% {}\n",
476 c.name,
477 dt,
478 c.compressed_bytes as f64 / 1e3,
479 c.bytes_per_feature,
480 100.0 * c.share,
481 c.encoding_note
482 ));
483 }
484 }
485
486 out.push_str("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n");
487 out
488}
489
490#[cfg(test)]
491mod tests {
492 use super::*;
493 use stt_core::arrow_tile::{encode_tile, ColumnarLayer, GeometryColumn, PropertyColumn};
494 use stt_core::curve::BlobOrdering;
495 use stt_core::metadata::Metadata;
496 use stt_core::pack::PackWriter;
497 use stt_core::tile::TileId;
498
499 fn line_layer(seed: u64, n: usize) -> ColumnarLayer {
503 let verts_per = 8usize;
504 let geometry: Vec<Vec<[f64; 2]>> = (0..n)
505 .map(|i| {
506 (0..verts_per)
507 .map(|v| {
508 [
509 -73.6 + (seed as f64) * 0.01 + v as f64 * 0.001,
510 45.5 + i as f64 * 0.002,
511 ]
512 })
513 .collect()
514 })
515 .collect();
516 let vertex_times: Vec<Vec<i64>> = (0..n)
517 .map(|_| (0..verts_per).map(|v| v as i64 * 50).collect())
518 .collect();
519 ColumnarLayer {
520 name: "default".to_string(),
521 feature_ids: (0..n as u64).map(|i| seed * 1000 + i).collect(),
522 start_times: vec![0; n],
523 end_times: vec![400; n],
524 geometry: GeometryColumn::LineString(geometry),
525 vertex_times: Some(vertex_times),
526 vertex_values: None,
527 triangles: None,
528 vertex_value_matrix: None,
529 properties: vec![
530 (
531 "speed".to_string(),
532 PropertyColumn::Numeric((0..n).map(|i| Some(i as f64 * 1.5)).collect()),
533 ),
534 (
535 "kind".to_string(),
536 PropertyColumn::Categorical(
537 (0..n)
538 .map(|i| Some(["bike", "ferry"][i % 2].to_string()))
539 .collect(),
540 ),
541 ),
542 ],
543 }
544 }
545
546 fn build_fixture(out: &std::path::Path) {
549 let mut w = PackWriter::create(out, BlobOrdering::Auto, 64 * 1024).unwrap();
550 let bucket = 3_600_000i64;
551 let dup = encode_tile(&[line_layer(7, 40)]).unwrap();
552 w.add_tile_full(
555 &TileId::new(5, 1, 1, 0),
556 0,
557 bucket - 1,
558 Some(0),
559 40,
560 Some(bucket as u64),
561 &dup,
562 )
563 .unwrap();
564 w.add_tile_full(
565 &TileId::new(5, 2, 1, bucket as u64),
566 bucket,
567 2 * bucket - 1,
568 Some(bucket),
569 40,
570 Some(bucket as u64),
571 &dup,
572 )
573 .unwrap();
574 let distinct = encode_tile(&[line_layer(9, 40)]).unwrap();
575 w.add_tile_full(
576 &TileId::new(5, 3, 1, 0),
577 0,
578 bucket - 1,
579 Some(0),
580 40,
581 Some(bucket as u64),
582 &distinct,
583 )
584 .unwrap();
585 let overview = encode_tile(&[line_layer(11, 40)]).unwrap();
587 w.add_tile_full(
588 &TileId::new(3, 0, 0, 0),
589 0,
590 bucket - 1,
591 Some(0),
592 40,
593 Some(bucket as u64),
594 &overview,
595 )
596 .unwrap();
597 let meta = Metadata::new("inspect-fixture")
598 .with_temporal_bucket_ms(bucket as u64)
599 .with_zoom_levels(3, 5);
600 w.finalize(&meta).unwrap();
601 }
602
603 #[test]
604 fn inspect_full_report_on_real_fixture() {
605 let dir = tempfile::tempdir().unwrap();
606 let out = dir.path().join("dataset");
607 build_fixture(&out);
608
609 let ts = PackedTileset::open(&out).unwrap();
610 let report = inspect(&ts, None).unwrap();
611
612 assert_eq!(report.tile_count, 4);
614 assert_eq!(report.per_zoom.len(), 2);
615 let z3 = &report.per_zoom[0];
616 let z5 = &report.per_zoom[1];
617 assert_eq!(
618 (z3.zoom, z3.entries, z3.distinct_blobs, z3.t_buckets),
619 (3, 1, 1, 1)
620 );
621 assert_eq!((z5.zoom, z5.entries, z5.t_buckets), (5, 3, 2));
622 assert_eq!(z5.distinct_blobs, 2);
624 assert!(z5.blob_bytes_max > 0);
625 assert!((z5.avg_blob_bytes - z5.blob_bytes_total as f64 / 3.0).abs() < 1e-9);
626
627 assert_eq!(report.dedup.entries, 4);
629 assert_eq!(report.dedup.distinct_blobs, 3);
630 assert!(report.dedup.dedup_ratio < 1.0);
631
632 assert!(
634 report.compression_ratio > 1.0,
635 "ratio {}",
636 report.compression_ratio
637 );
638 assert!(report.compressed_bytes > 0 && report.uncompressed_bytes > report.compressed_bytes);
639
640 assert!(!report.decode.sampled);
642 assert_eq!(report.decode.tiles_decoded, 4);
643 assert_eq!(report.decode.features_decoded, 160);
644 assert_eq!(report.decode.distinct_layer_schemas, 1);
645 assert_eq!(report.feature_count, 160);
646
647 let share_sum: f64 = report.per_column.iter().map(|c| c.share).sum();
650 assert!((share_sum - 1.0).abs() < 1e-9, "shares sum to {share_sum}");
651 let by_name = |n: &str| {
652 report
653 .per_column
654 .iter()
655 .find(|c| c.name == n)
656 .unwrap_or_else(|| panic!("column {n} missing"))
657 };
658 for name in ["geometry", "vertex_time", "speed", "kind", "id"] {
659 assert!(by_name(name).compressed_bytes > 0);
660 assert!(by_name(name).bytes_per_feature > 0.0);
661 }
662
663 assert_eq!(by_name("geometry").encoding_note, "plain f64 (unquantized)");
665 assert_eq!(by_name("speed").encoding_note, "plain f64 (unquantized)");
666 assert_eq!(by_name("kind").encoding_note, "dictionary-encoded");
667 assert_eq!(
668 by_name("vertex_time").encoding_note,
669 "u16 vertex-time deltas"
670 );
671
672 let text = format_text(&report);
674 assert!(text.contains("inspect-fixture"));
675 assert!(text.contains("geometry"));
676 assert!(text.contains("dedup: 4 entries -> 3 distinct blobs"));
677 assert!(!text.contains("sampled"));
678 }
679
680 #[test]
681 fn inspect_sampled_decode_keeps_directory_stats_total() {
682 let dir = tempfile::tempdir().unwrap();
683 let out = dir.path().join("dataset");
684 build_fixture(&out);
685 let ts = PackedTileset::open(&out).unwrap();
686
687 let report = inspect(&ts, Some(2)).unwrap();
689 assert!(report.decode.sampled);
690 assert_eq!(report.decode.tiles_decoded, 2);
691 assert_eq!(report.decode.features_decoded, 80);
692 assert_eq!(report.tile_count, 4);
694 assert_eq!(report.dedup.entries, 4);
695 assert_eq!(report.dedup.distinct_blobs, 3);
696 assert_eq!(report.per_zoom.iter().map(|z| z.entries).sum::<u64>(), 4);
697 let share_sum: f64 = report.per_column.iter().map(|c| c.share).sum();
699 assert!((share_sum - 1.0).abs() < 1e-9);
700 let rerun = inspect(&ts, Some(2)).unwrap();
702 assert_eq!(
703 rerun
704 .per_column
705 .iter()
706 .map(|c| (c.name.clone(), c.compressed_bytes))
707 .collect::<Vec<_>>(),
708 report
709 .per_column
710 .iter()
711 .map(|c| (c.name.clone(), c.compressed_bytes))
712 .collect::<Vec<_>>()
713 );
714 assert!(format_text(&report).contains("sampled"));
715
716 let none = inspect(&ts, Some(0)).unwrap();
718 assert_eq!(none.decode.tiles_decoded, 0);
719 assert!(none.per_column.is_empty());
720 assert_eq!(none.dedup.entries, 4);
721 let all = inspect(&ts, Some(100)).unwrap();
722 assert_eq!(all.decode.tiles_decoded, 4);
723 assert!(all.decode.sampled);
724 }
725
726 #[test]
727 fn report_serializes_to_json_and_back() {
728 let dir = tempfile::tempdir().unwrap();
729 let out = dir.path().join("dataset");
730 build_fixture(&out);
731 let ts = PackedTileset::open(&out).unwrap();
732 let report = inspect(&ts, None).unwrap();
733
734 let json = serde_json::to_string_pretty(&report).unwrap();
735 let back: InspectReport = serde_json::from_str(&json).unwrap();
736 assert_eq!(back.tile_count, report.tile_count);
737 assert_eq!(back.per_column.len(), report.per_column.len());
738 assert_eq!(back.dedup.distinct_blobs, report.dedup.distinct_blobs);
739 }
740}