core, graph: Rename GRAPH_AMP_MAX_BUFFER_SIZE to GRAPH_AMP_BUFFER_SIZE

Author: lutter

core/src/amp_subgraph/runner/context.rs

@@ -18,7 +18,7 @@ pub(in super::super) struct Context<AC> {
     pub(super) logger: Logger,
     pub(super) client: Arc<AC>,
     pub(super) store: Arc<dyn WritableStore>,
-    pub(super) max_buffer_size: usize,
+    pub(super) buffer_size: usize,
     pub(super) max_block_range: usize,
     pub(super) backoff: ExponentialBackoff,
     pub(super) deployment: DeploymentHash,
@@ -45,7 +45,7 @@ impl<AC> Context<AC> {
             logger,
             client,
             store,
-            max_buffer_size: env.max_buffer_size,
+            buffer_size: env.buffer_size,
             max_block_range: env.max_block_range,
             backoff,
             deployment,

core/src/amp_subgraph/runner/data_stream.rs

@@ -33,7 +33,7 @@ where
     let logger = cx.logger.new(slog::o!("process" => "new_data_stream"));
     let client = cx.client.cheap_clone();
     let manifest = cx.manifest.clone();
-    let max_buffer_size = cx.max_buffer_size;
+    let buffer_size = cx.buffer_size;
     let max_block_range = cx.max_block_range;
     let stopwatch = cx.metrics.stopwatch.cheap_clone();
 
@@ -74,7 +74,7 @@ where
                 &logger,
                 query_streams,
                 table_ptrs,
-                max_buffer_size,
+                buffer_size,
                 &stopwatch,
                 start_block,
             );
@@ -130,7 +130,7 @@ fn build_data_stream<E>(
     logger: &slog::Logger,
     query_streams: Vec<(String, BoxStream<'static, Result<ResponseBatch, E>>)>,
     table_ptrs: Arc<[TablePtr]>,
-    max_buffer_size: usize,
+    buffer_size: usize,
     stopwatch: &StopwatchMetrics,
     min_start_block: BlockNumber,
 ) -> BoxStream<'static, Result<(RecordBatchGroups, Arc<[TablePtr]>), Error>>
@@ -141,7 +141,7 @@ where
     let mut load_first_record_batch_group_section =
         Some(stopwatch.start_section("load_first_record_batch_group"));
 
-    StreamAggregator::new(logger, query_streams, max_buffer_size)
+    StreamAggregator::new(logger, query_streams, buffer_size)
         .map_ok(move |response| (response, table_ptrs.cheap_clone()))
         .map_err(Error::from)
         .map(move |result| {

docs/amp-powered-subgraphs.md

@@ -563,7 +563,7 @@ Amp-powered subgraphs feature introduces the following new ENV variables:
 
 - `GRAPH_AMP_FLIGHT_SERVICE_ADDRESS` – The address of the Amp Flight gRPC service. _Defaults to `None`, which disables support for Amp-powered subgraphs._
 - `GRAPH_AMP_FLIGHT_SERVICE_TOKEN` – Token used to authenticate Amp Flight gRPC service requests. _Defaults to `None`, which disables authentication._
-- `GRAPH_AMP_MAX_BUFFER_SIZE` – Maximum number of response batches to buffer in memory per stream for each SQL query. _Defaults to `1,000`._
+- `GRAPH_AMP_BUFFER_SIZE` – Maximum number of response batches to buffer in memory per stream for each SQL query. _Defaults to `1,000`._
 - `GRAPH_AMP_MAX_BLOCK_RANGE` – Maximum number of blocks to request per stream for each SQL query. _Defaults to `2,000,000`._
 - `GRAPH_AMP_QUERY_RETRY_MIN_DELAY_SECONDS` – Minimum time to wait before retrying a failed SQL query to the Amp server. _Defaults to `1` second._
 - `GRAPH_AMP_QUERY_RETRY_MAX_DELAY_SECONDS` – Maximum time to wait before retrying a failed SQL query to the Amp server. _Defaults to `600` seconds._

graph/src/amp/stream_aggregator/mod.rs

@@ -60,7 +60,7 @@ impl StreamAggregator {
     pub fn new<E>(
         logger: &Logger,
         named_streams: impl IntoIterator<Item = (String, BoxStream<'static, Result<ResponseBatch, E>>)>,
-        max_buffer_size: usize,
+        buffer_size: usize,
     ) -> Self
     where
         E: std::error::Error + IsDeterministic + Send + Sync + 'static,
@@ -103,7 +103,7 @@ impl StreamAggregator {
 
         Self {
             named_streams,
-            buffer: Buffer::new(num_streams, max_buffer_size),
+            buffer: Buffer::new(num_streams, buffer_size),
             logger,
             is_finalized: false,
             is_failed: false,

graph/src/amp/stream_aggregator/record_batch/buffer.rs

@@ -11,21 +11,21 @@ use super::{Aggregator, RecordBatchGroup, RecordBatchGroups, StreamRecordBatch};
 pub(in super::super) struct Buffer {
     aggregators: Vec<Aggregator>,
     num_streams: usize,
-    max_buffer_size: usize,
+    buffer_size: usize,
 }
 
 impl Buffer {
     /// Creates a new buffer that can handle exactly `num_streams` number of streams.
     ///
     /// Creates a new associated `Aggregator` for each stream.
-    /// The `max_buffer_size` specifies how many record batches for each stream can be buffered at most.
-    pub(in super::super) fn new(num_streams: usize, max_buffer_size: usize) -> Self {
+    /// The `buffer_size` specifies how many record batches for each stream can be buffered at most.
+    pub(in super::super) fn new(num_streams: usize, buffer_size: usize) -> Self {
         let aggregators = (0..num_streams).map(|_| Aggregator::new()).collect();
 
         Self {
             aggregators,
             num_streams,
-            max_buffer_size,
+            buffer_size,
         }
     }
 
@@ -130,7 +130,7 @@ impl Buffer {
     /// Panics if the `stream_index` is greater than the initialized number of streams.
     pub(in super::super) fn has_capacity(&self, stream_index: usize) -> bool {
         assert!(stream_index < self.num_streams);
-        self.aggregators[stream_index].len() < self.max_buffer_size
+        self.aggregators[stream_index].len() < self.buffer_size
     }
 
     /// Returns `true` if the stream `stream_index` is not allowed to make progress and

graph/src/env/amp.rs

@@ -7,7 +7,7 @@ pub struct AmpEnv {
     /// This is the maximum number of record batches that can be output by a single block.
     ///
     /// Defaults to `1,000`.
-    pub max_buffer_size: usize,
+    pub buffer_size: usize,
 
     /// Maximum number of blocks to request per stream for each SQL query.
     /// Limiting this value reduces load on the Amp server when processing heavy queries.
@@ -32,22 +32,22 @@ pub struct AmpEnv {
 }
 
 impl AmpEnv {
-    const DEFAULT_MAX_BUFFER_SIZE: usize = 1_000;
+    const DEFAULT_BUFFER_SIZE: usize = 1_000;
     const DEFAULT_MAX_BLOCK_RANGE: usize = 2_000_000;
     const DEFAULT_QUERY_RETRY_MIN_DELAY: Duration = Duration::from_secs(1);
     const DEFAULT_QUERY_RETRY_MAX_DELAY: Duration = Duration::from_secs(600);
 
     pub(super) fn new(raw_env: &super::Inner) -> Self {
         Self {
-            max_buffer_size: raw_env
-                .amp_max_buffer_size
+            buffer_size: raw_env
+                .amp_buffer_size
                 .and_then(|value| {
                     if value == 0 {
                         return None;
                     }
                     Some(value)
                 })
-                .unwrap_or(Self::DEFAULT_MAX_BUFFER_SIZE),
+                .unwrap_or(Self::DEFAULT_BUFFER_SIZE),
             max_block_range: raw_env
                 .amp_max_block_range
                 .map(|mut value| {

graph/src/env/mod.rs

@@ -608,8 +608,8 @@ struct Inner {
     )]
     disable_deployment_hash_validation: EnvVarBoolean,
 
-    #[envconfig(from = "GRAPH_AMP_MAX_BUFFER_SIZE")]
-    amp_max_buffer_size: Option<usize>,
+    #[envconfig(from = "GRAPH_AMP_BUFFER_SIZE")]
+    amp_buffer_size: Option<usize>,
     #[envconfig(from = "GRAPH_AMP_MAX_BLOCK_RANGE")]
     amp_max_block_range: Option<usize>,
     #[envconfig(from = "GRAPH_AMP_QUERY_RETRY_MIN_DELAY_SECONDS")]