// SPDX-License-Identifier: Apache-2.0

// SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership

// SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements

//! [`DFParser`]: DataFusion SQL Parser based on [`sqlparser`]

//!

//! This parser implements DataFusion specific statements such as

//! `CREATE EXTERNAL TABLE`

use datafusion_common::DataFusionError;

use datafusion_common::config::{ConfigNonZeroUsize, SqlParserOptions};

use datafusion_common::format::{ExplainFormat, ExplainStatementOptions};

use datafusion_common::{Diagnostic, Span, sql_err};

use sqlparser::ast::{ExprWithAlias, Ident, ObjectType, OrderByOptions, OrderBySort};

use sqlparser::tokenizer::TokenWithSpan;

use sqlparser::{

ast::{

ColumnDef, ColumnOptionDef, ObjectName, OrderByExpr, Query,

Statement as SQLStatement, TableConstraint, Value,

},

dialect::{Dialect, GenericDialect, keywords::Keyword},

parser::{Parser, ParserError},

tokenizer::{Token, Tokenizer, Word},

};

use std::collections::VecDeque;

use std::fmt;

use std::str::FromStr;

// Use `Parser::expected` instead, if possible

macro_rules! parser_err {

($MSG:expr $(; diagnostic = $DIAG:expr)?) => {{

let err = DataFusionError::from(ParserError::ParserError($MSG.to_string()));

$(

let err = err.with_diagnostic($DIAG);

)?

Err(err)

}};

}

fn parse_file_type(s: &str) -> Result<String, DataFusionError> {

Ok(s.to_uppercase())

}

/// DataFusion specific `EXPLAIN`

///

/// Supports both the legacy keyword form and, on dialects whose

/// [`Dialect::supports_explain_with_utility_options`] returns `true`

/// (PostgreSQL, DuckDB, etc.), the Postgres-style parenthesized option list:

///

/// ```sql

/// -- Legacy keyword form (any dialect)

/// EXPLAIN <ANALYZE> <VERBOSE> [FORMAT format] statement

///

/// -- Postgres-style option form (dialect-gated)

/// EXPLAIN (option [arg] [, ...]) statement

/// ```

///

/// See [`ExplainStatementOptions`] for the list of supported options in the

/// parenthesized form.

#[derive(Debug, Clone, PartialEq, Eq)]

pub struct ExplainStatement {

/// Normalized options parsed from either the legacy keyword form or the

/// parenthesized option list.

pub options: ExplainStatementOptions,

/// The statement to analyze. Note this is a DataFusion [`Statement`] (not a

/// [`sqlparser::ast::Statement`] so that we can use `EXPLAIN`, `COPY`, and other

/// DataFusion specific statements

pub statement: Box<Statement>,

}

impl fmt::Display for ExplainStatement {

fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

let Self { options, statement } = self;

// If only the legacy-era fields are set, print the legacy keyword

// form so existing round-trip tests continue to pass.

let uses_parenthesized = options.analyze_level.is_some()

|| options.analyze_categories.is_some()

|| options.show_statistics.is_some();

write!(f, "EXPLAIN ")?;

if uses_parenthesized {

// Emit a parenthesized option list.

let mut parts: Vec<String> = Vec::new();

if options.analyze {

parts.push("ANALYZE".to_string());

}

if options.verbose {

parts.push("VERBOSE".to_string());

}

if let Some(format) = &options.format {

parts.push(format!("FORMAT {format}"));

}

if let Some(level) = options.analyze_level {

parts.push(format!("LEVEL {level}"));

}

if let Some(cats) = &options.analyze_categories {

parts.push(format!("METRICS '{cats}'"));

}

if let Some(stats) = options.show_statistics {

parts.push(format!("COSTS {}", if stats { "ON" } else { "OFF" }));

}

write!(f, "({}) ", parts.join(", "))?;

} else {

if options.analyze {

write!(f, "ANALYZE ")?;

}

if options.verbose {

write!(f, "VERBOSE ")?;

}

if let Some(format) = &options.format {

write!(f, "FORMAT {format} ")?;

}

}

write!(f, "{statement}")

}

}

/// DataFusion extension DDL for `COPY`

///

/// # Syntax:

///

/// ```text

/// COPY <table_name | (<query>)>

/// TO

/// <destination_url>

/// (key_value_list)

/// ```

///

/// # Examples

///

/// ```sql

/// COPY lineitem TO 'lineitem'

/// STORED AS PARQUET (

/// partitions 16,

/// row_group_limit_rows 100000,

/// row_group_limit_bytes 200000

/// )

///

/// COPY (SELECT l_orderkey from lineitem) to 'lineitem.parquet';

/// ```

#[derive(Debug, Clone, PartialEq, Eq)]

pub struct CopyToStatement {

/// From where the data comes from

pub source: CopyToSource,

/// The URL to where the data is heading

pub target: String,

/// Partition keys

pub partitioned_by: Vec<String>,

/// File type (Parquet, NDJSON, CSV etc.)

pub stored_as: Option<String>,

/// Target specific options

pub options: Vec<(String, Value)>,

}

impl fmt::Display for CopyToStatement {

fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

let Self {

source,

target,

partitioned_by,

stored_as,

options,

..

} = self;

write!(f, "COPY {source} TO {target}")?;

if let Some(file_type) = stored_as {

write!(f, " STORED AS {file_type}")?;

}

if !partitioned_by.is_empty() {

write!(f, " PARTITIONED BY ({})", partitioned_by.join(", "))?;

}

if !options.is_empty() {

let opts: Vec<_> =

options.iter().map(|(k, v)| format!("'{k}' {v}")).collect();

write!(f, " OPTIONS ({})", opts.join(", "))?;

}

Ok(())

}

}

#[derive(Debug, Clone, PartialEq, Eq)]

pub enum CopyToSource {

/// `COPY <table> TO ...`

Relation(ObjectName),

/// COPY (...query...) TO ...

Query(Box<Query>),

}

impl fmt::Display for CopyToSource {

fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

match self {

CopyToSource::Relation(r) => write!(f, "{r}"),

CopyToSource::Query(q) => write!(f, "({q})"),

}

}

}

/// This type defines a lexicographical ordering.

pub(crate) type LexOrdering = Vec<OrderByExpr>;

/// DataFusion extension DDL for `CREATE EXTERNAL TABLE`

///

/// Syntax:

///

/// ```text

/// CREATE

/// [ OR REPLACE ]

/// EXTERNAL TABLE

/// [ IF NOT EXISTS ]

/// <TABLE_NAME>[ (<column_definition>) ]

/// STORED AS <file_type>

/// [ PARTITIONED BY (<column_definition list> | <column list>) ]

/// [ WITH ORDER (<ordered column list>)

/// [ OPTIONS (<key_value_list>) ]

/// LOCATION <literal> | LOCATION (<literal>[, ...])

///

/// <column_definition> := (<column_name> <data_type>, ...)

///

/// <column_list> := (<column_name>, ...)

///

/// <ordered_column_list> := (<column_name> <sort_clause>, ...)

///

/// <key_value_list> := (<literal> <literal, <literal> <literal>, ...)

/// ```

#[derive(Debug, Clone, PartialEq, Eq)]

pub struct CreateExternalTable {

/// Table name

pub name: ObjectName,

/// Optional schema

pub columns: Vec<ColumnDef>,

/// File type (Parquet, NDJSON, CSV, etc)

pub file_type: String,

/// Paths to files

pub locations: Vec<String>,

/// Partition Columns

pub table_partition_cols: Vec<String>,

/// Ordered expressions

pub order_exprs: Vec<LexOrdering>,

/// Option to not error if table already exists

pub if_not_exists: bool,

/// Option to replace table content if table already exists

pub or_replace: bool,

/// Whether the table is a temporary table

pub temporary: bool,

/// Infinite streams?

pub unbounded: bool,

/// Table(provider) specific options

pub options: Vec<(String, Value)>,

/// A table-level constraint

pub constraints: Vec<TableConstraint>,

}

impl fmt::Display for CreateExternalTable {

fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

write!(f, "CREATE ")?;

if self.or_replace {

write!(f, "OR REPLACE ")?;

}

write!(f, "EXTERNAL TABLE ")?;

if self.if_not_exists {

write!(f, "IF NOT EXISTS ")?;

}

write!(f, "{} ", self.name)?;

write!(f, "STORED AS {} ", self.file_type)?;

if !self.order_exprs.is_empty() {

write!(f, "WITH ORDER (")?;

let mut first = true;

for expr in self.order_exprs.iter().flatten() {

if !first {

write!(f, ", ")?;

}

write!(f, "{expr}")?;

first = false;

}

write!(f, ") ")?;

}

match self.locations.as_slice() {

[location] => write!(

f,

"LOCATION {}",

Value::SingleQuotedString(location.clone())

),

locations => {

write!(f, "LOCATION (")?;

for (idx, location) in locations.iter().enumerate() {

if idx > 0 {

write!(f, ", ")?;

}

write!(f, "{}", Value::SingleQuotedString(location.clone()))?;

}

write!(f, ")")

}

}

}

}

/// DataFusion extension `CREATE EXTERNAL CATALOG` statement.

///

/// ```sql

/// CREATE [OR REPLACE] EXTERNAL CATALOG [IF NOT EXISTS] <catalog_name>

/// STORED AS <catalog_type>

/// [ LOCATION <literal> ]

/// [ OPTIONS (<key_value_list>) ]

///

/// <key_value_list> := (<literal> <literal>, <literal> <literal>, ...)

/// ```

#[derive(Debug, Clone, PartialEq, Eq)]

pub struct CreateExternalCatalog {

/// Catalog name

pub catalog_name: ObjectName,

/// The key used to look up the registered `CatalogProviderFactory`

pub catalog_type: String,

/// The physical location of the catalog, if applicable

pub location: Option<String>,

/// Option to not error if catalog already exists

pub if_not_exists: bool,

/// Option to replace the catalog if it already exists

pub or_replace: bool,

/// Catalog(provider) specific options

pub options: Vec<(String, Value)>,

}

impl fmt::Display for CreateExternalCatalog {

fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

write!(f, "CREATE EXTERNAL CATALOG ")?;

if self.or_replace {

write!(f, "OR REPLACE ")?;

}

write!(f, "EXTERNAL CATALOG ")?;

if self.if_not_exists {

write!(f, "IF NOT EXISTS ")?;

}

write!(f, "{} ", self.catalog_name)?;

write!(f, "STORED AS {}", self.catalog_type)?;

if let Some(location) = &self.location {

write!(

f,

" LOCATION {}",

Value::SingleQuotedString(location.clone())

)?;

}

Ok(())

}

}

/// DataFusion extension `DROP CATALOG` statement.

#[derive(Debug, Clone, PartialEq, Eq)]

pub struct DropCatalog {

/// Catalog name

pub name: ObjectName,

/// Option to not error if the catalog does not exist

pub if_exists: bool,

/// Whether drop should cascade

pub cascade: bool,

}

impl fmt::Display for DropCatalog {

fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

write!(f, "DROP CATALOG ")?;

if self.if_exists {

write!(f, "IF EXISTS ")?;

}

write!(f, "{}", self.name)

}

}

/// DataFusion extension for `RESET`

#[derive(Debug, Clone, PartialEq, Eq)]

pub enum ResetStatement {

/// Reset a single configuration variable (stored as provided)

Variable(ObjectName),

}

impl fmt::Display for ResetStatement {

fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

match self {

ResetStatement::Variable(name) => write!(f, "RESET {name}"),

}

}

}

/// DataFusion SQL Statement.

///

/// This can either be a [`Statement`] from [`sqlparser`] from a

/// standard SQL dialect, or a DataFusion extension such as `CREATE

/// EXTERNAL TABLE`. See [`DFParser`] for more information.

///

/// [`Statement`]: sqlparser::ast::Statement

#[derive(Debug, Clone, PartialEq, Eq)]

pub enum Statement {

/// ANSI SQL AST node (from sqlparser-rs)

Statement(Box<SQLStatement>),

/// Extension: `CREATE EXTERNAL TABLE`

CreateExternalTable(CreateExternalTable),

/// Extension: `CREATE EXTERNAL CATALOG`

CreateExternalCatalog(CreateExternalCatalog),

/// Extension: `COPY TO`

CopyTo(CopyToStatement),

/// EXPLAIN for extensions

Explain(ExplainStatement),

/// Extension: `RESET`

Reset(ResetStatement),

}

impl fmt::Display for Statement {

fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

match self {

Statement::Statement(stmt) => write!(f, "{stmt}"),

Statement::CreateExternalTable(stmt) => write!(f, "{stmt}"),

Statement::CreateExternalCatalog(stmt) => write!(f, "{stmt}"),

Statement::CopyTo(stmt) => write!(f, "{stmt}"),

Statement::Explain(stmt) => write!(f, "{stmt}"),

Statement::Reset(stmt) => write!(f, "{stmt}"),

}

}

}

fn ensure_not_set<T>(field: Option<&T>, name: &str) -> Result<(), DataFusionError> {

if field.is_some() {

parser_err!(format!("{name} specified more than once",))?

}

Ok(())

}

/// DataFusion SQL Parser based on [`sqlparser`]

///

/// Parses DataFusion's SQL dialect, often delegating to [`sqlparser`]'s [`Parser`].

///

/// DataFusion mostly follows existing SQL dialects via

/// `sqlparser`. However, certain statements such as `COPY` and

/// `CREATE EXTERNAL TABLE` have special syntax in DataFusion. See

/// [`Statement`] for a list of this special syntax

pub struct DFParser<'a> {

pub parser: Parser<'a>,

options: SqlParserOptions,

/// Whether the configured dialect supports Postgres-style

/// `EXPLAIN (option, ...)` utility-option syntax. Cached here because

/// sqlparser's [`Parser::dialect`] field is private.

supports_explain_with_utility_options: bool,

}

/// Default recursion limit for SQL parsing. Higher than the `sqlparser` default

/// (50) for backwards compatibility with older versions.

///

/// Newer `sqlparser` versions added recursion guards to more parse functions, so

/// the same query consumes more depth than it used to.

const DEFAULT_RECURSION_LIMIT: usize = 51;

const DEFAULT_DIALECT: GenericDialect = GenericDialect {};

/// Builder for [`DFParser`]

///

/// # Example: Create and Parse SQL statements

/// ```

/// # use datafusion_sql::parser::DFParserBuilder;

/// # use datafusion_common::Result;

/// # fn test() -> Result<()> {

/// let mut parser = DFParserBuilder::new("SELECT * FROM foo; SELECT 1 + 2").build()?;

/// // parse the SQL into DFStatements

/// let statements = parser.parse_statements()?;

/// assert_eq!(statements.len(), 2);

/// # Ok(())

/// # }

/// ```

///

/// # Example: Create and Parse expression with a different dialect

/// ```

/// # use datafusion_sql::parser::DFParserBuilder;

/// # use datafusion_common::Result;

/// # use datafusion_sql::sqlparser::dialect::MySqlDialect;

/// # use datafusion_sql::sqlparser::ast::Expr;

/// # fn test() -> Result<()> {

/// let dialect = MySqlDialect {}; // Parse using MySQL dialect

/// let mut parser = DFParserBuilder::new("1 + 2")

/// .with_dialect(&dialect)

/// .build()?;

/// // parse 1+2 into an sqlparser::ast::Expr

/// let res = parser.parse_expr()?;

/// assert!(matches!(res.expr, Expr::BinaryOp { .. }));

/// # Ok(())

/// # }

/// ```

pub struct DFParserBuilder<'a, 'b> {

/// Parser input: either raw SQL or tokens

input: ParserInput<'a>,

/// The Dialect to use (defaults to [`GenericDialect`]

dialect: &'b dyn Dialect,

/// The recursion limit while parsing

recursion_limit: usize,

}

/// Describes a possible input for parser

pub enum ParserInput<'a> {

/// Raw SQL. Tokenization will be performed automatically as a

/// part of [`DFParserBuilder::build`]

Sql(&'a str),

/// Tokens

Tokens(Vec<TokenWithSpan>),

}

impl<'a> From<&'a str> for ParserInput<'a> {

fn from(sql: &'a str) -> Self {

Self::Sql(sql)

}

}

impl From<Vec<TokenWithSpan>> for ParserInput<'static> {

fn from(tokens: Vec<TokenWithSpan>) -> Self {

Self::Tokens(tokens)

}

}

impl<'a, 'b> DFParserBuilder<'a, 'b> {

/// Create a new parser builder for the specified tokens using the

/// [`GenericDialect`].

pub fn new(input: impl Into<ParserInput<'a>>) -> Self {

Self {

input: input.into(),

dialect: &DEFAULT_DIALECT,

recursion_limit: DEFAULT_RECURSION_LIMIT,

}

}

/// Adjust the parser builder's dialect. Defaults to [`GenericDialect`]

pub fn with_dialect(mut self, dialect: &'b dyn Dialect) -> Self {

self.dialect = dialect;

self

}

/// Adjust the recursion limit of sql parsing. Defaults to 51

pub fn with_recursion_limit(mut self, recursion_limit: usize) -> Self {

self.recursion_limit = recursion_limit;

self

}

/// Build resulting parser

pub fn build(self) -> Result<DFParser<'b>, DataFusionError> {

let tokens = match self.input {

ParserInput::Tokens(tokens) => tokens,

ParserInput::Sql(sql) => {

let mut tokenizer = Tokenizer::new(self.dialect, sql);

// Convert TokenizerError -> ParserError

tokenizer

.tokenize_with_location()

.map_err(ParserError::from)?

}

};

Ok(DFParser {

parser: Parser::new(self.dialect)

.with_tokens_with_locations(tokens)

.with_recursion_limit(self.recursion_limit),

options: SqlParserOptions {

recursion_limit: ConfigNonZeroUsize::try_new(self.recursion_limit)?,

..Default::default()

},

supports_explain_with_utility_options: self

.dialect

.supports_explain_with_utility_options(),

})

}

}

/// Returns true when `tok` is the start of a query / parenthesized query

/// group. Used to disambiguate `EXPLAIN (SELECT ...)` (a parenthesized query)

/// from `EXPLAIN (ANALYZE) SELECT ...` (a Postgres-style option list).

fn token_starts_query(tok: &Token) -> bool {

match tok {

Token::LParen => true,

Token::Word(Word { keyword, .. }) => matches!(

keyword,

Keyword::SELECT

| Keyword::WITH

| Keyword::VALUES

| Keyword::TABLE

| Keyword::INSERT

| Keyword::UPDATE

| Keyword::DELETE

| Keyword::MERGE

),

_ => false,

}

}

impl<'a> DFParser<'a> {

/// Parse a sql string into one or [`Statement`]s using the

/// [`GenericDialect`].

pub fn parse_sql(sql: &'a str) -> Result<VecDeque<Statement>, DataFusionError> {

let mut parser = DFParserBuilder::new(sql).build()?;

parser.parse_statements()

}

/// Parse a SQL string and produce one or more [`Statement`]s with

/// with the specified dialect.

pub fn parse_sql_with_dialect(

sql: &str,

dialect: &dyn Dialect,

) -> Result<VecDeque<Statement>, DataFusionError> {

let mut parser = DFParserBuilder::new(sql).with_dialect(dialect).build()?;

parser.parse_statements()

}

pub fn parse_sql_into_expr(sql: &str) -> Result<ExprWithAlias, DataFusionError> {

DFParserBuilder::new(sql).build()?.parse_into_expr()

}

pub fn parse_sql_into_expr_with_dialect(

sql: &str,

dialect: &dyn Dialect,

) -> Result<ExprWithAlias, DataFusionError> {

DFParserBuilder::new(sql)

.with_dialect(dialect)

.build()?

.parse_into_expr()

}

/// Parse a sql string into one or [`Statement`]s

pub fn parse_statements(&mut self) -> Result<VecDeque<Statement>, DataFusionError> {

let mut stmts = VecDeque::new();

let mut expecting_statement_delimiter = false;

loop {

// ignore empty statements (between successive statement delimiters)

while self.parser.consume_token(&Token::SemiColon) {

expecting_statement_delimiter = false;

}

if self.parser.peek_token() == Token::EOF {

break;

}

if expecting_statement_delimiter {

return self.expected("end of statement", &self.parser.peek_token());

}

let statement = self.parse_statement()?;

stmts.push_back(statement);

expecting_statement_delimiter = true;

}

Ok(stmts)

}

/// Report an unexpected token

fn expected<T>(

&self,

expected: &str,

found: &TokenWithSpan,

) -> Result<T, DataFusionError> {

let sql_parser_span = found.span;

let span = Span::try_from_sqlparser_span(sql_parser_span);

let diagnostic = Diagnostic::new_error(

format!("Expected: {expected}, found: {found}{}", found.span.start),

span,

);

parser_err!(

format!("Expected: {expected}, found: {found}{}", found.span.start);

diagnostic=

diagnostic

)

}

fn expect_token(

&mut self,

expected: &str,

token: &Token,

) -> Result<(), DataFusionError> {

let next_token = self.parser.peek_token_ref();

if next_token.token != *token {

self.expected(expected, next_token)

} else {

Ok(())

}

}

/// Parse a new expression

pub fn parse_statement(&mut self) -> Result<Statement, DataFusionError> {

match self.parser.peek_token().token {

Token::Word(w) => {

match w.keyword {

Keyword::CREATE => {

self.parser.next_token(); // CREATE

self.parse_create()

}

Keyword::COPY => {

if let Token::Word(w) = self.parser.peek_nth_token(1).token {

// use native parser for COPY INTO

if w.keyword == Keyword::INTO {

return self.parse_and_handle_statement();

}

}

self.parser.next_token(); // COPY

self.parse_copy()

}

Keyword::EXPLAIN => {

self.parser.next_token(); // EXPLAIN

self.parse_explain()

}

Keyword::RESET => {

self.parser.next_token(); // RESET

self.parse_reset()

}

Keyword::DROP

if self

.parser

.parse_keywords(&[Keyword::DROP, Keyword::CATALOG]) =>

{

self.parse_drop_catalog()

}

_ => {

// use sqlparser-rs parser

self.parse_and_handle_statement()

}

}

}

_ => {

// use the native parser

self.parse_and_handle_statement()

}

}

}

pub fn parse_expr(&mut self) -> Result<ExprWithAlias, DataFusionError> {

if let Token::Word(w) = self.parser.peek_token().token {

match w.keyword {

Keyword::CREATE | Keyword::COPY | Keyword::EXPLAIN => {

return parser_err!("Unsupported command in expression")?;

}

_ => {}

}

}

Ok(self.parser.parse_expr_with_alias()?)

}

/// Parses the entire SQL string into an expression.

///

/// In contrast to [`DFParser::parse_expr`], this function will report an error if the input

/// contains any trailing, unparsed tokens.

pub fn parse_into_expr(&mut self) -> Result<ExprWithAlias, DataFusionError> {

let expr = self.parse_expr()?;

self.expect_token("end of expression", &Token::EOF)?;

Ok(expr)

}

/// Helper method to parse a statement and handle errors consistently, especially for recursion limits

fn parse_and_handle_statement(&mut self) -> Result<Statement, DataFusionError> {

self.parser

.parse_statement()

.map(|stmt| Statement::Statement(Box::from(stmt)))

.map_err(|e| match e {

ParserError::RecursionLimitExceeded => DataFusionError::SQL(

Box::new(ParserError::RecursionLimitExceeded),

Some(format!(

" (current limit: {})",

self.options.recursion_limit

)),

),

other => DataFusionError::SQL(Box::new(other), None),

})

}

/// Parse a SQL `COPY TO` statement

pub fn parse_copy(&mut self) -> Result<Statement, DataFusionError> {

// parse as a query

let source = if self.parser.consume_token(&Token::LParen) {

let query = self.parser.parse_query()?;

self.parser.expect_token(&Token::RParen)?;

CopyToSource::Query(query)

} else {

// parse as table reference

let table_name = self.parser.parse_object_name(true)?;

CopyToSource::Relation(table_name)

};

#[derive(Default)]

struct Builder {

stored_as: Option<String>,

target: Option<String>,

partitioned_by: Option<Vec<String>>,

options: Option<Vec<(String, Value)>>,

}

let mut builder = Builder::default();

loop {

if let Some(keyword) = self.parser.parse_one_of_keywords(&[

Keyword::STORED,

Keyword::TO,

Keyword::PARTITIONED,

Keyword::OPTIONS,

Keyword::WITH,

]) {

match keyword {

Keyword::STORED => {

self.parser.expect_keyword(Keyword::AS)?;

ensure_not_set(builder.stored_as.as_ref(), "STORED AS")?;

builder.stored_as = Some(self.parse_file_format()?);

}

Keyword::TO => {

ensure_not_set(builder.target.as_ref(), "TO")?;

builder.target = Some(self.parser.parse_literal_string()?);

}

Keyword::WITH => {

self.parser.expect_keyword(Keyword::HEADER)?;

self.parser.expect_keyword(Keyword::ROW)?;

return parser_err!(

"WITH HEADER ROW clause is no longer in use. Please use the OPTIONS clause with 'format.has_header' set appropriately, e.g., OPTIONS ('format.has_header' 'true')"

)?;

}

Keyword::PARTITIONED => {

self.parser.expect_keyword(Keyword::BY)?;

ensure_not_set(

builder.partitioned_by.as_ref(),

"PARTITIONED BY",

)?;

builder.partitioned_by = Some(self.parse_partitions()?);

}

Keyword::OPTIONS => {

ensure_not_set(builder.options.as_ref(), "OPTIONS")?;

builder.options = Some(self.parse_value_options()?);

}

_ => {

unreachable!()

}

}

} else {

let token = self.parser.peek_token();

if token == Token::EOF || token == Token::SemiColon {

break;

} else {

return self.expected("end of statement or ;", &token)?;

}

}

}

let Some(target) = builder.target else {

return parser_err!("Missing TO clause in COPY statement")?;

};

Ok(Statement::CopyTo(CopyToStatement {

source,

target,

partitioned_by: builder.partitioned_by.unwrap_or(vec![]),

stored_as: builder.stored_as,

options: builder.options.unwrap_or(vec![]),

}))

}

/// Parse the next token as a key name for an option list

///

/// Note this is different than [`parse_literal_string`]

/// because it allows keywords as well as other non words

///

/// [`parse_literal_string`]: sqlparser::parser::Parser::parse_literal_string

pub fn parse_option_key(&mut self) -> Result<String, DataFusionError> {

let next_token = self.parser.next_token();

match next_token.token {

Token::Word(Word { value, .. }) => {

let mut parts = vec![value];

while self.parser.consume_token(&Token::Period) {

let next_token = self.parser.next_token();

if let Token::Word(Word { value, .. }) = next_token.token {

parts.push(value);

} else {

// Unquoted namespaced keys have to conform to the syntax

// "<WORD>[\.<WORD>]*". If we have a key that breaks this

// pattern, error out:

return self.expected("key name", &next_token);

}

}

Ok(parts.join("."))

}

Token::SingleQuotedString(s) => Ok(s),

Token::DoubleQuotedString(s) => Ok(s),

Token::EscapedStringLiteral(s) => Ok(s),

_ => self.expected("key name", &next_token),

}

}

/// Parse the next token as a value for an option list

///

/// Note this is different than [`parse_value`] as it allows any

/// word or keyword in this location.

///

/// [`parse_value`]: sqlparser::parser::Parser::parse_value

pub fn parse_option_value(&mut self) -> Result<Value, DataFusionError> {

let next_token = self.parser.next_token();

match next_token.token {

// e.g. things like "snappy" or "gzip" that may be keywords

Token::Word(word) => Ok(Value::SingleQuotedString(word.value)),

Token::SingleQuotedString(s) => Ok(Value::SingleQuotedString(s)),

Token::DoubleQuotedString(s) => Ok(Value::DoubleQuotedString(s)),

Token::EscapedStringLiteral(s) => Ok(Value::EscapedStringLiteral(s)),

Token::Number(n, l) => Ok(Value::Number(n, l)),

_ => self.expected("string or numeric value", &next_token),

}

}

/// Parse a SQL `EXPLAIN`

///

/// After the `EXPLAIN` keyword, if the dialect supports the Postgres-style

/// option list and the next non-whitespace token is `(`, we must

/// disambiguate between an option list (`EXPLAIN (ANALYZE) SELECT ...`)

/// and a parenthesized query (`EXPLAIN (SELECT ...)` or

/// `EXPLAIN (q1 EXCEPT q2) UNION ALL ...`).

pub fn parse_explain(&mut self) -> Result<Statement, DataFusionError> {

if self.supports_explain_with_utility_options

&& self.parser.peek_token().token == Token::LParen

&& !token_starts_query(&self.parser.peek_nth_token(1).token)

{

let raw = self.parser.parse_utility_options()?;

let options = ExplainStatementOptions::from_utility_options(&raw)?;

let statement = self.parse_statement()?;

return Ok(Statement::Explain(ExplainStatement {

statement: Box::new(statement),

options,

}));

}

// Legacy keyword form.

let analyze = self.parser.parse_keyword(Keyword::ANALYZE);

let verbose = self.parser.parse_keyword(Keyword::VERBOSE);

let format = self

.parse_explain_format()?

.map(|s| ExplainFormat::from_str(&s))

.transpose()?;

let statement = self.parse_statement()?;

let options = ExplainStatementOptions {

analyze,

verbose,

format,

..Default::default()

};

Ok(Statement::Explain(ExplainStatement {

statement: Box::new(statement),

options,

}))

}

/// Parse a SQL `RESET`

pub fn parse_reset(&mut self) -> Result<Statement, DataFusionError> {

let mut parts: Vec<String> = Vec::new();

let mut expecting_segment = true;

loop {

let next_token = self.parser.peek_token();

match &next_token.token {

Token::Word(word) => {

self.parser.next_token();

parts.push(word.value.clone());

expecting_segment = false;

}

Token::SingleQuotedString(s)

| Token::DoubleQuotedString(s)

| Token::EscapedStringLiteral(s) => {

self.parser.next_token();

parts.push(s.clone());

expecting_segment = false;

}

Token::Period => {

self.parser.next_token();

if expecting_segment || parts.is_empty() {

return self.expected("configuration parameter", &next_token);

}

expecting_segment = true;

}

Token::EOF | Token::SemiColon => break,

_ => return self.expected("configuration parameter", &next_token),

}

}

if parts.is_empty() || expecting_segment {

return self.expected("configuration parameter", &self.parser.peek_token());

}

let idents: Vec<Ident> = parts.into_iter().map(Ident::new).collect();

let variable = ObjectName::from(idents);

Ok(Statement::Reset(ResetStatement::Variable(variable)))

}

pub fn parse_explain_format(&mut self) -> Result<Option<String>, DataFusionError> {

if !self.parser.parse_keyword(Keyword::FORMAT) {

return Ok(None);

}

let next_token = self.parser.next_token();

let format = match next_token.token {

Token::Word(w) => Ok(w.value),

Token::SingleQuotedString(w) => Ok(w),

Token::DoubleQuotedString(w) => Ok(w),

_ => self.expected("an explain format such as TREE", &next_token),

}?;

Ok(Some(format))