alamb commented on code in PR #21566:
URL: https://github.com/apache/datafusion/pull/21566#discussion_r3723589625
##########
datafusion/datasource-parquet/src/opener/mod.rs:
##########
@@ -309,6 +315,109 @@ impl fmt::Debug for ParquetMorselizer {
}
}
+/// Scan-local cache for CPU-only pruning setup that can be reused across files
+/// with the same adapted expression inputs and physical schema.
+#[derive(Debug, Default)]
+pub(super) struct ParquetPruningSetupCache {
+ entries: Mutex<ParquetPruningSetupEntries>,
+}
+
+type ParquetPruningSetupEntries =
+ HashMap<ParquetPruningSetupCacheKey, ParquetPruningSetup>;
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+struct ParquetPruningSetupCacheKey {
+ // Schema coercions such as INT96 resolution and file-schema type coercions
+ // are included through the final physical schema used for adaptation.
+ logical_file_schema: SchemaRef,
+ physical_file_schema: SchemaRef,
+ // Page-index options are intentionally not part of this key because page
+ // pruning predicates are built after this cache entry is applied.
+ predicate_ptr: Option<usize>,
+ // The projection and predicate are scan-level inputs once literal column
+ // replacement has been ruled out, so pointer identity is stable within the
+ // scan and avoids structural expression hashing.
+ projection_expr_ptrs: Vec<usize>,
+}
+
+impl ParquetPruningSetupCacheKey {
+ fn new(
+ logical_file_schema: &SchemaRef,
+ physical_file_schema: &SchemaRef,
+ projection: &ProjectionExprs,
+ predicate: Option<&Arc<dyn PhysicalExpr>>,
+ ) -> Self {
+ Self {
+ logical_file_schema: Arc::clone(logical_file_schema),
+ physical_file_schema: Arc::clone(physical_file_schema),
+ predicate_ptr: predicate.map(physical_expr_ptr),
+ projection_expr_ptrs: projection
+ .iter()
+ .map(|expr| physical_expr_ptr(&expr.expr))
+ .collect(),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+struct ParquetPruningSetup {
+ projection: ProjectionExprs,
+ predicate: Option<Arc<dyn PhysicalExpr>>,
+ pruning_predicate: Option<Arc<PruningPredicate>>,
+}
+
+fn cache_lock_poisoned(context: &str, err: impl Display) -> DataFusionError {
Review Comment:
see comment above -- I suspect we can avoid this if we use parking_lot
##########
datafusion/datasource-parquet/src/opener/mod.rs:
##########
@@ -309,6 +315,109 @@ impl fmt::Debug for ParquetMorselizer {
}
}
+/// Scan-local cache for CPU-only pruning setup that can be reused across files
+/// with the same adapted expression inputs and physical schema.
+#[derive(Debug, Default)]
Review Comment:
Since opener.rs is already quite large, what do you think about putting this
into its own module? Perhaps something like
`datafusion/datasource-parquet/src/opener/pruning_cache.rs`
##########
datafusion/datasource-parquet/src/opener/mod.rs:
##########
@@ -39,22 +39,26 @@ use crate::{
use arrow::array::RecordBatch;
use arrow::datatypes::DataType;
use datafusion_datasource::morsel::{Morsel, MorselPlan, MorselPlanner,
Morselizer};
+use datafusion_functions::core::input_file_name::InputFileNameFunc;
use datafusion_physical_expr::projection::ProjectionExprs;
use datafusion_physical_expr_adapter::replace_columns_with_literals;
-use
datafusion_physical_expr_adapter::rewrite::rewrite_input_file_name_in_projection;
+use datafusion_physical_expr_adapter::rewrite::{
+ expr_references_scalar_udf, rewrite_input_file_name_in_projection,
+};
use std::collections::{HashMap, VecDeque};
-use std::fmt;
+use std::fmt::{self, Display};
use std::future::Future;
use std::mem;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex, MutexGuard};
Review Comment:
I think elsewhere we use parking_lot::Mutex which avoids the need for
poisoning checking
##########
datafusion/datasource-parquet/src/opener/mod.rs:
##########
@@ -309,6 +315,109 @@ impl fmt::Debug for ParquetMorselizer {
}
}
+/// Scan-local cache for CPU-only pruning setup that can be reused across files
+/// with the same adapted expression inputs and physical schema.
+#[derive(Debug, Default)]
+pub(super) struct ParquetPruningSetupCache {
+ entries: Mutex<ParquetPruningSetupEntries>,
+}
+
+type ParquetPruningSetupEntries =
+ HashMap<ParquetPruningSetupCacheKey, ParquetPruningSetup>;
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+struct ParquetPruningSetupCacheKey {
+ // Schema coercions such as INT96 resolution and file-schema type coercions
+ // are included through the final physical schema used for adaptation.
+ logical_file_schema: SchemaRef,
+ physical_file_schema: SchemaRef,
+ // Page-index options are intentionally not part of this key because page
+ // pruning predicates are built after this cache entry is applied.
+ predicate_ptr: Option<usize>,
+ // The projection and predicate are scan-level inputs once literal column
+ // replacement has been ruled out, so pointer identity is stable within the
+ // scan and avoids structural expression hashing.
+ projection_expr_ptrs: Vec<usize>,
+}
+
+impl ParquetPruningSetupCacheKey {
+ fn new(
+ logical_file_schema: &SchemaRef,
+ physical_file_schema: &SchemaRef,
+ projection: &ProjectionExprs,
+ predicate: Option<&Arc<dyn PhysicalExpr>>,
+ ) -> Self {
+ Self {
+ logical_file_schema: Arc::clone(logical_file_schema),
+ physical_file_schema: Arc::clone(physical_file_schema),
+ predicate_ptr: predicate.map(physical_expr_ptr),
+ projection_expr_ptrs: projection
+ .iter()
+ .map(|expr| physical_expr_ptr(&expr.expr))
+ .collect(),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+struct ParquetPruningSetup {
+ projection: ProjectionExprs,
+ predicate: Option<Arc<dyn PhysicalExpr>>,
+ pruning_predicate: Option<Arc<PruningPredicate>>,
+}
+
+fn cache_lock_poisoned(context: &str, err: impl Display) -> DataFusionError {
+ DataFusionError::External(Box::new(std::io::Error::other(format!(
+ "{context}: {err}"
+ ))))
+}
+
+impl ParquetPruningSetupCache {
+ fn entries(&self) -> Result<MutexGuard<'_, ParquetPruningSetupEntries>> {
+ self.entries.lock().map_err(|e| {
+ cache_lock_poisoned("Parquet pruning setup cache lock poisoned", e)
+ })
+ }
+
+ fn get_or_insert_with(
+ &self,
+ key: &ParquetPruningSetupCacheKey,
+ make_setup: impl FnOnce() -> Result<ParquetPruningSetup>,
+ ) -> Result<ParquetPruningSetup> {
+ if let Some(setup) = self.entries()?.get(key) {
+ return Ok(setup.clone());
+ }
+
+ // Compute outside the cache lock. Concurrent first misses for the same
Review Comment:
👍 makes sense
##########
datafusion/datasource-parquet/src/opener/mod.rs:
##########
@@ -309,6 +315,109 @@ impl fmt::Debug for ParquetMorselizer {
}
}
+/// Scan-local cache for CPU-only pruning setup that can be reused across files
+/// with the same adapted expression inputs and physical schema.
+#[derive(Debug, Default)]
+pub(super) struct ParquetPruningSetupCache {
+ entries: Mutex<ParquetPruningSetupEntries>,
+}
+
+type ParquetPruningSetupEntries =
+ HashMap<ParquetPruningSetupCacheKey, ParquetPruningSetup>;
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+struct ParquetPruningSetupCacheKey {
+ // Schema coercions such as INT96 resolution and file-schema type coercions
+ // are included through the final physical schema used for adaptation.
+ logical_file_schema: SchemaRef,
+ physical_file_schema: SchemaRef,
+ // Page-index options are intentionally not part of this key because page
+ // pruning predicates are built after this cache entry is applied.
+ predicate_ptr: Option<usize>,
+ // The projection and predicate are scan-level inputs once literal column
+ // replacement has been ruled out, so pointer identity is stable within the
+ // scan and avoids structural expression hashing.
+ projection_expr_ptrs: Vec<usize>,
+}
+
+impl ParquetPruningSetupCacheKey {
+ fn new(
+ logical_file_schema: &SchemaRef,
+ physical_file_schema: &SchemaRef,
+ projection: &ProjectionExprs,
+ predicate: Option<&Arc<dyn PhysicalExpr>>,
+ ) -> Self {
+ Self {
+ logical_file_schema: Arc::clone(logical_file_schema),
+ physical_file_schema: Arc::clone(physical_file_schema),
+ predicate_ptr: predicate.map(physical_expr_ptr),
+ projection_expr_ptrs: projection
+ .iter()
+ .map(|expr| physical_expr_ptr(&expr.expr))
+ .collect(),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+struct ParquetPruningSetup {
+ projection: ProjectionExprs,
+ predicate: Option<Arc<dyn PhysicalExpr>>,
+ pruning_predicate: Option<Arc<PruningPredicate>>,
+}
+
+fn cache_lock_poisoned(context: &str, err: impl Display) -> DataFusionError {
+ DataFusionError::External(Box::new(std::io::Error::other(format!(
+ "{context}: {err}"
+ ))))
+}
+
+impl ParquetPruningSetupCache {
+ fn entries(&self) -> Result<MutexGuard<'_, ParquetPruningSetupEntries>> {
+ self.entries.lock().map_err(|e| {
+ cache_lock_poisoned("Parquet pruning setup cache lock poisoned", e)
+ })
+ }
+
+ fn get_or_insert_with(
+ &self,
+ key: &ParquetPruningSetupCacheKey,
+ make_setup: impl FnOnce() -> Result<ParquetPruningSetup>,
+ ) -> Result<ParquetPruningSetup> {
+ if let Some(setup) = self.entries()?.get(key) {
+ return Ok(setup.clone());
+ }
+
+ // Compute outside the cache lock. Concurrent first misses for the same
+ // key may duplicate this CPU-only setup, but the first completed
insert
+ // still makes subsequent files reuse the cached entry. Reintroduce
+ // single-flight coordination only if profiling shows duplicate setup
is
+ // material.
+ let setup = make_setup()?;
+ self.entries()?.insert(key.clone(), setup.clone());
+ Ok(setup)
+ }
+}
+
+fn physical_expr_ptr(expr: &Arc<dyn PhysicalExpr>) -> usize {
+ Arc::as_ptr(expr) as *const () as usize
+}
+
+fn is_pruning_setup_reusable(
+ projection: &ProjectionExprs,
+ predicate: Option<&Arc<dyn PhysicalExpr>>,
+ has_literal_columns: bool,
+) -> bool {
+ let has_dynamic_predicate = predicate.is_some_and(|predicate| {
+ DynamicFilterTracking::classify(predicate).contains_dynamic_filter()
+ });
+ let has_input_file_name_projection = projection
+ .iter()
+ .any(|expr|
expr_references_scalar_udf::<InputFileNameFunc>(&expr.expr));
+
+ !has_literal_columns && !has_dynamic_predicate &&
!has_input_file_name_projection
Review Comment:
why can't we reuse the setup if there are literal columns?
##########
datafusion/datasource-parquet/src/opener/mod.rs:
##########
@@ -309,6 +315,109 @@ impl fmt::Debug for ParquetMorselizer {
}
}
+/// Scan-local cache for CPU-only pruning setup that can be reused across files
+/// with the same adapted expression inputs and physical schema.
+#[derive(Debug, Default)]
+pub(super) struct ParquetPruningSetupCache {
+ entries: Mutex<ParquetPruningSetupEntries>,
+}
+
+type ParquetPruningSetupEntries =
+ HashMap<ParquetPruningSetupCacheKey, ParquetPruningSetup>;
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+struct ParquetPruningSetupCacheKey {
+ // Schema coercions such as INT96 resolution and file-schema type coercions
+ // are included through the final physical schema used for adaptation.
+ logical_file_schema: SchemaRef,
+ physical_file_schema: SchemaRef,
+ // Page-index options are intentionally not part of this key because page
+ // pruning predicates are built after this cache entry is applied.
+ predicate_ptr: Option<usize>,
+ // The projection and predicate are scan-level inputs once literal column
+ // replacement has been ruled out, so pointer identity is stable within the
+ // scan and avoids structural expression hashing.
+ projection_expr_ptrs: Vec<usize>,
+}
+
+impl ParquetPruningSetupCacheKey {
+ fn new(
+ logical_file_schema: &SchemaRef,
+ physical_file_schema: &SchemaRef,
+ projection: &ProjectionExprs,
+ predicate: Option<&Arc<dyn PhysicalExpr>>,
+ ) -> Self {
+ Self {
+ logical_file_schema: Arc::clone(logical_file_schema),
+ physical_file_schema: Arc::clone(physical_file_schema),
+ predicate_ptr: predicate.map(physical_expr_ptr),
+ projection_expr_ptrs: projection
+ .iter()
+ .map(|expr| physical_expr_ptr(&expr.expr))
+ .collect(),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+struct ParquetPruningSetup {
+ projection: ProjectionExprs,
+ predicate: Option<Arc<dyn PhysicalExpr>>,
+ pruning_predicate: Option<Arc<PruningPredicate>>,
+}
+
+fn cache_lock_poisoned(context: &str, err: impl Display) -> DataFusionError {
+ DataFusionError::External(Box::new(std::io::Error::other(format!(
+ "{context}: {err}"
+ ))))
+}
+
+impl ParquetPruningSetupCache {
+ fn entries(&self) -> Result<MutexGuard<'_, ParquetPruningSetupEntries>> {
+ self.entries.lock().map_err(|e| {
+ cache_lock_poisoned("Parquet pruning setup cache lock poisoned", e)
+ })
+ }
+
+ fn get_or_insert_with(
+ &self,
+ key: &ParquetPruningSetupCacheKey,
+ make_setup: impl FnOnce() -> Result<ParquetPruningSetup>,
+ ) -> Result<ParquetPruningSetup> {
+ if let Some(setup) = self.entries()?.get(key) {
+ return Ok(setup.clone());
+ }
+
+ // Compute outside the cache lock. Concurrent first misses for the same
+ // key may duplicate this CPU-only setup, but the first completed
insert
+ // still makes subsequent files reuse the cached entry. Reintroduce
+ // single-flight coordination only if profiling shows duplicate setup
is
+ // material.
+ let setup = make_setup()?;
+ self.entries()?.insert(key.clone(), setup.clone());
+ Ok(setup)
+ }
+}
+
+fn physical_expr_ptr(expr: &Arc<dyn PhysicalExpr>) -> usize {
+ Arc::as_ptr(expr) as *const () as usize
+}
+
+fn is_pruning_setup_reusable(
Review Comment:
stylistically, I recommend putting this as a free function
`ParquetPruningSetupCache::is_pruning_setup_reusable` so that we keep the
code/logic about predicate reuse isolated as much as possible.
However I can see the argument for keeping it outside too, so no changes
needed
##########
datafusion/datasource-parquet/src/opener/mod.rs:
##########
@@ -439,6 +548,7 @@ struct PreparedParquetOpen {
/// the logical-with-virtual schema. `None` when no virtual columns were
/// requested.
virtual_state: Option<Arc<VirtualColumnsState>>,
+ pruning_setup_reusable: bool,
Review Comment:
If we add 2 fields, there is some chance they can get out of sync (e.g.
someone might forget to check `pruning_setup_reusable` before consulting
`pruning_setup_cache`
Rather than 2 new fields I wonder if you can model this as a single Option
(set to `None` when the pruning setup is not resuable) -- that way each call
site has to explicitly check before using pruning_setup_cache 🤔
```rust
pruning_setup_cache: Option<Arc<ParquetPruningSetupCache>>,
```
##########
datafusion/datasource-parquet/src/opener/mod.rs:
##########
@@ -309,6 +315,109 @@ impl fmt::Debug for ParquetMorselizer {
}
}
+/// Scan-local cache for CPU-only pruning setup that can be reused across files
+/// with the same adapted expression inputs and physical schema.
+#[derive(Debug, Default)]
+pub(super) struct ParquetPruningSetupCache {
+ entries: Mutex<ParquetPruningSetupEntries>,
+}
+
+type ParquetPruningSetupEntries =
+ HashMap<ParquetPruningSetupCacheKey, ParquetPruningSetup>;
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+struct ParquetPruningSetupCacheKey {
+ // Schema coercions such as INT96 resolution and file-schema type coercions
+ // are included through the final physical schema used for adaptation.
+ logical_file_schema: SchemaRef,
+ physical_file_schema: SchemaRef,
+ // Page-index options are intentionally not part of this key because page
+ // pruning predicates are built after this cache entry is applied.
Review Comment:
This seems reasonable -- however, I think we could use the same cache for
page pruning predicates too (as a follow on)
##########
datafusion/physical-expr-adapter/src/schema_rewriter.rs:
##########
@@ -179,6 +179,18 @@ pub trait PhysicalExprAdapterFactory: Send + Sync +
std::fmt::Debug {
logical_file_schema: SchemaRef,
physical_file_schema: SchemaRef,
) -> Result<Arc<dyn PhysicalExprAdapter>>;
+
+ /// Return true when rewritten expressions from this factory can be reused
+ /// for the same logical schema, physical schema, and input expressions.
Review Comment:
I think it would also help to explain what effect setting this to true has.
Something like
```rust
/// for the same logical schema, physical schema, and input expressions.
///
/// If this returns true, DataFusion will attempt to cache and reuse the
result of expressions adapted
/// using the [`PhysicalExprAdapter`] returned from [`Self::create`].
Otherwise the expressions
/// will potentially be converted multiple times.
```
##########
datafusion/datasource-parquet/src/opener/mod.rs:
##########
@@ -995,64 +1114,16 @@ impl MetadataLoadedParquetOpen {
)?;
}
- // Adapt the projection & filter predicate to the physical file schema.
- // This evaluates missing columns and inserts any necessary casts.
- // After rewriting to the file schema, further simplifications may be
possible.
- // For example, if `'a' = col_that_is_missing` becomes `'a' = NULL`
that can then be simplified to `FALSE`
- // and we can avoid doing any more work on the file (bloom filters,
loading the page index, etc.).
- // Additionally, if any casts were inserted we can move casts from the
column to the literal side:
- // `CAST(col AS INT) = 5` can become `col = CAST(5 AS <col type>)`,
which can be evaluated statically.
- //
- // When the schemas are identical and there is no predicate, the
- // rewriter is a no-op: column indices already match (partition
- // columns are appended after file columns in the table schema),
- // types are the same, and there are no missing columns. Skip the
- // tree walk entirely in that case.
- let needs_rewrite = prepared.predicate.is_some()
- || prepared.logical_file_schema != physical_file_schema;
- if needs_rewrite {
- // When virtual columns are requested, augment the logical and
- // physical schemas passed to the rewriter/simplifier with those
- // fields. The rewriter identity-rewrites references found in both
- // schemas, keeping virtual-column references as `Column` rather
- // than replacing them with null literals; the simplifier needs
- // them present so it can resolve their data types while walking
- // expression trees. We keep `physical_file_schema` itself as the
- // pure file schema so downstream predicate pushdown, pruning, and
- // row filter construction stay unaffected.
- let (logical_for_rewrite, physical_for_rewrite) =
- if let Some(state) = prepared.virtual_state.as_ref() {
- (
- Arc::clone(&state.logical_schema_with_virtual),
- append_fields(&physical_file_schema,
&state.virtual_columns),
- )
- } else {
- (
- Arc::clone(&prepared.logical_file_schema),
- Arc::clone(&physical_file_schema),
- )
- };
- let rewriter = prepared.expr_adapter_factory.create(
- Arc::clone(&logical_for_rewrite),
- Arc::clone(&physical_for_rewrite),
- )?;
- let simplifier =
PhysicalExprSimplifier::new(&physical_for_rewrite);
- prepared.predicate = prepared
- .predicate
- .map(|p| simplifier.simplify(rewriter.rewrite(p)?))
- .transpose()?;
- prepared.projection = prepared
- .projection
- .try_map_exprs(|p| simplifier.simplify(rewriter.rewrite(p)?))?;
- }
- prepared.physical_file_schema = Arc::clone(&physical_file_schema);
+ let pruning_setup = build_or_get_pruning_setup(&prepared,
&physical_file_schema)?;
Review Comment:
Would it make sense to move this logic on methods of `PreparedOpen` ?
Right now this code seems like it is looking at fields of `prepared` and
then updating fields on `prepared` -- if we made it a method on `PreparedOpen`
I think that would more naturally connect the code and the state that used it
##########
datafusion/datasource-parquet/src/opener/mod.rs:
##########
@@ -788,7 +899,13 @@ impl ParquetMorselizer {
let mut projection = self.projection.clone();
let mut predicate = self.predicate.clone();
- if !literal_columns.is_empty() {
+ let has_literal_columns = !literal_columns.is_empty();
Review Comment:
I think it would help here to explain the rationale for the checks in
`is_pruning_setup_reusable` -- for example it is not at all clear to me why we
can't cache literal columns
It may also be simpler to pass in `literal_columns` to
`is_pruning_setup_reusable` so you document all the requirements on reusable
setups in a single location
##########
datafusion/datasource-parquet/src/opener/mod.rs:
##########
@@ -2247,6 +2472,229 @@ mod test {
))
}
+ #[tokio::test]
+ async fn test_pruning_setup_cache_reuses_adapter_for_same_schema() {
+ let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
+ let table_schema =
+ Arc::new(Schema::new(vec![Field::new("a", DataType::Int64,
false)]));
+
+ let batch1 =
+ record_batch!(("a", Int32, vec![Some(1), Some(2),
Some(3)])).unwrap();
+ let batch2 =
+ record_batch!(("a", Int32, vec![Some(4), Some(5),
Some(6)])).unwrap();
+ let data_size1 = write_parquet(Arc::clone(&store), "file1.parquet",
batch1).await;
+ let data_size2 = write_parquet(Arc::clone(&store), "file2.parquet",
batch2).await;
+
+ let create_count = Arc::new(AtomicUsize::new(0));
+ let factory: Arc<dyn PhysicalExprAdapterFactory> = Arc::new(
+ CountingPhysicalExprAdapterFactory::new(Arc::clone(&create_count),
true),
+ );
+ let predicate = logical2physical(&col("a").gt(lit(0i64)),
&table_schema);
+
+ let morselizer = ParquetMorselizerBuilder::new()
+ .with_store(Arc::clone(&store))
+ .with_schema(table_schema)
+ .with_projection_indices(&[0])
+ .with_predicate(predicate)
+ .with_expr_adapter_factory(factory)
+ .build();
+
+ open_files_and_assert_row_count(
+ &morselizer,
+ [
+ PartitionedFile::new("file1.parquet",
u64::try_from(data_size1).unwrap()),
+ PartitionedFile::new("file2.parquet",
u64::try_from(data_size2).unwrap()),
+ ],
+ 3,
+ )
+ .await;
+
+ assert_eq!(
+ create_count.load(Ordering::SeqCst),
+ 1,
+ "same-schema files should reuse the cached pruning setup"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_pruning_setup_cache_skips_non_reusable_adapter() {
+ let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
+ let table_schema =
+ Arc::new(Schema::new(vec![Field::new("a", DataType::Int64,
false)]));
+
+ let batch1 =
Review Comment:
there is quite a bit of repeated setup code in these tests(e.g. the store,
table schema, batches) and it is hard for me to understand what is changing
from test to test
Could we perhaps factor it into a shared fixture to make it easer to read
the tests and their verification? So then it would be easier to see what
changes between each test and what is covered
##########
datafusion/physical-expr-adapter/src/schema_rewriter.rs:
##########
@@ -179,6 +179,18 @@ pub trait PhysicalExprAdapterFactory: Send + Sync +
std::fmt::Debug {
logical_file_schema: SchemaRef,
physical_file_schema: SchemaRef,
) -> Result<Arc<dyn PhysicalExprAdapter>>;
+
+ /// Return true when rewritten expressions from this factory can be reused
+ /// for the same logical schema, physical schema, and input expressions.
+ ///
+ /// Factories that opt in must not depend on factory-local mutable state or
+ /// other per-file inputs that are not represented by those rewrite inputs.
+ ///
+ /// Custom factories default to non-reusable because they may depend on
Review Comment:
I suppose this makes sense as an escape valve -- but as written this will
mean that anyone who implements PhysicalExprAdapterFactory will not get setup
pruning without overriding this. That might be good.
##########
datafusion/physical-expr-adapter/src/schema_rewriter.rs:
##########
@@ -195,6 +207,10 @@ impl PhysicalExprAdapterFactory for
DefaultPhysicalExprAdapterFactory {
physical_file_schema,
}))
}
+
+ fn supports_reusable_rewrites(&self) -> bool {
Review Comment:
maybe a comment saying this is save as the default rewriter has no state
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
