Fix rows synced and batch id display (#788)

In QRep UI we were showing Num Rows Synced but that number was in fact
rows pulled (rows in partition).
This PR fixes that and also introduces a separate column there for rows
synced.
Also fixes a bug where batch ID was not shown in CDC metrics

flow/activities/flowable.go

@@ -626,12 +626,12 @@ func (a *FlowableActivity) replicateQRepPartition(ctx context.Context,
 		shutdown <- struct{}{}
 	}()
 
-	res, err := dstConn.SyncQRepRecords(config, partition, stream)
+	rowsSynced, err := dstConn.SyncQRepRecords(config, partition, stream)
 	if err != nil {
 		return fmt.Errorf("failed to sync records: %w", err)
 	}
 
-	if res == 0 {
+	if rowsSynced == 0 {
 		log.WithFields(log.Fields{
 			"flowName": config.FlowJobName,
 		}).Infof("no records to push for partition %s\n", partition.PartitionId)
@@ -640,9 +640,15 @@ func (a *FlowableActivity) replicateQRepPartition(ctx context.Context,
 		if goroutineErr != nil {
 			return goroutineErr
 		}
+
+		err := a.CatalogMirrorMonitor.UpdateRowsSyncedForPartition(ctx, rowsSynced, runUUID, partition)
+		if err != nil {
+			return err
+		}
+
 		log.WithFields(log.Fields{
 			"flowName": config.FlowJobName,
-		}).Infof("pushed %d records\n", res)
+		}).Infof("pushed %d records\n", rowsSynced)
 	}
 
 	err = a.CatalogMirrorMonitor.UpdateEndTimeForPartition(ctx, runUUID, partition)
@@ -971,12 +977,12 @@ func (a *FlowableActivity) ReplicateXminPartition(ctx context.Context,
 		shutdown <- struct{}{}
 	}()
 
-	res, err := dstConn.SyncQRepRecords(config, partition, stream)
+	rowsSynced, err := dstConn.SyncQRepRecords(config, partition, stream)
 	if err != nil {
 		return 0, fmt.Errorf("failed to sync records: %w", err)
 	}
 
-	if res == 0 {
+	if rowsSynced == 0 {
 		log.WithFields(log.Fields{
 			"flowName": config.FlowJobName,
 		}).Info("no records to push for xmin\n")
@@ -985,9 +991,15 @@ func (a *FlowableActivity) ReplicateXminPartition(ctx context.Context,
 		if err != nil {
 			return 0, err
 		}
+
+		err = a.CatalogMirrorMonitor.UpdateRowsSyncedForPartition(ctx, rowsSynced, runUUID, partition)
+		if err != nil {
+			return 0, err
+		}
+
 		log.WithFields(log.Fields{
 			"flowName": config.FlowJobName,
-		}).Infof("pushed %d records\n", res)
+		}).Infof("pushed %d records\n", rowsSynced)
 	}
 
 	err = a.CatalogMirrorMonitor.UpdateEndTimeForPartition(ctx, runUUID, partition)

flow/connectors/utils/monitoring/monitoring.go

@@ -225,7 +225,7 @@ func (c *CatalogMirrorMonitor) UpdateStartTimeForQRepRun(ctx context.Context, ru
 		"UPDATE peerdb_stats.qrep_runs SET start_time=$1 WHERE run_uuid=$2",
 		time.Now(), runUUID)
 	if err != nil {
-		return fmt.Errorf("error while updating num_rows_to_sync for run_uuid %s in qrep_runs: %w", runUUID, err)
+		return fmt.Errorf("error while updating start time for run_uuid %s in qrep_runs: %w", runUUID, err)
 	}
 
 	return nil
@@ -240,7 +240,7 @@ func (c *CatalogMirrorMonitor) UpdateEndTimeForQRepRun(ctx context.Context, runU
 		"UPDATE peerdb_stats.qrep_runs SET end_time=$1 WHERE run_uuid=$2",
 		time.Now(), runUUID)
 	if err != nil {
-		return fmt.Errorf("error while updating num_rows_to_sync for run_uuid %s in qrep_runs: %w", runUUID, err)
+		return fmt.Errorf("error while updating end time for run_uuid %s in qrep_runs: %w", runUUID, err)
 	}
 
 	return nil
@@ -375,3 +375,17 @@ func (c *CatalogMirrorMonitor) UpdateEndTimeForPartition(ctx context.Context, ru
 	}
 	return nil
 }
+
+func (c *CatalogMirrorMonitor) UpdateRowsSyncedForPartition(ctx context.Context, rowsSynced int, runUUID string,
+	partition *protos.QRepPartition) error {
+	if c == nil || c.catalogConn == nil {
+		return nil
+	}
+
+	_, err := c.catalogConn.Exec(ctx, `UPDATE peerdb_stats.qrep_partitions SET rows_synced=$1
+	 WHERE run_uuid=$2 AND partition_uuid=$3`, rowsSynced, runUUID, partition.PartitionId)
+	if err != nil {
+		return fmt.Errorf("error while updating rows_synced in qrep_partitions: %w", err)
+	}
+	return nil
+}

nexus/catalog/migrations/V14__qrep_rows_synced.sql

@@ -0,0 +1,2 @@
+ALTER TABLE peerdb_stats.qrep_partitions
+ADD COLUMN rows_synced INTEGER;

ui/app/mirrors/edit/[mirrorId]/syncStatusTable.tsx

@@ -182,7 +182,7 @@ export const SyncStatusTable = ({ rows }: SyncStatusTableProps) => {
       {displayedRows.map((row) => (
         <TableRow key={row.batchId}>
           <TableCell>
-            <Label>{row.batchId}</Label>
+            <Label>{Number(row.batchId)}</Label>
           </TableCell>
           <TableCell>
             <Label>

ui/app/mirrors/status/qrep/[mirrorId]/page.tsx

@@ -35,7 +35,8 @@ export default async function QRepMirrorStatus({
       runUuid: run.run_uuid,
       startTime: run.start_time,
       endTime: run.end_time,
-      numRows: run.rows_in_partition,
+      pulledRows: run.rows_in_partition,
+      syncedRows: run.rows_synced,
       status: '',
     };
     return ret;
@@ -50,7 +51,7 @@ export default async function QRepMirrorStatus({
           partitionID: partition.partitionId,
           startTime: partition.startTime,
           endTime: partition.endTime,
-          numRows: partition.numRows,
+          numRows: partition.pulledRows,
         }))}
       />
       <br></br>

ui/app/mirrors/status/qrep/[mirrorId]/qrepStatusTable.tsx

@@ -17,7 +17,8 @@ export type QRepPartitionStatus = {
   status: string;
   startTime: Date | null;
   endTime: Date | null;
-  numRows: number | null;
+  pulledRows: number | null;
+  syncedRows: number | null;
 };
 
 function TimeOrProgressBar({ time }: { time: Date | null }) {
@@ -34,7 +35,8 @@ function RowPerPartition({
   status,
   startTime,
   endTime,
-  numRows,
+  pulledRows: numRows,
+  syncedRows,
 }: QRepPartitionStatus) {
   let duration = 'N/A';
   if (startTime && endTime) {
@@ -67,6 +69,9 @@ function RowPerPartition({
       <TableCell>
         <Label>{numRows}</Label>
       </TableCell>
+      <TableCell>
+        <Label>{syncedRows ?? 0}</Label>
+      </TableCell>
     </TableRow>
   );
 }
@@ -217,7 +222,8 @@ export default function QRepStatusTable({
             'Duration',
             'Start Time',
             'End Time',
-            'Num Rows Synced',
+            'Rows In Partition',
+            'Rows Synced',
           ].map((heading, index) => (
             <TableCell as='th' key={index}>
               <Label as='label' style={{ fontWeight: 'bold' }}>

ui/prisma/schema.prisma

@@ -119,6 +119,7 @@ model qrep_partitions {
   restart_count     Int
   metadata          Json?
   id                Int       @id @default(autoincrement())
+  rows_synced       Int?
   qrep_runs         qrep_runs @relation(fields: [flow_name, run_uuid], references: [flow_name, run_uuid], onDelete: Cascade, onUpdate: NoAction, map: "fk_qrep_partitions_run")
 
   @@unique([run_uuid, partition_uuid])
@@ -144,3 +145,17 @@ model qrep_runs {
   @@index([start_time], map: "idx_qrep_runs_start_time")
   @@schema("peerdb_stats")
 }
+
+model peer_slot_size {
+  id                  Int      @id @default(autoincrement())
+  slot_name           String
+  peer_name           String
+  redo_lsn            String?
+  restart_lsn         String?
+  confirmed_flush_lsn String?
+  slot_size           BigInt?
+  updated_at          DateTime @default(now()) @db.Timestamp(6)
+
+  @@index([slot_name], map: "index_slot_name")
+  @@schema("peerdb_stats")
+}