Refactor Redshift Append

clients/bigquery/bigquery.go

@@ -41,6 +41,10 @@ type Store struct {
 	db.Store
 }
 
+func (s *Store) Append(tableData *optimization.TableData) error {
+	return shared.Append(s, tableData, types.AppendOpts{})
+}
+
 func (s *Store) PrepareTemporaryTable(tableData *optimization.TableData, tableConfig *types.DwhTableConfig, tempTableID types.TableIdentifier, _ types.AdditionalSettings, createTempTable bool) error {
 	if createTempTable {
 		tempAlterTableArgs := ddl.AlterTableArgs{

clients/mssql/store.go

@@ -44,8 +44,7 @@ func (s *Store) Merge(tableData *optimization.TableData) error {
 }
 
 func (s *Store) Append(tableData *optimization.TableData) error {
-	tableID := s.IdentifierFor(tableData.TopicConfig(), tableData.Name())
-	return shared.Append(s, tableData, types.AppendOpts{TempTableID: tableID})
+	return shared.Append(s, tableData, types.AppendOpts{})
 }
 
 // specificIdentifierFor returns a MS SQL [TableIdentifier] for a [TopicConfig] + table name.

clients/redshift/staging.go

@@ -15,20 +15,21 @@ import (
 	"github.com/artie-labs/transfer/lib/s3lib"
 )
 
-func (s *Store) PrepareTemporaryTable(tableData *optimization.TableData, tableConfig *types.DwhTableConfig, tempTableID types.TableIdentifier, _ types.AdditionalSettings, _ bool) error {
-	// Redshift always creates a temporary table.
-	tempAlterTableArgs := ddl.AlterTableArgs{
-		Dwh:            s,
-		Tc:             tableConfig,
-		TableID:        tempTableID,
-		CreateTable:    true,
-		TemporaryTable: true,
-		ColumnOp:       constants.Add,
-		Mode:           tableData.Mode(),
-	}
+func (s *Store) PrepareTemporaryTable(tableData *optimization.TableData, tableConfig *types.DwhTableConfig, tempTableID types.TableIdentifier, _ types.AdditionalSettings, createTempTable bool) error {
+	if createTempTable {
+		tempAlterTableArgs := ddl.AlterTableArgs{
+			Dwh:            s,
+			Tc:             tableConfig,
+			TableID:        tempTableID,
+			CreateTable:    true,
+			TemporaryTable: true,
+			ColumnOp:       constants.Add,
+			Mode:           tableData.Mode(),
+		}
 
-	if err := tempAlterTableArgs.AlterTable(tableData.ReadOnlyInMemoryCols().GetColumns()...); err != nil {
-		return fmt.Errorf("failed to create temp table: %w", err)
+		if err := tempAlterTableArgs.AlterTable(tableData.ReadOnlyInMemoryCols().GetColumns()...); err != nil {
+			return fmt.Errorf("failed to create temp table: %w", err)
+		}
 	}
 
 	fp, err := s.loadTemporaryTable(tableData, tempTableID)

clients/redshift/writes.go

@@ -1,27 +1,13 @@
 package redshift
 
 import (
-	"fmt"
-
 	"github.com/artie-labs/transfer/clients/shared"
 	"github.com/artie-labs/transfer/lib/destination/types"
 	"github.com/artie-labs/transfer/lib/optimization"
 )
 
 func (s *Store) Append(tableData *optimization.TableData) error {
-	tableID := s.IdentifierFor(tableData.TopicConfig(), tableData.Name())
-
-	// Redshift is slightly different, we'll load and create the temporary table via shared.Append
-	// Then, we'll invoke `ALTER TABLE target APPEND FROM staging` to combine the diffs.
-	temporaryTableID := shared.TempTableID(tableID, tableData.TempTableSuffix())
-	if err := shared.Append(s, tableData, types.AppendOpts{TempTableID: temporaryTableID}); err != nil {
-		return err
-	}
-
-	_, err := s.Exec(
-		fmt.Sprintf(`ALTER TABLE %s APPEND FROM %s;`, tableID.FullyQualifiedName(), temporaryTableID.FullyQualifiedName()),
-	)
-	return err
+	return shared.Append(s, tableData, types.AppendOpts{})
 }
 
 func (s *Store) Merge(tableData *optimization.TableData) error {

clients/shared/append.go

@@ -16,17 +16,22 @@
 		return nil
 	}
 
-	tableID := dwh.IdentifierFor(tableData.TopicConfig(), tableData.Name())
 	tableConfig, err := dwh.GetTableConfig(tableData)
 	if err != nil {
 		return fmt.Errorf("failed to get table config: %w", err)
 	}
 
 	// We don't care about srcKeysMissing because we don't drop columns when we append.
-	_, targetKeysMissing := columns.Diff(tableData.ReadOnlyInMemoryCols(), tableConfig.Columns(),
-		tableData.TopicConfig().SoftDelete, tableData.TopicConfig().IncludeArtieUpdatedAt,
-		tableData.TopicConfig().IncludeDatabaseUpdatedAt, tableData.Mode())
+	_, targetKeysMissing := columns.Diff(
+		tableData.ReadOnlyInMemoryCols(),
+		tableConfig.Columns(),
+		tableData.TopicConfig().SoftDelete,
+		tableData.TopicConfig().IncludeArtieUpdatedAt,
+		tableData.TopicConfig().IncludeDatabaseUpdatedAt,
+		tableData.Mode(),
+	)
 
+	tableID := dwh.IdentifierFor(tableData.TopicConfig(), tableData.Name())
 	createAlterTableArgs := ddl.AlterTableArgs{
 		Dwh:         dwh,
 		Tc:          tableConfig,
@@ -46,9 +51,13 @@
 		return fmt.Errorf("failed to merge columns from destination: %w", err)
 	}
 
-	additionalSettings := types.AdditionalSettings{
-		AdditionalCopyClause: opts.AdditionalCopyClause,
-	}
-
-	return dwh.PrepareTemporaryTable(tableData, tableConfig, opts.TempTableID, additionalSettings, false)
+	return dwh.PrepareTemporaryTable(
+		tableData,
+		tableConfig,
+		tableID,
+		types.AdditionalSettings{
+			AdditionalCopyClause: opts.AdditionalCopyClause,
+		},
+		false,
+	)
 }

clients/snowflake/writes.go

@@ -25,10 +25,8 @@ func (s *Store) Append(tableData *optimization.TableData) error {
 			}
 		}
 
-		tableID := s.IdentifierFor(tableData.TopicConfig(), tableData.Name())
 		// TODO: For history mode - in the future, we could also have a separate stage name for history mode so we can enable parallel processing.
 		err = shared.Append(s, tableData, types.AppendOpts{
-			TempTableID:          tableID,
 			AdditionalCopyClause: `FILE_FORMAT = (TYPE = 'csv' FIELD_DELIMITER= '\t' FIELD_OPTIONALLY_ENCLOSED_BY='"' NULL_IF='\\N' EMPTY_FIELD_AS_NULL=FALSE) PURGE = TRUE`,
 		})
 	}

lib/destination/types/types.go

@@ -44,9 +44,6 @@ type AdditionalSettings struct {
 }
 
 type AppendOpts struct {
-	// TempTableID - sometimes the destination requires 2 steps to append to the table (e.g. Redshift), so we'll create and load the data into a staging table
-	// Redshift then has a separate step after `shared.Append(...)` to merge the two tables together.
-	TempTableID          TableIdentifier
 	AdditionalCopyClause string
 }