Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 0 additions & 5 deletions crates/rudb-bind/src/partition.rs
Original file line number Diff line number Diff line change
Expand Up @@ -223,11 +223,6 @@ pub(crate) fn partitioning(
"RETURN_STATS is not supported for the \"csv\" copy format",
));
}
if out.returns == Returns::Stats {
return Err(Error::not_implemented(
"COPY TO with the option return_stats is not supported yet",
));
}
if partitioned && !out.write_columns && out.columns.len() == names.len() {
return Err(Error::not_implemented(
"No column to write as all columns are specified as partition columns. \
Expand Down
2 changes: 1 addition & 1 deletion crates/rudb-parquet/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -51,4 +51,4 @@ pub use metadata::{
pub use page::{Body, DataV1, DataV2, Dictionary, Header};
pub use prune::{Bound, Footer, Op, Test, skips};
pub use reader::{Reader, read};
pub use write::{Writer, storage};
pub use write::{ColumnStatistics, Writer, storage};
73 changes: 71 additions & 2 deletions crates/rudb-parquet/src/write.rs
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@ struct Column {
struct Written {
values: i64,
nulls: i64,
nans: i64,
compressed: i64,
uncompressed: i64,
min: Option<Value>,
Expand All @@ -54,6 +55,27 @@ pub struct Writer<W: Write> {
groups: Vec<thrift::Writer>,
rows: i64,
created_by: String,
totals: Vec<ColumnStatistics>,
}

/// What one column of a file holds over all of its row groups, which is what a `COPY ... TO`
/// with `RETURN_STATS` reports.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct ColumnStatistics {
/// The name of the column.
pub name: String,
/// How many values it holds, the nulls among them.
pub values: i64,
/// How many of them are null.
pub nulls: i64,
/// How many are NaN, for a `FLOAT` or a `DOUBLE` column, and `None` for any other.
pub nans: Option<i64>,
/// How many bytes its column chunks take in the file, page headers and all.
pub size: i64,
/// The smallest value that is neither null nor NaN, of the type it was stored as.
pub min: Option<Value>,
/// The largest.
pub max: Option<Value>,
}

/// The type a column of `ty` has to be cast to before it is handed to a [`Writer`].
Expand Down Expand Up @@ -147,6 +169,14 @@ impl<W: Write> Writer<W> {
})?;
columns.push(Column { name: field.name.clone(), ty: field.ty.clone(), physical });
}
let totals = columns
.iter()
.map(|column| ColumnStatistics {
name: column.name.clone(),
nans: matches!(column.physical, Physical::Float | Physical::Double).then_some(0),
..ColumnStatistics::default()
})
.collect();
let mut writer = Self {
out,
at: 0,
Expand All @@ -155,6 +185,7 @@ impl<W: Write> Writer<W> {
groups: Vec::new(),
rows: 0,
created_by: created_by.to_string(),
totals,
};
writer.put(b"PAR1")?;
Ok(writer)
Expand Down Expand Up @@ -193,6 +224,17 @@ impl<W: Write> Writer<W> {
total += written.uncompressed;
compressed += written.compressed;
columns.push(self.chunk_meta(column, first, &written));
let totals = &mut self.totals[column];
totals.values += written.values;
totals.nulls += written.nulls;
totals.nans = totals.nans.map(|nans| nans + written.nans);
totals.size += written.compressed;
if let Some(min) = &written.min {
bound(&mut totals.min, min, Ordering::Less);
}
if let Some(max) = &written.max {
bound(&mut totals.max, max, Ordering::Greater);
}
}
let mut group = thrift::Writer::default();
group.list_of_nested(1, columns);
Expand Down Expand Up @@ -220,6 +262,11 @@ impl<W: Write> Writer<W> {
written.nulls += 1;
continue;
}
if matches!(value, Value::Double(value) if value.is_nan())
|| matches!(value, Value::Float(value) if value.is_nan())
{
written.nans += 1;
}
defined.push(true);
if column.physical == Physical::Boolean {
bits.push(matches!(value, Value::Boolean(true)));
Expand Down Expand Up @@ -292,12 +339,34 @@ impl<W: Write> Writer<W> {
chunk
}

/// What each column holds over the row groups written so far, in the order of the columns.
#[must_use]
pub fn statistics(&self) -> &[ColumnStatistics] {
&self.totals
}

/// How many row groups have been written so far.
#[must_use]
pub fn row_groups(&self) -> usize {
self.groups.len()
}

/// Writes the footer and hands back what the file was written to.
///
/// # Errors
///
/// If the write fails.
pub fn finish(mut self) -> Result<W> {
pub fn finish(self) -> Result<W> {
Ok(self.finish_sized()?.0)
}

/// Writes the footer and hands back what the file was written to and how many bytes the
/// footer's metadata took, which leaves out its length and the closing magic.
///
/// # Errors
///
/// If the write fails.
pub fn finish_sized(mut self) -> Result<(W, u64)> {
let mut schema = Vec::with_capacity(self.columns.len() + 1);
let mut root = thrift::Writer::default();
root.i32(3, 0);
Expand All @@ -320,7 +389,7 @@ impl<W: Write> Writer<W> {
self.out
.flush()
.map_err(|error| Error::io(format!("could not write a parquet file: {error}")))?;
Ok(self.out)
Ok((self.out, footer.len() as u64))
}
}

Expand Down
11 changes: 7 additions & 4 deletions crates/rudb/src/database.rs
Original file line number Diff line number Diff line change
Expand Up @@ -7152,10 +7152,13 @@ impl Shared {
let zone = session.session_time_zone();
let threads = self.inner.pool.threads();
let (rows, files) = crate::export::write(&copy, &result, zone, threads)?;
if copy.partitioned.returns == Returns::Files {
QueryResult::copied_files(rows, files)
} else {
QueryResult::changed(rows)
match copy.partitioned.returns {
Returns::Count => QueryResult::changed(rows),
Returns::Files => QueryResult::copied_files(
rows,
files.into_iter().map(|file| file.path).collect(),
),
Returns::Stats => crate::export::statistics(files),
}
}
Bound::Setting(setting)
Expand Down
Loading
Loading