feat: Task datasource and fixes (#3195)

## Changes
- Fixed the need to enclose `config` with $$
- Added full datasource support with tests
- Documentation and examples (v1 candidates; added for resource and
datasource)
  - Migration guide (now only for datasource)
  - Extracted schema for regular `in` filtering
- Describe wasn't added on purpose, because it seems to mirror values
from what we get from the SHOW command (can also add it, but for now it
doesn't provide any benefits)


## List from previous prs
- [ ] Apply comments
#3170
- [ ] Apply comments from
#3113
- [x] Cron
#3113 (comment)
- [ ] Resource logic
#3113 (comment)
- [ ] Refactor SDK suspend root logic for tasks (and overall suspend
logic in SDK/resource)
#3113 (comment)
- [ ] Move some of the logic to SDK (if possible)
- [ ]
#3170 (comment)
- [ ] Refactor task resuming in task resource (most likely with the use
of defer) because currently, there may be cases that error can cause
tasks to be not resumed.
- [ ] Tests
- [ ] External changes -
#3113 (comment)
- [ ] Add more complicated DAG structures to show the resource can
handle more complex structures
- [ ] Calling (`as` field) -
#3113 (comment)
- [ ] For showing how the DAG of tasks could be owned by a role other
than the one created with Terraform (also with less privileges, only to
run the task).
- [x] Check in one acceptance test why finalizer task in show_output is
not set (is that Snowflake or mapping error).
- [x] Data source
- [ ] Examples, documentation, and migration guide
- [ ] Keep manually changed files after regeneration
#3113 (comment)
- [x] Make config without $$ escapes needed
- [ ] Support session paramters
- [ ] Analyze non-deterministic test cases
- [ ] Check test tasks_gen_integration_test.go:937 (and see why it's non
deterministic).
- [ ] Re-generate and list all the issues with asserts and models

MIGRATION_GUIDE.md

@@ -6,6 +6,66 @@ across different versions.
 
 > [!TIP]
 > We highly recommend upgrading the versions one by one instead of bulk upgrades.
+ 
+## v0.98.0 ➞ v0.99.0
+
+### snowflake_task resource changes
+
+new fields:
+- `config` 
+
+### snowflake_tasks data source changes
+
+New filtering options:
+- `with_parameters`
+- `like`
+- `in`
+- `starts_with`
+- `root_only`
+- `limit`
+
+New output fields
+- `show_output`
+- `parameters`
+
+Breaking changes:
+- `database` and `schema` are right now under `in` field
+
+Before:
+```terraform
+data "snowflake_tasks" "old_tasks" {
+  database = "<database_name>"
+  schema = "<schema_name>"
+}
+```
+After:
+```terraform
+data "snowflake_tasks" "new_tasks" {
+  in {
+    # for IN SCHEMA specify:
+    schema = "<database_name>.<schema_name>"
+    
+    # for IN DATABASE specify:
+    database = "<database_name>"
+  }
+}
+```
+- `tasks` field now organizes output of show under `show_output` field and the output of show parameters under `parameters` field.
+
+Before:
+```terraform
+output "simple_output" {
+  value = data.snowflake_tasks.test.tasks[0].name
+}
+```
+After:
+```terraform
+output "simple_output" {
+  value = data.snowflake_tasks.test.tasks[0].show_output[0].name
+}
+```
+
+Please adjust your Terraform configuration files.
 
 ## v0.98.0 ➞ v0.99.0
 

docs/resources/task.md

@@ -5,6 +5,8 @@ description: |-
   Resource used to manage task objects. For more information, check task documentation https://docs.snowflake.com/en/user-guide/tasks-intro.
 ---
 
+!> **V1 release candidate** This resource was reworked and is a release candidate for the V1. We do not expect significant changes in it before the V1. We will welcome any feedback and adjust the resource if needed. Any errors reported will be resolved with a higher priority. We encourage checking this resource out before the V1 release. Please follow the [migration guide](https://github.com/Snowflake-Labs/terraform-provider-snowflake/blob/main/MIGRATION_GUIDE.md#v0980--v0990) to use it.
+
 # snowflake_task (Resource)
 
 Resource used to manage task objects. For more information, check [task documentation](https://docs.snowflake.com/en/user-guide/tasks-intro).
@@ -67,7 +69,6 @@ resource "snowflake_task" "test_task" {
   enabled                     = true
 }
 ```
-
 -> **Note** Instead of using fully_qualified_name, you can reference objects managed outside Terraform by constructing a correct ID, consult [identifiers guide](https://registry.terraform.io/providers/Snowflake-Labs/snowflake/latest/docs/guides/identifiers#new-computed-fully-qualified-name-field-in-resources).
 <!-- TODO(SNOW-1634854): include an example showing both methods-->
 

examples/data-sources/snowflake_tasks/data-source.tf

@@ -1,4 +1,120 @@
-data "snowflake_tasks" "current" {
-  database = "MYDB"
-  schema   = "MYSCHEMA"
-}
+# Simple usage
+data "snowflake_tasks" "simple" {
+}
+
+output "simple_output" {
+  value = data.snowflake_tasks.simple.tasks
+}
+
+# Filtering (like)
+data "snowflake_tasks" "like" {
+  like = "task-name"
+}
+
+output "like_output" {
+  value = data.snowflake_tasks.like.tasks
+}
+
+# Filtering (in - account - database - schema - application - application package)
+data "snowflake_tasks" "in_account" {
+  in {
+    account = true
+  }
+}
+
+data "snowflake_tasks" "in_database" {
+  in {
+    database = "<database_name>"
+  }
+}
+
+data "snowflake_tasks" "in_schema" {
+  in {
+    schema = "<database_name>.<schema_name>"
+  }
+}
+
+data "snowflake_tasks" "in_application" {
+  in {
+    application = "<application_name>"
+  }
+}
+
+data "snowflake_tasks" "in_application_package" {
+  in {
+    application_package = "<application_package_name>"
+  }
+}
+
+output "in_output" {
+  value = {
+    "account" : data.snowflake_tasks.in_account.tasks,
+    "database" : data.snowflake_tasks.in_database.tasks,
+    "schema" : data.snowflake_tasks.in_schema.tasks,
+    "application" : data.snowflake_tasks.in_application.tasks,
+    "application_package" : data.snowflake_tasks.in_application_package.tasks,
+  }
+}
+
+# Filtering (root only tasks)
+data "snowflake_tasks" "root_only" {
+  root_only = true
+}
+
+output "root_only_output" {
+  value = data.snowflake_tasks.root_only.tasks
+}
+
+# Filtering (starts_with)
+data "snowflake_tasks" "starts_with" {
+  starts_with = "task-"
+}
+
+output "starts_with_output" {
+  value = data.snowflake_tasks.starts_with.tasks
+}
+
+# Filtering (limit)
+data "snowflake_tasks" "limit" {
+  limit {
+    rows = 10
+    from = "task-"
+  }
+}
+
+output "limit_output" {
+  value = data.snowflake_tasks.limit.tasks
+}
+
+# Without additional data (to limit the number of calls make for every found task)
+data "snowflake_tasks" "only_show" {
+  # with_parameters is turned on by default and it calls SHOW PARAMETERS FOR task for every task found and attaches its output to tasks.*.parameters field
+  with_parameters = false
+}
+
+output "only_show_output" {
+  value = data.snowflake_tasks.only_show.tasks
+}
+
+# Ensure the number of tasks is equal to at least one element (with the use of postcondition)
+data "snowflake_tasks" "assert_with_postcondition" {
+  starts_with = "task-name"
+  lifecycle {
+    postcondition {
+      condition     = length(self.tasks) > 0
+      error_message = "there should be at least one task"
+    }
+  }
+}
+
+# Ensure the number of tasks is equal to at exactly one element (with the use of check block)
+check "task_check" {
+  data "snowflake_tasks" "assert_with_check_block" {
+    like = "task-name"
+  }
+
+  assert {
+    condition     = length(data.snowflake_tasks.assert_with_check_block.tasks) == 1
+    error_message = "tasks filtered by '${data.snowflake_tasks.assert_with_check_block.like}' returned ${length(data.snowflake_tasks.assert_with_check_block.tasks)} tasks where one was expected"
+  }
+}

pkg/acceptance/bettertestspoc/assert/resourceparametersassert/task_resource_parameters_ext.go

@@ -2,10 +2,23 @@ package resourceparametersassert
 
 import (
 	"strings"
+	"testing"
 
+	"github.com/Snowflake-Labs/terraform-provider-snowflake/pkg/acceptance/bettertestspoc/assert"
 	"github.com/Snowflake-Labs/terraform-provider-snowflake/pkg/sdk"
 )
 
+// TaskDatasourceParameters is a temporary workaround to have better parameter assertions in data source acceptance tests.
+func TaskDatasourceParameters(t *testing.T, name string) *TaskResourceParametersAssert {
+	t.Helper()
+
+	taskAssert := TaskResourceParametersAssert{
+		ResourceAssert: assert.NewDatasourceAssert("data."+name, "parameters", "tasks.0."),
+	}
+	taskAssert.AddAssertion(assert.ValueSet("parameters.#", "1"))
+	return &taskAssert
+}
+
 func (u *TaskResourceParametersAssert) HasAllDefaults() *TaskResourceParametersAssert {
 	return u.
 		HasSuspendTaskAfterNumFailures(10).

pkg/acceptance/bettertestspoc/assert/resourceshowoutputassert/task_show_output_ext.go

@@ -3,11 +3,28 @@ package resourceshowoutputassert
 import (
 	"fmt"
 	"strconv"
+	"testing"
 
 	"github.com/Snowflake-Labs/terraform-provider-snowflake/pkg/acceptance/bettertestspoc/assert"
 	"github.com/Snowflake-Labs/terraform-provider-snowflake/pkg/sdk"
 )
 
+// TaskDatasourceShowOutput is a temporary workaround to have better show output assertions in data source acceptance tests.
+func TaskDatasourceShowOutput(t *testing.T, name string) *TaskShowOutputAssert {
+	t.Helper()
+
+	taskAssert := TaskShowOutputAssert{
+		ResourceAssert: assert.NewDatasourceAssert("data."+name, "show_output", "tasks.0."),
+	}
+	taskAssert.AddAssertion(assert.ValueSet("show_output.#", "1"))
+	return &taskAssert
+}
+
+func (t *TaskShowOutputAssert) HasErrorIntegrationEmpty() *TaskShowOutputAssert {
+	t.AddAssertion(assert.ResourceShowOutputStringUnderlyingValueSet("error_integration", ""))
+	return t
+}
+
 func (t *TaskShowOutputAssert) HasCreatedOnNotEmpty() *TaskShowOutputAssert {
 	t.AddAssertion(assert.ResourceShowOutputValuePresent("created_on"))
 	return t
@@ -18,6 +35,11 @@ func (t *TaskShowOutputAssert) HasIdNotEmpty() *TaskShowOutputAssert {
 	return t
 }
 
+func (t *TaskShowOutputAssert) HasOwnerNotEmpty() *TaskShowOutputAssert {
+	t.AddAssertion(assert.ResourceShowOutputValuePresent("owner"))
+	return t
+}
+
 func (t *TaskShowOutputAssert) HasLastCommittedOnNotEmpty() *TaskShowOutputAssert {
 	t.AddAssertion(assert.ResourceShowOutputValuePresent("last_committed_on"))
 	return t

pkg/acceptance/helpers/ids_generator.go

@@ -81,6 +81,10 @@ func (c *IdsGenerator) RandomSchemaObjectIdentifierInSchema(schemaId sdk.Databas
 	return sdk.NewSchemaObjectIdentifierInSchema(schemaId, c.Alpha())
 }
 
+func (c *IdsGenerator) RandomSchemaObjectIdentifierInSchemaWithPrefix(prefix string, schemaId sdk.DatabaseObjectIdentifier) sdk.SchemaObjectIdentifier {
+	return sdk.NewSchemaObjectIdentifierInSchema(schemaId, c.AlphaWithPrefix(prefix))
+}
+
 func (c *IdsGenerator) RandomSchemaObjectIdentifierWithArgumentsOld(arguments ...sdk.DataType) sdk.SchemaObjectIdentifier {
 	return sdk.NewSchemaObjectIdentifierWithArgumentsOld(c.SchemaId().DatabaseName(), c.SchemaId().Name(), c.Alpha(), arguments)
 }

pkg/datasources/common.go

@@ -12,6 +12,37 @@ var likeSchema = &schema.Schema{
 	Description: "Filters the output with **case-insensitive** pattern, with support for SQL wildcard characters (`%` and `_`).",
 }
 
+var inSchema = &schema.Schema{
+	Type:        schema.TypeList,
+	Optional:    true,
+	Description: "IN clause to filter the list of objects",
+	MaxItems:    1,
+	Elem: &schema.Resource{
+		Schema: map[string]*schema.Schema{
+			"account": {
+				Type:         schema.TypeBool,
+				Optional:     true,
+				Description:  "Returns records for the entire account.",
+				ExactlyOneOf: []string{"in.0.account", "in.0.database", "in.0.schema"},
+			},
+			"database": {
+				Type:             schema.TypeString,
+				Optional:         true,
+				Description:      "Returns records for the current database in use or for a specified database.",
+				ExactlyOneOf:     []string{"in.0.account", "in.0.database", "in.0.schema"},
+				ValidateDiagFunc: resources.IsValidIdentifier[sdk.AccountObjectIdentifier](),
+			},
+			"schema": {
+				Type:             schema.TypeString,
+				Optional:         true,
+				Description:      "Returns records for the current schema in use or a specified schema. Use fully qualified name.",
+				ExactlyOneOf:     []string{"in.0.account", "in.0.database", "in.0.schema"},
+				ValidateDiagFunc: resources.IsValidIdentifier[sdk.DatabaseObjectIdentifier](),
+			},
+		},
+	},
+}
+
 var extendedInSchema = &schema.Schema{
 	Type:        schema.TypeList,
 	Optional:    true,
@@ -114,6 +145,29 @@ func handleLimitFrom(d *schema.ResourceData, setField **sdk.LimitFrom) {
 	}
 }
 
+func handleIn(d *schema.ResourceData, setField **sdk.In) error {
+	if v, ok := d.GetOk("in"); ok {
+		in := v.([]any)[0].(map[string]any)
+		accountValue, okAccount := in["account"]
+		databaseValue, okDatabase := in["database"]
+		schemaValue, okSchema := in["schema"]
+
+		switch {
+		case okAccount && accountValue.(bool):
+			*setField = &sdk.In{Account: sdk.Bool(true)}
+		case okDatabase && databaseValue.(string) != "":
+			*setField = &sdk.In{Database: sdk.NewAccountObjectIdentifier(databaseValue.(string))}
+		case okSchema && schemaValue.(string) != "":
+			schemaId, err := sdk.ParseDatabaseObjectIdentifier(schemaValue.(string))
+			if err != nil {
+				return err
+			}
+			*setField = &sdk.In{Schema: schemaId}
+		}
+	}
+	return nil
+}
+
 func handleExtendedIn(d *schema.ResourceData, setField **sdk.ExtendedIn) error {
 	if v, ok := d.GetOk("in"); ok {
 		in := v.([]any)[0].(map[string]any)