chore: revert synapse staging table with max varchar length (#5817)

Reverts #5775

Author: vyeshwanth

warehouse/integrations/azure-synapse/azure-synapse.go

@@ -82,10 +82,6 @@ var stringColumns = lo.Keys(lo.PickBy(azureSynapseDataTypesMapToRudder, func(_,
 	return value == "string"
 }))
 
-var azureSynapseColsWithVariableLength = lo.Keys(lo.PickBy(azureSynapseDataTypesMapToRudder, func(key, _ string) bool {
-	return key == "varchar" || key == "nvarchar" || key == "char" || key == "nchar"
-}))
-
 type AzureSynapse struct {
 	db                 *sqlmw.DB
 	namespace          string
@@ -214,67 +210,6 @@ func (*AzureSynapse) IsEmpty(_ context.Context, _ model.Warehouse) (empty bool,
 	return
 }
 
-func (as *AzureSynapse) createStagingTable(ctx context.Context, tableName string) (string, error) {
-	stagingTableName := warehouseutils.StagingTableName(
-		provider,
-		tableName,
-		tableNameLimit,
-	)
-
-	cols := warehouseutils.SortColumnKeysFromColumnMap(as.uploader.GetTableSchemaInWarehouse(tableName))
-
-	varcharCols, err := as.getStringColumnsWithVariableLength(ctx, tableName)
-	if err != nil {
-		return "", fmt.Errorf("getting varchar columns info: %w", err)
-	}
-
-	columnDefs := lo.Map(cols, func(name string, _ int) string {
-		if dataType, ok := varcharCols[name]; ok {
-			var targetType string
-			// use varchar for char and nvarchar for nchar
-			// this is because char and nchar have a max length of 8000 and 4000 respectively.
-			// row size limit is 8060 in Azure Synapse. To fix this, we use varchar and nvarchar with max length
-			switch dataType {
-			case "char":
-				targetType = "varchar"
-			case "nchar":
-				targetType = "nvarchar"
-			default:
-				targetType = string(dataType)
-			}
-			return fmt.Sprintf(`CAST('' AS %[1]s(max)) as %[2]s`, targetType, name)
-		}
-		return name
-	})
-
-	// The use of prepared statements for creating temporary tables is not suitable in this context.
-	// Temporary tables in SQL Server have a limited scope and are automatically purged after the transaction commits.
-	// Therefore, creating normal tables is chosen as an alternative.
-	//
-	// For more information on this behavior:
-	// - See the discussion at https://github.com/denisenkom/go-mssqldb/issues/149 regarding prepared statements.
-	// - Refer to Microsoft's documentation on temporary tables at
-	//   https://docs.microsoft.com/en-us/previous-versions/sql/sql-server-2008-r2/ms175528(v=sql.105)?redirectedfrom=MSDN.
-	createStagingTableStmt := fmt.Sprintf(`
-		SELECT TOP 0 %[3]s
-			INTO
-		%[1]s.%[2]s
-			FROM
-		%[1]s.%[4]s
-		`,
-		as.namespace,
-		stagingTableName,
-		strings.Join(columnDefs, ", "),
-		tableName,
-	)
-
-	if _, err := as.db.ExecContext(ctx, createStagingTableStmt); err != nil {
-		return "", fmt.Errorf("creating staging table: %w", err)
-	}
-
-	return stagingTableName, nil
-}
-
 func (as *AzureSynapse) loadTable(
 	ctx context.Context,
 	tableName string,
@@ -300,10 +235,32 @@ func (as *AzureSynapse) loadTable(
 		misc.RemoveFilePaths(fileNames...)
 	}()
 
+	stagingTableName := warehouseutils.StagingTableName(
+		provider,
+		tableName,
+		tableNameLimit,
+	)
+
+	// The use of prepared statements for creating temporary tables is not suitable in this context.
+	// Temporary tables in SQL Server have a limited scope and are automatically purged after the transaction commits.
+	// Therefore, creating normal tables is chosen as an alternative.
+	//
+	// For more information on this behavior:
+	// - See the discussion at https://github.com/denisenkom/go-mssqldb/issues/149 regarding prepared statements.
+	// - Refer to Microsoft's documentation on temporary tables at
+	//   https://docs.microsoft.com/en-us/previous-versions/sql/sql-server-2008-r2/ms175528(v=sql.105)?redirectedfrom=MSDN.
 	log.Debugw("creating staging table")
-	stagingTableName, err := as.createStagingTable(ctx, tableName)
-	if err != nil {
-		return nil, "", fmt.Errorf("creating staging table: %w", err)
+	createStagingTableStmt := fmt.Sprintf(`
+		SELECT
+		  TOP 0 * INTO %[1]s.%[2]s
+		FROM
+		  %[1]s.%[3]s;`,
+		as.namespace,
+		stagingTableName,
+		tableName,
+	)
+	if _, err = as.db.ExecContext(ctx, createStagingTableStmt); err != nil {
+		return nil, "", fmt.Errorf("creating temporary table: %w", err)
 	}
 
 	if !skipTempTableDelete {
@@ -395,46 +352,6 @@ func (as *AzureSynapse) loadTable(
 	}, stagingTableName, nil
 }
 
-// dataType is the columnn data type in Azure Synapse
-type dataType string
-
-// getStringColumnsWithVariableLength returns the column name and data type for all columns that are of type varchar, nvarchar, char, or nchar
-func (as *AzureSynapse) getStringColumnsWithVariableLength(ctx context.Context, tableName string) (map[string]dataType, error) {
-	dataTypes := "'" + strings.Join(azureSynapseColsWithVariableLength, "', '") + "'"
-	query := fmt.Sprintf(`
-		SELECT column_name, DATA_TYPE
-		FROM INFORMATION_SCHEMA.COLUMNS
-		WHERE TABLE_SCHEMA = @schema AND TABLE_NAME = @tableName AND DATA_TYPE IN (%s);
-	`,
-		dataTypes,
-	)
-
-	rows, err := as.db.QueryContext(ctx, query,
-		sql.Named("schema", as.namespace),
-		sql.Named("tableName", tableName),
-	)
-	if err != nil {
-		return nil, fmt.Errorf("querying string columns: %w", err)
-	}
-	defer func() { _ = rows.Close() }()
-
-	stringColumns := make(map[string]dataType)
-	for rows.Next() {
-		var (
-			columnName string
-			dataType   dataType
-		)
-		if err := rows.Scan(&columnName, &dataType); err != nil {
-			return nil, fmt.Errorf("scanning string columns: %w", err)
-		}
-		stringColumns[columnName] = dataType
-	}
-	if err := rows.Err(); err != nil {
-		return nil, fmt.Errorf("iterating string columns: %w", err)
-	}
-	return stringColumns, nil
-}
-
 // getVarcharLengthMap retrieves the maximum allowed length for varchar columns in a given table.
 // A `CHARACTER_MAXIMUM_LENGTH` of `-1` indicates that the column has the maximum possible length (i.e., `varchar(max)`).
 func (as *AzureSynapse) getVarcharLengthMap(ctx context.Context, tableName string) (map[string]int, error) {

warehouse/integrations/azure-synapse/azure_synapse_test.go

@@ -774,83 +774,6 @@ func TestIntegration(t *testing.T) {
 				{"7274e5db-f918-4efe-5322-872f66e235c5", "2022-12-15T06:53:49Z", "", "2022-12-15T06:53:49Z", "", "", ""},
 			})
 		})
-
-		t.Run("should truncate varchar, nvarchar, char, nchar columns", func(t *testing.T) {
-			tableName := "load_table_with_variable_character_columns"
-
-			// Schema for upload and warehouse
-			schemaInUpload := model.TableSchema{
-				"id":               "string",
-				"received_at":      "datetime",
-				"t_c_varchar_col":  "string",
-				"t_d_nvarchar_col": "string",
-				"t_e_char_col":     "string",
-				"t_f_nchar_col":    "string",
-			}
-			schemaInWarehouse := schemaInUpload
-			// t_c_varchar_col, t_d_nvarchar_col, t_e_char_col, t_f_nchar_col have more than 512 characters in the file
-			uploadOutput := whth.UploadLoadFile(t, fm, "testdata/loadcharcolumns.csv.gz", tableName)
-
-			loadFiles := []whutils.LoadFile{{Location: uploadOutput.Location}}
-			mockUploader := newMockUploader(t, loadFiles, tableName, schemaInUpload, schemaInWarehouse)
-
-			az := azuresynapse.New(config.New(), logger.NOP, stats.NOP)
-			err := az.Setup(ctx, warehouse, mockUploader)
-			require.NoError(t, err)
-
-			err = az.CreateSchema(ctx)
-			require.NoError(t, err)
-
-			// Create table with all character columns having length 512
-			_, err = db.ExecContext(ctx, fmt.Sprintf(`
-				CREATE TABLE %s.%s (
-					id varchar(64),
-					received_at datetimeoffset,
-					t_c_varchar_col varchar(512),
-					t_d_nvarchar_col nvarchar(512),
-					t_e_char_col char(512),
-					t_f_nchar_col nchar(512)
-				)
-			`, namespace, tableName))
-			require.NoError(t, err)
-
-			// Load data which should trigger truncation for all character columns
-			loadTableStat, err := az.LoadTable(ctx, tableName)
-			require.NoError(t, err)
-			require.Equal(t, loadTableStat.RowsInserted, int64(1))
-			require.Equal(t, loadTableStat.RowsUpdated, int64(0))
-
-			// Verify all character columns are truncated to 512 chars
-			records := whth.RetrieveRecordsFromWarehouse(t, db,
-				fmt.Sprintf(`
-					SELECT
-						id,
-						received_at,
-						t_c_varchar_col,
-						t_d_nvarchar_col,
-						t_e_char_col,
-						t_f_nchar_col
-					FROM %q.%q
-					ORDER BY id;
-				`,
-					namespace,
-					tableName,
-				),
-			)
-
-			// Each character column should contain exactly 512 'a' characters
-			expected512Chars := strings.Repeat("a", 512)
-			require.Equal(t, records, [][]string{
-				{
-					"6734e5db-f918-4efe-1421-872f66e235c5",
-					"2022-12-15T06:53:49Z",
-					expected512Chars,
-					expected512Chars,
-					expected512Chars,
-					expected512Chars,
-				},
-			})
-		})
 	})
 }