diff --git a/packages/graphql/lib/src/cache/cache.dart b/packages/graphql/lib/src/cache/cache.dart
index ff5cf169..f69659fc 100644
--- a/packages/graphql/lib/src/cache/cache.dart
+++ b/packages/graphql/lib/src/cache/cache.dart
@@ -204,9 +204,23 @@ class GraphQLCache extends NormalizingDataProxy {
   ///
   /// This allows for hierarchical optimism that is automatically cleaned up
   /// without having to tightly couple optimistic changes
+  ///
+  /// This is called on every network result as cleanup
   void removeOptimisticPatch(String removeId) {
+    final patchesToRemove = optimisticPatches
+        .where(
+          (patch) =>
+              patch.id == removeId || _parentPatchId(patch.id) == removeId,
+        )
+        .toList();
+
+    if (patchesToRemove.isEmpty) {
+      return;
+    }
+    // Only remove + mark broadcast requested if something was actually removed.
+    // This is to prevent unnecessary rebroadcasts
     optimisticPatches.removeWhere(
-      (patch) => patch.id == removeId || _parentPatchId(patch.id) == removeId,
+      (patch) => patchesToRemove.contains(patch),
     );
     broadcastRequested = true;
   }
diff --git a/packages/graphql/lib/src/core/observable_query.dart b/packages/graphql/lib/src/core/observable_query.dart
index 17d2b7f5..99d0ba61 100644
--- a/packages/graphql/lib/src/core/observable_query.dart
+++ b/packages/graphql/lib/src/core/observable_query.dart
@@ -93,8 +93,19 @@ class ObservableQuery<TParsed> {
   /// call [queryManager.maybeRebroadcastQueries] after all other [_onDataCallbacks]
   ///
   /// Automatically appended as an [OnData]
-  FutureOr<void> _maybeRebroadcast(QueryResult? result) =>
-      queryManager.maybeRebroadcastQueries(exclude: this);
+  FutureOr<void> _maybeRebroadcast(QueryResult? result) {
+    if (_onDataCallbacks.isEmpty &&
+        result?.hasException == true &&
+        result?.data == null) {
+      // We don't need to rebroadcast if there was an exception and there was no
+      // data. It's valid GQL to have data _and_ exception. If options.carryForwardDataOnException
+      // are true, this condition may never get hit.
+      // If there are onDataCallbacks, it's possible they modify cache and are
+      // depending on maybeRebroadcastQueries being called.
+      return false;
+    }
+    return queryManager.maybeRebroadcastQueries(exclude: this);
+  }
 
   /// The most recently seen result from this operation's stream
   QueryResult<TParsed>? latestResult;
@@ -242,12 +253,13 @@ class ObservableQuery<TParsed> {
   }
 
   /// Add a [result] to the [stream] unless it was created
-  /// before [lasestResult].
+  /// before [latestResult].
   ///
   /// Copies the [QueryResult.source] from the [latestResult]
   /// if it is set to `null`.
   ///
-  /// Called internally by the [QueryManager]
+  /// Called internally by the [QueryManager]. Do not call this directly except
+  /// for [QueryResult.loading]
   void addResult(QueryResult<TParsed> result, {bool fromRebroadcast = false}) {
     // don't overwrite results due to some async/optimism issue
     if (latestResult != null &&
diff --git a/packages/graphql/lib/src/core/query_manager.dart b/packages/graphql/lib/src/core/query_manager.dart
index 4b487712..a74578c9 100644
--- a/packages/graphql/lib/src/core/query_manager.dart
+++ b/packages/graphql/lib/src/core/query_manager.dart
@@ -311,14 +311,26 @@ class QueryManager {
       // we attempt to resolve the from the cache
       if (shouldRespondEagerlyFromCache(options.fetchPolicy) &&
           !queryResult.isOptimistic) {
-        final data = cache.readQuery(request, optimistic: false);
-        // we only push an eager query with data
-        if (data != null) {
-          queryResult = QueryResult(
-            options: options,
-            data: data,
+        final latestResult = _getQueryResultByRequest<TParsed>(request);
+        if (latestResult != null && latestResult.data != null) {
+          // we have a result already cached + deserialized for this request
+          // so we reuse it.
+          // latest result won't be for loading, it must contain data
+          queryResult = latestResult.copyWith(
             source: QueryResultSource.cache,
           );
+        } else {
+          // otherwise, we try to find the query in cache (which will require
+          // deserialization)
+          final data = cache.readQuery(request, optimistic: false);
+          // we only push an eager query with data
+          if (data != null) {
+            queryResult = QueryResult(
+              options: options,
+              data: data,
+              source: QueryResultSource.cache,
+            );
+          }
         }
 
         if (options.fetchPolicy == FetchPolicy.cacheOnly &&
@@ -358,6 +370,18 @@ class QueryManager {
     return queryResult;
   }
 
+  /// If a request already has a result associated with it in cache (as
+  /// determined by [ObservableQuery.latestResult]), we can return it without
+  /// needing to denormalize + parse again.
+  QueryResult<TParsed>? _getQueryResultByRequest<TParsed>(Request request) {
+    for (final query in queries.values) {
+      if (query.options.asRequest == request) {
+        return query.latestResult as QueryResult<TParsed>?;
+      }
+    }
+    return null;
+  }
+
   /// Refetch the [ObservableQuery] referenced by [queryId],
   /// overriding any present non-network-only [FetchPolicy].
   Future<QueryResult<TParsed>?> refetchQuery<TParsed>(String queryId) {
@@ -383,11 +407,11 @@ class QueryManager {
     return results;
   }
 
-  ObservableQuery<TParsed>? getQuery<TParsed>(String? queryId) {
-    if (!queries.containsKey(queryId)) {
+  ObservableQuery<TParsed>? getQuery<TParsed>(final String? queryId) {
+    if (!queries.containsKey(queryId) || queryId == null) {
       return null;
     }
-    final query = queries[queryId!];
+    final query = queries[queryId];
     if (query is ObservableQuery<TParsed>) {
       return query;
     }
@@ -402,16 +426,17 @@ class QueryManager {
   void addQueryResult<TParsed>(
     Request request,
     String? queryId,
-    QueryResult<TParsed> queryResult,
-  ) {
+    QueryResult<TParsed> queryResult, {
+    bool fromRebroadcast = false,
+  }) {
     final observableQuery = getQuery<TParsed>(queryId);
 
     if (observableQuery != null && !observableQuery.controller.isClosed) {
-      observableQuery.addResult(queryResult);
+      observableQuery.addResult(queryResult, fromRebroadcast: fromRebroadcast);
     }
   }
 
-  /// Create an optimstic result for the query specified by `queryId`, if it exists
+  /// Create an optimistic result for the query specified by `queryId`, if it exists
   QueryResult<TParsed> _getOptimisticQueryResult<TParsed>(
     Request request, {
     required String queryId,
@@ -463,27 +488,55 @@ class QueryManager {
       return false;
     }
 
-    final shouldBroadast = cache.shouldBroadcast(claimExecution: true);
+    final shouldBroadcast = cache.shouldBroadcast(claimExecution: true);
 
-    if (!shouldBroadast && !force) {
+    if (!shouldBroadcast && !force) {
       return false;
     }
 
-    for (var query in queries.values) {
-      if (query != exclude && query.isRebroadcastSafe) {
+    // If two ObservableQueries are backed by the same [Request], we only need
+    // to [readQuery] for it once.
+    final Map<Request, QueryResult<Object?>> diffQueryResultCache = {};
+    final Map<Request, bool> ignoreQueryResults = {};
+    for (final query in queries.values) {
+      final Request request = query.options.asRequest;
+      final cachedQueryResult = diffQueryResultCache[request];
+      if (query == exclude || !query.isRebroadcastSafe) {
+        continue;
+      }
+      if (cachedQueryResult != null) {
+        // We've already done the diff and denormalized, emit to the observable
+        addQueryResult(
+          request,
+          query.queryId,
+          cachedQueryResult,
+          fromRebroadcast: true,
+        );
+      } else if (ignoreQueryResults.containsKey(request)) {
+        // We've already seen this one and don't need to notify
+        continue;
+      } else {
+        // We haven't seen this one yet, denormalize from cache and diff
         final cachedData = cache.readQuery(
           query.options.asRequest,
           optimistic: query.options.policies.mergeOptimisticData,
         );
         if (_cachedDataHasChangedFor(query, cachedData)) {
-          query.addResult(
-            mapFetchResultToQueryResult(
-              Response(data: cachedData, response: {}),
-              query.options,
-              source: QueryResultSource.cache,
-            ),
+          // The data has changed
+          final queryResult = QueryResult(
+            data: cachedData,
+            options: query.options,
+            source: QueryResultSource.cache,
+          );
+          diffQueryResultCache[request] = queryResult;
+          addQueryResult(
+            request,
+            query.queryId,
+            queryResult,
             fromRebroadcast: true,
           );
+        } else {
+          ignoreQueryResults[request] = true;
         }
       }
     }
diff --git a/packages/graphql/lib/src/core/query_options.dart b/packages/graphql/lib/src/core/query_options.dart
index 529b0621..7a4a43b2 100644
--- a/packages/graphql/lib/src/core/query_options.dart
+++ b/packages/graphql/lib/src/core/query_options.dart
@@ -55,6 +55,37 @@ class QueryOptions<TParsed extends Object?> extends BaseOptions<TParsed> {
         onError,
       ];
 
+  /// Generic copyWith for all fields. There are other, more specific options:
+  /// - [copyWithPolicies] and [withFetchMoreOptions]
+  QueryOptions<TParsed> copyWithOptions({
+    DocumentNode? document,
+    String? operationName,
+    Map<String, dynamic>? variables,
+    FetchPolicy? fetchPolicy,
+    ErrorPolicy? errorPolicy,
+    CacheRereadPolicy? cacheRereadPolicy,
+    Object? optimisticResult,
+    Duration? pollInterval,
+    Context? context,
+    ResultParserFn<TParsed>? parserFn,
+    OnQueryComplete? onComplete,
+    OnQueryError? onError,
+  }) =>
+      QueryOptions<TParsed>(
+        document: document ?? this.document,
+        operationName: operationName ?? this.operationName,
+        variables: variables ?? this.variables,
+        fetchPolicy: fetchPolicy ?? this.fetchPolicy,
+        errorPolicy: errorPolicy ?? this.errorPolicy,
+        cacheRereadPolicy: cacheRereadPolicy ?? this.cacheRereadPolicy,
+        optimisticResult: optimisticResult ?? this.optimisticResult,
+        pollInterval: pollInterval ?? this.pollInterval,
+        context: context ?? this.context,
+        parserFn: parserFn ?? this.parserFn,
+        onComplete: onComplete ?? this.onComplete,
+        onError: onError ?? this.onError,
+      );
+
   QueryOptions<TParsed> withFetchMoreOptions(
     FetchMoreOptions fetchMoreOptions,
   ) =>
@@ -168,10 +199,13 @@ class WatchQueryOptions<TParsed extends Object?> extends QueryOptions<TParsed> {
           parserFn: parserFn,
         );
 
-  /// Whether or not to fetch results
+  /// Whether or not to fetch results every time a new listener is added.
+  /// If [eagerlyFetchResults] is `true`, fetch is triggered during instantiation.
   final bool fetchResults;
 
-  /// Whether to [fetchResults] immediately on instantiation.
+  /// Whether to [fetchResults] immediately on instantiation of [ObservableQuery].
+  /// If available, cache results are emitted when the first listener is added.
+  /// Network results are then emitted when they return to any attached listeners.
   /// Defaults to [fetchResults].
   final bool eagerlyFetchResults;
 
@@ -187,6 +221,40 @@ class WatchQueryOptions<TParsed extends Object?> extends QueryOptions<TParsed> {
         carryForwardDataOnException,
       ];
 
+  /// Generic copyWith for all fields. There are other, more specific options:
+  /// - [copyWithFetchPolicy], [copyWithVariables], etc
+  WatchQueryOptions<TParsed> copyWith({
+    DocumentNode? document,
+    String? operationName,
+    Map<String, dynamic>? variables,
+    FetchPolicy? fetchPolicy,
+    ErrorPolicy? errorPolicy,
+    CacheRereadPolicy? cacheRereadPolicy,
+    Object? optimisticResult,
+    Duration? pollInterval,
+    bool? fetchResults,
+    bool? carryForwardDataOnException,
+    bool? eagerlyFetchResults,
+    Context? context,
+    ResultParserFn<TParsed>? parserFn,
+  }) =>
+      WatchQueryOptions<TParsed>(
+        document: document ?? this.document,
+        operationName: operationName ?? this.operationName,
+        variables: variables ?? this.variables,
+        fetchPolicy: fetchPolicy ?? this.fetchPolicy,
+        errorPolicy: errorPolicy ?? this.errorPolicy,
+        cacheRereadPolicy: cacheRereadPolicy ?? this.cacheRereadPolicy,
+        optimisticResult: optimisticResult ?? this.optimisticResult,
+        pollInterval: pollInterval ?? this.pollInterval,
+        fetchResults: fetchResults ?? this.fetchResults,
+        eagerlyFetchResults: eagerlyFetchResults ?? this.eagerlyFetchResults,
+        carryForwardDataOnException:
+            carryForwardDataOnException ?? this.carryForwardDataOnException,
+        context: context ?? this.context,
+        parserFn: parserFn ?? this.parserFn,
+      );
+
   WatchQueryOptions<TParsed> copyWithFetchPolicy(
     FetchPolicy? fetchPolicy,
   ) =>
diff --git a/packages/graphql/lib/src/core/query_result.dart b/packages/graphql/lib/src/core/query_result.dart
index 5b5b5b38..da4922ca 100644
--- a/packages/graphql/lib/src/core/query_result.dart
+++ b/packages/graphql/lib/src/core/query_result.dart
@@ -48,7 +48,9 @@ class QueryResult<TParsed extends Object?> {
     this.context = const Context(),
     required this.parserFn,
     required this.source,
-  }) : timestamp = DateTime.now() {
+    TParsed? cachedParsedData,
+  })  : timestamp = DateTime.now(),
+        _cachedParsedData = cachedParsedData {
     _data = data;
   }
 
@@ -160,6 +162,23 @@ class QueryResult<TParsed extends Object?> {
     return _cachedParsedData = parserFn(data);
   }
 
+  QueryResult<TParsed> copyWith({
+    Map<String, dynamic>? data,
+    OperationException? exception,
+    Context? context,
+    QueryResultSource? source,
+    TParsed? cachedParsedData,
+  }) {
+    return QueryResult.internal(
+      data: data ?? this.data,
+      exception: exception ?? this.exception,
+      context: context ?? this.context,
+      parserFn: parserFn,
+      source: source ?? this.source,
+      cachedParsedData: cachedParsedData ?? _cachedParsedData,
+    );
+  }
+
   @override
   String toString() => 'QueryResult('
       'source: $source, '