substrait_explain/parser/

structural.rs

//! Parser for the structural part of the Substrait file format.
//!
//! This is the overall parser for parsing the text format. It is responsible
//! for tracking which section of the file we are currently parsing, and parsing
//! each line separately.

use std::fmt;

use substrait::proto::extensions::AdvancedExtension;
use substrait::proto::rel::RelType;
use substrait::proto::rel_common::EmitKind;
use substrait::proto::{
    AggregateRel, FetchRel, FilterRel, JoinRel, Plan, PlanRel, ProjectRel, ReadRel, Rel, RelRoot,
    SortRel, plan_rel,
};

use crate::extensions::{ExtensionRegistry, ExtensionType, SimpleExtensions, simple};
use crate::parser::common::{MessageParseError, ParsePair};
use crate::parser::errors::{ParseContext, ParseError, ParseResult};
use crate::parser::expressions::Name;
use crate::parser::extensions::{
    AdvExtInvocation, ExtensionInvocation, ExtensionParseError, ExtensionParser,
};
use crate::parser::relations::RelationParsingContext;
use crate::parser::{ErrorKind, ExpressionParser, RelationParsePair, Rule, unwrap_single_pair};

pub const PLAN_HEADER: &str = "=== Plan";

/// Represents an input line, trimmed of leading two-space indents and final
/// whitespace. Contains the number of indents and the trimmed line.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct IndentedLine<'a>(pub usize, pub &'a str);

impl<'a> From<&'a str> for IndentedLine<'a> {
    fn from(line: &'a str) -> Self {
        let line = line.trim_end();
        let mut spaces = 0;
        for c in line.chars() {
            if c == ' ' {
                spaces += 1;
            } else {
                break;
            }
        }

        let indents = spaces / 2;

        let (_, trimmed) = line.split_at(indents * 2);

        IndentedLine(indents, trimmed)
    }
}

/// An advanced-extension annotation (`+ Enh:Name[args]` or `+ Opt:Name[args]`)
/// that is attached to a relation node. The `pair` holds the `adv_extension`
/// grammar rule directly (already unwrapped from the outer `planNode`).
#[derive(Debug, Clone)]
pub struct AdvExt<'a> {
    pub pair: pest::iterators::Pair<'a, Rule>, // Rule::adv_extension
    pub line_no: i64,
}

/// A relation node in the plan tree, before conversion to a Substrait proto.
#[derive(Debug, Clone)]
pub struct RelationNode<'a> {
    pub pair: pest::iterators::Pair<'a, Rule>,
    pub line_no: i64,
    pub adv_extensions: Vec<AdvExt<'a>>,
    pub children: Vec<RelationNode<'a>>,
}

impl<'a> RelationNode<'a> {
    pub fn context(&self) -> ParseContext {
        ParseContext {
            line_no: self.line_no,
            line: self.pair.as_str().to_string(),
        }
    }
}

/// A parsed plan line: either a relation or an advanced-extension annotation.
///
/// Classification happens at construction time by inspecting the inner grammar
/// rule, so downstream code can use standard Rust pattern matching rather than
/// runtime rule inspection.
#[derive(Debug, Clone)]
pub enum LineNode<'a> {
    Relation(RelationNode<'a>),
    AdvExt(AdvExt<'a>),
}

impl<'a> LineNode<'a> {
    pub fn parse(line: &'a str, line_no: i64) -> Result<Self, ParseError> {
        let mut pairs: pest::iterators::Pairs<'a, Rule> =
            <ExpressionParser as pest::Parser<Rule>>::parse(Rule::planNode, line).map_err(|e| {
                ParseError::Plan(
                    ParseContext {
                        line_no,
                        line: line.to_string(),
                    },
                    MessageParseError::new("planNode", ErrorKind::InvalidValue, Box::new(e)),
                )
            })?;

        let outer = pairs.next().unwrap();
        assert!(pairs.next().is_none()); // Should be exactly one pair
        let inner = unwrap_single_pair(outer);

        Ok(if inner.as_rule() == Rule::adv_extension {
            LineNode::AdvExt(AdvExt {
                pair: inner,
                line_no,
            })
        } else {
            LineNode::Relation(RelationNode {
                pair: inner,
                line_no,
                adv_extensions: Vec::new(),
                children: Vec::new(),
            })
        })
    }

    /// Parse the line as a top-level relation at depth 0 (either root_relation or regular relation)
    pub fn parse_root(line: &'a str, line_no: i64) -> Result<Self, ParseError> {
        let mut pairs: pest::iterators::Pairs<'a, Rule> = <ExpressionParser as pest::Parser<
            Rule,
        >>::parse(
            Rule::top_level_relation, line
        )
        .map_err(|e| {
            ParseError::Plan(
                ParseContext::new(line_no, line.to_string()),
                MessageParseError::new("top_level_relation", ErrorKind::Syntax, Box::new(e)),
            )
        })?;

        let outer = pairs.next().unwrap();
        assert!(pairs.next().is_none());

        // top_level_relation is either root_relation or planNode.
        // If planNode, unwrap one more level to obtain the specific relation rule.
        let inner = unwrap_single_pair(outer);
        let pair = if inner.as_rule() == Rule::planNode {
            unwrap_single_pair(inner)
        } else {
            inner // root_relation
        };

        // planNode can technically include adv_extension; surface it as AdvExt so
        // TreeBuilder::add_line can produce the appropriate depth-0 error.
        if pair.as_rule() == Rule::adv_extension {
            return Ok(LineNode::AdvExt(AdvExt { pair, line_no })); // Where is the error produced or checked???
        }

        Ok(LineNode::Relation(RelationNode {
            pair,
            line_no,
            adv_extensions: Vec::new(),
            children: Vec::new(),
        }))
    }
}

/// Return the number of output columns for a relation.
///
/// Extension relations are excluded here intentionally: their output column
/// count is carried as the `usize` in the `(Rel, usize)` returned by
/// `build_rel` / `parse_extension_relation`, so they never reach this function.
fn get_output_field_count(rel: &Rel) -> usize {
    // An explicit emit mapping defines the exact emitted column count regardless
    // of relation type
    if let Some(EmitKind::Emit(e)) = get_emit_kind(rel) {
        return e.output_mapping.len();
    }

    match &rel.rel_type {
        Some(RelType::Read(read_rel)) => {
            // Read embeds the schema directly — count the typed fields.
            if let Some(schema) = read_rel.base_schema.as_ref()
                && let Some(s) = schema.r#struct.as_ref()
            {
                return s.types.len();
            }
            0
        }
        Some(RelType::Filter(filter_rel)) => {
            // Filter passes all input columns through unchanged.
            if let Some(input) = filter_rel.input.as_ref() {
                return get_output_field_count(input);
            }
            0
        }
        Some(RelType::Project(project_rel)) => {
            // Without an emit, Project's direct output is all input columns
            // followed by all computed expressions.
            if let Some(input) = project_rel.input.as_ref() {
                return get_output_field_count(input) + project_rel.expressions.len();
            }
            0
        }
        Some(RelType::Sort(sort_rel)) => {
            // Sort passes all input columns through unchanged.
            if let Some(input) = sort_rel.input.as_ref() {
                return get_output_field_count(input);
            }
            0
        }
        Some(RelType::Fetch(fetch_rel)) => {
            // Fetch (LIMIT/OFFSET) passes all input columns through unchanged.
            if let Some(input) = fetch_rel.input.as_ref() {
                return get_output_field_count(input);
            }
            0
        }
        Some(RelType::Join(join_rel)) => {
            // Join concatenates the columns of its two inputs.
            let left = join_rel.left.as_deref().map_or(0, get_output_field_count);
            let right = join_rel.right.as_deref().map_or(0, get_output_field_count);
            left + right
        }
        Some(RelType::Aggregate(agg_rel)) => {
            // Aggregate's direct output is the grouping expressions followed
            // by the measure results.
            agg_rel.grouping_expressions.len() + agg_rel.measures.len()
        }
        _ => unimplemented!("TODO: implement this for the specific type"),
    }
}

/// Extract the emit kind from a relation's `common` field, covering all
/// standard relation variants that carry it.
fn get_emit_kind(rel: &Rel) -> Option<&EmitKind> {
    match &rel.rel_type {
        Some(RelType::Read(r)) => r.common.as_ref(),
        Some(RelType::Filter(r)) => r.common.as_ref(),
        Some(RelType::Project(r)) => r.common.as_ref(),
        Some(RelType::Sort(r)) => r.common.as_ref(),
        Some(RelType::Fetch(r)) => r.common.as_ref(),
        Some(RelType::Join(r)) => r.common.as_ref(),
        Some(RelType::Aggregate(r)) => r.common.as_ref(),
        _ => unimplemented!("TODO: implement this for the specific type"),
    }
    .and_then(|c| c.emit_kind.as_ref())
}

/// Set the `advanced_extension` field on a [`Rel`], covering all standard
/// relation variants that carry the field directly.
fn set_advanced_extension(rel: &mut Rel, adv_ext: AdvancedExtension) {
    match &mut rel.rel_type {
        Some(RelType::Read(r)) => r.advanced_extension = Some(adv_ext),
        Some(RelType::Filter(r)) => r.advanced_extension = Some(adv_ext),
        Some(RelType::Project(r)) => r.advanced_extension = Some(adv_ext),
        Some(RelType::Aggregate(r)) => r.advanced_extension = Some(adv_ext),
        Some(RelType::Sort(r)) => r.advanced_extension = Some(adv_ext),
        Some(RelType::Fetch(r)) => r.advanced_extension = Some(adv_ext),
        Some(RelType::Join(r)) => r.advanced_extension = Some(adv_ext),
        Some(RelType::ExtensionLeaf(_)) => {
            unreachable!("Extension types do not have advanced extensions defined on them")
        }
        Some(RelType::ExtensionSingle(_)) => {
            unreachable!("Extension types do not have advanced extensions defined on them")
        }
        Some(RelType::ExtensionMulti(_)) => {
            unreachable!("Extension types do not have advanced extensions defined on them")
        }
        _ => {}
    }
}

#[derive(Copy, Clone, Debug)]
pub enum State {
    // The initial state, before we have parsed any lines.
    Initial,
    // The extensions section, after parsing the header and any other Extension lines.
    Extensions,
    // The plan section, after parsing the header and any other Plan lines.
    Plan,
}

impl fmt::Display for State {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{self:?}")
    }
}

// An in-progress tree builder, building the tree of relations.
#[derive(Debug, Clone, Default)]
pub struct TreeBuilder<'a> {
    // Current tree of nodes being built. These have been successfully parsed
    // into Pest pairs, but have not yet been converted to substrait plans.
    current: Option<RelationNode<'a>>,
    // Completed trees that have been built.
    completed: Vec<RelationNode<'a>>,
}

impl<'a> TreeBuilder<'a> {
    /// Traverse down the tree, always taking the last child at each level, until reaching the specified depth.
    pub fn get_at_depth(&mut self, depth: usize) -> Option<&mut RelationNode<'a>> {
        let mut node = self.current.as_mut()?;
        for _ in 0..depth {
            node = node.children.last_mut()?;
        }
        Some(node)
    }

    pub fn add_line(&mut self, depth: usize, node: LineNode<'a>) -> Result<(), ParseError> {
        match node {
            LineNode::Relation(rel_node) => {
                if depth == 0 {
                    if let Some(prev) = self.current.take() {
                        self.completed.push(prev);
                    }
                    self.current = Some(rel_node);
                    return Ok(());
                }

                let parent = match self.get_at_depth(depth - 1) {
                    None => {
                        return Err(ParseError::Plan(
                            rel_node.context(),
                            MessageParseError::invalid(
                                "relation",
                                rel_node.pair.as_span(),
                                format!("No parent found for depth {depth}"),
                            ),
                        ));
                    }
                    Some(parent) => parent,
                };

                parent.children.push(rel_node);
            }
            LineNode::AdvExt(adv_ext) => {
                if depth == 0 {
                    return Err(ParseError::ValidationError(format!(
                        "line {}: advanced extension annotations cannot appear at the top level",
                        adv_ext.line_no
                    )));
                }

                let parent = match self.get_at_depth(depth - 1) {
                    None => {
                        return Err(ParseError::ValidationError(format!(
                            "line {}: no parent found for advanced extension at depth {depth}",
                            adv_ext.line_no
                        )));
                    }
                    Some(parent) => parent,
                };

                if !parent.children.is_empty() {
                    return Err(ParseError::ValidationError(format!(
                        "line {}: advanced extension annotations (+ Enh: / + Opt:) must \
                         appear before child relations, not after",
                        adv_ext.line_no
                    )));
                }

                parent.adv_extensions.push(adv_ext);
            }
        }
        Ok(())
    }

    /// End of input - move any remaining nodes from stack to completed and
    /// return any trees in progress. Resets the builder to its initial state
    /// (empty)
    /// Move any remaining nodes from stack to completed
    pub fn finish(&mut self) -> Vec<RelationNode<'a>> {
        if let Some(node) = self.current.take() {
            self.completed.push(node);
        }
        std::mem::take(&mut self.completed)
    }
}

// Relation parsing component - handles converting LineNodes to Relations
#[derive(Debug, Clone, Default)]
pub struct RelationParser<'a> {
    tree: TreeBuilder<'a>,
}

impl<'a> RelationParser<'a> {
    pub fn parse_line(&mut self, line: IndentedLine<'a>, line_no: i64) -> Result<(), ParseError> {
        let IndentedLine(depth, line) = line;

        // Use parse_root for depth 0 (top-level relations), parse for other depths
        let node = if depth == 0 {
            LineNode::parse_root(line, line_no)?
        } else {
            LineNode::parse(line, line_no)?
        };

        self.tree.add_line(depth, node)
    }

    /// Parse a relation from a Pest pair of rule 'relation' into a Substrait
    /// Rel, together with its output column count.
    ///
    /// The `usize` in the return value is the number of columns this relation
    /// outputs.  Callers accumulate these counts across children and pass the
    /// sum in as `input_field_count` so that relations like `ProjectRel` can
    /// assign correct emit-mapping indices to their computed expressions.
    //
    // Clippy says a Vec<Box<…>> is unnecessary, as the Vec is already on the
    // heap, but this is what the protobuf requires so we allow it here
    #[allow(clippy::vec_box)]
    fn parse_relation(
        &self,
        extensions: &SimpleExtensions,
        registry: &ExtensionRegistry,
        line_no: i64,
        pair: pest::iterators::Pair<Rule>,
        child_relations: Vec<Box<substrait::proto::Rel>>,
        input_field_count: usize,
    ) -> Result<(substrait::proto::Rel, usize), ParseError> {
        // `pair` is already the specific relation rule (e.g. Rule::project_relation),
        // unwrapped from the outer planNode during LineNode construction.
        let (e, r, l, p_inner, cr, ic) = (
            extensions,
            registry,
            line_no,
            pair,
            child_relations,
            input_field_count,
        );

        match p_inner.as_rule() {
            Rule::read_relation => self.parse_rel::<ReadRel>(e, l, p_inner, cr, ic),
            Rule::filter_relation => self.parse_rel::<FilterRel>(e, l, p_inner, cr, ic),
            Rule::project_relation => self.parse_rel::<ProjectRel>(e, l, p_inner, cr, ic),
            Rule::aggregate_relation => self.parse_rel::<AggregateRel>(e, l, p_inner, cr, ic),
            Rule::sort_relation => self.parse_rel::<SortRel>(e, l, p_inner, cr, ic),
            Rule::fetch_relation => self.parse_rel::<FetchRel>(e, l, p_inner, cr, ic),
            Rule::join_relation => self.parse_rel::<JoinRel>(e, l, p_inner, cr, ic),
            Rule::extension_relation => self.parse_extension_relation(e, r, l, p_inner, cr),
            _ => unreachable!("unhandled relation rule: {:?}", p_inner.as_rule()),
        }
    }

    /// Parse a specific relation type, returning the relation and its output column count.
    ///
    /// The count is derived from the just-built relation via `get_output_field_count`.
    /// Extension relations are handled separately in `parse_extension_relation` because
    /// their count comes from the user-supplied output-columns list, not from a schema
    /// embedded in the protobuf.
    // Box is needed because Rel is a large enum and we need to pass ownership
    // through the RelationParsePair trait, which requires Box<Rel>.
    #[allow(clippy::vec_box)]
    fn parse_rel<T: RelationParsePair>(
        &self,
        extensions: &SimpleExtensions,
        line_no: i64,
        pair: pest::iterators::Pair<Rule>,
        child_relations: Vec<Box<substrait::proto::Rel>>,
        input_field_count: usize,
    ) -> Result<(substrait::proto::Rel, usize), ParseError> {
        assert_eq!(pair.as_rule(), T::rule());

        let line = pair.as_str();

        let rel_type =
            T::parse_pair_with_context(extensions, pair, child_relations, input_field_count);

        match rel_type {
            Ok(rel) => {
                let as_rel = rel.into_rel();
                let count = get_output_field_count(&as_rel);
                Ok((as_rel, count))
            }
            Err(e) => Err(ParseError::Plan(
                ParseContext::new(line_no, line.to_string()),
                e,
            )),
        }
    }

    /// Parse extension relations.
    /// Extension relations need the full parsing context for registry lookups and warning handling.
    // Box is needed because Rel is a large enum and we need to pass ownership
    // through the RelationParsePair trait, which requires Box<Rel>.
    #[allow(clippy::vec_box)]
    fn parse_extension_relation(
        &self,
        extensions: &SimpleExtensions,
        registry: &ExtensionRegistry,
        line_no: i64,
        pair: pest::iterators::Pair<Rule>,
        child_relations: Vec<Box<substrait::proto::Rel>>,
    ) -> Result<(substrait::proto::Rel, usize), ParseError> {
        assert_eq!(pair.as_rule(), Rule::extension_relation);

        let line = pair.as_str();
        let pair_span = pair.as_span();

        // Parse extension invocation, which includes the user-provided name
        let ExtensionInvocation {
            name,
            args: extension_args,
        } = ExtensionInvocation::parse_pair(pair);

        // Validate child count matches relation type
        let child_count = child_relations.len();
        extension_args
            .relation_type
            .validate_child_count(child_count)
            .map_err(|e| {
                ParseError::Plan(
                    ParseContext::new(line_no, line.to_string()),
                    MessageParseError::invalid("extension_relation", pair_span, e),
                )
            })?;

        let context = RelationParsingContext {
            extensions,
            registry,
            line_no,
            line,
        };

        let detail = context.resolve_extension_detail(&name, &extension_args)?;
        // Capture the output column count before consuming extension_args, so
        // it can be returned to the caller.
        let output_column_count = extension_args.output_columns.len();

        let rel = extension_args
            .relation_type
            .create_rel(detail, child_relations)
            .map_err(|e| {
                ParseError::Plan(
                    ParseContext::new(line_no, line.to_string()),
                    MessageParseError::invalid("extension_relation", pair_span, e),
                )
            })?;

        Ok((rel, output_column_count))
    }

    /// Convert a [`RelationNode`] into a Substrait Rel, together with its output column count.
    /// The returned `usize` is the number of columns this relation emits.
    fn build_rel(
        &self,
        extensions: &SimpleExtensions,
        registry: &ExtensionRegistry,
        node: RelationNode,
    ) -> Result<(substrait::proto::Rel, usize), ParseError> {
        let mut child_relations: Vec<Box<Rel>> = Vec::new();
        let mut input_field_count: usize = 0;

        for child in node.children {
            let (rel, count) = self.build_rel(extensions, registry, child)?;
            input_field_count += count;
            child_relations.push(Box::new(rel));
        }

        let (mut rel, output_count) = self.parse_relation(
            extensions,
            registry,
            node.line_no,
            node.pair,
            child_relations,
            input_field_count,
        )?;

        if !node.adv_extensions.is_empty() {
            let adv_ext =
                self.build_advanced_extension(extensions, registry, node.adv_extensions)?;
            set_advanced_extension(&mut rel, adv_ext);
        }

        Ok((rel, output_count))
    }

    /// Parse a list of [`AdvExt`] nodes into an [`AdvancedExtension`] proto.
    fn build_advanced_extension(
        &self,
        extensions: &SimpleExtensions,
        registry: &ExtensionRegistry,
        adv_exts: Vec<AdvExt>,
    ) -> Result<AdvancedExtension, ParseError> {
        let mut enhancement = None;
        let mut optimizations = Vec::new();

        for adv_ext in adv_exts {
            let line_no = adv_ext.line_no;
            let line = adv_ext.pair.as_str().to_string();
            let invocation = AdvExtInvocation::parse_pair(adv_ext.pair);
            let context = RelationParsingContext {
                extensions,
                registry,
                line_no,
                line: &line,
            };

            match invocation.ext_type {
                ExtensionType::Enhancement => {
                    let detail = context.resolve_adv_ext_detail(
                        ExtensionType::Enhancement,
                        &invocation.name,
                        &invocation.args,
                    )?;
                    if enhancement.is_some() {
                        return Err(ParseError::ValidationError(
                            "at most one enhancement per relation is allowed".to_string(),
                        ));
                    }
                    enhancement = Some(detail.into());
                }
                ExtensionType::Optimization => {
                    let detail = context.resolve_adv_ext_detail(
                        ExtensionType::Optimization,
                        &invocation.name,
                        &invocation.args,
                    )?;
                    optimizations.push(detail.into());
                }
                ExtensionType::Relation => {
                    unreachable!("Grammar restricts adv_ext_type to 'Enh' or 'Opt'")
                }
            }
        }

        Ok(AdvancedExtension {
            enhancement,
            optimization: optimizations,
        })
    }

    /// Build a tree of relations.
    fn build_plan_rel(
        &self,
        extensions: &SimpleExtensions,
        registry: &ExtensionRegistry,
        node: RelationNode,
    ) -> Result<PlanRel, ParseError> {
        // Plain relations are allowed as root relations; they just don't have names.
        if node.pair.as_rule() != Rule::root_relation {
            let (rel, _) = self.build_rel(extensions, registry, node)?;
            return Ok(PlanRel {
                rel_type: Some(plan_rel::RelType::Rel(rel)),
            });
        }

        // Named root relation.
        let context = node.context();
        let span = node.pair.as_span();

        // Parse the column names
        let column_names_pair = unwrap_single_pair(node.pair);
        assert_eq!(column_names_pair.as_rule(), Rule::root_name_list);

        let names: Vec<String> = column_names_pair
            .into_inner()
            .map(|name_pair| {
                assert_eq!(name_pair.as_rule(), Rule::name);
                Name::parse_pair(name_pair).0
            })
            .collect();

        let mut children = node.children;
        let child = match children.len() {
            1 => {
                let (rel, _) = self.build_rel(extensions, registry, children.pop().unwrap())?;
                rel
            }
            n => {
                return Err(ParseError::Plan(
                    context,
                    MessageParseError::invalid(
                        "root_relation",
                        span,
                        format!("Root relation must have exactly one child, found {n}"),
                    ),
                ));
            }
        };

        Ok(PlanRel {
            rel_type: Some(plan_rel::RelType::Root(RelRoot {
                names,
                input: Some(child),
            })),
        })
    }

    /// Build all the trees.
    fn build(
        mut self,
        extensions: &SimpleExtensions,
        registry: &ExtensionRegistry,
    ) -> Result<Vec<PlanRel>, ParseError> {
        let nodes = self.tree.finish();
        nodes
            .into_iter()
            .map(|n| self.build_plan_rel(extensions, registry, n))
            .collect::<Result<Vec<PlanRel>, ParseError>>()
    }
}

/// A parser for Substrait query plans in text format.
///
/// The `Parser` converts human-readable Substrait text format into Substrait
/// protobuf plans. It handles both the extensions section (which defines
/// functions, types, etc.) and the plan section (which defines the actual query
/// structure).
///
/// ## Usage
///
/// The simplest entry point is the static `parse()` method:
///
/// ```rust
/// use substrait_explain::parser::Parser;
///
/// let plan_text = r#"
/// === Plan
/// Root[c, d]
///   Project[$1, 42]
///     Read[schema.table => a:i64, b:string?]
/// "#;
///
/// let plan = Parser::parse(plan_text).unwrap();
/// ```
///
/// ## Input Format
///
/// The parser expects input in the following format:
///
/// ```text
/// === Extensions
/// URNs:
///   @  1: https://github.com/substrait-io/substrait/blob/main/extensions/functions_arithmetic.yaml
/// Functions:
///   # 10 @  1: add
/// === Plan
/// Root[columns]
///   Relation[arguments => columns]
///     ChildRelation[arguments => columns]
/// ```
///
/// - **Extensions section** (optional): Defines URNs and function/type declarations
/// - **Plan section** (required): Defines the query structure with indented relations
///
/// ## Error Handling
///
/// The parser provides detailed error information including:
/// - Line number where the error occurred
/// - The actual line content that failed to parse
/// - Specific error type and description
///
/// ```rust
/// use substrait_explain::parser::Parser;
///
/// let invalid_plan = r#"
/// === Plan
/// InvalidRelation[invalid syntax]
/// "#;
///
/// match Parser::parse(invalid_plan) {
///     Ok(plan) => println!("Successfully parsed"),
///     Err(e) => eprintln!("Parse error: {}", e),
/// }
/// ```
///
/// ## Supported Relations
///
/// The parser supports all standard Substrait relations:
/// - `Read[table => columns]` - Read from a table
/// - `Project[expressions]` - Project columns/expressions
/// - `Filter[condition => columns]` - Filter rows
/// - `Root[columns]` - Root relation with output columns
/// - And more...
///
/// ## Extensions Support
///
/// The parser fully supports Substrait Simple Extensions, allowing you to:
/// - Define custom functions with URNs and anchors
/// - Reference functions by name in expressions
/// - Use custom types and type variations
///
/// ```rust
/// use substrait_explain::parser::Parser;
///
/// let plan_with_extensions = r#"
/// === Extensions
/// URNs:
///   @  1: https://example.com/functions.yaml
/// Functions:
///   ## 10 @  1: my_custom_function
/// === Plan
/// Root[result]
///   Project[my_custom_function($0, $1)]
///     Read[table => col1:i32, col2:i32]
/// "#;
///
/// let plan = Parser::parse(plan_with_extensions).unwrap();
/// ```
///
/// ## Performance
///
/// The parser is designed for efficiency:
/// - Single-pass parsing with minimal allocations
/// - Early error detection and reporting
/// - Memory-efficient tree building
///
/// ## Thread Safety
///
/// `Parser` instances are not thread-safe and should not be shared between threads.
/// However, the static `parse()` method is safe to call from multiple threads.
#[derive(Debug)]
pub struct Parser<'a> {
    line_no: i64,
    state: State,
    extension_parser: ExtensionParser,
    extension_registry: ExtensionRegistry,
    relation_parser: RelationParser<'a>,
}
impl<'a> Default for Parser<'a> {
    fn default() -> Self {
        Self::new()
    }
}

impl<'a> Parser<'a> {
    /// Parse a Substrait plan from text format.
    ///
    /// This is the main entry point for parsing.
    ///
    /// The input should be in the Substrait text format, which consists of:
    /// - An optional extensions section starting with "=== Extensions"
    /// - A plan section starting with "=== Plan"
    /// - Indented relation definitions
    ///
    /// # Examples
    ///
    /// Simple parsing:
    /// ```rust
    /// use substrait_explain::parser::Parser;
    ///
    /// let plan_text = r#"
    /// === Plan
    /// Root[result]
    ///   Read[table => col:i32]
    /// "#;
    ///
    /// let plan = Parser::parse(plan_text).unwrap();
    /// assert_eq!(plan.relations.len(), 1);
    /// ```
    ///
    /// # Errors
    ///
    /// Returns a [`ParseError`] if the input cannot be parsed.
    pub fn parse(input: &str) -> ParseResult {
        Self::new().parse_plan(input)
    }

    /// Create a new parser with default configuration.
    pub fn new() -> Self {
        Self {
            line_no: 1,
            state: State::Initial,
            extension_parser: ExtensionParser::default(),
            extension_registry: ExtensionRegistry::new(),
            relation_parser: RelationParser::default(),
        }
    }

    /// Configure the parser to use the specified extension registry.
    pub fn with_extension_registry(mut self, registry: ExtensionRegistry) -> Self {
        self.extension_registry = registry;
        self
    }

    /// Parse a Substrait plan with the current parser configuration.
    pub fn parse_plan(mut self, input: &'a str) -> ParseResult {
        for line in input.lines() {
            if line.trim().is_empty() {
                self.line_no += 1;
                continue;
            }

            self.parse_line(line)?;
            self.line_no += 1;
        }

        let plan = self.build_plan()?;
        Ok(plan)
    }

    /// Parse a single line of input.
    fn parse_line(&mut self, line: &'a str) -> Result<(), ParseError> {
        let indented_line = IndentedLine::from(line);
        let line_no = self.line_no;
        let ctx = || ParseContext {
            line_no,
            line: line.to_string(),
        };

        match self.state {
            State::Initial => self.parse_initial(indented_line),
            State::Extensions => self
                .parse_extensions(indented_line)
                .map_err(|e| ParseError::Extension(ctx(), e)),
            State::Plan => {
                let IndentedLine(depth, line_str) = indented_line;

                // Parse the line
                let node = if depth == 0 {
                    LineNode::parse_root(line_str, line_no)?
                } else {
                    LineNode::parse(line_str, line_no)?
                };

                self.relation_parser.tree.add_line(depth, node)
            }
        }
    }

    /// Parse the initial line(s) of the input, which is either a blank line or
    /// the extensions or plan header.
    fn parse_initial(&mut self, line: IndentedLine) -> Result<(), ParseError> {
        match line {
            IndentedLine(0, l) if l.trim().is_empty() => {}
            IndentedLine(0, simple::EXTENSIONS_HEADER) => {
                self.state = State::Extensions;
            }
            IndentedLine(0, PLAN_HEADER) => {
                self.state = State::Plan;
            }
            IndentedLine(n, l) => {
                return Err(ParseError::Initial(
                    ParseContext::new(n as i64, l.to_string()),
                    MessageParseError::invalid(
                        "initial",
                        pest::Span::new(l, 0, l.len()).expect("Invalid span?!"),
                        format!("Unknown initial line: {l:?}"),
                    ),
                ));
            }
        }
        Ok(())
    }

    /// Parse a single line from the extensions section of the input, updating
    /// the parser state.
    fn parse_extensions(&mut self, line: IndentedLine<'_>) -> Result<(), ExtensionParseError> {
        if line == IndentedLine(0, PLAN_HEADER) {
            self.state = State::Plan;
            return Ok(());
        }
        self.extension_parser.parse_line(line)
    }

    /// Build the plan from the parser state with warning collection.
    fn build_plan(self) -> Result<Plan, ParseError> {
        let Parser {
            relation_parser,
            extension_parser,
            extension_registry,
            ..
        } = self;

        let extensions = extension_parser.extensions();

        // Parse the tree into relations
        let root_relations = relation_parser.build(extensions, &extension_registry)?;

        // Build the final plan
        Ok(Plan {
            extension_urns: extensions.to_extension_urns(),
            extensions: extensions.to_extension_declarations(),
            relations: root_relations,
            ..Default::default()
        })
    }
}

#[cfg(test)]
mod tests {
    use substrait::proto::extensions::simple_extension_declaration::MappingType;

    use super::*;
    use crate::extensions::simple::ExtensionKind;
    use crate::parser::extensions::ExtensionParserState;

    #[test]
    fn test_parse_basic_block() {
        let mut expected_extensions = SimpleExtensions::new();
        expected_extensions
            .add_extension_urn("/urn/common".to_string(), 1)
            .unwrap();
        expected_extensions
            .add_extension_urn("/urn/specific_funcs".to_string(), 2)
            .unwrap();
        expected_extensions
            .add_extension(ExtensionKind::Function, 1, 10, "func_a".to_string())
            .unwrap();
        expected_extensions
            .add_extension(ExtensionKind::Function, 2, 11, "func_b_special".to_string())
            .unwrap();
        expected_extensions
            .add_extension(ExtensionKind::Type, 1, 20, "SomeType".to_string())
            .unwrap();
        expected_extensions
            .add_extension(ExtensionKind::TypeVariation, 2, 30, "VarX".to_string())
            .unwrap();

        let mut parser = ExtensionParser::default();
        let input_block = r#"
URNs:
  @  1: /urn/common
  @  2: /urn/specific_funcs
Functions:
  # 10 @  1: func_a
  # 11 @  2: func_b_special
Types:
  # 20 @  1: SomeType
Type Variations:
  # 30 @  2: VarX
"#;

        for line_str in input_block.trim().lines() {
            parser
                .parse_line(IndentedLine::from(line_str))
                .unwrap_or_else(|e| panic!("Failed to parse line \'{line_str}\': {e:?}"));
        }

        assert_eq!(*parser.extensions(), expected_extensions);

        let extensions_str = parser.extensions().to_string("  ");
        // The writer adds the header; the ExtensionParser does not parse the
        // header, so we add it here for comparison.
        let expected_str = format!(
            "{}\n{}",
            simple::EXTENSIONS_HEADER,
            input_block.trim_start()
        );
        assert_eq!(extensions_str.trim(), expected_str.trim());
        // Check final state after all lines are processed.
        // The last significant line in input_block is a TypeVariation declaration.
        assert_eq!(
            parser.state(),
            ExtensionParserState::ExtensionDeclarations(ExtensionKind::TypeVariation)
        );

        // Check that a subsequent blank line correctly resets state to Extensions.
        parser.parse_line(IndentedLine(0, "")).unwrap();
        assert_eq!(parser.state(), ExtensionParserState::Extensions);
    }

    /// Test that we can parse a larger extensions block and it matches the input.
    #[test]
    fn test_parse_complete_extension_block() {
        let mut parser = ExtensionParser::default();
        let input_block = r#"
URNs:
  @  1: /urn/common
  @  2: /urn/specific_funcs
  @  3: /urn/types_lib
  @  4: /urn/variations_lib
Functions:
  # 10 @  1: func_a
  # 11 @  2: func_b_special
  # 12 @  1: func_c_common
Types:
  # 20 @  1: CommonType
  # 21 @  3: LibraryType
  # 22 @  1: AnotherCommonType
Type Variations:
  # 30 @  4: VarX
  # 31 @  4: VarY
"#;

        for line_str in input_block.trim().lines() {
            parser
                .parse_line(IndentedLine::from(line_str))
                .unwrap_or_else(|e| panic!("Failed to parse line \'{line_str}\': {e:?}"));
        }

        let extensions_str = parser.extensions().to_string("  ");
        // The writer adds the header; the ExtensionParser does not parse the
        // header, so we add it here for comparison.
        let expected_str = format!(
            "{}\n{}",
            simple::EXTENSIONS_HEADER,
            input_block.trim_start()
        );
        assert_eq!(extensions_str.trim(), expected_str.trim());
    }

    #[test]
    fn test_parse_relation_tree() {
        // Example plan with a Project, a Filter, and a Read, nested by indentation
        let plan = r#"=== Plan
Project[$0, $1, 42, 84]
  Filter[$2 => $0, $1]
    Read[my.table => a:i32, b:string?, c:boolean]
"#;
        let mut parser = Parser::default();
        for line in plan.lines() {
            parser.parse_line(line).unwrap();
        }

        // Complete the current tree to convert it to relations
        let plan = parser.build_plan().unwrap();

        let root_rel = &plan.relations[0].rel_type;
        let first_rel = match root_rel {
            Some(plan_rel::RelType::Rel(rel)) => rel,
            _ => panic!("Expected Rel type, got {root_rel:?}"),
        };
        // Root should be Project
        let project = match &first_rel.rel_type {
            Some(RelType::Project(p)) => p,
            other => panic!("Expected Project at root, got {other:?}"),
        };

        // Check that Project has Filter as input
        assert!(project.input.is_some());
        let filter_input = project.input.as_ref().unwrap();

        // Check that Filter has Read as input
        match &filter_input.rel_type {
            Some(RelType::Filter(_)) => {
                match &filter_input.rel_type {
                    Some(RelType::Filter(filter)) => {
                        assert!(filter.input.is_some());
                        let read_input = filter.input.as_ref().unwrap();

                        // Check that Read has no input (it's a leaf)
                        match &read_input.rel_type {
                            Some(RelType::Read(_)) => {}
                            other => panic!("Expected Read relation, got {other:?}"),
                        }
                    }
                    other => panic!("Expected Filter relation, got {other:?}"),
                }
            }
            other => panic!("Expected Filter relation, got {other:?}"),
        }
    }

    #[test]
    fn test_parse_root_relation() {
        // Test a plan with a Root relation
        let plan = r#"=== Plan
Root[result]
  Project[$0, $1]
    Read[my.table => a:i32, b:string?]
"#;
        let mut parser = Parser::default();
        for line in plan.lines() {
            parser.parse_line(line).unwrap();
        }

        let plan = parser.build_plan().unwrap();

        // Check that we have exactly one relation
        assert_eq!(plan.relations.len(), 1);

        let root_rel = &plan.relations[0].rel_type;
        let rel_root = match root_rel {
            Some(plan_rel::RelType::Root(rel_root)) => rel_root,
            other => panic!("Expected Root type, got {other:?}"),
        };

        // Check that the root has the correct name
        assert_eq!(rel_root.names, vec!["result"]);

        // Check that the root has a Project as input
        let project_input = match &rel_root.input {
            Some(rel) => rel,
            None => panic!("Root should have an input"),
        };

        let project = match &project_input.rel_type {
            Some(RelType::Project(p)) => p,
            other => panic!("Expected Project as root input, got {other:?}"),
        };

        // Check that Project has Read as input
        let read_input = match &project.input {
            Some(rel) => rel,
            None => panic!("Project should have an input"),
        };

        match &read_input.rel_type {
            Some(RelType::Read(_)) => {}
            other => panic!("Expected Read relation, got {other:?}"),
        }
    }

    #[test]
    fn test_parse_root_relation_no_names() {
        // Test a plan with a Root relation with no names
        let plan = r#"=== Plan
Root[]
  Project[$0, $1]
    Read[my.table => a:i32, b:string?]
"#;
        let mut parser = Parser::default();
        for line in plan.lines() {
            parser.parse_line(line).unwrap();
        }

        let plan = parser.build_plan().unwrap();

        let root_rel = &plan.relations[0].rel_type;
        let rel_root = match root_rel {
            Some(plan_rel::RelType::Root(rel_root)) => rel_root,
            other => panic!("Expected Root type, got {other:?}"),
        };

        // Check that the root has no names
        assert_eq!(rel_root.names, Vec::<String>::new());
    }

    #[test]
    fn test_parse_full_plan() {
        // Test a complete Substrait plan with extensions and relations
        let input = r#"
=== Extensions
URNs:
  @  1: /urn/common
  @  2: /urn/specific_funcs
Functions:
  # 10 @  1: func_a
  # 11 @  2: func_b_special
Types:
  # 20 @  1: SomeType
Type Variations:
  # 30 @  2: VarX

=== Plan
Project[$0, $1, 42, 84]
  Filter[$2 => $0, $1]
    Read[my.table => a:i32, b:string?, c:boolean]
"#;

        let plan = Parser::parse(input).unwrap();

        // Verify the plan structure
        assert_eq!(plan.extension_urns.len(), 2);
        assert_eq!(plan.extensions.len(), 4);
        assert_eq!(plan.relations.len(), 1);

        // Verify extension URIs
        let urn1 = &plan.extension_urns[0];
        assert_eq!(urn1.extension_urn_anchor, 1);
        assert_eq!(urn1.urn, "/urn/common");

        let urn2 = &plan.extension_urns[1];
        assert_eq!(urn2.extension_urn_anchor, 2);
        assert_eq!(urn2.urn, "/urn/specific_funcs");

        // Verify extensions
        let func1 = &plan.extensions[0];
        match &func1.mapping_type {
            Some(MappingType::ExtensionFunction(f)) => {
                assert_eq!(f.function_anchor, 10);
                assert_eq!(f.extension_urn_reference, 1);
                assert_eq!(f.name, "func_a");
            }
            other => panic!("Expected ExtensionFunction, got {other:?}"),
        }

        let func2 = &plan.extensions[1];
        match &func2.mapping_type {
            Some(MappingType::ExtensionFunction(f)) => {
                assert_eq!(f.function_anchor, 11);
                assert_eq!(f.extension_urn_reference, 2);
                assert_eq!(f.name, "func_b_special");
            }
            other => panic!("Expected ExtensionFunction, got {other:?}"),
        }

        let type1 = &plan.extensions[2];
        match &type1.mapping_type {
            Some(MappingType::ExtensionType(t)) => {
                assert_eq!(t.type_anchor, 20);
                assert_eq!(t.extension_urn_reference, 1);
                assert_eq!(t.name, "SomeType");
            }
            other => panic!("Expected ExtensionType, got {other:?}"),
        }

        let var1 = &plan.extensions[3];
        match &var1.mapping_type {
            Some(MappingType::ExtensionTypeVariation(v)) => {
                assert_eq!(v.type_variation_anchor, 30);
                assert_eq!(v.extension_urn_reference, 2);
                assert_eq!(v.name, "VarX");
            }
            other => panic!("Expected ExtensionTypeVariation, got {other:?}"),
        }

        // Verify the relation tree structure
        let root_rel = &plan.relations[0];
        match &root_rel.rel_type {
            Some(plan_rel::RelType::Rel(rel)) => {
                match &rel.rel_type {
                    Some(RelType::Project(project)) => {
                        // Verify Project relation
                        assert_eq!(project.expressions.len(), 2); // 42 and 84
                        assert!(project.input.is_some()); // Should have Filter as input

                        // Check the Filter input
                        let filter_input = project.input.as_ref().unwrap();
                        match &filter_input.rel_type {
                            Some(RelType::Filter(filter)) => {
                                assert!(filter.input.is_some()); // Should have Read as input

                                // Check the Read input
                                let read_input = filter.input.as_ref().unwrap();
                                match &read_input.rel_type {
                                    Some(RelType::Read(read)) => {
                                        // Verify Read relation
                                        let schema = read.base_schema.as_ref().unwrap();
                                        assert_eq!(schema.names.len(), 3);
                                        assert_eq!(schema.names[0], "a");
                                        assert_eq!(schema.names[1], "b");
                                        assert_eq!(schema.names[2], "c");

                                        let struct_ = schema.r#struct.as_ref().unwrap();
                                        assert_eq!(struct_.types.len(), 3);
                                    }
                                    other => panic!("Expected Read relation, got {other:?}"),
                                }
                            }
                            other => panic!("Expected Filter relation, got {other:?}"),
                        }
                    }
                    other => panic!("Expected Project relation, got {other:?}"),
                }
            }
            other => panic!("Expected Rel type, got {other:?}"),
        }
    }
}