diff --git a/results.cpp b/results.cpp
index e4ad126e..367a3c46 100644
--- a/results.cpp
+++ b/results.cpp
@@ -17,42 +17,288 @@
 ////////////////////////////////////////////////////////////////////////////
 
 #include "results.hpp"
-#import <stdexcept>
+
+#include <stdexcept>
 
 using namespace realm;
 
-Results::Results(SharedRealm &r, ObjectSchema &o, Query q, SortOrder s) :
-    realm(r), object_schema(o), backing_query(q), table_view(backing_query.find_all())
+Results::Results(SharedRealm r, Query q, SortOrder s)
+: m_realm(std::move(r))
+, m_query(std::move(q))
+, m_table(m_query.get_table().get())
+, m_sort(std::move(s))
+, m_mode(Mode::Query)
 {
-    setSort(std::move(s));
+}
+
+Results::Results(SharedRealm r, Table& table)
+: m_realm(std::move(r))
+, m_table(&table)
+, m_mode(Mode::Table)
+{
+}
+
+void Results::validate_read() const
+{
+    if (m_realm && !m_realm->check_thread())
+        throw Error::IncorrectThread;
+    if (m_table && !m_table->is_attached())
+        throw Error::Invalidated;
+}
+
+void Results::validate_write() const
+{
+    validate_read();
+    if (!m_realm || !m_realm->is_in_transaction())
+        throw Error::NotInWrite;
 }
 
 size_t Results::size()
 {
-    verify_attached();
-    return table_view.size();
-}
-
-void Results::setSort(SortOrder s)
-{
-    sort_order = std::make_unique<SortOrder>(std::move(s));
-    table_view.sort(sort_order->columnIndices, sort_order->ascending);
-}
-
-Row Results::get(std::size_t row_ndx)
-{
-    verify_attached();
-    if (row_ndx >= table_view.size()) {
-        throw std::out_of_range(std::string("Index ") + std::to_string(row_ndx) + " is outside of range 0..." +
-                               std::to_string(table_view.size()) + ".");
+    validate_read();
+    switch (m_mode) {
+        case Mode::Empty: return 0;
+        case Mode::Table: return m_table->size();
+        case Mode::Query: return m_query.count();
+        case Mode::TableView:
+            update_tableview();
+            return m_table_view.size();
     }
-    return table_view.get(row_ndx);
 }
 
-void Results::verify_attached()
+RowExpr Results::get(size_t row_ndx)
 {
-    if (!table_view.is_attached()) {
-        throw std::runtime_error("Tableview is not attached");
+    validate_read();
+    switch (m_mode) {
+        case Mode::Empty: break;
+        case Mode::Table:
+            if (row_ndx < m_table->size())
+                return m_table->get(row_ndx);
+            break;
+        case Mode::Query:
+        case Mode::TableView:
+            update_tableview();
+            if (row_ndx < m_table_view.size())
+                return m_table_view.get(row_ndx);
+            break;
     }
-    table_view.sync_if_needed();
+
+    throw Error::OutOfBoundsIndex;
+}
+
+util::Optional<RowExpr> Results::first()
+{
+    validate_read();
+    switch (m_mode) {
+        case Mode::Empty:
+            return none;
+        case Mode::Table:
+            return m_table->size() == 0 ? util::none : util::Optional<RowExpr>(m_table->front());
+        case Mode::Query:
+            update_tableview();
+            [[clang::fallthrough]];
+        case Mode::TableView:
+            return m_table_view.size() == 0 ? util::none : util::Optional<RowExpr>(m_table_view.front());
+    }
+}
+
+util::Optional<RowExpr> Results::last()
+{
+    validate_read();
+    switch (m_mode) {
+        case Mode::Empty:
+            return none;
+        case Mode::Table:
+            return m_table->size() == 0 ? util::none : util::Optional<RowExpr>(m_table->back());
+        case Mode::Query:
+            update_tableview();
+            [[clang::fallthrough]];
+        case Mode::TableView:
+            return m_table_view.size() == 0 ? util::none : util::Optional<RowExpr>(m_table_view.back());
+    }
+}
+
+void Results::update_tableview()
+{
+    validate_read();
+    switch (m_mode) {
+        case Mode::Empty:
+        case Mode::Table:
+            return;
+        case Mode::Query:
+            m_table_view = m_query.find_all();
+            if (m_sort) {
+                m_table_view.sort(m_sort.columnIndices, m_sort.ascending);
+            }
+            m_mode = Mode::TableView;
+            break;
+        case Mode::TableView:
+            m_table_view.sync_if_needed();
+            break;
+    }
+}
+
+size_t Results::index_of(Row const& row)
+{
+    validate_read();
+    if (!row) {
+        throw Error::DetachedAccessor;
+    }
+    if (m_table && row.get_table() != m_table) {
+        throw Error::IncorrectTable;
+    }
+    return index_of(row.get_index());
+}
+
+size_t Results::index_of(size_t row_ndx)
+{
+    validate_read();
+    switch (m_mode) {
+        case Mode::Empty:
+            return not_found;
+        case Mode::Table:
+            return row_ndx;
+        case Mode::Query:
+            if (!m_sort)
+                return m_query.count(row_ndx, row_ndx + 1) ? m_query.count(0, row_ndx) : not_found;
+            [[clang::fallthrough]];
+        case Mode::TableView:
+            update_tableview();
+            return m_table_view.find_by_source_ndx(row_ndx);
+    }
+
+}
+
+template<typename Int, typename Float, typename Double, typename DateTime>
+util::Optional<Mixed> Results::aggregate(size_t column, bool return_none_for_empty,
+                                         Int agg_int, Float agg_float,
+                                         Double agg_double, DateTime agg_datetime)
+{
+    validate_read();
+    if (!m_table)
+        return none;
+    if (column > m_table->get_column_count())
+        throw Error::OutOfBoundsIndex;
+
+    auto do_agg = [&](auto const& getter) -> util::Optional<Mixed> {
+        switch (m_mode) {
+            case Mode::Empty:
+                return none;
+            case Mode::Table:
+                if (return_none_for_empty && m_table->size() == 0)
+                    return none;
+                return util::Optional<Mixed>(getter(*m_table));
+            case Mode::Query:
+            case Mode::TableView:
+                update_tableview();
+                if (return_none_for_empty && m_table_view.size() == 0)
+                    return none;
+                return util::Optional<Mixed>(getter(m_table_view));
+        }
+    };
+
+    switch (m_table->get_column_type(column))
+    {
+        case type_DateTime: return do_agg(agg_datetime);
+        case type_Double: return do_agg(agg_double);
+        case type_Float: return do_agg(agg_float);
+        case type_Int: return do_agg(agg_int);
+        default:
+            throw Error::UnsupportedColumnType;
+    }
+}
+
+util::Optional<Mixed> Results::max(size_t column)
+{
+    return aggregate(column, true,
+                     [=](auto const& table) { return table.maximum_int(column); },
+                     [=](auto const& table) { return table.maximum_float(column); },
+                     [=](auto const& table) { return table.maximum_double(column); },
+                     [=](auto const& table) { return table.maximum_datetime(column); });
+}
+
+util::Optional<Mixed> Results::min(size_t column)
+{
+    return aggregate(column, true,
+                     [=](auto const& table) { return table.minimum_int(column); },
+                     [=](auto const& table) { return table.minimum_float(column); },
+                     [=](auto const& table) { return table.minimum_double(column); },
+                     [=](auto const& table) { return table.minimum_datetime(column); });
+}
+
+util::Optional<Mixed> Results::sum(size_t column)
+{
+    return aggregate(column, false,
+                     [=](auto const& table) { return table.sum_int(column); },
+                     [=](auto const& table) { return table.sum_float(column); },
+                     [=](auto const& table) { return table.sum_double(column); },
+                     [=](auto const&) -> util::None { throw Error::UnsupportedColumnType; });
+}
+
+util::Optional<Mixed> Results::average(size_t column)
+{
+    return aggregate(column, true,
+                     [=](auto const& table) { return table.average_int(column); },
+                     [=](auto const& table) { return table.average_float(column); },
+                     [=](auto const& table) { return table.average_double(column); },
+                     [=](auto const&) -> util::None { throw Error::UnsupportedColumnType; });
+}
+
+void Results::clear()
+{
+    switch (m_mode) {
+        case Mode::Empty:
+            return;
+        case Mode::Table:
+            validate_write();
+            m_table->clear();
+            break;
+        case Mode::Query:
+            // Not using Query:remove() because bulding the tableview and
+            // clearing it is actually significantly faster
+        case Mode::TableView:
+            validate_write();
+            update_tableview();
+            m_table_view.clear();
+            break;
+    }
+}
+
+Query Results::get_query() const
+{
+    validate_read();
+    switch (m_mode) {
+        case Mode::Empty:
+        case Mode::Query:
+        case Mode::TableView:
+            return m_query;
+        case Mode::Table:
+            return m_table->where();
+    }
+}
+
+TableView Results::get_tableview()
+{
+    validate_read();
+    switch (m_mode) {
+        case Mode::Empty:
+            return {};
+        case Mode::Query:
+        case Mode::TableView:
+            update_tableview();
+            return m_table_view;
+        case Mode::Table:
+            return m_table->where().find_all();
+    }
+    REALM_UNREACHABLE();
+}
+
+Results Results::sort(realm::SortOrder&& sort) const
+{
+    return Results(m_realm, get_query(), std::move(sort));
+}
+
+Results Results::filter(Query&& q) const
+{
+    return Results(m_realm, get_query().and_query(q), get_sort());
 }
diff --git a/results.hpp b/results.hpp
index fcb3e894..ef941b03 100644
--- a/results.hpp
+++ b/results.hpp
@@ -19,35 +19,140 @@
 #ifndef REALM_RESULTS_HPP
 #define REALM_RESULTS_HPP
 
-#import "shared_realm.hpp"
-#import <realm/table_view.hpp>
+#include "shared_realm.hpp"
+
+#include <realm/table_view.hpp>
+#include <realm/table.hpp>
+#include <realm/util/optional.hpp>
 
 namespace realm {
-    struct SortOrder {
-        std::vector<size_t> columnIndices;
-        std::vector<bool> ascending;
+template<typename T> class BasicRowExpr;
+using RowExpr = BasicRowExpr<Table>;
+class Mixed;
 
-        explicit operator bool() const {
-            return !columnIndices.empty();
-        }
+struct SortOrder {
+    std::vector<size_t> columnIndices;
+    std::vector<bool> ascending;
+
+    explicit operator bool() const
+    {
+        return !columnIndices.empty();
+    }
+};
+
+class Results {
+public:
+    // Results can be either be backed by nothing, a thin wrapper around a table,
+    // or a wrapper around a query and a sort order which creates and updates
+    // the tableview as needed
+    Results() = default;
+    Results(SharedRealm r, Table& table);
+    Results(SharedRealm r, Query q, SortOrder s = {});
+
+    // Results is copyable and moveable
+    Results(Results const&) = default;
+    Results(Results&&) = default;
+    Results& operator=(Results const&) = default;
+    Results& operator=(Results&&) = default;
+
+    // Get a query which will match the same rows as is contained in this Results
+    // Returned query will not be valid if the current mode is Empty
+    Query get_query() const;
+
+    // Get the currently applied sort order for this Results
+    SortOrder const& get_sort() const noexcept { return m_sort; }
+
+    // Get a tableview containing the same rows as this Results
+    TableView get_tableview();
+
+    // Get the size of this results
+    // Can be either O(1) or O(N) depending on the state of things
+    size_t size();
+
+    // Get the row accessor for the given index
+    // Throws OutOfBoundsIndex if index >= size()
+    RowExpr get(size_t index);
+
+    // Get a row accessor for the first/last row, or none if the results are empty
+    // More efficient than calling size()+get()
+    util::Optional<RowExpr> first();
+    util::Optional<RowExpr> last();
+
+    // Get the first index of the given row in this results, or not_found
+    // Throws DetachedAccessor if row is not attached
+    // Throws IncorrectTable if row belongs to a different table
+    size_t index_of(size_t row_ndx);
+    size_t index_of(Row const& row);
+
+    // Delete all of the rows in this Results from the Realm
+    // size() will always be zero afterwards
+    // Throws NotInWrite
+    void clear();
+
+    // Create a new Results by further filtering or sorting this Results
+    Results filter(Query&& q) const;
+    Results sort(SortOrder&& sort) const;
+
+    // Get the min/max/average/sum of the given column
+    // All but sum() returns none when there are zero matching rows
+    // sum() returns 0, except for when it returns none
+    // Throws UnsupportedColumnType for sum/average on datetime or non-numeric column
+    // Throws OutOfBoundsIndex for an out-of-bounds column
+    util::Optional<Mixed> max(size_t column);
+    util::Optional<Mixed> min(size_t column);
+    util::Optional<Mixed> average(size_t column);
+    util::Optional<Mixed> sum(size_t column);
+
+    enum class Mode {
+        Empty, // Backed by nothing (for missing tables)
+        Table, // Backed directly by a Table
+        Query, // Backed by a query that has not yet been turned into a TableView
+        TableView // Backed by a TableView created from a Query
+    };
+    // Get the currrent mode of the Results
+    // Ideally this would not be public but it's needed for some KVO stuff
+    Mode get_mode() const { return m_mode; }
+
+    // All errors thrown by Results are one of the following error codes
+    // Invalidated and IncorrectThread can be thrown by all non-noexcept functions
+    // Other errors are thrown only where indicated
+    enum class Error {
+        // The Results object has been invalidated (due to the Realm being invalidated)
+        Invalidated,
+        // The Results object is being accessed from the wrong thread
+        IncorrectThread,
+        // The containing Realm is not in a write transaction
+        NotInWrite,
+
+        // The input index parameter was out of bounds
+        OutOfBoundsIndex,
+        // The input Row object belongs to a different table
+        IncorrectTable,
+        // The input Row object is not attached
+        DetachedAccessor,
+        // The requested aggregate operation is not supported for the column type
+        UnsupportedColumnType
     };
 
-    static SortOrder s_defaultSort = {{}, {}};
+private:
+    SharedRealm m_realm;
+    Query m_query;
+    TableView m_table_view;
+    Table* m_table = nullptr;
+    SortOrder m_sort;
 
-    struct Results {
-        Results(SharedRealm &r, ObjectSchema &o, Query q, SortOrder s = s_defaultSort);
-        size_t size();
-        Row get(std::size_t row_ndx);
-        void verify_attached();
+    Mode m_mode = Mode::Empty;
 
-        SharedRealm realm;
-        ObjectSchema &object_schema;
-        Query backing_query;
-        TableView table_view;
-        std::unique_ptr<SortOrder> sort_order;
+    void validate_read() const;
+    void validate_write() const;
 
-        void setSort(SortOrder s);
-    };
+    void update_tableview();
+
+    template<typename Int, typename Float, typename Double, typename DateTime>
+    util::Optional<Mixed> aggregate(size_t column, bool return_none_for_empty,
+                                    Int agg_int, Float agg_float,
+                                    Double agg_double, DateTime agg_datetime);
+};
 }
 
 #endif /* REALM_RESULTS_HPP */
diff --git a/shared_realm.cpp b/shared_realm.cpp
index e8a497e3..1a670563 100644
--- a/shared_realm.cpp
+++ b/shared_realm.cpp
@@ -236,9 +236,14 @@ static void check_read_write(Realm *realm)
     }
 }
 
+bool Realm::check_thread() const noexcept
+{
+    return m_thread_id == std::this_thread::get_id();
+}
+
 void Realm::verify_thread() const
 {
-    if (m_thread_id != std::this_thread::get_id()) {
+    if (!check_thread()) {
         throw IncorrectThreadException("Realm accessed from incorrect thread.");
     }
 }
diff --git a/shared_realm.hpp b/shared_realm.hpp
index 0b51af5b..ef848fcb 100644
--- a/shared_realm.hpp
+++ b/shared_realm.hpp
@@ -19,7 +19,6 @@
 #ifndef REALM_REALM_HPP
 #define REALM_REALM_HPP
 
-#include <map>
 #include <memory>
 #include <thread>
 #include <vector>
@@ -99,6 +98,7 @@ namespace realm {
         bool compact();
 
         std::thread::id thread_id() const { return m_thread_id; }
+        bool check_thread() const noexcept;
         void verify_thread() const;
 
         // Close this Realm and remove it from the cache. Continuing to use a