timeseries-table-format 0.4.0

//! High-level time-series table abstraction.
//!
//! This module is the canonical home for the user-facing [`TimeSeriesTable`]
//! API.
//!
//! In v0.1 this is intentionally read-heavy and write-light:
//! - `open` reconstructs state from the transaction log,
//! - `create` bootstraps a fresh table with an initial metadata commit,
//! - append APIs handle schema enforcement, coverage sidecars, and OCC,
//! - range scans stream filtered record batches.

pub mod append;
/// Append profiling report types used by CLI benchmarks.
pub mod append_report;
pub mod coverage;
pub mod error;
mod optimize;
pub mod scan;

#[cfg(test)]
pub(crate) mod test_util;

#[cfg(test)]
mod latest_snapshot_tests;

use std::pin::Pin;

use arrow::array::RecordBatch;
use futures::Stream;
use snafu::prelude::*;

use crate::table::error::{
    AlreadyExistsSnafu, EmptyTableSnafu, IndexSpecSnafu, NotTimeSeriesSnafu,
    SchemaCompatibilitySnafu, TransactionLogSnafu, UnsupportedFormatVersionSnafu,
};

use crate::{
    metadata::{
        schema_compat::ensure_index_spec_matches_schema, table_metadata::TABLE_FORMAT_VERSION,
    },
    storage::TableLocation,
    transaction_log::{
        IndexSpec, LogAction, TableKind, TableMeta, TableState, TransactionLogStore,
    },
};

pub use error::TableError;
pub use optimize::OptimizeReport;

/// Stream of Arrow RecordBatch values from a time-series scan.
///
/// Batch and row order is unspecified.
pub type TimeSeriesScan = Pin<Box<dyn Stream<Item = Result<RecordBatch, TableError>> + Send>>;

/// High-level time-series table handle.
///
/// This is the main entry point for callers. It bundles:
/// - where the table is,
/// - how to talk to the transaction log,
/// - what the current committed state is,
/// - and the extracted time index spec.
#[derive(Debug, Clone)]
pub struct TimeSeriesTable {
    log: TransactionLogStore,
    state: TableState,
    index: IndexSpec,
}

impl TimeSeriesTable {
    /// Return the current committed table state.
    pub fn state(&self) -> &TableState {
        &self.state
    }

    /// Return a mutable reference to the current committed table state (crate-internal).
    ///
    /// This exists to support internal helpers (for example, tests) without
    /// exposing mutation to library callers.
    #[allow(dead_code)]
    pub(crate) fn state_mut(&mut self) -> &mut TableState {
        &mut self.state
    }

    /// Return the time index specification for this table.
    pub fn index_spec(&self) -> &IndexSpec {
        &self.index
    }

    /// Return the table location.
    pub fn location(&self) -> &TableLocation {
        self.log.location()
    }

    /// Open an existing time-series table at the given location.
    ///
    /// Steps:
    /// - Build a `TransactionLogStore` for the location.
    /// - Rebuild `TableState` from the transaction log.
    /// - Reject empty tables (version == 0).
    /// - Require `TableKind::TimeSeries` and extract `IndexSpec`.
    pub async fn open(location: TableLocation) -> Result<Self, TableError> {
        let log = TransactionLogStore::new(location.clone());

        // Early return for tables with no commits so we surface TableError::EmptyTable
        // instead of a lower-level corrupt state error.
        let current_version = log
            .load_current_version()
            .await
            .context(TransactionLogSnafu)?;

        if current_version == 0 {
            return EmptyTableSnafu.fail();
        }

        // Rebuild the snapshot of state from the log.
        let state = log
            .rebuild_table_state()
            .await
            .context(TransactionLogSnafu)?;

        // Extract the time index spec from TableMeta.kind.
        let index = match &state.table_meta.kind {
            TableKind::TimeSeries(spec) => spec.clone(),
            other => {
                return NotTimeSeriesSnafu {
                    kind: other.clone(),
                }
                .fail();
            }
        };

        Ok(Self { log, state, index })
    }

    /// Create a new time-series table at the given location.
    ///
    /// This:
    /// - Requires `table_meta.format_version` to match [`TABLE_FORMAT_VERSION`],
    /// - Requires `table_meta.kind` to be `TableKind::TimeSeries`,
    /// - Verifies that there are no existing commits (version must be 0),
    /// - Writes an initial commit with `UpdateTableMeta(table_meta.clone())`,
    /// - Returns a `TimeSeriesTable` with a fresh `TableState`.
    pub async fn create(
        location: TableLocation,
        table_meta: TableMeta,
    ) -> Result<Self, TableError> {
        if table_meta.format_version() != TABLE_FORMAT_VERSION {
            return UnsupportedFormatVersionSnafu {
                expected: TABLE_FORMAT_VERSION,
                found: table_meta.format_version(),
            }
            .fail();
        }

        // 1) Extract the time index spec from the provided metadata
        // and ensure this is actually a time-series table.
        let index = match &table_meta.kind {
            TableKind::TimeSeries(spec) => spec.clone(),
            other => {
                return NotTimeSeriesSnafu {
                    kind: other.clone(),
                }
                .fail();
            }
        };
        index.validate().context(IndexSpecSnafu)?;
        if let Some(schema) = &table_meta.logical_schema {
            ensure_index_spec_matches_schema(schema, &index).context(SchemaCompatibilitySnafu)?;
        }

        let log = TransactionLogStore::new(location.clone());

        // 2) Check that there are no existing commits. This keeps `create`
        // from silently appending to a pre-existing table.
        let current_version = log
            .load_current_version()
            .await
            .context(TransactionLogSnafu)?;

        if current_version != 0 {
            return AlreadyExistsSnafu { current_version }.fail();
        }

        // 3) Write the initial metadata commit at version 1.
        let actions = vec![LogAction::UpdateTableMeta(table_meta.clone())];

        let new_version = log
            .commit_with_expected_version(0, actions)
            .await
            .context(TransactionLogSnafu)?;

        debug_assert_eq!(new_version, 1);

        // 4) Rebuild state from the log so that `state` is guaranteed to be
        // consistent with what is on disk.
        let state = log
            .rebuild_table_state()
            .await
            .context(TransactionLogSnafu)?;
        Ok(Self { log, state, index })
    }

    /// Load the current log version from disk without mutating in-memory state.
    pub async fn current_version(&self) -> Result<u64, TableError> {
        self.log
            .load_current_version()
            .await
            .context(TransactionLogSnafu)
    }

    /// Rebuild and return the latest table state from the transaction log.
    pub async fn load_latest_state(&self) -> Result<TableState, TableError> {
        self.log
            .rebuild_table_state()
            .await
            .context(TransactionLogSnafu)
    }

    /// Refresh in-memory state if the log has advanced; returns true if updated.
    pub async fn refresh(&mut self) -> Result<bool, TableError> {
        let current = self
            .log
            .load_current_version()
            .await
            .context(TransactionLogSnafu)?;

        if current == self.state.version {
            return Ok(false);
        }

        let state = self
            .log
            .rebuild_table_state()
            .await
            .context(TransactionLogSnafu)?;

        let index = match &state.table_meta.kind {
            TableKind::TimeSeries(spec) => spec.clone(),
            other => {
                return NotTimeSeriesSnafu {
                    kind: other.clone(),
                }
                .fail();
            }
        };

        self.state = state;
        self.index = index;
        Ok(true)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    use crate::storage::{StorageLocation, layout};
    use crate::table::test_util::*;
    use crate::transaction_log::{CommitError, IndexKind, TimeBucket, TransactionLogStore};

    use tempfile::TempDir;

    #[tokio::test]
    async fn create_initializes_log_and_state() -> TestResult {
        let tmp = TempDir::new()?;
        let location = TableLocation::local(tmp.path());

        let meta = make_basic_table_meta();
        let table = TimeSeriesTable::create(location.clone(), meta).await?;

        // State should be at version 1 with no segments.
        assert_eq!(table.state().version, 1);
        assert_eq!(TABLE_FORMAT_VERSION, 6);
        assert_eq!(
            table.state().table_meta.format_version(),
            TABLE_FORMAT_VERSION
        );
        assert!(table.state().segments.is_empty());

        // Verify that the log layout exists on disk.
        let root = match table.location().storage() {
            StorageLocation::Local(p) => p.clone(),
        };

        let log_dir = root.join(layout::log_rel_dir());
        assert!(log_dir.is_dir());

        let current_path = root.join(layout::current_rel_path());
        let current_contents = tokio::fs::read_to_string(&current_path).await?;
        assert_eq!(current_contents.trim(), "1");

        Ok(())
    }

    #[tokio::test]
    async fn create_rejects_unsupported_format_without_writing_log() -> TestResult {
        let tmp = TempDir::new()?;
        let location = TableLocation::local(tmp.path());

        for found in [TABLE_FORMAT_VERSION - 1, TABLE_FORMAT_VERSION + 1] {
            let mut meta = make_basic_table_meta();
            meta.format_version = found;

            let err = TimeSeriesTable::create(location.clone(), meta)
                .await
                .expect_err("unsupported format version should be rejected");
            assert!(matches!(
                err,
                TableError::UnsupportedFormatVersion {
                    expected: TABLE_FORMAT_VERSION,
                    found: actual,
                } if actual == found
            ));
            assert!(!tmp.path().join(layout::log_rel_dir()).exists());
        }

        Ok(())
    }

    #[tokio::test]
    async fn open_round_trip_after_create() -> TestResult {
        let tmp = TempDir::new()?;
        let location = TableLocation::local(tmp.path());

        let meta = make_basic_table_meta();
        let created = TimeSeriesTable::create(location.clone(), meta).await?;

        let reopened = TimeSeriesTable::open(location.clone()).await?;

        assert_eq!(created.state().version, reopened.state().version);
        assert_eq!(created.index_spec(), reopened.index_spec());
        Ok(())
    }

    #[tokio::test]
    async fn open_rejects_every_non_current_format_with_typed_error() -> TestResult {
        for found in [TABLE_FORMAT_VERSION - 1, TABLE_FORMAT_VERSION + 1] {
            let tmp = TempDir::new()?;
            let location = TableLocation::local(tmp.path());
            let log = TransactionLogStore::new(location.clone());
            let mut meta = make_basic_table_meta();
            meta.format_version = found;
            log.commit_with_expected_version(0, vec![LogAction::UpdateTableMeta(meta)])
                .await?;

            let error = TimeSeriesTable::open(location)
                .await
                .expect_err("non-current table format must fail");

            assert!(matches!(
                error,
                TableError::TransactionLog {
                    source: CommitError::UnsupportedFormatVersion {
                        expected: TABLE_FORMAT_VERSION,
                        found: actual,
                    },
                } if actual == u64::from(found)
            ));
        }
        Ok(())
    }

    #[tokio::test]
    async fn open_empty_root_errors() -> TestResult {
        let tmp = TempDir::new()?;
        let location = TableLocation::local(tmp.path());

        // There is no CURRENT and no commits, so opening should fail.
        let result = TimeSeriesTable::open(location).await;
        assert!(matches!(result, Err(TableError::EmptyTable)));
        Ok(())
    }

    #[tokio::test]
    async fn create_fails_if_table_already_exists() -> TestResult {
        let tmp = TempDir::new()?;
        let location = TableLocation::local(tmp.path());

        let meta = make_basic_table_meta();
        let _first = TimeSeriesTable::create(location.clone(), meta.clone()).await?;

        // Second create should detect existing commits and fail.
        let result = TimeSeriesTable::create(location.clone(), meta).await;
        assert!(matches!(result, Err(TableError::AlreadyExists { .. })));
        Ok(())
    }

    #[tokio::test]
    async fn refresh_returns_false_when_no_new_commits() -> TestResult {
        let tmp = TempDir::new()?;
        let location = TableLocation::local(tmp.path());

        let meta = make_basic_table_meta();
        let mut table = TimeSeriesTable::create(location.clone(), meta).await?;

        let refreshed = table.refresh().await?;
        assert!(!refreshed);
        assert_eq!(table.state().version, 1);
        Ok(())
    }

    #[tokio::test]
    async fn refresh_updates_state_and_index_on_change() -> TestResult {
        let tmp = TempDir::new()?;
        let location = TableLocation::local(tmp.path());

        let meta = make_basic_table_meta();
        let mut table = TimeSeriesTable::create(location.clone(), meta.clone()).await?;

        let mut updated_meta = meta.clone();
        if let TableKind::TimeSeries(spec) = &mut updated_meta.kind {
            spec.kind = IndexKind::Timestamp {
                bucket: TimeBucket::Minutes(5),
                timezone: None,
            };
        }

        let log = TransactionLogStore::new(location.clone());
        let new_version = log
            .commit_with_expected_version(1, vec![LogAction::UpdateTableMeta(updated_meta.clone())])
            .await?;
        assert_eq!(new_version, 2);

        let refreshed = table.refresh().await?;
        assert!(refreshed);
        assert_eq!(table.state().version, 2);

        match &table.state().table_meta.kind {
            TableKind::TimeSeries(spec) => assert_eq!(
                spec.kind,
                IndexKind::Timestamp {
                    bucket: TimeBucket::Minutes(5),
                    timezone: None
                }
            ),
            other => panic!("expected time series table kind, got {other:?}"),
        }
        assert_eq!(
            table.index_spec().kind,
            IndexKind::Timestamp {
                bucket: TimeBucket::Minutes(5),
                timezone: None
            }
        );
        Ok(())
    }
}