chore: sync release v1.48.4 to main branch (#5812)

# Description

Syncing patch release v1.48.4 to main branch

**↓↓ Please review and edit commit overrides before merging ↓↓**

BEGIN_COMMIT_OVERRIDE
fix: warehouse cached schema mismatch (#5805)
END_COMMIT_OVERRIDE

---------

Co-authored-by: Aris Tzoumas <[email protected]>
Co-authored-by: Sudip Paul <[email protected]>
Co-authored-by: yeshwanth vuppu <[email protected]>

Author: 4 people

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## [1.48.4](https://github.com/rudderlabs/rudder-server/compare/v1.48.3...v1.48.4) (2025-05-05)
+
+
+### Bug Fixes
+
+* warehouse cached schema mismatch ([#5805](https://github.com/rudderlabs/rudder-server/issues/5805)) ([1d3abfd](https://github.com/rudderlabs/rudder-server/commit/1d3abfd961c12bc4a31c97f96b67f049d4a9cdde))
+
 ## [1.48.3](https://github.com/rudderlabs/rudder-server/compare/v1.48.2...v1.48.3) (2025-05-02)
 
 

warehouse/internal/repo/schema.go

@@ -53,36 +53,67 @@ func (sh *WHSchema) Insert(ctx context.Context, whSchema *model.WHSchema) (int64
 		return id, fmt.Errorf("marshaling schema: %w", err)
 	}
 
-	err = sh.db.QueryRowContext(ctx, `
-		INSERT INTO `+whSchemaTableName+` (
-          source_id, namespace, destination_id,
-		  destination_type, schema, created_at,
-		  updated_at, expires_at
+	err = (*repo)(sh).WithTx(ctx, func(tx *sqlmiddleware.Tx) error {
+		// update all schemas with the same destination_id and namespace but different source_id
+		// this is to ensure all the connections for a destination have the same schema copy
+		_, err = tx.ExecContext(ctx, `
+			UPDATE `+whSchemaTableName+`
+			SET
+				schema = $1,
+				updated_at = $2,
+				expires_at = $3
+			WHERE
+				destination_id = $4 AND
+				namespace = $5 AND
+				source_id != $6;
+		`,
+			schemaPayload,
+			now.UTC(),
+			whSchema.ExpiresAt,
+			whSchema.DestinationID,
+			whSchema.Namespace,
+			whSchema.SourceID,
 		)
-		VALUES
-		  ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT (
-			source_id, destination_id, namespace
-		  ) DO
-		UPDATE
-		SET
-		  schema = $5,
-		  updated_at = $7,
-		  expires_at = $8 RETURNING id;
-`,
-		whSchema.SourceID,
-		whSchema.Namespace,
-		whSchema.DestinationID,
-		whSchema.DestinationType,
-		schemaPayload,
-		now.UTC(),
-		now.UTC(),
-		whSchema.ExpiresAt,
-	).Scan(&id)
-	if err != nil {
-		return id, fmt.Errorf("inserting schema: %w", err)
-	}
+		if err != nil {
+			return fmt.Errorf("updating related schemas: %w", err)
+		}
+
+		// Then, insert/update the new schema using the unique constraint
+		err = tx.QueryRowContext(ctx, `
+			INSERT INTO `+whSchemaTableName+` (
+			  source_id, namespace, destination_id,
+			  destination_type, schema, created_at,
+			  updated_at, expires_at
+			)
+			VALUES
+			  ($1, $2, $3, $4, $5, $6, $7, $8) 
+			ON CONFLICT (
+				source_id, destination_id, namespace
+			) DO
+			UPDATE
+			SET
+			  schema = $5,
+			  updated_at = $7,
+			  expires_at = $8 
+			RETURNING id;
+		`,
+			whSchema.SourceID,
+			whSchema.Namespace,
+			whSchema.DestinationID,
+			whSchema.DestinationType,
+			schemaPayload,
+			now.UTC(),
+			now.UTC(),
+			whSchema.ExpiresAt,
+		).Scan(&id)
+		if err != nil {
+			return fmt.Errorf("inserting schema: %w", err)
+		}
+
+		return nil
+	})
 
-	return id, nil
+	return id, err
 }
 
 func (sh *WHSchema) GetForNamespace(ctx context.Context, sourceID, destID, namespace string) (model.WHSchema, error) {

warehouse/internal/repo/schema_test.go

@@ -178,4 +178,49 @@ func TestWHSchemasRepo(t *testing.T) {
 		require.NoError(t, err)
 		require.Equal(t, expiryTime, updatedSchema.ExpiresAt)
 	})
+
+	t.Run("Insert schema propagation to all connections with same destination_id and namespace", func(t *testing.T) {
+		// Create first connection schema
+		firstConnectionSchema := schema
+		firstID, err := r.Insert(ctx, &firstConnectionSchema)
+		require.NoError(t, err)
+		firstConnectionSchema.ID = firstID
+
+		// Create second connection schema with same destination_id and namespace
+		secondConnectionSchema := firstConnectionSchema
+		secondConnectionSchema.SourceID = "other_source_id"
+		secondConnectionSchema.ID = 0 // Reset ID for new insert
+		secondID, err := r.Insert(ctx, &secondConnectionSchema)
+		require.NoError(t, err)
+		secondConnectionSchema.ID = secondID
+
+		// Verify both connections have the same initial schema
+		firstRetrieved, err := r.GetForNamespace(ctx, firstConnectionSchema.SourceID, firstConnectionSchema.DestinationID, firstConnectionSchema.Namespace)
+		require.NoError(t, err)
+		require.Equal(t, firstConnectionSchema.Schema, firstRetrieved.Schema)
+
+		secondRetrieved, err := r.GetForNamespace(ctx, secondConnectionSchema.SourceID, secondConnectionSchema.DestinationID, secondConnectionSchema.Namespace)
+		require.NoError(t, err)
+		require.Equal(t, firstConnectionSchema.Schema, secondRetrieved.Schema)
+
+		// Update the first connection with new schema data
+		updatedSchema := firstConnectionSchema
+		updatedSchema.Schema = model.Schema{
+			"new_table": {
+				"new_column": "string",
+			},
+		}
+		updatedSchema.ID = 0 // Reset ID for new insert
+		_, err = r.Insert(ctx, &updatedSchema)
+		require.NoError(t, err)
+
+		// Verify both connections are updated with the new schema
+		firstRetrieved, err = r.GetForNamespace(ctx, firstConnectionSchema.SourceID, firstConnectionSchema.DestinationID, firstConnectionSchema.Namespace)
+		require.NoError(t, err)
+		require.Equal(t, updatedSchema.Schema, firstRetrieved.Schema)
+
+		secondRetrieved, err = r.GetForNamespace(ctx, secondConnectionSchema.SourceID, secondConnectionSchema.DestinationID, secondConnectionSchema.Namespace)
+		require.NoError(t, err)
+		require.Equal(t, updatedSchema.Schema, secondRetrieved.Schema)
+	})
 }