rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-11-04 01:46:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into chore/stable-toolchain-1.96

Browse Source
This commit is contained in:
Ning Sun
2026-09-23 16:08:11 +08:00
parent 9ad5f851b1 a35f9d5c78
commit 61559bb053
82 changed files with 895 additions and 843 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/rust.yml
+45 -12
View File
@@ -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
.gitignore
-2
View File
@@ -52,8 +52,6 @@ venv/
tests-fuzz/artifacts/
tests-fuzz/corpus/
# cargo-udeps reports
udeps-report.json
# Nix
.direnv
Cargo.lock
Generated
+1 -129
View File
@@ -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",
Cargo.toml
-2
View File
@@ -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"
Makefile
+2 -5
View File
@@ -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.
config/config.md
+2 -2
View File
@@ -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.<br/>Available values: "last_non_null", "last_row". |
| `pending_rows_batcher` | -- | -- | Ordinary-table batching for opted-in HTTP ingestion protocols.<br/>PENDING_ROWS_BATCH_SYNC defaults to true for both batchers. Set it to false to acknowledge<br/>queue admission without waiting for storage; later failures cannot be returned to the client.<br/>Omitted or empty protocols disables batching. Prom without metric engine uses this batcher.<br/>OTLP logs, traces and ordinary metrics use this batcher. |
| `pending_rows_batcher` | -- | -- | Ordinary-table batching for opted-in ingestion protocols.<br/>PENDING_ROWS_BATCH_SYNC defaults to true for both batchers. Set it to false to acknowledge<br/>queue admission without waiting for storage; later failures cannot be returned to the client.<br/>Omitted or empty protocols disables batching. Prom without metric engine uses this batcher.<br/>OTLP logs, traces and ordinary metrics use this batcher.<br/>MySQL and PostgreSQL use the same acknowledgement policy. Protocol timeouts are unchanged.<br/>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.<br/>Available values: "last_non_null", "last_row". |
| `pending_rows_batcher` | -- | -- | Ordinary-table batching for opted-in HTTP ingestion protocols.<br/>PENDING_ROWS_BATCH_SYNC defaults to true for both batchers. Set it to false to acknowledge<br/>queue admission without waiting for storage; later failures cannot be returned to the client.<br/>Omitted or empty protocols disables batching. Prom without metric engine uses this batcher.<br/>OTLP logs, traces and ordinary metrics use this batcher. |
| `pending_rows_batcher` | -- | -- | Ordinary-table batching for opted-in ingestion protocols.<br/>PENDING_ROWS_BATCH_SYNC defaults to true for both batchers. Set it to false to acknowledge<br/>queue admission without waiting for storage; later failures cannot be returned to the client.<br/>Omitted or empty protocols disables batching. Prom without metric engine uses this batcher.<br/>OTLP logs, traces and ordinary metrics use this batcher.<br/>MySQL and PostgreSQL use the same acknowledgement policy. Protocol timeouts are unchanged.<br/>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. |
config/frontend.example.toml
+5 -1
View File
@@ -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.
config/standalone.example.toml
+5 -1
View File
@@ -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.
flake.nix
+1 -1
View File
@@ -46,7 +46,7 @@
])
cargo-nextest
cargo-llvm-cov
cargo-udeps
cargo-shear
taplo
curl
gnuplot ## for cargo bench
scripts/fix-udeps.py
-265
View File
@@ -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()
src/catalog/Cargo.toml
-2
View File
@@ -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
src/cli/Cargo.toml
-8
View File
@@ -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
src/cli/src/metadata/snapshot.rs
+177 -41
View File
@@ -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::<Vec<_>>()
.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::<Vec<_>>()
.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::<Vec<_>>()
.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) {
src/client/Cargo.toml
-4
View File
@@ -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"
src/cmd/Cargo.toml
+1 -9
View File
@@ -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
src/common/config/Cargo.toml
-1
View File
@@ -29,4 +29,3 @@ datanode.workspace = true
meta-client.workspace = true
serde.workspace = true
temp-env = "0.3"
tempfile.workspace = true
src/common/datasource/src/file_format/csv.rs
+9 -10
View File
@@ -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())
src/common/datasource/src/file_format/json.rs
+5 -6
View File
@@ -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,
src/common/datasource/src/file_format/tests.rs
+12 -16
View File
@@ -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()));
src/common/datasource/src/test_util.rs
+16 -15
View File
@@ -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());
}
src/common/datasource/src/tests.rs
+11 -14
View File
@@ -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;
}
src/common/frontend/Cargo.toml
-5
View File
@@ -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
src/common/function/Cargo.toml
-5
View File
@@ -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
src/common/grpc-expr/Cargo.toml
-4
View File
@@ -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
src/common/mem-prof/Cargo.toml
+6
View File
@@ -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"]
src/common/meta/Cargo.toml
-2
View File
@@ -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
src/common/meta/src/snapshot.rs
+10 -30
View File
@@ -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 { .. })
}
}
src/common/procedure-test/Cargo.toml
-1
View File
@@ -10,5 +10,4 @@ workspace = true
[dependencies]
async-trait.workspace = true
common-procedure = { workspace = true, features = ["testing"] }
snafu.workspace = true
tokio.workspace = true
src/common/query/Cargo.toml
+5
View File
@@ -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"]
src/common/runtime/Cargo.toml
-2
View File
@@ -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
src/common/sql/Cargo.toml
-1
View File
@@ -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
src/common/stat/Cargo.toml
-1
View File
@@ -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
src/common/substrait/Cargo.toml
-5
View File
@@ -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
src/datanode/Cargo.toml
-3
View File
@@ -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
src/datatypes/Cargo.toml
+5
View File
@@ -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"]
src/file-engine/Cargo.toml
-1
View File
@@ -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 }
src/flow/Cargo.toml
-17
View File
@@ -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
src/flow/src/batching_mode/batching_execution.rs
+6
View File
@@ -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<usize>,
) -> ExecuteOnceOutcome;
/// Rewrite the completed query plan after incremental merging and before execution.
fn rewrite_plan(&self, _task: &BatchingTask, plan: LogicalPlan) -> Result<LogicalPlan> {
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) {}
src/flow/src/batching_mode/task.rs
+4
View File
@@ -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())
src/flow/src/batching_mode/task/test.rs
+93
View File
@@ -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<Vec<LogicalPlan>>,
}
#[async_trait::async_trait]
impl crate::BatchingExecution for PlanRecordingExecution {
async fn execute_once(
self: Arc<Self>,
_guard: BatchingExecutionGuard,
_task: &BatchingTask,
_engine: &QueryEngineRef,
_frontend: &Arc<FrontendClient>,
_max_window_cnt: Option<usize>,
) -> ExecuteOnceOutcome {
ExecuteOnceOutcome {
new_query: None,
result: Ok(None),
}
}
fn rewrite_plan(&self, _task: &BatchingTask, plan: LogicalPlan) -> crate::Result<LogicalPlan> {
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 {
src/flow/src/lib.rs
+4 -1
View File
@@ -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};
src/frontend/Cargo.toml
-6
View File
@@ -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
src/frontend/src/frontend.rs
+3 -8
View File
@@ -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]
);
},
);
src/frontend/src/instance/builder.rs
+1 -1
View File
@@ -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};
src/frontend/src/instance/logical_batcher.rs
+1 -1
View File
@@ -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;
src/frontend/src/server.rs
+42 -20
View File
@@ -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<dyn Server>;
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<dyn Server>;
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),
src/frontend/src/service_config/pending_rows_batcher.rs
+39 -5
View File
@@ -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<BatcherOptions>,
}
/// 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<BatchingProtocol>,
/// 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::<FrontendOptions>(&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 =
src/frontend/src/service_config/prom_store.rs
+1 -5
View File
@@ -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()
src/log-store/Cargo.toml
+6 -2
View File
@@ -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"]
src/meta-client/Cargo.toml
-4
View File
@@ -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
src/meta-srv/Cargo.toml
-10
View File
@@ -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"
src/metric-engine/Cargo.toml
+4 -8
View File
@@ -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"]
src/mito2/Cargo.toml
+6 -14
View File
@@ -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"]
src/mito2/src/cache/write_cache.rs
Vendored
+2 -1
View File
@@ -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("");
src/object-store/Cargo.toml
+5 -8
View File
@@ -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
src/object-store/src/lib.rs
+1
View File
@@ -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;
src/operator/Cargo.toml
-2
View File
@@ -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
src/partition/Cargo.toml
-4
View File
@@ -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
src/pipeline/Cargo.toml
-9
View File
@@ -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]]
src/query/Cargo.toml
-7
View File
@@ -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
src/query/src/promql/label_values.rs
+19 -10
View File
@@ -37,15 +37,15 @@ fn millis_since_epoch(time: SystemTime) -> Result<Timestamp> {
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<i64> {
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());
}
}
}
src/servers/Cargo.toml
+1 -3
View File
@@ -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"] }
src/servers/src/batcher.rs
+39
View File
@@ -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::<bool>().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::<BatchingProtocol>("\"prom\"").unwrap(),
BatchingProtocol::Prom
);
for name in ["sql", "unknown"] {
assert!(serde_json::from_str::<BatchingProtocol>(&format!("\"{name}\"")).is_err());
}
assert_eq!(
serde_json::from_str::<BatchingProtocol>("\"http_sql\"").unwrap(),
BatchingProtocol::HttpSql
);
}
}
src/servers/src/http.rs
+3 -32
View File
@@ -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<Router>,
@@ -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::<BatchingProtocol>("\"prom\"").unwrap(),
BatchingProtocol::Prom
);
for name in ["sql", "unknown"] {
assert!(serde_json::from_str::<BatchingProtocol>(&format!("\"{name}\"")).is_err());
}
assert_eq!(
serde_json::from_str::<BatchingProtocol>("\"http_sql\"").unwrap(),
BatchingProtocol::HttpSql
);
}
#[derive(Default)]
struct RecordingWriteHandler {
selections: Mutex<Vec<bool>>,
src/servers/src/mysql/handler.rs
+28 -3
View File
@@ -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<Result<Output>> {
if let Some(output) =
@@ -495,7 +509,7 @@ impl<W: AsyncWrite + Send + Sync + Unpin> AsyncMysqlShim<W> 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<W: AsyncWrite + Send + Sync + Unpin> AsyncMysqlShim<W> 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<W: AsyncWrite + Send + Sync + Unpin> AsyncMysqlShim<W> 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())
src/servers/src/mysql/server.rs
+10 -1
View File
@@ -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<Arc<ServerConfig>> {
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);
src/servers/src/postgres.rs
+8 -1
View File
@@ -80,6 +80,7 @@ pub struct PostgresServerHandlerInner {
force_tls: bool,
param_provider: Arc<GreptimeDBStartupParameters>,
batching_enabled: bool,
session: Arc<Session>,
query_parser: Arc<DefaultQueryParser>,
}
@@ -114,7 +115,12 @@ impl PgWireServerHandlers for PostgresServerHandler {
}
impl MakePostgresServerHandler {
fn make(&self, addr: Option<SocketAddr>, process_id: u32) -> PostgresServerHandler {
fn make(
&self,
addr: Option<SocketAddr>,
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)),
};
src/servers/src/postgres/handler.rs
+10 -2
View File
@@ -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<<C as Sink<PgWireBackendMessage>>::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<<C as Sink<PgWireBackendMessage>>::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()])
src/servers/src/postgres/server.rs
+11 -1
View File
@@ -38,6 +38,7 @@ pub struct PostgresServer {
make_handler: Arc<MakePostgresServerHandler>,
tls_server_config: Arc<ReloadableTlsServerConfig>,
keep_alive_secs: u64,
batching_enabled: bool,
bind_addr: Option<SocketAddr>,
process_manager: Option<ProcessManagerRef>,
}
@@ -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<Output = ()> + use<> {
let handler_maker = self.make_handler.clone();
let batching_enabled = self.batching_enabled;
let tls_server_config = self.tls_server_config.clone();
let process_manager = self.process_manager.clone();
accepting_stream.for_each(move |tcp_stream| {
@@ -102,7 +111,8 @@ impl PostgresServer {
let _handle = io_runtime.spawn(async move {
crate::metrics::METRIC_POSTGRES_CONNECTIONS.inc();
let pg_handler = Arc::new(handler_maker.make(addr, process_id));
let pg_handler =
Arc::new(handler_maker.make(addr, process_id, batching_enabled));
let r =
process_socket(io_stream, tls_acceptor.clone(), pg_handler).await;
crate::metrics::METRIC_POSTGRES_CONNECTIONS.dec();
src/servers/tests/http/prom_store_test.rs
+2 -1
View File
@@ -36,11 +36,12 @@ use datafusion_expr::LogicalPlan;
use prost::Message;
use query::parser::PromQuery;
use query::query_engine::DescribeResult;
use servers::batcher::BatchingProtocol;
use servers::error::{self, Result};
use servers::http::header::{CONTENT_ENCODING_SNAPPY, CONTENT_TYPE_PROTOBUF};
use servers::http::prom_store::PHYSICAL_TABLE_PARAM;
use servers::http::test_helpers::{TestClient, TestResponse};
use servers::http::{BatchingProtocol, HttpOptions, HttpServerBuilder};
use servers::http::{HttpOptions, HttpServerBuilder};
use servers::prom_remote_write::v2::test_util as remote_write_v2;
use servers::prom_remote_write::validation::PromValidationMode;
use servers::prom_store;
src/session/Cargo.toml
+6
View File
@@ -29,3 +29,9 @@ derive_more.workspace = true
snafu.workspace = true
sql.workspace = true
uuid.workspace = true
[package.metadata.cargo-shear]
# `common-error` is required by `common-macro`'s `stack_trace_debug`: the
# generated code references `::common_error` paths, so it must stay a direct
# dependency.
ignored = ["common-error"]
src/session/src/context.rs
+1 -1
View File
@@ -471,7 +471,7 @@ impl QueryContext {
self.logical_batching_enabled = enabled;
}
/// Whether the local HTTP entry point selected ordinary-table batching.
/// Whether the local protocol entry point selected ordinary-table batching.
pub fn batching_enabled(&self) -> bool {
self.batching_enabled
}
src/sql/Cargo.toml
-7
View File
@@ -16,25 +16,19 @@ arrow-buffer.workspace = true
chrono.workspace = true
common-base.workspace = true
common-catalog.workspace = true
common-decimal.workspace = true
common-error.workspace = true
common-macro.workspace = true
common-query.workspace = true
common-sql.workspace = true
common-telemetry.workspace = true
common-time.workspace = true
datafusion.workspace = true
datafusion-common.workspace = true
datafusion-expr.workspace = true
datafusion-physical-expr.workspace = true
datafusion-sql.workspace = true
datatypes.workspace = true
either.workspace = true
hex = "0.4"
humantime.workspace = true
iso8601 = "0.6.1"
itertools.workspace = true
jsonb.workspace = true
lazy_static.workspace = true
promql-parser.workspace = true
regex.workspace = true
@@ -43,7 +37,6 @@ serde_json.workspace = true
snafu.workspace = true
sqlparser.workspace = true
sqlparser_derive = "0.1"
store-api.workspace = true
table.workspace = true
uuid.workspace = true
src/standalone/src/options.rs
+3 -8
View File
@@ -217,6 +217,7 @@ mod tests {
use std::sync::Arc;
use common_event_recorder::EventTypeFilter;
use servers::batcher::BatchingProtocol;
use crate::options::*;
@@ -266,10 +267,7 @@ mod tests {
.logical_table
.unwrap()
.protocols,
vec![
servers::http::BatchingProtocol::Prom,
servers::http::BatchingProtocol::Otlp
]
vec![BatchingProtocol::Prom, BatchingProtocol::Otlp]
);
},
);
@@ -288,10 +286,7 @@ mod tests {
.unwrap();
assert_eq!(
options.pending_rows_batcher.table.protocols,
vec![
servers::http::BatchingProtocol::Influxdb,
servers::http::BatchingProtocol::HttpSql
]
vec![BatchingProtocol::Influxdb, BatchingProtocol::HttpSql]
);
},
);
src/store-api/Cargo.toml
-5
View File
@@ -36,15 +36,10 @@ prost.workspace = true
serde.workspace = true
serde_json.workspace = true
snafu.workspace = true
sqlparser.workspace = true
strum.workspace = true
tokio.workspace = true
uuid.workspace = true
[dev-dependencies]
async-stream.workspace = true
common-meta.workspace = true
serde_json.workspace = true
[package.metadata.cargo-udeps.ignore]
normal = ["aquamarine"]
src/table/Cargo.toml
-3
View File
@@ -34,13 +34,11 @@ derive_builder.workspace = true
futures.workspace = true
greptime-proto.workspace = true
humantime.workspace = true
humantime-serde.workspace = true
lazy_static.workspace = true
once_cell.workspace = true
paste.workspace = true
serde.workspace = true
snafu.workspace = true
sqlparser.workspace = true
store-api.workspace = true
tokio.workspace = true
@@ -48,4 +46,3 @@ tokio.workspace = true
common-test-util.workspace = true
parquet = { workspace = true, features = ["async"] }
serde_json.workspace = true
tokio-util = { version = "0.7", features = ["compat"] }
tests-fuzz/Cargo.toml
+9
View File
@@ -188,3 +188,12 @@ path = "utils/kafka_wal_helper.rs"
test = false
bench = false
doc = false
[package.metadata.cargo-shear]
# `common-error` is required by `common-macro`'s `stack_trace_debug`: the
# generated code references `::common_error` paths, so it must stay a direct
# dependency.
# `k8s-openapi` is a feature-pinning dependency: it enables the `v1_30`
# version feature required by the transitive `kube` -> `k8s-openapi` chain,
# but is never referenced by name in source code.
ignored = ["common-error", "k8s-openapi"]
tests-integration/Cargo.toml
+17 -22
View File
@@ -20,16 +20,11 @@ workspace = true
[dependencies]
api.workspace = true
arrow-flight.workspace = true
async-stream.workspace = true
async-trait.workspace = true
auth.workspace = true
axum.workspace = true
base64.workspace = true
cache.workspace = true
catalog.workspace = true
chrono.workspace = true
clap.workspace = true
cli.workspace = true
client.workspace = true
cmd.workspace = true
common-base.workspace = true
@@ -38,16 +33,12 @@ common-config.workspace = true
common-datasource.workspace = true
common-error.workspace = true
common-event-recorder.workspace = true
common-frontend.workspace = true
common-grpc.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
common-stat.workspace = true
common-telemetry.workspace = true
common-test-util.workspace = true
common-time.workspace = true
@@ -55,62 +46,59 @@ common-wal.workspace = true
datanode = { workspace = true }
datatypes.workspace = true
dotenv.workspace = true
flate2 = "1.0"
flow.workspace = true
frontend.workspace = true
futures.workspace = true
futures-util.workspace = true
http.workspace = true
hyper-util = { workspace = true, features = ["tokio"] }
log-query = { workspace = true }
loki-proto.workspace = true
meta-client.workspace = true
meta-srv = { workspace = true, features = ["mock"] }
mito2.workspace = true
object-store.workspace = true
operator.workspace = true
plugins.workspace = true
prost.workspace = true
query.workspace = true
rand.workspace = true
serde_json.workspace = true
servers.workspace = true
session.workspace = true
snafu.workspace = true
sql.workspace = true
sqlx = { workspace = true, features = [
"mysql",
"postgres",
"chrono",
] }
standalone.workspace = true
store-api.workspace = true
substrait.workspace = true
table.workspace = true
tempfile.workspace = true
time = "0.3"
tokio.workspace = true
tokio-stream = { workspace = true, features = ["net"] }
tonic.workspace = true
tower.workspace = true
uuid.workspace = true
zstd.workspace = true
[dev-dependencies]
base64.workspace = true
catalog = { workspace = true, features = ["testing"] }
chrono.workspace = true
clap.workspace = true
cli.workspace = true
client = { workspace = true, features = ["testing"] }
common-frontend.workspace = true
common-memory-manager.workspace = true
common-meta = { workspace = true, features = ["testing"] }
common-options.workspace = true
common-procedure-test.workspace = true
common-wal = { workspace = true, features = ["testing"] }
datafusion.workspace = true
datafusion-expr.workspace = true
either.workspace = true
flate2 = "1.0"
frontend = { workspace = true, features = ["testing"] }
hex.workspace = true
http.workspace = true
itertools.workspace = true
jsonb.workspace = true
log-query.workspace = true
loki-proto.workspace = true
mito2 = { workspace = true, features = ["test", "test-shared-fs-region-migration"] }
mysql_async = { version = "0.37", default-features = false, features = [
"time",
@@ -119,9 +107,9 @@ mysql_async = { version = "0.37", default-features = false, features = [
opentelemetry-proto.workspace = true
operator = { workspace = true, features = ["testing"] }
otel-arrow-rust.workspace = true
partition.workspace = true
paste.workspace = true
pipeline.workspace = true
plugins.workspace = true
prost.workspace = true
rand.workspace = true
rskafka.workspace = true
@@ -131,9 +119,16 @@ sea-query.workspace = true
servers = { workspace = true, features = ["testing"] }
session = { workspace = true, features = ["testing"] }
similar-asserts.workspace = true
sqlx = { workspace = true, features = [
"mysql",
"postgres",
"chrono",
] }
store-api.workspace = true
time = "0.3"
tokio-postgres = { workspace = true }
tokio-util.workspace = true
url = "2.3"
urlencoding = "2.1"
yaml-rust = "0.4"
zstd.workspace = true
tests-integration/src/otlp.rs
+1 -2
View File
@@ -514,8 +514,7 @@ WITH(
ExponentialHistogram, ExponentialHistogramDataPoint, exponential_histogram_data_point,
};
use prost::Message;
use servers::batcher::pending_rows_batch_sync_enabled;
use servers::http::BatchingProtocol;
use servers::batcher::{BatchingProtocol, pending_rows_batch_sync_enabled};
use servers::http::test_helpers::TestClient;
use session::protocol_ctx::{OtlpMetricCtx, ProtocolCtx};
tests-integration/src/standalone.rs
+10 -1
View File
@@ -90,6 +90,7 @@ pub struct GreptimeDbStandaloneBuilder {
auto_create_table: bool,
experimental_metric_export: bool,
logical_batcher: Option<BatcherOptions>,
table_batcher: BatcherOptions,
}
impl GreptimeDbStandaloneBuilder {
@@ -111,6 +112,7 @@ impl GreptimeDbStandaloneBuilder {
auto_create_table: true,
experimental_metric_export: false,
logical_batcher: None,
table_batcher: BatcherOptions::default(),
}
}
@@ -121,6 +123,13 @@ impl GreptimeDbStandaloneBuilder {
self
}
/// Configures ordinary-table batching for protocol integration tests.
#[must_use]
pub fn with_table_batcher(mut self, options: BatcherOptions) -> Self {
self.table_batcher = options;
self
}
/// Configures logical-table batching for integration tests.
#[must_use]
pub fn with_logical_batcher(mut self, options: BatcherOptions) -> Self {
@@ -389,7 +398,7 @@ impl GreptimeDbStandaloneBuilder {
experimental_metric_export: self.experimental_metric_export,
pending_rows_batcher: PendingRowsBatcherOptions {
logical_table: self.logical_batcher.clone(),
..Default::default()
table: self.table_batcher.clone(),
},
// Tests cover the descriptor, so they run with it enabled.
otlp: frontend::service_config::OtlpOptions {
tests-integration/tests/sql.rs
+181
View File
@@ -2672,3 +2672,184 @@ pub async fn test_declare_fetch_close_cursor(store_type: StorageType) {
let _ = fe_pg_server.shutdown().await;
guard.remove_all().await;
}
/// Keeps SQL and input rows fixed while changing only the selected batching protocols.
#[tokio::test(flavor = "multi_thread")]
async fn test_sql_batcher_alignment() {
use common_telemetry::{dump_metrics, info, init_default_ut_logging};
use frontend::server::Services;
use frontend::service_config::BatcherOptions;
use servers::batcher::BatchingProtocol;
use servers::mysql::server::MYSQL_SERVER;
use tests_integration::standalone::GreptimeDbStandaloneBuilder;
fn flushes() -> u64 {
dump_metrics()
.unwrap()
.lines()
.find_map(|line| {
line.strip_prefix("greptime_table_batcher_flush_total ")
.map(|value| value.parse().unwrap())
})
.unwrap_or(0)
}
init_default_ut_logging();
for protocols in [
vec![],
vec![BatchingProtocol::Mysql],
vec![BatchingProtocol::Postgres],
vec![BatchingProtocol::Mysql, BatchingProtocol::Postgres],
] {
let mysql_enabled = protocols.contains(&BatchingProtocol::Mysql);
let pg_enabled = protocols.contains(&BatchingProtocol::Postgres);
let mut instance = GreptimeDbStandaloneBuilder::new("sql_batcher_alignment")
.with_table_batcher(BatcherOptions {
protocols,
pending_rows_flush_interval: Duration::from_millis(10),
..Default::default()
})
.build()
.await;
let mut opts = instance.opts.clone();
opts.http.addr = "127.0.0.1:0".into();
opts.grpc.bind_addr = "127.0.0.1:0".into();
opts.mysql.addr = "127.0.0.1:0".into();
opts.postgres.addr = "127.0.0.1:0".into();
let mut servers = Services::new(opts, instance.fe_instance().clone(), Default::default())
.build()
.unwrap();
servers.start_all().await.unwrap();
let mysql_url = format!("mysql://{}/public", servers.addr(MYSQL_SERVER).unwrap());
let mut mysql = MySqlConnection::connect(&mysql_url).await.unwrap();
let (pg, connection) = tokio_postgres::connect(
&format!(
"postgres://{}/public",
servers.addr("POSTGRES_SERVER").unwrap()
),
NoTls,
)
.await
.unwrap();
let pg_task = tokio::spawn(async move { connection.await.unwrap() });
for round in 0..2 {
mysql.execute("CREATE TABLE batch_sql (ts TIMESTAMP TIME INDEX, host STRING PRIMARY KEY, val INT DEFAULT 7)").await.unwrap();
let started = std::time::Instant::now();
let before = flushes();
// COM_QUERY and COM_STMT_EXECUTE must both opt in.
assert_eq!(
mysql
.execute("INSERT INTO batch_sql (ts, host) VALUES (0, 'a')")
.await
.unwrap()
.rows_affected(),
1
);
assert_eq!(
sqlx::query("INSERT INTO batch_sql VALUES (0, ?, ?)")
.persistent(false)
.bind("b")
.bind(8_i32)
.execute(&mut mysql)
.await
.unwrap()
.rows_affected(),
1
);
assert_eq!(flushes() - before, if mysql_enabled { 2 } else { 0 });
let before = flushes();
let result = pg
.simple_query("INSERT INTO batch_sql (ts, host) VALUES (0, 'c')")
.await
.unwrap();
assert!(matches!(
result.as_slice(),
[SimpleQueryMessage::CommandComplete(1)]
));
let statement = pg
.prepare("INSERT INTO batch_sql VALUES (0, $1, $2)")
.await
.unwrap();
assert_eq!(pg.execute(&statement, &[&"d", &9_i32]).await.unwrap(), 1);
assert_eq!(flushes() - before, if pg_enabled { 2 } else { 0 });
// Independent connections can submit to the same worker concurrently.
let (mysql_result, pg_result) = tokio::join!(
mysql.execute("INSERT INTO batch_sql VALUES (0, 'e', 10)"),
pg.simple_query("INSERT INTO batch_sql VALUES (0, 'f', 11)")
);
assert_eq!(mysql_result.unwrap().rows_affected(), 1);
assert!(matches!(
pg_result.unwrap().as_slice(),
[SimpleQueryMessage::CommandComplete(1)]
));
// Errors must reach the client without affecting later requests.
assert!(
mysql
.execute("INSERT INTO batch_sql (missing) VALUES (1)")
.await
.is_err()
);
assert!(
pg.simple_query("INSERT INTO batch_sql (missing) VALUES (1)")
.await
.is_err()
);
let expected = vec![
("a".to_string(), 7_i32),
("b".to_string(), 8),
("c".to_string(), 7),
("d".to_string(), 9),
("e".to_string(), 10),
("f".to_string(), 11),
];
// No polling: successful protocol responses guarantee visibility.
let rows: Vec<(String, i32)> =
sqlx::query_as("SELECT host, val FROM batch_sql ORDER BY host")
.persistent(false)
.fetch_all(&mut mysql)
.await
.unwrap();
assert_eq!(rows, expected);
let rows: Vec<(String, i32)> = pg
.query("SELECT host, val FROM batch_sql ORDER BY host", &[])
.await
.unwrap()
.iter()
.map(|row| (row.get(0), row.get(1)))
.collect();
assert_eq!(rows, expected);
mysql.execute("CREATE TABLE batch_copy (ts TIMESTAMP TIME INDEX, host STRING PRIMARY KEY, val INT)").await.unwrap();
assert_eq!(
mysql
.execute("INSERT INTO batch_copy SELECT * FROM batch_sql")
.await
.unwrap()
.rows_affected(),
6
);
let rows: Vec<(String, i32)> =
sqlx::query_as("SELECT host, val FROM batch_copy ORDER BY host")
.persistent(false)
.fetch_all(&mut mysql)
.await
.unwrap();
assert_eq!(rows, expected);
info!(
"sql_batcher_alignment mysql={mysql_enabled} postgres={pg_enabled} round={round} elapsed={:?}",
started.elapsed()
);
mysql.execute("DROP TABLE batch_sql").await.unwrap();
mysql.execute("DROP TABLE batch_copy").await.unwrap();
}
mysql.close().await.unwrap();
drop(pg);
pg_task.await.unwrap();
servers.shutdown_all().await.unwrap();
instance.guard.remove_all().await;
}
}
tests/runner/Cargo.toml
-1
View File
@@ -13,7 +13,6 @@ clap.workspace = true
client.workspace = true
common-error.workspace = true
common-query.workspace = true
common-recordbatch.workspace = true
common-time.workspace = true
datatypes = { workspace = true }
flate2 = "1.0"