chore: upgrade nightly rust to 1.72

.github/workflows/ci.yml

@@ -42,7 +42,7 @@ env:
   CERESMETA_IMAGE_NAME: ceresdb/ceresmeta-server:latest
   CERESDB_IMAGE_NAME: ceresdb/ceresdb-server:latest
   LOCK_FILE: Cargo.lock
-  RUST_VERSION: nightly-2023-02-02
+  RUST_VERSION: nightly-2023-08-28
 
 jobs:
   style-check:

.github/workflows/coverage.yml

@@ -31,7 +31,7 @@ env:
   RUSTFLAGS: "-C debuginfo=1"
   CARGO_TERM_COLOR: always
   RUST_BACKTRACE: "1"
-  RUST_VERSION: nightly-2023-02-02
+  RUST_VERSION: nightly-2023-08-28
 
 jobs:
   coverage:

analytic_engine/src/compaction/picker.rs

@@ -543,7 +543,7 @@ impl TimeWindowPicker {
 
             let (left, _) = Self::get_window_bounds_in_millis(window, ts);
 
-            let bucket_files = buckets.entry(left).or_insert_with(Vec::new);
+            let bucket_files = buckets.entry(left).or_default();
 
             bucket_files.push(f.clone());
 
@@ -813,7 +813,7 @@ mod tests {
 
     #[test]
     fn test_time_window_picker() {
-        let picker_manager = PickerManager::default();
+        let picker_manager = PickerManager;
         let twp = picker_manager.get_picker(CompactionStrategy::Default);
         let mut ctx = PickerContext {
             segment_duration: Duration::from_millis(1000),

analytic_engine/src/compaction/scheduler.rs

@@ -311,7 +311,7 @@ impl SchedulerImpl {
             space_store,
             runtime: runtime.clone(),
             schedule_interval: config.schedule_interval.0,
-            picker_manager: PickerManager::default(),
+            picker_manager: PickerManager,
             max_ongoing_tasks: config.max_ongoing_tasks,
             max_unflushed_duration: config.max_unflushed_duration.0,
             write_sst_max_buffer_size,

analytic_engine/src/instance/engine.rs

@@ -407,12 +407,6 @@ impl Instance {
         let shard_id = request.shard_id;
         let mut table_ctxs = Vec::with_capacity(request.table_defs.len());
 
-        // Open tables.
-        struct TableInfo {
-            name: String,
-            id: TableId,
-        }
-
         let mut spaces_of_tables = Vec::with_capacity(request.table_defs.len());
         for table_def in request.table_defs {
             let context = SpaceContext {

analytic_engine/src/instance/wal_replayer.rs

@@ -217,7 +217,7 @@ impl TableBasedReplay {
         loop {
             // fetch entries to log_entry_buf
             let _timer = PULL_LOGS_DURATION_HISTOGRAM.start_timer();
-            let decoder = WalDecoder::default();
+            let decoder = WalDecoder;
             log_entry_buf = log_iter
                 .next_log_entries(decoder, log_entry_buf)
                 .await
@@ -309,7 +309,7 @@ impl RegionBasedReplay {
         // Split and replay logs.
         loop {
             let _timer = PULL_LOGS_DURATION_HISTOGRAM.start_timer();
-            let decoder = WalDecoder::default();
+            let decoder = WalDecoder;
             log_entry_buf = log_iter
                 .next_log_entries(decoder, log_entry_buf)
                 .await

analytic_engine/src/manifest/details.rs

@@ -595,8 +595,8 @@ struct ObjectStoreBasedSnapshotStore {
 }
 
 impl ObjectStoreBasedSnapshotStore {
-    const CURRENT_SNAPSHOT_NAME: &str = "current";
-    const SNAPSHOT_PATH_PREFIX: &str = "manifest/snapshot";
+    const CURRENT_SNAPSHOT_NAME: &'static str = "current";
+    const SNAPSHOT_PATH_PREFIX: &'static str = "manifest/snapshot";
 
     pub fn new(space_id: SpaceId, table_id: TableId, store: ObjectStoreRef) -> Self {
         let snapshot_path = Self::snapshot_path(space_id, table_id);

analytic_engine/src/setup.rs

@@ -429,7 +429,7 @@ async fn open_instance(
         manifest_storages,
         wal_manager,
         store_picker,
-        Arc::new(FactoryImpl::default()),
+        Arc::new(FactoryImpl),
     )
     .await
     .context(OpenInstance)?;

analytic_engine/src/sst/file.rs

@@ -369,7 +369,7 @@ struct FileHandleSet {
 
 impl FileHandleSet {
     fn latest(&self) -> Option<FileHandle> {
-        if let Some(file) = self.file_map.values().rev().next() {
+        if let Some(file) = self.file_map.values().next_back() {
             return Some(file.clone());
         }
         None

analytic_engine/src/sst/parquet/async_reader.rs

@@ -778,8 +778,7 @@ mod tests {
 
     impl MockRandomSenders {
         fn start_to_send(&mut self) {
-            while !self.tx_group.is_empty() {
-                let tx = self.tx_group.pop().unwrap();
+            while let Some(tx) = self.tx_group.pop() {
                 let test_data = self.test_datas.pop().unwrap();
                 tokio::spawn(async move {
                     for datum in test_data {

analytic_engine/src/table/version.rs

@@ -931,9 +931,9 @@ mod tests {
 
     #[test]
     fn test_table_version_sampling() {
-        let memtable = MemTableMocker::default().build();
+        let memtable = MemTableMocker.build();
         test_table_version_sampling_with_memtable(memtable);
-        let memtable = MemTableMocker::default().build_columnar();
+        let memtable = MemTableMocker.build_columnar();
         test_table_version_sampling_with_memtable(memtable);
     }
 
@@ -975,9 +975,9 @@ mod tests {
 
     #[test]
     fn test_table_version_sampling_switch() {
-        let memtable = MemTableMocker::default().build();
+        let memtable = MemTableMocker.build();
         test_table_version_sampling_switch_with_memtable(memtable);
-        let memtable = MemTableMocker::default().build_columnar();
+        let memtable = MemTableMocker.build_columnar();
         test_table_version_sampling_switch_with_memtable(memtable);
     }
 
@@ -1027,7 +1027,7 @@ mod tests {
     fn test_table_version_sampling_freeze() {
         let version = new_table_version();
 
-        let memtable = MemTableMocker::default().build();
+        let memtable = MemTableMocker.build();
         let schema = memtable.schema().clone();
 
         let memtable_id1 = 1;
@@ -1063,7 +1063,7 @@ mod tests {
         assert_eq!(memtable_id1, read_view.sampling_mem.as_ref().unwrap().id);
         assert!(read_view.sampling_mem.unwrap().freezed);
 
-        let memtable = MemTableMocker::default().build();
+        let memtable = MemTableMocker.build();
         let memtable_id2 = 2;
         let mem_state = MemTableState {
             mem: memtable,
@@ -1110,7 +1110,7 @@ mod tests {
     fn test_table_version_sampling_apply_edit() {
         let version = new_table_version();
 
-        let memtable = MemTableMocker::default().build();
+        let memtable = MemTableMocker.build();
 
         let memtable_id1 = 1;
         let sampling_mem = SamplingMemTable::new(memtable, memtable_id1);
@@ -1124,7 +1124,7 @@ mod tests {
             TimeRange::bucket_of(now, table_options::DEFAULT_SEGMENT_DURATION).unwrap();
 
         // Prepare mutable memtable.
-        let memtable = MemTableMocker::default().build();
+        let memtable = MemTableMocker.build();
         let memtable_id2 = 2;
         let mem_state = MemTableState {
             mem: memtable,

analytic_engine/src/table_options.rs

@@ -427,7 +427,7 @@ impl TableOptions {
                 SEGMENT_DURATION.to_string(),
                 self.segment_duration
                     .map(|v| v.to_string())
-                    .unwrap_or_else(String::new),
+                    .unwrap_or_default(),
             ),
             (UPDATE_MODE.to_string(), self.update_mode.to_string()),
             (ENABLE_TTL.to_string(), self.enable_ttl.to_string()),

analytic_engine/src/tests/util.rs

@@ -617,7 +617,7 @@ impl EngineBuildContext for RocksDBEngineBuildContext {
     type WalsOpener = RocksDBWalsOpener;
 
     fn wals_opener(&self) -> Self::WalsOpener {
-        RocksDBWalsOpener::default()
+        RocksDBWalsOpener
     }
 
     fn config(&self) -> Config {

benchmarks/src/merge_memtable_bench.rs

@@ -146,7 +146,7 @@ impl MergeMemTableBench {
         let table_id = self.table_id;
         let sequence = u64::MAX;
         let projected_schema = self.projected_schema.clone();
-        let sst_factory: SstFactoryRef = Arc::new(FactoryImpl::default());
+        let sst_factory: SstFactoryRef = Arc::new(FactoryImpl);
         let iter_options = IterOptions {
             batch_size: self.sst_read_options.num_rows_per_row_group,
         };

benchmarks/src/merge_sst_bench.rs

@@ -133,7 +133,7 @@ impl MergeSstBench {
         let table_id = self.table_id;
         let sequence = u64::MAX;
         let projected_schema = self.sst_read_options.projected_schema.clone();
-        let sst_factory: SstFactoryRef = Arc::new(FactoryImpl::default());
+        let sst_factory: SstFactoryRef = Arc::new(FactoryImpl);
         let iter_options = IterOptions {
             batch_size: self.sst_read_options.num_rows_per_row_group,
         };
@@ -189,7 +189,7 @@ impl MergeSstBench {
         let space_id = self.space_id;
         let table_id = self.table_id;
         let projected_schema = self.sst_read_options.projected_schema.clone();
-        let sst_factory: SstFactoryRef = Arc::new(FactoryImpl::default());
+        let sst_factory: SstFactoryRef = Arc::new(FactoryImpl);
 
         let request_id = RequestId::next_id();
         let store_picker: ObjectStorePickerRef = Arc::new(self.store.clone());

benchmarks/src/sst_tools.rs

@@ -228,7 +228,7 @@ pub async fn merge_sst(config: MergeSstConfig, runtime: Arc<Runtime>) {
     };
 
     let request_id = RequestId::next_id();
-    let sst_factory: SstFactoryRef = Arc::new(FactoryImpl::default());
+    let sst_factory: SstFactoryRef = Arc::new(FactoryImpl);
     let store_picker: ObjectStorePickerRef = Arc::new(store);
     let projected_schema = ProjectedSchema::no_projection(schema.clone());
     let sst_read_options = SstReadOptions {

cluster/src/shard_lock_manager.rs

@@ -524,7 +524,7 @@ impl ShardLock {
         lease_id: i64,
         expired_at: Instant,
         on_lock_expired: OnExpired,
-        etcd_client: &mut Client,
+        etcd_client: &Client,
         runtime: &RuntimeRef,
     ) -> Result<()>
     where

cluster/src/topology.rs

@@ -118,7 +118,7 @@ impl SchemaTopologies {
 
         self.topologies
             .entry(schema_name.to_string())
-            .or_insert_with(Default::default)
+            .or_default()
             .update_tables(tables);
 
         true

common_types/src/bitset.rs

@@ -298,7 +298,11 @@ mod tests {
     }
 
     fn iter_set_bools(bools: &[bool]) -> impl Iterator<Item = usize> + '_ {
-        bools.iter().enumerate().filter_map(|(x, y)| y.then(|| x))
+        bools
+            .iter()
+            .enumerate()
+            .filter(|&(_, y)| *y)
+            .map(|(x, _)| x)
     }
 
     #[test]
@@ -347,7 +351,7 @@ mod tests {
     }
 
     fn make_rng() -> StdRng {
-        let seed = OsRng::default().next_u64();
+        let seed = OsRng.next_u64();
         println!("Seed: {seed}");
         StdRng::seed_from_u64(seed)
     }

common_types/src/column_block.rs

@@ -1033,7 +1033,7 @@ macro_rules! define_column_block_builder {
 
                 /// Append the [DatumView] into the builder, the datum view should have same the data
                 /// type of builder
-                pub fn append_view<'a>(&mut self, datum: DatumView<'a>) -> Result<()> {
+                pub fn append_view(&mut self, datum: DatumView<'_>) -> Result<()> {
                     let given = datum.kind();
                     match self {
                         Self::Null { rows } => match datum {

common_types/src/column_schema.rs

@@ -602,7 +602,7 @@ mod tests {
 
     #[test]
     fn test_valid_tag_type() {
-        let invalid_tag_types = vec![DatumKind::Null, DatumKind::Float, DatumKind::Double];
+        let invalid_tag_types = [DatumKind::Null, DatumKind::Float, DatumKind::Double];
 
         for v in &DatumKind::VALUES {
             assert_eq!(
@@ -614,7 +614,7 @@ mod tests {
 
     #[test]
     fn test_valid_dictionary_type() {
-        let valid_dictionary_types = vec![DatumKind::String];
+        let valid_dictionary_types = [DatumKind::String];
 
         for v in &DatumKind::VALUES {
             assert_eq!(

components/codec/src/columnar/bool.rs

@@ -99,7 +99,7 @@ impl Encoding {
         }
     }
 
-    fn decode<B, F>(&self, buf: &mut B, f: F) -> Result<()>
+    fn decode<B, F>(&self, buf: &B, f: F) -> Result<()>
     where
         B: Buf,
         F: FnMut(bool) -> Result<()>,

components/codec/src/columnar/bytes.rs

@@ -125,7 +125,7 @@ impl Encoding {
     }
 
     /// The layout can be referred to the docs of [`Encoding`].
-    fn decode<B, F>(&self, ctx: DecodeContext<'_>, buf: &mut B, f: F) -> Result<()>
+    fn decode<B, F>(&self, ctx: DecodeContext<'_>, buf: &B, f: F) -> Result<()>
     where
         B: Buf,
         F: FnMut(Bytes) -> Result<()>,

components/codec/src/columnar/mod.rs

@@ -392,7 +392,7 @@ impl ColumnarDecoder {
 impl ColumnarDecoder {
     fn decode_with_nulls<B: Buf>(
         ctx: DecodeContext<'_>,
-        buf: &mut B,
+        buf: &B,
         num_datums: usize,
         datum_kind: DatumKind,
     ) -> Result<Vec<Datum>> {
@@ -570,7 +570,7 @@ mod tests {
 
     #[test]
     fn test_small_int() {
-        let datums = vec![10u32, 1u32, 2u32, 81u32, 82u32];
+        let datums = [10u32, 1u32, 2u32, 81u32, 82u32];
 
         check_encode_end_decode(
             10,

components/future_ext/src/cancel.rs

@@ -62,7 +62,7 @@ where
     fn drop(&mut self) {
         if !self.done {
             let inner = self.inner.take().unwrap();
-            let handle = self.runtime.spawn(async move { inner.await });
+            let handle = self.runtime.spawn(inner);
             drop(handle);
         }
     }

components/object_store/src/disk_cache.rs

@@ -132,7 +132,7 @@ struct Manifest {
 
 impl Manifest {
     const CURRENT_VERSION: usize = 2;
-    const FILE_NAME: &str = "manifest.json";
+    const FILE_NAME: &'static str = "manifest.json";
 
     #[inline]
     fn is_valid(&self, version: usize, page_size: usize) -> bool {
@@ -876,7 +876,7 @@ mod test {
 
         // remove cached values, then get again
         {
-            for range in vec![0..16, 16..32, 32..48, 48..64, 64..80, 80..96, 96..112] {
+            for range in [0..16, 16..32, 32..48, 48..64, 64..80, 80..96, 96..112] {
                 let data_cache = store
                     .inner
                     .cache
@@ -887,7 +887,7 @@ mod test {
                 assert!(test_file_exists(&store.cache_dir, &location, &range));
             }
 
-            for range in vec![16..32, 48..64, 80..96] {
+            for range in [16..32, 48..64, 80..96] {
                 let mut data_cache = store
                     .inner
                     .cache
@@ -1105,7 +1105,7 @@ mod test {
                     .await
                     .unwrap()
             };
-            for range in vec![16..32, 32..48, 48..64, 64..80, 80..96, 96..112] {
+            for range in [16..32, 32..48, 48..64, 64..80, 80..96, 96..112] {
                 let filename = DiskCacheStore::page_cache_name(&location, &range);
                 let cache = store.cache.meta_cache.lock(&filename);
                 assert!(cache.contains(&filename));

components/object_store/src/obkv/mod.rs

@@ -639,7 +639,7 @@ impl<T: TableKv> ObjectStore for ObkvObjectStore<T> {
             }));
         }
 
-        let iter = futures::stream::iter(meta_list.into_iter());
+        let iter = futures::stream::iter(meta_list);
         debug!(
             "ObkvObjectStore list operation, prefix:{path}, cost:{:?}",
             instant.elapsed()
@@ -682,7 +682,7 @@ impl<T: TableKv> ObjectStore for ObkvObjectStore<T> {
             }
         }
 
-        let common_prefixes = Vec::from_iter(common_prefixes.into_iter());
+        let common_prefixes = Vec::from_iter(common_prefixes);
         debug!(
             "ObkvObjectStore list_with_delimiter operation, prefix:{path}, cost:{:?}",
             instant.elapsed()