chore: cloud extract should always merge (#3800)

* chore: cloud extract should always merge

* chore: review comments

warehouse/integrations/snowflake/snowflake.go

@@ -59,6 +59,11 @@ var partitionKeyMap = map[string]string{
 	discardsTable:   `"ROW_ID", "COLUMN_NAME", "TABLE_NAME"`,
 }
 
+var mergeSourceCategoryMap = map[string]struct{}{
+	"cloud":           {},
+	"singer-protocol": {},
+}
+
 var (
 	usersTable              = whutils.ToProviderCase(whutils.SNOWFLAKE, whutils.UsersTable)
 	identifiesTable         = whutils.ToProviderCase(whutils.SNOWFLAKE, whutils.IdentifiesTable)
@@ -372,7 +377,7 @@ func (sf *Snowflake) loadTable(ctx context.Context, tableName string, tableSchem
 
 	// Truncating the columns by default to avoid size limitation errors
 	// https://docs.snowflake.com/en/sql-reference/sql/copy-into-table.html#copy-options-copyoptions
-	if sf.config.loadTableStrategy == loadTableStrategyAppendMode {
+	if sf.ShouldAppend() {
 		err = sf.copyInto(ctx, db, schemaIdentifier, tableName, sortedColumnNames, tableName, log)
 		if err != nil {
 			return tableLoadResp{}, err
@@ -815,6 +820,14 @@ func (sf *Snowflake) LoadIdentityMappingsTable(ctx context.Context) error {
 	return nil
 }
 
+// ShouldAppend returns true if the load table strategy is append mode and the source category is not in "mergeSourceCategoryMap"
+func (sf *Snowflake) ShouldAppend() bool {
+	sourceCategory := sf.Warehouse.Source.SourceDefinition.Category
+	_, isMergeCategory := mergeSourceCategoryMap[sourceCategory]
+
+	return !isMergeCategory && sf.config.loadTableStrategy == loadTableStrategyAppendMode
+}
+
 func (sf *Snowflake) LoadUserTables(ctx context.Context) map[string]error {
 	var (
 		identifiesSchema = sf.Uploader.GetTableSchemaInUpload(identifiesTable)
@@ -854,7 +867,7 @@ func (sf *Snowflake) LoadUserTables(ctx context.Context) map[string]error {
 	}
 
 	schemaIdentifier := sf.schemaIdentifier()
-	if sf.config.loadTableStrategy == loadTableStrategyAppendMode {
+	if sf.ShouldAppend() {
 		tmpIdentifiesStagingTable := whutils.StagingTableName(provider, identifiesTable, tableNameLimit)
 		sqlStatement := fmt.Sprintf(
 			`CREATE TEMPORARY TABLE %[1]s.%[2]q LIKE %[1]s.%[3]q;`,

warehouse/integrations/snowflake/snowflake_test.go

@@ -12,6 +12,12 @@ import (
 	"testing"
 	"time"
 
+	"github.com/rudderlabs/rudder-go-kit/config"
+	"github.com/rudderlabs/rudder-go-kit/logger"
+	"github.com/rudderlabs/rudder-go-kit/stats"
+	"github.com/rudderlabs/rudder-server/warehouse/integrations/snowflake"
+	"github.com/rudderlabs/rudder-server/warehouse/internal/model"
+
 	sfdb "github.com/snowflakedb/gosnowflake"
 	"github.com/stretchr/testify/require"
 
@@ -491,6 +497,72 @@ func TestIntegration(t *testing.T) {
 	})
 }
 
+func TestSnowflake_ShouldAppend(t *testing.T) {
+	testCases := []struct {
+		name              string
+		loadTableStrategy string
+		sourceCategory    string
+		expected          bool
+	}{
+		{
+			name:              "merge with event stream",
+			loadTableStrategy: "MERGE",
+			sourceCategory:    "event-stream",
+			expected:          false,
+		},
+		{
+			name:              "append with event-stream",
+			loadTableStrategy: "MERGE",
+			sourceCategory:    "event-stream",
+			expected:          false,
+		},
+		{
+			name:              "merge with extract cloud source",
+			loadTableStrategy: "MERGE",
+			sourceCategory:    "cloud",
+			expected:          false,
+		},
+		{
+			name:              "merge with extract singer protocol source",
+			loadTableStrategy: "MERGE",
+			sourceCategory:    "singer-protocol",
+			expected:          false,
+		},
+		{
+			name:              "append with extract cloud source",
+			loadTableStrategy: "APPEND",
+			sourceCategory:    "cloud",
+			expected:          false,
+		},
+		{
+			name:              "append with extract singer protocol source",
+			loadTableStrategy: "APPEND",
+			sourceCategory:    "singer-protocol",
+			expected:          false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			c := config.New()
+			c.Set("Warehouse.snowflake.loadTableStrategy", tc.loadTableStrategy)
+
+			sf, err := snowflake.New(c, logger.NOP, stats.Default)
+			require.NoError(t, err)
+
+			sf.Warehouse = model.Warehouse{
+				Source: backendconfig.SourceT{
+					SourceDefinition: backendconfig.SourceDefinitionT{
+						Category: tc.sourceCategory,
+					},
+				},
+			}
+
+			require.Equal(t, sf.ShouldAppend(), tc.expected)
+		})
+	}
+}
+
 func getSnowflakeDB(t testing.TB, dsn string) *sql.DB {
 	t.Helper()
 	db, err := sql.Open("snowflake", dsn)