diff --git a/docs/resources/unsafe_execute.md b/docs/resources/unsafe_execute.md
new file mode 100644
index 0000000000..379efc2eb5
--- /dev/null
+++ b/docs/resources/unsafe_execute.md
@@ -0,0 +1,62 @@
+---
+# generated by https://github.com/hashicorp/terraform-plugin-docs
+page_title: "snowflake_unsafe_execute Resource - terraform-provider-snowflake"
+subcategory: ""
+description: |-
+  Experimental resource used for testing purposes only. Allows to execute ANY SQL statement.
+---
+
+# snowflake_unsafe_execute (Resource)
+
+!> **Warning** This is a dangerous resource that allows executing **ANY** SQL statement. It may destroy resources if used incorrectly. It may behave incorrectly combined with other resources. Will be deleted in the upcoming versions. Use at your own risk.
+
+Experimental resource used for testing purposes only. Allows to execute ANY SQL statement.
+
+## Example Usage
+
+```terraform
+# create and destroy resource
+resource "snowflake_unsafe_execute" "test" {
+  execute = "CREATE DATABASE ABC"
+  revert  = "DROP DATABASE ABC"
+}
+
+# create and destroy resource using qualified name
+resource "snowflake_unsafe_execute" "test" {
+  execute = "CREATE DATABASE \"abc\""
+  revert  = "DROP DATABASE \"abc\""
+}
+
+# grant and revoke privilege USAGE to ROLE on database
+resource "snowflake_unsafe_execute" "test" {
+  execute = "GRANT USAGE ON DATABASE ABC TO ROLE XYZ"
+  revert  = "REVOKE USAGE ON DATABASE ABC FROM ROLE XYZ"
+}
+
+# grant and revoke with for_each
+variable "database_grants" {
+  type = list(object({
+    database_name = string
+    role_id       = string
+    privileges    = list(string)
+  }))
+}
+
+resource "snowflake_unsafe_execute" "test" {
+  for_each = { for index, db_grant in var.database_grants : index => db_grant }
+  execute  = "GRANT ${join(",", each.value.privileges)} ON DATABASE ${each.value.database_name} TO ROLE ${each.value.role_id}"
+  revert   = "REVOKE ${join(",", each.value.privileges)} ON DATABASE ${each.value.database_name} FROM ROLE ${each.value.role_id}"
+}
+```
+
+<!-- schema generated by tfplugindocs -->
+## Schema
+
+### Required
+
+- `execute` (String) SQL statement to execute.
+- `revert` (String) SQL statement to revert the execute statement. Invoked when resource is deleted.
+
+### Read-Only
+
+- `id` (String) The ID of this resource.
diff --git a/examples/resources/snowflake_unsafe_execute/resource.tf b/examples/resources/snowflake_unsafe_execute/resource.tf
new file mode 100644
index 0000000000..026ea127c9
--- /dev/null
+++ b/examples/resources/snowflake_unsafe_execute/resource.tf
@@ -0,0 +1,32 @@
+# create and destroy resource
+resource "snowflake_unsafe_execute" "test" {
+  execute = "CREATE DATABASE ABC"
+  revert  = "DROP DATABASE ABC"
+}
+
+# create and destroy resource using qualified name
+resource "snowflake_unsafe_execute" "test" {
+  execute = "CREATE DATABASE \"abc\""
+  revert  = "DROP DATABASE \"abc\""
+}
+
+# grant and revoke privilege USAGE to ROLE on database
+resource "snowflake_unsafe_execute" "test" {
+  execute = "GRANT USAGE ON DATABASE ABC TO ROLE XYZ"
+  revert  = "REVOKE USAGE ON DATABASE ABC FROM ROLE XYZ"
+}
+
+# grant and revoke with for_each
+variable "database_grants" {
+  type = list(object({
+    database_name = string
+    role_id       = string
+    privileges    = list(string)
+  }))
+}
+
+resource "snowflake_unsafe_execute" "test" {
+  for_each = { for index, db_grant in var.database_grants : index => db_grant }
+  execute  = "GRANT ${join(",", each.value.privileges)} ON DATABASE ${each.value.database_name} TO ROLE ${each.value.role_id}"
+  revert   = "REVOKE ${join(",", each.value.privileges)} ON DATABASE ${each.value.database_name} FROM ROLE ${each.value.role_id}"
+}
diff --git a/pkg/provider/provider.go b/pkg/provider/provider.go
index 497108179c..00f4d52ff0 100644
--- a/pkg/provider/provider.go
+++ b/pkg/provider/provider.go
@@ -247,6 +247,7 @@ func getResources() map[string]*schema.Resource {
 		"snowflake_tag_association":                resources.TagAssociation(),
 		"snowflake_tag_masking_policy_association": resources.TagMaskingPolicyAssociation(),
 		"snowflake_task":                           resources.Task(),
+		"snowflake_unsafe_execute":                 resources.UnsafeExecute(),
 		"snowflake_user":                           resources.User(),
 		"snowflake_user_ownership_grant":           resources.UserOwnershipGrant(),
 		"snowflake_user_public_keys":               resources.UserPublicKeys(),
diff --git a/pkg/resources/unsafe_execute.go b/pkg/resources/unsafe_execute.go
new file mode 100644
index 0000000000..dd0d7d7a48
--- /dev/null
+++ b/pkg/resources/unsafe_execute.go
@@ -0,0 +1,73 @@
+package resources
+
+import (
+	"database/sql"
+	"github.com/Snowflake-Labs/terraform-provider-snowflake/pkg/snowflake"
+	"log"
+
+	"github.com/hashicorp/go-uuid"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+)
+
+var unsafeExecuteSchema = map[string]*schema.Schema{
+	"execute": {
+		Type:        schema.TypeString,
+		Required:    true,
+		ForceNew:    true,
+		Description: "SQL statement to execute.",
+	},
+	"revert": {
+		Type:        schema.TypeString,
+		Required:    true,
+		Description: "SQL statement to revert the execute statement. Invoked when resource is deleted.",
+	},
+}
+
+func UnsafeExecute() *schema.Resource {
+	return &schema.Resource{
+		Create: ExecuteUnsafeSQLStatement,
+		Read:   schema.Noop,
+		Delete: RevertUnsafeSQLStatement,
+		Update: schema.Noop,
+
+		Schema: unsafeExecuteSchema,
+
+		DeprecationMessage: "Experimental resource. Will be deleted in the upcoming versions. Use at your own risk.",
+		Description:        "Experimental resource used for testing purposes only. Allows to execute ANY SQL statement.",
+	}
+}
+
+func ExecuteUnsafeSQLStatement(d *schema.ResourceData, meta interface{}) error {
+	db := meta.(*sql.DB)
+
+	id, err := uuid.GenerateUUID()
+	if err != nil {
+		return err
+	}
+
+	executeStatement := d.Get("execute").(string)
+	err = snowflake.Exec(db, executeStatement)
+	if err != nil {
+		return err
+	}
+
+	d.SetId(id)
+	log.Printf(`[DEBUG] SQL "%s" applied successfully\n`, executeStatement)
+
+	return nil
+}
+
+func RevertUnsafeSQLStatement(d *schema.ResourceData, meta interface{}) error {
+	db := meta.(*sql.DB)
+
+	revertStatement := d.Get("revert").(string)
+	err := snowflake.Exec(db, revertStatement)
+	if err != nil {
+		return err
+	}
+
+	d.SetId("")
+	log.Printf(`[DEBUG] SQL "%s" applied successfully\n`, revertStatement)
+
+	return nil
+}
diff --git a/pkg/resources/unsafe_execute_acceptance_test.go b/pkg/resources/unsafe_execute_acceptance_test.go
new file mode 100644
index 0000000000..47d8b12c7e
--- /dev/null
+++ b/pkg/resources/unsafe_execute_acceptance_test.go
@@ -0,0 +1,321 @@
+package resources_test
+
+import (
+	"crypto/rand"
+	"database/sql"
+	"errors"
+	"fmt"
+	"github.com/Snowflake-Labs/terraform-provider-snowflake/pkg/provider"
+	"github.com/Snowflake-Labs/terraform-provider-snowflake/pkg/snowflake"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/terraform"
+	"github.com/jmoiron/sqlx"
+	"log"
+	"math/big"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+var (
+	accTestDb  *sql.DB
+	accTestDbx *sqlx.DB
+)
+
+func init() {
+	db, err := provider.GetDatabaseHandleFromEnv()
+	if err != nil {
+		log.Fatalln(err)
+	}
+	dbx := sqlx.NewDb(db, "snowflake")
+	accTestDb = db
+	accTestDbx = dbx
+}
+
+func TestAcc_UnsafeExecute_basic(t *testing.T) {
+	id := generateUnsafeExecuteTestDatabaseName(t)
+	idLowerCase := strings.ToLower(generateUnsafeExecuteTestDatabaseName(t))
+	createDatabaseStatement := func(raw bool, id string) string {
+		if raw {
+			return fmt.Sprintf(`create database \"%s\"`, id)
+		}
+		return fmt.Sprintf("create database \"%s\"", id)
+	}
+	dropDatabaseStatement := func(raw bool, id string) string {
+		if raw {
+			return fmt.Sprintf(`drop database \"%s\"`, id)
+		}
+		return fmt.Sprintf("drop database \"%s\"", id)
+	}
+	resourceName := "snowflake_unsafe_execute.test"
+
+	resource.Test(t, resource.TestCase{
+		Providers:    providers(),
+		CheckDestroy: testAccCheckDatabaseExistence(t, id, false),
+		Steps: []resource.TestStep{
+			{
+				Config: schemaExecResourceConfig(createDatabaseStatement(true, id), dropDatabaseStatement(true, id)),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr(resourceName, "execute", createDatabaseStatement(false, id)),
+					resource.TestCheckResourceAttr(resourceName, "revert", dropDatabaseStatement(false, id)),
+					resource.TestCheckResourceAttrSet(resourceName, "id"),
+					testAccCheckDatabaseExistence(t, id, true),
+				),
+			},
+		},
+	})
+
+	resource.Test(t, resource.TestCase{
+		Providers:    providers(),
+		CheckDestroy: testAccCheckDatabaseExistence(t, idLowerCase, false),
+		Steps: []resource.TestStep{
+			{
+				Config: schemaExecResourceConfig(createDatabaseStatement(true, idLowerCase), dropDatabaseStatement(true, idLowerCase)),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr(resourceName, "execute", createDatabaseStatement(false, idLowerCase)),
+					resource.TestCheckResourceAttr(resourceName, "revert", dropDatabaseStatement(false, idLowerCase)),
+					resource.TestCheckResourceAttrSet(resourceName, "id"),
+					testAccCheckDatabaseExistence(t, idLowerCase, true),
+				),
+			},
+		},
+	})
+}
+
+func TestAcc_UnsafeExecute_revertUpdated(t *testing.T) {
+	id := generateUnsafeExecuteTestDatabaseName(t)
+	execute := fmt.Sprintf("create database %s", id)
+	revert := fmt.Sprintf("drop database %s", id)
+	notMatchingRevert := "select 1"
+	var savedId string
+	resourceName := "snowflake_unsafe_execute.test"
+
+	resource.Test(t, resource.TestCase{
+		Providers:    providers(),
+		CheckDestroy: testAccCheckDatabaseExistence(t, id, false),
+		Steps: []resource.TestStep{
+			{
+				Config: schemaExecResourceConfig(execute, notMatchingRevert),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr(resourceName, "execute", execute),
+					resource.TestCheckResourceAttr(resourceName, "revert", notMatchingRevert),
+					resource.TestCheckResourceAttrSet(resourceName, "id"),
+					resource.TestCheckResourceAttrWith(resourceName, "id", func(value string) error {
+						savedId = value
+						return nil
+					}),
+					testAccCheckDatabaseExistence(t, id, true),
+				),
+			},
+			{
+				Config: schemaExecResourceConfig(execute, revert),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr(resourceName, "execute", execute),
+					resource.TestCheckResourceAttr(resourceName, "revert", revert),
+					resource.TestCheckResourceAttrSet(resourceName, "id"),
+					resource.TestCheckResourceAttrWith(resourceName, "id", func(value string) error {
+						if savedId != value {
+							return errors.New("different id after revert update")
+						}
+						return nil
+					}),
+					testAccCheckDatabaseExistence(t, id, true),
+				),
+			},
+		},
+	})
+}
+
+func TestAcc_UnsafeExecute_executeUpdated(t *testing.T) {
+	id := generateUnsafeExecuteTestDatabaseName(t)
+	execute := fmt.Sprintf("create database %s", id)
+	revert := fmt.Sprintf("drop database %s", id)
+
+	newId := fmt.Sprintf("%s_2", id)
+	newExecute := fmt.Sprintf("create database %s", newId)
+	newRevert := fmt.Sprintf("drop database %s", newId)
+
+	var savedId string
+
+	resourceName := "snowflake_unsafe_execute.test"
+
+	resource.Test(t, resource.TestCase{
+		Providers: providers(),
+		CheckDestroy: func(state *terraform.State) error {
+			err := testAccCheckDatabaseExistence(t, id, false)(state)
+			if err != nil {
+				return err
+			}
+			err = testAccCheckDatabaseExistence(t, newId, false)(state)
+			if err != nil {
+				return err
+			}
+			return nil
+		},
+		Steps: []resource.TestStep{
+			{
+				Config: schemaExecResourceConfig(execute, revert),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr(resourceName, "execute", execute),
+					resource.TestCheckResourceAttr(resourceName, "revert", revert),
+					resource.TestCheckResourceAttrSet(resourceName, "id"),
+					resource.TestCheckResourceAttrWith(resourceName, "id", func(value string) error {
+						savedId = value
+						return nil
+					}),
+					testAccCheckDatabaseExistence(t, id, true),
+				),
+			},
+			{
+				Config: schemaExecResourceConfig(newExecute, newRevert),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr(resourceName, "execute", newExecute),
+					resource.TestCheckResourceAttr(resourceName, "revert", newRevert),
+					resource.TestCheckResourceAttrSet(resourceName, "id"),
+					resource.TestCheckResourceAttrWith(resourceName, "id", func(value string) error {
+						if savedId == value {
+							return errors.New("same id after execute update")
+						}
+						return nil
+					}),
+					testAccCheckDatabaseExistence(t, id, false),
+					testAccCheckDatabaseExistence(t, newId, true),
+				),
+			},
+		},
+	})
+}
+
+func TestAcc_UnsafeExecute_grants(t *testing.T) {
+	id := generateUnsafeExecuteTestDatabaseName(t)
+	roleId := generateUnsafeExecuteTestRoleName(t)
+	privilege := "CREATE SCHEMA"
+	execute := fmt.Sprintf("GRANT %s ON DATABASE %s TO ROLE %s", privilege, id, roleId)
+	revert := fmt.Sprintf("REVOKE %s ON DATABASE %s FROM ROLE %s", privilege, id, roleId)
+	resourceName := "snowflake_unsafe_execute.test"
+
+	resource.Test(t, resource.TestCase{
+		Providers: providers(),
+		CheckDestroy: func(state *terraform.State) error {
+			err := verifyGrantExists(t, roleId, privilege, false)(state)
+			dropResourcesForUnsafeExecuteTestCaseForGrants(t, id, roleId)
+			return err
+		},
+		Steps: []resource.TestStep{
+			{
+				PreConfig: func() { createResourcesForExecuteUnsafeTestCaseForGrants(t, id, roleId) },
+				Config:    schemaExecResourceConfig(execute, revert),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr(resourceName, "execute", execute),
+					resource.TestCheckResourceAttr(resourceName, "revert", revert),
+					resource.TestCheckResourceAttrSet(resourceName, "id"),
+					verifyGrantExists(t, roleId, privilege, true),
+				),
+			},
+		},
+	})
+}
+
+func schemaExecResourceConfig(exec string, revert string) string {
+	return fmt.Sprintf(`
+resource "snowflake_unsafe_execute" "test" {
+  execute = "%s"
+  revert = "%s"
+}
+`, exec, revert)
+}
+
+// generateUnsafeExecuteTestDatabaseName returns capitalized name on purpose.
+// Using small caps without escaping creates problem with later using sdk client which uses identifier that is escaped by default.
+func generateUnsafeExecuteTestDatabaseName(t *testing.T) string {
+	t.Helper()
+	id, err := rand.Int(rand.Reader, big.NewInt(10000))
+	if err != nil {
+		t.Fatalf("Failed to generate database id: %v", err)
+	}
+	return fmt.Sprintf("UNSAFE_EXECUTE_TEST_DATABASE_%d", id)
+}
+
+// generateUnsafeExecuteTestRoleName returns capitalized name on purpose.
+// Using small caps without escaping creates problem with later using sdk client which uses identifier that is escaped by default.
+func generateUnsafeExecuteTestRoleName(t *testing.T) string {
+	t.Helper()
+	id, err := rand.Int(rand.Reader, big.NewInt(10000))
+	if err != nil {
+		t.Fatalf("Failed to generate role id: %v", err)
+	}
+	return fmt.Sprintf("UNSAFE_EXECUTE_TEST_ROLE_%d", id)
+}
+
+func testAccCheckDatabaseExistence(t *testing.T, id string, shouldExist bool) func(state *terraform.State) error {
+	t.Helper()
+	return func(state *terraform.State) error {
+		_, err := snowflake.ListDatabase(accTestDbx, id)
+
+		if shouldExist {
+			if err != nil {
+				return fmt.Errorf("error while retrieving database %s, err = %w", id, err)
+			}
+		} else {
+			if err == nil {
+				return fmt.Errorf("database %v still exists", id)
+			}
+		}
+		return nil
+	}
+}
+
+func createResourcesForExecuteUnsafeTestCaseForGrants(t *testing.T, dbId string, roleId string) {
+	t.Helper()
+
+	createDatabaseSQL := snowflake.NewDatabaseBuilder(dbId).Create()
+	err := snowflake.Exec(accTestDb, createDatabaseSQL)
+	require.NoError(t, err)
+
+	createRoleSQL := snowflake.NewRoleBuilder(roleId).Create().Statement()
+	err = snowflake.Exec(accTestDb, createRoleSQL)
+	require.NoError(t, err)
+}
+
+func dropResourcesForUnsafeExecuteTestCaseForGrants(t *testing.T, dbId string, roleId string) {
+	t.Helper()
+
+	dropDatabaseSQL := snowflake.NewDatabaseBuilder(dbId).Drop()
+	err := snowflake.Exec(accTestDb, dropDatabaseSQL)
+	require.NoError(t, err)
+
+	dropRoleSQL := snowflake.NewRoleBuilder(roleId).Drop()
+	err = snowflake.Exec(accTestDb, dropRoleSQL)
+	require.NoError(t, err)
+}
+
+func verifyGrantExists(t *testing.T, roleId string, privilege string, shouldExist bool) func(state *terraform.State) error {
+	t.Helper()
+	return func(state *terraform.State) error {
+		grants, err := snowflake.ShowGrantsTo(accTestDb, "ROLE", roleId)
+		require.NoError(t, err)
+
+		if shouldExist {
+			require.Equal(t, 1, len(grants))
+
+			assert.True(t, grants[0].Privilege.Valid)
+			assert.Equal(t, privilege, grants[0].Privilege.String)
+
+			assert.True(t, grants[0].GrantedOn.Valid)
+			assert.Equal(t, "DATABASE", grants[0].GrantedOn.String)
+
+			assert.True(t, grants[0].GrantedTo.Valid)
+			assert.Equal(t, "ROLE", grants[0].GrantedTo.String)
+
+			assert.True(t, grants[0].GranteeName.Valid)
+			assert.Equal(t, roleId, grants[0].GranteeName.String)
+		} else {
+			require.Equal(t, 0, len(grants))
+		}
+
+		// it does not matter what we return, because we have assertions above
+		return nil
+	}
+}
diff --git a/pkg/snowflake/database.go b/pkg/snowflake/database.go
index a68c68d547..dd58b16ddd 100644
--- a/pkg/snowflake/database.go
+++ b/pkg/snowflake/database.go
@@ -261,6 +261,8 @@ type Database struct {
 	Comment       sql.NullString `db:"comment"`
 	Options       sql.NullString `db:"options"`
 	RetentionTime sql.NullString `db:"retention_time"`
+	Kind          sql.NullString `db:"kind"`
+	Budget        sql.NullString `db:"budget"`
 }
 
 func ScanDatabase(row *sqlx.Row) (*Database, error) {
@@ -304,13 +306,11 @@ func ListDatabase(sdb *sqlx.DB, databaseName string) (*Database, error) {
 		}
 		return nil, fmt.Errorf("unable to scan row for %s err = %w", stmt, err)
 	}
-	db := &Database{}
 	for _, d := range dbs {
 		d := d
 		if d.DBName.String == databaseName {
-			db = &d
-			break
+			return &d, nil
 		}
 	}
-	return db, nil
+	return nil, errors.New("database not found")
 }
diff --git a/templates/resources/unsafe_execute.md.tmpl b/templates/resources/unsafe_execute.md.tmpl
new file mode 100644
index 0000000000..7c0ab2cacf
--- /dev/null
+++ b/templates/resources/unsafe_execute.md.tmpl
@@ -0,0 +1,29 @@
+---
+# generated by https://github.com/hashicorp/terraform-plugin-docs
+page_title: "{{.Name}} {{.Type}} - {{.ProviderName}}"
+subcategory: ""
+description: |-
+{{ .Description | plainmarkdown | trimspace | prefixlines "  " }}
+---
+
+# {{.Name}} ({{.Type}})
+
+!> **Warning** This is a dangerous resource that allows executing **ANY** SQL statement. It may destroy resources if used incorrectly. It may behave incorrectly combined with other resources. Will be deleted in the upcoming versions. Use at your own risk.
+
+{{ .Description | trimspace }}
+
+{{ if .HasExample -}}
+## Example Usage
+
+{{ tffile (printf "examples/resources/%s/resource.tf" .Name)}}
+{{- end }}
+
+{{ .SchemaMarkdown | trimspace }}
+{{- if .HasImport }}
+
+## Import
+
+Import is supported using the following syntax:
+
+{{ printf "{{codefile \"shell\" %q}}" .ImportFile }}
+{{- end }}