fix: update tokio-postgres and correct explain/fetch cursor output schema (#8955)

* chore(deps): update tokio-postgres

* fix: describing fetch cursor and analyze

(cherry picked from commit b31f05eb59)
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
Ning Sun
2026-09-04 13:14:41 +08:00
committed by discord9
parent 9d34394c31
commit d071bdb796
9 changed files with 221 additions and 75 deletions
Generated
+153 -65
View File
@@ -1629,7 +1629,7 @@ dependencies = [
"maybe-owned",
"rustix 1.0.7",
"rustix-linux-procfs",
"windows-sys 0.61.2",
"windows-sys 0.60.2",
"winx",
]
@@ -2265,7 +2265,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "117725a109d387c937a1533ce01b450cbde6b88abceea8473c4d7a85853cda3c"
dependencies = [
"lazy_static",
"windows-sys 0.48.0",
"windows-sys 0.59.0",
]
[[package]]
@@ -4630,10 +4630,23 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb"
dependencies = [
"const-oid 0.9.6",
"der_derive",
"flagset",
"pem-rfc7468",
"zeroize",
]
[[package]]
name = "der_derive"
version = "0.7.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8034092389675178f570469e6c3b0465d3d30b4505c294a6550db47f3c17ad18"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.117",
]
[[package]]
name = "deranged"
version = "0.5.8"
@@ -5337,6 +5350,12 @@ version = "0.5.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1d674e81391d1e1ab681a28d99df07927c6d4aa5b027d7da16ba32d1d21ecd99"
[[package]]
name = "flagset"
version = "0.4.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b7ac824320a75a52197e8f2d787f6a38b6718bb6897a35142d749af3c0e8f4fe"
[[package]]
name = "flatbuffers"
version = "25.2.10"
@@ -5978,7 +5997,7 @@ dependencies = [
"cfg-if",
"js-sys",
"libc",
"wasi",
"wasi 0.11.1+wasi-snapshot-preview1",
"wasm-bindgen",
]
@@ -6669,7 +6688,7 @@ dependencies = [
"js-sys",
"log",
"wasm-bindgen",
"windows-core 0.61.2",
"windows-core 0.57.0",
]
[[package]]
@@ -7838,7 +7857,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "07033963ba89ebaf1584d767badaa2e8fcec21aedea6b8c0346d487d49c28667"
dependencies = [
"cfg-if",
"windows-targets 0.48.5",
"windows-targets 0.53.5",
]
[[package]]
@@ -7869,13 +7888,14 @@ checksum = "f9fbbcab51052fe104eb5e5d351cf728d30a5be1fe14d9be8a3b097481fb97de"
[[package]]
name = "libredox"
version = "0.1.4"
version = "0.1.23"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1580801010e535496706ba011c15f8532df6b42297d2e471fec38ceadd8c0638"
checksum = "8d8f1ea3f21fd3405dcaf6c9b5c1630af9afc422d9073ea39c5f6d6c772e08ed"
dependencies = [
"bitflags 2.12.1",
"libc",
"redox_syscall 0.5.13",
"plain",
"redox_syscall 0.9.3",
]
[[package]]
@@ -8545,7 +8565,7 @@ checksum = "02bd0af71c67b473010cbbc60715ee815645a4dc942899111f494b4b737d6fda"
dependencies = [
"libc",
"log",
"wasi",
"wasi 0.11.1+wasi-snapshot-preview1",
"windows-sys 0.61.2",
]
@@ -9106,7 +9126,7 @@ version = "0.50.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5"
dependencies = [
"windows-sys 0.61.2",
"windows-sys 0.59.0",
]
[[package]]
@@ -9300,7 +9320,7 @@ version = "0.7.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "77e878c846a8abae00dd069496dbe8751b16ac1c3d6bd2a7283a938e8228f90d"
dependencies = [
"proc-macro-crate 1.3.1",
"proc-macro-crate 3.3.0",
"proc-macro2",
"quote",
"syn 2.0.117",
@@ -9330,6 +9350,15 @@ dependencies = [
"objc2-encode",
]
[[package]]
name = "objc2-core-foundation"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2a180dd8642fa45cdb7dd721cd4c11b1cadd4929ce112ebd8b9f5803cc79d536"
dependencies = [
"bitflags 2.12.1",
]
[[package]]
name = "objc2-encode"
version = "4.1.0"
@@ -9346,6 +9375,15 @@ dependencies = [
"objc2",
]
[[package]]
name = "objc2-system-configuration"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7216bd11cbda54ccabcab84d523dc93b858ec75ecfb3a7d89513fa22464da396"
dependencies = [
"objc2-core-foundation",
]
[[package]]
name = "object"
version = "0.36.7"
@@ -10575,16 +10613,7 @@ dependencies = [
"tokio",
"tokio-rustls",
"tokio-util",
"x509-certificate 0.25.0",
]
[[package]]
name = "phf"
version = "0.11.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1fd6780a80ae0c52cc120a26a1a42c1ae51b247a253e4e06113d23d2c2edd078"
dependencies = [
"phf_shared 0.11.3",
"x509-certificate",
]
[[package]]
@@ -10837,6 +10866,12 @@ version = "0.3.32"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c"
[[package]]
name = "plain"
version = "0.2.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b4596b6d070b27117e987119b4dac604f3c58cfb0b191112e24771b2faeac1a6"
[[package]]
name = "plist"
version = "1.7.2"
@@ -10929,34 +10964,34 @@ dependencies = [
[[package]]
name = "postgres-protocol"
version = "0.6.8"
version = "0.6.12"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "76ff0abab4a9b844b93ef7b81f1efc0a366062aaef2cd702c76256b5dc075c54"
checksum = "08808e3c483c46e999108051c78334f473d5adb59d78bb80a1268c7e6aa6c514"
dependencies = [
"base64 0.22.1",
"byteorder",
"bytes",
"fallible-iterator",
"hmac 0.12.1",
"md-5 0.10.6",
"hmac 0.13.0",
"md-5 0.11.0",
"memchr",
"rand 0.9.4",
"sha2 0.10.9",
"rand 0.10.1",
"sha2 0.11.0",
"stringprep",
]
[[package]]
name = "postgres-types"
version = "0.2.9"
version = "0.2.14"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "613283563cd90e1dfc3518d548caee47e0e725455ed619881f5cf21f36de4b48"
checksum = "851ca9db4932932d69f3ea811b1abe63087a0f740a47692619dd40d4899b68be"
dependencies = [
"array-init",
"bytes",
"chrono",
"fallible-iterator",
"postgres-protocol",
"serde",
"serde_core",
"serde_json",
]
@@ -11322,7 +11357,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "22505a5c94da8e3b7c2996394d1c933236c4d743e81a410bcca4e6989fc066a4"
dependencies = [
"bytes",
"heck 0.5.0",
"heck 0.4.1",
"itertools 0.12.1",
"log",
"multimap",
@@ -11342,8 +11377,8 @@ version = "0.14.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ac6c3320f9abac597dcbc668774ef006702672474aad53c6d596b62e487b40b1"
dependencies = [
"heck 0.5.0",
"itertools 0.10.5",
"heck 0.4.1",
"itertools 0.14.0",
"log",
"multimap",
"once_cell",
@@ -11391,7 +11426,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d"
dependencies = [
"anyhow",
"itertools 0.10.5",
"itertools 0.14.0",
"proc-macro2",
"quote",
"syn 2.0.117",
@@ -11404,7 +11439,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9120690fafc389a67ba3803df527d0ec9cbbc9cc45e4cc20b332996dfb672425"
dependencies = [
"anyhow",
"itertools 0.10.5",
"itertools 0.14.0",
"proc-macro2",
"quote",
"syn 2.0.117",
@@ -12046,6 +12081,15 @@ dependencies = [
"bitflags 2.12.1",
]
[[package]]
name = "redox_syscall"
version = "0.9.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d678d17679829e73d371e96880897e98fee2ded7acc0a50bdf8af2affa4b2fe5"
dependencies = [
"bitflags 2.12.1",
]
[[package]]
name = "redox_users"
version = "0.4.6"
@@ -12859,7 +12903,7 @@ dependencies = [
"security-framework 3.7.0",
"security-framework-sys",
"webpki-root-certs",
"windows-sys 0.61.2",
"windows-sys 0.59.0",
]
[[package]]
@@ -13743,7 +13787,7 @@ version = "0.8.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1961e2ef424c1424204d3a5d6975f934f56b6d50ff5732382d84ebf460e147f7"
dependencies = [
"heck 0.5.0",
"heck 0.4.1",
"proc-macro2",
"quote",
"syn 2.0.117",
@@ -13772,7 +13816,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51"
dependencies = [
"libc",
"windows-sys 0.61.2",
"windows-sys 0.60.2",
]
[[package]]
@@ -14075,7 +14119,7 @@ dependencies = [
"stringprep",
"thiserror 2.0.17",
"tracing",
"whoami",
"whoami 1.6.0",
]
[[package]]
@@ -14114,7 +14158,7 @@ dependencies = [
"stringprep",
"thiserror 2.0.17",
"tracing",
"whoami",
"whoami 1.6.0",
]
[[package]]
@@ -14766,7 +14810,7 @@ dependencies = [
"getrandom 0.3.4",
"once_cell",
"rustix 1.0.7",
"windows-sys 0.61.2",
"windows-sys 0.59.0",
]
[[package]]
@@ -15131,6 +15175,27 @@ version = "0.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20"
[[package]]
name = "tls_codec"
version = "0.4.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0de2e01245e2bb89d6f05801c564fa27624dbd7b1846859876c7dad82e90bf6b"
dependencies = [
"tls_codec_derive",
"zeroize",
]
[[package]]
name = "tls_codec_derive"
version = "0.4.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2d2e76690929402faae40aebdda620a2c0e25dd6d3b9afe48867dfd95991f4bd"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.117",
]
[[package]]
name = "tokio"
version = "1.52.3"
@@ -15197,9 +15262,9 @@ dependencies = [
[[package]]
name = "tokio-postgres"
version = "0.7.13"
version = "0.7.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6c95d533c83082bb6490e0189acaa0bbeef9084e60471b696ca6988cd0541fb0"
checksum = "a528f7d280f6d5b9cd149635c8705b0dd049754bc67d81d31fa25169a93809d3"
dependencies = [
"async-trait",
"byteorder",
@@ -15210,29 +15275,29 @@ dependencies = [
"log",
"parking_lot 0.12.4",
"percent-encoding",
"phf 0.11.3",
"phf 0.13.1",
"pin-project-lite",
"postgres-protocol",
"postgres-types",
"rand 0.9.4",
"socket2 0.5.10",
"rand 0.10.1",
"socket2 0.6.4",
"tokio",
"tokio-util",
"whoami",
"whoami 2.1.3",
]
[[package]]
name = "tokio-postgres-rustls"
version = "0.12.0"
version = "0.14.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "04fb792ccd6bbcd4bba408eb8a292f70fc4a3589e5d793626f45190e6454b6ab"
checksum = "4c2ad44aa0ae96db89c4742212ed41645b2f597311ff6e1945542a4d9fadc2fb"
dependencies = [
"ring",
"rustls",
"sha2 0.11.0",
"tokio",
"tokio-postgres",
"tokio-rustls",
"x509-certificate 0.23.1",
"x509-cert",
]
[[package]]
@@ -16249,6 +16314,15 @@ version = "0.11.1+wasi-snapshot-preview1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
[[package]]
name = "wasi"
version = "0.14.7+wasi-0.2.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "883478de20367e224c0090af9cf5f9fa85bed63a95c1abf3afc5c083ebc06e8c"
dependencies = [
"wasip2",
]
[[package]]
name = "wasip2"
version = "1.0.2+wasi-0.2.9"
@@ -16273,6 +16347,15 @@ version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b8dad83b4f25e74f184f64c43b150b91efe7647395b42289f38e50566d82855b"
[[package]]
name = "wasite"
version = "1.0.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "66fe902b4a6b8028a753d5424909b764ccf79b7a209eac9bf97e59cda9f71a42"
dependencies = [
"wasi 0.14.7+wasi-0.2.4",
]
[[package]]
name = "wasm-bindgen"
version = "0.2.122"
@@ -16483,7 +16566,19 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6994d13118ab492c3c80c1f81928718159254c53c472bf9ce36f8dae4add02a7"
dependencies = [
"redox_syscall 0.5.13",
"wasite",
"wasite 0.1.0",
]
[[package]]
name = "whoami"
version = "2.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "626c4bac6755d76ffc12cb01b2eac751db1996b9e0041de9aa02c8c211ddc82c"
dependencies = [
"libc",
"libredox",
"objc2-system-configuration",
"wasite 1.0.2",
"web-sys",
]
@@ -16519,7 +16614,7 @@ version = "0.1.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cf221c93e13a30d793f7645a0e7762c55d169dbb0a49671918a2319d289b10bb"
dependencies = [
"windows-sys 0.48.0",
"windows-sys 0.59.0",
]
[[package]]
@@ -17092,22 +17187,15 @@ dependencies = [
]
[[package]]
name = "x509-certificate"
version = "0.23.1"
name = "x509-cert"
version = "0.2.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "66534846dec7a11d7c50a74b7cdb208b9a581cad890b7866430d438455847c85"
checksum = "1301e935010a701ae5f8655edc0ad17c44bad3ac5ce8c39185f75453b720ae94"
dependencies = [
"bcder",
"bytes",
"chrono",
"const-oid 0.9.6",
"der",
"hex",
"pem",
"ring",
"signature",
"spki",
"thiserror 1.0.69",
"zeroize",
"tls_codec",
]
[[package]]
+1 -1
View File
@@ -251,7 +251,7 @@ strum = { version = "0.27", features = ["derive"] }
sysinfo = "0.33"
tempfile = "3"
tokio = { version = "1.47", features = ["full"] }
tokio-postgres = "0.7"
tokio-postgres = "0.7.18"
tokio-rustls = { version = "0.26.2", default-features = false }
tokio-stream = "0.1"
tokio-util = { version = "0.7", features = ["io-util", "compat"] }
+1 -1
View File
@@ -89,7 +89,7 @@ strum.workspace = true
table = { workspace = true, features = ["testing"] }
tokio.workspace = true
tokio-postgres = { workspace = true, optional = true }
tokio-postgres-rustls = { version = "0.12", optional = true }
tokio-postgres-rustls = { version = "0.14", optional = true }
tonic.workspace = true
tracing.workspace = true
typetag.workspace = true
+9
View File
@@ -12,6 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use datatypes::schema::SchemaRef;
use futures::StreamExt;
use tokio::sync::Mutex;
@@ -28,12 +29,15 @@ struct Inner {
/// A cursor on RecordBatchStream that fetches data batch by batch
pub struct RecordBatchStreamCursor {
schema: SchemaRef,
inner: Mutex<Inner>,
}
impl RecordBatchStreamCursor {
pub fn new(stream: SendableRecordBatchStream) -> RecordBatchStreamCursor {
let schema = stream.schema();
Self {
schema,
inner: Mutex::new(Inner {
stream,
current_row_index: 0,
@@ -43,6 +47,11 @@ impl RecordBatchStreamCursor {
}
}
/// Returns the schema of the underlying record batch stream.
pub fn schema(&self) -> SchemaRef {
self.schema.clone()
}
/// Take `size` of row from the `RecordBatchStream` and create a new
/// `RecordBatch` for these rows.
pub async fn take(&self, size: usize) -> Result<RecordBatch> {
+16 -5
View File
@@ -46,6 +46,21 @@ const STAGE: &str = "stage";
const NODE: &str = "node";
const PLAN: &str = "plan";
/// Returns the fixed output schema of [`DistAnalyzeExec`]:
/// (`stage`: UInt32, `node`: UInt32, `plan`: Utf8).
///
/// Exposed so protocol-level `Describe` handlers can advertise the schema
/// clients will actually receive (the physical plan is rewritten in
/// `optimize_physical_plan` and its schema differs from the DataFusion
/// `Analyze` logical plan schema).
pub fn dist_analyze_output_schema() -> SchemaRef {
SchemaRef::new(Schema::new(vec![
Field::new(STAGE, DataType::UInt32, true),
Field::new(NODE, DataType::UInt32, true),
Field::new(PLAN, DataType::Utf8, true),
]))
}
#[derive(Debug)]
pub struct DistAnalyzeExec {
input: Arc<dyn ExecutionPlan>,
@@ -58,11 +73,7 @@ pub struct DistAnalyzeExec {
impl DistAnalyzeExec {
/// Create a new DistAnalyzeExec
pub fn new(input: Arc<dyn ExecutionPlan>, verbose: bool, format: AnalyzeFormat) -> Self {
let schema = SchemaRef::new(Schema::new(vec![
Field::new(STAGE, DataType::UInt32, true),
Field::new(NODE, DataType::UInt32, true),
Field::new(PLAN, DataType::Utf8, true),
]));
let schema = dist_analyze_output_schema();
let properties = Arc::new(Self::compute_properties(&input, schema.clone()));
Self {
input,
+1 -1
View File
@@ -45,7 +45,7 @@ pub(crate) mod test_util;
#[cfg(test)]
mod tests;
pub use crate::analyze::analyze_plan_metrics_to_json_value;
pub use crate::analyze::{analyze_plan_metrics_to_json_value, dist_analyze_output_schema};
pub use crate::datafusion::DfContextProviderAdapter;
pub use crate::query_engine::{
QueryEngine, QueryEngineContext, QueryEngineFactory, QueryEngineRef,
+2 -2
View File
@@ -166,8 +166,8 @@ serde_json.workspace = true
session = { workspace = true, features = ["testing"] }
table.workspace = true
tempfile = "3.0.0"
tokio-postgres = "0.7"
tokio-postgres-rustls = "0.12"
tokio-postgres.workspace = true
tokio-postgres-rustls = "0.14"
[target.'cfg(unix)'.dev-dependencies]
pprof = { version = "0.14", features = ["criterion", "flamegraph"] }
+29
View File
@@ -40,6 +40,7 @@ use pgwire::error::{ErrorInfo, PgWireError, PgWireResult};
use pgwire::messages::PgWireBackendMessage;
use pgwire::messages::copy::CopyData;
use pgwire::messages::data::DataRow;
use query::dist_analyze_output_schema;
use query::planner::DfLogicalPlanner;
use query::query_engine::DescribeResult;
use session::Session;
@@ -554,6 +555,16 @@ fn describe_fields(
session: &Arc<Session>,
) -> PgWireResult<Vec<FieldInfo>> {
match sql_plan {
// EXPLAIN ANALYZE: at execution time the physical plan is replaced with
// GreptimeDB's `DistAnalyzeExec` (see `optimize_physical_plan`), whose
// output schema (stage/node/plan) differs from the DataFusion `Analyze`
// logical plan schema (plan_type/plan). Describe with the schema the
// client will actually receive so the DataRow field count matches.
SqlPlan::Plan(LogicalPlan::Analyze(_), _) => {
let schema: Schema =
Schema::try_from(dist_analyze_output_schema()).map_err(convert_err)?;
schema_to_pg(&schema, format, None).map_err(convert_err)
}
// query
SqlPlan::Plan(plan, _) if !matches!(plan, LogicalPlan::Dml(_) | LogicalPlan::Ddl(_)) => {
let schema: Schema = plan.schema().clone().try_into().map_err(convert_err)?;
@@ -679,6 +690,24 @@ fn describe_fields(
Ok(vec![])
}
}
// FETCH cursor: return the cursor's schema so the RowDescription
// matches the DataRow field count sent during Execute.
SqlPlan::Statement(Statement::FetchCursor(fetch), _) => {
let cursor_name = fetch.cursor_name.to_string();
match session.get_cursor(&cursor_name) {
Some(cursor) => {
// `cursor.schema()` is the GreptimeDB `SchemaRef` captured
// when the cursor was declared; `schema_to_pg` accepts it
// directly (no DataFusion -> GreptimeDB conversion needed).
schema_to_pg(&cursor.schema(), format, None).map_err(convert_err)
}
None => {
// Cursor not found (e.g. DECLARE hasn't executed yet in
// an extended-protocol batch). Return NoData as a fallback.
Ok(vec![])
}
}
}
_ => {
// NoData
Ok(vec![])
+9
View File
@@ -111,6 +111,15 @@ impl Session {
.into()
}
/// Returns the cursor with the given name, if it exists.
///
/// Cursors are stored in the session's mutable inner data and shared across
/// all query contexts created from this session.
pub fn get_cursor(&self, name: &str) -> Option<Arc<RecordBatchStreamCursor>> {
let guard = self.mutable_inner.read().unwrap();
guard.cursors.get(name).cloned()
}
pub fn conn_info(&self) -> &ConnInfo {
&self.conn_info
}