cockroach/influxdb: CWE-407 findings

defects/influxdb/patch/influxdb-0001-persisted-files-dedup-quadratic.patch

@@ -0,0 +1,47 @@
+# UNDF:
+--- a/influxdb3_write/src/write_buffer/persisted_files.rs
++++ b/influxdb3_write/src/write_buffer/persisted_files.rs
+@@ -505,10 +505,15 @@ fn update_persisted_files_with_snapshot(
+                 if initial_load {
+                     file_count += new_parquet_files.len() as u64;
+                     table_files.extend(new_parquet_files.iter().cloned());
+                 } else {
+-                    let mut filtered_files: Vec<ParquetFile> = new_parquet_files
+-                        .iter()
+-                        .filter(|file| !table_files.contains(file))
+-                        .cloned()
+-                        .collect();
++                    // Build a HashSet of existing paths for O(1) dedup
++                    // instead of O(T) Vec::contains per file → O(F×T) total.
++                    let existing_paths: std::collections::HashSet<&str> =
++                        table_files.iter().map(|f| f.path.as_str()).collect();
++                    let mut filtered_files: Vec<ParquetFile> = new_parquet_files
++                        .iter()
++                        .filter(|file| !existing_paths.contains(file.path.as_str()))
++                        .cloned()
++                        .collect();
+                     file_count += filtered_files.len() as u64;
+                     table_files.append(&mut filtered_files);
+                 }
+@@ -423,10 +423,14 @@ impl Inner {
+     pub(crate) fn add_persisted_file(
+         &mut self,
+         db_id: &DbId,
+         table_id: &TableId,
+         parquet_file: &ParquetFile,
+     ) {
+         let existing_parquet_files = self
+             .files
+             .entry(*db_id)
+             .or_default()
+             .entry(*table_id)
+             .or_default();
+-        if !existing_parquet_files.contains(parquet_file) {
++        // Vec::contains is O(N); use path-based guard instead.
++        let already_present = existing_parquet_files
++            .iter()
++            .any(|f| f.path == parquet_file.path);
++        if !already_present {
+             self.parquet_files_row_count += parquet_file.row_count;
+             self.parquet_files_size_mb += as_mb(parquet_file.size_bytes);
+             existing_parquet_files.push(parquet_file.clone());

defects/influxdb/patch/influxdb-0002-table-def-series-key-quadratic.patch

@@ -0,0 +1,41 @@
+# UNDF:
+--- a/influxdb3_catalog/src/catalog/versions/v1.rs
++++ b/influxdb3_catalog/src/catalog/versions/v1.rs
+@@ -1442,10 +1442,16 @@ impl TableDefinitionV1 {
+     pub fn add_columns(
+         &mut self,
+         columns: Vec<(ColumnId, Arc<str>, InfluxColumnType)>,
+     ) -> Result<()> {
++        // Pre-build a HashSet of existing series key IDs for O(1) membership
++        // checks instead of O(K) Vec::contains per column → O(C×K) total.
++        let mut series_key_set: std::collections::HashSet<ColumnId> =
++            self.series_key.iter().copied().collect();
++
+         let mut cols = BTreeMap::new();
+         for col_def in self.columns.resource_iter().cloned() {
+             cols.insert(Arc::clone(&col_def.name), col_def);
+         }
+
+         let mut sort_key_changed = false;
+
+         for (id, name, column_type) in columns {
+             let nullable = name.as_ref() != TIME_COLUMN_NAME;
+             assert!(
+                 cols.insert(
+                     Arc::clone(&name),
+                     Arc::new(ColumnDefinition::new(
+                         id,
+                         Arc::clone(&name),
+                         column_type,
+                         nullable
+                     ))
+                 )
+                 .is_none(),
+                 "attempted to add existing column"
+             );
+             // add new tags to the series key in the order provided
+-            if matches!(column_type, InfluxColumnType::Tag) && !self.series_key.contains(&id) {
++            if matches!(column_type, InfluxColumnType::Tag) && series_key_set.insert(id) {
+                 self.tag_column_name_to_position_id
+                     .insert(Arc::clone(&name), self.series_key.len() as u8);
+                 self.series_key.push(id);