fix: return only the latest namespace entry when fetching tables from warehouse

warehouse/internal/repo/schema.go

@@ -216,7 +216,7 @@ func (sh *WHSchema) GetTablesForConnection(ctx context.Context, connections []wa
 				(source_id, destination_id) IN (` + strings.Join(sourceIDDestinationIDPairs, ", ") + `)
 			AND
 				schema::text <> '{}'::text
-			GROUP BY id
+			GROUP BY source_id, destination_id
 		)`
 	rows, err := sh.db.QueryContext(
 		ctx,

warehouse/internal/repo/schema_test.go

@@ -3,7 +3,6 @@ package repo_test
 import (
 	"context"
 	"errors"
-	"slices"
 	"testing"
 	"time"
 
@@ -122,12 +121,14 @@ func TestWHSchemasRepo(t *testing.T) {
 		connection := warehouseutils.SourceIDDestinationID{SourceID: sourceID, DestinationID: destinationID}
 		expectedTableNames, err := r.GetTablesForConnection(ctx, []warehouseutils.SourceIDDestinationID{connection})
 		require.NoError(t, err)
-		require.Equal(t, len(expectedTableNames), 1)
-		require.Equal(t, expectedTableNames[0].SourceID, sourceID)
-		require.Equal(t, expectedTableNames[0].DestinationID, destinationID)
-		require.Equal(t, expectedTableNames[0].Namespace, namespace)
-		require.True(t, slices.Contains(expectedTableNames[0].Tables, "table_name_1"))
-		require.True(t, slices.Contains(expectedTableNames[0].Tables, "table_name_2"))
+		require.ElementsMatch(t, expectedTableNames, []warehouseutils.FetchTableInfo{
+			{
+				SourceID:      sourceID,
+				DestinationID: destinationID,
+				Namespace:     namespace,
+				Tables:        []string{"table_name_1", "table_name_2"},
+			},
+		})
 
 		t.Log("cancelled context")
 		_, err = r.GetTablesForConnection(cancelledCtx, []warehouseutils.SourceIDDestinationID{connection})
@@ -142,5 +143,28 @@ func TestWHSchemasRepo(t *testing.T) {
 		t.Log("empty")
 		_, err = r.GetTablesForConnection(ctx, []warehouseutils.SourceIDDestinationID{})
 		require.EqualError(t, err, errors.New("no source id and destination id pairs provided").Error())
+
+		t.Log("multiple")
+		latestNamespace := "latest_namespace"
+		schemaLatest := model.WHSchema{
+			UploadID:        2,
+			SourceID:        sourceID,
+			Namespace:       latestNamespace,
+			DestinationID:   destinationID,
+			DestinationType: destinationType,
+			Schema:          schemaModel,
+			CreatedAt:       now,
+			UpdatedAt:       now,
+		}
+		_, err = r.Insert(ctx, &schemaLatest)
+		require.NoError(t, err)
+		expectedTableNames, err = r.GetTablesForConnection(ctx, []warehouseutils.SourceIDDestinationID{connection})
+		require.NoError(t, err)
+		require.ElementsMatch(t, expectedTableNames, []warehouseutils.FetchTableInfo{{
+			SourceID:      sourceID,
+			DestinationID: destinationID,
+			Namespace:     latestNamespace,
+			Tables:        []string{"table_name_1", "table_name_2"},
+		}}, "the new schema should not have changed the returned tables since we already had a schema for the same source id and destination id pair")
 	})
 }