[transfer] Call dedupe on complete (#356)

go.mod

@@ -4,7 +4,7 @@ go 1.22
 
 require (
 	github.com/DataDog/datadog-go v4.8.3+incompatible
-	github.com/artie-labs/transfer v1.22.37
+	github.com/artie-labs/transfer v1.22.38
 	github.com/aws/aws-sdk-go v1.44.327
 	github.com/aws/aws-sdk-go-v2/config v1.18.19
 	github.com/getsentry/sentry-go v0.27.0

go.sum

@@ -91,8 +91,8 @@ github.com/apache/thrift v0.0.0-20181112125854-24918abba929/go.mod h1:cp2SuWMxlE
 github.com/apache/thrift v0.14.2/go.mod h1:cp2SuWMxlEZw2r+iP2GNCdIi4C1qmUzdZFSVb+bacwQ=
 github.com/apache/thrift v0.16.0 h1:qEy6UW60iVOlUy+b9ZR0d5WzUWYGOo4HfopoyBaNmoY=
 github.com/apache/thrift v0.16.0/go.mod h1:PHK3hniurgQaNMZYaCLEqXKsYK8upmhPbmdP2FXSqgU=
-github.com/artie-labs/transfer v1.22.37 h1:NQ1avFmeamLDEoG6we09yo6u+Qd+7UJg/4zLhffbweQ=
-github.com/artie-labs/transfer v1.22.37/go.mod h1:Rgn0drCBpyirqCPZrc7+FKe4y9crhB32mzMdXQtft/g=
+github.com/artie-labs/transfer v1.22.38 h1:vrZ5mwL2MRNxV5TeH+u/SGtd6iheiSx0jhySZVHwEw4=
+github.com/artie-labs/transfer v1.22.38/go.mod h1:Rgn0drCBpyirqCPZrc7+FKe4y9crhB32mzMdXQtft/g=
 github.com/aws/aws-sdk-go v1.30.19/go.mod h1:5zCpMtNQVjRREroY7sYe8lOMRSxkhG6MZveU8YkpAk0=
 github.com/aws/aws-sdk-go v1.44.327 h1:ZS8oO4+7MOBLhkdwIhgtVeDzCeWOlTfKJS7EgggbIEY=
 github.com/aws/aws-sdk-go v1.44.327/go.mod h1:aVsgQcEevwlmQ7qHE9I3h+dtQgpqhFB+i8Phjh7fkwI=

writers/transfer/writer.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"log/slog"
 	"time"
 
 	"github.com/artie-labs/transfer/lib/artie"
@@ -161,6 +162,13 @@ func (w *Writer) OnComplete() error {
 	if err := w.flush("complete"); err != nil {
 		return err
 	}
-	// TODO: Run de-duplicate logic here as long as some amount of messages were written to the destination.
-	return nil
+
+	tableName, tableData, err := w.getTableData()
+	if err != nil {
+		return err
+	}
+
+	fqTableName := w.destination.ToFullyQualifiedName(tableData.TableData, true)
+	slog.Info("Running dedupe...", slog.String("table", tableName), slog.String("fqTable", fqTableName))
+	return w.destination.Dedupe(fqTableName)
 }