LAV-2760: Support ALTER TASK schedule and config properties (#3193)

* LAV-2760: support task schedule and config mutations

Persist ALTER TASK schedule/config changes in the task catalog, validate Cloud-captured value and graph rules, expose them through metadata/DDL, and let scheduler polls consume altered hour/cron schedules.

Co-authored-by: Sabir Akhadov <sabir.akhadov@localstack.cloud>
## Test matrix
interval seconds/minutes/hours × valid boundaries  ->  tests/queries/test_tasks.py::test_alter_task_schedule_config_value_spaces
cron expression/timezone × valid and invalid paths  ->  tests/queries/test_tasks.py::test_alter_task_schedule_config_value_spaces
CONFIG JSON object/non-object × valid and invalid paths  ->  tests/queries/test_tasks.py::test_alter_task_schedule_config_value_spaces
root started/suspended × SET/UNSET  ->  tests/queries/test_tasks.py::test_alter_task_schedule_config_state_and_predecessor_rules
predecessor task × SET/UNSET  ->  tests/queries/test_tasks.py::test_alter_task_schedule_config_state_and_predecessor_rules
missing task × IF EXISTS/plain  ->  tests/queries/test_tasks.py::test_alter_task_schedule_config_missing
SET/UNSET × SHOW/DESCRIBE/GET_DDL  ->  tests/queries/test_tasks.py::test_alter_task_schedule_config_set_unset_round_trip
interval-to-cron alteration × later scheduler poll  ->  tests/queries/test_tasks.py::test_scheduler_observes_altered_task_schedule

## Deviations
Extended the vendored CREATE TASK AST with CONFIG so GET_DDL output can parse and recreate the altered state; the hint only pointed at task DDL/catalog machinery but did not name the parser fork.

* LAV-2760: preserve legacy task catalog entry point

Keep the pre-CONFIG __snowflake$create_task signature as a forwarding overload for catalog migration and legacy-row compatibility.

Co-authored-by: Sabir Akhadov <sabir.akhadov@localstack.cloud>
## Test matrix
legacy task creation UDF × predecessor persistence  ->  tests/queries/test_tasks.py::test_task_legacy_predecessor_relations

## Deviations
Added a compatibility overload after the full task-file check exposed the existing direct catalog-UDF caller; this is an incidental compatibility fix required by the new CONFIG argument.

* LAV-2760: close task property parity gaps

Validate cron schedules before catalog persistence with the scheduler's parser plus Snowflake-specific day-of-week rules, accept Cloud-supported CONFIG quoting, safely emit CONFIG in GET_DDL, and prove recreated metadata and changed scheduler timing.\n\nSwept ALTER TASK schedule/config for the reviewer pattern: malformed/range/special cron fields, interval signs/decimals/overflow, timezone case and aliases, dollar/single quote CONFIG forms, combined UNSET, metadata round-trip, and stale scheduler timing are covered.\n\n## Test matrix\ninterval signed/decimal/overflow × valid/error -> tests/queries/test_tasks.py::test_alter_task_schedule_config_extended_value_spaces\ncron field bounds/names/ranges/lists/steps/L × valid/error -> tests/queries/test_tasks.py::test_alter_task_schedule_config_extended_value_spaces\ntimezone case/alias × valid -> tests/queries/test_tasks.py::test_alter_task_schedule_config_extended_value_spaces\nCONFIG dollar/single/apostrophe quoting × valid -> tests/queries/test_tasks.py::test_alter_task_schedule_config_extended_value_spaces\ncombined UNSET schedule/config × valid -> tests/queries/test_tasks.py::test_alter_task_schedule_config_extended_value_spaces\nSHOW CONFIG + GET_DDL apostrophe + recreate × valid -> tests/queries/test_tasks.py::test_alter_task_schedule_config_ddl_recreates_state\ninterval-to-cron × obsolete due time/later cron poll -> tests/queries/test_tasks.py::test_scheduler_altered_cron_replaces_interval_due_time

Co-authored-by: Sabir Akhadov <sabir.akhadov@localstack.cloud>

* LAV-2760: resolve task error carrier collision

Move schedule/config mutation carriers to an unused SQLSTATE range after merging main's sibling ALTER TASK property mappings.\n\n## Test matrix\nschedule/config errors × wire mapping -> tests/queries/test_tasks.py::test_alter_task_schedule_config_value_spaces\nextended interval/cron errors × wire mapping -> tests/queries/test_tasks.py::test_alter_task_schedule_config_extended_value_spaces

Co-authored-by: Sabir Akhadov <sabir.akhadov@localstack.cloud>

* LAV-2760: internalize task cron validator

Suppress the default pg_extern SQL declaration and register the Rust cron validator only as the internal __snowflake$valid_task_cron helper, so its native text/bool signature does not expand the public callable surface.

Swept the ALTER TASK schedule/config implementation for native-typed pg_extern helpers; valid_task_cron was the only instance and its call site now uses the internal name.

Co-authored-by: Sabir Akhadov <sabir.akhadov@localstack.cloud>
## Test matrix
cron schedule validation × accepted/error paths -> tests/queries/test_tasks.py::test_alter_task_schedule_config_extended_value_spaces
cron validator SQL surface × internal namespace only -> generated extension SQL and pg_proc catalog inspection

* LAV-2760: collapse task catalog compatibility overload

Keep the legacy 19-argument task catalog call compatible through a trailing defaulted CONFIG argument, while routing new CREATE TASK calls through the same single catalog function. This removes the multi-overload internal function that tripped the global any/null overload gate without exempting it.

Swept the task catalog SQL surface for the overload-gate pattern; __snowflake$create_task was the only new multi-overload helper and now has one signature.

Co-authored-by: Sabir Akhadov <sabir.akhadov@localstack.cloud>
## Test matrix
legacy CREATE TASK catalog call × omitted CONFIG -> tests/queries/test_tasks.py::test_task_legacy_predecessor_relations
CREATE TASK with CONFIG × DDL recreate -> tests/queries/test_tasks.py::test_alter_task_schedule_config_ddl_recreates_state
internal task helper × overload policy -> tests/queries/functions/test_overload_gate.py::test_every_multi_overload_function_carries_any_and_null

---------

Co-authored-by: spiral <spiral@localhost>
Co-authored-by: Sabir Akhadov <sabir.akhadov@localstack.cloud>

Author: 3 people

src/ast/mod.rs

@@ -5235,6 +5235,8 @@ pub enum Statement {
         warehouse: Option<Ident>,
         /// Optional `SCHEDULE = '<string>'` clause (raw inner string).
         schedule: Option<String>,
+        /// Optional `CONFIG = '<json object>'` clause.
+        config: Option<String>,
         /// Optional `AFTER <task>[, <task>, ...]` clause.
         after: Vec<ObjectName>,
         /// Optional `WHEN <expr>` clause.
@@ -8768,6 +8770,7 @@ impl fmt::Display for Statement {
                 name,
                 warehouse,
                 schedule,
+                config,
                 after,
                 when_condition,
                 suspend_task_after_num_failures,
@@ -8791,6 +8794,9 @@ impl fmt::Display for Statement {
                 if let Some(s) = schedule {
                     write!(f, " SCHEDULE = '{s}'")?;
                 }
+                if let Some(c) = config {
+                    write!(f, " CONFIG = '{c}'")?;
+                }
                 if !after.is_empty() {
                     write!(f, " AFTER ")?;
                     for (i, n) in after.iter().enumerate() {

src/keywords.rs

@@ -276,6 +276,7 @@ define_keywords!(
     COMPUTE,
     CONCURRENTLY,
     CONDITION,
+    CONFIG,
     CONFLICT,
     CONNECT,
     CONNECTION,

src/parser/mod.rs

@@ -6033,6 +6033,7 @@ impl<'a> Parser<'a> {
         let name = self.parse_object_name(false)?;
         let mut warehouse: Option<Ident> = None;
         let mut schedule: Option<String> = None;
+        let mut config: Option<String> = None;
         let mut after: Vec<ObjectName> = Vec::new();
         let mut when_condition: Option<Expr> = None;
         let mut suspend_task_after_num_failures: Option<u64> = None;
@@ -6052,6 +6053,7 @@ impl<'a> Parser<'a> {
                     name,
                     warehouse,
                     schedule,
+                    config,
                     after,
                     when_condition,
                     suspend_task_after_num_failures,
@@ -6070,6 +6072,9 @@ impl<'a> Parser<'a> {
             } else if self.parse_keyword(Keyword::SCHEDULE) {
                 self.expect_token(&Token::Eq)?;
                 schedule = Some(self.parse_literal_string()?);
+            } else if self.parse_keyword(Keyword::CONFIG) {
+                self.expect_token(&Token::Eq)?;
+                config = Some(self.parse_literal_string()?);
             } else if self.parse_keyword(Keyword::AFTER) {
                 after = self.parse_comma_separated(|p| p.parse_object_name(false))?;
             } else if self.parse_keyword(Keyword::WHEN) {
@@ -6097,7 +6102,7 @@ impl<'a> Parser<'a> {
                 comment = Some(self.parse_literal_string()?);
             } else {
                 return self.expected(
-                    "WAREHOUSE, SCHEDULE, AFTER, WHEN, SUSPEND_TASK_AFTER_NUM_FAILURES, COMMENT, or AS in CREATE TASK",
+                    "WAREHOUSE, SCHEDULE, CONFIG, AFTER, WHEN, SUSPEND_TASK_AFTER_NUM_FAILURES, COMMENT, or AS in CREATE TASK",
                     self.peek_token(),
                 );
             }