             .fetchTokens(in: [request], startImmediately: startImmediately, scheduler: scheduler)
             .mapFetch { try request.fetchCount($0) }
     }
+}
+
+// MARK: - FetchableRecord
+
+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.
+    ///
+    ///     let dbQueue = DatabaseQueue()
+    ///     let request = Player.all()
+    ///     request.rx
+    ///         .fetchAll(in: dbQueue)
+    ///         .subscribe(onNext: { players: [Player] in
+    ///             print("Players have changed")
+    ///         })
+    ///
+    /// By default, all records are emitted on the main dispatch queue. If you
+    /// give a *scheduler*, records 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: { players: [Player] in
+    ///             // on the main queue
+    ///             print("Players have changed")
+    ///         })
+    ///     // <- here "Players have changed" has been printed
+    ///
+    ///     // on any queue
+    ///     request.rx
+    ///         .fetchAll(in: dbQueue, scheduler: MainScheduler.instance)
+    ///         .subscribe(onNext: { players: [Player] in
+    ///             // on the main queue
+    ///             print("Players have changed")
+    ///         })
+    ///     // <- here "Players 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,
+        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.fetchAll($0, request) }
+                .distinctUntilChanged(==)
+                .map { (rows: [Row]) in rows.map { Base.RowDecoder.init(row: $0) } }
+        } else {
+            return AnyDatabaseWriter(writer).rx
+                .fetchTokens(in: [request], startImmediately: startImmediately, scheduler: scheduler)
+                .mapFetch { try request.fetchAll($0) }
+        }
+    }
     
-    /// Returns an Observable that emits an array of records 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).
+    /// 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.filter(key: 1)
+    ///     request.rx
+    ///         .fetchOne(in: dbQueue)
+    ///         .subscribe(onNext: { player: Player? in
+    ///             print("Player 1 has changed")
+    ///         })
+    ///
+    /// By default, all records are emitted on the main dispatch queue. If you
+    /// give a *scheduler*, records 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: { player: Player? in
+    ///             // on the main queue
+    ///             print("Player 1 has changed")
+    ///         })
+    ///     // <- here "Player 1 has changed" has been printed
+    ///
+    ///     // on any queue
+    ///     request.rx
+    ///         .fetchOne(in: dbQueue, scheduler: MainScheduler.instance)
+    ///         .subscribe(onNext: { player: Player? in
+    ///             // on the main queue
+    ///             print("Player 1 has changed")
+    ///         })
+    ///     // <- here "Player 1 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,
+        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 after each committed database
+    /// transaction that has modified the tables and columns fetched by
+    /// the request.
+    ///
+    ///     let dbQueue = DatabaseQueue()
+    ///     let request = SQLRequest<Row>("SELECT * FROM player")
+    ///     request.rx
+    ///         .fetchAll(in: dbQueue)
+    ///         .subscribe(onNext: { rows: [Row] in
+    ///             print("Players have changed")
+    ///         })
+    ///
+    /// By default, all rows are emitted on the main dispatch queue. If you
+    /// give a *scheduler*, rows 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: { rows: [Row] in
+    ///             // on the main queue
+    ///             print("Players have changed")
+    ///         })
+    ///     // <- here "Players have changed" has been printed
+    ///
+    ///     // on any queue
+    ///     request.rx
+    ///         .fetchAll(in: dbQueue, scheduler: MainScheduler.instance)
+    ///         .subscribe(onNext: { rows: [Row] in
+    ///             // on the main queue
+    ///             print("Players have changed")
+    ///         })
+    ///     // <- here "Players 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,
         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 after each committed database
+    /// transaction that has modified the tables and columns fetched by
+    /// the request.
+    ///
+    ///     let dbQueue = DatabaseQueue()
+    ///     let request = SQLRequest<Row>("SELECT * FROM player WHERE id = 1")
+    ///     request.rx
+    ///         .fetchOne(in: dbQueue)
+    ///         .subscribe(onNext: { row: Row? in
+    ///             print("Player 1 has changed")
+    ///         })
+    ///
+    /// By default, all rows are emitted on the main dispatch queue. If you
+    /// give a *scheduler*, rows 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: { row: Row? in
+    ///             // on the main queue
+    ///             print("Player 1 has changed")
+    ///         })
+    ///     // <- here "Player 1 has changed" has been printed
+    ///
+    ///     // on any queue
+    ///     request.rx
+    ///         .fetchOne(in: dbQueue, scheduler: MainScheduler.instance)
+    ///         .subscribe(onNext: { row: Row? in
+    ///             // on the main queue
+    ///             print("Player 1 has changed")
+    ///         })
+    ///     // <- here "Player 1 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,
+        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 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: 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,
+        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.fetchAll($0, request) }
-                .distinctUntilChanged(==)
-                .map { (rows: [Row]) in rows.map { Base.RowDecoder.init(row: $0) } }
+                .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)

}

extension Reactive where Base: FetchRequest & DatabaseRegionConvertible, Base.RowDecoder: _OptionalProtocol, Base.RowDecoder._Wrapped: 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,

        }
    }
    
    /// 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,
