feat(substrait): replace SessionContext with a trait (apache#13343)

* feat(substrait): replace SessionContext with SessionState

* feat(substrait): add logical plan context

* chore(substrait): add apache header

* docs: fix code in docs

* docs(substrait): rename and document context

* chore(substrait): context -> state

* chore: fmt

datafusion/core/src/execution/session_state.rs

@@ -296,7 +296,9 @@ impl SessionState {
             .resolve(&catalog.default_catalog, &catalog.default_schema)
     }
 
-    pub(crate) fn schema_for_ref(
+    /// Retrieve the [`SchemaProvider`] for a specific [`TableReference`], if it
+    /// esists.
+    pub fn schema_for_ref(
         &self,
         table_ref: impl Into<TableReference>,
     ) -> datafusion_common::Result<Arc<dyn SchemaProvider>> {

datafusion/substrait/Cargo.toml

@@ -34,6 +34,7 @@ workspace = true
 [dependencies]
 arrow-buffer = { workspace = true }
 async-recursion = "1.0"
+async-trait = { workspace = true }
 chrono = { workspace = true }
 datafusion = { workspace = true, default-features = true }
 itertools = { workspace = true }

datafusion/substrait/src/lib.rs

@@ -64,10 +64,10 @@
 //!  let plan = df.into_optimized_plan()?;
 //!
 //!  // Convert the plan into a substrait (protobuf) Plan
-//!  let substrait_plan = logical_plan::producer::to_substrait_plan(&plan, &ctx)?;
+//!  let substrait_plan = logical_plan::producer::to_substrait_plan(&plan, &ctx.state())?;
 //!
 //!  // Receive a substrait protobuf from somewhere, and turn it into a LogicalPlan
-//!  let logical_round_trip = logical_plan::consumer::from_substrait_plan(&ctx, &substrait_plan).await?;
+//!  let logical_round_trip = logical_plan::consumer::from_substrait_plan(&ctx.state(), &substrait_plan).await?;
 //!  let logical_round_trip = ctx.state().optimize(&logical_round_trip)?;
 //!  assert_eq!(format!("{:?}", plan), format!("{:?}", logical_round_trip));
 //! # Ok(())

datafusion/substrait/src/logical_plan/mod.rs

@@ -17,3 +17,4 @@
 
 pub mod consumer;
 pub mod producer;
+pub mod state;