sqlglot.transforms

   1from __future__ import annotations
   2
   3import typing as t
   4
   5from sqlglot import expressions as exp
   6from sqlglot.errors import UnsupportedError
   7from sqlglot.helper import find_new_name, name_sequence, seq_get
   8
   9
  10if t.TYPE_CHECKING:
  11    from sqlglot._typing import E
  12    from sqlglot.generator import Generator
  13
  14
  15class SqlHandler(t.Protocol):
  16    def __call__(self, expression: exp.Expr, *args: t.Any, **kwargs: t.Any) -> str: ...
  17
  18
  19def preprocess(
  20    transforms: list[t.Callable[[exp.Expr], exp.Expr]],
  21    generator: t.Callable[[Generator, exp.Expr], str] | None = None,
  22) -> t.Callable[[Generator, exp.Expr], str]:
  23    """
  24    Creates a new transform by chaining a sequence of transformations and converts the resulting
  25    expression to SQL, using either the "_sql" method corresponding to the resulting expression,
  26    or the appropriate `Generator.TRANSFORMS` function (when applicable -- see below).
  27
  28    Args:
  29        transforms: sequence of transform functions. These will be called in order.
  30
  31    Returns:
  32        Function that can be used as a generator transform.
  33    """
  34
  35    def _to_sql(self: Generator, expression: exp.Expr) -> str:
  36        expression_type = type(expression)
  37
  38        try:
  39            expression = transforms[0](expression)
  40            for transform in transforms[1:]:
  41                expression = transform(expression)
  42        except UnsupportedError as unsupported_error:
  43            self.unsupported(str(unsupported_error))
  44
  45        if generator:
  46            return generator(self, expression)
  47
  48        _sql_handler: SqlHandler | None = getattr(self, expression.key + "_sql", None)
  49        if _sql_handler:
  50            return _sql_handler(expression)
  51
  52        transforms_handler = self.TRANSFORMS.get(type(expression))
  53        if transforms_handler:
  54            if expression_type is type(expression):
  55                if isinstance(expression, exp.Func):
  56                    return self.function_fallback_sql(expression)
  57
  58                # Ensures we don't enter an infinite loop. This can happen when the original expression
  59                # has the same type as the final expression and there's no _sql method available for it,
  60                # because then it'd re-enter _to_sql.
  61                raise ValueError(
  62                    f"Expr type {expression.__class__.__name__} requires a _sql method in order to be transformed."
  63                )
  64
  65            return transforms_handler(self, expression)
  66
  67        raise ValueError(f"Unsupported expression type {expression.__class__.__name__}.")
  68
  69    return _to_sql
  70
  71
  72def unnest_generate_date_array_using_recursive_cte(expression: exp.Expr) -> exp.Expr:
  73    if isinstance(expression, exp.Select):
  74        count = 0
  75        recursive_ctes: list[exp.Expr] = []
  76
  77        for unnest in expression.find_all(exp.Unnest):
  78            if (
  79                not isinstance(unnest.parent, (exp.From, exp.Join))
  80                or len(unnest.expressions) != 1
  81                or not isinstance(unnest.expressions[0], exp.GenerateDateArray)
  82            ):
  83                continue
  84
  85            generate_date_array = unnest.expressions[0]
  86            start: exp.Expr | None = generate_date_array.args.get("start")
  87            end: exp.Expr | None = generate_date_array.args.get("end")
  88            step: exp.Expr | None = generate_date_array.args.get("step")
  89
  90            if not start or not end or not isinstance(step, exp.Interval):
  91                continue
  92
  93            alias: exp.TableAlias | None = unnest.args.get("alias")
  94            column_name: str = (
  95                alias.columns[0] if isinstance(alias, exp.TableAlias) else "date_value"
  96            )
  97
  98            start = exp.cast(start, "date")
  99            date_add = exp.func(
 100                "date_add", column_name, exp.Literal.number(step.name), step.args.get("unit")
 101            )
 102            cast_date_add = exp.cast(date_add, "date")
 103
 104            cte_name = "_generated_dates" + (f"_{count}" if count else "")
 105
 106            base_query = exp.select(start.as_(column_name))
 107            recursive_query = (
 108                exp.select(cast_date_add)
 109                .from_(cte_name)
 110                .where(cast_date_add <= exp.cast(end, "date"))
 111            )
 112            cte_query = base_query.union(recursive_query, distinct=False)
 113
 114            generate_dates_query = exp.select(column_name).from_(cte_name)
 115            unnest.replace(generate_dates_query.subquery(cte_name))
 116
 117            recursive_ctes.append(
 118                exp.alias_(exp.CTE(this=cte_query), cte_name, table=[column_name])
 119            )
 120            count += 1
 121
 122        if recursive_ctes:
 123            with_expression: exp.With = expression.args.get("with_") or exp.With()
 124            with_expression.set("recursive", True)
 125            with_expression.set("expressions", [*recursive_ctes, *with_expression.expressions])
 126            expression.set("with_", with_expression)
 127
 128    return expression
 129
 130
 131def unnest_generate_series(expression: exp.Expr) -> exp.Expr:
 132    """Unnests GENERATE_SERIES or SEQUENCE table references."""
 133    this = expression.this
 134    if isinstance(expression, exp.Table) and isinstance(this, exp.GenerateSeries):
 135        unnest = exp.Unnest(expressions=[this])
 136        if expression.alias:
 137            return exp.alias_(unnest, alias="_u", table=[expression.alias], copy=False)
 138
 139        return unnest
 140
 141    return expression
 142
 143
 144def eliminate_distinct_on(expression: exp.Expr) -> exp.Expr:
 145    """
 146    Convert SELECT DISTINCT ON statements to a subquery with a window function.
 147
 148    This is useful for dialects that don't support SELECT DISTINCT ON but support window functions.
 149
 150    Args:
 151        expression: the expression that will be transformed.
 152
 153    Returns:
 154        The transformed expression.
 155    """
 156    if (
 157        isinstance(expression, exp.Select)
 158        and expression.args.get("distinct")
 159        and isinstance(expression.args["distinct"].args.get("on"), exp.Tuple)
 160    ):
 161        row_number_window_alias = find_new_name(expression.named_selects, "_row_number")
 162
 163        distinct_cols = expression.args["distinct"].pop().args["on"].expressions
 164        window = exp.Window(this=exp.RowNumber(), partition_by=distinct_cols)
 165
 166        order: exp.Order | None = expression.args.get("order")
 167        if order:
 168            window.set("order", order.pop())
 169        else:
 170            window.set("order", exp.Order(expressions=[c.copy() for c in distinct_cols]))
 171
 172        expression.select(exp.alias_(window, row_number_window_alias), copy=False)
 173
 174        # We add aliases to the projections so that we can safely reference them in the outer query
 175        new_selects: list[exp.Expr] = []
 176        taken_names = {row_number_window_alias}
 177        for select in expression.selects[:-1]:
 178            if select.is_star:
 179                new_selects = [exp.Star()]
 180                break
 181
 182            if not isinstance(select, exp.Alias):
 183                alias = find_new_name(taken_names, select.output_name or "_col")
 184                quoted: bool | None = (
 185                    select.this.args.get("quoted") if isinstance(select, exp.Column) else None
 186                )
 187                select = select.replace(exp.alias_(select, alias, quoted=quoted))
 188
 189            taken_names.add(select.output_name)
 190            new_selects.append(select.args["alias"])
 191
 192        return (
 193            exp.select(*new_selects, copy=False)
 194            .from_(expression.subquery("_t", copy=False), copy=False)
 195            .where(exp.column(row_number_window_alias).eq(1), copy=False)
 196        )
 197
 198    return expression
 199
 200
 201def eliminate_qualify(expression: exp.Expr) -> exp.Expr:
 202    """
 203    Convert SELECT statements that contain the QUALIFY clause into subqueries, filtered equivalently.
 204
 205    The idea behind this transformation can be seen in Snowflake's documentation for QUALIFY:
 206    https://docs.snowflake.com/en/sql-reference/constructs/qualify
 207
 208    Some dialects don't support window functions in the WHERE clause, so we need to include them as
 209    projections in the subquery, in order to refer to them in the outer filter using aliases. Also,
 210    if a column is referenced in the QUALIFY clause but is not selected, we need to include it too,
 211    otherwise we won't be able to refer to it in the outer query's WHERE clause. Finally, if a
 212    newly aliased projection is referenced in the QUALIFY clause, it will be replaced by the
 213    corresponding expression to avoid creating invalid column references.
 214    """
 215    if isinstance(expression, exp.Select) and expression.args.get("qualify"):
 216        taken = set(expression.named_selects)
 217        for select in expression.selects:
 218            if not select.alias_or_name:
 219                alias = find_new_name(taken, "_c")
 220                select.replace(exp.alias_(select, alias))
 221                taken.add(alias)
 222
 223        def _select_alias_or_name(select: exp.Expr) -> str | exp.Column:
 224            alias_or_name = select.alias_or_name
 225            identifier = select.args.get("alias") or select.this
 226            if isinstance(identifier, exp.Identifier):
 227                return exp.column(alias_or_name, quoted=identifier.args.get("quoted"))
 228            return alias_or_name
 229
 230        outer_selects = exp.select(*map(_select_alias_or_name, expression.selects))
 231        qualify_filters: exp.Expr = expression.args["qualify"].pop().this
 232        expression_by_alias: dict[str, exp.Expr] = {
 233            select.alias: select.this
 234            for select in expression.selects
 235            if isinstance(select, exp.Alias)
 236        }
 237
 238        select_candidates = (exp.Window,) if expression.is_star else (exp.Window, exp.Column)
 239        for select_candidate in list(qualify_filters.find_all(*select_candidates)):
 240            if isinstance(select_candidate, exp.Window):
 241                if expression_by_alias:
 242                    for column in select_candidate.find_all(exp.Column):
 243                        expr = expression_by_alias.get(column.name)
 244                        if expr:
 245                            column.replace(expr)
 246
 247                alias = find_new_name(expression.named_selects, "_w")
 248                expression.select(exp.alias_(select_candidate, alias), copy=False)
 249                column = exp.column(alias)
 250
 251                if isinstance(select_candidate.parent, exp.Qualify):
 252                    qualify_filters = column
 253                else:
 254                    select_candidate.replace(column)
 255            elif select_candidate.name not in expression.named_selects:
 256                expression.select(select_candidate.copy(), copy=False)
 257
 258        return outer_selects.from_(expression.subquery(alias="_t", copy=False), copy=False).where(
 259            qualify_filters, copy=False
 260        )
 261
 262    return expression
 263
 264
 265def remove_precision_parameterized_types(expression: exp.Expr) -> exp.Expr:
 266    """
 267    Some dialects only allow the precision for parameterized types to be defined in the DDL and not in
 268    other expressions. This transforms removes the precision from parameterized types in expressions.
 269    """
 270    for node in expression.find_all(exp.DataType):
 271        node.set(
 272            "expressions", [e for e in node.expressions if not isinstance(e, exp.DataTypeParam)]
 273        )
 274
 275    return expression
 276
 277
 278def unqualify_unnest(expression: exp.Expr) -> exp.Expr:
 279    """Remove references to unnest table aliases, added by the optimizer's qualify_columns step."""
 280    from sqlglot.optimizer.scope import find_all_in_scope
 281
 282    if isinstance(expression, exp.Select):
 283        unnest_aliases = {
 284            unnest.alias
 285            for unnest in find_all_in_scope(expression, exp.Unnest)
 286            if isinstance(unnest.parent, (exp.From, exp.Join))
 287        }
 288        if unnest_aliases:
 289            for column in expression.find_all(exp.Column):
 290                leftmost_part = column.parts[0]
 291                if leftmost_part.arg_key != "this" and leftmost_part.this in unnest_aliases:
 292                    leftmost_part.pop()
 293
 294    return expression
 295
 296
 297def unnest_to_explode(
 298    expression: exp.Expr,
 299    unnest_using_arrays_zip: bool = True,
 300) -> exp.Expr:
 301    """Convert cross join unnest into lateral view explode."""
 302
 303    def _unnest_zip_exprs(
 304        u: exp.Unnest, unnest_exprs: list[exp.Expr], has_multi_expr: bool
 305    ) -> list[exp.Expr]:
 306        if has_multi_expr:
 307            if not unnest_using_arrays_zip:
 308                raise UnsupportedError("Cannot transpile UNNEST with multiple input arrays")
 309
 310            # Use INLINE(ARRAYS_ZIP(...)) for multiple expressions
 311            zip_exprs: list[exp.Expr] = [exp.Anonymous(this="ARRAYS_ZIP", expressions=unnest_exprs)]
 312            u.set("expressions", zip_exprs)
 313            return zip_exprs
 314        return unnest_exprs
 315
 316    def _udtf_type(u: exp.Unnest, has_multi_expr: bool) -> type[exp.Func]:
 317        if u.args.get("offset"):
 318            return exp.Posexplode
 319        return exp.Inline if has_multi_expr else exp.Explode
 320
 321    if isinstance(expression, exp.Select):
 322        from_ = expression.args.get("from_")
 323
 324        if from_ and isinstance(from_.this, exp.Unnest):
 325            unnest: exp.Unnest = from_.this
 326            alias: exp.TableAlias | None = unnest.args.get("alias")
 327            exprs: list[exp.Expr] = unnest.expressions
 328            has_multi_expr = len(exprs) > 1
 329            this, *_ = _unnest_zip_exprs(unnest, exprs, has_multi_expr)
 330
 331            columns: list[exp.Identifier] = alias.columns if alias else []
 332            offset: exp.Expr | None = unnest.args.get("offset")
 333            if offset:
 334                columns.insert(
 335                    0, offset if isinstance(offset, exp.Identifier) else exp.to_identifier("pos")
 336                )
 337
 338            unnest.replace(
 339                exp.Table(
 340                    this=_udtf_type(unnest, has_multi_expr)(this=this),
 341                    alias=exp.TableAlias(this=alias.this, columns=columns) if alias else None,
 342                )
 343            )
 344
 345        joins: list[exp.Join] = expression.args.get("joins") or []
 346        for join in list(joins):
 347            join_expr = join.this
 348
 349            is_lateral = isinstance(join_expr, exp.Lateral)
 350
 351            unnest = join_expr.this if is_lateral else join_expr
 352
 353            if isinstance(unnest, exp.Unnest):
 354                if is_lateral:
 355                    alias = join_expr.args.get("alias")
 356                else:
 357                    alias = unnest.args.get("alias")
 358
 359                if alias is None:
 360                    raise UnsupportedError(
 361                        "CROSS JOIN UNNEST to LATERAL VIEW EXPLODE transformation requires an alias"
 362                    )
 363
 364                exprs = unnest.expressions
 365                # The number of unnest.expressions will be changed by _unnest_zip_exprs, we need to record it here
 366                has_multi_expr = len(exprs) > 1
 367                exprs = _unnest_zip_exprs(unnest, exprs, has_multi_expr)
 368
 369                joins.remove(join)
 370
 371                alias_cols: list[exp.Identifier] = alias.columns
 372
 373                # # Handle UNNEST to LATERAL VIEW EXPLODE: Exception is raised when there are 0 or > 2 aliases
 374                # Spark LATERAL VIEW EXPLODE requires single alias for array/struct and two for Map type column unlike unnest in trino/presto which can take an arbitrary amount.
 375                # Refs: https://spark.apache.org/docs/latest/sql-ref-syntax-qry-select-lateral-view.html
 376
 377                if not has_multi_expr and len(alias_cols) not in (1, 2):
 378                    raise UnsupportedError(
 379                        "CROSS JOIN UNNEST to LATERAL VIEW EXPLODE transformation requires explicit column aliases"
 380                    )
 381
 382                offset = unnest.args.get("offset")
 383                if offset:
 384                    alias_cols.insert(
 385                        0,
 386                        offset if isinstance(offset, exp.Identifier) else exp.to_identifier("pos"),
 387                    )
 388
 389                for e, column in zip(exprs, alias_cols):
 390                    expression.append(
 391                        "laterals",
 392                        exp.Lateral(
 393                            this=_udtf_type(unnest, has_multi_expr)(this=e),
 394                            view=True,
 395                            alias=exp.TableAlias(this=alias.this, columns=alias_cols),
 396                        ),
 397                    )
 398
 399    return expression
 400
 401
 402def explode_projection_to_unnest(
 403    index_offset: int = 0,
 404) -> t.Callable[[exp.Expr], exp.Expr]:
 405    """Convert explode/posexplode projections into unnests."""
 406
 407    def _explode_projection_to_unnest(expression: exp.Expr) -> exp.Expr:
 408        if isinstance(expression, exp.Select):
 409            from sqlglot.optimizer.scope import Scope
 410
 411            taken_select_names = set(expression.named_selects)
 412            taken_source_names = {name for name, _ in Scope(expression).references}
 413
 414            def new_name(names: set[str], name: str) -> str:
 415                name = find_new_name(names, name)
 416                names.add(name)
 417                return name
 418
 419            arrays: list[exp.Condition] = []
 420            series_alias = new_name(taken_select_names, "pos")
 421            series = exp.alias_(
 422                exp.Unnest(
 423                    expressions=[exp.GenerateSeries(start=exp.Literal.number(index_offset))]
 424                ),
 425                new_name(taken_source_names, "_u"),
 426                table=[series_alias],
 427            )
 428
 429            # we use list here because expression.selects is mutated inside the loop
 430            for select in list(expression.selects):
 431                explode = select.find(exp.Explode)
 432
 433                if explode:
 434                    pos_alias: t.Any = ""
 435                    explode_alias: t.Any = ""
 436
 437                    if isinstance(select, exp.Alias):
 438                        explode_alias = select.args["alias"]
 439                        alias: exp.Expr = select
 440                    elif isinstance(select, exp.Aliases):
 441                        pos_alias = select.aliases[0]
 442                        explode_alias = select.aliases[1]
 443                        alias = select.replace(exp.alias_(select.this, "", copy=False))
 444                    else:
 445                        alias = select.replace(exp.alias_(select, ""))
 446                        explode = alias.find(exp.Explode)
 447                        assert explode
 448
 449                    is_posexplode = isinstance(explode, exp.Posexplode)
 450                    explode_arg = explode.this
 451
 452                    if isinstance(explode, exp.ExplodeOuter):
 453                        bracket = explode_arg[0]
 454                        bracket.set("safe", True)
 455                        bracket.set("offset", True)
 456                        explode_arg = exp.func(
 457                            "IF",
 458                            exp.func(
 459                                "ARRAY_SIZE", exp.func("COALESCE", explode_arg, exp.Array())
 460                            ).eq(0),
 461                            exp.array(bracket, copy=False),
 462                            explode_arg,
 463                        )
 464
 465                    # This ensures that we won't use [POS]EXPLODE's argument as a new selection
 466                    if isinstance(explode_arg, exp.Column):
 467                        taken_select_names.add(explode_arg.output_name)
 468
 469                    unnest_source_alias = new_name(taken_source_names, "_u")
 470
 471                    if not explode_alias:
 472                        explode_alias = new_name(taken_select_names, "col")
 473
 474                        if is_posexplode:
 475                            pos_alias = new_name(taken_select_names, "pos")
 476
 477                    if not pos_alias:
 478                        pos_alias = new_name(taken_select_names, "pos")
 479
 480                    alias.set("alias", exp.to_identifier(explode_alias))
 481
 482                    series_table_alias = series.args["alias"].this
 483                    column = exp.If(
 484                        this=exp.column(series_alias, table=series_table_alias).eq(
 485                            exp.column(pos_alias, table=unnest_source_alias)
 486                        ),
 487                        true=exp.column(explode_alias, table=unnest_source_alias),
 488                    )
 489
 490                    explode.replace(column)
 491
 492                    if is_posexplode:
 493                        expressions = expression.expressions
 494                        expressions.insert(
 495                            expressions.index(alias) + 1,
 496                            exp.If(
 497                                this=exp.column(series_alias, table=series_table_alias).eq(
 498                                    exp.column(pos_alias, table=unnest_source_alias)
 499                                ),
 500                                true=exp.column(pos_alias, table=unnest_source_alias),
 501                            ).as_(pos_alias),
 502                        )
 503                        expression.set("expressions", expressions)
 504
 505                    if not arrays:
 506                        if expression.args.get("from_"):
 507                            expression.join(series, copy=False, join_type="CROSS")
 508                        else:
 509                            expression.from_(series, copy=False)
 510
 511                    size: exp.Condition = exp.ArraySize(this=explode_arg.copy())
 512                    arrays.append(size)
 513
 514                    # trino doesn't support left join unnest with on conditions
 515                    # if it did, this would be much simpler
 516                    expression.join(
 517                        exp.alias_(
 518                            exp.Unnest(
 519                                expressions=[explode_arg.copy()],
 520                                offset=exp.to_identifier(pos_alias),
 521                            ),
 522                            unnest_source_alias,
 523                            table=[explode_alias],
 524                        ),
 525                        join_type="CROSS",
 526                        copy=False,
 527                    )
 528
 529                    if index_offset != 1:
 530                        size = size - 1
 531
 532                    expression.where(
 533                        exp.column(series_alias, table=series_table_alias)
 534                        .eq(exp.column(pos_alias, table=unnest_source_alias))
 535                        .or_(
 536                            (exp.column(series_alias, table=series_table_alias) > size).and_(
 537                                exp.column(pos_alias, table=unnest_source_alias).eq(size)
 538                            )
 539                        ),
 540                        copy=False,
 541                    )
 542
 543            if arrays:
 544                end: exp.Condition = exp.Greatest(this=arrays[0], expressions=arrays[1:])
 545
 546                if index_offset != 1:
 547                    end = end - (1 - index_offset)
 548                series.expressions[0].set("end", end)
 549
 550        return expression
 551
 552    return _explode_projection_to_unnest
 553
 554
 555def add_within_group_for_percentiles(expression: exp.Expr) -> exp.Expr:
 556    """Transforms percentiles by adding a WITHIN GROUP clause to them."""
 557    if (
 558        isinstance(expression, exp.PERCENTILES)
 559        and not isinstance(expression.parent, exp.WithinGroup)
 560        and expression.expression
 561    ):
 562        column = expression.this.pop()
 563        expression.set("this", expression.expression.pop())
 564        order = exp.Order(expressions=[exp.Ordered(this=column)])
 565        expression = exp.WithinGroup(this=expression, expression=order)
 566
 567    return expression
 568
 569
 570def remove_within_group_for_percentiles(expression: exp.Expr) -> exp.Expr:
 571    """Transforms percentiles by getting rid of their corresponding WITHIN GROUP clause."""
 572    if (
 573        isinstance(expression, exp.WithinGroup)
 574        and isinstance(expression.this, exp.PERCENTILES)
 575        and isinstance(expression.expression, exp.Order)
 576    ):
 577        quantile = expression.this.this
 578        input_value = t.cast(exp.Ordered, expression.find(exp.Ordered)).this
 579        return expression.replace(exp.ApproxQuantile(this=input_value, quantile=quantile))
 580
 581    return expression
 582
 583
 584def add_recursive_cte_column_names(expression: exp.Expr) -> exp.Expr:
 585    """Uses projection output names in recursive CTE definitions to define the CTEs' columns."""
 586    if isinstance(expression, exp.With) and expression.recursive:
 587        next_name = name_sequence("_c_")
 588
 589        for cte in expression.expressions:
 590            if not cte.args["alias"].columns:
 591                query = cte.this
 592                if isinstance(query, exp.SetOperation):
 593                    query = query.this
 594
 595                cte.args["alias"].set(
 596                    "columns",
 597                    [exp.to_identifier(s.alias_or_name or next_name()) for s in query.selects],
 598                )
 599
 600    return expression
 601
 602
 603def epoch_cast_to_ts(expression: exp.Expr) -> exp.Expr:
 604    """Replace 'epoch' in casts by the equivalent date literal."""
 605    if (
 606        isinstance(expression, (exp.Cast, exp.TryCast))
 607        and expression.name.lower() == "epoch"
 608        and expression.to.this in exp.DataType.TEMPORAL_TYPES
 609    ):
 610        expression.this.replace(exp.Literal.string("1970-01-01 00:00:00"))
 611
 612    return expression
 613
 614
 615def eliminate_semi_and_anti_joins(expression: exp.Expr) -> exp.Expr:
 616    """Convert SEMI and ANTI joins into equivalent forms that use EXIST instead."""
 617    if isinstance(expression, exp.Select):
 618        for join in list[exp.Join](expression.args.get("joins") or []):
 619            on: exp.Expr | None = join.args.get("on")
 620            if on and join.kind in ("SEMI", "ANTI"):
 621                subquery = exp.select("1").from_(join.this).where(on)
 622                exists: exp.Exists | exp.Not = exp.Exists(this=subquery)
 623                if join.kind == "ANTI":
 624                    exists = exists.not_(copy=False)
 625
 626                join.pop()
 627                expression.where(exists, copy=False)
 628
 629    return expression
 630
 631
 632def eliminate_full_outer_join(expression: exp.Expr) -> exp.Expr:
 633    """
 634    Converts a query with a FULL OUTER join to a union of identical queries that
 635    use LEFT/RIGHT OUTER joins instead. This transformation currently only works
 636    for queries that have a single FULL OUTER join.
 637    """
 638    if isinstance(expression, exp.Select):
 639        full_outer_joins: list[tuple[int, exp.Join]] = [
 640            (index, join)
 641            for index, join in enumerate[exp.Join](expression.args.get("joins") or [])
 642            if join.side == "FULL"
 643        ]
 644
 645        if len(full_outer_joins) == 1:
 646            expression_copy = expression.copy()
 647            expression.set("limit", None)
 648            index, full_outer_join = full_outer_joins[0]
 649
 650            tables = (expression.args["from_"].alias_or_name, full_outer_join.alias_or_name)
 651            join_conditions = full_outer_join.args.get("on") or exp.and_(
 652                *[
 653                    exp.column(col, tables[0]).eq(exp.column(col, tables[1]))
 654                    for col in t.cast(list[exp.Identifier], full_outer_join.args.get("using"))
 655                ]
 656            )
 657
 658            full_outer_join.set("side", "left")
 659            anti_join_clause = (
 660                exp.select("1").from_(expression.args["from_"]).where(join_conditions)
 661            )
 662            expression_copy.args["joins"][index].set("side", "right")
 663            expression_copy = expression_copy.where(exp.Exists(this=anti_join_clause).not_())
 664            expression_copy.set("with_", None)  # remove CTEs from RIGHT side
 665            expression.set("order", None)  # remove order by from LEFT side
 666
 667            return exp.union(expression, expression_copy, copy=False, distinct=False)
 668
 669    return expression
 670
 671
 672def move_ctes_to_top_level(expression: E) -> E:
 673    """
 674    Some dialects (e.g. Hive, T-SQL, Spark prior to version 3) only allow CTEs to be
 675    defined at the top-level, so for example queries like:
 676
 677        SELECT * FROM (WITH t(c) AS (SELECT 1) SELECT * FROM t) AS subq
 678
 679    are invalid in those dialects. This transformation can be used to ensure all CTEs are
 680    moved to the top level so that the final SQL code is valid from a syntax standpoint.
 681
 682    TODO: handle name clashes whilst moving CTEs (it can get quite tricky & costly).
 683    """
 684    top_level_with: exp.With | None = expression.args.get("with_")
 685    for inner_with in expression.find_all(exp.With):
 686        if inner_with.parent is expression:
 687            continue
 688
 689        if not top_level_with:
 690            top_level_with = inner_with.pop()
 691            expression.set("with_", top_level_with)
 692        else:
 693            if inner_with.recursive:
 694                top_level_with.set("recursive", True)
 695
 696            parent_cte = inner_with.find_ancestor(exp.CTE)
 697            inner_with.pop()
 698
 699            if parent_cte:
 700                i = top_level_with.expressions.index(parent_cte)
 701                top_level_with.expressions[i:i] = inner_with.expressions
 702                top_level_with.set("expressions", top_level_with.expressions)
 703            else:
 704                top_level_with.set(
 705                    "expressions", top_level_with.expressions + inner_with.expressions
 706                )
 707
 708    return expression
 709
 710
 711def ensure_bools(expression: exp.Expr) -> exp.Expr:
 712    """Converts numeric values used in conditions into explicit boolean expressions."""
 713    from sqlglot.optimizer.canonicalize import ensure_bools
 714
 715    def _ensure_bool(node: exp.Expr) -> None:
 716        if (
 717            node.is_number
 718            or (
 719                not isinstance(node, exp.SubqueryPredicate)
 720                and node.is_type(exp.DType.UNKNOWN, *exp.DataType.NUMERIC_TYPES)
 721            )
 722            or (isinstance(node, exp.Column) and not node.type)
 723        ):
 724            node.replace(node.neq(0))
 725
 726    for node in expression.walk():
 727        ensure_bools(node, _ensure_bool)
 728
 729    return expression
 730
 731
 732def unqualify_columns(expression: exp.Expr) -> exp.Expr:
 733    for column in expression.find_all(exp.Column):
 734        # We only wanna pop off the table, db, catalog args
 735        for part in column.parts[:-1]:
 736            part.pop()
 737
 738    return expression
 739
 740
 741def unqualify_pivot_fields(expression: exp.Expr) -> exp.Expr:
 742    """
 743    Some dialects only accept simple column names in a (UN)PIVOT's FOR clause and IN-list
 744    (Oracle raises ORA-01748), even though the aggregate itself may stay qualified.
 745
 746    Example:
 747        >>> from sqlglot import parse_one
 748        >>> expr = parse_one("SELECT * FROM tbl PIVOT (SUM(tbl.sales) FOR tbl.quarter IN ('Q1', 'Q2'))")
 749        >>> print(unqualify_pivot_fields(expr).sql(dialect="spark"))
 750        SELECT * FROM tbl PIVOT(SUM(tbl.sales) FOR quarter IN ('Q1', 'Q2'))
 751    """
 752    if isinstance(expression, exp.Pivot):
 753        expression.set("fields", [unqualify_columns(field) for field in expression.fields])
 754
 755    return expression
 756
 757
 758def remove_unique_constraints(expression: exp.Expr) -> exp.Expr:
 759    assert isinstance(expression, exp.Create)
 760    for constraint in expression.find_all(exp.UniqueColumnConstraint):
 761        if constraint.parent:
 762            constraint.parent.pop()
 763
 764    return expression
 765
 766
 767def ctas_with_tmp_tables_to_create_tmp_view(
 768    expression: exp.Expr,
 769    tmp_storage_provider: t.Callable[[exp.Expr], exp.Expr] = lambda e: e,
 770) -> exp.Expr:
 771    assert isinstance(expression, exp.Create)
 772    properties: exp.Properties | None = expression.args.get("properties")
 773    temporary = any(
 774        isinstance(prop, exp.TemporaryProperty)
 775        for prop in (properties.expressions if properties is not None else [])
 776    )
 777
 778    # CTAS with temp tables map to CREATE TEMPORARY VIEW
 779    if expression.kind == "TABLE" and temporary:
 780        if expression.expression:
 781            return exp.Create(
 782                kind="TEMPORARY VIEW",
 783                this=expression.this,
 784                expression=expression.expression,
 785            )
 786        return tmp_storage_provider(expression)
 787
 788    return expression
 789
 790
 791def move_schema_columns_to_partitioned_by(expression: exp.Expr) -> exp.Expr:
 792    """
 793    In Hive, the PARTITIONED BY property acts as an extension of a table's schema. When the
 794    PARTITIONED BY value is an array of column names, they are transformed into a schema.
 795    The corresponding columns are removed from the create statement.
 796    """
 797    assert isinstance(expression, exp.Create)
 798    schema = expression.this
 799    is_partitionable = expression.kind in {"TABLE", "VIEW"}
 800
 801    if isinstance(schema, exp.Schema) and is_partitionable:
 802        prop = expression.find(exp.PartitionedByProperty)
 803        if prop and prop.this and not isinstance(prop.this, exp.Schema):
 804            columns: set[str] = {v.name.upper() for v in prop.this.expressions}
 805            schema_exprs: list[exp.Expr] = schema.expressions
 806            partitions = [col for col in schema_exprs if col.name.upper() in columns]
 807            schema.set("expressions", [e for e in schema_exprs if e not in partitions])
 808            prop.replace(exp.PartitionedByProperty(this=exp.Schema(expressions=partitions)))
 809            expression.set("this", schema)
 810
 811    return expression
 812
 813
 814def move_partitioned_by_to_schema_columns(expression: exp.Expr) -> exp.Expr:
 815    """
 816    Spark 3 supports both "HIVEFORMAT" and "DATASOURCE" formats for CREATE TABLE.
 817
 818    Currently, SQLGlot uses the DATASOURCE format for Spark 3.
 819    """
 820    assert isinstance(expression, exp.Create)
 821    prop = expression.find(exp.PartitionedByProperty)
 822    if (
 823        prop
 824        and prop.this
 825        and isinstance(prop.this, exp.Schema)
 826        and all(isinstance(e, exp.ColumnDef) and e.kind for e in prop.this.expressions)
 827    ):
 828        prop_this = exp.Tuple(
 829            expressions=[exp.to_identifier(e.this) for e in prop.this.expressions]
 830        )
 831        schema: exp.Schema = expression.this
 832        for e in prop.this.expressions:
 833            schema.append("expressions", e)
 834        prop.set("this", prop_this)
 835
 836    return expression
 837
 838
 839def struct_kv_to_alias(expression: exp.Expr) -> exp.Expr:
 840    """Converts struct arguments to aliases, e.g. STRUCT(1 AS y)."""
 841    if isinstance(expression, exp.Struct):
 842        expression.set(
 843            "expressions",
 844            [
 845                exp.alias_(e.expression, e.this) if isinstance(e, exp.PropertyEQ) else e
 846                for e in expression.expressions
 847            ],
 848        )
 849
 850    return expression
 851
 852
 853def eliminate_join_marks(expression: exp.Expr) -> exp.Expr:
 854    """https://docs.oracle.com/cd/B19306_01/server.102/b14200/queries006.htm#sthref3178
 855
 856    1. You cannot specify the (+) operator in a query block that also contains FROM clause join syntax.
 857
 858    2. The (+) operator can appear only in the WHERE clause or, in the context of left-correlation (that is, when specifying the TABLE clause) in the FROM clause, and can be applied only to a column of a table or view.
 859
 860    The (+) operator does not produce an outer join if you specify one table in the outer query and the other table in an inner query.
 861
 862    You cannot use the (+) operator to outer-join a table to itself, although self joins are valid.
 863
 864    The (+) operator can be applied only to a column, not to an arbitrary expression. However, an arbitrary expression can contain one or more columns marked with the (+) operator.
 865
 866    A WHERE condition containing the (+) operator cannot be combined with another condition using the OR logical operator.
 867
 868    A WHERE condition cannot use the IN comparison condition to compare a column marked with the (+) operator with an expression.
 869
 870    A WHERE condition cannot compare any column marked with the (+) operator with a subquery.
 871
 872    -- example with WHERE
 873    SELECT d.department_name, sum(e.salary) as total_salary
 874    FROM departments d, employees e
 875    WHERE e.department_id(+) = d.department_id
 876    group by department_name
 877
 878    -- example of left correlation in select
 879    SELECT d.department_name, (
 880        SELECT SUM(e.salary)
 881            FROM employees e
 882            WHERE e.department_id(+) = d.department_id) AS total_salary
 883    FROM departments d;
 884
 885    -- example of left correlation in from
 886    SELECT d.department_name, t.total_salary
 887    FROM departments d, (
 888            SELECT SUM(e.salary) AS total_salary
 889            FROM employees e
 890            WHERE e.department_id(+) = d.department_id
 891        ) t
 892    """
 893
 894    from sqlglot.optimizer.scope import traverse_scope
 895    from sqlglot.optimizer.normalize import normalize, normalized
 896    from collections import defaultdict
 897
 898    # we go in reverse to check the main query for left correlation
 899    for scope in reversed(traverse_scope(expression)):
 900        query = scope.expression
 901
 902        where: exp.Expr | None = query.args.get("where")
 903        joins: list[exp.Join] = query.args.get("joins", [])
 904
 905        if not where or not any(c.args.get("join_mark") for c in where.find_all(exp.Column)):
 906            continue
 907
 908        # knockout: we do not support left correlation (see point 2)
 909        assert not scope.is_correlated_subquery, "Correlated queries are not supported"
 910
 911        # make sure we have AND of ORs to have clear join terms
 912        where = normalize(where.this)
 913        assert normalized(where), "Cannot normalize JOIN predicates"
 914        # dict of {name: list of join AND conditions}
 915        joins_ons: defaultdict[str, list[exp.Expr]] = defaultdict(list)
 916        for cond in [where] if not isinstance(where, exp.And) else where.flatten():
 917            join_cols = [col for col in cond.find_all(exp.Column) if col.args.get("join_mark")]
 918
 919            left_join_table = set(col.table for col in join_cols)
 920            if not left_join_table:
 921                continue
 922
 923            assert not (len(left_join_table) > 1), (
 924                "Cannot combine JOIN predicates from different tables"
 925            )
 926
 927            for col in join_cols:
 928                col.set("join_mark", False)
 929
 930            joins_ons[left_join_table.pop()].append(cond)
 931
 932        old_joins = {join.alias_or_name: join for join in joins}
 933        new_joins: dict[str, exp.Join] = {}
 934        query_from = query.args["from_"]
 935
 936        for table, predicates in joins_ons.items():
 937            join_what = old_joins.get(table, query_from).this.copy()
 938            new_joins[join_what.alias_or_name] = exp.Join(
 939                this=join_what, on=exp.and_(*predicates), kind="LEFT"
 940            )
 941
 942            for p in predicates:
 943                while isinstance(p.parent, exp.Paren):
 944                    p.parent.replace(p)
 945
 946                parent = p.parent
 947                p.pop()
 948                if isinstance(parent, exp.Binary):
 949                    left = parent.args.get("this")
 950                    parent.replace(parent.right if left is None else left)
 951                elif isinstance(parent, exp.Where):
 952                    parent.pop()
 953
 954        if query_from.alias_or_name in new_joins:
 955            only_old_joins: set[str] = old_joins.keys() - new_joins.keys()
 956            assert len(only_old_joins) >= 1, (
 957                "Cannot determine which table to use in the new FROM clause"
 958            )
 959
 960            new_from_name = list[str](only_old_joins)[0]
 961            query.set("from_", exp.From(this=old_joins[new_from_name].this))
 962
 963        if new_joins:
 964            for n, j in old_joins.items():  # preserve any other joins
 965                if n not in new_joins and n != query.args["from_"].name:
 966                    if not j.kind:
 967                        j.set("kind", "CROSS")
 968                    new_joins[n] = j
 969            query.set("joins", list(new_joins.values()))
 970
 971    return expression
 972
 973
 974def any_to_exists(expression: exp.Expr) -> exp.Expr:
 975    """
 976    Transform ANY operator to Spark's EXISTS
 977
 978    For example,
 979        - Postgres: SELECT * FROM tbl WHERE 5 > ANY(tbl.col)
 980        - Spark: SELECT * FROM tbl WHERE EXISTS(tbl.col, x -> x < 5)
 981
 982    Both ANY and EXISTS accept queries but currently only array expressions are supported for this
 983    transformation
 984    """
 985    if isinstance(expression, exp.Select):
 986        for any_expr in expression.find_all(exp.Any):
 987            this: exp.Expr = any_expr.this
 988            if isinstance(this, exp.Query) or isinstance(any_expr.parent, (exp.Like, exp.ILike)):
 989                continue
 990
 991            binop = any_expr.parent
 992            if isinstance(binop, exp.Binary):
 993                lambda_arg = exp.to_identifier("x")
 994                any_expr.replace(lambda_arg)
 995                lambda_expr = exp.Lambda(this=binop.copy(), expressions=[lambda_arg])
 996                binop.replace(exp.Exists(this=this.unnest(), expression=lambda_expr))
 997
 998    return expression
 999
1000
1001def eliminate_window_clause(expression: exp.Expr) -> exp.Expr:
1002    """Eliminates the `WINDOW` query clause by inling each named window."""
1003    windows: list[exp.Expr] | None = expression.args.get("windows")
1004    if isinstance(expression, exp.Select) and windows is not None:
1005        from sqlglot.optimizer.scope import find_all_in_scope
1006
1007        expression.set("windows", None)
1008
1009        window_expression: dict[str, exp.Expr] = {}
1010
1011        def _inline_inherited_window(window: exp.Expr) -> None:
1012            inherited_window = window_expression.get(window.alias.lower())
1013            if not inherited_window:
1014                return
1015
1016            window.set("alias", None)
1017            for key in ("partition_by", "order", "spec"):
1018                arg: exp.Expr | None = inherited_window.args.get(key)
1019                if arg is not None:
1020                    window.set(key, arg.copy())
1021
1022        for window in windows:
1023            _inline_inherited_window(window)
1024            window_expression[window.name.lower()] = window
1025
1026        for window in find_all_in_scope(expression, exp.Window):
1027            _inline_inherited_window(window)
1028
1029    return expression
1030
1031
1032def inherit_struct_field_names(expression: exp.Expr) -> exp.Expr:
1033    """
1034    Inherit field names from the first struct in an array.
1035
1036    BigQuery supports implicitly inheriting names from the first STRUCT in an array:
1037
1038    Example:
1039        ARRAY[
1040          STRUCT('Alice' AS name, 85 AS score),  -- defines names
1041          STRUCT('Bob', 92),                     -- inherits names
1042          STRUCT('Diana', 95)                    -- inherits names
1043        ]
1044
1045    This transformation makes the field names explicit on all structs by adding
1046    PropertyEQ nodes, in order to facilitate transpilation to other dialects.
1047
1048    Args:
1049        expression: The expression tree to transform
1050
1051    Returns:
1052        The modified expression with field names inherited in all structs
1053    """
1054    if (
1055        isinstance(expression, exp.Array)
1056        and expression.args.get("struct_name_inheritance")
1057        and isinstance(first_item := seq_get(expression.expressions, 0), exp.Struct)
1058        and all(isinstance(fld, exp.PropertyEQ) for fld in first_item.expressions)
1059    ):
1060        field_names: list[exp.Identifier] = [fld.this for fld in first_item.expressions]
1061
1062        # Apply field names to subsequent structs that don't have them
1063        for struct in expression.expressions[1:]:
1064            if not isinstance(struct, exp.Struct) or len(struct.expressions) != len(field_names):
1065                continue
1066
1067            # Convert unnamed expressions to PropertyEQ with inherited names
1068            new_expressions: list[exp.PropertyEQ] = []
1069            for i, expr in enumerate(struct.expressions):
1070                if not isinstance(expr, exp.PropertyEQ):
1071                    # Create PropertyEQ: field_name := value, preserving the type from the inner expression
1072                    property_eq = exp.PropertyEQ(
1073                        this=field_names[i].copy(),
1074                        expression=expr,
1075                    )
1076                    property_eq.type = expr.type
1077                    new_expressions.append(property_eq)
1078                else:
1079                    new_expressions.append(expr)
1080
1081            struct.set("expressions", new_expressions)
1082
1083    return expression