feat: Enable storing _cq_client_id. (#2046)

We need to be able to store `_cq_client_id` on every table for certain
use cases where we need to know where a row came from in terms of
multiplexed clients.

This PR adds the ability for the SDK and all its schedulers to store
`_cq_client_id` on all tables.

I added this feature to the example plugin in the SDK and it does store
it properly:

<img width="861" alt="Screenshot 2025-01-08 at 10 45 04"
src="https://github.com/user-attachments/assets/58705cf2-3e61-4585-bfbc-63abbccef555"
/>

If you want to try this yourself, please note that, very unfortunately,
the simple plugin also stores a `client_id` column that is unrelated
(and doesn't work, since it thinks it's supposed to be an integer).

Author: marianogappa

examples/simple_plugin/plugin/client.go

@@ -123,6 +123,7 @@ func getTables() schema.Tables {
 	}
 	for _, t := range tables {
 		schema.AddCqIDs(t)
+		schema.AddCqClientID(t)
 	}
 	return tables
 }

scheduler/queue/worker.go

@@ -147,6 +147,9 @@ func (w *worker) resolveResource(ctx context.Context, table *schema.Table, clien
 					atomic.AddUint64(&tableMetrics.Errors, 1)
 					return
 				}
+				if err := resolvedResource.StoreCQClientID(client.ID()); err != nil {
+					w.logger.Error().Err(err).Str("table", table.Name).Str("client", client.ID()).Msg("failed to store _cq_client_id")
+				}
 				if err := resolvedResource.Validate(); err != nil {
 					switch err.(type) {
 					case *schema.PKError:

scheduler/scheduler_dfs.go

@@ -188,6 +188,9 @@ func (s *syncClient) resolveResourcesDfs(ctx context.Context, table *schema.Tabl
 					atomic.AddUint64(&tableMetrics.Errors, 1)
 					return
 				}
+				if err := resolvedResource.StoreCQClientID(client.ID()); err != nil {
+					s.logger.Error().Err(err).Str("table", table.Name).Str("client", client.ID()).Msg("failed to store _cq_client_id")
+				}
 				if err := resolvedResource.Validate(); err != nil {
 					switch err.(type) {
 					case *schema.PKError:

schema/meta.go

@@ -29,6 +29,13 @@ var CqParentIDColumn = Column{
 	IgnoreInTests: true,
 }
 
+var CqClientIDColumn = Column{
+	Name:        "_cq_client_id",
+	Type:        arrow.BinaryTypes.String,
+	Description: "Internal CQ ID of the multiplexed client",
+	NotNull:     true,
+}
+
 // These columns are managed and populated by the destination plugin.
 var CqSyncTimeColumn = Column{
 	Name:        "_cq_sync_time",

schema/resource.go

@@ -123,6 +123,14 @@ func (r *Resource) storeCQID(value uuid.UUID) error {
 	return r.Set(CqIDColumn.Name, b)
 }
 
+func (r *Resource) StoreCQClientID(clientID string) error {
+	// We skip if _cq_client_id is not present.
+	if r.Table.Columns.Get(CqClientIDColumn.Name) == nil {
+		return nil
+	}
+	return r.Set(CqClientIDColumn.Name, clientID)
+}
+
 type PKError struct {
 	MissingPKs []string
 }

schema/table.go

@@ -127,6 +127,17 @@ func AddCqIDs(table *Table) {
 	}
 }
 
+// AddCqClientID adds the cq_client_id column to the table,
+// which is used to identify the multiplexed client that fetched the resource
+func AddCqClientID(t *Table) {
+	if t.Columns.Get(CqClientIDColumn.Name) == nil {
+		t.Columns = append(ColumnList{CqClientIDColumn}, t.Columns...)
+	}
+	for _, rel := range t.Relations {
+		AddCqClientID(rel)
+	}
+}
+
 // CqIDAsPK sets the cq_id column as primary key if it exists
 // and removes the primary key from all other columns
 func CqIDAsPK(t *Table) {