Struct AnalyticsConnection 

pub struct AnalyticsConnection<'a> { /* private fields */ }

Connection to the analytics store.

Provides access to the underlying object store for file uploads.

Implementations

impl<'a> AnalyticsConnection<'a>

pub fn mode(&self) -> &StoreMode

Get the store mode for split_batch operations.

pub fn get_pending_batch(&self, pipeline: &str) -> Batch

Get a clone of the pending rows for a pipeline. Returns default FileRows if pipeline has no pending rows.

pub fn set_pending_batch(&self, pipeline: &str, rows: Batch)

Set the pending rows for a pipeline after successful upload.

pub async fn write_to_object_store( &self, pipeline: &str, epoch: EpochId, checkpoint_range: Range<u64>, file_format: FileFormat, payload: PutPayload, ) -> Result<()>

Write a file to the object store.

Constructs the path from the provided parameters and delegates to the store mode.

pub async fn commit_batch<P: Processor>( &mut self, batch_from_framework: &[CheckpointRows], ) -> Result<usize>

Commit a batch of rows to the object store.

Background

The indexer framework has limitations that require us to handle batching and serialization in the store layer:

1. No minimum batch size: The framework supports max batch size but not min batch size, so there’s no way to defer commits until a batch reaches a certain size or to control which checkpoints end up in which output files.

2. No fan-out/fan-in for batch processing: The framework provides no way to fan out processing of a completed batch (e.g., CPU-intensive serialization) before committing it sequentially.

To work around these limitations, this store accumulates rows across checkpoint commits and manages its own batching logic (by checkpoint count or row count). Serialization is offloaded to background workers via spawn_blocking, allowing multiple batches to serialize in parallel while maintaining strict checkpoint ordering for uploads.

Commit Lifecycle

1. Accumulates rows in pending buffer
2. When batch threshold is reached, sends to background upload worker
3. Worker serializes (parallel) and uploads (sequential by checkpoint order)

Backpressure: If the upload channel is full, this method blocks.

Error Handling

The framework assumes commit_batch is atomic - if an error is returned, the transaction is rolled back and retried. This method never returns an error; object store write failures are retried internally. The implementation is idempotent, so framework retries would be safe anyway.

Trait Implementations

impl Connection for AnalyticsConnection<'_>

fn init_watermark<'life0, 'life1, 'async_trait>( &'life0 mut self, pipeline_task: &'life1 str, init_watermark: InitWatermark, ) -> Pin<Box<dyn Future<Output = Result<InitWatermark>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Initialize watermark.

In live mode: Watermarks are derived from file names, so just delegates to committer_watermark. In migration mode: Delegates to MigrationStore::init_watermark.

fn committer_watermark<'life0, 'life1, 'async_trait>( &'life0 mut self, pipeline: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<CommitterWatermark>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Determine the watermark.

In live mode: scans file names in the object store. In migration mode: reads from watermark metadata file.

fn set_committer_watermark<'life0, 'life1, 'async_trait>( &'life0 mut self, _pipeline_task: &'life1 str, watermark: CommitterWatermark, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Store the watermark for use in commit_batch.

Note: This doesn’t persist the watermark - that’s done by the upload worker after successful file uploads. This just captures the watermark so commit_batch can use it to detect file boundary crossings and end-of-processing.

fn reader_watermark<'life0, 'async_trait>( &'life0 mut self, _pipeline: &'static str, ) -> Pin<Box<dyn Future<Output = Result<Option<ReaderWatermark>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Given a pipeline, return the reader watermark from the database. This is used by the indexer to determine the new reader_lo or inclusive lower bound of available data.

fn pruner_watermark<'life0, 'async_trait>( &'life0 mut self, _pipeline: &'static str, _delay: Duration, ) -> Pin<Box<dyn Future<Output = Result<Option<PrunerWatermark>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Get the bounds for the region that the pruner is allowed to prune, and the time in milliseconds the pruner must wait before it can begin pruning data for the given pipeline. The pruner is allowed to prune the region between the returned pruner_hi (inclusive) and reader_lo (exclusive) after waiting until pruner_timestamp + delay has passed. This minimizes the possibility for the pruner to delete data still expected by inflight read requests.

fn set_reader_watermark<'life0, 'async_trait>( &'life0 mut self, _pipeline: &'static str, _reader_lo: u64, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Update the reader_lo of an existing watermark entry only if it raises reader_lo. Readers will reference this as the inclusive lower bound of available data for the corresponding pipeline.

fn set_pruner_watermark<'life0, 'async_trait>( &'life0 mut self, _pipeline: &'static str, _pruner_hi: u64, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Update the pruner watermark, returns true if the watermark was actually updated.