- Added :class:`.mysql.JSON` for MySQL 5.7. The JSON type provides

persistence of JSON values in MySQL as well as basic operator support
of "getitem" and "getpath", making use of the ``JSON_EXTRACT``
function in order to refer to individual paths in a JSON structure.
fixes #3547
- Added a new type to core :class:`.types.JSON`.  This is the
base of the PostgreSQL :class:`.postgresql.JSON` type as well as that
of the new :class:`.mysql.JSON` type, so that a PG/MySQL-agnostic
JSON column may be used.  The type features basic index and path
searching support.
fixes #3619
- reorganization of migration docs etc. to try to refer both to
the fixes to JSON that helps Postgresql while at the same time
indicating these are new features of the new base JSON type.
- a rework of the Array/Indexable system some more, moving things
that are specific to Array out of Indexable.
- new operators for JSON indexing added to core so that these can
be compiled by the PG and MySQL dialects individually
- rename sqltypes.Array to sqltypes.ARRAY - as there is no generic
Array implementation, this is an uppercase type for now, consistent
with the new sqltypes.JSON type that is also not a generic implementation.
There may need to be some convention change to handle the case of
datatypes that aren't generic, rely upon DB-native implementations,
but aren't necessarily all named the same thing.

19 files changed, 788 insertions, 273 deletions

diff --git a/lib/sqlalchemy/__init__.py b/lib/sqlalchemy/__init__.py
index 12d4e8d1c..40b8000e8 100644
--- a/lib/sqlalchemy/__init__.py
+++ b/lib/sqlalchemy/__init__.py
@@ -54,7 +54,7 @@ from .sql import (
     )
 
 from .types import (
-    Array,
+    ARRAY,
     BIGINT,
     BINARY,
     BLOB,
@@ -76,6 +76,7 @@ from .types import (
     INTEGER,
     Integer,
     Interval,
+    JSON,
     LargeBinary,
     NCHAR,
     NVARCHAR,
diff --git a/lib/sqlalchemy/dialects/mysql/__init__.py b/lib/sqlalchemy/dialects/mysql/__init__.py
index c1f78bd1d..ca204fcd1 100644
--- a/lib/sqlalchemy/dialects/mysql/__init__.py
+++ b/lib/sqlalchemy/dialects/mysql/__init__.py
@@ -15,7 +15,7 @@ base.dialect = mysqldb.dialect
 from .base import \
     BIGINT, BINARY, BIT, BLOB, BOOLEAN, CHAR, DATE, DATETIME, \
     DECIMAL, DOUBLE, ENUM, DECIMAL,\
-    FLOAT, INTEGER, INTEGER, LONGBLOB, LONGTEXT, MEDIUMBLOB, \
+    FLOAT, INTEGER, INTEGER, JSON, LONGBLOB, LONGTEXT, MEDIUMBLOB, \
     MEDIUMINT, MEDIUMTEXT, NCHAR, \
     NVARCHAR, NUMERIC, SET, SMALLINT, REAL, TEXT, TIME, TIMESTAMP, \
     TINYBLOB, TINYINT, TINYTEXT,\
@@ -24,8 +24,8 @@ from .base import \
 __all__ = (
     'BIGINT', 'BINARY', 'BIT', 'BLOB', 'BOOLEAN', 'CHAR', 'DATE', 'DATETIME',
     'DECIMAL', 'DOUBLE', 'ENUM', 'DECIMAL', 'FLOAT', 'INTEGER', 'INTEGER',
-    'LONGBLOB', 'LONGTEXT', 'MEDIUMBLOB', 'MEDIUMINT', 'MEDIUMTEXT', 'NCHAR',
-    'NVARCHAR', 'NUMERIC', 'SET', 'SMALLINT', 'REAL', 'TEXT', 'TIME',
+    'JSON', 'LONGBLOB', 'LONGTEXT', 'MEDIUMBLOB', 'MEDIUMINT', 'MEDIUMTEXT',
+    'NCHAR', 'NVARCHAR', 'NUMERIC', 'SET', 'SMALLINT', 'REAL', 'TEXT', 'TIME',
     'TIMESTAMP', 'TINYBLOB', 'TINYINT', 'TINYTEXT', 'VARBINARY', 'VARCHAR',
     'YEAR', 'dialect'
 )
diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index 66ef0d5e2..61c4a3fac 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -538,10 +538,11 @@ output::
 
 import re
 import sys
+import json
 
 from ... import schema as sa_schema
 from ... import exc, log, sql, util
-from ...sql import compiler
+from ...sql import compiler, elements
 from array import array as _array
 
 from ...engine import reflection
@@ -559,6 +560,7 @@ from .types import BIGINT, BIT, CHAR, DECIMAL, DATETIME, \
 from .types import _StringType, _IntegerType, _NumericType, \
     _FloatType, _MatchType
 from .enumerated import ENUM, SET
+from .json import JSON, JSONIndexType, JSONPathType
 
 
 RESERVED_WORDS = set(
@@ -663,7 +665,11 @@ colspecs = {
     sqltypes.Float: FLOAT,
     sqltypes.Time: TIME,
     sqltypes.Enum: ENUM,
-    sqltypes.MatchType: _MatchType
+    sqltypes.MatchType: _MatchType,
+    sqltypes.JSON: JSON,
+    sqltypes.JSON.JSONIndexType: JSONIndexType,
+    sqltypes.JSON.JSONPathType: JSONPathType
+
 }
 
 # Everything 3.23 through 5.1 excepting OpenGIS types.
@@ -683,6 +689,7 @@ ischema_names = {
     'float': FLOAT,
     'int': INTEGER,
     'integer': INTEGER,
+    'json': JSON,
     'longblob': LONGBLOB,
     'longtext': LONGTEXT,
     'mediumblob': MEDIUMBLOB,
@@ -728,6 +735,16 @@ class MySQLCompiler(compiler.SQLCompiler):
     def visit_sysdate_func(self, fn, **kw):
         return "SYSDATE()"
 
+    def visit_json_getitem_op_binary(self, binary, operator, **kw):
+        return "JSON_EXTRACT(%s, %s)" % (
+            self.process(binary.left),
+            self.process(binary.right))
+
+    def visit_json_path_getitem_op_binary(self, binary, operator, **kw):
+        return "JSON_EXTRACT(%s, %s)" % (
+            self.process(binary.left),
+            self.process(binary.right))
+
     def visit_concat_op_binary(self, binary, operator, **kw):
         return "concat(%s, %s)" % (self.process(binary.left),
                                    self.process(binary.right))
@@ -760,6 +777,8 @@ class MySQLCompiler(compiler.SQLCompiler):
             return self.dialect.type_compiler.process(adapted)
         elif isinstance(type_, sqltypes._Binary):
             return 'BINARY'
+        elif isinstance(type_, sqltypes.JSON):
+            return "JSON"
         elif isinstance(type_, sqltypes.NUMERIC):
             return self.dialect.type_compiler.process(
                 type_).replace('NUMERIC', 'DECIMAL')
@@ -1275,6 +1294,9 @@ class MySQLTypeCompiler(compiler.GenericTypeCompiler):
     def visit_VARBINARY(self, type_, **kw):
         return "VARBINARY(%d)" % type_.length
 
+    def visit_JSON(self, type_, **kw):
+        return "JSON"
+
     def visit_large_binary(self, type_, **kw):
         return self.visit_BLOB(type_)
 
@@ -1394,10 +1416,13 @@ class MySQLDialect(default.DefaultDialect):
         })
     ]
 
-    def __init__(self, isolation_level=None, **kwargs):
+    def __init__(self, isolation_level=None, json_serializer=None,
+                 json_deserializer=None, **kwargs):
         kwargs.pop('use_ansiquotes', None)   # legacy
         default.DefaultDialect.__init__(self, **kwargs)
         self.isolation_level = isolation_level
+        self._json_serializer = json_serializer
+        self._json_deserializer = json_deserializer
 
     def on_connect(self):
         if self.isolation_level is not None:
@@ -1564,6 +1589,10 @@ class MySQLDialect(default.DefaultDialect):
         default.DefaultDialect.initialize(self, connection)
 
     @property
+    def _is_mariadb(self):
+        return 'MariaDB' in self.server_version_info
+
+    @property
     def _supports_cast(self):
         return self.server_version_info is None or \
             self.server_version_info >= (4, 0, 2)
diff --git a/lib/sqlalchemy/dialects/mysql/json.py b/lib/sqlalchemy/dialects/mysql/json.py
new file mode 100644
index 000000000..a30cdc841
--- /dev/null
+++ b/lib/sqlalchemy/dialects/mysql/json.py
@@ -0,0 +1,90 @@
+# mysql/json.py
+# Copyright (C) 2005-2015 the SQLAlchemy authors and contributors
+# <see AUTHORS file>
+#
+# This module is part of SQLAlchemy and is released under
+# the MIT License: http://www.opensource.org/licenses/mit-license.php
+
+from __future__ import absolute_import
+
+import json
+
+from ...sql import elements
+from ... import types as sqltypes
+from ... import util
+
+
+class JSON(sqltypes.JSON):
+    """MySQL JSON type.
+
+    MySQL supports JSON as of version 5.7.  Note that MariaDB does **not**
+    support JSON at the time of this writing.
+
+    The :class:`.mysql.JSON` type supports persistence of JSON values
+    as well as the core index operations provided by :class:`.types.JSON`
+    datatype, by adapting the operations to render the ``JSON_EXTRACT``
+    function at the database level.
+
+    .. versionadded:: 1.1
+
+    """
+
+    @util.memoized_property
+    def _str_impl(self):
+        return sqltypes.String(convert_unicode=True)
+
+    def bind_processor(self, dialect):
+        string_process = self._str_impl.bind_processor(dialect)
+
+        json_serializer = dialect._json_serializer or json.dumps
+
+        def process(value):
+            if value is self.NULL:
+                value = None
+            elif isinstance(value, elements.Null) or (
+                value is None and self.none_as_null
+            ):
+                return None
+
+            serialized = json_serializer(value)
+            if string_process:
+                serialized = string_process(serialized)
+            return serialized
+
+        return process
+
+    def result_processor(self, dialect, coltype):
+        string_process = self._str_impl.result_processor(dialect, coltype)
+        json_deserializer = dialect._json_deserializer or json.loads
+
+        def process(value):
+            if value is None:
+                return None
+            if string_process:
+                value = string_process(value)
+            return json_deserializer(value)
+        return process
+
+
+class JSONIndexType(sqltypes.JSON.JSONIndexType):
+    def bind_processor(self, dialect):
+        def process(value):
+            if isinstance(value, int):
+                return "$[%s]" % value
+            else:
+                return '$."%s"' % value
+
+        return process
+
+
+class JSONPathType(sqltypes.JSON.JSONPathType):
+    def bind_processor(self, dialect):
+        def process(value):
+            return "$%s" % (
+                "".join([
+                    "[%s]" % elem if isinstance(elem, int)
+                    else '."%s"' % elem for elem in value
+                ])
+            )
+
+        return process
diff --git a/lib/sqlalchemy/dialects/postgresql/array.py b/lib/sqlalchemy/dialects/postgresql/array.py
index b88f139de..f4316d318 100644
--- a/lib/sqlalchemy/dialects/postgresql/array.py
+++ b/lib/sqlalchemy/dialects/postgresql/array.py
@@ -84,12 +84,20 @@ class array(expression.Tuple):
         super(array, self).__init__(*clauses, **kw)
         self.type = ARRAY(self.type)
 
-    def _bind_param(self, operator, obj):
-        return array([
-            expression.BindParameter(None, o, _compared_to_operator=operator,
-                                     _compared_to_type=self.type, unique=True)
-            for o in obj
-        ])
+    def _bind_param(self, operator, obj, _assume_scalar=False, type_=None):
+        if _assume_scalar or operator is operators.getitem:
+            # if getitem->slice were called, Indexable produces
+            # a Slice object from that
+            assert isinstance(obj, int)
+            return expression.BindParameter(
+                None, obj, _compared_to_operator=operator,
+                type_=type_,
+                _compared_to_type=self.type, unique=True)
+
+        else:
+            return array([
+                self._bind_param(operator, o, _assume_scalar=True, type_=type_)
+                for o in obj])
 
     def self_group(self, against=None):
         if (against in (
@@ -106,15 +114,15 @@ CONTAINED_BY = operators.custom_op("<@", precedence=5)
 OVERLAP = operators.custom_op("&&", precedence=5)
 
 
-class ARRAY(SchemaEventTarget, sqltypes.Array):
+class ARRAY(SchemaEventTarget, sqltypes.ARRAY):
 
     """Postgresql ARRAY type.
 
     .. versionchanged:: 1.1 The :class:`.postgresql.ARRAY` type is now
-       a subclass of the core :class:`.Array` type.
+       a subclass of the core :class:`.types.ARRAY` type.
 
     The :class:`.postgresql.ARRAY` type is constructed in the same way
-    as the core :class:`.Array` type; a member type is required, and a
+    as the core :class:`.types.ARRAY` type; a member type is required, and a
     number of dimensions is recommended if the type is to be used for more
     than one dimension::
 
@@ -125,9 +133,9 @@ class ARRAY(SchemaEventTarget, sqltypes.Array):
             )
 
     The :class:`.postgresql.ARRAY` type provides all operations defined on the
-    core :class:`.Array` type, including support for "dimensions", indexed
-    access, and simple matching such as :meth:`.Array.Comparator.any`
-    and :meth:`.Array.Comparator.all`.  :class:`.postgresql.ARRAY` class also
+    core :class:`.types.ARRAY` type, including support for "dimensions", indexed
+    access, and simple matching such as :meth:`.types.ARRAY.Comparator.any`
+    and :meth:`.types.ARRAY.Comparator.all`.  :class:`.postgresql.ARRAY` class also
     provides PostgreSQL-specific methods for containment operations, including
     :meth:`.postgresql.ARRAY.Comparator.contains`
     :meth:`.postgresql.ARRAY.Comparator.contained_by`,
@@ -144,20 +152,20 @@ class ARRAY(SchemaEventTarget, sqltypes.Array):
 
     .. seealso::
 
-        :class:`.types.Array` - base array type
+        :class:`.types.ARRAY` - base array type
 
         :class:`.postgresql.array` - produces a literal array value.
 
     """
 
-    class Comparator(sqltypes.Array.Comparator):
+    class Comparator(sqltypes.ARRAY.Comparator):
 
         """Define comparison operations for :class:`.ARRAY`.
 
         Note that these operations are in addition to those provided
-        by the base :class:`.types.Array.Comparator` class, including
-        :meth:`.types.Array.Comparator.any` and
-        :meth:`.types.Array.Comparator.all`.
+        by the base :class:`.types.ARRAY.Comparator` class, including
+        :meth:`.types.ARRAY.Comparator.any` and
+        :meth:`.types.ARRAY.Comparator.all`.
 
         """
 
diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index e9001f79a..3f9fcb27f 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -1056,6 +1056,16 @@ class PGCompiler(compiler.SQLCompiler):
             self.process(element.stop, **kw),
         )
 
+    def visit_json_getitem_op_binary(self, binary, operator, **kw):
+        return self._generate_generic_binary(
+            binary, " -> ", **kw
+        )
+
+    def visit_json_path_getitem_op_binary(self, binary, operator, **kw):
+        return self._generate_generic_binary(
+            binary, " #> ", **kw
+        )
+
     def visit_getitem_binary(self, binary, operator, **kw):
         return "%s[%s]" % (
             self.process(binary.left, **kw),
diff --git a/lib/sqlalchemy/dialects/postgresql/ext.py b/lib/sqlalchemy/dialects/postgresql/ext.py
index 1a443c2d7..66c7ed0e5 100644
--- a/lib/sqlalchemy/dialects/postgresql/ext.py
+++ b/lib/sqlalchemy/dialects/postgresql/ext.py
@@ -159,7 +159,7 @@ static/sql-createtable.html#SQL-CREATETABLE-EXCLUDE
 def array_agg(*arg, **kw):
     """Postgresql-specific form of :class:`.array_agg`, ensures
     return type is :class:`.postgresql.ARRAY` and not
-    the plain :class:`.types.Array`.
+    the plain :class:`.types.ARRAY`.
 
     .. versionadded:: 1.1
 
diff --git a/lib/sqlalchemy/dialects/postgresql/hstore.py b/lib/sqlalchemy/dialects/postgresql/hstore.py
index b7b0fc007..d2d20386a 100644
--- a/lib/sqlalchemy/dialects/postgresql/hstore.py
+++ b/lib/sqlalchemy/dialects/postgresql/hstore.py
@@ -12,34 +12,33 @@ from .array import ARRAY
 from ... import types as sqltypes
 from ...sql import functions as sqlfunc
 from ...sql import operators
-from ...sql.operators import custom_op
 from ... import util
 
 __all__ = ('HSTORE', 'hstore')
 
 
-INDEX = custom_op(
-    "->", precedence=5, natural_self_precedent=True
+GETITEM = operators.custom_op(
+    "->", precedence=15, natural_self_precedent=True,
 )
 
 HAS_KEY = operators.custom_op(
-    "?", precedence=5, natural_self_precedent=True
+    "?", precedence=15, natural_self_precedent=True
 )
 
 HAS_ALL = operators.custom_op(
-    "?&", precedence=5, natural_self_precedent=True
+    "?&", precedence=15, natural_self_precedent=True
 )
 
 HAS_ANY = operators.custom_op(
-    "?|", precedence=5, natural_self_precedent=True
+    "?|", precedence=15, natural_self_precedent=True
 )
 
 CONTAINS = operators.custom_op(
-    "@>", precedence=5, natural_self_precedent=True
+    "@>", precedence=15, natural_self_precedent=True
 )
 
 CONTAINED_BY = operators.custom_op(
-    "<@", precedence=5, natural_self_precedent=True
+    "<@", precedence=15, natural_self_precedent=True
 )
 
 
@@ -166,7 +165,7 @@ class HSTORE(sqltypes.Indexable, sqltypes.Concatenable, sqltypes.TypeEngine):
                 CONTAINED_BY, other, result_type=sqltypes.Boolean)
 
         def _setup_getitem(self, index):
-            return INDEX, index, self.type.text_type
+            return GETITEM, index, self.type.text_type
 
         def defined(self, key):
             """Boolean expression.  Test for presence of a non-NULL value for
diff --git a/lib/sqlalchemy/dialects/postgresql/json.py b/lib/sqlalchemy/dialects/postgresql/json.py
index 8a50270f5..6ff9fd88e 100644
--- a/lib/sqlalchemy/dialects/postgresql/json.py
+++ b/lib/sqlalchemy/dialects/postgresql/json.py
@@ -6,10 +6,10 @@
 # the MIT License: http://www.opensource.org/licenses/mit-license.php
 from __future__ import absolute_import
 
-import collections
 import json
+import collections
 
-from .base import ischema_names
+from .base import ischema_names, colspecs
 from ... import types as sqltypes
 from ...sql import operators
 from ...sql import elements
@@ -17,70 +17,68 @@ from ... import util
 
 __all__ = ('JSON', 'JSONB')
 
-
-# json : returns json
-INDEX = operators.custom_op(
-    "->", precedence=5, natural_self_precedent=True
-)
-
-# path operator: returns json
-PATHIDX = operators.custom_op(
-    "#>", precedence=5, natural_self_precedent=True
-)
-
-# json + astext: returns text
 ASTEXT = operators.custom_op(
-    "->>", precedence=5, natural_self_precedent=True
+    "->>", precedence=15, natural_self_precedent=True,
 )
 
-# path operator  + astext: returns text
-ASTEXT_PATHIDX = operators.custom_op(
-    "#>>", precedence=5, natural_self_precedent=True
+JSONPATH_ASTEXT = operators.custom_op(
+    "#>>", precedence=15, natural_self_precedent=True,
 )
 
+
 HAS_KEY = operators.custom_op(
-    "?", precedence=5, natural_self_precedent=True
+    "?", precedence=15, natural_self_precedent=True
 )
 
 HAS_ALL = operators.custom_op(
-    "?&", precedence=5, natural_self_precedent=True
+    "?&", precedence=15, natural_self_precedent=True
 )
 
 HAS_ANY = operators.custom_op(
-    "?|", precedence=5, natural_self_precedent=True
+    "?|", precedence=15, natural_self_precedent=True
 )
 
 CONTAINS = operators.custom_op(
-    "@>", precedence=5, natural_self_precedent=True
+    "@>", precedence=15, natural_self_precedent=True
 )
 
 CONTAINED_BY = operators.custom_op(
-    "<@", precedence=5, natural_self_precedent=True
+    "<@", precedence=15, natural_self_precedent=True
 )
 
 
-class JSON(sqltypes.Indexable, sqltypes.TypeEngine):
-    """Represent the Postgresql JSON type.
+class JSONPathType(sqltypes.JSON.JSONPathType):
+    def bind_processor(self, dialect):
+        def process(value):
+            assert isinstance(value, collections.Sequence)
+            tokens = [util.text_type(elem) for elem in value]
+            return "{%s}" % (", ".join(tokens))
 
-    The :class:`.JSON` type stores arbitrary JSON format data, e.g.::
+        return process
 
-        data_table = Table('data_table', metadata,
-            Column('id', Integer, primary_key=True),
-            Column('data', JSON)
-        )
+colspecs[sqltypes.JSON.JSONPathType] = JSONPathType
 
-        with engine.connect() as conn:
-            conn.execute(
-                data_table.insert(),
-                data = {"key1": "value1", "key2": "value2"}
-            )
 
-    :class:`.JSON` provides several operations:
+class JSON(sqltypes.JSON):
+    """Represent the Postgresql JSON type.
+
+    This type is a specialization of the Core-level :class:`.types.JSON`
+    type.   Be sure to read the documentation for :class:`.types.JSON` for
+    important tips regarding treatment of NULL values and ORM use.
+
+    .. versionchanged:: 1.1 :class:`.postgresql.JSON` is now a Postgresql-
+       specific specialization of the new :class:`.types.JSON` type.
+
+    The operators provided by the Postgresql version of :class:`.JSON`
+    include:
 
     * Index operations (the ``->`` operator)::
 
         data_table.c.data['some key']
 
+        data_table.c.data[5]
+
+
     * Index operations returning text (the ``->>`` operator)::
 
         data_table.c.data['some key'].astext == 'some value'
@@ -92,11 +90,11 @@ class JSON(sqltypes.Indexable, sqltypes.TypeEngine):
 
     * Path index operations (the ``#>`` operator)::
 
-        data_table.c.data[('key_1', 'key_2', ..., 'key_n')]
+        data_table.c.data[('key_1', 'key_2', 5, ..., 'key_n')]
 
     * Path index operations returning text (the ``#>>`` operator)::
 
-        data_table.c.data[('key_1', 'key_2', ..., 'key_n')].astext == \
+        data_table.c.data[('key_1', 'key_2', 5, ..., 'key_n')].astext == \
 'some value'
 
     .. versionchanged:: 1.1  The :meth:`.ColumnElement.cast` operator on
@@ -108,36 +106,6 @@ class JSON(sqltypes.Indexable, sqltypes.TypeEngine):
     :class:`.JSON` by default, so that further JSON-oriented instructions
     may be called upon the result type.
 
-    The :class:`.JSON` type, when used with the SQLAlchemy ORM, does not
-    detect in-place mutations to the structure.  In order to detect these, the
-    :mod:`sqlalchemy.ext.mutable` extension must be used.  This extension will
-    allow "in-place" changes to the datastructure to produce events which
-    will be detected by the unit of work.  See the example at :class:`.HSTORE`
-    for a simple example involving a dictionary.
-
-    When working with NULL values, the :class:`.JSON` type recommends the
-    use of two specific constants in order to differentiate between a column
-    that evaluates to SQL NULL, e.g. no value, vs. the JSON-encoded string
-    of ``"null"``.   To insert or select against a value that is SQL NULL,
-    use the constant :func:`.null`::
-
-        conn.execute(table.insert(), json_value=null())
-
-    To insert or select against a value that is JSON ``"null"``, use the
-    constant :attr:`.JSON.NULL`::
-
-        conn.execute(table.insert(), json_value=JSON.NULL)
-
-    The :class:`.JSON` type supports a flag
-    :paramref:`.JSON.none_as_null` which when set to True will result
-    in the Python constant ``None`` evaluating to the value of SQL
-    NULL, and when set to False results in the Python constant
-    ``None`` evaluating to the value of JSON ``"null"``.    The Python
-    value ``None`` may be used in conjunction with either
-    :attr:`.JSON.NULL` and :func:`.null` in order to indicate NULL
-    values, but care must be taken as to the value of the
-    :paramref:`.JSON.none_as_null` in these cases.
-
     Custom serializers and deserializers are specified at the dialect level,
     that is using :func:`.create_engine`.  The reason for this is that when
     using psycopg2, the DBAPI only allows serializers at the per-cursor
@@ -151,43 +119,16 @@ class JSON(sqltypes.Indexable, sqltypes.TypeEngine):
     When using the psycopg2 dialect, the json_deserializer is registered
     against the database using ``psycopg2.extras.register_default_json``.
 
-    .. versionadded:: 0.9
-
     .. seealso::
 
+        :class:`.types.JSON` - Core level JSON type
+
         :class:`.JSONB`
 
     """
 
-    __visit_name__ = 'JSON'
-
-    hashable = False
     astext_type = sqltypes.Text()
 
-    NULL = util.symbol('JSON_NULL')
-    """Describe the json value of NULL.
-
-    This value is used to force the JSON value of ``"null"`` to be
-    used as the value.   A value of Python ``None`` will be recognized
-    either as SQL NULL or JSON ``"null"``, based on the setting
-    of the :paramref:`.JSON.none_as_null` flag; the :attr:`.JSON.NULL`
-    constant can be used to always resolve to JSON ``"null"`` regardless
-    of this setting.  This is in contrast to the :func:`.sql.null` construct,
-    which always resolves to SQL NULL.  E.g.::
-
-        from sqlalchemy import null
-        from sqlalchemy.dialects.postgresql import JSON
-
-        obj1 = MyObject(json_value=null())  # will *always* insert SQL NULL
-        obj2 = MyObject(json_value=JSON.NULL)  # will *always* insert JSON string "null"
-
-        session.add_all([obj1, obj2])
-        session.commit()
-
-    .. versionadded:: 1.1
-
-    """
-
     def __init__(self, none_as_null=False, astext_type=None):
         """Construct a :class:`.JSON` type.
 
@@ -210,15 +151,14 @@ class JSON(sqltypes.Indexable, sqltypes.TypeEngine):
          :attr:`.JSON.Comparator.astext`
          accessor on indexed attributes.  Defaults to :class:`.types.Text`.
 
-         .. versionadded:: 1.1.0
+         .. versionadded:: 1.1
 
          """
-        self.none_as_null = none_as_null
+        super(JSON, self).__init__(none_as_null=none_as_null)
         if astext_type is not None:
             self.astext_type = astext_type
 
-    class Comparator(
-            sqltypes.Indexable.Comparator, sqltypes.Concatenable.Comparator):
+    class Comparator(sqltypes.JSON.Comparator):
         """Define comparison operations for :class:`.JSON`."""
 
         @property
@@ -235,69 +175,19 @@ class JSON(sqltypes.Indexable, sqltypes.TypeEngine):
                 :meth:`.ColumnElement.cast`
 
             """
-            against = self.expr.operator
-            if against is PATHIDX:
-                against = ASTEXT_PATHIDX
-            else:
-                against = ASTEXT
 
-            return self.expr.left.operate(
-                against, self.expr.right, result_type=self.type.astext_type)
-
-        def _setup_getitem(self, index):
-            if not isinstance(index, util.string_types):
-                assert isinstance(index, collections.Sequence)
-                tokens = [util.text_type(elem) for elem in index]
-                index = "{%s}" % (", ".join(tokens))
-                operator = PATHIDX
+            if isinstance(self.expr.right.type, sqltypes.JSON.JSONPathType):
+                return self.expr.left.operate(
+                    JSONPATH_ASTEXT,
+                    self.expr.right, result_type=self.type.astext_type)
             else:
-                operator = INDEX
-
-            return operator, index, self.type
+                return self.expr.left.operate(
+                    ASTEXT, self.expr.right, result_type=self.type.astext_type)
 
     comparator_factory = Comparator
 
-    @property
-    def should_evaluate_none(self):
-        return not self.none_as_null
-
-    def bind_processor(self, dialect):
-        json_serializer = dialect._json_serializer or json.dumps
-        if util.py2k:
-            encoding = dialect.encoding
-        else:
-            encoding = None
-
-        def process(value):
-            if value is self.NULL:
-                value = None
-            elif isinstance(value, elements.Null) or (
-                value is None and self.none_as_null
-            ):
-                return None
-            if encoding:
-                return json_serializer(value).encode(encoding)
-            else:
-                return json_serializer(value)
-
-        return process
-
-    def result_processor(self, dialect, coltype):
-        json_deserializer = dialect._json_deserializer or json.loads
-        if util.py2k:
-            encoding = dialect.encoding
-        else:
-            encoding = None
-
-        def process(value):
-            if value is None:
-                return None
-            if encoding:
-                value = value.decode(encoding)
-            return json_deserializer(value)
-        return process
-
 
+colspecs[sqltypes.JSON] = JSON
 ischema_names['json'] = JSON
 
 
diff --git a/lib/sqlalchemy/dialects/postgresql/psycopg2.py b/lib/sqlalchemy/dialects/postgresql/psycopg2.py
index d33554922..82fcc9054 100644
--- a/lib/sqlalchemy/dialects/postgresql/psycopg2.py
+++ b/lib/sqlalchemy/dialects/postgresql/psycopg2.py
@@ -534,6 +534,7 @@ class PGDialect_psycopg2(PGDialect):
             sqltypes.Enum: _PGEnum,  # needs force_unicode
             HSTORE: _PGHStore,
             JSON: _PGJSON,
+            sqltypes.JSON: _PGJSON,
             JSONB: _PGJSONB,
             UUID: _PGUUID
         }
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 2ca549267..2fe6ea02c 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -879,22 +879,28 @@ class SQLCompiler(Compiled):
         else:
             return text
 
+    def _get_operator_dispatch(self, operator_, qualifier1, qualifier2):
+        attrname = "visit_%s_%s%s" % (
+            operator_.__name__, qualifier1,
+            "_" + qualifier2 if qualifier2 else "")
+        return getattr(self, attrname, None)
+
     def visit_unary(self, unary, **kw):
         if unary.operator:
             if unary.modifier:
                 raise exc.CompileError(
                     "Unary expression does not support operator "
                     "and modifier simultaneously")
-            disp = getattr(self, "visit_%s_unary_operator" %
-                           unary.operator.__name__, None)
+            disp = self._get_operator_dispatch(
+                unary.operator, "unary", "operator")
             if disp:
                 return disp(unary, unary.operator, **kw)
             else:
                 return self._generate_generic_unary_operator(
                     unary, OPERATORS[unary.operator], **kw)
         elif unary.modifier:
-            disp = getattr(self, "visit_%s_unary_modifier" %
-                           unary.modifier.__name__, None)
+            disp = self._get_operator_dispatch(
+                unary.modifier, "unary", "modifier")
             if disp:
                 return disp(unary, unary.modifier, **kw)
             else:
@@ -928,7 +934,7 @@ class SQLCompiler(Compiled):
             kw['literal_binds'] = True
 
         operator_ = override_operator or binary.operator
-        disp = getattr(self, "visit_%s_binary" % operator_.__name__, None)
+        disp = self._get_operator_dispatch(operator_, "binary", None)
         if disp:
             return disp(binary, operator_, **kw)
         else:
diff --git a/lib/sqlalchemy/sql/default_comparator.py b/lib/sqlalchemy/sql/default_comparator.py
index 68ea5624e..ddb57da77 100644
--- a/lib/sqlalchemy/sql/default_comparator.py
+++ b/lib/sqlalchemy/sql/default_comparator.py
@@ -164,27 +164,7 @@ def _in_impl(expr, op, seq_or_selectable, negate_op, **kw):
 
 def _getitem_impl(expr, op, other, **kw):
     if isinstance(expr.type, type_api.INDEXABLE):
-        if isinstance(other, slice):
-            if expr.type.zero_indexes:
-                other = slice(
-                    other.start + 1,
-                    other.stop + 1,
-                    other.step
-                )
-            other = Slice(
-                _literal_as_binds(
-                    other.start, name=expr.key, type_=type_api.INTEGERTYPE),
-                _literal_as_binds(
-                    other.stop, name=expr.key, type_=type_api.INTEGERTYPE),
-                _literal_as_binds(
-                    other.step, name=expr.key, type_=type_api.INTEGERTYPE)
-            )
-        else:
-            if expr.type.zero_indexes:
-                other += 1
-
-        other = _literal_as_binds(
-            other, name=expr.key, type_=type_api.INTEGERTYPE)
+        other = _check_literal(expr, op, other)
         return _binary_operate(expr, op, other, **kw)
     else:
         _unsupported_impl(expr, op, other, **kw)
@@ -260,6 +240,8 @@ operator_lookup = {
     "mod": (_binary_operate,),
     "truediv": (_binary_operate,),
     "custom_op": (_binary_operate,),
+    "json_path_getitem_op": (_binary_operate, ),
+    "json_getitem_op": (_binary_operate, ),
     "concat_op": (_binary_operate,),
     "lt": (_boolean_compare, operators.ge),
     "le": (_boolean_compare, operators.gt),
@@ -295,7 +277,7 @@ operator_lookup = {
 }
 
 
-def _check_literal(expr, operator, other):
+def _check_literal(expr, operator, other, bindparam_type=None):
     if isinstance(other, (ColumnElement, TextClause)):
         if isinstance(other, BindParameter) and \
                 other.type._isnull:
@@ -310,7 +292,7 @@ def _check_literal(expr, operator, other):
     if isinstance(other, (SelectBase, Alias)):
         return other.as_scalar()
     elif not isinstance(other, Visitable):
-        return expr._bind_param(operator, other)
+        return expr._bind_param(operator, other, type_=bindparam_type)
     else:
         return other
 
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index 70046c66b..774e42609 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -682,9 +682,10 @@ class ColumnElement(operators.ColumnOperators, ClauseElement):
     def reverse_operate(self, op, other, **kwargs):
         return op(other, self.comparator, **kwargs)
 
-    def _bind_param(self, operator, obj):
+    def _bind_param(self, operator, obj, type_=None):
         return BindParameter(None, obj,
                              _compared_to_operator=operator,
+                             type_=type_,
                              _compared_to_type=self.type, unique=True)
 
     @property
@@ -1952,11 +1953,12 @@ class Tuple(ClauseList, ColumnElement):
     def _select_iterable(self):
         return (self, )
 
-    def _bind_param(self, operator, obj):
+    def _bind_param(self, operator, obj, type_=None):
         return Tuple(*[
             BindParameter(None, o, _compared_to_operator=operator,
-                          _compared_to_type=type_, unique=True)
-            for o, type_ in zip(obj, self._type_tuple)
+                          _compared_to_type=compared_to_type, unique=True,
+                          type_=type_)
+            for o, compared_to_type in zip(obj, self._type_tuple)
         ]).self_group()
 
 
@@ -3637,10 +3639,11 @@ class ColumnClause(Immutable, ColumnElement):
         else:
             return name
 
-    def _bind_param(self, operator, obj):
+    def _bind_param(self, operator, obj, type_=None):
         return BindParameter(self.key, obj,
                              _compared_to_operator=operator,
                              _compared_to_type=self.type,
+                             type_=type_,
                              unique=True)
 
     def _make_proxy(self, selectable, name=None, attach=True,
diff --git a/lib/sqlalchemy/sql/functions.py b/lib/sqlalchemy/sql/functions.py
index 6cfbd12b3..3c654bf67 100644
--- a/lib/sqlalchemy/sql/functions.py
+++ b/lib/sqlalchemy/sql/functions.py
@@ -256,16 +256,18 @@ class FunctionElement(Executable, ColumnElement, FromClause):
         """
         return self.select().execute()
 
-    def _bind_param(self, operator, obj):
+    def _bind_param(self, operator, obj, type_=None):
         return BindParameter(None, obj, _compared_to_operator=operator,
-                             _compared_to_type=self.type, unique=True)
+                             _compared_to_type=self.type, unique=True,
+                             type_=type_)
 
     def self_group(self, against=None):
         # for the moment, we are parenthesizing all array-returning
         # expressions against getitem.  This may need to be made
         # more portable if in the future we support other DBs
         # besides postgresql.
-        if against is operators.getitem:
+        if against is operators.getitem and \
+                isinstance(self.type, sqltypes.ARRAY):
             return Grouping(self)
         else:
             return super(FunctionElement, self).self_group(against=against)
@@ -423,10 +425,11 @@ class Function(FunctionElement):
 
         FunctionElement.__init__(self, *clauses, **kw)
 
-    def _bind_param(self, operator, obj):
+    def _bind_param(self, operator, obj, type_=None):
         return BindParameter(self.name, obj,
                              _compared_to_operator=operator,
                              _compared_to_type=self.type,
+                             type_=type_,
                              unique=True)
 
 
@@ -659,7 +662,7 @@ class array_agg(GenericFunction):
     """support for the ARRAY_AGG function.
 
     The ``func.array_agg(expr)`` construct returns an expression of
-    type :class:`.Array`.
+    type :class:`.types.ARRAY`.
 
     e.g.::
 
@@ -670,11 +673,11 @@ class array_agg(GenericFunction):
     .. seealso::
 
         :func:`.postgresql.array_agg` - PostgreSQL-specific version that
-        returns :class:`.ARRAY`, which has PG-specific operators added.
+        returns :class:`.postgresql.ARRAY`, which has PG-specific operators added.
 
     """
 
-    type = sqltypes.Array
+    type = sqltypes.ARRAY
 
     def __init__(self, *args, **kwargs):
         args = [_literal_as_binds(c) for c in args]
@@ -694,7 +697,7 @@ class OrderedSetAgg(GenericFunction):
         func_clauses = self.clause_expr.element
         order_by = sqlutil.unwrap_order_by(within_group.order_by)
         if self.array_for_multi_clause and len(func_clauses.clauses) > 1:
-            return sqltypes.Array(order_by[0].type)
+            return sqltypes.ARRAY(order_by[0].type)
         else:
             return order_by[0].type
 
@@ -719,7 +722,7 @@ class percentile_cont(OrderedSetAgg):
     modifier to supply a sort expression to operate upon.
 
     The return type of this function is the same as the sort expression,
-    or if the arguments are an array, an :class:`.Array` of the sort
+    or if the arguments are an array, an :class:`.types.ARRAY` of the sort
     expression's type.
 
     .. versionadded:: 1.1
@@ -736,7 +739,7 @@ class percentile_disc(OrderedSetAgg):
     modifier to supply a sort expression to operate upon.
 
     The return type of this function is the same as the sort expression,
-    or if the arguments are an array, an :class:`.Array` of the sort
+    or if the arguments are an array, an :class:`.types.ARRAY` of the sort
     expression's type.
 
     .. versionadded:: 1.1
diff --git a/lib/sqlalchemy/sql/operators.py b/lib/sqlalchemy/sql/operators.py
index da3576466..f4f90b664 100644
--- a/lib/sqlalchemy/sql/operators.py
+++ b/lib/sqlalchemy/sql/operators.py
@@ -12,7 +12,6 @@
 
 from .. import util
 
-
 from operator import (
     and_, or_, inv, add, mul, sub, mod, truediv, lt, le, ne, gt, ge, eq, neg,
     getitem, lshift, rshift
@@ -720,7 +719,6 @@ def istrue(a):
 def isfalse(a):
     raise NotImplementedError()
 
-
 def is_(a, b):
     return a.is_(b)
 
@@ -837,6 +835,14 @@ def nullslast_op(a):
     return a.nullslast()
 
 
+def json_getitem_op(a, b):
+    raise NotImplementedError()
+
+
+def json_path_getitem_op(a, b):
+    raise NotImplementedError()
+
+
 _commutative = set([eq, ne, add, mul])
 
 _comparison = set([eq, ne, lt, gt, ge, le, between_op, like_op])
@@ -879,7 +885,8 @@ def mirror(op):
 
 _associative = _commutative.union([concat_op, and_, or_])
 
-_natural_self_precedent = _associative.union([getitem])
+_natural_self_precedent = _associative.union([
+    getitem, json_getitem_op, json_path_getitem_op])
 """Operators where if we have (a op b) op c, we don't want to
 parenthesize (a op b).
 
@@ -894,6 +901,8 @@ _PRECEDENCE = {
     from_: 15,
     any_op: 15,
     all_op: 15,
+    json_getitem_op: 15,
+    json_path_getitem_op: 15,
     getitem: 15,
     mul: 8,
     truediv: 8,
diff --git a/lib/sqlalchemy/sql/sqltypes.py b/lib/sqlalchemy/sql/sqltypes.py
index 4abb9b15a..b65d39ba1 100644
--- a/lib/sqlalchemy/sql/sqltypes.py
+++ b/lib/sqlalchemy/sql/sqltypes.py
@@ -11,9 +11,12 @@
 
 import datetime as dt
 import codecs
+import collections
+import json
 
+from . import elements
 from .type_api import TypeEngine, TypeDecorator, to_instance
-from .elements import quoted_name, TypeCoerce as type_coerce, _defer_name
+from .elements import quoted_name, TypeCoerce as type_coerce, _defer_name, Slice, _literal_as_binds
 from .. import exc, util, processors
 from .base import _bind_or_error, SchemaEventTarget
 from . import operators
@@ -85,20 +88,16 @@ class Indexable(object):
 
     """
 
-    zero_indexes = False
-    """if True, Python zero-based indexes should be interpreted as one-based
-    on the SQL expression side."""
-
     class Comparator(TypeEngine.Comparator):
 
         def _setup_getitem(self, index):
             raise NotImplementedError()
 
         def __getitem__(self, index):
-            operator, adjusted_right_expr, result_type = \
+            adjusted_op, adjusted_right_expr, result_type = \
                 self._setup_getitem(index)
             return self.operate(
-                operator,
+                adjusted_op,
                 adjusted_right_expr,
                 result_type=result_type
             )
@@ -1496,7 +1495,221 @@ class Interval(_DateAffinity, TypeDecorator):
         return self.impl.coerce_compared_value(op, value)
 
 
-class Array(Indexable, Concatenable, TypeEngine):
+class JSON(Indexable, TypeEngine):
+    """Represent a SQL JSON type.
+
+    .. note::  :class:`.types.JSON` is provided as a facade for vendor-specific
+       JSON types.  Since it supports JSON SQL operations, it only
+       works on backends that have an actual JSON type, currently
+       Postgresql as well as certain versions of MySQL.
+
+    :class:`.types.JSON` is part of the Core in support of the growing
+    popularity of native JSON datatypes.
+
+    The :class:`.types.JSON` type stores arbitrary JSON format data, e.g.::
+
+        data_table = Table('data_table', metadata,
+            Column('id', Integer, primary_key=True),
+            Column('data', JSON)
+        )
+
+        with engine.connect() as conn:
+            conn.execute(
+                data_table.insert(),
+                data = {"key1": "value1", "key2": "value2"}
+            )
+
+    The base :class:`.types.JSON` provides these two operations:
+
+    * Keyed index operations::
+
+        data_table.c.data['some key']
+
+    * Integer index operations::
+
+        data_table.c.data[3]
+
+    * Path index operations::
+
+        data_table.c.data[('key_1', 'key_2', 5, ..., 'key_n')]
+
+    Additional operations are available from the dialect-specific versions
+    of :class:`.types.JSON`, such as :class:`.postgresql.JSON` and
+    :class:`.postgresql.JSONB`, each of which offer more operators than
+    just the basic type.
+
+    Index operations return an expression object whose type defaults to
+    :class:`.JSON` by default, so that further JSON-oriented instructions
+    may be called upon the result type.
+
+    The :class:`.JSON` type, when used with the SQLAlchemy ORM, does not
+    detect in-place mutations to the structure.  In order to detect these, the
+    :mod:`sqlalchemy.ext.mutable` extension must be used.  This extension will
+    allow "in-place" changes to the datastructure to produce events which
+    will be detected by the unit of work.  See the example at :class:`.HSTORE`
+    for a simple example involving a dictionary.
+
+    When working with NULL values, the :class:`.JSON` type recommends the
+    use of two specific constants in order to differentiate between a column
+    that evaluates to SQL NULL, e.g. no value, vs. the JSON-encoded string
+    of ``"null"``.   To insert or select against a value that is SQL NULL,
+    use the constant :func:`.null`::
+
+        from sqlalchemy import null
+        conn.execute(table.insert(), json_value=null())
+
+    To insert or select against a value that is JSON ``"null"``, use the
+    constant :attr:`.JSON.NULL`::
+
+        conn.execute(table.insert(), json_value=JSON.NULL)
+
+    The :class:`.JSON` type supports a flag
+    :paramref:`.JSON.none_as_null` which when set to True will result
+    in the Python constant ``None`` evaluating to the value of SQL
+    NULL, and when set to False results in the Python constant
+    ``None`` evaluating to the value of JSON ``"null"``.    The Python
+    value ``None`` may be used in conjunction with either
+    :attr:`.JSON.NULL` and :func:`.null` in order to indicate NULL
+    values, but care must be taken as to the value of the
+    :paramref:`.JSON.none_as_null` in these cases.
+
+    .. seealso::
+
+        :class:`.postgresql.JSON`
+
+        :class:`.postgresql.JSONB`
+
+        :class:`.mysql.JSON`
+
+    .. versionadded:: 1.1
+
+
+    """
+    __visit_name__ = 'JSON'
+
+    hashable = False
+    NULL = util.symbol('JSON_NULL')
+    """Describe the json value of NULL.
+
+    This value is used to force the JSON value of ``"null"`` to be
+    used as the value.   A value of Python ``None`` will be recognized
+    either as SQL NULL or JSON ``"null"``, based on the setting
+    of the :paramref:`.JSON.none_as_null` flag; the :attr:`.JSON.NULL`
+    constant can be used to always resolve to JSON ``"null"`` regardless
+    of this setting.  This is in contrast to the :func:`.sql.null` construct,
+    which always resolves to SQL NULL.  E.g.::
+
+        from sqlalchemy import null
+        from sqlalchemy.dialects.postgresql import JSON
+
+        obj1 = MyObject(json_value=null())  # will *always* insert SQL NULL
+        obj2 = MyObject(json_value=JSON.NULL)  # will *always* insert JSON string "null"
+
+        session.add_all([obj1, obj2])
+        session.commit()
+
+    """
+
+    def __init__(self, none_as_null=False):
+        """Construct a :class:`.types.JSON` type.
+
+        :param none_as_null=False: if True, persist the value ``None`` as a
+         SQL NULL value, not the JSON encoding of ``null``.   Note that
+         when this flag is False, the :func:`.null` construct can still
+         be used to persist a NULL value::
+
+             from sqlalchemy import null
+             conn.execute(table.insert(), data=null())
+
+         .. seealso::
+
+              :attr:`.types.JSON.NULL`
+
+         """
+        self.none_as_null = none_as_null
+
+    class JSONIndexType(TypeEngine):
+        """Placeholder for the datatype of a JSON index value.
+
+        This allows execution-time processing of JSON index values
+        for special syntaxes.
+
+        """
+
+    class JSONPathType(TypeEngine):
+        """Placeholder type for JSON path operations.
+
+        This allows execution-time processing of a path-based
+        index value into a specific SQL syntax.
+
+        """
+
+    class Comparator(Indexable.Comparator, Concatenable.Comparator):
+        """Define comparison operations for :class:`.types.JSON`."""
+
+        @util.dependencies('sqlalchemy.sql.default_comparator')
+        def _setup_getitem(self, default_comparator, index):
+            if not isinstance(index, util.string_types) and \
+                    isinstance(index, collections.Sequence):
+                index = default_comparator._check_literal(
+                    self.expr, operators.json_path_getitem_op,
+                    index, bindparam_type=JSON.JSONPathType
+                )
+
+                operator = operators.json_path_getitem_op
+            else:
+                index = default_comparator._check_literal(
+                    self.expr, operators.json_getitem_op,
+                    index, bindparam_type=JSON.JSONIndexType
+                )
+                operator = operators.json_getitem_op
+
+            return operator, index, self.type
+
+    comparator_factory = Comparator
+
+    @property
+    def should_evaluate_none(self):
+        return not self.none_as_null
+
+    @util.memoized_property
+    def _str_impl(self):
+        return String(convert_unicode=True)
+
+    def bind_processor(self, dialect):
+        string_process = self._str_impl.bind_processor(dialect)
+
+        json_serializer = dialect._json_serializer or json.dumps
+
+        def process(value):
+            if value is self.NULL:
+                value = None
+            elif isinstance(value, elements.Null) or (
+                value is None and self.none_as_null
+            ):
+                return None
+
+            serialized = json_serializer(value)
+            if string_process:
+                serialized = string_process(serialized)
+            return serialized
+
+        return process
+
+    def result_processor(self, dialect, coltype):
+        string_process = self._str_impl.result_processor(dialect, coltype)
+        json_deserializer = dialect._json_deserializer or json.loads
+
+        def process(value):
+            if value is None:
+                return None
+            if string_process:
+                value = string_process(value)
+            return json_deserializer(value)
+        return process
+
+
+class ARRAY(Indexable, Concatenable, TypeEngine):
     """Represent a SQL Array type.
 
     .. note::  This type serves as the basis for all ARRAY operations.
@@ -1506,17 +1719,17 @@ class Array(Indexable, Concatenable, TypeEngine):
        with PostgreSQL, as it provides additional operators specific
        to that backend.
 
-    :class:`.Array` is part of the Core in support of various SQL standard
+    :class:`.types.ARRAY` is part of the Core in support of various SQL standard
     functions such as :class:`.array_agg` which explicitly involve arrays;
     however, with the exception of the PostgreSQL backend and possibly
     some third-party dialects, no other SQLAlchemy built-in dialect has
     support for this type.
 
-    An :class:`.Array` type is constructed given the "type"
+    An :class:`.types.ARRAY` type is constructed given the "type"
     of element::
 
         mytable = Table("mytable", metadata,
-                Column("data", Array(Integer))
+                Column("data", ARRAY(Integer))
             )
 
     The above type represents an N-dimensional array,
@@ -1529,11 +1742,11 @@ class Array(Indexable, Concatenable, TypeEngine):
                 data=[1,2,3]
         )
 
-    The :class:`.Array` type can be constructed given a fixed number
+    The :class:`.types.ARRAY` type can be constructed given a fixed number
     of dimensions::
 
         mytable = Table("mytable", metadata,
-                Column("data", Array(Integer, dimensions=2))
+                Column("data", ARRAY(Integer, dimensions=2))
             )
 
     Sending a number of dimensions is optional, but recommended if the
@@ -1555,10 +1768,10 @@ class Array(Indexable, Concatenable, TypeEngine):
             >>> expr = table.c.column[5]  # returns ARRAY(Integer, dimensions=1)
             >>> expr = expr[6]  # returns Integer
 
-    For 1-dimensional arrays, an :class:`.Array` instance with no
+    For 1-dimensional arrays, an :class:`.types.ARRAY` instance with no
     dimension parameter will generally assume single-dimensional behaviors.
 
-    SQL expressions of type :class:`.Array` have support for "index" and
+    SQL expressions of type :class:`.types.ARRAY` have support for "index" and
     "slice" behavior.  The Python ``[]`` operator works normally here, given
     integer indexes or slices.  Arrays default to 1-based indexing.
     The operator produces binary expression
@@ -1575,9 +1788,9 @@ class Array(Indexable, Concatenable, TypeEngine):
             mytable.c.data[2:7]: [1, 2, 3]
         })
 
-    The :class:`.Array` type also provides for the operators
-    :meth:`.Array.Comparator.any` and :meth:`.Array.Comparator.all`.
-    The PostgreSQL-specific version of :class:`.Array` also provides additional
+    The :class:`.types.ARRAY` type also provides for the operators
+    :meth:`.types.ARRAY.Comparator.any` and :meth:`.types.ARRAY.Comparator.all`.
+    The PostgreSQL-specific version of :class:`.types.ARRAY` also provides additional
     operators.
 
     .. versionadded:: 1.1.0
@@ -1589,9 +1802,13 @@ class Array(Indexable, Concatenable, TypeEngine):
     """
     __visit_name__ = 'ARRAY'
 
+    zero_indexes = False
+    """if True, Python zero-based indexes should be interpreted as one-based
+    on the SQL expression side."""
+
     class Comparator(Indexable.Comparator, Concatenable.Comparator):
 
-        """Define comparison operations for :class:`.Array`.
+        """Define comparison operations for :class:`.types.ARRAY`.
 
         More operators are available on the dialect-specific form
         of this type.  See :class:`.postgresql.ARRAY.Comparator`.
@@ -1601,11 +1818,32 @@ class Array(Indexable, Concatenable, TypeEngine):
         def _setup_getitem(self, index):
             if isinstance(index, slice):
                 return_type = self.type
-            elif self.type.dimensions is None or self.type.dimensions == 1:
-                return_type = self.type.item_type
+                if self.type.zero_indexes:
+                    index = slice(
+                        index.start + 1,
+                        index.stop + 1,
+                        index.step
+                    )
+                index = Slice(
+                    _literal_as_binds(
+                        index.start, name=self.expr.key,
+                        type_=type_api.INTEGERTYPE),
+                    _literal_as_binds(
+                        index.stop, name=self.expr.key,
+                        type_=type_api.INTEGERTYPE),
+                    _literal_as_binds(
+                        index.step, name=self.expr.key,
+                        type_=type_api.INTEGERTYPE)
+                )
             else:
-                adapt_kw = {'dimensions': self.type.dimensions - 1}
-                return_type = self.type.adapt(self.type.__class__, **adapt_kw)
+                if self.type.zero_indexes:
+                    index += 1
+                if self.type.dimensions is None or self.type.dimensions == 1:
+                    return_type = self.type.item_type
+                else:
+                    adapt_kw = {'dimensions': self.type.dimensions - 1}
+                    return_type = self.type.adapt(
+                        self.type.__class__, **adapt_kw)
 
             return operators.getitem, index, return_type
 
@@ -1635,7 +1873,7 @@ class Array(Indexable, Concatenable, TypeEngine):
 
                 :func:`.sql.expression.any_`
 
-                :meth:`.Array.Comparator.all`
+                :meth:`.types.ARRAY.Comparator.all`
 
             """
             operator = operator if operator else operators.eq
@@ -1670,7 +1908,7 @@ class Array(Indexable, Concatenable, TypeEngine):
 
                 :func:`.sql.expression.all_`
 
-                :meth:`.Array.Comparator.any`
+                :meth:`.types.ARRAY.Comparator.any`
 
             """
             operator = operator if operator else operators.eq
@@ -1683,18 +1921,18 @@ class Array(Indexable, Concatenable, TypeEngine):
 
     def __init__(self, item_type, as_tuple=False, dimensions=None,
                  zero_indexes=False):
-        """Construct an :class:`.Array`.
+        """Construct an :class:`.types.ARRAY`.
 
         E.g.::
 
-          Column('myarray', Array(Integer))
+          Column('myarray', ARRAY(Integer))
 
         Arguments are:
 
         :param item_type: The data type of items of this array. Note that
           dimensionality is irrelevant here, so multi-dimensional arrays like
-          ``INTEGER[][]``, are constructed as ``Array(Integer)``, not as
-          ``Array(Array(Integer))`` or such.
+          ``INTEGER[][]``, are constructed as ``ARRAY(Integer)``, not as
+          ``ARRAY(ARRAY(Integer))`` or such.
 
         :param as_tuple=False: Specify whether return results
           should be converted to tuples from lists.  This parameter is
@@ -1706,7 +1944,7 @@ class Array(Indexable, Concatenable, TypeEngine):
          on the database, how it goes about interpreting Python and
          result values, as well as how expression behavior in conjunction
          with the "getitem" operator works.  See the description at
-         :class:`.Array` for additional detail.
+         :class:`.types.ARRAY` for additional detail.
 
         :param zero_indexes=False: when True, index values will be converted
          between Python zero-based and SQL one-based indexes, e.g.
@@ -1714,7 +1952,7 @@ class Array(Indexable, Concatenable, TypeEngine):
          to the database.
 
         """
-        if isinstance(item_type, Array):
+        if isinstance(item_type, ARRAY):
             raise ValueError("Do not nest ARRAY types; ARRAY(basetype) "
                              "handles multi-dimensional arrays of basetype")
         if isinstance(item_type, type):
diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index 15bfad831..87c776e8c 100644
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -487,6 +487,12 @@ class SuiteRequirements(Requirements):
         return exclusions.open()
 
     @property
+    def json_type(self):
+        """target platform implements a native JSON type."""
+
+        return exclusions.closed()
+
+    @property
     def precision_numerics_general(self):
         """target backend has general support for moderately high-precision
         numerics."""
diff --git a/lib/sqlalchemy/testing/suite/test_types.py b/lib/sqlalchemy/testing/suite/test_types.py
index 230aeb1e9..c6e882fb5 100644
--- a/lib/sqlalchemy/testing/suite/test_types.py
+++ b/lib/sqlalchemy/testing/suite/test_types.py
@@ -5,7 +5,7 @@ from ..assertions import eq_
 from ..config import requirements
 from sqlalchemy import Integer, Unicode, UnicodeText, select
 from sqlalchemy import Date, DateTime, Time, MetaData, String, \
-    Text, Numeric, Float, literal, Boolean
+    Text, Numeric, Float, literal, Boolean, cast, null, JSON
 from ..schema import Table, Column
 from ... import testing
 import decimal
@@ -586,7 +586,246 @@ class BooleanTest(_LiteralRoundTripFixture, fixtures.TablesTest):
         )
 
 
-__all__ = ('UnicodeVarcharTest', 'UnicodeTextTest',
+class JSONTest(_LiteralRoundTripFixture, fixtures.TablesTest):
+    __requires__ = 'json_type',
+    __backend__ = True
+
+    datatype = JSON
+
+    data1 = {
+        "key1": "value1",
+        "key2": "value2"
+    }
+
+    data2 = {
+        "Key 'One'": "value1",
+        "key two": "value2",
+        "key three": "value ' three '"
+    }
+
+    data3 = {
+        "key1": [1, 2, 3],
+        "key2": ["one", "two", "three"],
+        "key3": [{"four": "five"}, {"six": "seven"}]
+    }
+
+    data4 = ["one", "two", "three"]
+
+    data5 = {
+        "nested": {
+            "elem1": [
+                {"a": "b", "c": "d"},
+                {"e": "f", "g": "h"}
+            ],
+            "elem2": {
+                "elem3": {"elem4": "elem5"}
+            }
+        }
+    }
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table('data_table', metadata,
+              Column('id', Integer, primary_key=True),
+              Column('name', String(30), nullable=False),
+              Column('data', cls.datatype),
+              Column('nulldata', cls.datatype(none_as_null=True))
+              )
+
+    def test_round_trip_data1(self):
+        self._test_round_trip(self.data1)
+
+    def _test_round_trip(self, data_element):
+        data_table = self.tables.data_table
+
+        config.db.execute(
+            data_table.insert(),
+            {'name': 'row1', 'data': data_element}
+        )
+
+        row = config.db.execute(
+            select([
+                data_table.c.data,
+            ])
+        ).first()
+
+        eq_(row, (data_element, ))
+
+    def test_round_trip_none_as_sql_null(self):
+        col = self.tables.data_table.c['nulldata']
+
+        with config.db.connect() as conn:
+            conn.execute(
+                self.tables.data_table.insert(),
+                {"name": "r1", "data": None}
+            )
+
+            eq_(
+                conn.scalar(
+                    select([self.tables.data_table.c.name]).
+                    where(col.is_(null()))
+                ),
+                "r1"
+            )
+
+            eq_(
+                conn.scalar(
+                    select([col])
+                ),
+                None
+            )
+
+    def test_round_trip_json_null_as_json_null(self):
+        col = self.tables.data_table.c['data']
+
+        with config.db.connect() as conn:
+            conn.execute(
+                self.tables.data_table.insert(),
+                {"name": "r1", "data": JSON.NULL}
+            )
+
+            eq_(
+                conn.scalar(
+                    select([self.tables.data_table.c.name]).
+                    where(cast(col, String) == 'null')
+                ),
+                "r1"
+            )
+
+            eq_(
+                conn.scalar(
+                    select([col])
+                ),
+                None
+            )
+
+    def test_round_trip_none_as_json_null(self):
+        col = self.tables.data_table.c['data']
+
+        with config.db.connect() as conn:
+            conn.execute(
+                self.tables.data_table.insert(),
+                {"name": "r1", "data": None}
+            )
+
+            eq_(
+                conn.scalar(
+                    select([self.tables.data_table.c.name]).
+                    where(cast(col, String) == 'null')
+                ),
+                "r1"
+            )
+
+            eq_(
+                conn.scalar(
+                    select([col])
+                ),
+                None
+            )
+
+    def _criteria_fixture(self):
+        config.db.execute(
+            self.tables.data_table.insert(),
+            [{"name": "r1", "data": self.data1},
+             {"name": "r2", "data": self.data2},
+             {"name": "r3", "data": self.data3},
+             {"name": "r4", "data": self.data4},
+             {"name": "r5", "data": self.data5}]
+        )
+
+    def _test_index_criteria(self, crit, expected):
+        self._criteria_fixture()
+        with config.db.connect() as conn:
+            eq_(
+                conn.scalar(
+                    select([self.tables.data_table.c.name]).
+                    where(crit)
+                ),
+                expected
+            )
+
+    def test_crit_spaces_in_key(self):
+        col = self.tables.data_table.c['data']
+        self._test_index_criteria(
+            cast(col["key two"], String) == '"value2"',
+            "r2"
+        )
+
+    def test_crit_simple_int(self):
+        col = self.tables.data_table.c['data']
+        self._test_index_criteria(
+            cast(col[1], String) == '"two"',
+            "r4"
+        )
+
+    def test_crit_mixed_path(self):
+        col = self.tables.data_table.c['data']
+        self._test_index_criteria(
+            cast(col[("key3", 1, "six")], String) == '"seven"',
+            "r3"
+        )
+
+    def test_crit_string_path(self):
+        col = self.tables.data_table.c['data']
+        self._test_index_criteria(
+            cast(col[("nested", "elem2", "elem3", "elem4")], String)
+            == '"elem5"',
+            "r5"
+        )
+
+    def test_unicode_round_trip(self):
+        s = select([
+            cast(
+                {
+                    util.u('réveillé'): util.u('réveillé'),
+                    "data": {"k1": util.u('drôle')}
+                },
+                self.datatype
+            )
+        ])
+        eq_(
+            config.db.scalar(s),
+            {
+                util.u('réveillé'): util.u('réveillé'),
+                "data": {"k1": util.u('drôle')}
+            },
+        )
+
+    def test_eval_none_flag_orm(self):
+        from sqlalchemy.ext.declarative import declarative_base
+        from sqlalchemy.orm import Session
+
+        Base = declarative_base()
+
+        class Data(Base):
+            __table__ = self.tables.data_table
+
+        s = Session(testing.db)
+
+        d1 = Data(name='d1', data=None, nulldata=None)
+        s.add(d1)
+        s.commit()
+
+        s.bulk_insert_mappings(
+            Data, [{"name": "d2", "data": None, "nulldata": None}]
+        )
+        eq_(
+            s.query(
+                cast(self.tables.data_table.c.data, String),
+                cast(self.tables.data_table.c.nulldata, String)
+            ).filter(self.tables.data_table.c.name == 'd1').first(),
+            ("null", None)
+        )
+        eq_(
+            s.query(
+                cast(self.tables.data_table.c.data, String),
+                cast(self.tables.data_table.c.nulldata, String)
+            ).filter(self.tables.data_table.c.name == 'd2').first(),
+            ("null", None)
+        )
+
+
+__all__ = ('UnicodeVarcharTest', 'UnicodeTextTest', 'JSONTest',
            'DateTest', 'DateTimeTest', 'TextTest',
            'NumericTest', 'IntegerTest',
            'DateTimeHistoricTest', 'DateTimeCoercedToDateTimeTest',
diff --git a/lib/sqlalchemy/types.py b/lib/sqlalchemy/types.py
index d82e683d9..ac6d3b439 100644
--- a/lib/sqlalchemy/types.py
+++ b/lib/sqlalchemy/types.py
@@ -17,7 +17,7 @@ __all__ = ['TypeEngine', 'TypeDecorator', 'UserDefinedType',
            'SmallInteger', 'BigInteger', 'Numeric', 'Float', 'DateTime',
            'Date', 'Time', 'LargeBinary', 'Binary', 'Boolean', 'Unicode',
            'Concatenable', 'UnicodeText', 'PickleType', 'Interval', 'Enum',
-           'Indexable', 'Array']
+           'Indexable', 'ARRAY', 'JSON']
 
 from .sql.type_api import (
     adapt_type,
@@ -28,7 +28,7 @@ from .sql.type_api import (
     UserDefinedType
 )
 from .sql.sqltypes import (
-    Array,
+    ARRAY,
     BIGINT,
     BINARY,
     BLOB,
@@ -53,6 +53,7 @@ from .sql.sqltypes import (
     INTEGER,
     Integer,
     Interval,
+    JSON,
     LargeBinary,
     MatchType,
     NCHAR,