berkaysynnada commented on code in PR #15503: URL: https://github.com/apache/datafusion/pull/15503#discussion_r2030225208
########## datafusion/core/tests/physical_optimizer/partition_statistics.rs: ########## @@ -0,0 +1,244 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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. + +#[cfg(test)] +mod test { + use arrow_schema::{DataType, Field, Schema, SortOptions}; + use datafusion::datasource::listing::ListingTable; + use datafusion::prelude::SessionContext; + use datafusion_catalog::TableProvider; + use datafusion_common::stats::Precision; + use datafusion_common::{ColumnStatistics, ScalarValue, Statistics}; + use datafusion_execution::config::SessionConfig; + use datafusion_expr_common::operator::Operator; + use datafusion_physical_expr::expressions::{binary, lit, Column}; + use datafusion_physical_expr_common::physical_expr::PhysicalExpr; + use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr}; + use datafusion_physical_plan::filter::FilterExec; + use datafusion_physical_plan::projection::ProjectionExec; + use datafusion_physical_plan::sorts::sort::SortExec; + use datafusion_physical_plan::union::UnionExec; + use datafusion_physical_plan::ExecutionPlan; + use std::sync::Arc; + + /// Creates a test table with statistics from the test data directory. + /// + /// This function: + /// - Creates an external table from './tests/data/test_statistics_per_partition' + /// - If we set the `target_partition` to `2, the data contains 2 partitions, each with 2 rows Review Comment: ```suggestion /// - If we set the `target_partition` to 2, the data contains 2 partitions, each with 2 rows ``` ########## datafusion/physical-plan/src/execution_plan.rs: ########## @@ -427,6 +427,16 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync { Ok(Statistics::new_unknown(&self.schema())) } + /// Returns statistics for each partition of this `ExecutionPlan` node. + /// If statistics are not available, returns an array of + /// [`Statistics::new_unknown`] for each partition. + fn statistics_by_partition(&self) -> Result<Vec<Statistics>> { Review Comment: Another alternative is completely separating table statistics from partition statistics, if their methods are like more distinct, and not used commonly very often ########## datafusion/datasource/src/source.rs: ########## @@ -175,6 +175,23 @@ impl ExecutionPlan for DataSourceExec { self.data_source.statistics() } + fn statistics_by_partition(&self) -> datafusion_common::Result<Vec<Statistics>> { + let mut statistics = vec![ + Statistics::new_unknown(&self.schema()); + self.properties().partitioning.partition_count() + ]; + if let Some(file_config) = + self.data_source.as_any().downcast_ref::<FileScanConfig>() + { + for (idx, file_group) in file_config.file_groups.iter().enumerate() { + if let Some(stat) = file_group.statistics() { + statistics[idx] = stat.clone(); Review Comment: I believe we should fix this as soon as possible, since the Statistics part of the codebase is under heavy focus at these moments by many people, and in the near future, I expect that we will have many Statistics related PR's and developments. So, to not bring an inherent regression, we should bring the infra to a safely extensible state. ########## datafusion/datasource/src/source.rs: ########## @@ -175,6 +175,23 @@ impl ExecutionPlan for DataSourceExec { self.data_source.statistics() } + fn statistics_by_partition(&self) -> datafusion_common::Result<Vec<Statistics>> { + let mut statistics = vec![ + Statistics::new_unknown(&self.schema()); + self.properties().partitioning.partition_count() + ]; + if let Some(file_config) = + self.data_source.as_any().downcast_ref::<FileScanConfig>() + { + for (idx, file_group) in file_config.file_groups.iter().enumerate() { + if let Some(stat) = file_group.statistics() { + statistics[idx] = stat.clone(); Review Comment: Wrapping the Statistics with `Arc<>'`s can be a solution? Some structs like `FileGroup`, `PartitionedData` etc. caches the Statistics. So, if the source operators can access those, they should return over them. However, for other intermediate operators, perhaps we can utilize `PlanProperties`? The Statistics will be initiated once and cached like other planning properties ########## datafusion/physical-plan/src/execution_plan.rs: ########## @@ -427,6 +427,16 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync { Ok(Statistics::new_unknown(&self.schema())) } + /// Returns statistics for each partition of this `ExecutionPlan` node. + /// If statistics are not available, returns an array of + /// [`Statistics::new_unknown`] for each partition. + fn statistics_by_partition(&self) -> Result<Vec<Statistics>> { Review Comment: I'm still on the side of unifying these two API's. Maybe you can have a better proposal, but those are what is in my mind now: If we don't prefer dealing with new enum's or structs for Statistics, we can modify the API such: ```rust fn statistics(&self, partition: Option<usize>) -> Result<Statistics> ``` This would give the option of not implementing all operators and express clearly what we have, and get all stuff closer. If we prefer using the same API, but enriching the `Statistics` definition, then: 1) Rename `Statistics` as `TableStatistics` in the statistics() API. 2) ```rust struct TableStatistics { first_partition: Statistics others: Vec<Statistics> } ``` This option requires some methods for ease of use for TableStatistics as you imagine ########## datafusion/core/tests/physical_optimizer/partition_statistics.rs: ########## @@ -0,0 +1,317 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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. + +#[cfg(test)] +mod test { + use arrow_schema::{DataType, Field, Schema, SortOptions}; + use datafusion::datasource::listing::ListingTable; + use datafusion::prelude::SessionContext; + use datafusion_catalog::TableProvider; + use datafusion_common::stats::Precision; + use datafusion_common::{ScalarValue, Statistics}; + use datafusion_execution::config::SessionConfig; + use datafusion_expr_common::operator::Operator; + use datafusion_physical_expr::expressions::{binary, lit, Column}; + use datafusion_physical_expr_common::physical_expr::PhysicalExpr; + use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr}; + use datafusion_physical_plan::filter::FilterExec; + use datafusion_physical_plan::projection::ProjectionExec; + use datafusion_physical_plan::sorts::sort::SortExec; + use datafusion_physical_plan::union::UnionExec; + use datafusion_physical_plan::ExecutionPlan; + use std::sync::Arc; + + async fn generate_listing_table_with_statistics( Review Comment: This function returns an ExecutionPlan. This name is a bit misleading I think -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: github-unsubscr...@datafusion.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: github-unsubscr...@datafusion.apache.org For additional commands, e-mail: github-h...@datafusion.apache.org
