[DDL] Standing up AlterTableDropColumns and removing AlterTableArgs

clients/shared/merge.go

@@ -6,7 +6,6 @@ import (
 	"log/slog"
 	"time"
 
-	"github.com/artie-labs/transfer/lib/config/constants"
 	"github.com/artie-labs/transfer/lib/destination"
 	"github.com/artie-labs/transfer/lib/destination/ddl"
 	"github.com/artie-labs/transfer/lib/destination/types"
@@ -44,25 +43,15 @@ func Merge(ctx context.Context, dwh destination.DataWarehouse, tableData *optimi
 		}
 	} else {
 		if err = AlterTableAddColumns(ctx, dwh, tableConfig, tableID, targetKeysMissing); err != nil {
-			return fmt.Errorf("failed to alter table: %w", err)
+			return fmt.Errorf("failed to add columns for table %q: %w", tableID.Table(), err)
 		}
 	}
 
-	// Keys that exist in DWH, but not in our CDC stream.
-	deleteAlterTableArgs := ddl.AlterTableArgs{
-		Dialect:                dwh.Dialect(),
-		Tc:                     tableConfig,
-		TableID:                tableID,
-		ColumnOp:               constants.Delete,
-		ContainOtherOperations: tableData.ContainOtherOperations(),
-		CdcTime:                tableData.LatestCDCTs,
-		Mode:                   tableData.Mode(),
-	}
-
-	if err = deleteAlterTableArgs.AlterTable(dwh, srcKeysMissing...); err != nil {
-		return fmt.Errorf("failed to apply alter table: %w", err)
+	if err = AlterTableDropColumns(ctx, dwh, tableConfig, tableID, srcKeysMissing, tableData.LatestCDCTs, tableData.ContainOtherOperations()); err != nil {
+		return fmt.Errorf("failed to drop columns for table %q: %w", tableID.Table(), err)
 	}
 
+	// TODO: Examine whether [AuditColumnsToDelete] still needs to be called.
 	tableConfig.AuditColumnsToDelete(srcKeysMissing)
 	if err = tableData.MergeColumnsFromDestination(tableConfig.GetColumns()...); err != nil {
 		return fmt.Errorf("failed to merge columns from destination: %w", err)

clients/shared/table.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
+	"time"
 
 	"github.com/artie-labs/transfer/lib/config/constants"
 	"github.com/artie-labs/transfer/lib/destination"
@@ -61,3 +62,35 @@ func AlterTableAddColumns(ctx context.Context, dwh destination.DataWarehouse, tc
 	tc.MutateInMemoryColumns(constants.Add, colsToAdd...)
 	return nil
 }
+
+func AlterTableDropColumns(ctx context.Context, dwh destination.DataWarehouse, tc *types.DwhTableConfig, tableID sql.TableIdentifier, cols []columns.Column, cdcTime time.Time, containOtherOperations bool) error {
+	if len(cols) == 0 {
+		return nil
+	}
+
+	var colsToDrop []columns.Column
+	for _, col := range cols {
+		if tc.ShouldDeleteColumn(col.Name(), cdcTime, containOtherOperations) {
+			colsToDrop = append(colsToDrop, col)
+		}
+	}
+
+	if len(colsToDrop) == 0 {
+		return nil
+	}
+
+	for _, colToDrop := range colsToDrop {
+		query, err := ddl.BuildAlterTableDropColumns(dwh.Dialect(), tableID, colToDrop)
+		if err != nil {
+			return fmt.Errorf("failed to build alter table drop columns: %w", err)
+		}
+
+		slog.Info("[DDL] Executing query", slog.String("query", query))
+		if _, err = dwh.ExecContext(ctx, query); err != nil {
+			return fmt.Errorf("failed to alter table: %w", err)
+		}
+	}
+
+	tc.MutateInMemoryColumns(constants.Delete, colsToDrop...)
+	return nil
+}

lib/destination/ddl/ddl.go

@@ -4,17 +4,19 @@ import (
 	"fmt"
 	"log/slog"
 	"strings"
-	"time"
 
 	bigQueryDialect "github.com/artie-labs/transfer/clients/bigquery/dialect"
 	"github.com/artie-labs/transfer/lib/config"
 	"github.com/artie-labs/transfer/lib/config/constants"
 	"github.com/artie-labs/transfer/lib/destination"
-	"github.com/artie-labs/transfer/lib/destination/types"
 	"github.com/artie-labs/transfer/lib/sql"
 	"github.com/artie-labs/transfer/lib/typing/columns"
 )
 
+func shouldCreatePrimaryKey(col columns.Column, mode config.Mode, createTable bool) bool {
+	return col.PrimaryKey() && mode == config.Replication && createTable
+}
+
 func BuildCreateTableSQL(dialect sql.Dialect, tableIdentifier sql.TableIdentifier, temporaryTable bool, mode config.Mode, columns []columns.Column) (string, error) {
 	if len(columns) == 0 {
 		return "", fmt.Errorf("no columns provided")
@@ -83,87 +85,10 @@ func BuildAlterTableAddColumns(dialect sql.Dialect, tableID sql.TableIdentifier,
 	return parts, nil
 }
 
-type AlterTableArgs struct {
-	Dialect sql.Dialect
-	Tc      *types.DwhTableConfig
-	// ContainsOtherOperations - this is sourced from tableData `containOtherOperations`
-	ContainOtherOperations bool
-	TableID                sql.TableIdentifier
-	ColumnOp               constants.ColumnOperation
-	Mode                   config.Mode
-	CdcTime                time.Time
-}
-
-func (a AlterTableArgs) Validate() error {
-	if a.Dialect == nil {
-		return fmt.Errorf("dialect cannot be nil")
-	}
-
-	if !(a.Mode == config.History || a.Mode == config.Replication) {
-		return fmt.Errorf("unexpected mode: %s", a.Mode.String())
-	}
-
-	return nil
-}
-
-func shouldCreatePrimaryKey(col columns.Column, mode config.Mode, createTable bool) bool {
-	return col.PrimaryKey() && mode == config.Replication && createTable
-}
-
-func (a AlterTableArgs) buildStatements(cols ...columns.Column) ([]string, []columns.Column) {
-	var mutateCol []columns.Column
-	// It's okay to combine since args.ColumnOp only takes one of: `Delete` or `Add`
-	var colSQLParts []string
-	for _, col := range cols {
-		if col.ShouldSkip() {
-			// Let's not modify the table if the column kind is invalid
-			continue
-		}
-
-		if a.ColumnOp == constants.Delete {
-			if !a.Tc.ShouldDeleteColumn(col.Name(), a.CdcTime, a.ContainOtherOperations) {
-				continue
-			}
-		}
-
-		mutateCol = append(mutateCol, col)
-		switch a.ColumnOp {
-		case constants.Add:
-			colSQLParts = append(colSQLParts, fmt.Sprintf("%s %s", a.Dialect.QuoteIdentifier(col.Name()), a.Dialect.DataTypeForKind(col.KindDetails, col.PrimaryKey())))
-		case constants.Delete:
-			colSQLParts = append(colSQLParts, a.Dialect.QuoteIdentifier(col.Name()))
-		}
-	}
-
-	var alterStatements []string
-	for _, colSQLPart := range colSQLParts {
-		alterStatements = append(alterStatements, a.Dialect.BuildAlterColumnQuery(a.TableID, a.ColumnOp, colSQLPart))
-	}
-
-	return alterStatements, mutateCol
-}
-
-func (a AlterTableArgs) AlterTable(dwh destination.DataWarehouse, cols ...columns.Column) error {
-	if err := a.Validate(); err != nil {
-		return err
-	}
-
-	if len(cols) == 0 {
-		return nil
-	}
-
-	alterStatements, mutateCol := a.buildStatements(cols...)
-	for _, sqlQuery := range alterStatements {
-		slog.Info("DDL - executing sql", slog.String("query", sqlQuery))
-		if _, err := dwh.Exec(sqlQuery); err != nil {
-			if !a.Dialect.IsColumnAlreadyExistsErr(err) {
-				return fmt.Errorf("failed to apply ddl, sql: %q, err: %w", sqlQuery, err)
-			}
-		}
+func BuildAlterTableDropColumns(dialect sql.Dialect, tableID sql.TableIdentifier, col columns.Column) (string, error) {
+	if col.ShouldSkip() {
+		return "", fmt.Errorf("received an invalid column %q", col.Name())
 	}
 
-	// createTable = false since it all successfully updated.
-	a.Tc.MutateInMemoryColumns(a.ColumnOp, mutateCol...)
-
-	return nil
+	return dialect.BuildAlterColumnQuery(tableID, constants.Delete, dialect.QuoteIdentifier(col.Name())), nil
 }