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

Author: am6010

warehouse/api/http.go

@@ -172,8 +172,6 @@ func (a *Api) addMasterEndpoints(ctx context.Context, r chi.Router) {
 
 			r.Post("/jobs", a.logMiddleware(a.sourceManager.InsertJobHandler))       // TODO: add degraded mode
 			r.Get("/jobs/status", a.logMiddleware(a.sourceManager.StatusJobHandler)) // TODO: add degraded mode
-
-			r.Get("/fetch-tables", a.logMiddleware(a.fetchTablesHandler)) // TODO: Remove this endpoint once sources change is released
 		})
 	})
 	r.Route("/internal", func(r chi.Router) {

warehouse/api/http_test.go

@@ -862,7 +862,6 @@ func TestHTTPApi(t *testing.T) {
 
 			t.Run("fetch tables", func(t *testing.T) {
 				for _, u := range []string{
-					fmt.Sprintf("%s/v1/warehouse/fetch-tables", serverURL),
 					fmt.Sprintf("%s/internal/v1/warehouse/fetch-tables", serverURL),
 				} {
 					req, err := http.NewRequest(http.MethodGet, u, bytes.NewReader([]byte(`

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

@@ -142,5 +142,50 @@ 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"
+		sourceID2 := "source_id_2"
+		destinationID2 := "destination_id_2"
+		connection2 := warehouseutils.SourceIDDestinationID{SourceID: sourceID2, DestinationID: destinationID2}
+		schemaLatest := model.WHSchema{
+			UploadID:        2,
+			SourceID:        sourceID,
+			Namespace:       latestNamespace,
+			DestinationID:   destinationID,
+			DestinationType: destinationType,
+			Schema:          schemaModel,
+			CreatedAt:       now,
+			UpdatedAt:       now,
+		}
+		schema2 := model.WHSchema{
+			UploadID:        3,
+			SourceID:        sourceID2,
+			Namespace:       namespace,
+			DestinationID:   destinationID2,
+			DestinationType: destinationType,
+			Schema:          schemaModel,
+			CreatedAt:       now,
+			UpdatedAt:       now,
+		}
+		_, err = r.Insert(ctx, &schemaLatest)
+		require.NoError(t, err)
+		_, err = r.Insert(ctx, &schema2)
+		require.NoError(t, err)
+		expectedTableNames, err = r.GetTablesForConnection(ctx, []warehouseutils.SourceIDDestinationID{connection, connection2})
+		require.NoError(t, err)
+		require.Equal(t, len(expectedTableNames), 2)
+		require.Contains(t, expectedTableNames, warehouseutils.FetchTableInfo{
+			SourceID:      sourceID,
+			DestinationID: destinationID,
+			Namespace:     latestNamespace,
+			Tables:        []string{"table_name_1", "table_name_2"},
+		})
+		require.Contains(t, expectedTableNames, warehouseutils.FetchTableInfo{
+			SourceID:      sourceID2,
+			DestinationID: destinationID2,
+			Namespace:     namespace,
+			Tables:        []string{"table_name_1", "table_name_2"},
+		})
 	})
 }