Checkpoint.

artie-labs · Jun 17, 2024 · 8a7d7e9 · 8a7d7e9
1 parent 8eeff08
commit 8a7d7e9
Show file tree

Hide file tree

Showing 4 changed files with 13 additions and 2 deletions.
diff --git a/clients/mssql/dialect/dialect.go b/clients/mssql/dialect/dialect.go
@@ -169,6 +169,10 @@ func (md MSSQLDialect) BuildIsNotToastValueExpression(tableAlias constants.Table
 	return fmt.Sprintf("COALESCE(%s, '') != '%s'", colName, constants.ToastUnavailableValuePlaceholder)
 }
 
+func (MSSQLDialect) GetDedupeTableQuery(tableID sql.TableIdentifier, primaryKeys []string) string {
+	panic("not implemented")
+}
+
 func (MSSQLDialect) BuildDedupeQueries(tableID, stagingTableID sql.TableIdentifier, primaryKeys []string, topicConfig kafkalib.TopicConfig) []string {
 	panic("not implemented") // We don't currently support deduping for MS SQL.
 }

diff --git a/clients/redshift/dialect/dialect.go b/clients/redshift/dialect/dialect.go
@@ -135,6 +135,10 @@ func (rd RedshiftDialect) BuildIsNotToastValueExpression(tableAlias constants.Ta
 	return fmt.Sprintf("COALESCE(%s != '%s', true)", colName, constants.ToastUnavailableValuePlaceholder)
 }
 
+func (rd RedshiftDialect) GetDedupeTableQuery(tableID sql.TableIdentifier, _ []string) string {
+	return fmt.Sprintf(`( SELECT DISTINCT * FROM %s )`, tableID.FullyQualifiedName())
+}
+
 func (rd RedshiftDialect) BuildDedupeQueries(tableID, stagingTableID sql.TableIdentifier, primaryKeys []string, topicConfig kafkalib.TopicConfig) []string {
 	primaryKeysEscaped := sql.QuoteIdentifiers(primaryKeys, rd)
 

diff --git a/clients/shared/merge.go b/clients/shared/merge.go
@@ -111,8 +111,7 @@ func Merge(dwh destination.DataWarehouse, tableData *optimization.TableData, opt
 		}
 	}
 
-	temporaryTableName := temporaryTableID.FullyQualifiedName()
-	subQuery := temporaryTableName
+	subQuery := temporaryTableID.FullyQualifiedName()
 	if opts.SubQueryDedupe {
 		subQuery = dwh.Dialect().GetDedupeTableQuery(temporaryTableID, tableData.PrimaryKeys())
 	}

diff --git a/clients/snowflake/dialect/dialect.go b/clients/snowflake/dialect/dialect.go
@@ -153,6 +153,10 @@ func (sd SnowflakeDialect) BuildIsNotToastValueExpression(tableAlias constants.T
 	return fmt.Sprintf("COALESCE(%s != '%s', true)", colName, constants.ToastUnavailableValuePlaceholder)
 }
 
+func (SnowflakeDialect) GetDedupeTableQuery(tableID sql.TableIdentifier, primaryKeys []string) string {
+	panic("not implemented")
+}
+
 func (sd SnowflakeDialect) BuildDedupeQueries(tableID, stagingTableID sql.TableIdentifier, primaryKeys []string, topicConfig kafkalib.TopicConfig) []string {
 	primaryKeysEscaped := sql.QuoteIdentifiers(primaryKeys, sd)