fix: Arrow Retain and Release fixes (#795)

This fixes a memory leak detected by the `file` and `s3` destination
tests, and refactors the retain/release logic a little bit.

Quoting the Go Arrow docs:

> If you send an object over a channel, you must call Retain before
sending it as the receiver is assumed to own the object and will later
call Release when it no longer needs the object.

I think we should call `Retain` as close as possible to where a record
is sent over a channel.

Author: hermanschaaf

internal/servers/destination/v0/destinations.go

@@ -141,6 +141,7 @@ func (s *Server) Write2(msg pb.Destination_Write2Server) error {
 		select {
 		case resources <- convertedResource:
 		case <-ctx.Done():
+			convertedResource.Release()
 			close(resources)
 			if err := eg.Wait(); err != nil {
 				return status.Errorf(codes.Internal, "Context done: %v and failed to wait for plugin: %v", ctx.Err(), err)

plugins/destination/managed_writer.go

@@ -70,6 +70,9 @@ func (p *Plugin) flush(ctx context.Context, metrics *Metrics, table *arrow.Schem
 		p.logger.Info().Str("table", tableName).Int("len", batchSize).Dur("duration", time.Since(start)).Msg("batch written successfully")
 		atomic.AddUint64(&metrics.Writes, uint64(batchSize))
 	}
+	for _, r := range resources {
+		r.Release()
+	}
 }
 
 func (*Plugin) removeDuplicatesByPK(table *arrow.Schema, resources []arrow.Record) []arrow.Record {
@@ -81,18 +84,19 @@ func (*Plugin) removeDuplicatesByPK(table *arrow.Schema, resources []arrow.Recor
 
 	pks := make(map[string]struct{}, len(resources))
 	res := make([]arrow.Record, 0, len(resources))
-	var reported bool
 	for _, r := range resources {
+		if r.NumRows() > 1 {
+			panic(fmt.Sprintf("record with more than 1 row: %d", r.NumRows()))
+		}
 		key := pk.String(r)
 		_, ok := pks[key]
-		switch {
-		case !ok:
+		if !ok {
 			pks[key] = struct{}{}
 			res = append(res, r)
 			continue
-		case reported:
-			continue
 		}
+		// duplicate, release early
+		r.Release()
 	}
 
 	return res