scopedb_client/
table.rs

1// Copyright 2024 ScopeDB, Inc.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use crate::CatalogTable;
16use crate::Client;
17use crate::Error;
18use crate::append_stream::AppendStreamBuilder;
19use crate::protocol::AppendRowsResult;
20
21/// A table selected for writes or catalog reads.
22///
23/// The table must already exist for writes. Database and schema default to
24/// `scopedb` and `public`; override them with [`Self::with_database`] and
25/// [`Self::with_schema`].
26#[derive(Debug, Clone)]
27pub struct Table {
28    client: Client,
29    database: Option<String>,
30    schema: Option<String>,
31    table: String,
32}
33
34impl Table {
35    pub(crate) fn new(client: Client, table: String) -> Self {
36        Self {
37            client,
38            database: None,
39            schema: None,
40            table,
41        }
42    }
43
44    pub fn with_database(mut self, database: impl Into<String>) -> Self {
45        self.database = Some(database.into());
46        self
47    }
48
49    pub fn with_schema(mut self, schema: impl Into<String>) -> Self {
50        self.schema = Some(schema.into());
51        self
52    }
53
54    pub fn identifier(&self) -> String {
55        [
56            self.database.as_deref().unwrap_or("scopedb"),
57            self.schema.as_deref().unwrap_or("public"),
58            &self.table,
59        ]
60        .map(quote_ident)
61        .join(".")
62    }
63
64    pub async fn drop(&self) -> Result<(), Error> {
65        self.client
66            .statement(format!("DROP TABLE {}", self.identifier()))
67            .execute()
68            .await
69            .map(|_| ())
70    }
71
72    /// Writes newline-delimited JSON to this table in one HTTP attempt.
73    ///
74    /// Each line must be a JSON object matching the destination columns. A request
75    /// allows at most 8 MiB of uncompressed NDJSON and 200,000 rows. The SDK sends
76    /// it with zstd compression and does not automatically retry it.
77    ///
78    /// Integer inputs must fit in a signed 64-bit integer, including values in
79    /// `uint` columns and nested JSON. Larger integers are rejected without
80    /// narrowing; query results still support the full `u64` range.
81    ///
82    /// ```no_run
83    /// # async fn example(client: scopedb_client::Client) -> Result<(), scopedb_client::Error> {
84    /// let result = client
85    ///     .table("events")
86    ///     .append(r#"{"id": 1, "name": "page_view"}"#)
87    ///     .await?;
88    /// println!("Written: {} rows", result.num_rows_inserted);
89    /// # Ok(())
90    /// # }
91    /// ```
92    pub async fn append(&self, ndjson: impl Into<String>) -> Result<AppendRowsResult, Error> {
93        self.client
94            .append_rows(
95                self.database.as_deref().unwrap_or("scopedb"),
96                self.schema.as_deref().unwrap_or("public"),
97                &self.table,
98                ndjson,
99            )
100            .await
101    }
102
103    /// Builds an asynchronous, bounded and concurrent append stream for this table.
104    pub fn append_stream(&self) -> AppendStreamBuilder {
105        AppendStreamBuilder::new(
106            self.client.clone(),
107            self.database
108                .clone()
109                .unwrap_or_else(|| "scopedb".to_string()),
110            self.schema.clone().unwrap_or_else(|| "public".to_string()),
111            self.table.clone(),
112        )
113    }
114
115    /// Selects this table's complete definition and ETag for conditional changes.
116    pub fn catalog(&self) -> crate::CatalogResource<CatalogTable> {
117        self.client.catalog().table(
118            self.database.as_deref().unwrap_or("scopedb"),
119            self.schema.as_deref().unwrap_or("public"),
120            &self.table,
121        )
122    }
123}
124
125fn quote_ident(input: &str) -> String {
126    let quote = '`';
127    let mut out = String::with_capacity(input.len() + 2);
128    out.push(quote);
129    for ch in input.chars() {
130        match ch {
131            '\t' => out.push_str("\\t"),
132            '\n' => out.push_str("\\n"),
133            '\r' => out.push_str("\\r"),
134            '\\' => out.push_str("\\\\"),
135            c if c == quote => {
136                out.push('\\');
137                out.push(c);
138            }
139            c if c < '\u{20}' => out.push_str(&format!("\\x{:02x}", c as u32)),
140            c => out.push(c),
141        }
142    }
143    out.push(quote);
144    out
145}
146
147#[cfg(test)]
148mod tests {
149    use super::quote_ident;
150
151    #[test]
152    fn identifiers_always_use_the_same_defaults_as_append_and_catalog() {
153        let client = crate::Client::builder("https://example.invalid")
154            .build()
155            .unwrap();
156        assert_eq!(
157            client.table("events").identifier(),
158            "`scopedb`.`public`.`events`"
159        );
160        assert_eq!(
161            client
162                .table("events")
163                .with_database("analytics")
164                .identifier(),
165            "`analytics`.`public`.`events`"
166        );
167        assert_eq!(
168            client.table("events").with_schema("archive").identifier(),
169            "`scopedb`.`archive`.`events`"
170        );
171    }
172
173    #[test]
174    fn test_quote_ident() {
175        assert_eq!(quote_ident("plain"), "`plain`");
176        assert_eq!(quote_ident("a`b"), "`a\\`b`");
177        assert_eq!(quote_ident("a\nb"), "`a\\nb`");
178    }
179}
scopedb_client/table.rs

scopedb_client/
table.rs
