fix: accumulate planner observations instead of overwriting them

foreign_session, foreign_provider, and foreign_plan were written with
store, so each one described only the most recent plan. Their accessors
are named foreign_*_observed, which asks whether the thing was ever
seen, and the tests assert them after running more than one query. The
existing tests passed by luck.

Reproduced: after scanning a foreign provider and then running
SELECT 1, foreign_provider_observed goes from True back to False.

Writes them with fetch_or so a later plan cannot retract what an earlier
one observed. plan_calls already accumulated, used_fallback only ever
stores true so it was already cumulative, and last_max_rows is
deliberately last-wins as its name says. Documents that split on the
struct, since it is the kind of thing that gets "tidied" back.

Confirmed non-vacuous: with store restored, exactly the new test fails
and the other 22 pass.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: timsaucer

examples/datafusion-ffi-query-planner-example/python/tests/_test_three_library_query_planner.py

@@ -398,6 +398,36 @@ def test_a_session_fallback_delegates_to_its_installed_planner():
     assert first.plan_calls() > 0
 
 
+def test_observations_accumulate_across_queries():
+    """A later plain query must not retract what an earlier query observed.
+
+    The ``*_observed`` accessors answer "was this ever seen". They are written
+    with ``fetch_or`` rather than ``store`` so a query that touches no foreign
+    object cannot clear a flag an earlier one set. Written with ``store``,
+    ``SELECT 1`` here clears ``foreign_provider_observed``, and every other
+    test asserting these flags after more than one query is a coincidence away
+    from failing.
+    """
+    ctx, _logical_codec, _physical_codec = configured_context(max_rows=3)
+    planner = MyQueryPlanner()
+    ctx = ctx.with_query_planner(planner)
+
+    ctx.sql('SELECT "A" FROM numbers ORDER BY "A"').collect()
+    assert planner.foreign_session_observed()
+    assert planner.foreign_provider_observed()
+    assert planner.foreign_plan_observed()
+
+    # Touches no table, so this plan has no foreign provider of its own.
+    ctx.sql("SELECT 1").collect()
+    assert planner.foreign_session_observed()
+    assert planner.foreign_provider_observed()
+    assert planner.foreign_plan_observed()
+
+    # `last_max_rows` is deliberately not cumulative; it reports the last plan.
+    assert planner.last_max_rows() == 3
+    assert planner.plan_calls() >= 2
+
+
 def test_second_planner_replaces_the_first():
     """A session holds exactly one planner, so installing another replaces it."""
     ctx, _logical_codec, _physical_codec = configured_context(max_rows=2)

examples/datafusion-ffi-query-planner-example/src/planner.rs

@@ -41,13 +41,24 @@ use pyo3::types::PyCapsule;
 
 use crate::config::MyPlannerConfig;
 
+/// What the planner saw, accumulated across every call rather than reset each
+/// time.
+///
+/// Two kinds of field here, and mixing them up is easy. `last_max_rows`
+/// reports the most recent value, as its name says. Everything else is
+/// cumulative: a count, or a "did this ever happen" flag written with
+/// `fetch_or` so a later plan cannot retract an earlier observation. Tests
+/// assert after running more than one query, so a flag that only described the
+/// most recent plan would be answering a different question than the one its
+/// accessor name asks.
 #[derive(Default)]
 struct PlannerObservations {
     plan_calls: AtomicUsize,
     last_max_rows: AtomicUsize,
     foreign_session: AtomicBool,
     foreign_provider: AtomicBool,
     foreign_plan: AtomicBool,
+    /// Only ever set to `true`, so it is already cumulative.
     used_fallback: AtomicBool,
 }
 
@@ -156,10 +167,14 @@ impl QueryPlanner for DistributedQueryPlanner {
         session: &dyn Session,
     ) -> datafusion::common::Result<Arc<dyn ExecutionPlan>> {
         self.observations.plan_calls.fetch_add(1, Ordering::SeqCst);
+        // `fetch_or`, not `store`: these answer "was this ever seen", so a
+        // later plan that happens not to touch a foreign object must not
+        // retract what an earlier one observed. A bare `SELECT 1` after a
+        // scan of a foreign provider would otherwise clear the flag.
         self.observations
             .foreign_session
-            .store(session.as_any().is::<ForeignSession>(), Ordering::SeqCst);
-        self.observations.foreign_provider.store(
+            .fetch_or(session.as_any().is::<ForeignSession>(), Ordering::SeqCst);
+        self.observations.foreign_provider.fetch_or(
             logical_plan_has_foreign_provider(logical_plan),
             Ordering::SeqCst,
         );
@@ -184,7 +199,7 @@ impl QueryPlanner for DistributedQueryPlanner {
         };
         self.observations
             .foreign_plan
-            .store(physical_plan_has_foreign_plan(&plan), Ordering::SeqCst);
+            .fetch_or(physical_plan_has_foreign_plan(&plan), Ordering::SeqCst);
 
         Ok(Arc::new(GlobalLimitExec::new(
             plan,