move TABLESAMPLE SQL surface out of core, into RelationPlanner

Background: PR apache#17843 (merged Dec 9 2025) added a `RelationPlanner`
extension API specifically motivated by TABLESAMPLE, and the blog post
"Extending SQL in DataFusion: from ->> to TABLESAMPLE" canonizes that
as the way to add SQL syntax extensions. Two prior in-core attempts
(apache#16505 Spark-style, apache#16325 random()-rewrite) closed without merging
on semantics-fragmentation grounds. alamb's stated reason for keeping
the surface out of core — sampling semantics differ widely across
DBs, plus WASM bloat — applies directly to the parsing we had baked
into `datafusion/sql/src/relation/mod.rs`.

This commit reshapes the PR around that:

- Revert the inline TABLESAMPLE parsing in `relation/mod.rs`.
- Drop the SQL planning unit tests that exercised the inline parsing.
- Replace the hacky inline `Sample` fallback in
  `DefaultPhysicalPlanner` with a public `SamplePhysicalPlanner`
  `ExtensionPlanner` that callers register alongside their
  `RelationPlanner`. Same effect, cleaner shape.
- Move the e2e fixture from a sqllogictest `.slt` (which can't
  register a `RelationPlanner`) to a Rust integration test under
  `datafusion/core/tests/parquet/tablesample.rs`. The test uses a
  small inline `RelationPlanner` to emit the core `Sample` node,
  proving the cube-root pushdown works end to end against parquet.

What stays in core (the actual contribution):

- `Sample` logical extension node and `SampleExec` placeholder.
- `SampleSpec` / `SamplePushdownResult` / `FileSourceSampleResult`
  types and the `try_push_sample` trait method on `ExecutionPlan` and
  `FileSource`.
- Per-node Passthrough overrides (filter, projection, coalesce,
  repartition, non-fetch sort).
- `ParquetSource::try_push_sample` cube-root absorption +
  `DataSourceExec::try_push_sample` rebuilding `file_groups`.
- `SamplePushdown` optimizer rule registered in the default pipeline.

These are the building blocks any SQL surface (the existing example
under `datafusion-examples/examples/relation_planner/table_sample.rs`
included) can use to push sampling into the source instead of
filtering per-batch with `random() < p`.

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

Author: adriangb

datafusion/core/src/physical_planner.rs

@@ -237,6 +237,77 @@ pub trait ExtensionPlanner {
     }
 }
 
+/// [`ExtensionPlanner`] that lowers the [`Sample`] logical extension
+/// node into a [`SampleExec`] physical node.
+///
+/// The pushdown machinery (cube-root absorption into `ParquetSource`,
+/// the `SamplePushdown` optimizer rule, per-node `Passthrough`
+/// overrides) is wired into the default optimizer pipeline, so once
+/// a `Sample` reaches the physical planner it will be pushed into
+/// the source — but it has to *get* there first. Register this
+/// planner on a [`SessionStateBuilder`] / [`DefaultPhysicalPlanner`]
+/// alongside whichever [`RelationPlanner`] (or other front-end) you
+/// use to emit the `Sample` logical node:
+///
+/// ```rust,ignore
+/// use std::sync::Arc;
+/// use datafusion::physical_planner::{DefaultPhysicalPlanner, SamplePhysicalPlanner};
+///
+/// let planner = DefaultPhysicalPlanner::with_extension_planners(vec![
+///     Arc::new(SamplePhysicalPlanner),
+/// ]);
+/// ```
+///
+/// `Sample` is intentionally not built into [`DefaultPhysicalPlanner`]
+/// by default: that would force callers who use a different SQL
+/// surface (or who don't want any TABLESAMPLE story) to carry the
+/// physical glue. The planner is small enough to opt into.
+///
+/// [`Sample`]: datafusion_expr::logical_plan::sample::Sample
+/// [`SampleExec`]: datafusion_physical_plan::sample::SampleExec
+/// [`RelationPlanner`]: datafusion_expr::planner::RelationPlanner
+/// [`SessionStateBuilder`]: crate::execution::session_state::SessionStateBuilder
+#[derive(Debug, Default)]
+pub struct SamplePhysicalPlanner;
+
+#[async_trait]
+impl ExtensionPlanner for SamplePhysicalPlanner {
+    async fn plan_extension(
+        &self,
+        _planner: &dyn PhysicalPlanner,
+        node: &dyn UserDefinedLogicalNode,
+        _logical_inputs: &[&LogicalPlan],
+        physical_inputs: &[Arc<dyn ExecutionPlan>],
+        _session_state: &SessionState,
+    ) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+        let Some(sample) = node
+            .as_any()
+            .downcast_ref::<datafusion_expr::logical_plan::sample::Sample>()
+        else {
+            return Ok(None);
+        };
+        if physical_inputs.len() != 1 {
+            return plan_err!(
+                "Sample expects exactly one input; got {}",
+                physical_inputs.len()
+            );
+        }
+        let method = match sample.method {
+            datafusion_expr::logical_plan::sample::SampleMethod::System => {
+                datafusion_physical_plan::sample_pushdown::SampleMethod::System
+            }
+        };
+        Ok(Some(Arc::new(
+            datafusion_physical_plan::sample::SampleExec::new(
+                Arc::clone(&physical_inputs[0]),
+                method,
+                sample.fraction,
+                sample.seed,
+            ),
+        )))
+    }
+}
+
 /// Default single node physical query planner that converts a
 /// `LogicalPlan` to an `ExecutionPlan` suitable for execution.
 ///
@@ -1810,35 +1881,6 @@ impl DefaultPhysicalPlanner {
                         .await?;
                 }
 
-                // Built-in fallback: the `Sample` extension node is part of
-                // core SQL (TABLESAMPLE), so plan it without requiring a
-                // user-installed ExtensionPlanner. User planners run first,
-                // so a custom override still wins.
-                if maybe_plan.is_none()
-                    && let Some(sample) = node
-                        .as_any()
-                        .downcast_ref::<datafusion_expr::logical_plan::sample::Sample>(
-                    )
-                {
-                    if children.len() != 1 {
-                        return plan_err!(
-                            "Sample expects exactly one input; got {}",
-                            children.len()
-                        );
-                    }
-                    let method = match sample.method {
-                        datafusion_expr::logical_plan::sample::SampleMethod::System =>
-                            datafusion_physical_plan::sample_pushdown::SampleMethod::System,
-                    };
-                    maybe_plan =
-                        Some(Arc::new(datafusion_physical_plan::sample::SampleExec::new(
-                            Arc::clone(&children[0]),
-                            method,
-                            sample.fraction,
-                            sample.seed,
-                        )) as Arc<dyn ExecutionPlan>);
-                }
-
                 let plan = match maybe_plan {
                     Some(v) => Ok(v),
                     _ => plan_err!(

datafusion/core/tests/parquet/mod.rs

@@ -57,6 +57,7 @@ mod page_pruning;
 mod row_group_pruning;
 mod schema;
 mod schema_coercion;
+mod tablesample;
 mod utils;
 
 #[cfg(test)]