fix: Update to plugin-pb v1.5.0 (#1026)

Updated to cloudquery/plugin-pb-go#46

Author: yevgenypats

go.mod

@@ -5,7 +5,7 @@ go 1.19
 require (
 	github.com/apache/arrow/go/v13 v13.0.0-20230622042343-ec413b7763fe
 	github.com/bradleyjkemp/cupaloy/v2 v2.8.0
-	github.com/cloudquery/plugin-pb-go v1.4.0
+	github.com/cloudquery/plugin-pb-go v1.5.0
 	github.com/cloudquery/plugin-sdk/v2 v2.7.0
 	github.com/getsentry/sentry-go v0.20.0
 	github.com/goccy/go-json v0.10.0

go.sum

@@ -42,8 +42,8 @@ github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMn
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
 github.com/cloudquery/arrow/go/v13 v13.0.0-20230626001500-065602842c3a h1:O/FNq1+8YlWzHYNj2tokFQyja6GXsQBdkuvLMdpuaSw=
 github.com/cloudquery/arrow/go/v13 v13.0.0-20230626001500-065602842c3a/go.mod h1:W69eByFNO0ZR30q1/7Sr9d83zcVZmF2MiP3fFYAWJOc=
-github.com/cloudquery/plugin-pb-go v1.4.0 h1:sfy0oWSFac2JCJQJuKoR+8flZGKkEoUVORwZDNM3aiI=
-github.com/cloudquery/plugin-pb-go v1.4.0/go.mod h1:NbWAtT2BzJQ9+XUWwh3IKBg3MOeV9ZEpHoHNAQ/YDV8=
+github.com/cloudquery/plugin-pb-go v1.5.0 h1:A/RE1U1l34W5T+JlXJzrHz0IMzfpdUK4VSg+J1Hw0gw=
+github.com/cloudquery/plugin-pb-go v1.5.0/go.mod h1:NbWAtT2BzJQ9+XUWwh3IKBg3MOeV9ZEpHoHNAQ/YDV8=
 github.com/cloudquery/plugin-sdk/v2 v2.7.0 h1:hRXsdEiaOxJtsn/wZMFQC9/jPfU1MeMK3KF+gPGqm7U=
 github.com/cloudquery/plugin-sdk/v2 v2.7.0/go.mod h1:pAX6ojIW99b/Vg4CkhnsGkRIzNaVEceYMR+Bdit73ug=
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=

internal/memdb/memdb.go

@@ -191,7 +191,7 @@ func (c *client) Close(context.Context) error {
 
 func (c *client) deleteStale(_ context.Context, msg *message.WriteDeleteStale) {
 	var filteredTable []arrow.Record
-	tableName := msg.Table.Name
+	tableName := msg.TableName
 	for i, row := range c.memoryDB[tableName] {
 		sc := row.Schema()
 		indices := sc.FieldIndices(schema.CqSourceNameColumn.Name)

internal/servers/destination/v0/destinations.go

@@ -243,7 +243,7 @@ func (s *Server) DeleteStale(ctx context.Context, req *pb.DeleteStale_Request) (
 		bldr.Field(table.Columns.Index(schema.CqSourceNameColumn.Name)).(*array.StringBuilder).Append(req.Source)
 		bldr.Field(table.Columns.Index(schema.CqSyncTimeColumn.Name)).(*array.TimestampBuilder).AppendTime(req.Timestamp.AsTime())
 		msgs <- &message.WriteDeleteStale{
-			Table:      table,
+			TableName:  table.Name,
 			SourceName: req.Source,
 			SyncTime:   req.Timestamp.AsTime(),
 		}

internal/servers/destination/v1/destinations.go

@@ -226,7 +226,7 @@ func (s *Server) DeleteStale(ctx context.Context, req *pb.DeleteStale_Request) (
 		bldr.Field(table.Columns.Index(schema.CqSourceNameColumn.Name)).(*array.StringBuilder).Append(req.Source)
 		bldr.Field(table.Columns.Index(schema.CqSyncTimeColumn.Name)).(*array.TimestampBuilder).AppendTime(req.Timestamp.AsTime())
 		msgs <- &message.WriteDeleteStale{
-			Table:      table,
+			TableName:  table.Name,
 			SourceName: req.Source,
 			SyncTime:   req.Timestamp.AsTime(),
 		}

internal/servers/plugin/v3/plugin.go

@@ -176,18 +176,8 @@ func (s *Server) Write(msg pb.Plugin_WriteServer) error {
 				Record: record,
 			}
 		case *pb.Write_Request_Delete:
-			sc, err := pb.NewSchemaFromBytes(pbMsg.Delete.Table)
-			if err != nil {
-				pbMsgConvertErr = status.Errorf(codes.InvalidArgument, "failed to create schema from bytes: %v", err)
-				break
-			}
-			table, err := schema.NewTableFromArrowSchema(sc)
-			if err != nil {
-				pbMsgConvertErr = status.Errorf(codes.InvalidArgument, "failed to create table from schema: %v", err)
-				break
-			}
 			pluginMessage = &message.WriteDeleteStale{
-				Table:      table,
+				TableName:  pbMsg.Delete.TableName,
 				SourceName: pbMsg.Delete.SourceName,
 				SyncTime:   pbMsg.Delete.SyncTime.AsTime(),
 			}

message/write_message.go

@@ -95,19 +95,19 @@ func (m WriteInserts) GetRecordsForTable(table *schema.Table) []arrow.Record {
 // The message indicates that the destination needs to run something like "DELETE FROM table WHERE _cq_source_name=$1 and sync_time < $2"
 type WriteDeleteStale struct {
 	writeBaseMessage
-	Table      *schema.Table
+	TableName  string
 	SourceName string
 	SyncTime   time.Time
 }
 
 func (m WriteDeleteStale) GetTable() *schema.Table {
-	return m.Table
+	return &schema.Table{Name: m.TableName}
 }
 
 type WriteDeleteStales []*WriteDeleteStale
 
 func (m WriteDeleteStales) Exists(tableName string) bool {
 	return slices.ContainsFunc(m, func(msg *WriteDeleteStale) bool {
-		return msg.Table.Name == tableName
+		return msg.TableName == tableName
 	})
 }

plugin/testing_write_delete.go

@@ -53,7 +53,7 @@ func (s *WriterTestSuite) testDeleteStale(ctx context.Context) error {
 	bldr.Field(1).(*array.TimestampBuilder).AppendTime(syncTime.Add(time.Second))
 
 	if err := s.plugin.writeOne(ctx, &message.WriteDeleteStale{
-		Table:      table,
+		TableName:  table.Name,
 		SourceName: "test",
 		SyncTime:   syncTime,
 	}); err != nil {

writers/batchwriter/batchwriter.go

@@ -251,7 +251,7 @@ func (w *BatchWriter) Write(ctx context.Context, msgs <-chan message.WriteMessag
 			if err := w.flushMigrateTables(ctx); err != nil {
 				return err
 			}
-			w.flushInsert(m.Table.Name)
+			w.flushInsert(m.TableName)
 			w.deleteStaleLock.Lock()
 			w.deleteStaleMessages = append(w.deleteStaleMessages, m)
 			l := len(w.deleteStaleMessages)

writers/mixedbatchwriter/mixedbatchwriter_test.go

@@ -95,12 +95,12 @@ func TestMixedBatchWriter(t *testing.T) {
 
 	// message to delete stale from table1
 	msgDeleteStale1 := &message.WriteDeleteStale{
-		Table:      table1,
+		TableName:  table1.Name,
 		SourceName: "my-source",
 		SyncTime:   time.Now(),
 	}
 	msgDeleteStale2 := &message.WriteDeleteStale{
-		Table:      table1,
+		TableName:  table1.Name,
 		SourceName: "my-source",
 		SyncTime:   time.Now(),
 	}