-module(kura_ets_query).
-moduledoc """
Plan interpreter for the ETS backend. Executes the `{kura_ets, Plan}`
terms emitted by `kura_ets_dialect` against the pool's ETS tables and
returns pgo-shaped results (`#{command, rows, num_rows}`).

## Storage model

One `set` table per source table. Objects are `{Key, Row}` where `Row`
is a map of dumped field values (as produced by kura's type system on
the write path) and `Key` is the primary-key value — a bare value for
single-column keys, a tuple for composite keys, or a unique reference
when no primary key can be derived from the schema.

Missing columns are materialized as `null` at insert (with schema
defaults applied first), mirroring what a SQL backend would return,
so `is_nil` conditions and `kura_db:load_row/2` behave identically.

## Filtering

Queries are evaluated with a plain scan-and-filter over the table
(`ets:tab2list/1`), not with match specifications: kura's where
grammar (like/ilike, composite in, between, nested and/or/not) maps
poorly onto match-spec guards, and for the in-memory datasets this
backend targets a scan is fast enough and far simpler to verify.
One exception: when the top-level wheres pin every primary-key column
with an equality, the row is fetched with `ets:lookup/2` (the storage
key is the primary key), making point reads O(1) — the cache-aside
hot path.
Comparison values are passed through `kura_types:dump/2` when the
schema declares a type for the field, so they compare against the
stored (dumped) representation the same way SQL params are encoded by
a database client.

SQL null semantics are approximated: a comparison against a null
(`undefined`/`null`) field value is false, except for the explicit
`is_nil` / `is_not_nil` conditions.

Unsupported constructs (joins, group_by, having, CTEs, combinations,
subqueries, fragments, `matches`, locks, prefix multitenancy) yield
`{error, {kura_ets_unsupported, What}}`.
""".

-include_lib("kura/include/kura.hrl").

-export([run/2]).

-define(AGGS, [count, sum, avg, min, max]).

-spec run(atom(), kura_ets_dialect:plan()) -> dynamic().
run(Pool, Plan) ->
    try
        do_run(Pool, Plan)
    catch
        throw:{kura_ets_unsupported, _} = Err -> {error, Err}
    end.

%%----------------------------------------------------------------------
%% SELECT
%%----------------------------------------------------------------------

do_run(Pool, {select, Q = #kura_query{from = From}}) ->
    ok = assert_supported(Q),
    with_table(Pool, From, fun(Tid) ->
        Types = types_of(From),
        Matched =
            case pk_lookup(Tid, From, Q#kura_query.wheres, Types) of
                {hit, PkRows} -> PkRows;
                scan -> filter_rows(Tid, Q#kura_query.wheres, Types)
            end,
        Rows =
            case aggregate_select(Q#kura_query.select) of
                true ->
                    aggregate(Q#kura_query.select, Matched);
                false ->
                    Sorted = order(Matched, Q#kura_query.order_bys),
                    Deduped = distinct_on(Sorted, Q#kura_query.distinct),
                    Projected = project(Deduped, Q#kura_query.select),
                    Distinct = distinct_all(Projected, Q#kura_query.distinct),
                    limit(offset(Distinct, Q#kura_query.offset), Q#kura_query.limit)
            end,
        #{command => select, rows => Rows, num_rows => length(Rows)}
    end);
%%----------------------------------------------------------------------
%% INSERT
%%----------------------------------------------------------------------

do_run(Pool, {insert, Schema, Data, Opts}) ->
    with_table(Pool, Schema, fun(Tid) ->
        Row = fill_defaults(Schema, Data),
        Key = row_key(Schema, Row),
        insert_one(Tid, Key, Row, maps:get(on_conflict, Opts, undefined), Schema)
    end);
do_run(Pool, {insert_all, Schema, Rows0, Opts}) ->
    with_table(Pool, Schema, fun(Tid) ->
        Rows = [fill_defaults(Schema, R) || R <- Rows0],
        Objects = [{row_key(Schema, R), R} || R <- Rows],
        case insert_many(Tid, Objects, maps:get(on_conflict, Opts, undefined), Schema) of
            {ok, Inserted} ->
                Base = #{command => insert, num_rows => length(Inserted)},
                case maps:get(returning, Opts, false) of
                    false ->
                        Base#{rows => []};
                    true ->
                        Base#{rows => Inserted};
                    Fields when is_list(Fields) ->
                        Base#{rows => [maps:with(Fields, R) || R <- Inserted]}
                end;
            {error, _} = Err ->
                Err
        end
    end);
%%----------------------------------------------------------------------
%% UPDATE / DELETE by primary key
%%----------------------------------------------------------------------

do_run(Pool, {update, Schema, Changes, KeyClauses}) ->
    with_table(Pool, Schema, fun(Tid) ->
        Types = types_of(Schema),
        case find_by_clauses(Tid, KeyClauses, Types) of
            [] ->
                #{command => update, rows => [], num_rows => 0};
            [{Key, Old} | _] ->
                New = maps:merge(Old, Changes),
                case replace_row(Tid, Key, Schema, New) of
                    ok -> #{command => update, rows => [New], num_rows => 1};
                    {error, _} = Err -> Err
                end
        end
    end);
do_run(Pool, {delete, Schema, KeyClauses}) ->
    with_table(Pool, Schema, fun(Tid) ->
        Types = types_of(Schema),
        case find_by_clauses(Tid, KeyClauses, Types) of
            [] ->
                #{command => delete, rows => [], num_rows => 0};
            [{Key, Old} | _] ->
                true = ets:delete(Tid, Key),
                #{command => delete, rows => [Old], num_rows => 1}
        end
    end);
%%----------------------------------------------------------------------
%% Bulk UPDATE / DELETE
%%----------------------------------------------------------------------

do_run(Pool, {update_all, Q = #kura_query{from = From}, SetMap}) ->
    ok = assert_supported(Q),
    with_table(Pool, From, fun(Tid) ->
        Types = types_of(From),
        Changes = dump_values(SetMap, Types),
        Matched = filter_objects(Tid, Q#kura_query.wheres, Types),
        Result = lists:foldl(
            fun
                ({Key, Old}, N) when is_integer(N) ->
                    case replace_row(Tid, Key, From, maps:merge(Old, Changes)) of
                        ok -> N + 1;
                        {error, _} = Err -> Err
                    end;
                (_, Err) ->
                    Err
            end,
            0,
            Matched
        ),
        case Result of
            N when is_integer(N) -> #{command => update, rows => [], num_rows => N};
            {error, _} = Err -> Err
        end
    end);
do_run(Pool, {delete_all, Q = #kura_query{from = From}}) ->
    ok = assert_supported(Q),
    with_table(Pool, From, fun(Tid) ->
        Types = types_of(From),
        Matched = filter_objects(Tid, Q#kura_query.wheres, Types),
        lists:foreach(fun({Key, _}) -> true = ets:delete(Tid, Key) end, Matched),
        #{command => delete, rows => [], num_rows => length(Matched)}
    end).

%%----------------------------------------------------------------------
%% Insert internals
%%----------------------------------------------------------------------

insert_one(Tid, Key, Row, undefined, Schema) ->
    case ets:insert_new(Tid, {Key, Row}) of
        true -> #{command => insert, rows => [Row], num_rows => 1};
        false -> unique_violation(Schema)
    end;
insert_one(Tid, Key, Row, {_Target, nothing}, _Schema) ->
    case ets:insert_new(Tid, {Key, Row}) of
        true -> #{command => insert, rows => [Row], num_rows => 1};
        false -> #{command => insert, rows => [], num_rows => 0}
    end;
insert_one(Tid, Key, Row, {_Target, replace_all}, _Schema) ->
    New = upsert(Tid, Key, Row, fun(Old) -> maps:merge(Old, Row) end),
    #{command => insert, rows => [New], num_rows => 1};
insert_one(Tid, Key, Row, {_Target, {replace, Fields}}, _Schema) ->
    New = upsert(Tid, Key, Row, fun(Old) -> maps:merge(Old, maps:with(Fields, Row)) end),
    #{command => insert, rows => [New], num_rows => 1};
insert_one(_Tid, _Key, _Row, Other, _Schema) ->
    throw({kura_ets_unsupported, {on_conflict, Other}}).

%% The conflict target is not checked against the schema: ETS enforces
%% uniqueness only on the primary key, so every conflict clause is
%% interpreted as targeting the primary key.
upsert(Tid, Key, Row, MergeFun) ->
    New =
        case ets:lookup(Tid, Key) of
            [] -> Row;
            [{_, Old}] -> MergeFun(Old)
        end,
    true = ets:insert(Tid, {Key, New}),
    New.

insert_many(Tid, Objects, undefined, Schema) ->
    case ets:insert_new(Tid, Objects) of
        true -> {ok, [Row || {_, Row} <- Objects]};
        false -> unique_violation(Schema)
    end;
insert_many(Tid, Objects, {_Target, nothing}, _Schema) ->
    Inserted = [Row || {Key, Row} <- Objects, ets:insert_new(Tid, {Key, Row})],
    {ok, Inserted};
insert_many(Tid, Objects, {_Target, replace_all}, _Schema) ->
    {ok, [upsert(Tid, Key, Row, fun(Old) -> maps:merge(Old, Row) end) || {Key, Row} <- Objects]};
insert_many(Tid, Objects, {_Target, {replace, Fields}}, _Schema) ->
    {ok, [
        upsert(Tid, Key, Row, fun(Old) -> maps:merge(Old, maps:with(Fields, Row)) end)
     || {Key, Row} <- Objects
    ]};
insert_many(_Tid, _Objects, Other, _Schema) ->
    throw({kura_ets_unsupported, {on_conflict, Other}}).

%% Reinsert a row under its (possibly changed) primary key.
replace_row(Tid, OldKey, Schema, New) ->
    case pk_key(Schema, New) of
        {ok, NewKey} when NewKey =/= OldKey ->
            case ets:member(Tid, NewKey) of
                true ->
                    unique_violation(Schema);
                false ->
                    true = ets:delete(Tid, OldKey),
                    true = ets:insert(Tid, {NewKey, New}),
                    ok
            end;
        _ ->
            true = ets:insert(Tid, {OldKey, New}),
            ok
    end.

unique_violation(Schema) ->
    Table = table_of(Schema),
    {error, #{
        code => ~"23505",
        constraint => <<Table/binary, "_pkey">>,
        message => ~"duplicate key value violates unique constraint"
    }}.

%%----------------------------------------------------------------------
%% Row keys and defaults
%%----------------------------------------------------------------------

row_key(Schema, Row) ->
    case pk_key(Schema, Row) of
        {ok, Key} -> Key;
        error -> erlang:make_ref()
    end.

pk_key(Schema, Row) ->
    try kura_schema:key(Schema) of
        [Field] -> {ok, maps:get(Field, Row, null)};
        Fields -> {ok, list_to_tuple([maps:get(F, Row, null) || F <- Fields])}
    catch
        _:_ -> error
    end.

%% Apply schema defaults and materialize missing columns as null, like
%% a SQL insert would.
fill_defaults(Schema, Data) ->
    try Schema:fields() of
        Fields ->
            lists:foldl(
                fun
                    (#kura_field{virtual = true}, Acc) ->
                        Acc;
                    (#kura_field{name = N, type = T, default = D}, Acc) ->
                        case maps:is_key(N, Acc) of
                            true -> Acc;
                            false -> Acc#{N => dump_default(T, D)}
                        end
                end,
                Data,
                Fields
            )
    catch
        _:_ -> Data
    end.

dump_default(_Type, undefined) ->
    null;
dump_default(Type, Default) ->
    case kura_types:dump(Type, Default) of
        {ok, V} -> V;
        {error, _} -> Default
    end.

%%----------------------------------------------------------------------
%% Table / schema resolution
%%----------------------------------------------------------------------

with_table(Pool, SchemaOrTable, Fun) ->
    case kura_ets_pool:ensure_table(Pool, table_of(SchemaOrTable)) of
        {ok, Tid} -> Fun(Tid);
        {error, _} = Err -> Err
    end.

table_of(Mod) when is_atom(Mod) ->
    case code:ensure_loaded(Mod) of
        {module, Mod} ->
            case erlang:function_exported(Mod, table, 0) of
                true -> Mod:table();
                false -> atom_to_binary(Mod, utf8)
            end;
        _ ->
            atom_to_binary(Mod, utf8)
    end.

types_of(SchemaOrTable) ->
    try
        kura_schema:field_types(SchemaOrTable)
    catch
        _:_ -> #{}
    end.

assert_supported(#kura_query{joins = [_ | _]}) ->
    throw({kura_ets_unsupported, joins});
assert_supported(#kura_query{group_bys = [_ | _]}) ->
    throw({kura_ets_unsupported, group_by});
assert_supported(#kura_query{havings = [_ | _]}) ->
    throw({kura_ets_unsupported, having});
assert_supported(#kura_query{ctes = [_ | _]}) ->
    throw({kura_ets_unsupported, ctes});
assert_supported(#kura_query{combinations = [_ | _]}) ->
    throw({kura_ets_unsupported, combinations});
assert_supported(#kura_query{lock = Lock}) when Lock =/= undefined ->
    throw({kura_ets_unsupported, lock});
assert_supported(#kura_query{prefix = Prefix}) when Prefix =/= undefined ->
    throw({kura_ets_unsupported, prefix});
assert_supported(#kura_query{}) ->
    ok.

%%----------------------------------------------------------------------
%% Filtering
%%----------------------------------------------------------------------

filter_rows(Tid, Wheres, Types) ->
    [Row || {_, Row} <- filter_objects(Tid, Wheres, Types)].

%% Point-lookup fast path: when the top-level where conditions contain
%% an equality for every primary-key column, the storage key is fully
%% determined and the row can be fetched with ets:lookup instead of a
%% table scan. The full condition list is still evaluated against the
%% fetched row, so extra conditions (soft-delete filters, tenant
%% attributes, conflicting duplicates) keep their exact semantics.
pk_lookup(Tid, Schema, Wheres, Types) ->
    try kura_schema:key(Schema) of
        KeyFields ->
            case eq_values(Wheres, KeyFields, Types) of
                {ok, [Single]} ->
                    pk_fetch(Tid, Single, Wheres, Types);
                {ok, Composite} ->
                    pk_fetch(Tid, list_to_tuple(Composite), Wheres, Types);
                error ->
                    scan
            end
    catch
        _:_ -> scan
    end.

pk_fetch(Tid, Key, Wheres, Types) ->
    {hit, [Row || {_, Row} <- ets:lookup(Tid, Key), matches(Wheres, Row, Types)]}.

%% Extract one (dumped) equality value per key column from the
%% top-level (implicitly ANDed) conditions. Any key column without an
%% equality means the key is underdetermined -> scan.
eq_values(Wheres, KeyFields, Types) ->
    Eqs = lists:foldl(
        fun
            ({F, '=', V}, Acc) when is_atom(F) ->
                Acc#{F => dump_value(F, V, Types)};
            ({F, V}, Acc) when
                is_atom(F),
                F =/= 'and',
                F =/= 'or',
                F =/= 'not',
                V =/= is_nil,
                V =/= is_not_nil
            ->
                Acc#{F => dump_value(F, V, Types)};
            (_, Acc) ->
                Acc
        end,
        #{},
        Wheres
    ),
    case lists:all(fun(F) -> maps:is_key(F, Eqs) end, KeyFields) of
        true -> {ok, [maps:get(F, Eqs) || F <- KeyFields]};
        false -> error
    end.

filter_objects(Tid, Wheres, Types) ->
    [Obj || {_, Row} = Obj <- ets:tab2list(Tid), matches(Wheres, Row, Types)].

find_by_clauses(Tid, KeyClauses, Types) ->
    filter_objects(Tid, KeyClauses, Types).

matches(Conditions, Row, Types) ->
    lists:all(fun(C) -> eval(C, Row, Types) end, Conditions).

eval({'and', Conditions}, Row, Types) ->
    lists:all(fun(C) -> eval(C, Row, Types) end, Conditions);
eval({'or', Conditions}, Row, Types) ->
    lists:any(fun(C) -> eval(C, Row, Types) end, Conditions);
eval({'not', Condition}, Row, Types) ->
    not eval(Condition, Row, Types);
eval({Field, is_nil}, Row, _Types) when is_atom(Field) ->
    is_null(field(Field, Row));
eval({Field, is_not_nil}, Row, _Types) when is_atom(Field) ->
    not is_null(field(Field, Row));
eval({Field, '=', Value}, Row, Types) when is_atom(Field) ->
    eval({Field, Value}, Row, Types);
eval({Field, '!=', Value}, Row, Types) when is_atom(Field) ->
    non_null_cmp(field(Field, Row), dump_value(Field, Value, Types), fun(A, B) -> A /= B end);
eval({Field, '<', Value}, Row, Types) when is_atom(Field) ->
    non_null_cmp(field(Field, Row), dump_value(Field, Value, Types), fun(A, B) -> A < B end);
eval({Field, '>', Value}, Row, Types) when is_atom(Field) ->
    non_null_cmp(field(Field, Row), dump_value(Field, Value, Types), fun(A, B) -> A > B end);
eval({Field, '<=', Value}, Row, Types) when is_atom(Field) ->
    non_null_cmp(field(Field, Row), dump_value(Field, Value, Types), fun(A, B) -> A =< B end);
eval({Field, '>=', Value}, Row, Types) when is_atom(Field) ->
    non_null_cmp(field(Field, Row), dump_value(Field, Value, Types), fun(A, B) -> A >= B end);
eval({Field, like, Pattern}, Row, _Types) when is_atom(Field) ->
    like(field(Field, Row), Pattern, []);
eval({Field, ilike, Pattern}, Row, _Types) when is_atom(Field) ->
    like(field(Field, Row), Pattern, [caseless]);
eval({Field, between, {Low, High}}, Row, Types) when is_atom(Field) ->
    V = field(Field, Row),
    L = dump_value(Field, Low, Types),
    H = dump_value(Field, High, Types),
    non_null_cmp(V, L, fun(A, B) -> A >= B end) andalso
        non_null_cmp(V, H, fun(A, B) -> A =< B end);
eval({Field, in, Values}, Row, Types) when is_atom(Field), is_list(Values) ->
    V = field(Field, Row),
    lists:any(fun(Val) -> sql_eq(V, dump_value(Field, Val, Types)) end, Values);
eval({Field, not_in, Values}, Row, Types) when is_atom(Field), is_list(Values) ->
    V = field(Field, Row),
    not is_null(V) andalso
        not lists:any(fun(Val) -> sql_eq(V, dump_value(Field, Val, Types)) end, Values);
eval({[Col], in, Tuples}, Row, Types) when is_atom(Col), is_list(Tuples) ->
    eval({Col, in, [element(1, T) || T <- Tuples]}, Row, Types);
eval({Cols, in, Tuples}, Row, Types) when is_list(Cols), is_list(Tuples) ->
    RowVals = [field(C, Row) || C <- Cols],
    lists:any(
        fun(Tuple) ->
            Vals = [dump_value(C, V, Types) || {C, V} <- lists:zip(Cols, tuple_to_list(Tuple))],
            lists:all(fun({A, B}) -> sql_eq(A, B) end, lists:zip(RowVals, Vals))
        end,
        Tuples
    );
eval({Field, Value}, Row, Types) when is_atom(Field) ->
    sql_eq(field(Field, Row), dump_value(Field, Value, Types));
eval(Condition, _Row, _Types) ->
    throw({kura_ets_unsupported, {condition, Condition}}).

field(Field, Row) ->
    maps:get(Field, Row, undefined).

is_null(undefined) -> true;
is_null(null) -> true;
is_null(_) -> false.

%% SQL comparisons against NULL are never true.
sql_eq(A, B) ->
    non_null_cmp(A, B, fun(X, Y) -> X == Y end).

non_null_cmp(A, B, Cmp) ->
    not is_null(A) andalso not is_null(B) andalso Cmp(A, B).

%% Encode a comparison value the way a SQL client would encode a query
%% parameter, so it compares against the stored (dumped) form.
dump_value(Field, Value, Types) ->
    case Types of
        #{Field := Type} ->
            case kura_types:dump(Type, Value) of
                {ok, Dumped} -> Dumped;
                {error, _} -> Value
            end;
        #{} ->
            Value
    end.

dump_values(Map, Types) ->
    maps:map(fun(Field, Value) -> dump_value(Field, Value, Types) end, Map).

%%----------------------------------------------------------------------
%% LIKE / ILIKE
%%----------------------------------------------------------------------

like(Value, Pattern, ReOpts) when is_binary(Value), is_binary(Pattern) ->
    RE = like_to_regex(Pattern),
    case re:run(Value, RE, [{capture, none}, unicode | ReOpts]) of
        match -> true;
        _ -> false
    end;
like(_Value, _Pattern, _ReOpts) ->
    false.

like_to_regex(Pattern) ->
    Inner = <<<<(like_char(C))/binary>> || <<C/utf8>> <= Pattern>>,
    <<"^", Inner/binary, "$">>.

like_char($%) ->
    <<".*">>;
like_char($_) ->
    <<".">>;
like_char(C) ->
    case lists:member(C, ".^$*+?()[]{}|\\") of
        true -> <<$\\, C/utf8>>;
        false -> <<C/utf8>>
    end.

%%----------------------------------------------------------------------
%% Ordering / projection / distinct / paging
%%----------------------------------------------------------------------

order(Rows, []) ->
    Rows;
order(Rows, Orders) ->
    lists:sort(fun(A, B) -> order_cmp(Orders, A, B) end, Rows).

order_cmp([], _A, _B) ->
    true;
order_cmp([{Field, Dir} | Rest], A, B) ->
    Va = field(Field, A),
    Vb = field(Field, B),
    case {Va == Vb, Dir} of
        {true, _} -> order_cmp(Rest, A, B);
        {false, asc} -> Va < Vb;
        {false, desc} -> Va > Vb
    end.

%% DISTINCT ON (fields): keep the first row per field-value group,
%% preserving sort order. Runs before projection so the distinct
%% fields need not be selected.
distinct_on(Rows, Fields) when is_list(Fields) ->
    dedupe(Rows, fun(Row) -> maps:with(Fields, Row) end);
distinct_on(Rows, _) ->
    Rows.

%% Plain DISTINCT: dedupe identical projected rows.
distinct_all(Rows, true) ->
    dedupe(Rows, fun(Row) -> Row end);
distinct_all(Rows, _) ->
    Rows.

dedupe(Rows, KeyFun) ->
    {Deduped, _} = lists:foldl(
        fun(Row, {Acc, Seen}) ->
            Key = KeyFun(Row),
            case maps:is_key(Key, Seen) of
                true -> {Acc, Seen};
                false -> {[Row | Acc], Seen#{Key => true}}
            end
        end,
        {[], #{}},
        Rows
    ),
    lists:reverse(Deduped).

project(Rows, []) ->
    Rows;
project(Rows, {exprs, Exprs}) ->
    Pairs = [expr_pair(E) || E <- Exprs],
    [maps:from_list([{Alias, field(F, Row)} || {Alias, F} <- Pairs]) || Row <- Rows];
project(Rows, Fields) when is_list(Fields) ->
    case lists:all(fun is_atom/1, Fields) of
        true -> [maps:with(Fields, Row) || Row <- Rows];
        false -> throw({kura_ets_unsupported, {select, Fields}})
    end.

expr_pair({Alias, Field}) when is_atom(Alias), is_atom(Field) ->
    {Alias, Field};
expr_pair(Expr) ->
    throw({kura_ets_unsupported, {select_expr, Expr}}).

offset(Rows, undefined) -> Rows;
offset(Rows, N) when N >= length(Rows) -> [];
offset(Rows, N) -> lists:nthtail(N, Rows).

limit(Rows, undefined) -> Rows;
limit(Rows, N) -> lists:sublist(Rows, N).

%%----------------------------------------------------------------------
%% Aggregates
%%----------------------------------------------------------------------

aggregate_select(Select) when is_list(Select), Select =/= [] ->
    case lists:any(fun is_agg/1, Select) of
        true ->
            lists:all(fun is_agg/1, Select) orelse
                throw({kura_ets_unsupported, {select, Select}}),
            true;
        false ->
            false
    end;
aggregate_select(_) ->
    false.

is_agg({Agg, Field}) when is_atom(Field) -> lists:member(Agg, ?AGGS);
is_agg(_) -> false.

aggregate(Select, Rows) ->
    [lists:foldl(fun(Spec, Acc) -> agg(Spec, Rows, Acc) end, #{}, Select)].

agg({count, '*'}, Rows, Acc) ->
    Acc#{count => length(Rows)};
agg({count, Field}, Rows, Acc) ->
    Acc#{count => length(agg_values(Field, Rows))};
agg({sum, Field}, Rows, Acc) ->
    Acc#{sum => sum_or_null(agg_values(Field, Rows))};
agg({avg, Field}, Rows, Acc) ->
    Acc#{
        avg =>
            case agg_values(Field, Rows) of
                [] -> null;
                Vs -> lists:sum(Vs) / length(Vs)
            end
    };
agg({min, Field}, Rows, Acc) ->
    Acc#{min => min_max_or_null(fun lists:min/1, agg_values(Field, Rows))};
agg({max, Field}, Rows, Acc) ->
    Acc#{max => min_max_or_null(fun lists:max/1, agg_values(Field, Rows))}.

agg_values(Field, Rows) ->
    [V || Row <- Rows, V <- [field(Field, Row)], not is_null(V)].

sum_or_null([]) -> null;
sum_or_null(Vs) -> lists:sum(Vs).

min_max_or_null(_Fun, []) -> null;
min_max_or_null(Fun, Vs) -> Fun(Vs).