fix: deltalake syncs failing for columns with unhandled data type (#5467)

warehouse/integrations/deltalake/deltalake.go

@@ -11,6 +11,8 @@
 	"strings"
 	"time"
 
+	"github.com/samber/lo"
+
 	"github.com/rudderlabs/sqlconnect-go/sqlconnect"
 	sqlconnectconfig "github.com/rudderlabs/sqlconnect-go/sqlconnect/config"
 
@@ -35,8 +37,7 @@
 )
 
 const (
-	schemaNotFound       = "[SCHEMA_NOT_FOUND]"
-	columnsAlreadyExists = "already exists in root"
+	schemaNotFound = "[SCHEMA_NOT_FOUND]"
 )
 
 const (
@@ -526,6 +527,15 @@
 
 // AddColumns adds columns to the table.
 func (d *Deltalake) AddColumns(ctx context.Context, tableName string, columnsInfo []warehouseutils.ColumnInfo) error {
+	tableSchema, err := d.fetchTableAttributes(ctx, tableName)
+	if err != nil {
+		return fmt.Errorf("fetch table attributes: %w", err)
+	}
+	columnsToAddInfo := lo.Filter(columnsInfo, func(columnInfo warehouseutils.ColumnInfo, _ int) bool {
+		_, ok := tableSchema[columnInfo.Name]
+		return !ok
+	})
+
 	var queryBuilder strings.Builder
 
 	queryBuilder.WriteString(fmt.Sprintf(`
@@ -536,33 +546,14 @@
 		tableName,
 	))
 
-	for _, columnInfo := range columnsInfo {
+	for _, columnInfo := range columnsToAddInfo {
 		queryBuilder.WriteString(fmt.Sprintf(` %s %s,`, columnInfo.Name, dataTypesMap[columnInfo.Type]))
 	}
 
 	query := strings.TrimSuffix(queryBuilder.String(), ",")
 	query += ");"
 
-	_, err := d.DB.ExecContext(ctx, query)
-
-	// Handle error in case of single column
-	if len(columnsInfo) == 1 {
-		if err != nil && strings.Contains(err.Error(), columnsAlreadyExists) {
-			d.logger.Infow("column already exists",
-				logfield.SourceID, d.Warehouse.Source.ID,
-				logfield.SourceType, d.Warehouse.Source.SourceDefinition.Name,
-				logfield.DestinationID, d.Warehouse.Destination.ID,
-				logfield.DestinationType, d.Warehouse.Destination.DestinationDefinition.Name,
-				logfield.WorkspaceID, d.Warehouse.WorkspaceID,
-				logfield.Namespace, d.Namespace,
-				logfield.TableName, tableName,
-				logfield.ColumnName, columnsInfo[0].Name,
-				logfield.Error, err.Error(),
-			)
-			return nil
-		}
-	}
-
+	_, err = d.DB.ExecContext(ctx, query)
 	if err != nil {
 		return fmt.Errorf("adding columns: %w", err)
 	}

warehouse/integrations/deltalake/deltalake_test.go

@@ -1319,6 +1319,75 @@ func TestIntegration(t *testing.T) {
 			}
 		})
 	})
+
+	t.Run("Add columns to table", func(t *testing.T) {
+		tableName := "test_add_existing_columns"
+		namespace := whth.RandSchema(destType)
+
+		loadFiles := []whutils.LoadFile{{Location: "dummy_location"}}
+		mockUploader := newMockUploader(t, loadFiles, tableName, nil, nil, whutils.LoadFileTypeCsv, false, false)
+		ctx := context.Background()
+
+		d := deltalake.New(config.New(), logger.NOP, stats.NOP)
+		warehouse := model.Warehouse{
+			Destination: backendconfig.DestinationT{
+				ID: "test_destination_id",
+				DestinationDefinition: backendconfig.DestinationDefinitionT{
+					Name: destType,
+				},
+				Config: map[string]any{
+					"host":           credentials.Host,
+					"port":           credentials.Port,
+					"path":           credentials.Path,
+					"token":          credentials.Token,
+					"namespace":      namespace,
+					"bucketProvider": whutils.AzureBlob,
+					"containerName":  credentials.ContainerName,
+					"accountName":    credentials.AccountName,
+					"accountKey":     credentials.AccountKey,
+				},
+			},
+			Namespace: namespace,
+		}
+		err := d.Setup(ctx, warehouse, mockUploader)
+		require.NoError(t, err)
+
+		err = d.CreateSchema(ctx)
+		require.NoError(t, err)
+		t.Cleanup(func() {
+			dropSchema(t, d.DB.DB, d.Namespace)
+		})
+		columns := map[string]string{
+			"col1": "int",
+			"col2": "string",
+		}
+
+		err = d.CreateTable(ctx, tableName, columns)
+		require.NoError(t, err)
+
+		columnsToAdd := []whutils.ColumnInfo{
+			{
+				Name: "col1",
+				Type: "int",
+			},
+			{
+				Name: "col3",
+				Type: "int",
+			},
+		}
+
+		err = d.AddColumns(ctx, tableName, columnsToAdd)
+		require.NoError(t, err)
+
+		schema, err := d.FetchSchema(ctx)
+		require.NoError(t, err)
+		require.Contains(t, schema, tableName)
+		require.Equal(t, model.TableSchema{
+			"col1": "int",
+			"col2": "string",
+			"col3": "int",
+		}, schema[tableName])
+	})
 }
 
 func tableMetadata(t *testing.T, db *sql.DB, namespace, tableName string) map[string]string {