transforms: list[t.Callable[[exp.Expr], exp.Expr]],

generator: t.Callable[[Generator, exp.Expr], str] | None = None,

) -> t.Callable[[Generator, exp.Expr], str]:

"""

def _to_sql(self: Generator, expression: exp.Expr) -> str:

expression_type = type(expression)

try:

expression = transforms[0](expression)

for transform in transforms[1:]:

expression = transform(expression)

except UnsupportedError as unsupported_error:

self.unsupported(str(unsupported_error))

if generator:

return generator(self, expression)

_sql_handler: SqlHandler | None = getattr(self, expression.key + "_sql", None)

if _sql_handler:

return _sql_handler(expression)

transforms_handler = self.TRANSFORMS.get(type(expression))

if transforms_handler:

if expression_type is type(expression):

if isinstance(expression, exp.Func):

return self.function_fallback_sql(expression)

# Ensures we don't enter an infinite loop. This can happen when the original expression

# has the same type as the final expression and there's no _sql method available for it,

# because then it'd re-enter _to_sql.

raise ValueError(

f"Expr type {expression.__class__.__name__} requires a _sql method in order to be transformed."

)

return transforms_handler(self, expression)

raise ValueError(f"Unsupported expression type {expression.__class__.__name__}.")

if isinstance(expression, exp.Select):

count = 0

recursive_ctes: list[exp.Expr] = []

for unnest in expression.find_all(exp.Unnest):

if (

not isinstance(unnest.parent, (exp.From, exp.Join))

or len(unnest.expressions) != 1

or not isinstance(unnest.expressions[0], exp.GenerateDateArray)

):

continue

generate_date_array = unnest.expressions[0]

start: exp.Expr | None = generate_date_array.args.get("start")

end: exp.Expr | None = generate_date_array.args.get("end")

step: exp.Expr | None = generate_date_array.args.get("step")

if not start or not end or not isinstance(step, exp.Interval):

continue

alias: exp.TableAlias | None = unnest.args.get("alias")

column_name: str = (

alias.columns[0] if isinstance(alias, exp.TableAlias) else "date_value"

)

start = exp.cast(start, "date")

date_add = exp.func(

"date_add", column_name, exp.Literal.number(step.name), step.args.get("unit")

)

cast_date_add = exp.cast(date_add, "date")

cte_name = "_generated_dates" + (f"_{count}" if count else "")

base_query = exp.select(start.as_(column_name))

recursive_query = (

exp.select(cast_date_add)

.from_(cte_name)

.where(cast_date_add <= exp.cast(end, "date"))

)

cte_query = base_query.union(recursive_query, distinct=False)

generate_dates_query = exp.select(column_name).from_(cte_name)

unnest.replace(generate_dates_query.subquery(cte_name))

recursive_ctes.append(

exp.alias_(exp.CTE(this=cte_query), cte_name, table=[column_name])

)

count += 1

if recursive_ctes:

with_expression: exp.With = expression.args.get("with_") or exp.With()

with_expression.set("recursive", True)

with_expression.set("expressions", [*recursive_ctes, *with_expression.expressions])

expression.set("with_", with_expression)

"""Unnests GENERATE_SERIES or SEQUENCE table references."""

this = expression.this

if isinstance(expression, exp.Table) and isinstance(this, exp.GenerateSeries):

unnest = exp.Unnest(expressions=[this])

if expression.alias:

return exp.alias_(unnest, alias="_u", table=[expression.alias], copy=False)

return unnest

"""

if (

isinstance(expression, exp.Select)

and expression.args.get("distinct")

and isinstance(expression.args["distinct"].args.get("on"), exp.Tuple)

):

row_number_window_alias = find_new_name(expression.named_selects, "_row_number")

distinct_cols = expression.args["distinct"].pop().args["on"].expressions

window = exp.Window(this=exp.RowNumber(), partition_by=distinct_cols)

order: exp.Order | None = expression.args.get("order")

if order:

window.set("order", order.pop())

else:

window.set("order", exp.Order(expressions=[c.copy() for c in distinct_cols]))

expression.select(exp.alias_(window, row_number_window_alias), copy=False)

# We add aliases to the projections so that we can safely reference them in the outer query

new_selects: list[exp.Expr] = []

taken_names = {row_number_window_alias}

for select in expression.selects[:-1]:

if select.is_star:

new_selects = [exp.Star()]

break

if not isinstance(select, exp.Alias):

alias = find_new_name(taken_names, select.output_name or "_col")

quoted: bool | None = (

select.this.args.get("quoted") if isinstance(select, exp.Column) else None

)

select = select.replace(exp.alias_(select, alias, quoted=quoted))

taken_names.add(select.output_name)

new_selects.append(select.args["alias"])

return (

exp.select(*new_selects, copy=False)

.from_(expression.subquery("_t", copy=False), copy=False)

.where(exp.column(row_number_window_alias).eq(1), copy=False)

)

"""

if isinstance(expression, exp.Select) and expression.args.get("qualify"):

taken = set(expression.named_selects)

for select in expression.selects:

if not select.alias_or_name:

alias = find_new_name(taken, "_c")

select.replace(exp.alias_(select, alias))

taken.add(alias)

def _select_alias_or_name(select: exp.Expr) -> str | exp.Column:

alias_or_name = select.alias_or_name

identifier = select.args.get("alias") or select.this

if isinstance(identifier, exp.Identifier):

return exp.column(alias_or_name, quoted=identifier.args.get("quoted"))

return alias_or_name

outer_selects = exp.select(*map(_select_alias_or_name, expression.selects))

qualify_filters: exp.Expr = expression.args["qualify"].pop().this

expression_by_alias: dict[str, exp.Expr] = {

select.alias: select.this

for select in expression.selects

if isinstance(select, exp.Alias)

}

select_candidates = (exp.Window,) if expression.is_star else (exp.Window, exp.Column)

for select_candidate in list(qualify_filters.find_all(*select_candidates)):

if isinstance(select_candidate, exp.Window):

if expression_by_alias:

for column in select_candidate.find_all(exp.Column):

expr = expression_by_alias.get(column.name)

if expr:

column.replace(expr)

alias = find_new_name(expression.named_selects, "_w")

expression.select(exp.alias_(select_candidate, alias), copy=False)

column = exp.column(alias)

if isinstance(select_candidate.parent, exp.Qualify):

qualify_filters = column

else:

select_candidate.replace(column)

elif select_candidate.name not in expression.named_selects:

expression.select(select_candidate.copy(), copy=False)

return outer_selects.from_(expression.subquery(alias="_t", copy=False), copy=False).where(

qualify_filters, copy=False

)

"""

for node in expression.find_all(exp.DataType):

node.set(

"expressions", [e for e in node.expressions if not isinstance(e, exp.DataTypeParam)]

)

"""Remove references to unnest table aliases, added by the optimizer's qualify_columns step."""

from sqlglot.optimizer.scope import find_all_in_scope

if isinstance(expression, exp.Select):

unnest_aliases = {

unnest.alias

for unnest in find_all_in_scope(expression, exp.Unnest)

if isinstance(unnest.parent, (exp.From, exp.Join))

}

if unnest_aliases:

for column in expression.find_all(exp.Column):

leftmost_part = column.parts[0]

if leftmost_part.arg_key != "this" and leftmost_part.this in unnest_aliases:

leftmost_part.pop()

expression: exp.Expr,

unnest_using_arrays_zip: bool = True,

) -> exp.Expr:

"""Convert cross join unnest into lateral view explode."""

def _unnest_zip_exprs(

u: exp.Unnest, unnest_exprs: list[exp.Expr], has_multi_expr: bool

) -> list[exp.Expr]:

if has_multi_expr:

if not unnest_using_arrays_zip:

raise UnsupportedError("Cannot transpile UNNEST with multiple input arrays")

# Use INLINE(ARRAYS_ZIP(...)) for multiple expressions

zip_exprs: list[exp.Expr] = [exp.Anonymous(this="ARRAYS_ZIP", expressions=unnest_exprs)]

u.set("expressions", zip_exprs)

return zip_exprs

return unnest_exprs

def _udtf_type(u: exp.Unnest, has_multi_expr: bool) -> type[exp.Func]:

if u.args.get("offset"):

return exp.Posexplode

return exp.Inline if has_multi_expr else exp.Explode

if isinstance(expression, exp.Select):

from_ = expression.args.get("from_")

if from_ and isinstance(from_.this, exp.Unnest):

unnest: exp.Unnest = from_.this

alias: exp.TableAlias | None = unnest.args.get("alias")

exprs: list[exp.Expr] = unnest.expressions

has_multi_expr = len(exprs) > 1

this, *_ = _unnest_zip_exprs(unnest, exprs, has_multi_expr)

columns: list[exp.Identifier] = alias.columns if alias else []

offset: exp.Expr | None = unnest.args.get("offset")

if offset:

columns.insert(

0, offset if isinstance(offset, exp.Identifier) else exp.to_identifier("pos")

)

unnest.replace(

exp.Table(

this=_udtf_type(unnest, has_multi_expr)(this=this),

alias=exp.TableAlias(this=alias.this, columns=columns) if alias else None,

)

)

joins: list[exp.Join] = expression.args.get("joins") or []

for join in list(joins):

join_expr = join.this

is_lateral = isinstance(join_expr, exp.Lateral)

unnest = join_expr.this if is_lateral else join_expr

if isinstance(unnest, exp.Unnest):

if is_lateral:

alias = join_expr.args.get("alias")

else:

alias = unnest.args.get("alias")

if alias is None:

raise UnsupportedError(

"CROSS JOIN UNNEST to LATERAL VIEW EXPLODE transformation requires an alias"

)

exprs = unnest.expressions

# The number of unnest.expressions will be changed by _unnest_zip_exprs, we need to record it here

has_multi_expr = len(exprs) > 1

exprs = _unnest_zip_exprs(unnest, exprs, has_multi_expr)

joins.remove(join)

alias_cols: list[exp.Identifier] = alias.columns

# # Handle UNNEST to LATERAL VIEW EXPLODE: Exception is raised when there are 0 or > 2 aliases

# 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.

# Refs: https://spark.apache.org/docs/latest/sql-ref-syntax-qry-select-lateral-view.html

if not has_multi_expr and len(alias_cols) not in (1, 2):

raise UnsupportedError(

"CROSS JOIN UNNEST to LATERAL VIEW EXPLODE transformation requires explicit column aliases"

)

offset = unnest.args.get("offset")

if offset:

alias_cols.insert(

0,

offset if isinstance(offset, exp.Identifier) else exp.to_identifier("pos"),

)

for e, column in zip(exprs, alias_cols):

expression.append(

"laterals",

exp.Lateral(

this=_udtf_type(unnest, has_multi_expr)(this=e),

view=True,

alias=exp.TableAlias(this=alias.this, columns=alias_cols),

),

)

index_offset: int = 0,

) -> t.Callable[[exp.Expr], exp.Expr]:

"""Convert explode/posexplode projections into unnests."""

def _explode_projection_to_unnest(expression: exp.Expr) -> exp.Expr:

if isinstance(expression, exp.Select):

from sqlglot.optimizer.scope import Scope

taken_select_names = set(expression.named_selects)

taken_source_names = {name for name, _ in Scope(expression).references}

def new_name(names: set[str], name: str) -> str:

name = find_new_name(names, name)

names.add(name)

return name

arrays: list[exp.Condition] = []

series_alias = new_name(taken_select_names, "pos")

series = exp.alias_(

exp.Unnest(

expressions=[exp.GenerateSeries(start=exp.Literal.number(index_offset))]

),

new_name(taken_source_names, "_u"),

table=[series_alias],

)

# we use list here because expression.selects is mutated inside the loop

for select in list(expression.selects):

explode = select.find(exp.Explode)

if explode:

pos_alias: t.Any = ""

explode_alias: t.Any = ""

if isinstance(select, exp.Alias):

explode_alias = select.args["alias"]

alias: exp.Expr = select

elif isinstance(select, exp.Aliases):

pos_alias = select.aliases[0]

explode_alias = select.aliases[1]

alias = select.replace(exp.alias_(select.this, "", copy=False))

else:

alias = select.replace(exp.alias_(select, ""))

explode = alias.find(exp.Explode)

assert explode

is_posexplode = isinstance(explode, exp.Posexplode)

explode_arg = explode.this

if isinstance(explode, exp.ExplodeOuter):

bracket = explode_arg[0]

bracket.set("safe", True)

bracket.set("offset", True)

explode_arg = exp.func(

"IF",

exp.func(

"ARRAY_SIZE", exp.func("COALESCE", explode_arg, exp.Array())

).eq(0),

exp.array(bracket, copy=False),

explode_arg,

)

# This ensures that we won't use [POS]EXPLODE's argument as a new selection

if isinstance(explode_arg, exp.Column):

taken_select_names.add(explode_arg.output_name)

unnest_source_alias = new_name(taken_source_names, "_u")

if not explode_alias:

explode_alias = new_name(taken_select_names, "col")

if is_posexplode:

pos_alias = new_name(taken_select_names, "pos")

if not pos_alias:

pos_alias = new_name(taken_select_names, "pos")

alias.set("alias", exp.to_identifier(explode_alias))

series_table_alias = series.args["alias"].this

column = exp.If(

this=exp.column(series_alias, table=series_table_alias).eq(

exp.column(pos_alias, table=unnest_source_alias)

),

true=exp.column(explode_alias, table=unnest_source_alias),

)

explode.replace(column)

if is_posexplode:

expressions = expression.expressions

expressions.insert(

expressions.index(alias) + 1,

exp.If(

this=exp.column(series_alias, table=series_table_alias).eq(

exp.column(pos_alias, table=unnest_source_alias)

),

true=exp.column(pos_alias, table=unnest_source_alias),

).as_(pos_alias),

)

expression.set("expressions", expressions)

if not arrays:

if expression.args.get("from_"):

expression.join(series, copy=False, join_type="CROSS")

else:

expression.from_(series, copy=False)

size: exp.Condition = exp.ArraySize(this=explode_arg.copy())

arrays.append(size)

# trino doesn't support left join unnest with on conditions

# if it did, this would be much simpler

expression.join(

exp.alias_(

exp.Unnest(

expressions=[explode_arg.copy()],

offset=exp.to_identifier(pos_alias),

),

unnest_source_alias,

table=[explode_alias],

),

join_type="CROSS",

copy=False,

)

if index_offset != 1:

size = size - 1

expression.where(

exp.column(series_alias, table=series_table_alias)

.eq(exp.column(pos_alias, table=unnest_source_alias))

.or_(

(exp.column(series_alias, table=series_table_alias) > size).and_(

exp.column(pos_alias, table=unnest_source_alias).eq(size)

)

),

copy=False,

)

if arrays:

end: exp.Condition = exp.Greatest(this=arrays[0], expressions=arrays[1:])

if index_offset != 1:

end = end - (1 - index_offset)

series.expressions[0].set("end", end)

return expression

"""Transforms percentiles by adding a WITHIN GROUP clause to them."""

if (

isinstance(expression, exp.PERCENTILES)

and not isinstance(expression.parent, exp.WithinGroup)

and expression.expression

):

column = expression.this.pop()

expression.set("this", expression.expression.pop())

order = exp.Order(expressions=[exp.Ordered(this=column)])

expression = exp.WithinGroup(this=expression, expression=order)

"""Transforms percentiles by getting rid of their corresponding WITHIN GROUP clause."""

if (

isinstance(expression, exp.WithinGroup)

and isinstance(expression.this, exp.PERCENTILES)

and isinstance(expression.expression, exp.Order)

):

quantile = expression.this.this

input_value = t.cast(exp.Ordered, expression.find(exp.Ordered)).this

return expression.replace(exp.ApproxQuantile(this=input_value, quantile=quantile))

"""Uses projection output names in recursive CTE definitions to define the CTEs' columns."""

if isinstance(expression, exp.With) and expression.recursive:

next_name = name_sequence("_c_")

for cte in expression.expressions:

if not cte.args["alias"].columns:

query = cte.this

if isinstance(query, exp.SetOperation):

query = query.this

cte.args["alias"].set(

"columns",

[exp.to_identifier(s.alias_or_name or next_name()) for s in query.selects],

)

"""Replace 'epoch' in casts by the equivalent date literal."""

if (

isinstance(expression, (exp.Cast, exp.TryCast))

and expression.name.lower() == "epoch"

and expression.to.this in exp.DataType.TEMPORAL_TYPES

):

expression.this.replace(exp.Literal.string("1970-01-01 00:00:00"))

"""Convert SEMI and ANTI joins into equivalent forms that use EXIST instead."""

if isinstance(expression, exp.Select):

for join in list[exp.Join](expression.args.get("joins") or []):

on: exp.Expr | None = join.args.get("on")

if on and join.kind in ("SEMI", "ANTI"):

subquery = exp.select("1").from_(join.this).where(on)

exists: exp.Exists | exp.Not = exp.Exists(this=subquery)

if join.kind == "ANTI":

exists = exists.not_(copy=False)

join.pop()

expression.where(exists, copy=False)

"""

if isinstance(expression, exp.Select):

full_outer_joins: list[tuple[int, exp.Join]] = [

(index, join)

for index, join in enumerate[exp.Join](expression.args.get("joins") or [])

if join.side == "FULL"

]

if len(full_outer_joins) == 1:

expression_copy = expression.copy()

expression.set("limit", None)

index, full_outer_join = full_outer_joins[0]

tables = (expression.args["from_"].alias_or_name, full_outer_join.alias_or_name)

join_conditions = full_outer_join.args.get("on") or exp.and_(

*[

exp.column(col, tables[0]).eq(exp.column(col, tables[1]))

for col in t.cast(list[exp.Identifier], full_outer_join.args.get("using"))

]

)

full_outer_join.set("side", "left")

anti_join_clause = (

exp.select("1").from_(expression.args["from_"]).where(join_conditions)

)

expression_copy.args["joins"][index].set("side", "right")

expression_copy = expression_copy.where(exp.Exists(this=anti_join_clause).not_())

expression_copy.set("with_", None) # remove CTEs from RIGHT side

expression.set("order", None) # remove order by from LEFT side

return exp.union(expression, expression_copy, copy=False, distinct=False)

"""

top_level_with: exp.With | None = expression.args.get("with_")

for inner_with in expression.find_all(exp.With):

if inner_with.parent is expression:

continue

if not top_level_with:

top_level_with = inner_with.pop()

expression.set("with_", top_level_with)

else:

if inner_with.recursive:

top_level_with.set("recursive", True)

parent_cte = inner_with.find_ancestor(exp.CTE)

inner_with.pop()

if parent_cte:

i = top_level_with.expressions.index(parent_cte)

top_level_with.expressions[i:i] = inner_with.expressions

top_level_with.set("expressions", top_level_with.expressions)

else:

top_level_with.set(

"expressions", top_level_with.expressions + inner_with.expressions

)

"""Converts numeric values used in conditions into explicit boolean expressions."""

from sqlglot.optimizer.canonicalize import ensure_bools

def _ensure_bool(node: exp.Expr) -> None:

if (

node.is_number

or (

not isinstance(node, exp.SubqueryPredicate)

and node.is_type(exp.DType.UNKNOWN, *exp.DataType.NUMERIC_TYPES)

)

or (isinstance(node, exp.Column) and not node.type)

):

node.replace(node.neq(0))

for node in expression.walk():

ensure_bools(node, _ensure_bool)

for column in expression.find_all(exp.Column):

# We only wanna pop off the table, db, catalog args

for part in column.parts[:-1]:

part.pop()

"""

if isinstance(expression, exp.Pivot):

expression.set("fields", [unqualify_columns(field) for field in expression.fields])

assert isinstance(expression, exp.Create)

for constraint in expression.find_all(exp.UniqueColumnConstraint):

if constraint.parent:

constraint.parent.pop()

expression: exp.Expr,

tmp_storage_provider: t.Callable[[exp.Expr], exp.Expr] = lambda e: e,

) -> exp.Expr:

assert isinstance(expression, exp.Create)

properties: exp.Properties | None = expression.args.get("properties")

temporary = any(

isinstance(prop, exp.TemporaryProperty)

for prop in (properties.expressions if properties is not None else [])

)

# CTAS with temp tables map to CREATE TEMPORARY VIEW

if expression.kind == "TABLE" and temporary:

if expression.expression:

return exp.Create(

kind="TEMPORARY VIEW",

this=expression.this,

expression=expression.expression,

)

return tmp_storage_provider(expression)

"""

assert isinstance(expression, exp.Create)

schema = expression.this

is_partitionable = expression.kind in {"TABLE", "VIEW"}

if isinstance(schema, exp.Schema) and is_partitionable:

prop = expression.find(exp.PartitionedByProperty)

if prop and prop.this and not isinstance(prop.this, exp.Schema):

columns: set[str] = {v.name.upper() for v in prop.this.expressions}

schema_exprs: list[exp.Expr] = schema.expressions

partitions = [col for col in schema_exprs if col.name.upper() in columns]

schema.set("expressions", [e for e in schema_exprs if e not in partitions])

prop.replace(exp.PartitionedByProperty(this=exp.Schema(expressions=partitions)))

expression.set("this", schema)

"""

assert isinstance(expression, exp.Create)

prop = expression.find(exp.PartitionedByProperty)

if (

prop

and prop.this

and isinstance(prop.this, exp.Schema)

and all(isinstance(e, exp.ColumnDef) and e.kind for e in prop.this.expressions)

):

prop_this = exp.Tuple(

expressions=[exp.to_identifier(e.this) for e in prop.this.expressions]

)

schema: exp.Schema = expression.this

for e in prop.this.expressions:

schema.append("expressions", e)

prop.set("this", prop_this)

"""Converts struct arguments to aliases, e.g. STRUCT(1 AS y)."""

if isinstance(expression, exp.Struct):

expression.set(

"expressions",

[

exp.alias_(e.expression, e.this) if isinstance(e, exp.PropertyEQ) else e

for e in expression.expressions

],

)

"""https://docs.oracle.com/cd/B19306_01/server.102/b14200/queries006.htm#sthref3178

from sqlglot.optimizer.scope import traverse_scope

from sqlglot.optimizer.normalize import normalize, normalized

from collections import defaultdict

# we go in reverse to check the main query for left correlation

for scope in reversed(traverse_scope(expression)):

query = scope.expression

where: exp.Expr | None = query.args.get("where")

joins: list[exp.Join] = query.args.get("joins", [])

if not where or not any(c.args.get("join_mark") for c in where.find_all(exp.Column)):

continue

# knockout: we do not support left correlation (see point 2)

assert not scope.is_correlated_subquery, "Correlated queries are not supported"

# make sure we have AND of ORs to have clear join terms

where = normalize(where.this)

assert normalized(where), "Cannot normalize JOIN predicates"

# dict of {name: list of join AND conditions}

joins_ons: defaultdict[str, list[exp.Expr]] = defaultdict(list)

for cond in [where] if not isinstance(where, exp.And) else where.flatten():

join_cols = [col for col in cond.find_all(exp.Column) if col.args.get("join_mark")]

left_join_table = set(col.table for col in join_cols)

if not left_join_table:

continue

assert not (len(left_join_table) > 1), (

"Cannot combine JOIN predicates from different tables"

)

for col in join_cols:

col.set("join_mark", False)

joins_ons[left_join_table.pop()].append(cond)

old_joins = {join.alias_or_name: join for join in joins}

new_joins: dict[str, exp.Join] = {}

query_from = query.args["from_"]

for table, predicates in joins_ons.items():

join_what = old_joins.get(table, query_from).this.copy()

new_joins[join_what.alias_or_name] = exp.Join(

this=join_what, on=exp.and_(*predicates), kind="LEFT"

)

for p in predicates:

while isinstance(p.parent, exp.Paren):

p.parent.replace(p)

parent = p.parent

p.pop()

if isinstance(parent, exp.Binary):

left = parent.args.get("this")

parent.replace(parent.right if left is None else left)

elif isinstance(parent, exp.Where):

parent.pop()

if query_from.alias_or_name in new_joins:

only_old_joins: set[str] = old_joins.keys() - new_joins.keys()

assert len(only_old_joins) >= 1, (

"Cannot determine which table to use in the new FROM clause"

)

new_from_name = list[str](only_old_joins)[0]

query.set("from_", exp.From(this=old_joins[new_from_name].this))

if new_joins:

for n, j in old_joins.items(): # preserve any other joins

if n not in new_joins and n != query.args["from_"].name:

if not j.kind:

j.set("kind", "CROSS")

new_joins[n] = j

query.set("joins", list(new_joins.values()))

"""

if isinstance(expression, exp.Select):

for any_expr in expression.find_all(exp.Any):

this: exp.Expr = any_expr.this

if isinstance(this, exp.Query) or isinstance(any_expr.parent, (exp.Like, exp.ILike)):

continue

binop = any_expr.parent

if isinstance(binop, exp.Binary):

lambda_arg = exp.to_identifier("x")

any_expr.replace(lambda_arg)

lambda_expr = exp.Lambda(this=binop.copy(), expressions=[lambda_arg])

binop.replace(exp.Exists(this=this.unnest(), expression=lambda_expr))