fix: deltalake fetch table attributes (#6455)

# Description

- There seems to be some breaking changes as `DESCRIBE QUERY TABLE` for
Databricks started returning 4 columns instead of 3 columns: col_name,
data_type, comment and metadata. [Docs
link.](https://docs.databricks.com/aws/en/sql/language-manual/sql-ref-syntax-aux-describe-table)

<img width="633" height="104" alt="Screenshot 2025-10-23 at 8 30 45 AM"
src="https://github.com/user-attachments/assets/60d76c9a-a50f-48fe-abee-1f6abb2e5399"
/>

- Dynamically checking for col_name and data_type and populating table
schema.


## Linear Ticket

- Resolves WAR-1216

## Security

- [x] The code changed/added as part of this pull request won't create
any security issues with how the software is being used.

Author: achettyiitr

warehouse/integrations/deltalake/deltalake.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/rudderlabs/rudder-go-kit/jsonrs"
 	obskit "github.com/rudderlabs/rudder-observability-kit/go/labels"
+
 	"github.com/rudderlabs/rudder-server/utils/misc"
 	warehouseclient "github.com/rudderlabs/rudder-server/warehouse/client"
 	sqlmiddleware "github.com/rudderlabs/rudder-server/warehouse/integrations/middleware/sqlquerywrapper"
@@ -404,31 +405,62 @@ func (d *Deltalake) sendStatForMissingDatatype(missingDatatype string) {
 
 // fetchTableAttributes fetches the attributes of a table
 func (d *Deltalake) fetchTableAttributes(ctx context.Context, tableName string) (model.TableSchema, error) {
-	tableSchema := make(model.TableSchema)
-
 	query := fmt.Sprintf(`DESCRIBE QUERY TABLE %s.%s;`, d.Namespace, tableName)
 
 	rows, err := d.DB.QueryContext(ctx, query)
 	if err != nil {
-		return nil, fmt.Errorf("executing fetching table attributes: %w", err)
+		return nil, fmt.Errorf("executing describe query table: %w", err)
 	}
 	defer func() { _ = rows.Close() }()
 
+	columns, err := rows.Columns()
+	if err != nil {
+		return nil, fmt.Errorf("getting rows columns: %w", err)
+	}
+	columnsMap := make(map[string]int)
+	for i, col := range columns {
+		columnsMap[col] = i
+	}
+
+	columnNameIndex, ok := columnsMap["col_name"]
+	if !ok {
+		return nil, errors.New("column name index not found")
+	}
+	datatypeIndex, ok := columnsMap["data_type"]
+	if !ok {
+		return nil, errors.New("datatype index not found")
+	}
+
+	scanDest := make([]any, len(columns))
+	scanPtrs := make([]any, len(columns))
+	for i := range scanDest {
+		scanPtrs[i] = &scanDest[i]
+	}
+
+	tableSchema := make(model.TableSchema)
+
 	for rows.Next() {
-		var (
-			colName, datatype string
-			comment           sql.NullString
-		)
+		if err := rows.Scan(scanPtrs...); err != nil {
+			return nil, fmt.Errorf("scanning row: %w", err)
+		}
+
+		colName, ok := scanDest[columnNameIndex].(string)
+		if !ok {
+			return nil, fmt.Errorf("column name is not a string: %v", scanDest[columnNameIndex])
+		}
 
-		if err = rows.Scan(&colName, &datatype, &comment); err != nil {
-			return nil, fmt.Errorf("processing fetched table attributes: %w", err)
+		datatype, ok := scanDest[datatypeIndex].(string)
+		if !ok {
+			return nil, fmt.Errorf("datatype is not a string: %v", scanDest[datatypeIndex])
 		}
 
 		tableSchema[colName] = datatype
 	}
+
 	if err = rows.Err(); err != nil {
-		return nil, fmt.Errorf("processing fetched table attributes: %w", err)
+		return nil, fmt.Errorf("iterating rows: %w", err)
 	}
+
 	return tableSchema, nil
 }