Merge branch 'master' into dependabot/go_modules/golang.org/x/net-0.36.0

Author: Tang8330

clients/redshift/dialect/default.go

@@ -1,7 +1,15 @@
 package dialect
 
-import "github.com/artie-labs/transfer/lib/sql"
+import (
+	"fmt"
+
+	"github.com/artie-labs/transfer/lib/sql"
+)
 
 func (RedshiftDialect) GetDefaultValueStrategy() sql.DefaultValueStrategy {
 	return sql.Backfill
 }
+
+func (RedshiftDialect) BuildBackfillQuery(tableID sql.TableIdentifier, escapedColumn string, defaultValue any) string {
+	return fmt.Sprintf(`UPDATE %s SET %s = %v WHERE %s IS NULL;`, tableID.FullyQualifiedName(), escapedColumn, defaultValue, escapedColumn)
+}

clients/redshift/dialect/default_test.go

@@ -0,0 +1,18 @@
+package dialect
+
+import (
+	"testing"
+
+	"github.com/artie-labs/transfer/lib/typing"
+	"github.com/artie-labs/transfer/lib/typing/columns"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestRedshiftDialect_BuildBackfillQuery(t *testing.T) {
+	_dialect := RedshiftDialect{}
+
+	tableID := NewTableIdentifier("{SCHEMA}", "{TABLE}")
+	col := columns.NewColumn("{COLUMN}", typing.String)
+
+	assert.Equal(t, `UPDATE {SCHEMA}."{table}" SET "{column}" = {DEFAULT_VALUE} WHERE "{column}" IS NULL;`, _dialect.BuildBackfillQuery(tableID, _dialect.QuoteIdentifier(col.Name()), "{DEFAULT_VALUE}"))
+}

clients/shared/default_value.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	bigQueryDialect "github.com/artie-labs/transfer/clients/bigquery/dialect"
+	redshiftDialect "github.com/artie-labs/transfer/clients/redshift/dialect"
 	"github.com/artie-labs/transfer/lib/destination"
 	"github.com/artie-labs/transfer/lib/sql"
 	"github.com/artie-labs/transfer/lib/typing"
@@ -72,23 +73,30 @@ func DefaultValue(column columns.Column, dialect sql.Dialect) (any, error) {
 }
 
 func BackfillColumn(dest destination.Destination, column columns.Column, tableID sql.TableIdentifier) error {
-	switch dest.Dialect().GetDefaultValueStrategy() {
+	dialect := dest.Dialect()
+	switch dialect.GetDefaultValueStrategy() {
 	case sql.Backfill:
 		if !column.ShouldBackfill() {
 			// If we don't need to backfill, don't backfill.
 			return nil
 		}
 
-		defaultVal, err := DefaultValue(column, dest.Dialect())
+		defaultVal, err := DefaultValue(column, dialect)
 		if err != nil {
 			return fmt.Errorf("failed to escape default value: %w", err)
 		}
 
-		escapedCol := dest.Dialect().QuoteIdentifier(column.Name())
+		escapedCol := dialect.QuoteIdentifier(column.Name())
 		query := fmt.Sprintf(`UPDATE %s as t SET t.%s = %v WHERE t.%s IS NULL;`,
 			// UPDATE table as t SET t.col = default_val WHERE t.col IS NULL
 			tableID.FullyQualifiedName(), escapedCol, defaultVal, escapedCol,
 		)
+
+		if rd, ok := dialect.(redshiftDialect.RedshiftDialect); ok {
+			// Redshift UPDATE does not support table aliasing nor do we need it. Redshift will not throw an ambiguous error if the table and column name are the same.
+			query = rd.BuildBackfillQuery(tableID, escapedCol, defaultVal)
+		}
+
 		slog.Info("Backfilling column",
 			slog.String("colName", column.Name()),
 			slog.String("query", query),
@@ -100,7 +108,7 @@ func BackfillColumn(dest destination.Destination, column columns.Column, tableID
 		}
 
 		query = fmt.Sprintf(`COMMENT ON COLUMN %s.%s IS '%v';`, tableID.FullyQualifiedName(), escapedCol, `{"backfilled": true}`)
-		if _, ok := dest.Dialect().(bigQueryDialect.BigQueryDialect); ok {
+		if _, ok := dialect.(bigQueryDialect.BigQueryDialect); ok {
 			query = fmt.Sprintf("ALTER TABLE %s ALTER COLUMN %s SET OPTIONS (description=`%s`);",
 				// ALTER TABLE table ALTER COLUMN col set OPTIONS (description=...)
 				tableID.FullyQualifiedName(), escapedCol, `{"backfilled": true}`,