WIP.

clients/snowflake/dialect/dialect.go

@@ -2,6 +2,7 @@ package dialect
 
 import (
 	"fmt"
+	"path/filepath"
 	"strings"
 
 	"github.com/artie-labs/transfer/lib/config/constants"
@@ -152,3 +153,8 @@ FROM
 WHERE
     UPPER(table_schema) = UPPER(?) AND table_name ILIKE ?`, dbName), []any{schemaName, "%" + constants.ArtiePrefix + "%"}
 }
+
+func (SnowflakeDialect) BuildRemoveAllFilesFromStage(stageName string, path string) string {
+	// https://docs.snowflake.com/en/sql-reference/sql/remove
+	return fmt.Sprintf("REMOVE @%s", filepath.Join(stageName, path))
+}

clients/snowflake/dialect/dialect_test.go

@@ -225,3 +225,20 @@ WHEN MATCHED AND stg."__ARTIE_DELETE" THEN DELETE
 WHEN MATCHED AND IFNULL(stg."__ARTIE_DELETE", false) = false THEN UPDATE SET "GROUP"=stg."GROUP","ID"=stg."ID","START"=stg."START","UPDATED_AT"=stg."UPDATED_AT"
 WHEN NOT MATCHED AND IFNULL(stg."__ARTIE_DELETE", false) = false THEN INSERT ("GROUP","ID","START","UPDATED_AT") VALUES (stg."GROUP",stg."ID",stg."START",stg."UPDATED_AT");`, statements[0])
 }
+
+func TestSnowflakeDialect_BuildRemoveAllFilesFromStage(t *testing.T) {
+	{
+		// Stage name only, no path
+		assert.Equal(t,
+			"REMOVE @STAGE_NAME",
+			SnowflakeDialect{}.BuildRemoveAllFilesFromStage("STAGE_NAME", ""),
+		)
+	}
+	{
+		// Stage name and path
+		assert.Equal(t,
+			"REMOVE @STAGE_NAME/path1/subpath2",
+			SnowflakeDialect{}.BuildRemoveAllFilesFromStage("STAGE_NAME", "path1/subpath2"),
+		)
+	}
+}

clients/snowflake/staging.go

@@ -86,6 +86,7 @@ func (s *Store) PrepareTemporaryTable(ctx context.Context, tableData *optimizati
 	}
 
 	if _, err = s.Exec(copyCommand); err != nil {
+
 		return fmt.Errorf("failed to run copy into temporary table: %w", err)
 	}