chore: sync release v1.58.3 to main branch (#6345)

# Description

Syncing patch release v1.58.3 to main branch

**↓↓ Please review and edit commit overrides before merging ↓↓**

BEGIN_COMMIT_OVERRIDE
fix(batchrouter): support configurable column ordering in CSVs in sftp
(#6342)
END_COMMIT_OVERRIDE

---------

Co-authored-by: Akash Chetty <[email protected]>
Co-authored-by: Dilip Kola <[email protected]>
Co-authored-by: Aris Tzoumas <[email protected]>

Author: 4 people

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## [1.58.3](https://github.com/rudderlabs/rudder-server/compare/v1.58.2...v1.58.3) (2025-09-15)
+
+
+### Bug Fixes
+
+* **batchrouter:** support configurable column ordering in CSVs in sftp ([#6342](https://github.com/rudderlabs/rudder-server/issues/6342)) ([1c0cf56](https://github.com/rudderlabs/rudder-server/commit/1c0cf56395a071b588c8d67400acff4bd6eb05fa))
+
 ## [1.58.2](https://github.com/rudderlabs/rudder-server/compare/v1.58.1...v1.58.2) (2025-09-11)
 
 

go.mod

@@ -220,7 +220,7 @@ require (
 	github.com/go-playground/validator/v10 v10.23.0 // indirect
 	github.com/go-sql-driver/mysql v1.8.1 // indirect
 	github.com/go-task/slim-sprig/v3 v3.0.0 // indirect
-	github.com/go-viper/mapstructure/v2 v2.4.0 // indirect
+	github.com/go-viper/mapstructure/v2 v2.4.0
 	github.com/goccy/go-json v0.10.5 // indirect
 	github.com/goccy/go-reflect v1.2.0 // indirect
 	github.com/godbus/dbus v0.0.0-20190726142602-4481cbc300e2 // indirect

router/batchrouter/asyncdestinationmanager/sftp/manager.go

@@ -5,6 +5,7 @@ import (
 	"os"
 	"time"
 
+	"github.com/go-viper/mapstructure/v2"
 	"github.com/tidwall/gjson"
 
 	"github.com/rudderlabs/rudder-go-kit/logger"
@@ -58,7 +59,7 @@ func (d *defaultManager) Upload(asyncDestStruct *common.AsyncDestinationStruct)
 	fileFormat := result.Get("fileFormat").String()
 
 	// Generate temporary file based on the destination's file format
-	jsonOrCSVFilePath, err := generateFile(textFilePath, fileFormat)
+	jsonOrCSVFilePath, err := generateFile(textFilePath, fileFormat, d.config.SortColumnNames)
 	if err != nil {
 		return generateErrorOutput(fmt.Sprintf("error generating temporary file: %v", err.Error()), asyncDestStruct.ImportingJobIDs, destinationID)
 	}
@@ -98,16 +99,21 @@ func (d *defaultManager) Upload(asyncDestStruct *common.AsyncDestinationStruct)
 	}
 }
 
-func newDefaultManager(logger logger.Logger, statsFactory stats.Stats, fileManager sftp.FileManager) *defaultManager {
+func newDefaultManager(logger logger.Logger, statsFactory stats.Stats, fileManager sftp.FileManager, config destConfig) *defaultManager {
 	return &defaultManager{
 		FileManager:  fileManager,
 		logger:       logger.Child("SFTP").Child("Manager"),
 		statsFactory: statsFactory,
+		config:       config,
 	}
 }
 
 func newInternalManager(logger logger.Logger, statsFactory stats.Stats, destination *backendconfig.DestinationT) (common.AsyncUploadAndTransformManager, error) {
-	sshConfig, err := createSSHConfig(destination)
+	var config destConfig
+	if err := mapstructure.Decode(destination.Config, &config); err != nil {
+		return nil, fmt.Errorf("unmarshalling destination config: %w", err)
+	}
+	sshConfig, err := createSSHConfig(config)
 	if err != nil {
 		return nil, fmt.Errorf("creating SSH config: %w", err)
 	}
@@ -117,7 +123,8 @@ func newInternalManager(logger logger.Logger, statsFactory stats.Stats, destinat
 		return nil, fmt.Errorf("creating file manager: %w", err)
 	}
 
-	return newDefaultManager(logger, statsFactory, fileManager), nil
+	manager := newDefaultManager(logger, statsFactory, fileManager, config)
+	return manager, nil
 }
 
 func NewManager(logger logger.Logger, statsFactory stats.Stats, destination *backendconfig.DestinationT) (common.AsyncDestinationManager, error) {