Skip to content

Commit e50769a

Browse files
isumisum
authored
Extend supported types for first and last aggregations (#6225)
* feat(graph, store): extend supported types for first and last aggregations * fix(store): rename migration and remove generate script
1 parent 4b9a858 commit e50769a

7 files changed

Lines changed: 267 additions & 34 deletions

File tree

‎graph/src/schema/input/mod.rs‎

Lines changed: 88 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -788,7 +788,7 @@ pub struct Aggregate {
788788
}
789789

790790
impl Aggregate {
791-
fn new(_schema: &Schema, name: &str, field_type: &s::Type, dir: &s::Directive) -> Self {
791+
fn new(schema: &Schema, name: &str, field_type: &s::Type, dir: &s::Directive) -> Self {
792792
let func = dir
793793
.argument("fn")
794794
.unwrap()
@@ -818,7 +818,7 @@ impl Aggregate {
818818
arg,
819819
cumulative,
820820
field_type: field_type.clone(),
821-
value_type: field_type.get_base_type().parse().unwrap(),
821+
value_type: Field::scalar_value_type(schema, field_type),
822822
}
823823
}
824824

@@ -2366,27 +2366,63 @@ mod validations {
23662366
}
23672367
}
23682368

2369-
fn aggregate_fields_are_numbers(agg_type: &s::ObjectType, errors: &mut Vec<Err>) {
2369+
fn aggregate_field_types(
2370+
schema: &Schema,
2371+
agg_type: &s::ObjectType,
2372+
errors: &mut Vec<Err>,
2373+
) {
2374+
fn is_first_last(agg_directive: &s::Directive) -> bool {
2375+
match agg_directive.argument(kw::FUNC) {
2376+
Some(s::Value::Enum(func) | s::Value::String(func)) => {
2377+
func == AggregateFn::First.as_str()
2378+
|| func == AggregateFn::Last.as_str()
2379+
}
2380+
_ => false,
2381+
}
2382+
}
2383+
23702384
let errs = agg_type
23712385
.fields
23722386
.iter()
2373-
.filter(|field| field.find_directive(kw::AGGREGATE).is_some())
2374-
.map(|field| match field.field_type.value_type() {
2375-
Ok(vt) => {
2376-
if vt.is_numeric() {
2377-
Ok(())
2378-
} else {
2379-
Err(Err::NonNumericAggregate(
2387+
.filter_map(|field| {
2388+
field
2389+
.find_directive(kw::AGGREGATE)
2390+
.map(|agg_directive| (field, agg_directive))
2391+
})
2392+
.map(|(field, agg_directive)| {
2393+
let is_first_last = is_first_last(agg_directive);
2394+
2395+
match field.field_type.value_type() {
2396+
Ok(value_type) if value_type.is_numeric() => Ok(()),
2397+
Ok(ValueType::Bytes | ValueType::String) if is_first_last => Ok(()),
2398+
Ok(_) if is_first_last => Err(Err::InvalidFirstLastAggregate(
2399+
agg_type.name.clone(),
2400+
field.name.clone(),
2401+
)),
2402+
Ok(_) => Err(Err::NonNumericAggregate(
2403+
agg_type.name.to_owned(),
2404+
field.name.to_owned(),
2405+
)),
2406+
Err(_) => {
2407+
if is_first_last
2408+
&& schema
2409+
.entity_types
2410+
.iter()
2411+
.find(|entity_type| {
2412+
entity_type.name.eq(field.field_type.get_base_type())
2413+
})
2414+
.is_some()
2415+
{
2416+
return Ok(());
2417+
}
2418+
2419+
Err(Err::FieldTypeUnknown(
23802420
agg_type.name.to_owned(),
23812421
field.name.to_owned(),
2422+
field.field_type.get_base_type().to_owned(),
23822423
))
23832424
}
23842425
}
2385-
Err(_) => Err(Err::FieldTypeUnknown(
2386-
agg_type.name.to_owned(),
2387-
field.name.to_owned(),
2388-
field.field_type.get_base_type().to_owned(),
2389-
)),
23902426
})
23912427
.filter_map(|err| err.err());
23922428
errors.extend(errs);
@@ -2519,16 +2555,10 @@ mod validations {
25192555
continue;
25202556
}
25212557
};
2522-
let field_type = match field.field_type.value_type() {
2523-
Ok(field_type) => field_type,
2524-
Err(_) => {
2525-
errors.push(Err::NonNumericAggregate(
2526-
agg_type.name.to_owned(),
2527-
field.name.to_owned(),
2528-
));
2529-
continue;
2530-
}
2531-
};
2558+
2559+
let is_first_last =
2560+
matches!(func, AggregateFn::First | AggregateFn::Last);
2561+
25322562
// It would be nicer to use a proper struct here
25332563
// and have that implement
25342564
// `sqlexpr::ExprVisitor` but we need access to
@@ -2539,6 +2569,18 @@ mod validations {
25392569
let arg_type = match source.field(ident) {
25402570
Some(arg_field) => match arg_field.field_type.value_type() {
25412571
Ok(arg_type) if arg_type.is_numeric() => arg_type,
2572+
Ok(ValueType::Bytes | ValueType::String)
2573+
if is_first_last =>
2574+
{
2575+
return Ok(());
2576+
}
2577+
Err(_)
2578+
if is_first_last
2579+
&& arg_field.field_type.get_base_type()
2580+
== field.field_type.get_base_type() =>
2581+
{
2582+
return Ok(());
2583+
}
25422584
Ok(_) | Err(_) => {
25432585
return Err(Err::AggregationNonNumericArg(
25442586
agg_type.name.to_owned(),
@@ -2556,15 +2598,27 @@ mod validations {
25562598
));
25572599
}
25582600
};
2559-
if arg_type > field_type {
2560-
return Err(Err::AggregationNonMatchingArg(
2561-
agg_type.name.to_owned(),
2562-
field.name.to_owned(),
2563-
arg.to_owned(),
2564-
arg_type.to_str().to_owned(),
2565-
field_type.to_str().to_owned(),
2566-
));
2601+
2602+
match field.field_type.value_type() {
2603+
Ok(field_type) if field_type.is_numeric() => {
2604+
if arg_type > field_type {
2605+
return Err(Err::AggregationNonMatchingArg(
2606+
agg_type.name.to_owned(),
2607+
field.name.to_owned(),
2608+
arg.to_owned(),
2609+
arg_type.to_str().to_owned(),
2610+
field_type.to_str().to_owned(),
2611+
));
2612+
}
2613+
}
2614+
Ok(_) | Err(_) => {
2615+
return Err(Err::NonNumericAggregate(
2616+
agg_type.name.to_owned(),
2617+
field.name.to_owned(),
2618+
));
2619+
}
25672620
}
2621+
25682622
Ok(())
25692623
};
25702624
if let Err(mut errs) = sqlexpr::parse(arg, check_ident) {
@@ -2661,7 +2715,7 @@ mod validations {
26612715
errors.push(err);
26622716
}
26632717
no_derived_fields(agg_type, &mut errors);
2664-
aggregate_fields_are_numbers(agg_type, &mut errors);
2718+
aggregate_field_types(self, agg_type, &mut errors);
26652719
aggregate_directive(self, agg_type, &mut errors);
26662720
// check timeseries directive has intervals and args
26672721
aggregation_intervals(agg_type, &mut errors);

‎graph/src/schema/mod.rs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -123,6 +123,8 @@ pub enum SchemaValidationError {
123123
TimestampFieldMissing(String),
124124
#[error("Aggregation {0}, field{1}: aggregates must use a numeric type, one of Int, Int8, BigInt, and BigDecimal")]
125125
NonNumericAggregate(String, String),
126+
#[error("Aggregation '{0}', field '{1}': first/last aggregates must use a numeric, byte array, string or a reference type")]
127+
InvalidFirstLastAggregate(String, String),
126128
#[error("Aggregation {0} is missing the `source` argument")]
127129
AggregationMissingSource(String),
128130
#[error(

‎graph/src/schema/test_schemas/ts_invalid_first_type_reference.graphql‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
# fail: FieldTypeUnknown("Stats", "firstBlockNumber", "BlockNumber")
2+
3+
type Data @entity(timeseries: true) {
4+
id: Int8!
5+
timestamp: Timestamp!
6+
blockNumber: Int8!
7+
}
8+
9+
type Stats @aggregation(intervals: ["hour", "day"], source: "Data") {
10+
id: Int8!
11+
timestamp: Timestamp!
12+
13+
firstBlockNumber: BlockNumber! @aggregate(fn: "first", arg: "blockNumber")
14+
}

‎graph/src/schema/test_schemas/ts_invalid_last_type_reference.graphql‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
# fail: FieldTypeUnknown("Stats", "lastBlockNumber", "BlockNumber")
2+
3+
type Data @entity(timeseries: true) {
4+
id: Int8!
5+
timestamp: Timestamp!
6+
blockNumber: Int8!
7+
}
8+
9+
type Stats @aggregation(intervals: ["hour", "day"], source: "Data") {
10+
id: Int8!
11+
timestamp: Timestamp!
12+
13+
lastBlockNumber: BlockNumber! @aggregate(fn: "last", arg: "blockNumber")
14+
}

‎graph/src/schema/test_schemas/ts_valid_non_numeric_first_last.graphql‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
# valid: Non-numeric first and last aggregations
2+
3+
type EntityA @entity(immutable: true) {
4+
id: ID!
5+
}
6+
7+
type EntityB @entity(immutable: true) {
8+
id: Int8!
9+
}
10+
11+
type EntityC @entity(immutable: true) {
12+
id: Bytes!
13+
}
14+
15+
type Data @entity(timeseries: true) {
16+
id: Int8!
17+
timestamp: Timestamp!
18+
fieldA: EntityA!
19+
fieldB: EntityB!
20+
fieldC: EntityC!
21+
fieldD: String!
22+
fieldE: Bytes!
23+
}
24+
25+
type Stats @aggregation(intervals: ["hour", "day"], source: "Data") {
26+
id: Int8!
27+
timestamp: Timestamp!
28+
29+
firstA: EntityA! @aggregate(fn: "first", arg: "fieldA")
30+
lastA: EntityA! @aggregate(fn: "last", arg: "fieldA")
31+
32+
firstB: EntityB! @aggregate(fn: "first", arg: "fieldB")
33+
lastB: EntityB! @aggregate(fn: "last", arg: "fieldB")
34+
35+
firstC: EntityC! @aggregate(fn: "first", arg: "fieldC")
36+
lastC: EntityC! @aggregate(fn: "last", arg: "fieldC")
37+
38+
firstD: String! @aggregate(fn: "first", arg: "fieldD")
39+
lastD: String! @aggregate(fn: "last", arg: "fieldD")
40+
41+
firstE: Bytes! @aggregate(fn: "first", arg: "fieldE")
42+
lastE: Bytes! @aggregate(fn: "last", arg: "fieldE")
43+
}

‎store/postgres/migrations/2026-01-09-164812_extend_arg_min_max/down.sql‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
-- This file was generated by generate.sh in this directory
2+
set search_path = public;
3+
drop aggregate arg_min_text(text_and_value);
4+
drop aggregate arg_max_text(text_and_value);
5+
drop function arg_from_text_and_value(text_and_value);
6+
drop function arg_max_agg_text(text_and_value, text_and_value);
7+
drop function arg_min_agg_text(text_and_value, text_and_value);
8+
drop type text_and_value;
9+
drop aggregate arg_min_bytea(bytea_and_value);
10+
drop aggregate arg_max_bytea(bytea_and_value);
11+
drop function arg_from_bytea_and_value(bytea_and_value);
12+
drop function arg_max_agg_bytea(bytea_and_value, bytea_and_value);
13+
drop function arg_min_agg_bytea(bytea_and_value, bytea_and_value);
14+
drop type bytea_and_value;

‎store/postgres/migrations/2026-01-09-164812_extend_arg_min_max/up.sql‎

Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,92 @@
1+
-- This file was generated by generate.sh in this directory
2+
set search_path = public;
3+
create type public.text_and_value as (
4+
arg text,
5+
value int8
6+
);
7+
8+
create or replace function arg_min_agg_text (a text_and_value, b text_and_value)
9+
returns text_and_value
10+
language sql immutable strict parallel safe as
11+
'select case when a.arg is null then b
12+
when b.arg is null then a
13+
when a.value <= b.value then a
14+
else b end';
15+
16+
create or replace function arg_max_agg_text (a text_and_value, b text_and_value)
17+
returns text_and_value
18+
language sql immutable strict parallel safe as
19+
'select case when a.arg is null then b
20+
when b.arg is null then a
21+
when a.value > b.value then a
22+
else b end';
23+
24+
create function arg_from_text_and_value(a text_and_value)
25+
returns text
26+
language sql immutable strict parallel safe as
27+
'select a.arg';
28+
29+
create aggregate arg_min_text (text_and_value) (
30+
sfunc = arg_min_agg_text,
31+
stype = text_and_value,
32+
finalfunc = arg_from_text_and_value,
33+
parallel = safe
34+
);
35+
36+
comment on aggregate arg_min_text(text_and_value) is
37+
'For ''select arg_min_text((arg, value)) from ..'' return the arg for the smallest value';
38+
39+
create aggregate arg_max_text (text_and_value) (
40+
sfunc = arg_max_agg_text,
41+
stype = text_and_value,
42+
finalfunc = arg_from_text_and_value,
43+
parallel = safe
44+
);
45+
46+
comment on aggregate arg_max_text(text_and_value) is
47+
'For ''select arg_max_text((arg, value)) from ..'' return the arg for the largest value';
48+
create type public.bytea_and_value as (
49+
arg bytea,
50+
value int8
51+
);
52+
53+
create or replace function arg_min_agg_bytea (a bytea_and_value, b bytea_and_value)
54+
returns bytea_and_value
55+
language sql immutable strict parallel safe as
56+
'select case when a.arg is null then b
57+
when b.arg is null then a
58+
when a.value <= b.value then a
59+
else b end';
60+
61+
create or replace function arg_max_agg_bytea (a bytea_and_value, b bytea_and_value)
62+
returns bytea_and_value
63+
language sql immutable strict parallel safe as
64+
'select case when a.arg is null then b
65+
when b.arg is null then a
66+
when a.value > b.value then a
67+
else b end';
68+
69+
create function arg_from_bytea_and_value(a bytea_and_value)
70+
returns bytea
71+
language sql immutable strict parallel safe as
72+
'select a.arg';
73+
74+
create aggregate arg_min_bytea (bytea_and_value) (
75+
sfunc = arg_min_agg_bytea,
76+
stype = bytea_and_value,
77+
finalfunc = arg_from_bytea_and_value,
78+
parallel = safe
79+
);
80+
81+
comment on aggregate arg_min_bytea(bytea_and_value) is
82+
'For ''select arg_min_bytea((arg, value)) from ..'' return the arg for the smallest value';
83+
84+
create aggregate arg_max_bytea (bytea_and_value) (
85+
sfunc = arg_max_agg_bytea,
86+
stype = bytea_and_value,
87+
finalfunc = arg_from_bytea_and_value,
88+
parallel = safe
89+
);
90+
91+
comment on aggregate arg_max_bytea(bytea_and_value) is
92+
'For ''select arg_max_bytea((arg, value)) from ..'' return the arg for the largest value';

0 commit comments

Comments
 (0)