Fixed #18757, #14462, #21565 -- Reworked database-python type conversions

Complete rework of translating data values from database

Deprecation of SubfieldBase, removal of resolve_columns and
convert_values in favour of a more general converter based approach and
public API Field.from_db_value(). Now works seamlessly with aggregation,
.values() and raw queries.

Thanks to akaariai in particular for extensive advice and inspiration,
also to shaib, manfre and timograham for their reviews.

23 files changed, 259 insertions, 376 deletions

diff --git a/django/contrib/gis/db/backends/mysql/compiler.py b/django/contrib/gis/db/backends/mysql/compiler.py
deleted file mode 100644
index 74e666c165..0000000000
--- a/django/contrib/gis/db/backends/mysql/compiler.py
+++ /dev/null
@@ -1,41 +0,0 @@
-from django.contrib.gis.db.models.sql.compiler import GeoSQLCompiler as BaseGeoSQLCompiler
-from django.db.backends.mysql import compiler
-
-SQLCompiler = compiler.SQLCompiler
-
-
-class GeoSQLCompiler(BaseGeoSQLCompiler, SQLCompiler):
-    def resolve_columns(self, row, fields=()):
-        """
-        Integrate the cases handled both by the base GeoSQLCompiler and the
-        main MySQL compiler (converting 0/1 to True/False for boolean fields).
-
-        Refs #15169.
-
-        """
-        row = BaseGeoSQLCompiler.resolve_columns(self, row, fields)
-        return SQLCompiler.resolve_columns(self, row, fields)
-
-
-class SQLInsertCompiler(compiler.SQLInsertCompiler, GeoSQLCompiler):
-    pass
-
-
-class SQLDeleteCompiler(compiler.SQLDeleteCompiler, GeoSQLCompiler):
-    pass
-
-
-class SQLUpdateCompiler(compiler.SQLUpdateCompiler, GeoSQLCompiler):
-    pass
-
-
-class SQLAggregateCompiler(compiler.SQLAggregateCompiler, GeoSQLCompiler):
-    pass
-
-
-class SQLDateCompiler(compiler.SQLDateCompiler, GeoSQLCompiler):
-    pass
-
-
-class SQLDateTimeCompiler(compiler.SQLDateTimeCompiler, GeoSQLCompiler):
-    pass
diff --git a/django/contrib/gis/db/backends/mysql/operations.py b/django/contrib/gis/db/backends/mysql/operations.py
index 989b5a1292..60583a8e94 100644
--- a/django/contrib/gis/db/backends/mysql/operations.py
+++ b/django/contrib/gis/db/backends/mysql/operations.py
@@ -6,7 +6,7 @@ from django.contrib.gis.db.backends.base import BaseSpatialOperations
 
 class MySQLOperations(DatabaseOperations, BaseSpatialOperations):
 
-    compiler_module = 'django.contrib.gis.db.backends.mysql.compiler'
+    compiler_module = 'django.contrib.gis.db.models.sql.compiler'
     mysql = True
     name = 'mysql'
     select = 'AsText(%s)'
diff --git a/django/contrib/gis/db/models/fields.py b/django/contrib/gis/db/models/fields.py
index 6002f67ef3..2e0e13abf9 100644
--- a/django/contrib/gis/db/models/fields.py
+++ b/django/contrib/gis/db/models/fields.py
@@ -197,6 +197,11 @@ class GeometryField(Field):
         else:
             return geom
 
+    def from_db_value(self, value, connection):
+        if value is not None:
+            value = Geometry(value)
+        return value
+
     def get_srid(self, geom):
         """
         Returns the default SRID for the given geometry, taking into account
diff --git a/django/contrib/gis/db/models/query.py b/django/contrib/gis/db/models/query.py
index eb1d9fdeb4..a975b110b2 100644
--- a/django/contrib/gis/db/models/query.py
+++ b/django/contrib/gis/db/models/query.py
@@ -1,5 +1,5 @@
 from django.db import connections
-from django.db.models.query import QuerySet, ValuesQuerySet, ValuesListQuerySet
+from django.db.models.query import QuerySet
 
 from django.contrib.gis.db.models import aggregates
 from django.contrib.gis.db.models.fields import get_srid_info, PointField, LineStringField
@@ -18,19 +18,6 @@ class GeoQuerySet(QuerySet):
         super(GeoQuerySet, self).__init__(model=model, query=query, using=using, hints=hints)
         self.query = query or GeoQuery(self.model)
 
-    def values(self, *fields):
-        return self._clone(klass=GeoValuesQuerySet, setup=True, _fields=fields)
-
-    def values_list(self, *fields, **kwargs):
-        flat = kwargs.pop('flat', False)
-        if kwargs:
-            raise TypeError('Unexpected keyword arguments to values_list: %s'
-                    % (list(kwargs),))
-        if flat and len(fields) > 1:
-            raise TypeError("'flat' is not valid when values_list is called with more than one field.")
-        return self._clone(klass=GeoValuesListQuerySet, setup=True, flat=flat,
-                           _fields=fields)
-
     ### GeoQuerySet Methods ###
     def area(self, tolerance=0.05, **kwargs):
         """
@@ -767,16 +754,3 @@ class GeoQuerySet(QuerySet):
             return self.query.get_compiler(self.db)._field_column(geo_field, parent_model._meta.db_table)
         else:
             return self.query.get_compiler(self.db)._field_column(geo_field)
-
-
-class GeoValuesQuerySet(ValuesQuerySet):
-    def __init__(self, *args, **kwargs):
-        super(GeoValuesQuerySet, self).__init__(*args, **kwargs)
-        # This flag tells `resolve_columns` to run the values through
-        # `convert_values`.  This ensures that Geometry objects instead
-        # of string values are returned with `values()` or `values_list()`.
-        self.query.geo_values = True
-
-
-class GeoValuesListQuerySet(GeoValuesQuerySet, ValuesListQuerySet):
-    pass
diff --git a/django/contrib/gis/db/models/sql/compiler.py b/django/contrib/gis/db/models/sql/compiler.py
index 351a3171b0..e721514085 100644
--- a/django/contrib/gis/db/models/sql/compiler.py
+++ b/django/contrib/gis/db/models/sql/compiler.py
@@ -1,12 +1,6 @@
-import datetime
-
-from django.conf import settings
-from django.db.backends.utils import truncate_name, typecast_date, typecast_timestamp
+from django.db.backends.utils import truncate_name
 from django.db.models.sql import compiler
-from django.db.models.sql.constants import MULTI
 from django.utils import six
-from django.utils.six.moves import zip, zip_longest
-from django.utils import timezone
 
 SQLCompiler = compiler.SQLCompiler
 
@@ -153,38 +147,13 @@ class GeoSQLCompiler(compiler.SQLCompiler):
                     col_aliases.add(field.column)
         return result, aliases
 
-    def resolve_columns(self, row, fields=()):
-        """
-        This routine is necessary so that distances and geometries returned
-        from extra selection SQL get resolved appropriately into Python
-        objects.
-        """
-        values = []
-        aliases = list(self.query.extra_select)
-
-        # Have to set a starting row number offset that is used for
-        # determining the correct starting row index -- needed for
-        # doing pagination with Oracle.
-        rn_offset = 0
-        if self.connection.ops.oracle:
-            if self.query.high_mark is not None or self.query.low_mark:
-                rn_offset = 1
-        index_start = rn_offset + len(aliases)
-
-        # Converting any extra selection values (e.g., geometries and
-        # distance objects added by GeoQuerySet methods).
-        values = [self.query.convert_values(v,
-                               self.query.extra_select_fields.get(a, None),
-                               self.connection)
-                  for v, a in zip(row[rn_offset:index_start], aliases)]
-        if self.connection.ops.oracle or getattr(self.query, 'geo_values', False):
-            # We resolve the rest of the columns if we're on Oracle or if
-            # the `geo_values` attribute is defined.
-            for value, field in zip_longest(row[index_start:], fields):
-                values.append(self.query.convert_values(value, field, self.connection))
-        else:
-            values.extend(row[index_start:])
-        return tuple(values)
+    def get_converters(self, fields):
+        converters = super(GeoSQLCompiler, self).get_converters(fields)
+        for i, alias in enumerate(self.query.extra_select):
+            field = self.query.extra_select_fields.get(alias)
+            if field:
+                converters[i] = ([], [field.from_db_value], field)
+        return converters
 
     #### Routines unique to GeoQuery ####
     def get_extra_select_format(self, alias):
@@ -268,55 +237,8 @@ class SQLAggregateCompiler(compiler.SQLAggregateCompiler, GeoSQLCompiler):
 
 
 class SQLDateCompiler(compiler.SQLDateCompiler, GeoSQLCompiler):
-    """
-    This is overridden for GeoDjango to properly cast date columns, since
-    `GeoQuery.resolve_columns` is used for spatial values.
-    See #14648, #16757.
-    """
-    def results_iter(self):
-        if self.connection.ops.oracle:
-            from django.db.models.fields import DateTimeField
-            fields = [DateTimeField()]
-        else:
-            needs_string_cast = self.connection.features.needs_datetime_string_cast
-
-        offset = len(self.query.extra_select)
-        for rows in self.execute_sql(MULTI):
-            for row in rows:
-                date = row[offset]
-                if self.connection.ops.oracle:
-                    date = self.resolve_columns(row, fields)[offset]
-                elif needs_string_cast:
-                    date = typecast_date(str(date))
-                if isinstance(date, datetime.datetime):
-                    date = date.date()
-                yield date
+    pass
 
 
 class SQLDateTimeCompiler(compiler.SQLDateTimeCompiler, GeoSQLCompiler):
-    """
-    This is overridden for GeoDjango to properly cast date columns, since
-    `GeoQuery.resolve_columns` is used for spatial values.
-    See #14648, #16757.
-    """
-    def results_iter(self):
-        if self.connection.ops.oracle:
-            from django.db.models.fields import DateTimeField
-            fields = [DateTimeField()]
-        else:
-            needs_string_cast = self.connection.features.needs_datetime_string_cast
-
-        offset = len(self.query.extra_select)
-        for rows in self.execute_sql(MULTI):
-            for row in rows:
-                datetime = row[offset]
-                if self.connection.ops.oracle:
-                    datetime = self.resolve_columns(row, fields)[offset]
-                elif needs_string_cast:
-                    datetime = typecast_timestamp(str(datetime))
-                # Datetimes are artificially returned in UTC on databases that
-                # don't support time zone. Restore the zone used in the query.
-                if settings.USE_TZ:
-                    datetime = datetime.replace(tzinfo=None)
-                    datetime = timezone.make_aware(datetime, self.query.tzinfo)
-                yield datetime
+    pass
diff --git a/django/contrib/gis/db/models/sql/conversion.py b/django/contrib/gis/db/models/sql/conversion.py
index ee4f0e9e33..2908828b80 100644
--- a/django/contrib/gis/db/models/sql/conversion.py
+++ b/django/contrib/gis/db/models/sql/conversion.py
@@ -1,32 +1,53 @@
 """
-This module holds simple classes used by GeoQuery.convert_values
-to convert geospatial values from the database.
+This module holds simple classes to convert geospatial values from the
+database.
 """
 
+from django.contrib.gis.geometry.backend import Geometry
+from django.contrib.gis.measure import Area, Distance
+
 
 class BaseField(object):
     empty_strings_allowed = True
 
-    def get_internal_type(self):
-        "Overloaded method so OracleQuery.convert_values doesn't balk."
-        return None
-
 
 class AreaField(BaseField):
     "Wrapper for Area values."
     def __init__(self, area_att):
         self.area_att = area_att
 
+    def from_db_value(self, value, connection):
+        if value is not None:
+            value = Area(**{self.area_att: value})
+        return value
+
+    def get_internal_type(self):
+        return 'AreaField'
+
 
 class DistanceField(BaseField):
     "Wrapper for Distance values."
     def __init__(self, distance_att):
         self.distance_att = distance_att
 
+    def from_db_value(self, value, connection):
+        if value is not None:
+            value = Distance(**{self.distance_att: value})
+        return value
+
+    def get_internal_type(self):
+        return 'DistanceField'
+
 
 class GeomField(BaseField):
     """
     Wrapper for Geometry values.  It is a lightweight alternative to
     using GeometryField (which requires an SQL query upon instantiation).
     """
-    pass
+    def from_db_value(self, value, connection):
+        if value is not None:
+            value = Geometry(value)
+        return value
+
+    def get_internal_type(self):
+        return 'GeometryField'
diff --git a/django/contrib/gis/db/models/sql/query.py b/django/contrib/gis/db/models/sql/query.py
index b3be7a00f1..e82f17b8a3 100644
--- a/django/contrib/gis/db/models/sql/query.py
+++ b/django/contrib/gis/db/models/sql/query.py
@@ -5,9 +5,7 @@ from django.contrib.gis.db.models.constants import ALL_TERMS
 from django.contrib.gis.db.models.fields import GeometryField
 from django.contrib.gis.db.models.lookups import GISLookup
 from django.contrib.gis.db.models.sql import aggregates as gis_aggregates
-from django.contrib.gis.db.models.sql.conversion import AreaField, DistanceField, GeomField
-from django.contrib.gis.geometry.backend import Geometry
-from django.contrib.gis.measure import Area, Distance
+from django.contrib.gis.db.models.sql.conversion import GeomField
 
 
 class GeoQuery(sql.Query):
@@ -38,32 +36,6 @@ class GeoQuery(sql.Query):
         obj.extra_select_fields = self.extra_select_fields.copy()
         return obj
 
-    def convert_values(self, value, field, connection):
-        """
-        Using the same routines that Oracle does we can convert our
-        extra selection objects into Geometry and Distance objects.
-        TODO: Make converted objects 'lazy' for less overhead.
-        """
-        if connection.ops.oracle:
-            # Running through Oracle's first.
-            value = super(GeoQuery, self).convert_values(value, field or GeomField(), connection)
-
-        if value is None:
-            # Output from spatial function is NULL (e.g., called
-            # function on a geometry field with NULL value).
-            pass
-        elif isinstance(field, DistanceField):
-            # Using the field's distance attribute, can instantiate
-            # `Distance` with the right context.
-            value = Distance(**{field.distance_att: value})
-        elif isinstance(field, AreaField):
-            value = Area(**{field.area_att: value})
-        elif isinstance(field, (GeomField, GeometryField)) and value:
-            value = Geometry(value)
-        elif field is not None:
-            return super(GeoQuery, self).convert_values(value, field, connection)
-        return value
-
     def get_aggregation(self, using, force_subq=False):
         # Remove any aggregates marked for reduction from the subquery
         # and move them to the outer AggregateQuery.
diff --git a/django/contrib/gis/tests/geoapp/models.py b/django/contrib/gis/tests/geoapp/models.py
index 415dd568ad..59e54272fd 100644
--- a/django/contrib/gis/tests/geoapp/models.py
+++ b/django/contrib/gis/tests/geoapp/models.py
@@ -66,3 +66,18 @@ class MinusOneSRID(models.Model):
 
     class Meta:
         app_label = 'geoapp'
+
+
+class NonConcreteField(models.IntegerField):
+
+    def db_type(self, connection):
+        return None
+
+    def get_attname_column(self):
+        attname, column = super(NonConcreteField, self).get_attname_column()
+        return attname, None
+
+
+class NonConcreteModel(NamedModel):
+    non_concrete = NonConcreteField()
+    point = models.PointField(geography=True)
diff --git a/django/contrib/gis/tests/geoapp/tests.py b/django/contrib/gis/tests/geoapp/tests.py
index 8fd912fb9e..6b4206a40f 100644
--- a/django/contrib/gis/tests/geoapp/tests.py
+++ b/django/contrib/gis/tests/geoapp/tests.py
@@ -13,9 +13,7 @@ from django.utils import six
 if HAS_GEOS:
     from django.contrib.gis.geos import (fromstr, GEOSGeometry,
         Point, LineString, LinearRing, Polygon, GeometryCollection)
-
-    from .models import Country, City, PennsylvaniaCity, State, Track
-    from .models import Feature, MinusOneSRID
+    from .models import Country, City, PennsylvaniaCity, State, Track, NonConcreteModel, Feature, MinusOneSRID
 
 
 def postgis_bug_version():
@@ -754,10 +752,5 @@ class GeoQuerySetTest(TestCase):
         self.assertEqual(None, qs.unionagg(field_name='point'))
 
     def test_non_concrete_field(self):
-        pkfield = City._meta.get_field_by_name('id')[0]
-        orig_pkfield_col = pkfield.column
-        pkfield.column = None
-        try:
-            list(City.objects.all())
-        finally:
-            pkfield.column = orig_pkfield_col
+        NonConcreteModel.objects.create(point=Point(0, 0), name='name')
+        list(NonConcreteModel.objects.all())
diff --git a/django/contrib/gis/tests/relatedapp/models.py b/django/contrib/gis/tests/relatedapp/models.py
index 3b4364f7b9..5221bbe0b1 100644
--- a/django/contrib/gis/tests/relatedapp/models.py
+++ b/django/contrib/gis/tests/relatedapp/models.py
@@ -71,3 +71,8 @@ class Article(SimpleModel):
 class Book(SimpleModel):
     title = models.CharField(max_length=100)
     author = models.ForeignKey(Author, related_name='books', null=True)
+
+
+class Event(SimpleModel):
+    name = models.CharField(max_length=100)
+    when = models.DateTimeField()
diff --git a/django/contrib/gis/tests/relatedapp/tests.py b/django/contrib/gis/tests/relatedapp/tests.py
index 67dd07b583..3f856ca508 100644
--- a/django/contrib/gis/tests/relatedapp/tests.py
+++ b/django/contrib/gis/tests/relatedapp/tests.py
@@ -4,13 +4,15 @@ from django.contrib.gis.geos import HAS_GEOS
 from django.contrib.gis.tests.utils import no_oracle
 from django.db import connection
 from django.test import TestCase, skipUnlessDBFeature
+from django.test.utils import override_settings
+from django.utils import timezone
 
 if HAS_GEOS:
     from django.contrib.gis.db.models import Collect, Count, Extent, F, Union
     from django.contrib.gis.geometry.backend import Geometry
     from django.contrib.gis.geos import GEOSGeometry, Point, MultiPoint
 
-    from .models import City, Location, DirectoryEntry, Parcel, Book, Author, Article
+    from .models import City, Location, DirectoryEntry, Parcel, Book, Author, Article, Event
 
 
 @skipUnlessDBFeature("gis_enabled")
@@ -183,6 +185,12 @@ class RelatedGeoModelTest(TestCase):
             self.assertEqual(m.point, d['point'])
             self.assertEqual(m.point, t[1])
 
+    @override_settings(USE_TZ=True)
+    def test_07b_values(self):
+        "Testing values() and values_list() with aware datetime. See #21565."
+        Event.objects.create(name="foo", when=timezone.now())
+        list(Event.objects.values_list('when'))
+
     def test08_defer_only(self):
         "Testing defer() and only() on Geographic models."
         qs = Location.objects.all()
diff --git a/django/db/backends/__init__.py b/django/db/backends/__init__.py
index ca58bce3c0..70e024a19b 100644
--- a/django/db/backends/__init__.py
+++ b/django/db/backends/__init__.py
@@ -1190,20 +1190,13 @@ class BaseDatabaseOperations(object):
             second = timezone.make_aware(second, tz)
         return [first, second]
 
-    def convert_values(self, value, field):
-        """
-        Coerce the value returned by the database backend into a consistent type
-        that is compatible with the field type.
+    def get_db_converters(self, internal_type):
+        """Get a list of functions needed to convert field data.
+
+        Some field types on some backends do not provide data in the correct
+        format, this is the hook for coverter functions.
         """
-        if value is None or field is None:
-            return value
-        internal_type = field.get_internal_type()
-        if internal_type == 'FloatField':
-            return float(value)
-        elif (internal_type and (internal_type.endswith('IntegerField')
-                                 or internal_type == 'AutoField')):
-            return int(value)
-        return value
+        return []
 
     def check_aggregate_support(self, aggregate_func):
         """Check that the backend supports the provided aggregate
diff --git a/django/db/backends/mysql/base.py b/django/db/backends/mysql/base.py
index a7fe9b6104..9ab6fbf69c 100644
--- a/django/db/backends/mysql/base.py
+++ b/django/db/backends/mysql/base.py
@@ -394,6 +394,17 @@ class DatabaseOperations(BaseDatabaseOperations):
             return 'POW(%s)' % ','.join(sub_expressions)
         return super(DatabaseOperations, self).combine_expression(connector, sub_expressions)
 
+    def get_db_converters(self, internal_type):
+        converters = super(DatabaseOperations, self).get_db_converters(internal_type)
+        if internal_type in ['BooleanField', 'NullBooleanField']:
+            converters.append(self.convert_booleanfield_value)
+        return converters
+
+    def convert_booleanfield_value(self, value, field):
+        if value in (0, 1):
+            value = bool(value)
+        return value
+
 
 class DatabaseWrapper(BaseDatabaseWrapper):
     vendor = 'mysql'
diff --git a/django/db/backends/mysql/compiler.py b/django/db/backends/mysql/compiler.py
index 85c045fff2..56bea51add 100644
--- a/django/db/backends/mysql/compiler.py
+++ b/django/db/backends/mysql/compiler.py
@@ -1,18 +1,7 @@
 from django.db.models.sql import compiler
-from django.utils.six.moves import zip_longest
 
 
 class SQLCompiler(compiler.SQLCompiler):
-    def resolve_columns(self, row, fields=()):
-        values = []
-        index_extra_select = len(self.query.extra_select)
-        for value, field in zip_longest(row[index_extra_select:], fields):
-            if (field and field.get_internal_type() in ("BooleanField", "NullBooleanField") and
-                    value in (0, 1)):
-                value = bool(value)
-            values.append(value)
-        return row[:index_extra_select] + tuple(values)
-
     def as_subquery_condition(self, alias, columns, qn):
         qn2 = self.connection.ops.quote_name
         sql, params = self.as_sql()
diff --git a/django/db/backends/mysql/validation.py b/django/db/backends/mysql/validation.py
index cd7503a352..d69cb7e3e0 100644
--- a/django/db/backends/mysql/validation.py
+++ b/django/db/backends/mysql/validation.py
@@ -17,6 +17,10 @@ class DatabaseValidation(BaseDatabaseValidation):
         if getattr(field, 'rel', None) is None:
             field_type = field.db_type(connection)
 
+            # Ignore any non-concrete fields
+            if field_type is None:
+                return errors
+
             if (field_type.startswith('varchar')  # Look for CharFields...
                     and field.unique  # ... that are unique
                     and (field.max_length is None or int(field.max_length) > 255)):
diff --git a/django/db/backends/oracle/base.py b/django/db/backends/oracle/base.py
index 5ceddf40d8..596e326110 100644
--- a/django/db/backends/oracle/base.py
+++ b/django/db/backends/oracle/base.py
@@ -250,49 +250,64 @@ WHEN (new.%(col_name)s IS NULL)
             sql = field_name    # Cast to DATE removes sub-second precision.
         return sql, []
 
-    def convert_values(self, value, field):
-        if isinstance(value, Database.LOB):
-            value = value.read()
-            if field and field.get_internal_type() == 'TextField':
-                value = force_text(value)
+    def get_db_converters(self, internal_type):
+        converters = super(DatabaseOperations, self).get_db_converters(internal_type)
+        if internal_type == 'TextField':
+            converters.append(self.convert_textfield_value)
+        elif internal_type == 'BinaryField':
+            converters.append(self.convert_binaryfield_value)
+        elif internal_type in ['BooleanField', 'NullBooleanField']:
+            converters.append(self.convert_booleanfield_value)
+        elif internal_type == 'DecimalField':
+            converters.append(self.convert_decimalfield_value)
+        elif internal_type == 'DateField':
+            converters.append(self.convert_datefield_value)
+        elif internal_type == 'TimeField':
+            converters.append(self.convert_timefield_value)
+        converters.append(self.convert_empty_values)
+        return converters
 
+    def convert_empty_values(self, value, field):
         # Oracle stores empty strings as null. We need to undo this in
         # order to adhere to the Django convention of using the empty
         # string instead of null, but only if the field accepts the
         # empty string.
-        if value is None and field and field.empty_strings_allowed:
+        if value is None and field.empty_strings_allowed:
+            value = ''
             if field.get_internal_type() == 'BinaryField':
                 value = b''
-            else:
-                value = ''
-        # Convert 1 or 0 to True or False
-        elif value in (1, 0) and field and field.get_internal_type() in ('BooleanField', 'NullBooleanField'):
+        return value
+
+    def convert_textfield_value(self, value, field):
+        if isinstance(value, Database.LOB):
+            value = force_text(value.read())
+        return value
+
+    def convert_binaryfield_value(self, value, field):
+        if isinstance(value, Database.LOB):
+            value = force_bytes(value.read())
+        return value
+
+    def convert_booleanfield_value(self, value, field):
+        if value in (1, 0):
             value = bool(value)
-        # Force floats to the correct type
-        elif value is not None and field and field.get_internal_type() == 'FloatField':
-            value = float(value)
-        # Convert floats to decimals
-        elif value is not None and field and field.get_internal_type() == 'DecimalField':
+        return value
+
+    def convert_decimalfield_value(self, value, field):
+        if value is not None:
             value = backend_utils.typecast_decimal(field.format_number(value))
-        # cx_Oracle always returns datetime.datetime objects for
-        # DATE and TIMESTAMP columns, but Django wants to see a
-        # python datetime.date, .time, or .datetime.  We use the type
-        # of the Field to determine which to cast to, but it's not
-        # always available.
-        # As a workaround, we cast to date if all the time-related
-        # values are 0, or to time if the date is 1/1/1900.
-        # This could be cleaned a bit by adding a method to the Field
-        # classes to normalize values from the database (the to_python
-        # method is used for validation and isn't what we want here).
-        elif isinstance(value, Database.Timestamp):
-            if field and field.get_internal_type() == 'DateTimeField':
-                pass
-            elif field and field.get_internal_type() == 'DateField':
-                value = value.date()
-            elif field and field.get_internal_type() == 'TimeField' or (value.year == 1900 and value.month == value.day == 1):
-                value = value.time()
-            elif value.hour == value.minute == value.second == value.microsecond == 0:
-                value = value.date()
+        return value
+
+    # cx_Oracle always returns datetime.datetime objects for
+    # DATE and TIMESTAMP columns, but Django wants to see a
+    # python datetime.date, .time, or .datetime.
+    def convert_datefield_value(self, value, field):
+        if isinstance(value, Database.Timestamp):
+            return value.date()
+
+    def convert_timefield_value(self, value, field):
+        if isinstance(value, Database.Timestamp):
+            value = value.time()
         return value
 
     def deferrable_sql(self):
diff --git a/django/db/backends/oracle/compiler.py b/django/db/backends/oracle/compiler.py
index 0e4af85f22..0ce833a237 100644
--- a/django/db/backends/oracle/compiler.py
+++ b/django/db/backends/oracle/compiler.py
@@ -1,23 +1,7 @@
 from django.db.models.sql import compiler
-from django.utils.six.moves import zip_longest
 
 
 class SQLCompiler(compiler.SQLCompiler):
-    def resolve_columns(self, row, fields=()):
-        # If this query has limit/offset information, then we expect the
-        # first column to be an extra "_RN" column that we need to throw
-        # away.
-        if self.query.high_mark is not None or self.query.low_mark:
-            rn_offset = 1
-        else:
-            rn_offset = 0
-        index_start = rn_offset + len(self.query.extra_select)
-        values = [self.query.convert_values(v, None, connection=self.connection)
-                  for v in row[rn_offset:index_start]]
-        for value, field in zip_longest(row[index_start:], fields):
-            values.append(self.query.convert_values(value, field, connection=self.connection))
-        return tuple(values)
-
     def as_sql(self, with_limits=True, with_col_aliases=False):
         """
         Creates the SQL for this query. Returns the SQL string and list
@@ -48,7 +32,7 @@ class SQLCompiler(compiler.SQLCompiler):
             high_where = ''
             if self.query.high_mark is not None:
                 high_where = 'WHERE ROWNUM <= %d' % (self.query.high_mark,)
-            sql = 'SELECT * FROM (SELECT ROWNUM AS "_RN", "_SUB".* FROM (%s) "_SUB" %s) WHERE "_RN" > %d' % (sql, high_where, self.query.low_mark)
+            sql = 'SELECT * FROM (SELECT "_SUB".*, ROWNUM AS "_RN" FROM (%s) "_SUB" %s) WHERE "_RN" > %d' % (sql, high_where, self.query.low_mark)
 
         return sql, params
 
diff --git a/django/db/backends/sqlite3/base.py b/django/db/backends/sqlite3/base.py
index 3b61a8a1e5..42c1e5e177 100644
--- a/django/db/backends/sqlite3/base.py
+++ b/django/db/backends/sqlite3/base.py
@@ -263,27 +263,36 @@ class DatabaseOperations(BaseDatabaseOperations):
 
         return six.text_type(value)
 
-    def convert_values(self, value, field):
-        """SQLite returns floats when it should be returning decimals,
-        and gets dates and datetimes wrong.
-        For consistency with other backends, coerce when required.
-        """
-        if value is None:
-            return None
-
-        internal_type = field.get_internal_type()
-        if internal_type == 'DecimalField':
-            return backend_utils.typecast_decimal(field.format_number(value))
-        elif internal_type and internal_type.endswith('IntegerField') or internal_type == 'AutoField':
-            return int(value)
+    def get_db_converters(self, internal_type):
+        converters = super(DatabaseOperations, self).get_db_converters(internal_type)
+        if internal_type == 'DateTimeField':
+            converters.append(self.convert_datetimefield_value)
         elif internal_type == 'DateField':
-            return parse_date(value)
-        elif internal_type == 'DateTimeField':
-            return parse_datetime_with_timezone_support(value)
+            converters.append(self.convert_datefield_value)
         elif internal_type == 'TimeField':
-            return parse_time(value)
+            converters.append(self.convert_timefield_value)
+        elif internal_type == 'DecimalField':
+            converters.append(self.convert_decimalfield_value)
+        return converters
+
+    def convert_decimalfield_value(self, value, field):
+        if value is not None:
+            value = backend_utils.typecast_decimal(field.format_number(value))
+        return value
+
+    def convert_datefield_value(self, value, field):
+        if value is not None and not isinstance(value, datetime.date):
+            value = parse_date(value)
+        return value
+
+    def convert_datetimefield_value(self, value, field):
+        if value is not None and not isinstance(value, datetime.datetime):
+            value = parse_datetime_with_timezone_support(value)
+        return value
 
-        # No field, or the field isn't known to be a decimal or integer
+    def convert_timefield_value(self, value, field):
+        if value is not None and not isinstance(value, datetime.time):
+            value = parse_time(value)
         return value
 
     def bulk_insert_sql(self, fields, num_values):
diff --git a/django/db/models/fields/__init__.py b/django/db/models/fields/__init__.py
index 4509f4139c..c41d4eaad2 100644
--- a/django/db/models/fields/__init__.py
+++ b/django/db/models/fields/__init__.py
@@ -558,6 +558,11 @@ class Field(RegisterLookupMixin):
     def db_type_suffix(self, connection):
         return connection.creation.data_types_suffix.get(self.get_internal_type())
 
+    def get_db_converters(self, connection):
+        if hasattr(self, 'from_db_value'):
+            return [self.from_db_value]
+        return []
+
     @property
     def unique(self):
         return self._unique or self.primary_key
diff --git a/django/db/models/fields/subclassing.py b/django/db/models/fields/subclassing.py
index c87cc144d0..4233106429 100644
--- a/django/db/models/fields/subclassing.py
+++ b/django/db/models/fields/subclassing.py
@@ -7,6 +7,10 @@ to_python() and the other necessary methods and everything will work
 seamlessly.
 """
 
+import warnings
+
+from django.utils.deprecation import RemovedInDjango20Warning
+
 
 class SubfieldBase(type):
     """
@@ -14,6 +18,9 @@ class SubfieldBase(type):
     has the descriptor protocol attached to it.
     """
     def __new__(cls, name, bases, attrs):
+        warnings.warn("SubfieldBase has been deprecated. Use Field.from_db_value instead.",
+                  RemovedInDjango20Warning)
+
         new_class = super(SubfieldBase, cls).__new__(cls, name, bases, attrs)
         new_class.contribute_to_class = make_contrib(
             new_class, attrs.get('contribute_to_class')
diff --git a/django/db/models/query.py b/django/db/models/query.py
index f96cce2713..be5892a371 100644
--- a/django/db/models/query.py
+++ b/django/db/models/query.py
@@ -1560,7 +1560,6 @@ class RawQuerySet(object):
         compiler = connections[db].ops.compiler('SQLCompiler')(
             self.query, connections[db], db
         )
-        need_resolv_columns = hasattr(compiler, 'resolve_columns')
 
         query = iter(self.query)
 
@@ -1578,11 +1577,11 @@ class RawQuerySet(object):
                 model_cls = deferred_class_factory(self.model, skip)
             else:
                 model_cls = self.model
-            if need_resolv_columns:
-                fields = [self.model_fields.get(c, None) for c in self.columns]
+            fields = [self.model_fields.get(c, None) for c in self.columns]
+            converters = compiler.get_converters(fields)
             for values in query:
-                if need_resolv_columns:
-                    values = compiler.resolve_columns(values, fields)
+                if converters:
+                    values = compiler.apply_converters(values, converters)
                 # Associate fields to values
                 model_init_values = [values[pos] for pos in model_init_pos]
                 instance = model_cls.from_db(db, model_init_names, model_init_values)
diff --git a/django/db/models/sql/compiler.py b/django/db/models/sql/compiler.py
index 82438355cb..ee5e319360 100644
--- a/django/db/models/sql/compiler.py
+++ b/django/db/models/sql/compiler.py
@@ -690,12 +690,34 @@ class SQLCompiler(object):
         self.query.deferred_to_data(columns, self.query.deferred_to_columns_cb)
         return columns
 
+    def get_converters(self, fields):
+        converters = {}
+        index_extra_select = len(self.query.extra_select)
+        for i, field in enumerate(fields):
+            if field:
+                backend_converters = self.connection.ops.get_db_converters(field.get_internal_type())
+                field_converters = field.get_db_converters(self.connection)
+                if backend_converters or field_converters:
+                    converters[index_extra_select + i] = (backend_converters, field_converters, field)
+        return converters
+
+    def apply_converters(self, row, converters):
+        row = list(row)
+        for pos, (backend_converters, field_converters, field) in converters.items():
+            value = row[pos]
+            for converter in backend_converters:
+                value = converter(value, field)
+            for converter in field_converters:
+                value = converter(value, self.connection)
+            row[pos] = value
+        return tuple(row)
+
     def results_iter(self):
         """
         Returns an iterator over the results from executing this query.
         """
-        resolve_columns = hasattr(self, 'resolve_columns')
         fields = None
+        converters = None
         has_aggregate_select = bool(self.query.aggregate_select)
         for rows in self.execute_sql(MULTI):
             for row in rows:
@@ -703,39 +725,40 @@ class SQLCompiler(object):
                     loaded_fields = self.query.get_loaded_field_names().get(self.query.model, set()) or self.query.select
                     aggregate_start = len(self.query.extra_select) + len(loaded_fields)
                     aggregate_end = aggregate_start + len(self.query.aggregate_select)
-                if resolve_columns:
-                    if fields is None:
-                        # We only set this up here because
-                        # related_select_cols isn't populated until
-                        # execute_sql() has been called.
+                if fields is None:
+                    # We only set this up here because
+                    # related_select_cols isn't populated until
+                    # execute_sql() has been called.
 
-                        # We also include types of fields of related models that
-                        # will be included via select_related() for the benefit
-                        # of MySQL/MySQLdb when boolean fields are involved
-                        # (#15040).
+                    # We also include types of fields of related models that
+                    # will be included via select_related() for the benefit
+                    # of MySQL/MySQLdb when boolean fields are involved
+                    # (#15040).
 
-                        # This code duplicates the logic for the order of fields
-                        # found in get_columns(). It would be nice to clean this up.
-                        if self.query.select:
-                            fields = [f.field for f in self.query.select]
-                        elif self.query.default_cols:
-                            fields = self.query.get_meta().concrete_fields
-                        else:
-                            fields = []
-                        fields = fields + [f.field for f in self.query.related_select_cols]
+                    # This code duplicates the logic for the order of fields
+                    # found in get_columns(). It would be nice to clean this up.
+                    if self.query.select:
+                        fields = [f.field for f in self.query.select]
+                    elif self.query.default_cols:
+                        fields = self.query.get_meta().concrete_fields
+                    else:
+                        fields = []
+                    fields = fields + [f.field for f in self.query.related_select_cols]
 
-                        # If the field was deferred, exclude it from being passed
-                        # into `resolve_columns` because it wasn't selected.
-                        only_load = self.deferred_to_columns()
-                        if only_load:
-                            fields = [f for f in fields if f.model._meta.db_table not in only_load or
-                                      f.column in only_load[f.model._meta.db_table]]
-                        if has_aggregate_select:
-                            # pad None in to fields for aggregates
-                            fields = fields[:aggregate_start] + [
-                                None for x in range(0, aggregate_end - aggregate_start)
-                            ] + fields[aggregate_start:]
-                    row = self.resolve_columns(row, fields)
+                    # If the field was deferred, exclude it from being passed
+                    # into `get_converters` because it wasn't selected.
+                    only_load = self.deferred_to_columns()
+                    if only_load:
+                        fields = [f for f in fields if f.model._meta.db_table not in only_load or
+                                  f.column in only_load[f.model._meta.db_table]]
+                    if has_aggregate_select:
+                        # pad None in to fields for aggregates
+                        fields = fields[:aggregate_start] + [
+                            None for x in range(0, aggregate_end - aggregate_start)
+                        ] + fields[aggregate_start:]
+                    converters = self.get_converters(fields)
+                if converters:
+                    row = self.apply_converters(row, converters)
 
                 if has_aggregate_select:
                     row = tuple(row[:aggregate_start]) + tuple(
@@ -1092,22 +1115,13 @@ class SQLDateCompiler(SQLCompiler):
         """
         Returns an iterator over the results from executing this query.
         """
-        resolve_columns = hasattr(self, 'resolve_columns')
-        if resolve_columns:
-            from django.db.models.fields import DateField
-            fields = [DateField()]
-        else:
-            from django.db.backends.utils import typecast_date
-            needs_string_cast = self.connection.features.needs_datetime_string_cast
+        from django.db.models.fields import DateField
+        converters = self.get_converters([DateField()])
 
         offset = len(self.query.extra_select)
         for rows in self.execute_sql(MULTI):
             for row in rows:
-                date = row[offset]
-                if resolve_columns:
-                    date = self.resolve_columns(row, fields)[offset]
-                elif needs_string_cast:
-                    date = typecast_date(str(date))
+                date = self.apply_converters(row, converters)[offset]
                 if isinstance(date, datetime.datetime):
                     date = date.date()
                 yield date
@@ -1118,22 +1132,13 @@ class SQLDateTimeCompiler(SQLCompiler):
         """
         Returns an iterator over the results from executing this query.
         """
-        resolve_columns = hasattr(self, 'resolve_columns')
-        if resolve_columns:
-            from django.db.models.fields import DateTimeField
-            fields = [DateTimeField()]
-        else:
-            from django.db.backends.utils import typecast_timestamp
-            needs_string_cast = self.connection.features.needs_datetime_string_cast
+        from django.db.models.fields import DateTimeField
+        converters = self.get_converters([DateTimeField()])
 
         offset = len(self.query.extra_select)
         for rows in self.execute_sql(MULTI):
             for row in rows:
-                datetime = row[offset]
-                if resolve_columns:
-                    datetime = self.resolve_columns(row, fields)[offset]
-                elif needs_string_cast:
-                    datetime = typecast_timestamp(str(datetime))
+                datetime = self.apply_converters(row, converters)[offset]
                 # Datetimes are artificially returned in UTC on databases that
                 # don't support time zone. Restore the zone used in the query.
                 if settings.USE_TZ:
diff --git a/django/db/models/sql/query.py b/django/db/models/sql/query.py
index 9ec84d64cd..e6708715d3 100644
--- a/django/db/models/sql/query.py
+++ b/django/db/models/sql/query.py
@@ -54,15 +54,6 @@ class RawQuery(object):
     def clone(self, using):
         return RawQuery(self.sql, using, params=self.params)
 
-    def convert_values(self, value, field, connection):
-        """Convert the database-returned value into a type that is consistent
-        across database backends.
-
-        By default, this defers to the underlying backend operations, but
-        it can be overridden by Query classes for specific backends.
-        """
-        return connection.ops.convert_values(value, field)
-
     def get_columns(self):
         if self.cursor is None:
             self._execute_query()
@@ -308,15 +299,6 @@ class Query(object):
             obj._setup_query()
         return obj
 
-    def convert_values(self, value, field, connection):
-        """Convert the database-returned value into a type that is consistent
-        across database backends.
-
-        By default, this defers to the underlying backend operations, but
-        it can be overridden by Query classes for specific backends.
-        """
-        return connection.ops.convert_values(value, field)
-
     def resolve_aggregate(self, value, aggregate, connection):
         """Resolve the value of aggregates returned by the database to
         consistent (and reasonable) types.
@@ -337,7 +319,13 @@ class Query(object):
             return float(value)
         else:
             # Return value depends on the type of the field being processed.
-            return self.convert_values(value, aggregate.field, connection)
+            backend_converters = connection.ops.get_db_converters(aggregate.field.get_internal_type())
+            field_converters = aggregate.field.get_db_converters(connection)
+            for converter in backend_converters:
+                value = converter(value, aggregate.field)
+            for converter in field_converters:
+                value = converter(value, connection)
+            return value
 
     def get_aggregation(self, using, force_subq=False):
         """