--- /dev/null
+.. change::
+ :tags: bug, sql
+ :tickets: 7471
+
+ Added an additional lookup step to the compiler which will track all FROM
+ clauses which are tables, that may have the same name shared in multiple
+ schemas where one of the schemas is the implicit "default" schema; in this
+ case, the table name when referring to that name without a schema
+ qualification will be rendered with an anonymous alias name at the compiler
+ level in order to disambiguate the two (or more) names. The approach of
+ schema-qualifying the normally unqualified name with the server-detected
+ "default schema name" value was also considered, however this approach
+ doesn't apply to Oracle nor is it accepted by SQL Server, nor would it work
+ with multiple entries in the PostgreSQL search path. The name collision
+ issue resolved here has been identified as affecting at least Oracle,
+ PostgreSQL, SQL Server, MySQL and MariaDB.
+
self._setup_for_generate()
SelectState.__init__(self, self.statement, compiler, **kw)
-
return self
def _dump_option_struct(self):
"""
- __slots__ = ("statement",)
+ __slots__ = ("statement", "_ambiguous_table_name_map")
plugins = {}
add_to_result_map=None,
include_table=True,
result_map_targets=(),
+ ambiguous_table_name_map=None,
**kwargs,
):
name = orig_name = column.name
else:
schema_prefix = ""
tablename = table.name
+
+ if (
+ not effective_schema
+ and ambiguous_table_name_map
+ and tablename in ambiguous_table_name_map
+ ):
+ tablename = ambiguous_table_name_map[tablename]
+
if isinstance(tablename, elements._truncated_label):
tablename = self._truncated_identifier("alias", tablename)
compile_state = select_stmt._compile_state_factory(
select_stmt, self, **kwargs
)
+ kwargs[
+ "ambiguous_table_name_map"
+ ] = compile_state._ambiguous_table_name_map
+
select_stmt = compile_state.statement
toplevel = not self.stack
fromhints=None,
use_schema=True,
from_linter=None,
+ ambiguous_table_name_map=None,
**kwargs,
):
if from_linter:
)
else:
ret = self.preparer.quote(table.name)
+
+ if (
+ not effective_schema
+ and ambiguous_table_name_map
+ and table.name in ambiguous_table_name_map
+ ):
+ anon_name = self._truncated_identifier(
+ "alias", ambiguous_table_name_map[table.name]
+ )
+
+ ret = ret + self.get_render_as_alias_suffix(
+ self.preparer.format_alias(None, anon_name)
+ )
+
if fromhints and table in fromhints:
ret = self.format_from_hint_text(
ret, table, fromhints[table], iscrud
is_clause_element = True
is_selectable = False
+ _is_table = False
_is_textual = False
_is_from_clause = False
_is_returns_rows = False
named_with_column = True
+ _is_table = True
+
implicit_returning = False
""":class:`_expression.TableClause`
doesn't support having a primary key or column
return go
def _get_froms(self, statement):
+ self._ambiguous_table_name_map = ambiguous_table_name_map = {}
+
return self._normalize_froms(
itertools.chain(
itertools.chain.from_iterable(
self.from_clauses,
),
check_statement=statement,
+ ambiguous_table_name_map=ambiguous_table_name_map,
)
@classmethod
- def _normalize_froms(cls, iterable_of_froms, check_statement=None):
+ def _normalize_froms(
+ cls,
+ iterable_of_froms,
+ check_statement=None,
+ ambiguous_table_name_map=None,
+ ):
"""given an iterable of things to select FROM, reduce them to what
would actually render in the FROM clause of a SELECT.
froms = []
for item in iterable_of_froms:
+
if item._is_subquery and item.element is check_statement:
raise exc.InvalidRequestError(
"select() construct refers to itself as a FROM"
# using a list to maintain ordering
froms = [f for f in froms if f not in toremove]
+ if ambiguous_table_name_map is not None:
+ ambiguous_table_name_map.update(
+ (
+ fr.name,
+ _anonymous_label.safe_construct(
+ hash(fr.name), fr.name
+ ),
+ )
+ for item in froms
+ for fr in item._from_objects
+ if fr._is_table
+ and fr.schema
+ and fr.name not in ambiguous_table_name_map
+ )
+
return froms
def _get_display_froms(
eq_(set(fa), set([(3, 3, 4), (4, 4, 5), (5, 4, 6)]))
+class SameNamedSchemaTableTest(fixtures.TablesTest):
+ """tests for #7471"""
+
+ __backend__ = True
+
+ __requires__ = ("schemas",)
+
+ @classmethod
+ def define_tables(cls, metadata):
+ Table(
+ "some_table",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ schema=config.test_schema,
+ )
+ Table(
+ "some_table",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column(
+ "some_table_id",
+ Integer,
+ # ForeignKey("%s.some_table.id" % config.test_schema),
+ nullable=False,
+ ),
+ )
+
+ @classmethod
+ def insert_data(cls, connection):
+ some_table, some_table_schema = cls.tables(
+ "some_table", "%s.some_table" % config.test_schema
+ )
+ connection.execute(some_table_schema.insert(), {"id": 1})
+ connection.execute(some_table.insert(), {"id": 1, "some_table_id": 1})
+
+ def test_simple_join_both_tables(self, connection):
+ some_table, some_table_schema = self.tables(
+ "some_table", "%s.some_table" % config.test_schema
+ )
+
+ eq_(
+ connection.execute(
+ select(some_table, some_table_schema).join_from(
+ some_table,
+ some_table_schema,
+ some_table.c.some_table_id == some_table_schema.c.id,
+ )
+ ).first(),
+ (1, 1, 1),
+ )
+
+ def test_simple_join_whereclause_only(self, connection):
+ some_table, some_table_schema = self.tables(
+ "some_table", "%s.some_table" % config.test_schema
+ )
+
+ eq_(
+ connection.execute(
+ select(some_table)
+ .join_from(
+ some_table,
+ some_table_schema,
+ some_table.c.some_table_id == some_table_schema.c.id,
+ )
+ .where(some_table.c.id == 1)
+ ).first(),
+ (1, 1),
+ )
+
+ def test_subquery(self, connection):
+ some_table, some_table_schema = self.tables(
+ "some_table", "%s.some_table" % config.test_schema
+ )
+
+ subq = (
+ select(some_table)
+ .join_from(
+ some_table,
+ some_table_schema,
+ some_table.c.some_table_id == some_table_schema.c.id,
+ )
+ .where(some_table.c.id == 1)
+ .subquery()
+ )
+
+ eq_(
+ connection.execute(
+ select(some_table, subq.c.id)
+ .join_from(
+ some_table,
+ subq,
+ some_table.c.some_table_id == subq.c.id,
+ )
+ .where(some_table.c.id == 1)
+ ).first(),
+ (1, 1, 1),
+ )
+
+
class JoinTest(fixtures.TablesTest):
__backend__ = True
basically, PostgreSQL.
+ TODO: what does this mean? all the backends have a "default"
+ schema
+
"""
return only_on(["postgresql"])
render_schema_translate=True,
)
+ def test_schema_non_schema_disambiguation(self):
+ """test #7471"""
+
+ t1 = table("some_table", column("id"), column("q"))
+ t2 = table("some_table", column("id"), column("p"), schema="foo")
+
+ self.assert_compile(
+ select(t1, t2),
+ "SELECT some_table_1.id, some_table_1.q, "
+ "foo.some_table.id AS id_1, foo.some_table.p "
+ "FROM some_table AS some_table_1, foo.some_table",
+ )
+
+ self.assert_compile(
+ select(t1, t2).set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL),
+ # the original "tablename_colname" label is preserved despite
+ # the alias of some_table
+ "SELECT some_table_1.id AS some_table_id, some_table_1.q AS "
+ "some_table_q, foo.some_table.id AS foo_some_table_id, "
+ "foo.some_table.p AS foo_some_table_p "
+ "FROM some_table AS some_table_1, foo.some_table",
+ )
+
+ self.assert_compile(
+ select(t1, t2).join_from(t1, t2, t1.c.id == t2.c.id),
+ "SELECT some_table_1.id, some_table_1.q, "
+ "foo.some_table.id AS id_1, foo.some_table.p "
+ "FROM some_table AS some_table_1 "
+ "JOIN foo.some_table ON some_table_1.id = foo.some_table.id",
+ )
+
+ self.assert_compile(
+ select(t1, t2).where(t1.c.id == t2.c.id),
+ "SELECT some_table_1.id, some_table_1.q, "
+ "foo.some_table.id AS id_1, foo.some_table.p "
+ "FROM some_table AS some_table_1, foo.some_table "
+ "WHERE some_table_1.id = foo.some_table.id",
+ )
+
+ self.assert_compile(
+ select(t1).where(t1.c.id == t2.c.id),
+ "SELECT some_table_1.id, some_table_1.q "
+ "FROM some_table AS some_table_1, foo.some_table "
+ "WHERE some_table_1.id = foo.some_table.id",
+ )
+
+ subq = select(t1).where(t1.c.id == t2.c.id).subquery()
+ self.assert_compile(
+ select(t2).select_from(t2).join(subq, t2.c.id == subq.c.id),
+ "SELECT foo.some_table.id, foo.some_table.p "
+ "FROM foo.some_table JOIN "
+ "(SELECT some_table_1.id AS id, some_table_1.q AS q "
+ "FROM some_table AS some_table_1, foo.some_table "
+ "WHERE some_table_1.id = foo.some_table.id) AS anon_1 "
+ "ON foo.some_table.id = anon_1.id",
+ )
+
+ self.assert_compile(
+ select(t1, subq.c.id)
+ .select_from(t1)
+ .join(subq, t1.c.id == subq.c.id),
+ # some_table is only aliased inside the subquery. this is not
+ # any challenge for the compiler, just checking as this is a new
+ # source of aliasing.
+ "SELECT some_table.id, some_table.q, anon_1.id AS id_1 "
+ "FROM some_table "
+ "JOIN (SELECT some_table_1.id AS id, some_table_1.q AS q "
+ "FROM some_table AS some_table_1, foo.some_table "
+ "WHERE some_table_1.id = foo.some_table.id) AS anon_1 "
+ "ON some_table.id = anon_1.id",
+ )
+
def test_alias(self):
a = alias(table4, "remtable")
self.assert_compile(
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
