WIP.

clients/redshift/dialect/dialect.go

@@ -147,6 +147,10 @@ func (RedshiftDialect) BuildAlterColumnQuery(tableID sql.TableIdentifier, column
 	return fmt.Sprintf("ALTER TABLE %s %s COLUMN %s", tableID.FullyQualifiedName(), columnOp, colSQLPart)
 }
 
+func (rd RedshiftDialect) BuildIncreaseStringPrecisionQuery(tableID sql.TableIdentifier, column columns.Column, newPrecision int32) string {
+	return fmt.Sprintf("ALTER TABLE %s ALTER COLUMN %s TYPE VARCHAR(%d)", tableID.FullyQualifiedName(), rd.QuoteIdentifier(column.Name()), newPrecision)
+}
+
 func (rd RedshiftDialect) BuildIsNotToastValueExpression(tableAlias constants.TableAlias, column columns.Column) string {
 	colName := sql.QuoteTableAliasColumn(tableAlias, column, rd)
 	if column.KindDetails == typing.Struct {

clients/redshift/dialect/dialect_test.go

@@ -195,6 +195,16 @@ func TestRedshiftDialect_BuildAlterColumnQuery(t *testing.T) {
 	)
 }
 
+func TestRedshiftDialect_BuildIncreaseStringPrecisionQuery(t *testing.T) {
+	fakeTableID := &mocks.FakeTableIdentifier{}
+	fakeTableID.FullyQualifiedNameReturns("{TABLE}")
+
+	assert.Equal(t,
+		`ALTER TABLE {TABLE} ALTER COLUMN "{column}" TYPE VARCHAR(12345)`,
+		RedshiftDialect{}.BuildIncreaseStringPrecisionQuery(fakeTableID, columns.NewColumn("{COLUMN}", typing.String), 12345),
+	)
+}
+
 func TestRedshiftDialect_BuildIsNotToastValueExpression(t *testing.T) {
 	assert.Equal(t,
 		`COALESCE(tbl."bar" != '__debezium_unavailable_value', true)`,

lib/config/types.go

@@ -39,6 +39,8 @@ type Kafka struct {
 type SharedDestinationSettings struct {
 	// TruncateExceededValues - This will truncate exceeded values instead of replacing it with `__artie_exceeded_value`
 	TruncateExceededValues bool `yaml:"truncateExceededValues"`
+	// IncreaseStringPrecision - This will increase the string precision to the maximum allowed by the destination.
+	IncreaseStringPrecision bool `yaml:"increaseStringPrecision"`
 }
 
 type Config struct {