fix: use source-id for getting cached schemas (#6268)

# Description

- Use sourceID for getting schemas instead of ID to be more
deterministic in cases where schemas are available for the same
destinationID and namespace.

## Linear Ticket

- Resolves WAR-1064

## 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/internal/repo/schema.go

@@ -5,10 +5,11 @@ import (
 	"database/sql"
 	"errors"
 	"fmt"
-	"reflect"
 	"strings"
 	"time"
 
+	"github.com/google/go-cmp/cmp"
+
 	"github.com/rudderlabs/rudder-go-kit/config"
 	"github.com/rudderlabs/rudder-go-kit/jsonrs"
 
@@ -234,8 +235,9 @@ func (sh *WHSchema) GetForNamespace(ctx context.Context, destID, namespace strin
 	if err != nil {
 		return model.WHSchema{}, err
 	}
-	if !reflect.DeepEqual(originalSchema.Schema, tableLevelSchemas) {
-		return model.WHSchema{}, errors.New("parent schema does not match parent schema")
+	diff := cmp.Diff(originalSchema.Schema, tableLevelSchemas)
+	if len(diff) > 0 {
+		return model.WHSchema{}, fmt.Errorf("parent schema does not match: %s", diff)
 	}
 	return originalSchema, nil
 }
@@ -247,7 +249,7 @@ func (sh *WHSchema) getForNamespace(ctx context.Context, destID, namespace strin
 		namespace = $2 AND
 		table_name = ''
 	ORDER BY
-		id DESC;
+		source_id DESC;
 	`
 
 	rows, err := sh.db.QueryContext(
@@ -329,13 +331,13 @@ func (sh *WHSchema) getTableLevelSchemasForNamespaceWithTx(ctx context.Context,
 				schema,
 				ROW_NUMBER() OVER (
 					PARTITION BY destination_id, namespace, table_name
-					ORDER BY id DESC
+					ORDER BY source_id DESC
 				) as rn
 			FROM ` + whSchemaTableName + `
 			WHERE
-				destination_id = $1
-				AND namespace = $2
-				AND table_name != ''
+				destination_id = $1 AND
+				namespace = $2 AND
+				table_name != ''
 		) t
 		WHERE rn = 1;
 	`

warehouse/internal/repo/schema_test.go

@@ -443,6 +443,61 @@ func TestWHSchemasRepo(t *testing.T) {
 	}
 }
 
+func TestWHSchemasRepo_GetForNamespace(t *testing.T) {
+	t.Run("SourceID ordering", func(t *testing.T) {
+		db, ctx := setupDB(t), context.Background()
+		now := time.Now().Truncate(time.Second).UTC()
+		r := repo.NewWHSchemas(db, config.New(), repo.WithNow(func() time.Time {
+			return now
+		}))
+
+		require.NoError(t, r.Insert(ctx, &model.WHSchema{
+			SourceID:        "source_id_2",
+			Namespace:       "namespace_1",
+			DestinationID:   "destination_id_1",
+			DestinationType: "destination_type_1",
+			Schema: model.Schema{
+				"table_name_1": {
+					"column_name_1": "string",
+					"column_name_2": "int",
+					"column_name_3": "boolean",
+				},
+			},
+		}))
+		require.NoError(t, r.Insert(ctx, &model.WHSchema{
+			SourceID:        "source_id_1",
+			Namespace:       "namespace_1",
+			DestinationID:   "destination_id_1",
+			DestinationType: "destination_type_1",
+			Schema: model.Schema{
+				"table_name_1": {
+					"column_name_1": "string",
+					"column_name_2": "int",
+					"column_name_3": "boolean",
+				},
+			},
+			CreatedAt: now,
+			UpdatedAt: now,
+		}))
+
+		expectedSchema, err := r.GetForNamespace(ctx, "destination_id_1", "namespace_1")
+		require.NoError(t, err)
+		require.Equal(t, "source_id_2", expectedSchema.SourceID)
+		require.Equal(t, "namespace_1", expectedSchema.Namespace)
+		require.Equal(t, "destination_id_1", expectedSchema.DestinationID)
+		require.Equal(t, "destination_type_1", expectedSchema.DestinationType)
+		require.Equal(t, model.Schema{
+			"table_name_1": {
+				"column_name_1": "string",
+				"column_name_2": "int",
+				"column_name_3": "boolean",
+			},
+		}, expectedSchema.Schema)
+		require.Equal(t, now, expectedSchema.CreatedAt)
+		require.Equal(t, now, expectedSchema.UpdatedAt)
+	})
+}
+
 func TestWHSchemasRepo_GetDestinationNamespaces(t *testing.T) {
 	destinationID := "test_destination_id"
 	conf := config.New()