infino 0.5.6

A fast retrieval engine that stores data on object storage and runs SQL, full-text search, and vector search over it from a single system — search-on-Parquet.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204

// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The Infino Authors

//! [`ConnectOptions`] — storage + cache configuration the URI scheme
//! can't carry (credentials, region, endpoint, disk cache). Passed to
//! [`connect_with`](crate::connect_with); plain [`connect`](crate::connect)
//! uses the default.

use std::{collections::HashMap, path::PathBuf};

use crate::supertable::{Consistency, reader_cache::ColdFetchMode as InternalColdFetchMode};

/// How a disk-cache miss is serviced when reading cold superfiles from
/// object storage. The cache-servicing modes need a cache
/// ([`ConnectOptions::with_cache_dir`]); `RangeOnly` is the cacheless
/// path and does not currently use a disk cache.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum ColdFetchMode {
    /// Parallel range-GETs that tee into both the live query and the
    /// cache fill — 1× object-store bandwidth per cold miss.
    HybridWithPrefetch,
    /// Range-GETs straight through with no cache fill — best for
    /// query-once / stateless callers.
    RangeOnly,
    /// A lazy reader serves the query immediately (a few range-GETs);
    /// the full superfile is downloaded to the cache in the background.
    /// Lowest cold-query latency — the default.
    #[default]
    LazyForegroundWithBackgroundFill,
}

impl ColdFetchMode {
    pub(crate) fn to_internal(self) -> InternalColdFetchMode {
        match self {
            ColdFetchMode::HybridWithPrefetch => InternalColdFetchMode::HybridWithPrefetch,
            ColdFetchMode::RangeOnly => InternalColdFetchMode::RangeOnly,
            ColdFetchMode::LazyForegroundWithBackgroundFill => {
                InternalColdFetchMode::LazyForegroundWithBackgroundFill
            }
        }
    }
}

/// Storage configuration for [`connect_with`](crate::connect_with).
///
/// The storage **backend** is derived from the URI scheme passed to
/// `connect` (`s3://…`, `az://…`, `file://…`, `memory://`, or a bare
/// path), not from these options — `ConnectOptions` carries only what
/// the URI can't express. The common cases need no options:
/// `connect("./data")` and `connect("s3://bucket/prefix")` (ambient
/// cloud identity) both work with the default.
#[derive(Debug, Clone, Default)]
pub struct ConnectOptions {
    /// Credentials/tuning for the URI-selected backend, keyed by
    /// `object_store` config strings. Empty → ambient cloud identity.
    pub(crate) storage_options: HashMap<String, String>,
    /// Disk-cache root. `None` (default) → caching off; cold reads go
    /// straight to object storage. Set → a local NVMe tier under this
    /// directory, per table (`<cache_dir>/<table>`).
    pub(crate) cache_dir: Option<PathBuf>,
    /// Disk-cache byte budget. `None` → the cache's built-in default.
    /// Applies per table.
    pub(crate) cache_budget_bytes: Option<u64>,
    /// Cold-fetch strategy when the disk cache is enabled.
    pub(crate) cold_fetch_mode: ColdFetchMode,
    /// Per-connection memory (heap) budget in bytes. `None` (default) tracks
    /// usage without enforcing; `Some(n)` enforces a ceiling so one connection
    /// can't exhaust process memory. Applies to the whole connection, shared
    /// across supertables.
    pub(crate) connection_memory_budget_bytes: Option<u64>,
    /// Read-consistency policy for every table opened or created on this
    /// connection (see [`Consistency`]). Default:
    /// [`Consistency::BoundedStaleness`] with a 1s window — the engine default,
    /// which amortizes the per-query manifest-pointer re-check. Set
    /// [`Consistency::Strong`] for a pointer re-check on every query.
    pub(crate) read_consistency: Consistency,
    /// Probe the backend at `connect`. Default `false`; opt in for
    /// fail-fast on bad credentials.
    pub(crate) validate: bool,
    /// API key for a hosted (`https://…`) connect target, sent as a bearer
    /// credential on every request. Ignored by local (object-store) backends.
    /// When unset, a hosted connection falls back to the `INFINO_API_KEY`
    /// environment variable.
    pub(crate) api_key: Option<String>,
}

impl ConnectOptions {
    /// Default options — ambient credentials for object-store backends,
    /// disk cache off.
    pub fn new() -> Self {
        Self::default()
    }

    /// Enable a local disk cache rooted at `dir` (off by default). Cold
    /// superfile reads are cached to NVMe; per table, under
    /// `<dir>/<table>`. No effect on `memory://` catalogs.
    pub fn with_cache_dir(mut self, dir: impl Into<PathBuf>) -> Self {
        self.cache_dir = Some(dir.into());
        self
    }

    /// Set the disk-cache byte budget (per table). Defaults to the
    /// cache's built-in budget when unset. Only meaningful with
    /// [`with_cache_dir`](Self::with_cache_dir).
    pub fn with_cache_budget_bytes(mut self, bytes: u64) -> Self {
        self.cache_budget_bytes = Some(bytes);
        self
    }

    /// Choose how cold misses are serviced (see [`ColdFetchMode`]). Only
    /// meaningful with [`with_cache_dir`](Self::with_cache_dir).
    /// `RangeOnly` is the cacheless path and is rejected if a `cache_dir` is set.
    pub fn with_cold_fetch_mode(mut self, mode: ColdFetchMode) -> Self {
        self.cold_fetch_mode = mode;
        self
    }

    /// Set a per-connection memory budget, in bytes. Unset (the default)
    /// tracks usage without enforcing; a positive value enforces a ceiling so
    /// one connection can't exhaust process memory. Shared across all of the
    /// connection's tables.
    pub fn with_connection_memory_budget_bytes(mut self, bytes: u64) -> Self {
        self.connection_memory_budget_bytes = Some(bytes);
        self
    }

    /// Set one storage option (e.g. `aws_access_key_id`,
    /// `azure_storage_account_key`). An unknown or cross-backend key
    /// errors at connect time. Chainable.
    pub fn with_storage_option(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
        self.storage_options.insert(key.into(), value.into());
        self
    }

    /// Set the read-consistency policy for tables on this connection (see
    /// [`Consistency`]). Unset defaults to [`Consistency::BoundedStaleness`]
    /// with a 1s window (the engine default); pass [`Consistency::Strong`] to
    /// re-check the manifest pointer on every query. Chainable.
    pub fn with_read_consistency(mut self, consistency: Consistency) -> Self {
        self.read_consistency = consistency;
        self
    }

    /// Probe the object store at `connect` (default `false`). `true`
    /// fails fast on bad credentials instead of on first use.
    pub fn with_validate(mut self, validate: bool) -> Self {
        self.validate = validate;
        self
    }

    /// Set the API key for a hosted (`https://<host>/<db>`) connect target,
    /// sent as a bearer credential. Ignored by local backends. When unset, a
    /// hosted connection falls back to the `INFINO_API_KEY` environment
    /// variable. Chainable.
    pub fn with_api_key(mut self, key: impl Into<String>) -> Self {
        self.api_key = Some(key.into());
        self
    }

    /// The configured API key, if any (used by the hosted transport). Only the
    /// `remote` transport reads this, so it is dead code in a build without it.
    #[cfg_attr(not(feature = "remote"), allow(dead_code))]
    pub(crate) fn api_key(&self) -> Option<&str> {
        self.api_key.as_deref()
    }
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use super::*;

    #[test]
    fn read_consistency_defaults_to_bounded_staleness() {
        assert_eq!(
            ConnectOptions::new().read_consistency,
            Consistency::BoundedStaleness(Duration::from_secs(1)),
            "unset read consistency defaults to the engine's BoundedStaleness(1s)"
        );
    }

    #[test]
    fn with_read_consistency_overrides_the_default() {
        let opts = ConnectOptions::new().with_read_consistency(Consistency::Strong);
        assert_eq!(opts.read_consistency, Consistency::Strong);
    }

    #[test]
    fn with_storage_option_round_trips() {
        let o = ConnectOptions::new().with_storage_option("aws_region", "us-east-1");
        assert_eq!(
            o.storage_options.get("aws_region").map(String::as_str),
            Some("us-east-1")
        );
    }

    #[test]
    fn with_api_key_round_trips() {
        let o = ConnectOptions::new().with_api_key("ik_test");
        assert_eq!(o.api_key(), Some("ik_test"));
        assert_eq!(ConnectOptions::new().api_key(), None);
    }
}