Deprecate textual column matching in Row

Deprecate query.instances() without a context

Deprecate string alias with contains_eager()

Deprecated the behavior by which a :class:`.Column` can be used as the key
in a result set row lookup, when that :class:`.Column` is not part of the
SQL selectable that is being selected; that is, it is only matched on name.
A deprecation warning is now emitted for this case.   Various ORM use
cases, such as those involving :func:`.text` constructs, have been improved
so that this fallback logic is avoided in most cases.

Calling the :meth:`.Query.instances` method without passing a
:class:`.QueryContext` is deprecated.   The original use case for this was
that a :class:`.Query` could yield ORM objects when given only the entities
to be selected as well as a DBAPI cursor object.  However, for this to work
correctly there is essential metadata that is passed from a SQLAlchemy
:class:`.ResultProxy` that is derived from the mapped column expressions,
which comes originally from the :class:`.QueryContext`.   To retrieve ORM
results from arbitrary SELECT statements, the :meth:`.Query.from_statement`
method should be used.

Note there is a small bump in test_zoomark because the
column._label is being calculated for each of those columns within
baseline_3_properties, as it is now part of the result map.
This label can't be calculated when the column is attached
to the table because it needs to have all the columns present
to do this correctly.  Another approach here would be to
pre-load the _label before the test runs however the zoomark
tests don't have an easy place for this to happen and it's
not really worth it.

Fixes: #4877
Fixes: #4719
Change-Id: I9bd29e72e6dce7c855651d69ba68d7383469acbc

9 files changed, 150 insertions, 44 deletions

diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index 54f0043c4..4ef656817 100644
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -1783,6 +1783,11 @@ class MSSQLCompiler(compiler.SQLCompiler):
         return super(MSSQLCompiler, self).visit_binary(binary, **kwargs)
 
     def returning_clause(self, stmt, returning_cols):
+        # SQL server returning clause requires that the columns refer to
+        # the virtual table names "inserted" or "deleted".   Here, we make
+        # a simple alias of our table with that name, and then adapt the
+        # columns we have from the list of RETURNING columns to that new name
+        # so that they render as "inserted.<colname>" / "deleted.<colname>".
 
         if self.isinsert or self.isupdate:
             target = stmt.table.alias("inserted")
@@ -1791,9 +1796,21 @@ class MSSQLCompiler(compiler.SQLCompiler):
 
         adapter = sql_util.ClauseAdapter(target)
 
+        # adapter.traverse() takes a column from our target table and returns
+        # the one that is linked to the "inserted" / "deleted" tables.  So  in
+        # order to retrieve these values back from the result  (e.g. like
+        # row[column]), tell the compiler to also add the original unadapted
+        # column to the result map.   Before #4877, these were  (unknowingly)
+        # falling back using string name matching in the result set which
+        # necessarily used an expensive KeyError in order to match.
+
         columns = [
             self._label_select_column(
-                None, adapter.traverse(c), True, False, {}
+                None,
+                adapter.traverse(c),
+                True,
+                False,
+                {"result_map_targets": (c,)},
             )
             for c in expression._select_iterables(returning_cols)
         ]
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index eac593125..5a6e5c72e 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -653,6 +653,7 @@ class DefaultExecutionContext(interfaces.ExecutionContext):
             compiled._result_columns,
             compiled._ordered_columns,
             compiled._textual_ordered_columns,
+            compiled._loose_column_name_matching,
         )
 
         self.unicode_statement = util.text_type(compiled)
diff --git a/lib/sqlalchemy/engine/result.py b/lib/sqlalchemy/engine/result.py
index 90c884f94..af5303658 100644
--- a/lib/sqlalchemy/engine/result.py
+++ b/lib/sqlalchemy/engine/result.py
@@ -254,12 +254,15 @@ class ResultMetaData(object):
                 result_columns,
                 cols_are_ordered,
                 textual_ordered,
+                loose_column_name_matching,
             ) = context.result_column_struct
             num_ctx_cols = len(result_columns)
         else:
             result_columns = (
                 cols_are_ordered
-            ) = num_ctx_cols = textual_ordered = False
+            ) = (
+                num_ctx_cols
+            ) = loose_column_name_matching = textual_ordered = False
 
         # merge cursor.description with the column info
         # present in the compiled structure, if any
@@ -270,6 +273,7 @@ class ResultMetaData(object):
             num_ctx_cols,
             cols_are_ordered,
             textual_ordered,
+            loose_column_name_matching,
         )
 
         self._keymap = {}
@@ -388,6 +392,7 @@ class ResultMetaData(object):
         num_ctx_cols,
         cols_are_ordered,
         textual_ordered,
+        loose_column_name_matching,
     ):
         """Merge a cursor.description with compiled result column information.
 
@@ -482,7 +487,10 @@ class ResultMetaData(object):
                 # compiled SQL with a mismatch of description cols
                 # vs. compiled cols, or textual w/ unordered columns
                 raw_iterator = self._merge_cols_by_name(
-                    context, cursor_description, result_columns
+                    context,
+                    cursor_description,
+                    result_columns,
+                    loose_column_name_matching,
                 )
             else:
                 # no compiled SQL, just a raw string
@@ -587,13 +595,18 @@ class ResultMetaData(object):
 
             yield idx, colname, mapped_type, coltype, obj, untranslated
 
-    def _merge_cols_by_name(self, context, cursor_description, result_columns):
+    def _merge_cols_by_name(
+        self,
+        context,
+        cursor_description,
+        result_columns,
+        loose_column_name_matching,
+    ):
         dialect = context.dialect
         case_sensitive = dialect.case_sensitive
         match_map = self._create_description_match_map(
-            result_columns, case_sensitive
+            result_columns, case_sensitive, loose_column_name_matching
         )
-
         self.matched_on_name = True
         for (
             idx,
@@ -622,7 +635,10 @@ class ResultMetaData(object):
 
     @classmethod
     def _create_description_match_map(
-        cls, result_columns, case_sensitive=True
+        cls,
+        result_columns,
+        case_sensitive=True,
+        loose_column_name_matching=False,
     ):
         """when matching cursor.description to a set of names that are present
         in a Compiled object, as is the case with TextualSelect, get all the
@@ -631,22 +647,29 @@ class ResultMetaData(object):
 
         d = {}
         for elem in result_columns:
-            key, rec = (
-                elem[RM_RENDERED_NAME],
-                (elem[RM_NAME], elem[RM_OBJECTS], elem[RM_TYPE]),
-            )
+            key = elem[RM_RENDERED_NAME]
             if not case_sensitive:
                 key = key.lower()
             if key in d:
-                # conflicting keyname, just double up the list
-                # of objects.  this will cause an "ambiguous name"
-                # error if an attempt is made by the result set to
-                # access.
+                # conflicting keyname - just add the column-linked objects
+                # to the existing record.  if there is a duplicate column
+                # name in the cursor description, this will allow all of those
+                # objects to raise an ambiguous column error
                 e_name, e_obj, e_type = d[key]
-                d[key] = e_name, e_obj + rec[1], e_type
+                d[key] = e_name, e_obj + elem[RM_OBJECTS], e_type
             else:
-                d[key] = rec
-
+                d[key] = (elem[RM_NAME], elem[RM_OBJECTS], elem[RM_TYPE])
+
+            if loose_column_name_matching:
+                # when using a textual statement with an unordered set
+                # of columns that line up, we are expecting the user
+                # to be using label names in the SQL that match to the column
+                # expressions.  Enable more liberal matching for this case;
+                # duplicate keys that are ambiguous will be fixed later.
+                for r_key in elem[RM_OBJECTS]:
+                    d.setdefault(
+                        r_key, (elem[RM_NAME], elem[RM_OBJECTS], elem[RM_TYPE])
+                    )
         return d
 
     def _key_fallback(self, key, raiseerr=True):
@@ -688,6 +711,22 @@ class ResultMetaData(object):
                         break
                 else:
                     result = None
+            if result is not None:
+                if result[MD_OBJECTS] is _UNPICKLED:
+                    util.warn_deprecated(
+                        "Retreiving row values using Column objects from a "
+                        "row that was unpickled is deprecated; adequate "
+                        "state cannot be pickled for this to be efficient.   "
+                        "This usage will raise KeyError in a future release."
+                    )
+                else:
+                    util.warn_deprecated(
+                        "Retreiving row values using Column objects with only "
+                        "matching names as keys is deprecated, and will raise "
+                        "KeyError in a future release; only Column "
+                        "objects that are explicitly part of the statement "
+                        "object should be used."
+                    )
         if result is None:
             if raiseerr:
                 raise exc.NoSuchColumnError(
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index 0480bfb01..92f9ee952 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -3159,12 +3159,6 @@ class Query(Generative):
 
         """
         statement = coercions.expect(roles.SelectStatementRole, statement)
-
-        # TODO: coercions above should have this handled
-        assert isinstance(
-            statement, (expression.TextClause, expression.SelectBase)
-        )
-
         self._statement = statement
 
     def first(self):
@@ -3395,21 +3389,21 @@ class Query(Generative):
             ]
         ]
 
-    def instances(self, cursor, __context=None):
-        """Given a ResultProxy cursor as returned by connection.execute(),
-        return an ORM result as an iterator.
+    def instances(self, result_proxy, context=None):
+        """Return an ORM result given a :class:`.ResultProxy` and
+        :class:`.QueryContext`.
 
-        e.g.::
-
-            result = engine.execute("select * from users")
-            for u in session.query(User).instances(result):
-                print u
         """
-        context = __context
         if context is None:
+            util.warn_deprecated(
+                "Using the Query.instances() method without a context "
+                "is deprecated and will be disallowed in a future release.  "
+                "Please make use of :meth:`.Query.from_statement` "
+                "for linking ORM results to arbitrary select constructs."
+            )
             context = QueryContext(self)
 
-        return loading.instances(self, cursor, context)
+        return loading.instances(self, result_proxy, context)
 
     def merge_result(self, iterator, load=True):
         """Merge a result into this :class:`.Query` object's Session.
@@ -3824,6 +3818,22 @@ class Query(Generative):
         context = QueryContext(self)
 
         if context.statement is not None:
+            if isinstance(context.statement, expression.TextClause):
+                # setup for all entities, including contains_eager entities.
+                for entity in self._entities:
+                    entity.setup_context(self, context)
+                context.statement = expression.TextualSelect(
+                    context.statement,
+                    context.primary_columns,
+                    positional=False,
+                )
+            else:
+                # allow TextualSelect with implicit columns as well
+                # as select() with ad-hoc columns, see test_query::TextTest
+                self._from_obj_alias = sql.util.ColumnAdapter(
+                    context.statement, adapt_on_names=True
+                )
+
             return context
 
         context.labels = not for_statement or self._with_labels
@@ -4603,7 +4613,9 @@ class _ColumnEntity(_QueryEntity):
         if ("fetch_column", self) in context.attributes:
             column = context.attributes[("fetch_column", self)]
         else:
-            column = query._adapt_clause(self.column, False, True)
+            column = self.column
+            if query._from_obj_alias:
+                column = query._from_obj_alias.columns[column]
 
         if column._annotations:
             # annotated columns perform more slowly in compiler and
@@ -4697,6 +4709,7 @@ class QueryContext(object):
                 self.statement = query._statement.apply_labels()
             else:
                 self.statement = query._statement
+            self.order_by = None
         else:
             self.statement = None
             self.from_clause = query._from_obj
diff --git a/lib/sqlalchemy/orm/strategy_options.py b/lib/sqlalchemy/orm/strategy_options.py
index c50b7d041..00c91dab5 100644
--- a/lib/sqlalchemy/orm/strategy_options.py
+++ b/lib/sqlalchemy/orm/strategy_options.py
@@ -1012,6 +1012,13 @@ def contains_eager(loadopt, attr, alias=None):
         if not isinstance(alias, str):
             info = inspect(alias)
             alias = info.selectable
+        else:
+            util.warn_deprecated(
+                "Passing a string name for the 'alias' argument to "
+                "'contains_eager()` is deprecated, and will not work in a "
+                "future release.  Please use a sqlalchemy.alias() or "
+                "sqlalchemy.orm.aliased() construct."
+            )
 
     elif getattr(attr, "_of_type", None):
         ot = inspect(attr._of_type)
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 5e432a74c..1381e734c 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -490,6 +490,13 @@ class SQLCompiler(Compiled):
     True unless using an unordered TextualSelect.
     """
 
+    _loose_column_name_matching = False
+    """tell the result object that the SQL staement is textual, wants to match
+    up to Column objects, and may be using the ._label in the SELECT rather
+    than the base name.
+
+    """
+
     _numeric_binds = False
     """
     True if paramstyle is "numeric".  This paramstyle is trickier than
@@ -799,6 +806,7 @@ class SQLCompiler(Compiled):
         within_label_clause=False,
         within_columns_clause=False,
         render_label_as_label=None,
+        result_map_targets=(),
         **kw
     ):
         # only render labels within the columns clause
@@ -820,7 +828,7 @@ class SQLCompiler(Compiled):
                 add_to_result_map(
                     labelname,
                     label.name,
-                    (label, labelname) + label._alt_names,
+                    (label, labelname) + label._alt_names + result_map_targets,
                     label.type,
                 )
 
@@ -847,7 +855,12 @@ class SQLCompiler(Compiled):
         )
 
     def visit_column(
-        self, column, add_to_result_map=None, include_table=True, **kwargs
+        self,
+        column,
+        add_to_result_map=None,
+        include_table=True,
+        result_map_targets=(),
+        **kwargs
     ):
         name = orig_name = column.name
         if name is None:
@@ -859,7 +872,10 @@ class SQLCompiler(Compiled):
 
         if add_to_result_map is not None:
             add_to_result_map(
-                name, orig_name, (column, name, column.key), column.type
+                name,
+                orig_name,
+                (column, name, column.key, column._label) + result_map_targets,
+                column.type,
             )
 
         if is_literal:
@@ -948,6 +964,13 @@ class SQLCompiler(Compiled):
             self._ordered_columns = (
                 self._textual_ordered_columns
             ) = taf.positional
+
+            # enable looser result column matching when the SQL text links to
+            # Column objects by name only
+            self._loose_column_name_matching = not taf.positional and bool(
+                taf.column_args
+            )
+
             for c in taf.column_args:
                 self.process(
                     c,
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index 3045cb84e..8ee157b6f 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -4278,7 +4278,13 @@ class ColumnClause(roles.LabeledColumnExprRole, Immutable, ColumnElement):
                 label = quoted_name(label, t.name.quote)
 
             # ensure the label name doesn't conflict with that
-            # of an existing column
+            # of an existing column.   note that this implies that any
+            # Column must **not** set up its _label before its parent table
+            # has all of its other Column objects set up.  There are several
+            # tables in the test suite which will fail otherwise; example:
+            # table "owner" has columns "name" and "owner_name".  Therefore
+            # column owner.name cannot use the label "owner_name", it has
+            # to be "owner_name_1".
             if label in t.c:
                 _label = label
                 counter = 1
@@ -4339,7 +4345,6 @@ class ColumnClause(roles.LabeledColumnExprRole, Immutable, ColumnElement):
         c._proxies = [self]
         if selectable._is_clone_of is not None:
             c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
-
         return c.key, c
 
 
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index b41a77622..33ba95717 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -4449,7 +4449,10 @@ class TextualSelect(SelectBase):
 
     def __init__(self, text, columns, positional=False):
         self.element = text
-        self.column_args = columns
+        # convert for ORM attributes->columns, etc
+        self.column_args = [
+            coercions.expect(roles.ColumnsClauseRole, c) for c in columns
+        ]
         self.positional = positional
 
     @SelectBase._memoized_property
diff --git a/lib/sqlalchemy/sql/util.py b/lib/sqlalchemy/sql/util.py
index 780cdc7b2..5aeed0c1c 100644
--- a/lib/sqlalchemy/sql/util.py
+++ b/lib/sqlalchemy/sql/util.py
@@ -798,9 +798,7 @@ class ClauseAdapter(visitors.ReplacingCloningVisitor):
                 if newcol is not None:
                     return newcol
         if self.adapt_on_names and newcol is None:
-            # TODO: this should be changed to .exported_columns if and
-            # when we need to be able to adapt a plain Select statement
-            newcol = self.selectable.c.get(col.name)
+            newcol = self.selectable.exported_columns.get(col.name)
         return newcol
 
     def replace(self, col):