connect, session: Spark Connect group 1 translator arms and sql_command (P3-T012a part 2b, PR-12; stacked on #772)

changelog.d/774.added.md

@@ -0,0 +1 @@
+Spark Connect translates local relations, declared file schemas, qualified subquery aliases, nested literals, expression strings and `spark.sql()` (sql_command), as part of group 1 (PR-12 part 2b, #774).

crates/krishiv-connect-server/CLAUDE.md

@@ -50,6 +50,9 @@ Paths are relative to this crate.
   operation's `Replay` under the ahead budget (`krishiv.connect.execute.aheadBytes`) and memory consumer
   `connect_responses` (the Arrow payload is `Bytes`, shared with the copy sent), the
   response order (schema, batches, metrics, `ResultComplete` when reattachable); commands are 56K00 until T012;
+  `request_plan` (shared with `AnalyzePlan`) translates twice when the plan holds local relations: the first pass records
+  them, `Session::local_relation` builds them (cached ones are `NOT_FOUND_CACHED_LOCAL_RELATION` until the artifact
+  store lands), the second pass resolves them (`krishiv-connect` CLAUDE.md);
   `src/execute/arrow.rs` Spark's Arrow shapes, batch limits, IPC streams, codecs and chunks;
   `src/execute/schema.rs` the `schema` response (`DataTypeProtoConverter.toConnectProtoType`)
 - `src/codec.rs` `WireCodec`: requests decode only through `wire::decode_request` (bounded size, count, depth, fields)

crates/krishiv-connect-server/src/analyze.rs

@@ -7,7 +7,7 @@
 //! - `schema`: the analyzed schema, the same `DataType` `ExecutePlan` sends;
 //! - `explain`: `DataFrame::explain` in the requested mode;
 //! - `tree_string`: Spark's `StructType.treeString` (`analyze::types`);
-//! - `is_local`: whether the translated root is an inline table (Spark's
+//! - `is_local`: whether the translated root is a local relation (Spark's
 //!   `LocalRelation`); `is_streaming`: always false (no streaming sources);
 //! - `input_files`: the paths of the file scans, as written;
 //! - `spark_version`: the session profile's release (`4.2.0` for 4.2);
@@ -68,13 +68,13 @@ pub async fn analyze_plan(
 /// recurse on the async worker's stack (translation runs a deep plan on a
 /// big stack of its own).
 async fn frame(
-    session: &krishiv_session::Session,
+    entry: &SessionEntry,
     plan: Option<&pb::Plan>,
     limits: &ExecuteLimits,
 ) -> Result<DataFrame, KrishivError> {
     let empty = pb::Plan::default();
-    let logical = request_plan(session, plan.unwrap_or(&empty), limits).await?;
-    Ok(session.from_plan(&logical))
+    let logical = request_plan(entry, plan.unwrap_or(&empty), limits).await?;
+    Ok(entry.session().from_plan(&logical))
 }
 
 async fn analyze(
@@ -91,45 +91,47 @@ async fn analyze(
             .into());
         }
         Some(Analyze::Schema(s)) => {
-            let df = frame(session, s.plan.as_ref(), limits).await?;
+            let df = frame(entry, s.plan.as_ref(), limits).await?;
             res::Result::Schema(res::Schema {
                 schema: Some(schema::proto_schema(&*df.schema().await?)?),
             })
         }
         Some(Analyze::Explain(e)) => {
             // Spark builds the Dataset (translation, analysis) before it
             // reads the mode.
-            let df = frame(session, e.plan.as_ref(), limits).await?;
+            let df = frame(entry, e.plan.as_ref(), limits).await?;
             df.analyzed().await?;
             let mode = explain_mode(e.explain_mode)?;
             res::Result::Explain(res::Explain {
                 explain_string: df.explain(mode).await?,
             })
         }
         Some(Analyze::TreeString(t)) => {
-            let df = frame(session, t.plan.as_ref(), limits).await?;
+            let df = frame(entry, t.plan.as_ref(), limits).await?;
             let spark = schema::spark_schema(&*df.schema().await?)?;
             res::Result::TreeString(res::TreeString {
                 tree_string: types::tree_string(&spark, t.level),
             })
         }
         Some(Analyze::IsLocal(l)) => {
-            // Spark: the analyzed plan is a `LocalRelation`. Krishiv's
-            // analyzer lowers inline rows to `OneRowRelation` (as for
-            // `SELECT 1`), so the question is asked of the translated plan
-            // (an inline table, as `LocalRelation` will translate in T012);
-            // SQL `VALUES` is SQL text until analysis and answers false.
-            // The plan is still analyzed so that an invalid one fails as in
-            // Spark.
+            // Spark: the analyzed plan is a `LocalRelation`. A `local_relation` translates to
+            // a `Read` of a local relation (its user schema already applied eagerly, so the
+            // analyzed plan is that bare relation either way); Krishiv's analyzer lowers inline
+            // rows to `OneRowRelation` (as for `SELECT 1`), so the question is asked of the
+            // translated plan; SQL `VALUES` is SQL text until analysis and answers false. The
+            // plan is still analyzed so that an invalid one fails as in Spark.
             let empty = pb::Plan::default();
-            let logical = request_plan(session, l.plan.as_ref().unwrap_or(&empty), limits).await?;
-            let is_local = matches!(logical.root().node, PlanNode::InlineTable { .. });
+            let logical = request_plan(entry, l.plan.as_ref().unwrap_or(&empty), limits).await?;
+            let is_local = matches!(
+                logical.root().node,
+                PlanNode::InlineTable { .. } | PlanNode::Read(ReadSource::Local(_))
+            );
             session.from_plan(&logical).analyzed().await?;
             res::Result::IsLocal(res::IsLocal { is_local })
         }
         Some(Analyze::IsStreaming(s)) => {
             // Analyzed so that an invalid plan fails as in Spark.
-            frame(session, s.plan.as_ref(), limits)
+            frame(entry, s.plan.as_ref(), limits)
                 .await?
                 .analyzed()
                 .await?;
@@ -138,7 +140,7 @@ async fn analyze(
             })
         }
         Some(Analyze::InputFiles(f)) => {
-            let analyzed = frame(session, f.plan.as_ref(), limits)
+            let analyzed = frame(entry, f.plan.as_ref(), limits)
                 .await?
                 .analyzed()
                 .await?;
@@ -157,15 +159,14 @@ async fn analyze(
             )?)?),
         }),
         Some(Analyze::SameSemantics(s)) => {
-            let target =
-                semantic_hash(frame(session, s.target_plan.as_ref(), limits).await?).await?;
-            let other = semantic_hash(frame(session, s.other_plan.as_ref(), limits).await?).await?;
+            let target = semantic_hash(frame(entry, s.target_plan.as_ref(), limits).await?).await?;
+            let other = semantic_hash(frame(entry, s.other_plan.as_ref(), limits).await?).await?;
             res::Result::SameSemantics(res::SameSemantics {
                 result: target == other,
             })
         }
         Some(Analyze::SemanticHash(s)) => {
-            let hash = semantic_hash(frame(session, s.plan.as_ref(), limits).await?).await?;
+            let hash = semantic_hash(frame(entry, s.plan.as_ref(), limits).await?).await?;
             res::Result::SemanticHash(res::SemanticHash {
                 result: fold_hash(hash),
             })

crates/krishiv-connect-server/src/analyze/types.rs

@@ -183,28 +183,7 @@ pub fn json_to_ddl(json: &str) -> Result<String, KrishivError> {
     }
 }
 
-/// The planner's `parseDatatypeString` (`SparkConnectPlanner.scala:1511-1528`):
-/// a table schema (`a INT, b STRING`), else a data type (`int`), else the
-/// text as the fields of a struct; the first failure is the error.
-///
-/// Krishiv parses the table schema as the fields of a `STRUCT<...>`, so
-/// clauses only a table schema takes (`DEFAULT`, `GENERATED`) are refused,
-/// and the error is the data type's (Spark reports the table schema's): a
-/// recorded difference.
-///
-/// # Errors
-/// The parse error of the text as a data type.
-pub fn parse_datatype_string(text: &str) -> Result<DataType, KrishivError> {
-    if let Ok(t) = krishiv_types::parse_data_type(&format!("STRUCT<{text}>")) {
-        return Ok(t);
-    }
-    match krishiv_types::parse_data_type(text) {
-        Ok(t) => Ok(t),
-        Err(e) => {
-            krishiv_types::parse_data_type(&format!("STRUCT<{}>", text.trim())).map_err(|_| e)
-        }
-    }
-}
+pub use krishiv_connect::translate::parse_datatype_string;
 
 #[cfg(test)]
 mod tests;