use anyhow::{Context, Result};
use arrow_array::Array;
use arrow_array::RecordBatch;
use arrow_cast::display::array_value_to_string;
use arrow_schema::Schema;
use rayon::prelude::*;
use crate::config::RuntimeConfig;
use crate::parsers::{
structured::{
constants::limits::{
TABULAR_BPR_SKINNY_MAX_BYTES, TABULAR_BYTE_SCALE_MIN_RETAIN_FRAC,
TABULAR_BYTE_SCALE_PCT_PER_DECADE, TABULAR_COLUMN_CHUNK,
TABULAR_FULL_SCAN_MAX_FILE_BYTES, TABULAR_FULL_SCAN_MAX_ROWS,
TABULAR_SAMPLE_BYTE_THRESHOLD,
},
table_sample_profile,
},
traits::AdaptiveParallel,
};
use crate::results::{
BooleanStats, DateStats, MergeColumnStatsInput, NumericStats, merge_column_stats,
};
pub use crate::parsers::structured::constants::limits::TABULAR_COL_SCALE_NUMERATOR;
#[must_use]
pub(crate) fn effective_sample_rows_after_file_byte_scaling(base: usize, file_bytes: u64) -> usize {
if base <= 1 {
return 1;
}
if file_bytes <= TABULAR_SAMPLE_BYTE_THRESHOLD {
return base;
}
let ratio = (file_bytes as f64) / (TABULAR_SAMPLE_BYTE_THRESHOLD as f64);
let k = ratio.log10().floor() as i32;
let k = k.max(1);
let retain = (1.0_f64 - TABULAR_BYTE_SCALE_PCT_PER_DECADE * f64::from(k))
.max(TABULAR_BYTE_SCALE_MIN_RETAIN_FRAC);
let scaled = (base as f64 * retain).floor() as usize;
scaled.max(1)
}
#[must_use]
pub(crate) fn effective_tabular_sample_rows(base: usize, column_count: usize) -> usize {
if base <= 1 {
return 1;
}
let cc = column_count.max(1);
(base.saturating_mul(TABULAR_COL_SCALE_NUMERATOR) / cc.max(TABULAR_COL_SCALE_NUMERATOR)).max(1)
}
#[must_use]
pub(crate) fn tabular_effective_sample_rows(
base: usize,
file_bytes: u64,
column_count: usize,
row_count: Option<usize>,
) -> usize {
let after_file = match row_count {
None | Some(0) => effective_sample_rows_after_file_byte_scaling(base, file_bytes),
Some(rc) => {
let bpr = file_bytes / (rc as u64);
let skinny = bpr <= TABULAR_BPR_SKINNY_MAX_BYTES;
let small_file = file_bytes <= TABULAR_FULL_SCAN_MAX_FILE_BYTES;
if skinny && small_file && rc <= TABULAR_FULL_SCAN_MAX_ROWS {
rc
} else if skinny && small_file {
effective_sample_rows_after_file_byte_scaling(base, file_bytes).min(rc)
} else {
effective_sample_rows_after_file_byte_scaling(base, file_bytes)
}
}
};
effective_tabular_sample_rows(after_file, column_count.max(1))
}
pub(crate) fn record_batch_row_to_strings(batch: &RecordBatch, row: usize) -> Result<Vec<String>> {
let mut out = Vec::with_capacity(batch.num_columns());
for col in 0..batch.num_columns() {
let arr = batch.column(col);
if arr.is_null(row) {
out.push(String::new());
} else {
out.push(array_value_to_string(arr.as_ref(), row)?);
}
}
Ok(out)
}
pub(crate) fn record_batch_all_rows_as_strings(
batch: &RecordBatch,
config: &RuntimeConfig,
) -> Result<Vec<Vec<String>>> {
let n = batch.num_rows();
if n == 0 {
return Ok(Vec::new());
}
if n >= config.min_collection_size_for_chunking {
(0..n)
.collect::<Vec<_>>()
.par_iter_adaptive(config)
.map(|i| record_batch_row_to_strings(batch, i))
.collect()
} else {
let mut rows = Vec::with_capacity(n);
for i in 0..n {
rows.push(record_batch_row_to_strings(batch, i)?);
}
Ok(rows)
}
}
pub(crate) fn record_batches_to_string_sample<E>(
batches: impl Iterator<Item = Result<RecordBatch, E>>,
max_sample: usize,
config: &RuntimeConfig,
decode_ctx: &'static str,
) -> Result<Vec<Vec<String>>>
where
E: std::error::Error + Send + Sync + 'static,
{
let mut sample_data = Vec::new();
for batch in batches {
let batch = batch.context(decode_ctx)?;
if sample_data.len() >= max_sample {
break;
}
let rows = record_batch_all_rows_as_strings(&batch, config)?;
for row in rows {
if sample_data.len() >= max_sample {
break;
}
sample_data.push(row);
}
}
Ok(sample_data)
}
#[must_use]
pub(crate) fn schema_column_names(schema: &Schema) -> Vec<String> {
schema.fields().iter().map(|f| f.name().clone()).collect()
}
#[must_use]
pub(crate) fn schema_arrow_dtype_strings(schema: &Schema) -> Vec<String> {
schema
.fields()
.iter()
.map(|f| format!("{}", f.data_type()))
.collect()
}
pub(crate) struct TabularSampleStats {
pub column_types: Option<Vec<String>>,
pub null_percentages: Option<Vec<f64>>,
pub unique_counts: Option<Vec<usize>>,
pub numeric_stats: Option<Vec<Option<NumericStats>>>,
pub date_stats: Option<Vec<Option<DateStats>>>,
pub boolean_stats: Option<Vec<Option<BooleanStats>>>,
}
fn narrow_rows_for_columns(
sample_data: &[Vec<String>],
col_start: usize,
col_end: usize,
) -> Vec<Vec<String>> {
let mut out = Vec::with_capacity(sample_data.len());
for row in sample_data {
let mut slice = Vec::with_capacity(col_end - col_start);
for i in col_start..col_end {
slice.push(row.get(i).cloned().unwrap_or_default());
}
out.push(slice);
}
out
}
fn tabular_stats_from_sample_one_chunk(
sample_data: &[Vec<String>],
column_count: usize,
config: &RuntimeConfig,
) -> TabularSampleStats {
let types = table_sample_profile::infer_column_types(sample_data, column_count, config);
let (null_percentages, unique_counts) =
table_sample_profile::compute_column_statistics(sample_data, column_count, config);
let (numeric_stats, date_stats, boolean_stats) =
table_sample_profile::compute_type_specific_statistics(
sample_data,
&types,
column_count,
config,
);
TabularSampleStats {
column_types: Some(types),
null_percentages: Some(null_percentages),
unique_counts: Some(unique_counts),
numeric_stats: Some(numeric_stats),
date_stats: Some(date_stats),
boolean_stats: Some(boolean_stats),
}
}
pub(crate) fn tabular_stats_from_sample(
sample_data: &[Vec<String>],
column_count: usize,
config: &RuntimeConfig,
) -> TabularSampleStats {
if sample_data.is_empty() || column_count == 0 {
return TabularSampleStats {
column_types: None,
null_percentages: None,
unique_counts: None,
numeric_stats: None,
date_stats: None,
boolean_stats: None,
};
}
if column_count <= TABULAR_COLUMN_CHUNK {
return tabular_stats_from_sample_one_chunk(sample_data, column_count, config);
}
let mut types = vec![String::new(); column_count];
let mut null_pct = vec![0.0f64; column_count];
let mut uniq = vec![0usize; column_count];
let mut numeric_stats = vec![None; column_count];
let mut date_stats = vec![None; column_count];
let mut boolean_stats = vec![None; column_count];
for col_start in (0..column_count).step_by(TABULAR_COLUMN_CHUNK) {
let col_end = (col_start + TABULAR_COLUMN_CHUNK).min(column_count);
let chunk_len = col_end - col_start;
let narrow = narrow_rows_for_columns(sample_data, col_start, col_end);
let ts = tabular_stats_from_sample_one_chunk(&narrow, chunk_len, config);
if let (Some(t), Some(np), Some(u), Some(ns), Some(ds), Some(bs)) = (
ts.column_types,
ts.null_percentages,
ts.unique_counts,
ts.numeric_stats,
ts.date_stats,
ts.boolean_stats,
) {
types[col_start..col_end].clone_from_slice(&t);
null_pct[col_start..col_end].clone_from_slice(&np);
uniq[col_start..col_end].clone_from_slice(&u);
for (i, v) in ns.into_iter().enumerate() {
numeric_stats[col_start + i] = v;
}
for (i, v) in ds.into_iter().enumerate() {
date_stats[col_start + i] = v;
}
for (i, v) in bs.into_iter().enumerate() {
boolean_stats[col_start + i] = v;
}
}
}
TabularSampleStats {
column_types: Some(types),
null_percentages: Some(null_pct),
unique_counts: Some(uniq),
numeric_stats: Some(numeric_stats),
date_stats: Some(date_stats),
boolean_stats: Some(boolean_stats),
}
}
#[must_use]
pub(crate) fn columns_from_tabular_sample(
column_count: usize,
column_names: Option<Vec<String>>,
ts: TabularSampleStats,
physical_types: Option<Vec<String>>,
) -> Option<Vec<crate::results::ColumnStat>> {
merge_column_stats(&MergeColumnStatsInput {
column_count,
column_names,
column_types: ts.column_types,
null_percentages: ts.null_percentages,
unique_counts: ts.unique_counts,
numeric_stats: ts.numeric_stats,
date_stats: ts.date_stats,
boolean_stats: ts.boolean_stats,
physical_types,
})
}