LAV-2128: snowflake_task AFTER/predecessor relations don't round-trip → OnTask_Discussion2877 step 2/4 refresh-plan drift (#2723)

* LAV-2128: round-trip task predecessor relations

Persist fully qualified, lexically ordered task predecessor JSON for CREATE TASK and ALTER TASK ADD/REMOVE AFTER, and project matching SHOW/DESC relation metadata. Add a real-Snowflake snapshot covering create, add, remove, ordering, and root-task empty relations.

* LAV-2128: fix task parser test construction

Construct expected task names through ObjectName's conversion API so the vendored parser test builds with the current multipart-name representation.

* LAV-2128: preserve legacy task predecessor rows

Normalize legacy comma-separated predecessor values when projecting or mutating task relations, while retaining ordered JSON for new writes. Add a Snowflake-captured regression for upgrading and mutating an existing row and document both catalog encodings.

## Deviations

- **Catalog representation**: the architecture document described comma-separated storage. The earlier implementation retained JSON storage for new rows; this round documents it and adds transparent compatibility for legacy comma-separated rows instead of migrating catalog data eagerly.

* LAV-2128: qualify two-part legacy predecessors

Qualify two-part legacy task predecessor catalog values with the task database before SHOW/DESC projection and ADD/REMOVE normalization. Exercise the path by injecting a two-part legacy value in the existing Snowflake-captured compat test.

* LAV-2128: update task alter callers and docs

Pass the predecessor argument from SYSTEM$TASK_DEPENDENTS_ENABLE and document the complete ALTER TASK UDF surface, including predecessor mutations.

---------

Co-authored-by: root <root@ip-10-42-44-3.eu-west-1.compute.internal>
Co-authored-by: root <root@ip-10-42-42-116.eu-west-1.compute.internal>

Author: 3 people

src/ast/mod.rs

@@ -14869,21 +14869,35 @@ impl fmt::Display for AlterAccountOperation {
 /// Action for [`Statement::AlterTask`].
 ///
 /// See <https://docs.snowflake.com/en/sql-reference/sql/alter-task>.
-#[derive(Debug, Copy, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 #[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
 pub enum AlterTaskAction {
     /// `RESUME`
     Resume,
     /// `SUSPEND`
     Suspend,
+    /// `ADD AFTER <task> [, <task> ...]`
+    AddAfter(Vec<ObjectName>),
+    /// `REMOVE AFTER <task> [, <task> ...]`
+    RemoveAfter(Vec<ObjectName>),
 }
 
 impl fmt::Display for AlterTaskAction {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         match self {
             AlterTaskAction::Resume => write!(f, "RESUME"),
             AlterTaskAction::Suspend => write!(f, "SUSPEND"),
+            AlterTaskAction::AddAfter(tasks) => write!(
+                f,
+                "ADD AFTER {}",
+                display_comma_separated(tasks)
+            ),
+            AlterTaskAction::RemoveAfter(tasks) => write!(
+                f,
+                "REMOVE AFTER {}",
+                display_comma_separated(tasks)
+            ),
         }
     }
 }

src/parser/mod.rs

@@ -12379,16 +12379,29 @@ impl<'a> Parser<'a> {
         Ok(Statement::AlterAccount { name, operation })
     }
 
-    /// Parse `ALTER TASK [IF EXISTS] <name> { RESUME | SUSPEND }`.
+    /// Parse `ALTER TASK [IF EXISTS] <name> { RESUME | SUSPEND | { ADD | REMOVE } AFTER ... }`.
     pub fn parse_alter_task(&mut self) -> Result<Statement, ParserError> {
         let if_exists = self.parse_keywords(&[Keyword::IF, Keyword::EXISTS]);
         let name = self.parse_object_name(false)?;
         let action = if self.parse_keyword(Keyword::RESUME) {
             AlterTaskAction::Resume
         } else if self.parse_keyword(Keyword::SUSPEND) {
             AlterTaskAction::Suspend
+        } else if self.parse_keyword(Keyword::ADD) {
+            self.expect_keyword(Keyword::AFTER)?;
+            AlterTaskAction::AddAfter(
+                self.parse_comma_separated(|parser| parser.parse_object_name(false))?,
+            )
+        } else if self.parse_keyword(Keyword::REMOVE) {
+            self.expect_keyword(Keyword::AFTER)?;
+            AlterTaskAction::RemoveAfter(
+                self.parse_comma_separated(|parser| parser.parse_object_name(false))?,
+            )
         } else {
-            return self.expected("RESUME or SUSPEND after ALTER TASK", self.peek_token());
+            return self.expected(
+                "RESUME, SUSPEND, ADD AFTER, or REMOVE AFTER after ALTER TASK",
+                self.peek_token(),
+            );
         };
         Ok(Statement::AlterTask {
             if_exists,

tests/sqlparser_snowflake.rs

@@ -8173,6 +8173,29 @@ fn test_alter_task_resume() {
         }
         _ => unreachable!(),
     }
+
+    for (sql, expected) in [
+        (
+            "ALTER TASK foo ADD AFTER root, db.schema.other",
+            AlterTaskAction::AddAfter(vec![
+                ObjectName::from(vec![Ident::new("root")]),
+                ObjectName::from(vec![
+                    Ident::new("db"),
+                    Ident::new("schema"),
+                    Ident::new("other"),
+                ]),
+            ]),
+        ),
+        (
+            "ALTER TASK foo REMOVE AFTER root",
+            AlterTaskAction::RemoveAfter(vec![ObjectName::from(vec![Ident::new("root")])]),
+        ),
+    ] {
+        match snowflake().verified_stmt(sql) {
+            Statement::AlterTask { action, .. } => assert_eq!(expected, action),
+            _ => unreachable!(),
+        }
+    }
 }
 
 #[test]