graph, store: Require function syntax for aggregations

Author: lutter

docs/implementation/sql-interface.md

@@ -20,8 +20,11 @@ The body of the request must contain the following keys:
   result is sent in the response
 
 The SQL query can use all the tables of the given subgraph. Table and
-attribute names are snake-cased from their form in the GraphQL schema, so
-that data for `SomeDailyStuff` is stored in a table `some_daily_stuff`.
+attribute names for normal `@entity` types are snake-cased from their form
+in the GraphQL schema, so that data for `SomeDailyStuff` is stored in a
+table `some_daily_stuff`. For `@aggregation` types, the table can be
+accessed as `<aggregation>(<interval>)`, for example, `my_stats('hour')` for
+`type MyStats @aggregation(..) { .. }`
 
 The query can use fairly arbitrary SQL, including aggregations and most
 functions built into PostgreSQL.

graph/src/schema/input/mod.rs

@@ -1383,6 +1383,14 @@ impl InputSchema {
             .any(|ti| matches!(ti, TypeInfo::Aggregation(_)))
     }
 
+    pub fn aggregation_names(&self) -> impl Iterator<Item = &str> {
+        self.inner
+            .type_infos
+            .iter()
+            .filter_map(TypeInfo::aggregation)
+            .map(|agg_type| self.inner.pool.get(agg_type.name).unwrap())
+    }
+
     pub fn entity_fulltext_definitions(
         &self,
         entity: &str,

store/postgres/src/relational.rs

@@ -39,7 +39,8 @@ use graph::data_source::CausalityRegion;
 use graph::internal_error;
 use graph::prelude::{q, EntityQuery, StopwatchMetrics, ENV_VARS};
 use graph::schema::{
-    EntityKey, EntityType, Field, FulltextConfig, FulltextDefinition, InputSchema,
+    AggregationInterval, EntityKey, EntityType, Field, FulltextConfig, FulltextDefinition,
+    InputSchema,
 };
 use graph::slog::warn;
 use index::IndexList;
@@ -1156,6 +1157,27 @@ impl Layout {
         Ok(rollups)
     }
 
+    /// Given an aggregation name that is already snake-cased like `stats`
+    /// (for an an aggregation `type Stats @aggregation(..)`) and an
+    /// interval, return the table that holds the aggregated data, like
+    /// `stats_hour`.
+    pub fn aggregation_table(
+        &self,
+        aggregation: &str,
+        interval: AggregationInterval,
+    ) -> Option<&Table> {
+        let sql_name = format!("{}_{interval}", aggregation);
+        self.table(&sql_name)
+    }
+
+    /// Return true if the layout has an aggregation with the given name
+    /// like `stats` (already snake_cased)
+    pub fn has_aggregation(&self, aggregation: &str) -> bool {
+        self.input_schema
+            .aggregation_names()
+            .any(|agg_name| SqlName::from(agg_name).as_str() == aggregation)
+    }
+
     /// Roll up all timeseries for each entry in `block_times`. The overall
     /// effect is that all buckets that end after `last_rollup` and before
     /// the last entry in `block_times` are filled. This will fill all

store/postgres/src/sql/parser.rs

@@ -43,7 +43,7 @@ mod test {
 
     use super::Parser;
 
-    const TEST_GQL: &str = "
+    const TEST_GQL: &str = r#"
         type Swap @entity(immutable: true) {
             id: Bytes!
             timestamp: BigInt!
@@ -68,7 +68,19 @@ mod test {
             name: String!
             decimals: Int!
         }
-    ";
+
+        type Data @entity(timeseries: true) {
+            id: Int8!
+            timestamp: Timestamp!
+            price: Int!
+        }
+
+        type Stats @aggregation(intervals: ["hour", "day"], source: "Data") {
+            id: Int8!
+            timestamp: Timestamp!
+            sum: BigDecimal! @aggregate(fn: "sum", arg: "price")
+        }
+    "#;
 
     fn parse_and_validate(sql: &str) -> Result<String, anyhow::Error> {
         let parser = Parser::new(Arc::new(make_layout(TEST_GQL)), BLOCK_NUMBER_MAX);

store/postgres/src/sql/parser_tests.yaml

@@ -98,3 +98,30 @@
 - name: unknown tables forbidden
   sql: select * from unknown_table
   err: Unknown table unknown_table
+- name: qualified tables are forbidden
+  sql: select * from pg_catalog.pg_class
+  err: "Qualified table names are not supported: pg_catalog.pg_class"
+- name: aggregation tables are hidden
+  sql: select * from stats_hour
+  err: Unknown table stats_hour
+- name: CTEs take precedence
+  sql: with stats_hour as (select 1) select * from stats_hour
+  ok: WITH stats_hour AS (SELECT 1) SELECT * FROM stats_hour
+- name: aggregation tables use function syntax
+  sql: select * from stats('hour')
+  ok: SELECT * FROM (SELECT "id", "timestamp", "sum" FROM "sgd0815"."stats_hour" WHERE block$ <= 2147483647) AS stats_hour
+- name: unknown aggregation interval
+  sql: select * from stats('fortnight')
+  err: Unknown aggregation interval `fortnight` for table stats
+- name: aggregation tables with empty arg
+  sql: select * from stats('')
+  err: Unknown aggregation interval `` for table stats
+- name: aggregation tables with no args
+  sql: select * from stats()
+  err: Invalid syntax for aggregation stats
+- name: aggregation tables with multiple args
+  sql: select * from stats('hour', 'day')
+  err: Invalid syntax for aggregation stats
+- name: aggregation tables with alias
+  sql: select * from stats('hour') as sh
+  ok: SELECT * FROM (SELECT "id", "timestamp", "sum" FROM "sgd0815"."stats_hour" WHERE block$ <= 2147483647) AS sh

store/postgres/src/sql/validation.rs

@@ -1,7 +1,8 @@
 use graph::prelude::BlockNumber;
+use graph::schema::AggregationInterval;
 use sqlparser::ast::{
-    Expr, Ident, ObjectName, Offset, Query, SetExpr, Statement, TableAlias, TableFactor, Value,
-    VisitMut, VisitorMut,
+    Expr, FunctionArg, FunctionArgExpr, Ident, ObjectName, Offset, Query, SetExpr, Statement,
+    TableAlias, TableFactor, Value, VisitMut, VisitorMut,
 };
 use sqlparser::parser::Parser;
 use std::result::Result;
@@ -22,12 +23,18 @@ pub enum Error {
     NotSelectQuery,
     #[error("Unknown table {0}")]
     UnknownTable(String),
+    #[error("Unknown aggregation interval `{1}` for table {0}")]
+    UnknownAggregationInterval(String, String),
+    #[error("Invalid syntax for aggregation {0}")]
+    InvalidAggregationSyntax(String),
     #[error("Only constant numbers are supported for LIMIT and OFFSET.")]
     UnsupportedLimitOffset,
     #[error("The limit of {0} is greater than the maximum allowed limit of {1}.")]
     UnsupportedLimit(u32, u32),
     #[error("The offset of {0} is greater than the maximum allowed offset of {1}.")]
     UnsupportedOffset(u32, u32),
+    #[error("Qualified table names are not supported: {0}")]
+    NoQualifiedTables(String),
 }
 
 pub struct Validator<'a> {
@@ -151,25 +158,79 @@ impl VisitorMut for Validator<'_> {
         &mut self,
         table_factor: &mut TableFactor,
     ) -> ControlFlow<Self::Break> {
+        /// Check whether `args` is a single string argument and return that
+        /// string
+        fn extract_string_arg(args: &Vec<FunctionArg>) -> Option<String> {
+            if args.len() != 1 {
+                return None;
+            }
+            match &args[0] {
+                FunctionArg::Unnamed(FunctionArgExpr::Expr(Expr::Value(
+                    Value::SingleQuotedString(s),
+                ))) => Some(s.clone()),
+                _ => None,
+            }
+        }
+
         if let TableFactor::Table {
             name, args, alias, ..
         } = table_factor
         {
-            if args.is_some() {
-                return self.validate_function_name(name);
+            if name.0.len() != 1 {
+                // We do not support schema qualified table names
+                return ControlFlow::Break(Error::NoQualifiedTables(name.to_string()));
             }
-            let table = if let Some(table_name) = name.0.last() {
-                let name = &table_name.value;
-                let Some(table) = self.layout.table(name) else {
-                    if self.ctes.contains(name) {
-                        return ControlFlow::Continue(());
-                    } else {
-                        return ControlFlow::Break(Error::UnknownTable(name.to_string()));
-                    }
-                };
-                table
-            } else {
+            let table_name = &name.0[0].value;
+
+            // CTES override subgraph tables
+            if self.ctes.contains(&table_name.to_lowercase()) && args.is_none() {
                 return ControlFlow::Continue(());
+            }
+
+            let table = match (self.layout.table(table_name), args) {
+                (None, None) => {
+                    return ControlFlow::Break(Error::UnknownTable(table_name.clone()));
+                }
+                (Some(_), Some(_)) => {
+                    // Table exists but has args, must be a function
+                    return self.validate_function_name(&name);
+                }
+                (None, Some(args)) => {
+                    // Table does not exist but has args, is either an
+                    // aggregation table in the form <name>(<interval>) or
+                    // must be a function
+
+                    if !self.layout.has_aggregation(table_name) {
+                        // Not an aggregation, must be a function
+                        return self.validate_function_name(&name);
+                    }
+
+                    let Some(intv) = extract_string_arg(args) else {
+                        // Looks like an aggregation, but argument is not a single string
+                        return ControlFlow::Break(Error::InvalidAggregationSyntax(
+                            table_name.clone(),
+                        ));
+                    };
+                    let Some(intv) = intv.parse::<AggregationInterval>().ok() else {
+                        return ControlFlow::Break(Error::UnknownAggregationInterval(
+                            table_name.clone(),
+                            intv,
+                        ));
+                    };
+
+                    let Some(table) = self.layout.aggregation_table(table_name, intv) else {
+                        return self.validate_function_name(&name);
+                    };
+                    table
+                }
+                (Some(table), None) => {
+                    if !table.object.is_object_type() {
+                        // Interfaces and aggregations can not be queried
+                        // with the table name directly
+                        return ControlFlow::Break(Error::UnknownTable(table_name.clone()));
+                    }
+                    table
+                }
             };
 
             // Change 'from table [as alias]' to 'from (select {columns} from table) as alias'