Clean.

clients/snowflake/staging.go

@@ -90,7 +90,7 @@ func (s *Store) PrepareTemporaryTable(ctx context.Context, tableData *optimizati
 		// This is because [PURGE = TRUE] will only delete the staging files upon a successful COPY INTO.
 		// We also only need to do this for non-temp tables because these staging files will linger, since we create a new temporary table per attempt.
 		if !createTempTable {
-			if _, deleteErr := s.ExecContext(ctx, s.dialect().BuildRemoveFilesFromStage(tempTableID.FullyQualifiedName(), "")); deleteErr != nil {
+			if _, deleteErr := s.ExecContext(ctx, s.dialect().BuildRemoveFilesFromStage(addPrefixToTableName(tempTableID, "%"), "")); deleteErr != nil {
 				slog.Warn("Failed to remove all files from stage", slog.Any("deleteErr", deleteErr))
 			}
 		}

clients/snowflake/staging_test.go

@@ -22,6 +22,13 @@ import (
 	"github.com/artie-labs/transfer/lib/typing/columns"
 )
 
+func (s *SnowflakeTestSuite) TestBuildRemoveFilesFromStage() {
+	table := dialect.NewTableIdentifier("db", "schema", "table")
+
+	query := s.stageStore.dialect().BuildRemoveFilesFromStage(addPrefixToTableName(table, "%"), "")
+	assert.Equal(s.T(), `REMOVE @db.schema."%table"`, query)
+}
+
 func (s *SnowflakeTestSuite) TestReplaceExceededValues() {
 	// String + OptionalStringPrecision not set + equal to max LOB length:
 	assert.Equal(s.T(), strings.Repeat("a", 16777216), replaceExceededValues(strings.Repeat("a", 16777216), typing.String))