Propose Result as immediate replacement for ResultProxy

As progress is made on the _future.Result, including breaking
it out such that DBAPI behaviors are local to specific
implementations, it becomes apparent that the Result object
is a functional superset of ResultProxy and that basic
operations like fetchone(), fetchall(), and fetchmany()
behave pretty much exactly the same way on the new object.
Reorganize things so that ResultProxy is now referred to
as LegacyCursorResult, which subclasses CursorResult
that represents the DBAPI-cursor version of Result,
making use of a multiple inheritance pattern so that
the functionality of Result is also available in non-DBAPI
contexts, as will be necessary for some ORM
patterns.

Additionally propose the composition system for Result
that will form the basis for ORM-alternative result
systems such as horizontal sharding and dogpile cache.
As ORM results will soon be coming directly from
instances of Result, these extensions will instead
build their own ResultFetchStrategies that perform
the special steps to create composed or cached
result sets.

Also considering at the moment not emitting deprecation
warnings for fetchXYZ() methods; the immediate issue
is Keystone tests are calling upon it, but as the
implementations here are proving to be not in any
kind of conflict with how Result works, there's
not too much issue leaving them around and deprecating
at some later point.

References: #5087
References: #4395
Fixes: #4959
Change-Id: I8091919d45421e3f53029b8660427f844fee0228

9 files changed, 1196 insertions, 182 deletions

diff --git a/test/aaa_profiling/test_memusage.py b/test/aaa_profiling/test_memusage.py
index ad4db4545..59564e5bb 100644
--- a/test/aaa_profiling/test_memusage.py
+++ b/test/aaa_profiling/test_memusage.py
@@ -1355,6 +1355,71 @@ class CycleTest(_fixtures.FixtureTest):
 
         go()
 
+    def test_result_fetchone(self):
+        User, Address = self.classes("User", "Address")
+        configure_mappers()
+
+        s = Session()
+
+        stmt = s.query(User).join(User.addresses).statement
+
+        @assert_cycles()
+        def go():
+            result = s.execute(stmt)
+            while True:
+                row = result.fetchone()
+                if row is None:
+                    break
+
+        go()
+
+    def test_result_fetchall(self):
+        User, Address = self.classes("User", "Address")
+        configure_mappers()
+
+        s = Session()
+
+        stmt = s.query(User).join(User.addresses).statement
+
+        @assert_cycles()
+        def go():
+            result = s.execute(stmt)
+            rows = result.fetchall()  # noqa
+
+        go()
+
+    def test_result_fetchmany(self):
+        User, Address = self.classes("User", "Address")
+        configure_mappers()
+
+        s = Session()
+
+        stmt = s.query(User).join(User.addresses).statement
+
+        @assert_cycles()
+        def go():
+            result = s.execute(stmt)
+            for partition in result.partitions(3):
+                pass
+
+        go()
+
+    def test_result_fetchmany_unique(self):
+        User, Address = self.classes("User", "Address")
+        configure_mappers()
+
+        s = Session()
+
+        stmt = s.query(User).join(User.addresses).statement
+
+        @assert_cycles()
+        def go():
+            result = s.execute(stmt)
+            for partition in result.unique().partitions(3):
+                pass
+
+        go()
+
     def test_core_select(self):
         User, Address = self.classes("User", "Address")
         configure_mappers()
diff --git a/test/aaa_profiling/test_resultset.py b/test/aaa_profiling/test_resultset.py
index fee4daeb2..abc10d2f6 100644
--- a/test/aaa_profiling/test_resultset.py
+++ b/test/aaa_profiling/test_resultset.py
@@ -8,8 +8,8 @@ from sqlalchemy import String
 from sqlalchemy import Table
 from sqlalchemy import testing
 from sqlalchemy import Unicode
-from sqlalchemy.engine.result import LegacyRow
-from sqlalchemy.engine.result import Row
+from sqlalchemy.engine.row import LegacyRow
+from sqlalchemy.engine.row import Row
 from sqlalchemy.testing import AssertsExecutionResults
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import fixtures
diff --git a/test/base/test_result.py b/test/base/test_result.py
new file mode 100644
index 000000000..097e65307
--- /dev/null
+++ b/test/base/test_result.py
@@ -0,0 +1,790 @@
+from sqlalchemy import exc
+from sqlalchemy import testing
+from sqlalchemy.engine import result
+from sqlalchemy.engine.row import Row
+from sqlalchemy.testing import assert_raises
+from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import eq_
+from sqlalchemy.testing import fixtures
+from sqlalchemy.testing import is_false
+from sqlalchemy.testing import is_true
+from sqlalchemy.testing.util import picklers
+
+
+class ResultTupleTest(fixtures.TestBase):
+    def _fixture(self, values, labels):
+        return result.result_tuple(labels)(values)
+
+    def test_empty(self):
+        keyed_tuple = self._fixture([], [])
+        eq_(str(keyed_tuple), "()")
+        eq_(len(keyed_tuple), 0)
+
+        eq_(list(keyed_tuple._mapping.keys()), [])
+        eq_(keyed_tuple._fields, ())
+        eq_(keyed_tuple._asdict(), {})
+
+    def test_values_none_labels(self):
+        keyed_tuple = self._fixture([1, 2], [None, None])
+        eq_(str(keyed_tuple), "(1, 2)")
+        eq_(len(keyed_tuple), 2)
+
+        eq_(list(keyed_tuple._mapping.keys()), [])
+        eq_(keyed_tuple._fields, ())
+        eq_(keyed_tuple._asdict(), {})
+
+        eq_(keyed_tuple[0], 1)
+        eq_(keyed_tuple[1], 2)
+
+    def test_creation(self):
+        keyed_tuple = self._fixture([1, 2], ["a", "b"])
+        eq_(str(keyed_tuple), "(1, 2)")
+        eq_(list(keyed_tuple._mapping.keys()), ["a", "b"])
+        eq_(keyed_tuple._fields, ("a", "b"))
+        eq_(keyed_tuple._asdict(), {"a": 1, "b": 2})
+
+    def test_index_access(self):
+        keyed_tuple = self._fixture([1, 2], ["a", "b"])
+        eq_(keyed_tuple[0], 1)
+        eq_(keyed_tuple[1], 2)
+
+        def should_raise():
+            keyed_tuple[2]
+
+        assert_raises(IndexError, should_raise)
+
+    def test_negative_index_access(self):
+        keyed_tuple = self._fixture([1, 2], ["a", "b"])
+        eq_(keyed_tuple[-1], 2)
+        eq_(keyed_tuple[-2:-1], (1,))
+
+    def test_slice_access(self):
+        keyed_tuple = self._fixture([1, 2], ["a", "b"])
+        eq_(keyed_tuple[0:2], (1, 2))
+
+    def test_getter(self):
+        keyed_tuple = self._fixture([1, 2, 3], ["a", "b", "c"])
+
+        getter = keyed_tuple._parent._getter("b")
+        eq_(getter(keyed_tuple), 2)
+
+        getter = keyed_tuple._parent._getter(2)
+        eq_(getter(keyed_tuple), 3)
+
+    def test_tuple_getter(self):
+        keyed_tuple = self._fixture([1, 2, 3], ["a", "b", "c"])
+
+        getter = keyed_tuple._parent._row_as_tuple_getter(["b", "c"])
+        eq_(getter(keyed_tuple), (2, 3))
+
+        getter = keyed_tuple._parent._row_as_tuple_getter([2, 0, 1])
+        eq_(getter(keyed_tuple), (3, 1, 2))
+
+    def test_attribute_access(self):
+        keyed_tuple = self._fixture([1, 2], ["a", "b"])
+        eq_(keyed_tuple.a, 1)
+        eq_(keyed_tuple.b, 2)
+
+        def should_raise():
+            keyed_tuple.c
+
+        assert_raises(AttributeError, should_raise)
+
+    def test_contains(self):
+        keyed_tuple = self._fixture(["x", "y"], ["a", "b"])
+
+        is_true("x" in keyed_tuple)
+        is_false("z" in keyed_tuple)
+
+        is_true("z" not in keyed_tuple)
+        is_false("x" not in keyed_tuple)
+
+        # we don't do keys
+        is_false("a" in keyed_tuple)
+        is_false("z" in keyed_tuple)
+        is_true("a" not in keyed_tuple)
+        is_true("z" not in keyed_tuple)
+
+    def test_none_label(self):
+        keyed_tuple = self._fixture([1, 2, 3], ["a", None, "b"])
+        eq_(str(keyed_tuple), "(1, 2, 3)")
+
+        eq_(list(keyed_tuple._mapping.keys()), ["a", "b"])
+        eq_(keyed_tuple._fields, ("a", "b"))
+        eq_(keyed_tuple._asdict(), {"a": 1, "b": 3})
+
+        # attribute access: can't get at value 2
+        eq_(keyed_tuple.a, 1)
+        eq_(keyed_tuple.b, 3)
+
+        # index access: can get at value 2
+        eq_(keyed_tuple[0], 1)
+        eq_(keyed_tuple[1], 2)
+        eq_(keyed_tuple[2], 3)
+
+    def test_duplicate_labels(self):
+        keyed_tuple = self._fixture([1, 2, 3], ["a", "b", "b"])
+        eq_(str(keyed_tuple), "(1, 2, 3)")
+
+        eq_(list(keyed_tuple._mapping.keys()), ["a", "b", "b"])
+        eq_(keyed_tuple._fields, ("a", "b", "b"))
+        eq_(keyed_tuple._asdict(), {"a": 1, "b": 3})
+
+        # attribute access: can't get at value 2
+        eq_(keyed_tuple.a, 1)
+        eq_(keyed_tuple.b, 3)
+
+        # index access: can get at value 2
+        eq_(keyed_tuple[0], 1)
+        eq_(keyed_tuple[1], 2)
+        eq_(keyed_tuple[2], 3)
+
+    def test_immutable(self):
+        keyed_tuple = self._fixture([1, 2], ["a", "b"])
+        eq_(str(keyed_tuple), "(1, 2)")
+
+        eq_(keyed_tuple.a, 1)
+
+        # eh
+        # assert_raises(AttributeError, setattr, keyed_tuple, "a", 5)
+
+        def should_raise():
+            keyed_tuple[0] = 100
+
+        assert_raises(TypeError, should_raise)
+
+    def test_serialize(self):
+
+        keyed_tuple = self._fixture([1, 2, 3], ["a", None, "b"])
+
+        for loads, dumps in picklers():
+            kt = loads(dumps(keyed_tuple))
+
+            eq_(str(kt), "(1, 2, 3)")
+
+            eq_(list(kt._mapping.keys()), ["a", "b"])
+            eq_(kt._fields, ("a", "b"))
+            eq_(kt._asdict(), {"a": 1, "b": 3})
+
+
+class ResultTest(fixtures.TestBase):
+    def _fixture(
+        self,
+        extras=None,
+        alt_row=None,
+        num_rows=None,
+        default_filters=None,
+        data=None,
+    ):
+
+        if data is None:
+            data = [(1, 1, 1), (2, 1, 2), (1, 3, 2), (4, 1, 2)]
+        if num_rows is not None:
+            data = data[:num_rows]
+
+        res = result.IteratorResult(
+            result.SimpleResultMetaData(["a", "b", "c"], extra=extras),
+            iter(data),
+        )
+        if default_filters:
+            res._metadata._unique_filters = default_filters
+
+        if alt_row:
+            res._process_row = alt_row
+
+        return res
+
+    def test_index_extra(self):
+        ex1a, ex1b, ex2, ex3a, ex3b = (
+            object(),
+            object(),
+            object(),
+            object(),
+            object(),
+        )
+
+        result = self._fixture(extras=[(ex1a, ex1b), (ex2,), (ex3a, ex3b,)])
+        eq_(
+            result.columns(ex2, ex3b).columns(ex3a).all(),
+            [(1,), (2,), (2,), (2,)],
+        )
+
+        result = self._fixture(extras=[(ex1a, ex1b), (ex2,), (ex3a, ex3b,)])
+        eq_([row._mapping[ex1b] for row in result], [1, 2, 1, 4])
+
+        result = self._fixture(extras=[(ex1a, ex1b), (ex2,), (ex3a, ex3b,)])
+        eq_(
+            [
+                dict(r)
+                for r in result.columns(ex2, ex3b).columns(ex3a).mappings()
+            ],
+            [{"c": 1}, {"c": 2}, {"c": 2}, {"c": 2}],
+        )
+
+    def test_unique_default_filters(self):
+        result = self._fixture(
+            default_filters=[lambda x: x < 4, lambda x: x, lambda x: True]
+        )
+
+        eq_(result.unique().all(), [(1, 1, 1), (1, 3, 2), (4, 1, 2)])
+
+    def test_unique_default_filters_rearrange_scalar(self):
+        result = self._fixture(
+            default_filters=[lambda x: x < 4, lambda x: x, lambda x: True]
+        )
+
+        eq_(result.unique().scalars(1).all(), [1, 3])
+
+    def test_unique_default_filters_rearrange_order(self):
+        result = self._fixture(
+            default_filters=[lambda x: x < 4, lambda x: x, lambda x: True]
+        )
+
+        eq_(
+            result.unique().columns("b", "a", "c").all(),
+            [(1, 1, 1), (3, 1, 2), (1, 4, 2)],
+        )
+
+    def test_unique_default_filters_rearrange_twice(self):
+        # test that the default uniqueness filter is reconfigured
+        # each time columns() is called
+        result = self._fixture(
+            default_filters=[lambda x: x < 4, lambda x: x, lambda x: True]
+        )
+
+        result = result.unique()
+
+        # 1, 1, 1  ->   True, 1, True
+        eq_(result.fetchone(), (1, 1, 1))
+
+        # now rearrange for b, a, c
+        # 1, 2, 2  ->   1, True, True
+        # 3, 1, 2  ->   3, True, True
+        result = result.columns("b", "a", "c")
+        eq_(result.fetchone(), (3, 1, 2))
+
+        # now rearrange for c, a
+        # 2, 4  -> True, False
+        result = result.columns("c", "a")
+        eq_(result.fetchall(), [(2, 4)])
+
+    def test_unique_scalars_all(self):
+        result = self._fixture()
+
+        eq_(result.unique().scalars(1).all(), [1, 3])
+
+    def test_unique_mappings_all(self):
+        result = self._fixture()
+
+        def uniq(row):
+            return row[0]
+
+        eq_(
+            result.unique(uniq).mappings().all(),
+            [
+                {"a": 1, "b": 1, "c": 1},
+                {"a": 2, "b": 1, "c": 2},
+                {"a": 4, "b": 1, "c": 2},
+            ],
+        )
+
+    def test_unique_filtered_all(self):
+        result = self._fixture()
+
+        def uniq(row):
+            return row[0]
+
+        eq_(result.unique(uniq).all(), [(1, 1, 1), (2, 1, 2), (4, 1, 2)])
+
+    def test_unique_scalars_many(self):
+        result = self._fixture()
+
+        result = result.unique().scalars(1)
+
+        eq_(result.fetchmany(2), [1, 3])
+        eq_(result.fetchmany(2), [])
+
+    def test_unique_filtered_many(self):
+        result = self._fixture()
+
+        def uniq(row):
+            return row[0]
+
+        result = result.unique(uniq)
+
+        eq_(result.fetchmany(2), [(1, 1, 1), (2, 1, 2)])
+        eq_(result.fetchmany(2), [(4, 1, 2)])
+        eq_(result.fetchmany(2), [])
+
+    def test_unique_scalars_many_none(self):
+        result = self._fixture()
+
+        result = result.unique().scalars(1)
+
+        # this assumes the default fetchmany behavior of all() for
+        # the ListFetchStrategy
+        eq_(result.fetchmany(None), [1, 3])
+        eq_(result.fetchmany(None), [])
+
+    def test_unique_scalars_iterate(self):
+        result = self._fixture()
+
+        result = result.unique().scalars(1)
+
+        eq_(list(result), [1, 3])
+
+    def test_unique_filtered_iterate(self):
+        result = self._fixture()
+
+        def uniq(row):
+            return row[0]
+
+        result = result.unique(uniq)
+
+        eq_(list(result), [(1, 1, 1), (2, 1, 2), (4, 1, 2)])
+
+    def test_all(self):
+        result = self._fixture()
+
+        eq_(result.all(), [(1, 1, 1), (2, 1, 2), (1, 3, 2), (4, 1, 2)])
+
+        eq_(result.all(), [])
+
+    def test_many_then_all(self):
+        result = self._fixture()
+
+        eq_(result.fetchmany(3), [(1, 1, 1), (2, 1, 2), (1, 3, 2)])
+        eq_(result.all(), [(4, 1, 2)])
+
+        eq_(result.all(), [])
+
+    def test_scalars(self):
+        result = self._fixture()
+
+        eq_(list(result.scalars()), [1, 2, 1, 4])
+
+        result = self._fixture()
+
+        eq_(list(result.scalars(2)), [1, 2, 2, 2])
+
+    def test_scalars_no_fetchone(self):
+        result = self._fixture()
+
+        result = result.scalars()
+
+        assert_raises_message(
+            exc.InvalidRequestError,
+            r"Can't use fetchone\(\) when returning scalar values; ",
+            result.fetchone,
+        )
+
+        # mappings() switches the flag off
+        eq_(result.mappings().fetchone(), {"a": 1})
+
+    def test_first(self):
+        result = self._fixture()
+
+        row = result.first()
+        eq_(row, (1, 1, 1))
+
+        eq_(result.all(), [])
+
+    def test_one(self):
+        result = self._fixture(num_rows=1)
+
+        row = result.one()
+        eq_(row, (1, 1, 1))
+
+    def test_scalar_one(self):
+        result = self._fixture(num_rows=1)
+
+        row = result.scalars().one()
+        eq_(row, 1)
+
+    def test_scalar_one_none(self):
+        result = self._fixture(num_rows=0)
+
+        result = result.scalars()
+        assert_raises_message(
+            exc.NoResultFound,
+            "No row was found when one was required",
+            result.one,
+        )
+
+    def test_one_none(self):
+        result = self._fixture(num_rows=0)
+
+        assert_raises_message(
+            exc.NoResultFound,
+            "No row was found when one was required",
+            result.one,
+        )
+
+    def test_one_or_none_none(self):
+        result = self._fixture(num_rows=0)
+
+        eq_(result.one_or_none(), None)
+
+    def test_one_raise_mutiple(self):
+        result = self._fixture(num_rows=2)
+
+        assert_raises_message(
+            exc.MultipleResultsFound,
+            "Multiple rows were found when exactly one was required",
+            result.one,
+        )
+
+    def test_one_or_none_raise_multiple(self):
+        result = self._fixture(num_rows=2)
+
+        assert_raises_message(
+            exc.MultipleResultsFound,
+            "Multiple rows were found when one or none was required",
+            result.one_or_none,
+        )
+
+    def test_scalar(self):
+        result = self._fixture()
+
+        eq_(result.scalar(), 1)
+
+        eq_(result.all(), [])
+
+    def test_partition(self):
+        result = self._fixture()
+
+        r = []
+        for partition in result.partitions(2):
+            r.append(list(partition))
+        eq_(r, [[(1, 1, 1), (2, 1, 2)], [(1, 3, 2), (4, 1, 2)]])
+
+        eq_(result.all(), [])
+
+    def test_partition_unique_yield_per(self):
+        result = self._fixture()
+
+        r = []
+        for partition in result.unique().yield_per(2).partitions():
+            r.append(list(partition))
+        eq_(r, [[(1, 1, 1), (2, 1, 2)], [(1, 3, 2), (4, 1, 2)]])
+
+        eq_(result.all(), [])
+
+    def test_partition_yield_per(self):
+        result = self._fixture()
+
+        r = []
+        for partition in result.yield_per(2).partitions():
+            r.append(list(partition))
+        eq_(r, [[(1, 1, 1), (2, 1, 2)], [(1, 3, 2), (4, 1, 2)]])
+
+        eq_(result.all(), [])
+
+    def test_columns(self):
+        result = self._fixture()
+
+        result = result.columns("b", "c")
+        eq_(result.keys(), ["b", "c"])
+        eq_(result.all(), [(1, 1), (1, 2), (3, 2), (1, 2)])
+
+    def test_columns_ints(self):
+        result = self._fixture()
+
+        eq_(result.columns(1, -2).all(), [(1, 1), (1, 1), (3, 3), (1, 1)])
+
+    def test_columns_again(self):
+        result = self._fixture()
+
+        eq_(
+            result.columns("b", "c", "a").columns(1, 2).all(),
+            [(1, 1), (2, 2), (2, 1), (2, 4)],
+        )
+
+    def test_mappings(self):
+        result = self._fixture()
+
+        eq_(
+            [dict(r) for r in result.mappings()],
+            [
+                {"a": 1, "b": 1, "c": 1},
+                {"a": 2, "b": 1, "c": 2},
+                {"a": 1, "b": 3, "c": 2},
+                {"a": 4, "b": 1, "c": 2},
+            ],
+        )
+
+    def test_columns_with_mappings(self):
+        result = self._fixture()
+        eq_(
+            [dict(r) for r in result.columns("b", "c").mappings().all()],
+            [
+                {"b": 1, "c": 1},
+                {"b": 1, "c": 2},
+                {"b": 3, "c": 2},
+                {"b": 1, "c": 2},
+            ],
+        )
+
+    def test_alt_row_fetch(self):
+        class AppleRow(Row):
+            def apple(self):
+                return "apple"
+
+        result = self._fixture(alt_row=AppleRow)
+
+        row = result.all()[0]
+        eq_(row.apple(), "apple")
+
+    def test_alt_row_transform(self):
+        class AppleRow(Row):
+            def apple(self):
+                return "apple"
+
+        result = self._fixture(alt_row=AppleRow)
+
+        row = result.columns("c", "a").all()[2]
+        eq_(row.apple(), "apple")
+        eq_(row, (2, 1))
+
+    def test_scalar_none_iterate(self):
+        result = self._fixture(
+            data=[
+                (1, None, 2),
+                (3, 4, 5),
+                (3, None, 5),
+                (3, None, 5),
+                (3, 4, 5),
+            ]
+        )
+
+        result = result.scalars(1)
+        eq_(list(result), [None, 4, None, None, 4])
+
+    def test_scalar_none_many(self):
+        result = self._fixture(
+            data=[
+                (1, None, 2),
+                (3, 4, 5),
+                (3, None, 5),
+                (3, None, 5),
+                (3, 4, 5),
+            ]
+        )
+
+        result = result.scalars(1)
+
+        eq_(result.fetchmany(3), [None, 4, None])
+        eq_(result.fetchmany(5), [None, 4])
+
+    def test_scalar_none_all(self):
+        result = self._fixture(
+            data=[
+                (1, None, 2),
+                (3, 4, 5),
+                (3, None, 5),
+                (3, None, 5),
+                (3, 4, 5),
+            ]
+        )
+
+        result = result.scalars(1)
+        eq_(result.all(), [None, 4, None, None, 4])
+
+    def test_scalar_none_all_unique(self):
+        result = self._fixture(
+            data=[
+                (1, None, 2),
+                (3, 4, 5),
+                (3, None, 5),
+                (3, None, 5),
+                (3, 4, 5),
+            ]
+        )
+
+        result = result.scalars(1).unique()
+        eq_(result.all(), [None, 4])
+
+    def test_scalar_none_one(self):
+        result = self._fixture(data=[(1, None, 2)])
+
+        result = result.scalars(1).unique()
+
+        # one is returning None, see?
+        eq_(result.one(), None)
+
+    def test_scalar_none_one_or_none(self):
+        result = self._fixture(data=[(1, None, 2)])
+
+        result = result.scalars(1).unique()
+
+        # the orm.Query can actually do this right now, so we sort of
+        # have to allow for this unforuntately, unless we want to raise?
+        eq_(result.one_or_none(), None)
+
+    def test_scalar_none_first(self):
+        result = self._fixture(data=[(1, None, 2)])
+
+        result = result.scalars(1).unique()
+        eq_(result.first(), None)
+
+    def test_freeze(self):
+        result = self._fixture()
+
+        frozen = result.freeze()
+
+        r1 = frozen()
+        eq_(r1.fetchall(), [(1, 1, 1), (2, 1, 2), (1, 3, 2), (4, 1, 2)])
+        eq_(r1.fetchall(), [])
+
+        r2 = frozen()
+        eq_(r1.fetchall(), [])
+        eq_(r2.fetchall(), [(1, 1, 1), (2, 1, 2), (1, 3, 2), (4, 1, 2)])
+        eq_(r2.fetchall(), [])
+
+    def test_columns_unique_freeze(self):
+        result = self._fixture()
+
+        result = result.columns("b", "c").unique()
+
+        frozen = result.freeze()
+
+        r1 = frozen()
+        eq_(r1.fetchall(), [(1, 1), (1, 2), (3, 2)])
+
+    def test_columns_freeze(self):
+        result = self._fixture()
+
+        result = result.columns("b", "c")
+
+        frozen = result.freeze()
+
+        r1 = frozen()
+        eq_(r1.fetchall(), [(1, 1), (1, 2), (3, 2), (1, 2)])
+
+        r2 = frozen().unique()
+        eq_(r2.fetchall(), [(1, 1), (1, 2), (3, 2)])
+
+    def test_scalars_freeze(self):
+        result = self._fixture()
+
+        result = result.scalars(1)
+
+        frozen = result.freeze()
+
+        r1 = frozen()
+        eq_(r1.fetchall(), [1, 1, 3, 1])
+
+        r2 = frozen().unique()
+        eq_(r2.fetchall(), [1, 3])
+
+
+class MergeResultTest(fixtures.TestBase):
+    @testing.fixture
+    def merge_fixture(self):
+
+        r1 = result.IteratorResult(
+            result.SimpleResultMetaData(["user_id", "user_name"]),
+            iter([(7, "u1"), (8, "u2")]),
+        )
+        r2 = result.IteratorResult(
+            result.SimpleResultMetaData(["user_id", "user_name"]),
+            iter([(9, "u3")]),
+        )
+        r3 = result.IteratorResult(
+            result.SimpleResultMetaData(["user_id", "user_name"]),
+            iter([(10, "u4"), (11, "u5")]),
+        )
+        r4 = result.IteratorResult(
+            result.SimpleResultMetaData(["user_id", "user_name"]),
+            iter([(12, "u6")]),
+        )
+
+        return r1, r2, r3, r4
+
+    @testing.fixture
+    def dupe_fixture(self):
+
+        r1 = result.IteratorResult(
+            result.SimpleResultMetaData(["x", "y", "z"]),
+            iter([(1, 2, 1), (2, 2, 1)]),
+        )
+        r2 = result.IteratorResult(
+            result.SimpleResultMetaData(["x", "y", "z"]),
+            iter([(3, 1, 2), (3, 3, 3)]),
+        )
+
+        return r1, r2
+
+    def test_merge_results(self, merge_fixture):
+        r1, r2, r3, r4 = merge_fixture
+
+        result = r1.merge(r2, r3, r4)
+
+        eq_(result.keys(), ["user_id", "user_name"])
+        row = result.fetchone()
+        eq_(row, (7, "u1"))
+        result.close()
+
+    def test_fetchall(self, merge_fixture):
+        r1, r2, r3, r4 = merge_fixture
+
+        result = r1.merge(r2, r3, r4)
+        eq_(
+            result.fetchall(),
+            [
+                (7, "u1"),
+                (8, "u2"),
+                (9, "u3"),
+                (10, "u4"),
+                (11, "u5"),
+                (12, "u6"),
+            ],
+        )
+
+    def test_first(self, merge_fixture):
+        r1, r2, r3, r4 = merge_fixture
+
+        result = r1.merge(r2, r3, r4)
+        eq_(
+            result.first(), (7, "u1"),
+        )
+
+    def test_columns(self, merge_fixture):
+        r1, r2, r3, r4 = merge_fixture
+
+        result = r1.merge(r2, r3, r4)
+        eq_(
+            result.columns("user_name").fetchmany(4),
+            [("u1",), ("u2",), ("u3",), ("u4",)],
+        )
+        result.close()
+
+    def test_merge_scalars(self, merge_fixture):
+        r1, r2, r3, r4 = merge_fixture
+
+        for r in (r1, r2, r3, r4):
+            r.scalars(0)
+
+        result = r1.merge(r2, r3, r4)
+
+        eq_(result.all(), [7, 8, 9, 10, 11, 12])
+
+    def test_merge_unique(self, dupe_fixture):
+        r1, r2 = dupe_fixture
+
+        r1.scalars("y")
+        r2.scalars("y")
+        result = r1.merge(r2)
+
+        # uniqued 2, 2, 1, 3
+        eq_(result.unique().all(), [2, 1, 3])
+
+    def test_merge_preserve_unique(self, dupe_fixture):
+        r1, r2 = dupe_fixture
+
+        r1.unique().scalars("y")
+        r2.unique().scalars("y")
+        result = r1.merge(r2)
+
+        # unique takes place
+        eq_(result.all(), [2, 1, 3])
diff --git a/test/base/test_utils.py b/test/base/test_utils.py
index 246199861..96a9f955a 100644
--- a/test/base/test_utils.py
+++ b/test/base/test_utils.py
@@ -9,7 +9,6 @@ from sqlalchemy import exc
 from sqlalchemy import sql
 from sqlalchemy import testing
 from sqlalchemy import util
-from sqlalchemy.engine import result
 from sqlalchemy.sql import column
 from sqlalchemy.sql.base import DedupeColumnCollection
 from sqlalchemy.testing import assert_raises
@@ -33,144 +32,6 @@ from sqlalchemy.util import timezone
 from sqlalchemy.util import WeakSequence
 
 
-class _KeyedTupleTest(object):
-    def _fixture(self, values, labels):
-        raise NotImplementedError()
-
-    def test_empty(self):
-        keyed_tuple = self._fixture([], [])
-        eq_(str(keyed_tuple), "()")
-        eq_(len(keyed_tuple), 0)
-
-        eq_(list(keyed_tuple._mapping.keys()), [])
-        eq_(keyed_tuple._fields, ())
-        eq_(keyed_tuple._asdict(), {})
-
-    def test_values_none_labels(self):
-        keyed_tuple = self._fixture([1, 2], [None, None])
-        eq_(str(keyed_tuple), "(1, 2)")
-        eq_(len(keyed_tuple), 2)
-
-        eq_(list(keyed_tuple._mapping.keys()), [])
-        eq_(keyed_tuple._fields, ())
-        eq_(keyed_tuple._asdict(), {})
-
-        eq_(keyed_tuple[0], 1)
-        eq_(keyed_tuple[1], 2)
-
-    def test_creation(self):
-        keyed_tuple = self._fixture([1, 2], ["a", "b"])
-        eq_(str(keyed_tuple), "(1, 2)")
-        eq_(list(keyed_tuple._mapping.keys()), ["a", "b"])
-        eq_(keyed_tuple._fields, ("a", "b"))
-        eq_(keyed_tuple._asdict(), {"a": 1, "b": 2})
-
-    def test_index_access(self):
-        keyed_tuple = self._fixture([1, 2], ["a", "b"])
-        eq_(keyed_tuple[0], 1)
-        eq_(keyed_tuple[1], 2)
-
-        def should_raise():
-            keyed_tuple[2]
-
-        assert_raises(IndexError, should_raise)
-
-    def test_slice_access(self):
-        keyed_tuple = self._fixture([1, 2], ["a", "b"])
-        eq_(keyed_tuple[0:2], (1, 2))
-
-    def test_attribute_access(self):
-        keyed_tuple = self._fixture([1, 2], ["a", "b"])
-        eq_(keyed_tuple.a, 1)
-        eq_(keyed_tuple.b, 2)
-
-        def should_raise():
-            keyed_tuple.c
-
-        assert_raises(AttributeError, should_raise)
-
-    def test_contains(self):
-        keyed_tuple = self._fixture(["x", "y"], ["a", "b"])
-
-        is_true("x" in keyed_tuple)
-        is_false("z" in keyed_tuple)
-
-        is_true("z" not in keyed_tuple)
-        is_false("x" not in keyed_tuple)
-
-        # we don't do keys
-        is_false("a" in keyed_tuple)
-        is_false("z" in keyed_tuple)
-        is_true("a" not in keyed_tuple)
-        is_true("z" not in keyed_tuple)
-
-    def test_none_label(self):
-        keyed_tuple = self._fixture([1, 2, 3], ["a", None, "b"])
-        eq_(str(keyed_tuple), "(1, 2, 3)")
-
-        eq_(list(keyed_tuple._mapping.keys()), ["a", "b"])
-        eq_(keyed_tuple._fields, ("a", "b"))
-        eq_(keyed_tuple._asdict(), {"a": 1, "b": 3})
-
-        # attribute access: can't get at value 2
-        eq_(keyed_tuple.a, 1)
-        eq_(keyed_tuple.b, 3)
-
-        # index access: can get at value 2
-        eq_(keyed_tuple[0], 1)
-        eq_(keyed_tuple[1], 2)
-        eq_(keyed_tuple[2], 3)
-
-    def test_duplicate_labels(self):
-        keyed_tuple = self._fixture([1, 2, 3], ["a", "b", "b"])
-        eq_(str(keyed_tuple), "(1, 2, 3)")
-
-        eq_(list(keyed_tuple._mapping.keys()), ["a", "b", "b"])
-        eq_(keyed_tuple._fields, ("a", "b", "b"))
-        eq_(keyed_tuple._asdict(), {"a": 1, "b": 3})
-
-        # attribute access: can't get at value 2
-        eq_(keyed_tuple.a, 1)
-        eq_(keyed_tuple.b, 3)
-
-        # index access: can get at value 2
-        eq_(keyed_tuple[0], 1)
-        eq_(keyed_tuple[1], 2)
-        eq_(keyed_tuple[2], 3)
-
-    def test_immutable(self):
-        keyed_tuple = self._fixture([1, 2], ["a", "b"])
-        eq_(str(keyed_tuple), "(1, 2)")
-
-        eq_(keyed_tuple.a, 1)
-
-        # eh
-        # assert_raises(AttributeError, setattr, keyed_tuple, "a", 5)
-
-        def should_raise():
-            keyed_tuple[0] = 100
-
-        assert_raises(TypeError, should_raise)
-
-    def test_serialize(self):
-
-        keyed_tuple = self._fixture([1, 2, 3], ["a", None, "b"])
-
-        for loads, dumps in picklers():
-            kt = loads(dumps(keyed_tuple))
-
-            eq_(str(kt), "(1, 2, 3)")
-
-            eq_(list(kt._mapping.keys()), ["a", "b"])
-            eq_(kt._fields, ("a", "b"))
-            eq_(kt._asdict(), {"a": 1, "b": 3})
-
-
-class LWKeyedTupleTest(_KeyedTupleTest, fixtures.TestBase):
-    def _fixture(self, values, labels):
-        return result.result_tuple(labels)(values)
-
-
 class WeakSequenceTest(fixtures.TestBase):
     @testing.requires.predictable_gc
     def test_cleanout_elements(self):
diff --git a/test/engine/test_execute.py b/test/engine/test_execute.py
index f694a251c..a8d7b397e 100644
--- a/test/engine/test_execute.py
+++ b/test/engine/test_execute.py
@@ -2370,7 +2370,7 @@ class HandleErrorTest(fixtures.TestBase):
             the_conn.append(connection)
 
         with mock.patch(
-            "sqlalchemy.engine.result.BaseResult.__init__",
+            "sqlalchemy.engine.cursor.BaseCursorResult.__init__",
             Mock(side_effect=tsa.exc.InvalidRequestError("duplicate col")),
         ):
             assert_raises(
@@ -2403,7 +2403,7 @@ class HandleErrorTest(fixtures.TestBase):
         conn = engine.connect()
 
         with mock.patch(
-            "sqlalchemy.engine.result.BaseResult.__init__",
+            "sqlalchemy.engine.cursor.BaseCursorResult.__init__",
             Mock(side_effect=tsa.exc.InvalidRequestError("duplicate col")),
         ):
             assert_raises(
diff --git a/test/orm/test_unitofworkv2.py b/test/orm/test_unitofworkv2.py
index 5ae0dc5dc..2bd908c8f 100644
--- a/test/orm/test_unitofworkv2.py
+++ b/test/orm/test_unitofworkv2.py
@@ -1642,7 +1642,7 @@ class BasicStaleChecksTest(fixtures.MappedTest):
             config.db.dialect, "supports_sane_multi_rowcount", False
         ):
             with patch(
-                "sqlalchemy.engine.result.ResultProxy.rowcount", rowcount
+                "sqlalchemy.engine.cursor.CursorResult.rowcount", rowcount
             ):
                 Parent, Child = self._fixture()
                 sess = Session()
@@ -1672,7 +1672,7 @@ class BasicStaleChecksTest(fixtures.MappedTest):
             config.db.dialect, "supports_sane_multi_rowcount", False
         ):
             with patch(
-                "sqlalchemy.engine.result.ResultProxy.rowcount", rowcount
+                "sqlalchemy.engine.cursor.CursorResult.rowcount", rowcount
             ):
                 Parent, Child = self._fixture()
                 sess = Session()
@@ -1702,7 +1702,7 @@ class BasicStaleChecksTest(fixtures.MappedTest):
             config.db.dialect, "supports_sane_multi_rowcount", False
         ):
             with patch(
-                "sqlalchemy.engine.result.ResultProxy.rowcount", rowcount
+                "sqlalchemy.engine.cursor.CursorResult.rowcount", rowcount
             ):
                 Parent, Child = self._fixture()
                 sess = Session()
diff --git a/test/orm/test_versioning.py b/test/orm/test_versioning.py
index 1c540145b..2a7d09fad 100644
--- a/test/orm/test_versioning.py
+++ b/test/orm/test_versioning.py
@@ -451,7 +451,7 @@ class VersioningTest(fixtures.MappedTest):
 
         with patch.object(
             config.db.dialect, "supports_sane_multi_rowcount", False
-        ), patch("sqlalchemy.engine.result.ResultProxy.rowcount", rowcount):
+        ), patch("sqlalchemy.engine.cursor.CursorResult.rowcount", rowcount):
 
             Foo = self.classes.Foo
             s1 = self._fixture()
diff --git a/test/sql/test_compiler.py b/test/sql/test_compiler.py
index 440eaa05a..4b0b58b7e 100644
--- a/test/sql/test_compiler.py
+++ b/test/sql/test_compiler.py
@@ -5020,7 +5020,7 @@ class ResultMapTest(fixtures.TestBase):
         )
 
     def test_nested_api(self):
-        from sqlalchemy.engine.result import CursorResultMetaData
+        from sqlalchemy.engine.cursor import CursorResultMetaData
 
         stmt2 = select([table2]).subquery()
 
diff --git a/test/sql/test_resultset.py b/test/sql/test_resultset.py
index 1611dc1ba..a0c456056 100644
--- a/test/sql/test_resultset.py
+++ b/test/sql/test_resultset.py
@@ -25,8 +25,8 @@ from sqlalchemy import type_coerce
 from sqlalchemy import TypeDecorator
 from sqlalchemy import util
 from sqlalchemy import VARCHAR
+from sqlalchemy.engine import cursor as _cursor
 from sqlalchemy.engine import default
-from sqlalchemy.engine import result as _result
 from sqlalchemy.engine import Row
 from sqlalchemy.ext.compiler import compiles
 from sqlalchemy.future import select as future_select
@@ -43,7 +43,6 @@ from sqlalchemy.testing import eq_
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import in_
 from sqlalchemy.testing import is_
-from sqlalchemy.testing import is_false
 from sqlalchemy.testing import is_true
 from sqlalchemy.testing import le_
 from sqlalchemy.testing import ne_
@@ -55,7 +54,7 @@ from sqlalchemy.testing.schema import Table
 from sqlalchemy.util import collections_abc
 
 
-class ResultProxyTest(fixtures.TablesTest):
+class CursorResultTest(fixtures.TablesTest):
     __backend__ = True
 
     @classmethod
@@ -564,9 +563,9 @@ class ResultProxyTest(fixtures.TablesTest):
 
         # these proxies don't work with no cursor.description present.
         # so they don't apply to this test at the moment.
-        # result.FullyBufferedResultProxy,
-        # result.BufferedRowResultProxy,
-        # result.BufferedColumnResultProxy
+        # result.FullyBufferedCursorResult,
+        # result.BufferedRowCursorResult,
+        # result.BufferedColumnCursorResult
 
         users = self.tables.users
 
@@ -578,7 +577,9 @@ class ResultProxyTest(fixtures.TablesTest):
             lambda r: r.scalar(),
             lambda r: r.fetchmany(),
             lambda r: r._getter("user"),
-            lambda r: r._has_key("user"),
+            lambda r: r.keys(),
+            lambda r: r.columns("user"),
+            lambda r: r.cursor_strategy.fetchone(r),
         ]:
             trans = conn.begin()
             result = conn.execute(users.insert(), user_id=1)
@@ -648,11 +649,17 @@ class ResultProxyTest(fixtures.TablesTest):
         result = testing.db.execute(text("update users set user_id=5"))
         connection = result.connection
         assert connection.closed
+
         assert_raises_message(
             exc.ResourceClosedError,
             "This result object does not return rows.",
             result.fetchone,
         )
+        assert_raises_message(
+            exc.ResourceClosedError,
+            "This result object does not return rows.",
+            result.keys,
+        )
 
     def test_row_case_sensitive(self, connection):
         row = connection.execute(
@@ -788,16 +795,6 @@ class ResultProxyTest(fixtures.TablesTest):
             lambda: r._mapping["user_id"],
         )
 
-        result = connection.execute(users.outerjoin(addresses).select())
-        result = _result.BufferedColumnResultProxy(result.context)
-        r = result.first()
-        assert isinstance(r, _result.BufferedColumnRow)
-        assert_raises_message(
-            exc.InvalidRequestError,
-            "Ambiguous column name",
-            lambda: r._mapping["user_id"],
-        )
-
     @testing.requires.duplicate_names_in_cursor_description
     def test_ambiguous_column_by_col(self, connection):
         users = self.tables.users
@@ -1031,7 +1028,43 @@ class ResultProxyTest(fixtures.TablesTest):
         eq_(odict_row.values(), list(mapping_row.values()))
         eq_(odict_row.items(), list(mapping_row.items()))
 
-    def test_keys(self, connection):
+    @testing.combinations(
+        (lambda result: result),
+        (lambda result: result.first(),),
+        (lambda result: result.first()._mapping),
+        argnames="get_object",
+    )
+    def test_keys(self, connection, get_object):
+        users = self.tables.users
+        addresses = self.tables.addresses
+
+        connection.execute(users.insert(), user_id=1, user_name="foo")
+        result = connection.execute(users.select())
+
+        obj = get_object(result)
+
+        # Row still has a .keys() method as well as LegacyRow
+        # as in 1.3.x, the KeyedTuple object also had a keys() method.
+        # it emits a 2.0 deprecation warning.
+        keys = obj.keys()
+
+        # in 1.4, keys() is now a view that includes support for testing
+        # of columns and other objects
+        eq_(len(keys), 2)
+        eq_(list(keys), ["user_id", "user_name"])
+        eq_(keys, ["user_id", "user_name"])
+        ne_(keys, ["user_name", "user_id"])
+        in_("user_id", keys)
+        not_in_("foo", keys)
+        in_(users.c.user_id, keys)
+        not_in_(0, keys)
+        not_in_(addresses.c.user_id, keys)
+        not_in_(addresses.c.address, keys)
+
+        if isinstance(obj, Row):
+            eq_(obj._fields, ("user_id", "user_name"))
+
+    def test_row_mapping_keys(self, connection):
         users = self.tables.users
 
         connection.execute(users.insert(), user_id=1, user_name="foo")
@@ -1041,6 +1074,10 @@ class ResultProxyTest(fixtures.TablesTest):
         eq_(list(row._mapping.keys()), ["user_id", "user_name"])
         eq_(row._fields, ("user_id", "user_name"))
 
+        in_("user_id", row.keys())
+        not_in_("foo", row.keys())
+        in_(users.c.user_id, row.keys())
+
     def test_row_keys_legacy_dont_warn(self, connection):
         users = self.tables.users
 
@@ -1645,7 +1682,6 @@ class KeyTargetingTest(fixtures.TablesTest):
         )
 
         result = connection.execute(stmt)
-        is_false(result._metadata.matched_on_name)
 
         # ensure the result map is the same number of cols so we can
         # use positional targeting
@@ -2032,7 +2068,7 @@ class PositionalTextTest(fixtures.TablesTest):
         )
 
 
-class AlternateResultProxyTest(fixtures.TablesTest):
+class AlternateCursorResultTest(fixtures.TablesTest):
     __requires__ = ("sqlite",)
 
     @classmethod
@@ -2139,41 +2175,41 @@ class AlternateResultProxyTest(fixtures.TablesTest):
         )
 
     def test_basic_plain(self):
-        self._test_proxy(_result.DefaultCursorFetchStrategy)
+        self._test_proxy(_cursor.CursorFetchStrategy)
 
     def test_basic_buffered_row_result_proxy(self):
-        self._test_proxy(_result.BufferedRowCursorFetchStrategy)
+        self._test_proxy(_cursor.BufferedRowCursorFetchStrategy)
 
     def test_basic_fully_buffered_result_proxy(self):
-        self._test_proxy(_result.FullyBufferedCursorFetchStrategy)
+        self._test_proxy(_cursor.FullyBufferedCursorFetchStrategy)
 
     def test_basic_buffered_column_result_proxy(self):
-        self._test_proxy(_result.DefaultCursorFetchStrategy)
+        self._test_proxy(_cursor.CursorFetchStrategy)
 
     def test_resultprocessor_plain(self):
-        self._test_result_processor(_result.DefaultCursorFetchStrategy, False)
+        self._test_result_processor(_cursor.CursorFetchStrategy, False)
 
     def test_resultprocessor_plain_cached(self):
-        self._test_result_processor(_result.DefaultCursorFetchStrategy, True)
+        self._test_result_processor(_cursor.CursorFetchStrategy, True)
 
     def test_resultprocessor_buffered_row(self):
         self._test_result_processor(
-            _result.BufferedRowCursorFetchStrategy, False
+            _cursor.BufferedRowCursorFetchStrategy, False
         )
 
     def test_resultprocessor_buffered_row_cached(self):
         self._test_result_processor(
-            _result.BufferedRowCursorFetchStrategy, True
+            _cursor.BufferedRowCursorFetchStrategy, True
         )
 
     def test_resultprocessor_fully_buffered(self):
         self._test_result_processor(
-            _result.FullyBufferedCursorFetchStrategy, False
+            _cursor.FullyBufferedCursorFetchStrategy, False
         )
 
     def test_resultprocessor_fully_buffered_cached(self):
         self._test_result_processor(
-            _result.FullyBufferedCursorFetchStrategy, True
+            _cursor.FullyBufferedCursorFetchStrategy, True
         )
 
     def _test_result_processor(self, cls, use_cache):
@@ -2196,7 +2232,7 @@ class AlternateResultProxyTest(fixtures.TablesTest):
 
     @testing.fixture
     def row_growth_fixture(self):
-        with self._proxy_fixture(_result.BufferedRowCursorFetchStrategy):
+        with self._proxy_fixture(_cursor.BufferedRowCursorFetchStrategy):
             with self.engine.connect() as conn:
                 conn.execute(
                     self.table.insert(),
@@ -2237,10 +2273,230 @@ class AlternateResultProxyTest(fixtures.TablesTest):
                 assertion[idx] = result.cursor_strategy._bufsize
             le_(len(result.cursor_strategy._rowbuffer), max_size)
 
-        eq_(checks, assertion)
+    def test_buffered_fetchmany_fixed(self, row_growth_fixture):
+        """The BufferedRow cursor strategy will defer to the fetchmany
+        size passed when given rather than using the buffer growth
+        heuristic.
+
+        """
+        result = row_growth_fixture.execute(self.table.select())
+        eq_(len(result.cursor_strategy._rowbuffer), 1)
+
+        rows = result.fetchmany(300)
+        eq_(len(rows), 300)
+        eq_(len(result.cursor_strategy._rowbuffer), 0)
+
+        rows = result.fetchmany(300)
+        eq_(len(rows), 300)
+        eq_(len(result.cursor_strategy._rowbuffer), 0)
+
+        bufsize = result.cursor_strategy._bufsize
+        result.fetchone()
+
+        # the fetchone() caused it to buffer a full set of rows
+        eq_(len(result.cursor_strategy._rowbuffer), bufsize - 1)
+
+        # assert partitions uses fetchmany(), therefore controlling
+        # how the buffer is used
+        lens = []
+        for partition in result.partitions(180):
+            lens.append(len(partition))
+            eq_(len(result.cursor_strategy._rowbuffer), 0)
+
+        for lp in lens[0:-1]:
+            eq_(lp, 180)
+
+    def test_buffered_fetchmany_yield_per(self, connection):
+        table = self.tables.test
+
+        connection.execute(
+            table.insert(),
+            [{"x": i, "y": "t_%d" % i} for i in range(15, 3000)],
+        )
+
+        result = connection.execute(table.select())
+        assert isinstance(result.cursor_strategy, _cursor.CursorFetchStrategy)
 
+        result.fetchmany(5)
 
-class FutureResultTest(fixtures.FutureEngineMixin, fixtures.TablesTest):
+        result = result.yield_per(100)
+        assert isinstance(
+            result.cursor_strategy, _cursor.BufferedRowCursorFetchStrategy
+        )
+        eq_(result.cursor_strategy._bufsize, 100)
+        eq_(result.cursor_strategy._growth_factor, 0)
+        eq_(len(result.cursor_strategy._rowbuffer), 0)
+
+        result.fetchone()
+        eq_(len(result.cursor_strategy._rowbuffer), 99)
+
+        for i, row in enumerate(result):
+            if i == 188:
+                break
+
+        # buffer of 98, plus buffer of 99 - 89, 10 rows
+        eq_(len(result.cursor_strategy._rowbuffer), 10)
+
+    def test_buffered_fetchmany_yield_per_all(self, connection):
+        table = self.tables.test
+
+        connection.execute(
+            table.insert(),
+            [{"x": i, "y": "t_%d" % i} for i in range(15, 500)],
+        )
+
+        result = connection.execute(table.select())
+        assert isinstance(result.cursor_strategy, _cursor.CursorFetchStrategy)
+
+        result.fetchmany(5)
+
+        result = result.yield_per(0)
+        assert isinstance(
+            result.cursor_strategy, _cursor.BufferedRowCursorFetchStrategy
+        )
+        eq_(result.cursor_strategy._bufsize, 0)
+        eq_(result.cursor_strategy._growth_factor, 0)
+        eq_(len(result.cursor_strategy._rowbuffer), 0)
+
+        result.fetchone()
+        eq_(len(result.cursor_strategy._rowbuffer), 490)
+
+        for i, row in enumerate(result):
+            if i == 188:
+                break
+
+        eq_(len(result.cursor_strategy._rowbuffer), 301)
+
+        # already buffered, so this doesn't change things
+        result.yield_per(10)
+
+        result.fetchmany(5)
+        eq_(len(result.cursor_strategy._rowbuffer), 296)
+
+
+class MergeCursorResultTest(fixtures.TablesTest):
+    __backend__ = True
+
+    __requires__ = ("independent_cursors",)
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            "users",
+            metadata,
+            Column("user_id", INT, primary_key=True, autoincrement=False),
+            Column("user_name", VARCHAR(20)),
+            test_needs_acid=True,
+        )
+
+    @classmethod
+    def insert_data(cls, connection):
+        users = cls.tables.users
+
+        connection.execute(
+            users.insert(),
+            [
+                {"user_id": 7, "user_name": "u1"},
+                {"user_id": 8, "user_name": "u2"},
+                {"user_id": 9, "user_name": "u3"},
+                {"user_id": 10, "user_name": "u4"},
+                {"user_id": 11, "user_name": "u5"},
+                {"user_id": 12, "user_name": "u6"},
+            ],
+        )
+
+    @testing.fixture
+    def merge_fixture(self):
+        users = self.tables.users
+
+        def results(connection):
+
+            r1 = connection.execute(
+                users.select()
+                .where(users.c.user_id.in_([7, 8]))
+                .order_by(users.c.user_id)
+            )
+            r2 = connection.execute(
+                users.select()
+                .where(users.c.user_id.in_([9]))
+                .order_by(users.c.user_id)
+            )
+            r3 = connection.execute(
+                users.select()
+                .where(users.c.user_id.in_([10, 11]))
+                .order_by(users.c.user_id)
+            )
+            r4 = connection.execute(
+                users.select()
+                .where(users.c.user_id.in_([12]))
+                .order_by(users.c.user_id)
+            )
+            return r1, r2, r3, r4
+
+        return results
+
+    def test_merge_results(self, connection, merge_fixture):
+        r1, r2, r3, r4 = merge_fixture(connection)
+
+        result = r1.merge(r2, r3, r4)
+
+        eq_(result.keys(), ["user_id", "user_name"])
+        row = result.fetchone()
+        eq_(row, (7, "u1"))
+        result.close()
+
+    def test_close(self, connection, merge_fixture):
+        r1, r2, r3, r4 = merge_fixture(connection)
+
+        result = r1.merge(r2, r3, r4)
+
+        for r in [result, r1, r2, r3, r4]:
+            assert not r.closed
+
+        result.close()
+        for r in [result, r1, r2, r3, r4]:
+            assert r.closed
+
+    def test_fetchall(self, connection, merge_fixture):
+        r1, r2, r3, r4 = merge_fixture(connection)
+
+        result = r1.merge(r2, r3, r4)
+        eq_(
+            result.fetchall(),
+            [
+                (7, "u1"),
+                (8, "u2"),
+                (9, "u3"),
+                (10, "u4"),
+                (11, "u5"),
+                (12, "u6"),
+            ],
+        )
+        for r in [r1, r2, r3, r4]:
+            assert r._soft_closed
+
+    def test_first(self, connection, merge_fixture):
+        r1, r2, r3, r4 = merge_fixture(connection)
+
+        result = r1.merge(r2, r3, r4)
+        eq_(
+            result.first(), (7, "u1"),
+        )
+        for r in [r1, r2, r3, r4]:
+            assert r.closed
+
+    def test_columns(self, connection, merge_fixture):
+        r1, r2, r3, r4 = merge_fixture(connection)
+
+        result = r1.merge(r2, r3, r4)
+        eq_(
+            result.columns("user_name").fetchmany(4),
+            [("u1",), ("u2",), ("u3",), ("u4",)],
+        )
+        result.close()
+
+
+class GenerativeResultTest(fixtures.TablesTest):
     __backend__ = True
 
     @classmethod
@@ -2303,7 +2559,49 @@ class FutureResultTest(fixtures.FutureEngineMixin, fixtures.TablesTest):
         result = connection.execute(
             future_select(users).order_by(users.c.user_id)
         )
-        eq_(result.columns(*columns).all(), expected)
+
+        all_ = result.columns(*columns).all()
+        eq_(all_, expected)
+
+        # ensure Row / LegacyRow comes out with .columns
+        assert type(all_[0]) is result._process_row
+
+    def test_columns_twice(self, connection):
+        users = self.tables.users
+        connection.execute(
+            users.insert(),
+            [{"user_id": 7, "user_name": "jack", "x": 1, "y": 2}],
+        )
+
+        result = connection.execute(
+            future_select(users).order_by(users.c.user_id)
+        )
+
+        all_ = (
+            result.columns("x", "y", "user_name", "user_id")
+            .columns("user_name", "x")
+            .all()
+        )
+        eq_(all_, [("jack", 1)])
+
+        # ensure Row / LegacyRow comes out with .columns
+        assert type(all_[0]) is result._process_row
+
+    def test_columns_plus_getter(self, connection):
+        users = self.tables.users
+        connection.execute(
+            users.insert(),
+            [{"user_id": 7, "user_name": "jack", "x": 1, "y": 2}],
+        )
+
+        result = connection.execute(
+            future_select(users).order_by(users.c.user_id)
+        )
+
+        result = result.columns("x", "y", "user_name")
+        getter = result._metadata._getter("y")
+
+        eq_(getter(result.first()), 2)
 
     def test_partitions(self, connection):
         users = self.tables.users