Struct Table 

pub struct Table { /* private fields */ }

A table selected for writes or catalog reads.

The table must already exist for writes. Database and schema default to scopedb and public; override them with Self::with_database and Self::with_schema.

Implementations

impl Table

pub fn with_database(self, database: impl Into<String>) -> Self

Examples found in repository

examples/common/mod.rs (line 63)

pub fn write_table(client: &Client) -> Result<Table, io::Error> {
    let table = std::env::var("SCOPEDB_TABLE")
        .ok()
        .filter(|value| !value.is_empty())
        .ok_or_else(|| {
            io::Error::new(
                io::ErrorKind::InvalidInput,
                "set SCOPEDB_TABLE to a disposable table before running this write example",
            )
        })?;

    Ok(client
        .table(table)
        .with_database(database())
        .with_schema(schema()))
}

examples/table.rs (line 23)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table_name = std::env::var("SCOPEDB_TABLE").unwrap_or_else(|_| "events".to_string());
    let table = client
        .table(table_name)
        .with_database(common::database())
        .with_schema(common::schema());

    println!("identifier: {}", table.identifier());

    let description = table.catalog().get().await?.definition;
    for column in description.columns {
        println!("{}: {:?}", column.name, column.data_type);
    }

    Ok(())
}

pub fn with_schema(self, schema: impl Into<String>) -> Self

Examples found in repository

examples/common/mod.rs (line 64)

pub fn write_table(client: &Client) -> Result<Table, io::Error> {
    let table = std::env::var("SCOPEDB_TABLE")
        .ok()
        .filter(|value| !value.is_empty())
        .ok_or_else(|| {
            io::Error::new(
                io::ErrorKind::InvalidInput,
                "set SCOPEDB_TABLE to a disposable table before running this write example",
            )
        })?;

    Ok(client
        .table(table)
        .with_database(database())
        .with_schema(schema()))
}

examples/table.rs (line 24)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table_name = std::env::var("SCOPEDB_TABLE").unwrap_or_else(|_| "events".to_string());
    let table = client
        .table(table_name)
        .with_database(common::database())
        .with_schema(common::schema());

    println!("identifier: {}", table.identifier());

    let description = table.catalog().get().await?.definition;
    for column in description.columns {
        println!("{}: {:?}", column.name, column.data_type);
    }

    Ok(())
}

pub fn identifier(&self) -> String

Examples found in repository

examples/table.rs (line 26)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table_name = std::env::var("SCOPEDB_TABLE").unwrap_or_else(|_| "events".to_string());
    let table = client
        .table(table_name)
        .with_database(common::database())
        .with_schema(common::schema());

    println!("identifier: {}", table.identifier());

    let description = table.catalog().get().await?.definition;
    for column in description.columns {
        println!("{}: {:?}", column.name, column.data_type);
    }

    Ok(())
}

pub async fn drop(&self) -> Result<(), Error>

pub async fn append( &self, ndjson: impl Into<String>, ) -> Result<AppendRowsResult, Error>

Writes newline-delimited JSON to this table in one HTTP attempt.

Each line must be a JSON object matching the destination columns. A request allows at most 8 MiB of uncompressed NDJSON and 200,000 rows. The SDK sends it with zstd compression and does not automatically retry it.

Integer inputs must fit in a signed 64-bit integer, including values in uint columns and nested JSON. Larger integers are rejected without narrowing; query results still support the full u64 range.

let result = client
    .table("events")
    .append(r#"{"id": 1, "name": "page_view"}"#)
    .await?;
println!("Written: {} rows", result.num_rows_inserted);

Examples found in repository

examples/append.rs (line 44)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // Direct append sends exactly one caller-owned NDJSON request. Each value
    // is serialized as one line; JSON arrays are not accepted by this API.
    let ndjson = [
        serde_json::json!({
            "id": 1,
            "occurred_at": jiff::Timestamp::now(),
            "name": "first",
        }),
        serde_json::json!({
            "id": 2,
            "occurred_at": jiff::Timestamp::now(),
            "name": "second",
        }),
    ]
    .iter()
    .map(serde_json::to_string)
    .collect::<Result<Vec<_>, _>>()?
    .join("\n");

    match table.append(ndjson).await {
        Ok(result) => {
            println!("committed remotely: {} rows", result.num_rows_inserted);
            Ok(())
        }
        Err(error) => {
            if error.kind() == ErrorKind::AppendRowsFailed
                && matches!(
                    error.append_details().map(|details| details.status),
                    Some(AppendState::Unknown)
                )
            {
                eprintln!("commit outcome is unknown; reconcile before replaying this payload");
            }
            Err(error.into())
        }
    }
}

pub fn append_stream(&self) -> AppendStreamBuilder

Builds an asynchronous, bounded and concurrent append stream for this table.

Examples found in repository

examples/append_stream.rs (line 21)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;
    let stream = table.append_stream().build()?;

    let admitted = stream
        .send_all([
            serde_json::json!({
                "id": 1,
                "occurred_at": jiff::Timestamp::now(),
                "name": "first",
            }),
            serde_json::json!({
                "id": 2,
                "occurred_at": jiff::Timestamp::now(),
                "name": "second",
            }),
        ])
        .await?;
    println!("accepted locally: {} rows", admitted.accepted_rows);

    // send_all() confirms only local admission. With the default strict policy,
    // successful shutdown confirms at least one commit acknowledgement per row.
    // Retrying an unknown outcome can insert duplicates.
    let report = stream.shutdown().await?;
    println!("committed remotely: {} rows", report.committed_rows);

    Ok(())
}

examples/import_ndjson.rs (line 29)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let path = std::env::args()
        .nth(1)
        .ok_or("usage: import_ndjson <events.ndjson>")?;
    let mut input = BufReader::new(File::open(path)?);
    let client = common::client()?;
    let stream = common::write_table(&client)?.append_stream().build()?;
    let admitted: Result<(), Box<dyn std::error::Error>> = async {
        let mut line_number = 0;
        loop {
            // Bound a single input record as well as the stream's queued bytes.
            let mut line = String::new();
            let bytes = input
                .by_ref()
                .take(8 * 1024 * 1024 + 1)
                .read_line(&mut line)?;
            if bytes == 0 {
                break;
            }
            line_number += 1;
            if bytes > 8 * 1024 * 1024 {
                return Err(format!("line {line_number} exceeds 8 MiB").into());
            }
            if line.trim().is_empty() {
                continue;
            }
            let row: serde_json::Value = serde_json::from_str(&line)
                .map_err(|error| format!("line {line_number}: {error}"))?;
            stream.send(&row).await?;
        }
        Ok(())
    }
    .await;
    // Settle the accepted prefix even if parsing or admission failed midway.
    let settled = stream.shutdown().await;
    println!("delivery: {:?}", stream.stats());
    settled?;
    admitted?;
    Ok(())
}

examples/telemetry.rs (line 50)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // Continue mode is explicitly best effort: failed or ambiguous batches are
    // accounted for, released, and do not stop later telemetry batches.
    let telemetry = table
        .append_stream()
        .failure_policy(AppendFailurePolicy::Continue)
        .circuit_breaker(Some(AppendCircuitBreakerOptions::default()))
        .target_batch_bytes(1024 * 1024)
        .max_batch_rows(10_000)
        .flush_interval(Duration::from_secs(1))
        .max_buffered_bytes(32 * 1024 * 1024)
        // This limit applies to each stream; include every application replica.
        .max_concurrent_batches(1)
        .attempt_timeout(Duration::from_secs(10))
        .on_batch_failure(|event| {
            // A synchronous observer; use a separate diagnostics sink.
            eprintln!("telemetry append failure: {event:?}");
        })
        .build()?;

    let mut dropped_locally = 0_u64;
    for request in 1..=100 {
        if !track(&telemetry, request) {
            dropped_locally += 1;
        }
    }
    println!("locally dropped: {dropped_locally}");

    // In a service, stop and join producer tasks before this settlement barrier.
    let report = telemetry.shutdown().await?;
    if report.outcome != AppendDeliveryOutcome::Ok {
        eprintln!("telemetry loss or ambiguity: {report:?}");
    }
    println!("final stream stats: {:?}", telemetry.stats());

    Ok(())
}

examples/bulk_append.rs (line 30)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // The default Stop policy gives strict barriers. Local memory and request
    // concurrency are bounded independently; concurrent batches have no
    // defined commit order.
    let stream = table
        .append_stream()
        .target_batch_bytes(32 * 1024)
        .max_batch_rows(10_000)
        .flush_interval(Duration::from_secs(1))
        .max_buffered_bytes(1024 * 1024)
        .max_concurrent_batches(4)
        .attempt_timeout(Duration::from_secs(30))
        .build()?;

    let rows = (1..=10_000).map(|id| {
        serde_json::json!({
            "id": id,
            "occurred_at": jiff::Timestamp::now(),
            "name": format!("example-{id}"),
        })
    });

    let admission_error = match stream.send_all(rows).await {
        Ok(result) => {
            println!("accepted locally: {} rows", result.accepted_rows);
            None
        }
        Err(error) => {
            // There is no stream-wide abort or rollback. Settle the already
            // accepted prefix even if row production or admission stopped.
            eprintln!("row admission stopped; settling the accepted prefix: {error}");
            Some(error)
        }
    };

    match stream.shutdown().await {
        Ok(report) => {
            println!("committed remotely: {} rows", report.committed_rows);
            println!("final stream stats: {:?}", stream.stats());
        }
        Err(error) => {
            if matches!(
                error.append_details().map(|details| details.status),
                Some(AppendState::Unknown)
            ) {
                eprintln!("commit outcome is unknown; reconcile before replaying");
            } else {
                eprintln!("bulk append failed: {error}");
            }
            eprintln!("final stream stats: {:?}", stream.stats());
            return Err(error.into());
        }
    }

    if let Some(error) = admission_error {
        return Err(error.into());
    }
    Ok(())
}

pub fn catalog(&self) -> CatalogResource<CatalogTable>

Selects this table’s complete definition and ETag for conditional changes.

Examples found in repository

examples/table.rs (line 28)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table_name = std::env::var("SCOPEDB_TABLE").unwrap_or_else(|_| "events".to_string());
    let table = client
        .table(table_name)
        .with_database(common::database())
        .with_schema(common::schema());

    println!("identifier: {}", table.identifier());

    let description = table.catalog().get().await?.definition;
    for column in description.columns {
        println!("{}: {:?}", column.name, column.data_type);
    }

    Ok(())
}

Trait Implementations

impl Clone for Table

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for Table

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.