Method _columns_query

lib/sqlalchemy/dialects/postgresql/base.py:4129–4292

(self, schema, has_filter_names, scope, kind)

    @lru_cache()
    def _columns_query(self, schema, has_filter_names, scope, kind):
        # NOTE: the query with the default and identity options scalar
        # subquery is faster than trying to use outer joins for them
        generated = (
            pg_catalog.pg_attribute.c.attgenerated.label("generated")
            if self.server_version_info >= (12,)
            else sql.null().label("generated")
        )
        if self.server_version_info >= (10,):
            # join lateral performs worse (~2x slower) than a scalar_subquery
            # also the subquery can be run only if the column is an identity
            identity = sql.case(
                (  # attidentity != '' is required or it will reflect also
                    # serial columns as identity.
                    pg_catalog.pg_attribute.c.attidentity != "",
                    select(
                        sql.func.json_build_object(
                            "always",
                            pg_catalog.pg_attribute.c.attidentity == "a",
                            "start",
                            pg_catalog.pg_sequence.c.seqstart,
                            "increment",
                            pg_catalog.pg_sequence.c.seqincrement,
                            "minvalue",
                            pg_catalog.pg_sequence.c.seqmin,
                            "maxvalue",
                            pg_catalog.pg_sequence.c.seqmax,
                            "cache",
                            pg_catalog.pg_sequence.c.seqcache,
                            "cycle",
                            pg_catalog.pg_sequence.c.seqcycle,
                            type_=sqltypes.JSON(),
                        )
                    )
                    .select_from(pg_catalog.pg_sequence)
                    .where(
                        # not needed but pg seems to like it
                        pg_catalog.pg_attribute.c.attidentity != "",
                        pg_catalog.pg_sequence.c.seqrelid
                        == sql.cast(
                            sql.cast(
                                pg_catalog.pg_get_serial_sequence(
                                    sql.cast(
                                        sql.cast(
                                            pg_catalog.pg_attribute.c.attrelid,
                                            REGCLASS,
                                        ),
                                        TEXT,
                                    ),
                                    pg_catalog.pg_attribute.c.attname,
                                ),
                                REGCLASS,
                            ),
                            OID,
                        ),
                    )
                    .correlate(pg_catalog.pg_attribute)
                    .scalar_subquery(),