From 5668fdc358ea6d14752473ed08729cf62924717a Mon Sep 17 00:00:00 2001
From: Kaushik Iska <iska.kaushik@gmail.com>
Date: Thu, 14 Sep 2023 13:41:48 -0400
Subject: [PATCH] fix ref to val

---
 flow/connectors/postgres/cdc.go | 4 ++--
 flow/model/model.go             | 3 ++-
 2 files changed, 4 insertions(+), 3 deletions(-)

diff --git a/flow/connectors/postgres/cdc.go b/flow/connectors/postgres/cdc.go
index 1ba00ad453..22ca6e516e 100644
--- a/flow/connectors/postgres/cdc.go
+++ b/flow/connectors/postgres/cdc.go
@@ -228,7 +228,7 @@ func (p *PostgresCDCSource) consumeStream(
 
 					tablePkeyVal := model.TableWithPkey{
 						TableName:  tableName,
-						PkeyColVal: pkeyColVal,
+						PkeyColVal: *pkeyColVal,
 					}
 					_, ok := records.TablePKeyLastSeen[tablePkeyVal]
 					if !ok {
@@ -252,7 +252,7 @@ func (p *PostgresCDCSource) consumeStream(
 					}
 					tablePkeyVal := model.TableWithPkey{
 						TableName:  tableName,
-						PkeyColVal: pkeyColVal,
+						PkeyColVal: *pkeyColVal,
 					}
 					records.Records = append(records.Records, rec)
 					// all columns will be set in insert record, so add it to the map
diff --git a/flow/model/model.go b/flow/model/model.go
index 6764ca87bd..3984ad2848 100644
--- a/flow/model/model.go
+++ b/flow/model/model.go
@@ -145,6 +145,7 @@ func (r *RecordItems) ToJSON() (string, error) {
 	if err != nil {
 		return "", err
 	}
+
 	return string(jsonBytes), nil
 }
 
@@ -233,7 +234,7 @@ func (r *DeleteRecord) GetItems() *RecordItems {
 
 type TableWithPkey struct {
 	TableName  string
-	PkeyColVal interface{}
+	PkeyColVal qvalue.QValue
 }
 
 type RecordBatch struct {