localstack
diff --git a/‎src/ast/mod.rs‎
Lines changed: 15 additions & 0 deletions b/‎src/ast/mod.rs‎
Lines changed: 15 additions & 0 deletions
diff --git a/‎src/parser/mod.rs‎
Lines changed: 29 additions & 12 deletions b/‎src/parser/mod.rs‎
Lines changed: 29 additions & 12 deletions
@@ -5279,6 +5279,8 @@ pub enum Statement {
         task_auto_retry_attempts: Option<u64>,
         /// Optional `COMMENT = '<string>'` clause.
         comment: Option<String>,
+        /// Session parameters applied while the task body executes.
+        session_parameters: KeyValueOptions,
         /// Body executed by the task.
         sql_body: Box<Statement>,
     },
@@ -8840,6 +8842,7 @@ impl fmt::Display for Statement {
                 allow_overlapping_execution,
                 task_auto_retry_attempts,
                 comment,
+                session_parameters,
                 sql_body,
             } => {
                 write!(
@@ -8908,6 +8911,9 @@ impl fmt::Display for Statement {
                 if let Some(c) = comment {
                     write!(f, " COMMENT = '{c}'")?;
                 }
+                if !session_parameters.options.is_empty() {
+                    write!(f, " {session_parameters}")?;
+                }
                 write!(f, " AS {sql_body}")
             }
             Statement::AlterTask {
@@ -15849,6 +15855,10 @@ pub enum AlterTaskAction {
     SetOverlapPolicy(String),
     /// `UNSET OVERLAP_POLICY`
     UnsetOverlapPolicy,
+    /// `SET <session_parameter> = <value> [, ...]`
+    SetSessionParameters(KeyValueOptions),
+    /// `UNSET <session_parameter> [, ...]`
+    UnsetSessionParameters(Vec<Ident>),
 }
 
 impl fmt::Display for AlterTaskAction {
@@ -15906,6 +15916,11 @@ impl fmt::Display for AlterTaskAction {
                 write!(f, "SET OVERLAP_POLICY = {policy}")
             }
             AlterTaskAction::UnsetOverlapPolicy => write!(f, "UNSET OVERLAP_POLICY"),
+            AlterTaskAction::SetSessionParameters(options) => write!(f, "SET {options}"),
+            AlterTaskAction::UnsetSessionParameters(params) => {
+                write!(f, "UNSET ")?;
+                display_comma_separated(params).fmt(f)
+            }
         }
     }
 }
 
@@ -6051,6 +6051,10 @@ impl<'a> Parser<'a> {
         let mut allow_overlapping_execution: Option<bool> = None;
         let mut task_auto_retry_attempts: Option<u64> = None;
         let mut comment: Option<String> = None;
+        let mut session_parameters = KeyValueOptions {
+            delimiter: KeyValueOptionsDelimiter::Space,
+            options: Vec::new(),
+        };
 
         loop {
             if self.parse_keyword(Keyword::AS) {
@@ -6077,6 +6081,7 @@ impl<'a> Parser<'a> {
                     allow_overlapping_execution,
                     task_auto_retry_attempts,
                     comment,
+                    session_parameters,
                     sql_body,
                 });
             }
@@ -6133,10 +6138,16 @@ impl<'a> Parser<'a> {
                 self.expect_token(&Token::Eq)?;
                 comment = Some(self.parse_literal_string()?);
             } else {
-                return self.expected(
-                    "WAREHOUSE, SCHEDULE, CONFIG, AFTER, WHEN, SUSPEND_TASK_AFTER_NUM_FAILURES, COMMENT, or AS in CREATE TASK",
-                    self.peek_token(),
-                );
+                let token = self.next_token();
+                let Token::Word(word) = token.token else {
+                    return self.expected("task property or AS", token);
+                };
+                session_parameters
+                    .options
+                    .push(self.parse_key_value_option(&word, false)?);
+                if self.consume_token(&Token::Comma) {
+                    session_parameters.delimiter = KeyValueOptionsDelimiter::Comma;
+                }
             }
         }
     }
@@ -12973,10 +12984,17 @@ impl<'a> Parser<'a> {
                     },
                 )
             } else {
-                return self.expected(
-                    "WAREHOUSE, USER_TASK_MANAGED_INITIAL_WAREHOUSE_SIZE, or OVERLAP_POLICY after ALTER TASK SET",
-                    self.peek_token(),
-                );
+                let options = self.parse_comma_separated(|parser| {
+                    let word = parser.next_token();
+                    let Token::Word(word) = word.token else {
+                        return parser.expected("session parameter name", word);
+                    };
+                    parser.parse_key_value_option(&word, false)
+                })?;
+                AlterTaskAction::SetSessionParameters(KeyValueOptions {
+                    delimiter: KeyValueOptionsDelimiter::Comma,
+                    options,
+                })
             }
         } else if self.parse_keyword(Keyword::UNSET) {
             if self.parse_keyword(Keyword::WAREHOUSE) {
@@ -12994,10 +13012,9 @@ impl<'a> Parser<'a> {
             } else if self.parse_keyword(Keyword::OVERLAP_POLICY) {
                 AlterTaskAction::UnsetOverlapPolicy
             } else {
-                return self.expected(
-                    "WAREHOUSE, USER_TASK_MANAGED_INITIAL_WAREHOUSE_SIZE, or OVERLAP_POLICY after ALTER TASK UNSET",
-                    self.peek_token(),
-                );
+                AlterTaskAction::UnsetSessionParameters(
+                    self.parse_comma_separated(|parser| parser.parse_identifier())?,
+                )
             }
         } else {
             return self.expected(