diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml
index e4483b4a42b..5778d3856bb 100644
--- a/.github/workflows/rust.yml
+++ b/.github/workflows/rust.yml
@@ -24,7 +24,7 @@ on:
name: Rust CI
env:
- CARGO_UDEPS_VERSION: "0.1.61"
+ CARGO_SHEAR_VERSION: "1.13.4"
concurrency:
group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
@@ -84,20 +84,11 @@ jobs:
- uses: actions/checkout@v4
with:
persist-credentials: false
- - uses: arduino/setup-protoc@v3
- with:
- repo-token: ${{ secrets.GITHUB_TOKEN }}
- uses: actions-rust-lang/setup-rust-toolchain@v1
- - name: Rust Cache
- uses: Swatinem/rust-cache@v2
- with:
- shared-key: "check-udeps"
- cache-all-crates: "true"
- save-if: ${{ github.ref == 'refs/heads/main' }}
- - name: Install cargo-udeps
+ - name: Install cargo-shear
uses: taiki-e/install-action@v2
with:
- tool: cargo-udeps@${{ env.CARGO_UDEPS_VERSION }}
+ tool: cargo-shear@${{ env.CARGO_SHEAR_VERSION }}
fallback: none
- name: Check unused dependencies
run: make check-udeps
@@ -157,6 +148,48 @@ jobs:
env:
RUSTFLAGS: "-D warnings"
+ test-windows:
+ if: ${{ github.repository == 'GreptimeTeam/greptimedb' && (github.event_name != 'pull_request' || github.event.pull_request.draft == false) }}
+ name: Test Windows
+ runs-on: windows-2022
+ timeout-minutes: 90
+ env:
+ CARGO_INCREMENTAL: 0
+ CARGO_PROFILE_DEV_DEBUG: 0
+ CARGO_PROFILE_TEST_DEBUG: 0
+ RUSTFLAGS: "-D warnings -C linker=lld-link"
+ RUST_BACKTRACE: 1
+ RUST_MIN_STACK: 8388608 # 8MB
+ steps:
+ - run: git config --global core.autocrlf false
+ - uses: actions/checkout@v4
+ with:
+ persist-credentials: false
+ - uses: ./.github/actions/setup-cyborg
+ - uses: arduino/setup-protoc@v3
+ with:
+ repo-token: ${{ secrets.GITHUB_TOKEN }}
+ - uses: KyleMayes/install-llvm-action@v1
+ with:
+ version: "14.0"
+ - uses: actions-rust-lang/setup-rust-toolchain@v1
+ with:
+ cache: false
+ - name: Rust Cache
+ uses: Swatinem/rust-cache@v2
+ with:
+ shared-key: "test-windows"
+ cache-all-crates: "true"
+ save-if: ${{ github.ref == 'refs/heads/main' }}
+ - name: Install Cargo Nextest
+ uses: taiki-e/install-action@nextest
+ - name: Install WSL distribution
+ uses: Vampire/setup-wsl@v2
+ with:
+ distribution: Ubuntu-22.04
+ - name: Run Windows tests
+ run: cargo nextest run --locked --workspace -F dashboard --no-fail-fast
+
conflict-check:
if: ${{ github.repository == 'GreptimeTeam/greptimedb' && (github.event_name != 'pull_request' || github.event.pull_request.draft == false) }}
name: Check for conflict
diff --git a/.gitignore b/.gitignore
index 90893f8aac1..dd60d96224e 100644
--- a/.gitignore
+++ b/.gitignore
@@ -52,8 +52,6 @@ venv/
tests-fuzz/artifacts/
tests-fuzz/corpus/
-# cargo-udeps reports
-udeps-report.json
# Nix
.direnv
diff --git a/Cargo.lock b/Cargo.lock
index ea55f1b297d..6c7916d8194 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1673,7 +1673,6 @@ dependencies = [
"async-trait",
"bytes",
"cache",
- "catalog",
"chrono",
"common-base",
"common-catalog",
@@ -1701,7 +1700,6 @@ dependencies = [
"lazy_static",
"meta-client",
"moka",
- "object-store",
"partition",
"paste",
"prometheus 0.14.0",
@@ -2006,10 +2004,8 @@ version = "1.3.0-alpha.1"
dependencies = [
"async-stream",
"async-trait",
- "auth",
"base64 0.22.1",
"cache",
- "catalog",
"chrono",
"clap",
"client",
@@ -2022,13 +2018,10 @@ dependencies = [
"common-macro",
"common-meta",
"common-procedure",
- "common-query",
- "common-recordbatch",
"common-runtime",
"common-telemetry",
"common-test-util",
"common-time",
- "common-version",
"common-wal",
"datatypes",
"etcd-client",
@@ -2048,7 +2041,6 @@ dependencies = [
"serde",
"serde_json",
"servers",
- "session",
"snafu 0.8.6",
"sql",
"standalone",
@@ -2056,7 +2048,6 @@ dependencies = [
"table",
"tempfile",
"tokio",
- "tracing-appender",
"url",
"uuid",
]
@@ -2074,7 +2065,6 @@ dependencies = [
"common-catalog",
"common-error",
"common-grpc",
- "common-grpc-expr",
"common-macro",
"common-meta",
"common-query",
@@ -2100,7 +2090,6 @@ dependencies = [
"tokio",
"tokio-stream",
"tonic 0.14.2",
- "tracing",
]
[[package]]
@@ -2150,10 +2139,8 @@ dependencies = [
"api",
"async-trait",
"auth",
- "base64 0.22.1",
"cache",
"catalog",
- "chrono",
"clap",
"cli",
"client",
@@ -2170,7 +2157,6 @@ dependencies = [
"common-options",
"common-procedure",
"common-query",
- "common-recordbatch",
"common-runtime",
"common-stat",
"common-telemetry",
@@ -2184,7 +2170,6 @@ dependencies = [
"datafusion-physical-plan",
"datanode",
"datatypes",
- "either",
"etcd-client",
"file-engine",
"flow",
@@ -2192,7 +2177,6 @@ dependencies = [
"futures",
"hex",
"human-panic",
- "humantime",
"lazy_static",
"log-store",
"meta-client",
@@ -2216,20 +2200,17 @@ dependencies = [
"serde",
"serde_json",
"servers",
- "session",
"similar-asserts",
"smallvec",
"snafu 0.8.6",
"sqlparser",
"standalone",
"store-api",
- "table",
"temp-env",
"tempfile",
"tikv-jemallocator",
"tokio",
"toml 0.8.23",
- "tonic 0.14.2",
"tracing-appender",
]
@@ -2355,7 +2336,6 @@ dependencies = [
"serde_json",
"snafu 0.8.6",
"temp-env",
- "tempfile",
"toml 0.8.23",
]
@@ -2456,14 +2436,11 @@ dependencies = [
"common-grpc",
"common-macro",
"common-meta",
- "common-time",
"greptime-proto",
- "humantime",
"meta-client",
"serde",
"serde_json",
"snafu 0.8.6",
- "tokio",
"tonic 0.14.2",
]
@@ -2474,22 +2451,18 @@ dependencies = [
"ahash 0.8.12",
"api",
"approx 0.5.1",
- "arc-swap",
"arrow 59.2.0",
- "arrow-cast 59.2.0",
"arrow-schema 59.2.0",
"async-trait",
"axum 0.8.4",
"bincode",
"catalog",
- "chrono",
"common-base",
"common-catalog",
"common-error",
"common-macro",
"common-meta",
"common-query",
- "common-runtime",
"common-telemetry",
"common-time",
"common-version",
@@ -2514,7 +2487,6 @@ dependencies = [
"memchr",
"mito-codec",
"nalgebra",
- "num",
"num-traits",
"paste",
"pretty_assertions",
@@ -2591,7 +2563,6 @@ name = "common-grpc-expr"
version = "1.3.0-alpha.1"
dependencies = [
"api",
- "common-base",
"common-catalog",
"common-error",
"common-macro",
@@ -2599,7 +2570,6 @@ dependencies = [
"common-sql",
"common-time",
"datatypes",
- "paste",
"prost 0.14.1",
"snafu 0.8.6",
"store-api",
@@ -2675,7 +2645,6 @@ dependencies = [
"common-wal",
"common-workload",
"datafusion-common",
- "datafusion-expr",
"datatypes",
"deadpool",
"deadpool-postgres",
@@ -2684,7 +2653,6 @@ dependencies = [
"flexbuffers",
"futures",
"futures-util",
- "greptime-proto",
"hex",
"humantime-serde",
"itertools 0.14.0",
@@ -2779,7 +2747,6 @@ version = "1.3.0-alpha.1"
dependencies = [
"async-trait",
"common-procedure",
- "snafu 0.8.6",
"tokio",
]
@@ -2852,13 +2819,11 @@ dependencies = [
"lazy_static",
"num_cpus",
"once_cell",
- "parking_lot 0.12.4",
"paste",
"pin-project",
"prometheus 0.14.0",
"ratelimit",
"serde",
- "serde_json",
"snafu 0.8.6",
"tempfile",
"tokio",
@@ -2885,7 +2850,6 @@ dependencies = [
"common-decimal",
"common-error",
"common-macro",
- "common-telemetry",
"common-time",
"datafusion-sql",
"datatypes",
@@ -2903,7 +2867,6 @@ dependencies = [
"common-base",
"common-runtime",
"common-telemetry",
- "lazy_static",
"nix 0.30.1",
"num_cpus",
"prometheus 0.14.0",
@@ -4533,7 +4496,6 @@ dependencies = [
"async-trait",
"bytes",
"cache",
- "client",
"common-base",
"common-catalog",
"common-config",
@@ -4541,7 +4503,6 @@ dependencies = [
"common-error",
"common-function",
"common-greptimedb-telemetry",
- "common-grpc",
"common-macro",
"common-meta",
"common-options",
@@ -5332,7 +5293,6 @@ dependencies = [
"common-datasource",
"common-error",
"common-macro",
- "common-query",
"common-recordbatch",
"common-telemetry",
"common-test-util",
@@ -5472,19 +5432,14 @@ dependencies = [
"arrow 59.2.0",
"arrow-flight",
"arrow-schema 59.2.0",
- "async-recursion",
"async-trait",
- "bytes",
"catalog",
"chrono",
"client",
"common-base",
"common-catalog",
"common-config",
- "common-datasource",
- "common-decimal",
"common-error",
- "common-frontend",
"common-function",
"common-grpc",
"common-macro",
@@ -5497,42 +5452,31 @@ dependencies = [
"common-telemetry",
"common-time",
"common-version",
- "config",
"datafusion",
"datafusion-common",
"datafusion-expr",
"datafusion-physical-expr",
- "datafusion-substrait",
"datatypes",
- "enum_dispatch",
"futures",
"get-size2",
"greptime-proto",
"hostname 0.4.1",
- "http 1.5.0",
"humantime-serde",
"itertools 0.14.0",
"lazy_static",
"meta-client",
- "nom 7.1.3",
- "num-traits",
- "num_cpus",
"operator",
- "partition",
"pretty_assertions",
"prometheus 0.14.0",
- "prost 0.14.1",
"query",
"rand 0.9.4",
"serde",
"serde_json",
"servers",
"session",
- "smallvec",
"snafu 0.8.6",
"sql",
"store-api",
- "strum 0.27.1",
"substrait 1.3.0-alpha.1",
"table",
"tokio",
@@ -5633,7 +5577,6 @@ name = "frontend"
version = "1.3.0-alpha.1"
dependencies = [
"api",
- "arc-swap",
"arrow-flight",
"async-stream",
"async-trait",
@@ -5657,7 +5600,6 @@ dependencies = [
"common-memory-manager",
"common-meta",
"common-options",
- "common-procedure",
"common-query",
"common-recordbatch",
"common-runtime",
@@ -5672,13 +5614,11 @@ dependencies = [
"datatypes",
"futures",
"hostname 0.4.1",
- "humantime",
"humantime-serde",
"hyper-util",
"lazy_static",
"log-query",
"meta-client",
- "num_cpus",
"opentelemetry-proto",
"opentelemetry-semantic-conventions 0.31.0",
"operator",
@@ -5689,7 +5629,6 @@ dependencies = [
"promql-parser",
"prost 0.14.1",
"query",
- "rand 0.9.4",
"reqwest 0.13.4",
"serde",
"serde_json",
@@ -5708,7 +5647,6 @@ dependencies = [
"tonic 0.14.2",
"tower 0.5.2",
"tracing",
- "uuid",
]
[[package]]
@@ -8121,7 +8059,6 @@ dependencies = [
"derive_builder 0.20.2",
"futures",
"futures-util",
- "itertools 0.14.0",
"lazy_static",
"object-store",
"pin-project",
@@ -8465,20 +8402,16 @@ dependencies = [
"common-options",
"common-telemetry",
"common-time",
- "datatypes",
"futures",
- "futures-util",
"humantime-serde",
"meta-srv",
"rand 0.9.4",
"serde",
"serde_json",
- "session",
"snafu 0.8.6",
"tokio",
"tokio-stream",
"tonic 0.14.2",
- "tower 0.5.2",
"tracing",
"tracing-subscriber",
]
@@ -8490,9 +8423,7 @@ dependencies = [
"api",
"async-trait",
"axum 0.8.4",
- "axum-extra",
"axum-macros",
- "bytes",
"chrono",
"clap",
"client",
@@ -8525,19 +8456,14 @@ dependencies = [
"futures",
"h2 0.3.26",
"hostname 0.4.1",
- "http-body-util",
- "humantime",
"humantime-serde",
"hyper-util",
"itertools 0.14.0",
"lazy_static",
- "local-ip-address",
"once_cell",
"ordered-float 4.6.0",
- "parking_lot 0.12.4",
"partition",
"prometheus 0.14.0",
- "prost 0.14.1",
"rand 0.9.4",
"regex",
"rskafka",
@@ -8555,7 +8481,6 @@ dependencies = [
"toml 0.8.23",
"tonic 0.14.2",
"tower 0.5.2",
- "tower-http 0.6.11",
"tracing",
"tracing-subscriber",
"typetag",
@@ -8603,7 +8528,6 @@ dependencies = [
"common-recordbatch",
"common-runtime",
"common-telemetry",
- "common-test-util",
"common-time",
"common-wal",
"criterion 0.7.0",
@@ -8723,9 +8647,7 @@ dependencies = [
"bytes",
"chrono",
"common-base",
- "common-config",
"common-datasource",
- "common-decimal",
"common-error",
"common-function",
"common-grpc",
@@ -8764,7 +8686,6 @@ dependencies = [
"object-store",
"parquet",
"partition",
- "paste",
"pin-project",
"prometheus 0.14.0",
"prost 0.14.1",
@@ -8776,11 +8697,9 @@ dependencies = [
"rskafka",
"rstest",
"rstest_reuse",
- "scopeguard",
"serde",
"serde_json",
"serde_with",
- "session",
"smallvec",
"snafu 0.8.6",
"store-api",
@@ -9562,10 +9481,8 @@ name = "object-store"
version = "1.3.0-alpha.1"
dependencies = [
"anyhow",
- "async-trait",
"bytes",
"cap-std",
- "chrono",
"common-base",
"common-error",
"common-macro",
@@ -9575,13 +9492,12 @@ dependencies = [
"derive_builder 0.20.2",
"futures",
"humantime-serde",
- "lazy_static",
+ "object-store",
"object_store",
"object_store_opendal",
"opendal",
"opendal-http-transport-reqwest",
"prometheus 0.14.0",
- "rand 0.9.4",
"reqwest 0.13.4",
"serde",
"snafu 0.8.6",
@@ -10096,7 +10012,6 @@ dependencies = [
"ahash 0.8.12",
"api",
"arrow 59.2.0",
- "arrow-ipc 59.2.0",
"async-stream",
"async-trait",
"axum 0.8.4",
@@ -10143,7 +10058,6 @@ dependencies = [
"path-slash",
"percent-encoding",
"prometheus 0.14.0",
- "prost 0.14.1",
"prost-types 0.14.1",
"query",
"regex",
@@ -10492,19 +10406,15 @@ name = "partition"
version = "1.3.0-alpha.1"
dependencies = [
"api",
- "async-trait",
"common-base",
"common-error",
"common-macro",
"common-meta",
- "common-query",
- "common-telemetry",
"criterion 0.7.0",
"datafusion-common",
"datafusion-expr",
"datafusion-physical-expr",
"datatypes",
- "itertools 0.14.0",
"moka",
"rand 0.8.5",
"serde",
@@ -10893,29 +10803,22 @@ dependencies = [
"api",
"arrow 59.2.0",
"arrow-schema 59.2.0",
- "async-trait",
"catalog",
"chrono",
"chrono-tz",
"common-catalog",
"common-error",
- "common-function",
"common-macro",
"common-meta",
"common-query",
"common-recordbatch",
- "common-runtime",
"common-telemetry",
"common-time",
"criterion 0.7.0",
- "crossbeam-utils",
"csv",
- "dashmap",
"datafusion",
"datafusion-common",
"datafusion-expr",
- "datafusion-functions",
- "datafusion-physical-expr",
"datatypes",
"dyn-fmt",
"enum_dispatch",
@@ -10930,10 +10833,8 @@ dependencies = [
"once_cell",
"operator",
"ordered-float 4.6.0",
- "paste",
"prometheus 0.14.0",
"query",
- "rayon",
"regex",
"serde",
"serde_json",
@@ -11867,7 +11768,6 @@ version = "1.3.0-alpha.1"
dependencies = [
"ahash 0.8.12",
"api",
- "arc-swap",
"arrow 59.2.0",
"arrow-schema 59.2.0",
"async-recursion",
@@ -11901,12 +11801,10 @@ dependencies = [
"datafusion-proto",
"datafusion-sql",
"datatypes",
- "either",
"fastrand",
"futures",
"futures-util",
"greptime-proto",
- "humantime",
"itertools 0.14.0",
"jsonb",
"lazy_static",
@@ -11915,11 +11813,8 @@ dependencies = [
"meter-macros",
"moka",
"nalgebra",
- "num",
- "num-traits",
"object-store",
"once_cell",
- "parking_lot 0.12.4",
"partition",
"paste",
"pretty_assertions",
@@ -11939,7 +11834,6 @@ dependencies = [
"substrait 1.3.0-alpha.1",
"table",
"tokio",
- "tokio-stream",
"toml 0.8.23",
"tracing",
"uuid",
@@ -13582,7 +13476,6 @@ dependencies = [
"common-catalog",
"common-decimal",
"common-error",
- "common-frontend",
"common-grpc",
"common-macro",
"common-mem-prof",
@@ -13602,7 +13495,6 @@ dependencies = [
"common-version",
"criterion 0.5.1",
"csv",
- "dashmap",
"datafusion",
"datafusion-common",
"datafusion-expr",
@@ -14070,25 +13962,19 @@ dependencies = [
"common-base",
"common-catalog",
"common-datasource",
- "common-decimal",
"common-error",
"common-macro",
"common-query",
"common-sql",
- "common-telemetry",
"common-time",
"datafusion",
"datafusion-common",
"datafusion-expr",
- "datafusion-physical-expr",
"datafusion-sql",
"datatypes",
"either",
- "hex",
- "humantime",
"iso8601",
"itertools 0.14.0",
- "jsonb",
"lazy_static",
"pretty_assertions",
"promql-parser",
@@ -14098,7 +13984,6 @@ dependencies = [
"snafu 0.8.6",
"sqlparser",
"sqlparser_derive 0.1.1",
- "store-api",
"table",
"uuid",
]
@@ -14130,7 +14015,6 @@ dependencies = [
"client",
"common-error",
"common-query",
- "common-recordbatch",
"common-time",
"datatypes",
"flate2",
@@ -14458,7 +14342,6 @@ dependencies = [
"api",
"aquamarine",
"arrow-schema 59.2.0",
- "async-stream",
"async-trait",
"bytes",
"common-base",
@@ -14485,7 +14368,6 @@ dependencies = [
"serde",
"serde_json",
"snafu 0.8.6",
- "sqlparser",
"strum 0.27.1",
"tokio",
"uuid",
@@ -14654,16 +14536,13 @@ dependencies = [
"common-error",
"common-function",
"common-macro",
- "common-telemetry",
"datafusion",
"datafusion-common",
"datafusion-expr",
"datafusion-substrait",
- "datatypes",
"promql",
"prost 0.14.1",
"snafu 0.8.6",
- "tokio",
]
[[package]]
@@ -14812,7 +14691,6 @@ dependencies = [
"futures",
"greptime-proto",
"humantime",
- "humantime-serde",
"lazy_static",
"once_cell",
"parquet",
@@ -14820,10 +14698,8 @@ dependencies = [
"serde",
"serde_json",
"snafu 0.8.6",
- "sqlparser",
"store-api",
"tokio",
- "tokio-util",
]
[[package]]
@@ -15121,7 +14997,6 @@ version = "1.3.0-alpha.1"
dependencies = [
"api",
"arrow-flight",
- "async-stream",
"async-trait",
"auth",
"axum 0.8.4",
@@ -15149,7 +15024,6 @@ dependencies = [
"common-query",
"common-recordbatch",
"common-runtime",
- "common-stat",
"common-telemetry",
"common-test-util",
"common-time",
@@ -15180,7 +15054,6 @@ dependencies = [
"opentelemetry-proto",
"operator",
"otel-arrow-rust",
- "partition",
"paste",
"pipeline",
"plugins",
@@ -15195,7 +15068,6 @@ dependencies = [
"servers",
"session",
"similar-asserts",
- "snafu 0.8.6",
"sql",
"sqlx",
"standalone",
diff --git a/Cargo.toml b/Cargo.toml
index 7c5c53d91c8..5cdc3cf63a2 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -104,7 +104,6 @@ aquamarine = "0.6"
arrow = { version = "=59.2.0", features = ["prettyprint"] }
arrow-array = { version = "=59.2.0", default-features = false, features = ["chrono-tz"] }
arrow-buffer = "=59.2.0"
-arrow-cast = "=59.2.0"
arrow-flight = "=59.2.0"
arrow-ipc = { version = "=59.2.0", default-features = false, features = ["lz4", "zstd"] }
arrow-schema = { version = "=59.2.0", features = ["serde"] }
@@ -236,7 +235,6 @@ rskafka = { git = "https://github.com/influxdata/rskafka.git", rev = "80f42eb15e
rstest = "0.25"
rstest_reuse = "0.7"
rust_decimal = "1.33"
-rustc-hash = "2.0"
rustls = { version = "0.23.25", default-features = false }
ryu = "1.0"
sea-query = "0.32"
diff --git a/Makefile b/Makefile
index 7176eafd0bb..1379d86683c 100644
--- a/Makefile
+++ b/Makefile
@@ -233,14 +233,11 @@ fix-clippy: ## Fix clippy violations.
.PHONY: check-udeps
check-udeps: ## Check unused dependencies.
- cargo udeps --workspace --all-targets
+ cargo shear
.PHONY: fix-udeps
fix-udeps: ## Remove unused dependencies automatically.
- @echo "Running cargo-udeps to find unused dependencies..."
- @cargo udeps --workspace --all-targets --output json > udeps-report.json || true
- @echo "Removing unused dependencies..."
- @python3 scripts/fix-udeps.py udeps-report.json
+ cargo shear --fix
.PHONY: fmt-check
fmt-check: ## Check code format.
diff --git a/config/config.md b/config/config.md
index d736c24d848..c33918a8573 100644
--- a/config/config.md
+++ b/config/config.md
@@ -75,7 +75,7 @@
| `influxdb` | -- | -- | InfluxDB protocol options. |
| `influxdb.enable` | Bool | `true` | Whether to enable InfluxDB protocol in HTTP API. |
| `influxdb.default_merge_mode` | String | `last_non_null` | Default merge mode for tables automatically created by InfluxDB protocol.
Available values: "last_non_null", "last_row". |
-| `pending_rows_batcher` | -- | -- | Ordinary-table batching for opted-in HTTP ingestion protocols.
PENDING_ROWS_BATCH_SYNC defaults to true for both batchers. Set it to false to acknowledge
queue admission without waiting for storage; later failures cannot be returned to the client.
Omitted or empty protocols disables batching. Prom without metric engine uses this batcher.
OTLP logs, traces and ordinary metrics use this batcher. |
+| `pending_rows_batcher` | -- | -- | Ordinary-table batching for opted-in ingestion protocols.
PENDING_ROWS_BATCH_SYNC defaults to true for both batchers. Set it to false to acknowledge
queue admission without waiting for storage; later failures cannot be returned to the client.
Omitted or empty protocols disables batching. Prom without metric engine uses this batcher.
OTLP logs, traces and ordinary metrics use this batcher.
MySQL and PostgreSQL use the same acknowledgement policy. Protocol timeouts are unchanged.
Single-connection writes and INSERT SELECT may incur additional flush waits. |
| `pending_rows_batcher.pending_rows_flush_interval` | String | `0s` | Flush interval measured from the first pending submission. Zero disables batching. |
| `pending_rows_batcher.max_batch_rows` | Integer | `100000` | Flush after a complete submission reaches this row threshold. |
| `pending_rows_batcher.max_concurrent_flushes` | Integer | `256` | Maximum concurrent flushes shared by the frontend batcher. |
@@ -343,7 +343,7 @@
| `influxdb` | -- | -- | InfluxDB protocol options. |
| `influxdb.enable` | Bool | `true` | Whether to enable InfluxDB protocol in HTTP API. |
| `influxdb.default_merge_mode` | String | `last_non_null` | Default merge mode for tables automatically created by InfluxDB protocol.
Available values: "last_non_null", "last_row". |
-| `pending_rows_batcher` | -- | -- | Ordinary-table batching for opted-in HTTP ingestion protocols.
PENDING_ROWS_BATCH_SYNC defaults to true for both batchers. Set it to false to acknowledge
queue admission without waiting for storage; later failures cannot be returned to the client.
Omitted or empty protocols disables batching. Prom without metric engine uses this batcher.
OTLP logs, traces and ordinary metrics use this batcher. |
+| `pending_rows_batcher` | -- | -- | Ordinary-table batching for opted-in ingestion protocols.
PENDING_ROWS_BATCH_SYNC defaults to true for both batchers. Set it to false to acknowledge
queue admission without waiting for storage; later failures cannot be returned to the client.
Omitted or empty protocols disables batching. Prom without metric engine uses this batcher.
OTLP logs, traces and ordinary metrics use this batcher.
MySQL and PostgreSQL use the same acknowledgement policy. Protocol timeouts are unchanged.
Single-connection writes and INSERT SELECT may incur additional flush waits. |
| `pending_rows_batcher.pending_rows_flush_interval` | String | `0s` | Flush interval measured from the first pending submission. Zero disables batching. |
| `pending_rows_batcher.max_batch_rows` | Integer | `100000` | Flush after a complete submission reaches this row threshold. |
| `pending_rows_batcher.max_concurrent_flushes` | Integer | `256` | Maximum concurrent flushes shared by the frontend batcher. |
diff --git a/config/frontend.example.toml b/config/frontend.example.toml
index 275db3105c5..055acbb93bd 100644
--- a/config/frontend.example.toml
+++ b/config/frontend.example.toml
@@ -231,11 +231,13 @@ enable = true
## Available values: "last_non_null", "last_row".
default_merge_mode = "last_non_null"
-## Ordinary-table batching for opted-in HTTP ingestion protocols.
+## Ordinary-table batching for opted-in ingestion protocols.
## PENDING_ROWS_BATCH_SYNC defaults to true for both batchers. Set it to false to acknowledge
## queue admission without waiting for storage; later failures cannot be returned to the client.
## Omitted or empty protocols disables batching. Prom without metric engine uses this batcher.
## OTLP logs, traces and ordinary metrics use this batcher.
+## MySQL and PostgreSQL use the same acknowledgement policy. Protocol timeouts are unchanged.
+## Single-connection writes and INSERT SELECT may incur additional flush waits.
[pending_rows_batcher]
# protocols = [
# "influxdb",
@@ -246,6 +248,8 @@ default_merge_mode = "last_non_null"
# "splunk",
# "elasticsearch",
# "http_sql",
+# "mysql",
+# "postgres",
# "prom",
# ]
## Flush interval measured from the first pending submission. Zero disables batching.
diff --git a/config/standalone.example.toml b/config/standalone.example.toml
index 257f48eda2c..357ec5904f7 100644
--- a/config/standalone.example.toml
+++ b/config/standalone.example.toml
@@ -210,11 +210,13 @@ enable = true
## Available values: "last_non_null", "last_row".
default_merge_mode = "last_non_null"
-## Ordinary-table batching for opted-in HTTP ingestion protocols.
+## Ordinary-table batching for opted-in ingestion protocols.
## PENDING_ROWS_BATCH_SYNC defaults to true for both batchers. Set it to false to acknowledge
## queue admission without waiting for storage; later failures cannot be returned to the client.
## Omitted or empty protocols disables batching. Prom without metric engine uses this batcher.
## OTLP logs, traces and ordinary metrics use this batcher.
+## MySQL and PostgreSQL use the same acknowledgement policy. Protocol timeouts are unchanged.
+## Single-connection writes and INSERT SELECT may incur additional flush waits.
[pending_rows_batcher]
# protocols = [
# "influxdb",
@@ -225,6 +227,8 @@ default_merge_mode = "last_non_null"
# "splunk",
# "elasticsearch",
# "http_sql",
+# "mysql",
+# "postgres",
# "prom",
# ]
## Flush interval measured from the first pending submission. Zero disables batching.
diff --git a/flake.nix b/flake.nix
index 5b099427988..fbba48ed373 100644
--- a/flake.nix
+++ b/flake.nix
@@ -46,7 +46,7 @@
])
cargo-nextest
cargo-llvm-cov
- cargo-udeps
+ cargo-shear
taplo
curl
gnuplot ## for cargo bench
diff --git a/scripts/fix-udeps.py b/scripts/fix-udeps.py
deleted file mode 100755
index 7547b897361..00000000000
--- a/scripts/fix-udeps.py
+++ /dev/null
@@ -1,265 +0,0 @@
-# Copyright 2023 Greptime Team
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-import json
-import os
-import re
-import sys
-
-
-def load_udeps_report(report_path):
- try:
- with open(report_path, "r") as f:
- return json.load(f)
- except FileNotFoundError:
- print(f"Error: Report file '{report_path}' not found.")
- return None
- except json.JSONDecodeError as e:
- print(f"Error: Invalid JSON in report file: {e}")
- return None
-
-
-def extract_unused_dependencies(report):
- """
- Extract and organize unused dependencies from the cargo-udeps JSON report.
-
- The cargo-udeps report has this structure:
- {
- "unused_deps": {
- "package_name v0.1.0 (/path/to/package)": {
- "normal": ["dep1", "dep2"],
- "development": ["dev_dep1"],
- "build": ["build_dep1"],
- "manifest_path": "/path/to/Cargo.toml"
- }
- }
- }
-
- Args:
- report (dict): The parsed JSON report from cargo-udeps
-
- Returns:
- dict: Organized unused dependencies by package name:
- {
- "package_name": {
- "dependencies": [("dep1", "normal"), ("dev_dep1", "dev")],
- "manifest_path": "/path/to/Cargo.toml"
- }
- }
- """
- if not report or "unused_deps" not in report:
- return {}
-
- unused_deps = {}
- for package_full_name, deps_info in report["unused_deps"].items():
- package_name = package_full_name.split(" ")[0]
-
- all_unused = []
- if deps_info.get("normal"):
- all_unused.extend([(dep, "normal") for dep in deps_info["normal"]])
- if deps_info.get("development"):
- all_unused.extend([(dep, "dev") for dep in deps_info["development"]])
- if deps_info.get("build"):
- all_unused.extend([(dep, "build") for dep in deps_info["build"]])
-
- if all_unused:
- unused_deps[package_name] = {
- "dependencies": all_unused,
- "manifest_path": deps_info.get("manifest_path", "unknown"),
- }
-
- return unused_deps
-
-
-def get_section_pattern(dep_type):
- """
- Get regex patterns to identify different dependency sections in Cargo.toml.
-
- Args:
- dep_type (str): Type of dependency ("normal", "dev", or "build")
-
- Returns:
- list: List of regex patterns to match the appropriate section headers
-
- """
- patterns = {
- "normal": [r"\[dependencies\]", r"\[dependencies\..*?\]"],
- "dev": [r"\[dev-dependencies\]", r"\[dev-dependencies\..*?\]"],
- "build": [r"\[build-dependencies\]", r"\[build-dependencies\..*?\]"],
- }
- return patterns.get(dep_type, [])
-
-
-def remove_dependency_line(content, dep_name, section_start, section_end):
- """
- Remove a dependency line from a specific section of a Cargo.toml file.
-
- Args:
- content (str): The entire content of the Cargo.toml file
- dep_name (str): Name of the dependency to remove (e.g., "serde", "tokio")
- section_start (int): Starting position of the section in the content
- section_end (int): Ending position of the section in the content
-
- Returns:
- tuple: (new_content, removed) where:
- - new_content (str): The modified content with dependency removed
- - removed (bool): True if dependency was found and removed, False otherwise
-
- Example input content format:
- content = '''
- [package]
- name = "my-crate"
- version = "0.1.0"
-
- [dependencies]
- serde = "1.0"
- tokio = { version = "1.0", features = ["full"] }
- serde_json.workspace = true
-
- [dev-dependencies]
- tempfile = "3.0"
- '''
-
- # If dep_name = "serde", section_start = start of [dependencies],
- # section_end = start of [dev-dependencies], this function will:
- # 1. Extract the section: "serde = "1.0"\ntokio = { version = "1.0", features = ["full"] }\nserde_json.workspace = true\n"
- # 2. Find and remove the line: "serde = "1.0""
- # 3. Return the modified content with that line removed
- """
- section_content = content[section_start:section_end]
-
- dep_patterns = [
- rf"^{re.escape(dep_name)}\s*=.*$", # e.g., "serde = "1.0""
- rf"^{re.escape(dep_name)}\.workspace\s*=.*$", # e.g., "serde_json.workspace = true"
- ]
-
- for pattern in dep_patterns:
- match = re.search(pattern, section_content, re.MULTILINE)
- if match:
- line_start = section_start + match.start() # Start of the matched line
- line_end = section_start + match.end() # End of the matched line
-
- if line_end < len(content) and content[line_end] == "\n":
- line_end += 1
-
- return content[:line_start] + content[line_end:], True
-
- return content, False
-
-
-def remove_dependency_from_toml(file_path, dep_name, dep_type):
- """
- Remove a specific dependency from a Cargo.toml file.
-
- Args:
- file_path (str): Path to the Cargo.toml file
- dep_name (str): Name of the dependency to remove
- dep_type (str): Type of dependency ("normal", "dev", or "build")
-
- Returns:
- bool: True if dependency was successfully removed, False otherwise
- """
- try:
- with open(file_path, "r") as f:
- content = f.read()
-
- section_patterns = get_section_pattern(dep_type)
- if not section_patterns:
- return False
-
- for pattern in section_patterns:
- section_match = re.search(pattern, content, re.IGNORECASE)
- if not section_match:
- continue
-
- section_start = section_match.end()
- next_section = re.search(r"\n\s*\[", content[section_start:])
- section_end = (
- section_start + next_section.start() if next_section else len(content)
- )
-
- new_content, removed = remove_dependency_line(
- content, dep_name, section_start, section_end
- )
- if removed:
- with open(file_path, "w") as f:
- f.write(new_content)
- return True
-
- return False
-
- except Exception as e:
- print(f"Error processing {file_path}: {e}")
- return False
-
-
-def process_unused_dependencies(unused_deps):
- """
- Process and remove all unused dependencies from their respective Cargo.toml files.
-
- Args:
- unused_deps (dict): Dictionary of unused dependencies organized by package:
- {
- "package_name": {
- "dependencies": [("dep1", "normal"), ("dev_dep1", "dev")],
- "manifest_path": "/path/to/Cargo.toml"
- }
- }
-
- """
- if not unused_deps:
- print("No unused dependencies found.")
- return
-
- total_removed = 0
- total_failed = 0
-
- for package, info in unused_deps.items():
- deps = info["dependencies"]
- manifest_path = info["manifest_path"]
-
- if not os.path.exists(manifest_path):
- print(f"Manifest file not found: {manifest_path}")
- total_failed += len(deps)
- continue
-
- for dep, dep_type in deps:
- if remove_dependency_from_toml(manifest_path, dep, dep_type):
- print(f"Removed {dep} from {package}")
- total_removed += 1
- else:
- print(f"Failed to remove {dep} from {package}")
- total_failed += 1
-
- print(f"Removed {total_removed} dependencies")
- if total_failed > 0:
- print(f"Failed to remove {total_failed} dependencies")
-
-
-def main():
- if len(sys.argv) > 1:
- report_path = sys.argv[1]
- else:
- report_path = "udeps-report.json"
-
- report = load_udeps_report(report_path)
- if report is None:
- sys.exit(1)
-
- unused_deps = extract_unused_dependencies(report)
- process_unused_dependencies(unused_deps)
-
-
-if __name__ == "__main__":
- main()
diff --git a/src/catalog/Cargo.toml b/src/catalog/Cargo.toml
index f46887e148a..fe42ddc2d13 100644
--- a/src/catalog/Cargo.toml
+++ b/src/catalog/Cargo.toml
@@ -61,9 +61,7 @@ tokio-stream.workspace = true
[dev-dependencies]
cache.workspace = true
-catalog = { workspace = true, features = ["testing"] }
chrono.workspace = true
common-meta = { workspace = true, features = ["testing"] }
common-query = { workspace = true, features = ["testing"] }
-object-store.workspace = true
tokio.workspace = true
diff --git a/src/cli/Cargo.toml b/src/cli/Cargo.toml
index 587e5b8cffa..8178fa1b0b5 100644
--- a/src/cli/Cargo.toml
+++ b/src/cli/Cargo.toml
@@ -18,10 +18,8 @@ workspace = true
[dependencies]
async-stream.workspace = true
async-trait.workspace = true
-auth.workspace = true
base64.workspace = true
cache.workspace = true
-catalog.workspace = true
chrono.workspace = true
clap.workspace = true
client = { workspace = true, features = ["testing"] }
@@ -34,14 +32,11 @@ common-grpc.workspace = true
common-macro.workspace = true
common-meta.workspace = true
common-procedure.workspace = true
-common-query.workspace = true
-common-recordbatch.workspace = true
common-runtime.workspace = true
common-telemetry = { workspace = true, features = [
"deadlock_detection",
] }
common-time.workspace = true
-common-version.workspace = true
common-wal.workspace = true
datatypes.workspace = true
etcd-client.workspace = true
@@ -61,21 +56,18 @@ reqwest.workspace = true
serde.workspace = true
serde_json.workspace = true
servers.workspace = true
-session.workspace = true
snafu.workspace = true
sql.workspace = true
standalone.workspace = true
store-api.workspace = true
table.workspace = true
tokio.workspace = true
-tracing-appender.workspace = true
url.workspace = true
uuid.workspace = true
[dev-dependencies]
common-meta = { workspace = true, features = ["testing"] }
common-test-util.workspace = true
-common-version.workspace = true
object-store = { workspace = true, features = ["testing"] }
serde.workspace = true
tempfile.workspace = true
diff --git a/src/cli/src/metadata/snapshot.rs b/src/cli/src/metadata/snapshot.rs
index 59b2599ad9c..9c010c2235b 100644
--- a/src/cli/src/metadata/snapshot.rs
+++ b/src/cli/src/metadata/snapshot.rs
@@ -12,11 +12,13 @@
// See the License for the specific language governing permissions and
// limitations under the License.
+use std::path::{Component, Path, PathBuf};
+
use async_trait::async_trait;
use clap::{Parser, Subcommand};
use common_error::ext::BoxedError;
use common_meta::snapshot::MetadataSnapshotManager;
-use object_store::{ObjectStore, services};
+use object_store::ObjectStore;
use crate::Tool;
use crate::common::{ObjectStoreConfig, StoreConfig, new_fs_object_store};
@@ -270,25 +272,44 @@ fn build_object_store_and_resolve_file_path(
fs_root: &str,
file_path: &str,
) -> Result<(ObjectStore, String), BoxedError> {
- let object_store = object_store.build().map_err(BoxedError::new)?;
- let object_store = match object_store {
- Some(object_store) => object_store,
- None => new_fs_object_store(fs_root)?,
- };
+ if let Some(object_store) = object_store.build().map_err(BoxedError::new)? {
+ return Ok((object_store, file_path.to_string()));
+ }
- let file_path = if object_store.info().scheme() == services::FS_SCHEME {
- resolve_relative_path_with_current_dir(file_path).map_err(BoxedError::new)?
+ let absolute_path =
+ resolve_relative_path_with_current_dir(file_path).map_err(BoxedError::new)?;
+ let (root, key) = split_fs_snapshot_path(fs_root, &absolute_path);
+ Ok((new_fs_object_store(&root)?, key))
+}
+
+/// Separates a local absolute path into an operator root and a relative key.
+fn split_fs_snapshot_path(fs_root: &str, absolute_path: &str) -> (String, String) {
+ let path = Path::new(absolute_path);
+ // OpenDAL keys cannot contain a Windows drive/UNC prefix or a root directory.
+ // Keep the filesystem anchor in the operator root, not in the object key.
+ let anchor: PathBuf = path
+ .components()
+ .take_while(|c| matches!(c, Component::Prefix(_) | Component::RootDir))
+ .collect();
+ let key = path
+ .components()
+ .filter(|c| !matches!(c, Component::Prefix(_) | Component::RootDir))
+ .map(|c| c.as_os_str().to_string_lossy())
+ .collect::>()
+ .join("/");
+ // Preserve the existing custom-root semantics: cwd-resolved keys are rooted
+ // under --dir. The default root follows the input path's drive on Windows.
+ let root = if fs_root == "/" {
+ anchor.to_string_lossy()
} else {
- file_path.to_string()
+ fs_root.into()
};
-
- Ok((object_store, file_path))
+ (root.into_owned(), key)
}
#[cfg(test)]
mod tests {
use std::env;
- use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
@@ -311,44 +332,159 @@ mod tests {
#[tokio::test]
async fn test_cmd_resolve_file_path() {
- common_telemetry::init_default_ut_logging();
- let cmd = RestoreCommand::parse_from([
- "",
- "--file_name",
- "metadata_snapshot.metadata.fb",
- "--backend",
- "memory-store",
- "--store-addrs",
- "memory://",
- ]);
- let tool = cmd.build().await.unwrap();
let current_dir = env::current_dir().unwrap();
- let file_path = current_dir.join("metadata_snapshot.metadata.fb");
- assert_eq!(tool.file_path, file_path.to_string_lossy().to_string());
+ for file_path in [
+ "metadata_snapshot.metadata.fb".to_string(),
+ current_dir
+ .join("metadata_snapshot.metadata.fb")
+ .display()
+ .to_string(),
+ ] {
+ let cmd = RestoreCommand::parse_from([
+ "",
+ "--file_name",
+ &file_path,
+ "--backend",
+ "memory-store",
+ "--store-addrs",
+ "memory://",
+ ]);
+ let tool = cmd.build().await.unwrap();
+ assert!(!Path::new(&tool.file_path).is_absolute());
+ assert!(
+ Path::new(&tool.file_path)
+ .components()
+ .all(|c| matches!(c, Component::Normal(_)))
+ );
+ let expected = current_dir
+ .components()
+ .filter_map(|c| match c {
+ Component::Normal(part) => Some(part.to_str().unwrap()),
+ _ => None,
+ })
+ .chain(["metadata_snapshot.metadata.fb"])
+ .collect::>()
+ .join("/");
+ assert_eq!(tool.file_path, expected);
+ }
+ }
+ #[cfg(windows)]
+ #[test]
+ fn test_split_windows_snapshot_path() {
+ for (path, root, key) in [
+ (
+ r"C:\backup\snapshot.metadata.fb",
+ r"C:\",
+ "backup/snapshot.metadata.fb",
+ ),
+ (
+ r"D:\备份 dir\snapshot.metadata.fb",
+ r"D:\",
+ "备份 dir/snapshot.metadata.fb",
+ ),
+ (
+ r"\\server\share\backup\snapshot.metadata.fb",
+ r"\\server\share\",
+ "backup/snapshot.metadata.fb",
+ ),
+ (
+ r"\\?\C:\backup\snapshot.metadata.fb",
+ r"\\?\C:\",
+ "backup/snapshot.metadata.fb",
+ ),
+ ] {
+ assert_eq!(
+ split_fs_snapshot_path("/", path),
+ (root.to_string(), key.to_string())
+ );
+ assert_eq!(
+ split_fs_snapshot_path(r"D:\root", path),
+ (r"D:\root".to_string(), key.to_string())
+ );
+ }
+ }
+
+ #[tokio::test]
+ async fn test_fs_snapshot_paths() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ for name in ["snapshot.metadata.fb", "nested dir/快照.metadata.fb"] {
+ let path = temp_dir.path().join(name);
+ let (store, key) = build_object_store_and_resolve_file_path(
+ ObjectStoreConfig::default(),
+ "/",
+ path.to_str().unwrap(),
+ )
+ .unwrap();
+ assert!(!path.exists());
+ store.write(&key, "snapshot").await.unwrap();
+ assert_eq!(std::fs::read(&path).unwrap(), b"snapshot");
+ assert_eq!(store.read(&key).await.unwrap().to_bytes(), &b"snapshot"[..]);
+ }
+ }
+
+ #[tokio::test]
+ async fn test_fs_snapshot_custom_root() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let (store, key) = build_object_store_and_resolve_file_path(
+ ObjectStoreConfig::default(),
+ temp_dir.path().to_str().unwrap(),
+ "snapshot.metadata.fb",
+ )
+ .unwrap();
+ let current_dir = env::current_dir().unwrap();
+ let expected = current_dir
+ .components()
+ .filter_map(|c| match c {
+ Component::Normal(part) => Some(part.to_str().unwrap()),
+ _ => None,
+ })
+ .chain(["snapshot.metadata.fb"])
+ .collect::>()
+ .join("/");
+ assert_eq!(key, expected);
+ store.write(&key, "snapshot").await.unwrap();
+ assert_eq!(
+ std::fs::read(temp_dir.path().join(expected)).unwrap(),
+ b"snapshot"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_fs_snapshot_rejects_parent_traversal() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let path = temp_dir.path().join("../snapshot.metadata.fb");
+ let (store, key) = build_object_store_and_resolve_file_path(
+ ObjectStoreConfig::default(),
+ "/",
+ path.to_str().unwrap(),
+ )
+ .unwrap();
+ assert!(key.split('/').any(|part| part == ".."));
+ let err = store.write(&key, "snapshot").await.unwrap_err();
+ assert_eq!(err.kind(), object_store::ErrorKind::NotFound);
+ assert!(err.to_string().contains("path escapes the configured root"));
+ }
+
+ #[test]
+ fn test_remote_snapshot_key_unchanged() {
let cmd = RestoreCommand::parse_from([
"",
- "--file_name",
- "metadata_snapshot.metadata.fb",
"--backend",
"memory-store",
"--store-addrs",
"memory://",
+ "--s3",
+ "--s3-bucket",
+ "snapshots",
+ "--s3-region",
+ "us-east-1",
]);
- let tool = cmd.build().await.unwrap();
- assert_eq!(tool.file_path, file_path.to_string_lossy().to_string());
-
- let cmd = RestoreCommand::parse_from([
- "",
- "--file_name",
- "metadata_snapshot.metadata.fb",
- "--backend",
- "memory-store",
- "--store-addrs",
- "memory://",
- ]);
- let tool = cmd.build().await.unwrap();
- assert_eq!(tool.file_path, file_path.to_string_lossy().to_string());
+ let key = "nested/snapshot.metadata.fb";
+ let (store, resolved) =
+ build_object_store_and_resolve_file_path(cmd.object_store, "/ignored", key).unwrap();
+ assert_eq!(store.info().scheme(), object_store::services::S3_SCHEME);
+ assert_eq!(resolved, key);
}
async fn setup_backup_file(object_store: ObjectStore, file_path: &str) {
diff --git a/src/client/Cargo.toml b/src/client/Cargo.toml
index 5b479b1fbf2..36c3f60a9a4 100644
--- a/src/client/Cargo.toml
+++ b/src/client/Cargo.toml
@@ -45,7 +45,3 @@ substrait.workspace = true
tokio.workspace = true
tokio-stream = { workspace = true, features = ["net"] }
tonic.workspace = true
-
-[dev-dependencies]
-common-grpc-expr.workspace = true
-tracing = "0.1"
diff --git a/src/cmd/Cargo.toml b/src/cmd/Cargo.toml
index fff268dea45..367b86809d8 100644
--- a/src/cmd/Cargo.toml
+++ b/src/cmd/Cargo.toml
@@ -43,10 +43,8 @@ workspace = true
api.workspace = true
async-trait.workspace = true
auth.workspace = true
-base64.workspace = true
cache.workspace = true
catalog.workspace = true
-chrono.workspace = true
clap.workspace = true
cli.workspace = true
client.workspace = true
@@ -63,7 +61,6 @@ common-meta.workspace = true
common-options.workspace = true
common-procedure.workspace = true
common-query.workspace = true
-common-recordbatch.workspace = true
common-runtime.workspace = true
common-stat.workspace = true
common-telemetry = { workspace = true, features = [
@@ -78,19 +75,16 @@ datafusion-physical-plan.workspace = true
datafusion_object_store.workspace = true
datanode.workspace = true
datatypes.workspace = true
-either = "1.15"
etcd-client.workspace = true
flow.workspace = true
frontend = { workspace = true, default-features = false }
futures.workspace = true
hex.workspace = true
human-panic = "2.0"
-humantime.workspace = true
lazy_static.workspace = true
log-store.workspace = true
meta-client.workspace = true
meta-srv.workspace = true
-metric-engine.workspace = true
mito-codec.workspace = true
mito2.workspace = true
moka = { workspace = true, features = ["future"] }
@@ -107,16 +101,13 @@ reqwest.workspace = true
serde.workspace = true
serde_json.workspace = true
servers.workspace = true
-session.workspace = true
smallvec.workspace = true
snafu.workspace = true
sqlparser.workspace = true
standalone.workspace = true
store-api.workspace = true
-table.workspace = true
tokio.workspace = true
toml.workspace = true
-tonic.workspace = true
tracing-appender.workspace = true
[target.'cfg(unix)'.dependencies]
@@ -134,6 +125,7 @@ client = { workspace = true, features = ["testing"] }
common-test-util.workspace = true
common-version.workspace = true
file-engine.workspace = true
+metric-engine.workspace = true
mito2.workspace = true
serde.workspace = true
similar-asserts.workspace = true
diff --git a/src/common/config/Cargo.toml b/src/common/config/Cargo.toml
index 27b238add73..ad356716867 100644
--- a/src/common/config/Cargo.toml
+++ b/src/common/config/Cargo.toml
@@ -29,4 +29,3 @@ datanode.workspace = true
meta-client.workspace = true
serde.workspace = true
temp-env = "0.3"
-tempfile.workspace = true
diff --git a/src/common/datasource/src/file_format/csv.rs b/src/common/datasource/src/file_format/csv.rs
index d016d4eabe9..d4cfb45088c 100644
--- a/src/common/datasource/src/file_format/csv.rs
+++ b/src/common/datasource/src/file_format/csv.rs
@@ -647,16 +647,15 @@ mod tests {
let compressed_file_name =
format!("test_compressed_csv.{}", compression_type.file_extension());
let compressed_file_path = temp_dir.path().join(&compressed_file_name);
- let compressed_file_path_str = compressed_file_path.to_str().unwrap();
// Create a simple file store for testing
- let store = test_store("/");
+ let store = test_store(temp_dir.path().to_str().unwrap());
// Export CSV with compression
let rows = stream_to_csv(
Box::pin(DfRecordBatchStreamAdapter::new(recordbatches.as_stream())),
store,
- compressed_file_path_str,
+ &compressed_file_name,
1024,
1,
&format,
@@ -708,13 +707,13 @@ mod tests {
}
// Verify the compressed file can be decompressed and content matches original data
- let store = test_store("/");
+ let store = test_store(temp_dir.path().to_str().unwrap());
let schema = Arc::new(
CsvFormat {
compression_type,
..Default::default()
}
- .infer_schema(&store, compressed_file_path_str)
+ .infer_schema(&store, &compressed_file_name)
.await
.unwrap(),
);
@@ -722,7 +721,7 @@ mod tests {
let stream = file_to_stream(
&store,
- compressed_file_path_str,
+ &compressed_file_name,
csv_source.clone(),
None,
compression_type,
@@ -761,13 +760,13 @@ mod tests {
)
.unwrap();
- let store = test_store("/");
+ let store = test_store(temp_dir.path().to_str().unwrap());
let schema = Arc::new(arrow_schema::Schema::new(vec![
Field::new("id", DataType::UInt32, false),
Field::new("name", DataType::Utf8, false),
Field::new("value", DataType::Float64, false),
]));
- let path = csv_file_path.to_str().unwrap();
+ let path = "input.csv";
let stream =
tolerant_csv_stream(&store, path, schema, vec![0, 1, 2], &CsvFormat::default())
@@ -793,13 +792,13 @@ mod tests {
let csv_file_path = temp_dir.path().join("input.csv");
std::fs::write(&csv_file_path, "id,name,value\n1,Alice,10.5\n2,Bob\n").unwrap();
- let store = test_store("/");
+ let store = test_store(temp_dir.path().to_str().unwrap());
let schema = Arc::new(arrow_schema::Schema::new(vec![
Field::new("id", DataType::UInt32, false),
Field::new("name", DataType::Utf8, false),
Field::new("value", DataType::Float64, false),
]));
- let path = csv_file_path.to_str().unwrap();
+ let path = "input.csv";
let stream =
tolerant_csv_stream(&store, path, schema, vec![0, 1, 2], &CsvFormat::default())
diff --git a/src/common/datasource/src/file_format/json.rs b/src/common/datasource/src/file_format/json.rs
index 3c24a4b7e27..23ce461fcb2 100644
--- a/src/common/datasource/src/file_format/json.rs
+++ b/src/common/datasource/src/file_format/json.rs
@@ -313,16 +313,15 @@ mod tests {
let compressed_file_name =
format!("test_compressed_json.{}", compression_type.file_extension());
let compressed_file_path = temp_dir.path().join(&compressed_file_name);
- let compressed_file_path_str = compressed_file_path.to_str().unwrap();
// Create a simple file store for testing
- let store = test_store("/");
+ let store = test_store(temp_dir.path().to_str().unwrap());
// Export JSON with compression
let rows = stream_to_json(
Box::pin(DfRecordBatchStreamAdapter::new(recordbatches.as_stream())),
store,
- compressed_file_path_str,
+ &compressed_file_name,
1024,
1,
&format,
@@ -374,13 +373,13 @@ mod tests {
}
// Verify the compressed file can be decompressed and content matches original data
- let store = test_store("/");
+ let store = test_store(temp_dir.path().to_str().unwrap());
let schema = Arc::new(
JsonFormat {
compression_type,
..Default::default()
}
- .infer_schema(&store, compressed_file_path_str)
+ .infer_schema(&store, &compressed_file_name)
.await
.unwrap(),
);
@@ -388,7 +387,7 @@ mod tests {
let stream = file_to_stream(
&store,
- compressed_file_path_str,
+ &compressed_file_name,
json_source.clone(),
None,
compression_type,
diff --git a/src/common/datasource/src/file_format/tests.rs b/src/common/datasource/src/file_format/tests.rs
index 27b86e1b114..14be2b1777f 100644
--- a/src/common/datasource/src/file_format/tests.rs
+++ b/src/common/datasource/src/file_format/tests.rs
@@ -66,13 +66,12 @@ impl Test<'_> {
#[tokio::test]
async fn test_json_opener() {
- let store = test_store("/");
+ let root = find_workspace_path("/src/common/datasource/tests");
+ let store = test_store(root.to_str().unwrap());
let schema = basic_schema_with_time_format();
let file_source = Arc::new(JsonSource::new(schema)).with_batch_size(test_util::TEST_BATCH_SIZE);
- let path = &find_workspace_path("/src/common/datasource/tests/json/basic.json")
- .display()
- .to_string();
+ let path = "json/basic.json";
let tests = [
Test {
@@ -108,11 +107,10 @@ async fn test_json_opener() {
#[tokio::test]
async fn test_csv_opener() {
- let store = test_store("/");
+ let root = find_workspace_path("/src/common/datasource/tests");
+ let store = test_store(root.to_str().unwrap());
let schema = basic_schema_with_time_format();
- let path = &find_workspace_path("/src/common/datasource/tests/csv/basic.csv")
- .display()
- .to_string();
+ let path = "csv/basic.csv";
let file_source = CsvSource::new(schema).with_batch_size(test_util::TEST_BATCH_SIZE);
@@ -150,13 +148,12 @@ async fn test_csv_opener() {
#[tokio::test(flavor = "multi_thread")]
async fn test_parquet_exec() {
- let store = test_store("/");
+ let root = find_workspace_path("/src/common/datasource/tests");
+ let store = test_store(root.to_str().unwrap());
let schema = test_basic_schema();
- let path = &find_workspace_path("/src/common/datasource/tests/parquet/basic.parquet")
- .display()
- .to_string();
+ let path = "parquet/basic.parquet";
let parquet_source = ParquetSource::new(schema)
.with_parquet_file_reader_factory(Arc::new(DefaultParquetFileReaderFactory::new(store)));
@@ -191,11 +188,10 @@ async fn test_parquet_exec() {
#[tokio::test]
async fn test_orc_opener() {
- let path = &find_workspace_path("/src/common/datasource/tests/orc/test.orc")
- .display()
- .to_string();
+ let path = "orc/test.orc";
- let store = test_store("/");
+ let root = find_workspace_path("/src/common/datasource/tests");
+ let store = test_store(root.to_str().unwrap());
let schema = Arc::new(OrcFormat.infer_schema(&store, path).await.unwrap());
let file_source = Arc::new(OrcSource::new(schema.into()));
diff --git a/src/common/datasource/src/test_util.rs b/src/common/datasource/src/test_util.rs
index 4cefda88890..f95d24930b2 100644
--- a/src/common/datasource/src/test_util.rs
+++ b/src/common/datasource/src/test_util.rs
@@ -56,8 +56,8 @@ pub fn test_store(root: &str) -> ObjectStore {
pub fn test_tmp_store(root: &str) -> (ObjectStore, TempDir) {
let dir = create_temp_dir(root);
- let builder = Fs::default();
- (ObjectStore::new(builder.root("/")).unwrap(), dir)
+ let store = test_store(dir.path().to_str().unwrap());
+ (store, dir)
}
pub fn test_basic_schema() -> SchemaRef {
@@ -94,9 +94,11 @@ pub(crate) fn scan_config(
.build()
}
-pub async fn setup_stream_to_json_test(origin_path: &str, threshold: impl Fn(usize) -> usize) {
- let store = test_store("/");
-
+pub async fn setup_stream_to_json_test(
+ store: &ObjectStore,
+ origin_path: &str,
+ threshold: impl Fn(usize) -> usize,
+) {
let schema = basic_schema_with_time_format();
let json_opener = JsonOpener::new(
@@ -117,9 +119,9 @@ pub async fn setup_stream_to_json_test(origin_path: &str, threshold: impl Fn(usi
.build()
.unwrap();
- let (tmp_store, dir) = test_tmp_store("test_stream_to_json");
+ let (tmp_store, _dir) = test_tmp_store("test_stream_to_json");
- let output_path = format!("{}/{}", dir.path().display(), "output");
+ let output_path = "output";
let json_format = JsonFormat::default();
@@ -127,7 +129,7 @@ pub async fn setup_stream_to_json_test(origin_path: &str, threshold: impl Fn(usi
stream_to_json(
Box::pin(stream),
tmp_store.clone(),
- &output_path,
+ output_path,
threshold(size),
8,
&json_format,
@@ -136,18 +138,17 @@ pub async fn setup_stream_to_json_test(origin_path: &str, threshold: impl Fn(usi
.is_ok()
);
- let written = tmp_store.read(&output_path).await.unwrap();
+ let written = tmp_store.read(output_path).await.unwrap();
let origin = store.read(origin_path).await.unwrap();
assert_eq_lines(written.to_vec(), origin.to_vec());
}
pub async fn setup_stream_to_csv_test(
+ store: &ObjectStore,
origin_path: &str,
format_path: &str,
threshold: impl Fn(usize) -> usize,
) {
- let store = test_store("/");
-
let schema = basic_schema_with_time_format();
let csv_source = CsvSource::new(schema).with_batch_size(TEST_BATCH_SIZE);
@@ -168,9 +169,9 @@ pub async fn setup_stream_to_csv_test(
.build()
.unwrap();
- let (tmp_store, dir) = test_tmp_store("test_stream_to_csv");
+ let (tmp_store, _dir) = test_tmp_store("test_stream_to_csv");
- let output_path = format!("{}/{}", dir.path().display(), "output");
+ let output_path = "output";
let csv_format = CsvFormat {
timestamp_format: Some("%m-%d-%Y".to_string()),
@@ -183,7 +184,7 @@ pub async fn setup_stream_to_csv_test(
stream_to_csv(
Box::pin(stream),
tmp_store.clone(),
- &output_path,
+ output_path,
threshold(size),
8,
&csv_format,
@@ -192,7 +193,7 @@ pub async fn setup_stream_to_csv_test(
.is_ok()
);
- let written = tmp_store.read(&output_path).await.unwrap();
+ let written = tmp_store.read(output_path).await.unwrap();
let format_expect = store.read(format_path).await.unwrap();
assert_eq_lines(written.to_vec(), format_expect.to_vec());
}
diff --git a/src/common/datasource/src/tests.rs b/src/common/datasource/src/tests.rs
index 8be9add7c88..b983ef86627 100644
--- a/src/common/datasource/src/tests.rs
+++ b/src/common/datasource/src/tests.rs
@@ -18,34 +18,31 @@ use crate::test_util;
#[tokio::test]
async fn test_stream_to_json() {
- let origin_path = &find_workspace_path("/src/common/datasource/tests/json/basic.json")
- .display()
- .to_string();
+ let root = find_workspace_path("/src/common/datasource/tests/json");
+ let store = test_util::test_store(root.to_str().unwrap());
+ let origin_path = "basic.json";
// A small threshold
// Triggers the flush each writes
- test_util::setup_stream_to_json_test(origin_path, |size| size / 2).await;
+ test_util::setup_stream_to_json_test(&store, origin_path, |size| size / 2).await;
// A large threshold
// Only triggers the flush at last
- test_util::setup_stream_to_json_test(origin_path, |size| size * 2).await;
+ test_util::setup_stream_to_json_test(&store, origin_path, |size| size * 2).await;
}
#[tokio::test]
async fn test_stream_to_csv() {
- let origin_path = &find_workspace_path("/src/common/datasource/tests/csv/basic.csv")
- .display()
- .to_string();
-
- let format_path = &find_workspace_path("/src/common/datasource/tests/csv/basic_format.csv")
- .display()
- .to_string();
+ let root = find_workspace_path("/src/common/datasource/tests/csv");
+ let store = test_util::test_store(root.to_str().unwrap());
+ let origin_path = "basic.csv";
+ let format_path = "basic_format.csv";
// A small threshold
// Triggers the flush each writes
- test_util::setup_stream_to_csv_test(origin_path, format_path, |size| size / 2).await;
+ test_util::setup_stream_to_csv_test(&store, origin_path, format_path, |size| size / 2).await;
// A large threshold
// Only triggers the flush at last
- test_util::setup_stream_to_csv_test(origin_path, format_path, |size| size * 2).await;
+ test_util::setup_stream_to_csv_test(&store, origin_path, format_path, |size| size * 2).await;
}
diff --git a/src/common/frontend/Cargo.toml b/src/common/frontend/Cargo.toml
index 32a5146b566..26092ae13f7 100644
--- a/src/common/frontend/Cargo.toml
+++ b/src/common/frontend/Cargo.toml
@@ -12,14 +12,9 @@ common-event-recorder.workspace = true
common-grpc.workspace = true
common-macro.workspace = true
common-meta.workspace = true
-common-time.workspace = true
greptime-proto.workspace = true
-humantime.workspace = true
meta-client.workspace = true
serde.workspace = true
serde_json.workspace = true
snafu.workspace = true
tonic.workspace = true
-
-[dev-dependencies]
-tokio.workspace = true
diff --git a/src/common/function/Cargo.toml b/src/common/function/Cargo.toml
index ee6cfac9dfb..3a5da30c136 100644
--- a/src/common/function/Cargo.toml
+++ b/src/common/function/Cargo.toml
@@ -17,21 +17,17 @@ ai_functions = ["dep:futures", "dep:reqwest"]
[dependencies]
ahash.workspace = true
api.workspace = true
-arc-swap = "1.0"
arrow.workspace = true
-arrow-cast.workspace = true
arrow-schema.workspace = true
async-trait.workspace = true
bincode = "=1.3.3"
catalog.workspace = true
-chrono.workspace = true
common-base.workspace = true
common-catalog.workspace = true
common-error.workspace = true
common-macro.workspace = true
common-meta.workspace = true
common-query.workspace = true
-common-runtime.workspace = true
common-telemetry.workspace = true
common-time.workspace = true
common-version.workspace = true
@@ -55,7 +51,6 @@ jsonb.workspace = true
memchr = "2.7"
mito-codec.workspace = true
nalgebra.workspace = true
-num = "0.4"
num-traits = "0.2"
paste.workspace = true
regex.workspace = true
diff --git a/src/common/grpc-expr/Cargo.toml b/src/common/grpc-expr/Cargo.toml
index a689de7c018..b735e8f9ba8 100644
--- a/src/common/grpc-expr/Cargo.toml
+++ b/src/common/grpc-expr/Cargo.toml
@@ -9,7 +9,6 @@ workspace = true
[dependencies]
api.workspace = true
-common-base.workspace = true
common-catalog.workspace = true
common-error.workspace = true
common-macro.workspace = true
@@ -21,6 +20,3 @@ prost.workspace = true
snafu.workspace = true
store-api.workspace = true
table.workspace = true
-
-[dev-dependencies]
-paste.workspace = true
diff --git a/src/common/mem-prof/Cargo.toml b/src/common/mem-prof/Cargo.toml
index 10b67cf7fa7..46d959b02cc 100644
--- a/src/common/mem-prof/Cargo.toml
+++ b/src/common/mem-prof/Cargo.toml
@@ -26,3 +26,9 @@ jemalloc-pprof-mappings = { version = "0.7", package = "mappings" } # for get th
[target.'cfg(not(windows))'.dependencies.tikv-jemalloc-sys]
features = ["stats", "profiling", "unprefixed_malloc_on_supported_platforms"]
version = "0.7"
+
+[package.metadata.cargo-shear]
+# `tikv-jemalloc-sys` is a link-only dependency: it enables jemalloc
+# profiling/stats features and pins the linked jemalloc instance, but is
+# never referenced by name in source code.
+ignored = ["tikv-jemalloc-sys"]
diff --git a/src/common/meta/Cargo.toml b/src/common/meta/Cargo.toml
index 8deb3d00814..bd974deb4cf 100644
--- a/src/common/meta/Cargo.toml
+++ b/src/common/meta/Cargo.toml
@@ -46,7 +46,6 @@ common-time.workspace = true
common-wal.workspace = true
common-workload.workspace = true
datafusion-common.workspace = true
-datafusion-expr.workspace = true
datatypes.workspace = true
deadpool = { workspace = true, optional = true }
deadpool-postgres = { workspace = true, optional = true }
@@ -55,7 +54,6 @@ etcd-client.workspace = true
flexbuffers = "25.2"
futures.workspace = true
futures-util.workspace = true
-greptime-proto.workspace = true
hex.workspace = true
humantime-serde.workspace = true
itertools.workspace = true
diff --git a/src/common/meta/src/snapshot.rs b/src/common/meta/src/snapshot.rs
index f35c86e2e7b..3582b0acecb 100644
--- a/src/common/meta/src/snapshot.rs
+++ b/src/common/meta/src/snapshot.rs
@@ -408,7 +408,6 @@ mod tests {
async fn test_dump_and_restore() {
common_telemetry::init_default_ut_logging();
let (temp_dir, kv_backend, manager) = test_env("test_dump_and_restore");
- let temp_path = temp_dir.path();
for i in 0..10 {
kv_backend
@@ -420,32 +419,16 @@ mod tests {
.await
.unwrap();
}
- let dump_path = temp_path.join("snapshot");
- manager
- .dump(&format!(
- "{}/metadata_snapshot",
- &dump_path.as_path().display().to_string()
- ))
- .await
- .unwrap();
+ manager.dump("snapshot/metadata_snapshot").await.unwrap();
+ let snapshot_key = "snapshot/metadata_snapshot.metadata.fb";
+ assert!(temp_dir.path().join("data").join(snapshot_key).is_file());
// Clean up the kv backend
kv_backend.clear();
- let err = manager
- .dump(&format!(
- "{}/metadata_snapshot.metadata.fb",
- &dump_path.as_path().display().to_string()
- ))
- .await
- .unwrap_err();
+ let err = manager.dump(snapshot_key).await.unwrap_err();
assert_matches!(err, Error::Unexpected { .. });
assert!(err.to_string().contains("already exists"));
- let restore_path = dump_path
- .join("metadata_snapshot.metadata.fb")
- .as_path()
- .display()
- .to_string();
- manager.restore(&restore_path).await.unwrap();
+ manager.restore(snapshot_key).await.unwrap();
for i in 0..10 {
let key = format!("test_{}", i);
@@ -456,14 +439,11 @@ mod tests {
#[tokio::test]
async fn test_restore_from_nonexistent_file() {
- let (temp_dir, _kv_backend, manager) = test_env("test_restore_from_nonexistent_file");
- let restore_path = temp_dir
- .path()
- .join("nonexistent.metadata.fb")
- .as_path()
- .display()
- .to_string();
- let err = manager.restore(&restore_path).await.unwrap_err();
+ let (_temp_dir, _kv_backend, manager) = test_env("test_restore_from_nonexistent_file");
+ let err = manager
+ .restore("nonexistent.metadata.fb")
+ .await
+ .unwrap_err();
assert_matches!(err, Error::ReadObject { .. })
}
}
diff --git a/src/common/procedure-test/Cargo.toml b/src/common/procedure-test/Cargo.toml
index e445a235bc9..907035dd052 100644
--- a/src/common/procedure-test/Cargo.toml
+++ b/src/common/procedure-test/Cargo.toml
@@ -10,5 +10,4 @@ workspace = true
[dependencies]
async-trait.workspace = true
common-procedure = { workspace = true, features = ["testing"] }
-snafu.workspace = true
tokio.workspace = true
diff --git a/src/common/query/Cargo.toml b/src/common/query/Cargo.toml
index 9706098826f..4acdfafcb6c 100644
--- a/src/common/query/Cargo.toml
+++ b/src/common/query/Cargo.toml
@@ -39,3 +39,8 @@ common-base.workspace = true
futures-util.workspace = true
serde_json.workspace = true
tokio.workspace = true
+
+[package.metadata.cargo-shear]
+# `sqlparser` is required by `sqlparser_derive`: the generated code references
+# `::sqlparser` paths, so it must stay a direct dependency.
+ignored = ["sqlparser"]
diff --git a/src/common/runtime/Cargo.toml b/src/common/runtime/Cargo.toml
index 927aa17b9c2..410173f6445 100644
--- a/src/common/runtime/Cargo.toml
+++ b/src/common/runtime/Cargo.toml
@@ -25,13 +25,11 @@ futures.workspace = true
lazy_static.workspace = true
num_cpus.workspace = true
once_cell.workspace = true
-parking_lot.workspace = true
paste.workspace = true
pin-project.workspace = true
prometheus.workspace = true
ratelimit.workspace = true
serde.workspace = true
-serde_json.workspace = true
snafu.workspace = true
tempfile.workspace = true
tokio.workspace = true
diff --git a/src/common/sql/Cargo.toml b/src/common/sql/Cargo.toml
index 3a66ec2d04e..318685996d5 100644
--- a/src/common/sql/Cargo.toml
+++ b/src/common/sql/Cargo.toml
@@ -10,7 +10,6 @@ common-base.workspace = true
common-decimal.workspace = true
common-error.workspace = true
common-macro.workspace = true
-common-telemetry.workspace = true
common-time.workspace = true
datafusion-sql.workspace = true
datatypes.workspace = true
diff --git a/src/common/stat/Cargo.toml b/src/common/stat/Cargo.toml
index 2a99107c909..fcf927e9cca 100644
--- a/src/common/stat/Cargo.toml
+++ b/src/common/stat/Cargo.toml
@@ -8,7 +8,6 @@ license.workspace = true
common-base.workspace = true
common-runtime.workspace = true
common-telemetry.workspace = true
-lazy_static.workspace = true
num_cpus.workspace = true
prometheus.workspace = true
sysinfo.workspace = true
diff --git a/src/common/substrait/Cargo.toml b/src/common/substrait/Cargo.toml
index a4e78475051..a56b56b8301 100644
--- a/src/common/substrait/Cargo.toml
+++ b/src/common/substrait/Cargo.toml
@@ -13,7 +13,6 @@ bytes.workspace = true
common-error.workspace = true
common-function.workspace = true
common-macro.workspace = true
-common-telemetry.workspace = true
datafusion.workspace = true
datafusion-common.workspace = true
datafusion-expr.workspace = true
@@ -21,7 +20,3 @@ datafusion-substrait.workspace = true
promql.workspace = true
prost.workspace = true
snafu.workspace = true
-
-[dev-dependencies]
-datatypes.workspace = true
-tokio.workspace = true
diff --git a/src/datanode/Cargo.toml b/src/datanode/Cargo.toml
index 6b78858bd2f..7c126764723 100644
--- a/src/datanode/Cargo.toml
+++ b/src/datanode/Cargo.toml
@@ -16,7 +16,6 @@ api.workspace = true
arrow-flight.workspace = true
async-trait.workspace = true
bytes.workspace = true
-client.workspace = true
common-base.workspace = true
common-catalog.workspace = true
common-config.workspace = true
@@ -24,7 +23,6 @@ common-datasource.workspace = true
common-error.workspace = true
common-function.workspace = true
common-greptimedb-telemetry.workspace = true
-common-grpc.workspace = true
common-macro.workspace = true
common-meta.workspace = true
common-options.workspace = true
@@ -75,7 +73,6 @@ tracing.workspace = true
[dev-dependencies]
cache.workspace = true
-client.workspace = true
common-meta = { workspace = true, features = ["testing"] }
common-query.workspace = true
common-test-util.workspace = true
diff --git a/src/datatypes/Cargo.toml b/src/datatypes/Cargo.toml
index 124b1e0ea64..d1dd70acef1 100644
--- a/src/datatypes/Cargo.toml
+++ b/src/datatypes/Cargo.toml
@@ -37,3 +37,8 @@ serde_json.workspace = true
snafu.workspace = true
sqlparser.workspace = true
sqlparser_derive = "0.1"
+
+[package.metadata.cargo-shear]
+# `sqlparser` is required by `sqlparser_derive`: the generated code references
+# `::sqlparser` paths, so it must stay a direct dependency.
+ignored = ["sqlparser"]
diff --git a/src/file-engine/Cargo.toml b/src/file-engine/Cargo.toml
index ec904addf1e..dda5fb4221c 100644
--- a/src/file-engine/Cargo.toml
+++ b/src/file-engine/Cargo.toml
@@ -18,7 +18,6 @@ common-catalog.workspace = true
common-datasource.workspace = true
common-error.workspace = true
common-macro.workspace = true
-common-query.workspace = true
common-recordbatch.workspace = true
common-telemetry.workspace = true
common-test-util = { workspace = true, optional = true }
diff --git a/src/flow/Cargo.toml b/src/flow/Cargo.toml
index 4586e612693..d0ac5eee9ac 100644
--- a/src/flow/Cargo.toml
+++ b/src/flow/Cargo.toml
@@ -11,18 +11,13 @@ workspace = true
api.workspace = true
arrow.workspace = true
arrow-schema.workspace = true
-async-recursion = "1.0"
async-trait.workspace = true
-bytes.workspace = true
catalog.workspace = true
chrono.workspace = true
client.workspace = true
common-base.workspace = true
common-config.workspace = true
-common-datasource.workspace = true
-common-decimal.workspace = true
common-error.workspace = true
-common-frontend.workspace = true
common-function.workspace = true
common-grpc.workspace = true
common-macro.workspace = true
@@ -35,41 +30,30 @@ common-stat.workspace = true
common-telemetry.workspace = true
common-time.workspace = true
common-version.workspace = true
-config.workspace = true
datafusion.workspace = true
datafusion-common.workspace = true
datafusion-expr.workspace = true
datafusion-physical-expr.workspace = true
-datafusion-substrait.workspace = true
datatypes.workspace = true
-enum_dispatch = "0.3"
futures.workspace = true
get-size2 = "0.1.2"
greptime-proto.workspace = true
hostname.workspace = true
-http.workspace = true
humantime-serde.workspace = true
itertools.workspace = true
lazy_static.workspace = true
meta-client.workspace = true
-nom = "7.1.3"
-num-traits = "0.2"
-num_cpus.workspace = true
operator.workspace = true
-partition.workspace = true
prometheus.workspace = true
-prost.workspace = true
query.workspace = true
rand.workspace = true
serde.workspace = true
serde_json.workspace = true
servers.workspace = true
session.workspace = true
-smallvec.workspace = true
snafu.workspace = true
sql.workspace = true
store-api.workspace = true
-strum.workspace = true
substrait.workspace = true
table.workspace = true
tokio.workspace = true
@@ -81,7 +65,6 @@ catalog = { workspace = true, features = ["testing"] }
common-catalog.workspace = true
common-meta = { workspace = true, features = ["testing"] }
pretty_assertions.workspace = true
-prost.workspace = true
query.workspace = true
session.workspace = true
table.workspace = true
diff --git a/src/flow/src/batching_mode/batching_execution.rs b/src/flow/src/batching_mode/batching_execution.rs
index 0a699a227be..298522b8137 100644
--- a/src/flow/src/batching_mode/batching_execution.rs
+++ b/src/flow/src/batching_mode/batching_execution.rs
@@ -16,6 +16,7 @@
use std::sync::Arc;
+use datafusion_expr::LogicalPlan;
use query::QueryEngineRef;
use table::TableRef;
@@ -37,6 +38,11 @@ pub trait BatchingExecution: Send + Sync + 'static {
max_window_cnt: Option,
) -> ExecuteOnceOutcome;
+ /// Rewrite the completed query plan after incremental merging and before execution.
+ fn rewrite_plan(&self, _task: &BatchingTask, plan: LogicalPlan) -> Result {
+ Ok(plan)
+ }
+
/// Retire this execution instance, rejecting new work and requesting that any
/// retained local work stop. This is not an acknowledgement of remote quiescence.
fn stop(&self) {}
diff --git a/src/flow/src/batching_mode/task.rs b/src/flow/src/batching_mode/task.rs
index c0ee6d1c1b5..346d39d1d46 100644
--- a/src/flow/src/batching_mode/task.rs
+++ b/src/flow/src/batching_mode/task.rs
@@ -828,6 +828,10 @@ impl BatchingTask {
return Ok(None);
}
let plan = incremental_plan.unwrap_or_else(|| plan.clone());
+ let plan = match &self.execution {
+ Some(execution) => execution.rewrite_plan(self, plan)?,
+ None => plan,
+ };
let extensions = self
.build_flow_query_extensions(incremental_safe, coverage.is_incremental_delta())
diff --git a/src/flow/src/batching_mode/task/test.rs b/src/flow/src/batching_mode/task/test.rs
index 7a56c70823d..7add59664a5 100644
--- a/src/flow/src/batching_mode/task/test.rs
+++ b/src/flow/src/batching_mode/task/test.rs
@@ -143,6 +143,99 @@ impl crate::BatchingExecution for RetainingExecution {
}
}
+/// Records the completed plan handed to it by the task.
+#[derive(Default)]
+struct PlanRecordingExecution {
+ plans: std::sync::Mutex>,
+}
+
+#[async_trait::async_trait]
+impl crate::BatchingExecution for PlanRecordingExecution {
+ async fn execute_once(
+ self: Arc,
+ _guard: BatchingExecutionGuard,
+ _task: &BatchingTask,
+ _engine: &QueryEngineRef,
+ _frontend: &Arc,
+ _max_window_cnt: Option,
+ ) -> ExecuteOnceOutcome {
+ ExecuteOnceOutcome {
+ new_query: None,
+ result: Ok(None),
+ }
+ }
+
+ fn rewrite_plan(&self, _task: &BatchingTask, plan: LogicalPlan) -> crate::Result {
+ self.plans.lock().unwrap().push(plan.clone());
+ Ok(plan)
+ }
+}
+
+#[tokio::test]
+async fn test_execution_hook_receives_the_completed_incremental_plan() {
+ let sink_table = "hook_sink";
+ let query = "SELECT max(number) AS number, ts FROM numbers_with_ts GROUP BY ts";
+ let TestTaskParts {
+ task, query_engine, ..
+ } = new_test_task_engine_and_plan_with_query(query, sink_table).await;
+ register_auto_created_aggregate_sink(&query_engine, sink_table);
+
+ let ctx = task.state.read().unwrap().query_ctx.clone();
+ let plan = sql_to_df_plan(ctx, query_engine.clone(), query, true)
+ .await
+ .unwrap();
+ let (sink, _) = get_table_info_df_schema(
+ query_engine.engine_state().catalog_manager().clone(),
+ [
+ "greptime".to_string(),
+ "public".to_string(),
+ sink_table.to_string(),
+ ],
+ )
+ .await
+ .unwrap();
+ let table_provider = Arc::new(DfTableProviderAdapter::new(sink));
+ let table_source = Arc::new(DefaultTableSource::new(table_provider));
+ let dml_plan = LogicalPlan::Dml(DmlStatement::new(
+ datafusion_common::TableReference::bare("test"),
+ table_source,
+ WriteOp::Insert(datafusion_expr::dml::InsertOp::Append),
+ Arc::new(plan),
+ ));
+
+ task.state
+ .write()
+ .unwrap()
+ .advance_checkpoints(HashMap::from([(1_u64, 10_u64)]));
+
+ let execution = Arc::new(PlanRecordingExecution::default());
+ let task = task.with_execution(Some(execution.clone()));
+ let (frontend_client, _handler) =
+ FrontendClient::from_empty_grpc_handler(QueryOptions::default());
+ let frontend = Arc::new(frontend_client);
+ let dirty_restore = DirtyRestore::Unscoped(dirty_range(10, 15));
+
+ let _ = task
+ .execute_plan_unlocked(
+ &query_engine,
+ &frontend,
+ &dml_plan,
+ &dirty_restore,
+ &QueryCoverage::IncrementalDelta,
+ )
+ .await;
+
+ let recorded = execution.plans.lock().unwrap();
+ let plan = recorded
+ .first()
+ .expect("the execution hook must see the plan that is dispatched");
+ let plan_text = plan.to_string();
+ assert!(
+ plan_text.contains("Left Join"),
+ "the hook must receive the completed delta-sink merge plan, got:\n{plan_text}"
+ );
+}
+
#[tokio::test]
async fn test_execution_delegate_dispatch_is_serialized() {
let TestTaskParts {
diff --git a/src/flow/src/lib.rs b/src/flow/src/lib.rs
index e9ef90e09d6..36dc7c21818 100644
--- a/src/flow/src/lib.rs
+++ b/src/flow/src/lib.rs
@@ -48,7 +48,10 @@ pub use batching_mode::task::{
QueryCoverage, TaskArgs,
};
pub use batching_mode::time_window::{TimeWindowExpr, find_time_window_expr};
-pub use batching_mode::utils::sql_to_df_plan;
+pub use batching_mode::utils::{
+ IncrementalAggregateAnalysis, IncrementalAggregateMergeColumn, IncrementalAggregateMergeOp,
+ analyze_incremental_aggregate_plan, sql_to_df_plan,
+};
pub(crate) use engine::{CreateFlowArgs, FlowId, TableName};
pub use error::{Error, Result};
pub use server::{FlownodeBuilder, FlownodeInstance, FlownodeServer, FlownodeServiceBuilder};
diff --git a/src/frontend/Cargo.toml b/src/frontend/Cargo.toml
index b59e1e83a4a..1aa057d0fcd 100644
--- a/src/frontend/Cargo.toml
+++ b/src/frontend/Cargo.toml
@@ -19,7 +19,6 @@ workspace = true
[dependencies]
api.workspace = true
-arc-swap = "1.0"
async-stream.workspace = true
async-trait.workspace = true
auth.workspace = true
@@ -42,7 +41,6 @@ common-macro.workspace = true
common-memory-manager.workspace = true
common-meta.workspace = true
common-options.workspace = true
-common-procedure.workspace = true
common-query.workspace = true
common-recordbatch.workspace = true
common-runtime.workspace = true
@@ -56,12 +54,10 @@ datafusion-expr.workspace = true
datatypes.workspace = true
futures.workspace = true
hostname.workspace = true
-humantime.workspace = true
humantime-serde.workspace = true
lazy_static.workspace = true
log-query.workspace = true
meta-client.workspace = true
-num_cpus.workspace = true
opentelemetry-proto.workspace = true
opentelemetry-semantic-conventions = { version = "0.31", features = ["semconv_experimental"] }
operator.workspace = true
@@ -72,7 +68,6 @@ prometheus.workspace = true
promql-parser.workspace = true
prost.workspace = true
query.workspace = true
-rand.workspace = true
serde.workspace = true
serde_json.workspace = true
servers.workspace = true
@@ -100,4 +95,3 @@ serde_json.workspace = true
strfmt = "0.2"
temp-env.workspace = true
tower.workspace = true
-uuid.workspace = true
diff --git a/src/frontend/src/frontend.rs b/src/frontend/src/frontend.rs
index 72bf322b1f1..51e291c7906 100644
--- a/src/frontend/src/frontend.rs
+++ b/src/frontend/src/frontend.rs
@@ -206,6 +206,7 @@ mod tests {
use futures::Stream;
use meta_client::MetaClientRef;
use meta_client::client::MetaClientBuilder;
+ use servers::batcher::BatchingProtocol;
use servers::grpc::{FlightCompression, GRPC_SERVER};
use servers::http::HTTP_SERVER;
use servers::http::result::greptime_result_v1::GreptimedbV1Response;
@@ -249,10 +250,7 @@ mod tests {
.logical_table
.unwrap()
.protocols,
- vec![
- servers::http::BatchingProtocol::Prom,
- servers::http::BatchingProtocol::Otlp
- ]
+ vec![BatchingProtocol::Prom, BatchingProtocol::Otlp]
);
},
);
@@ -270,10 +268,7 @@ mod tests {
FrontendOptions::load_layered_options(None, "FRONTEND_BATCHER_TEST").unwrap();
assert_eq!(
options.pending_rows_batcher.table.protocols,
- vec![
- servers::http::BatchingProtocol::Influxdb,
- servers::http::BatchingProtocol::HttpSql
- ]
+ vec![BatchingProtocol::Influxdb, BatchingProtocol::HttpSql]
);
},
);
diff --git a/src/frontend/src/instance/builder.rs b/src/frontend/src/instance/builder.rs
index d9b86ac5a05..d261daed48e 100644
--- a/src/frontend/src/instance/builder.rs
+++ b/src/frontend/src/instance/builder.rs
@@ -50,8 +50,8 @@ use partition::manager::PartitionRuleManager;
use pipeline::pipeline_operator::PipelineOperator;
use query::QueryEngineFactory;
use query::region_query::RegionQueryHandlerFactoryRef;
+use servers::batcher::BatchingProtocol;
use servers::batcher::table::TablePendingRowsBatcher;
-use servers::http::BatchingProtocol;
use snafu::{OptionExt, ResultExt};
use crate::error::{self, DataFusionSnafu, ExternalSnafu, Result};
diff --git a/src/frontend/src/instance/logical_batcher.rs b/src/frontend/src/instance/logical_batcher.rs
index 28ed19b3e14..e68dcf644d4 100644
--- a/src/frontend/src/instance/logical_batcher.rs
+++ b/src/frontend/src/instance/logical_batcher.rs
@@ -16,9 +16,9 @@ use std::sync::{Arc, Weak};
use api::v1::ColumnSchema;
use async_trait::async_trait;
+use servers::batcher::BatchingProtocol;
use servers::batcher::logical_table::{LogicalTablePendingRowsBatcher, PendingRowsSchemaAlterer};
use servers::error::{BatcherChannelClosedSnafu, Result};
-use servers::http::BatchingProtocol;
use session::context::QueryContextRef;
use snafu::OptionExt;
diff --git a/src/frontend/src/server.rs b/src/frontend/src/server.rs
index 0a6612ba31c..fb1ab2f2603 100644
--- a/src/frontend/src/server.rs
+++ b/src/frontend/src/server.rs
@@ -24,7 +24,7 @@ use common_base::Plugins;
use common_config::Configurable;
use common_telemetry::{info, warn};
use meta_client::MetaClientOptions;
-use servers::batcher::pending_rows_batch_sync_enabled;
+use servers::batcher::{BatchingProtocol, pending_rows_batch_sync_enabled};
use servers::error::Error as ServerError;
use servers::grpc::builder::GrpcServerBuilder;
use servers::grpc::flight::FlightCraftRef;
@@ -34,7 +34,7 @@ use servers::grpc::{GrpcOptions, GrpcServer};
use servers::http::event::LogValidatorRef;
use servers::http::result::error_result::ErrorResponse;
use servers::http::utils::router::RouterConfigurator;
-use servers::http::{BatchingProtocol, HttpOptions, HttpServer, HttpServerBuilder};
+use servers::http::{HttpOptions, HttpServer, HttpServerBuilder};
use servers::interceptor::LogIngestInterceptorRef;
use servers::metrics_handler::MetricsHandler;
use servers::mysql::server::{MysqlServer, MysqlSpawnConfig, MysqlSpawnRef};
@@ -369,6 +369,15 @@ where
}
}
+ let table_batcher = opts.table_batcher_options();
+ let batching_enabled = table_batcher.pending_rows_batching_enabled();
+ let mysql_batching =
+ batching_enabled && table_batcher.protocols.contains(&BatchingProtocol::Mysql);
+ let postgres_batching = batching_enabled
+ && table_batcher
+ .protocols
+ .contains(&BatchingProtocol::Postgres);
+
if opts.mysql.enable {
// Init MySQL server
let opts = &opts.mysql;
@@ -384,13 +393,16 @@ where
let mysql_server = MysqlServer::create_server(
common_runtime::global_runtime(),
Arc::new(MysqlSpawnRef::new(instance.clone(), user_provider.clone())),
- Arc::new(MysqlSpawnConfig::new(
- opts.tls.should_force_tls(),
- tls_server_config,
- opts.keep_alive.as_secs(),
- opts.reject_no_database.unwrap_or(false),
- opts.prepared_stmt_cache_size,
- )),
+ Arc::new(
+ MysqlSpawnConfig::new(
+ opts.tls.should_force_tls(),
+ tls_server_config,
+ opts.keep_alive.as_secs(),
+ opts.reject_no_database.unwrap_or(false),
+ opts.prepared_stmt_cache_size,
+ )
+ .with_batching_enabled(mysql_batching),
+ ),
Some(instance.process_manager().clone()),
);
handlers.insert((mysql_server, mysql_addr));
@@ -407,15 +419,18 @@ where
maybe_watch_server_tls_config(tls_server_config.clone()).context(StartServerSnafu)?;
- let pg_server = Box::new(PostgresServer::new(
- instance.clone(),
- opts.tls.should_force_tls(),
- tls_server_config,
- opts.keep_alive.as_secs(),
- common_runtime::global_runtime(),
- user_provider.clone(),
- Some(self.instance.process_manager().clone()),
- )) as Box;
+ let pg_server = Box::new(
+ PostgresServer::new(
+ instance.clone(),
+ opts.tls.should_force_tls(),
+ tls_server_config,
+ opts.keep_alive.as_secs(),
+ common_runtime::global_runtime(),
+ user_provider.clone(),
+ Some(self.instance.process_manager().clone()),
+ )
+ .with_batching_enabled(postgres_batching),
+ ) as Box;
handlers.insert((pg_server, pg_addr));
}
@@ -452,8 +467,11 @@ fn effective_http_options_with_sync(opts: &FrontendOptions, batch_sync: bool) ->
let common_enabled = batch_sync
&& shared.pending_rows_batching_enabled()
&& shared.protocols.iter().any(|protocol| {
- *protocol != BatchingProtocol::Prom
- || (prom_store.enable && !prom_store.with_metric_engine)
+ !matches!(
+ protocol,
+ BatchingProtocol::Mysql | BatchingProtocol::Postgres
+ ) && (*protocol != BatchingProtocol::Prom
+ || (prom_store.enable && !prom_store.with_metric_engine))
});
let common_interval = common_enabled.then_some(shared.pending_rows_flush_interval);
let prom_interval = (prom_store.pending_rows_batching_enabled() && batch_sync)
@@ -655,6 +673,10 @@ mod tests {
expected_secs,
) in [
(vec![BatchingProtocol::Prom], false, false, 5, 2, 1, 1),
+ (vec![BatchingProtocol::Mysql], false, false, 5, 0, 1, 1),
+ (vec![BatchingProtocol::Postgres], false, false, 5, 0, 1, 1),
+ (vec![BatchingProtocol::Mysql], false, true, 5, 0, 1, 1),
+ (vec![BatchingProtocol::Postgres], false, true, 5, 0, 1, 1),
(vec![BatchingProtocol::Prom], true, false, 5, 2, 1, 1),
(vec![BatchingProtocol::Prom], true, true, 5, 2, 1, 6),
(vec![BatchingProtocol::Influxdb], true, false, 5, 2, 1, 1),
diff --git a/src/frontend/src/service_config/pending_rows_batcher.rs b/src/frontend/src/service_config/pending_rows_batcher.rs
index 61572015d7c..b671ee9a272 100644
--- a/src/frontend/src/service_config/pending_rows_batcher.rs
+++ b/src/frontend/src/service_config/pending_rows_batcher.rs
@@ -17,7 +17,7 @@ use std::time::Duration;
use common_batcher::flush_policy::timing::TimingFlushPolicy;
use serde::{Deserialize, Serialize};
-use servers::http::BatchingProtocol;
+use servers::batcher::BatchingProtocol;
use tokio::sync::Semaphore;
use crate::frontend::FrontendOptions;
@@ -38,11 +38,11 @@ pub struct PendingRowsBatcherOptions {
pub logical_table: Option,
}
-/// Write batching controls shared by HTTP ingestion protocols.
+/// Write batching controls shared by ingestion protocols.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(default)]
pub struct BatcherOptions {
- /// HTTP write protocols sharing this batcher; empty disables all entrances.
+ /// Write protocols sharing this batcher; empty disables all entrances.
pub protocols: Vec,
/// Time from the first pending submission to a timed flush. Zero disables batching.
#[serde(with = "humantime_serde")]
@@ -238,6 +238,8 @@ mod tests {
"opentsdb",
"elasticsearch",
"splunk",
+ "mysql",
+ "postgres",
] {
assert!(
toml::from_str::(&format!(
@@ -251,9 +253,11 @@ mod tests {
#[test]
fn test_protocols() {
let options: BatcherOptions = toml::from_str(
- "protocols = ['influxdb', 'opentsdb', 'otlp', 'logs', 'loki', 'splunk', 'elasticsearch', 'http_sql', 'prom']",
+ "protocols = ['influxdb', 'opentsdb', 'otlp', 'logs', 'loki', 'splunk', 'elasticsearch', 'http_sql', 'prom', 'mysql', 'postgres']",
).unwrap();
- assert_eq!(options.protocols.len(), 9);
+ assert_eq!(options.protocols.len(), 11);
+ assert!(options.protocols.contains(&BatchingProtocol::Mysql));
+ assert!(options.protocols.contains(&BatchingProtocol::Postgres));
assert!(options.protocols.contains(&BatchingProtocol::HttpSql));
assert!(BatcherOptions::default().protocols.is_empty());
for invalid in ["sql", "jaeger", "unknown"] {
@@ -289,6 +293,36 @@ mod tests {
);
}
+ #[test]
+ fn test_worker_capacity_override() {
+ let options: FrontendOptions = toml::from_str(
+ r#"
+[pending_rows_batcher]
+worker_channel_capacity = 12345
+[pending_rows_batcher.logical_table]
+worker_channel_capacity = 12345
+[prom_store]
+enable = true
+with_metric_engine = true
+worker_channel_capacity = 12345
+"#,
+ )
+ .unwrap();
+ assert_eq!(
+ options.pending_rows_batcher.table.worker_channel_capacity,
+ 12_345
+ );
+ assert_eq!(
+ options
+ .pending_rows_batcher
+ .logical_table
+ .unwrap()
+ .worker_channel_capacity,
+ 12_345
+ );
+ assert_eq!(options.prom_store.worker_channel_capacity, 12_345);
+ }
+
#[test]
fn test_partial_options_and_zero_controls() {
let options: BatcherOptions =
diff --git a/src/frontend/src/service_config/prom_store.rs b/src/frontend/src/service_config/prom_store.rs
index 2fdd05043db..3d00f48cdda 100644
--- a/src/frontend/src/service_config/prom_store.rs
+++ b/src/frontend/src/service_config/prom_store.rs
@@ -108,7 +108,6 @@ mod tests {
use crate::service_config::prom_store::{
default_flow_notification_queue_capacity, default_max_batch_rows,
default_max_concurrent_flushes, default_max_inflight_requests,
- default_worker_channel_capacity,
};
#[test]
@@ -160,10 +159,7 @@ mod tests {
default.max_concurrent_flushes,
default_max_concurrent_flushes()
);
- assert_eq!(
- default.worker_channel_capacity,
- default_worker_channel_capacity()
- );
+ assert_eq!(default.worker_channel_capacity, 65_536);
assert_eq!(
default.max_inflight_requests,
default_max_inflight_requests()
diff --git a/src/log-store/Cargo.toml b/src/log-store/Cargo.toml
index 15560b06b2c..5d068ca344d 100644
--- a/src/log-store/Cargo.toml
+++ b/src/log-store/Cargo.toml
@@ -31,7 +31,6 @@ delta-encoding = "0.4"
derive_builder.workspace = true
futures.workspace = true
futures-util.workspace = true
-itertools.workspace = true
lazy_static.workspace = true
object-store.workspace = true
pin-project.workspace = true
@@ -49,8 +48,13 @@ tokio.workspace = true
common-meta = { workspace = true, features = ["testing"] }
common-test-util.workspace = true
common-wal = { workspace = true, features = ["testing"] }
-itertools.workspace = true
object-store = { workspace = true, features = ["testing"] }
rand.workspace = true
rskafka = { workspace = true, features = ["unstable-fuzzing"] }
uuid.workspace = true
+
+[package.metadata.cargo-shear]
+# `protobuf` is required by the code generated by `build.rs`: the generated
+# protobuf bindings (included via `include!`) reference `::protobuf` paths,
+# but no source file does.
+ignored = ["protobuf"]
diff --git a/src/meta-client/Cargo.toml b/src/meta-client/Cargo.toml
index 984f434cc51..9863b32bbbf 100644
--- a/src/meta-client/Cargo.toml
+++ b/src/meta-client/Cargo.toml
@@ -19,7 +19,6 @@ common-options.workspace = true
common-telemetry.workspace = true
common-time.workspace = true
futures.workspace = true
-futures-util.workspace = true
humantime-serde.workspace = true
rand.workspace = true
serde.workspace = true
@@ -31,10 +30,7 @@ tonic.workspace = true
[dev-dependencies]
common-event-recorder.workspace = true
-datatypes.workspace = true
futures.workspace = true
meta-srv = { workspace = true, features = ["mock"] }
-session.workspace = true
-tower.workspace = true
tracing = "0.1"
tracing-subscriber.workspace = true
diff --git a/src/meta-srv/Cargo.toml b/src/meta-srv/Cargo.toml
index e6585505db4..6baca0934e1 100644
--- a/src/meta-srv/Cargo.toml
+++ b/src/meta-srv/Cargo.toml
@@ -14,16 +14,11 @@ enterprise = ["common-meta/enterprise"]
[lints]
workspace = true
-[target.'cfg(not(target_os = "android"))'.dependencies]
-local-ip-address.workspace = true
-
[dependencies]
api.workspace = true
async-trait.workspace = true
axum.workspace = true
-axum-extra.workspace = true
axum-macros.workspace = true
-bytes.workspace = true
chrono.workspace = true
clap.workspace = true
client.workspace = true
@@ -55,18 +50,14 @@ etcd-client.workspace = true
futures.workspace = true
h2 = "0.3"
hostname.workspace = true
-http-body-util = "0.1"
-humantime.workspace = true
humantime-serde.workspace = true
hyper-util = { workspace = true, features = ["tokio"] }
itertools.workspace = true
lazy_static.workspace = true
once_cell.workspace = true
ordered-float.workspace = true
-parking_lot.workspace = true
partition.workspace = true
prometheus.workspace = true
-prost.workspace = true
rand.workspace = true
regex.workspace = true
rskafka.workspace = true
@@ -87,7 +78,6 @@ tokio-stream = { workspace = true, features = ["net"] }
toml.workspace = true
tonic.workspace = true
tower.workspace = true
-tower-http.workspace = true
tracing.workspace = true
typetag.workspace = true
url = "2.3"
diff --git a/src/metric-engine/Cargo.toml b/src/metric-engine/Cargo.toml
index 0005a4dc543..92000549832 100644
--- a/src/metric-engine/Cargo.toml
+++ b/src/metric-engine/Cargo.toml
@@ -14,7 +14,7 @@ async-stream.workspace = true
async-trait.workspace = true
base64.workspace = true
bytes.workspace = true
-fxhash = "0.2"
+chrono.workspace = true
common-base.workspace = true
common-error.workspace = true
common-grpc.workspace = true
@@ -24,10 +24,10 @@ common-recordbatch.workspace = true
common-runtime.workspace = true
common-telemetry.workspace = true
common-time.workspace = true
-chrono.workspace = true
datafusion.workspace = true
datatypes.workspace = true
futures-util.workspace = true
+fxhash = "0.2"
humantime-serde.workspace = true
itertools.workspace = true
lazy_static = "1.4"
@@ -45,18 +45,14 @@ tokio.workspace = true
tracing.workspace = true
[dev-dependencies]
+common-function = { workspace = true }
common-meta = { workspace = true, features = ["testing"] }
-common-test-util.workspace = true
-mito2 = { workspace = true, features = ["test"] }
common-wal = { workspace = true }
criterion = { workspace = true, features = ["async", "async_tokio"] }
+mito2 = { workspace = true, features = ["test"] }
mur3 = "0.1"
-common-function = { workspace = true }
partition = { workspace = true }
[[bench]]
name = "bench_tsid_generator"
harness = false
-
-[package.metadata.cargo-udeps.ignore]
-normal = ["aquamarine"]
diff --git a/src/mito2/Cargo.toml b/src/mito2/Cargo.toml
index 2be72af57cf..82a2884549a 100644
--- a/src/mito2/Cargo.toml
+++ b/src/mito2/Cargo.toml
@@ -20,7 +20,6 @@ api.workspace = true
aquamarine.workspace = true
arrow-schema.workspace = true
async-channel = "1.9"
-common-stat.workspace = true
async-stream.workspace = true
async-trait.workspace = true
base64.workspace = true
@@ -28,58 +27,55 @@ bytemuck.workspace = true
bytes.workspace = true
chrono.workspace = true
common-base.workspace = true
-common-config.workspace = true
common-datasource.workspace = true
-common-decimal.workspace = true
common-error.workspace = true
-common-grpc.workspace = true
common-function.workspace = true
+common-grpc.workspace = true
common-macro.workspace = true
common-memory-manager.workspace = true
common-meta.workspace = true
common-query.workspace = true
common-recordbatch.workspace = true
common-runtime.workspace = true
+common-stat.workspace = true
common-telemetry.workspace = true
common-test-util = { workspace = true, optional = true }
common-time.workspace = true
common-wal.workspace = true
crc32fast = "1"
crossbeam-utils.workspace = true
+dashmap.workspace = true
datafusion.workspace = true
datafusion-common.workspace = true
datafusion-expr.workspace = true
datatypes.workspace = true
-dashmap.workspace = true
derive_more.workspace = true
dotenv.workspace = true
either.workspace = true
futures.workspace = true
+greptime-proto.workspace = true
humantime.workspace = true
humantime-serde.workspace = true
index.workspace = true
itertools.workspace = true
-greptime-proto.workspace = true
lazy_static = "1.4"
log-store = { workspace = true }
mito-codec.workspace = true
moka = { workspace = true, features = ["sync", "future"] }
object-store = { workspace = true, features = ["testing"] }
parquet = { workspace = true, features = ["async", "variant_experimental"] }
-paste.workspace = true
+partition.workspace = true
pin-project.workspace = true
prometheus.workspace = true
prost.workspace = true
-partition.workspace = true
puffin.workspace = true
rand.workspace = true
rayon = "1.10"
-roaring = { version = "0.10", optional = true }
regex.workspace = true
+roaring = { version = "0.10", optional = true }
rskafka = { workspace = true, optional = true }
rstest = { workspace = true, optional = true }
rstest_reuse = { workspace = true, optional = true }
-scopeguard = "1.2"
serde.workspace = true
serde_json.workspace = true
serde_with.workspace = true
@@ -108,7 +104,6 @@ object-store = { workspace = true, features = ["services-memory"] }
rskafka.workspace = true
rstest.workspace = true
rstest_reuse.workspace = true
-session.workspace = true
toml.workspace = true
[[bench]]
@@ -159,6 +154,3 @@ required-features = ["testing"]
name = "bench_dense_index_update"
harness = false
required-features = ["testing"]
-
-[package.metadata.cargo-udeps.ignore]
-normal = ["aquamarine"]
diff --git a/src/mito2/src/cache/write_cache.rs b/src/mito2/src/cache/write_cache.rs
index 0e96c2f16a2..87ab4f0081a 100644
--- a/src/mito2/src/cache/write_cache.rs
+++ b/src/mito2/src/cache/write_cache.rs
@@ -790,7 +790,8 @@ mod tests {
async fn test_read_metadata_from_write_cache() {
common_telemetry::init_default_ut_logging();
let mut env = TestEnv::new().await;
- let data_home = env.data_home().display().to_string();
+ // Object keys are relative to the mock store root, not native paths.
+ let data_home = "write-cache-test".to_string();
let mock_store = env.init_object_store_manager();
let local_dir = create_temp_dir("");
diff --git a/src/object-store/Cargo.toml b/src/object-store/Cargo.toml
index 412f3563298..4083271a967 100644
--- a/src/object-store/Cargo.toml
+++ b/src/object-store/Cargo.toml
@@ -10,23 +10,18 @@ workspace = true
[features]
mysql-object-store = ["opendal/services-mysql"]
services-memory = ["opendal/services-memory"]
-testing = ["derive_builder"]
+testing = ["derive_builder", "uuid"]
[dependencies]
-async-trait.workspace = true
bytes.workspace = true
cap-std.workspace = true
-chrono.workspace = true
common-base.workspace = true
common-error.workspace = true
common-macro.workspace = true
common-runtime.workspace = true
common-telemetry.workspace = true
-datafusion_object_store.workspace = true
derive_builder = { workspace = true, optional = true }
-futures.workspace = true
humantime-serde.workspace = true
-lazy_static.workspace = true
opendal = { version = "0.59.2", features = [
"layers-tracing",
"layers-prometheus",
@@ -43,14 +38,16 @@ reqwest.workspace = true
serde.workspace = true
snafu.workspace = true
tokio.workspace = true
-uuid.workspace = true
+uuid = { workspace = true, optional = true }
[dev-dependencies]
anyhow = "1.0"
common-telemetry.workspace = true
common-test-util.workspace = true
+datafusion_object_store.workspace = true
+futures.workspace = true
+object-store = { workspace = true, features = ["testing"] }
object_store_opendal.workspace = true
-rand.workspace = true
tempfile.workspace = true
tokio.workspace = true
toml.workspace = true
diff --git a/src/object-store/src/lib.rs b/src/object-store/src/lib.rs
index 05c344485c6..2de2b570082 100644
--- a/src/object-store/src/lib.rs
+++ b/src/object-store/src/lib.rs
@@ -26,6 +26,7 @@ pub mod layers;
pub mod manager;
mod metrics;
pub mod secure_fs;
+#[cfg(feature = "testing")]
pub mod test_util;
pub mod util;
diff --git a/src/operator/Cargo.toml b/src/operator/Cargo.toml
index ea1bc990f74..72f953e0f89 100644
--- a/src/operator/Cargo.toml
+++ b/src/operator/Cargo.toml
@@ -22,7 +22,6 @@ workspace = true
ahash.workspace = true
api.workspace = true
arrow.workspace = true
-arrow-ipc.workspace = true
async-stream.workspace = true
async-trait.workspace = true
bytes.workspace = true
@@ -66,7 +65,6 @@ object-store.workspace = true
partition.workspace = true
percent-encoding.workspace = true
prometheus.workspace = true
-prost.workspace = true
prost-types = { workspace = true, optional = true }
query.workspace = true
regex.workspace = true
diff --git a/src/partition/Cargo.toml b/src/partition/Cargo.toml
index a8e3a8ae11a..6773cc5878b 100644
--- a/src/partition/Cargo.toml
+++ b/src/partition/Cargo.toml
@@ -9,18 +9,14 @@ workspace = true
[dependencies]
api.workspace = true
-async-trait.workspace = true
common-base.workspace = true
common-error.workspace = true
common-macro.workspace = true
common-meta.workspace = true
-common-query.workspace = true
-common-telemetry.workspace = true
datafusion-common.workspace = true
datafusion-expr.workspace = true
datafusion-physical-expr.workspace = true
datatypes.workspace = true
-itertools.workspace = true
moka = { workspace = true, features = ["future"] }
serde.workspace = true
serde_json.workspace = true
diff --git a/src/pipeline/Cargo.toml b/src/pipeline/Cargo.toml
index 620ac36ac77..9a355f4c8dc 100644
--- a/src/pipeline/Cargo.toml
+++ b/src/pipeline/Cargo.toml
@@ -14,28 +14,21 @@ ahash.workspace = true
api.workspace = true
arrow.workspace = true
arrow-schema.workspace = true
-async-trait.workspace = true
catalog.workspace = true
chrono.workspace = true
chrono-tz.workspace = true
common-catalog.workspace = true
common-error.workspace = true
-common-function.workspace = true
common-macro.workspace = true
common-meta.workspace = true
common-query.workspace = true
common-recordbatch.workspace = true
-common-runtime.workspace = true
common-telemetry.workspace = true
common-time.workspace = true
-crossbeam-utils.workspace = true
csv = "1.3"
-dashmap.workspace = true
datafusion.workspace = true
datafusion-common.workspace = true
datafusion-expr.workspace = true
-datafusion-functions.workspace = true
-datafusion-physical-expr.workspace = true
datatypes.workspace = true
dyn-fmt = "0.4"
enum_dispatch = "0.3"
@@ -50,7 +43,6 @@ moka = { workspace = true, features = ["future"] }
once_cell.workspace = true
operator.workspace = true
ordered-float.workspace = true
-paste.workspace = true
prometheus.workspace = true
query.workspace = true
regex.workspace = true
@@ -68,7 +60,6 @@ yaml-rust = "0.4"
[dev-dependencies]
catalog = { workspace = true, features = ["testing"] }
criterion = { workspace = true, features = ["html_reports"] }
-rayon = "1.0"
session = { workspace = true, features = ["testing"] }
[[bench]]
diff --git a/src/query/Cargo.toml b/src/query/Cargo.toml
index 480c95a5adb..e81d93b7781 100644
--- a/src/query/Cargo.toml
+++ b/src/query/Cargo.toml
@@ -16,7 +16,6 @@ vector_index = []
[dependencies]
ahash.workspace = true
api.workspace = true
-arc-swap = "1.0"
arrow.workspace = true
arrow-schema.workspace = true
async-recursion = "1.0"
@@ -50,11 +49,9 @@ datafusion-physical-expr.workspace = true
datafusion-proto.workspace = true
datafusion-sql.workspace = true
datatypes.workspace = true
-either.workspace = true
futures.workspace = true
futures-util.workspace = true
greptime-proto.workspace = true
-humantime.workspace = true
itertools.workspace = true
jsonb.workspace = true
lazy_static.workspace = true
@@ -63,7 +60,6 @@ meter-core.workspace = true
meter-macros.workspace = true
object-store.workspace = true
once_cell.workspace = true
-parking_lot.workspace = true
partition.workspace = true
prometheus.workspace = true
promql.workspace = true
@@ -93,13 +89,10 @@ common-query = { workspace = true, features = ["testing"] }
fastrand = "2.0"
moka = { workspace = true, features = ["future"] }
nalgebra.workspace = true
-num = "0.4"
-num-traits = "0.2"
paste.workspace = true
pretty_assertions.workspace = true
rand.workspace = true
session = { workspace = true, features = ["testing"] }
store-api.workspace = true
table = { workspace = true, features = ["testing"] }
-tokio-stream.workspace = true
toml.workspace = true
diff --git a/src/query/src/promql/label_values.rs b/src/query/src/promql/label_values.rs
index 44c173477bb..841666758e4 100644
--- a/src/query/src/promql/label_values.rs
+++ b/src/query/src/promql/label_values.rs
@@ -37,15 +37,15 @@ fn millis_since_epoch(time: SystemTime) -> Result {
Ok(duration) => (duration.as_millis(), false),
Err(earlier) => (earlier.duration().as_millis(), true),
};
- let millis = i64::try_from(millis)
- .ok()
- .with_context(|| SystemTimeOutOfRangeSnafu { time })?;
+ let millis =
+ signed_millis(millis, before_epoch).with_context(|| SystemTimeOutOfRangeSnafu { time })?;
+ Ok(Timestamp::new_millisecond(millis))
+}
- Ok(Timestamp::new_millisecond(if before_epoch {
- -millis
- } else {
- millis
- }))
+/// Converts a millisecond magnitude and epoch direction without platform time limits.
+fn signed_millis(millis: u128, before_epoch: bool) -> Option {
+ let millis = i64::try_from(millis).ok()?;
+ Some(if before_epoch { -millis } else { millis })
}
fn build_time_filter(time_index_expr: Expr, start: Timestamp, end: Timestamp) -> Expr {
@@ -139,7 +139,16 @@ mod tests {
#[test]
fn millis_beyond_i64_are_rejected() {
- let time = UNIX_EPOCH + Duration::from_secs(1 << 60);
- assert!(millis_since_epoch(time).is_err());
+ let limit = i64::MAX as u128;
+ assert_eq!(signed_millis(limit, false), Some(i64::MAX));
+ assert_eq!(signed_millis(limit, true), Some(-i64::MAX));
+ assert_eq!(signed_millis(limit + 1, false), None);
+ assert_eq!(signed_millis(limit + 1, true), None);
+
+ // Windows cannot represent this SystemTime. The conversion boundary is
+ // tested above on every platform; exercise the wrapper where possible.
+ if let Some(time) = UNIX_EPOCH.checked_add(Duration::from_millis(i64::MAX as u64 + 1)) {
+ assert!(millis_since_epoch(time).is_err());
+ }
}
}
diff --git a/src/servers/Cargo.toml b/src/servers/Cargo.toml
index 0e2d896db8c..06648254a06 100644
--- a/src/servers/Cargo.toml
+++ b/src/servers/Cargo.toml
@@ -40,7 +40,6 @@ common-batcher.workspace = true
common-catalog.workspace = true
common-decimal.workspace = true
common-error.workspace = true
-common-frontend.workspace = true
common-grpc.workspace = true
common-macro.workspace = true
common-mem-prof = { workspace = true, optional = true }
@@ -58,7 +57,6 @@ common-telemetry.workspace = true
common-time.workspace = true
common-version = { workspace = true, features = ["codec"] }
csv = "1.3"
-dashmap.workspace = true
datafusion.workspace = true
datafusion-common.workspace = true
datafusion-expr.workspace = true
@@ -123,7 +121,6 @@ serde.workspace = true
serde_json.workspace = true
session.workspace = true
simd-json.workspace = true
-simdutf8 = "0.1"
smallvec.workspace = true
snafu.workspace = true
snap = "1"
@@ -169,6 +166,7 @@ rand.workspace = true
serde_json.workspace = true
servers = { workspace = true, features = ["testing"] }
session = { workspace = true, features = ["testing"] }
+simdutf8 = "0.1"
table.workspace = true
tempfile = "3.0.0"
tokio = { workspace = true, features = ["test-util"] }
diff --git a/src/servers/src/batcher.rs b/src/servers/src/batcher.rs
index 6b6ca8f7ee7..3cd9a831aaa 100644
--- a/src/servers/src/batcher.rs
+++ b/src/servers/src/batcher.rs
@@ -22,6 +22,8 @@ pub mod table;
#[cfg(test)]
mod test_util;
+use serde::{Deserialize, Serialize};
+
/// Controls whether batching waits for storage before replying to the client.
const PENDING_ROWS_BATCH_SYNC_ENV: &str = "PENDING_ROWS_BATCH_SYNC";
@@ -39,3 +41,40 @@ pub fn pending_rows_batch_sync_enabled() -> bool {
.and_then(|v| v.parse::().ok())
.unwrap_or(true)
}
+
+/// Ingestion protocols that can opt into pending-row batching.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum BatchingProtocol {
+ Prom,
+ Influxdb,
+ Opentsdb,
+ Otlp,
+ Logs,
+ Loki,
+ Splunk,
+ Elasticsearch,
+ HttpSql,
+ Mysql,
+ Postgres,
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn test_protocol_names_reject_unknown_values() {
+ assert_eq!(
+ serde_json::from_str::("\"prom\"").unwrap(),
+ BatchingProtocol::Prom
+ );
+ for name in ["sql", "unknown"] {
+ assert!(serde_json::from_str::(&format!("\"{name}\"")).is_err());
+ }
+ assert_eq!(
+ serde_json::from_str::("\"http_sql\"").unwrap(),
+ BatchingProtocol::HttpSql
+ );
+ }
+}
diff --git a/src/servers/src/http.rs b/src/servers/src/http.rs
index 4d168111ade..f947ec1621b 100644
--- a/src/servers/src/http.rs
+++ b/src/servers/src/http.rs
@@ -53,6 +53,7 @@ use tower_http::trace::TraceLayer;
use self::authorize::AuthState;
use self::result::table_result::TableResponse;
+use crate::batcher::BatchingProtocol;
use crate::batcher::logical_table::LogicalTablePendingRowsBatcher;
use crate::elasticsearch;
use crate::error::{
@@ -189,22 +190,6 @@ pub(crate) enum HttpServerKind {
Api,
}
-/// HTTP write protocols eligible for the shared pending-row batcher.
-/// Prometheus uses this selector only when metric-engine storage is disabled.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(rename_all = "snake_case")]
-pub enum BatchingProtocol {
- Prom,
- Influxdb,
- Opentsdb,
- Otlp,
- Logs,
- Loki,
- Splunk,
- Elasticsearch,
- HttpSql,
-}
-
#[derive(Default)]
pub struct HttpServer {
router: StdMutex,
@@ -2478,9 +2463,10 @@ mod batching_tests {
use common_query::Output;
use session::context::QueryContextRef;
+ use crate::batcher::BatchingProtocol;
use crate::error::Result as ServerResult;
use crate::http::test_helpers::TestClient;
- use crate::http::{BatchingProtocol, HttpOptions, HttpServerBuilder};
+ use crate::http::{HttpOptions, HttpServerBuilder};
use crate::influxdb::InfluxdbRequest;
use crate::opentsdb::codec::DataPoint;
use crate::query_handler::{InfluxdbLineProtocolHandler, OpentsdbProtocolHandler};
@@ -2523,21 +2509,6 @@ mod batching_tests {
}
}
- #[test]
- fn test_protocol_names_reject_unknown_values() {
- assert_eq!(
- serde_json::from_str::("\"prom\"").unwrap(),
- BatchingProtocol::Prom
- );
- for name in ["sql", "unknown"] {
- assert!(serde_json::from_str::(&format!("\"{name}\"")).is_err());
- }
- assert_eq!(
- serde_json::from_str::("\"http_sql\"").unwrap(),
- BatchingProtocol::HttpSql
- );
- }
-
#[derive(Default)]
struct RecordingWriteHandler {
selections: Mutex>,
diff --git a/src/servers/src/mysql/handler.rs b/src/servers/src/mysql/handler.rs
index 3764591beaa..d0dd1f25279 100644
--- a/src/servers/src/mysql/handler.rs
+++ b/src/servers/src/mysql/handler.rs
@@ -85,6 +85,7 @@ pub struct MysqlInstanceShim {
prepared_stmts_counter: AtomicU32,
process_id: u32,
prepared_stmt_cache_size: usize,
+ batching_enabled: bool,
}
impl MysqlInstanceShim {
@@ -122,9 +123,22 @@ impl MysqlInstanceShim {
prepared_stmts_counter: AtomicU32::new(1),
process_id,
prepared_stmt_cache_size,
+ batching_enabled: false,
}
}
+ /// Enables ordinary-table batching for this connection.
+ pub fn with_batching_enabled(mut self, enabled: bool) -> Self {
+ self.batching_enabled = enabled;
+ self
+ }
+
+ fn new_query_context(&self) -> QueryContextRef {
+ let mut ctx = self.session.new_query_context();
+ Arc::make_mut(&mut ctx).set_batching_enabled(self.batching_enabled);
+ ctx
+ }
+
#[tracing::instrument(skip_all, name = "mysql::do_query")]
async fn do_query(&self, query: &str, query_ctx: QueryContextRef) -> Vec> {
if let Some(output) =
@@ -495,7 +509,7 @@ impl AsyncMysqlShim for MysqlInstanceShi
raw_query: &'a str,
w: StatementMetaWriter<'a, W>,
) -> Result<()> {
- let query_ctx = self.session.new_query_context();
+ let query_ctx = self.new_query_context();
let stmt_id = self.prepared_stmts_counter.fetch_add(1, Ordering::Relaxed);
let stmt_key = uuid::Uuid::from_u128(stmt_id as u128).to_string();
let (params, columns) = match self
@@ -525,7 +539,7 @@ impl AsyncMysqlShim for MysqlInstanceShi
) -> Result<()> {
self.session.clear_warnings();
- let query_ctx = self.session.new_query_context();
+ let query_ctx = self.new_query_context();
let db = query_ctx.get_db_string();
let _timer = crate::metrics::METRIC_MYSQL_QUERY_TIMER
.with_label_values(&[crate::metrics::METRIC_MYSQL_BINQUERY, db.as_str()])
@@ -569,7 +583,7 @@ impl AsyncMysqlShim for MysqlInstanceShi
query: &'a str,
writer: QueryResultWriter<'a, W>,
) -> Result<()> {
- let query_ctx = self.session.new_query_context();
+ let query_ctx = self.new_query_context();
let db = query_ctx.get_db_string();
let _timer = crate::metrics::METRIC_MYSQL_QUERY_TIMER
.with_label_values(&[crate::metrics::METRIC_MYSQL_TEXTQUERY, db.as_str()])
@@ -1039,6 +1053,17 @@ mod tests {
)
}
+ #[test]
+ fn test_batching_context() {
+ for enabled in [false, true] {
+ let shim = create_shim().with_batching_enabled(enabled);
+ let ctx = shim.new_query_context();
+ assert_eq!(ctx.batching_enabled(), enabled);
+ assert!(!ctx.logical_batching_enabled());
+ assert_eq!(ctx.channel(), Channel::Mysql);
+ }
+ }
+
fn statement_with_transformed_placeholders(query: &str) -> Statement {
let mut statements =
ParserContext::create_with_dialect(query, &MySqlDialect {}, ParseOptions::default())
diff --git a/src/servers/src/mysql/server.rs b/src/servers/src/mysql/server.rs
index 0f48cb0d598..811fc9827ec 100644
--- a/src/servers/src/mysql/server.rs
+++ b/src/servers/src/mysql/server.rs
@@ -86,6 +86,7 @@ pub struct MysqlSpawnConfig {
reject_no_database: bool,
// prepared statement cache capacity
prepared_stmt_cache_size: usize,
+ batching_enabled: bool,
}
impl MysqlSpawnConfig {
@@ -102,9 +103,16 @@ impl MysqlSpawnConfig {
keep_alive_secs,
reject_no_database,
prepared_stmt_cache_size,
+ batching_enabled: false,
}
}
+ /// Enables ordinary-table batching for connections accepted by this server.
+ pub fn with_batching_enabled(mut self, enabled: bool) -> Self {
+ self.batching_enabled = enabled;
+ self
+ }
+
fn tls(&self) -> Option> {
self.tls.get_config()
}
@@ -213,7 +221,8 @@ impl MysqlServer {
stream.peer_addr()?,
process_id,
spawn_config.prepared_stmt_cache_size,
- );
+ )
+ .with_batching_enabled(spawn_config.batching_enabled);
let (mut r, w) = stream.into_split();
let mut w = BufWriter::with_capacity(DEFAULT_RESULT_SET_WRITE_BUFFER_SIZE, w);
diff --git a/src/servers/src/postgres.rs b/src/servers/src/postgres.rs
index 58ef4fdd7b7..1cd5907157b 100644
--- a/src/servers/src/postgres.rs
+++ b/src/servers/src/postgres.rs
@@ -80,6 +80,7 @@ pub struct PostgresServerHandlerInner {
force_tls: bool,
param_provider: Arc,
+ batching_enabled: bool,
session: Arc,
query_parser: Arc,
}
@@ -114,7 +115,12 @@ impl PgWireServerHandlers for PostgresServerHandler {
}
impl MakePostgresServerHandler {
- fn make(&self, addr: Option, process_id: u32) -> PostgresServerHandler {
+ fn make(
+ &self,
+ addr: Option,
+ process_id: u32,
+ batching_enabled: bool,
+ ) -> PostgresServerHandler {
let session = Arc::new(Session::new(
addr,
Channel::Postgres,
@@ -127,6 +133,7 @@ impl MakePostgresServerHandler {
force_tls: self.force_tls,
param_provider: self.param_provider.clone(),
+ batching_enabled,
session: session.clone(),
query_parser: Arc::new(DefaultQueryParser::new(self.query_handler.clone(), session)),
};
diff --git a/src/servers/src/postgres/handler.rs b/src/servers/src/postgres/handler.rs
index e618c55614b..7bf1edca705 100644
--- a/src/servers/src/postgres/handler.rs
+++ b/src/servers/src/postgres/handler.rs
@@ -59,6 +59,14 @@ use crate::postgres::utils::convert_err;
use crate::postgres::{PostgresServerHandlerInner, fixtures};
use crate::query_handler::sql::ServerSqlQueryHandlerRef;
+impl PostgresServerHandlerInner {
+ fn new_query_context(&self) -> QueryContextRef {
+ let mut ctx = self.session.new_query_context();
+ Arc::make_mut(&mut ctx).set_batching_enabled(self.batching_enabled);
+ ctx
+ }
+}
+
#[async_trait]
impl SimpleQueryHandler for PostgresServerHandlerInner {
#[tracing::instrument(skip_all, fields(protocol = "postgres"))]
@@ -68,7 +76,7 @@ impl SimpleQueryHandler for PostgresServerHandlerInner {
C::Error: Debug,
PgWireError: From<>::Error>,
{
- let query_ctx = self.session.new_query_context();
+ let query_ctx = self.new_query_context();
let db = query_ctx.get_db_string();
let _timer = crate::metrics::METRIC_POSTGRES_QUERY_TIMER
.with_label_values(&[crate::metrics::METRIC_POSTGRES_SIMPLE_QUERY, db.as_str()])
@@ -425,7 +433,7 @@ impl ExtendedQueryHandler for PostgresServerHandlerInner {
C::Error: Debug,
PgWireError: From<>::Error>,
{
- let query_ctx = self.session.new_query_context();
+ let query_ctx = self.new_query_context();
let db = query_ctx.get_db_string();
let _timer = crate::metrics::METRIC_POSTGRES_QUERY_TIMER
.with_label_values(&[crate::metrics::METRIC_POSTGRES_EXTENDED_QUERY, db.as_str()])
diff --git a/src/servers/src/postgres/server.rs b/src/servers/src/postgres/server.rs
index 09bc5015a0b..6e4aa942f9d 100644
--- a/src/servers/src/postgres/server.rs
+++ b/src/servers/src/postgres/server.rs
@@ -38,6 +38,7 @@ pub struct PostgresServer {
make_handler: Arc,
tls_server_config: Arc,
keep_alive_secs: u64,
+ batching_enabled: bool,
bind_addr: Option,
process_manager: Option,
}
@@ -66,17 +67,25 @@ impl PostgresServer {
make_handler,
tls_server_config,
keep_alive_secs,
+ batching_enabled: false,
bind_addr: None,
process_manager,
}
}
+ /// Enables ordinary-table batching for connections accepted by this server.
+ pub fn with_batching_enabled(mut self, enabled: bool) -> Self {
+ self.batching_enabled = enabled;
+ self
+ }
+
fn accept(
&self,
io_runtime: Runtime,
accepting_stream: AbortableStream,
) -> impl Future