fix(core, graph): simplify SQL query requirements

Only require block number columns and try to load
block hashes and timestamps from the source tables

Author: isum

core/src/amp_subgraph/runner/data_processing.rs

@@ -215,7 +215,7 @@ fn decode_block_timestamp(record_batches: &[StreamRecordBatch]) -> Result<DateTi
 
     for record_batch in record_batches {
         match auto_block_timestamp_decoder(&record_batch.record_batch) {
-            Ok(decoder) => {
+            Ok((_, decoder)) => {
                 return decoder
                     .decode(0)
                     .map_err(|e| Error::Deterministic(e))?

core/src/amp_subgraph/runner/data_stream.rs

@@ -70,7 +70,7 @@ where
             }
 
             for (j, table) in data_source.transformer.tables.iter().enumerate() {
-                let query = table.query.with_block_range_filter(block_range);
+                let query = table.query.build_with_block_range(block_range);
 
                 query_streams.push(cx.client.query(&cx.logger, query, None));
                 query_streams_table_ptr.push((i, j));

graph/src/amp/codec/utils.rs

@@ -10,61 +10,85 @@ use crate::amp::common::column_aliases;
 
 pub fn auto_block_number_decoder<'a>(
     record_batch: &'a RecordBatch,
-) -> Result<Box<dyn Decoder<Option<BlockNumber>> + 'a>> {
-    let column_index = column_index(record_batch, column_aliases::BLOCK_NUMBER)
-        .context("failed to find block numbers column")?;
+) -> Result<(&'static str, Box<dyn Decoder<Option<BlockNumber>> + 'a>)> {
+    let (&column_name, column_index) = find_column(record_batch, column_aliases::BLOCK_NUMBER)
+        .with_context(|| {
+            format!(
+                "failed to find block numbers column; expected one of: {}",
+                column_aliases::BLOCK_NUMBER.join(", ")
+            )
+        })?;
 
     block_number_decoder(record_batch, column_index)
+        .map(|decoder| (column_name, decoder))
+        .with_context(|| format!("column '{column_name}' is not valid"))
 }
 
 pub fn block_number_decoder<'a>(
     record_batch: &'a RecordBatch,
     column_index: usize,
 ) -> Result<Box<dyn Decoder<Option<BlockNumber>> + 'a>> {
-    column_decoder::<UInt64Array, BlockNumber>(record_batch, column_index)
+    column_decoder::<UInt64Array, BlockNumber>(record_batch, column_index, false)
 }
 
 pub fn auto_block_hash_decoder<'a>(
     record_batch: &'a RecordBatch,
-) -> Result<Box<dyn Decoder<Option<BlockHash>> + 'a>> {
-    let column_index = column_index(record_batch, column_aliases::BLOCK_HASH)
-        .context("failed to find block hashes column")?;
+) -> Result<(&'static str, Box<dyn Decoder<Option<BlockHash>> + 'a>)> {
+    let (&column_name, column_index) = find_column(record_batch, column_aliases::BLOCK_HASH)
+        .with_context(|| {
+            format!(
+                "failed to find block hashes column; expected one of: {}",
+                column_aliases::BLOCK_HASH.join(", ")
+            )
+        })?;
 
     block_hash_decoder(record_batch, column_index)
+        .map(|decoder| (column_name, decoder))
+        .with_context(|| format!("column '{column_name}' is not valid"))
 }
 
 pub fn block_hash_decoder<'a>(
     record_batch: &'a RecordBatch,
     column_index: usize,
 ) -> Result<Box<dyn Decoder<Option<BlockHash>> + 'a>> {
-    column_decoder::<FixedSizeBinaryArray, BlockHash>(record_batch, column_index)
+    column_decoder::<FixedSizeBinaryArray, BlockHash>(record_batch, column_index, false)
 }
 
 pub fn auto_block_timestamp_decoder<'a>(
     record_batch: &'a RecordBatch,
-) -> Result<Box<dyn Decoder<Option<DateTime<Utc>>> + 'a>> {
-    let column_index = column_index(record_batch, column_aliases::BLOCK_TIMESTAMP)
-        .context("failed to find block timestamps column")?;
+) -> Result<(&'static str, Box<dyn Decoder<Option<DateTime<Utc>>> + 'a>)> {
+    let (&column_name, column_index) = find_column(record_batch, column_aliases::BLOCK_TIMESTAMP)
+        .with_context(|| {
+        format!(
+            "failed to find block timestamps column; expected one of: {}",
+            column_aliases::BLOCK_TIMESTAMP.join(", ")
+        )
+    })?;
 
     block_timestamp_decoder(record_batch, column_index)
+        .map(|decoder| (column_name, decoder))
+        .with_context(|| format!("column '{column_name}' is not valid"))
 }
 
 pub fn block_timestamp_decoder<'a>(
     record_batch: &'a RecordBatch,
     column_index: usize,
 ) -> Result<Box<dyn Decoder<Option<DateTime<Utc>>> + 'a>> {
-    column_decoder::<TimestampNanosecondArray, DateTime<Utc>>(record_batch, column_index)
+    column_decoder::<TimestampNanosecondArray, DateTime<Utc>>(record_batch, column_index, false)
 }
 
-pub fn column_index(
+pub fn find_column<T>(
     record_batch: &RecordBatch,
-    column_names: impl IntoIterator<Item = impl AsRef<str>>,
-) -> Option<usize> {
+    column_names: impl IntoIterator<Item = T>,
+) -> Option<(T, usize)>
+where
+    T: AsRef<str>,
+{
     let schema_ref = record_batch.schema_ref();
 
     for column_name in column_names {
         if let Some((column_index, _)) = schema_ref.column_with_name(column_name.as_ref()) {
-            return Some(column_index);
+            return Some((column_name, column_index));
         }
     }
 
@@ -74,16 +98,22 @@ pub fn column_index(
 pub fn column_decoder<'a, T: 'static, U>(
     record_batch: &'a RecordBatch,
     column_index: usize,
+    nullable: bool,
 ) -> Result<Box<dyn Decoder<Option<U>> + 'a>>
 where
     T: Array,
     ArrayDecoder<'a, T>: Decoder<Option<U>>,
 {
     if column_index >= record_batch.num_columns() {
-        bail!("column {column_index} does not exist");
+        bail!("column does not exist");
     }
 
     let array = record_batch.column(column_index);
+
+    if !nullable && array.is_nullable() {
+        bail!("column must not have nullable values");
+    }
+
     let decoder = ArrayDecoder::<T>::new(array)?;
 
     Ok(Box::new(decoder))

graph/src/amp/manifest/data_source/mod.rs

@@ -8,7 +8,7 @@ use arrow::datatypes::Schema;
 use semver::Version;
 
 use crate::{
-    amp::{common::Ident, sql::Query},
+    amp::{common::Ident, sql::BlockRangeQueryBuilder},
     data::subgraph::SPEC_VERSION_1_5_0,
 };
 
@@ -106,7 +106,7 @@ pub struct Table {
     /// The SQL query that executes on the Amp server.
     ///
     /// The data resulting from this SQL query execution transforms into subgraph entities.
-    pub query: Query,
+    pub query: BlockRangeQueryBuilder,
 
     /// The Arrow schema of this transformed table SQL query.
     ///