feat: Rework streams data source (#3151)

<!-- Feel free to delete comments as you fill this in -->
- rework `streams` data source
- extract some common resource schema to be reused in next data sources
<!-- summary of changes -->

## Test Plan
<!-- detail ways in which this PR has been tested or needs to be tested
-->
* [x] acceptance tests
<!-- add more below if you think they are relevant -->

## References
<!-- issues documentation links, etc  -->
https://docs.snowflake.com/en/sql-reference/sql/show-streams

## TODO
- handle external change on stream type

MIGRATION_GUIDE.md

@@ -9,6 +9,24 @@ across different versions.
 
 ## v0.97.0 ➞ v0.98.0
 
+### snowflake_streams data source changes
+New filtering options:
+- `like`
+- `in`
+- `starts_with`
+- `limit`
+- `with_describe`
+
+New output fields
+- `show_output`
+- `describe_output`
+
+Breaking changes:
+- `database` and `schema` are right now under `in` field
+- `streams` field now organizes output of show under `show_output` field and the output of describe under `describe_output` field.
+
+Please adjust your Terraform configuration files.
+
 ### *(behavior change)* Provider configuration rework
 On our road to v1, we have decided to rework configuration to address the most common issues (see a [roadmap entry](https://github.com/Snowflake-Labs/terraform-provider-snowflake/blob/main/ROADMAP.md#providers-configuration-rework)). We have created a list of topics we wanted to address before v1. We will prepare an announcement soon. The following subsections describe the things addressed in the v0.98.0.
 

docs/data-sources/streams.md

@@ -2,12 +2,14 @@
 page_title: "snowflake_streams Data Source - terraform-provider-snowflake"
 subcategory: ""
 description: |-
-  
+  Datasource used to get details of filtered streams. Filtering is aligned with the current possibilities for SHOW STREAMS https://docs.snowflake.com/en/sql-reference/sql/show-streams query. The results of SHOW and DESCRIBE are encapsulated in one output collection streams.
 ---
 
-# snowflake_streams (Data Source)
+!> **V1 release candidate** This data source 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 data source if needed. Any errors reported will be resolved with a higher priority. We encourage checking this data source out before the V1 release. Please follow the [migration guide](https://github.com/Snowflake-Labs/terraform-provider-snowflake/blob/main/MIGRATION_GUIDE.md#v0970--v0980) to use it.
 
+# snowflake_streams (Data Source)
 
+Datasource used to get details of filtered streams. Filtering is aligned with the current possibilities for [SHOW STREAMS](https://docs.snowflake.com/en/sql-reference/sql/show-streams) query. The results of SHOW and DESCRIBE are encapsulated in one output collection `streams`.
 
 ## Example Usage
 
@@ -21,23 +23,90 @@ data "snowflake_streams" "current" {
 <!-- schema generated by tfplugindocs -->
 ## Schema
 
-### Required
+### Optional
 
-- `database` (String) The database from which to return the streams from.
-- `schema` (String) The schema from which to return the streams from.
+- `in` (Block List, Max: 1) IN clause to filter the list of objects (see [below for nested schema](#nestedblock--in))
+- `like` (String) Filters the output with **case-insensitive** pattern, with support for SQL wildcard characters (`%` and `_`).
+- `limit` (Block List, Max: 1) Limits the number of rows returned. If the `limit.from` is set, then the limit wll start from the first element matched by the expression. The expression is only used to match with the first element, later on the elements are not matched by the prefix, but you can enforce a certain pattern with `starts_with` or `like`. (see [below for nested schema](#nestedblock--limit))
+- `starts_with` (String) Filters the output with **case-sensitive** characters indicating the beginning of the object name.
+- `with_describe` (Boolean) Runs DESC STREAM for each user returned by SHOW STREAMS. The output of describe is saved to the description field. By default this value is set to true.
 
 ### Read-Only
 
 - `id` (String) The ID of this resource.
-- `streams` (List of Object) The streams in the schema (see [below for nested schema](#nestedatt--streams))
+- `streams` (List of Object) Holds the aggregated output of all streams details queries. (see [below for nested schema](#nestedatt--streams))
+
+<a id="nestedblock--in"></a>
+### Nested Schema for `in`
+
+Optional:
+
+- `account` (Boolean) Returns records for the entire account.
+- `application` (String) Returns records for the specified application.
+- `application_package` (String) Returns records for the specified application package.
+- `database` (String) Returns records for the current database in use or for a specified database.
+- `schema` (String) Returns records for the current schema in use or a specified schema. Use fully qualified name.
+
+
+<a id="nestedblock--limit"></a>
+### Nested Schema for `limit`
+
+Required:
+
+- `rows` (Number) The maximum number of rows to return.
+
+Optional:
+
+- `from` (String) Specifies a **case-sensitive** pattern that is used to match object name. After the first match, the limit on the number of rows will be applied.
+
 
 <a id="nestedatt--streams"></a>
 ### Nested Schema for `streams`
 
 Read-Only:
 
+- `describe_output` (List of Object) (see [below for nested schema](#nestedobjatt--streams--describe_output))
+- `show_output` (List of Object) (see [below for nested schema](#nestedobjatt--streams--show_output))
+
+<a id="nestedobjatt--streams--describe_output"></a>
+### Nested Schema for `streams.describe_output`
+
+Read-Only:
+
+- `base_tables` (List of String)
+- `comment` (String)
+- `created_on` (String)
+- `database_name` (String)
+- `invalid_reason` (String)
+- `mode` (String)
+- `name` (String)
+- `owner` (String)
+- `owner_role_type` (String)
+- `schema_name` (String)
+- `source_type` (String)
+- `stale` (Boolean)
+- `stale_after` (String)
+- `table_name` (String)
+- `type` (String)
+
+
+<a id="nestedobjatt--streams--show_output"></a>
+### Nested Schema for `streams.show_output`
+
+Read-Only:
+
+- `base_tables` (List of String)
 - `comment` (String)
-- `database` (String)
+- `created_on` (String)
+- `database_name` (String)
+- `invalid_reason` (String)
+- `mode` (String)
 - `name` (String)
-- `schema` (String)
-- `table` (String)
+- `owner` (String)
+- `owner_role_type` (String)
+- `schema_name` (String)
+- `source_type` (String)
+- `stale` (Boolean)
+- `stale_after` (String)
+- `table_name` (String)
+- `type` (String)

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

@@ -30,7 +30,7 @@ func (s *StreamShowOutputAssert) HasStaleAfterNotEmpty() *StreamShowOutputAssert
 	return s
 }
 
-func (s *StreamShowOutputAssert) HasBaseTables(ids []sdk.SchemaObjectIdentifier) *StreamShowOutputAssert {
+func (s *StreamShowOutputAssert) HasBaseTables(ids ...sdk.SchemaObjectIdentifier) *StreamShowOutputAssert {
 	s.AddAssertion(assert.ResourceShowOutputValueSet("base_tables.#", strconv.FormatInt(int64(len(ids)), 10)))
 	for i := range ids {
 		s.AddAssertion(assert.ResourceShowOutputValueSet(fmt.Sprintf("base_tables.%d", i), ids[i].FullyQualifiedName()))

pkg/datasources/common.go

@@ -1,10 +1,89 @@
 package datasources
 
 import (
+	"github.com/Snowflake-Labs/terraform-provider-snowflake/pkg/resources"
 	"github.com/Snowflake-Labs/terraform-provider-snowflake/pkg/sdk"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
 )
 
+var likeSchema = &schema.Schema{
+	Type:        schema.TypeString,
+	Optional:    true,
+	Description: "Filters the output with **case-insensitive** pattern, with support for SQL wildcard characters (`%` and `_`).",
+}
+
+var extendedInSchema = &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", "in.0.application", "in.0.application_package"},
+			},
+			"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", "in.0.application", "in.0.application_package"},
+				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", "in.0.application", "in.0.application_package"},
+				ValidateDiagFunc: resources.IsValidIdentifier[sdk.DatabaseObjectIdentifier](),
+			},
+			"application": {
+				Type:             schema.TypeString,
+				Optional:         true,
+				Description:      "Returns records for the specified application.",
+				ExactlyOneOf:     []string{"in.0.account", "in.0.database", "in.0.schema", "in.0.application", "in.0.application_package"},
+				ValidateDiagFunc: resources.IsValidIdentifier[sdk.AccountObjectIdentifier](),
+			},
+			"application_package": {
+				Type:             schema.TypeString,
+				Optional:         true,
+				Description:      "Returns records for the specified application package.",
+				ExactlyOneOf:     []string{"in.0.account", "in.0.database", "in.0.schema", "in.0.application", "in.0.application_package"},
+				ValidateDiagFunc: resources.IsValidIdentifier[sdk.AccountObjectIdentifier](),
+			},
+		},
+	},
+}
+
+var startsWithSchema = &schema.Schema{
+	Type:        schema.TypeString,
+	Optional:    true,
+	Description: "Filters the output with **case-sensitive** characters indicating the beginning of the object name.",
+}
+
+var limitFromSchema = &schema.Schema{
+	Type:        schema.TypeList,
+	Optional:    true,
+	Description: "Limits the number of rows returned. If the `limit.from` is set, then the limit wll start from the first element matched by the expression. The expression is only used to match with the first element, later on the elements are not matched by the prefix, but you can enforce a certain pattern with `starts_with` or `like`.",
+	MaxItems:    1,
+	Elem: &schema.Resource{
+		Schema: map[string]*schema.Schema{
+			"rows": {
+				Type:        schema.TypeInt,
+				Required:    true,
+				Description: "The maximum number of rows to return.",
+			},
+			"from": {
+				Type:        schema.TypeString,
+				Optional:    true,
+				Description: "Specifies a **case-sensitive** pattern that is used to match object name. After the first match, the limit on the number of rows will be applied.",
+			},
+		},
+	},
+}
+
 func handleLike(d *schema.ResourceData, setField **sdk.Like) {
 	if likePattern, ok := d.GetOk("like"); ok {
 		*setField = &sdk.Like{
@@ -13,6 +92,12 @@ func handleLike(d *schema.ResourceData, setField **sdk.Like) {
 	}
 }
 
+func handleStartsWith(d *schema.ResourceData, setField **string) {
+	if startsWith, ok := d.GetOk("starts_with"); ok {
+		*setField = sdk.String(startsWith.(string))
+	}
+}
+
 func handleLimitFrom(d *schema.ResourceData, setField **sdk.LimitFrom) {
 	if v, ok := d.GetOk("limit"); ok {
 		l := v.([]any)[0].(map[string]any)