diff --git a/Cargo.lock b/Cargo.lock
index acc8b7aa7fe4..0b36b1283553 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2200,10 +2200,14 @@ dependencies = [
  "bytes",
  "dashmap",
  "datafusion",
+ "datafusion-common",
+ "datafusion-expr",
  "datafusion-physical-expr-adapter",
  "datafusion-proto",
+ "datafusion-sql",
  "env_logger",
  "futures",
+ "insta",
  "log",
  "mimalloc",
  "nix",
diff --git a/datafusion-examples/Cargo.toml b/datafusion-examples/Cargo.toml
index 57b6af6b52f8..e445184291eb 100644
--- a/datafusion-examples/Cargo.toml
+++ b/datafusion-examples/Cargo.toml
@@ -46,10 +46,14 @@ dashmap = { workspace = true }
 # note only use main datafusion crate for examples
 base64 = "0.22.1"
 datafusion = { workspace = true, default-features = true, features = ["parquet_encryption"] }
+datafusion-common = { workspace = true }
+datafusion-expr = { workspace = true }
 datafusion-physical-expr-adapter = { workspace = true }
 datafusion-proto = { workspace = true }
+datafusion-sql = { workspace = true }
 env_logger = { workspace = true }
 futures = { workspace = true }
+insta = { workspace = true }
 log = { workspace = true }
 mimalloc = { version = "0.1", default-features = false }
 object_store = { workspace = true, features = ["aws", "http"] }
diff --git a/datafusion-examples/README.md b/datafusion-examples/README.md
index 0bf313ab755d..a8611c8c3460 100644
--- a/datafusion-examples/README.md
+++ b/datafusion-examples/README.md
@@ -86,6 +86,9 @@ cargo run --example dataframe -- dataframe
 - [`examples/external_dependency/query_aws_s3.rs`](examples/external_dependency/query_aws_s3.rs): Configure `object_store` and run a query against files stored in AWS S3
 - [`examples/data_io/query_http_csv.rs`](examples/data_io/query_http_csv.rs): Configure `object_store` and run a query against files via HTTP
 - [`examples/builtin_functions/regexp.rs`](examples/builtin_functions/regexp.rs): Examples of using regular expression functions
+- [`examples/relation_planner/match_recognize.rs`](examples/relation_planner/match_recognize.rs): Use custom relation planner to implement MATCH_RECOGNIZE pattern matching
+- [`examples/relation_planner/pivot_unpivot.rs`](examples/relation_planner/pivot_unpivot.rs): Use custom relation planner to implement PIVOT and UNPIVOT operations
+- [`examples/relation_planner/table_sample.rs`](examples/relation_planner/table_sample.rs): Use custom relation planner to implement TABLESAMPLE clause
 - [`examples/data_io/remote_catalog.rs`](examples/data_io/remote_catalog.rs): Examples of interfacing with a remote catalog (e.g. over a network)
 - [`examples/udf/simple_udaf.rs`](examples/udf/simple_udaf.rs): Define and invoke a User Defined Aggregate Function (UDAF)
 - [`examples/udf/simple_udf.rs`](examples/udf/simple_udf.rs): Define and invoke a User Defined Scalar Function (UDF)
diff --git a/datafusion-examples/examples/relation_planner/main.rs b/datafusion-examples/examples/relation_planner/main.rs
new file mode 100644
index 000000000000..ff9c93de7b69
--- /dev/null
+++ b/datafusion-examples/examples/relation_planner/main.rs
@@ -0,0 +1,141 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! # Relation Planner Examples
+//!
+//! These examples demonstrate how to use custom relation planners to extend
+//! DataFusion's SQL syntax with custom table operators.
+//!
+//! ## Usage
+//! ```bash
+//! cargo run --example relation_planner -- [match_recognize|pivot_unpivot|table_sample]
+//! ```
+//!
+//! Each subcommand runs a corresponding example:
+//! - `match_recognize` — MATCH_RECOGNIZE pattern matching on event streams
+//! - `pivot_unpivot` — PIVOT and UNPIVOT operations for reshaping data
+//! - `table_sample` — TABLESAMPLE clause for sampling rows from tables
+//!
+//! ## Snapshot Testing
+//!
+//! These examples use [insta](https://insta.rs) for inline snapshot assertions.
+//! If query output changes, regenerate the snapshots with:
+//! ```bash
+//! cargo insta test --example relation_planner --accept
+//! ```
+
+mod match_recognize;
+mod pivot_unpivot;
+mod table_sample;
+
+use std::str::FromStr;
+
+use datafusion::error::{DataFusionError, Result};
+
+enum ExampleKind {
+    MatchRecognize,
+    PivotUnpivot,
+    TableSample,
+}
+
+impl AsRef<str> for ExampleKind {
+    fn as_ref(&self) -> &str {
+        match self {
+            Self::MatchRecognize => "match_recognize",
+            Self::PivotUnpivot => "pivot_unpivot",
+            Self::TableSample => "table_sample",
+        }
+    }
+}
+
+impl FromStr for ExampleKind {
+    type Err = DataFusionError;
+
+    fn from_str(s: &str) -> Result<Self> {
+        match s {
+            "match_recognize" => Ok(Self::MatchRecognize),
+            "pivot_unpivot" => Ok(Self::PivotUnpivot),
+            "table_sample" => Ok(Self::TableSample),
+            _ => Err(DataFusionError::Execution(format!("Unknown example: {s}"))),
+        }
+    }
+}
+
+impl ExampleKind {
+    const ALL: [Self; 3] = [Self::MatchRecognize, Self::PivotUnpivot, Self::TableSample];
+
+    const EXAMPLE_NAME: &str = "relation_planner";
+
+    fn variants() -> Vec<&'static str> {
+        Self::ALL.iter().map(|x| x.as_ref()).collect()
+    }
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let usage = format!(
+        "Usage: cargo run --example {} -- [{}]",
+        ExampleKind::EXAMPLE_NAME,
+        ExampleKind::variants().join("|")
+    );
+
+    let arg = std::env::args().nth(1).ok_or_else(|| {
+        eprintln!("{usage}");
+        DataFusionError::Execution("Missing argument".to_string())
+    })?;
+
+    if arg == "all" {
+        for example in ExampleKind::ALL {
+            match example {
+                ExampleKind::MatchRecognize => match_recognize::match_recognize().await?,
+                ExampleKind::PivotUnpivot => pivot_unpivot::pivot_unpivot().await?,
+                ExampleKind::TableSample => table_sample::table_sample().await?,
+            }
+        }
+    } else {
+        match arg.parse::<ExampleKind>()? {
+            ExampleKind::MatchRecognize => match_recognize::match_recognize().await?,
+            ExampleKind::PivotUnpivot => pivot_unpivot::pivot_unpivot().await?,
+            ExampleKind::TableSample => table_sample::table_sample().await?,
+        }
+    }
+
+    Ok(())
+}
+
+/// Test wrappers that enable `cargo insta test --example relation_planner --accept`
+/// to regenerate inline snapshots. Without these, insta cannot run the examples
+/// in test mode since they only have `main()` functions.
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn test_match_recognize() {
+        match_recognize::match_recognize().await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_pivot_unpivot() {
+        pivot_unpivot::pivot_unpivot().await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_table_sample() {
+        table_sample::table_sample().await.unwrap();
+    }
+}
diff --git a/datafusion-examples/examples/relation_planner/match_recognize.rs b/datafusion-examples/examples/relation_planner/match_recognize.rs
new file mode 100644
index 000000000000..7eeddb1f9742
--- /dev/null
+++ b/datafusion-examples/examples/relation_planner/match_recognize.rs
@@ -0,0 +1,406 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! # MATCH_RECOGNIZE Example
+//!
+//! This example demonstrates implementing SQL `MATCH_RECOGNIZE` pattern matching
+//! using a custom [`RelationPlanner`]. Unlike the [`pivot_unpivot`] example that
+//! rewrites SQL to standard operations, this example creates a **custom logical
+//! plan node** (`MiniMatchRecognizeNode`) to represent the operation.
+//!
+//! ## Supported Syntax
+//!
+//! ```sql
+//! SELECT * FROM events
+//!   MATCH_RECOGNIZE (
+//!     PARTITION BY region
+//!     MEASURES SUM(price) AS total, AVG(price) AS average
+//!     PATTERN (A B+ C)
+//!     DEFINE
+//!       A AS price < 100,
+//!       B AS price BETWEEN 100 AND 200,
+//!       C AS price > 200
+//!   ) AS matches
+//! ```
+//!
+//! ## Architecture
+//!
+//! This example demonstrates **logical planning only**. Physical execution would
+//! require implementing an [`ExecutionPlan`] (see the [`table_sample`] example
+//! for a complete implementation with physical planning).
+//!
+//! ```text
+//! SQL Query
+//!     │
+//!     ▼
+//! ┌─────────────────────────────────────┐
+//! │ MatchRecognizePlanner               │
+//! │ (RelationPlanner trait)             │
+//! │                                     │
+//! │ • Parses MATCH_RECOGNIZE syntax     │
+//! │ • Creates MiniMatchRecognizeNode    │
+//! │ • Converts SQL exprs to DataFusion  │
+//! └─────────────────────────────────────┘
+//!     │
+//!     ▼
+//! ┌─────────────────────────────────────┐
+//! │ MiniMatchRecognizeNode              │
+//! │ (UserDefinedLogicalNode)            │
+//! │                                     │
+//! │ • measures: [(alias, expr), ...]    │
+//! │ • definitions: [(symbol, expr), ...]│
+//! └─────────────────────────────────────┘
+//! ```
+//!
+//! [`pivot_unpivot`]: super::pivot_unpivot
+//! [`table_sample`]: super::table_sample
+//! [`ExecutionPlan`]: datafusion::physical_plan::ExecutionPlan
+
+use std::{any::Any, cmp::Ordering, hash::Hasher, sync::Arc};
+
+use arrow::array::{ArrayRef, Float64Array, Int32Array, StringArray};
+use arrow::record_batch::RecordBatch;
+use datafusion::prelude::*;
+use datafusion_common::{DFSchemaRef, Result};
+use datafusion_expr::{
+    logical_plan::{Extension, InvariantLevel, LogicalPlan},
+    planner::{
+        PlannedRelation, RelationPlanner, RelationPlannerContext, RelationPlanning,
+    },
+    Expr, UserDefinedLogicalNode,
+};
+use datafusion_sql::sqlparser::ast::TableFactor;
+use insta::assert_snapshot;
+
+// ============================================================================
+// Example Entry Point
+// ============================================================================
+
+/// Runs the MATCH_RECOGNIZE examples demonstrating pattern matching on event streams.
+///
+/// Note: This example demonstrates **logical planning only**. Physical execution
+/// would require additional implementation of an [`ExecutionPlan`].
+pub async fn match_recognize() -> Result<()> {
+    let ctx = SessionContext::new();
+    ctx.register_relation_planner(Arc::new(MatchRecognizePlanner))?;
+    register_sample_data(&ctx)?;
+
+    println!("MATCH_RECOGNIZE Example (Logical Planning Only)");
+    println!("================================================\n");
+
+    run_examples(&ctx).await
+}
+
+async fn run_examples(ctx: &SessionContext) -> Result<()> {
+    // Example 1: Basic MATCH_RECOGNIZE with MEASURES and DEFINE
+    // Demonstrates: Aggregate measures over matched rows
+    let plan = run_example(
+        ctx,
+        "Example 1: MATCH_RECOGNIZE with aggregations",
+        r#"SELECT * FROM events
+           MATCH_RECOGNIZE (
+             PARTITION BY 1
+             MEASURES SUM(price) AS total_price, AVG(price) AS avg_price
+             PATTERN (A)
+             DEFINE A AS price > 10
+           ) AS matches"#,
+    )
+    .await?;
+    assert_snapshot!(plan, @r"
+    Projection: matches.price
+      SubqueryAlias: matches
+        MiniMatchRecognize measures=[total_price := sum(events.price), avg_price := avg(events.price)] define=[a := events.price > Int64(10)]
+          TableScan: events
+    ");
+
+    // Example 2: Stock price pattern detection
+    // Demonstrates: Real-world use case finding prices above threshold
+    let plan = run_example(
+        ctx,
+        "Example 2: Detect high stock prices",
+        r#"SELECT * FROM stock_prices
+           MATCH_RECOGNIZE (
+             MEASURES
+               MIN(price) AS min_price,
+               MAX(price) AS max_price,
+               AVG(price) AS avg_price
+             PATTERN (HIGH)
+             DEFINE HIGH AS price > 151.0
+           ) AS trends"#,
+    )
+    .await?;
+    assert_snapshot!(plan, @r"
+    Projection: trends.symbol, trends.price
+      SubqueryAlias: trends
+        MiniMatchRecognize measures=[min_price := min(stock_prices.price), max_price := max(stock_prices.price), avg_price := avg(stock_prices.price)] define=[high := stock_prices.price > Float64(151)]
+          TableScan: stock_prices
+    ");
+
+    Ok(())
+}
+
+/// Helper to run a single example query and display the logical plan.
+async fn run_example(ctx: &SessionContext, title: &str, sql: &str) -> Result<String> {
+    println!("{title}:\n{sql}\n");
+    let plan = ctx.sql(sql).await?.into_unoptimized_plan();
+    let plan_str = plan.display_indent().to_string();
+    println!("{plan_str}\n");
+    Ok(plan_str)
+}
+
+/// Register test data tables.
+fn register_sample_data(ctx: &SessionContext) -> Result<()> {
+    // events: simple price series
+    ctx.register_batch(
+        "events",
+        RecordBatch::try_from_iter(vec![(
+            "price",
+            Arc::new(Int32Array::from(vec![5, 12, 8, 15, 20])) as ArrayRef,
+        )])?,
+    )?;
+
+    // stock_prices: realistic stock data
+    ctx.register_batch(
+        "stock_prices",
+        RecordBatch::try_from_iter(vec![
+            (
+                "symbol",
+                Arc::new(StringArray::from(vec!["DDOG", "DDOG", "DDOG", "DDOG"]))
+                    as ArrayRef,
+            ),
+            (
+                "price",
+                Arc::new(Float64Array::from(vec![150.0, 155.0, 152.0, 158.0])),
+            ),
+        ])?,
+    )?;
+
+    Ok(())
+}
+
+// ============================================================================
+// Logical Plan Node: MiniMatchRecognizeNode
+// ============================================================================
+
+/// A custom logical plan node representing MATCH_RECOGNIZE operations.
+///
+/// This is a simplified implementation that captures the essential structure:
+/// - `measures`: Aggregate expressions computed over matched rows
+/// - `definitions`: Symbol definitions (predicate expressions)
+///
+/// A production implementation would also include:
+/// - Pattern specification (regex-like pattern)
+/// - Partition and order by clauses
+/// - Output mode (ONE ROW PER MATCH, ALL ROWS PER MATCH)
+/// - After match skip strategy
+#[derive(Debug)]
+struct MiniMatchRecognizeNode {
+    input: Arc<LogicalPlan>,
+    schema: DFSchemaRef,
+    /// Measures: (alias, aggregate_expr)
+    measures: Vec<(String, Expr)>,
+    /// Symbol definitions: (symbol_name, predicate_expr)
+    definitions: Vec<(String, Expr)>,
+}
+
+impl UserDefinedLogicalNode for MiniMatchRecognizeNode {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn name(&self) -> &str {
+        "MiniMatchRecognize"
+    }
+
+    fn inputs(&self) -> Vec<&LogicalPlan> {
+        vec![&self.input]
+    }
+
+    fn schema(&self) -> &DFSchemaRef {
+        &self.schema
+    }
+
+    fn check_invariants(&self, _check: InvariantLevel) -> Result<()> {
+        Ok(())
+    }
+
+    fn expressions(&self) -> Vec<Expr> {
+        self.measures
+            .iter()
+            .chain(&self.definitions)
+            .map(|(_, expr)| expr.clone())
+            .collect()
+    }
+
+    fn fmt_for_explain(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "MiniMatchRecognize")?;
+
+        if !self.measures.is_empty() {
+            write!(f, " measures=[")?;
+            for (i, (alias, expr)) in self.measures.iter().enumerate() {
+                if i > 0 {
+                    write!(f, ", ")?;
+                }
+                write!(f, "{alias} := {expr}")?;
+            }
+            write!(f, "]")?;
+        }
+
+        if !self.definitions.is_empty() {
+            write!(f, " define=[")?;
+            for (i, (symbol, expr)) in self.definitions.iter().enumerate() {
+                if i > 0 {
+                    write!(f, ", ")?;
+                }
+                write!(f, "{symbol} := {expr}")?;
+            }
+            write!(f, "]")?;
+        }
+
+        Ok(())
+    }
+
+    fn with_exprs_and_inputs(
+        &self,
+        exprs: Vec<Expr>,
+        inputs: Vec<LogicalPlan>,
+    ) -> Result<Arc<dyn UserDefinedLogicalNode>> {
+        let expected_len = self.measures.len() + self.definitions.len();
+        if exprs.len() != expected_len {
+            return Err(datafusion_common::plan_datafusion_err!(
+                "MiniMatchRecognize: expected {expected_len} expressions, got {}",
+                exprs.len()
+            ));
+        }
+
+        let input = inputs.into_iter().next().ok_or_else(|| {
+            datafusion_common::plan_datafusion_err!(
+                "MiniMatchRecognize requires exactly one input"
+            )
+        })?;
+
+        let (measure_exprs, definition_exprs) = exprs.split_at(self.measures.len());
+
+        let measures = self
+            .measures
+            .iter()
+            .zip(measure_exprs)
+            .map(|((alias, _), expr)| (alias.clone(), expr.clone()))
+            .collect();
+
+        let definitions = self
+            .definitions
+            .iter()
+            .zip(definition_exprs)
+            .map(|((symbol, _), expr)| (symbol.clone(), expr.clone()))
+            .collect();
+
+        Ok(Arc::new(Self {
+            input: Arc::new(input),
+            schema: Arc::clone(&self.schema),
+            measures,
+            definitions,
+        }))
+    }
+
+    fn dyn_hash(&self, state: &mut dyn Hasher) {
+        state.write_usize(Arc::as_ptr(&self.input) as usize);
+        state.write_usize(self.measures.len());
+        state.write_usize(self.definitions.len());
+    }
+
+    fn dyn_eq(&self, other: &dyn UserDefinedLogicalNode) -> bool {
+        other.as_any().downcast_ref::<Self>().is_some_and(|o| {
+            Arc::ptr_eq(&self.input, &o.input)
+                && self.measures == o.measures
+                && self.definitions == o.definitions
+        })
+    }
+
+    fn dyn_ord(&self, other: &dyn UserDefinedLogicalNode) -> Option<Ordering> {
+        if self.dyn_eq(other) {
+            Some(Ordering::Equal)
+        } else {
+            None
+        }
+    }
+}
+
+// ============================================================================
+// Relation Planner: MatchRecognizePlanner
+// ============================================================================
+
+/// Relation planner that creates `MiniMatchRecognizeNode` for MATCH_RECOGNIZE queries.
+#[derive(Debug)]
+struct MatchRecognizePlanner;
+
+impl RelationPlanner for MatchRecognizePlanner {
+    fn plan_relation(
+        &self,
+        relation: TableFactor,
+        ctx: &mut dyn RelationPlannerContext,
+    ) -> Result<RelationPlanning> {
+        let TableFactor::MatchRecognize {
+            table,
+            measures,
+            symbols,
+            alias,
+            ..
+        } = relation
+        else {
+            return Ok(RelationPlanning::Original(relation));
+        };
+
+        // Plan the input table
+        let input = ctx.plan(*table)?;
+        let schema = input.schema().clone();
+
+        // Convert MEASURES: SQL expressions → DataFusion expressions
+        let planned_measures: Vec<(String, Expr)> = measures
+            .iter()
+            .map(|m| {
+                let alias = ctx.normalize_ident(m.alias.clone());
+                let expr = ctx.sql_to_expr(m.expr.clone(), schema.as_ref())?;
+                Ok((alias, expr))
+            })
+            .collect::<Result<_>>()?;
+
+        // Convert DEFINE: symbol definitions → DataFusion expressions
+        let planned_definitions: Vec<(String, Expr)> = symbols
+            .iter()
+            .map(|s| {
+                let name = ctx.normalize_ident(s.symbol.clone());
+                let expr = ctx.sql_to_expr(s.definition.clone(), schema.as_ref())?;
+                Ok((name, expr))
+            })
+            .collect::<Result<_>>()?;
+
+        // Create the custom node
+        let node = MiniMatchRecognizeNode {
+            input: Arc::new(input),
+            schema,
+            measures: planned_measures,
+            definitions: planned_definitions,
+        };
+
+        let plan = LogicalPlan::Extension(Extension {
+            node: Arc::new(node),
+        });
+
+        Ok(RelationPlanning::Planned(PlannedRelation::new(plan, alias)))
+    }
+}
diff --git a/datafusion-examples/examples/relation_planner/pivot_unpivot.rs b/datafusion-examples/examples/relation_planner/pivot_unpivot.rs
new file mode 100644
index 000000000000..64158bb384ef
--- /dev/null
+++ b/datafusion-examples/examples/relation_planner/pivot_unpivot.rs
@@ -0,0 +1,568 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! # PIVOT and UNPIVOT Example
+//!
+//! This example demonstrates implementing SQL `PIVOT` and `UNPIVOT` operations
+//! using a custom [`RelationPlanner`]. Unlike the other examples that create
+//! custom logical/physical nodes, this example shows how to **rewrite** SQL
+//! constructs into equivalent standard SQL operations:
+//!
+//! ## Supported Syntax
+//!
+//! ```sql
+//! -- PIVOT: Transform rows into columns
+//! SELECT * FROM sales
+//!   PIVOT (SUM(amount) FOR quarter IN ('Q1', 'Q2', 'Q3', 'Q4'))
+//!
+//! -- UNPIVOT: Transform columns into rows
+//! SELECT * FROM wide_table
+//!   UNPIVOT (value FOR name IN (col1, col2, col3))
+//! ```
+//!
+//! ## Rewrite Strategy
+//!
+//! **PIVOT** is rewritten to `GROUP BY` with `CASE` expressions:
+//! ```sql
+//! -- Original:
+//! SELECT * FROM sales PIVOT (SUM(amount) FOR quarter IN ('Q1', 'Q2'))
+//!
+//! -- Rewritten to:
+//! SELECT region,
+//!        SUM(CASE quarter WHEN 'Q1' THEN amount END) AS Q1,
+//!        SUM(CASE quarter WHEN 'Q2' THEN amount END) AS Q2
+//! FROM sales
+//! GROUP BY region
+//! ```
+//!
+//! **UNPIVOT** is rewritten to `UNION ALL` of projections:
+//! ```sql
+//! -- Original:
+//! SELECT * FROM wide UNPIVOT (sales FOR quarter IN (q1, q2))
+//!
+//! -- Rewritten to:
+//! SELECT region, 'q1' AS quarter, q1 AS sales FROM wide
+//! UNION ALL
+//! SELECT region, 'q2' AS quarter, q2 AS sales FROM wide
+//! ```
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, Int64Array, StringArray};
+use arrow::record_batch::RecordBatch;
+use datafusion::prelude::*;
+use datafusion_common::{plan_datafusion_err, Result, ScalarValue};
+use datafusion_expr::{
+    case, col, lit,
+    logical_plan::builder::LogicalPlanBuilder,
+    planner::{
+        PlannedRelation, RelationPlanner, RelationPlannerContext, RelationPlanning,
+    },
+    Expr,
+};
+use datafusion_sql::sqlparser::ast::{NullInclusion, PivotValueSource, TableFactor};
+use insta::assert_snapshot;
+
+// ============================================================================
+// Example Entry Point
+// ============================================================================
+
+/// Runs the PIVOT/UNPIVOT examples demonstrating data reshaping operations.
+pub async fn pivot_unpivot() -> Result<()> {
+    let ctx = SessionContext::new();
+    ctx.register_relation_planner(Arc::new(PivotUnpivotPlanner))?;
+    register_sample_data(&ctx)?;
+
+    println!("PIVOT and UNPIVOT Example");
+    println!("=========================\n");
+
+    run_examples(&ctx).await
+}
+
+async fn run_examples(ctx: &SessionContext) -> Result<()> {
+    // ----- PIVOT Examples -----
+
+    // Example 1: Basic PIVOT
+    // Transforms: (region, quarter, amount) → (region, Q1, Q2)
+    let results = run_example(
+        ctx,
+        "Example 1: Basic PIVOT",
+        r#"SELECT * FROM quarterly_sales
+           PIVOT (SUM(amount) FOR quarter IN ('Q1', 'Q2')) AS p
+           ORDER BY region"#,
+    )
+    .await?;
+    assert_snapshot!(results, @r"
+    +--------+------+------+
+    | region | Q1   | Q2   |
+    +--------+------+------+
+    | North  | 1000 | 1500 |
+    | South  | 1200 | 1300 |
+    +--------+------+------+
+    ");
+
+    // Example 2: PIVOT with multiple aggregates
+    // Creates columns for each (aggregate, value) combination
+    let results = run_example(
+        ctx,
+        "Example 2: PIVOT with multiple aggregates",
+        r#"SELECT * FROM quarterly_sales
+           PIVOT (SUM(amount), AVG(amount) FOR quarter IN ('Q1', 'Q2')) AS p
+           ORDER BY region"#,
+    )
+    .await?;
+    assert_snapshot!(results, @r"
+    +--------+--------+--------+--------+--------+
+    | region | sum_Q1 | sum_Q2 | avg_Q1 | avg_Q2 |
+    +--------+--------+--------+--------+--------+
+    | North  | 1000   | 1500   | 1000.0 | 1500.0 |
+    | South  | 1200   | 1300   | 1200.0 | 1300.0 |
+    +--------+--------+--------+--------+--------+
+    ");
+
+    // Example 3: PIVOT with multiple grouping columns
+    // Non-pivot, non-aggregate columns become GROUP BY columns
+    let results = run_example(
+        ctx,
+        "Example 3: PIVOT with multiple grouping columns",
+        r#"SELECT * FROM product_sales
+           PIVOT (SUM(amount) FOR quarter IN ('Q1', 'Q2')) AS p
+           ORDER BY region, product"#,
+    )
+    .await?;
+    assert_snapshot!(results, @r"
+    +--------+----------+-----+-----+
+    | region | product  | Q1  | Q2  |
+    +--------+----------+-----+-----+
+    | North  | ProductA | 500 |     |
+    | North  | ProductB | 500 |     |
+    | South  | ProductA |     | 650 |
+    +--------+----------+-----+-----+
+    ");
+
+    // ----- UNPIVOT Examples -----
+
+    // Example 4: Basic UNPIVOT
+    // Transforms: (region, q1, q2) → (region, quarter, sales)
+    let results = run_example(
+        ctx,
+        "Example 4: Basic UNPIVOT",
+        r#"SELECT * FROM wide_sales
+           UNPIVOT (sales FOR quarter IN (q1 AS 'Q1', q2 AS 'Q2')) AS u
+           ORDER BY quarter, region"#,
+    )
+    .await?;
+    assert_snapshot!(results, @r"
+    +--------+---------+-------+
+    | region | quarter | sales |
+    +--------+---------+-------+
+    | North  | Q1      | 1000  |
+    | South  | Q1      | 1200  |
+    | North  | Q2      | 1500  |
+    | South  | Q2      | 1300  |
+    +--------+---------+-------+
+    ");
+
+    // Example 5: UNPIVOT with INCLUDE NULLS
+    // By default, UNPIVOT excludes rows where the value column is NULL.
+    // INCLUDE NULLS keeps them (same result here since no NULLs in data).
+    let results = run_example(
+        ctx,
+        "Example 5: UNPIVOT INCLUDE NULLS",
+        r#"SELECT * FROM wide_sales
+           UNPIVOT INCLUDE NULLS (sales FOR quarter IN (q1 AS 'Q1', q2 AS 'Q2')) AS u
+           ORDER BY quarter, region"#,
+    )
+    .await?;
+    assert_snapshot!(results, @r"
+    +--------+---------+-------+
+    | region | quarter | sales |
+    +--------+---------+-------+
+    | North  | Q1      | 1000  |
+    | South  | Q1      | 1200  |
+    | North  | Q2      | 1500  |
+    | South  | Q2      | 1300  |
+    +--------+---------+-------+
+    ");
+
+    // Example 6: PIVOT with column projection
+    // Standard SQL operations work seamlessly after PIVOT
+    let results = run_example(
+        ctx,
+        "Example 6: PIVOT with projection",
+        r#"SELECT region FROM quarterly_sales
+           PIVOT (SUM(amount) FOR quarter IN ('Q1', 'Q2')) AS p
+           ORDER BY region"#,
+    )
+    .await?;
+    assert_snapshot!(results, @r"
+    +--------+
+    | region |
+    +--------+
+    | North  |
+    | South  |
+    +--------+
+    ");
+
+    Ok(())
+}
+
+/// Helper to run a single example query and capture results.
+async fn run_example(ctx: &SessionContext, title: &str, sql: &str) -> Result<String> {
+    println!("{title}:\n{sql}\n");
+    let df = ctx.sql(sql).await?;
+    println!("{}\n", df.logical_plan().display_indent());
+
+    let batches = df.collect().await?;
+    let results = arrow::util::pretty::pretty_format_batches(&batches)?.to_string();
+    println!("{results}\n");
+
+    Ok(results)
+}
+
+/// Register test data tables.
+fn register_sample_data(ctx: &SessionContext) -> Result<()> {
+    // quarterly_sales: normalized sales data (region, quarter, amount)
+    ctx.register_batch(
+        "quarterly_sales",
+        RecordBatch::try_from_iter(vec![
+            (
+                "region",
+                Arc::new(StringArray::from(vec!["North", "North", "South", "South"]))
+                    as ArrayRef,
+            ),
+            (
+                "quarter",
+                Arc::new(StringArray::from(vec!["Q1", "Q2", "Q1", "Q2"])),
+            ),
+            (
+                "amount",
+                Arc::new(Int64Array::from(vec![1000, 1500, 1200, 1300])),
+            ),
+        ])?,
+    )?;
+
+    // product_sales: sales with additional grouping dimension
+    ctx.register_batch(
+        "product_sales",
+        RecordBatch::try_from_iter(vec![
+            (
+                "region",
+                Arc::new(StringArray::from(vec!["North", "North", "South"])) as ArrayRef,
+            ),
+            (
+                "quarter",
+                Arc::new(StringArray::from(vec!["Q1", "Q1", "Q2"])),
+            ),
+            (
+                "product",
+                Arc::new(StringArray::from(vec!["ProductA", "ProductB", "ProductA"])),
+            ),
+            ("amount", Arc::new(Int64Array::from(vec![500, 500, 650]))),
+        ])?,
+    )?;
+
+    // wide_sales: denormalized/wide format (for UNPIVOT)
+    ctx.register_batch(
+        "wide_sales",
+        RecordBatch::try_from_iter(vec![
+            (
+                "region",
+                Arc::new(StringArray::from(vec!["North", "South"])) as ArrayRef,
+            ),
+            ("q1", Arc::new(Int64Array::from(vec![1000, 1200]))),
+            ("q2", Arc::new(Int64Array::from(vec![1500, 1300]))),
+        ])?,
+    )?;
+
+    Ok(())
+}
+
+// ============================================================================
+// Relation Planner: PivotUnpivotPlanner
+// ============================================================================
+
+/// Relation planner that rewrites PIVOT and UNPIVOT into standard SQL.
+#[derive(Debug)]
+struct PivotUnpivotPlanner;
+
+impl RelationPlanner for PivotUnpivotPlanner {
+    fn plan_relation(
+        &self,
+        relation: TableFactor,
+        ctx: &mut dyn RelationPlannerContext,
+    ) -> Result<RelationPlanning> {
+        match relation {
+            TableFactor::Pivot {
+                table,
+                aggregate_functions,
+                value_column,
+                value_source,
+                alias,
+                ..
+            } => plan_pivot(
+                ctx,
+                *table,
+                &aggregate_functions,
+                &value_column,
+                value_source,
+                alias,
+            ),
+
+            TableFactor::Unpivot {
+                table,
+                value,
+                name,
+                columns,
+                null_inclusion,
+                alias,
+            } => plan_unpivot(
+                ctx,
+                *table,
+                &value,
+                name,
+                &columns,
+                null_inclusion.as_ref(),
+                alias,
+            ),
+
+            other => Ok(RelationPlanning::Original(other)),
+        }
+    }
+}
+
+// ============================================================================
+// PIVOT Implementation
+// ============================================================================
+
+/// Rewrite PIVOT to GROUP BY with CASE expressions.
+fn plan_pivot(
+    ctx: &mut dyn RelationPlannerContext,
+    table: TableFactor,
+    aggregate_functions: &[datafusion_sql::sqlparser::ast::ExprWithAlias],
+    value_column: &[datafusion_sql::sqlparser::ast::Expr],
+    value_source: PivotValueSource,
+    alias: Option<datafusion_sql::sqlparser::ast::TableAlias>,
+) -> Result<RelationPlanning> {
+    // Plan the input table
+    let input = ctx.plan(table)?;
+    let schema = input.schema();
+
+    // Parse aggregate functions
+    let aggregates: Vec<Expr> = aggregate_functions
+        .iter()
+        .map(|agg| ctx.sql_to_expr(agg.expr.clone(), schema.as_ref()))
+        .collect::<Result<_>>()?;
+
+    // Get the pivot column (only single-column pivot supported)
+    if value_column.len() != 1 {
+        return Err(plan_datafusion_err!(
+            "Only single-column PIVOT is supported"
+        ));
+    }
+    let pivot_col = ctx.sql_to_expr(value_column[0].clone(), schema.as_ref())?;
+    let pivot_col_name = extract_column_name(&pivot_col)?;
+
+    // Parse pivot values
+    let pivot_values = match value_source {
+        PivotValueSource::List(list) => list
+            .iter()
+            .map(|item| {
+                let alias = item
+                    .alias
+                    .as_ref()
+                    .map(|id| ctx.normalize_ident(id.clone()));
+                let expr = ctx.sql_to_expr(item.expr.clone(), schema.as_ref())?;
+                Ok((alias, expr))
+            })
+            .collect::<Result<Vec<_>>>()?,
+        _ => {
+            return Err(plan_datafusion_err!(
+                "Dynamic PIVOT (ANY/Subquery) is not supported"
+            ));
+        }
+    };
+
+    // Determine GROUP BY columns (non-pivot, non-aggregate columns)
+    let agg_input_cols: Vec<&str> = aggregates
+        .iter()
+        .filter_map(|agg| {
+            if let Expr::AggregateFunction(f) = agg {
+                f.params.args.first().and_then(|e| {
+                    if let Expr::Column(c) = e {
+                        Some(c.name.as_str())
+                    } else {
+                        None
+                    }
+                })
+            } else {
+                None
+            }
+        })
+        .collect();
+
+    let group_by_cols: Vec<Expr> = schema
+        .fields()
+        .iter()
+        .map(|f| f.name().as_str())
+        .filter(|name| *name != pivot_col_name.as_str() && !agg_input_cols.contains(name))
+        .map(col)
+        .collect();
+
+    // Build CASE expressions for each (aggregate, pivot_value) pair
+    let mut pivot_exprs = Vec::new();
+    for agg in &aggregates {
+        let Expr::AggregateFunction(agg_fn) = agg else {
+            continue;
+        };
+        let Some(agg_input) = agg_fn.params.args.first().cloned() else {
+            continue;
+        };
+
+        for (value_alias, pivot_value) in &pivot_values {
+            // CASE pivot_col WHEN pivot_value THEN agg_input END
+            let case_expr = case(col(&pivot_col_name))
+                .when(pivot_value.clone(), agg_input.clone())
+                .end()?;
+
+            // Wrap in aggregate function
+            let pivoted = agg_fn.func.call(vec![case_expr]);
+
+            // Determine column alias
+            let value_str = value_alias
+                .clone()
+                .unwrap_or_else(|| expr_to_string(pivot_value));
+            let col_alias = if aggregates.len() > 1 {
+                format!("{}_{}", agg_fn.func.name(), value_str)
+            } else {
+                value_str
+            };
+
+            pivot_exprs.push(pivoted.alias(col_alias));
+        }
+    }
+
+    let plan = LogicalPlanBuilder::from(input)
+        .aggregate(group_by_cols, pivot_exprs)?
+        .build()?;
+
+    Ok(RelationPlanning::Planned(PlannedRelation::new(plan, alias)))
+}
+
+// ============================================================================
+// UNPIVOT Implementation
+// ============================================================================
+
+/// Rewrite UNPIVOT to UNION ALL of projections.
+fn plan_unpivot(
+    ctx: &mut dyn RelationPlannerContext,
+    table: TableFactor,
+    value: &datafusion_sql::sqlparser::ast::Expr,
+    name: datafusion_sql::sqlparser::ast::Ident,
+    columns: &[datafusion_sql::sqlparser::ast::ExprWithAlias],
+    null_inclusion: Option<&NullInclusion>,
+    alias: Option<datafusion_sql::sqlparser::ast::TableAlias>,
+) -> Result<RelationPlanning> {
+    // Plan the input table
+    let input = ctx.plan(table)?;
+    let schema = input.schema();
+
+    // Output column names
+    let value_col_name = value.to_string();
+    let name_col_name = ctx.normalize_ident(name);
+
+    // Parse columns to unpivot: (source_column, label)
+    let unpivot_cols: Vec<(String, String)> = columns
+        .iter()
+        .map(|c| {
+            let label = c
+                .alias
+                .as_ref()
+                .map(|id| ctx.normalize_ident(id.clone()))
+                .unwrap_or_else(|| c.expr.to_string());
+            let expr = ctx.sql_to_expr(c.expr.clone(), schema.as_ref())?;
+            let col_name = extract_column_name(&expr)?;
+            Ok((col_name.to_string(), label))
+        })
+        .collect::<Result<_>>()?;
+
+    // Columns to preserve (not being unpivoted)
+    let keep_cols: Vec<&str> = schema
+        .fields()
+        .iter()
+        .map(|f| f.name().as_str())
+        .filter(|name| !unpivot_cols.iter().any(|(c, _)| c == *name))
+        .collect();
+
+    // Build UNION ALL: one SELECT per unpivot column
+    if unpivot_cols.is_empty() {
+        return Err(plan_datafusion_err!("UNPIVOT requires at least one column"));
+    }
+
+    let mut union_inputs: Vec<_> = unpivot_cols
+        .iter()
+        .map(|(col_name, label)| {
+            let mut projection: Vec<Expr> = keep_cols.iter().map(|c| col(*c)).collect();
+            projection.push(lit(label.clone()).alias(&name_col_name));
+            projection.push(col(col_name).alias(&value_col_name));
+
+            LogicalPlanBuilder::from(input.clone())
+                .project(projection)?
+                .build()
+        })
+        .collect::<Result<_>>()?;
+
+    // Combine with UNION ALL
+    let mut plan = union_inputs.remove(0);
+    for branch in union_inputs {
+        plan = LogicalPlanBuilder::from(plan).union(branch)?.build()?;
+    }
+
+    // Apply EXCLUDE NULLS filter (default behavior)
+    let exclude_nulls = null_inclusion.is_none()
+        || matches!(null_inclusion, Some(&NullInclusion::ExcludeNulls));
+    if exclude_nulls {
+        plan = LogicalPlanBuilder::from(plan)
+            .filter(col(&value_col_name).is_not_null())?
+            .build()?;
+    }
+
+    Ok(RelationPlanning::Planned(PlannedRelation::new(plan, alias)))
+}
+
+// ============================================================================
+// Helpers
+// ============================================================================
+
+/// Extract column name from an expression.
+fn extract_column_name(expr: &Expr) -> Result<String> {
+    match expr {
+        Expr::Column(c) => Ok(c.name.clone()),
+        _ => Err(plan_datafusion_err!(
+            "Expected column reference, got {expr}"
+        )),
+    }
+}
+
+/// Convert an expression to a string for use as column alias.
+fn expr_to_string(expr: &Expr) -> String {
+    match expr {
+        Expr::Literal(ScalarValue::Utf8(Some(s)), _) => s.clone(),
+        Expr::Literal(v, _) => v.to_string(),
+        other => other.to_string(),
+    }
+}
diff --git a/datafusion-examples/examples/relation_planner/table_sample.rs b/datafusion-examples/examples/relation_planner/table_sample.rs
new file mode 100644
index 000000000000..a3ae5396eda1
--- /dev/null
+++ b/datafusion-examples/examples/relation_planner/table_sample.rs
@@ -0,0 +1,846 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! # TABLESAMPLE Example
+//!
+//! This example demonstrates implementing SQL `TABLESAMPLE` support using
+//! DataFusion's extensibility APIs.
+//!
+//! This is a working `TABLESAMPLE` implementation that can serve as a starting
+//! point for your own projects. It also works as a template for adding other
+//! custom SQL operators, covering the full pipeline from parsing to execution.
+//!
+//! It shows how to:
+//!
+//! 1. **Parse** TABLESAMPLE syntax via a custom [`RelationPlanner`]
+//! 2. **Plan** sampling as a custom logical node ([`TableSamplePlanNode`])
+//! 3. **Execute** sampling via a custom physical operator ([`SampleExec`])
+//!
+//! ## Supported Syntax
+//!
+//! ```sql
+//! -- Bernoulli sampling (each row has N% chance of selection)
+//! SELECT * FROM table TABLESAMPLE BERNOULLI(10 PERCENT)
+//!
+//! -- Fractional sampling (0.0 to 1.0)
+//! SELECT * FROM table TABLESAMPLE (0.1)
+//!
+//! -- Row count limit
+//! SELECT * FROM table TABLESAMPLE (100 ROWS)
+//!
+//! -- Reproducible sampling with a seed
+//! SELECT * FROM table TABLESAMPLE (10 PERCENT) REPEATABLE(42)
+//! ```
+//!
+//! ## Architecture
+//!
+//! ```text
+//! ┌─────────────────────────────────────────────────────────────────┐
+//! │                         SQL Query                               │
+//! │  SELECT * FROM t TABLESAMPLE BERNOULLI(10 PERCENT) REPEATABLE(1)│
+//! └─────────────────────────────────────────────────────────────────┘
+//!                                │
+//!                                ▼
+//! ┌─────────────────────────────────────────────────────────────────┐
+//! │                    TableSamplePlanner                           │
+//! │    (RelationPlanner: parses TABLESAMPLE, creates logical node)  │
+//! └─────────────────────────────────────────────────────────────────┘
+//!                                │
+//!                                ▼
+//! ┌─────────────────────────────────────────────────────────────────┐
+//! │                   TableSamplePlanNode                           │
+//! │         (UserDefinedLogicalNode: stores sampling params)        │
+//! └─────────────────────────────────────────────────────────────────┘
+//!                                │
+//!                                ▼
+//! ┌─────────────────────────────────────────────────────────────────┐
+//! │                TableSampleExtensionPlanner                      │
+//! │       (ExtensionPlanner: creates physical execution plan)       │
+//! └─────────────────────────────────────────────────────────────────┘
+//!                                │
+//!                                ▼
+//! ┌─────────────────────────────────────────────────────────────────┐
+//! │                        SampleExec                               │
+//! │    (ExecutionPlan: performs actual row sampling at runtime)     │
+//! └─────────────────────────────────────────────────────────────────┘
+//! ```
+
+use std::{
+    any::Any,
+    fmt::{self, Debug, Formatter},
+    hash::{Hash, Hasher},
+    ops::{Add, Div, Mul, Sub},
+    pin::Pin,
+    str::FromStr,
+    sync::Arc,
+    task::{Context, Poll},
+};
+
+use arrow::{
+    array::{ArrayRef, Int32Array, RecordBatch, StringArray, UInt32Array},
+    compute,
+};
+use arrow_schema::SchemaRef;
+use futures::{
+    ready,
+    stream::{Stream, StreamExt},
+};
+use rand::{rngs::StdRng, Rng, SeedableRng};
+use tonic::async_trait;
+
+use datafusion::{
+    execution::{
+        context::QueryPlanner, RecordBatchStream, SendableRecordBatchStream,
+        SessionState, SessionStateBuilder, TaskContext,
+    },
+    physical_expr::EquivalenceProperties,
+    physical_plan::{
+        metrics::{BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet, RecordOutput},
+        DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties,
+    },
+    physical_planner::{DefaultPhysicalPlanner, ExtensionPlanner, PhysicalPlanner},
+    prelude::*,
+};
+use datafusion_common::{
+    internal_err, not_impl_err, plan_datafusion_err, plan_err, DFSchemaRef,
+    DataFusionError, Result, Statistics,
+};
+use datafusion_expr::{
+    logical_plan::{Extension, LogicalPlan, LogicalPlanBuilder},
+    planner::{
+        PlannedRelation, RelationPlanner, RelationPlannerContext, RelationPlanning,
+    },
+    UserDefinedLogicalNode, UserDefinedLogicalNodeCore,
+};
+use datafusion_sql::sqlparser::ast::{
+    self, TableFactor, TableSampleMethod, TableSampleUnit,
+};
+use insta::assert_snapshot;
+
+// ============================================================================
+// Example Entry Point
+// ============================================================================
+
+/// Runs the TABLESAMPLE examples demonstrating various sampling techniques.
+pub async fn table_sample() -> Result<()> {
+    // Build session with custom query planner for physical planning
+    let state = SessionStateBuilder::new()
+        .with_default_features()
+        .with_query_planner(Arc::new(TableSampleQueryPlanner))
+        .build();
+
+    let ctx = SessionContext::new_with_state(state);
+
+    // Register custom relation planner for logical planning
+    ctx.register_relation_planner(Arc::new(TableSamplePlanner))?;
+    register_sample_data(&ctx)?;
+
+    println!("TABLESAMPLE Example");
+    println!("===================\n");
+
+    run_examples(&ctx).await
+}
+
+async fn run_examples(ctx: &SessionContext) -> Result<()> {
+    // Example 1: Baseline - full table scan
+    let results = run_example(
+        ctx,
+        "Example 1: Full table (baseline)",
+        "SELECT * FROM sample_data",
+    )
+    .await?;
+    assert_snapshot!(results, @r"
+    +---------+---------+
+    | column1 | column2 |
+    +---------+---------+
+    | 1       | row_1   |
+    | 2       | row_2   |
+    | 3       | row_3   |
+    | 4       | row_4   |
+    | 5       | row_5   |
+    | 6       | row_6   |
+    | 7       | row_7   |
+    | 8       | row_8   |
+    | 9       | row_9   |
+    | 10      | row_10  |
+    +---------+---------+
+    ");
+
+    // Example 2: Percentage-based Bernoulli sampling
+    // REPEATABLE(seed) ensures deterministic results for snapshot testing
+    let results = run_example(
+        ctx,
+        "Example 2: BERNOULLI percentage sampling",
+        "SELECT * FROM sample_data TABLESAMPLE BERNOULLI(30 PERCENT) REPEATABLE(123)",
+    )
+    .await?;
+    assert_snapshot!(results, @r"
+    +---------+---------+
+    | column1 | column2 |
+    +---------+---------+
+    | 1       | row_1   |
+    | 2       | row_2   |
+    | 7       | row_7   |
+    | 8       | row_8   |
+    +---------+---------+
+    ");
+
+    // Example 3: Fractional sampling (0.0 to 1.0)
+    // REPEATABLE(seed) ensures deterministic results for snapshot testing
+    let results = run_example(
+        ctx,
+        "Example 3: Fractional sampling",
+        "SELECT * FROM sample_data TABLESAMPLE (0.5) REPEATABLE(456)",
+    )
+    .await?;
+    assert_snapshot!(results, @r"
+    +---------+---------+
+    | column1 | column2 |
+    +---------+---------+
+    | 2       | row_2   |
+    | 4       | row_4   |
+    | 8       | row_8   |
+    +---------+---------+
+    ");
+
+    // Example 4: Row count limit (deterministic, no seed needed)
+    let results = run_example(
+        ctx,
+        "Example 4: Row count limit",
+        "SELECT * FROM sample_data TABLESAMPLE (3 ROWS)",
+    )
+    .await?;
+    assert_snapshot!(results, @r"
+    +---------+---------+
+    | column1 | column2 |
+    +---------+---------+
+    | 1       | row_1   |
+    | 2       | row_2   |
+    | 3       | row_3   |
+    +---------+---------+
+    ");
+
+    // Example 5: Sampling combined with filtering
+    let results = run_example(
+        ctx,
+        "Example 5: Sampling with WHERE clause",
+        "SELECT * FROM sample_data TABLESAMPLE (5 ROWS) WHERE column1 > 2",
+    )
+    .await?;
+    assert_snapshot!(results, @r"
+    +---------+---------+
+    | column1 | column2 |
+    +---------+---------+
+    | 3       | row_3   |
+    | 4       | row_4   |
+    | 5       | row_5   |
+    +---------+---------+
+    ");
+
+    // Example 6: Sampling in JOIN queries
+    // REPEATABLE(seed) ensures deterministic results for snapshot testing
+    let results = run_example(
+        ctx,
+        "Example 6: Sampling in JOINs",
+        r#"SELECT t1.column1, t2.column1, t1.column2, t2.column2
+           FROM sample_data t1 TABLESAMPLE (0.7) REPEATABLE(789)
+           JOIN sample_data t2 TABLESAMPLE (0.7) REPEATABLE(123)
+           ON t1.column1 = t2.column1"#,
+    )
+    .await?;
+    assert_snapshot!(results, @r"
+    +---------+---------+---------+---------+
+    | column1 | column1 | column2 | column2 |
+    +---------+---------+---------+---------+
+    | 2       | 2       | row_2   | row_2   |
+    | 5       | 5       | row_5   | row_5   |
+    | 7       | 7       | row_7   | row_7   |
+    | 8       | 8       | row_8   | row_8   |
+    | 10      | 10      | row_10  | row_10  |
+    +---------+---------+---------+---------+
+    ");
+
+    Ok(())
+}
+
+/// Helper to run a single example query and capture results.
+async fn run_example(ctx: &SessionContext, title: &str, sql: &str) -> Result<String> {
+    println!("{title}:\n{sql}\n");
+    let df = ctx.sql(sql).await?;
+    println!("{}\n", df.logical_plan().display_indent());
+
+    let batches = df.collect().await?;
+    let results = arrow::util::pretty::pretty_format_batches(&batches)?.to_string();
+    println!("{results}\n");
+
+    Ok(results)
+}
+
+/// Register test data: 10 rows with column1=1..10 and column2="row_1".."row_10"
+fn register_sample_data(ctx: &SessionContext) -> Result<()> {
+    let column1: ArrayRef = Arc::new(Int32Array::from((1..=10).collect::<Vec<i32>>()));
+    let column2: ArrayRef = Arc::new(StringArray::from(
+        (1..=10).map(|i| format!("row_{i}")).collect::<Vec<_>>(),
+    ));
+    let batch =
+        RecordBatch::try_from_iter(vec![("column1", column1), ("column2", column2)])?;
+    ctx.register_batch("sample_data", batch)?;
+    Ok(())
+}
+
+// ============================================================================
+// Logical Planning: TableSamplePlanner + TableSamplePlanNode
+// ============================================================================
+
+/// Relation planner that intercepts `TABLESAMPLE` clauses in SQL and creates
+/// [`TableSamplePlanNode`] logical nodes.
+#[derive(Debug)]
+struct TableSamplePlanner;
+
+impl RelationPlanner for TableSamplePlanner {
+    fn plan_relation(
+        &self,
+        relation: TableFactor,
+        context: &mut dyn RelationPlannerContext,
+    ) -> Result<RelationPlanning> {
+        // Only handle Table relations with TABLESAMPLE clause
+        let TableFactor::Table {
+            sample: Some(sample),
+            alias,
+            name,
+            args,
+            with_hints,
+            version,
+            with_ordinality,
+            partitions,
+            json_path,
+            index_hints,
+        } = relation
+        else {
+            return Ok(RelationPlanning::Original(relation));
+        };
+
+        // Extract sample spec (handles both before/after alias positions)
+        let sample = match sample {
+            ast::TableSampleKind::BeforeTableAlias(s)
+            | ast::TableSampleKind::AfterTableAlias(s) => s,
+        };
+
+        // Validate sampling method
+        if let Some(method) = &sample.name {
+            if *method != TableSampleMethod::Bernoulli
+                && *method != TableSampleMethod::Row
+            {
+                return not_impl_err!(
+                    "Sampling method {} is not supported (only BERNOULLI and ROW)",
+                    method
+                );
+            }
+        }
+
+        // Offset sampling (ClickHouse-style) not supported
+        if sample.offset.is_some() {
+            return not_impl_err!(
+                "TABLESAMPLE with OFFSET is not supported (requires total row count)"
+            );
+        }
+
+        // Parse optional REPEATABLE seed
+        let seed = sample
+            .seed
+            .map(|s| {
+                s.value.to_string().parse::<u64>().map_err(|_| {
+                    plan_datafusion_err!("REPEATABLE seed must be an integer")
+                })
+            })
+            .transpose()?;
+
+        // Plan the underlying table without the sample clause
+        let base_relation = TableFactor::Table {
+            sample: None,
+            alias: alias.clone(),
+            name,
+            args,
+            with_hints,
+            version,
+            with_ordinality,
+            partitions,
+            json_path,
+            index_hints,
+        };
+        let input = context.plan(base_relation)?;
+
+        // Handle bucket sampling (Hive-style: TABLESAMPLE(BUCKET x OUT OF y))
+        if let Some(bucket) = sample.bucket {
+            if bucket.on.is_some() {
+                return not_impl_err!(
+                    "TABLESAMPLE BUCKET with ON clause requires CLUSTERED BY table"
+                );
+            }
+            let bucket_num: u64 =
+                bucket.bucket.to_string().parse().map_err(|_| {
+                    plan_datafusion_err!("bucket number must be an integer")
+                })?;
+            let total: u64 =
+                bucket.total.to_string().parse().map_err(|_| {
+                    plan_datafusion_err!("bucket total must be an integer")
+                })?;
+
+            let fraction = bucket_num as f64 / total as f64;
+            let plan = TableSamplePlanNode::new(input, fraction, seed).into_plan();
+            return Ok(RelationPlanning::Planned(PlannedRelation::new(plan, alias)));
+        }
+
+        // Handle quantity-based sampling
+        let Some(quantity) = sample.quantity else {
+            return plan_err!(
+                "TABLESAMPLE requires a quantity (percentage, fraction, or row count)"
+            );
+        };
+
+        match quantity.unit {
+            // TABLESAMPLE (N ROWS) - exact row limit
+            Some(TableSampleUnit::Rows) => {
+                let rows = parse_quantity::<i64>(&quantity.value)?;
+                if rows < 0 {
+                    return plan_err!("row count must be non-negative, got {}", rows);
+                }
+                let plan = LogicalPlanBuilder::from(input)
+                    .limit(0, Some(rows as usize))?
+                    .build()?;
+                Ok(RelationPlanning::Planned(PlannedRelation::new(plan, alias)))
+            }
+
+            // TABLESAMPLE (N PERCENT) - percentage sampling
+            Some(TableSampleUnit::Percent) => {
+                let percent = parse_quantity::<f64>(&quantity.value)?;
+                let fraction = percent / 100.0;
+                let plan = TableSamplePlanNode::new(input, fraction, seed).into_plan();
+                Ok(RelationPlanning::Planned(PlannedRelation::new(plan, alias)))
+            }
+
+            // TABLESAMPLE (N) - fraction if <1.0, row limit if >=1.0
+            None => {
+                let value = parse_quantity::<f64>(&quantity.value)?;
+                if value < 0.0 {
+                    return plan_err!("sample value must be non-negative, got {}", value);
+                }
+                let plan = if value >= 1.0 {
+                    // Interpret as row limit
+                    LogicalPlanBuilder::from(input)
+                        .limit(0, Some(value as usize))?
+                        .build()?
+                } else {
+                    // Interpret as fraction
+                    TableSamplePlanNode::new(input, value, seed).into_plan()
+                };
+                Ok(RelationPlanning::Planned(PlannedRelation::new(plan, alias)))
+            }
+        }
+    }
+}
+
+/// Parse a SQL expression as a numeric value (supports basic arithmetic).
+fn parse_quantity<T>(expr: &ast::Expr) -> Result<T>
+where
+    T: FromStr + Add<Output = T> + Sub<Output = T> + Mul<Output = T> + Div<Output = T>,
+{
+    eval_numeric_expr(expr)
+        .ok_or_else(|| plan_datafusion_err!("invalid numeric expression: {:?}", expr))
+}
+
+/// Recursively evaluate numeric SQL expressions.
+fn eval_numeric_expr<T>(expr: &ast::Expr) -> Option<T>
+where
+    T: FromStr + Add<Output = T> + Sub<Output = T> + Mul<Output = T> + Div<Output = T>,
+{
+    match expr {
+        ast::Expr::Value(v) => match &v.value {
+            ast::Value::Number(n, _) => n.to_string().parse().ok(),
+            _ => None,
+        },
+        ast::Expr::BinaryOp { left, op, right } => {
+            let l = eval_numeric_expr::<T>(left)?;
+            let r = eval_numeric_expr::<T>(right)?;
+            match op {
+                ast::BinaryOperator::Plus => Some(l + r),
+                ast::BinaryOperator::Minus => Some(l - r),
+                ast::BinaryOperator::Multiply => Some(l * r),
+                ast::BinaryOperator::Divide => Some(l / r),
+                _ => None,
+            }
+        }
+        _ => None,
+    }
+}
+
+/// Custom logical plan node representing a TABLESAMPLE operation.
+///
+/// Stores sampling parameters (bounds, seed) and wraps the input plan.
+/// Gets converted to [`SampleExec`] during physical planning.
+#[derive(Debug, Clone, Hash, Eq, PartialEq, PartialOrd)]
+struct TableSamplePlanNode {
+    input: LogicalPlan,
+    lower_bound: HashableF64,
+    upper_bound: HashableF64,
+    seed: u64,
+}
+
+impl TableSamplePlanNode {
+    /// Create a new sampling node with the given fraction (0.0 to 1.0).
+    fn new(input: LogicalPlan, fraction: f64, seed: Option<u64>) -> Self {
+        Self {
+            input,
+            lower_bound: HashableF64(0.0),
+            upper_bound: HashableF64(fraction),
+            seed: seed.unwrap_or_else(rand::random),
+        }
+    }
+
+    /// Wrap this node in a LogicalPlan::Extension.
+    fn into_plan(self) -> LogicalPlan {
+        LogicalPlan::Extension(Extension {
+            node: Arc::new(self),
+        })
+    }
+}
+
+impl UserDefinedLogicalNodeCore for TableSamplePlanNode {
+    fn name(&self) -> &str {
+        "TableSample"
+    }
+
+    fn inputs(&self) -> Vec<&LogicalPlan> {
+        vec![&self.input]
+    }
+
+    fn schema(&self) -> &DFSchemaRef {
+        self.input.schema()
+    }
+
+    fn expressions(&self) -> Vec<Expr> {
+        vec![]
+    }
+
+    fn fmt_for_explain(&self, f: &mut Formatter) -> fmt::Result {
+        write!(
+            f,
+            "Sample: bounds=[{}, {}], seed={}",
+            self.lower_bound.0, self.upper_bound.0, self.seed
+        )
+    }
+
+    fn with_exprs_and_inputs(
+        &self,
+        _exprs: Vec<Expr>,
+        mut inputs: Vec<LogicalPlan>,
+    ) -> Result<Self> {
+        Ok(Self {
+            input: inputs.swap_remove(0),
+            lower_bound: self.lower_bound,
+            upper_bound: self.upper_bound,
+            seed: self.seed,
+        })
+    }
+}
+
+/// Wrapper for f64 that implements Hash and Eq (required for LogicalPlan).
+#[derive(Debug, Clone, Copy, PartialOrd)]
+struct HashableF64(f64);
+
+impl PartialEq for HashableF64 {
+    fn eq(&self, other: &Self) -> bool {
+        self.0.to_bits() == other.0.to_bits()
+    }
+}
+
+impl Eq for HashableF64 {}
+
+impl Hash for HashableF64 {
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        self.0.to_bits().hash(state);
+    }
+}
+
+// ============================================================================
+// Physical Planning: TableSampleQueryPlanner + TableSampleExtensionPlanner
+// ============================================================================
+
+/// Custom query planner that registers [`TableSampleExtensionPlanner`] to
+/// convert [`TableSamplePlanNode`] into [`SampleExec`].
+#[derive(Debug)]
+struct TableSampleQueryPlanner;
+
+#[async_trait]
+impl QueryPlanner for TableSampleQueryPlanner {
+    async fn create_physical_plan(
+        &self,
+        logical_plan: &LogicalPlan,
+        session_state: &SessionState,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
+        let planner = DefaultPhysicalPlanner::with_extension_planners(vec![Arc::new(
+            TableSampleExtensionPlanner,
+        )]);
+        planner
+            .create_physical_plan(logical_plan, session_state)
+            .await
+    }
+}
+
+/// Extension planner that converts [`TableSamplePlanNode`] to [`SampleExec`].
+struct TableSampleExtensionPlanner;
+
+#[async_trait]
+impl ExtensionPlanner for TableSampleExtensionPlanner {
+    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) = node.as_any().downcast_ref::<TableSamplePlanNode>()
+        else {
+            return Ok(None);
+        };
+
+        let exec = SampleExec::try_new(
+            Arc::clone(&physical_inputs[0]),
+            sample_node.lower_bound.0,
+            sample_node.upper_bound.0,
+            sample_node.seed,
+        )?;
+        Ok(Some(Arc::new(exec)))
+    }
+}
+
+// ============================================================================
+// Physical Execution: SampleExec + BernoulliSampler
+// ============================================================================
+
+/// Physical execution plan that samples rows from its input using Bernoulli sampling.
+///
+/// Each row is independently selected with probability `(upper_bound - lower_bound)`
+/// and appears at most once.
+#[derive(Debug, Clone)]
+pub struct SampleExec {
+    input: Arc<dyn ExecutionPlan>,
+    lower_bound: f64,
+    upper_bound: f64,
+    seed: u64,
+    metrics: ExecutionPlanMetricsSet,
+    cache: PlanProperties,
+}
+
+impl SampleExec {
+    /// Create a new SampleExec with Bernoulli sampling (without replacement).
+    ///
+    /// # Arguments
+    /// * `input` - The input execution plan
+    /// * `lower_bound` - Lower bound of sampling range (typically 0.0)
+    /// * `upper_bound` - Upper bound of sampling range (0.0 to 1.0)
+    /// * `seed` - Random seed for reproducible sampling
+    pub fn try_new(
+        input: Arc<dyn ExecutionPlan>,
+        lower_bound: f64,
+        upper_bound: f64,
+        seed: u64,
+    ) -> Result<Self> {
+        if lower_bound < 0.0 || upper_bound > 1.0 || lower_bound > upper_bound {
+            return internal_err!(
+                "Sampling bounds must satisfy 0.0 <= lower <= upper <= 1.0, got [{}, {}]",
+                lower_bound,
+                upper_bound
+            );
+        }
+
+        let cache = PlanProperties::new(
+            EquivalenceProperties::new(input.schema()),
+            input.properties().partitioning.clone(),
+            input.properties().emission_type,
+            input.properties().boundedness,
+        );
+
+        Ok(Self {
+            input,
+            lower_bound,
+            upper_bound,
+            seed,
+            metrics: ExecutionPlanMetricsSet::new(),
+            cache,
+        })
+    }
+
+    /// Create a sampler for the given partition.
+    fn create_sampler(&self, partition: usize) -> BernoulliSampler {
+        let seed = self.seed.wrapping_add(partition as u64);
+        BernoulliSampler::new(self.lower_bound, self.upper_bound, seed)
+    }
+}
+
+impl DisplayAs for SampleExec {
+    fn fmt_as(&self, _t: DisplayFormatType, f: &mut Formatter) -> fmt::Result {
+        write!(
+            f,
+            "SampleExec: bounds=[{}, {}], seed={}",
+            self.lower_bound, self.upper_bound, self.seed
+        )
+    }
+}
+
+impl ExecutionPlan for SampleExec {
+    fn name(&self) -> &'static str {
+        "SampleExec"
+    }
+
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn properties(&self) -> &PlanProperties {
+        &self.cache
+    }
+
+    fn maintains_input_order(&self) -> Vec<bool> {
+        // Sampling preserves row order (rows are filtered, not reordered)
+        vec![true]
+    }
+
+    fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
+        vec![&self.input]
+    }
+
+    fn with_new_children(
+        self: Arc<Self>,
+        mut children: Vec<Arc<dyn ExecutionPlan>>,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
+        Ok(Arc::new(Self::try_new(
+            children.swap_remove(0),
+            self.lower_bound,
+            self.upper_bound,
+            self.seed,
+        )?))
+    }
+
+    fn execute(
+        &self,
+        partition: usize,
+        context: Arc<TaskContext>,
+    ) -> Result<SendableRecordBatchStream> {
+        Ok(Box::pin(SampleStream {
+            input: self.input.execute(partition, context)?,
+            sampler: self.create_sampler(partition),
+            metrics: BaselineMetrics::new(&self.metrics, partition),
+        }))
+    }
+
+    fn metrics(&self) -> Option<MetricsSet> {
+        Some(self.metrics.clone_inner())
+    }
+
+    fn partition_statistics(&self, partition: Option<usize>) -> Result<Statistics> {
+        let mut stats = self.input.partition_statistics(partition)?;
+        let ratio = self.upper_bound - self.lower_bound;
+
+        // Scale statistics by sampling ratio (inexact due to randomness)
+        stats.num_rows = stats
+            .num_rows
+            .map(|n| (n as f64 * ratio) as usize)
+            .to_inexact();
+        stats.total_byte_size = stats
+            .total_byte_size
+            .map(|n| (n as f64 * ratio) as usize)
+            .to_inexact();
+
+        Ok(stats)
+    }
+}
+
+/// Bernoulli sampler: includes each row with probability `(upper - lower)`.
+/// This is sampling **without replacement** - each row appears at most once.
+struct BernoulliSampler {
+    lower_bound: f64,
+    upper_bound: f64,
+    rng: StdRng,
+}
+
+impl BernoulliSampler {
+    fn new(lower_bound: f64, upper_bound: f64, seed: u64) -> Self {
+        Self {
+            lower_bound,
+            upper_bound,
+            rng: StdRng::seed_from_u64(seed),
+        }
+    }
+
+    fn sample(&mut self, batch: &RecordBatch) -> Result<RecordBatch> {
+        let range = self.upper_bound - self.lower_bound;
+        if range <= 0.0 {
+            return Ok(RecordBatch::new_empty(batch.schema()));
+        }
+
+        // Select rows where random value falls in [lower, upper)
+        let indices: Vec<u32> = (0..batch.num_rows())
+            .filter(|_| {
+                let r: f64 = self.rng.random();
+                r >= self.lower_bound && r < self.upper_bound
+            })
+            .map(|i| i as u32)
+            .collect();
+
+        if indices.is_empty() {
+            return Ok(RecordBatch::new_empty(batch.schema()));
+        }
+
+        compute::take_record_batch(batch, &UInt32Array::from(indices))
+            .map_err(DataFusionError::from)
+    }
+}
+
+/// Stream adapter that applies sampling to each batch.
+struct SampleStream {
+    input: SendableRecordBatchStream,
+    sampler: BernoulliSampler,
+    metrics: BaselineMetrics,
+}
+
+impl Stream for SampleStream {
+    type Item = Result<RecordBatch>;
+
+    fn poll_next(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Option<Self::Item>> {
+        match ready!(self.input.poll_next_unpin(cx)) {
+            Some(Ok(batch)) => {
+                let elapsed = self.metrics.elapsed_compute().clone();
+                let _timer = elapsed.timer();
+                let result = self.sampler.sample(&batch);
+                Poll::Ready(Some(result.record_output(&self.metrics)))
+            }
+            Some(Err(e)) => Poll::Ready(Some(Err(e))),
+            None => Poll::Ready(None),
+        }
+    }
+}
+
+impl RecordBatchStream for SampleStream {
+    fn schema(&self) -> SchemaRef {
+        self.input.schema()
+    }
+}
diff --git a/datafusion/core/src/execution/context/mod.rs b/datafusion/core/src/execution/context/mod.rs
index 55f92b747a3f..a29313254283 100644
--- a/datafusion/core/src/execution/context/mod.rs
+++ b/datafusion/core/src/execution/context/mod.rs
@@ -80,6 +80,8 @@ use datafusion_execution::disk_manager::{
 use datafusion_execution::registry::SerializerRegistry;
 pub use datafusion_execution::TaskContext;
 pub use datafusion_expr::execution_props::ExecutionProps;
+#[cfg(feature = "sql")]
+use datafusion_expr::planner::RelationPlanner;
 use datafusion_expr::simplify::SimplifyContext;
 use datafusion_expr::{
     expr_rewriter::FunctionRewrite,
@@ -1436,6 +1438,18 @@ impl SessionContext {
         self.state.write().register_udwf(Arc::new(f)).ok();
     }
 
+    #[cfg(feature = "sql")]
+    /// Registers a [`RelationPlanner`] to customize SQL table-factor planning.
+    ///
+    /// Planners are invoked in reverse registration order, allowing newer
+    /// planners to take precedence over existing ones.
+    pub fn register_relation_planner(
+        &self,
+        planner: Arc<dyn RelationPlanner>,
+    ) -> Result<()> {
+        self.state.write().register_relation_planner(planner)
+    }
+
     /// Deregisters a UDF within this context.
     pub fn deregister_udf(&self, name: &str) {
         self.state.write().deregister_udf(name).ok();
diff --git a/datafusion/core/src/execution/session_state.rs b/datafusion/core/src/execution/session_state.rs
index c1bf6708708f..2466f9544bd1 100644
--- a/datafusion/core/src/execution/session_state.rs
+++ b/datafusion/core/src/execution/session_state.rs
@@ -53,7 +53,7 @@ use datafusion_expr::execution_props::ExecutionProps;
 use datafusion_expr::expr_rewriter::FunctionRewrite;
 use datafusion_expr::planner::ExprPlanner;
 #[cfg(feature = "sql")]
-use datafusion_expr::planner::TypePlanner;
+use datafusion_expr::planner::{RelationPlanner, TypePlanner};
 use datafusion_expr::registry::{FunctionRegistry, SerializerRegistry};
 use datafusion_expr::simplify::SimplifyInfo;
 #[cfg(feature = "sql")]
@@ -139,6 +139,8 @@ pub struct SessionState {
     analyzer: Analyzer,
     /// Provides support for customizing the SQL planner, e.g. to add support for custom operators like `->>` or `?`
     expr_planners: Vec<Arc<dyn ExprPlanner>>,
+    #[cfg(feature = "sql")]
+    relation_planners: Vec<Arc<dyn RelationPlanner>>,
     /// Provides support for customizing the SQL type planning
     #[cfg(feature = "sql")]
     type_planner: Option<Arc<dyn TypePlanner>>,
@@ -210,6 +212,9 @@ impl Debug for SessionState {
             .field("cache_factory", &self.cache_factory)
             .field("expr_planners", &self.expr_planners);
 
+        #[cfg(feature = "sql")]
+        let ret = ret.field("relation_planners", &self.relation_planners);
+
         #[cfg(feature = "sql")]
         let ret = ret.field("type_planner", &self.type_planner);
 
@@ -600,6 +605,24 @@ impl SessionState {
         &self.expr_planners
     }
 
+    #[cfg(feature = "sql")]
+    /// Returns the registered relation planners in priority order.
+    pub fn relation_planners(&self) -> &[Arc<dyn RelationPlanner>] {
+        &self.relation_planners
+    }
+
+    #[cfg(feature = "sql")]
+    /// Registers a [`RelationPlanner`] to customize SQL relation planning.
+    ///
+    /// Newly registered planners are given higher priority than existing ones.
+    pub fn register_relation_planner(
+        &mut self,
+        planner: Arc<dyn RelationPlanner>,
+    ) -> datafusion_common::Result<()> {
+        self.relation_planners.insert(0, planner);
+        Ok(())
+    }
+
     /// Returns the [`QueryPlanner`] for this session
     pub fn query_planner(&self) -> &Arc<dyn QueryPlanner + Send + Sync> {
         &self.query_planner
@@ -943,6 +966,8 @@ pub struct SessionStateBuilder {
     analyzer: Option<Analyzer>,
     expr_planners: Option<Vec<Arc<dyn ExprPlanner>>>,
     #[cfg(feature = "sql")]
+    relation_planners: Option<Vec<Arc<dyn RelationPlanner>>>,
+    #[cfg(feature = "sql")]
     type_planner: Option<Arc<dyn TypePlanner>>,
     optimizer: Option<Optimizer>,
     physical_optimizers: Option<PhysicalOptimizer>,
@@ -981,6 +1006,8 @@ impl SessionStateBuilder {
             analyzer: None,
             expr_planners: None,
             #[cfg(feature = "sql")]
+            relation_planners: None,
+            #[cfg(feature = "sql")]
             type_planner: None,
             optimizer: None,
             physical_optimizers: None,
@@ -1032,6 +1059,8 @@ impl SessionStateBuilder {
             analyzer: Some(existing.analyzer),
             expr_planners: Some(existing.expr_planners),
             #[cfg(feature = "sql")]
+            relation_planners: Some(existing.relation_planners),
+            #[cfg(feature = "sql")]
             type_planner: existing.type_planner,
             optimizer: Some(existing.optimizer),
             physical_optimizers: Some(existing.physical_optimizers),
@@ -1172,6 +1201,16 @@ impl SessionStateBuilder {
         self
     }
 
+    #[cfg(feature = "sql")]
+    /// Sets the [`RelationPlanner`]s used to customize SQL relation planning.
+    pub fn with_relation_planners(
+        mut self,
+        relation_planners: Vec<Arc<dyn RelationPlanner>>,
+    ) -> Self {
+        self.relation_planners = Some(relation_planners);
+        self
+    }
+
     /// Set the [`TypePlanner`] used to customize the behavior of the SQL planner.
     #[cfg(feature = "sql")]
     pub fn with_type_planner(mut self, type_planner: Arc<dyn TypePlanner>) -> Self {
@@ -1395,6 +1434,8 @@ impl SessionStateBuilder {
             analyzer,
             expr_planners,
             #[cfg(feature = "sql")]
+            relation_planners,
+            #[cfg(feature = "sql")]
             type_planner,
             optimizer,
             physical_optimizers,
@@ -1426,6 +1467,8 @@ impl SessionStateBuilder {
             analyzer: analyzer.unwrap_or_default(),
             expr_planners: expr_planners.unwrap_or_default(),
             #[cfg(feature = "sql")]
+            relation_planners: relation_planners.unwrap_or_default(),
+            #[cfg(feature = "sql")]
             type_planner,
             optimizer: optimizer.unwrap_or_default(),
             physical_optimizers: physical_optimizers.unwrap_or_default(),
@@ -1563,6 +1606,12 @@ impl SessionStateBuilder {
         &mut self.expr_planners
     }
 
+    #[cfg(feature = "sql")]
+    /// Returns a mutable reference to the current [`RelationPlanner`] list.
+    pub fn relation_planners(&mut self) -> &mut Option<Vec<Arc<dyn RelationPlanner>>> {
+        &mut self.relation_planners
+    }
+
     /// Returns the current type_planner value
     #[cfg(feature = "sql")]
     pub fn type_planner(&mut self) -> &mut Option<Arc<dyn TypePlanner>> {
@@ -1743,6 +1792,10 @@ impl ContextProvider for SessionContextProvider<'_> {
         self.state.expr_planners()
     }
 
+    fn get_relation_planners(&self) -> &[Arc<dyn RelationPlanner>] {
+        self.state.relation_planners()
+    }
+
     fn get_type_planner(&self) -> Option<Arc<dyn TypePlanner>> {
         if let Some(type_planner) = &self.state.type_planner {
             Some(Arc::clone(type_planner))
diff --git a/datafusion/core/tests/user_defined/mod.rs b/datafusion/core/tests/user_defined/mod.rs
index 515a82ab6c30..bc9949f5d681 100644
--- a/datafusion/core/tests/user_defined/mod.rs
+++ b/datafusion/core/tests/user_defined/mod.rs
@@ -36,5 +36,8 @@ mod user_defined_table_functions;
 /// Tests for Expression Planner
 mod expr_planner;
 
+/// Tests for Relation Planner extensions
+mod relation_planner;
+
 /// Tests for insert operations
 mod insert_operation;
diff --git a/datafusion/core/tests/user_defined/relation_planner.rs b/datafusion/core/tests/user_defined/relation_planner.rs
new file mode 100644
index 000000000000..dc06c783db7c
--- /dev/null
+++ b/datafusion/core/tests/user_defined/relation_planner.rs
@@ -0,0 +1,518 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Tests for the RelationPlanner extension point
+
+use std::sync::Arc;
+
+use arrow::array::{Int64Array, RecordBatch, StringArray};
+use arrow::datatypes::{DataType, Field, Schema};
+use datafusion::catalog::memory::MemTable;
+use datafusion::common::test_util::batches_to_string;
+use datafusion::prelude::*;
+use datafusion_common::{Result, ScalarValue};
+use datafusion_expr::logical_plan::builder::LogicalPlanBuilder;
+use datafusion_expr::planner::{
+    PlannedRelation, RelationPlanner, RelationPlannerContext, RelationPlanning,
+};
+use datafusion_expr::Expr;
+use datafusion_sql::sqlparser::ast::TableFactor;
+use insta::assert_snapshot;
+
+// ============================================================================
+// Test Planners - Example Implementations
+// ============================================================================
+
+// The planners in this section are deliberately minimal, static examples used
+// only for tests. In real applications a `RelationPlanner` would typically
+// construct richer logical plans tailored to external systems or custom
+// semantics rather than hard-coded in-memory tables.
+//
+// For more realistic examples, see `datafusion-examples/examples/relation_planner/`:
+// - `table_sample.rs`: Full TABLESAMPLE implementation (parsing → execution)
+// - `pivot_unpivot.rs`: PIVOT/UNPIVOT via SQL rewriting
+// - `match_recognize.rs`: MATCH_RECOGNIZE logical planning
+
+/// Helper to build simple static values-backed virtual tables used by the
+/// example planners below.
+fn plan_static_values_table(
+    relation: TableFactor,
+    table_name: &str,
+    column_name: &str,
+    values: Vec<ScalarValue>,
+) -> Result<RelationPlanning> {
+    match relation {
+        TableFactor::Table { name, alias, .. }
+            if name.to_string().eq_ignore_ascii_case(table_name) =>
+        {
+            let rows = values
+                .into_iter()
+                .map(|v| vec![Expr::Literal(v, None)])
+                .collect::<Vec<_>>();
+
+            let plan = LogicalPlanBuilder::values(rows)?
+                .project(vec![col("column1").alias(column_name)])?
+                .build()?;
+
+            Ok(RelationPlanning::Planned(PlannedRelation::new(plan, alias)))
+        }
+        other => Ok(RelationPlanning::Original(other)),
+    }
+}
+
+/// Example planner that provides a virtual `numbers` table with values
+/// 1, 2, 3.
+#[derive(Debug)]
+struct NumbersPlanner;
+
+impl RelationPlanner for NumbersPlanner {
+    fn plan_relation(
+        &self,
+        relation: TableFactor,
+        _context: &mut dyn RelationPlannerContext,
+    ) -> Result<RelationPlanning> {
+        plan_static_values_table(
+            relation,
+            "numbers",
+            "number",
+            vec![
+                ScalarValue::Int64(Some(1)),
+                ScalarValue::Int64(Some(2)),
+                ScalarValue::Int64(Some(3)),
+            ],
+        )
+    }
+}
+
+/// Example planner that provides a virtual `colors` table with three string
+/// values: `red`, `green`, `blue`.
+#[derive(Debug)]
+struct ColorsPlanner;
+
+impl RelationPlanner for ColorsPlanner {
+    fn plan_relation(
+        &self,
+        relation: TableFactor,
+        _context: &mut dyn RelationPlannerContext,
+    ) -> Result<RelationPlanning> {
+        plan_static_values_table(
+            relation,
+            "colors",
+            "color",
+            vec![
+                ScalarValue::Utf8(Some("red".into())),
+                ScalarValue::Utf8(Some("green".into())),
+                ScalarValue::Utf8(Some("blue".into())),
+            ],
+        )
+    }
+}
+
+/// Alternative implementation of `numbers` (returns 100, 200) used to
+/// demonstrate planner precedence (last registered planner wins).
+#[derive(Debug)]
+struct AlternativeNumbersPlanner;
+
+impl RelationPlanner for AlternativeNumbersPlanner {
+    fn plan_relation(
+        &self,
+        relation: TableFactor,
+        _context: &mut dyn RelationPlannerContext,
+    ) -> Result<RelationPlanning> {
+        plan_static_values_table(
+            relation,
+            "numbers",
+            "number",
+            vec![ScalarValue::Int64(Some(100)), ScalarValue::Int64(Some(200))],
+        )
+    }
+}
+
+/// Example planner that intercepts nested joins and samples both sides (limit 2)
+/// before joining, demonstrating recursive planning with `context.plan()`.
+#[derive(Debug)]
+struct SamplingJoinPlanner;
+
+impl RelationPlanner for SamplingJoinPlanner {
+    fn plan_relation(
+        &self,
+        relation: TableFactor,
+        context: &mut dyn RelationPlannerContext,
+    ) -> Result<RelationPlanning> {
+        match relation {
+            TableFactor::NestedJoin {
+                table_with_joins,
+                alias,
+                ..
+            } if table_with_joins.joins.len() == 1 => {
+                // Use context.plan() to recursively plan both sides
+                // This ensures other planners (like NumbersPlanner) can handle them
+                let left = context.plan(table_with_joins.relation.clone())?;
+                let right = context.plan(table_with_joins.joins[0].relation.clone())?;
+
+                // Sample each table to 2 rows
+                let left_sampled =
+                    LogicalPlanBuilder::from(left).limit(0, Some(2))?.build()?;
+
+                let right_sampled =
+                    LogicalPlanBuilder::from(right).limit(0, Some(2))?.build()?;
+
+                // Cross join: 2 rows × 2 rows = 4 rows (instead of 3×3=9 without sampling)
+                let plan = LogicalPlanBuilder::from(left_sampled)
+                    .cross_join(right_sampled)?
+                    .build()?;
+
+                Ok(RelationPlanning::Planned(PlannedRelation::new(plan, alias)))
+            }
+            other => Ok(RelationPlanning::Original(other)),
+        }
+    }
+}
+
+/// Example planner that never handles any relation and always delegates by
+/// returning `RelationPlanning::Original`.
+#[derive(Debug)]
+struct PassThroughPlanner;
+
+impl RelationPlanner for PassThroughPlanner {
+    fn plan_relation(
+        &self,
+        relation: TableFactor,
+        _context: &mut dyn RelationPlannerContext,
+    ) -> Result<RelationPlanning> {
+        // Never handles anything - always delegates
+        Ok(RelationPlanning::Original(relation))
+    }
+}
+
+/// Example planner that shows how planners can block specific constructs and
+/// surface custom error messages by rejecting `UNNEST` relations (here framed
+/// as a mock premium feature check).
+#[derive(Debug)]
+struct PremiumFeaturePlanner;
+
+impl RelationPlanner for PremiumFeaturePlanner {
+    fn plan_relation(
+        &self,
+        relation: TableFactor,
+        _context: &mut dyn RelationPlannerContext,
+    ) -> Result<RelationPlanning> {
+        match relation {
+            TableFactor::UNNEST { .. } => Err(datafusion_common::DataFusionError::Plan(
+                "UNNEST is a premium feature! Please upgrade to DataFusion Pro™ \
+                     to unlock advanced array operations."
+                    .to_string(),
+            )),
+            other => Ok(RelationPlanning::Original(other)),
+        }
+    }
+}
+
+// ============================================================================
+// Test Helpers - SQL Execution
+// ============================================================================
+
+/// Execute SQL and return results with better error messages.
+async fn execute_sql(ctx: &SessionContext, sql: &str) -> Result<Vec<RecordBatch>> {
+    let df = ctx.sql(sql).await?;
+    df.collect().await
+}
+
+/// Execute SQL and convert to string format for snapshot comparison.
+async fn execute_sql_to_string(ctx: &SessionContext, sql: &str) -> String {
+    let batches = execute_sql(ctx, sql)
+        .await
+        .expect("SQL execution should succeed");
+    batches_to_string(&batches)
+}
+
+// ============================================================================
+// Test Helpers - Context Builders
+// ============================================================================
+
+/// Create a SessionContext with a catalog table containing Int64 and Utf8 columns.
+///
+/// Creates a table with the specified name and sample data for fallback/integration tests.
+fn create_context_with_catalog_table(
+    table_name: &str,
+    id_values: Vec<i64>,
+    name_values: Vec<&str>,
+) -> SessionContext {
+    let ctx = SessionContext::new();
+
+    let schema = Arc::new(Schema::new(vec![
+        Field::new("id", DataType::Int64, false),
+        Field::new("name", DataType::Utf8, false),
+    ]));
+
+    let batch = RecordBatch::try_new(
+        schema.clone(),
+        vec![
+            Arc::new(Int64Array::from(id_values)),
+            Arc::new(StringArray::from(name_values)),
+        ],
+    )
+    .unwrap();
+
+    let table = MemTable::try_new(schema, vec![vec![batch]]).unwrap();
+    ctx.register_table(table_name, Arc::new(table)).unwrap();
+
+    ctx
+}
+
+/// Create a SessionContext with a simple single-column Int64 table.
+///
+/// Useful for basic tests that need a real catalog table.
+fn create_context_with_simple_table(
+    table_name: &str,
+    values: Vec<i64>,
+) -> SessionContext {
+    let ctx = SessionContext::new();
+
+    let schema = Arc::new(Schema::new(vec![Field::new(
+        "value",
+        DataType::Int64,
+        true,
+    )]));
+
+    let batch =
+        RecordBatch::try_new(schema.clone(), vec![Arc::new(Int64Array::from(values))])
+            .unwrap();
+
+    let table = MemTable::try_new(schema, vec![vec![batch]]).unwrap();
+    ctx.register_table(table_name, Arc::new(table)).unwrap();
+
+    ctx
+}
+
+// ============================================================================
+// TESTS: Ordered from Basic to Complex
+// ============================================================================
+
+/// Comprehensive test suite for RelationPlanner extension point.
+/// Tests are ordered from simplest smoke test to most complex scenarios.
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Small extension trait to make test setup read fluently.
+    trait TestSessionExt {
+        fn with_planner<P: RelationPlanner + 'static>(self, planner: P) -> Self;
+    }
+
+    impl TestSessionExt for SessionContext {
+        fn with_planner<P: RelationPlanner + 'static>(self, planner: P) -> Self {
+            self.register_relation_planner(Arc::new(planner)).unwrap();
+            self
+        }
+    }
+
+    /// Session context with only the `NumbersPlanner` registered.
+    fn ctx_with_numbers() -> SessionContext {
+        SessionContext::new().with_planner(NumbersPlanner)
+    }
+
+    /// Session context with virtual tables (`numbers`, `colors`) and the
+    /// `SamplingJoinPlanner` registered for nested joins.
+    fn ctx_with_virtual_tables_and_sampling() -> SessionContext {
+        SessionContext::new()
+            .with_planner(NumbersPlanner)
+            .with_planner(ColorsPlanner)
+            .with_planner(SamplingJoinPlanner)
+    }
+
+    // Basic smoke test: virtual table can be queried like a regular table.
+    #[tokio::test]
+    async fn virtual_table_basic_select() {
+        let ctx = ctx_with_numbers();
+
+        let result = execute_sql_to_string(&ctx, "SELECT * FROM numbers").await;
+
+        assert_snapshot!(result, @r"
+        +--------+
+        | number |
+        +--------+
+        | 1      |
+        | 2      |
+        | 3      |
+        +--------+");
+    }
+
+    // Virtual table supports standard SQL operations (projection, filter, aggregation).
+    #[tokio::test]
+    async fn virtual_table_filters_and_aggregation() {
+        let ctx = ctx_with_numbers();
+
+        let filtered = execute_sql_to_string(
+            &ctx,
+            "SELECT number * 10 AS scaled FROM numbers WHERE number > 1",
+        )
+        .await;
+
+        assert_snapshot!(filtered, @r"
+        +--------+
+        | scaled |
+        +--------+
+        | 20     |
+        | 30     |
+        +--------+");
+
+        let aggregated = execute_sql_to_string(
+            &ctx,
+            "SELECT COUNT(*) as count, SUM(number) as total, AVG(number) as average \
+             FROM numbers",
+        )
+        .await;
+
+        assert_snapshot!(aggregated, @r"
+        +-------+-------+---------+
+        | count | total | average |
+        +-------+-------+---------+
+        | 3     | 6     | 2.0     |
+        +-------+-------+---------+");
+    }
+
+    // Multiple planners can coexist and each handles its own virtual table.
+    #[tokio::test]
+    async fn multiple_planners_virtual_tables() {
+        let ctx = SessionContext::new()
+            .with_planner(NumbersPlanner)
+            .with_planner(ColorsPlanner);
+
+        let result1 = execute_sql_to_string(&ctx, "SELECT * FROM numbers").await;
+        assert_snapshot!(result1, @r"
+        +--------+
+        | number |
+        +--------+
+        | 1      |
+        | 2      |
+        | 3      |
+        +--------+");
+
+        let result2 = execute_sql_to_string(&ctx, "SELECT * FROM colors").await;
+        assert_snapshot!(result2, @r"
+        +-------+
+        | color |
+        +-------+
+        | red   |
+        | green |
+        | blue  |
+        +-------+");
+    }
+
+    // Last registered planner for the same table name takes precedence (LIFO).
+    #[tokio::test]
+    async fn lifo_precedence_last_planner_wins() {
+        let ctx = SessionContext::new()
+            .with_planner(AlternativeNumbersPlanner)
+            .with_planner(NumbersPlanner);
+
+        let result = execute_sql_to_string(&ctx, "SELECT * FROM numbers").await;
+
+        // CustomValuesPlanner registered last, should win (returns 1,2,3 not 100,200)
+        assert_snapshot!(result, @r"
+        +--------+
+        | number |
+        +--------+
+        | 1      |
+        | 2      |
+        | 3      |
+        +--------+");
+    }
+
+    // Pass-through planner delegates to the catalog without changing behavior.
+    #[tokio::test]
+    async fn delegation_pass_through_to_catalog() {
+        let ctx = create_context_with_simple_table("real_table", vec![42])
+            .with_planner(PassThroughPlanner);
+
+        let result = execute_sql_to_string(&ctx, "SELECT * FROM real_table").await;
+
+        assert_snapshot!(result, @r"
+        +-------+
+        | value |
+        +-------+
+        | 42    |
+        +-------+");
+    }
+
+    // Catalog is used when no planner claims the relation.
+    #[tokio::test]
+    async fn catalog_fallback_when_no_planner() {
+        let ctx =
+            create_context_with_catalog_table("users", vec![1, 2], vec!["Alice", "Bob"])
+                .with_planner(NumbersPlanner);
+
+        let result = execute_sql_to_string(&ctx, "SELECT * FROM users ORDER BY id").await;
+
+        assert_snapshot!(result, @r"
+        +----+-------+
+        | id | name  |
+        +----+-------+
+        | 1  | Alice |
+        | 2  | Bob   |
+        +----+-------+");
+    }
+
+    // Planners can block specific constructs and surface custom error messages.
+    #[tokio::test]
+    async fn error_handling_premium_feature_blocking() {
+        // Verify UNNEST works without planner
+        let ctx_without_planner = SessionContext::new();
+        let result =
+            execute_sql(&ctx_without_planner, "SELECT * FROM UNNEST(ARRAY[1, 2, 3])")
+                .await
+                .expect("UNNEST should work by default");
+        assert_eq!(result.len(), 1);
+
+        // Same query with blocking planner registered
+        let ctx = SessionContext::new().with_planner(PremiumFeaturePlanner);
+
+        // Verify UNNEST is now rejected
+        let error = execute_sql(&ctx, "SELECT * FROM UNNEST(ARRAY[1, 2, 3])")
+            .await
+            .expect_err("UNNEST should be rejected");
+
+        let error_msg = error.to_string();
+        assert!(
+            error_msg.contains("premium feature") && error_msg.contains("DataFusion Pro"),
+            "Expected custom rejection message, got: {error_msg}"
+        );
+    }
+
+    // SamplingJoinPlanner recursively calls `context.plan()` on both sides of a
+    // nested join before sampling, exercising recursive relation planning.
+    #[tokio::test]
+    async fn recursive_planning_sampling_join() {
+        let ctx = ctx_with_virtual_tables_and_sampling();
+
+        let result =
+            execute_sql_to_string(&ctx, "SELECT * FROM (numbers JOIN colors ON true)")
+                .await;
+
+        // SamplingJoinPlanner limits each side to 2 rows: 2×2=4 (not 3×3=9)
+        assert_snapshot!(result, @r"
+        +--------+-------+
+        | number | color |
+        +--------+-------+
+        | 1      | red   |
+        | 1      | green |
+        | 2      | red   |
+        | 2      | green |
+        +--------+-------+");
+    }
+}
diff --git a/datafusion/expr/src/planner.rs b/datafusion/expr/src/planner.rs
index 794c394d11d4..d5311c21a117 100644
--- a/datafusion/expr/src/planner.rs
+++ b/datafusion/expr/src/planner.rs
@@ -21,6 +21,8 @@ use std::fmt::Debug;
 use std::sync::Arc;
 
 use crate::expr::NullTreatment;
+#[cfg(feature = "sql")]
+use crate::logical_plan::LogicalPlan;
 use crate::{
     AggregateUDF, Expr, GetFieldAccess, ScalarUDF, SortExpr, TableSource, WindowFrame,
     WindowFunctionDefinition, WindowUDF,
@@ -30,6 +32,8 @@ use datafusion_common::{
     DFSchema, Result, TableReference, config::ConfigOptions,
     file_options::file_type::FileType, not_impl_err,
 };
+#[cfg(feature = "sql")]
+use sqlparser::ast::{Expr as SQLExpr, Ident, ObjectName, TableAlias, TableFactor};
 
 /// Provides the `SQL` query planner meta-data about tables and
 /// functions referenced in SQL statements, without a direct dependency on the
@@ -83,6 +87,12 @@ pub trait ContextProvider {
         &[]
     }
 
+    /// Return [`RelationPlanner`] extensions for planning table factors
+    #[cfg(feature = "sql")]
+    fn get_relation_planners(&self) -> &[Arc<dyn RelationPlanner>] {
+        &[]
+    }
+
     /// Return [`TypePlanner`] extensions for planning data types
     #[cfg(feature = "sql")]
     fn get_type_planner(&self) -> Option<Arc<dyn TypePlanner>> {
@@ -324,6 +334,85 @@ pub enum PlannerResult<T> {
     Original(T),
 }
 
+/// Result of planning a relation with [`RelationPlanner`]
+#[cfg(feature = "sql")]
+#[derive(Debug, Clone)]
+pub struct PlannedRelation {
+    /// The logical plan for the relation
+    pub plan: LogicalPlan,
+    /// Optional table alias for the relation
+    pub alias: Option<TableAlias>,
+}
+
+#[cfg(feature = "sql")]
+impl PlannedRelation {
+    /// Create a new `PlannedRelation` with the given plan and alias
+    pub fn new(plan: LogicalPlan, alias: Option<TableAlias>) -> Self {
+        Self { plan, alias }
+    }
+}
+
+/// Result of attempting to plan a relation with extension planners
+#[cfg(feature = "sql")]
+#[derive(Debug)]
+pub enum RelationPlanning {
+    /// The relation was successfully planned by an extension planner
+    Planned(PlannedRelation),
+    /// No extension planner handled the relation, return it for default processing
+    Original(TableFactor),
+}
+
+/// Customize planning SQL table factors to [`LogicalPlan`]s.
+#[cfg(feature = "sql")]
+pub trait RelationPlanner: Debug + Send + Sync {
+    /// Plan a table factor into a [`LogicalPlan`].
+    ///
+    /// Returning [`RelationPlanning::Planned`] short-circuits further planning and uses the
+    /// provided plan. Returning [`RelationPlanning::Original`] allows the next registered planner,
+    /// or DataFusion's default logic, to handle the relation.
+    fn plan_relation(
+        &self,
+        relation: TableFactor,
+        context: &mut dyn RelationPlannerContext,
+    ) -> Result<RelationPlanning>;
+}
+
+/// Provides utilities for relation planners to interact with DataFusion's SQL
+/// planner.
+///
+/// This trait provides SQL planning utilities specific to relation planning,
+/// such as converting SQL expressions to logical expressions and normalizing
+/// identifiers. It uses composition to provide access to session context via
+/// [`ContextProvider`].
+#[cfg(feature = "sql")]
+pub trait RelationPlannerContext {
+    /// Provides access to the underlying context provider for reading session
+    /// configuration, accessing tables, functions, and other metadata.
+    fn context_provider(&self) -> &dyn ContextProvider;
+
+    /// Plans the specified relation through the full planner pipeline, starting
+    /// from the first registered relation planner.
+    fn plan(&mut self, relation: TableFactor) -> Result<LogicalPlan>;
+
+    /// Converts a SQL expression into a logical expression using the current
+    /// planner context.
+    fn sql_to_expr(&mut self, expr: SQLExpr, schema: &DFSchema) -> Result<Expr>;
+
+    /// Converts a SQL expression into a logical expression without DataFusion
+    /// rewrites.
+    fn sql_expr_to_logical_expr(
+        &mut self,
+        expr: SQLExpr,
+        schema: &DFSchema,
+    ) -> Result<Expr>;
+
+    /// Normalizes an identifier according to session settings.
+    fn normalize_ident(&self, ident: Ident) -> String;
+
+    /// Normalizes a SQL object name into a [`TableReference`].
+    fn object_name_to_table_reference(&self, name: ObjectName) -> Result<TableReference>;
+}
+
 /// Customize planning SQL types to DataFusion (Arrow) types.
 #[cfg(feature = "sql")]
 pub trait TypePlanner: Debug + Send + Sync {
diff --git a/datafusion/sql/src/relation/mod.rs b/datafusion/sql/src/relation/mod.rs
index 9dfa078701d3..6a06b2d52e7a 100644
--- a/datafusion/sql/src/relation/mod.rs
+++ b/datafusion/sql/src/relation/mod.rs
@@ -24,19 +24,122 @@ use datafusion_common::{
     not_impl_err, plan_err, DFSchema, Diagnostic, Result, Span, Spans, TableReference,
 };
 use datafusion_expr::builder::subquery_alias;
+use datafusion_expr::planner::{
+    PlannedRelation, RelationPlannerContext, RelationPlanning,
+};
 use datafusion_expr::{expr::Unnest, Expr, LogicalPlan, LogicalPlanBuilder};
 use datafusion_expr::{Subquery, SubqueryAlias};
 use sqlparser::ast::{FunctionArg, FunctionArgExpr, Spanned, TableFactor};
 
 mod join;
 
+struct SqlToRelRelationContext<'a, 'b, S: ContextProvider> {
+    planner: &'a SqlToRel<'b, S>,
+    planner_context: &'a mut PlannerContext,
+}
+
+// Implement RelationPlannerContext
+impl<'a, 'b, S: ContextProvider> RelationPlannerContext
+    for SqlToRelRelationContext<'a, 'b, S>
+{
+    fn context_provider(&self) -> &dyn ContextProvider {
+        self.planner.context_provider
+    }
+
+    fn plan(&mut self, relation: TableFactor) -> Result<LogicalPlan> {
+        self.planner.create_relation(relation, self.planner_context)
+    }
+
+    fn sql_to_expr(
+        &mut self,
+        expr: sqlparser::ast::Expr,
+        schema: &DFSchema,
+    ) -> Result<Expr> {
+        self.planner.sql_to_expr(expr, schema, self.planner_context)
+    }
+
+    fn sql_expr_to_logical_expr(
+        &mut self,
+        expr: sqlparser::ast::Expr,
+        schema: &DFSchema,
+    ) -> Result<Expr> {
+        self.planner
+            .sql_expr_to_logical_expr(expr, schema, self.planner_context)
+    }
+
+    fn normalize_ident(&self, ident: sqlparser::ast::Ident) -> String {
+        self.planner.ident_normalizer.normalize(ident)
+    }
+
+    fn object_name_to_table_reference(
+        &self,
+        name: sqlparser::ast::ObjectName,
+    ) -> Result<TableReference> {
+        self.planner.object_name_to_table_reference(name)
+    }
+}
+
 impl<S: ContextProvider> SqlToRel<'_, S> {
-    /// Create a `LogicalPlan` that scans the named relation
+    /// Create a `LogicalPlan` that scans the named relation.
+    ///
+    /// First tries any registered extension planners. If no extension handles
+    /// the relation, falls back to the default planner.
     fn create_relation(
         &self,
         relation: TableFactor,
         planner_context: &mut PlannerContext,
     ) -> Result<LogicalPlan> {
+        let planned_relation =
+            match self.create_extension_relation(relation, planner_context)? {
+                RelationPlanning::Planned(planned) => planned,
+                RelationPlanning::Original(original) => {
+                    self.create_default_relation(original, planner_context)?
+                }
+            };
+
+        let optimized_plan = optimize_subquery_sort(planned_relation.plan)?.data;
+        if let Some(alias) = planned_relation.alias {
+            self.apply_table_alias(optimized_plan, alias)
+        } else {
+            Ok(optimized_plan)
+        }
+    }
+
+    fn create_extension_relation(
+        &self,
+        relation: TableFactor,
+        planner_context: &mut PlannerContext,
+    ) -> Result<RelationPlanning> {
+        let planners = self.context_provider.get_relation_planners();
+        if planners.is_empty() {
+            return Ok(RelationPlanning::Original(relation));
+        }
+
+        let mut current_relation = relation;
+        for planner in planners.iter() {
+            let mut context = SqlToRelRelationContext {
+                planner: self,
+                planner_context,
+            };
+
+            match planner.plan_relation(current_relation, &mut context)? {
+                RelationPlanning::Planned(planned) => {
+                    return Ok(RelationPlanning::Planned(planned));
+                }
+                RelationPlanning::Original(original) => {
+                    current_relation = original;
+                }
+            }
+        }
+
+        Ok(RelationPlanning::Original(current_relation))
+    }
+
+    fn create_default_relation(
+        &self,
+        relation: TableFactor,
+        planner_context: &mut PlannerContext,
+    ) -> Result<PlannedRelation> {
         let relation_span = relation.span();
         let (plan, alias) = match relation {
             TableFactor::Table {
@@ -190,13 +293,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
                 );
             }
         };
-
-        let optimized_plan = optimize_subquery_sort(plan)?.data;
-        if let Some(alias) = alias {
-            self.apply_table_alias(optimized_plan, alias)
-        } else {
-            Ok(optimized_plan)
-        }
+        Ok(PlannedRelation::new(plan, alias))
     }
 
     pub(crate) fn create_relation_subquery(
