APIE-532 - Tableflow Upsert topic (#3179)

tmalikconfluent · web-flow · commit 4f479b79aa6d · 2025-09-12T12:04:08.000-05:00
diff --git a/go.mod b/go.mod
@@ -50,7 +50,7 @@ require (
 	github.com/confluentinc/ccloud-sdk-go-v2/service-quota v0.2.0
 	github.com/confluentinc/ccloud-sdk-go-v2/srcm v0.7.3
 	github.com/confluentinc/ccloud-sdk-go-v2/sso v0.0.1
-	github.com/confluentinc/ccloud-sdk-go-v2/tableflow v0.2.0
+	github.com/confluentinc/ccloud-sdk-go-v2/tableflow v0.4.0
 	github.com/confluentinc/ccloud-sdk-go-v2/usm v0.1.0
 	github.com/confluentinc/cmf-sdk-go v0.0.4
 	github.com/confluentinc/confluent-kafka-go/v2 v2.8.0
diff --git a/go.sum b/go.sum
@@ -258,8 +258,8 @@ github.com/confluentinc/ccloud-sdk-go-v2/srcm v0.7.3 h1:ozdDSJHruQIgtxS5hwz8Rp8p
 github.com/confluentinc/ccloud-sdk-go-v2/srcm v0.7.3/go.mod h1:cD0AeCMBAWBesmWxWCMgVYNABYgHJ/ahCj7b4HP2R2I=
 github.com/confluentinc/ccloud-sdk-go-v2/sso v0.0.1 h1:WZJYfgXJrvTIYQpCFps/qHF7T8ekgPlX/SFqx4EY2zQ=
 github.com/confluentinc/ccloud-sdk-go-v2/sso v0.0.1/go.mod h1:kB+MXWYYg9ohrTCb27LlfpTbuexAzyYAmum105ow0ho=
-github.com/confluentinc/ccloud-sdk-go-v2/tableflow v0.2.0 h1:ALDmQhIbzKjOz8xVgOx0ud89P3YFiOWff6IpXggkEdc=
-github.com/confluentinc/ccloud-sdk-go-v2/tableflow v0.2.0/go.mod h1:9eUl1KxQurBT6HgO/Whd7v2mfCBeqkA7FV8IGAjCOPY=
+github.com/confluentinc/ccloud-sdk-go-v2/tableflow v0.4.0 h1:QFp1J5P4o3I28c4eXU12gZFC2YauG8XozxvHBXltzFo=
+github.com/confluentinc/ccloud-sdk-go-v2/tableflow v0.4.0/go.mod h1:WRCZS91/w+RlmsBwHsxcTGW8zIquRAJNnzkJtuWk2yY=
 github.com/confluentinc/ccloud-sdk-go-v2/usm v0.1.0 h1:rF9cKecDCowq+oDWjf8rSpXXZHAnVXowIsT/OXF4MOI=
 github.com/confluentinc/ccloud-sdk-go-v2/usm v0.1.0/go.mod h1:umhEDvQp/5h0ALKBpYTQOmFwaWrvilnbE8Rkzh6oJ4Q=
 github.com/confluentinc/cmf-sdk-go v0.0.4 h1:IAzACCIgcp0OAah9pvr6xtqaLUoQNoorxynNzIH5dQQ=
diff --git a/internal/tableflow/command_topic.go b/internal/tableflow/command_topic.go
@@ -35,6 +35,7 @@ type topicOut struct {
 	TablePath             string `human:"Table Path" serialized:"table_path"`
 	Phase                 string `human:"Phase" serialized:"phase"`
 	ErrorMessage          string `human:"Error Message,omitempty" serialized:"error_message,omitempty"`
+	WriteMode             string `human:"Write Mode,omitempty" serialized:"write_mode,omitempty"`
 }
 
 func (c *command) newTopicCommand() *cobra.Command {
@@ -131,6 +132,7 @@ func printTopicTable(cmd *cobra.Command, topic tableflowv1.TableflowV1TableflowT
 		Suspended:             topic.Spec.GetSuspended(),
 		Phase:                 topic.Status.GetPhase(),
 		ErrorMessage:          topic.Status.GetErrorMessage(),
+		WriteMode:             topic.Status.GetWriteMode(),
 	}
 
 	if storageType == byos {
diff --git a/internal/tableflow/command_topic_list.go b/internal/tableflow/command_topic_list.go
@@ -63,6 +63,7 @@ func (c *command) list(cmd *cobra.Command, _ []string) error {
 			Suspended:             topic.Spec.GetSuspended(),
 			Phase:                 topic.Status.GetPhase(),
 			ErrorMessage:          topic.Status.GetErrorMessage(),
+			WriteMode:             topic.Status.GetWriteMode(),
 		}
 
 		if storageType == byos {
diff --git a/test/fixtures/output/tableflow/topic/describe-topic-json.golden b/test/fixtures/output/tableflow/topic/describe-topic-json.golden
@@ -13,5 +13,6 @@
   "suspended": false,
   "table_formats": "ICEBERG",
   "table_path": "s3://dummy-bucket-name-1//10011010/11101100/org-1/env-2/lkc-3/v1/tableId",
-  "phase": "RUNNING"
+  "phase": "RUNNING",
+  "write_mode": "UPSERT"
 }
diff --git a/test/fixtures/output/tableflow/topic/describe-topic.golden b/test/fixtures/output/tableflow/topic/describe-topic.golden
@@ -14,4 +14,5 @@
 | Table Formats           | ICEBERG                                                                  |
 | Table Path              | s3://dummy-bucket-name-1//10011010/11101100/org-1/env-2/lkc-3/v1/tableId |
 | Phase                   | RUNNING                                                                  |
+| Write Mode              | UPSERT                                                                   |
 +-------------------------+--------------------------------------------------------------------------+
diff --git a/test/fixtures/output/tableflow/topic/enable-topic-byob.golden b/test/fixtures/output/tableflow/topic/enable-topic-byob.golden
@@ -14,4 +14,5 @@
 | Table Formats           | ICEBERG                                                                  |
 | Table Path              | s3://dummy-bucket-name-1//10011010/11101100/org-1/env-2/lkc-3/v1/tableId |
 | Phase                   | RUNNING                                                                  |
+| Write Mode              | APPEND                                                                   |
 +-------------------------+--------------------------------------------------------------------------+
diff --git a/test/fixtures/output/tableflow/topic/enable-topic-managed.golden b/test/fixtures/output/tableflow/topic/enable-topic-managed.golden
@@ -11,4 +11,5 @@
 | Table Formats           | DELTA                                                                    |
 | Table Path              | s3://dummy-bucket-name-1//10011010/11101100/org-1/env-2/lkc-3/v1/tableId |
 | Phase                   | RUNNING                                                                  |
+| Write Mode              | APPEND                                                                   |
 +-------------------------+--------------------------------------------------------------------------+
diff --git a/test/fixtures/output/tableflow/topic/list-topic-json.golden b/test/fixtures/output/tableflow/topic/list-topic-json.golden
@@ -14,7 +14,8 @@
     "suspended": false,
     "table_formats": "ICEBERG",
     "table_path": "s3://dummy-bucket-name-1//10011010/11101100/org-1/env-2/lkc-3/v1/tableId",
-    "phase": "RUNNING"
+    "phase": "RUNNING",
+    "write_mode": "UPSERT"
   },
   {
     "kafka_cluster": "lkc-123456",
@@ -28,6 +29,7 @@
     "suspended": false,
     "table_formats": "DELTA",
     "table_path": "s3://dummy-bucket-name-1//10011010/11101100/org-1/env-2/lkc-3/v1/tableId",
-    "phase": "RUNNING"
+    "phase": "RUNNING",
+    "write_mode": "APPEND"
   }
 ]
diff --git a/test/fixtures/output/tableflow/topic/list-topic.golden b/test/fixtures/output/tableflow/topic/list-topic.golden
@@ -1,4 +1,4 @@
-  Kafka Cluster |  Topic Name   | Enable Compaction | Enable Partitioning | Environment | Record Failure Strategy | Retention Ms | Storage Type | Provider Integration ID | Bucket Name | Bucket Region | Suspended | Table Formats |                                Table Path                                |  Phase  | Error Message  
-----------------+---------------+-------------------+---------------------+-------------+-------------------------+--------------+--------------+-------------------------+-------------+---------------+-----------+---------------+--------------------------------------------------------------------------+---------+----------------
-  lkc-123456    | topic-byob    | true              | true                | env-596     | SKIP                    | 604800000    | BYOS         | cspi-stgce89r7          | bucket_1    | us-east-1     | false     | ICEBERG       | s3://dummy-bucket-name-1//10011010/11101100/org-1/env-2/lkc-3/v1/tableId | RUNNING |                
-  lkc-123456    | topic-managed | true              | true                | env-596     | SUSPEND                 | 604800000    | MANAGED      |                         |             |               | false     | DELTA         | s3://dummy-bucket-name-1//10011010/11101100/org-1/env-2/lkc-3/v1/tableId | RUNNING |                
+  Kafka Cluster |  Topic Name   | Enable Compaction | Enable Partitioning | Environment | Record Failure Strategy | Retention Ms | Storage Type | Provider Integration ID | Bucket Name | Bucket Region | Suspended | Table Formats |                                Table Path                                |  Phase  | Error Message | Write Mode  
+----------------+---------------+-------------------+---------------------+-------------+-------------------------+--------------+--------------+-------------------------+-------------+---------------+-----------+---------------+--------------------------------------------------------------------------+---------+---------------+-------------
+  lkc-123456    | topic-byob    | true              | true                | env-596     | SKIP                    | 604800000    | BYOS         | cspi-stgce89r7          | bucket_1    | us-east-1     | false     | ICEBERG       | s3://dummy-bucket-name-1//10011010/11101100/org-1/env-2/lkc-3/v1/tableId | RUNNING |               | UPSERT      
+  lkc-123456    | topic-managed | true              | true                | env-596     | SUSPEND                 | 604800000    | MANAGED      |                         |             |               | false     | DELTA         | s3://dummy-bucket-name-1//10011010/11101100/org-1/env-2/lkc-3/v1/tableId | RUNNING |               | APPEND      
diff --git a/test/fixtures/output/tableflow/topic/update-topic.golden b/test/fixtures/output/tableflow/topic/update-topic.golden
@@ -14,4 +14,5 @@
 | Table Formats           | ICEBERG                                                                  |
 | Table Path              | s3://dummy-bucket-name-1//10011010/11101100/org-1/env-2/lkc-3/v1/tableId |
 | Phase                   | RUNNING                                                                  |
+| Write Mode              | UPSERT                                                                   |
 +-------------------------+--------------------------------------------------------------------------+
diff --git a/test/test-server/tableflow_handlers.go b/test/test-server/tableflow_handlers.go
@@ -80,7 +80,7 @@ func handleTableflowTopicsCreate(t *testing.T, environment string) http.HandlerF
 		require.NoError(t, err)
 
 		tableflowTopic.Spec.Config.SetEnableCompaction(true) //readonly attribute, always true
-		tableflowTopic.Status = tableflowv1.NewTableflowV1TableflowTopicStatus()
+		tableflowTopic.Status = tableflowv1.NewTableflowV1TableflowTopicStatus("APPEND")
 		tableflowTopic.Status.SetPhase("RUNNING")
 		tableflowTopic.Spec.SetEnvironment(tableflowv1.GlobalObjectReference{Id: environment})
 
@@ -190,6 +190,7 @@ func getTopicByob(display_name, environmentId, clusterId string) tableflowv1.Tab
 		Status: &tableflowv1.TableflowV1TableflowTopicStatus{
 			Phase: tableflowv1.PtrString("RUNNING"),
 			//ErrorMessage: tableflowv1.PtrString(""),
+			WriteMode: "UPSERT",
 		},
 	}
 }
@@ -218,6 +219,7 @@ func getTopicManaged(display_name, environmentId, clusterId string) tableflowv1.
 		Status: &tableflowv1.TableflowV1TableflowTopicStatus{
 			Phase: tableflowv1.PtrString("RUNNING"),
 			//ErrorMessage: tableflowv1.PtrString(""),
+			WriteMode: "APPEND",
 		},
 	}
 }

Original file line number	Diff line number	Diff line change
`@@ -35,6 +35,7 @@ type topicOut struct {`
`35`	`35`	TablePath string `human:"Table Path" serialized:"table_path"`
`36`	`36`	Phase string `human:"Phase" serialized:"phase"`
`37`	`37`	ErrorMessage string `human:"Error Message,omitempty" serialized:"error_message,omitempty"`
	`38`	+ WriteMode string `human:"Write Mode,omitempty" serialized:"write_mode,omitempty"`
`38`	`39`	`}`
`39`	`40`
`40`	`41`	`func (c command) newTopicCommand() cobra.Command {`
`@@ -131,6 +132,7 @@ func printTopicTable(cmd *cobra.Command, topic tableflowv1.TableflowV1TableflowT`
`131`	`132`	`Suspended: topic.Spec.GetSuspended(),`
`132`	`133`	`Phase: topic.Status.GetPhase(),`
`133`	`134`	`ErrorMessage: topic.Status.GetErrorMessage(),`
	`135`	`+ WriteMode: topic.Status.GetWriteMode(),`
`134`	`136`	`}`
`135`	`137`
`136`	`138`	`if storageType == byos {`
Original file line number	Diff line number	Diff line change
`@@ -63,6 +63,7 @@ func (c command) list(cmd cobra.Command, _ []string) error {`
`63`	`63`	`Suspended: topic.Spec.GetSuspended(),`
`64`	`64`	`Phase: topic.Status.GetPhase(),`
`65`	`65`	`ErrorMessage: topic.Status.GetErrorMessage(),`
	`66`	`+ WriteMode: topic.Status.GetWriteMode(),`
`66`	`67`	`}`
`67`	`68`
`68`	`69`	`if storageType == byos {`
Original file line number	Diff line number	Diff line change
`@@ -13,5 +13,6 @@`
`13`	`13`	`"suspended": false,`
`14`	`14`	`"table_formats": "ICEBERG",`
`15`	`15`	`"table_path": "s3://dummy-bucket-name-1//10011010/11101100/org-1/env-2/lkc-3/v1/tableId",`
`16`		`- "phase": "RUNNING"`
	`16`	`+ "phase": "RUNNING",`
	`17`	`+ "write_mode": "UPSERT"`
`17`	`18`	`}`
Original file line number	Diff line number	Diff line change
`@@ -14,7 +14,8 @@`
`14`	`14`	`"suspended": false,`
`15`	`15`	`"table_formats": "ICEBERG",`
`16`	`16`	`"table_path": "s3://dummy-bucket-name-1//10011010/11101100/org-1/env-2/lkc-3/v1/tableId",`
`17`		`- "phase": "RUNNING"`
	`17`	`+ "phase": "RUNNING",`
	`18`	`+ "write_mode": "UPSERT"`
`18`	`19`	`},`
`19`	`20`	`{`
`20`	`21`	`"kafka_cluster": "lkc-123456",`
`@@ -28,6 +29,7 @@`
`28`	`29`	`"suspended": false,`
`29`	`30`	`"table_formats": "DELTA",`
`30`	`31`	`"table_path": "s3://dummy-bucket-name-1//10011010/11101100/org-1/env-2/lkc-3/v1/tableId",`
`31`		`- "phase": "RUNNING"`
	`32`	`+ "phase": "RUNNING",`
	`33`	`+ "write_mode": "APPEND"`
`32`	`34`	`}`
`33`	`35`	`]`