Method sql

crates/integrations/datafusion/src/sql_context.rs:294–430

Execute a SQL statement. ALTER TABLE is handled by Paimon directly; everything else is delegated to DataFusion.

(&self, sql: &str)

    /// Execute a SQL statement. ALTER TABLE is handled by Paimon directly;
    /// everything else is delegated to DataFusion.
    pub async fn sql(&self, sql: &str) -> DFResult<DataFrame> {
        let is_create_table = looks_like_create_table(sql);
        let (rewritten_sql, partition_keys) = if is_create_table {
            extract_partition_by(sql)?
        } else {
            (sql.to_string(), vec![])
        };
        if contains_time_travel_keyword(&rewritten_sql) {
            // Time-travel queries are not DDL; skip our own parsing and handle directly.
            return self.handle_time_travel_query(&rewritten_sql).await;
        }

        let statements = Parser::parse_sql(&GenericDialect {}, &rewritten_sql)
            .map_err(|e| DataFusionError::Plan(format!("SQL parse error: {e}")))?;

        if statements.len() != 1 {
            return Err(DataFusionError::Plan(
                "Expected exactly one SQL statement".to_string(),
            ));
        }

        match &statements[0] {
            Statement::CreateTable(create_table) => {
                if create_table.temporary {
                    self.handle_create_temp_table(create_table).await
                } else {
                    let (catalog, _catalog_name, _) =
                        self.resolve_catalog_and_table(&create_table.name)?;
                    self.handle_create_table(&catalog, create_table, partition_keys)
                        .await
                }
            }
            Statement::AlterTable(alter_table) => {
                let (catalog, _catalog_name, _) =
                    self.resolve_catalog_and_table(&alter_table.name)?;
                self.handle_alter_table(
                    &catalog,
                    &alter_table.name,
                    &alter_table.operations,
                    alter_table.if_exists,
                )
                .await
            }
            Statement::Merge(merge) => self.handle_merge_into(merge).await,
            Statement::Update(update) => self.handle_update(update).await,
            Statement::Delete(delete) => self.handle_delete(delete).await,
            Statement::Insert(insert)
                if insert.overwrite
                    && insert.partitioned.as_ref().is_some_and(|p| !p.is_empty()) =>
            {
                self.handle_insert_overwrite_partition(insert).await
            }
            Statement::Set(Set::SingleAssignment {
                variable, values, ..
            }) => {
                let key = variable.to_string();
                let key = key.trim_matches('\'').trim_matches('"');
                if let Some(paimon_key) = key.strip_prefix("paimon.") {