Record slot size: select distinct (#1428)

Turns out we create a flows record per source/destination table pair

flow/activities/flowable.go

@@ -852,7 +852,7 @@ func (a *FlowableActivity) SendWALHeartbeat(ctx context.Context) error {
 }
 
 func (a *FlowableActivity) RecordSlotSizes(ctx context.Context) error {
-	rows, err := a.CatalogPool.Query(ctx, "SELECT flows.name, flows.config_proto FROM flows WHERE query_string IS NULL")
+	rows, err := a.CatalogPool.Query(ctx, "SELECT DISTINCT ON (name) name, config_proto FROM flows WHERE query_string IS NULL")
 	if err != nil {
 		return err
 	}