Merged zzzeek/sqlalchemy into master

11 files changed, 591 insertions, 184 deletions

diff --git a/lib/sqlalchemy/sql/__init__.py b/lib/sqlalchemy/sql/__init__.py
index fa2cf2399..f4ad3ec00 100644
--- a/lib/sqlalchemy/sql/__init__.py
+++ b/lib/sqlalchemy/sql/__init__.py
@@ -66,6 +66,7 @@ from .expression import (
     union,
     union_all,
     update,
+    within_group
 )
 
 from .visitors import ClauseVisitor
diff --git a/lib/sqlalchemy/sql/base.py b/lib/sqlalchemy/sql/base.py
index eed079238..48b9a8a2b 100644
--- a/lib/sqlalchemy/sql/base.py
+++ b/lib/sqlalchemy/sql/base.py
@@ -449,11 +449,10 @@ class ColumnCollection(util.OrderedProperties):
 
     """
 
-    __slots__ = '_all_col_set', '_all_columns'
+    __slots__ = '_all_columns'
 
     def __init__(self, *columns):
         super(ColumnCollection, self).__init__()
-        object.__setattr__(self, '_all_col_set', util.column_set())
         object.__setattr__(self, '_all_columns', [])
         for c in columns:
             self.add(c)
@@ -482,14 +481,11 @@ class ColumnCollection(util.OrderedProperties):
             other = self[column.name]
             if other.name == other.key:
                 remove_col = other
-                self._all_col_set.remove(other)
                 del self._data[other.key]
 
         if column.key in self._data:
             remove_col = self._data[column.key]
-            self._all_col_set.remove(remove_col)
 
-        self._all_col_set.add(column)
         self._data[column.key] = column
         if remove_col is not None:
             self._all_columns[:] = [column if c is remove_col
@@ -534,7 +530,6 @@ class ColumnCollection(util.OrderedProperties):
             # in a _make_proxy operation
             util.memoized_property.reset(value, "proxy_set")
 
-        self._all_col_set.add(value)
         self._all_columns.append(value)
         self._data[key] = value
 
@@ -543,22 +538,20 @@ class ColumnCollection(util.OrderedProperties):
 
     def remove(self, column):
         del self._data[column.key]
-        self._all_col_set.remove(column)
         self._all_columns[:] = [
             c for c in self._all_columns if c is not column]
 
     def update(self, iter):
         cols = list(iter)
+        all_col_set = set(self._all_columns)
         self._all_columns.extend(
-            c for label, c in cols if c not in self._all_col_set)
-        self._all_col_set.update(c for label, c in cols)
+            c for label, c in cols if c not in all_col_set)
         self._data.update((label, c) for label, c in cols)
 
     def extend(self, iter):
         cols = list(iter)
-        self._all_columns.extend(c for c in cols if c not in
-                                 self._all_col_set)
-        self._all_col_set.update(cols)
+        all_col_set = set(self._all_columns)
+        self._all_columns.extend(c for c in cols if c not in all_col_set)
         self._data.update((c.key, c) for c in cols)
 
     __hash__ = None
@@ -584,22 +577,18 @@ class ColumnCollection(util.OrderedProperties):
     def __setstate__(self, state):
         object.__setattr__(self, '_data', state['_data'])
         object.__setattr__(self, '_all_columns', state['_all_columns'])
-        object.__setattr__(
-            self, '_all_col_set', util.column_set(state['_all_columns']))
 
     def contains_column(self, col):
-        # this has to be done via set() membership
-        return col in self._all_col_set
+        existing = self._data.get(col.key)
+        return existing is not None and hash(existing) == hash(col)
 
     def as_immutable(self):
-        return ImmutableColumnCollection(
-            self._data, self._all_col_set, self._all_columns)
+        return ImmutableColumnCollection(self._data, self._all_columns)
 
 
 class ImmutableColumnCollection(util.ImmutableProperties, ColumnCollection):
-    def __init__(self, data, colset, all_columns):
+    def __init__(self, data, all_columns):
         util.ImmutableProperties.__init__(self, data)
-        object.__setattr__(self, '_all_col_set', colset)
         object.__setattr__(self, '_all_columns', all_columns)
 
     extend = remove = util.ImmutableProperties._immutable
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 6766c99b7..492999d16 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -167,25 +167,39 @@ class Compiled(object):
     _cached_metadata = None
 
     def __init__(self, dialect, statement, bind=None,
+                 schema_translate_map=None,
                  compile_kwargs=util.immutabledict()):
-        """Construct a new ``Compiled`` object.
+        """Construct a new :class:`.Compiled` object.
 
-        :param dialect: ``Dialect`` to compile against.
+        :param dialect: :class:`.Dialect` to compile against.
 
-        :param statement: ``ClauseElement`` to be compiled.
+        :param statement: :class:`.ClauseElement` to be compiled.
 
         :param bind: Optional Engine or Connection to compile this
           statement against.
 
+        :param schema_translate_map: dictionary of schema names to be
+         translated when forming the resultant SQL
+
+         .. versionadded:: 1.1
+
+         .. seealso::
+
+            :ref:`schema_translating`
+
         :param compile_kwargs: additional kwargs that will be
          passed to the initial call to :meth:`.Compiled.process`.
 
-         .. versionadded:: 0.8
 
         """
 
         self.dialect = dialect
         self.bind = bind
+        self.preparer = self.dialect.identifier_preparer
+        if schema_translate_map:
+            self.preparer = self.preparer._with_schema_translate(
+                schema_translate_map)
+
         if statement is not None:
             self.statement = statement
             self.can_execute = statement.supports_execution
@@ -286,12 +300,11 @@ class _CompileLabel(visitors.Visitable):
     def self_group(self, **kw):
         return self
 
-class SQLCompiler(Compiled):
 
-    """Default implementation of Compiled.
+class SQLCompiler(Compiled):
+    """Default implementation of :class:`.Compiled`.
 
-    Compiles ClauseElements into SQL strings.   Uses a similar visit
-    paradigm as visitors.ClauseVisitor but implements its own traversal.
+    Compiles :class:`.ClauseElement` objects into SQL strings.
 
     """
 
@@ -305,6 +318,8 @@ class SQLCompiler(Compiled):
     INSERT/UPDATE/DELETE
     """
 
+    isplaintext = False
+
     returning = None
     """holds the "returning" collection of columns if
     the statement is CRUD and defines returning columns
@@ -330,19 +345,34 @@ class SQLCompiler(Compiled):
     driver/DB enforces this
     """
 
+    _textual_ordered_columns = False
+    """tell the result object that the column names as rendered are important,
+    but they are also "ordered" vs. what is in the compiled object here.
+    """
+
+    _ordered_columns = True
+    """
+    if False, means we can't be sure the list of entries
+    in _result_columns is actually the rendered order.  Usually
+    True unless using an unordered TextAsFrom.
+    """
+
     def __init__(self, dialect, statement, column_keys=None,
                  inline=False, **kwargs):
-        """Construct a new ``DefaultCompiler`` object.
+        """Construct a new :class:`.SQLCompiler` object.
 
-        dialect
-          Dialect to be used
+        :param dialect: :class:`.Dialect` to be used
 
-        statement
-          ClauseElement to be compiled
+        :param statement: :class:`.ClauseElement` to be compiled
 
-        column_keys
-          a list of column names to be compiled into an INSERT or UPDATE
-          statement.
+        :param column_keys:  a list of column names to be compiled into an
+         INSERT or UPDATE statement.
+
+        :param inline: whether to generate INSERT statements as "inline", e.g.
+         not formatted to return any generated defaults
+
+        :param kwargs: additional keyword arguments to be consumed by the
+         superclass.
 
         """
         self.column_keys = column_keys
@@ -368,11 +398,6 @@ class SQLCompiler(Compiled):
         # column targeting
         self._result_columns = []
 
-        # if False, means we can't be sure the list of entries
-        # in _result_columns is actually the rendered order.   This
-        # gets flipped when we use TextAsFrom, for example.
-        self._ordered_columns = True
-
         # true if the paramstyle is positional
         self.positional = dialect.positional
         if self.positional:
@@ -381,8 +406,6 @@ class SQLCompiler(Compiled):
 
         self.ctes = None
 
-        # an IdentifierPreparer that formats the quoting of identifiers
-        self.preparer = dialect.identifier_preparer
         self.label_length = dialect.label_length \
             or dialect.max_identifier_length
 
@@ -649,8 +672,11 @@ class SQLCompiler(Compiled):
         if table is None or not include_table or not table.named_with_column:
             return name
         else:
-            if table.schema:
-                schema_prefix = self.preparer.quote_schema(table.schema) + '.'
+            effective_schema = self.preparer.schema_for_object(table)
+
+            if effective_schema:
+                schema_prefix = self.preparer.quote_schema(
+                    effective_schema) + '.'
             else:
                 schema_prefix = ''
             tablename = table.name
@@ -688,6 +714,9 @@ class SQLCompiler(Compiled):
             else:
                 return self.bindparam_string(name, **kw)
 
+        if not self.stack:
+            self.isplaintext = True
+
         # un-escape any \:params
         return BIND_PARAMS_ESC.sub(
             lambda m: m.group(1),
@@ -711,7 +740,8 @@ class SQLCompiler(Compiled):
             ) or entry.get('need_result_map_for_nested', False)
 
         if populate_result_map:
-            self._ordered_columns = False
+            self._ordered_columns = \
+                self._textual_ordered_columns = taf.positional
             for c in taf.column_args:
                 self.process(c, within_columns_clause=True,
                              add_to_result_map=self._add_to_result_map)
@@ -873,22 +903,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:
@@ -922,7 +958,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:
@@ -1298,7 +1334,7 @@ class SQLCompiler(Compiled):
             add_to_result_map = lambda keyname, name, objects, type_: \
                 self._add_to_result_map(
                     keyname, name,
-                    objects + (column,), type_)
+                    (column,) + objects, type_)
         else:
             col_expr = column
             if populate_result_map:
@@ -1386,7 +1422,7 @@ class SQLCompiler(Compiled):
         """Rewrite any "a JOIN (b JOIN c)" expression as
         "a JOIN (select * from b JOIN c) AS anon", to support
         databases that can't parse a parenthesized join correctly
-        (i.e. sqlite the main one).
+        (i.e. sqlite < 3.7.16).
 
         """
         cloned = {}
@@ -1801,8 +1837,10 @@ class SQLCompiler(Compiled):
     def visit_table(self, table, asfrom=False, iscrud=False, ashint=False,
                     fromhints=None, use_schema=True, **kwargs):
         if asfrom or ashint:
-            if use_schema and getattr(table, "schema", None):
-                ret = self.preparer.quote_schema(table.schema) + \
+            effective_schema = self.preparer.schema_for_object(table)
+
+            if use_schema and effective_schema:
+                ret = self.preparer.quote_schema(effective_schema) + \
                     "." + self.preparer.quote(table.name)
             else:
                 ret = self.preparer.quote(table.name)
@@ -2080,6 +2118,30 @@ class SQLCompiler(Compiled):
             self.preparer.format_savepoint(savepoint_stmt)
 
 
+class StrSQLCompiler(SQLCompiler):
+    """"a compiler subclass with a few non-standard SQL features allowed.
+
+    Used for stringification of SQL statements when a real dialect is not
+    available.
+
+    """
+
+    def visit_getitem_binary(self, binary, operator, **kw):
+        return "%s[%s]" % (
+            self.process(binary.left, **kw),
+            self.process(binary.right, **kw)
+        )
+
+    def returning_clause(self, stmt, returning_cols):
+
+        columns = [
+            self._label_select_column(None, c, True, False, {})
+            for c in elements._select_iterables(returning_cols)
+        ]
+
+        return 'RETURNING ' + ', '.join(columns)
+
+
 class DDLCompiler(Compiled):
 
     @util.memoized_property
@@ -2090,10 +2152,6 @@ class DDLCompiler(Compiled):
     def type_compiler(self):
         return self.dialect.type_compiler
 
-    @property
-    def preparer(self):
-        return self.dialect.identifier_preparer
-
     def construct_params(self, params=None):
         return None
 
@@ -2103,7 +2161,7 @@ class DDLCompiler(Compiled):
         if isinstance(ddl.target, schema.Table):
             context = context.copy()
 
-            preparer = self.dialect.identifier_preparer
+            preparer = self.preparer
             path = preparer.format_table_seq(ddl.target)
             if len(path) == 1:
                 table, sch = path[0], ''
@@ -2129,7 +2187,7 @@ class DDLCompiler(Compiled):
 
     def visit_create_table(self, create):
         table = create.element
-        preparer = self.dialect.identifier_preparer
+        preparer = self.preparer
 
         text = "\nCREATE "
         if table._prefixes:
@@ -2256,9 +2314,12 @@ class DDLCompiler(Compiled):
             index, include_schema=True)
 
     def _prepared_index_name(self, index, include_schema=False):
-        if include_schema and index.table is not None and index.table.schema:
-            schema = index.table.schema
-            schema_name = self.preparer.quote_schema(schema)
+        if index.table is not None:
+            effective_schema = self.preparer.schema_for_object(index.table)
+        else:
+            effective_schema = None
+        if include_schema and effective_schema:
+            schema_name = self.preparer.quote_schema(effective_schema)
         else:
             schema_name = None
 
@@ -2386,7 +2447,7 @@ class DDLCompiler(Compiled):
         return text
 
     def visit_foreign_key_constraint(self, constraint):
-        preparer = self.dialect.identifier_preparer
+        preparer = self.preparer
         text = ""
         if constraint.name is not None:
             formatted_name = self.preparer.format_constraint(constraint)
@@ -2603,6 +2664,17 @@ class GenericTypeCompiler(TypeCompiler):
         return type_.get_col_spec(**kw)
 
 
+class StrSQLTypeCompiler(GenericTypeCompiler):
+    def __getattr__(self, key):
+        if key.startswith("visit_"):
+            return self._visit_unknown
+        else:
+            raise AttributeError(key)
+
+    def _visit_unknown(self, type_, **kw):
+        return "%s" % type_.__class__.__name__
+
+
 class IdentifierPreparer(object):
 
     """Handle quoting and case-folding of identifiers based on options."""
@@ -2613,6 +2685,8 @@ class IdentifierPreparer(object):
 
     illegal_initial_characters = ILLEGAL_INITIAL_CHARACTERS
 
+    schema_for_object = schema._schema_getter(None)
+
     def __init__(self, dialect, initial_quote='"',
                  final_quote=None, escape_quote='"', omit_schema=False):
         """Construct a new ``IdentifierPreparer`` object.
@@ -2637,6 +2711,12 @@ class IdentifierPreparer(object):
         self.omit_schema = omit_schema
         self._strings = {}
 
+    def _with_schema_translate(self, schema_translate_map):
+        prep = self.__class__.__new__(self.__class__)
+        prep.__dict__.update(self.__dict__)
+        prep.schema_for_object = schema._schema_getter(schema_translate_map)
+        return prep
+
     def _escape_identifier(self, value):
         """Escape an identifier.
 
@@ -2709,9 +2789,12 @@ class IdentifierPreparer(object):
 
     def format_sequence(self, sequence, use_schema=True):
         name = self.quote(sequence.name)
+
+        effective_schema = self.schema_for_object(sequence)
+
         if (not self.omit_schema and use_schema and
-                sequence.schema is not None):
-            name = self.quote_schema(sequence.schema) + "." + name
+                effective_schema is not None):
+            name = self.quote_schema(effective_schema) + "." + name
         return name
 
     def format_label(self, label, name=None):
@@ -2740,9 +2823,12 @@ class IdentifierPreparer(object):
         if name is None:
             name = table.name
         result = self.quote(name)
+
+        effective_schema = self.schema_for_object(table)
+
         if not self.omit_schema and use_schema \
-                and getattr(table, "schema", None):
-            result = self.quote_schema(table.schema) + "." + result
+                and effective_schema:
+            result = self.quote_schema(effective_schema) + "." + result
         return result
 
     def format_schema(self, name, quote=None):
@@ -2781,9 +2867,11 @@ class IdentifierPreparer(object):
         # ('database', 'owner', etc.) could override this and return
         # a longer sequence.
 
+        effective_schema = self.schema_for_object(table)
+
         if not self.omit_schema and use_schema and \
-                getattr(table, 'schema', None):
-            return (self.quote_schema(table.schema),
+                effective_schema:
+            return (self.quote_schema(effective_schema),
                     self.format_table(table, use_schema=False))
         else:
             return (self.format_table(table, use_schema=False), )
diff --git a/lib/sqlalchemy/sql/ddl.py b/lib/sqlalchemy/sql/ddl.py
index 71018f132..7953b61b8 100644
--- a/lib/sqlalchemy/sql/ddl.py
+++ b/lib/sqlalchemy/sql/ddl.py
@@ -679,13 +679,16 @@ class SchemaGenerator(DDLBase):
 
     def _can_create_table(self, table):
         self.dialect.validate_identifier(table.name)
-        if table.schema:
-            self.dialect.validate_identifier(table.schema)
+        effective_schema = self.connection.schema_for_object(table)
+        if effective_schema:
+            self.dialect.validate_identifier(effective_schema)
         return not self.checkfirst or \
             not self.dialect.has_table(self.connection,
-                                       table.name, schema=table.schema)
+                                       table.name, schema=effective_schema)
 
     def _can_create_sequence(self, sequence):
+        effective_schema = self.connection.schema_for_object(sequence)
+
         return self.dialect.supports_sequences and \
             (
                 (not self.dialect.sequences_optional or
@@ -695,7 +698,7 @@ class SchemaGenerator(DDLBase):
                     not self.dialect.has_sequence(
                         self.connection,
                         sequence.name,
-                        schema=sequence.schema)
+                        schema=effective_schema)
                 )
             )
 
@@ -882,12 +885,14 @@ class SchemaDropper(DDLBase):
 
     def _can_drop_table(self, table):
         self.dialect.validate_identifier(table.name)
-        if table.schema:
-            self.dialect.validate_identifier(table.schema)
+        effective_schema = self.connection.schema_for_object(table)
+        if effective_schema:
+            self.dialect.validate_identifier(effective_schema)
         return not self.checkfirst or self.dialect.has_table(
-            self.connection, table.name, schema=table.schema)
+            self.connection, table.name, schema=effective_schema)
 
     def _can_drop_sequence(self, sequence):
+        effective_schema = self.connection.schema_for_object(sequence)
         return self.dialect.supports_sequences and \
             ((not self.dialect.sequences_optional or
               not sequence.optional) and
@@ -895,7 +900,7 @@ class SchemaDropper(DDLBase):
                  self.dialect.has_sequence(
                      self.connection,
                      sequence.name,
-                     schema=sequence.schema))
+                     schema=effective_schema))
              )
 
     def visit_index(self, index):
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..fe2fecce8 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -429,7 +429,7 @@ class ClauseElement(Visitable):
                 dialect = self.bind.dialect
                 bind = self.bind
             else:
-                dialect = default.DefaultDialect()
+                dialect = default.StrCompileDialect()
         return self._compiler(dialect, bind=bind, **kw)
 
     def _compiler(self, dialect, **kw):
@@ -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
@@ -1275,17 +1276,16 @@ class TextClause(Executable, ClauseElement):
             for id, name in connection.execute(t):
                 print(id, name)
 
-        The :func:`.text` construct is used internally in cases when
-        a literal string is specified for part of a larger query, such as
-        when a string is specified to the :meth:`.Select.where` method of
-        :class:`.Select`.  In those cases, the same
-        bind parameter syntax is applied::
+        The :func:`.text` construct is used in cases when
+        a literal string SQL fragment is specified as part of a larger query,
+        such as for the WHERE clause of a SELECT statement::
 
-            s = select([users.c.id, users.c.name]).where("id=:user_id")
+            s = select([users.c.id, users.c.name]).where(text("id=:user_id"))
             result = connection.execute(s, user_id=12)
 
-        Using :func:`.text` explicitly usually implies the construction
-        of a full, standalone statement.   As such, SQLAlchemy refers
+        :func:`.text` is also used for the construction
+        of a full, standalone statement using plain text.
+        As such, SQLAlchemy refers
         to it as an :class:`.Executable` object, and it supports
         the :meth:`Executable.execution_options` method.  For example,
         a :func:`.text` construct that should be subject to "autocommit"
@@ -1360,6 +1360,12 @@ class TextClause(Executable, ClauseElement):
           .. deprecated:: 0.9.0 the :meth:`.TextClause.columns` method
              supersedes the ``typemap`` argument to :func:`.text`.
 
+        .. seealso::
+
+            :ref:`sqlexpression_text` - in the Core tutorial
+
+            :ref:`orm_tutorial_literal_sql` - in the ORM tutorial
+
         """
         stmt = TextClause(text, bind=bind)
         if bindparams:
@@ -1485,9 +1491,17 @@ class TextClause(Executable, ClauseElement):
                         mytable.join(stmt, mytable.c.name == stmt.c.name)
                     ).where(stmt.c.id > 5)
 
-        Above, we used untyped :func:`.column` elements.  These can also have
-        types specified, which will impact how the column behaves in
-        expressions as well as determining result set behavior::
+        Above, we pass a series of :func:`.column` elements to the
+        :meth:`.TextClause.columns` method positionally.  These :func:`.column`
+        elements now become first class elements upon the :attr:`.TextAsFrom.c`
+        column collection, just like any other selectable.
+
+        The column expressions we pass to :meth:`.TextClause.columns` may
+        also be typed; when we do so, these :class:`.TypeEngine` objects become
+        the effective return type of the column, so that SQLAlchemy's
+        result-set-processing systems may be used on the return values.
+        This is often needed for types such as date or boolean types, as well
+        as for unicode processing on some dialect configurations::
 
             stmt = text("SELECT id, name, timestamp FROM some_table")
             stmt = stmt.columns(
@@ -1499,9 +1513,8 @@ class TextClause(Executable, ClauseElement):
             for id, name, timestamp in connection.execute(stmt):
                 print(id, name, timestamp)
 
-        Keyword arguments allow just the names and types of columns to be
-        specified, where the :func:`.column` elements will be generated
-        automatically::
+        As a shortcut to the above syntax, keyword arguments referring to
+        types alone may be used, if only type conversion is needed::
 
             stmt = text("SELECT id, name, timestamp FROM some_table")
             stmt = stmt.columns(
@@ -1513,6 +1526,31 @@ class TextClause(Executable, ClauseElement):
             for id, name, timestamp in connection.execute(stmt):
                 print(id, name, timestamp)
 
+        The positional form of :meth:`.TextClause.columns` also provides
+        the unique feature of **positional column targeting**, which is
+        particularly useful when using the ORM with complex textual queries.
+        If we specify the columns from our model to :meth:`.TextClause.columns`,
+        the result set will match to those columns positionally, meaning the
+        name or origin of the column in the textual SQL doesn't matter::
+
+            stmt = text("SELECT users.id, addresses.id, users.id, "
+                 "users.name, addresses.email_address AS email "
+                 "FROM users JOIN addresses ON users.id=addresses.user_id "
+                 "WHERE users.id = 1").columns(
+                    User.id,
+                    Address.id,
+                    Address.user_id,
+                    User.name,
+                    Address.email_address
+                 )
+
+            query = session.query(User).from_statement(stmt).options(
+                contains_eager(User.addresses))
+
+        .. versionadded:: 1.1 the :meth:`.TextClause.columns` method now
+           offers positional column targeting in the result set when
+           the column expressions are passed purely positionally.
+
         The :meth:`.TextClause.columns` method provides a direct
         route to calling :meth:`.FromClause.alias` as well as
         :meth:`.SelectBase.cte` against a textual SELECT statement::
@@ -1526,15 +1564,22 @@ class TextClause(Executable, ClauseElement):
            :meth:`.TextClause.columns` method.  This method supersedes the
            ``typemap`` argument to :func:`.text`.
 
+
         """
 
-        input_cols = [
+        positional_input_cols = [
             ColumnClause(col.key, types.pop(col.key))
             if col.key in types
             else col
             for col in cols
-        ] + [ColumnClause(key, type_) for key, type_ in types.items()]
-        return selectable.TextAsFrom(self, input_cols)
+        ]
+        keyed_input_cols = [
+            ColumnClause(key, type_) for key, type_ in types.items()]
+
+        return selectable.TextAsFrom(
+            self,
+            positional_input_cols + keyed_input_cols,
+            positional=bool(positional_input_cols) and not keyed_input_cols)
 
     @property
     def type(self):
@@ -1952,11 +1997,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 +3683,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/schema.py b/lib/sqlalchemy/sql/schema.py
index b244d746c..0626cb2b4 100644
--- a/lib/sqlalchemy/sql/schema.py
+++ b/lib/sqlalchemy/sql/schema.py
@@ -30,20 +30,19 @@ as components in SQL expressions.
 """
 from __future__ import absolute_import
 
-import inspect
 from .. import exc, util, event, inspection
 from .base import SchemaEventTarget, DialectKWArgs
+import operator
 from . import visitors
 from . import type_api
 from .base import _bind_or_error, ColumnCollection
-from .elements import ClauseElement, ColumnClause, _truncated_label, \
+from .elements import ClauseElement, ColumnClause, \
     _as_truncated, TextClause, _literal_as_text,\
-    ColumnElement, _find_columns, quoted_name
+    ColumnElement, quoted_name
 from .selectable import TableClause
 import collections
 import sqlalchemy
 from . import ddl
-import types
 
 RETAIN_SCHEMA = util.symbol('retain_schema')
 
@@ -3862,3 +3861,52 @@ class ThreadLocalMetaData(MetaData):
         for e in self.__engines.values():
             if hasattr(e, 'dispose'):
                 e.dispose()
+
+
+class _SchemaTranslateMap(object):
+    """Provide translation of schema names based on a mapping.
+
+    Also provides helpers for producing cache keys and optimized
+    access when no mapping is present.
+
+    Used by the :paramref:`.Connection.execution_options.schema_translate_map`
+    feature.
+
+    .. versionadded:: 1.1
+
+
+    """
+    __slots__ = 'map_', '__call__', 'hash_key', 'is_default'
+
+    _default_schema_getter = operator.attrgetter("schema")
+
+    def __init__(self, map_):
+        self.map_ = map_
+        if map_ is not None:
+            def schema_for_object(obj):
+                effective_schema = self._default_schema_getter(obj)
+                effective_schema = map_.get(effective_schema, effective_schema)
+                return effective_schema
+            self.__call__ = schema_for_object
+            self.hash_key = ";".join(
+                "%s=%s" % (k, map_[k])
+                for k in sorted(map_, key=str)
+            )
+            self.is_default = False
+        else:
+            self.hash_key = 0
+            self.__call__ = self._default_schema_getter
+            self.is_default = True
+
+    @classmethod
+    def _schema_getter(cls, map_):
+        if map_ is None:
+            return _default_schema_map
+        elif isinstance(map_, _SchemaTranslateMap):
+            return map_
+        else:
+            return _SchemaTranslateMap(map_)
+
+_default_schema_map = _SchemaTranslateMap(None)
+_schema_getter = _SchemaTranslateMap._schema_getter
+
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 73341053d..1955fc934 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -3420,9 +3420,10 @@ class TextAsFrom(SelectBase):
 
     _textual = True
 
-    def __init__(self, text, columns):
+    def __init__(self, text, columns, positional=False):
         self.element = text
         self.column_args = columns
+        self.positional = positional
 
     @property
     def _bind(self):
diff --git a/lib/sqlalchemy/sql/sqltypes.py b/lib/sqlalchemy/sql/sqltypes.py
index 4abb9b15a..84bfca026 100644
--- a/lib/sqlalchemy/sql/sqltypes.py
+++ b/lib/sqlalchemy/sql/sqltypes.py
@@ -11,9 +11,13 @@
 
 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 +89,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
             )
@@ -909,9 +909,9 @@ class LargeBinary(_Binary):
 
     """A type for large binary byte data.
 
-    The Binary type generates BLOB or BYTEA when tables are created,
-    and also converts incoming values using the ``Binary`` callable
-    provided by each DB-API.
+    The :class:`.LargeBinary` type corresponds to a large and/or unlengthed
+    binary type for the target platform, such as BLOB on MySQL and BYTEA for
+    Postgresql.  It also handles the necessary conversions for the DBAPI.
 
     """
 
@@ -922,13 +922,8 @@ class LargeBinary(_Binary):
         Construct a LargeBinary type.
 
         :param length: optional, a length for the column for use in
-          DDL statements, for those BLOB types that accept a length
-          (i.e. MySQL).  It does *not* produce a *lengthed* BINARY/VARBINARY
-          type - use the BINARY/VARBINARY types specifically for those.
-          May be safely omitted if no ``CREATE
-          TABLE`` will be issued.  Certain databases may require a
-          *length* for use in DDL, and will raise an exception when
-          the ``CREATE TABLE`` DDL is issued.
+          DDL statements, for those binary types that accept a length,
+          such as the MySQL BLOB type.
 
         """
         _Binary.__init__(self, length=length)
@@ -1496,7 +1491,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 +1715,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 +1738,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 +1764,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 +1784,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 +1798,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 +1814,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 +1869,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 +1904,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 +1917,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 +1940,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 +1948,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):