mirror of
https://github.com/lancedb/lancedb.git
synced 2026-08-29 09:28:27 +00:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Will Jones
|
b7dbef971e | ||
|
Will Jones
|
71202dd3e6 | ||
|
Will Jones
|
0ab435f2d7 |
Generated
+22
-22
@@ -3456,7 +3456,7 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
|
||||
[[package]]
|
||||
name = "fsst"
|
||||
version = "11.0.0-beta.15"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.15#8064b3a27dc4e05a6ab6ceb439fa1be9950e00eb"
|
||||
source = "git+https://github.com/lance-format/lance.git?rev=d7b1d570461c6d2adde8f3a84ae88db4823c726f#d7b1d570461c6d2adde8f3a84ae88db4823c726f"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"rand 0.9.5",
|
||||
@@ -4816,7 +4816,7 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a"
|
||||
[[package]]
|
||||
name = "lance"
|
||||
version = "11.0.0-beta.15"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.15#8064b3a27dc4e05a6ab6ceb439fa1be9950e00eb"
|
||||
source = "git+https://github.com/lance-format/lance.git?rev=d7b1d570461c6d2adde8f3a84ae88db4823c726f#d7b1d570461c6d2adde8f3a84ae88db4823c726f"
|
||||
dependencies = [
|
||||
"arc-swap",
|
||||
"arrow",
|
||||
@@ -4889,7 +4889,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "lance-arrow"
|
||||
version = "11.0.0-beta.15"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.15#8064b3a27dc4e05a6ab6ceb439fa1be9950e00eb"
|
||||
source = "git+https://github.com/lance-format/lance.git?rev=d7b1d570461c6d2adde8f3a84ae88db4823c726f#d7b1d570461c6d2adde8f3a84ae88db4823c726f"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
@@ -4911,7 +4911,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "lance-arrow-scalar"
|
||||
version = "58.0.0"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.15#8064b3a27dc4e05a6ab6ceb439fa1be9950e00eb"
|
||||
source = "git+https://github.com/lance-format/lance.git?rev=d7b1d570461c6d2adde8f3a84ae88db4823c726f#d7b1d570461c6d2adde8f3a84ae88db4823c726f"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
@@ -4925,7 +4925,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "lance-arrow-stats"
|
||||
version = "58.0.0"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.15#8064b3a27dc4e05a6ab6ceb439fa1be9950e00eb"
|
||||
source = "git+https://github.com/lance-format/lance.git?rev=d7b1d570461c6d2adde8f3a84ae88db4823c726f#d7b1d570461c6d2adde8f3a84ae88db4823c726f"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-schema",
|
||||
@@ -4935,7 +4935,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "lance-bitpacking"
|
||||
version = "11.0.0-beta.15"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.15#8064b3a27dc4e05a6ab6ceb439fa1be9950e00eb"
|
||||
source = "git+https://github.com/lance-format/lance.git?rev=d7b1d570461c6d2adde8f3a84ae88db4823c726f#d7b1d570461c6d2adde8f3a84ae88db4823c726f"
|
||||
dependencies = [
|
||||
"arrayref",
|
||||
"crunchy",
|
||||
@@ -4946,7 +4946,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "lance-core"
|
||||
version = "11.0.0-beta.15"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.15#8064b3a27dc4e05a6ab6ceb439fa1be9950e00eb"
|
||||
source = "git+https://github.com/lance-format/lance.git?rev=d7b1d570461c6d2adde8f3a84ae88db4823c726f#d7b1d570461c6d2adde8f3a84ae88db4823c726f"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
@@ -4984,7 +4984,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "lance-datafusion"
|
||||
version = "11.0.0-beta.15"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.15#8064b3a27dc4e05a6ab6ceb439fa1be9950e00eb"
|
||||
source = "git+https://github.com/lance-format/lance.git?rev=d7b1d570461c6d2adde8f3a84ae88db4823c726f#d7b1d570461c6d2adde8f3a84ae88db4823c726f"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-array",
|
||||
@@ -5014,7 +5014,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "lance-datagen"
|
||||
version = "11.0.0-beta.15"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.15#8064b3a27dc4e05a6ab6ceb439fa1be9950e00eb"
|
||||
source = "git+https://github.com/lance-format/lance.git?rev=d7b1d570461c6d2adde8f3a84ae88db4823c726f#d7b1d570461c6d2adde8f3a84ae88db4823c726f"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-array",
|
||||
@@ -5032,7 +5032,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "lance-derive"
|
||||
version = "11.0.0-beta.15"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.15#8064b3a27dc4e05a6ab6ceb439fa1be9950e00eb"
|
||||
source = "git+https://github.com/lance-format/lance.git?rev=d7b1d570461c6d2adde8f3a84ae88db4823c726f#d7b1d570461c6d2adde8f3a84ae88db4823c726f"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
@@ -5042,7 +5042,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "lance-encoding"
|
||||
version = "11.0.0-beta.15"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.15#8064b3a27dc4e05a6ab6ceb439fa1be9950e00eb"
|
||||
source = "git+https://github.com/lance-format/lance.git?rev=d7b1d570461c6d2adde8f3a84ae88db4823c726f#d7b1d570461c6d2adde8f3a84ae88db4823c726f"
|
||||
dependencies = [
|
||||
"arrow-arith",
|
||||
"arrow-array",
|
||||
@@ -5076,7 +5076,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "lance-file"
|
||||
version = "11.0.0-beta.15"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.15#8064b3a27dc4e05a6ab6ceb439fa1be9950e00eb"
|
||||
source = "git+https://github.com/lance-format/lance.git?rev=d7b1d570461c6d2adde8f3a84ae88db4823c726f#d7b1d570461c6d2adde8f3a84ae88db4823c726f"
|
||||
dependencies = [
|
||||
"arrow-arith",
|
||||
"arrow-array",
|
||||
@@ -5108,7 +5108,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "lance-index"
|
||||
version = "11.0.0-beta.15"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.15#8064b3a27dc4e05a6ab6ceb439fa1be9950e00eb"
|
||||
source = "git+https://github.com/lance-format/lance.git?rev=d7b1d570461c6d2adde8f3a84ae88db4823c726f#d7b1d570461c6d2adde8f3a84ae88db4823c726f"
|
||||
dependencies = [
|
||||
"arc-swap",
|
||||
"arrow",
|
||||
@@ -5173,7 +5173,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "lance-index-core"
|
||||
version = "11.0.0-beta.15"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.15#8064b3a27dc4e05a6ab6ceb439fa1be9950e00eb"
|
||||
source = "git+https://github.com/lance-format/lance.git?rev=d7b1d570461c6d2adde8f3a84ae88db4823c726f#d7b1d570461c6d2adde8f3a84ae88db4823c726f"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-schema",
|
||||
@@ -5196,7 +5196,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "lance-io"
|
||||
version = "11.0.0-beta.15"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.15#8064b3a27dc4e05a6ab6ceb439fa1be9950e00eb"
|
||||
source = "git+https://github.com/lance-format/lance.git?rev=d7b1d570461c6d2adde8f3a84ae88db4823c726f#d7b1d570461c6d2adde8f3a84ae88db4823c726f"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-array",
|
||||
@@ -5233,7 +5233,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "lance-linalg"
|
||||
version = "11.0.0-beta.15"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.15#8064b3a27dc4e05a6ab6ceb439fa1be9950e00eb"
|
||||
source = "git+https://github.com/lance-format/lance.git?rev=d7b1d570461c6d2adde8f3a84ae88db4823c726f#d7b1d570461c6d2adde8f3a84ae88db4823c726f"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-schema",
|
||||
@@ -5248,7 +5248,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "lance-namespace"
|
||||
version = "11.0.0-beta.15"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.15#8064b3a27dc4e05a6ab6ceb439fa1be9950e00eb"
|
||||
source = "git+https://github.com/lance-format/lance.git?rev=d7b1d570461c6d2adde8f3a84ae88db4823c726f#d7b1d570461c6d2adde8f3a84ae88db4823c726f"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"async-trait",
|
||||
@@ -5261,7 +5261,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "lance-namespace-impls"
|
||||
version = "11.0.0-beta.15"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.15#8064b3a27dc4e05a6ab6ceb439fa1be9950e00eb"
|
||||
source = "git+https://github.com/lance-format/lance.git?rev=d7b1d570461c6d2adde8f3a84ae88db4823c726f#d7b1d570461c6d2adde8f3a84ae88db4823c726f"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-ipc",
|
||||
@@ -5315,7 +5315,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "lance-select"
|
||||
version = "11.0.0-beta.15"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.15#8064b3a27dc4e05a6ab6ceb439fa1be9950e00eb"
|
||||
source = "git+https://github.com/lance-format/lance.git?rev=d7b1d570461c6d2adde8f3a84ae88db4823c726f#d7b1d570461c6d2adde8f3a84ae88db4823c726f"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
@@ -5330,7 +5330,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "lance-table"
|
||||
version = "11.0.0-beta.15"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.15#8064b3a27dc4e05a6ab6ceb439fa1be9950e00eb"
|
||||
source = "git+https://github.com/lance-format/lance.git?rev=d7b1d570461c6d2adde8f3a84ae88db4823c726f#d7b1d570461c6d2adde8f3a84ae88db4823c726f"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-array",
|
||||
@@ -5371,7 +5371,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "lance-testing"
|
||||
version = "11.0.0-beta.15"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.15#8064b3a27dc4e05a6ab6ceb439fa1be9950e00eb"
|
||||
source = "git+https://github.com/lance-format/lance.git?rev=d7b1d570461c6d2adde8f3a84ae88db4823c726f#d7b1d570461c6d2adde8f3a84ae88db4823c726f"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-schema",
|
||||
@@ -5385,7 +5385,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "lance-tokenizer"
|
||||
version = "11.0.0-beta.15"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.15#8064b3a27dc4e05a6ab6ceb439fa1be9950e00eb"
|
||||
source = "git+https://github.com/lance-format/lance.git?rev=d7b1d570461c6d2adde8f3a84ae88db4823c726f#d7b1d570461c6d2adde8f3a84ae88db4823c726f"
|
||||
dependencies = [
|
||||
"frostem",
|
||||
"icu_segmenter",
|
||||
|
||||
+17
-14
@@ -13,20 +13,23 @@ categories = ["database-implementations"]
|
||||
rust-version = "1.91.0"
|
||||
|
||||
[workspace.dependencies]
|
||||
lance = { "version" = "=11.0.0-beta.15", default-features = false, "tag" = "v11.0.0-beta.15", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-core = { "version" = "=11.0.0-beta.15", "tag" = "v11.0.0-beta.15", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datagen = { "version" = "=11.0.0-beta.15", "tag" = "v11.0.0-beta.15", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-file = { "version" = "=11.0.0-beta.15", "tag" = "v11.0.0-beta.15", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-io = { "version" = "=11.0.0-beta.15", default-features = false, "tag" = "v11.0.0-beta.15", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-index = { "version" = "=11.0.0-beta.15", "tag" = "v11.0.0-beta.15", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-linalg = { "version" = "=11.0.0-beta.15", "tag" = "v11.0.0-beta.15", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace = { "version" = "=11.0.0-beta.15", "tag" = "v11.0.0-beta.15", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace-impls = { "version" = "=11.0.0-beta.15", default-features = false, "tag" = "v11.0.0-beta.15", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-table = { "version" = "=11.0.0-beta.15", "tag" = "v11.0.0-beta.15", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-testing = { "version" = "=11.0.0-beta.15", "tag" = "v11.0.0-beta.15", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datafusion = { "version" = "=11.0.0-beta.15", "tag" = "v11.0.0-beta.15", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-encoding = { "version" = "=11.0.0-beta.15", "tag" = "v11.0.0-beta.15", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-arrow = { "version" = "=11.0.0-beta.15", "tag" = "v11.0.0-beta.15", "git" = "https://github.com/lance-format/lance.git" }
|
||||
# TEMPORARY: `ObjectStore::read_dir_page` is not in a lance release yet, so these point at
|
||||
# lance-format/lance#8606 cherry-picked onto the v11.0.0-beta.15 tag. Put the tag back once that
|
||||
# PR has merged and shipped in a release.
|
||||
lance = { "version" = "=11.0.0-beta.15", default-features = false, "rev" = "d7b1d570461c6d2adde8f3a84ae88db4823c726f", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-core = { "version" = "=11.0.0-beta.15", "rev" = "d7b1d570461c6d2adde8f3a84ae88db4823c726f", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datagen = { "version" = "=11.0.0-beta.15", "rev" = "d7b1d570461c6d2adde8f3a84ae88db4823c726f", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-file = { "version" = "=11.0.0-beta.15", "rev" = "d7b1d570461c6d2adde8f3a84ae88db4823c726f", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-io = { "version" = "=11.0.0-beta.15", default-features = false, "rev" = "d7b1d570461c6d2adde8f3a84ae88db4823c726f", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-index = { "version" = "=11.0.0-beta.15", "rev" = "d7b1d570461c6d2adde8f3a84ae88db4823c726f", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-linalg = { "version" = "=11.0.0-beta.15", "rev" = "d7b1d570461c6d2adde8f3a84ae88db4823c726f", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace = { "version" = "=11.0.0-beta.15", "rev" = "d7b1d570461c6d2adde8f3a84ae88db4823c726f", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace-impls = { "version" = "=11.0.0-beta.15", default-features = false, "rev" = "d7b1d570461c6d2adde8f3a84ae88db4823c726f", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-table = { "version" = "=11.0.0-beta.15", "rev" = "d7b1d570461c6d2adde8f3a84ae88db4823c726f", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-testing = { "version" = "=11.0.0-beta.15", "rev" = "d7b1d570461c6d2adde8f3a84ae88db4823c726f", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datafusion = { "version" = "=11.0.0-beta.15", "rev" = "d7b1d570461c6d2adde8f3a84ae88db4823c726f", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-encoding = { "version" = "=11.0.0-beta.15", "rev" = "d7b1d570461c6d2adde8f3a84ae88db4823c726f", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-arrow = { "version" = "=11.0.0-beta.15", "rev" = "d7b1d570461c6d2adde8f3a84ae88db4823c726f", "git" = "https://github.com/lance-format/lance.git" }
|
||||
ahash = "0.8"
|
||||
# Note that this one does not include pyarrow
|
||||
arrow = { version = "58.0.0", optional = false }
|
||||
|
||||
@@ -1755,7 +1755,8 @@ mod tests {
|
||||
|
||||
let page = db.list_tables().limit(10).execute().await.unwrap();
|
||||
assert_eq!(page.tables, names[..10]);
|
||||
// The token is opaque: a caller checks that there is one, not what it says.
|
||||
// The token is opaque and is not a table name: it is whatever resumes the store
|
||||
// the database sits on, so a caller checks that there is one, not what it says.
|
||||
assert!(page.page_token.is_some());
|
||||
|
||||
// Walking in pages has to reach every table exactly once, with nothing lost
|
||||
|
||||
@@ -13,7 +13,7 @@ use lance::dataset::{ReadParams, WriteMode, builder::DatasetBuilder};
|
||||
use lance::io::{ObjectStore, ObjectStoreParams, WrappingObjectStore};
|
||||
use lance_datafusion::utils::StreamingWriteSource;
|
||||
use lance_file::version::LanceFileVersion;
|
||||
use lance_io::object_store::{StorageOptionsAccessor, StorageOptionsProvider};
|
||||
use lance_io::object_store::{ReadDirOptions, StorageOptionsAccessor, StorageOptionsProvider};
|
||||
use lance_table::io::commit::commit_handler_from_url;
|
||||
use object_store::local::LocalFileSystem;
|
||||
use snafu::ResultExt;
|
||||
@@ -281,6 +281,21 @@ impl std::fmt::Display for ListingDatabase {
|
||||
}
|
||||
|
||||
const LANCE_EXTENSION: &str = "lance";
|
||||
|
||||
/// The table a listed child of the database names, or `None` if the child is not a table.
|
||||
///
|
||||
/// A table is the directory `<name>.lance`; a loose file or any other directory under the
|
||||
/// database prefix belongs to something else. `dir_suffix` is `.lance`, built once by the
|
||||
/// caller rather than per child.
|
||||
/// The table a listed child directory holds, or `None` if it is not a table at all.
|
||||
///
|
||||
/// Only directories are considered, so a loose object named like a table is not one.
|
||||
fn table_name(location: &object_store::path::Path, dir_suffix: &str) -> Option<String> {
|
||||
location
|
||||
.filename()?
|
||||
.strip_suffix(dir_suffix)
|
||||
.map(String::from)
|
||||
}
|
||||
const ENGINE: &str = "engine";
|
||||
const MIRRORED_STORE: &str = "mirroredStore";
|
||||
|
||||
@@ -988,51 +1003,75 @@ impl Database for ListingDatabase {
|
||||
Ok(f)
|
||||
}
|
||||
|
||||
/// List the tables in the database, a page at a time.
|
||||
///
|
||||
/// The page token and the page size go into the object store's list request rather than
|
||||
/// being applied to a full listing, so a page costs what the page holds and not what the
|
||||
/// database holds. Stores with no paginated list API list the level in full and page it
|
||||
/// locally, which is what every store did before.
|
||||
///
|
||||
/// The token is opaque and is only meaningful to the store that issued it: it carries a
|
||||
/// continuation token where the store has one. It is not a table name, and a caller must
|
||||
/// not construct one. A page can be shorter than `limit` and still be followed by more, so
|
||||
/// the token is what ends a walk.
|
||||
///
|
||||
/// Tables come back in the order the store lists directories, which is by key. That
|
||||
/// differs from sorting by name only between a name and one that extends it:
|
||||
/// `users-archive` precedes `users`, because the `-` of `users-archive.lance` sorts below
|
||||
/// the `.` of `users.lance`.
|
||||
async fn list_tables(&self, request: ListTablesRequest) -> Result<ListTablesResponse> {
|
||||
if request.id.as_ref().map(|v| !v.is_empty()).unwrap_or(false) {
|
||||
return self.namespace_database().list_tables(request).await;
|
||||
}
|
||||
let mut f = self
|
||||
.object_store
|
||||
.read_dir(self.base_path.clone())
|
||||
.await?
|
||||
.iter()
|
||||
.map(Path::new)
|
||||
.filter(|path| {
|
||||
let is_lance = path
|
||||
.extension()
|
||||
.and_then(|e| e.to_str())
|
||||
.map(|e| e == LANCE_EXTENSION);
|
||||
is_lance.unwrap_or(false)
|
||||
})
|
||||
.filter_map(|p| p.file_stem().and_then(|s| s.to_str().map(String::from)))
|
||||
.collect::<Vec<String>>();
|
||||
f.sort();
|
||||
let limit = request.limit.map(|limit| limit.max(0) as usize);
|
||||
let dir_suffix = format!(".{LANCE_EXTENSION}");
|
||||
let mut tables = Vec::new();
|
||||
let mut page_token = request.page_token.filter(|token| !token.is_empty());
|
||||
|
||||
// Handle pagination with page_token
|
||||
if let Some(ref page_token) = request.page_token {
|
||||
let index = f
|
||||
.iter()
|
||||
.position(|name| name.as_str() > page_token.as_str())
|
||||
.unwrap_or(f.len());
|
||||
f.drain(0..index);
|
||||
// A page of nothing: the store rejects a limit of zero, and no table was handed over
|
||||
// for a token to resume after.
|
||||
if limit == Some(0) {
|
||||
return Ok(ListTablesResponse {
|
||||
context: None,
|
||||
tables,
|
||||
page_token: None,
|
||||
});
|
||||
}
|
||||
|
||||
// Determine if there's a next page. The token is the last name of this page,
|
||||
// not the first of the next one: the next page resumes strictly after the
|
||||
// token, so naming the next page's first entry would skip it.
|
||||
let next_page_token = match request.limit {
|
||||
Some(limit) if f.len() > limit as usize => {
|
||||
f.truncate(limit as usize);
|
||||
f.last().cloned()
|
||||
loop {
|
||||
// Ask only for what the page still has room for, so a database holding more
|
||||
// than one page costs one request per page rather than one per table.
|
||||
let listing = self
|
||||
.object_store
|
||||
.read_dir_page(
|
||||
self.base_path.clone(),
|
||||
ReadDirOptions {
|
||||
page_token: page_token.take(),
|
||||
limit: limit.map(|limit| limit - tables.len()),
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
page_token = listing.page_token;
|
||||
// Only child directories can be tables, and the store already separates them
|
||||
// out, so the objects in the page are not looked at.
|
||||
tables.extend(
|
||||
listing
|
||||
.result
|
||||
.common_prefixes
|
||||
.iter()
|
||||
.filter_map(|location| table_name(location, &dir_suffix)),
|
||||
);
|
||||
// Children that are not tables leave the page short of the limit, so keep
|
||||
// going until the page is full or the database runs out.
|
||||
if page_token.is_none() || limit.is_none_or(|limit| tables.len() >= limit) {
|
||||
break;
|
||||
}
|
||||
_ => None,
|
||||
};
|
||||
}
|
||||
|
||||
Ok(ListTablesResponse {
|
||||
context: None,
|
||||
tables: f,
|
||||
page_token: next_page_token,
|
||||
tables,
|
||||
page_token,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -1306,6 +1345,171 @@ mod tests {
|
||||
use tokio::sync::Barrier;
|
||||
use tokio::time::timeout;
|
||||
|
||||
async fn create_tables(db: &ListingDatabase, names: &[&str]) {
|
||||
let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
|
||||
for name in names {
|
||||
db.create_table(CreateTableRequest {
|
||||
name: name.to_string(),
|
||||
namespace_path: vec![],
|
||||
data: Box::new(RecordBatch::new_empty(schema.clone())) as Box<dyn Scannable>,
|
||||
mode: CreateTableMode::Create,
|
||||
write_options: Default::default(),
|
||||
location: None,
|
||||
namespace_client: None,
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
/// Every table in the database, taken `limit` at a time, which is how a caller walks a
|
||||
/// listing: the token ends the walk, never a short page.
|
||||
async fn walk(db: &ListingDatabase, limit: Option<i32>) -> Vec<String> {
|
||||
let mut seen = Vec::new();
|
||||
let mut page_token = None;
|
||||
loop {
|
||||
let page = db
|
||||
.list_tables(ListTablesRequest {
|
||||
limit,
|
||||
page_token,
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
seen.extend(page.tables);
|
||||
page_token = page.page_token;
|
||||
if page_token.is_none() {
|
||||
return seen;
|
||||
}
|
||||
assert!(
|
||||
seen.len() < 100,
|
||||
"the walk is serving tables more than once"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// Paging with the returned token has to visit every table exactly once, whatever the
|
||||
/// page size, with nothing lost or repeated at a boundary.
|
||||
#[rstest::rstest]
|
||||
#[tokio::test]
|
||||
async fn test_list_tables_pages_over_every_table_once(#[values(1, 2, 3, 5, 10)] limit: i32) {
|
||||
let (_tempdir, db) = setup_database().await;
|
||||
create_tables(&db, &["a", "b", "c", "d", "e"]).await;
|
||||
|
||||
assert_eq!(walk(&db, Some(limit)).await, vec!["a", "b", "c", "d", "e"]);
|
||||
}
|
||||
|
||||
/// The token is opaque: it is whatever resumes the store the database sits on, not a
|
||||
/// table name. Callers hand it back and nothing else.
|
||||
///
|
||||
/// Nothing validates a token, so one invented by a caller is read as a position rather
|
||||
/// than refused — which is why the token has to come back from a previous page.
|
||||
#[tokio::test]
|
||||
async fn test_the_page_token_is_not_a_table_name() {
|
||||
let (_tempdir, db) = setup_database().await;
|
||||
create_tables(&db, &["a", "b", "c"]).await;
|
||||
|
||||
let page = db
|
||||
.list_tables(ListTablesRequest {
|
||||
limit: Some(1),
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(page.tables, vec!["a"]);
|
||||
let token = page.page_token.expect("two tables are still to come");
|
||||
assert_ne!(token, "a");
|
||||
|
||||
// Handing it back is the only thing a caller does with it, and it resumes.
|
||||
let rest = db
|
||||
.list_tables(ListTablesRequest {
|
||||
page_token: Some(token),
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(rest.tables, vec!["b", "c"]);
|
||||
}
|
||||
|
||||
/// A limit the listing does not fill leaves no token behind, so a caller paging by token
|
||||
/// stops without asking for an empty page.
|
||||
#[tokio::test]
|
||||
async fn test_a_listing_that_runs_out_has_no_token() {
|
||||
let (_tempdir, db) = setup_database().await;
|
||||
create_tables(&db, &["a", "b"]).await;
|
||||
|
||||
let page = db
|
||||
.list_tables(ListTablesRequest {
|
||||
limit: Some(10),
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(page.tables, vec!["a", "b"]);
|
||||
assert_eq!(page.page_token, None);
|
||||
}
|
||||
|
||||
/// An empty page token means "from the start", which is how a client looping on a token
|
||||
/// spells its first request.
|
||||
#[tokio::test]
|
||||
async fn test_an_empty_page_token_lists_from_the_start() {
|
||||
let (_tempdir, db) = setup_database().await;
|
||||
create_tables(&db, &["a", "b"]).await;
|
||||
|
||||
let page = db
|
||||
.list_tables(ListTablesRequest {
|
||||
page_token: Some(String::new()),
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(page.tables, vec!["a", "b"]);
|
||||
}
|
||||
|
||||
/// Listing follows the order the object store lists directories in, so a name that
|
||||
/// extends another comes first: the `-` of `users-archive.lance` sorts below the `.` of
|
||||
/// `users.lance`. Pagination pushes its cursor into the list request, so it cannot report
|
||||
/// an order other than the one it resumes in.
|
||||
#[tokio::test]
|
||||
async fn test_listing_order_follows_the_store_not_the_table_name() {
|
||||
let (_tempdir, db) = setup_database().await;
|
||||
create_tables(&db, &["users", "users-archive", "users.old"]).await;
|
||||
|
||||
assert_eq!(
|
||||
walk(&db, None).await,
|
||||
vec!["users-archive", "users", "users.old"]
|
||||
);
|
||||
// And paging reports the same order, so a walk sees each table once.
|
||||
assert_eq!(
|
||||
walk(&db, Some(1)).await,
|
||||
vec!["users-archive", "users", "users.old"]
|
||||
);
|
||||
}
|
||||
|
||||
/// Only directories named `<name>.lance` are tables; loose files and other directories
|
||||
/// under the database prefix are not. A page spent on them is filled from the next one,
|
||||
/// so a page holding only non-tables does not read as an empty database.
|
||||
#[tokio::test]
|
||||
async fn test_listing_ignores_non_table_children() {
|
||||
let (tempdir, db) = setup_database().await;
|
||||
create_tables(&db, &["real"]).await;
|
||||
std::fs::write(tempdir.path().join("aaa-loose.lance"), b"not a table").unwrap();
|
||||
create_dir_all(tempdir.path().join("aaa-scratch")).unwrap();
|
||||
|
||||
let page = db
|
||||
.list_tables(ListTablesRequest {
|
||||
limit: Some(1),
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(page.tables, vec!["real"]);
|
||||
}
|
||||
|
||||
async fn setup_database() -> (tempfile::TempDir, ListingDatabase) {
|
||||
let tempdir = tempdir().unwrap();
|
||||
let uri = tempdir.path().to_str().unwrap();
|
||||
|
||||
@@ -7,6 +7,7 @@ use std::{fmt::Formatter, sync::Arc};
|
||||
|
||||
use futures::{StreamExt, TryFutureExt, stream::BoxStream};
|
||||
use lance::io::WrappingObjectStore;
|
||||
use object_store::list::PaginatedListStore;
|
||||
use object_store::{
|
||||
CopyOptions, Error, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta,
|
||||
ObjectStore, ObjectStoreExt, PutMultipartOptions, PutOptions, PutPayload, PutResult, Result,
|
||||
@@ -187,6 +188,16 @@ impl WrappingObjectStore for MirroringObjectStoreWrapper {
|
||||
secondary: self.secondary.clone(),
|
||||
})
|
||||
}
|
||||
|
||||
// Only writes are mirrored, and a listing reads, so a pushed-down listing sees the same
|
||||
// primary this wrapper would have read from.
|
||||
fn wrap_paginated(
|
||||
&self,
|
||||
_store_prefix: &str,
|
||||
original: Arc<dyn PaginatedListStore>,
|
||||
) -> Option<Arc<dyn PaginatedListStore>> {
|
||||
Some(original)
|
||||
}
|
||||
}
|
||||
|
||||
// windows pathing can't be simply concatenated
|
||||
|
||||
@@ -57,6 +57,16 @@ impl WrappingObjectStore for IoStatsHolder {
|
||||
stats: self.0.clone(),
|
||||
})
|
||||
}
|
||||
|
||||
// This exists to count requests, so it gives up the pushdown rather than let a listing
|
||||
// go around the counter.
|
||||
fn wrap_paginated(
|
||||
&self,
|
||||
_store_prefix: &str,
|
||||
_original: Arc<dyn object_store::list::PaginatedListStore>,
|
||||
) -> Option<Arc<dyn object_store::list::PaginatedListStore>> {
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
impl IoTrackingStore {
|
||||
|
||||
@@ -33,7 +33,7 @@ use super::client::{
|
||||
ClientConfig, HeaderProvider, HttpSend, RequestResultExt, RestfulLanceDbClient, Sender,
|
||||
};
|
||||
use super::table::RemoteTable;
|
||||
use super::util::parse_server_version;
|
||||
use super::util::{SERVER_VERSION_HEADER, parse_server_version};
|
||||
use super::{ARROW_STREAM_CONTENT_TYPE, extract_job_id};
|
||||
|
||||
// Request structure for the remote clone table API
|
||||
@@ -86,6 +86,10 @@ impl ServerVersion {
|
||||
pub fn support_blobs(&self) -> bool {
|
||||
self.0 >= semver::Version::new(0, 5, 0)
|
||||
}
|
||||
|
||||
pub fn support_paginated_list_tables(&self) -> bool {
|
||||
self.0 >= semver::Version::new(0, 6, 0)
|
||||
}
|
||||
}
|
||||
|
||||
pub const OPT_REMOTE_PREFIX: &str = "remote_database_";
|
||||
@@ -207,6 +211,14 @@ pub struct RemoteDatabase<S: HttpSend = Sender> {
|
||||
namespace_context_provider: Option<Arc<dyn DynamicContextProvider>>,
|
||||
/// TLS configuration for mTLS support
|
||||
tls_config: Option<super::client::TlsConfig>,
|
||||
/// Whether this server serves the `/v2` table listing, learned once per connection.
|
||||
///
|
||||
/// This holds the one answer it is asked for rather than the server version it was
|
||||
/// derived from. A server that sends no version header is indistinguishable from one
|
||||
/// running the oldest version we know of, and caching that as a version would let a
|
||||
/// stripped header switch off multivector, structural FTS, multipart write and blobs for
|
||||
/// the life of the connection. A missing header can only cost a listing its pushdown.
|
||||
serves_paginated_list: tokio::sync::OnceCell<bool>,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
@@ -325,11 +337,45 @@ impl RemoteDatabase {
|
||||
namespace_headers,
|
||||
namespace_context_provider,
|
||||
tls_config: client_config.tls_config,
|
||||
serves_paginated_list: tokio::sync::OnceCell::new(),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl<S: HttpSend> RemoteDatabase<S> {
|
||||
/// Whether this server serves the `/v2` table listing, asking it once per connection.
|
||||
///
|
||||
/// The answer has to be known before the first listing rather than learned from it. The
|
||||
/// two listing routes resume from different things, so a walk that started on one cannot
|
||||
/// finish on the other, and a walk that learned the answer from its own first page would
|
||||
/// do exactly that: page one goes to `/v1` for want of an answer, and page two, now
|
||||
/// holding one, hands `/v2` a token `/v1` minted.
|
||||
///
|
||||
/// `/v1/version` is the question. A server too old to serve it still answers, because the
|
||||
/// version header is on its 404 as well, and a server that is not Phalanx sends no header
|
||||
/// at all, which is the answer for every implementation of the namespace spec: they serve
|
||||
/// `/v1` only.
|
||||
async fn serves_paginated_list(&self) -> Result<bool> {
|
||||
self.serves_paginated_list
|
||||
.get_or_try_init(|| async {
|
||||
let req = self.client.get("/v1/version");
|
||||
// Deliberately not `check_response`: a 404 is a useful answer here, and its
|
||||
// headers carry the version just as a 200's do.
|
||||
let (request_id, rsp) = self.client.send(req).await?;
|
||||
let Some(version) = rsp.headers().get(SERVER_VERSION_HEADER) else {
|
||||
return Ok(false);
|
||||
};
|
||||
let version = version.to_str().map_err(|e| Error::Http {
|
||||
source: e.into(),
|
||||
request_id,
|
||||
status_code: Some(rsp.status()),
|
||||
})?;
|
||||
Ok(ServerVersion::parse(version)?.support_paginated_list_tables())
|
||||
})
|
||||
.await
|
||||
.copied()
|
||||
}
|
||||
|
||||
async fn submit_drop_table(
|
||||
&self,
|
||||
name: &str,
|
||||
@@ -366,6 +412,7 @@ mod test_utils {
|
||||
namespace_headers: HashMap::new(),
|
||||
namespace_context_provider: None,
|
||||
tls_config: None,
|
||||
serves_paginated_list: tokio::sync::OnceCell::new(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -388,6 +435,7 @@ mod test_utils {
|
||||
namespace_headers: config.extra_headers.clone(),
|
||||
namespace_context_provider,
|
||||
tls_config: config.tls_config.clone(),
|
||||
serves_paginated_list: tokio::sync::OnceCell::new(),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -676,9 +724,17 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
|
||||
async fn list_tables(&self, request: ListTablesRequest) -> Result<ListTablesResponse> {
|
||||
let namespace_parts = request.id.as_deref().unwrap_or(&[]);
|
||||
let namespace_id = build_namespace_identifier(namespace_parts, &self.client.id_delimiter);
|
||||
let mut req = self
|
||||
.client
|
||||
.get(&format!("/v1/namespace/{}/table/list", namespace_id));
|
||||
// v1 takes a table name in `page_token` to resume after, which `table_names` callers
|
||||
// build themselves from the last name they saw, so its meaning cannot change. Only a
|
||||
// token the server minted can resume a listing in the store, and that contract needs a
|
||||
// route no name-passing caller reaches. Servers that do not serve it keep the listing
|
||||
// they have always served, which is correct and merely slower.
|
||||
let path = if self.serves_paginated_list().await? {
|
||||
format!("/v2/namespace/{}/table/list", namespace_id)
|
||||
} else {
|
||||
format!("/v1/namespace/{}/table/list", namespace_id)
|
||||
};
|
||||
let mut req = self.client.get(&path);
|
||||
|
||||
if let Some(limit) = request.limit {
|
||||
req = req.query(&[("limit", limit)]);
|
||||
@@ -1137,7 +1193,7 @@ impl From<StorageOptions> for RemoteOptions {
|
||||
// `table_names` is deprecated but still supported, so its tests still call it.
|
||||
#[allow(deprecated)]
|
||||
mod tests {
|
||||
use super::{NamespaceHeaderProviderContext, build_cache_key};
|
||||
use super::{NamespaceHeaderProviderContext, SERVER_VERSION_HEADER, build_cache_key};
|
||||
use std::collections::HashMap;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::{Arc, OnceLock};
|
||||
@@ -1666,6 +1722,125 @@ mod tests {
|
||||
assert_eq!(names, vec!["table1", "table2"]);
|
||||
}
|
||||
|
||||
/// Answer the capability probe with a version, or with no header at all for `None`.
|
||||
fn version_probe(version: Option<&str>) -> http::Response<String> {
|
||||
let builder = http::Response::builder().status(if version.is_some() { 200 } else { 404 });
|
||||
match version {
|
||||
Some(version) => builder
|
||||
.header(SERVER_VERSION_HEADER, version)
|
||||
.body(format!(r#"{{"version": "{version}"}}"#))
|
||||
.unwrap(),
|
||||
None => builder.body(String::new()).unwrap(),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_list_tables_uses_the_opaque_token_route() {
|
||||
// `table_names` keeps /v1, where `page_token` is a table name to resume after. Only
|
||||
// /v2 round-trips a server-minted token, so that is where this API has to go.
|
||||
let conn = Connection::new_with_handler(|request| {
|
||||
if request.url().path() == "/v1/version" {
|
||||
return version_probe(Some("0.6.0"));
|
||||
}
|
||||
assert_eq!(request.url().path(), "/v2/namespace/$/table/list");
|
||||
assert_eq!(request.url().query(), None);
|
||||
|
||||
http::Response::builder()
|
||||
.status(200)
|
||||
.body(r#"{"tables": ["table1"], "page_token": "opaque=="}"#.to_string())
|
||||
.unwrap()
|
||||
});
|
||||
let page = conn.list_tables().execute().await.unwrap();
|
||||
assert_eq!(page.tables, vec!["table1"]);
|
||||
assert_eq!(page.page_token.as_deref(), Some("opaque=="));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_list_tables_resumes_with_the_server_token() {
|
||||
let conn = Connection::new_with_handler(|request| {
|
||||
if request.url().path() == "/v1/version" {
|
||||
return version_probe(Some("0.6.0"));
|
||||
}
|
||||
assert_eq!(request.url().path(), "/v2/namespace/ns1$ns2/table/list");
|
||||
assert_eq!(request.url().query(), Some("page_token=opaque%3D%3D"));
|
||||
|
||||
http::Response::builder()
|
||||
.status(200)
|
||||
.body(r#"{"tables": ["table2"]}"#.to_string())
|
||||
.unwrap()
|
||||
});
|
||||
let page = conn
|
||||
.list_tables()
|
||||
.namespace(vec!["ns1".to_string(), "ns2".to_string()])
|
||||
.page_token("opaque==")
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(page.tables, vec!["table2"]);
|
||||
assert_eq!(page.page_token, None);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_a_server_without_the_route_keeps_the_v1_listing() {
|
||||
// Every implementation of the namespace spec is this case: no version header, and only
|
||||
// /v1 to serve. Asking it for /v2 would 404 a listing that /v1 can answer.
|
||||
let conn = Connection::new_with_handler(|request| {
|
||||
if request.url().path() == "/v1/version" {
|
||||
return version_probe(None);
|
||||
}
|
||||
assert_eq!(request.url().path(), "/v1/namespace/$/table/list");
|
||||
|
||||
http::Response::builder()
|
||||
.status(200)
|
||||
.body(r#"{"tables": ["table1"]}"#.to_string())
|
||||
.unwrap()
|
||||
});
|
||||
let page = conn.list_tables().execute().await.unwrap();
|
||||
assert_eq!(page.tables, vec!["table1"]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_a_server_older_than_the_route_keeps_the_v1_listing() {
|
||||
let conn = Connection::new_with_handler(|request| {
|
||||
if request.url().path() == "/v1/version" {
|
||||
return version_probe(Some("0.5.0"));
|
||||
}
|
||||
assert_eq!(request.url().path(), "/v1/namespace/$/table/list");
|
||||
|
||||
http::Response::builder()
|
||||
.status(200)
|
||||
.body(r#"{"tables": ["table1"]}"#.to_string())
|
||||
.unwrap()
|
||||
});
|
||||
assert_eq!(
|
||||
conn.list_tables().execute().await.unwrap().tables,
|
||||
vec!["table1"]
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_the_probe_is_asked_once_per_connection() {
|
||||
// A walk resumes from what its first page returned, so every page of it has to reach
|
||||
// the same route. Re-asking per request would let a rolling deploy answer differently
|
||||
// mid-walk and hand /v2 a token /v1 minted.
|
||||
let probes = Arc::new(AtomicUsize::new(0));
|
||||
let counted = probes.clone();
|
||||
let conn = Connection::new_with_handler(move |request| {
|
||||
if request.url().path() == "/v1/version" {
|
||||
counted.fetch_add(1, Ordering::SeqCst);
|
||||
return version_probe(Some("0.6.0"));
|
||||
}
|
||||
http::Response::builder()
|
||||
.status(200)
|
||||
.body(r#"{"tables": ["table1"]}"#.to_string())
|
||||
.unwrap()
|
||||
});
|
||||
for _ in 0..3 {
|
||||
conn.list_tables().execute().await.unwrap();
|
||||
}
|
||||
assert_eq!(probes.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_table_names_with_nested_namespace() {
|
||||
// When namespace is vec!["ns1", "ns2"], should use /v1/namespace/ns1$ns2/table/list
|
||||
|
||||
@@ -47,10 +47,15 @@ pub fn stream_as_body(data: SendableRecordBatchStream) -> Result<reqwest::Body>
|
||||
Ok(reqwest::Body::wrap_stream(stream))
|
||||
}
|
||||
|
||||
/// The response header a Phalanx server stamps its version onto.
|
||||
///
|
||||
/// A global layer adds it, so it is on error responses as well as successful ones.
|
||||
pub const SERVER_VERSION_HEADER: &str = "phalanx-version";
|
||||
|
||||
pub fn parse_server_version(req_id: &str, rsp: &Response) -> Result<ServerVersion> {
|
||||
let version = rsp
|
||||
.headers()
|
||||
.get("phalanx-version")
|
||||
.get(SERVER_VERSION_HEADER)
|
||||
.map(|v| {
|
||||
let v = v.to_str().map_err(|e| crate::Error::Http {
|
||||
source: e.into(),
|
||||
|
||||
@@ -4065,6 +4065,16 @@ mod tests {
|
||||
parent_list_calls: self.parent_list_calls.clone(),
|
||||
})
|
||||
}
|
||||
|
||||
// The guard counts listings through `wrap`, so a pushed-down listing would slip
|
||||
// past it.
|
||||
fn wrap_paginated(
|
||||
&self,
|
||||
_store_prefix: &str,
|
||||
_original: Arc<dyn object_store::list::PaginatedListStore>,
|
||||
) -> Option<Arc<dyn object_store::list::PaginatedListStore>> {
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -4168,6 +4178,15 @@ mod tests {
|
||||
self.called.store(true, Ordering::Relaxed);
|
||||
original
|
||||
}
|
||||
|
||||
// Hands the store back untouched, so a listing has nothing to go around.
|
||||
fn wrap_paginated(
|
||||
&self,
|
||||
_store_prefix: &str,
|
||||
original: Arc<dyn object_store::list::PaginatedListStore>,
|
||||
) -> Option<Arc<dyn object_store::list::PaginatedListStore>> {
|
||||
Some(original)
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
Reference in New Issue
Block a user