# SPDX-FileCopyrightText: 2025 ash_neo4j contributors # # SPDX-License-Identifier: MIT defmodule AshNeo4j.QueryHelper do require Logger alias AshNeo4j.Cypher alias AshNeo4j.Cypher.Query alias AshNeo4j.Resource.Info, as: ResourceInfo alias AshNeo4j.ResourceMapping @moduledoc """ AshNeo4j DataLayer QueryHelper """ @doc """ Queries nodes, using Ash.Query """ @spec query_nodes(struct()) :: {:error, any()} | {:ok, any()} def query_nodes(ash_query) when is_struct(ash_query) do case Map.get(ash_query, :combination_of, []) do [] -> run_simple_query(ash_query) combinations -> run_combination_query(ash_query, combinations) end end defp run_simple_query(ash_query) do mapping = ResourceInfo.mapping(ash_query.resource) # A predicate we couldn't form (e.g. an unresolvable traverse, #342) is an # `{:error, _}` that doesn't match `%Query{}` — `with` passes it straight # through rather than running a fabricated query. with %Query{} = base <- build_query(ash_query, mapping), {:ok, {terms, sort_params}} <- sort_terms(ash_query, mapping) do query = base |> Query.merge_params(sort_params) |> Query.paginate_nodes(terms, ash_query.offset, ash_query.limit) |> Query.add_order_by(terms) run_cypher_query(query) end end defp run_combination_query(ash_query, combinations) do mapping = ResourceInfo.mapping(ash_query.resource) case classify_combination(combinations) do {:ok, :native, union_type, branch_dl_queries} -> run_native_combination(ash_query, mapping, branch_dl_queries, union_type) {:ok, :in_memory, all_types, branch_dl_queries} -> run_in_memory_combination(ash_query, mapping, branch_dl_queries, all_types) {:error, reason} -> {:error, reason} end end defp run_native_combination(ash_query, mapping, branch_dl_queries, union_type) do branch_queries = branch_dl_queries |> Enum.with_index() |> Enum.map(fn {branch_dl_query, idx} -> build_branch_query(branch_dl_query, mapping, "b#{idx}_", :nodes) end) with nil <- Enum.find(branch_queries, &match?({:error, _}, &1)), {:ok, {terms, sort_params}} <- sort_terms(ash_query, mapping) do query = branch_queries |> Query.combination_block(union_type: union_type) |> Query.merge_params(sort_params) |> Query.paginate_nodes(terms, ash_query.offset, ash_query.limit) |> Query.add_order_by(terms) run_cypher_query(query) end end defp run_in_memory_combination(ash_query, mapping, branch_dl_queries, all_types) do branch_id_results = branch_dl_queries |> Enum.with_index() |> Enum.map(fn {branch_dl_query, idx} -> with %Query{} = query <- build_branch_query(branch_dl_query, mapping, "b#{idx}_", :ids), {:ok, %Bolty.Response{results: results}} <- Cypher.run(query) do {:ok, MapSet.new(results, &Map.get(&1, "sid"))} end end) case Enum.find(branch_id_results, &match?({:error, _}, &1)) do nil -> [base_set | rest_sets] = Enum.map(branch_id_results, fn {:ok, s} -> s end) rest_types = tl(all_types) keep_set = apply_set_ops(base_set, Enum.zip(rest_types, rest_sets)) keep_ids = MapSet.to_list(keep_set) if keep_ids == [] do {:ok, []} else with {:ok, {terms, sort_params}} <- sort_terms(ash_query, mapping) do final = mapping.label_pair |> Query.node_read_by_ids(keep_ids) |> Query.merge_params(sort_params) |> Query.paginate_nodes(terms, ash_query.offset, ash_query.limit) |> Query.add_order_by(terms) run_cypher_query(final) end end {:error, reason} -> {:error, reason} end end defp apply_set_ops(initial_set, []), do: initial_set defp apply_set_ops(running_set, [{op, branch_set} | rest]) do new_running = case op do # :union and :union_all are equivalent over id sets — MapSet dedups # regardless. The Cypher-level distinction only matters when the # native CALL+UNION pushdown path runs. :union -> MapSet.union(running_set, branch_set) :union_all -> MapSet.union(running_set, branch_set) :intersect -> MapSet.intersection(running_set, branch_set) :except -> MapSet.difference(running_set, branch_set) end apply_set_ops(new_running, rest) end defp run_cypher_query(query) do case Cypher.run(query) do {:ok, %Bolty.Response{results: results}} -> {:ok, results} {:error, error} -> {:error, AshNeo4j.Error.Neo4j.from_bolt(error)} end end # Walks the combination list and decides the execution path. # First element must be `:base`. All-`:union` or all-`:union_all` subsequents # take the native CALL+UNION pushdown path. Anything else (mixed union types, # or any `:intersect` / `:except`) takes the in-memory orchestration path. defp classify_combination([{:base, base_query} | rest]) do types = Enum.map(rest, &elem(&1, 0)) branch_queries = [base_query | Enum.map(rest, &elem(&1, 1))] cond do types == [] -> {:ok, :native, :union_all, branch_queries} Enum.all?(types, &(&1 == :union)) -> {:ok, :native, :union, branch_queries} Enum.all?(types, &(&1 == :union_all)) -> {:ok, :native, :union_all, branch_queries} true -> {:ok, :in_memory, [:base | types], branch_queries} end end defp classify_combination(_), do: {:error, AshNeo4j.Error.Internal.exception(detail: "combination_of must start with :base")} # builder: :nodes | :ids — which branch_node_read variant to use. # Returns the branch Cypher.Query, or `{:error, _}` if a branch predicate # can't be formed (threaded up by the combination runners). defp build_branch_query(branch_dl_query, %ResourceMapping{} = mapping, param_prefix, builder) do with {:ok, conditions} <- extract_branch_conditions(branch_dl_query, mapping) do build = if builder == :ids, do: &Query.branch_node_read_ids/3, else: &Query.branch_node_read/3 build.(mapping.label_pair, conditions, param_prefix: param_prefix) end end defp extract_branch_conditions(branch_dl_query, %ResourceMapping{} = mapping) do case branch_dl_query.filter do nil -> {:ok, []} filter -> simple_filter = Ash.Filter.to_simple_filter(filter, skip_invalid?: true) predicates = simple_filter |> Map.get(:predicates, []) |> Enum.reject(fn pred -> match?(%Ash.Query.Ref{attribute: %Ash.Query.Calculation{}}, Map.get(pred, :left)) end) to_conditions(mapping, predicates) end end defp build_query(ash_query, %ResourceMapping{} = mapping) do expression = if ash_query.filter, do: Map.get(ash_query.filter, :expression) cond do ash_query.filter == nil -> Query.node_read(mapping.label_pair) # A filter that *is* a single `fragment(...)` (the Cypher escape hatch, #33) — # render it straight into the WHERE. match?(%Ash.Query.Function.Fragment{}, expression) -> build_fragment_query(expression, mapping) # A fragment combined with other conditions can't be rendered yet, and raw # Cypher can't be evaluated in-memory — refuse rather than silently mis-answer. contains_fragment?(expression) -> {:error, AshNeo4j.Error.UnsupportedFilterFragment.exception(resource: mapping.module, reason: :combined)} true -> simple_filter = Ash.Filter.to_simple_filter(ash_query.filter, skip_invalid?: true) predicates = simple_filter |> Map.get(:predicates, []) |> Enum.reject(fn pred -> match?(%Ash.Query.Ref{attribute: %Ash.Query.Calculation{}}, Map.get(pred, :left)) end) if predicates == [] do Logger.debug("AshNeo4j.QueryHelper: filter #{inspect(ash_query.filter)} is not a simple filter") Query.node_read(mapping.label_pair) else build_filtered_query(mapping, predicates) end end end # Renders a single `fragment(...)` (#33) to a node read with the fragment as the # WHERE: `MATCH (s:Label) WHERE OPTIONAL MATCH (s)-[r]-(d) RETURN …`. # Refuses (no silent in-memory fallback) when an argument isn't a plain attribute # reference or literal. defp build_fragment_query(%Ash.Query.Function.Fragment{} = fragment, %ResourceMapping{} = mapping) do case render_fragment(fragment, mapping) do {:ok, {where, params}} -> Query.node_read_fragment(mapping.label_pair, where, params) {:error, reason} -> {:error, AshNeo4j.Error.UnsupportedFilterFragment.exception(resource: mapping.module, reason: reason)} end end @doc false # Walks the fragment tokens — `{:raw, cypher}` literal (author-controlled), an # attribute `{:expr, ref}` → `s.`, a literal `{:expr, value}` → a bound # `$frag_N` param (injection-safe). Other argument shapes are refused. Public for # testing the render of a `fragment(...)` without a live (APOC) server. def render_fragment(%Ash.Query.Function.Fragment{arguments: tokens}, %ResourceMapping{} = mapping) do tokens |> Enum.reduce_while({:ok, {"", %{}, 0}}, fn token, {:ok, {acc_str, acc_params, index}} -> case render_fragment_token(token, mapping, index) do {:ok, piece, params, next_index} -> {:cont, {:ok, {acc_str <> piece, Map.merge(acc_params, params), next_index}}} {:error, _} = error -> {:halt, error} end end) |> case do {:ok, {where, params, _index}} -> {:ok, {where, params}} error -> error end end defp render_fragment_token({:raw, cypher}, _mapping, index) when is_binary(cypher), do: {:ok, cypher, %{}, index} defp render_fragment_token({:expr, %Ash.Query.Ref{attribute: attribute, relationship_path: []}}, mapping, index) do name = Map.get(attribute, :name, attribute) if Ash.Resource.Info.attribute(mapping.module, name) do {:ok, "s.#{Keyword.get(mapping.properties, name, name)}", %{}, index} else {:error, {:unsupported_argument, attribute}} end end defp render_fragment_token({:expr, value}, _mapping, index) when is_number(value) or is_binary(value) or is_boolean(value) do key = "frag_#{index}" {:ok, "$#{key}", %{key => value}, index + 1} end defp render_fragment_token({:expr, other}, _mapping, _index), do: {:error, {:unsupported_argument, other}} defp contains_fragment?(%Ash.Query.Function.Fragment{}), do: true defp contains_fragment?(%_struct{} = struct), do: struct |> Map.from_struct() |> Map.values() |> Enum.any?(&contains_fragment?/1) defp contains_fragment?(list) when is_list(list), do: Enum.any?(list, &contains_fragment?/1) defp contains_fragment?(%{} = map), do: map |> Map.values() |> Enum.any?(&contains_fragment?/1) defp contains_fragment?(_), do: false defp build_filtered_query(%ResourceMapping{} = mapping, predicates) do case Enum.find(predicates, &traverse_predicate?/1) do nil -> build_relationship_or_property_query(mapping, predicates) traverse_predicate -> build_traversal_query(mapping, traverse_predicate) end end # A reached-node predicate over a traversal: a field comparison (#321), a # field aggregate (#338), or a spatial predicate with the traversal as its # geometry argument (#330). defp traverse_predicate?(%{left: %AshNeo4j.Functions.Traverse{arguments: [_chain, {agg, field}]}}) when agg in [:min, :max, :avg, :sum] and is_atom(field), do: true defp traverse_predicate?(%{left: %AshNeo4j.Functions.Traverse{arguments: [_chain, field]}}) when is_atom(field), do: true defp traverse_predicate?(%AshNeo4j.Functions.StDwithin{arguments: [%AshNeo4j.Functions.Traverse{} | _]}), do: true defp traverse_predicate?(%AshNeo4j.Functions.StContains{arguments: [%AshNeo4j.Functions.Traverse{} | _]}), do: true defp traverse_predicate?(%{left: %AshNeo4j.Functions.StDistance{arguments: [%AshNeo4j.Functions.Traverse{} | _]}}), do: true defp traverse_predicate?(%{left: %AshNeo4j.Functions.StDistanceInMeters{arguments: [%AshNeo4j.Functions.Traverse{} | _]}}), do: true defp traverse_predicate?(_), do: false # Membership over the reached set (#334): `traverse(^chain, :exists) == true|false`. # `:exists` renders to `EXISTS {}` / `NOT EXISTS {}` — no reached-node field, so # no reached-resource typing needed (composes over reverse chains too). defp build_traversal_query(%ResourceMapping{} = mapping, %{ operator: operator, left: %AshNeo4j.Functions.Traverse{arguments: [chain, :exists]}, right: value }) when operator in [:==, :!=] and is_boolean(value) do with {:ok, {segments, _reached}} <- chain_segments(mapping, chain) do exists? = if operator == :==, do: value, else: not value traversal_predicate(mapping, segments, {:exists, exists?}) end end # Cardinality over the reached set (#334): `traverse(^chain, :count) n`. defp build_traversal_query(%ResourceMapping{} = mapping, %{ operator: operator, left: %AshNeo4j.Functions.Traverse{arguments: [chain, :count]}, right: value }) when operator in [:==, :!=, :<, :<=, :>, :>=] and is_integer(value) do with {:ok, {segments, _reached}} <- chain_segments(mapping, chain) do traversal_predicate(mapping, segments, {:count, operator, value}) end end # Field aggregate over the reached set (#338): `traverse(^chain, {:min, :field}) value`. # Reads `d.field`, so it needs the reached resource's property mapping — a # forward relationship-name chain resolves it; a reverse-terminal chain (#336) # gives `reached = nil` and falls back to an unfiltered read. defp build_traversal_query(%ResourceMapping{} = mapping, %{ operator: operator, left: %AshNeo4j.Functions.Traverse{arguments: [chain, {agg, field}]}, right: value }) when agg in [:min, :max, :avg, :sum] and operator in [:==, :!=, :<, :<=, :>, :>=] do with {:ok, {segments, reached}} <- chain_segments(mapping, chain) do traversal_aggregate(mapping, segments, reached, agg, field, operator, value) end end # Reached-node field comparison: `traverse(^chain, :field) value`. defp build_traversal_query(%ResourceMapping{} = mapping, %{ operator: operator, left: %AshNeo4j.Functions.Traverse{arguments: [chain, field]}, right: value }) do with {:ok, {segments, reached}} <- chain_segments(mapping, chain) do cond do is_nil(reached) -> unresolvable(mapping, :unresolved_reached, %{chain: chain, field: field}) not mapped_property?(reached, field) -> unresolvable(mapping, :unmapped_property, %{reached: reached, field: field}) true -> condition = {property_name(ResourceInfo.mapping(reached), field), operator, to_param_value(value), false} traversal_query(mapping, segments, [condition]) end end end # Composition (#330/#332): a spatial predicate with the traversal as its geometry # argument — render it against the reached node. Needs a resolvable reached # resource (relationship-name chain) for the reached geo attribute. defp build_traversal_query(%ResourceMapping{} = mapping, %AshNeo4j.Functions.StDwithin{ arguments: [%AshNeo4j.Functions.Traverse{arguments: [chain, field]}, test_point, threshold] }) when is_number(threshold) do spatial_traversal(mapping, chain, fn reached -> geo_condition(reached, field, test_point, fn param -> {point_property(reached, field), :st_dwithin, {param, threshold}, false} end) end) end defp build_traversal_query(%ResourceMapping{} = mapping, %{ operator: operator, left: %distance{arguments: [%AshNeo4j.Functions.Traverse{arguments: [chain, field]}, test_point]}, right: threshold }) when distance in [AshNeo4j.Functions.StDistance, AshNeo4j.Functions.StDistanceInMeters] and is_number(threshold) do spatial_traversal(mapping, chain, fn reached -> geo_condition(reached, field, test_point, fn param -> {point_property(reached, field), :st_distance, {operator, param, threshold}, false} end) end) end defp build_traversal_query(%ResourceMapping{} = mapping, %AshNeo4j.Functions.StContains{ arguments: [%AshNeo4j.Functions.Traverse{arguments: [chain, field]}, %Geo.Point{} = point] }) do spatial_traversal(mapping, chain, fn reached -> if polygon_attribute?(reached, field) do {property_name(reached, field), :st_contains, to_param_value(point), false} end end) end # Unsupported traverse predicate shape — can't form it, so return an error # rather than a fabricated/unfiltered read (#342). defp build_traversal_query(%ResourceMapping{} = mapping, predicate) do unresolvable(mapping, :unsupported_predicate, %{predicate: predicate}) end # Resolves the chain + reached resource, applies `condition_fn` against the # reached mapping (`nil` = not applicable), then builds the traversal read. defp spatial_traversal(%ResourceMapping{} = mapping, chain, condition_fn) do with {:ok, {segments, reached}} <- chain_segments(mapping, chain) do if is_nil(reached) do unresolvable(mapping, :unresolved_reached, %{chain: chain}) else case condition_fn.(ResourceInfo.mapping(reached)) do nil -> unresolvable(mapping, :unmapped_property, %{reached: reached}) {:error, _} = error -> error condition -> traversal_query(mapping, segments, [condition]) end end end end # resolve_chain + map an unresolved hop to a returned `UnresolvableTraversal` # (#342), so the filter path never runs a query with a fabricated edge label. defp chain_segments(%ResourceMapping{} = mapping, chain) do case resolve_chain(mapping.module, chain) do {:ok, result} -> {:ok, result} {:error, {:unresolved_hop, hop}} -> unresolvable(mapping, :unresolved_hop, %{hop: hop, chain: chain}) end end defp traversal_query(%ResourceMapping{} = mapping, [], _conditions), do: unresolvable(mapping, :empty_chain, %{}) defp traversal_query(%ResourceMapping{} = mapping, segments, conditions) do Query.traversal_read(mapping.label_pair, segments, conditions) end defp traversal_predicate(%ResourceMapping{} = mapping, [], _agg), do: unresolvable(mapping, :empty_chain, %{}) defp traversal_predicate(%ResourceMapping{} = mapping, segments, agg) do Query.traversal_predicate_read(mapping.label_pair, segments, agg) end defp traversal_aggregate(%ResourceMapping{} = mapping, [], _reached, _agg, _field, _op, _value), do: unresolvable(mapping, :empty_chain, %{}) # No resolved reached resource (e.g. reverse-terminal chain, #336) — can't map # the reached field to a property. defp traversal_aggregate(%ResourceMapping{} = mapping, _segments, nil, agg, field, _op, _value), do: unresolvable(mapping, :unresolved_reached, %{field: field, aggregate: agg}) defp traversal_aggregate(%ResourceMapping{} = mapping, segments, reached, agg, field, op, value) do if mapped_property?(reached, field) do prop = property_name(ResourceInfo.mapping(reached), field) Query.traversal_aggregate_read(mapping.label_pair, segments, {agg, prop, op, to_param_value(value)}) else unresolvable(mapping, :unmapped_property, %{reached: reached, field: field}) end end # Builds the `{:error, %UnresolvableTraversal{}}` a data layer returns when a # traverse predicate can't be formed — `:reason` distinguishes the failure # mode, `:context` carries the specifics. defp unresolvable(%ResourceMapping{module: module}, reason, context) do {:error, AshNeo4j.Error.UnresolvableTraversal.exception(world: module, reason: reason, context: context)} end # True when `field` resolves to a property in the reached resource's mapping. defp mapped_property?(resource, field) do Keyword.has_key?(ResourceInfo.mapping(resource).properties, field) end @doc """ Resolves a hop chain to `{:ok, {[{edge_label, direction, dest_label}], reached_resource}}`, threading the current resource so relationship-name hops resolve at each step. `reached_resource` is `nil` once an explicit-edge hop breaks the resource chain. Returns `{:error, {:unresolved_hop, hop}}` when a relationship-name hop names no declared `relate` edge (so it can't be a Cypher edge label) — instead of silently using the name as the label (#342). Public so read-time consumers (e.g. the projection calculation) can turn a `chain` opt into Cypher path segments for `Cypher.Query.related_nodes/4`. """ @spec resolve_chain(module(), list()) :: {:ok, {[{atom(), atom(), atom() | nil}], module() | nil}} | {:error, {:unresolved_hop, term()}} def resolve_chain(resource, chain) when is_list(chain) do Enum.reduce_while(chain, {:ok, {[], resource}}, fn hop, {:ok, {acc, current}} -> case resolve_hop(current, hop) do {:error, _} = error -> {:halt, error} {segment, next} -> {:cont, {:ok, {acc ++ [segment], next}}} end end) end def resolve_chain(_resource, _), do: {:ok, {[], nil}} # Relationship-name hop — resolve via `relate` on the current resource. `:forward` # walks the declared edge direction, `:reverse` flips it. defp resolve_hop(resource, {direction, rel_name}) when is_atom(rel_name) and not is_nil(resource) do case ResourceInfo.node_relationship(resource, rel_name) do {_name, edge_label, edge_direction, dest_label} -> cypher_direction = if direction == :reverse, do: flip_direction(edge_direction), else: edge_direction # A `relate` edge is declared *out of* this resource, so `:reverse` of it # has no well-defined reached type — the honest typed reverse is the # explicit-edge form. Forward resolves to the relationship's destination. {dest, next} = if direction == :forward, do: {dest_label, relationship_destination(resource, rel_name)}, else: {nil, nil} {{edge_label, cypher_direction, dest}, next} _ -> # `rel_name` is not a declared `relate` edge — it can't be an edge label, # so refuse rather than fabricate one (#342). Use `{:edge, label}` to # name a raw edge explicitly. {:error, {:unresolved_hop, {direction, rel_name}}} end end # Explicit edge hop — `{:edge, label}` or `{:edge, label, dest_label}`. A given # dest label resolves to its resource so the reached node is typed. defp resolve_hop(_resource, {direction, {:edge, label}}), do: {{label, hop_direction(direction), nil}, nil} defp resolve_hop(_resource, {direction, {:edge, label, dest}}), do: {{label, hop_direction(direction), dest}, AshNeo4j.resource_for_label(dest)} # Any other hop shape — incl. a relationship-name hop once the resource chain # has broken (`current` is nil) — is unresolvable. defp resolve_hop(_resource, hop), do: {:error, {:unresolved_hop, hop}} defp relationship_destination(resource, rel_name) do case Ash.Resource.Info.relationship(resource, rel_name) do %{destination: destination} -> destination _ -> nil end end defp hop_direction(:forward), do: :outgoing defp hop_direction(:reverse), do: :incoming defp flip_direction(:outgoing), do: :incoming defp flip_direction(:incoming), do: :outgoing defp flip_direction(other), do: other defp build_relationship_or_property_query(%ResourceMapping{} = mapping, predicates) do relationship_predicates = Enum.filter(predicates, fn predicate -> if Map.has_key?(predicate, :operator) and ref_or_atom?(predicate.left) do prop = property_name(mapping, predicate.left) predicate.operator in [:==, :in] and ResourceInfo.relationship(mapping.module, prop) != nil else false end end) property_predicates = predicates -- relationship_predicates cond do Enum.empty?(relationship_predicates) -> with {:ok, conditions} <- to_conditions(mapping, property_predicates) do Query.node_read_filtered(mapping.label_pair, conditions) end length(relationship_predicates) == 1 -> predicate = hd(relationship_predicates) prop = property_name(mapping, predicate.left) relationship_name = elem(ResourceInfo.relationship(mapping.module, prop), 1) relationship = Ash.Resource.Info.relationship(mapping.module, relationship_name) edge = Enum.find(mapping.edges, &(&1.relationship == relationship_name)) dest_label = ResourceInfo.label(relationship.destination) dest_property = ResourceInfo.convert_to_property_name(relationship.destination, relationship.destination_attribute) Query.relationship_read( mapping.label_pair, edge.label, edge.direction, dest_label, dest_property, predicate.operator, to_param_value(predicate.right) ) true -> Logger.debug("AshNeo4j.QueryHelper: combination of predicates #{inspect(predicates)} not supported") Query.node_read(mapping.label_pair) end end @doc """ Renders an update's `changeset.filter` guard (the "only-update-if" predicate, #361) to `{:ok, conditions}` for the update `WHERE`, or `{:error, %AshNeo4j.Error.UnsupportedUpdateFilter{}}` when it can't be rendered **in full**. Stance: never under-guard. The guard is only pushed down when the filter is a conjunction (`and`-only — no `or`/`not`) of supported attribute predicates that each render; a calculation ref or any predicate the pushdown drops makes the whole guard unsupported, so the data layer returns rather than applies the update unguarded. `nil` (no guard) is `{:ok, []}`. """ @spec guard_conditions(ResourceMapping.t(), Ash.Filter.t() | nil) :: {:ok, list()} | {:error, struct()} def guard_conditions(%ResourceMapping{}, nil), do: {:ok, []} def guard_conditions(%ResourceMapping{module: module} = mapping, filter) do predicates = filter |> Ash.Filter.to_simple_filter(skip_invalid?: true) |> Map.get(:predicates, []) if simple_conjunction?(filter) and predicates != [] and not Enum.any?(predicates, &calculation_ref?/1) do case to_conditions(mapping, predicates) do # Every predicate rendered — a `nil` (unhandled) drop shortens the list. {:ok, conditions} when length(conditions) == length(predicates) -> {:ok, conditions} {:ok, _partial} -> {:error, unsupported_changeset_filter(module, filter)} {:error, _} = error -> error end else {:error, unsupported_changeset_filter(module, filter)} end end defp unsupported_changeset_filter(module, filter) do AshNeo4j.Error.UnsupportedChangesetFilter.exception(resource: module, filter: filter) end @doc """ Renders `changeset.atomics` (a keyword of `{field, Ash.Expr}`, #361) into `{:ok, {set_expressions, params}}` for the update `SET`, where each entry is `"n. = "` evaluated against the **live** node — or `{:error, %AshNeo4j.Error.UnsupportedAtomic{}}` when an expression can't be rendered. Renders: arithmetic (`+ - * /`), comparisons, `if/3` (⇒ `CASE`), string concat (`<>` ⇒ `+`) and `string_trim` (⇒ `trim`, Ash's string cast), attribute refs (`n.`) and scalar literals — numbers/strings/booleans/nil, and atoms (e.g. `Ash.Type.Atom` enum values) bound as their stored string form. An expression node it doesn't cover (e.g. another Ash function) is refused rather than mis-written (stance a). Param keys are `am_`-prefixed and threaded across all atomics so they stay unique. """ @spec render_atomic_sets(module(), ResourceMapping.t(), keyword()) :: {:ok, {[binary()], map()}} | {:error, struct()} def render_atomic_sets(_resource, %ResourceMapping{}, []), do: {:ok, {[], %{}}} def render_atomic_sets(resource, %ResourceMapping{module: module} = mapping, atomics) do Enum.reduce_while(atomics, {:ok, {[], %{}}}, fn {field, expr}, {:ok, {sets, params}} -> with {:ok, hydrated} <- Ash.Filter.hydrate_refs(expr, %{resource: resource, public?: false}), {:ok, cypher, params} <- render_atomic_node(mapping, hydrated, params) do {:cont, {:ok, {sets ++ ["n.#{property_name(mapping, field)} = #{cypher}"], params}}} else _ -> {:halt, {:error, AshNeo4j.Error.UnsupportedAtomic.exception(resource: module, expression: expr)}} end end) end # `+ - * /` (parenthesised) and comparisons, dispatched on the `:operator` atom. @atomic_binary_ops %{ +: "+", -: "-", *: "*", /: "/", >: ">", >=: ">=", <: "<", <=: "<=", ==: "=", !=: "<>" } defp render_atomic_node(mapping, %Ash.Query.Ref{} = ref, params), do: {:ok, "n.#{property_name(mapping, ref)}", params} defp render_atomic_node(mapping, %Ash.Query.Function.If{arguments: [condition, then | rest]}, params) do with {:ok, c, params} <- render_atomic_node(mapping, condition, params), {:ok, t, params} <- render_atomic_node(mapping, then, params), {:ok, e, params} <- render_atomic_else(mapping, rest, params) do {:ok, "CASE WHEN #{c} THEN #{t} ELSE #{e} END", params} end end # String atomics: Ash casts them as `if trim(expr) == "" then null else trim(expr)` # (empty-string ⇒ nil). `string_trim/1` ⇒ Cypher `trim/1`, and string `<>` ⇒ `+`. defp render_atomic_node(mapping, %Ash.Query.Function.StringTrim{arguments: [inner]}, params) do with {:ok, s, params} <- render_atomic_node(mapping, inner, params) do {:ok, "trim(#{s})", params} end end defp render_atomic_node(mapping, %Ash.Query.Operator.Basic.Concat{left: left, right: right}, params) do with {:ok, l, params} <- render_atomic_node(mapping, left, params), {:ok, r, params} <- render_atomic_node(mapping, right, params) do {:ok, "(#{l} + #{r})", params} end end defp render_atomic_node(mapping, %{operator: op, left: left, right: right}, params) when is_map_key(@atomic_binary_ops, op) do with {:ok, l, params} <- render_atomic_node(mapping, left, params), {:ok, r, params} <- render_atomic_node(mapping, right, params) do {:ok, "(#{l} #{@atomic_binary_ops[op]} #{r})", params} end end # Atom literal (e.g. an `Ash.Type.Atom` enum value) — stored as its string form, # matching how the data layer writes atom attributes (Neo4j has no atom type). defp render_atomic_node(_mapping, atom, params) when is_atom(atom) and atom not in [nil, true, false] do key = "am_#{map_size(params)}" {:ok, "$#{key}", Map.put(params, key, Atom.to_string(atom))} end defp render_atomic_node(_mapping, literal, params) when is_number(literal) or is_binary(literal) or is_boolean(literal) or is_nil(literal) do key = "am_#{map_size(params)}" {:ok, "$#{key}", Map.put(params, key, literal)} end defp render_atomic_node(_mapping, other, _params), do: {:error, other} defp render_atomic_else(_mapping, [], params), do: {:ok, "null", params} defp render_atomic_else(mapping, [else_], params), do: render_atomic_node(mapping, else_, params) # True when the filter is an `and`-only tree of leaf predicates — no `or`/`not`, # so `to_simple_filter`'s flattened predicate list represents it faithfully. defp simple_conjunction?(%Ash.Filter{expression: expression}), do: simple_conjunction?(expression) defp simple_conjunction?(nil), do: true defp simple_conjunction?(%Ash.Query.BooleanExpression{op: :and, left: left, right: right}), do: simple_conjunction?(left) and simple_conjunction?(right) defp simple_conjunction?(%Ash.Query.BooleanExpression{op: :or}), do: false defp simple_conjunction?(%Ash.Query.Not{}), do: false defp simple_conjunction?(_leaf), do: true defp calculation_ref?(predicate) do match?(%Ash.Query.Ref{attribute: %Ash.Query.Calculation{}}, Map.get(predicate, :left)) end defp to_conditions(%ResourceMapping{} = mapping, predicates) do predicates |> Enum.map(fn # st_distance(attr, ^p) ^n and the st_distance_in_meters alias — # nested function in comparison, pushed down as point.distance comparison # ONLY when attr is a Point attribute (Neo4j point.distance is point-to-point). # Point's primary stored value is at ".point" (the symmetric split # introduced in #274 — see Type.Point.primary_suffix/0); the pushdown # references that suffixed property, not the bare attribute name. %{operator: op, left: %AshNeo4j.Functions.StDistance{arguments: [ref, test_point]}, right: threshold} when op in [:<, :<=, :>, :>=, :==, :!=] and is_number(threshold) -> if point_attribute?(mapping, ref) do geo_condition(mapping, ref, test_point, fn param -> {point_property(mapping, ref), :st_distance, {op, param, threshold}, false} end) end %{operator: op, left: %AshNeo4j.Functions.StDistanceInMeters{arguments: [ref, test_point]}, right: threshold} when op in [:<, :<=, :>, :>=, :==, :!=] and is_number(threshold) -> if point_attribute?(mapping, ref) do geo_condition(mapping, ref, test_point, fn param -> {point_property(mapping, ref), :st_distance, {op, param, threshold}, false} end) end %{operator: op, left: left} = predicate when is_struct(left, Ash.Query.Ref) or is_atom(left) -> prop = property_name(mapping, predicate.left) val = if op == :is_nil, do: predicate.right, else: to_param_value(predicate.right) ci? = case_insensitive?(mapping, predicate.left, predicate.right) {prop, op, val, ci?} %{name: :contains} = predicate -> argument = hd(predicate.arguments) value = hd(tl(predicate.arguments)) prop = property_name(mapping, argument) ci? = case_insensitive?(mapping, argument, value) {prop, :contains, to_param_value(value), ci?} %{name: :st_contains, arguments: [ref, value]} -> if polygon_attribute?(mapping, ref) do prop = property_name(mapping, ref) case value do %Geo.Point{} = point -> {prop, :st_contains, to_param_value(point), false} %Geo.Polygon{} = poly -> {prop, :st_contains_box, polygon_bbox_corners(poly), false} _other -> # other forms — skip pushdown, let in-memory eval handle it nil end end %{name: :st_dwithin, arguments: [ref, test_point, threshold]} when is_number(threshold) -> if point_attribute?(mapping, ref) do geo_condition(mapping, ref, test_point, fn param -> {point_property(mapping, ref), :st_dwithin, {param, threshold}, false} end) end %{operator: op, left: %AshNeo4j.Functions.VectorSimilarity{arguments: [ref, query_vec]}, right: threshold} when op in [:<, :<=, :>, :>=, :==, :!=] and is_number(threshold) -> with :ok <- AshNeo4j.Cypher.require_cypher25() do {property_name(mapping, ref), :vector_similarity, {op, to_vector_param(query_vec), threshold}, false} end %{operator: op, left: %AshNeo4j.Functions.VectorCosineDistance{arguments: [ref, query_vec]}, right: threshold} when op in [:<, :<=, :>, :>=, :==, :!=] and is_number(threshold) -> with :ok <- AshNeo4j.Cypher.require_cypher25() do {property_name(mapping, ref), :vector_cosine_distance, {op, to_vector_param(query_vec), threshold}, false} end predicate -> Logger.debug("AshNeo4j.QueryHelper: predicate #{inspect(predicate)} not handled") nil end) |> finalize_conditions() end # The condition list, or the first `{:error, _}` a builder produced (e.g. a # geo dimension mismatch) — so the data layer returns it rather than running a # query missing that predicate (#350). defp finalize_conditions(mapped) do case Enum.find(mapped, &match?({:error, _}, &1)) do {:error, _} = error -> error nil -> {:ok, Enum.reject(mapped, &is_nil/1)} end end # Builds a spatial condition from a dimension-checked geo param, or returns the # `{:error, %GeoDimensionMismatch{}}` to thread up. defp geo_condition(mapping, ref, value, builder) do case to_geo_param(mapping, ref, value) do {:error, _} = error -> error {:ok, param} -> builder.(param) end end # True when the referenced attribute is a Point-shaped Geo attribute — # 2D (`geo_types` includes `:point`) or 3D (`:point_z`, #270). Spatial # predicates push down to Neo4j's native `point.distance` / `point.withinBBox` # only for Point attributes; other geometries get bbox-prefilter pushdown via # their own companions. defp point_attribute?(mapping, ref_or_atom) do geo_attribute_with_type?(mapping, ref_or_atom, :point) or geo_attribute_with_type?(mapping, ref_or_atom, :point_z) end # Converts a spatial test geometry to a Bolty param, guarding that its # coordinate dimension matches the attribute's. Neo4j returns `null` (then # silently drops rows) for a 2D/3D mix, so a mismatch raises # `AshNeo4j.Error.GeoDimensionMismatch` here instead (#270). Bridge worlds # explicitly with `AshNeo4j.Geo.force_2d/1`. defp to_geo_param(mapping, ref, value) do types = geo_types_of(mapping, ref) vd = value_dim(value) cond do vd == 3 and :point in types and :point_z not in types -> {:error, AshNeo4j.Error.GeoDimensionMismatch.exception(attr_dim: 2, value_dim: 3)} vd == 2 and :point_z in types and :point not in types -> {:error, AshNeo4j.Error.GeoDimensionMismatch.exception(attr_dim: 3, value_dim: 2)} true -> {:ok, to_param_value(value)} end end defp geo_types_of(%ResourceMapping{module: module}, ref) do name = attribute_name(ref) case name && Ash.Resource.Info.attribute(module, name) do %{constraints: constraints} -> case Keyword.get(constraints, :geo_types) do types when is_list(types) -> types type when is_atom(type) -> [type] _ -> [] end _ -> [] end end defp value_dim(%Geo.PointZ{}), do: 3 defp value_dim(%Geo.Point{}), do: 2 defp value_dim(%{coordinates: coords}), do: geo_coord_dim(coords) defp value_dim(_), do: 2 defp geo_coord_dim(t) when is_tuple(t), do: tuple_size(t) defp geo_coord_dim([head | _]), do: geo_coord_dim(head) defp geo_coord_dim(_), do: 2 # True when the referenced attribute is a Polygon-shaped Geo attribute. # st_contains pushdown uses the polygon's bbox companions (`.bbSW`/ # `.bbNE`) for indexed prefilter via `point.withinBBox`. defp polygon_attribute?(mapping, ref_or_atom), do: geo_attribute_with_type?(mapping, ref_or_atom, :polygon) defp geo_attribute_with_type?(%ResourceMapping{module: module}, ref_or_atom, geo_type) do name = attribute_name(ref_or_atom) if name do case Ash.Resource.Info.attribute(module, name) do %{constraints: constraints} -> case Keyword.get(constraints, :geo_types) do types when is_list(types) -> geo_type in types ^geo_type -> true _ -> false end _ -> false end else false end end # Derives {sw, ne} Bolty Points from a Geo.Polygon's exterior ring, # for use in `:st_contains_box` cypher pushdown (two ANDed # `point.withinBBox` calls — sufficient for testing whether the # polygon's bbox fits inside the attribute's bbox). defp polygon_bbox_corners(%Geo.Polygon{coordinates: [exterior | _]}) do xs = Enum.map(exterior, &elem(&1, 0)) ys = Enum.map(exterior, &elem(&1, 1)) sw = Bolty.Types.Point.create(:wgs_84, Enum.min(xs), Enum.min(ys)) ne = Bolty.Types.Point.create(:wgs_84, Enum.max(xs), Enum.max(ys)) {sw, ne} end defp attribute_name(%Ash.Query.Ref{} = ref), do: Ash.Query.Ref.name(ref) defp attribute_name(atom) when is_atom(atom), do: atom defp attribute_name(_), do: nil # Builds the ORDER BY terms and any params they reference. # # Returns `{terms, params}` where each term is `{order_expression, :asc | :desc}` # with `order_expression` a fully-formed Cypher expression (including the `s.` # prefix). Plain properties render to `"s."`; a `calc(vector_similarity(…))` # or `calc(vector_cosine_distance(…))` sort renders to the corresponding scalar # expression with the query embedding bound as a param. Other calculations are # dropped (Ash evaluates them in-memory). defp sort_terms(ash_query, %ResourceMapping{} = mapping) do case ash_query.sort do sort when sort in [nil, []] -> {:ok, {[], %{}}} sort -> sort |> Enum.with_index() |> Enum.reduce_while({:ok, {[], %{}}}, fn {{name, order}, index}, {:ok, {terms, params}} -> case sort_term(mapping, name, order, index) do nil -> {:cont, {:ok, {terms, params}}} {:error, _} = error -> {:halt, error} {term, term_params} -> {:cont, {:ok, {terms ++ [term], Map.merge(params, term_params)}}} end end) end end # vector_similarity / vector_cosine_distance sort — pushed down as the scalar # expression, with the query embedding bound as a param. defp sort_term(mapping, %Ash.Query.Calculation{module: Ash.Resource.Calculation.Expression, opts: opts}, order, index) do case Keyword.get(opts, :expr) do %Ash.Query.Call{name: fname, args: [ref, query_vec]} when fname in [:vector_similarity, :vector_cosine_distance] -> with :ok <- AshNeo4j.Cypher.require_cypher25() do prop = property_name(mapping, ref) key = "sort_#{Cypher.sanitize_param(prop)}_#{index}_vec" expr = Cypher.vector_scalar(fname, :s, prop, "$#{key}") {{expr, order}, %{key => to_vector_param(query_vec)}} end _ -> nil end end # Any other calculation — Ash evaluates it in-memory, so drop from pushdown. defp sort_term(_mapping, %Ash.Query.Calculation{}, _order, _index), do: nil # Plain property sort. defp sort_term(mapping, name, order, _index) do prop = Keyword.get(mapping.properties, name, name) {{"s.#{prop}", order}, %{}} end defp ref_or_atom?(%Ash.Query.Ref{}), do: true defp ref_or_atom?(value) when is_atom(value), do: true defp ref_or_atom?(_), do: false defp property_name(%ResourceMapping{} = mapping, ref_or_atom) do attr_name = case ref_or_atom do %Ash.Query.Ref{} -> Ash.Query.Ref.name(ref_or_atom) atom when is_atom(atom) -> atom end Keyword.get(mapping.properties, attr_name, attr_name) |> to_string() end defp to_param_value(%Ash.CiString{} = v), do: Ash.CiString.value(v) defp to_param_value(%MapSet{} = ms), do: MapSet.to_list(ms) defp to_param_value(%Geo.Point{coordinates: {x, y}}), do: Bolty.Types.Point.create(:wgs_84, x, y) defp to_param_value(%Geo.PointZ{coordinates: {x, y, z}}), do: Bolty.Types.Point.create(:wgs_84, x, y, z) defp to_param_value(value), do: value # The query embedding is passed as a plain LIST param — what # `vector.similarity.cosine/2` expects, and consistent with list storage. defp to_vector_param(value) when is_list(value), do: Enum.map(value, &(&1 / 1)) defp to_vector_param(%Bolty.Types.Vector{data: data}), do: Enum.map(data, &(&1 / 1)) # Builds the on-disk property name for a Point attribute under the symmetric # split — `.point` is where the native Neo4j POINT lives (the indexable # primary), so spatial pushdown predicates reference that suffixed name. defp point_property(%ResourceMapping{} = mapping, ref) do "#{property_name(mapping, ref)}.point" end defp case_insensitive?(%ResourceMapping{} = mapping, predicate_left, predicate_right) do ResourceInfo.attribute_type(mapping.module, predicate_left) in [Ash.Type.CiString, :ci_string] or match?(%Ash.CiString{}, predicate_right) end end