         }
     }
     
-    /// Returns an Observable that emits an optional record after each committed
-    /// database transaction that has modified the tables and columns fetched by
-    /// the request.
-    ///
-    /// All records are emitted on *scheduler*, which defaults to
-    /// `MainScheduler.instance`. If you set *startImmediately* to true (the
-    /// default value), the first element is emitted right upon subscription.
-    ///
-    /// - parameter writer: A DatabaseWriter (DatabaseQueue or DatabasePool).
-    /// - parameter startImmediately: When true (the default), the first
-    ///   element is emitted right upon subscription.
-    /// - parameter scheduler: The scheduler on which elements are emitted
-    ///   (default is MainScheduler.instance).
-    public func fetchOne(
-        in writer: DatabaseWriter,
-        startImmediately: Bool = true,
-        scheduler: ImmediateSchedulerType? = nil,
-        distinctUntilChanged: Bool = false)
-        -> Observable<Base.RowDecoder?>
-    {
-        let request = base
-        if distinctUntilChanged {
-            return AnyDatabaseWriter(writer).rx
-                .fetchTokens(in: [request], startImmediately: startImmediately, scheduler: scheduler)
-                .mapFetch { try Row.fetchOne($0, request) }
-                .distinctUntilChanged(==)
-                .map { (row: Row?) in row.map { Base.RowDecoder.init(row: $0) } }
-        } else {
-            return AnyDatabaseWriter(writer).rx
-                .fetchTokens(in: [request], startImmediately: startImmediately, scheduler: scheduler)
-                .mapFetch { try request.fetchOne($0) }
-        }
-    }
-}
-
-// MARK: - Row
-
-extension Reactive where Base: FetchRequest & DatabaseRegionConvertible, Base.RowDecoder: Row {
-    /// Returns an Observable that emits an array of rows after each
-    /// committed database transaction that has modified the tables and columns
-    /// fetched by the request.
-    ///
-    /// All arrays are emitted on *scheduler*, which defaults to
-    /// `MainScheduler.instance`. If you set *startImmediately* to true (the
-    /// default value), the first element is emitted right upon subscription.
-    ///
-    /// - parameter writer: A DatabaseWriter (DatabaseQueue or DatabasePool).
-    /// - parameter startImmediately: When true (the default), the first
-    ///   element is emitted right upon subscription.
-    /// - parameter scheduler: The scheduler on which elements are emitted
-    ///   (default is MainScheduler.instance).
-    public func fetchAll(
-        in writer: DatabaseWriter,
-        startImmediately: Bool = true,
-        scheduler: ImmediateSchedulerType? = nil,
-        distinctUntilChanged: Bool = false)
-        -> Observable<[Row]>
-    {
-        let request = base
-        if distinctUntilChanged {
-            return AnyDatabaseWriter(writer).rx
-                .fetchTokens(in: [request], startImmediately: startImmediately, scheduler: scheduler)
-                .mapFetch { try request.fetchAll($0) }
-                .distinctUntilChanged(==)
-        } else {
-            return AnyDatabaseWriter(writer).rx
-                .fetchTokens(in: [request], startImmediately: startImmediately, scheduler: scheduler)
-                .mapFetch { try request.fetchAll($0) }
-        }
-    }
-    
-    /// Returns an Observable that emits an optional row after each committed
-    /// database transaction that has modified the tables and columns fetched by
-    /// the request.
-    ///
-    /// All rows are emitted on *scheduler*, which defaults to
-    /// `MainScheduler.instance`. If you set *startImmediately* to true (the
-    /// default value), the first element is emitted right upon subscription.
-    ///
-    /// - parameter writer: A DatabaseWriter (DatabaseQueue or DatabasePool).
-    /// - parameter startImmediately: When true (the default), the first
-    ///   element is emitted right upon subscription.
-    /// - parameter scheduler: The scheduler on which elements are emitted
-    ///   (default is MainScheduler.instance).
-    public func fetchOne(
-        in writer: DatabaseWriter,
-        startImmediately: Bool = true,
-        scheduler: ImmediateSchedulerType? = nil,
-        distinctUntilChanged: Bool = false)
-        -> Observable<Row?>
-    {
-        let request = base
-        if distinctUntilChanged {
-            return AnyDatabaseWriter(writer).rx
-                .fetchTokens(in: [request], startImmediately: startImmediately, scheduler: scheduler)
-                .mapFetch { try request.fetchOne($0) }
-                .distinctUntilChanged(==)
-        } else {
-            return AnyDatabaseWriter(writer).rx
-                .fetchTokens(in: [request], startImmediately: startImmediately, scheduler: scheduler)
-                .mapFetch { try request.fetchOne($0) }
-        }
-        
-    }
-}
-
-// MARK: - DatabaseValueConvertible
-
-extension Reactive where Base: FetchRequest & DatabaseRegionConvertible, Base.RowDecoder: DatabaseValueConvertible {
-    /// Returns an Observable that emits an array of values after each
-    /// committed database transaction that has modified the tables and columns
-    /// fetched by the request.
-    ///
-    /// All arrays are emitted on *scheduler*, which defaults to
-    /// `MainScheduler.instance`. If you set *startImmediately* to true (the
-    /// default value), the first element is emitted right upon subscription.
-    ///
-    /// - parameter writer: A DatabaseWriter (DatabaseQueue or DatabasePool).
-    /// - parameter startImmediately: When true (the default), the first
-    ///   element is emitted right upon subscription.
-    /// - parameter scheduler: The scheduler on which elements are emitted
-    ///   (default is MainScheduler.instance).
-    public func fetchAll(
-        in writer: DatabaseWriter,
-        startImmediately: Bool = true,
-        scheduler: ImmediateSchedulerType? = nil,
-        distinctUntilChanged: Bool = false)
-        -> Observable<[Base.RowDecoder]>
-    {
-        let request = base
-        if distinctUntilChanged {
-            return AnyDatabaseWriter(writer).rx
-                .fetchTokens(in: [request], startImmediately: startImmediately, scheduler: scheduler)
-                .mapFetch { try DatabaseValue.fetchAll($0, request) }
-                .distinctUntilChanged(==)
-                .map { (dbValues: [DatabaseValue]) in dbValues.map { $0.losslessConvert() as Base.RowDecoder } }
-        } else {
-            return AnyDatabaseWriter(writer).rx
-                .fetchTokens(in: [request], startImmediately: startImmediately, scheduler: scheduler)
-                .mapFetch { try request.fetchAll($0) }
-        }
-    }
-    
-    /// Returns an Observable that emits an optional value after each committed
-    /// database transaction that has modified the tables and columns fetched by
-    /// the request.
-    ///
-    /// All values are emitted on *scheduler*, which defaults to
-    /// `MainScheduler.instance`. If you set *startImmediately* to true (the
-    /// default value), the first element is emitted right upon subscription.
-    ///
-    /// - parameter writer: A DatabaseWriter (DatabaseQueue or DatabasePool).
-    /// - parameter startImmediately: When true (the default), the first
-    ///   element is emitted right upon subscription.
-    /// - parameter scheduler: The scheduler on which elements are emitted
-    ///   (default is MainScheduler.instance).
+    /// Returns an Observable that emits after each committed database
+    /// transaction that has modified the tables and columns fetched by
+    /// the request.
+    ///
+    ///     let dbQueue = DatabaseQueue()
+    ///     let request = SQLRequest<String>("SELECT name FROM player ORDER BY score DESC")
+    ///     request.rx
+    ///         .fetchOne(in: dbQueue)
+    ///         .subscribe(onNext: { bestPlayerName: String? in
+    ///             print("Best player has changed")
+    ///         })
+    ///
+    /// By default, all values are emitted on the main dispatch queue. If you
+    /// give a *scheduler*, values are emitted on that scheduler.
+    ///
+    /// If you set *startImmediately* to true (the default value), the first
+    /// element is emitted right upon subscription. It is *synchronously*
+    /// emitted if and only if the observable is subscribed on the main queue,
+    /// and is given a nil *scheduler* argument:
+    ///
+    ///     // on the main queue
+    ///     request.rx
+    ///         .fetchOne(in: dbQueue)
+    ///         .subscribe(onNext: { bestPlayerName: String? in
+    ///             // on the main queue
+    ///             print("Best player has changed")
+    ///         })
+    ///     // <- here "Best player has changed" has been printed
+    ///
+    ///     // on any queue
+    ///     request.rx
+    ///         .fetchOne(in: dbQueue, scheduler: MainScheduler.instance)
+    ///         .subscribe(onNext: { bestPlayerName: String? in
+    ///             // on the main queue
+    ///             print("Best player has changed")
+    ///         })
+    ///     // <- here "Best player has changed" may not be printed yet
+    ///
+    /// - parameter writer: A DatabaseWriter (DatabaseQueue or DatabasePool).
+    /// - parameter startImmediately: When true (the default), the first
+    ///   element is emitted right upon subscription.
+    /// - parameter scheduler: The eventual scheduler on which elements
+    ///   are emitted.
     public func fetchOne(
         in writer: DatabaseWriter,
         startImmediately: Bool = true,

#endif
import RxSwift

// MARK: - FetchableRecord

// TODO: consider performing distinctUntilChanged comparisons in some background queue
extension Reactive where Base: FetchRequest & DatabaseRegionConvertible, Base.RowDecoder: FetchableRecord {
    /// Returns an Observable that emits after each committed database
    /// transaction that has modified the tables and columns fetched by
    /// the request.
    ///
    /// All values are emitted on *scheduler*, which defaults to
    /// `MainScheduler.instance`. If you set *startImmediately* to true (the
    /// default value), the first element is emitted right upon subscription.
    ///
    ///     let dbQueue = DatabaseQueue()
    ///     let request = Person.all()
    ///     request.rx
    ///         .fetchCount(in: dbQueue)
    ///         .subscribe(onNext: { count in
    ///             print("Number of persons: \(count)")
    ///         })
    ///     // Prints "Number of persons: 0"
    ///
    ///     try dbQueue.inDatabase { db in
    ///         try Person(name: "Arthur").insert(db)
    ///         // Prints "Number of persons: 1"
    ///         try Person(name: "Barbara").insert(db)
    ///         // Prints "Number of persons: 2"
    ///     }
    ///
    ///     try dbQueue.inTransaction { db in
    ///         try Person(name: "Craig").insert(db)
    ///         try Person(name: "David").insert(db)
    ///         return .commit
    ///     }
    ///     // Prints "Number of persons: 4"
    ///
    /// - parameter writer: A DatabaseWriter (DatabaseQueue or DatabasePool).
    /// - parameter startImmediately: When true (the default), the first
    ///   element is emitted right upon subscription.
    /// - parameter scheduler: The scheduler on which elements are emitted
    ///   (default is MainScheduler.instance).
    public func fetchCount(
        in writer: DatabaseWriter,
        startImmediately: Bool = true,

}

extension Reactive where Base: FetchRequest & DatabaseRegionConvertible, Base.RowDecoder: _OptionalProtocol, Base.RowDecoder._Wrapped: DatabaseValueConvertible {
    /// Returns an Observable that emits after each committed database
    /// transaction that has modified the tables and columns fetched by
    /// the request.
    ///
    ///     let dbQueue = DatabaseQueue()
    ///     let request = Player.select(Column("name")).asRequest(of: Optional<String>.self)
    ///     request.rx
    ///         .fetchAll(in: dbQueue)
    ///         .subscribe(onNext: { names: [String?] in
    ///             print("Player names have changed")
    ///         })
    ///
    /// By default, all values are emitted on the main dispatch queue. If you
    /// give a *scheduler*, values are emitted on that scheduler.
    ///
    /// If you set *startImmediately* to true (the default value), the first
    /// element is emitted right upon subscription. It is *synchronously*
    /// emitted if and only if the observable is subscribed on the main queue,
    /// and is given a nil *scheduler* argument:
    ///
    ///     // on the main queue
    ///     request.rx
    ///         .fetchAll(in: dbQueue)
    ///         .subscribe(onNext: { names: [String?] in
    ///             // on the main queue
    ///             print("Player names have changed")
    ///         })
    ///     // <- here "Player names have changed" has been printed
    ///
    ///     // on any queue
    ///     request.rx
    ///         .fetchAll(in: dbQueue, scheduler: MainScheduler.instance)
    ///         .subscribe(onNext: { names: [String?] in
    ///             // on the main queue
    ///             print("Player names have changed")
    ///         })
    ///     // <- here "Player names have changed" may not be printed yet
    ///
    /// - parameter writer: A DatabaseWriter (DatabaseQueue or DatabasePool).
    /// - parameter startImmediately: When true (the default), the first
    ///   element is emitted right upon subscription.
    /// - parameter scheduler: The eventual scheduler on which elements
    ///   are emitted.
    public func fetchAll(
        in writer: DatabaseWriter,
        startImmediately: Bool = true,
