From 20619273eede955f62f252cc2e01e3038c27b0bd Mon Sep 17 00:00:00 2001 From: Ning Sun Date: Thu, 24 Sep 2026 08:37:43 +0000 Subject: [PATCH] chore(toolchain): switch to stable Rust 1.96.1 and remove all nightly feature gates (#9303) * chore(toolchain): switch to stable Rust 1.96.1 and remove all nightly feature gates Move the workspace from the pinned nightly-2026-03-21 to stable 1.96.1 and drop all 23 '#![feature]' gates across 13 crates, rewriting the still-unstable API usages with stable equivalents: - try_blocks: closures / an async block (table, query, common-function, servers) - duration_constructors: Duration::from_secs(n * 86400) / (n * 60) - iterator_try_collect: collect::, _>>() - box_patterns: as_deref() + matches! chains (sql) - error_iter: error_chain_root() source-chain walker (common-error); sources() includes the error itself, so the walker never panics - int_roundings: div_floor -> div_euclid (equal for positive divisors) - iter_partition_in_place: stable sort_by_key partition helper (index) - hash_set_entry: HashSet::insert bool / contains+insert - trait_alias: lifetime-parameterized dyn FnOnce type aliases (puffin) - string_from_utf8_lossy_owned: from_utf8_lossy(&v).into_owned() - never_type: Infallible (common-recordbatch) - debug_closure_helpers: closure-backed DebugFmt newtype (mito2) - binary_heap_pop_if: peek().is_some_and() + pop() - exclusive_wrapper: drop Exclusive; C: Send + Unpin already in bounds - stmt_expr_attributes: stale gate, no usages Also fix release-dev-builder-images.yaml, which parsed rust-toolchain.toml with a date-only regex and would produce empty image versions with a stable channel; it now extracts the full channel token. Dev-builder images verified against stable 1.96.1 (image build, default-toolchain behavior, binstall/nextest, riscv64 and android targets, in-image cargo check). Validated on 1.96.1: cargo check --workspace --all-targets, clippy --workspace --all-targets --all-features -D warnings, cargo fmt --check, and nextest on all 13 affected crates (4586 passed). Part of #9289. Depends on #9298 (fuzz nightly quarantine) merging first. Signed-off-by: Ning Sun * chore: update flake checksum * chore: use wild for linker in flake --------- Signed-off-by: Ning Sun --- .../workflows/release-dev-builder-images.yaml | 5 +- flake.nix | 4 +- rust-toolchain.toml | 2 +- src/common/error/src/ext.rs | 18 +++- src/common/error/src/lib.rs | 2 - src/common/event-recorder/src/lib.rs | 2 - src/common/event-recorder/src/recorder.rs | 4 +- src/common/function/src/lib.rs | 2 - src/common/function/src/scalars/matches.rs | 5 +- src/common/recordbatch/src/adapter.rs | 2 +- src/common/recordbatch/src/lib.rs | 2 - src/common/telemetry/src/lib.rs | 2 - src/common/telemetry/src/logging.rs | 2 +- src/common/wal/src/config.rs | 2 +- src/common/wal/src/config/kafka/common.rs | 2 +- src/index/src/bloom_filter/creator.rs | 8 +- src/index/src/inverted_index/search.rs | 10 ++ .../search/fst_apply/keys_apply.rs | 12 +-- .../search/index_apply/predicates_apply.rs | 7 +- src/index/src/lib.rs | 3 - src/meta-srv/src/gc/options.rs | 10 +- .../src/handler/persist_stats_handler.rs | 4 +- src/meta-srv/src/lib.rs | 3 - src/meta-srv/src/metasrv.rs | 2 +- src/meta-srv/src/metasrv/builder.rs | 2 +- src/meta-srv/src/procedure/repartition.rs | 2 +- .../src/procedure/wal_prune/manager.rs | 17 ++-- src/mito2/src/compaction/run.rs | 6 +- src/mito2/src/lib.rs | 4 - src/mito2/src/memtable/time_partition.rs | 2 +- src/mito2/src/sst/file.rs | 50 ++++++---- .../src/etl/transform/transformer/greptime.rs | 6 +- src/pipeline/src/lib.rs | 2 - src/pipeline/src/tablesuffix.rs | 2 +- src/puffin/src/lib.rs | 2 - src/puffin/src/puffin_manager/stager.rs | 8 +- .../puffin_manager/stager/bounded_stager.rs | 14 +-- src/query/src/lib.rs | 6 -- src/query/src/log_query/planner.rs | 16 ++-- src/query/src/optimizer/windowed_sort.rs | 8 +- src/query/src/planner.rs | 8 +- src/query/src/range_select/plan.rs | 4 +- src/query/src/window_sort.rs | 7 +- .../src/http/result/prometheus_resp.rs | 7 +- src/servers/src/lib.rs | 3 - src/servers/src/postgres/auth_handler.rs | 7 +- src/sql/src/lib.rs | 2 - src/sql/src/statements/insert.rs | 95 ++++++++++--------- src/table/src/lib.rs | 2 - src/table/src/table/scan.rs | 8 +- 50 files changed, 209 insertions(+), 196 deletions(-) diff --git a/.github/workflows/release-dev-builder-images.yaml b/.github/workflows/release-dev-builder-images.yaml index bebd5ed61e7..41325f61ab9 100644 --- a/.github/workflows/release-dev-builder-images.yaml +++ b/.github/workflows/release-dev-builder-images.yaml @@ -59,7 +59,10 @@ jobs: commitShortSHA=`echo ${{ github.sha }} | cut -c1-8` buildTime=`date +%Y%m%d%H%M%S` BUILD_VERSION="$commitShortSHA-$buildTime" - RUST_TOOLCHAIN_VERSION=$(cat rust-toolchain.toml | grep -Eo '[0-9]{4}-[0-9]{2}-[0-9]{2}') + # Extract the full channel string (e.g. "1.96.1" or + # "nightly-2026-03-21") so the image version always carries the + # toolchain it was built with. + RUST_TOOLCHAIN_VERSION=$(grep -E '^channel' rust-toolchain.toml | cut -d'"' -f2) IMAGE_VERSION="${RUST_TOOLCHAIN_VERSION}-${BUILD_VERSION}" echo "VERSION=${IMAGE_VERSION}" >> $GITHUB_ENV echo "version=$IMAGE_VERSION" >> $GITHUB_OUTPUT diff --git a/flake.nix b/flake.nix index 2d95d8dcfd6..6adb69678b2 100644 --- a/flake.nix +++ b/flake.nix @@ -22,7 +22,7 @@ lib = nixpkgs.lib; rustToolchain = fenix.packages.${system}.fromToolchainName { name = (lib.importTOML ./rust-toolchain.toml).toolchain.channel; - sha256 = "sha256-rboGKQLH4eDuiY01SINOqmXUFUNr9F4awoFZGzib17o="; + sha256 = "sha256-h+t2xTBz5yt2YIO+1VMIIGlCU7gyp2LYOFvaV1nwOXU="; }; in { @@ -34,7 +34,7 @@ gcc protobuf gnumake - mold + wild (rustToolchain.withComponents [ "cargo" "clippy" diff --git a/rust-toolchain.toml b/rust-toolchain.toml index d16edecca80..2d45363a5be 100644 --- a/rust-toolchain.toml +++ b/rust-toolchain.toml @@ -1,2 +1,2 @@ [toolchain] -channel = "nightly-2026-03-21" +channel = "1.96.1" diff --git a/src/common/error/src/ext.rs b/src/common/error/src/ext.rs index 59895551156..0c267dc15b2 100644 --- a/src/common/error/src/ext.rs +++ b/src/common/error/src/ext.rs @@ -21,6 +21,20 @@ use std::sync::Arc; use serde::{Deserialize, Deserializer, Serializer}; use snafu::{FromString, Snafu}; +/// Returns the root cause of an error's source chain, i.e. the last error +/// reachable via [`std::error::Error::source`]. For an error without any +/// source the error itself is returned. +/// +/// Mirrors `err.sources().last().unwrap()` (unstable `error_iter`, which +/// yields the error itself followed by its sources). +fn error_chain_root(err: &dyn std::error::Error) -> &dyn std::error::Error { + let mut root = err; + while let Some(source) = root.source() { + root = source; + } + root +} + use crate::status_code::StatusCode; /// Describes whether an error instance is safe and useful to retry. @@ -148,7 +162,7 @@ pub trait ErrorExt: StackError { _ => { let error = self.last(); if let Some(external_error) = error.source() { - let external_root = external_error.sources().last().unwrap(); + let external_root = error_chain_root(external_error); if error.transparent() { format!("{external_root}") @@ -169,7 +183,7 @@ pub trait ErrorExt: StackError { { let error = self.last(); if let Some(external_error) = error.source() { - let external_root = external_error.sources().last().unwrap(); + let external_root = error_chain_root(external_error); Some(external_root) } else { None diff --git a/src/common/error/src/lib.rs b/src/common/error/src/lib.rs index eeedc637133..29ae5c3b1cc 100644 --- a/src/common/error/src/lib.rs +++ b/src/common/error/src/lib.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. -#![feature(error_iter)] - pub mod ext; pub mod mock; pub mod status_code; diff --git a/src/common/event-recorder/src/lib.rs b/src/common/event-recorder/src/lib.rs index 7ee71722229..0a5d051a01e 100644 --- a/src/common/event-recorder/src/lib.rs +++ b/src/common/event-recorder/src/lib.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. -#![feature(duration_constructors)] - pub mod context; pub mod error; pub mod event_table; diff --git a/src/common/event-recorder/src/recorder.rs b/src/common/event-recorder/src/recorder.rs index 3ea5c7e55a8..040b2fbcbb3 100644 --- a/src/common/event-recorder/src/recorder.rs +++ b/src/common/event-recorder/src/recorder.rs @@ -108,9 +108,9 @@ fn event_type_filter_is_all(event_types: &EventTypeFilterRef) -> bool { /// The time interval for flushing batched events to the event handler. pub const DEFAULT_FLUSH_INTERVAL_SECONDS: Duration = Duration::from_secs(5); /// The default TTL(90 days) for the events table. -const DEFAULT_EVENTS_TABLE_TTL: Duration = Duration::from_days(90); +const DEFAULT_EVENTS_TABLE_TTL: Duration = Duration::from_secs(90 * 86400); /// The default compaction time window for the events table. -pub const DEFAULT_COMPACTION_TIME_WINDOW: Duration = Duration::from_days(1); +pub const DEFAULT_COMPACTION_TIME_WINDOW: Duration = Duration::from_secs(86400); // The capacity of the tokio channel for transmitting events to background processor. const DEFAULT_CHANNEL_SIZE: usize = 2048; // The size of the buffer for batching events before flushing to event handler. diff --git a/src/common/function/src/lib.rs b/src/common/function/src/lib.rs index 336f8d489ca..7309dba981d 100644 --- a/src/common/function/src/lib.rs +++ b/src/common/function/src/lib.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. -#![feature(try_blocks)] - mod admin; mod flush_flow; mod macros; diff --git a/src/common/function/src/scalars/matches.rs b/src/common/function/src/scalars/matches.rs index 14889957ee9..109ad1daeb2 100644 --- a/src/common/function/src/scalars/matches.rs +++ b/src/common/function/src/scalars/matches.rs @@ -1119,11 +1119,12 @@ mod test { ]; for (query, expected) in cases { - let result: Result<()> = try { + let result: Result<()> = (|| { let parser = ParserContext { stack: vec![] }; let ast = parser.parse_pattern(query)?; let _ast = ast.transform_ast()?; - }; + Ok(()) + })(); assert!(result.is_err(), "{query}"); let actual_error = result.unwrap_err().to_string(); diff --git a/src/common/recordbatch/src/adapter.rs b/src/common/recordbatch/src/adapter.rs index a7f6e4a10ac..f8d645bd3e1 100644 --- a/src/common/recordbatch/src/adapter.rs +++ b/src/common/recordbatch/src/adapter.rs @@ -570,7 +570,7 @@ impl MetricCollector { } impl ExecutionPlanVisitor for MetricCollector { - type Error = !; + type Error = std::convert::Infallible; fn pre_visit(&mut self, plan: &dyn ExecutionPlan) -> std::result::Result { // skip if no metric available diff --git a/src/common/recordbatch/src/lib.rs b/src/common/recordbatch/src/lib.rs index 4fb37cd5b11..0ccb8a4c28e 100644 --- a/src/common/recordbatch/src/lib.rs +++ b/src/common/recordbatch/src/lib.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. -#![feature(never_type)] - pub mod adapter; pub mod cursor; pub mod error; diff --git a/src/common/telemetry/src/lib.rs b/src/common/telemetry/src/lib.rs index 26bf5d53b3c..6daf2662abe 100644 --- a/src/common/telemetry/src/lib.rs +++ b/src/common/telemetry/src/lib.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. -#![feature(duration_constructors)] - pub mod logging; mod macros; pub mod metric; diff --git a/src/common/telemetry/src/logging.rs b/src/common/telemetry/src/logging.rs index a9b50930786..dd99e901f75 100644 --- a/src/common/telemetry/src/logging.rs +++ b/src/common/telemetry/src/logging.rs @@ -348,7 +348,7 @@ impl Default for SlowQueryOptions { record_type: SlowQueriesRecordType::SystemTable, threshold: Duration::from_secs(30), sample_ratio: 1.0, - ttl: Duration::from_days(90), + ttl: Duration::from_secs(90 * 86400), } } } diff --git a/src/common/wal/src/config.rs b/src/common/wal/src/config.rs index 343df193ab4..f79a28607cc 100644 --- a/src/common/wal/src/config.rs +++ b/src/common/wal/src/config.rs @@ -234,7 +234,7 @@ mod tests { create_topic_timeout: Duration::from_secs(30), }, auto_create_topics: true, - auto_prune_interval: Duration::from_mins(30), + auto_prune_interval: Duration::from_secs(30 * 60), auto_prune_logical_delete: false, auto_prune_parallelism: 10, flush_trigger_size: ReadableSize::mb(512), diff --git a/src/common/wal/src/config/kafka/common.rs b/src/common/wal/src/config/kafka/common.rs index cf632fbdce3..daa5607e476 100644 --- a/src/common/wal/src/config/kafka/common.rs +++ b/src/common/wal/src/config/kafka/common.rs @@ -38,7 +38,7 @@ pub const DEFAULT_BACKOFF_CONFIG: BackoffConfig = BackoffConfig { }; /// Default interval for auto WAL pruning. -pub const DEFAULT_AUTO_PRUNE_INTERVAL: Duration = Duration::from_mins(30); +pub const DEFAULT_AUTO_PRUNE_INTERVAL: Duration = Duration::from_secs(30 * 60); /// Default mode for auto WAL pruning. pub const DEFAULT_AUTO_PRUNE_LOGICAL_DELETE: bool = false; /// Default limit for concurrent auto pruning tasks. diff --git a/src/index/src/bloom_filter/creator.rs b/src/index/src/bloom_filter/creator.rs index df23ca2352a..07f4f977caf 100644 --- a/src/index/src/bloom_filter/creator.rs +++ b/src/index/src/bloom_filter/creator.rs @@ -161,10 +161,10 @@ impl BloomFilterCreator { if let Some(elem) = elem { let old_len = self.cur_seg_distinct_elems.len(); - // A borrowed entry lookup avoids hashing new values twice and only - // allocates when the value is absent from the current segment. - self.cur_seg_distinct_elems - .get_or_insert_with(elem, <[u8]>::to_vec); + // Only allocate when the value is absent from the current segment. + if !self.cur_seg_distinct_elems.contains(elem) { + self.cur_seg_distinct_elems.insert(elem.to_vec()); + } if self.cur_seg_distinct_elems.len() != old_len { self.cur_seg_distinct_elems_mem_usage += elem.len(); self.global_memory_usage diff --git a/src/index/src/inverted_index/search.rs b/src/index/src/inverted_index/search.rs index 8e28440c7e2..581135c2d4e 100644 --- a/src/index/src/inverted_index/search.rs +++ b/src/index/src/inverted_index/search.rs @@ -16,3 +16,13 @@ pub mod fst_apply; pub mod fst_values_mapper; pub mod index_apply; pub mod predicate; + +/// Partitions `slice` in place so that elements matching `pred` come first, +/// preserving the original relative order within each group. Returns the +/// number of matching elements. +/// +/// Stable replacement for the unstable `Iterator::partition_in_place`. +pub(crate) fn partition_in_place(slice: &mut [T], mut pred: impl FnMut(&T) -> bool) -> usize { + slice.sort_by_key(|x| !pred(x)); + slice.iter().filter(|x| pred(x)).count() +} diff --git a/src/index/src/inverted_index/search/fst_apply/keys_apply.rs b/src/index/src/inverted_index/search/fst_apply/keys_apply.rs index d565749452d..b68f58cdfd3 100644 --- a/src/index/src/inverted_index/search/fst_apply/keys_apply.rs +++ b/src/index/src/inverted_index/search/fst_apply/keys_apply.rs @@ -68,16 +68,16 @@ impl KeysFstApplier { } fn split_at_in_lists(predicates: &mut [Predicate]) -> (&mut [Predicate], &mut [Predicate]) { - let in_list_index = predicates - .iter_mut() - .partition_in_place(|p| matches!(p, Predicate::InList(_))); + let in_list_index = crate::inverted_index::search::partition_in_place(predicates, |p| { + matches!(p, Predicate::InList(_)) + }); predicates.split_at_mut(in_list_index) } fn split_at_ranges(predicates: &mut [Predicate]) -> (&mut [Predicate], &mut [Predicate]) { - let range_index = predicates - .iter_mut() - .partition_in_place(|p| matches!(p, Predicate::Range(_))); + let range_index = crate::inverted_index::search::partition_in_place(predicates, |p| { + matches!(p, Predicate::Range(_)) + }); predicates.split_at_mut(range_index) } diff --git a/src/index/src/inverted_index/search/index_apply/predicates_apply.rs b/src/index/src/inverted_index/search/index_apply/predicates_apply.rs index eaea8cfb95c..5fcc962ca7d 100644 --- a/src/index/src/inverted_index/search/index_apply/predicates_apply.rs +++ b/src/index/src/inverted_index/search/index_apply/predicates_apply.rs @@ -127,9 +127,10 @@ impl PredicatesIndexApplier { let mut fst_appliers = Vec::with_capacity(predicates.len()); // InList predicates are applied first to benefit from higher selectivity. - let in_list_index = predicates - .iter_mut() - .partition_in_place(|(_, ps)| ps.iter().any(|p| matches!(p, Predicate::InList(_)))); + let in_list_index = + crate::inverted_index::search::partition_in_place(&mut predicates, |(_, ps)| { + ps.iter().any(|p| matches!(p, Predicate::InList(_))) + }); let mut iter = predicates.into_iter(); for _ in 0..in_list_index { let (column_name, predicates) = iter.next().unwrap(); diff --git a/src/index/src/lib.rs b/src/index/src/lib.rs index 2c1ceb4e3f6..571866ade90 100644 --- a/src/index/src/lib.rs +++ b/src/index/src/lib.rs @@ -12,9 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. -#![feature(iter_partition_in_place)] -#![feature(hash_set_entry)] - pub mod bitmap; pub mod bloom_filter; pub mod error; diff --git a/src/meta-srv/src/gc/options.rs b/src/meta-srv/src/gc/options.rs index fb43bd6dce1..298fb30bfd8 100644 --- a/src/meta-srv/src/gc/options.rs +++ b/src/meta-srv/src/gc/options.rs @@ -38,7 +38,7 @@ impl Default for SoftDropGcOptions { fn default() -> Self { Self { enable: false, - retention: Duration::from_days(7), + retention: Duration::from_secs(7 * 86400), } } } @@ -242,7 +242,7 @@ mod tests { assert!(!options.experimental_soft_drop.enable); assert_eq!( - Duration::from_days(7), + Duration::from_secs(7 * 86400), options.experimental_soft_drop.retention ); } @@ -254,7 +254,7 @@ mod tests { enable: true, experimental_soft_drop: SoftDropGcOptions { enable: true, - retention: Duration::from_days(1), + retention: Duration::from_secs(86400), }, ..Default::default() }; @@ -269,7 +269,7 @@ mod tests { enable: true, experimental_soft_drop: SoftDropGcOptions { enable: true, - retention: Duration::from_days(1), + retention: Duration::from_secs(86400), }, ..Default::default() }; @@ -284,7 +284,7 @@ mod tests { let options = GcSchedulerOptions { experimental_soft_drop: SoftDropGcOptions { enable: true, - retention: Duration::from_days(1), + retention: Duration::from_secs(86400), }, ..Default::default() }; diff --git a/src/meta-srv/src/handler/persist_stats_handler.rs b/src/meta-srv/src/handler/persist_stats_handler.rs index 04d63e26e2f..472072aa42b 100644 --- a/src/meta-srv/src/handler/persist_stats_handler.rs +++ b/src/meta-srv/src/handler/persist_stats_handler.rs @@ -154,9 +154,9 @@ fn align_ts(ts: i64, interval: Duration) -> i64 { impl PersistStatsHandler { /// Creates a new [`PersistStatsHandler`]. pub fn new(inserter: Box, mut persist_interval: Duration) -> Self { - if persist_interval < Duration::from_mins(10) { + if persist_interval < Duration::from_secs(10 * 60) { warn!("persist_interval is less than 10 minutes, set to 10 minutes"); - persist_interval = Duration::from_mins(10); + persist_interval = Duration::from_secs(10 * 60); } Self { diff --git a/src/meta-srv/src/lib.rs b/src/meta-srv/src/lib.rs index 88e8466be39..948cab240d4 100644 --- a/src/meta-srv/src/lib.rs +++ b/src/meta-srv/src/lib.rs @@ -12,9 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. -#![feature(hash_set_entry)] -#![feature(duration_constructors)] - pub mod bootstrap; pub mod cache_invalidator; pub mod cluster; diff --git a/src/meta-srv/src/metasrv.rs b/src/meta-srv/src/metasrv.rs index 65a079bec17..86aee9f8b82 100644 --- a/src/meta-srv/src/metasrv.rs +++ b/src/meta-srv/src/metasrv.rs @@ -122,7 +122,7 @@ impl Default for StatsPersistenceOptions { fn default() -> Self { Self { ttl: Duration::ZERO, - interval: Duration::from_mins(10), + interval: Duration::from_secs(10 * 60), } } } diff --git a/src/meta-srv/src/metasrv/builder.rs b/src/meta-srv/src/metasrv/builder.rs index cbc859716a4..57e7deb3db6 100644 --- a/src/meta-srv/src/metasrv/builder.rs +++ b/src/meta-srv/src/metasrv/builder.rs @@ -92,7 +92,7 @@ use crate::utils::database::DatabaseOperator; use crate::utils::insert_forwarder::InsertForwarder; /// The time window for twcs compaction of the region stats table. -const REGION_STATS_TABLE_TWCS_COMPACTION_TIME_WINDOW: Duration = Duration::from_days(1); +const REGION_STATS_TABLE_TWCS_COMPACTION_TIME_WINDOW: Duration = Duration::from_secs(86400); // TODO(fys): try use derive_builder macro pub struct MetasrvBuilder { diff --git a/src/meta-srv/src/procedure/repartition.rs b/src/meta-srv/src/procedure/repartition.rs index f42d8aacd0c..008c0ce05bb 100644 --- a/src/meta-srv/src/procedure/repartition.rs +++ b/src/meta-srv/src/procedure/repartition.rs @@ -107,7 +107,7 @@ pub struct PersistentContext { } fn default_timeout() -> Duration { - Duration::from_mins(2) + Duration::from_secs(2 * 60) } impl PersistentContext { diff --git a/src/meta-srv/src/procedure/wal_prune/manager.rs b/src/meta-srv/src/procedure/wal_prune/manager.rs index e398be93e45..4de21086906 100644 --- a/src/meta-srv/src/procedure/wal_prune/manager.rs +++ b/src/meta-srv/src/procedure/wal_prune/manager.rs @@ -13,7 +13,6 @@ // limitations under the License. use std::collections::HashSet; -use std::collections::hash_set::Entry; use std::fmt::{Debug, Formatter}; use std::sync::{Arc, RwLock}; @@ -46,15 +45,13 @@ impl WalPruneProcedureTracker { /// consume acquire a semaphore permit for the given topic name. pub fn insert_running_procedure(&self, topic_name: String) -> Option { let mut running_procedures = self.running_procedures.write().unwrap(); - match running_procedures.entry(topic_name.clone()) { - Entry::Occupied(_) => None, - Entry::Vacant(entry) => { - entry.insert(); - Some(WalPruneProcedureGuard { - topic_name, - running_procedures: self.running_procedures.clone(), - }) - } + if running_procedures.insert(topic_name.clone()) { + Some(WalPruneProcedureGuard { + topic_name, + running_procedures: self.running_procedures.clone(), + }) + } else { + None } } diff --git a/src/mito2/src/compaction/run.rs b/src/mito2/src/compaction/run.rs index 2ad9f2f1eba..664005b9af3 100644 --- a/src/mito2/src/compaction/run.rs +++ b/src/mito2/src/compaction/run.rs @@ -475,7 +475,11 @@ where for item in items { let (start, _) = item.range(); - while let Some(run) = runs_sort_by_end.pop_if(|x| x.run.end.unwrap() <= start) { + while runs_sort_by_end + .peek() + .is_some_and(|x| x.run.end.unwrap() <= start) + { + let run = runs_sort_by_end.pop().unwrap(); runs_sort_by_index.push(Wrapper(run)); } diff --git a/src/mito2/src/lib.rs b/src/mito2/src/lib.rs index 4f4409afd8a..dcbfe60dea5 100644 --- a/src/mito2/src/lib.rs +++ b/src/mito2/src/lib.rs @@ -16,10 +16,6 @@ //! //! Mito is the a region engine to store timeseries data. -#![feature(debug_closure_helpers)] -#![feature(duration_constructors)] -#![feature(binary_heap_pop_if)] - #[cfg(any(test, feature = "test"))] #[cfg_attr(feature = "test", allow(unused))] pub mod test_util; diff --git a/src/mito2/src/memtable/time_partition.rs b/src/mito2/src/memtable/time_partition.rs index a6b27cdd0f3..0ca88ca9a70 100644 --- a/src/mito2/src/memtable/time_partition.rs +++ b/src/mito2/src/memtable/time_partition.rs @@ -43,7 +43,7 @@ use crate::memtable::{KeyValues, MemtableBuilderRef, MemtableId, MemtableRef}; use crate::sst::{FlatSchemaOptions, to_flat_sst_arrow_schema}; /// Initial time window if not specified. -const INITIAL_TIME_WINDOW: Duration = Duration::from_days(1); +const INITIAL_TIME_WINDOW: Duration = Duration::from_secs(86400); /// A partition holds rows with timestamps between `[min, max)`. #[derive(Debug, Clone)] diff --git a/src/mito2/src/sst/file.rs b/src/mito2/src/sst/file.rs index c18d6195f19..87684586fbf 100644 --- a/src/mito2/src/sst/file.rs +++ b/src/mito2/src/sst/file.rs @@ -298,20 +298,33 @@ fn is_false(value: &bool) -> bool { !*value } +/// Formats a debug field with a custom closure, as a stable replacement for +/// the unstable `DebugStruct::field_with`. +struct DebugFmt) -> fmt::Result>(F); + +impl) -> fmt::Result> std::fmt::Debug for DebugFmt { + fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { + (self.0)(f) + } +} + impl Debug for FileMeta { fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { let mut debug_struct = f.debug_struct("FileMeta"); debug_struct .field("region_id", &self.region_id) - .field_with("file_id", |f| write!(f, "{} ", self.file_id)) - .field_with("time_range", |f| { - write!( - f, - "({}, {}) ", - self.time_range.0.to_iso8601_string(), - self.time_range.1.to_iso8601_string() - ) - }) + .field("file_id", &DebugFmt(|f| write!(f, "{} ", self.file_id))) + .field( + "time_range", + &DebugFmt(|f| { + write!( + f, + "({}, {}) ", + self.time_range.0.to_iso8601_string(), + self.time_range.1.to_iso8601_string() + ) + }), + ) .field("level", &self.level) .field("file_size", &ReadableSize(self.file_size)) .field( @@ -327,14 +340,17 @@ impl Debug for FileMeta { debug_struct .field("num_rows", &self.num_rows) .field("num_row_groups", &self.num_row_groups) - .field_with("sequence", |f| match self.sequence { - None => { - write!(f, "None") - } - Some(seq) => { - write!(f, "{}", seq) - } - }) + .field( + "sequence", + &DebugFmt(|f| match self.sequence { + None => { + write!(f, "None") + } + Some(seq) => { + write!(f, "{}", seq) + } + }), + ) .field("partition_expr", &self.partition_expr) .field("num_series", &self.num_series); if self.primary_key_min.is_some() || self.primary_key_max.is_some() { diff --git a/src/pipeline/src/etl/transform/transformer/greptime.rs b/src/pipeline/src/etl/transform/transformer/greptime.rs index bbe3949a540..eb25071c91c 100644 --- a/src/pipeline/src/etl/transform/transformer/greptime.rs +++ b/src/pipeline/src/etl/transform/transformer/greptime.rs @@ -741,9 +741,9 @@ fn resolve_value( &ConcreteDataType::string_datatype(), schema_info, )?; - Some(ValueData::StringValue(String::from_utf8_lossy_owned( - v.to_vec(), - ))) + Some(ValueData::StringValue( + String::from_utf8_lossy(&v).into_owned(), + )) } VrlValue::Regex(v) => { diff --git a/src/pipeline/src/lib.rs b/src/pipeline/src/lib.rs index 5d1acb7462e..15954b18f98 100644 --- a/src/pipeline/src/lib.rs +++ b/src/pipeline/src/lib.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. -#![feature(string_from_utf8_lossy_owned)] - mod dispatcher; pub mod error; mod etl; diff --git a/src/pipeline/src/tablesuffix.rs b/src/pipeline/src/tablesuffix.rs index 01ce4202cfc..4a72f49bbc6 100644 --- a/src/pipeline/src/tablesuffix.rs +++ b/src/pipeline/src/tablesuffix.rs @@ -56,7 +56,7 @@ impl TableSuffixTemplate { let v = val.get(key.as_str())?; match v { VrlValue::Integer(v) => Some(v.to_string()), - VrlValue::Bytes(v) => Some(String::from_utf8_lossy_owned(v.to_vec())), + VrlValue::Bytes(v) => Some(String::from_utf8_lossy(v).into_owned()), _ => None, } }) diff --git a/src/puffin/src/lib.rs b/src/puffin/src/lib.rs index 96a8421f98b..2be956e43dc 100644 --- a/src/puffin/src/lib.rs +++ b/src/puffin/src/lib.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. -#![feature(trait_alias)] - pub mod blob_metadata; pub mod error; pub mod file_format; diff --git a/src/puffin/src/puffin_manager/stager.rs b/src/puffin/src/puffin_manager/stager.rs index 512e94f4e85..8a0adb34bca 100644 --- a/src/puffin/src/puffin_manager/stager.rs +++ b/src/puffin/src/puffin_manager/stager.rs @@ -43,13 +43,13 @@ pub type DirWriterProviderRef = Box; /// /// `Stager` will provide a `BoxWriter` that the caller of `get_blob` /// can use to write the blob into the staging area. -pub trait InitBlobFn = FnOnce(BoxWriter) -> WriteResult; +pub type InitBlobFn<'a> = dyn FnOnce(BoxWriter) -> WriteResult + Send + Sync + 'a; /// Function that initializes a directory. /// /// `Stager` will provide a `DirWriterProvider` that the caller of `get_dir` /// can use to write files inside the directory into the staging area. -pub trait InitDirFn = FnOnce(DirWriterProviderRef) -> WriteResult; +pub type InitDirFn<'a> = dyn FnOnce(DirWriterProviderRef) -> WriteResult + Send + Sync + 'a; /// `Stager` manages the staging area for the puffin files. #[async_trait] @@ -67,7 +67,7 @@ pub trait Stager: Send + Sync { &self, handle: &Self::FileHandle, key: &str, - init_factory: Box, + init_factory: Box>, ) -> Result; /// Retrieves a directory, initializing it if necessary using the provided `init_fn`. @@ -79,7 +79,7 @@ pub trait Stager: Send + Sync { &self, handle: &Self::FileHandle, key: &str, - init_fn: Box, + init_fn: Box>, ) -> Result<(Self::Dir, DirMetrics)>; /// Stores a directory in the staging area. diff --git a/src/puffin/src/puffin_manager/stager/bounded_stager.rs b/src/puffin/src/puffin_manager/stager/bounded_stager.rs index dfb9285452b..e0952b2892e 100644 --- a/src/puffin/src/puffin_manager/stager/bounded_stager.rs +++ b/src/puffin/src/puffin_manager/stager/bounded_stager.rs @@ -146,7 +146,7 @@ impl Stager for BoundedStager { &self, handle: &Self::FileHandle, key: &str, - init_fn: Box, + init_fn: Box>, ) -> Result { let handle_str = handle.to_string(); let cache_key = Self::encode_cache_key(&handle_str, key); @@ -202,7 +202,7 @@ impl Stager for BoundedStager { &self, handle: &Self::FileHandle, key: &str, - init_fn: Box, + init_fn: Box>, ) -> Result<(Self::Dir, DirMetrics)> { let handle_str = handle.to_string(); @@ -332,10 +332,7 @@ impl BoundedStager { BASE64_URL_SAFE.encode(hash) } - async fn write_blob( - target_path: &PathBuf, - init_fn: Box, - ) -> Result { + async fn write_blob(target_path: &PathBuf, init_fn: Box>) -> Result { // To guarantee the atomicity of writing the file, we need to write // the file to a temporary file first... let tmp_path = target_path.with_extension(TMP_EXTENSION); @@ -354,10 +351,7 @@ impl BoundedStager { Ok(size) } - async fn write_dir( - target_path: &PathBuf, - init_fn: Box, - ) -> Result { + async fn write_dir(target_path: &PathBuf, init_fn: Box>) -> Result { // To guarantee the atomicity of writing the directory, we need to write // the directory to a temporary directory first... let tmp_base = target_path.with_extension(TMP_EXTENSION); diff --git a/src/query/src/lib.rs b/src/query/src/lib.rs index 4b2a2d5264d..aa83c8b8f0e 100644 --- a/src/query/src/lib.rs +++ b/src/query/src/lib.rs @@ -12,12 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. -#![feature(int_roundings)] -#![feature(try_blocks)] -#![feature(stmt_expr_attributes)] -#![feature(iterator_try_collect)] -#![feature(box_patterns)] - mod analyze; pub mod datafusion; pub mod dist_plan; diff --git a/src/query/src/log_query/planner.rs b/src/query/src/log_query/planner.rs index 5bc8d034a12..0855974241e 100644 --- a/src/query/src/log_query/planner.rs +++ b/src/query/src/log_query/planner.rs @@ -142,7 +142,7 @@ impl LogQueryPlanner { let exprs = filters .iter() .filter_map(|filter| self.build_filters(filter, schema).transpose()) - .try_collect::>()?; + .collect::, _>>()?; if exprs.is_empty() { Ok(None) } else { @@ -153,7 +153,7 @@ impl LogQueryPlanner { let exprs = filters .iter() .filter_map(|filter| self.build_filters(filter, schema).transpose()) - .try_collect::>()?; + .collect::, _>>()?; if exprs.is_empty() { Ok(None) } else { @@ -191,7 +191,7 @@ impl LogQueryPlanner { self.build_content_filter_with_expr(col_expr.clone(), filter, &df_schema) .transpose() }) - .try_collect::>()?; + .collect::, _>>()?; if filter_exprs.is_empty() { return Ok(Some(col_expr.is_true())); @@ -285,7 +285,7 @@ impl LogQueryPlanner { self.build_content_filter_with_expr(col_expr.clone(), filter, schema) .transpose() }) - .try_collect::>()?; + .collect::, _>>()?; if exprs.is_empty() { return Ok(None); @@ -326,19 +326,19 @@ impl LogQueryPlanner { let args = args .iter() .map(|expr| self.log_expr_to_df_expr(expr, schema)) - .try_collect::>()?; + .collect::, _>>()?; if let Some(alias) = alias { Ok(aggr_fn.call(args).alias(alias)) } else { Ok(aggr_fn.call(args)) } }) - .try_collect::>()?; + .collect::, _>>()?; let group_exprs = by .iter() .map(|expr| self.log_expr_to_df_expr(expr, schema)) - .try_collect::>()?; + .collect::, _>>()?; Ok((aggr_expr, group_exprs)) } @@ -380,7 +380,7 @@ impl LogQueryPlanner { let args = args .iter() .map(|expr| self.log_expr_to_df_expr(expr, schema)) - .try_collect::>()?; + .collect::, _>>()?; let func = self.session_state.scalar_functions().get(name).context( UnknownScalarFunctionSnafu { name: name.to_string(), diff --git a/src/query/src/optimizer/windowed_sort.rs b/src/query/src/optimizer/windowed_sort.rs index 3e4ccaf91c4..800bc103d38 100644 --- a/src/query/src/optimizer/windowed_sort.rs +++ b/src/query/src/optimizer/windowed_sort.rs @@ -199,12 +199,12 @@ fn fetch_partition_range(input: Arc) -> DataFusionResult, _>()?; + .collect::, _>>()?; let (min_val, max_val) = (typed_sorted_range[0].clone(), typed_sorted_range[1].clone()); @@ -2682,7 +2682,10 @@ mod test { let exec_stream = exec.execute(0, Arc::new(TaskContext::default())).unwrap(); let real_output = exec_stream.collect::>().await; - let real_output: Vec<_> = real_output.into_iter().try_collect().unwrap(); + let real_output: Vec<_> = real_output + .into_iter() + .collect::, _>>() + .unwrap(); real_output } } diff --git a/src/servers/src/http/result/prometheus_resp.rs b/src/servers/src/http/result/prometheus_resp.rs index 518d0f65e32..05956716d4b 100644 --- a/src/servers/src/http/result/prometheus_resp.rs +++ b/src/servers/src/http/result/prometheus_resp.rs @@ -231,7 +231,7 @@ impl PrometheusJsonResponse { ) -> Self { // Hold the collector while a streaming result is consumed. let collector = query_id.and_then(get_promql_annotation_collector); - let response: Result = try { + let response: Result = async { let result = result?; let mut resp = match result.data { @@ -259,8 +259,9 @@ impl PrometheusJsonResponse { resp.resp_metrics = re; } - resp - }; + Ok(resp) + } + .await; let result_type_string = result_type.to_string(); diff --git a/src/servers/src/lib.rs b/src/servers/src/lib.rs index 1106f89d3e9..2fed3d43de0 100644 --- a/src/servers/src/lib.rs +++ b/src/servers/src/lib.rs @@ -12,9 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. -#![feature(try_blocks)] -#![feature(exclusive_wrapper)] - use datafusion_expr::LogicalPlan; use sql::statements::statement::Statement; // Re-export for use in add_service! macro diff --git a/src/servers/src/postgres/auth_handler.rs b/src/servers/src/postgres/auth_handler.rs index 8a7b7272447..1fa075a8e17 100644 --- a/src/servers/src/postgres/auth_handler.rs +++ b/src/servers/src/postgres/auth_handler.rs @@ -13,7 +13,6 @@ // limitations under the License. use std::fmt::Debug; -use std::sync::Exclusive; use ::auth::{ BEARER_TOKEN_USER, Identity, Password, PgAuthInfo, PgScramSha256Verifier, UserInfoRef, @@ -252,7 +251,7 @@ impl StartupHandler for PostgresServerHandlerInner { auth::save_startup_parameters_to_metadata(client, startup); // check if db is valid - match resolve_db_info(Exclusive::new(client), self.query_handler.clone()).await? { + match resolve_db_info(client, self.query_handler.clone()).await? { DbResolution::Resolved(catalog, schema) => { let metadata = client.metadata_mut(); let _ = metadata.insert(super::METADATA_CATALOG.to_owned(), catalog); @@ -595,13 +594,13 @@ enum DbResolution { /// A function extracted to resolve lifetime and readability issues: async fn resolve_db_info( - client: Exclusive<&mut C>, + client: &mut C, query_handler: ServerSqlQueryHandlerRef, ) -> PgWireResult where C: ClientInfo + Unpin + Send, { - let db_ref = client.into_inner().metadata().get(super::METADATA_DATABASE); + let db_ref = client.metadata().get(super::METADATA_DATABASE); if let Some(db) = db_ref { let (catalog, schema) = parse_catalog_and_schema_from_db_string(db); if query_handler diff --git a/src/sql/src/lib.rs b/src/sql/src/lib.rs index b017f798f3f..e6bfe327902 100644 --- a/src/sql/src/lib.rs +++ b/src/sql/src/lib.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. -#![feature(box_patterns)] - pub mod ast; pub mod dialect; pub mod error; diff --git a/src/sql/src/statements/insert.rs b/src/sql/src/statements/insert.rs index 52318cddbf4..9c069bde43d 100644 --- a/src/sql/src/statements/insert.rs +++ b/src/sql/src/statements/insert.rs @@ -14,7 +14,7 @@ use serde::Serialize; use sqlparser::ast::{ - Insert as SpInsert, ObjectName, ObjectNamePart, Parens, Query, SetExpr, Statement, TableObject, + Insert as SpInsert, ObjectName, ObjectNamePart, Parens, SetExpr, Statement, TableObject, UnaryOperator, ValueWithSpan, Values, }; use sqlparser::parser::ParserError; @@ -70,14 +70,18 @@ impl Insert { /// Extracts the literal insert statement body if possible pub fn values_body(&self) -> Result>> { match &self.inner { - Statement::Insert(SpInsert { - source: - Some(box Query { - body: box SetExpr::Values(Values { rows, .. }), - .. - }), - .. - }) => sql_exprs_to_values(rows), + Statement::Insert(SpInsert { source, .. }) => { + let rows = source + .as_deref() + .and_then(|query| match query.body.as_ref() { + SetExpr::Values(Values { rows, .. }) => Some(rows), + _ => None, + }); + match rows { + Some(rows) => sql_exprs_to_values(rows), + None => unreachable!(), + } + } _ => unreachable!(), } } @@ -86,36 +90,37 @@ impl Insert { /// The rules is the same as function `values_body()`. pub fn can_extract_values(&self) -> bool { match &self.inner { - Statement::Insert(SpInsert { - source: - Some(box Query { - body: box SetExpr::Values(Values { rows, .. }), - .. - }), - .. - }) => rows.iter().all(|es| { - es.iter().all(|expr| match expr { - Expr::Value(_) => true, - Expr::Identifier(ident) => { - if ident.quote_style.is_none() { - ident.value.to_lowercase() == "default" - } else { - ident.quote_style == Some('"') - } - } - Expr::UnaryOp { op, expr } => { - matches!(op, UnaryOperator::Minus | UnaryOperator::Plus) - && matches!( - &**expr, - Expr::Value(ValueWithSpan { - value: Value::Number(_, _), - .. - }) - ) - } - _ => false, + Statement::Insert(SpInsert { source, .. }) => source + .as_deref() + .and_then(|query| match query.body.as_ref() { + SetExpr::Values(Values { rows, .. }) => Some(rows), + _ => None, }) - }), + .is_some_and(|rows| { + rows.iter().all(|es| { + es.iter().all(|expr| match expr { + Expr::Value(_) => true, + Expr::Identifier(ident) => { + if ident.quote_style.is_none() { + ident.value.to_lowercase() == "default" + } else { + ident.quote_style == Some('"') + } + } + Expr::UnaryOp { op, expr } => { + matches!(op, UnaryOperator::Minus | UnaryOperator::Plus) + && matches!( + &**expr, + Expr::Value(ValueWithSpan { + value: Value::Number(_, _), + .. + }) + ) + } + _ => false, + }) + }) + }), _ => false, } } @@ -124,7 +129,7 @@ impl Insert { pub fn has_non_values_query_source(&self) -> bool { match &self.inner { Statement::Insert(SpInsert { - source: Some(box query), + source: Some(query), .. }) => !matches!(&*query.body, SetExpr::Values(_)), _ => false, @@ -134,9 +139,9 @@ impl Insert { pub fn query_body(&self) -> Result> { Ok(match &self.inner { Statement::Insert(SpInsert { - source: Some(box query), + source: Some(query), .. - }) => Some(query.clone().try_into()?), + }) => Some(query.as_ref().clone().try_into()?), _ => None, }) } @@ -383,11 +388,9 @@ mod tests { let q = insert.query_body().unwrap().unwrap(); assert!(insert.has_non_values_query_source()); assert!(matches!( - q.inner, - Query { - body: box SetExpr::Select { .. }, - .. - } + &q.inner, + sqlparser::ast::Query { body, .. } + if matches!(body.as_ref(), SetExpr::Select { .. }) )); } _ => unreachable!(), diff --git a/src/table/src/lib.rs b/src/table/src/lib.rs index abc7dc83002..9968ee4dc97 100644 --- a/src/table/src/lib.rs +++ b/src/table/src/lib.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. -#![feature(try_blocks)] - pub mod dist_table; pub mod error; pub mod metadata; diff --git a/src/table/src/table/scan.rs b/src/table/src/table/scan.rs index 934f838a33a..2db051a1e9c 100644 --- a/src/table/src/table/scan.rs +++ b/src/table/src/table/scan.rs @@ -144,8 +144,8 @@ impl RegionScanExec { )) }) .collect::>(); - let ts_col: Option = try { - PhysicalSortExpr::new( + let ts_col: Option = (|| { + Some(PhysicalSortExpr::new( Arc::new( Column::new_with_schema( &metadata.time_index_column().column_schema.name, @@ -157,8 +157,8 @@ impl RegionScanExec { descending: false, nulls_first: true, }, - ) - }; + )) + })(); let eq_props = match request.distribution { Some(TimeSeriesDistribution::PerSeries) => {