From 5d88e941afa461102ed9afa291ed672b442663b2 Mon Sep 17 00:00:00 2001
From: Mark C <mark@Marks-MacBook-Pro.local>
Date: Fri, 17 Apr 2026 13:43:37 -0700
Subject: [PATCH] Partial update support in common

---
 connectors/common/base.go   | 18 +++++++++++++++---
 internal/app/progress.go    |  3 +++
 protocol/iface/transport.go |  2 ++
 3 files changed, 20 insertions(+), 3 deletions(-)

diff --git a/connectors/common/base.go b/connectors/common/base.go
index babc7578..efe253ea 100644
--- a/connectors/common/base.go
+++ b/connectors/common/base.go
@@ -798,6 +798,8 @@ func (c *connector) StartReadToChannel(flowId iface.FlowID, options iface.Connec
 								mutationType = iface.MutationType_Update
 							case adiomv1.UpdateType_UPDATE_TYPE_DELETE:
 								mutationType = iface.MutationType_Delete
+							case adiomv1.UpdateType_UPDATE_TYPE_PARTIAL_UPDATE:
+								mutationType = iface.MutationType_Apply
 							}
 
 							dataChannel <- iface.DataMessage{
@@ -822,6 +824,8 @@ func (c *connector) StartReadToChannel(flowId iface.FlowID, options iface.Connec
 							mutationType = iface.MutationType_Update
 						case adiomv1.UpdateType_UPDATE_TYPE_DELETE:
 							mutationType = iface.MutationType_Delete
+						case adiomv1.UpdateType_UPDATE_TYPE_PARTIAL_UPDATE:
+							mutationType = iface.MutationType_Apply
 						}
 
 						dataChannel <- iface.DataMessage{
@@ -843,6 +847,8 @@ func (c *connector) StartReadToChannel(flowId iface.FlowID, options iface.Connec
 						mutationType = iface.MutationType_Update
 					case adiomv1.UpdateType_UPDATE_TYPE_DELETE:
 						mutationType = iface.MutationType_Delete
+					case adiomv1.UpdateType_UPDATE_TYPE_PARTIAL_UPDATE:
+						mutationType = iface.MutationType_Apply
 					}
 					c.progressTracker.UpdateChangeStreamProgressTracking()
 					readerProgress.changeStreamEvents.Add(1)
@@ -1123,21 +1129,27 @@ func (c *connector) ProcessDataMessages(dataMsgs []iface.DataMessage) error {
 			c.progressTracker.UpdateWriteLSN(dataMsg.SeqNum)
 		case iface.MutationType_Insert:
 			msgs = append(msgs, &adiomv1.Update{
-				Id:   *&dataMsg.Id,
+				Id:   dataMsg.Id,
 				Type: adiomv1.UpdateType_UPDATE_TYPE_INSERT,
 				Data: *dataMsg.Data,
 			})
 		case iface.MutationType_Update:
 			msgs = append(msgs, &adiomv1.Update{
-				Id:   *&dataMsg.Id,
+				Id:   dataMsg.Id,
 				Type: adiomv1.UpdateType_UPDATE_TYPE_UPDATE,
 				Data: *dataMsg.Data,
 			})
 		case iface.MutationType_Delete:
 			msgs = append(msgs, &adiomv1.Update{
-				Id:   *&dataMsg.Id,
+				Id:   dataMsg.Id,
 				Type: adiomv1.UpdateType_UPDATE_TYPE_DELETE,
 			})
+		case iface.MutationType_Apply:
+			msgs = append(msgs, &adiomv1.Update{
+				Id:   dataMsg.Id,
+				Type: adiomv1.UpdateType_UPDATE_TYPE_PARTIAL_UPDATE,
+				Data: *dataMsg.Data,
+			})
 		default:
 			slog.Error(fmt.Sprintf("unsupported operation type during batch: %v", dataMsg.MutationType))
 		}
diff --git a/internal/app/progress.go b/internal/app/progress.go
index beddfb87..878c7aed 100644
--- a/internal/app/progress.go
+++ b/internal/app/progress.go
@@ -50,7 +50,10 @@ func (tv *TViewDetails) SetUpDisplay(app *tview.Application, errorText *tview.Te
 		AddItem(progressBarTextView, 1, 1, false).
 		AddItem(errorText, 0, 1, false)
 	tv.root = root //indices are 0, 1, 2, 3, corresponding to header, table, progressBar, and errorLogs respectively
+	table.SetFixed(1, 0)             // Keep the header row fixed
+	table.SetSelectable(true, false) // Allow row selection (vertical scrolling)
 	tv.app.SetRoot(root, true)
+	tv.app.SetFocus(table) // Set focus to the table (so that the user can scroll through it)
 }
 
 // Get the latest status report based on the runner progress struct and update the tview components accordingly
diff --git a/protocol/iface/transport.go b/protocol/iface/transport.go
index c2f0a030..2f2ec28a 100644
--- a/protocol/iface/transport.go
+++ b/protocol/iface/transport.go
@@ -42,6 +42,8 @@ const (
 	MutationType_Ignore
 
 	MutationType_Barrier
+
+	MutationType_Apply
 )
 
 const (