diff --git a/src/client/spark_api/api/Update.cs b/src/client/spark_api/api/Update.cs
index 7f31478f3..0785e0e3a 100644
--- a/src/client/spark_api/api/Update.cs
+++ b/src/client/spark_api/api/Update.cs
@@ -719,45 +719,46 @@ namespace bts.api.core {
 
       // Constructs a new dictionary so not as cheap as Select on the Update<> type
       public static IObservable<TableUpdate<T2, TKey>>
-         Select<T1, T2, TKey>(this IObservable<TableUpdate<T1, TKey>> src, Func<T1, T2> map, Func<T2, TKey> key) where TKey : IEquatable<TKey> {
-         var cache = ImmutableDictionary<TKey, T2>.Empty;
-         return src.Select(update => {
-            var added = update.added.Select(map);
-            var removed = update.removed.Select(map);
-            var changes = update.changes.Select(ch => new Change<T2>(map(ch.Old), map(ch.New)));
-            var builder = cache.ToBuilder();
-            builder.RemoveRange(removed.Select(key));
-            builder.AddRange(added.Select(t => new KeyValuePair<TKey,T2>(key(t), t)));
-            foreach (var (o,n) in changes) {
-               Debug.Assert(key(o).Equals(key(n)));
-               builder[key(o)] = n;
-            }
-            cache = builder.ToImmutable();
+         Select<T1, T2, TKey>(this IObservable<TableUpdate<T1, TKey>> src, Func<T1, T2> map, Func<T2, TKey> key) where TKey : IEquatable<TKey> =>
+         Observable.Create<TableUpdate<T2, TKey>>(observer => {
+            var cache = ImmutableDictionary<TKey, T2>.Empty;
+            return src.Subscribe(update => {
+               var added = update.added.Select(map);
+               var removed = update.removed.Select(map);
+               var changes = update.changes.Select(ch => new Change<T2>(map(ch.Old), map(ch.New)));
+               var builder = cache.ToBuilder();
+               builder.RemoveRange(removed.Select(key));
+               builder.AddRange(added.Select(t => new KeyValuePair<TKey, T2>(key(t), t)));
+               foreach (var (o, n) in changes) {
+                  Debug.Assert(key(o).Equals(key(n)));
+                  builder[key(o)] = n;
+               }
+               cache = builder.ToImmutable();
 
-            return new TableUpdate<T2, TKey>(cache, added, changes, removed, key);
+               observer.OnNext(new TableUpdate<T2, TKey>(cache, added, changes, removed, key));
+            });
          });
-      }
 
       // Constructs a new dictionary so not as cheap as Select on the Update<> type
       public static IObservable<TableUpdate<T2, TKey2>>
-         SelectKeyValue<T1, T2, TKey, TKey2>(this IObservable<TableUpdate<T1, TKey>> src, Func<T1, T2> map, Func<T2, TKey2> key) where TKey : IEquatable<TKey> where TKey2 : IEquatable<TKey2> {
-         var cache = ImmutableDictionary<TKey2, T2>.Empty;
-         return src.Select(update => {
-            var added = update.added.Select(map);
-            var removed = update.removed.Select(map);
-            var changes = update.changes.Select(ch => new Change<T2>(map(ch.Old), map(ch.New)));
-            var builder = cache.ToBuilder();
-            builder.RemoveRange(removed.Select(key));
-            builder.AddRange(added.Select(t => new KeyValuePair<TKey2,T2>(key(t), t)));
-            foreach (var (o,n) in changes) {
-               Debug.Assert(key(o).Equals(key(n)));
-               builder[key(o)] = n;
-            }
-            cache = builder.ToImmutable();
-
-            return new TableUpdate<T2, TKey2>(cache, added, changes, removed, key);
+         SelectKeyValue<T1, T2, TKey, TKey2>(this IObservable<TableUpdate<T1, TKey>> src, Func<T1, T2> map, Func<T2, TKey2> key) where TKey : IEquatable<TKey> where TKey2 : IEquatable<TKey2> =>
+         Observable.Create<TableUpdate<T2, TKey2>>(observer => {
+            var cache = ImmutableDictionary<TKey2, T2>.Empty;
+            return src.Subscribe(update => {
+               var added = update.added.Select(map);
+               var removed = update.removed.Select(map);
+               var changes = update.changes.Select(ch => new Change<T2>(map(ch.Old), map(ch.New)));
+               var builder = cache.ToBuilder();
+               builder.RemoveRange(removed.Select(key));
+               builder.AddRange(added.Select(t => new KeyValuePair<TKey2, T2>(key(t), t)));
+               foreach (var (o, n) in changes) {
+                  Debug.Assert(key(o).Equals(key(n)));
+                  builder[key(o)] = n;
+               }
+               cache = builder.ToImmutable();
+               observer.OnNext(new TableUpdate<T2, TKey2>(cache, added, changes, removed, key));
+            });
          });
-      }
 
       public static IObservable<TableUpdate<(TLeft left, TRight right), TKey>> LeftJoin<TLeft, TRight, TKey>(this IObservable<TableUpdate<TLeft, TKey>> left_source, IObservable<TableUpdate<TRight, TKey>> right_source,
          Func<TLeft, TKey> left_key, Func<TRight, TKey> right_key, TRight right_default = default)