@@ -456,4 +456,75 @@ mod tests {
 
         Ok(())
     }
+
+    /// A [`ViewTable`] that is not inlined: each read plans the saved query in
+    /// `scan`.
+    ///
+    /// [`ViewTable`]: crate::datasource::ViewTable
+    #[derive(Debug)]
+    struct NotInlinedView(crate::datasource::ViewTable);
+
+    #[async_trait::async_trait]
+    impl crate::datasource::TableProvider for NotInlinedView {
+        fn schema(&self) -> arrow::datatypes::SchemaRef {
+            self.0.schema()
+        }
+
+        fn table_type(&self) -> crate::datasource::TableType {
+            self.0.table_type()
+        }
+
+        async fn scan(
+            &self,
+            state: &dyn crate::catalog::Session,
+            projection: Option<&[usize]>,
+            filters: &[datafusion_expr::Expr],
+            limit: Option<usize>,
+        ) -> Result<std::sync::Arc<dyn crate::physical_plan::ExecutionPlan>> {
+            self.0.scan(state, projection, filters, limit).await
+        }
+    }
+
+    #[tokio::test]
+    async fn query_not_inlined_view_with_materialized_cte_twice() -> Result<()> {
+        // sqlparser only parses `[NOT] MATERIALIZED` for the PostgreSQL dialect.
+        let session_ctx = SessionContext::new_with_config(
+            SessionConfig::new().set_str("datafusion.sql_parser.dialect", "PostgreSQL"),
+        );
+        session_ctx
+            .sql("CREATE TABLE t AS VALUES (1), (2)")
+            .await?
+            .collect()
+            .await?;
+        let plan = session_ctx
+            .sql(
+                "WITH r AS MATERIALIZED (SELECT column1 AS a FROM t) \
+                 SELECT a FROM r",
+            )
+            .await?
+            .into_optimized_plan()?;
+        let view = crate::datasource::ViewTable::new(plan, None);
+        session_ctx.register_table("w", std::sync::Arc::new(NotInlinedView(view)))?;
+
+        // Each read of `w` plans the saved query, which has the same CTE id
+        // each time.
+        let results = session_ctx
+            .sql("SELECT x.a AS x, y.a AS y FROM w x, w y ORDER BY x, y")
+            .await?
+            .collect()
+            .await?;
+
+        insta::assert_snapshot!(batches_to_string(&results),@r"
+        +---+---+
+        | x | y |
+        +---+---+
+        | 1 | 1 |
+        | 1 | 2 |
+        | 2 | 1 |
+        | 2 | 2 |
+        +---+---+
+        ");
+
+        Ok(())
+    }
 }
@@ -45,6 +45,9 @@ use crate::physical_plan::joins::{
     PartitionMode, SortMergeJoinExec,
 };
 use crate::physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
+use crate::physical_plan::materialized_cte::{
+    MaterializedCteBuffer, MaterializedCteExec, MaterializedCteScanExec,
+};
 use crate::physical_plan::projection::{ProjectionExec, ProjectionExpr};
 use crate::physical_plan::repartition::RepartitionExec;
 use crate::physical_plan::sorts::sort::SortExec;
@@ -230,14 +233,160 @@ impl DefaultPhysicalPlanner {
         {
             return Ok(plan);
         }
+        let renumbered = renumber_materialized_ctes(logical_plan)?;
+        let logical_plan = renumbered.as_ref().unwrap_or(logical_plan);
         let plan = self
             .create_initial_plan(logical_plan, session_state)
             .await?;
+        let plan = bind_materialized_cte_scans(plan)?;
 
         self.optimize_physical_plan(plan, session_state, |_, _| {})
     }
 }
 
+/// Give each occurrence of a [`MaterializedCte`] in `plan` a new id, and
+/// update the scans in its continuation, including the scans in subqueries.
+///
+/// The id is copied when a logical plan is cloned, so one CTE can occur more
+/// than once, for example when a view that declares it is referenced two
+/// times, or when a DataFrame is joined to itself. Each occurrence is a
+/// separate CTE, as in PostgreSQL.
+///
+/// The ids must also be new for each planning run, not only unique within
+/// `plan`. A table provider can plan a saved logical plan in its `scan`, as a
+/// [`ViewTable`] that is not inlined does. Each such scan is a separate
+/// planning run of the same saved plan, and its CTEs end up in the same
+/// physical plan as the CTEs of the outer run.
+///
+/// Returns `None` when `plan` has no materialized CTE.
+///
+/// [`MaterializedCte`]: datafusion_expr::MaterializedCte
+/// [`ViewTable`]: crate::datasource::ViewTable
+fn renumber_materialized_ctes(plan: &LogicalPlan) -> Result<Option<LogicalPlan>> {
+    fn as_cte(plan: &LogicalPlan) -> Option<&datafusion_expr::MaterializedCte> {
+        match plan {
+            LogicalPlan::Extension(Extension { node }) => node.as_any().downcast_ref(),
+            _ => None,
+        }
+    }
+
+    let mut has_cte = false;
+    plan.apply_with_subqueries(|node| {
+        if as_cte(node).is_some() {
+            has_cte = true;
+            return Ok(TreeNodeRecursion::Stop);
+        }
+        Ok(TreeNodeRecursion::Continue)
+    })?;
+    if !has_cte {
+        return Ok(None);
+    }
+
+    plan.clone()
+        .transform_down_with_subqueries(|node| {
+            let Some(cte) = as_cte(&node) else {
+                return Ok(Transformed::no(node));
+            };
+            let old_id = cte.id;
+            let new_id = datafusion_expr::MaterializedCteId::next();
+            let continuation = cte
+                .continuation
+                .clone()
+                .transform_down_with_subqueries(|node| {
+                    // A nested occurrence of the same CTE owns the scans below
+                    // it, and is renumbered on its own.
+                    if as_cte(&node).is_some_and(|nested| nested.id == old_id) {
+                        return Ok(Transformed::new(
+                            node,
+                            false,
+                            TreeNodeRecursion::Jump,
+                        ));
+                    }
+                    let LogicalPlan::Extension(Extension { node: scan }) = &node else {
+                        return Ok(Transformed::no(node));
+                    };
+                    match scan
+                        .as_any()
+                        .downcast_ref::<datafusion_expr::MaterializedCteScan>()
+                    {
+                        Some(scan) if scan.id == old_id => {
+                            let scan = datafusion_expr::MaterializedCteScan {
+                                id: new_id,
+                                ..scan.clone()
+                            };
+                            Ok(Transformed::yes(LogicalPlan::Extension(Extension {
+                                node: Arc::new(scan),
+                            })))
+                        }
+                        _ => Ok(Transformed::no(node)),
+                    }
+                })?
+                .data;
+            let cte = datafusion_expr::MaterializedCte {
+                id: new_id,
+                name: cte.name.clone(),
+                cte: cte.cte.clone(),
+                continuation,
+            };
+            Ok(Transformed::yes(LogicalPlan::Extension(Extension {
+                node: Arc::new(cte),
+            })))
+        })
+        .map(|t| Some(t.data))
+}
+
+/// Point every [`MaterializedCteScanExec`] at the buffer of the
+/// [`MaterializedCteExec`] with the same id.
+///
+/// [`renumber_materialized_ctes`] makes the ids unique before the physical
+/// plan is created.
+///
+/// A scan that is already bound was planned and bound by a separate planning
+/// run, such as the `scan` of a [`ViewTable`], and is left as is. Binding it
+/// again would count its partitions twice in the buffer.
+///
+/// Scans are planned before the node that owns their CTE, and a scan inside a
+/// scalar subquery is planned in a separate subtree, so the binding is done
+/// once on the whole initial plan.
+///
+/// [`ViewTable`]: crate::datasource::ViewTable
+fn bind_materialized_cte_scans(
+    plan: Arc<dyn ExecutionPlan>,
+) -> Result<Arc<dyn ExecutionPlan>> {
+    let mut buffers = HashMap::new();
+    plan.apply(|node| {
+        if let Some(cte) = node.downcast_ref::<MaterializedCteExec>()
+            && buffers
+                .insert(cte.buffer().id(), Arc::clone(cte.buffer()))
+                .is_some()
+        {
+            return internal_err!(
+                "Two MaterializedCteExec nodes have the id {}",
+                cte.buffer().id()
+            );
+        }
+        Ok(TreeNodeRecursion::Continue)
+    })?;
+    if buffers.is_empty() {
+        return Ok(plan);
+    }
+    plan.transform_up(|node| {
+        let Some(scan) = node.downcast_ref::<MaterializedCteScanExec>() else {
+            return Ok(Transformed::no(node));
+        };
+        if scan.is_bound() {
+            return Ok(Transformed::no(node));
+        }
+        let Some(buffer) = buffers.get(&scan.id()) else {
+            return internal_err!("MaterializedCteScanExec {} has no CTE", scan.id());
+        };
+        Ok(Transformed::yes(
+            Arc::new(scan.bind(Arc::clone(buffer))) as Arc<dyn ExecutionPlan>
+        ))
+    })
+    .map(|t| t.data)
+}
+
 #[derive(Debug)]
 struct ExecutionPlanChild {
     /// Index needed to order children of parent to ensure consistency with original
@@ -1904,6 +2053,58 @@ impl DefaultPhysicalPlanner {
 
             // N Children
             LogicalPlan::Union(_) => UnionExec::try_new(children.vec())?,
+            LogicalPlan::Extension(Extension { node })
+                if node.as_any().is::<datafusion_expr::MaterializedCte>() =>
+            {
+                let cte = node
+                    .as_any()
+                    .downcast_ref::<datafusion_expr::MaterializedCte>()
+                    .unwrap();
+                let [body, continuation] = children.two()?;
+                // A CTE that nothing reads is never evaluated.
+                let mut is_referenced = false;
+                cte.continuation.apply_with_subqueries(|node| {
+                    if let LogicalPlan::Extension(Extension { node }) = node
+                        && let Some(scan) =
+                            node.as_any()
+                                .downcast_ref::<datafusion_expr::MaterializedCteScan>()
+                        && scan.id == cte.id
+                    {
+                        is_referenced = true;
+                        return Ok(TreeNodeRecursion::Stop);
+                    }
+                    Ok(TreeNodeRecursion::Continue)
+                })?;
+                if !is_referenced {
+                    return Ok(continuation);
+                }
+                // The body is fully buffered before a scan yields a row, so an
+                // unbounded body would never produce output.
+                if body.boundedness().is_unbounded() {
+                    return plan_err!(
+                        "MATERIALIZED CTE {} reads an unbounded source, which cannot be materialized",
+                        cte.name
+                    );
+                }
+                let buffer =
+                    Arc::new(MaterializedCteBuffer::new(cte.id.as_u64(), &cte.name));
+                Arc::new(MaterializedCteExec::new(body, continuation, buffer))
+            }
+            LogicalPlan::Extension(Extension { node })
+                if node.as_any().is::<datafusion_expr::MaterializedCteScan>() =>
+            {
+                let scan = node
+                    .as_any()
+                    .downcast_ref::<datafusion_expr::MaterializedCteScan>()
+                    .unwrap();
+                // Unbound until `bind_materialized_cte_scans` runs on the whole plan.
+                Arc::new(MaterializedCteScanExec::new(
+                    scan.id.as_u64(),
+                    &scan.name,
+                    Arc::clone(scan.schema.inner()),
+                    session_state.config().target_partitions(),
+                ))
+            }
             LogicalPlan::Extension(Extension { node }) => {
                 let mut maybe_plan = None;
                 let children = children.vec();