PostgreSQL: add support for CREATE AGGREGATE

fuzz/fuzz_seeds/create_aggregate.sql

@@ -0,0 +1,21 @@
+CREATE AGGREGATE public.my_avg(numeric) (
+    SFUNC = numeric_avg_accum,
+    STYPE = internal,
+    SSPACE = 128,
+    INITCOND = '0',
+    FINALFUNC = numeric_avg,
+    FINALFUNC_EXTRA,
+    FINALFUNC_MODIFY = SHAREABLE,
+    COMBINEFUNC = numeric_avg_combine,
+    SERIALFUNC = numeric_avg_serialize,
+    DESERIALFUNC = numeric_avg_deserialize,
+    MSFUNC = numeric_avg_accum,
+    MINVFUNC = numeric_accum_inv,
+    MSTYPE = internal,
+    MSSPACE = 64,
+    MFINALFUNC = numeric_avg,
+    MFINALFUNC_EXTRA,
+    MFINALFUNC_MODIFY = READ_WRITE,
+    MINITCOND = '0',
+    PARALLEL = safe
+)

fuzz/fuzz_seeds/create_aggregate_args.sql

@@ -0,0 +1 @@
+CREATE AGGREGATE s.my_agg(VARIADIC "any" ORDER BY VARIADIC "any") (SFUNC = s.f, STYPE = integer[], INITCOND = '{}', SORTOP = <, HYPOTHETICAL)

fuzz/fuzz_seeds/create_aggregate_legacy.sql

@@ -0,0 +1 @@
+CREATE AGGREGATE my_sum (BASETYPE = int4, SFUNC = int4pl, STYPE = int4, INITCOND = '0')

fuzz/fuzz_seeds/create_aggregate_sortop.sql

@@ -0,0 +1 @@
+CREATE AGGREGATE public.my_min(integer) (SFUNC = int4smaller, STYPE = integer, SORTOP = OPERATOR(pg_catalog.<))

fuzz/fuzz_seeds/create_aggregate_star.sql

@@ -0,0 +1 @@
+CREATE OR REPLACE AGGREGATE my_count(*) (SFUNC = int8inc, STYPE = bigint, INITCOND = '0')

fuzz/fuzz_seeds/postgres_2316_aggregate_output_arg.sql

@@ -0,0 +1 @@
+CREATE AGGREGATE a(OUT INT)(SFUNC=f,STYPE=INT)

src/ast/ddl.rs

@@ -6112,3 +6112,182 @@ impl Spanned for CreateForeignTable {
         )
     }
 }
+
+/// CREATE AGGREGATE statement.
+/// See <https://www.postgresql.org/docs/current/sql-createaggregate.html>
+#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+pub struct CreateAggregate {
+    /// True if `OR REPLACE` was specified.
+    pub or_replace: bool,
+    /// The aggregate name (can be schema-qualified).
+    pub name: ObjectName,
+    /// The input arguments, without output modes or defaults.
+    pub args: CreateAggregateArgs,
+    /// The options listed inside the required parentheses after the argument
+    /// list (e.g. `SFUNC`, `STYPE`, `FINALFUNC`, `PARALLEL`, …).
+    pub options: Vec<CreateAggregateOption>,
+}
+
+impl fmt::Display for CreateAggregate {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "CREATE")?;
+        if self.or_replace {
+            write!(f, " OR REPLACE")?;
+        }
+        write!(f, " AGGREGATE {}", self.name)?;
+        match &self.args {
+            CreateAggregateArgs::Legacy => {}
+            CreateAggregateArgs::Star => write!(f, " (*)")?,
+            CreateAggregateArgs::List(args) => write!(f, " ({})", display_comma_separated(args))?,
+            CreateAggregateArgs::OrderedSet { direct, aggregated } => {
+                write!(f, " (")?;
+                if !direct.is_empty() {
+                    write!(f, "{} ", display_comma_separated(direct))?;
+                }
+                write!(f, "ORDER BY {})", display_comma_separated(aggregated))?;
+            }
+        }
+        write!(f, " ({})", display_comma_separated(&self.options))
+    }
+}
+
+/// The argument list of a [`CreateAggregate`] statement.
+#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+pub enum CreateAggregateArgs {
+    /// No argument-list parentheses: the historical form that packs everything,
+    /// usually including a `BASETYPE` option, into a single list.
+    Legacy,
+    /// The wildcard form `(*)`, used by zero-argument aggregates such as
+    /// `count(*)`.
+    Star,
+    /// An explicit argument list: `(NUMERIC)`,
+    /// `(input INT, VARIADIC tail TEXT)`.
+    List(Vec<OperateFunctionArg>),
+    /// PostgreSQL ordered-set form: args before `ORDER BY` are direct args.
+    OrderedSet {
+        /// Arguments before `ORDER BY`.
+        direct: Vec<OperateFunctionArg>,
+        /// Arguments after `ORDER BY`.
+        aggregated: Vec<OperateFunctionArg>,
+    },
+}
+
+impl From<CreateAggregate> for crate::ast::Statement {
+    fn from(v: CreateAggregate) -> Self {
+        crate::ast::Statement::CreateAggregate(v)
+    }
+}
+
+/// A single option in a `CREATE AGGREGATE` options list.
+///
+/// See <https://www.postgresql.org/docs/current/sql-createaggregate.html>
+#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+pub enum CreateAggregateOption {
+    /// `SFUNC = state_transition_function`
+    StateTransitionFunction(ObjectName),
+    /// `STYPE = state_data_type`
+    StateDataType(DataType),
+    /// `SSPACE = state_data_size` (in bytes)
+    StateDataSize(u64),
+    /// `FINALFUNC = final_function`
+    FinalFunction(ObjectName),
+    /// `FINALFUNC_EXTRA`. Passes extra dummy arguments to the final function.
+    FinalFunctionExtra,
+    /// `FINALFUNC_MODIFY = { READ_ONLY | SHAREABLE | READ_WRITE }`
+    FinalFunctionModify(AggregateModifyKind),
+    /// `COMBINEFUNC = combine_function`
+    CombineFunction(ObjectName),
+    /// `SERIALFUNC = serial_function`
+    SerialFunction(ObjectName),
+    /// `DESERIALFUNC = deserial_function`
+    DeserialFunction(ObjectName),
+    /// `INITCOND = initial_condition` (a string literal)
+    InitialCondition(ValueWithSpan),
+    /// `MSFUNC = moving_state_transition_function`
+    MovingStateTransitionFunction(ObjectName),
+    /// `MINVFUNC = moving_inverse_transition_function`
+    MovingInverseTransitionFunction(ObjectName),
+    /// `MSTYPE = moving_state_data_type`
+    MovingStateDataType(DataType),
+    /// `MSSPACE = moving_state_data_size` (in bytes)
+    MovingStateDataSize(u64),
+    /// `MFINALFUNC = moving_final_function`
+    MovingFinalFunction(ObjectName),
+    /// `MFINALFUNC_EXTRA`
+    MovingFinalFunctionExtra,
+    /// `MFINALFUNC_MODIFY = { READ_ONLY | SHAREABLE | READ_WRITE }`
+    MovingFinalFunctionModify(AggregateModifyKind),
+    /// `MINITCOND = moving_initial_condition` (a string literal)
+    MovingInitialCondition(ValueWithSpan),
+    /// `SORTOP = sort_operator`
+    SortOperator(ObjectName),
+    /// `PARALLEL = { SAFE | RESTRICTED | UNSAFE }`
+    Parallel(FunctionParallel),
+    /// `HYPOTHETICAL`. Marks the aggregate as hypothetical-set.
+    Hypothetical,
+    /// `BASETYPE = base_type` (old aggregate syntax).
+    BaseType(DataType),
+}
+
+impl fmt::Display for CreateAggregateOption {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            Self::StateTransitionFunction(name) => write!(f, "SFUNC = {name}"),
+            Self::StateDataType(data_type) => write!(f, "STYPE = {data_type}"),
+            Self::StateDataSize(size) => write!(f, "SSPACE = {size}"),
+            Self::FinalFunction(name) => write!(f, "FINALFUNC = {name}"),
+            Self::FinalFunctionExtra => write!(f, "FINALFUNC_EXTRA"),
+            Self::FinalFunctionModify(kind) => write!(f, "FINALFUNC_MODIFY = {kind}"),
+            Self::CombineFunction(name) => write!(f, "COMBINEFUNC = {name}"),
+            Self::SerialFunction(name) => write!(f, "SERIALFUNC = {name}"),
+            Self::DeserialFunction(name) => write!(f, "DESERIALFUNC = {name}"),
+            Self::InitialCondition(cond) => write!(f, "INITCOND = {cond}"),
+            Self::MovingStateTransitionFunction(name) => write!(f, "MSFUNC = {name}"),
+            Self::MovingInverseTransitionFunction(name) => write!(f, "MINVFUNC = {name}"),
+            Self::MovingStateDataType(data_type) => write!(f, "MSTYPE = {data_type}"),
+            Self::MovingStateDataSize(size) => write!(f, "MSSPACE = {size}"),
+            Self::MovingFinalFunction(name) => write!(f, "MFINALFUNC = {name}"),
+            Self::MovingFinalFunctionExtra => write!(f, "MFINALFUNC_EXTRA"),
+            Self::MovingFinalFunctionModify(kind) => write!(f, "MFINALFUNC_MODIFY = {kind}"),
+            Self::MovingInitialCondition(cond) => write!(f, "MINITCOND = {cond}"),
+            Self::SortOperator(name) if name.0.len() > 1 => {
+                write!(f, "SORTOP = OPERATOR({name})")
+            }
+            Self::SortOperator(name) => write!(f, "SORTOP = {name}"),
+            Self::Parallel(parallel) => write!(f, "PARALLEL = {}", parallel.as_str()),
+            Self::Hypothetical => write!(f, "HYPOTHETICAL"),
+            Self::BaseType(data_type) => write!(f, "BASETYPE = {data_type}"),
+        }
+    }
+}
+
+/// Modifier kind for `FINALFUNC_MODIFY` / `MFINALFUNC_MODIFY` in `CREATE AGGREGATE`.
+///
+/// See <https://www.postgresql.org/docs/current/sql-createaggregate.html>
+#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+pub enum AggregateModifyKind {
+    /// The final function does not modify the transition state.
+    ReadOnly,
+    /// The transition state may be shared between aggregate calls.
+    Shareable,
+    /// The final function may modify the transition state.
+    ReadWrite,
+}
+
+impl fmt::Display for AggregateModifyKind {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            Self::ReadOnly => write!(f, "READ_ONLY"),
+            Self::Shareable => write!(f, "SHAREABLE"),
+            Self::ReadWrite => write!(f, "READ_WRITE"),
+        }
+    }
+}

src/ast/mod.rs

@@ -62,16 +62,17 @@ pub use self::dcl::{
     SetConfigValue, Use,
 };
 pub use self::ddl::{
-    Alignment, AlterCollation, AlterCollationOperation, AlterColumnOperation, AlterConnectorOwner,
-    AlterFunction, AlterFunctionAction, AlterFunctionKind, AlterFunctionOperation,
-    AlterIndexOperation, AlterOperator, AlterOperatorClass, AlterOperatorClassOperation,
-    AlterOperatorFamily, AlterOperatorFamilyOperation, AlterOperatorOperation, AlterPolicy,
-    AlterPolicyOperation, AlterSchema, AlterSchemaOperation, AlterTable, AlterTableAlgorithm,
-    AlterTableLock, AlterTableOperation, AlterTableType, AlterTextSearch, AlterTextSearchOperation,
-    AlterTextSearchOption, AlterType, AlterTypeAddValue, AlterTypeAddValuePosition,
-    AlterTypeOperation, AlterTypeRename, AlterTypeRenameValue, ClusteredBy, ColumnDef,
-    ColumnOption, ColumnOptionDef, ColumnOptions, ColumnPolicy, ColumnPolicyProperty,
-    ConstraintCharacteristics, CreateCollation, CreateCollationDefinition, CreateConnector,
+    AggregateModifyKind, Alignment, AlterCollation, AlterCollationOperation, AlterColumnOperation,
+    AlterConnectorOwner, AlterFunction, AlterFunctionAction, AlterFunctionKind,
+    AlterFunctionOperation, AlterIndexOperation, AlterOperator, AlterOperatorClass,
+    AlterOperatorClassOperation, AlterOperatorFamily, AlterOperatorFamilyOperation,
+    AlterOperatorOperation, AlterPolicy, AlterPolicyOperation, AlterSchema, AlterSchemaOperation,
+    AlterTable, AlterTableAlgorithm, AlterTableLock, AlterTableOperation, AlterTableType,
+    AlterTextSearch, AlterTextSearchOperation, AlterTextSearchOption, AlterType, AlterTypeAddValue,
+    AlterTypeAddValuePosition, AlterTypeOperation, AlterTypeRename, AlterTypeRenameValue,
+    ClusteredBy, ColumnDef, ColumnOption, ColumnOptionDef, ColumnOptions, ColumnPolicy,
+    ColumnPolicyProperty, ConstraintCharacteristics, CreateAggregate, CreateAggregateArgs,
+    CreateAggregateOption, CreateCollation, CreateCollationDefinition, CreateConnector,
     CreateDomain, CreateExtension, CreateForeignTable, CreateFunction, CreateIndex, CreateOperator,
     CreateOperatorClass, CreateOperatorFamily, CreatePolicy, CreatePolicyCommand, CreatePolicyType,
     CreateTable, CreateTextSearch, CreateTrigger, CreateView, Deduplicate, DeferrableInitial,
@@ -3848,6 +3849,11 @@ pub enum Statement {
     /// See [PostgreSQL](https://www.postgresql.org/docs/current/textsearch-intro.html)
     CreateTextSearch(CreateTextSearch),
     /// ```sql
+    /// CREATE AGGREGATE
+    /// ```
+    /// See [PostgreSQL](https://www.postgresql.org/docs/current/sql-createaggregate.html)
+    CreateAggregate(CreateAggregate),
+    /// ```sql
     /// ALTER TABLE
     /// ```
     AlterTable(AlterTable),
@@ -5686,6 +5692,7 @@ impl fmt::Display for Statement {
                 create_operator_family.fmt(f)
             }
             Statement::CreateOperatorClass(create_operator_class) => create_operator_class.fmt(f),
+            Statement::CreateAggregate(create_aggregate) => create_aggregate.fmt(f),
             Statement::CreateTextSearch(create_text_search) => create_text_search.fmt(f),
             Statement::AlterTable(alter_table) => write!(f, "{alter_table}"),
             Statement::AlterIndex { name, operation } => {
@@ -10352,16 +10359,23 @@ pub enum FunctionParallel {
     Safe,
 }
 
-impl fmt::Display for FunctionParallel {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+impl FunctionParallel {
+    /// Returns the bare keyword for this parallel mode, without the `PARALLEL` prefix.
+    pub fn as_str(&self) -> &'static str {
         match self {
-            FunctionParallel::Unsafe => write!(f, "PARALLEL UNSAFE"),
-            FunctionParallel::Restricted => write!(f, "PARALLEL RESTRICTED"),
-            FunctionParallel::Safe => write!(f, "PARALLEL SAFE"),
+            FunctionParallel::Unsafe => "UNSAFE",
+            FunctionParallel::Restricted => "RESTRICTED",
+            FunctionParallel::Safe => "SAFE",
         }
     }
 }
 
+impl fmt::Display for FunctionParallel {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "PARALLEL {}", self.as_str())
+    }
+}
+
 /// [BigQuery] Determinism specifier used in a UDF definition.
 ///
 /// [BigQuery]: https://cloud.google.com/bigquery/docs/reference/standard-sql/data-definition-language#syntax_11

src/ast/spans.rs

@@ -18,8 +18,8 @@
 use crate::{
     ast::{
         ddl::AlterSchema, query::SelectItemQualifiedWildcardKind, AlterSchemaOperation, AlterTable,
-        ColumnOptions, CreateOperator, CreateOperatorClass, CreateOperatorFamily, CreateView,
-        ExportData, Owner, TypedString,
+        ColumnOptions, CreateAggregate, CreateAggregateArgs, CreateAggregateOption, CreateOperator,
+        CreateOperatorClass, CreateOperatorFamily, CreateView, ExportData, Owner, TypedString,
     },
     tokenizer::TokenWithSpan,
 };
@@ -41,9 +41,9 @@ use super::{
     LimitClause, MatchRecognizePattern, Measure, Merge, MergeAction, MergeClause, MergeInsertExpr,
     MergeInsertKind, MergeUpdateExpr, MergeUpdateKind, NamedParenthesizedList,
     NamedWindowDefinition, ObjectName, ObjectNamePart, Offset, OnConflict, OnConflictAction,
-    OnInsert, OpenStatement, OrderBy, OrderByExpr, OrderByKind, OutputClause, Parens, Partition,
-    PartitionBoundValue, PivotValueSource, ProjectionSelect, Query, RaiseStatement,
-    RaiseStatementValue, ReferentialAction, RenameSelectItem, ReplaceSelectElement,
+    OnInsert, OpenStatement, OperateFunctionArg, OrderBy, OrderByExpr, OrderByKind, OutputClause,
+    Parens, Partition, PartitionBoundValue, PivotValueSource, ProjectionSelect, Query,
+    RaiseStatement, RaiseStatementValue, ReferentialAction, RenameSelectItem, ReplaceSelectElement,
     ReplaceSelectItem, Select, SelectInto, SelectItem, SetExpr, Setting, SqlOption, Statement,
     Subscript, SymbolDefinition, TableAlias, TableAliasColumnDef, TableConstraint, TableFactor,
     TableObject, TableOptionsClustered, TableWithJoins, Update, UpdateTableFromKind, Use, Values,
@@ -524,6 +524,7 @@ impl Spanned for Statement {
             Statement::Vacuum(..) => Span::empty(),
             Statement::AlterUser(..) => Span::empty(),
             Statement::Reset(..) => Span::empty(),
+            Statement::CreateAggregate(create_aggregate) => create_aggregate.span(),
         }
     }
 }
@@ -2554,6 +2555,68 @@ impl Spanned for AlterTable {
     }
 }
 
+impl Spanned for CreateAggregate {
+    fn span(&self) -> Span {
+        union_spans(
+            core::iter::once(self.name.span())
+                .chain(core::iter::once(self.args.span()))
+                .chain(self.options.iter().map(|option| option.span())),
+        )
+    }
+}
+
+impl Spanned for CreateAggregateArgs {
+    fn span(&self) -> Span {
+        match self {
+            CreateAggregateArgs::Legacy | CreateAggregateArgs::Star => Span::empty(),
+            CreateAggregateArgs::List(args) => union_spans(args.iter().map(|arg| arg.span())),
+            CreateAggregateArgs::OrderedSet { direct, aggregated } => {
+                union_spans(direct.iter().chain(aggregated).map(|arg| arg.span()))
+            }
+        }
+    }
+}
+
+impl Spanned for OperateFunctionArg {
+    fn span(&self) -> Span {
+        union_spans(
+            self.name
+                .iter()
+                .map(|name| name.span)
+                .chain(self.default_expr.iter().map(|expr| expr.span())),
+        )
+    }
+}
+
+impl Spanned for CreateAggregateOption {
+    fn span(&self) -> Span {
+        match self {
+            CreateAggregateOption::StateTransitionFunction(name)
+            | CreateAggregateOption::FinalFunction(name)
+            | CreateAggregateOption::CombineFunction(name)
+            | CreateAggregateOption::SerialFunction(name)
+            | CreateAggregateOption::DeserialFunction(name)
+            | CreateAggregateOption::MovingStateTransitionFunction(name)
+            | CreateAggregateOption::MovingInverseTransitionFunction(name)
+            | CreateAggregateOption::MovingFinalFunction(name)
+            | CreateAggregateOption::SortOperator(name) => name.span(),
+            CreateAggregateOption::InitialCondition(value)
+            | CreateAggregateOption::MovingInitialCondition(value) => value.span,
+            CreateAggregateOption::StateDataType(_)
+            | CreateAggregateOption::StateDataSize(_)
+            | CreateAggregateOption::FinalFunctionExtra
+            | CreateAggregateOption::FinalFunctionModify(_)
+            | CreateAggregateOption::MovingStateDataType(_)
+            | CreateAggregateOption::MovingStateDataSize(_)
+            | CreateAggregateOption::MovingFinalFunctionExtra
+            | CreateAggregateOption::MovingFinalFunctionModify(_)
+            | CreateAggregateOption::Parallel(_)
+            | CreateAggregateOption::Hypothetical
+            | CreateAggregateOption::BaseType(_) => Span::empty(),
+        }
+    }
+}
+
 impl Spanned for CreateOperator {
     fn span(&self) -> Span {
         Span::empty()