Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Retry to use Arc<Statistic> in PartitionedFile again #11894

Closed
wants to merge 6 commits into from
Closed

Retry to use Arc<Statistic> in PartitionedFile again #11894

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
ca0afb3
swich statistic to arc.
Rachelint Aug 8, 2024
e814d54
fix readme.
Rachelint Aug 8, 2024
67b4e49
fix pb.
Rachelint Aug 8, 2024
1af3df6
fix readme.
Rachelint Aug 8, 2024
33a7a90
fix fmt.
Rachelint Aug 8, 2024
fcbfa92
release the arc stats before executing.
Rachelint Aug 8, 2024
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
2 changes: 1 addition & 1 deletion datafusion/core/src/datasource/listing/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,7 @@ pub struct PartitionedFile {
///
/// DataFusion relies on these statistics for planning (in particular to sort file groups),
/// so if they are incorrect, incorrect answers may result.
pub statistics: Option<Statistics>,
pub statistics: Option<Arc<Statistics>>,
/// An optional field for user defined per object metadata
pub extensions: Option<Arc<dyn std::any::Any + Send + Sync>>,
}
Expand Down
4 changes: 2 additions & 2 deletions datafusion/core/src/datasource/physical_plan/file_scan_config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1193,7 +1193,7 @@ mod tests {
},
partition_values: vec![ScalarValue::from(file.date)],
range: None,
statistics: Some(Statistics {
statistics: Some(Arc::new(Statistics {
num_rows: Precision::Absent,
total_byte_size: Precision::Absent,
column_statistics: file
Expand All @@ -1213,7 +1213,7 @@ mod tests {
.unwrap_or_default()
})
.collect::<Vec<_>>(),
}),
})),
extensions: None,
}
}
Expand Down
8 changes: 7 additions & 1 deletion datafusion/core/src/datasource/physical_plan/parquet/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -354,7 +354,7 @@ impl ParquetExecBuilder {
schema_adapter_factory,
} = self;

let base_config = file_scan_config;
let mut base_config = file_scan_config;
debug!("Creating ParquetExec, files: {:?}, projection {:?}, predicate: {:?}, limit: {:?}",
base_config.file_groups, base_config.projection, predicate, base_config.limit);

Expand Down Expand Up @@ -391,6 +391,12 @@ impl ParquetExecBuilder {
&projected_output_ordering,
&base_config,
);

base_config
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

that is an interesting idea

Copy link
Contributor Author

@Rachelint Rachelint Aug 15, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

that is an interesting idea

But it still seem to make some long queries slower...

After finish the refactor work about blocked accumulators sketch. I plan to try to reduce the cost of the expansive clone scalar values like ScalarValue::String(String -> Arc<str>), and see if it can faster the shorts and don't slower the longs...

.file_groups
.iter_mut()
.for_each(|g| g.iter_mut().for_each(|f| f.statistics = None));

ParquetExec {
base_config,
projected_statistics,
Expand Down
4 changes: 2 additions & 2 deletions datafusion/core/src/datasource/statistics.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,7 @@ pub async fn get_statistics_with_limit(

if let Some(first_file) = all_files.next().await {
let (mut file, file_stats) = first_file?;
file.statistics = Some(file_stats.as_ref().clone());
file.statistics = Some(file_stats.clone());
result_files.push(file);

// First file, we set them directly from the file statistics.
Expand All @@ -83,7 +83,7 @@ pub async fn get_statistics_with_limit(
if conservative_num_rows <= limit.unwrap_or(usize::MAX) {
while let Some(current) = all_files.next().await {
let (mut file, file_stats) = current?;
file.statistics = Some(file_stats.as_ref().clone());
file.statistics = Some(file_stats.clone());
result_files.push(file);
if !collect_stats {
continue;
Expand Down
6 changes: 5 additions & 1 deletion datafusion/proto/src/physical_plan/from_proto.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -559,7 +559,11 @@ impl TryFrom<&protobuf::PartitionedFile> for PartitionedFile {
.map(|v| v.try_into())
.collect::<Result<Vec<_>, _>>()?,
range: val.range.as_ref().map(|v| v.try_into()).transpose()?,
statistics: val.statistics.as_ref().map(|v| v.try_into()).transpose()?,
statistics: val
.statistics
.as_ref()
.map(|v| v.try_into().map(|v| Arc::new(v)))
.transpose()?,
extensions: None,
})
}
Expand Down
2 changes: 1 addition & 1 deletion datafusion/proto/src/physical_plan/to_proto.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -527,7 +527,7 @@ impl TryFrom<&PartitionedFile> for protobuf::PartitionedFile {
.map(|v| v.try_into())
.collect::<Result<Vec<_>, _>>()?,
range: pf.range.as_ref().map(|r| r.try_into()).transpose()?,
statistics: pf.statistics.as_ref().map(|s| s.into()),
statistics: pf.statistics.as_ref().map(|s| s.as_ref().into()),
})
}
}
Expand Down
Loading