Address Copilot review: guard cache write against older-generation clobber

A slow `current()` call that observed an older `inner` generation could
finish after a concurrent caller has published a newer entry, and its
unconditional write would replace that newer entry with a stale one.
Semantics were still correct (later readers see the mismatch and
recompute), but subsequent generation-matched calls would miss the cache
and redo the remap unnecessarily.

Guard the write: only publish if the observed generation is at least
as fresh as whatever is currently cached.

Author: zhuqi-lucas

datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs

@@ -269,10 +269,22 @@ impl DynamicFilterPhysicalExpr {
         // 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)?;
-        // A concurrent thread may have populated the cache with the same
-        // generation; either wins the write is fine — both are semantically
-        // equivalent because the inner generation hasn't changed.
-        *self.current_cache.write() = Some((generation, Arc::clone(&remapped)));
+        // 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,
+                None => true,
+            };
+            if should_write {
+                *cache = Some((generation, Arc::clone(&remapped)));
+            }
+        }
         Ok(remapped)
     }