Checkpoint.

clients/bigquery/bigquery.go

@@ -42,8 +42,23 @@ type Store struct {
 	db.Store
 }
 
-func (s *Store) Append(tableData *optimization.TableData) error {
-	return shared.Append(s, tableData, types.AdditionalSettings{})
+func (s *Store) Append(tableData *optimization.TableData, useTempTable bool) 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)
+	if err := shared.Append(s, tableData, types.AdditionalSettings{
+		UseTempTable: true,
+		TempTableID:  temporaryTableID,
+	}); err != nil {
+		return err
+	}
+
+	_, err := s.Exec(
+		fmt.Sprintf(`ALTER TABLE %s APPEND FROM %s;`, tableID.FullyQualifiedName(), temporaryTableID.FullyQualifiedName()),
+	)
+	return err
 }
 
 func (s *Store) PrepareTemporaryTable(tableData *optimization.TableData, tableConfig *types.DwhTableConfig, tempTableID sql.TableIdentifier, _ types.AdditionalSettings, createTempTable bool) error {

clients/mssql/store.go

@@ -43,7 +43,7 @@ func (s *Store) Merge(tableData *optimization.TableData) error {
 	return shared.Merge(s, tableData, types.MergeOpts{})
 }
 
-func (s *Store) Append(tableData *optimization.TableData) error {
+func (s *Store) Append(tableData *optimization.TableData, _ bool) error {
 	return shared.Append(s, tableData, types.AdditionalSettings{})
 }
 

clients/redshift/redshift.go

@@ -28,7 +28,7 @@ type Store struct {
 	db.Store
 }
 
-func (s *Store) Append(tableData *optimization.TableData) error {
+func (s *Store) Append(tableData *optimization.TableData, _ bool) error {
 	return shared.Append(s, tableData, types.AdditionalSettings{})
 }
 

clients/shared/append.go

@@ -56,6 +56,6 @@ func Append(dwh destination.DataWarehouse, tableData *optimization.TableData, op
 		tableConfig,
 		tableID,
 		opts,
-		false,
+		opts.UseTempTable,
 	)
 }

clients/snowflake/writes.go

@@ -13,7 +13,7 @@ import (
 	"github.com/artie-labs/transfer/lib/typing/columns"
 )
 
-func (s *Store) Append(tableData *optimization.TableData) error {
+func (s *Store) Append(tableData *optimization.TableData, _ bool) error {
 	var err error
 	for i := 0; i < maxRetries; i++ {
 		if i > 0 {

lib/destination/dwh.go

@@ -14,7 +14,7 @@ import (
 type DataWarehouse interface {
 	Dialect() sqllib.Dialect
 	Merge(tableData *optimization.TableData) error
-	Append(tableData *optimization.TableData) error
+	Append(tableData *optimization.TableData, useTempTable bool) error
 	Dedupe(tableID sqllib.TableIdentifier, primaryKeys []string, includeArtieUpdatedAt bool) error
 	Exec(query string, args ...any) (sql.Result, error)
 	Query(query string, args ...any) (*sql.Rows, error)
@@ -30,7 +30,7 @@ type DataWarehouse interface {
 
 type Baseline interface {
 	Merge(tableData *optimization.TableData) error
-	Append(tableData *optimization.TableData) error
+	Append(tableData *optimization.TableData, useTempTable bool) error
 	IsRetryableError(err error) bool
 	IdentifierFor(topicConfig kafkalib.TopicConfig, table string) sqllib.TableIdentifier
 }

lib/destination/types/types.go

@@ -42,4 +42,7 @@ type MergeOpts struct {
 
 type AdditionalSettings struct {
 	AdditionalCopyClause string
+
+	UseTempTable bool
+	TempTableID  sql.TableIdentifier
 }

processes/consumer/flush.go

@@ -89,7 +89,7 @@ func Flush(ctx context.Context, inMemDB *models.DatabaseData, dest destination.B
 			action := "merge"
 			// Merge or Append depending on the mode.
 			if _tableData.Mode() == config.History {
-				err = dest.Append(_tableData.TableData)
+				err = dest.Append(_tableData.TableData, false)
 				action = "append"
 			} else {
 				err = dest.Merge(_tableData.TableData)