### Running the WAL Writer Loop and Asserting Results in Rust Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/wal/writer.rs.html This example demonstrates the setup and execution of the main WAL writer loop. It includes initializing the writer with configuration, running it until stalled, and asserting the final results and captured snapshot. ```rust let result = run_writer_loop::( WriterLoopInit { exec: Arc::new(NoopExecutor::default()), storage, cfg, metrics: Arc::clone(&metrics), queue_depth: queue_depth_writer, initial_segment_seq: 0, initial_frame_seq: frame::INITIAL_FRAME_SEQ, }, receiver, ) .await; *result_cell_clone.borrow_mut() = Some(result); ``` -------------------------------- ### Example: Scan with Predicate and Limit Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/db/scan.rs.html Demonstrates setting up an in-memory Tonbo DB, ingesting data, and performing a scan with a filter and a limit. This example requires Tokio and Arrow crates. ```rust use std::sync::Arc; use arrow_array::{Int32Array, RecordBatch, StringArray}; use arrow_schema::{DataType, Field, Schema}; use fusio::executor::tokio::TokioExecutor; use tonbo:: ColumnRef, Predicate, ScalarValue, db::{DB, DbBuilder}, schema::SchemaBuilder, }; #[tokio::main] async fn main() -> Result<(), Box> { let schema = Arc::new(Schema::new(vec![ Field::new("id", DataType::Utf8, false), Field::new("value", DataType::Int32, false), ])); let config = SchemaBuilder::from_schema(Arc::clone(&schema)) .primary_key("id") .build()?; // In-memory DB; use DbBuilder::on_disk/object_store for durable backends. let db: DB = DbBuilder::from_schema_key_name(schema, "id")? .in_memory("scan-example")? .open_with_executor(Arc::new(TokioExecutor::default())) .await?; // Seed data let batch = RecordBatch::try_new( Arc::clone(&config.schema()), vec![ Arc::new(StringArray::from(vec!["a", "b"])) as _, Arc::new(Int32Array::from(vec![1, 2])) as _, ], )?; db.ingest(batch).await?; // Scan with predicate + limit let pred = Predicate::eq(ColumnRef::new("id"), ScalarValue::from("a")); let batches = db.scan().filter(pred).limit(1).collect().await?; assert_eq!(batches.iter().map(|b| b.num_rows()).sum::(), 1); Ok(()) } ``` -------------------------------- ### Example: Building an In-Memory DB Source: https://docs.rs/tonbo/0.4.0-a0/tonbo/db/struct.DbBuilder.html Demonstrates how to create an in-memory database instance using DbBuilder, specifying a schema and key name. ```rust use std::sync::Arc; use arrow_schema::{DataType, Field, Schema}; use fusio::{executor::tokio::TokioExecutor, mem::fs::InMemoryFs}; use tonbo::{ db::{DB, DbBuilder}, schema::SchemaBuilder, }; #[tokio::main] async fn main() -> Result<(), Box> { let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Utf8, false)])); let _db: DB = DbBuilder::from_schema_key_name(schema, "id")? .in_memory("builder-example")? .build() .await?; Ok(()) } ``` -------------------------------- ### Initialize Manifest and Context Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/manifest/driver.rs.html Initializes a new LeaseStore, a ManifestContext, and opens a Manifest instance. This is a setup step for manifest operations. ```rust let lease = LeaseStoreImpl::new(fs, "catalog/leases", BackoffPolicy::default(), timer); let context = Arc::new(ManifestContext::new(DefaultExecutor::default())); Manifest::open(Stores::new(head, segment, checkpoint, lease), context) ``` -------------------------------- ### Example Usage of DB Source: https://docs.rs/tonbo/0.4.0-a0/tonbo/db/struct.DB.html Demonstrates creating a DB instance using the builder API, cloning it, and beginning a transaction. Requires TokioExecutor and InMemoryFs. ```rust use std::sync::Arc; use arrow_schema::{DataType, Field, Schema}; use fusio::{executor::tokio::TokioExecutor, mem::fs::InMemoryFs}; use tonbo::{db::DB, schema::SchemaBuilder}; #[tokio::main] async fn main() -> Result<(), Box> { let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Utf8, false)])); let config = SchemaBuilder::from_schema(Arc::clone(&schema)) .primary_key("id") .build()?; let db: DB = DB::::builder(config) .in_memory("example-db")? .build() .await?; // Clone is cheap (just Arc clone) let _db2 = db.clone(); // Begin a transaction let _tx = db.begin_transaction().await?; Ok(()) } ``` -------------------------------- ### DB::builder Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/db/mod.rs.html Starts the construction of a new database instance using a fluent builder API. ```APIDOC ## DB::builder ### Description Begins constructing a DB through the fluent builder API. ### Method `fn builder(config: DynModeConfig) -> DbBuilder` ### Parameters #### Query Parameters - **config** (`DynModeConfig`): The configuration for the database mode. ### Response - `DbBuilder`: A new `DbBuilder` instance to configure and create the database. ``` -------------------------------- ### Build Database with In-Memory Storage Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/db/builder.rs.html Example demonstrating how to build a database using an in-memory filesystem. This is useful for testing or temporary data storage. ```rust use std::sync::Arc; use arrow_schema::{DataType, Field, Schema}; use fusio::{executor::tokio::TokioExecutor, mem::fs::InMemoryFs}; use tonbo:: db::{DB, DbBuilder}, schema::SchemaBuilder, ; #[tokio::main] async fn main() -> Result<(), Box> { let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Utf8, false)])); let _db: DB = DbBuilder::from_schema_key_name(schema, "id")? .in_memory("builder-example")? .build() .await?; Ok(()) } ``` -------------------------------- ### Example: Creating a User Database with DbBuilder Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/db/builder.rs.html Demonstrates how to use `DbBuilder::from_schema` to create a database for a `User` struct, specifying the storage path and opening the database asynchronously. ```rust use tonbo::{db::DbBuilder, typed_arrow::{Record, schema::SchemaMeta}}; #[derive(Record)] struct User { #[metadata(k = "tonbo.key", v = "true")] id: String, name: String, } let db = DbBuilder::from_schema(User::schema())? .on_disk("/tmp/users")? .open() .await?; ``` -------------------------------- ### begin_snapshot Source: https://docs.rs/tonbo/0.4.0-a0/tonbo/db/struct.DB.html Starts a read-only snapshot for executing queries. This provides a consistent view of the database at a specific point in time. ```APIDOC ## pub async fn begin_snapshot(&self) -> Result ### Description Begin a read-only snapshot for queries. ### Method `async fn begin_snapshot(&self)` ### Parameters None ### Returns A `Result` containing a `TxSnapshot` on success or a `SnapshotError` on failure. ``` -------------------------------- ### Example: Creating and Executing a Scan Query Source: https://docs.rs/tonbo/0.4.0-a0/tonbo/db/struct.ScanBuilder.html Demonstrates how to create an in-memory database, ingest data, and perform a scan with a filter and limit using ScanBuilder. ```rust use std::sync::Arc; use arrow_array::{Int32Array, RecordBatch, StringArray}; use arrow_schema::{DataType, Field, Schema}; use fusio::{executor::tokio::TokioExecutor, mem::fs::InMemoryFs}; use tonbo:: ColumnRef, Predicate, ScalarValue, db::{DB, DbBuilder}, schema::SchemaBuilder, }; #[tokio::main] async fn main() -> Result<(), Box> { let schema = Arc::new(Schema::new(vec![ Field::new("id", DataType::Utf8, false), Field::new("value", DataType::Int32, false), ])); let config = SchemaBuilder::from_schema(Arc::clone(&schema)) .primary_key("id") .build()?; // In-memory DB; use DbBuilder::on_disk/object_store for durable backends. let db: DB = DbBuilder::from_schema_key_name(schema, "id")? .in_memory("scan-example")? .open_with_executor(Arc::new(TokioExecutor::default())) .await?; // Seed data let batch = RecordBatch::try_new( Arc::clone(&config.schema()), vec![ Arc::new(StringArray::from(vec!["a", "b"])) as _, Arc::new(Int32Array::from(vec![1, 2])) as _, ], )?; db.ingest(batch).await?; // Scan with predicate + limit let pred = Predicate::eq(ColumnRef::new("id"), ScalarValue::from("a")); let batches = db.scan().filter(pred).limit(1).collect().await?; assert_eq!(batches.iter().map(|b| b.num_rows()).sum::(), 1); Ok(()) } ``` -------------------------------- ### Start DB Builder Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/db/mod.rs.html Initializes the fluent builder API for constructing a new database instance. Use this to configure and create a DB with specific settings. ```rust pub fn builder(config: DynModeConfig) -> DbBuilder { DbBuilder::new(config) } ``` -------------------------------- ### Setup for Tombstone Scan Test Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/ondisk/scan.rs.html Initializes temporary directory, filesystem, and schemas for testing tombstone behavior. Writes data and delete sidecar Parquet files. ```rust let tmpdir = tempdir().expect("tempdir"); let fs: Arc = Arc::new(LocalFs {}); let root = Path::from(tmpdir.path().to_string_lossy().to_string()); // User schema for data extractor let user_schema = Arc::new(Schema::new(vec![ Field::new("id", DataType::Utf8, false), Field::new("v", DataType::Int32, false), ])); let data_extractor = crate::extractor::projection_for_field(Arc::clone(&user_schema), 0).expect("extractor"); // Key-only schema for delete extractor let key_schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Utf8, false)])); let delete_extractor = crate::extractor::projection_for_field(Arc::clone(&key_schema), 0) .expect("delete extractor"); // Write data with only 'b' let data_path = root.child("data.parquet"); write_data_parquet( Arc::clone(&fs), data_path.clone(), vec![("b", 2, 20)], // only data row ) .await; // Write delete sidecar with tombstones for 'a', 'c', 'd' let delete_path = root.child("delete.parquet"); write_delete_parquet( Arc::clone(&fs), delete_path.clone(), vec![ ("a", 10), // tombstone-only (comes before 'b') ("c", 30), // tombstone-only (comes after 'b') ("d", 40), // tombstone-only (comes after 'b') ], ) .await; ``` -------------------------------- ### Initialize SchemaBuilder from Arrow Schema Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/schema.rs.html Start building a schema configuration by providing an Arrow schema reference. This is the initial step before declaring key fields. ```rust use std::sync::Arc; use arrow_schema::{Schema, SchemaRef}; use crate::mode::DynModeConfig; use crate::extractor::KeyExtractError; // Assuming SchemaBuilder and other necessary imports are available let schema: SchemaRef = Arc::new(Schema::new(vec![])); // Example schema let builder = SchemaBuilder::from_schema(schema); ``` -------------------------------- ### Start Scan Query Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/db/mod.rs.html Begins the construction of a scan query. Use the returned ScanBuilder to define the criteria for retrieving data from the database. ```rust pub fn scan(&self) -> ScanBuilder<'_, FS, E> { ScanBuilder::new(&self.inner) } ``` -------------------------------- ### Get Prefers Staged Rows Over Ingested Rows Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/transaction/mod.rs.html Demonstrates that a 'get' operation within a transaction will retrieve staged (upserted) rows instead of previously ingested rows with the same key. ```rust async fn get_prefers_staged_rows() { let (db, schema) = make_db().await; ingest_rows( &db, &schema, vec![DynRow(vec![ Some(DynCell::Str("k1".into())), Some(DynCell::I32(10)), ])], ) .await; let mut tx: TestTx = db.begin_transaction().await.expect("tx"); tx.upsert(DynRow(vec![ Some(DynCell::Str("k1".into())), Some(DynCell::I32(42)), ])) .expect("stage"); let value = tx .get(&KeyOwned::from("k1")) .await .expect("get") .expect("row"); match &value.0[1] { Some(DynCell::I32(v)) => assert_eq!(*v, 42), other => panic!("unexpected cell: {other:?}"), } } ``` -------------------------------- ### WalStorage Initialization and Segment Opening Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/wal/storage.rs.html Demonstrates the initialization of WalStorage with a DynFs and a root path, followed by ensuring the directory exists and opening a specific segment. This is part of a test setup. ```rust #[tokio::test(flavor = "multi_thread")] async fn list_segments_only_falls_back_with_hint() { let base = Arc::new(InMemoryFs::new()); let base_dyn: Arc = base.clone(); let root = Path::parse("wal-s3").expect("root path"); let instrumented = Arc::new(S3ProbeFs::new(base_dyn, root.clone())); let fs_dyn: Arc = instrumented.clone(); let storage = WalStorage::new(fs_dyn, root.clone()); storage .ensure_dir(storage.root()) .await .expect("ensure dir"); let mut segment = storage.open_segment(7).await.expect("open segment"); ``` -------------------------------- ### Get a row with staged mutations applied Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/transaction/mod.rs.html Use `get` to retrieve the latest visible row for a given key, considering staged mutations. If a staged mutation is a delete, `None` is returned. Otherwise, staged upserts are returned. ```rust pub async fn get(&self, key: &KeyOwned) -> Result, TransactionError> { if let Some(mutation) = self.staged.get(key) { return Ok(match mutation { DynMutation::Upsert(row) => Some(clone_dyn_row(row)), DynMutation::Delete(_) => None, }); } let rows = self.read_mutable_rows(&*self.handle)?; Ok(rows.get(key).map(clone_dyn_row)) } ``` -------------------------------- ### deref Source: https://docs.rs/tonbo/0.4.0-a0/tonbo/db/struct.ScanBuilder.html Dereferences the given pointer to get a reference to the object. ```APIDOC ## unsafe fn deref<'a>(ptr: usize) -> &'a T ### Description Dereferences the given pointer. ### Method (Not specified, likely a method call) ### Parameters - **ptr** (usize) - Description: The pointer to dereference. ``` -------------------------------- ### Get Commit Timestamp Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/manifest/domain.rs.html Returns the commit timestamp of the manifest. ```rust pub(crate) fn commit_timestamp(&self) -> Timestamp { self.commit_timestamp } ``` -------------------------------- ### DynModeConfig Get Schema Source: https://docs.rs/tonbo/0.4.0-a0/tonbo/db/struct.DynModeConfig.html Clones and returns the schema associated with this DynModeConfig. ```rust pub fn schema(&self) -> SchemaRef ``` -------------------------------- ### Begin Read-Only Snapshot Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/db/mod.rs.html Starts a read-only snapshot of the database. This allows for querying the database state at a specific point in time without affecting ongoing transactions. ```rust pub async fn begin_snapshot(&self) -> Result { self.inner.begin_snapshot().await } ``` -------------------------------- ### Get ScanStream Priority Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/query/stream/mod.rs.html Returns the SourcePriority associated with each variant of the ScanStream. ```rust impl<'t, E: Executor> ScanStream<'t, E> { pub(crate) fn priority(&self) -> SourcePriority { match self { ScanStream::Txn { .. } => SourcePriority::Txn, ScanStream::Mutable { .. } => SourcePriority::Mutable, ScanStream::Immutable { .. } => SourcePriority::Immutable, ScanStream::SsTable { .. } => SourcePriority::Sstable, } } } ``` -------------------------------- ### Initializing and Planning with LeveledCompactionPlanner Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/compaction/planner.rs.html Demonstrates how to create a LeveledCompactionPlanner with specific configuration and use it to plan a compaction task based on a given snapshot. Asserts that the generated task has no key range. ```rust let snapshot = CompactionSnapshot::new(vec![level1]); let planner = LeveledCompactionPlanner::new(LeveledPlannerConfig { l0_trigger: usize::MAX, l0_max_inputs: 0, level_thresholds: vec![0], max_inputs_per_task: 1, l0_max_bytes: None, level_max_bytes: Vec::new(), max_task_bytes: None, }); let task = planner.plan(&snapshot).expect("plan"); assert!(task.key_range.is_none()); ``` -------------------------------- ### deref_mut Source: https://docs.rs/tonbo/0.4.0-a0/tonbo/db/struct.ScanBuilder.html Mutably dereferences the given pointer to get a mutable reference to the object. ```APIDOC ## unsafe fn deref_mut<'a>(ptr: usize) -> &'a mut T ### Description Mutably dereferences the given pointer. ### Method (Not specified, likely a method call) ### Parameters - **ptr** (usize) - Description: The pointer to mutably dereference. ``` -------------------------------- ### Get KeyRow Length Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/key/row.rs.html Returns the number of components contained within the KeyRow. ```rust pub fn len(&self) -> usize { self.raw.len() } ``` -------------------------------- ### Initialize WalStorage Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/wal/storage.rs.html Creates a new `WalStorage` instance with a specific filesystem and root directory. ```rust pub(crate) fn new(fs: Arc, root: Path) -> Self { Self { fs, root } } ``` -------------------------------- ### DB::scan Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/db/mod.rs.html Starts building a scan query to retrieve data from the database. ```APIDOC ## DB::scan ### Description Start building a scan query. ### Method `fn scan(&self) -> ScanBuilder<'_, FS, E>` ### Parameters None ### Response - `ScanBuilder<'_, FS, E>`: A builder for constructing scan queries. ``` -------------------------------- ### Get WAL Segments Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/manifest/domain.rs.html Returns a slice of the WAL segment references associated with the manifest. ```rust pub(crate) fn wal_segments(&self) -> &[WalSegmentRef] { &self.wal_segments } ``` -------------------------------- ### Initialize Filesystem Manifest Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/manifest/bootstrap.rs.html Constructs a manifest rooted under a specified path using a provided filesystem backend. Ensures necessary directories are created. ```rust pub(crate) async fn init_fs_manifest( fs: FS, root: &Path, executor: E, ) -> ManifestResult> where FS: super::ManifestFs, E: Executor + Timer + Clone + 'static, HeadStoreImpl: fusio_manifest::HeadStore, SegmentStoreImpl: fusio_manifest::SegmentIo, CheckpointStoreImpl: fusio_manifest::CheckpointStore, LeaseStoreImpl: fusio_manifest::LeaseStore, ::File: fusio::durability::FileCommit, { let manifest_root = append_segment(root, "manifest"); let version_root = append_segment(&manifest_root, "version"); let catalog_root = append_segment(&manifest_root, "catalog"); let gc_root = append_segment(&manifest_root, "gc"); ensure_manifest_dirs::(&version_root).await?; ensure_manifest_dirs::(&catalog_root).await?; ensure_manifest_dirs::(&gc_root).await?; let version_manifest = open_manifest_instance::(fs.clone(), &version_root, executor.clone()); let catalog_manifest = open_manifest_instance::(fs.clone(), &catalog_root, executor.clone()); let gc_manifest = open_manifest_instance::(fs.clone(), &gc_root, executor); let tonbo = TonboManifest::new(version_manifest, catalog_manifest, gc_manifest); tonbo.init_catalog().await?; Ok(tonbo) } ``` -------------------------------- ### Initialize WAL Storage Configuration Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/wal/writer.rs.html Sets up the WAL configuration, including queue size, sync policy, segment backend, and state store. ```rust let mut cfg = WalConfig::default(); cfg.queue_size = 2; cfg.sync = WalSyncPolicy::Always; cfg.segment_backend = fs_writer; let state_store: Arc = Arc::new(FsWalStateStore::new(fs_cas)); cfg.state_store = Some(Arc::clone(&state_store)); ``` -------------------------------- ### Get SSTables Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/manifest/domain.rs.html Returns a slice of all SSTable entries stored in the manifest, organized by level. ```rust pub(crate) fn ssts(&self) -> &[Vec] { &self.ssts } ``` -------------------------------- ### Initialize and Use Tonbo DB Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/db/mod.rs.html Demonstrates how to build and use the Tonbo DB with an in-memory file system and a Tokio executor. Shows schema creation, primary key definition, and transaction initiation. ```rust use std::sync::Arc; use arrow_schema::{DataType, Field, Schema}; use fusio::{executor::tokio::TokioExecutor, mem::fs::InMemoryFs}; use tonbo::{db::DB, schema::SchemaBuilder}; #[tokio::main] async fn main() -> Result<(), Box> { let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Utf8, false)])); let config = SchemaBuilder::from_schema(Arc::clone(&schema)) .primary_key("id") .build()?; let db: DB = DB::::builder(config) .in_memory("example-db")? .build() .await?; // Clone is cheap (just Arc clone) let _db2 = db.clone(); // Begin a transaction let _tx = db.begin_transaction().await?; Ok(()) } ``` -------------------------------- ### Get Predicate Node Source: https://docs.rs/tonbo/0.4.0-a0/tonbo/db/struct.Predicate.html Retrieves a reference to the underlying PredicateNode of a Predicate. This is useful for inspecting the internal structure. ```rust pub fn kind(&self) -> &PredicateNode ``` -------------------------------- ### begin_snapshot Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/db/mod.rs.html Opens a read-only snapshot pinned to the current manifest head. This provides a consistent view of the database at the latest committed state. ```APIDOC ## begin_snapshot ### Description Opens a read-only snapshot pinned to the current manifest head. This provides a consistent view of the database at the latest committed state. ### Method `async fn` ### Endpoint N/A (SDK method) ### Parameters None ### Response #### Success Response - `Result`: A `TxSnapshot` representing the current database state, or a `SnapshotError` if an error occurred. #### Response Example ```json // Success response is a TxSnapshot object // Error response is a SnapshotError object ``` ``` -------------------------------- ### Get Alignment for S3Spec Pointer Source: https://docs.rs/tonbo/0.4.0-a0/tonbo/db/struct.S3Spec.html Defines the alignment requirement for pointers to S3Spec. This is part of the Pointable trait. ```rust const ALIGN: usize ``` -------------------------------- ### Initialize StorageLayout Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/db/builder.rs.html Creates a new StorageLayout instance with the provided filesystem, optional CAS, and root path. ```rust let dyn_fs: Arc = fs.clone(); Self { fs, dyn_fs, cas, root, } ``` -------------------------------- ### Access WAL State Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/wal/state.rs.html Provides methods to get a reference to the current WAL state or a mutable reference. ```rust /// Access the current state snapshot. #[inline] pub fn state(&self) -> &WalState { &self.state } /// Mutably access the state snapshot. #[inline] pub fn state_mut(&mut self) -> &mut WalState { &mut self.state } ``` -------------------------------- ### Bare Manifest Initialization Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/manifest/driver.rs.html Sets up a minimal in-memory manifest instance for testing purposes, including necessary stores and context. ```rust fn bare_manifest() -> TestManifest { let fs = InMemoryFs::new(); let head = HeadStoreImpl::new(fs.clone(), "head.json"); let segment = SegmentStoreImpl::new(fs.clone(), "segments"); let checkpoint = CheckpointStoreImpl::new(fs.clone(), ""); let timer = DefaultExecutor::default(); let lease = LeaseStoreImpl::new(fs, "", BackoffPolicy::default(), timer); let context = Arc::new(ManifestContext::new(DefaultExecutor::default())); Manifest::open(Stores::new(head, segment, checkpoint, lease), context) } ``` -------------------------------- ### Get Table ID Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/manifest/domain.rs.html Returns a reference to the table ID. This function is only available when the 'tokio' feature is enabled. ```rust #[cfg(all(test, feature = "tokio"))] pub(crate) fn table_id(&self) -> &TableId { &self.table_id } ``` -------------------------------- ### WAL Replay with Absolute Consistency Mode Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/wal/replay.rs.html Demonstrates setting up and scanning with WAL recovery in AbsoluteConsistency mode. This mode is currently unimplemented and expected to return an error. ```rust { let mut cfg = WalConfig::default(); cfg.segment_backend = fs_dyn; cfg.state_store = Some(Arc::new(FsWalStateStore::new(fs_cas))); cfg.recovery = WalRecoveryMode::AbsoluteConsistency; let replayer = Replayer::new(cfg); let err = replayer.scan().await.expect_err("mode unimplemented"); assert!(matches!( err, WalError::Unimplemented("wal recovery mode AbsoluteConsistency is not implemented") )); } ``` -------------------------------- ### SchemaBuilder::from_schema Source: https://docs.rs/tonbo/0.4.0-a0/tonbo/schema/struct.SchemaBuilder.html Initializes a SchemaBuilder from an Arrow schema reference. This is the starting point for defining primary keys. ```APIDOC ## SchemaBuilder::from_schema ### Description Start a builder from an Arrow schema reference. ### Method `SchemaBuilder::from_schema` ### Parameters * `schema` (SchemaRef) - The Arrow schema reference to build from. ``` -------------------------------- ### builder Source: https://docs.rs/tonbo/0.4.0-a0/tonbo/db/struct.DB.html Initiates the construction of a DB instance using a fluent builder API, starting with a provided configuration. ```APIDOC ## pub fn builder(config: DynModeConfig) -> DbBuilder ### Description Begin constructing a DB through the fluent builder API. ### Method `fn builder(config: DynModeConfig)` ### Parameters * **config** (DynModeConfig) - The configuration for the database mode. ### Returns A `DbBuilder` instance to continue building the DB. ``` -------------------------------- ### SSTable Configuration Builder Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/ondisk/sstable.rs.html Shows how to construct an SsTableConfig with essential parameters like schema, file system, and root path, and how to customize it further. ```rust impl SsTableConfig { /// Build a new configuration bundle for an SSTable. pub fn new(schema: SchemaRef, fs: Arc, root: Path) -> Self { Self { schema, target_level: 0, compression: SsTableCompression::default(), fs, root, key_extractor: None, merge_iteration_budget: DEFAULT_MERGE_ITERATION_BUDGET, } } /// Set the target compaction level for the table being produced. pub fn with_target_level(mut self, level: usize) -> Self { self.target_level = level; self } /// Choose the compression strategy applied to persisted pages. pub fn with_compression(mut self, compression: SsTableCompression) -> Self { self.compression = compression; self } /// Access the Arrow schema used by this table. pub fn schema(&self) -> &SchemaRef { &self.schema } /// Access the target level hint used when enqueuing this table into the version set. pub fn target_level(&self) -> usize { self.target_level } /// Access the configured compression scheme. pub fn compression(&self) -> SsTableCompression { self.compression } /// Access the Parquet-backed store used to create or read tables. pub fn fs(&self) -> &Arc { &self.fs } /// Root directory prefix for all SSTable files. pub fn root(&self) -> &Path { &self.root } /// Build a level-scoped path under the SST root. pub(crate) fn level_dir(&self, level: usize) -> Result { let level_name = format!("L{level}"); let part = PathPart::parse(&level_name) ``` -------------------------------- ### Initialize Replayer with Configuration Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/wal/replay.rs.html Creates a new Replayer instance. It requires a WalConfig which includes the segment backend and directory. The storage facade is initialized internally. ```rust pub fn new(cfg: WalConfig) -> Self { let storage = WalStorage::new(Arc::clone(&cfg.segment_backend), cfg.dir.clone()); Self { cfg, storage } } ``` -------------------------------- ### Get Table Metadata Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/manifest/driver.rs.html Retrieves the metadata for a specific table from the catalog. Handles cases where metadata might be missing. ```rust pub(crate) async fn table_meta(&self, table: TableId) -> ManifestResult { let session = self.inner.session_read().await?; let key = CatalogKey::TableMeta { table_id: table }; let value = match session.get(&key).await? { Some(value) => value, None => { session.end().await?; return Err(ManifestError::Invariant( "catalog metadata missing for table_id", )); } }; ::validate_key_value(&key, &value)?; let meta = TableMeta::try_from(value)?; session.end().await?; Ok(meta) } ``` -------------------------------- ### Initialize Local File System Manifest Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/manifest/bootstrap.rs.html Initializes a manifest on the local file system using the provided path and executor. This is a smoke test to ensure basic initialization works. ```rust #[tokio::test(flavor = "multi_thread")] async fn init_disk_manifest_smoke() { let dir = tempfile::tempdir().expect("temp dir"); let path = FusioPath::from_filesystem_path(dir.path()).expect("fs path"); init_fs_manifest(LocalFs {}, &path, TokioExecutor::default()) .await .expect("disk manifest"); } ``` -------------------------------- ### Get WAL Configuration Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/db/builder.rs.html Retrieves a reference to the WAL configuration if present. Returns None if WAL is not configured or not applicable. ```rust pub fn wal_config(&self) -> Option<&WalConfig> { self.wal_config.as_ref() } ``` -------------------------------- ### Get Table Name Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/db/builder.rs.html Retrieves a reference to the table name if it has been set. Returns None if no table name is configured. ```rust pub fn table_name(&self) -> Option<&String> { self.table_name.as_ref() } ``` -------------------------------- ### Building a Concrete Compaction Planner Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/compaction/planner.rs.html The `build` method on `CompactionStrategy` constructs a concrete `CompactionPlannerKind` based on the chosen strategy. ```rust impl CompactionStrategy { /// Build a concrete planner for the selected strategy. pub(crate) fn build(self) -> CompactionPlannerKind { match self { Self::Leveled(cfg) => { CompactionPlannerKind::Leveled(LeveledCompactionPlanner::new(cfg)) } } } } ``` -------------------------------- ### Get Batch Count Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/inmem/mutable/memtable.rs.html Returns the current number of batches being processed in the memtable. This is a simple counter for internal tracking. ```rust #[cfg(all(test, feature = "tokio"))] pub(crate) fn batch_count(&self) -> usize { self.batch_cursor.load(Ordering::Relaxed) } ``` -------------------------------- ### Manifest::open Constructor Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/manifest/driver.rs.html Constructs a new manifest wrapper by opening it with the provided storage stores and manifest context. This method initializes the inner fusio-manifest. ```rust impl Manifest where C: ManifestCodec, HS: HeadStore + MaybeSend + MaybeSync + 'static, SS: SegmentIo + MaybeSend + MaybeSync + 'static, CS: CheckpointStore + MaybeSend + MaybeSync + 'static, LS: LeaseStore + MaybeSend + MaybeSync + 'static, E: Executor + Timer + Clone + 'static, { /// Construct a new manifest wrapper from the provided stores and context. #[must_use] pub(super) fn open( stores: Stores, ctx: Arc>, ) -> Self { Self { inner: FusioManifest::new_with_context( stores.head, stores.segment, stores.checkpoint, stores.lease, ctx, ), } } } ``` -------------------------------- ### Getting Table ID Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/db/mod.rs.html Retrieves the Table ID registered in the manifest for this database. This is typically used in test scenarios. ```rust /// Table ID registered in the manifest for this DB. #[cfg(test)] pub fn table_id(&self) -> TableId { self.manifest_table } ``` -------------------------------- ### Get Cloned WAL Floor Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/manifest/domain.rs.html Returns a cloned optional WAL segment reference representing the floor of the WAL. ```rust pub(crate) fn cloned_wal_floor(&self) -> Option { self.wal_floor.clone() } ``` -------------------------------- ### Create New Storage Configuration Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/db/builder.rs.html Initializes a new storage configuration. WAL configuration is automatically set for durable backends. ```rust pub fn new(fs: Arc, root: Path, durability: DurabilityClass) -> Self { let wal_config = if durability.is_durable() { Some(WalConfig::default()) } else { None }; Self { fs, root, table_name: None, wal_config, durability, create_layout: false, } } ``` -------------------------------- ### Create new DbBuilder Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/db/builder.rs.html Creates a new instance of DbBuilder with an Unconfigured state. This is the starting point for configuring the database. ```rust pub(super) fn new(mode_config: DynModeConfig) -> Self { Self { mode_config, state: Unconfigured, compaction_strategy: CompactionStrategy::default(), compaction_interval: None, compaction_loop_cfg: None, minor_compaction: None, } } ``` -------------------------------- ### Initialize In-Memory WAL Environment Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/wal/writer.rs.html Sets up an in-memory WAL storage and configuration. This is useful for testing WAL functionality without interacting with the actual file system. ```rust fn in_memory_env( queue_size: usize, sync: WalSyncPolicy, root: &str, ) -> (WalStorage, WalConfig) { let backend = Arc::new(InMemoryFs::new()); let fs_dyn: Arc = backend.clone(); let fs_cas: Arc = backend.clone(); let storage = WalStorage::new(Arc::clone(&fs_dyn), Path::parse(root).expect("path")); let mut cfg = WalConfig::default(); cfg.queue_size = queue_size; cfg.sync = sync; cfg.segment_backend = fs_dyn; cfg.state_store = Some(Arc::new(FsWalStateStore::new(fs_cas))); (storage, cfg) } ``` -------------------------------- ### txn_begin Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/wal/mod.rs.html Logs a transaction begin marker with a given provisional ID. This initiates a new transaction in the WAL. ```APIDOC ## txn_begin ### Description Log a transaction begin marker. ### Method `async fn txn_begin(&self, provisional_id: u64) -> WalResult>` ### Parameters #### Path Parameters - **provisional_id** (u64) - Required - The provisional ID for the transaction. ### Response #### Success Response - **WalTicket** : A ticket representing the submitted transaction command. ### Response Example ```json { "example": "WalTicket object" } ``` ``` -------------------------------- ### Drain Attached Batches Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/inmem/mutable/memtable.rs.html Retrieves and returns any attached batches from the MemTable. This example asserts that exactly one batch was drained. ```rust // Drain attached batches let drained = m.into_attached_batches(); assert_eq!(drained.len(), 1); ``` -------------------------------- ### Get Write-Ahead Log (WAL) Handle Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/wal/mod.rs.html Returns an optional reference to the WAL handle. This can be used to interact with the WAL after it has been initialized. ```rust fn wal(&self) -> Option<&WalHandle> { self.wal_handle() } ``` -------------------------------- ### Initialize Write-Ahead Log (WAL) Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/wal/mod.rs.html Initializes the WAL by creating storage, loading its state, determining the next sequence numbers for payloads and frames, and spawning a writer task. This is used when setting up the WAL for the first time or resuming from a previous state. ```rust let storage = storage::WalStorage::new(Arc::clone(&cfg.segment_backend), cfg.dir.clone()); let wal_state_handle = storage.load_state_handle(cfg.state_store.as_ref()).await?; let wal_state_hint = wal_state_handle .as_ref() .and_then(|handle| handle.state().last_segment_seq); let tail_metadata = storage.tail_metadata_with_hint(wal_state_hint).await?; let next_payload_seq = tail_metadata .as_ref() .and_then(|meta| meta.last_provisional_id) .map(|last| last.saturating_add(1)) .unwrap_or(frame::INITIAL_FRAME_SEQ); let initial_frame_seq = tail_metadata .as_ref() .and_then(|meta| meta.last_frame_seq) .map(|seq| seq.saturating_add(1)) .unwrap_or(frame::INITIAL_FRAME_SEQ); let metrics = Arc::new(E::rw_lock(WalMetrics::default())); let writer = writer::spawn_writer( Arc::clone(self.executor()), storage, cfg.clone(), Arc::clone(&metrics), 0, initial_frame_seq, ); let (sender, queue_depth, join) = writer.into_parts(); let handle = WalHandle::from_parts( sender, queue_depth, join, next_payload_seq, Arc::clone(&metrics), ); self.set_wal_config(Some(cfg.clone())); self.set_wal_handle(Some(handle.clone())); Ok(handle) ``` -------------------------------- ### Get Number of Rows in DynBatchPayload Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/wal/mod.rs.html Retrieves the number of rows from a `DynBatchPayload`, whether it's a row batch or a delete batch. ```rust impl DynBatchPayload { /// Number of rows carried by the batch. pub fn num_rows(&self) -> usize { match self { DynBatchPayload::Row { batch, .. } | DynBatchPayload::Delete { batch } => { batch.num_rows() } } } } ``` -------------------------------- ### Create New Compaction Driver Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/compaction/driver.rs.html Initializes a new CompactionDriver with the provided manifest, table ID, WAL configuration, and WAL handle. This is the entry point for creating a driver instance. ```rust pub(crate) fn new( manifest: TonboManifest, table_id: TableId, wal_config: Option, wal_handle: Option>, ) -> Self { Self { manifest, table_id, wal_config, wal_handle, } } ``` -------------------------------- ### Get the durability class of a transaction Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/transaction/mod.rs.html Retrieve the `TransactionDurability` setting for the current transaction. This indicates how the transaction's writes are persisted. ```rust pub fn durability(&self) -> TransactionDurability { self.durability } ``` -------------------------------- ### Initialize and Run WAL Writer Loop Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/wal/writer.rs.html Initializes and runs the WAL writer loop within a local pool. This snippet demonstrates setting up the writer with necessary configurations, metrics, and channels, then spawning it as a local task. It also includes the logic for receiving acknowledgments and asserting the final state. ```rust let (mut sender, receiver) = mpsc::channel(cfg.queue_size); let queue_depth = Arc::new(AtomicUsize::new(0)); let queue_depth_writer = Arc::clone(&queue_depth); let result_cell: Rc>>> = Rc::new(RefCell::new(None)); let result_cell_clone = Rc::clone(&result_cell); let mut pool = LocalPool::new(); let spawner = pool.spawner(); spawner .spawn_local(async move { let result = run_writer_loop::( WriterLoopInit { exec: Arc::new(NoopExecutor::default()), storage, cfg, metrics: Arc::clone(&metrics), queue_depth: queue_depth_writer, initial_segment_seq: 0, initial_frame_seq: frame::INITIAL_FRAME_SEQ, }, receiver, ) .await; *result_cell_clone.borrow_mut() = Some(result); }) .expect("spawn writer"); let (append_rx, commit_rx) = queue_autocommit( &mut sender, &queue_depth, 99, 100, sample_commands(&sample_batch(), 11, 99), ); let ack_cell: Rc>>> = Rc::new(RefCell::new(None)); let ack_cell_clone = Rc::clone(&ack_cell); spawner .spawn_local(async move { let append_ack = append_rx.await.expect("append ack"); append_ack.expect("append ack ok"); let commit_ack = commit_rx.await.expect("commit ack"); *ack_cell_clone.borrow_mut() = Some(commit_ack); }) .expect("spawn ack"); sender.close_channel(); pool.run(); let ack = ack_cell .borrow() .clone() .expect("ack result") .expect("ack ok"); assert_eq!(ack.last_seq, frame::INITIAL_FRAME_SEQ + 1); let writer_result = result_cell.borrow().clone().expect("writer result"); assert!(writer_result.is_ok()); let metrics_guard = metrics_reader.read().await; assert_eq!(metrics_guard.queue_depth, 0); assert!(metrics_guard.bytes_written > 0); assert!(metrics_guard.sync_operations >= 1); ``` -------------------------------- ### Initialize DynKeyProjection Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/extractor/extractors.rs.html Constructs a new DynKeyProjection, ensuring the key projection requires at least one column and that all specified columns have supported data types. ```rust fn new(schema: SchemaRef, columns: Vec) -> Result { if columns.is_empty() { return Err(KeyExtractError::Arrow(ArrowError::ComputeError( "key projection requires at least one column".to_string(), ))); } for &col in &columns { ensure_supported_type(schema.field(col).data_type(), col)?; } let projection = DynProjection::from_indices(schema.as_ref(), columns.clone()).map_err(map_view_err)?; let key_fields = columns .iter() .map(|&idx| schema.field(idx).clone()) .collect::>(); let key_schema = Arc::new(Schema::new(key_fields)); let key_columns: Arc<[usize]> = columns.into(); Ok(Self { dyn_schema: DynSchema::from_ref(schema.clone()), schema, key_schema, projection, key_columns, }) } ``` -------------------------------- ### Get Mutable WAL Configuration Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/db/builder.rs.html Provides mutable access to the WAL configuration if present. Allows modification of WAL settings. ```rust pub fn wal_config_mut(&mut self) -> Option<&mut WalConfig> { self.wal_config.as_mut() } ``` -------------------------------- ### Get Durability Class Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/db/builder.rs.html Retrieves the durability class of the current storage configuration. This indicates whether the storage is volatile or durable. ```rust pub fn durability(&self) -> DurabilityClass { self.durability } ``` -------------------------------- ### init Source: https://docs.rs/tonbo/0.4.0-a0/tonbo/db/struct.ScanBuilder.html Initializes a Pointable object with the given initializer. ```APIDOC ## unsafe fn init(init: ::Init) -> usize ### Description Initializes a with the given initializer. ### Method (Not specified, likely a method call) ### Parameters - **init** (::Init) - Description: The initializer for the object. ``` -------------------------------- ### Get Memtable Batch Count (Test) Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/inmem/mutable/memtable.rs.html Returns the current number of batches stored in the memtable, primarily for testing and debugging. ```rust self.inner.read().batch_count() ``` -------------------------------- ### Prepare Storage Layout Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/db/builder.rs.html Executes the preparation of the storage layout, including creating directories, if enabled. This is an asynchronous operation. ```rust pub async fn prepare(&self) -> Result<(), DbBuildError> where ``` -------------------------------- ### Get Number of Immutable Segments Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/db/mod.rs.html Returns the count of immutable segments currently attached to the database, ordered from oldest to newest. ```rust pub(crate) fn num_immutable_segments(&self) -> usize { self.seal_state_lock().immutables.len() } ``` -------------------------------- ### Get Tombstone Watermark (Conditional) Source: https://docs.rs/tonbo/0.4.0-a0/src/tonbo/manifest/domain.rs.html Returns the tombstone watermark as an optional u64. This function is only available when the 'tokio' feature is enabled. ```rust #[cfg(all(test, feature = "tokio"))] pub(crate) fn tombstone_watermark(&self) -> Option { self.tombstone_watermark } ```