Merge pull request MaterializeInc#33472 from petrosagg/forbid-iterator-zip

*: disallow using `Iterator::zip`

Author: petrosagg

Cargo.lock

@@ -4,6 +4,7 @@ await-holding-invalid-types = [
 ]
 
 disallowed-methods = [
+    { path = "std::iter::Iterator::zip", reason = "use `Itertools::zip_eq` instead" },
     { path = "std::panic::catch_unwind", reason = "use `mz_ore::panic::catch_unwind` instead" },
     { path = "futures::FutureExt::catch_unwind", reason = "use `mz_ore::future::FutureExt::catch_unwind` instead" },
 

clippy.toml

@@ -2808,7 +2808,7 @@ mod tests {
                 let full_name = conn_catalog.resolve_full_name(item.name());
                 let actual_desc = item.desc(&full_name).expect("invalid item type");
                 for (index, ((actual_name, actual_typ), (expected_name, expected_typ))) in
-                    actual_desc.iter().zip(expected_desc.iter()).enumerate()
+                    actual_desc.iter().zip_eq(expected_desc.iter()).enumerate()
                 {
                     assert_eq!(
                         actual_name, expected_name,

src/adapter/src/catalog.rs

@@ -791,7 +791,7 @@ fn add_new_remove_old_builtin_items_migration(
     let new_builtin_ids = txn.allocate_system_item_ids(usize_to_u64(new_builtins.len()))?;
     let new_builtins: Vec<_> = new_builtins
         .into_iter()
-        .zip(new_builtin_ids.clone())
+        .zip_eq(new_builtin_ids.clone())
         .collect();
 
     // Look for migrated builtins.

src/adapter/src/catalog/open.rs

@@ -3229,7 +3229,7 @@ impl Coordinator {
         );
 
         let catalog = self.catalog_mut();
-        for (id, plan) in catalog_ids.into_iter().zip(dataflows) {
+        for (id, plan) in catalog_ids.into_iter().zip_eq(dataflows) {
             catalog.set_physical_plan(id, plan);
         }
 

src/adapter/src/coord.rs

@@ -467,7 +467,7 @@ impl Coordinator {
             .allocate_user_ids(u64::cast_from(subsource_stmts.len()), id_ts)
             .await?;
         for (subsource_stmt, (item_id, global_id)) in
-            subsource_stmts.into_iter().zip(ids.into_iter())
+            subsource_stmts.into_iter().zip_eq(ids.into_iter())
         {
             let s = self.plan_subsource(session, &params, subsource_stmt, item_id, global_id)?;
             subsource_plans.push(s);
@@ -622,7 +622,7 @@ impl Coordinator {
             .catalog_mut()
             .allocate_user_ids(u64::cast_from(subsource_stmts.len()), id_ts)
             .await?;
-        for (stmt, (item_id, global_id)) in subsource_stmts.into_iter().zip(ids.into_iter()) {
+        for (stmt, (item_id, global_id)) in subsource_stmts.into_iter().zip_eq(ids.into_iter()) {
             let plan = self.plan_subsource(ctx.session(), &params, stmt, item_id, global_id)?;
             create_source_plans.push(plan);
         }

src/adapter/src/coord/sequencer/inner.rs

@@ -10,6 +10,7 @@
 //! Various utility methods used by the [`Coordinator`]. Ideally these are all
 //! put in more meaningfully named modules.
 
+use itertools::Itertools;
 use mz_adapter_types::connection::ConnectionId;
 use mz_ore::now::EpochMillis;
 use mz_repr::{Diff, GlobalId, ScalarType};
@@ -76,7 +77,7 @@ impl Coordinator {
         let params = params
             .datums
             .into_iter()
-            .zip(params.execute_types)
+            .zip_eq(params.execute_types)
             .collect();
         let result_formats = vec![mz_pgwire_common::Format::Text; desc.arity()];
         let logging = session.mint_logging(sql, Some(&stmt), now);

src/adapter/src/coord/sql.rs

@@ -9,6 +9,7 @@
 
 use std::fmt::Debug;
 
+use itertools::Itertools;
 use mz_catalog::durable::{DurableCatalogError, FenceError};
 use mz_compute_client::controller::error::{
     CollectionUpdateError, DataflowCreationError, InstanceMissing, PeekError, ReadPolicyError,
@@ -463,7 +464,7 @@ pub fn verify_datum_desc(
         return Err(AdapterError::Internal(msg));
     }
 
-    for (i, (d, t)) in datums.iter().zip(col_types).enumerate() {
+    for (i, (d, t)) in datums.iter().zip_eq(col_types).enumerate() {
         if !d.is_instance_of(t) {
             let msg = format!(
                 "internal error: column {} is not of expected type {:?}: {:?}",

src/adapter/src/util.rs

@@ -15,6 +15,7 @@ arrow = { version = "55.2.0", default-features = false }
 chrono = { version = "0.4.39", default-features = false, features = ["std"] }
 dec = { version = "0.4.9", features = ["num-traits"] }
 half = "2"
+itertools = "0.14.0"
 mz-repr = { path = "../repr" }
 mz-ore = { path = "../ore" }
 num-traits = "0.2"

src/arrow-util/Cargo.toml

@@ -22,6 +22,7 @@ use arrow::datatypes::{
 use arrow::error::ArrowError;
 use arrow::record_batch::RecordBatch;
 use chrono::Timelike;
+use itertools::Itertools;
 use mz_ore::cast::CastFrom;
 use mz_repr::adt::jsonb::JsonbRef;
 use mz_repr::{Datum, RelationDesc, Row, ScalarType};
@@ -123,7 +124,7 @@ impl ArrowBuilder {
     /// Appends a row to the builder.
     /// Errors if the row contains an unimplemented or out-of-range value.
     pub fn add_row(&mut self, row: &Row) -> Result<(), anyhow::Error> {
-        for (col, datum) in self.columns.iter_mut().zip(row.iter()) {
+        for (col, datum) in self.columns.iter_mut().zip_eq(row.iter()) {
             col.append_datum(datum)?;
         }
         self.row_size_bytes += row.byte_len();