mirror of
https://github.com/lancedb/lancedb.git
synced 2026-09-13 08:42:20 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 lancedb automation
|
beaf2e35dc |
+1
-1
@@ -1,5 +1,5 @@
|
||||
[tool.bumpversion]
|
||||
current_version = "0.27.1"
|
||||
current_version = "0.27.0-beta.5"
|
||||
parse = """(?x)
|
||||
(?P<major>0|[1-9]\\d*)\\.
|
||||
(?P<minor>0|[1-9]\\d*)\\.
|
||||
|
||||
@@ -7,7 +7,6 @@ on:
|
||||
pull_request:
|
||||
paths:
|
||||
- Cargo.toml
|
||||
- Cargo.lock
|
||||
- nodejs/**
|
||||
- rust/**
|
||||
- docs/src/js/**
|
||||
|
||||
@@ -19,7 +19,6 @@ on:
|
||||
paths:
|
||||
- .github/workflows/npm-publish.yml
|
||||
- Cargo.toml # Change in dependency frequently breaks builds
|
||||
- Cargo.lock
|
||||
|
||||
concurrency:
|
||||
group: ${{ github.workflow }}-${{ github.ref }}
|
||||
@@ -125,12 +124,7 @@ jobs:
|
||||
pre_build: |-
|
||||
set -e &&
|
||||
apt-get update &&
|
||||
apt-get install -y protobuf-compiler pkg-config &&
|
||||
# The base image (manylinux2014-cross) sets TARGET_CC to the old
|
||||
# GCC 4.8 cross-compiler. aws-lc-sys checks TARGET_CC before CC,
|
||||
# so it picks up GCC even though the napi-rs image sets CC=clang.
|
||||
# Override to use the image's clang-18 which supports -fuse-ld=lld.
|
||||
export TARGET_CC=clang TARGET_CXX=clang++
|
||||
apt-get install -y protobuf-compiler pkg-config
|
||||
- target: x86_64-unknown-linux-musl
|
||||
# This one seems to need some extra memory
|
||||
host: ubuntu-2404-8x-x64
|
||||
@@ -150,10 +144,9 @@ jobs:
|
||||
set -e &&
|
||||
apt-get update &&
|
||||
apt-get install -y protobuf-compiler pkg-config &&
|
||||
export TARGET_CC=clang TARGET_CXX=clang++ &&
|
||||
# The manylinux2014 sysroot has glibc 2.17 headers which lack
|
||||
# AT_HWCAP2 (added in Linux 3.17). Define it for aws-lc-sys.
|
||||
export CFLAGS="$CFLAGS -DAT_HWCAP2=26" &&
|
||||
# https://github.com/aws/aws-lc-rs/issues/737#issuecomment-2725918627
|
||||
ln -s /usr/aarch64-unknown-linux-gnu/lib/gcc/aarch64-unknown-linux-gnu/4.8.5/crtbeginS.o /usr/aarch64-unknown-linux-gnu/aarch64-unknown-linux-gnu/sysroot/usr/lib/crtbeginS.o &&
|
||||
ln -s /usr/aarch64-unknown-linux-gnu/lib/gcc /usr/aarch64-unknown-linux-gnu/aarch64-unknown-linux-gnu/sysroot/usr/lib/gcc &&
|
||||
rustup target add aarch64-unknown-linux-gnu
|
||||
- target: aarch64-unknown-linux-musl
|
||||
host: ubuntu-2404-8x-x64
|
||||
@@ -273,7 +266,7 @@ jobs:
|
||||
- target: x86_64-unknown-linux-gnu
|
||||
host: ubuntu-latest
|
||||
- target: aarch64-unknown-linux-gnu
|
||||
host: ubuntu-2404-8x-arm64
|
||||
host: buildjet-16vcpu-ubuntu-2204-arm
|
||||
node:
|
||||
- '20'
|
||||
runs-on: ${{ matrix.settings.host }}
|
||||
|
||||
@@ -9,7 +9,6 @@ on:
|
||||
paths:
|
||||
- .github/workflows/pypi-publish.yml
|
||||
- Cargo.toml # Change in dependency frequently breaks builds
|
||||
- Cargo.lock
|
||||
|
||||
env:
|
||||
PIP_EXTRA_INDEX_URL: "https://pypi.fury.io/lance-format/ https://pypi.fury.io/lancedb/"
|
||||
|
||||
@@ -7,7 +7,6 @@ on:
|
||||
pull_request:
|
||||
paths:
|
||||
- Cargo.toml
|
||||
- Cargo.lock
|
||||
- python/**
|
||||
- rust/**
|
||||
- .github/workflows/python.yml
|
||||
|
||||
@@ -7,7 +7,6 @@ on:
|
||||
pull_request:
|
||||
paths:
|
||||
- Cargo.toml
|
||||
- Cargo.lock
|
||||
- rust/**
|
||||
- .github/workflows/rust.yml
|
||||
|
||||
@@ -207,14 +206,14 @@ jobs:
|
||||
- name: Downgrade dependencies
|
||||
# These packages have newer requirements for MSRV
|
||||
run: |
|
||||
cargo update -p aws-sdk-bedrockruntime --precise 1.77.0
|
||||
cargo update -p aws-sdk-dynamodb --precise 1.68.0
|
||||
cargo update -p aws-config --precise 1.6.0
|
||||
cargo update -p aws-sdk-kms --precise 1.63.0
|
||||
cargo update -p aws-sdk-s3 --precise 1.79.0
|
||||
cargo update -p aws-sdk-sso --precise 1.62.0
|
||||
cargo update -p aws-sdk-ssooidc --precise 1.63.0
|
||||
cargo update -p aws-sdk-sts --precise 1.63.0
|
||||
cargo update -p aws-sdk-bedrockruntime --precise 1.64.0
|
||||
cargo update -p aws-sdk-dynamodb --precise 1.55.0
|
||||
cargo update -p aws-config --precise 1.5.10
|
||||
cargo update -p aws-sdk-kms --precise 1.51.0
|
||||
cargo update -p aws-sdk-s3 --precise 1.65.0
|
||||
cargo update -p aws-sdk-sso --precise 1.50.0
|
||||
cargo update -p aws-sdk-ssooidc --precise 1.51.0
|
||||
cargo update -p aws-sdk-sts --precise 1.51.0
|
||||
cargo update -p home --precise 0.5.9
|
||||
- name: cargo +${{ matrix.msrv }} check
|
||||
env:
|
||||
|
||||
Generated
+43
-45
@@ -3070,8 +3070,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
|
||||
|
||||
[[package]]
|
||||
name = "fsst"
|
||||
version = "4.1.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.1.0-beta.3#244c721504c6ef0b4c2f9700a342509976898d6e"
|
||||
version = "4.0.0-beta.11"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.11#e8109ad0e1bba185a206c7b36daae4d47f587d7a"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"rand 0.9.2",
|
||||
@@ -3852,7 +3852,7 @@ dependencies = [
|
||||
"libc",
|
||||
"percent-encoding",
|
||||
"pin-project-lite",
|
||||
"socket2 0.6.0",
|
||||
"socket2 0.5.10",
|
||||
"system-configuration",
|
||||
"tokio",
|
||||
"tower-service",
|
||||
@@ -4241,8 +4241,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance"
|
||||
version = "4.1.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.1.0-beta.3#244c721504c6ef0b4c2f9700a342509976898d6e"
|
||||
version = "4.0.0-beta.11"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.11#e8109ad0e1bba185a206c7b36daae4d47f587d7a"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-arith",
|
||||
@@ -4308,8 +4308,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-arrow"
|
||||
version = "4.1.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.1.0-beta.3#244c721504c6ef0b4c2f9700a342509976898d6e"
|
||||
version = "4.0.0-beta.11"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.11#e8109ad0e1bba185a206c7b36daae4d47f587d7a"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
@@ -4329,8 +4329,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-bitpacking"
|
||||
version = "4.1.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.1.0-beta.3#244c721504c6ef0b4c2f9700a342509976898d6e"
|
||||
version = "4.0.0-beta.11"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.11#e8109ad0e1bba185a206c7b36daae4d47f587d7a"
|
||||
dependencies = [
|
||||
"arrayref",
|
||||
"paste",
|
||||
@@ -4339,8 +4339,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-core"
|
||||
version = "4.1.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.1.0-beta.3#244c721504c6ef0b4c2f9700a342509976898d6e"
|
||||
version = "4.0.0-beta.11"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.11#e8109ad0e1bba185a206c7b36daae4d47f587d7a"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
@@ -4377,8 +4377,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-datafusion"
|
||||
version = "4.1.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.1.0-beta.3#244c721504c6ef0b4c2f9700a342509976898d6e"
|
||||
version = "4.0.0-beta.11"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.11#e8109ad0e1bba185a206c7b36daae4d47f587d7a"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-array",
|
||||
@@ -4408,8 +4408,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-datagen"
|
||||
version = "4.1.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.1.0-beta.3#244c721504c6ef0b4c2f9700a342509976898d6e"
|
||||
version = "4.0.0-beta.11"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.11#e8109ad0e1bba185a206c7b36daae4d47f587d7a"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-array",
|
||||
@@ -4427,8 +4427,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-encoding"
|
||||
version = "4.1.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.1.0-beta.3#244c721504c6ef0b4c2f9700a342509976898d6e"
|
||||
version = "4.0.0-beta.11"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.11#e8109ad0e1bba185a206c7b36daae4d47f587d7a"
|
||||
dependencies = [
|
||||
"arrow-arith",
|
||||
"arrow-array",
|
||||
@@ -4465,8 +4465,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-file"
|
||||
version = "4.1.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.1.0-beta.3#244c721504c6ef0b4c2f9700a342509976898d6e"
|
||||
version = "4.0.0-beta.11"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.11#e8109ad0e1bba185a206c7b36daae4d47f587d7a"
|
||||
dependencies = [
|
||||
"arrow-arith",
|
||||
"arrow-array",
|
||||
@@ -4498,8 +4498,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-index"
|
||||
version = "4.1.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.1.0-beta.3#244c721504c6ef0b4c2f9700a342509976898d6e"
|
||||
version = "4.0.0-beta.11"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.11#e8109ad0e1bba185a206c7b36daae4d47f587d7a"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-arith",
|
||||
@@ -4513,7 +4513,6 @@ dependencies = [
|
||||
"bitpacking",
|
||||
"bitvec",
|
||||
"bytes",
|
||||
"chrono",
|
||||
"crossbeam-queue",
|
||||
"datafusion",
|
||||
"datafusion-common",
|
||||
@@ -4563,8 +4562,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-io"
|
||||
version = "4.1.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.1.0-beta.3#244c721504c6ef0b4c2f9700a342509976898d6e"
|
||||
version = "4.0.0-beta.11"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.11#e8109ad0e1bba185a206c7b36daae4d47f587d7a"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-arith",
|
||||
@@ -4605,8 +4604,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-linalg"
|
||||
version = "4.1.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.1.0-beta.3#244c721504c6ef0b4c2f9700a342509976898d6e"
|
||||
version = "4.0.0-beta.11"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.11#e8109ad0e1bba185a206c7b36daae4d47f587d7a"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
@@ -4622,22 +4621,21 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-namespace"
|
||||
version = "4.1.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.1.0-beta.3#244c721504c6ef0b4c2f9700a342509976898d6e"
|
||||
version = "4.0.0-beta.11"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.11#e8109ad0e1bba185a206c7b36daae4d47f587d7a"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"async-trait",
|
||||
"bytes",
|
||||
"lance-core",
|
||||
"lance-namespace-reqwest-client",
|
||||
"serde",
|
||||
"snafu 0.9.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "lance-namespace-impls"
|
||||
version = "4.1.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.1.0-beta.3#244c721504c6ef0b4c2f9700a342509976898d6e"
|
||||
version = "4.0.0-beta.11"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.11#e8109ad0e1bba185a206c7b36daae4d47f587d7a"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-ipc",
|
||||
@@ -4668,9 +4666,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-namespace-reqwest-client"
|
||||
version = "0.6.1"
|
||||
version = "0.5.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ee2e48de899e2931afb67fcddd0a08e439bf5d8b6ea2a2ed9cb8f4df669bd5cc"
|
||||
checksum = "3ad4c947349acd6e37e984eba0254588bd894e6128434338b9e6904e56fb4633"
|
||||
dependencies = [
|
||||
"reqwest",
|
||||
"serde",
|
||||
@@ -4681,8 +4679,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-table"
|
||||
version = "4.1.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.1.0-beta.3#244c721504c6ef0b4c2f9700a342509976898d6e"
|
||||
version = "4.0.0-beta.11"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.11#e8109ad0e1bba185a206c7b36daae4d47f587d7a"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-array",
|
||||
@@ -4721,8 +4719,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-testing"
|
||||
version = "4.1.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.1.0-beta.3#244c721504c6ef0b4c2f9700a342509976898d6e"
|
||||
version = "4.0.0-beta.11"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.11#e8109ad0e1bba185a206c7b36daae4d47f587d7a"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-schema",
|
||||
@@ -4733,7 +4731,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lancedb"
|
||||
version = "0.27.1"
|
||||
version = "0.27.0-beta.5"
|
||||
dependencies = [
|
||||
"ahash",
|
||||
"anyhow",
|
||||
@@ -4815,7 +4813,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lancedb-nodejs"
|
||||
version = "0.27.1"
|
||||
version = "0.27.0-beta.5"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-ipc",
|
||||
@@ -4835,7 +4833,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lancedb-python"
|
||||
version = "0.30.1"
|
||||
version = "0.30.0-beta.5"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"async-trait",
|
||||
@@ -6445,7 +6443,7 @@ version = "0.14.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "343d3bd7056eda839b03204e68deff7d1b13aba7af2b2fd16890697274262ee7"
|
||||
dependencies = [
|
||||
"heck 0.5.0",
|
||||
"heck 0.4.1",
|
||||
"itertools 0.14.0",
|
||||
"log",
|
||||
"multimap",
|
||||
@@ -6634,7 +6632,7 @@ dependencies = [
|
||||
"quinn-udp",
|
||||
"rustc-hash",
|
||||
"rustls 0.23.31",
|
||||
"socket2 0.6.0",
|
||||
"socket2 0.5.10",
|
||||
"thiserror 2.0.17",
|
||||
"tokio",
|
||||
"tracing",
|
||||
@@ -6671,7 +6669,7 @@ dependencies = [
|
||||
"cfg_aliases",
|
||||
"libc",
|
||||
"once_cell",
|
||||
"socket2 0.6.0",
|
||||
"socket2 0.5.10",
|
||||
"tracing",
|
||||
"windows-sys 0.60.2",
|
||||
]
|
||||
@@ -7737,7 +7735,7 @@ version = "0.8.9"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c1c97747dbf44bb1ca44a561ece23508e99cb592e862f22222dcf42f51d1e451"
|
||||
dependencies = [
|
||||
"heck 0.5.0",
|
||||
"heck 0.4.1",
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.114",
|
||||
@@ -7749,7 +7747,7 @@ version = "0.9.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "54254b8531cafa275c5e096f62d48c81435d1015405a91198ddb11e967301d40"
|
||||
dependencies = [
|
||||
"heck 0.5.0",
|
||||
"heck 0.4.1",
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.114",
|
||||
|
||||
+14
-14
@@ -15,20 +15,20 @@ categories = ["database-implementations"]
|
||||
rust-version = "1.91.0"
|
||||
|
||||
[workspace.dependencies]
|
||||
lance = { "version" = "=4.1.0-beta.3", default-features = false, "tag" = "v4.1.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-core = { "version" = "=4.1.0-beta.3", "tag" = "v4.1.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datagen = { "version" = "=4.1.0-beta.3", "tag" = "v4.1.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-file = { "version" = "=4.1.0-beta.3", "tag" = "v4.1.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-io = { "version" = "=4.1.0-beta.3", default-features = false, "tag" = "v4.1.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-index = { "version" = "=4.1.0-beta.3", "tag" = "v4.1.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-linalg = { "version" = "=4.1.0-beta.3", "tag" = "v4.1.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace = { "version" = "=4.1.0-beta.3", "tag" = "v4.1.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace-impls = { "version" = "=4.1.0-beta.3", default-features = false, "tag" = "v4.1.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-table = { "version" = "=4.1.0-beta.3", "tag" = "v4.1.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-testing = { "version" = "=4.1.0-beta.3", "tag" = "v4.1.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datafusion = { "version" = "=4.1.0-beta.3", "tag" = "v4.1.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-encoding = { "version" = "=4.1.0-beta.3", "tag" = "v4.1.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-arrow = { "version" = "=4.1.0-beta.3", "tag" = "v4.1.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance = { "version" = "=4.0.0-beta.11", default-features = false, "tag" = "v4.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-core = { "version" = "=4.0.0-beta.11", "tag" = "v4.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datagen = { "version" = "=4.0.0-beta.11", "tag" = "v4.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-file = { "version" = "=4.0.0-beta.11", "tag" = "v4.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-io = { "version" = "=4.0.0-beta.11", default-features = false, "tag" = "v4.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-index = { "version" = "=4.0.0-beta.11", "tag" = "v4.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-linalg = { "version" = "=4.0.0-beta.11", "tag" = "v4.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace = { "version" = "=4.0.0-beta.11", "tag" = "v4.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace-impls = { "version" = "=4.0.0-beta.11", default-features = false, "tag" = "v4.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-table = { "version" = "=4.0.0-beta.11", "tag" = "v4.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-testing = { "version" = "=4.0.0-beta.11", "tag" = "v4.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datafusion = { "version" = "=4.0.0-beta.11", "tag" = "v4.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-encoding = { "version" = "=4.0.0-beta.11", "tag" = "v4.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-arrow = { "version" = "=4.0.0-beta.11", "tag" = "v4.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
|
||||
ahash = "0.8"
|
||||
# Note that this one does not include pyarrow
|
||||
arrow = { version = "57.2", optional = false }
|
||||
|
||||
@@ -3,7 +3,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import functools
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
@@ -27,7 +26,6 @@ SEMVER_RE = re.compile(
|
||||
)
|
||||
|
||||
|
||||
@functools.total_ordering
|
||||
@dataclass(frozen=True)
|
||||
class SemVer:
|
||||
major: int
|
||||
@@ -158,9 +156,7 @@ def read_current_version(repo_root: Path) -> str:
|
||||
|
||||
|
||||
def determine_latest_tag(tags: Iterable[TagInfo]) -> TagInfo:
|
||||
# Stable releases (no prerelease) are always preferred over pre-releases.
|
||||
# Within each group, standard semver ordering applies.
|
||||
return max(tags, key=lambda tag: (not tag.semver.prerelease, tag.semver))
|
||||
return max(tags, key=lambda tag: tag.semver)
|
||||
|
||||
|
||||
def write_outputs(args: argparse.Namespace, payload: dict) -> None:
|
||||
|
||||
+1
-1
@@ -1,7 +1,7 @@
|
||||
version: "3.9"
|
||||
services:
|
||||
localstack:
|
||||
image: localstack/localstack:4.0
|
||||
image: localstack/localstack:3.3
|
||||
ports:
|
||||
- 4566:4566
|
||||
environment:
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
mkdocs==1.6.1
|
||||
mkdocs==1.5.3
|
||||
mkdocs-jupyter==0.24.1
|
||||
mkdocs-material==9.6.23
|
||||
mkdocs-material==9.5.3
|
||||
mkdocs-autorefs>=0.5,<=1.0
|
||||
mkdocstrings[python]>=0.24,<1.0
|
||||
mkdocstrings[python]==0.25.2
|
||||
griffe>=0.40,<1.0
|
||||
mkdocs-render-swagger-plugin>=0.1.0
|
||||
pydantic>=2.0,<3.0
|
||||
|
||||
@@ -14,7 +14,7 @@ Add the following dependency to your `pom.xml`:
|
||||
<dependency>
|
||||
<groupId>com.lancedb</groupId>
|
||||
<artifactId>lancedb-core</artifactId>
|
||||
<version>0.27.1</version>
|
||||
<version>0.27.0-beta.5</version>
|
||||
</dependency>
|
||||
```
|
||||
|
||||
|
||||
@@ -485,7 +485,19 @@ Modeled after ``VACUUM`` in PostgreSQL.
|
||||
- Prune: Removes old versions of the dataset
|
||||
- Index: Optimizes the indices, adding new data to existing indices
|
||||
|
||||
The frequency an application should call optimize is based on the frequency of
|
||||
Experimental API
|
||||
----------------
|
||||
|
||||
The optimization process is undergoing active development and may change.
|
||||
Our goal with these changes is to improve the performance of optimization and
|
||||
reduce the complexity.
|
||||
|
||||
That being said, it is essential today to run optimize if you want the best
|
||||
performance. It should be stable and safe to use in production, but it our
|
||||
hope that the API may be simplified (or not even need to be called) in the
|
||||
future.
|
||||
|
||||
The frequency an application shoudl call optimize is based on the frequency of
|
||||
data modifications. If data is frequently added, deleted, or updated then
|
||||
optimize should be run frequently. A good rule of thumb is to run optimize if
|
||||
you have added or modified 100,000 or more records or run more than 20 data
|
||||
|
||||
@@ -37,12 +37,3 @@ tbl.optimize({cleanupOlderThan: new Date()});
|
||||
```ts
|
||||
deleteUnverified: boolean;
|
||||
```
|
||||
|
||||
Because they may be part of an in-progress transaction, files newer than
|
||||
7 days old are not deleted by default. If you are sure that there are no
|
||||
in-progress transactions, then you can set this to true to delete all
|
||||
files older than `cleanupOlderThan`.
|
||||
|
||||
**WARNING**: This should only be set to true if you can guarantee that
|
||||
no other process is currently working on this dataset. Otherwise the
|
||||
dataset could be put into a corrupted state.
|
||||
|
||||
@@ -8,7 +8,7 @@
|
||||
<parent>
|
||||
<groupId>com.lancedb</groupId>
|
||||
<artifactId>lancedb-parent</artifactId>
|
||||
<version>0.27.1-final.0</version>
|
||||
<version>0.27.0-beta.5</version>
|
||||
<relativePath>../pom.xml</relativePath>
|
||||
</parent>
|
||||
|
||||
|
||||
+2
-2
@@ -6,7 +6,7 @@
|
||||
|
||||
<groupId>com.lancedb</groupId>
|
||||
<artifactId>lancedb-parent</artifactId>
|
||||
<version>0.27.1-final.0</version>
|
||||
<version>0.27.0-beta.5</version>
|
||||
<packaging>pom</packaging>
|
||||
<name>${project.artifactId}</name>
|
||||
<description>LanceDB Java SDK Parent POM</description>
|
||||
@@ -28,7 +28,7 @@
|
||||
<properties>
|
||||
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
|
||||
<arrow.version>15.0.0</arrow.version>
|
||||
<lance-core.version>4.1.0-beta.3</lance-core.version>
|
||||
<lance-core.version>4.0.0-beta.11</lance-core.version>
|
||||
<spotless.skip>false</spotless.skip>
|
||||
<spotless.version>2.30.0</spotless.version>
|
||||
<spotless.java.googlejavaformat.version>1.7</spotless.java.googlejavaformat.version>
|
||||
|
||||
+1
-1
@@ -1,7 +1,7 @@
|
||||
[package]
|
||||
name = "lancedb-nodejs"
|
||||
edition.workspace = true
|
||||
version = "0.27.1"
|
||||
version = "0.27.0-beta.5"
|
||||
license.workspace = true
|
||||
description.workspace = true
|
||||
repository.workspace = true
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@lancedb/lancedb-darwin-arm64",
|
||||
"version": "0.27.1",
|
||||
"version": "0.27.0-beta.5",
|
||||
"os": ["darwin"],
|
||||
"cpu": ["arm64"],
|
||||
"main": "lancedb.darwin-arm64.node",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@lancedb/lancedb-linux-arm64-gnu",
|
||||
"version": "0.27.1",
|
||||
"version": "0.27.0-beta.5",
|
||||
"os": ["linux"],
|
||||
"cpu": ["arm64"],
|
||||
"main": "lancedb.linux-arm64-gnu.node",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@lancedb/lancedb-linux-arm64-musl",
|
||||
"version": "0.27.1",
|
||||
"version": "0.27.0-beta.5",
|
||||
"os": ["linux"],
|
||||
"cpu": ["arm64"],
|
||||
"main": "lancedb.linux-arm64-musl.node",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@lancedb/lancedb-linux-x64-gnu",
|
||||
"version": "0.27.1",
|
||||
"version": "0.27.0-beta.5",
|
||||
"os": ["linux"],
|
||||
"cpu": ["x64"],
|
||||
"main": "lancedb.linux-x64-gnu.node",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@lancedb/lancedb-linux-x64-musl",
|
||||
"version": "0.27.1",
|
||||
"version": "0.27.0-beta.5",
|
||||
"os": ["linux"],
|
||||
"cpu": ["x64"],
|
||||
"main": "lancedb.linux-x64-musl.node",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@lancedb/lancedb-win32-arm64-msvc",
|
||||
"version": "0.27.1",
|
||||
"version": "0.27.0-beta.5",
|
||||
"os": [
|
||||
"win32"
|
||||
],
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@lancedb/lancedb-win32-x64-msvc",
|
||||
"version": "0.27.1",
|
||||
"version": "0.27.0-beta.5",
|
||||
"os": ["win32"],
|
||||
"cpu": ["x64"],
|
||||
"main": "lancedb.win32-x64-msvc.node",
|
||||
|
||||
Generated
+2
-2
@@ -1,12 +1,12 @@
|
||||
{
|
||||
"name": "@lancedb/lancedb",
|
||||
"version": "0.27.1",
|
||||
"version": "0.27.0-beta.5",
|
||||
"lockfileVersion": 3,
|
||||
"requires": true,
|
||||
"packages": {
|
||||
"": {
|
||||
"name": "@lancedb/lancedb",
|
||||
"version": "0.27.1",
|
||||
"version": "0.27.0-beta.5",
|
||||
"cpu": [
|
||||
"x64",
|
||||
"arm64"
|
||||
|
||||
+1
-1
@@ -11,7 +11,7 @@
|
||||
"ann"
|
||||
],
|
||||
"private": false,
|
||||
"version": "0.27.1",
|
||||
"version": "0.27.0-beta.5",
|
||||
"main": "dist/index.js",
|
||||
"exports": {
|
||||
".": "./dist/index.js",
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
[tool.bumpversion]
|
||||
current_version = "0.30.1"
|
||||
current_version = "0.30.0-beta.5"
|
||||
parse = """(?x)
|
||||
(?P<major>0|[1-9]\\d*)\\.
|
||||
(?P<minor>0|[1-9]\\d*)\\.
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "lancedb-python"
|
||||
version = "0.30.1"
|
||||
version = "0.30.0-beta.5"
|
||||
edition.workspace = true
|
||||
description = "Python bindings for LanceDB"
|
||||
license.workspace = true
|
||||
|
||||
@@ -2205,8 +2205,8 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
|
||||
self._vector_query.select(self._columns)
|
||||
self._fts_query.select(self._columns)
|
||||
if self._where:
|
||||
self._vector_query.where(self._where, not self._postfilter)
|
||||
self._fts_query.where(self._where, not self._postfilter)
|
||||
self._vector_query.where(self._where, self._postfilter)
|
||||
self._fts_query.where(self._where, self._postfilter)
|
||||
if self._with_row_id:
|
||||
self._vector_query.with_row_id(True)
|
||||
self._fts_query.with_row_id(True)
|
||||
|
||||
@@ -4751,16 +4751,7 @@ class IndexStatistics:
|
||||
num_indexed_rows: int
|
||||
num_unindexed_rows: int
|
||||
index_type: Literal[
|
||||
"IVF_FLAT",
|
||||
"IVF_SQ",
|
||||
"IVF_PQ",
|
||||
"IVF_RQ",
|
||||
"IVF_HNSW_SQ",
|
||||
"IVF_HNSW_PQ",
|
||||
"FTS",
|
||||
"BTREE",
|
||||
"BITMAP",
|
||||
"LABEL_LIST",
|
||||
"IVF_PQ", "IVF_HNSW_PQ", "IVF_HNSW_SQ", "FTS", "BTREE", "BITMAP", "LABEL_LIST"
|
||||
]
|
||||
distance_type: Optional[Literal["l2", "cosine", "dot"]] = None
|
||||
num_indices: Optional[int] = None
|
||||
|
||||
@@ -177,60 +177,6 @@ async def test_analyze_plan(table: AsyncTable):
|
||||
assert "metrics=" in res
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def table_with_id(tmpdir_factory) -> Table:
|
||||
tmp_path = str(tmpdir_factory.mktemp("data"))
|
||||
db = lancedb.connect(tmp_path)
|
||||
data = pa.table(
|
||||
{
|
||||
"id": pa.array([1, 2, 3, 4], type=pa.int64()),
|
||||
"text": pa.array(["a", "b", "cat", "dog"]),
|
||||
"vector": pa.array(
|
||||
[[0.1, 0.1], [2, 2], [-0.1, -0.1], [0.5, -0.5]],
|
||||
type=pa.list_(pa.float32(), list_size=2),
|
||||
),
|
||||
}
|
||||
)
|
||||
table = db.create_table("test_with_id", data)
|
||||
table.create_fts_index("text", with_position=False, use_tantivy=False)
|
||||
return table
|
||||
|
||||
|
||||
def test_hybrid_prefilter_explain_plan(table_with_id: Table):
|
||||
"""
|
||||
Verify that the prefilter logic is not inverted in LanceHybridQueryBuilder.
|
||||
"""
|
||||
plan_prefilter = (
|
||||
table_with_id.search(query_type="hybrid")
|
||||
.vector([0.0, 0.0])
|
||||
.text("dog")
|
||||
.where("id = 1", prefilter=True)
|
||||
.limit(2)
|
||||
.explain_plan(verbose=True)
|
||||
)
|
||||
|
||||
plan_postfilter = (
|
||||
table_with_id.search(query_type="hybrid")
|
||||
.vector([0.0, 0.0])
|
||||
.text("dog")
|
||||
.where("id = 1", prefilter=False)
|
||||
.limit(2)
|
||||
.explain_plan(verbose=True)
|
||||
)
|
||||
|
||||
# prefilter=True: filter is pushed into the LanceRead scan.
|
||||
# The FTS sub-plan exposes this as "full_filter=id = Int64(1)" inside LanceRead.
|
||||
assert "full_filter=id = Int64(1)" in plan_prefilter, (
|
||||
f"Should push the filter into the scan.\nPlan:\n{plan_prefilter}"
|
||||
)
|
||||
|
||||
# prefilter=False: filter is applied as a separate FilterExec after the search.
|
||||
# The filter must NOT be embedded in the scan.
|
||||
assert "full_filter=id = Int64(1)" not in plan_postfilter, (
|
||||
f"Should NOT push the filter into the scan.\nPlan:\n{plan_postfilter}"
|
||||
)
|
||||
|
||||
|
||||
def test_normalize_scores():
|
||||
cases = [
|
||||
(pa.array([0.1, 0.4]), pa.array([0.0, 1.0])),
|
||||
|
||||
@@ -3,7 +3,6 @@
|
||||
|
||||
from datetime import timedelta
|
||||
import random
|
||||
from typing import get_args, get_type_hints
|
||||
|
||||
import pyarrow as pa
|
||||
import pytest
|
||||
@@ -23,7 +22,6 @@ from lancedb.index import (
|
||||
HnswSq,
|
||||
FTS,
|
||||
)
|
||||
from lancedb.table import IndexStatistics
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
@@ -285,23 +283,3 @@ async def test_create_index_with_binary_vectors(binary_table: AsyncTable):
|
||||
for v in range(256):
|
||||
res = await binary_table.query().nearest_to([v] * 128).to_arrow()
|
||||
assert res["id"][0].as_py() == v
|
||||
|
||||
|
||||
def test_index_statistics_index_type_lists_all_supported_values():
|
||||
expected_index_types = {
|
||||
"IVF_FLAT",
|
||||
"IVF_SQ",
|
||||
"IVF_PQ",
|
||||
"IVF_RQ",
|
||||
"IVF_HNSW_SQ",
|
||||
"IVF_HNSW_PQ",
|
||||
"FTS",
|
||||
"BTREE",
|
||||
"BITMAP",
|
||||
"LABEL_LIST",
|
||||
}
|
||||
|
||||
assert (
|
||||
set(get_args(get_type_hints(IndexStatistics)["index_type"]))
|
||||
== expected_index_types
|
||||
)
|
||||
|
||||
@@ -147,12 +147,7 @@ class TrackingNamespace(LanceNamespace):
|
||||
This simulates a credential rotation system where each call returns
|
||||
new credentials that expire after credential_expires_in_seconds.
|
||||
"""
|
||||
# Start from base storage options (endpoint, region, allow_http, etc.)
|
||||
# because DirectoryNamespace returns None for storage_options from
|
||||
# describe_table/declare_table when no credential vendor is configured.
|
||||
modified = copy.deepcopy(self.base_storage_options)
|
||||
if storage_options:
|
||||
modified.update(storage_options)
|
||||
modified = copy.deepcopy(storage_options) if storage_options else {}
|
||||
|
||||
# Increment credentials to simulate rotation
|
||||
modified["aws_access_key_id"] = f"AKID_{count}"
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "lancedb"
|
||||
version = "0.27.1"
|
||||
version = "0.27.0-beta.5"
|
||||
edition.workspace = true
|
||||
description = "LanceDB: A serverless, low-latency vector database for AI applications"
|
||||
license.workspace = true
|
||||
|
||||
@@ -72,10 +72,6 @@ impl ServerVersion {
|
||||
pub fn support_structural_fts(&self) -> bool {
|
||||
self.0 >= semver::Version::new(0, 3, 0)
|
||||
}
|
||||
|
||||
pub fn support_multipart_write(&self) -> bool {
|
||||
self.0 >= semver::Version::new(0, 4, 0)
|
||||
}
|
||||
}
|
||||
|
||||
pub const OPT_REMOTE_PREFIX: &str = "remote_database_";
|
||||
|
||||
@@ -10,7 +10,6 @@ use super::ARROW_STREAM_CONTENT_TYPE;
|
||||
use super::client::RequestResultExt;
|
||||
use super::client::{HttpSend, RestfulLanceDbClient, Sender};
|
||||
use super::db::ServerVersion;
|
||||
use crate::data::scannable::{PeekedScannable, Scannable, estimate_write_partitions};
|
||||
use crate::index::Index;
|
||||
use crate::index::IndexStatistics;
|
||||
use crate::index::waiter::wait_for_index;
|
||||
@@ -24,7 +23,7 @@ use crate::table::MergeResult;
|
||||
use crate::table::Tags;
|
||||
use crate::table::UpdateResult;
|
||||
use crate::table::query::create_multi_vector_plan;
|
||||
use crate::table::{AnyQuery, Filter, PreprocessingOutput, TableStatistics};
|
||||
use crate::table::{AnyQuery, Filter, TableStatistics};
|
||||
use crate::utils::background_cache::BackgroundCache;
|
||||
use crate::utils::{supported_btree_data_type, supported_vector_data_type};
|
||||
use crate::{DistanceType, Error};
|
||||
@@ -44,7 +43,7 @@ use async_trait::async_trait;
|
||||
use datafusion_common::DataFusionError;
|
||||
use datafusion_physical_plan::stream::RecordBatchStreamAdapter;
|
||||
use datafusion_physical_plan::{ExecutionPlan, RecordBatchStream, SendableRecordBatchStream};
|
||||
use futures::{StreamExt, TryStreamExt};
|
||||
use futures::TryStreamExt;
|
||||
use http::header::CONTENT_TYPE;
|
||||
use http::{HeaderName, StatusCode};
|
||||
use lance::arrow::json::{JsonDataType, JsonSchema};
|
||||
@@ -615,66 +614,6 @@ impl<S: HttpSend> RemoteTable<S> {
|
||||
Ok(bodies)
|
||||
}
|
||||
|
||||
async fn create_multipart_write(&self) -> Result<String> {
|
||||
let request = self.client.post(&format!(
|
||||
"/v1/table/{}/multipart_write/create",
|
||||
self.identifier
|
||||
));
|
||||
let (request_id, response) = self.send(request, true).await?;
|
||||
let response = self.check_table_response(&request_id, response).await?;
|
||||
let body = response.text().await.err_to_http(request_id.clone())?;
|
||||
let parsed: serde_json::Value = serde_json::from_str(&body).map_err(|e| Error::Http {
|
||||
source: format!("Failed to parse multipart create response: {}", e).into(),
|
||||
request_id,
|
||||
status_code: None,
|
||||
})?;
|
||||
parsed["upload_id"]
|
||||
.as_str()
|
||||
.map(|s| s.to_string())
|
||||
.ok_or_else(|| Error::Http {
|
||||
source: "Missing upload_id in multipart create response".into(),
|
||||
request_id: String::new(),
|
||||
status_code: None,
|
||||
})
|
||||
}
|
||||
|
||||
async fn complete_multipart_write(&self, upload_id: &str) -> Result<AddResult> {
|
||||
let request = self
|
||||
.client
|
||||
.post(&format!(
|
||||
"/v1/table/{}/multipart_write/complete",
|
||||
self.identifier
|
||||
))
|
||||
.query(&[("upload_id", upload_id)]);
|
||||
let (request_id, response) = self.send(request, true).await?;
|
||||
let response = self.check_table_response(&request_id, response).await?;
|
||||
let body = response.text().await.err_to_http(request_id.clone())?;
|
||||
let parsed: serde_json::Value = serde_json::from_str(&body).map_err(|e| Error::Http {
|
||||
source: format!("Failed to parse multipart complete response: {}", e).into(),
|
||||
request_id,
|
||||
status_code: None,
|
||||
})?;
|
||||
let version = parsed["version"].as_u64().ok_or_else(|| Error::Http {
|
||||
source: "Missing version in multipart complete response".into(),
|
||||
request_id: String::new(),
|
||||
status_code: None,
|
||||
})?;
|
||||
Ok(AddResult { version })
|
||||
}
|
||||
|
||||
async fn abort_multipart_write(&self, upload_id: &str) -> Result<()> {
|
||||
let request = self
|
||||
.client
|
||||
.post(&format!(
|
||||
"/v1/table/{}/multipart_write/abort",
|
||||
self.identifier
|
||||
))
|
||||
.query(&[("upload_id", upload_id)]);
|
||||
let (request_id, response) = self.send(request, true).await?;
|
||||
self.check_table_response(&request_id, response).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn check_mutable(&self) -> Result<()> {
|
||||
let read_guard = self.version.read().await;
|
||||
match *read_guard {
|
||||
@@ -878,19 +817,6 @@ mod test_utils {
|
||||
}
|
||||
|
||||
pub fn new_mock_with_config<F, T>(name: String, handler: F, config: ClientConfig) -> Self
|
||||
where
|
||||
F: Fn(reqwest::Request) -> http::Response<T> + Send + Sync + 'static,
|
||||
T: Into<reqwest::Body>,
|
||||
{
|
||||
Self::new_mock_with_version_and_config(name, handler, None, config)
|
||||
}
|
||||
|
||||
pub fn new_mock_with_version_and_config<F, T>(
|
||||
name: String,
|
||||
handler: F,
|
||||
version: Option<semver::Version>,
|
||||
config: ClientConfig,
|
||||
) -> Self
|
||||
where
|
||||
F: Fn(reqwest::Request) -> http::Response<T> + Send + Sync + 'static,
|
||||
T: Into<reqwest::Body>,
|
||||
@@ -901,7 +827,7 @@ mod test_utils {
|
||||
name: name.clone(),
|
||||
namespace: vec![],
|
||||
identifier: name,
|
||||
server_version: version.map(ServerVersion).unwrap_or_default(),
|
||||
server_version: ServerVersion::default(),
|
||||
version: RwLock::new(None),
|
||||
location: RwLock::new(None),
|
||||
schema_cache: BackgroundCache::new(SCHEMA_CACHE_TTL, SCHEMA_CACHE_REFRESH_WINDOW),
|
||||
@@ -910,185 +836,6 @@ mod test_utils {
|
||||
}
|
||||
}
|
||||
|
||||
impl<S: HttpSend + 'static> RemoteTable<S> {
|
||||
fn is_retryable_write_error(&self, err: &Error) -> bool {
|
||||
match err {
|
||||
Error::Http {
|
||||
source,
|
||||
status_code,
|
||||
..
|
||||
} => {
|
||||
// Don't retry read errors (is_body/is_decode): the
|
||||
// server may have committed the write already, and
|
||||
// without an idempotency key we'd duplicate data.
|
||||
source
|
||||
.downcast_ref::<reqwest::Error>()
|
||||
.is_some_and(|e| e.is_connect())
|
||||
|| status_code.is_some_and(|s| self.client.retry_config.statuses.contains(&s))
|
||||
}
|
||||
// send_with_retry exhausted its internal retries on a retryable
|
||||
// status. The outer loop can still retry the whole operation with
|
||||
// a fresh session.
|
||||
Error::Retry { status_code, .. } => {
|
||||
status_code.is_some_and(|s| self.client.retry_config.statuses.contains(&s))
|
||||
}
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
|
||||
async fn add_single_partition(&self, output: PreprocessingOutput) -> Result<AddResult> {
|
||||
use crate::remote::retry::RetryCounter;
|
||||
|
||||
let mut insert: Arc<dyn ExecutionPlan> = Arc::new(RemoteInsertExec::new(
|
||||
self.name.clone(),
|
||||
self.identifier.clone(),
|
||||
self.client.clone(),
|
||||
output.plan,
|
||||
output.overwrite,
|
||||
));
|
||||
|
||||
let mut retry_counter =
|
||||
RetryCounter::new(&self.client.retry_config, uuid::Uuid::new_v4().to_string());
|
||||
|
||||
loop {
|
||||
let stream = execute_plan(insert.clone(), Default::default())?;
|
||||
let result: Result<Vec<_>> = stream.try_collect().await.map_err(Error::from);
|
||||
|
||||
match result {
|
||||
Ok(_) => {
|
||||
let add_result = insert
|
||||
.as_any()
|
||||
.downcast_ref::<RemoteInsertExec<S>>()
|
||||
.and_then(|i| i.add_result())
|
||||
.unwrap_or(AddResult { version: 0 });
|
||||
|
||||
if output.overwrite {
|
||||
self.invalidate_schema_cache();
|
||||
}
|
||||
|
||||
return Ok(add_result);
|
||||
}
|
||||
Err(err) if output.rescannable && self.is_retryable_write_error(&err) => {
|
||||
retry_counter.increment_from_error(err)?;
|
||||
tokio::time::sleep(retry_counter.next_sleep_time()).await;
|
||||
insert = insert.reset_state()?;
|
||||
continue;
|
||||
}
|
||||
Err(err) => return Err(err),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn add_multipart(
|
||||
&self,
|
||||
output: PreprocessingOutput,
|
||||
num_partitions: usize,
|
||||
) -> Result<AddResult> {
|
||||
use crate::remote::retry::RetryCounter;
|
||||
|
||||
let mut retry_counter =
|
||||
RetryCounter::new(&self.client.retry_config, uuid::Uuid::new_v4().to_string());
|
||||
|
||||
loop {
|
||||
let upload_id = self.create_multipart_write().await?;
|
||||
|
||||
let result = self
|
||||
.execute_multipart_inserts(&upload_id, &output, num_partitions)
|
||||
.await;
|
||||
|
||||
match result {
|
||||
Ok(()) => match self.complete_multipart_write(&upload_id).await {
|
||||
Ok(result) => {
|
||||
if output.overwrite {
|
||||
self.invalidate_schema_cache();
|
||||
}
|
||||
return Ok(result);
|
||||
}
|
||||
Err(e) => {
|
||||
if let Err(abort_err) = self.abort_multipart_write(&upload_id).await {
|
||||
log::warn!(
|
||||
"Failed to abort multipart write {}: {}",
|
||||
upload_id,
|
||||
abort_err
|
||||
);
|
||||
}
|
||||
if output.rescannable && self.is_retryable_write_error(&e) {
|
||||
retry_counter.increment_from_error(e)?;
|
||||
tokio::time::sleep(retry_counter.next_sleep_time()).await;
|
||||
continue;
|
||||
}
|
||||
return Err(e);
|
||||
}
|
||||
},
|
||||
Err(e) => {
|
||||
if let Err(abort_err) = self.abort_multipart_write(&upload_id).await {
|
||||
log::warn!(
|
||||
"Failed to abort multipart write {}: {}",
|
||||
upload_id,
|
||||
abort_err
|
||||
);
|
||||
}
|
||||
if output.rescannable && self.is_retryable_write_error(&e) {
|
||||
retry_counter.increment_from_error(e)?;
|
||||
tokio::time::sleep(retry_counter.next_sleep_time()).await;
|
||||
continue;
|
||||
}
|
||||
return Err(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn execute_multipart_inserts(
|
||||
&self,
|
||||
upload_id: &str,
|
||||
output: &PreprocessingOutput,
|
||||
num_partitions: usize,
|
||||
) -> Result<()> {
|
||||
let plan = Arc::new(
|
||||
datafusion_physical_plan::repartition::RepartitionExec::try_new(
|
||||
output.plan.clone(),
|
||||
datafusion_physical_plan::Partitioning::RoundRobinBatch(num_partitions),
|
||||
)?,
|
||||
) as Arc<dyn ExecutionPlan>;
|
||||
|
||||
let insert = Arc::new(RemoteInsertExec::new_multipart(
|
||||
self.name.clone(),
|
||||
self.identifier.clone(),
|
||||
self.client.clone(),
|
||||
plan,
|
||||
output.overwrite,
|
||||
upload_id.to_string(),
|
||||
));
|
||||
|
||||
let task_ctx = Arc::new(datafusion_execution::TaskContext::default());
|
||||
let mut join_set = tokio::task::JoinSet::new();
|
||||
for partition in 0..num_partitions {
|
||||
let exec = insert.clone();
|
||||
let ctx = task_ctx.clone();
|
||||
join_set.spawn(async move {
|
||||
let mut stream = exec
|
||||
.execute(partition, ctx)
|
||||
.map_err(|e| -> Error { e.into() })?;
|
||||
while let Some(batch) = stream.next().await {
|
||||
batch.map_err(|e| -> Error { e.into() })?;
|
||||
}
|
||||
Ok::<_, Error>(())
|
||||
});
|
||||
}
|
||||
|
||||
// JoinSet aborts all remaining tasks when dropped, so if we return
|
||||
// early on error the orphaned tasks are automatically cancelled.
|
||||
while let Some(result) = join_set.join_next().await {
|
||||
result.map_err(|e| Error::Runtime {
|
||||
message: format!("Insert task panicked: {}", e),
|
||||
})??;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl<S: HttpSend> BaseTable for RemoteTable<S> {
|
||||
fn as_any(&self) -> &dyn std::any::Any {
|
||||
@@ -1239,44 +986,74 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
|
||||
status_code: None,
|
||||
})
|
||||
}
|
||||
async fn add(&self, mut add: AddDataBuilder) -> Result<AddResult> {
|
||||
async fn add(&self, add: AddDataBuilder) -> Result<AddResult> {
|
||||
use crate::remote::retry::RetryCounter;
|
||||
|
||||
self.check_mutable().await?;
|
||||
|
||||
let table_schema = self.schema().await?;
|
||||
let table_def = TableDefinition::try_from_rich_schema(table_schema.clone())?;
|
||||
|
||||
let num_partitions = if let Some(parallelism) = add.write_parallelism {
|
||||
if parallelism > 1 && self.server_version.support_multipart_write() {
|
||||
parallelism
|
||||
} else {
|
||||
1
|
||||
}
|
||||
} else if self.server_version.support_multipart_write() {
|
||||
// Peek at the first batch to estimate write partitions, same as NativeTable.
|
||||
let mut peeked = PeekedScannable::new(add.data);
|
||||
let n = if let Some(first_batch) = peeked.peek().await {
|
||||
let max_partitions = lance_core::utils::tokio::get_num_compute_intensive_cpus();
|
||||
estimate_write_partitions(
|
||||
first_batch.get_array_memory_size(),
|
||||
first_batch.num_rows(),
|
||||
peeked.num_rows(),
|
||||
max_partitions,
|
||||
)
|
||||
} else {
|
||||
1
|
||||
};
|
||||
add.data = Box::new(peeked);
|
||||
n
|
||||
} else {
|
||||
1
|
||||
};
|
||||
|
||||
let output = add.into_plan(&table_schema, &table_def)?;
|
||||
|
||||
if num_partitions > 1 {
|
||||
self.add_multipart(output, num_partitions).await
|
||||
} else {
|
||||
self.add_single_partition(output).await
|
||||
let mut insert: Arc<dyn ExecutionPlan> = Arc::new(RemoteInsertExec::new(
|
||||
self.name.clone(),
|
||||
self.identifier.clone(),
|
||||
self.client.clone(),
|
||||
output.plan,
|
||||
output.overwrite,
|
||||
));
|
||||
|
||||
let mut retry_counter =
|
||||
RetryCounter::new(&self.client.retry_config, uuid::Uuid::new_v4().to_string());
|
||||
|
||||
loop {
|
||||
let stream = execute_plan(insert.clone(), Default::default())?;
|
||||
let result: Result<Vec<_>> = stream.try_collect().await.map_err(Error::from);
|
||||
|
||||
match result {
|
||||
Ok(_) => {
|
||||
let add_result = insert
|
||||
.as_any()
|
||||
.downcast_ref::<RemoteInsertExec<S>>()
|
||||
.and_then(|i| i.add_result())
|
||||
.unwrap_or(AddResult { version: 0 });
|
||||
|
||||
if output.overwrite {
|
||||
self.invalidate_schema_cache();
|
||||
}
|
||||
|
||||
return Ok(add_result);
|
||||
}
|
||||
Err(err) if output.rescannable => {
|
||||
let retryable = match &err {
|
||||
Error::Http {
|
||||
source,
|
||||
status_code,
|
||||
..
|
||||
} => {
|
||||
// Don't retry read errors (is_body/is_decode): the
|
||||
// server may have committed the write already, and
|
||||
// without an idempotency key we'd duplicate data.
|
||||
source
|
||||
.downcast_ref::<reqwest::Error>()
|
||||
.is_some_and(|e| e.is_connect())
|
||||
|| status_code
|
||||
.is_some_and(|s| self.client.retry_config.statuses.contains(&s))
|
||||
}
|
||||
_ => false,
|
||||
};
|
||||
|
||||
if retryable {
|
||||
retry_counter.increment_from_error(err)?;
|
||||
tokio::time::sleep(retry_counter.next_sleep_time()).await;
|
||||
insert = insert.reset_state()?;
|
||||
continue;
|
||||
}
|
||||
|
||||
return Err(err);
|
||||
}
|
||||
Err(err) => return Err(err),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2034,7 +1811,6 @@ mod tests {
|
||||
|
||||
use super::*;
|
||||
|
||||
use crate::remote::client::{ClientConfig, RetryConfig};
|
||||
use crate::table::AddDataMode;
|
||||
|
||||
use arrow::{array::AsArray, compute::concat_batches, datatypes::Int32Type};
|
||||
@@ -5055,516 +4831,4 @@ mod tests {
|
||||
assert_eq!(data.len(), 1);
|
||||
assert_eq!(data[0].as_ref().unwrap(), &expected_data);
|
||||
}
|
||||
|
||||
fn schema_json() -> &'static str {
|
||||
r#"{"fields": [{"name": "id", "type": {"type": "int32"}, "nullable": true}]}"#
|
||||
}
|
||||
|
||||
fn simple_describe_response() -> http::Response<String> {
|
||||
http::Response::builder()
|
||||
.status(200)
|
||||
.body(format!(r#"{{"version": 1, "schema": {}}}"#, schema_json()))
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_multipart_write_happy_path() {
|
||||
use std::sync::Mutex;
|
||||
|
||||
let create_count = Arc::new(AtomicUsize::new(0));
|
||||
let insert_count = Arc::new(AtomicUsize::new(0));
|
||||
let complete_count = Arc::new(AtomicUsize::new(0));
|
||||
let abort_count = Arc::new(AtomicUsize::new(0));
|
||||
let upload_ids = Arc::new(Mutex::new(Vec::<String>::new()));
|
||||
|
||||
let create_count_c = create_count.clone();
|
||||
let insert_count_c = insert_count.clone();
|
||||
let complete_count_c = complete_count.clone();
|
||||
let abort_count_c = abort_count.clone();
|
||||
let upload_ids_c = upload_ids.clone();
|
||||
|
||||
let table = Table::new_with_handler_version(
|
||||
"my_table",
|
||||
semver::Version::new(0, 4, 0),
|
||||
move |request| {
|
||||
let path = request.url().path();
|
||||
let query = request.url().query().unwrap_or("");
|
||||
|
||||
if path == "/v1/table/my_table/describe/" {
|
||||
return simple_describe_response();
|
||||
}
|
||||
|
||||
if path == "/v1/table/my_table/multipart_write/create" {
|
||||
create_count_c.fetch_add(1, Ordering::SeqCst);
|
||||
return http::Response::builder()
|
||||
.status(200)
|
||||
.body(r#"{"upload_id": "test-upload-123"}"#.to_string())
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
if path == "/v1/table/my_table/insert/" {
|
||||
insert_count_c.fetch_add(1, Ordering::SeqCst);
|
||||
let uid = url::form_urlencoded::parse(query.as_bytes())
|
||||
.find(|(k, _)| k == "upload_id")
|
||||
.map(|(_, v)| v.to_string());
|
||||
upload_ids_c
|
||||
.lock()
|
||||
.unwrap()
|
||||
.push(uid.expect("missing upload_id on insert"));
|
||||
return http::Response::builder()
|
||||
.status(200)
|
||||
.body(r#"{"version": 1}"#.to_string())
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
if path == "/v1/table/my_table/multipart_write/complete" {
|
||||
complete_count_c.fetch_add(1, Ordering::SeqCst);
|
||||
let uid = url::form_urlencoded::parse(query.as_bytes())
|
||||
.find(|(k, _)| k == "upload_id")
|
||||
.map(|(_, v)| v.to_string());
|
||||
upload_ids_c
|
||||
.lock()
|
||||
.unwrap()
|
||||
.push(uid.expect("missing upload_id on complete"));
|
||||
return http::Response::builder()
|
||||
.status(200)
|
||||
.body(r#"{"version": 5}"#.to_string())
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
if path == "/v1/table/my_table/multipart_write/abort" {
|
||||
abort_count_c.fetch_add(1, Ordering::SeqCst);
|
||||
return http::Response::builder()
|
||||
.status(200)
|
||||
.body(String::new())
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
panic!("Unexpected request path: {}", path);
|
||||
},
|
||||
);
|
||||
|
||||
let batch = record_batch!(("id", Int32, [1, 2, 3])).unwrap();
|
||||
let result = table
|
||||
.add(vec![batch])
|
||||
.write_parallelism(2)
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result.version, 5);
|
||||
assert_eq!(create_count.load(Ordering::SeqCst), 1);
|
||||
assert!(
|
||||
insert_count.load(Ordering::SeqCst) > 1,
|
||||
"Expected multiple insert calls, got {}",
|
||||
insert_count.load(Ordering::SeqCst)
|
||||
);
|
||||
assert_eq!(complete_count.load(Ordering::SeqCst), 1);
|
||||
assert_eq!(abort_count.load(Ordering::SeqCst), 0);
|
||||
|
||||
let ids = upload_ids.lock().unwrap();
|
||||
assert!(
|
||||
ids.iter().all(|id| id == "test-upload-123"),
|
||||
"All requests should use the same upload_id, got: {:?}",
|
||||
*ids
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_multipart_write_fallback_old_server() {
|
||||
let insert_count = Arc::new(AtomicUsize::new(0));
|
||||
let create_count = Arc::new(AtomicUsize::new(0));
|
||||
|
||||
let insert_count_c = insert_count.clone();
|
||||
let create_count_c = create_count.clone();
|
||||
|
||||
// Server version 0.3.0 does not support multipart writes
|
||||
let table = Table::new_with_handler_version(
|
||||
"my_table",
|
||||
semver::Version::new(0, 3, 0),
|
||||
move |request| {
|
||||
let path = request.url().path();
|
||||
|
||||
if path == "/v1/table/my_table/describe/" {
|
||||
return simple_describe_response();
|
||||
}
|
||||
|
||||
if path.contains("multipart_write") {
|
||||
create_count_c.fetch_add(1, Ordering::SeqCst);
|
||||
panic!("Should not call multipart write endpoints on old server");
|
||||
}
|
||||
|
||||
if path == "/v1/table/my_table/insert/" {
|
||||
let query = request.url().query().unwrap_or("");
|
||||
assert!(
|
||||
!query.contains("upload_id"),
|
||||
"Should not have upload_id for old server"
|
||||
);
|
||||
insert_count_c.fetch_add(1, Ordering::SeqCst);
|
||||
return http::Response::builder()
|
||||
.status(200)
|
||||
.body(r#"{"version": 2}"#.to_string())
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
panic!("Unexpected request path: {}", path);
|
||||
},
|
||||
);
|
||||
|
||||
let batch = record_batch!(("id", Int32, [1, 2, 3])).unwrap();
|
||||
let result = table
|
||||
.add(vec![batch])
|
||||
.write_parallelism(2)
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result.version, 2);
|
||||
assert_eq!(create_count.load(Ordering::SeqCst), 0);
|
||||
assert_eq!(insert_count.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_multipart_write_small_data_single_partition() {
|
||||
let insert_count = Arc::new(AtomicUsize::new(0));
|
||||
let create_count = Arc::new(AtomicUsize::new(0));
|
||||
|
||||
let insert_count_c = insert_count.clone();
|
||||
let create_count_c = create_count.clone();
|
||||
|
||||
let table = Table::new_with_handler_version(
|
||||
"my_table",
|
||||
semver::Version::new(0, 4, 0),
|
||||
move |request| {
|
||||
let path = request.url().path();
|
||||
|
||||
if path == "/v1/table/my_table/describe/" {
|
||||
return simple_describe_response();
|
||||
}
|
||||
|
||||
if path.contains("multipart_write") {
|
||||
create_count_c.fetch_add(1, Ordering::SeqCst);
|
||||
panic!("Should not call multipart write endpoints for small data");
|
||||
}
|
||||
|
||||
if path == "/v1/table/my_table/insert/" {
|
||||
let query = request.url().query().unwrap_or("");
|
||||
assert!(
|
||||
!query.contains("upload_id"),
|
||||
"Should not have upload_id for small data"
|
||||
);
|
||||
insert_count_c.fetch_add(1, Ordering::SeqCst);
|
||||
return http::Response::builder()
|
||||
.status(200)
|
||||
.body(r#"{"version": 2}"#.to_string())
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
panic!("Unexpected request path: {}", path);
|
||||
},
|
||||
);
|
||||
|
||||
// Small data: only 3 rows
|
||||
let batch = record_batch!(("id", Int32, [1, 2, 3])).unwrap();
|
||||
let result = table.add(vec![batch]).execute().await.unwrap();
|
||||
|
||||
assert_eq!(result.version, 2);
|
||||
assert_eq!(create_count.load(Ordering::SeqCst), 0);
|
||||
assert_eq!(insert_count.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_multipart_write_abort_on_insert_failure() {
|
||||
let create_count = Arc::new(AtomicUsize::new(0));
|
||||
let insert_count = Arc::new(AtomicUsize::new(0));
|
||||
let complete_count = Arc::new(AtomicUsize::new(0));
|
||||
let abort_count = Arc::new(AtomicUsize::new(0));
|
||||
|
||||
let create_count_c = create_count.clone();
|
||||
let insert_count_c = insert_count.clone();
|
||||
let complete_count_c = complete_count.clone();
|
||||
let abort_count_c = abort_count.clone();
|
||||
|
||||
let table = Table::new_with_handler_version(
|
||||
"my_table",
|
||||
semver::Version::new(0, 4, 0),
|
||||
move |request| {
|
||||
let path = request.url().path();
|
||||
|
||||
if path == "/v1/table/my_table/describe/" {
|
||||
return simple_describe_response();
|
||||
}
|
||||
|
||||
if path == "/v1/table/my_table/multipart_write/create" {
|
||||
create_count_c.fetch_add(1, Ordering::SeqCst);
|
||||
return http::Response::builder()
|
||||
.status(200)
|
||||
.body(r#"{"upload_id": "test-upload-456"}"#.to_string())
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
if path == "/v1/table/my_table/insert/" {
|
||||
let count = insert_count_c.fetch_add(1, Ordering::SeqCst);
|
||||
// Fail on the first insert with non-retryable status
|
||||
if count == 0 {
|
||||
return http::Response::builder()
|
||||
.status(400)
|
||||
.body("Bad Request".to_string())
|
||||
.unwrap();
|
||||
}
|
||||
return http::Response::builder()
|
||||
.status(200)
|
||||
.body(r#"{"version": 1}"#.to_string())
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
if path == "/v1/table/my_table/multipart_write/complete" {
|
||||
complete_count_c.fetch_add(1, Ordering::SeqCst);
|
||||
return http::Response::builder()
|
||||
.status(200)
|
||||
.body(r#"{"version": 5}"#.to_string())
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
if path == "/v1/table/my_table/multipart_write/abort" {
|
||||
abort_count_c.fetch_add(1, Ordering::SeqCst);
|
||||
return http::Response::builder()
|
||||
.status(200)
|
||||
.body(String::new())
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
panic!("Unexpected request path: {}", path);
|
||||
},
|
||||
);
|
||||
|
||||
let batch = record_batch!(("id", Int32, [1, 2, 3])).unwrap();
|
||||
let result = table.add(vec![batch]).write_parallelism(2).execute().await;
|
||||
|
||||
assert!(result.is_err());
|
||||
assert_eq!(create_count.load(Ordering::SeqCst), 1);
|
||||
assert_eq!(complete_count.load(Ordering::SeqCst), 0);
|
||||
assert_eq!(abort_count.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_multipart_write_abort_on_complete_failure() {
|
||||
let abort_count = Arc::new(AtomicUsize::new(0));
|
||||
let abort_count_c = abort_count.clone();
|
||||
|
||||
let table = Table::new_with_handler_version(
|
||||
"my_table",
|
||||
semver::Version::new(0, 4, 0),
|
||||
move |request| {
|
||||
let path = request.url().path();
|
||||
|
||||
if path == "/v1/table/my_table/describe/" {
|
||||
return simple_describe_response();
|
||||
}
|
||||
|
||||
if path == "/v1/table/my_table/multipart_write/create" {
|
||||
return http::Response::builder()
|
||||
.status(200)
|
||||
.body(r#"{"upload_id": "test-upload-789"}"#.to_string())
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
if path == "/v1/table/my_table/insert/" {
|
||||
return http::Response::builder()
|
||||
.status(200)
|
||||
.body(r#"{"version": 1}"#.to_string())
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
if path == "/v1/table/my_table/multipart_write/complete" {
|
||||
return http::Response::builder()
|
||||
.status(400)
|
||||
.body("Bad Request".to_string())
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
if path == "/v1/table/my_table/multipart_write/abort" {
|
||||
abort_count_c.fetch_add(1, Ordering::SeqCst);
|
||||
return http::Response::builder()
|
||||
.status(200)
|
||||
.body(String::new())
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
panic!("Unexpected request path: {}", path);
|
||||
},
|
||||
);
|
||||
|
||||
let batch = record_batch!(("id", Int32, [1, 2, 3])).unwrap();
|
||||
let result = table.add(vec![batch]).write_parallelism(2).execute().await;
|
||||
|
||||
assert!(result.is_err());
|
||||
assert_eq!(abort_count.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
|
||||
fn retry_config_no_backoff() -> ClientConfig {
|
||||
ClientConfig {
|
||||
retry_config: RetryConfig {
|
||||
retries: Some(3),
|
||||
connect_retries: Some(3),
|
||||
read_retries: Some(3),
|
||||
backoff_factor: Some(0.0),
|
||||
backoff_jitter: Some(0.0),
|
||||
statuses: Some(vec![502, 503]),
|
||||
},
|
||||
..Default::default()
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_multipart_write_retry_on_partition_failure() {
|
||||
// All inserts for the first upload session return 503 (retryable).
|
||||
// After exhausting internal retries, the outer loop retries with a
|
||||
// new session and succeeds.
|
||||
let create_count = Arc::new(AtomicUsize::new(0));
|
||||
let complete_count = Arc::new(AtomicUsize::new(0));
|
||||
let abort_count = Arc::new(AtomicUsize::new(0));
|
||||
|
||||
let create_count_c = create_count.clone();
|
||||
let complete_count_c = complete_count.clone();
|
||||
let abort_count_c = abort_count.clone();
|
||||
|
||||
let table = Table::new_with_handler_version_and_config(
|
||||
"my_table",
|
||||
semver::Version::new(0, 4, 0),
|
||||
move |request| {
|
||||
let path = request.url().path();
|
||||
let query = request.url().query().unwrap_or("");
|
||||
|
||||
if path == "/v1/table/my_table/describe/" {
|
||||
return simple_describe_response();
|
||||
}
|
||||
|
||||
if path == "/v1/table/my_table/multipart_write/create" {
|
||||
let n = create_count_c.fetch_add(1, Ordering::SeqCst);
|
||||
let body = format!(r#"{{"upload_id": "upload-{}"}}"#, n + 1);
|
||||
return http::Response::builder().status(200).body(body).unwrap();
|
||||
}
|
||||
|
||||
if path == "/v1/table/my_table/insert/" {
|
||||
// Fail all inserts for the first session
|
||||
if query.contains("upload_id=upload-1") {
|
||||
return http::Response::builder()
|
||||
.status(503)
|
||||
.body("Service Unavailable".to_string())
|
||||
.unwrap();
|
||||
}
|
||||
return http::Response::builder()
|
||||
.status(200)
|
||||
.body(r#"{"version": 1}"#.to_string())
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
if path == "/v1/table/my_table/multipart_write/complete" {
|
||||
complete_count_c.fetch_add(1, Ordering::SeqCst);
|
||||
return http::Response::builder()
|
||||
.status(200)
|
||||
.body(r#"{"version": 7}"#.to_string())
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
if path == "/v1/table/my_table/multipart_write/abort" {
|
||||
abort_count_c.fetch_add(1, Ordering::SeqCst);
|
||||
return http::Response::builder()
|
||||
.status(200)
|
||||
.body(String::new())
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
panic!("Unexpected request path: {}", path);
|
||||
},
|
||||
retry_config_no_backoff(),
|
||||
);
|
||||
|
||||
let batch = record_batch!(("id", Int32, [1, 2, 3])).unwrap();
|
||||
let result = table
|
||||
.add(vec![batch])
|
||||
.write_parallelism(2)
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result.version, 7);
|
||||
assert_eq!(create_count.load(Ordering::SeqCst), 2);
|
||||
assert_eq!(abort_count.load(Ordering::SeqCst), 1);
|
||||
assert_eq!(complete_count.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_multipart_write_retry_on_complete_failure() {
|
||||
// Complete returns 503 for the first session, succeeds for the second.
|
||||
let create_count = Arc::new(AtomicUsize::new(0));
|
||||
let abort_count = Arc::new(AtomicUsize::new(0));
|
||||
|
||||
let create_count_c = create_count.clone();
|
||||
let abort_count_c = abort_count.clone();
|
||||
|
||||
let table = Table::new_with_handler_version_and_config(
|
||||
"my_table",
|
||||
semver::Version::new(0, 4, 0),
|
||||
move |request| {
|
||||
let path = request.url().path();
|
||||
let query = request.url().query().unwrap_or("");
|
||||
|
||||
if path == "/v1/table/my_table/describe/" {
|
||||
return simple_describe_response();
|
||||
}
|
||||
|
||||
if path == "/v1/table/my_table/multipart_write/create" {
|
||||
let n = create_count_c.fetch_add(1, Ordering::SeqCst);
|
||||
let body = format!(r#"{{"upload_id": "upload-{}"}}"#, n + 1);
|
||||
return http::Response::builder().status(200).body(body).unwrap();
|
||||
}
|
||||
|
||||
if path == "/v1/table/my_table/insert/" {
|
||||
return http::Response::builder()
|
||||
.status(200)
|
||||
.body(r#"{"version": 1}"#.to_string())
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
if path == "/v1/table/my_table/multipart_write/complete" {
|
||||
// Fail complete for first session
|
||||
if query.contains("upload_id=upload-1") {
|
||||
return http::Response::builder()
|
||||
.status(503)
|
||||
.body("Service Unavailable".to_string())
|
||||
.unwrap();
|
||||
}
|
||||
return http::Response::builder()
|
||||
.status(200)
|
||||
.body(r#"{"version": 9}"#.to_string())
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
if path == "/v1/table/my_table/multipart_write/abort" {
|
||||
abort_count_c.fetch_add(1, Ordering::SeqCst);
|
||||
return http::Response::builder()
|
||||
.status(200)
|
||||
.body(String::new())
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
panic!("Unexpected request path: {}", path);
|
||||
},
|
||||
retry_config_no_backoff(),
|
||||
);
|
||||
|
||||
let batch = record_batch!(("id", Int32, [1, 2, 3])).unwrap();
|
||||
let result = table
|
||||
.add(vec![batch])
|
||||
.write_parallelism(2)
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result.version, 9);
|
||||
assert_eq!(create_count.load(Ordering::SeqCst), 2);
|
||||
assert_eq!(abort_count.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -12,9 +12,7 @@ use datafusion_common::{DataFusionError, Result as DataFusionResult};
|
||||
use datafusion_execution::{SendableRecordBatchStream, TaskContext};
|
||||
use datafusion_physical_expr::EquivalenceProperties;
|
||||
use datafusion_physical_plan::stream::RecordBatchStreamAdapter;
|
||||
use datafusion_physical_plan::{
|
||||
DisplayAs, DisplayFormatType, ExecutionPlan, ExecutionPlanProperties, PlanProperties,
|
||||
};
|
||||
use datafusion_physical_plan::{DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties};
|
||||
use futures::StreamExt;
|
||||
use http::header::CONTENT_TYPE;
|
||||
|
||||
@@ -27,12 +25,10 @@ use crate::table::datafusion::insert::COUNT_SCHEMA;
|
||||
|
||||
/// ExecutionPlan for inserting data into a remote LanceDB table.
|
||||
///
|
||||
/// Streams data as Arrow IPC to `/v1/table/{id}/insert/` endpoint.
|
||||
///
|
||||
/// When `upload_id` is set, inserts are staged as part of a multipart write
|
||||
/// session and the plan supports multiple partitions for parallel uploads.
|
||||
/// Without `upload_id`, the plan requires a single partition and commits
|
||||
/// immediately.
|
||||
/// This plan:
|
||||
/// 1. Requires single partition (no parallel remote inserts yet)
|
||||
/// 2. Streams data as Arrow IPC to `/v1/table/{id}/insert/` endpoint
|
||||
/// 3. Stores AddResult for retrieval after execution
|
||||
#[derive(Debug)]
|
||||
pub struct RemoteInsertExec<S: HttpSend = Sender> {
|
||||
table_name: String,
|
||||
@@ -42,11 +38,10 @@ pub struct RemoteInsertExec<S: HttpSend = Sender> {
|
||||
overwrite: bool,
|
||||
properties: PlanProperties,
|
||||
add_result: Arc<Mutex<Option<AddResult>>>,
|
||||
upload_id: Option<String>,
|
||||
}
|
||||
|
||||
impl<S: HttpSend + 'static> RemoteInsertExec<S> {
|
||||
/// Create a new single-partition RemoteInsertExec.
|
||||
/// Create a new RemoteInsertExec.
|
||||
pub fn new(
|
||||
table_name: String,
|
||||
identifier: String,
|
||||
@@ -54,49 +49,10 @@ impl<S: HttpSend + 'static> RemoteInsertExec<S> {
|
||||
input: Arc<dyn ExecutionPlan>,
|
||||
overwrite: bool,
|
||||
) -> Self {
|
||||
Self::new_inner(table_name, identifier, client, input, overwrite, None)
|
||||
}
|
||||
|
||||
/// Create a multi-partition RemoteInsertExec for use with multipart writes.
|
||||
///
|
||||
/// Each partition's insert is staged under the given `upload_id` without
|
||||
/// committing. The caller is responsible for calling the complete (or abort)
|
||||
/// endpoint after all partitions finish.
|
||||
pub fn new_multipart(
|
||||
table_name: String,
|
||||
identifier: String,
|
||||
client: RestfulLanceDbClient<S>,
|
||||
input: Arc<dyn ExecutionPlan>,
|
||||
overwrite: bool,
|
||||
upload_id: String,
|
||||
) -> Self {
|
||||
Self::new_inner(
|
||||
table_name,
|
||||
identifier,
|
||||
client,
|
||||
input,
|
||||
overwrite,
|
||||
Some(upload_id),
|
||||
)
|
||||
}
|
||||
|
||||
fn new_inner(
|
||||
table_name: String,
|
||||
identifier: String,
|
||||
client: RestfulLanceDbClient<S>,
|
||||
input: Arc<dyn ExecutionPlan>,
|
||||
overwrite: bool,
|
||||
upload_id: Option<String>,
|
||||
) -> Self {
|
||||
let num_partitions = if upload_id.is_some() {
|
||||
input.output_partitioning().partition_count()
|
||||
} else {
|
||||
1
|
||||
};
|
||||
let schema = COUNT_SCHEMA.clone();
|
||||
let properties = PlanProperties::new(
|
||||
EquivalenceProperties::new(schema),
|
||||
datafusion_physical_plan::Partitioning::UnknownPartitioning(num_partitions),
|
||||
datafusion_physical_plan::Partitioning::UnknownPartitioning(1),
|
||||
datafusion_physical_plan::execution_plan::EmissionType::Final,
|
||||
datafusion_physical_plan::execution_plan::Boundedness::Bounded,
|
||||
);
|
||||
@@ -109,7 +65,6 @@ impl<S: HttpSend + 'static> RemoteInsertExec<S> {
|
||||
overwrite,
|
||||
properties,
|
||||
add_result: Arc::new(Mutex::new(None)),
|
||||
upload_id,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -219,11 +174,8 @@ impl<S: HttpSend + 'static> ExecutionPlan for RemoteInsertExec<S> {
|
||||
}
|
||||
|
||||
fn required_input_distribution(&self) -> Vec<datafusion_physical_plan::Distribution> {
|
||||
if self.upload_id.is_some() {
|
||||
vec![datafusion_physical_plan::Distribution::UnspecifiedDistribution]
|
||||
} else {
|
||||
vec![datafusion_physical_plan::Distribution::SinglePartition]
|
||||
}
|
||||
// Until we have a separate commit endpoint, we need to do all inserts in a single partition
|
||||
vec![datafusion_physical_plan::Distribution::SinglePartition]
|
||||
}
|
||||
|
||||
fn benefits_from_input_partitioning(&self) -> Vec<bool> {
|
||||
@@ -239,13 +191,12 @@ impl<S: HttpSend + 'static> ExecutionPlan for RemoteInsertExec<S> {
|
||||
"RemoteInsertExec requires exactly one child".to_string(),
|
||||
));
|
||||
}
|
||||
Ok(Arc::new(Self::new_inner(
|
||||
Ok(Arc::new(Self::new(
|
||||
self.table_name.clone(),
|
||||
self.identifier.clone(),
|
||||
self.client.clone(),
|
||||
children[0].clone(),
|
||||
self.overwrite,
|
||||
self.upload_id.clone(),
|
||||
)))
|
||||
}
|
||||
|
||||
@@ -254,20 +205,18 @@ impl<S: HttpSend + 'static> ExecutionPlan for RemoteInsertExec<S> {
|
||||
partition: usize,
|
||||
context: Arc<TaskContext>,
|
||||
) -> DataFusionResult<SendableRecordBatchStream> {
|
||||
if self.upload_id.is_none() && partition != 0 {
|
||||
if partition != 0 {
|
||||
return Err(DataFusionError::Internal(
|
||||
"RemoteInsertExec only supports single partition execution without upload_id"
|
||||
.to_string(),
|
||||
"RemoteInsertExec only supports single partition execution".to_string(),
|
||||
));
|
||||
}
|
||||
|
||||
let input_stream = self.input.execute(partition, context)?;
|
||||
let input_stream = self.input.execute(0, context)?;
|
||||
let client = self.client.clone();
|
||||
let identifier = self.identifier.clone();
|
||||
let overwrite = self.overwrite;
|
||||
let add_result = self.add_result.clone();
|
||||
let table_name = self.table_name.clone();
|
||||
let upload_id = self.upload_id.clone();
|
||||
|
||||
let stream = futures::stream::once(async move {
|
||||
let mut request = client
|
||||
@@ -277,9 +226,6 @@ impl<S: HttpSend + 'static> ExecutionPlan for RemoteInsertExec<S> {
|
||||
if overwrite {
|
||||
request = request.query(&[("mode", "overwrite")]);
|
||||
}
|
||||
if let Some(ref uid) = upload_id {
|
||||
request = request.query(&[("upload_id", uid.as_str())]);
|
||||
}
|
||||
|
||||
let (error_tx, mut error_rx) = tokio::sync::oneshot::channel();
|
||||
let body = Self::stream_as_http_body(input_stream, error_tx)?;
|
||||
@@ -316,30 +262,28 @@ impl<S: HttpSend + 'static> ExecutionPlan for RemoteInsertExec<S> {
|
||||
|
||||
let (request_id, response) = result?;
|
||||
|
||||
// For multipart writes, the staging response is not the final
|
||||
// version. Only parse AddResult for non-multipart inserts.
|
||||
if upload_id.is_none() {
|
||||
let body_text = response.text().await.map_err(|e| {
|
||||
let body_text = response.text().await.map_err(|e| {
|
||||
DataFusionError::External(Box::new(Error::Http {
|
||||
source: Box::new(e),
|
||||
request_id: request_id.clone(),
|
||||
status_code: None,
|
||||
}))
|
||||
})?;
|
||||
|
||||
let parsed_result = if body_text.trim().is_empty() {
|
||||
// Backward compatible with old servers
|
||||
AddResult { version: 0 }
|
||||
} else {
|
||||
serde_json::from_str(&body_text).map_err(|e| {
|
||||
DataFusionError::External(Box::new(Error::Http {
|
||||
source: Box::new(e),
|
||||
source: format!("Failed to parse add response: {}", e).into(),
|
||||
request_id: request_id.clone(),
|
||||
status_code: None,
|
||||
}))
|
||||
})?;
|
||||
|
||||
let parsed_result = if body_text.trim().is_empty() {
|
||||
// Backward compatible with old servers
|
||||
AddResult { version: 0 }
|
||||
} else {
|
||||
serde_json::from_str(&body_text).map_err(|e| {
|
||||
DataFusionError::External(Box::new(Error::Http {
|
||||
source: format!("Failed to parse add response: {}", e).into(),
|
||||
request_id: request_id.clone(),
|
||||
status_code: None,
|
||||
}))
|
||||
})?
|
||||
};
|
||||
})?
|
||||
};
|
||||
|
||||
{
|
||||
let mut res_lock = add_result.lock().map_err(|_| {
|
||||
DataFusionError::Execution("Failed to acquire lock for add_result".to_string())
|
||||
})?;
|
||||
|
||||
+13
-48
@@ -75,8 +75,6 @@ pub mod query;
|
||||
pub mod schema_evolution;
|
||||
pub mod update;
|
||||
use crate::index::waiter::wait_for_index;
|
||||
#[cfg(feature = "remote")]
|
||||
pub(crate) use add_data::PreprocessingOutput;
|
||||
pub use add_data::{AddDataBuilder, AddDataMode, AddResult, NaNVectorBehavior};
|
||||
pub use chrono::Duration;
|
||||
pub use delete::DeleteResult;
|
||||
@@ -442,34 +440,6 @@ mod test_utils {
|
||||
embedding_registry: Arc::new(MemoryRegistry::new()),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn new_with_handler_version_and_config<T>(
|
||||
name: impl Into<String>,
|
||||
version: semver::Version,
|
||||
handler: impl Fn(reqwest::Request) -> http::Response<T> + Clone + Send + Sync + 'static,
|
||||
config: crate::remote::ClientConfig,
|
||||
) -> Self
|
||||
where
|
||||
T: Into<reqwest::Body>,
|
||||
{
|
||||
let inner = Arc::new(
|
||||
crate::remote::table::RemoteTable::new_mock_with_version_and_config(
|
||||
name.into(),
|
||||
handler.clone(),
|
||||
Some(version),
|
||||
config.clone(),
|
||||
),
|
||||
);
|
||||
let database = Arc::new(crate::remote::db::RemoteDatabase::new_mock_with_config(
|
||||
handler, config,
|
||||
));
|
||||
Self {
|
||||
inner,
|
||||
database: Some(database),
|
||||
// Registry is unused.
|
||||
embedding_registry: Arc::new(MemoryRegistry::new()),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2228,26 +2198,21 @@ impl BaseTable for NativeTable {
|
||||
|
||||
let table_schema = Schema::from(&ds.schema().clone());
|
||||
|
||||
let num_partitions = if let Some(parallelism) = add.write_parallelism {
|
||||
parallelism
|
||||
// Peek at the first batch to estimate a good partition count for
|
||||
// write parallelism.
|
||||
let mut peeked = PeekedScannable::new(add.data);
|
||||
let num_partitions = if let Some(first_batch) = peeked.peek().await {
|
||||
let max_partitions = lance_core::utils::tokio::get_num_compute_intensive_cpus();
|
||||
estimate_write_partitions(
|
||||
first_batch.get_array_memory_size(),
|
||||
first_batch.num_rows(),
|
||||
peeked.num_rows(),
|
||||
max_partitions,
|
||||
)
|
||||
} else {
|
||||
// Peek at the first batch to estimate a good partition count for
|
||||
// write parallelism.
|
||||
let mut peeked = PeekedScannable::new(add.data);
|
||||
let n = if let Some(first_batch) = peeked.peek().await {
|
||||
let max_partitions = lance_core::utils::tokio::get_num_compute_intensive_cpus();
|
||||
estimate_write_partitions(
|
||||
first_batch.get_array_memory_size(),
|
||||
first_batch.num_rows(),
|
||||
peeked.num_rows(),
|
||||
max_partitions,
|
||||
)
|
||||
} else {
|
||||
1
|
||||
};
|
||||
add.data = Box::new(peeked);
|
||||
n
|
||||
1
|
||||
};
|
||||
add.data = Box::new(peeked);
|
||||
|
||||
let output = add.into_plan(&table_schema, &table_def)?;
|
||||
|
||||
|
||||
@@ -52,7 +52,6 @@ pub struct AddDataBuilder {
|
||||
pub(crate) write_options: WriteOptions,
|
||||
pub(crate) on_nan_vectors: NaNVectorBehavior,
|
||||
pub(crate) embedding_registry: Option<Arc<dyn EmbeddingRegistry>>,
|
||||
pub(crate) write_parallelism: Option<usize>,
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for AddDataBuilder {
|
||||
@@ -78,7 +77,6 @@ impl AddDataBuilder {
|
||||
write_options: WriteOptions::default(),
|
||||
on_nan_vectors: NaNVectorBehavior::default(),
|
||||
embedding_registry,
|
||||
write_parallelism: None,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -103,22 +101,7 @@ impl AddDataBuilder {
|
||||
self
|
||||
}
|
||||
|
||||
/// Set the number of parallel write streams.
|
||||
///
|
||||
/// By default, the number of streams is estimated from the data size.
|
||||
/// Setting this to `1` disables parallel writes.
|
||||
pub fn write_parallelism(mut self, parallelism: usize) -> Self {
|
||||
self.write_parallelism = Some(parallelism);
|
||||
self
|
||||
}
|
||||
|
||||
pub async fn execute(self) -> Result<AddResult> {
|
||||
if self.write_parallelism.map(|p| p == 0).unwrap_or(false) {
|
||||
return Err(Error::InvalidInput {
|
||||
message: "write_parallelism must be greater than 0".to_string(),
|
||||
});
|
||||
}
|
||||
|
||||
self.parent.clone().add(self).await
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user