feat: Add stream on directory table

CREATING_ISSUES.md

@@ -202,7 +202,20 @@ resource "snowflake_grant_privileges_to_account_role" "grant_on_procedure" {
   account_role_name = snowflake_account_role.name
   on_schema_object {
     object_type = "PROCEDURE"
-    object_name = "\"${snowflake_database.database.name}\".\"${snowflake_schema.schema.name}\".\"${snowflake_procedure.procedure.name}(NUMBER, VARCHAR)\""
+    object_name = "\"${snowflake_database.database.name}\".\"${snowflake_schema.schema.name}\".\"${snowflake_procedure.procedure.name}\"(NUMBER, VARCHAR)"
+  }
+}
+```
+
+If you manage the procedure in Terraform, you can use `fully_qualified_name` field:
+
+```terraform
+resource "snowflake_grant_privileges_to_account_role" "grant_on_procedure" {
+  privileges        = ["USAGE"]
+  account_role_name = snowflake_account_role.name
+  on_schema_object {
+    object_type = "PROCEDURE"
+    object_name = snowflake_procedure.procedure_name.fully_qualified_name
   }
 }
 ```

MIGRATION_GUIDE.md

@@ -9,13 +9,14 @@ across different versions.
 
 ## v0.97.0 ➞ v0.98.0
 
+#### *(behavior change)* handling copy_grants
+Currently, resources like `snowflake_view`, `snowflake_stream_on_table`, `snowflake_stream_on_external_table` and `snowflake_stream_on_directory_table`  support `copy_grants` field corresponding with `COPY GRANTS` during `CREATE`. The current behavior is that, when a change leading for recreation is detected (meaning a change that can not be handled by ALTER, but only by `CREATE OR REPLACE`), `COPY GRANTS` are used during recreation when `copy_grants` is set to `true`. Changing this field without changes in other field results in a noop because in this case there is no need to recreate a resource.
+
+### *(new feature)* recovering stale streams
+Starting from this version, the provider detects stale streams for `snowflake_stream_on_table`, `snowflake_stream_on_external_table` and `snowflake_stream_on_directory_table` and recreates them (optionally with `copy_grants`) to recover them. To handle this correctly, a new computed-only field `stale` has been added to these resource, indicating whether a stream is stale.
+
 ### *(new feature)* snowflake_stream_on_directory_table resource
-Continuing changes made in v0.96, to enhance clarity and functionality, the new resource `snowflake_stream_on_directory_table` has been introduced to replace the previous `snowflake_stream`. Recognizing that the old resource carried multiple responsibilities within a single entity, we opted to divide it into more specialized resources.
-The newly introduced resources are aligned with the latest Snowflake documentation at the time of implementation, and adhere to our [new conventions](#general-changes).
-This segregation was based on the object on which the stream is created. The mapping between SQL statements and the resources is the following:
-- `ON TABLE <table_name>` -> `snowflake_stream_on_table` (added in v0.96)
-- `ON EXTERNAL TABLE <external_table_name>` -> `snowflake_stream_on_external_table` (this was previously not supported, added in v0.96)
-- `ON STAGE <stage_name>` -> `snowflake_stream_on_directory_table`
+Continuing changes made in [v0.97](#v0960--v0970), the new resource `snowflake_stream_on_directory_table` has been introduced to replace the previous `snowflake_stream` for streams on directory tables.
 
 To use the new `stream_on_directory_table`, change the old `stream` from
 ```terraform
@@ -56,7 +57,6 @@ The newly introduced resources are aligned with the latest Snowflake documentati
 This segregation was based on the object on which the stream is created. The mapping between SQL statements and the resources is the following:
 - `ON TABLE <table_name>` -> `snowflake_stream_on_table`
 - `ON EXTERNAL TABLE <external_table_name>` -> `snowflake_stream_on_external_table` (this was previously not supported)
-- `ON STAGE <stage_name>` -> `snowflake_stream_on_directory_table`
 
 The resources for streams on directory tables and streams on views will be implemented in the future releases.
 

docs/guides/identifiers.md

@@ -15,7 +15,7 @@ For example, instead of writing
 ```
 object_name = “\”${snowflake_table.database}\”.\”${snowflake_table.schema}\”.\”${snowflake_table.name}\””
 # for procedures
-object_name = “\”${snowflake_procedure.database}\”.\”${snowflake_procedure.schema}\”.\”${snowflake_procedure.name}(NUMBER, VARCHAR)\””
+object_name = “\”${snowflake_procedure.database}\”.\”${snowflake_procedure.schema}\”.\”${snowflake_procedure.name}\"(NUMBER, VARCHAR)”
 ```
 
 now we can write

docs/resources/stream_on_directory_table.md

@@ -90,7 +90,7 @@ Read-Only:
 - `owner_role_type` (String)
 - `schema_name` (String)
 - `source_type` (String)
-- `stale` (String)
+- `stale` (Boolean)
 - `stale_after` (String)
 - `table_name` (String)
 - `type` (String)
@@ -112,7 +112,7 @@ Read-Only:
 - `owner_role_type` (String)
 - `schema_name` (String)
 - `source_type` (String)
-- `stale` (String)
+- `stale` (Boolean)
 - `stale_after` (String)
 - `table_name` (String)
 - `type` (String)

docs/resources/stream_on_external_table.md

@@ -129,7 +129,7 @@ Read-Only:
 - `owner_role_type` (String)
 - `schema_name` (String)
 - `source_type` (String)
-- `stale` (String)
+- `stale` (Boolean)
 - `stale_after` (String)
 - `table_name` (String)
 - `type` (String)
@@ -151,7 +151,7 @@ Read-Only:
 - `owner_role_type` (String)
 - `schema_name` (String)
 - `source_type` (String)
-- `stale` (String)
+- `stale` (Boolean)
 - `stale_after` (String)
 - `table_name` (String)
 - `type` (String)

docs/resources/stream_on_table.md

@@ -114,7 +114,7 @@ Read-Only:
 - `owner_role_type` (String)
 - `schema_name` (String)
 - `source_type` (String)
-- `stale` (String)
+- `stale` (Boolean)
 - `stale_after` (String)
 - `table_name` (String)
 - `type` (String)
@@ -136,7 +136,7 @@ Read-Only:
 - `owner_role_type` (String)
 - `schema_name` (String)
 - `source_type` (String)
-- `stale` (String)
+- `stale` (Boolean)
 - `stale_after` (String)
 - `table_name` (String)
 - `type` (String)

pkg/acceptance/helpers/stage_client.go

@@ -44,9 +44,9 @@ func (c *StageClient) CreateStageWithURL(t *testing.T, id sdk.SchemaObjectIdenti
 	return stage, c.DropStageFunc(t, id)
 }
 
-func (c *StageClient) CreateStageWithDirectory(t *testing.T, schemaId sdk.DatabaseObjectIdentifier) (*sdk.Stage, func()) {
+func (c *StageClient) CreateStageWithDirectory(t *testing.T) (*sdk.Stage, func()) {
 	t.Helper()
-	id := c.ids.RandomSchemaObjectIdentifierInSchema(schemaId)
+	id := c.ids.RandomSchemaObjectIdentifier()
 	return c.CreateStageWithRequest(t, sdk.NewCreateInternalStageRequest(id).WithDirectoryTableOptions(sdk.NewInternalDirectoryTableOptionsRequest().WithEnable(sdk.Bool(true))))
 }
 

pkg/resources/custom_diffs.go

@@ -196,6 +196,8 @@ func RecreateWhenUserTypeChangedExternally(userType sdk.UserType) schema.Customi
 	}
 }
 
+// RecreateWhenStreamIsStale detects when the stream is stale, and sets a `false` value for `stale` field.
+// This means that the provider can detect that change in `stale` from `true` to `false`, where `false` is our desired state.
 func RecreateWhenStreamIsStale() schema.CustomizeDiffFunc {
 	return func(_ context.Context, diff *schema.ResourceDiff, _ interface{}) error {
 		if old, _ := diff.GetChange("stale"); old.(bool) {

pkg/resources/resource_helpers_create.go

@@ -65,7 +65,7 @@ func attributeDirectValueCreate[T any](d *schema.ResourceData, key string, creat
 func copyGrantsAttributeCreate(d *schema.ResourceData, isOrReplace bool, orReplaceField, copyGrantsField **bool) error {
 	if isOrReplace {
 		*orReplaceField = sdk.Bool(true)
-		if d.Get("copy_grants").(bool) {
+		if d.GetRawConfig().AsValueMap()["copy_grants"].True() {
 			*copyGrantsField = sdk.Bool(true)
 		}
 	}

pkg/resources/stream_common.go

@@ -40,7 +40,7 @@ var streamCommonSchema = map[string]*schema.Schema{
 		Default:     false,
 		Description: "Retains the access permissions from the original stream when a stream is recreated using the OR REPLACE clause. That is sometimes used when the provider detects changes for fields that can not be changed by ALTER. This value will not have any effect when creating a new stream.",
 		// Changing ONLY copy grants should have no effect. It is only used as an "option" during CREATE OR REPLACE - when other attributes change, it's not an object state. There is no point in recreating the object when only this field is changed.
-		DiffSuppressFunc: IgnoreAlways,
+		DiffSuppressFunc: IgnoreAfterCreation,
 	},
 	"stale": {
 		Type:        schema.TypeBool,
@@ -201,15 +201,11 @@ func handleStreamRead(d *schema.ResourceData,
 	stream *sdk.Stream,
 	streamDescription *sdk.Stream,
 ) error {
-	stale, err := booleanStringToBool(*stream.Stale)
-	if err != nil {
-		return err
-	}
 	return errors.Join(
 		d.Set("comment", stream.Comment),
 		d.Set(ShowOutputAttributeName, []map[string]any{schemas.StreamToSchema(stream)}),
 		d.Set(DescribeOutputAttributeName, []map[string]any{schemas.StreamDescriptionToSchema(*streamDescription)}),
 		d.Set(FullyQualifiedNameAttributeName, id.FullyQualifiedName()),
-		d.Set("stale", stale),
+		d.Set("stale", stream.Stale),
 	)
 }

pkg/resources/stream_on_directory_table_acceptance_test.go

@@ -29,7 +29,7 @@ func TestAcc_StreamOnDirectoryTable_Basic(t *testing.T) {
 	resourceId := helpers.EncodeResourceIdentifier(id)
 	resourceName := "snowflake_stream_on_directory_table.test"
 
-	stage, cleanupStage := acc.TestClient().Stage.CreateStageWithDirectory(t, acc.TestClient().Ids.SchemaId())
+	stage, cleanupStage := acc.TestClient().Stage.CreateStageWithDirectory(t)
 	t.Cleanup(cleanupStage)
 
 	baseModel := func() *model.StreamOnDirectoryTableModel {
@@ -272,7 +272,7 @@ func TestAcc_StreamOnDirectoryTable_CopyGrants(t *testing.T) {
 
 	var createdOn string
 
-	stage, cleanupStage := acc.TestClient().Stage.CreateStageWithDirectory(t, acc.TestClient().Ids.SchemaId())
+	stage, cleanupStage := acc.TestClient().Stage.CreateStageWithDirectory(t)
 	t.Cleanup(cleanupStage)
 
 	model := model.StreamOnDirectoryTable("test", id.DatabaseName(), id.Name(), id.SchemaName(), stage.ID().FullyQualifiedName())
@@ -299,7 +299,7 @@ func TestAcc_StreamOnDirectoryTable_CopyGrants(t *testing.T) {
 					HasNameString(id.Name()),
 					assert.Check(resource.TestCheckResourceAttrWith(resourceName, "show_output.0.created_on", func(value string) error {
 						if value != createdOn {
-							return fmt.Errorf("view was recreated")
+							return fmt.Errorf("stream was recreated")
 						}
 						return nil
 					})),
@@ -311,7 +311,7 @@ func TestAcc_StreamOnDirectoryTable_CopyGrants(t *testing.T) {
 					HasNameString(id.Name()),
 					assert.Check(resource.TestCheckResourceAttrWith(resourceName, "show_output.0.created_on", func(value string) error {
 						if value != createdOn {
-							return fmt.Errorf("view was recreated")
+							return fmt.Errorf("stream was recreated")
 						}
 						return nil
 					})),
@@ -321,6 +321,86 @@ func TestAcc_StreamOnDirectoryTable_CopyGrants(t *testing.T) {
 	})
 }
 
+func TestAcc_StreamOnDirectoryTable_CheckGrantsAfterRecreation(t *testing.T) {
+	_ = testenvs.GetOrSkipTest(t, testenvs.EnableAcceptance)
+	acc.TestAccPreCheck(t)
+	id := acc.TestClient().Ids.RandomSchemaObjectIdentifier()
+	resourceName := "snowflake_stream_on_directory_table.test"
+
+	stage, cleanupStage := acc.TestClient().Stage.CreateStageWithDirectory(t)
+	t.Cleanup(cleanupStage)
+
+	stage2, cleanupStage2 := acc.TestClient().Stage.CreateStageWithDirectory(t)
+	t.Cleanup(cleanupStage2)
+
+	role, cleanupRole := acc.TestClient().Role.CreateRole(t)
+	t.Cleanup(cleanupRole)
+
+	model1 := model.StreamOnDirectoryTable("test", id.DatabaseName(), id.Name(), id.SchemaName(), stage.ID().FullyQualifiedName()).WithCopyGrants(true)
+	model1WithoutCopyGrants := model.StreamOnDirectoryTable("test", id.DatabaseName(), id.Name(), id.SchemaName(), stage.ID().FullyQualifiedName())
+	model2 := model.StreamOnDirectoryTable("test", id.DatabaseName(), id.Name(), id.SchemaName(), stage2.ID().FullyQualifiedName()).WithCopyGrants(true)
+
+	resource.Test(t, resource.TestCase{
+		ProtoV6ProviderFactories: acc.TestAccProtoV6ProviderFactories,
+		PreCheck:                 func() { acc.TestAccPreCheck(t) },
+		TerraformVersionChecks: []tfversion.TerraformVersionCheck{
+			tfversion.RequireAbove(tfversion.Version1_5_0),
+		},
+		CheckDestroy: acc.CheckDestroy(t, resources.StreamOnDirectoryTable),
+		Steps: []resource.TestStep{
+			{
+				Config: config.FromModel(t, model1) + grantStreamPrivilegesConfig(resourceName, role.ID()),
+				Check: resource.ComposeAggregateTestCheckFunc(
+					// there should be more than one privilege, because we applied grant all privileges and initially there's always one which is ownership
+					resource.TestCheckResourceAttr("data.snowflake_grants.grants", "grants.#", "2"),
+					resource.TestCheckResourceAttr("data.snowflake_grants.grants", "grants.1.privilege", "SELECT"),
+				),
+			},
+			{
+				Config: config.FromModel(t, model2) + grantStreamPrivilegesConfig(resourceName, role.ID()),
+				Check: resource.ComposeAggregateTestCheckFunc(
+					resource.TestCheckResourceAttr("data.snowflake_grants.grants", "grants.#", "2"),
+					resource.TestCheckResourceAttr("data.snowflake_grants.grants", "grants.1.privilege", "SELECT"),
+				),
+			},
+			{
+				Config:             config.FromModel(t, model1WithoutCopyGrants) + grantStreamPrivilegesConfig(resourceName, role.ID()),
+				ExpectNonEmptyPlan: true,
+				ConfigPlanChecks: resource.ConfigPlanChecks{
+					PostApplyPostRefresh: []plancheck.PlanCheck{
+						plancheck.ExpectResourceAction("snowflake_grant_privileges_to_account_role.grant", plancheck.ResourceActionUpdate),
+					},
+				},
+				Check: resource.ComposeAggregateTestCheckFunc(
+					resource.TestCheckResourceAttr("data.snowflake_grants.grants", "grants.#", "1"),
+				),
+			},
+		},
+	})
+}
+
+func grantStreamPrivilegesConfig(resourceName string, roleId sdk.AccountObjectIdentifier) string {
+	return fmt.Sprintf(`
+resource "snowflake_grant_privileges_to_account_role" "grant" {
+  privileges        = ["SELECT"]
+  account_role_name = %[1]s
+  on_schema_object {
+    object_type = "STREAM"
+    object_name = %[2]s.fully_qualified_name
+  }
+}
+
+data "snowflake_grants" "grants" {
+  depends_on = [snowflake_grant_privileges_to_account_role.grant, %[2]s]
+  grants_on {
+    object_type = "STREAM"
+    object_name = %[2]s.fully_qualified_name
+  }
+}`, roleId.FullyQualifiedName(), resourceName)
+}
+
+// TODO (SNOW-1737932): Setting schema parameters related to retention time seems to have no affect on streams on directory tables.
+// Adjust this test after this is fixed on Snowflake side.
 func TestAcc_StreamOnDirectoryTable_RecreateWhenStale(t *testing.T) {
 	_ = testenvs.GetOrSkipTest(t, testenvs.EnableAcceptance)
 	acc.TestAccPreCheck(t)
@@ -336,7 +416,7 @@ func TestAcc_StreamOnDirectoryTable_RecreateWhenStale(t *testing.T) {
 	t.Cleanup(cleanupSchema)
 	id := acc.TestClient().Ids.RandomSchemaObjectIdentifierInSchema(schema.ID())
 
-	stage, cleanupStage := acc.TestClient().Stage.CreateStageWithDirectory(t, acc.TestClient().Ids.SchemaId())
+	stage, cleanupStage := acc.TestClient().Stage.CreateStageWithDirectory(t)
 	t.Cleanup(cleanupStage)
 
 	model := model.StreamOnDirectoryTable("test", id.DatabaseName(), id.Name(), id.SchemaName(), stage.ID().FullyQualifiedName())
@@ -347,8 +427,6 @@ func TestAcc_StreamOnDirectoryTable_RecreateWhenStale(t *testing.T) {
 		},
 		CheckDestroy: acc.CheckDestroy(t, resources.StreamOnDirectoryTable),
 		Steps: []resource.TestStep{
-			// TODO (SNOW-1737932): Setting schema parameters related to retention time seems to have no affect on streams on directory tables.
-			// Adjust this test after this is fixed on Snowflake side.
 			{
 				Config: config.FromModel(t, model),
 				Check: assert.AssertThat(t, resourceassert.StreamOnDirectoryTableResource(t, resourceName).