timeseries_table_format/table/operations/
open.rs

1//! Opening an existing time-series table.
2
3use snafu::{ResultExt, Snafu};
4
5use crate::{
6    storage::TableLocation,
7    table::{TableError, TimeSeriesTable},
8    transaction_log::{CommitError, TableKind, TransactionLogStore},
9};
10
11/// Errors owned by a table open operation.
12#[derive(Debug, Snafu)]
13#[snafu(module, visibility(pub(crate)))]
14#[non_exhaustive]
15pub enum OpenTableError {
16    /// The table location contains no commits.
17    #[snafu(display("Cannot open table with no commits"))]
18    EmptyTable,
19
20    /// The loaded metadata does not describe a time-series table.
21    #[snafu(display("Table kind is {kind:?}, expected a time-series table"))]
22    NotTimeSeries {
23        /// Loaded table kind.
24        kind: TableKind,
25    },
26
27    /// Reading or replaying the transaction log failed.
28    #[snafu(context(false), display("Table open transaction-log error: {source}"))]
29    Commit {
30        /// Complete transaction-log failure.
31        #[snafu(source, backtrace)]
32        source: CommitError,
33    },
34}
35
36impl TimeSeriesTable {
37    /// Open an existing time-series table at the given location.
38    #[tracing::instrument(
39        name = "table.open",
40        target = "timeseries_table_format::table",
41        level = "debug",
42        skip_all,
43        fields(
44            table_version = tracing::field::Empty,
45            index_kind = tracing::field::Empty,
46            outcome = tracing::field::Empty
47        )
48    )]
49    pub async fn open(location: TableLocation) -> Result<Self, TableError> {
50        let result: Result<Self, OpenTableError> = async {
51            let log = TransactionLogStore::new(location);
52            let current_version = log
53                .load_current_version()
54                .await
55                .map_err(OpenTableError::from)?;
56            tracing::Span::current().record("table_version", current_version);
57            if current_version == 0 {
58                return Err(OpenTableError::EmptyTable);
59            }
60
61            let state = log
62                .rebuild_table_state()
63                .await
64                .map_err(OpenTableError::from)?;
65            let index = match &state.table_meta.kind {
66                TableKind::TimeSeries(index) => index.clone(),
67                kind => return Err(OpenTableError::NotTimeSeries { kind: kind.clone() }),
68            };
69            tracing::Span::current().record("index_kind", index.kind.name());
70
71            Ok(Self { log, state, index })
72        }
73        .await;
74        tracing::Span::current().record(
75            "outcome",
76            if result.is_ok() {
77                "succeeded"
78            } else {
79                "failed"
80            },
81        );
82        result.context(crate::table::error::OpenSnafu)
83    }
84}
85
86#[cfg(test)]
87mod tests {
88    use super::*;
89    use crate::{
90        metadata::protocol::TABLE_PROTOCOL_VERSION,
91        table::test_util::{
92            TestResult, TraceCapture, assert_capture_excludes, assert_debug_span, assert_no_event,
93            captured_span, make_basic_table_meta,
94        },
95        transaction_log::{LogAction, TableProtocolError, TransactionLogStore},
96    };
97    use tempfile::TempDir;
98
99    #[tokio::test]
100    async fn open_round_trips_a_created_table() -> TestResult {
101        let tmp = TempDir::new()?;
102        let location = TableLocation::local(tmp.path());
103        let created = TimeSeriesTable::create(location.clone(), make_basic_table_meta()).await?;
104        let capture = TraceCapture::default();
105
106        let reopened = capture.run(TimeSeriesTable::open(location)).await?;
107
108        assert_eq!(created.state().version, reopened.state().version);
109        assert_eq!(created.index_spec(), reopened.index_spec());
110        assert_debug_span(
111            &capture,
112            "table.open",
113            &[
114                ("table_version", Some("1")),
115                ("index_kind", Some("timestamp")),
116                ("outcome", Some("succeeded")),
117            ],
118        );
119        assert_eq!(
120            captured_span(&capture, "table.open").target,
121            "timeseries_table_format::table"
122        );
123        assert_no_event(&capture, "table.open");
124        assert_capture_excludes(&capture, &[&tmp.path().display().to_string()]);
125        Ok(())
126    }
127
128    #[tokio::test]
129    async fn open_rejects_an_empty_location() -> TestResult {
130        let tmp = TempDir::new()?;
131        let capture = TraceCapture::default();
132
133        let error = capture
134            .run(TimeSeriesTable::open(TableLocation::local(tmp.path())))
135            .await
136            .expect_err("empty table must fail");
137
138        assert!(matches!(
139            error,
140            TableError::Open {
141                source: OpenTableError::EmptyTable
142            }
143        ));
144        assert_debug_span(
145            &capture,
146            "table.open",
147            &[
148                ("table_version", Some("0")),
149                ("index_kind", None),
150                ("outcome", Some("failed")),
151            ],
152        );
153        Ok(())
154    }
155
156    #[tokio::test]
157    async fn open_preserves_protocol_and_table_kind_failures() -> TestResult {
158        for found in [TABLE_PROTOCOL_VERSION - 1, TABLE_PROTOCOL_VERSION + 1] {
159            let tmp = TempDir::new()?;
160            let location = TableLocation::local(tmp.path());
161            let mut meta = make_basic_table_meta();
162            meta.protocol_version = found;
163            TransactionLogStore::new(location.clone())
164                .commit_with_expected_version(0, vec![LogAction::UpdateTableMeta(meta)])
165                .await?;
166
167            assert!(matches!(
168                TimeSeriesTable::open(location)
169                    .await
170                    .expect_err("unsupported protocol must fail"),
171                TableError::Open {
172                    source: OpenTableError::Commit {
173                        source: CommitError::Protocol {
174                            source: TableProtocolError::UnsupportedVersion {
175                                expected: TABLE_PROTOCOL_VERSION,
176                                found: actual,
177                            },
178                            ..
179                        }
180                    }
181                } if actual == u64::from(found)
182            ));
183        }
184
185        let tmp = TempDir::new()?;
186        let location = TableLocation::local(tmp.path());
187        let mut meta = make_basic_table_meta();
188        meta.kind = TableKind::Generic;
189        TransactionLogStore::new(location.clone())
190            .commit_with_expected_version(0, vec![LogAction::UpdateTableMeta(meta)])
191            .await?;
192
193        assert!(matches!(
194            TimeSeriesTable::open(location)
195                .await
196                .expect_err("generic table must fail"),
197            TableError::Open {
198                source: OpenTableError::NotTimeSeries {
199                    kind: TableKind::Generic
200                }
201            }
202        ));
203        Ok(())
204    }
205
206    #[tokio::test]
207    async fn open_applies_reader_requirements_without_requiring_writer_support() -> TestResult {
208        let writer_tmp = TempDir::new()?;
209        let writer_location = TableLocation::local(writer_tmp.path());
210        let mut writer_meta = make_basic_table_meta();
211        writer_meta
212            .required_writer_features
213            .insert("future_writer".to_string());
214        TransactionLogStore::new(writer_location.clone())
215            .commit_with_expected_version(0, vec![LogAction::UpdateTableMeta(writer_meta)])
216            .await?;
217
218        let table = TimeSeriesTable::open(writer_location).await?;
219        assert_eq!(
220            table.state().table_meta.required_writer_features(),
221            &["future_writer".to_string()].into_iter().collect()
222        );
223
224        let reader_tmp = TempDir::new()?;
225        let reader_location = TableLocation::local(reader_tmp.path());
226        let mut reader_meta = make_basic_table_meta();
227        reader_meta
228            .required_reader_features
229            .insert("future_reader".to_string());
230        TransactionLogStore::new(reader_location.clone())
231            .commit_with_expected_version(0, vec![LogAction::UpdateTableMeta(reader_meta)])
232            .await?;
233
234        assert!(matches!(
235            TimeSeriesTable::open(reader_location)
236                .await
237                .expect_err("unknown reader feature must reject open"),
238            TableError::Open {
239                source: OpenTableError::Commit {
240                    source: CommitError::Protocol {
241                        source: TableProtocolError::UnsupportedReaderFeatures { features },
242                        ..
243                    }
244                }
245            } if features == ["future_reader"]
246        ));
247        Ok(())
248    }
249}
timeseries_table_format/table/operations/open.rs

timeseries_table_format/table/operations/
open.rs
