adriangb commented on code in PR #23532:
URL: https://github.com/apache/datafusion/pull/23532#discussion_r3572062748
##########
datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs:
##########
@@ -232,9 +263,45 @@ impl DynamicFilterPhysicalExpr {
/// Get the current expression.
/// This will return the current expression with any children
/// remapped to match calls to [`PhysicalExpr::with_new_children`].
+ ///
+ /// Called per batch on the RowFilter path (via
+ /// [`PhysicalExpr::evaluate`]). The remap walk is O(tree size) and, for
+ /// dynamic filters that carry a large `InListExpr` (join key IN list),
+ /// dominated by `InListExpr::with_new_children` cloning the whole list.
+ /// The inner generation only changes when [`Self::update`] fires, so we
+ /// cache the remapped expression per generation and return it directly
+ /// on subsequent per-batch calls.
pub fn current(&self) -> Result<Arc<dyn PhysicalExpr>> {
- let expr = Arc::clone(self.inner.read().expr());
- Self::remap_children(&self.children, self.remapped_children.as_ref(),
expr)
+ // Fast path: cache hit for the current generation.
+ let (expr, generation) = {
+ let inner = self.inner.read();
+ (Arc::clone(inner.expr()), inner.generation)
+ };
+ if let Some((cached_gen, cached_expr)) =
self.current_cache.read().as_ref()
+ && *cached_gen == generation
+ {
+ return Ok(Arc::clone(cached_expr));
+ }
+ // Slow path: (re)compute the remap and store it under a write lock.
+ let remapped =
+ Self::remap_children(&self.children,
self.remapped_children.as_ref(), expr)?;
+ // Only publish our result if it is at least as fresh as whatever is
+ // currently cached. Without this guard a slow computation that
+ // observed an older `inner` could clobber a newer entry that a
+ // concurrent caller has already published (see #23532 review), which
+ // would force subsequent readers to redo the remap for the newer
+ // generation.
+ {
+ let mut cache = self.current_cache.write();
+ let should_write = match cache.as_ref() {
+ Some((cached_gen, _)) => generation >= *cached_gen,
Review Comment:
Is `>=` the right choice here or would `>` result in less updates and still
be correct?
##########
datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs:
##########
@@ -232,9 +263,45 @@ impl DynamicFilterPhysicalExpr {
/// Get the current expression.
/// This will return the current expression with any children
/// remapped to match calls to [`PhysicalExpr::with_new_children`].
+ ///
+ /// Called per batch on the RowFilter path (via
+ /// [`PhysicalExpr::evaluate`]). The remap walk is O(tree size) and, for
+ /// dynamic filters that carry a large `InListExpr` (join key IN list),
+ /// dominated by `InListExpr::with_new_children` cloning the whole list.
+ /// The inner generation only changes when [`Self::update`] fires, so we
+ /// cache the remapped expression per generation and return it directly
+ /// on subsequent per-batch calls.
pub fn current(&self) -> Result<Arc<dyn PhysicalExpr>> {
- let expr = Arc::clone(self.inner.read().expr());
- Self::remap_children(&self.children, self.remapped_children.as_ref(),
expr)
+ // Fast path: cache hit for the current generation.
+ let (expr, generation) = {
+ let inner = self.inner.read();
+ (Arc::clone(inner.expr()), inner.generation)
+ };
+ if let Some((cached_gen, cached_expr)) =
self.current_cache.read().as_ref()
+ && *cached_gen == generation
+ {
+ return Ok(Arc::clone(cached_expr));
+ }
+ // Slow path: (re)compute the remap and store it under a write lock.
+ let remapped =
+ Self::remap_children(&self.children,
self.remapped_children.as_ref(), expr)?;
Review Comment:
Do we have tests that hit this under concurrency? I'm not sure how to inject
behavior to make sure we hit all concurrency scenarios. We could also make a
test that launches many threads and hot loop races them and asserts that some
property is upheld. I just feel we should derisk this looking okay under single
threaded test scenarios but then breaking under production when someone runs
with 128 partitions or whatever.
--
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]
