You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@arrow.apache.org by al...@apache.org on 2023/03/17 10:51:28 UTC
[arrow-datafusion] branch main updated: fix dataframe only boolean/binary column got error on describe (#5585)
This is an automated email from the ASF dual-hosted git repository.
alamb pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 89c2d4305 fix dataframe only boolean/binary column got error on describe (#5585)
89c2d4305 is described below
commit 89c2d43054fcaf32185a5c377d68458112113c25
Author: zhenxing jiang <ji...@gmail.com>
AuthorDate: Fri Mar 17 18:51:20 2023 +0800
fix dataframe only boolean/binary column got error on describe (#5585)
---
datafusion/core/src/dataframe.rs | 185 ++++++++++++++++++-------------------
datafusion/core/tests/dataframe.rs | 35 ++++++-
2 files changed, 123 insertions(+), 97 deletions(-)
diff --git a/datafusion/core/src/dataframe.rs b/datafusion/core/src/dataframe.rs
index 1b8b742c2..e3e2987d3 100644
--- a/datafusion/core/src/dataframe.rs
+++ b/datafusion/core/src/dataframe.rs
@@ -344,91 +344,70 @@ impl DataFrame {
//collect recordBatch
let describe_record_batch = vec![
// count aggregation
- self.clone()
- .aggregate(
- vec![],
- original_schema_fields
- .clone()
- .map(|f| count(col(f.name())).alias(f.name()))
- .collect::<Vec<_>>(),
- )?
- .collect()
- .await?,
+ self.clone().aggregate(
+ vec![],
+ original_schema_fields
+ .clone()
+ .map(|f| count(col(f.name())).alias(f.name()))
+ .collect::<Vec<_>>(),
+ ),
// null_count aggregation
- self.clone()
- .aggregate(
- vec![],
- original_schema_fields
- .clone()
- .map(|f| count(is_null(col(f.name()))).alias(f.name()))
- .collect::<Vec<_>>(),
- )?
- .collect()
- .await?,
+ self.clone().aggregate(
+ vec![],
+ original_schema_fields
+ .clone()
+ .map(|f| count(is_null(col(f.name()))).alias(f.name()))
+ .collect::<Vec<_>>(),
+ ),
// mean aggregation
- self.clone()
- .aggregate(
- vec![],
- original_schema_fields
- .clone()
- .filter(|f| f.data_type().is_numeric())
- .map(|f| avg(col(f.name())).alias(f.name()))
- .collect::<Vec<_>>(),
- )?
- .collect()
- .await?,
+ self.clone().aggregate(
+ vec![],
+ original_schema_fields
+ .clone()
+ .filter(|f| f.data_type().is_numeric())
+ .map(|f| avg(col(f.name())).alias(f.name()))
+ .collect::<Vec<_>>(),
+ ),
// std aggregation
- self.clone()
- .aggregate(
- vec![],
- original_schema_fields
- .clone()
- .filter(|f| f.data_type().is_numeric())
- .map(|f| stddev(col(f.name())).alias(f.name()))
- .collect::<Vec<_>>(),
- )?
- .collect()
- .await?,
+ self.clone().aggregate(
+ vec![],
+ original_schema_fields
+ .clone()
+ .filter(|f| f.data_type().is_numeric())
+ .map(|f| stddev(col(f.name())).alias(f.name()))
+ .collect::<Vec<_>>(),
+ ),
// min aggregation
- self.clone()
- .aggregate(
- vec![],
- original_schema_fields
- .clone()
- .filter(|f| {
- !matches!(f.data_type(), DataType::Binary | DataType::Boolean)
- })
- .map(|f| min(col(f.name())).alias(f.name()))
- .collect::<Vec<_>>(),
- )?
- .collect()
- .await?,
+ self.clone().aggregate(
+ vec![],
+ original_schema_fields
+ .clone()
+ .filter(|f| {
+ !matches!(f.data_type(), DataType::Binary | DataType::Boolean)
+ })
+ .map(|f| min(col(f.name())).alias(f.name()))
+ .collect::<Vec<_>>(),
+ ),
// max aggregation
- self.clone()
- .aggregate(
- vec![],
- original_schema_fields
- .clone()
- .filter(|f| {
- !matches!(f.data_type(), DataType::Binary | DataType::Boolean)
- })
- .map(|f| max(col(f.name())).alias(f.name()))
- .collect::<Vec<_>>(),
- )?
- .collect()
- .await?,
+ self.clone().aggregate(
+ vec![],
+ original_schema_fields
+ .clone()
+ .filter(|f| {
+ !matches!(f.data_type(), DataType::Binary | DataType::Boolean)
+ })
+ .map(|f| max(col(f.name())).alias(f.name()))
+ .collect::<Vec<_>>(),
+ ),
// median aggregation
- self.clone()
- .aggregate(
- vec![],
- original_schema_fields
- .clone()
- .filter(|f| f.data_type().is_numeric())
- .map(|f| median(col(f.name())).alias(f.name()))
- .collect::<Vec<_>>(),
- )?
- .collect()
- .await?,
+ self.clone().aggregate(
+ vec![],
+ original_schema_fields
+ .clone()
+ .filter(|f| f.data_type().is_numeric())
+ .map(|f| median(col(f.name())).alias(f.name()))
+ .collect::<Vec<_>>(),
+ ),
];
// first column with function names
@@ -437,24 +416,44 @@ impl DataFrame {
))];
for field in original_schema_fields {
let mut array_datas = vec![];
- for record_batch in describe_record_batch.iter() {
- // safe unwrap since aggregate record batches should have at least 1 record
- let column = record_batch.get(0).unwrap().column_by_name(field.name());
- match column {
- Some(c) => {
- if field.data_type().is_numeric() {
- array_datas.push(cast(c, &DataType::Float64)?);
- } else {
- array_datas.push(cast(c, &DataType::Utf8)?);
+ for result in describe_record_batch.iter() {
+ let array_ref = match result {
+ Ok(df) => {
+ let batchs = df.clone().collect().await;
+ match batchs {
+ Ok(batchs)
+ if batchs.len() == 1
+ && batchs[0]
+ .column_by_name(field.name())
+ .is_some() =>
+ {
+ let column =
+ batchs[0].column_by_name(field.name()).unwrap();
+ if field.data_type().is_numeric() {
+ cast(column, &DataType::Float64)?
+ } else {
+ cast(column, &DataType::Utf8)?
+ }
+ }
+ _ => Arc::new(StringArray::from_slice(["null"])),
}
}
- //if None mean the column cannot be min/max aggregation
- None => {
- array_datas.push(Arc::new(StringArray::from_slice(["null"])));
+ //Handling error when only boolean/binary column, and in other cases
+ Err(err)
+ if err.to_string().contains(
+ "Error during planning: \
+ Aggregate requires at least one grouping \
+ or aggregate expression",
+ ) =>
+ {
+ Arc::new(StringArray::from_slice(["null"]))
}
- }
+ Err(other_err) => {
+ panic!("{other_err}")
+ }
+ };
+ array_datas.push(array_ref);
}
-
array_ref_vec.push(concat(
array_datas
.iter()
diff --git a/datafusion/core/tests/dataframe.rs b/datafusion/core/tests/dataframe.rs
index 82bd0d844..a09f458c8 100644
--- a/datafusion/core/tests/dataframe.rs
+++ b/datafusion/core/tests/dataframe.rs
@@ -39,12 +39,15 @@ use datafusion_expr::{avg, col, count, lit, max, sum, Expr, ExprSchemable};
async fn describe() -> Result<()> {
let ctx = SessionContext::new();
let testdata = datafusion::test_util::parquet_test_data();
+ ctx.register_parquet(
+ "alltypes_tiny_pages",
+ &format!("{testdata}/alltypes_tiny_pages.parquet"),
+ ParquetReadOptions::default(),
+ )
+ .await?;
let describe_record_batch = ctx
- .read_parquet(
- &format!("{testdata}/alltypes_tiny_pages.parquet"),
- ParquetReadOptions::default(),
- )
+ .table("alltypes_tiny_pages")
.await?
.describe()
.await?
@@ -67,6 +70,30 @@ async fn describe() -> Result<()> {
];
assert_batches_eq!(expected, &describe_record_batch);
+ //add test case for only boolean boolean/binary column
+ let result = ctx
+ .sql("select 'a' as a,true as b")
+ .await?
+ .describe()
+ .await?
+ .collect()
+ .await?;
+ #[rustfmt::skip]
+ let expected = vec![
+ "+------------+------+------+",
+ "| describe | a | b |",
+ "+------------+------+------+",
+ "| count | 1 | 1 |",
+ "| null_count | 1 | 1 |",
+ "| mean | null | null |",
+ "| std | null | null |",
+ "| min | a | null |",
+ "| max | a | null |",
+ "| median | null | null |",
+ "+------------+------+------+",
+ ];
+ assert_batches_eq!(expected, &result);
+
Ok(())
}