use std::path::{Path, PathBuf};
const DATA_EXTENSIONS: &[&str] = &[
"parquet", "csv", "tsv", "txt", "json", "ndjson", "jsonl", "ipc", "arrow", "feather", "avro",
"orc", "xlsx", "xls", "xlsm",
];
const COMPRESSION_EXTENSIONS: &[&str] = &["gz", "bz2", "xz", "zst", "zstd"];
const HIVE_PROBE_LIMIT: usize = 8;
pub const MAX_ENTRIES_PER_DIR: usize = 5_000;
const MAX_CLASSIFY_PER_DIR: usize = 64;
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum EntryKind {
File,
Hive,
MultiFile,
Directory,
Unknown,
}
impl EntryKind {
pub fn label(self) -> &'static str {
match self {
EntryKind::File => "",
EntryKind::Hive => "hive",
EntryKind::MultiFile => "multi",
EntryKind::Directory => "dir",
EntryKind::Unknown => "",
}
}
pub fn is_dataset(self) -> bool {
!matches!(self, EntryKind::Directory)
}
}
#[derive(Debug, Clone)]
pub struct Entry {
pub path: PathBuf,
pub kind: EntryKind,
pub name: String,
pub size: Option<u64>,
pub modified: Option<std::time::SystemTime>,
pub rows: Option<usize>,
pub cols: Option<usize>,
pub columns: Vec<String>,
pub cost: Cost,
}
#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct Cost {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub source: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub uncompressed: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub codec: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub row_groups: Option<usize>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub partitions: Option<Partitions>,
}
#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct Partitions {
pub keys: Vec<String>,
pub first_key_values: Vec<String>,
pub count: usize,
pub more: bool,
}
impl Entry {
pub fn directory(path: &Path) -> Self {
Self::new(path.to_path_buf(), EntryKind::Directory)
}
pub fn for_test(path: &Path, name: &str) -> Self {
let mut entry = Self::new(path.to_path_buf(), EntryKind::File);
entry.name = name.to_string();
entry
}
pub(crate) fn new(path: PathBuf, kind: EntryKind) -> Self {
let name = path
.file_name()
.map(|n| n.to_string_lossy().into_owned())
.unwrap_or_else(|| path.to_string_lossy().into_owned());
Self {
path,
kind,
name,
size: None,
modified: None,
rows: None,
cols: None,
columns: Vec::new(),
cost: Cost::default(),
}
}
pub(crate) fn with_fs_metadata(mut self, meta: &std::fs::Metadata) -> Self {
if meta.is_file() {
self.size = Some(meta.len());
}
self.modified = meta.modified().ok();
self
}
}
pub fn is_data_file(path: &Path) -> bool {
let Some(name) = path.file_name().and_then(|n| n.to_str()) else {
return false;
};
let lower = name.to_ascii_lowercase();
let mut parts: Vec<&str> = lower.rsplit('.').collect();
parts.reverse();
if parts.len() < 2 {
return false;
}
let mut idx = parts.len() - 1;
if COMPRESSION_EXTENSIONS.contains(&parts[idx]) && idx > 1 {
idx -= 1;
}
DATA_EXTENSIONS.contains(&parts[idx])
}
fn is_partition_dir(path: &Path) -> bool {
path.file_name()
.and_then(|n| n.to_str())
.map(|n| {
matches!(n.find('='), Some(i) if i > 0)
})
.unwrap_or(false)
}
pub fn classify_directory(path: &Path) -> EntryKind {
let Ok(iter) = std::fs::read_dir(path) else {
return EntryKind::Directory;
};
let mut partitions = 0usize;
let mut data_files = 0usize;
let mut seen = 0usize;
let mut extension: Option<String> = None;
let mut mixed_extensions = false;
for entry in iter.flatten() {
let entry_path = entry.path();
let name = entry.file_name();
if name.to_string_lossy().starts_with('.') || name == "_SUCCESS" {
continue;
}
if entry_path.is_dir() {
if is_partition_dir(&entry_path) {
partitions += 1;
}
} else if is_data_file(&entry_path) {
data_files += 1;
let ext = entry_path
.extension()
.and_then(|e| e.to_str())
.map(|e| e.to_ascii_lowercase());
match (&extension, ext) {
(None, Some(e)) => extension = Some(e),
(Some(current), Some(e)) if *current != e => mixed_extensions = true,
_ => {}
}
}
seen += 1;
if seen >= HIVE_PROBE_LIMIT {
break;
}
}
if partitions > 0 && partitions >= data_files {
return EntryKind::Hive;
}
let homogeneous = data_files > 1 && !mixed_extensions;
let mostly_data = data_files * 2 >= seen;
if homogeneous && mostly_data {
EntryKind::MultiFile
} else {
EntryKind::Directory
}
}
pub fn scan_dir(dir: &Path) -> Vec<Entry> {
scan_dir_bounded(dir).entries
}
#[derive(Debug, Clone, Default)]
pub struct Scan {
pub entries: Vec<Entry>,
pub truncated: bool,
}
pub fn scan_dir_bounded(dir: &Path) -> Scan {
let Ok(iter) = std::fs::read_dir(dir) else {
return Scan::default();
};
let mut entries = Vec::new();
let mut classified = 0usize;
let mut seen = 0usize;
let mut truncated = false;
for dir_entry in iter.flatten().take(MAX_ENTRIES_PER_DIR + 1) {
seen += 1;
if seen > MAX_ENTRIES_PER_DIR {
truncated = true;
break;
}
let path = dir_entry.path();
let name = dir_entry.file_name();
if name.to_string_lossy().starts_with('.') {
continue;
}
let Ok(meta) = dir_entry.metadata() else {
continue;
};
let kind = if meta.is_dir() {
if classified < MAX_CLASSIFY_PER_DIR {
classified += 1;
classify_directory(&path)
} else {
EntryKind::Directory
}
} else if meta.is_file() && is_data_file(&path) {
EntryKind::File
} else {
continue;
};
entries.push(Entry::new(path, kind).with_fs_metadata(&meta));
}
sort_entries(&mut entries);
Scan { entries, truncated }
}
fn sort_entries(entries: &mut [Entry]) {
entries.sort_by(|a, b| {
let group = |k: EntryKind| if k.is_dataset() { 0 } else { 1 };
group(a.kind).cmp(&group(b.kind)).then_with(|| {
a.name
.to_ascii_lowercase()
.cmp(&b.name.to_ascii_lowercase())
})
});
}
const MAX_FOOTERS_PER_DATASET: usize = 64;
pub fn enrich(entry: &mut Entry) {
match entry.kind {
EntryKind::File => enrich_parquet(entry),
EntryKind::Hive | EntryKind::MultiFile => enrich_dataset(entry),
EntryKind::Directory | EntryKind::Unknown => {}
}
}
fn enrich_dataset(entry: &mut Entry) {
if entry.kind == EntryKind::Hive {
entry.cost.partitions = partition_layout(&entry.path);
}
entry.size = None;
let mut files = Vec::new();
collect_parquet_files(&entry.path, 0, &mut files);
if files.is_empty() || files.len() > MAX_FOOTERS_PER_DATASET {
if let Some(first) = files.first() {
if let Some(meta) = crate::widgets::info::read_parquet_metadata(first) {
entry.cols = Some(meta.schema_descr.columns().len());
entry.columns = column_names(&meta);
physical_facts(&meta, &mut entry.cost);
entry.cost.uncompressed = None;
}
}
return;
}
let mut rows = 0usize;
let mut cols = None;
let mut bytes = 0u64;
let mut columns = Vec::new();
let mut cost = Cost::default();
let mut uncompressed = 0u64;
let mut row_groups = 0usize;
for file in &files {
let Some(meta) = crate::widgets::info::read_parquet_metadata(file) else {
return; };
rows += meta.num_rows;
cols.get_or_insert(meta.schema_descr.columns().len());
if columns.is_empty() {
columns = column_names(&meta);
}
let mut per_file = Cost::default();
physical_facts(&meta, &mut per_file);
uncompressed += per_file.uncompressed.unwrap_or(0);
row_groups += per_file.row_groups.unwrap_or(0);
if cost.codec.is_none() {
cost.codec = per_file.codec;
}
if let Ok(m) = std::fs::metadata(file) {
bytes += m.len();
}
}
entry.rows = Some(rows);
entry.cols = cols;
entry.size = Some(bytes);
entry.columns = columns;
cost.uncompressed = (uncompressed > 0).then_some(uncompressed);
cost.row_groups = (row_groups > 0).then_some(row_groups);
cost.partitions = entry.cost.partitions.take();
entry.cost = cost;
}
fn collect_parquet_files(dir: &Path, depth: u8, out: &mut Vec<PathBuf>) {
if depth > 4 || out.len() > MAX_FOOTERS_PER_DATASET {
return;
}
let Ok(iter) = std::fs::read_dir(dir) else {
return;
};
let mut subdirs = Vec::new();
for entry in iter.flatten() {
let path = entry.path();
if path.is_dir() {
subdirs.push(path);
} else if path
.extension()
.and_then(|e| e.to_str())
.map(|e| e.eq_ignore_ascii_case("parquet"))
.unwrap_or(false)
&& is_regular_file(&path)
{
out.push(path);
if out.len() > MAX_FOOTERS_PER_DATASET {
return;
}
}
}
subdirs.sort();
for sub in subdirs {
collect_parquet_files(&sub, depth + 1, out);
if out.len() > MAX_FOOTERS_PER_DATASET {
return;
}
}
}
pub fn enrich_parquet(entry: &mut Entry) {
if entry.kind != EntryKind::File {
return;
}
let is_parquet = entry
.path
.extension()
.and_then(|e| e.to_str())
.map(|e| e.eq_ignore_ascii_case("parquet"))
.unwrap_or(false);
if !is_parquet {
return;
}
if !is_regular_file(&entry.path) {
return;
}
if let Some(meta) = crate::widgets::info::read_parquet_metadata(&entry.path) {
entry.rows = Some(meta.num_rows);
entry.cols = Some(meta.schema_descr.columns().len());
entry.columns = column_names(&meta);
physical_facts(&meta, &mut entry.cost);
}
}
pub fn physical_facts(meta: &crate::widgets::info::ParquetMetadataCache, cost: &mut Cost) {
if meta.row_groups.is_empty() {
return;
}
cost.row_groups = Some(meta.row_groups.len());
let mut uncompressed: u64 = 0;
let mut codecs: Vec<String> = Vec::new();
for rg in &meta.row_groups {
uncompressed = uncompressed.saturating_add(rg.total_byte_size() as u64);
for cc in rg.parquet_columns() {
let codec = format!("{:?}", cc.compression()).to_lowercase();
if !codecs.contains(&codec) {
codecs.push(codec);
}
}
}
if uncompressed > 0 {
cost.uncompressed = Some(uncompressed);
}
cost.codec = match codecs.len() {
0 => None,
1 => Some(codecs.remove(0)),
n => Some(format!("mixed ({n})")),
};
}
const MAX_PARTITION_DIRS: usize = 512;
pub fn partition_layout(dir: &Path) -> Option<Partitions> {
let iter = std::fs::read_dir(dir).ok()?;
let mut values: Vec<String> = Vec::new();
let mut keys: Vec<String> = Vec::new();
let mut count = 0usize;
let mut more = false;
for entry in iter.flatten() {
if count >= MAX_PARTITION_DIRS {
more = true;
break;
}
let name = entry.file_name().to_string_lossy().into_owned();
let Some((key, value)) = name.split_once('=') else {
continue;
};
if !entry.path().is_dir() {
continue;
}
if keys.is_empty() {
keys.push(key.to_string());
keys.extend(nested_keys(&entry.path()));
}
values.push(value.to_string());
count += 1;
}
if keys.is_empty() {
return None;
}
values.sort();
values.dedup();
Some(Partitions {
keys,
first_key_values: values,
count,
more,
})
}
fn nested_keys(dir: &Path) -> Vec<String> {
let mut keys = Vec::new();
let mut current = dir.to_path_buf();
for _ in 0..6 {
let Ok(iter) = std::fs::read_dir(¤t) else {
break;
};
let Some(child) = iter
.flatten()
.find(|e| e.file_name().to_string_lossy().contains('=') && e.path().is_dir())
else {
break;
};
let name = child.file_name().to_string_lossy().into_owned();
let Some((key, _)) = name.split_once('=') else {
break;
};
keys.push(key.to_string());
current = child.path();
}
keys
}
pub fn format_size(bytes: u64) -> String {
const UNITS: &[&str] = &["B", "KB", "MB", "GB", "TB"];
let mut value = bytes as f64;
let mut unit = 0;
while value >= 1024.0 && unit < UNITS.len() - 1 {
value /= 1024.0;
unit += 1;
}
if unit == 0 {
format!("{} {}", bytes, UNITS[0])
} else if value >= 100.0 {
format!("{:.0} {}", value, UNITS[unit])
} else {
format!("{:.1} {}", value, UNITS[unit])
}
}
pub fn format_rows(rows: usize) -> String {
let r = rows as f64;
if rows >= 1_000_000_000 {
format!("{:.1}B", r / 1e9)
} else if rows >= 1_000_000 {
format!("{:.1}M", r / 1e6)
} else if rows >= 10_000 {
format!("{:.0}k", r / 1e3)
} else if rows >= 1_000 {
let mut out = String::new();
let digits = rows.to_string();
for (i, c) in digits.chars().enumerate() {
if i > 0 && (digits.len() - i).is_multiple_of(3) {
out.push(',');
}
out.push(c);
}
out
} else {
rows.to_string()
}
}
pub fn format_age(t: std::time::SystemTime) -> String {
let Ok(elapsed) = t.elapsed() else {
return String::new();
};
let secs = elapsed.as_secs();
if secs < 60 {
"now".to_string()
} else if secs < 3600 {
format!("{}m", secs / 60)
} else if secs < 86_400 {
format!("{}h", secs / 3600)
} else if secs < 86_400 * 365 {
format!("{}d", secs / 86_400)
} else {
format!("{}y", secs / (86_400 * 365))
}
}
pub type SchemaPreview = Vec<(String, polars::prelude::DataType)>;
fn first_parquet_under(dir: &Path, depth: u8) -> Option<PathBuf> {
if depth > 4 {
return None;
}
let mut subdirs = Vec::new();
for entry in std::fs::read_dir(dir).ok()?.flatten().take(64) {
let path = entry.path();
if path.is_dir() {
subdirs.push(path);
} else if path
.extension()
.and_then(|e| e.to_str())
.map(|e| e.eq_ignore_ascii_case("parquet"))
.unwrap_or(false)
&& is_regular_file(&path)
{
return Some(path);
}
}
subdirs.sort();
subdirs
.into_iter()
.take(4)
.find_map(|d| first_parquet_under(&d, depth + 1))
}
pub fn column_names(meta: &crate::widgets::info::ParquetMetadataCache) -> Vec<String> {
meta.schema_descr
.columns()
.iter()
.map(|c| c.path_in_schema.join("."))
.collect()
}
fn is_regular_file(path: &Path) -> bool {
std::fs::metadata(path)
.map(|m| m.file_type().is_file())
.unwrap_or(false)
}
pub fn schema_preview(entry: &Entry) -> Option<SchemaPreview> {
use polars::prelude::{ParquetReader, Schema, SchemaExt, SerReader};
let file_path = match entry.kind {
EntryKind::File => {
let is_parquet = entry
.path
.extension()
.and_then(|e| e.to_str())
.map(|e| e.eq_ignore_ascii_case("parquet"))
.unwrap_or(false);
if !is_parquet {
return None;
}
entry.path.clone()
}
EntryKind::Hive | EntryKind::MultiFile => first_parquet_under(&entry.path, 0)?,
EntryKind::Directory | EntryKind::Unknown => return None,
};
if !is_regular_file(&file_path) {
return None;
}
let file = std::fs::File::open(&file_path).ok()?;
let mut reader = ParquetReader::new(file);
let arrow_schema = reader.schema().ok()?;
let schema = Schema::from_arrow_schema(arrow_schema.as_ref());
Some(
schema
.iter()
.map(|(name, dtype)| (name.to_string(), dtype.clone()))
.collect(),
)
}