From 75ec7f572bbec1bc173ff4a6f31507f8bc5bff75 Mon Sep 17 00:00:00 2001 From: poonai Date: Mon, 19 Sep 2022 13:22:00 +0530 Subject: [PATCH 1/2] GroupByMap is introduced, which will use hasmap or array based on the key size. if the key size is bigger than u8 then HashMap is been used otherwise array is been used. Signed-off-by: poonai --- .../physical_plan/aggregates/groupby_map.rs | 18 ++ .../core/src/physical_plan/aggregates/mod.rs | 2 +- .../src/physical_plan/aggregates/row_hash.rs | 165 ++++++++++++++---- .../core/src/physical_plan/hash_utils.rs | 9 + datafusion/row/src/accessor.rs | 2 +- 5 files changed, 161 insertions(+), 35 deletions(-) create mode 100644 datafusion/core/src/physical_plan/aggregates/groupby_map.rs diff --git a/datafusion/core/src/physical_plan/aggregates/groupby_map.rs b/datafusion/core/src/physical_plan/aggregates/groupby_map.rs new file mode 100644 index 0000000000000..c11b1ff28e62b --- /dev/null +++ b/datafusion/core/src/physical_plan/aggregates/groupby_map.rs @@ -0,0 +1,18 @@ +use arrow::datatypes::Schema; +// Copyright 2022 Balaji (rbalajis25@gmail.com) +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +use arrow::datatypes::DataType; +use datafusion_row::layout::{RowLayout, RowType}; +use hashbrown::raw::RawTable; + diff --git a/datafusion/core/src/physical_plan/aggregates/mod.rs b/datafusion/core/src/physical_plan/aggregates/mod.rs index 88eda63edeef0..ae354e1638aa8 100644 --- a/datafusion/core/src/physical_plan/aggregates/mod.rs +++ b/datafusion/core/src/physical_plan/aggregates/mod.rs @@ -16,7 +16,6 @@ // under the License. //! Aggregates functionalities - use crate::execution::context::TaskContext; use crate::physical_plan::aggregates::hash::GroupedHashAggregateStream; use crate::physical_plan::aggregates::no_grouping::AggregateStream; @@ -43,6 +42,7 @@ use std::sync::Arc; mod hash; mod no_grouping; mod row_hash; +mod groupby_map; use crate::physical_plan::aggregates::row_hash::GroupedHashAggregateStreamV2; pub use datafusion_expr::AggregateFunction; diff --git a/datafusion/core/src/physical_plan/aggregates/row_hash.rs b/datafusion/core/src/physical_plan/aggregates/row_hash.rs index d1c61cd287f9d..925a426c2602a 100644 --- a/datafusion/core/src/physical_plan/aggregates/row_hash.rs +++ b/datafusion/core/src/physical_plan/aggregates/row_hash.rs @@ -16,7 +16,6 @@ // under the License. //! Hash aggregation through row format - use std::sync::Arc; use std::task::{Context, Poll}; use std::vec; @@ -32,10 +31,11 @@ use crate::physical_plan::aggregates::{ evaluate_group_by, evaluate_many, group_schema, AccumulatorItemV2, AggregateMode, PhysicalGroupBy, }; -use crate::physical_plan::hash_utils::create_row_hashes; +use crate::physical_plan::hash_utils::{create_row_hash, create_row_hashes}; use crate::physical_plan::metrics::{BaselineMetrics, RecordOutput}; use crate::physical_plan::{aggregates, AggregateExpr, PhysicalExpr}; use crate::physical_plan::{RecordBatchStream, SendableRecordBatchStream}; +use arrow::datatypes::DataType; use arrow::compute::cast; use arrow::datatypes::Schema; @@ -128,12 +128,12 @@ impl GroupedHashAggregateStreamV2 { input, group_by, accumulators, - group_schema, + group_schema: group_schema.clone(), aggr_schema, aggr_layout, baseline_metrics, aggregate_expressions, - aggr_state: Default::default(), + aggr_state: AggregationState::new(group_schema), random_state: Default::default(), finished: false, }) @@ -225,7 +225,7 @@ fn group_aggregate_batch( let grouping_by_values = evaluate_group_by(grouping_set, &batch)?; for group_values in grouping_by_values { - let group_rows: Vec> = create_group_rows(group_values, group_schema); + let mut group_rows: Vec> = create_group_rows(group_values, group_schema); // evaluate the aggregation expressions. // We could evaluate them after the `take`, but since we need to evaluate all @@ -243,45 +243,75 @@ fn group_aggregate_batch( let mut batch_hashes = vec![0; batch.num_rows()]; create_row_hashes(&group_rows, random_state, &mut batch_hashes)?; - for (row, hash) in batch_hashes.into_iter().enumerate() { - let AggregationState { map, group_states } = aggr_state; - - let entry = map.get_mut(hash, |(_hash, group_idx)| { - // verify that a group that we are inserting with hash is - // actually the same key value as the group in - // existing_idx (aka group_values @ row) - let group_state = &group_states[*group_idx]; - group_rows[row] == group_state.group_by_values - }); - - match entry { - // Existing entry for this group value - Some((_hash, group_idx)) => { - let group_state = &mut group_states[*group_idx]; + for (row_idx, row) in group_rows.iter_mut().enumerate() { + let AggregationState{ map, group_states} = aggr_state; + + let map_idx = map.map_idx_for_row(row); + match map.get_group_idx(row, map_idx, group_states){ + Some(group_idx) => { + let group_state = &mut group_states[group_idx]; // 1.3 if group_state.indices.is_empty() { - groups_with_rows.push(*group_idx); + groups_with_rows.push(group_idx); }; - group_state.indices.push(row as u32); // remember this row - } - // 1.2 Need to create new entry + group_state.indices.push(row_idx as u32); // remember this row + }, None => { - // Add new entry to group_states and save newly created index - let group_state = RowGroupState { - group_by_values: group_rows[row].clone(), + // Add new entry to group_states and save newly created index + let group_state = RowGroupState { + group_by_values: row.clone(), aggregation_buffer: vec![0; state_layout.fixed_part_width()], - indices: vec![row as u32], // 1.3 + indices: vec![row_idx as u32], // 1.3 }; let group_idx = group_states.len(); group_states.push(group_state); groups_with_rows.push(group_idx); // for hasher function, use precomputed hash value - map.insert(hash, (hash, group_idx), |(hash, _group_idx)| *hash); + map.update_group_idx(map_idx, group_idx); } - }; + } } + // for (row, hash) in batch_hashes.into_iter().enumerate() { + // let AggregationState { map, group_states } = aggr_state; + + // let entry = map.get_mut(hash, |(_hash, group_idx)| { + // // verify that a group that we are inserting with hash is + // // actually the same key value as the group in + // // existing_idx (aka group_values @ row) + // let group_state = &group_states[*group_idx]; + // group_rows[row] == group_state.group_by_values + // }); + + // match entry { + // // Existing entry for this group value + // Some((_hash, group_idx)) => { + // let group_state = &mut group_states[*group_idx]; + // // 1.3 + // if group_state.indices.is_empty() { + // groups_with_rows.push(*group_idx); + // }; + // group_state.indices.push(row as u32); // remember this row + // } + // // 1.2 Need to create new entry + // None => { + // // Add new entry to group_states and save newly created index + // let group_state = RowGroupState { + // group_by_values: group_rows[row].clone(), + // aggregation_buffer: vec![0; state_layout.fixed_part_width()], + // indices: vec![row as u32], // 1.3 + // }; + // let group_idx = group_states.len(); + // group_states.push(group_state); + // groups_with_rows.push(group_idx); + + // // for hasher function, use precomputed hash value + // map.insert(hash, (hash, group_idx), |(hash, _group_idx)| *hash); + // } + // }; + // } + // Collect all indices + offsets based on keys in this vec let mut batch_indices: UInt32Builder = UInt32Builder::with_capacity(0); let mut offsets = vec![0]; @@ -379,8 +409,7 @@ struct RowGroupState { indices: Vec, } -/// The state of all the groups -#[derive(Default)] + struct AggregationState { /// Logically maps group values to an index in `group_states` /// @@ -389,12 +418,19 @@ struct AggregationState { /// /// keys: u64 hashes of the GroupValue /// values: (hash, index into `group_states`) - map: RawTable<(u64, usize)>, + map: GroupByMap, /// State for each group group_states: Vec, } + +impl AggregationState { + fn new(schema: Arc) -> AggregationState { + AggregationState { map: choose_group_by_map(schema), group_states: Default::default() } + } +} + impl std::fmt::Debug for AggregationState { fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { // hashes are not store inline, so could only get values @@ -486,3 +522,66 @@ fn read_as_batch(rows: &[Vec], schema: &Schema, row_type: RowType) -> Vec; u8::MAX as usize], Arc)), + Hash((RawTable<(u64, usize)>, RandomState)), +} + +fn choose_group_by_map(schema: Arc) -> GroupByMap { + if schema.fields().len() > 1 + || !matches!( + schema.field(0).data_type(), + DataType::Boolean | DataType::UInt8 + ) + { + return GroupByMap::Hash(Default::default()); + } + GroupByMap::DirectIndexing(( + [None; u8::MAX as usize], + Arc::new(RowLayout::new(&schema, RowType::Compact)), + )) +} + +impl GroupByMap { + fn map_idx_for_row(&mut self, row: &mut Vec) -> u64 { + match self { + GroupByMap::Hash((_, random_state)) => create_row_hash(row, random_state), + GroupByMap::DirectIndexing((_, layout)) => { + let mut accessor = RowAccessor::new_from_layout(layout.clone()); + accessor.point_to(0, row.as_mut_slice()); + accessor.get_u8(0) as u64 + } + } + } + + fn get_group_idx( + &mut self, + row: &Vec, + map_idx: u64, + group_states: &mut Vec, + ) -> Option { + match self { + GroupByMap::DirectIndexing((map, _)) => { + map[map_idx as usize].map(|idx| idx as usize) + } + GroupByMap::Hash((map, _)) => { + let entry = map.get_mut(map_idx, |(_hash, group_idx)| { + *row == group_states[*group_idx].group_by_values + }); + entry.map(|(_hash, group_idx)| *group_idx) + } + } + } + + fn update_group_idx(&mut self, map_idx: u64, group_idx: usize) { + match self { + GroupByMap::DirectIndexing((map, _)) => { + map[map_idx as usize] = Some(group_idx as u8) + } + GroupByMap::Hash((map, _)) => { + map.insert(map_idx, (map_idx, group_idx), |(hash, _group_idx)| *hash); + } + } + } +} diff --git a/datafusion/core/src/physical_plan/hash_utils.rs b/datafusion/core/src/physical_plan/hash_utils.rs index 61e72b07c551f..c6a650cfdaf35 100644 --- a/datafusion/core/src/physical_plan/hash_utils.rs +++ b/datafusion/core/src/physical_plan/hash_utils.rs @@ -305,6 +305,15 @@ pub fn create_row_hashes<'a>( Ok(hashes_buffer) } +#[cfg(not(feature = "force_hash_collisions"))] +pub fn create_row_hash(row: &Vec, random_state: &RandomState) -> u64 { + random_state.hash_one(row) +} + +#[cfg(feature = "force_hash_collisions")] +pub fn create_row_hash(row: &Vec, random_state: &RandomState) -> u64 { + 0 +} /// Creates hash values for every row, based on the values in the /// columns. /// diff --git a/datafusion/row/src/accessor.rs b/datafusion/row/src/accessor.rs index f8e34578dbdac..f114207c6546e 100644 --- a/datafusion/row/src/accessor.rs +++ b/datafusion/row/src/accessor.rs @@ -178,7 +178,7 @@ impl<'a> RowAccessor<'a> { value[0] != 0 } - fn get_u8(&self, idx: usize) -> u8 { + pub fn get_u8(&self, idx: usize) -> u8 { self.assert_index_valid(idx); let offset = self.field_offsets()[idx]; self.data[self.base_offset + offset] From 28d9588a1ea39cf9c7d3f06fa4639ff8fff7a06d Mon Sep 17 00:00:00 2001 From: poonai Date: Wed, 21 Sep 2022 10:38:35 +0530 Subject: [PATCH 2/2] clean up and remove groupby_map.rs Signed-off-by: poonai --- .../physical_plan/aggregates/groupby_map.rs | 18 --- .../core/src/physical_plan/aggregates/mod.rs | 1 - .../src/physical_plan/aggregates/row_hash.rs | 132 +++++++----------- datafusion/core/tests/sql/group_by.rs | 35 +++++ datafusion/row/src/layout.rs | 1 - 5 files changed, 83 insertions(+), 104 deletions(-) delete mode 100644 datafusion/core/src/physical_plan/aggregates/groupby_map.rs diff --git a/datafusion/core/src/physical_plan/aggregates/groupby_map.rs b/datafusion/core/src/physical_plan/aggregates/groupby_map.rs deleted file mode 100644 index c11b1ff28e62b..0000000000000 --- a/datafusion/core/src/physical_plan/aggregates/groupby_map.rs +++ /dev/null @@ -1,18 +0,0 @@ -use arrow::datatypes::Schema; -// Copyright 2022 Balaji (rbalajis25@gmail.com) -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. -use arrow::datatypes::DataType; -use datafusion_row::layout::{RowLayout, RowType}; -use hashbrown::raw::RawTable; - diff --git a/datafusion/core/src/physical_plan/aggregates/mod.rs b/datafusion/core/src/physical_plan/aggregates/mod.rs index ae354e1638aa8..b6a0320c12f6a 100644 --- a/datafusion/core/src/physical_plan/aggregates/mod.rs +++ b/datafusion/core/src/physical_plan/aggregates/mod.rs @@ -42,7 +42,6 @@ use std::sync::Arc; mod hash; mod no_grouping; mod row_hash; -mod groupby_map; use crate::physical_plan::aggregates::row_hash::GroupedHashAggregateStreamV2; pub use datafusion_expr::AggregateFunction; diff --git a/datafusion/core/src/physical_plan/aggregates/row_hash.rs b/datafusion/core/src/physical_plan/aggregates/row_hash.rs index 925a426c2602a..aa0503a1bce9b 100644 --- a/datafusion/core/src/physical_plan/aggregates/row_hash.rs +++ b/datafusion/core/src/physical_plan/aggregates/row_hash.rs @@ -84,7 +84,6 @@ pub(crate) struct GroupedHashAggregateStreamV2 { aggr_layout: Arc, baseline_metrics: BaselineMetrics, - random_state: RandomState, finished: bool, } @@ -128,13 +127,12 @@ impl GroupedHashAggregateStreamV2 { input, group_by, accumulators, - group_schema: group_schema.clone(), - aggr_schema, + group_schema, + aggr_schema: aggr_schema.clone(), aggr_layout, baseline_metrics, aggregate_expressions, - aggr_state: AggregationState::new(group_schema), - random_state: Default::default(), + aggr_state: AggregationState::new(aggr_schema), finished: false, }) } @@ -160,7 +158,6 @@ impl Stream for GroupedHashAggregateStreamV2 { let timer = elapsed_compute.timer(); let result = group_aggregate_batch( &this.mode, - &this.random_state, &this.group_by, &mut this.accumulators, &this.group_schema, @@ -212,7 +209,6 @@ impl RecordBatchStream for GroupedHashAggregateStreamV2 { #[allow(clippy::too_many_arguments)] fn group_aggregate_batch( mode: &AggregateMode, - random_state: &RandomState, grouping_set: &PhysicalGroupBy, accumulators: &mut [AccumulatorItemV2], group_schema: &Schema, @@ -239,15 +235,12 @@ fn group_aggregate_batch( // track which entries in `aggr_state` have rows in this batch to aggregate let mut groups_with_rows = vec![]; - // 1.1 Calculate the group keys for the group values - let mut batch_hashes = vec![0; batch.num_rows()]; - create_row_hashes(&group_rows, random_state, &mut batch_hashes)?; - for (row_idx, row) in group_rows.iter_mut().enumerate() { - let AggregationState{ map, group_states} = aggr_state; - - let map_idx = map.map_idx_for_row(row); - match map.get_group_idx(row, map_idx, group_states){ + let AggregationState { map, group_states } = aggr_state; + + // 1.1 Calculate the group keys for the group values + let hash = map.generate_hash(row); + match map.get_group_idx(row, hash, group_states) { Some(group_idx) => { let group_state = &mut group_states[group_idx]; // 1.3 @@ -255,10 +248,10 @@ fn group_aggregate_batch( groups_with_rows.push(group_idx); }; group_state.indices.push(row_idx as u32); // remember this row - }, + } None => { - // Add new entry to group_states and save newly created index - let group_state = RowGroupState { + // 1.2 Add new entry to group_states and save newly created index + let group_state = RowGroupState { group_by_values: row.clone(), aggregation_buffer: vec![0; state_layout.fixed_part_width()], indices: vec![row_idx as u32], // 1.3 @@ -267,51 +260,12 @@ fn group_aggregate_batch( group_states.push(group_state); groups_with_rows.push(group_idx); - // for hasher function, use precomputed hash value - map.update_group_idx(map_idx, group_idx); + // store the index using computed hash. + map.store_group_idx(hash, group_idx); } } } - // for (row, hash) in batch_hashes.into_iter().enumerate() { - // let AggregationState { map, group_states } = aggr_state; - - // let entry = map.get_mut(hash, |(_hash, group_idx)| { - // // verify that a group that we are inserting with hash is - // // actually the same key value as the group in - // // existing_idx (aka group_values @ row) - // let group_state = &group_states[*group_idx]; - // group_rows[row] == group_state.group_by_values - // }); - - // match entry { - // // Existing entry for this group value - // Some((_hash, group_idx)) => { - // let group_state = &mut group_states[*group_idx]; - // // 1.3 - // if group_state.indices.is_empty() { - // groups_with_rows.push(*group_idx); - // }; - // group_state.indices.push(row as u32); // remember this row - // } - // // 1.2 Need to create new entry - // None => { - // // Add new entry to group_states and save newly created index - // let group_state = RowGroupState { - // group_by_values: group_rows[row].clone(), - // aggregation_buffer: vec![0; state_layout.fixed_part_width()], - // indices: vec![row as u32], // 1.3 - // }; - // let group_idx = group_states.len(); - // group_states.push(group_state); - // groups_with_rows.push(group_idx); - - // // for hasher function, use precomputed hash value - // map.insert(hash, (hash, group_idx), |(hash, _group_idx)| *hash); - // } - // }; - // } - // Collect all indices + offsets based on keys in this vec let mut batch_indices: UInt32Builder = UInt32Builder::with_capacity(0); let mut offsets = vec![0]; @@ -409,32 +363,27 @@ struct RowGroupState { indices: Vec, } - struct AggregationState { /// Logically maps group values to an index in `group_states` - /// - /// Uses the raw API of hashbrown to avoid actually storing the - /// keys in the table - /// - /// keys: u64 hashes of the GroupValue - /// values: (hash, index into `group_states`) map: GroupByMap, /// State for each group group_states: Vec, } - impl AggregationState { - fn new(schema: Arc) -> AggregationState { - AggregationState { map: choose_group_by_map(schema), group_states: Default::default() } + fn new(agg_schema: Arc) -> AggregationState { + AggregationState { + map: choose_group_by_map(agg_schema), + group_states: Default::default(), + } } } impl std::fmt::Debug for AggregationState { fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { // hashes are not store inline, so could only get values - let map_string = "RawTable"; + let map_string = self.map.strategy_type(); f.debug_struct("AggregationState") .field("map", &map_string) .field("group_states", &self.group_states) @@ -523,28 +472,34 @@ fn read_as_batch(rows: &[Vec], schema: &Schema, row_type: RowType) -> Vec; u8::MAX as usize], Arc)), Hash((RawTable<(u64, usize)>, RandomState)), } +/// chooses group by map based on the key length. fn choose_group_by_map(schema: Arc) -> GroupByMap { - if schema.fields().len() > 1 - || !matches!( + if schema.fields().len() == 1 + && matches!( schema.field(0).data_type(), DataType::Boolean | DataType::UInt8 ) { - return GroupByMap::Hash(Default::default()); + return GroupByMap::DirectIndexing(( + [None; u8::MAX as usize], + Arc::new(RowLayout::new(&schema, RowType::Compact)), + )); } - GroupByMap::DirectIndexing(( - [None; u8::MAX as usize], - Arc::new(RowLayout::new(&schema, RowType::Compact)), - )) + GroupByMap::Hash(Default::default()) } impl GroupByMap { - fn map_idx_for_row(&mut self, row: &mut Vec) -> u64 { + /// generates hash based on the given row. it'll be used to store or retrive + /// group index. + fn generate_hash(&mut self, row: &mut Vec) -> u64 { match self { GroupByMap::Hash((_, random_state)) => create_row_hash(row, random_state), GroupByMap::DirectIndexing((_, layout)) => { @@ -555,18 +510,19 @@ impl GroupByMap { } } + /// retrive group index for the given hash and row. fn get_group_idx( &mut self, row: &Vec, - map_idx: u64, + hash: u64, group_states: &mut Vec, ) -> Option { match self { GroupByMap::DirectIndexing((map, _)) => { - map[map_idx as usize].map(|idx| idx as usize) + map[hash as usize].map(|idx| idx as usize) } GroupByMap::Hash((map, _)) => { - let entry = map.get_mut(map_idx, |(_hash, group_idx)| { + let entry = map.get_mut(hash, |(_hash, group_idx)| { *row == group_states[*group_idx].group_by_values }); entry.map(|(_hash, group_idx)| *group_idx) @@ -574,14 +530,22 @@ impl GroupByMap { } } - fn update_group_idx(&mut self, map_idx: u64, group_idx: usize) { + /// store the group index for the given hash. + fn store_group_idx(&mut self, hash: u64, group_idx: usize) { match self { GroupByMap::DirectIndexing((map, _)) => { - map[map_idx as usize] = Some(group_idx as u8) + map[hash as usize] = Some(group_idx as u8) } GroupByMap::Hash((map, _)) => { - map.insert(map_idx, (map_idx, group_idx), |(hash, _group_idx)| *hash); + map.insert(hash, (hash, group_idx), |(hash, _group_idx)| *hash); } } } + + fn strategy_type(&self) -> &str { + match self { + GroupByMap::DirectIndexing(..) => "DirectIndexing", + GroupByMap::Hash(..) => "RawTable", + } + } } diff --git a/datafusion/core/tests/sql/group_by.rs b/datafusion/core/tests/sql/group_by.rs index 2e1007be81c9e..d1bde7e9b63aa 100644 --- a/datafusion/core/tests/sql/group_by.rs +++ b/datafusion/core/tests/sql/group_by.rs @@ -753,3 +753,38 @@ async fn csv_query_group_by_order_by_avg_group_by_substr() -> Result<()> { assert_batches_sorted_eq!(expected, &actual); Ok(()) } + +#[tokio::test] +async fn test_u8_columns() -> Result<()> { + let schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::UInt8, true), + Field::new("b", DataType::UInt8, false), + ])); + let col_a = UInt8Array::from(vec![1, 1, 2, 3, 4, 5]); + let col_b = UInt8Array::from(vec![1, 1, 1, 1, 1, 1]); + + let batch = + RecordBatch::try_new(schema.clone(), vec![Arc::new(col_a), Arc::new(col_b)]) + .unwrap(); + + let provider = MemTable::try_new(schema, vec![vec![batch]]).unwrap(); + let ctx = SessionContext::new(); + ctx.register_table("t", Arc::new(provider)).unwrap(); + let results = plan_and_collect(&ctx, "SELECT a, count(b) FROM t GROUP BY a") + .await + .expect("ran plan correctly"); + + let expected = vec![ + "+---+------------+", + "| a | COUNT(t.b) |", + "+---+------------+", + "| 1 | 2 |", + "| 2 | 1 |", + "| 3 | 1 |", + "| 4 | 1 |", + "| 5 | 1 |", + "+---+------------+", + ]; + assert_batches_sorted_eq!(expected, &results); + Ok(()) +} diff --git a/datafusion/row/src/layout.rs b/datafusion/row/src/layout.rs index 1518df9bf55a2..a14ed738bc3fa 100644 --- a/datafusion/row/src/layout.rs +++ b/datafusion/row/src/layout.rs @@ -204,7 +204,6 @@ pub fn row_supported(schema: &Schema, row_type: RowType) -> bool { fn supported_type(dt: &DataType, row_type: RowType) -> bool { use DataType::*; - match row_type { RowType::Compact => { matches!(