refactor: remove conversion step in validator (#25515)

The write validator now builds the row data destined for the WAL directly, vs. creating an intermediate type to hold row data.
influxdata · Nov 4, 2024 · da294a2 · da294a2
1 parent ec01934
commit da294a2
Show file tree

Hide file tree

Showing 3 changed files with 190 additions and 146 deletions.
diff --git a/influxdb3_write/src/lib.rs b/influxdb3_write/src/lib.rs
@@ -343,13 +343,14 @@ mod test_helpers {
         let db_name = NamespaceName::new(db_name).unwrap();
         let result = WriteValidator::initialize(db_name.clone(), catalog, 0)
             .unwrap()
-            .v1_parse_lines_and_update_schema(lp, false)
-            .unwrap()
-            .convert_lines_to_buffer(
+            .v1_parse_lines_and_update_schema(
+                lp,
+                false,
                 Time::from_timestamp_nanos(0),
-                Gen1Duration::new_5m(),
                 Precision::Nanosecond,
-            );
+            )
+            .unwrap()
+            .convert_lines_to_buffer(Gen1Duration::new_5m());
 
         result.valid_data
     }

diff --git a/influxdb3_write/src/write_buffer/mod.rs b/influxdb3_write/src/write_buffer/mod.rs
@@ -225,8 +225,8 @@ impl WriteBufferImpl {
             self.catalog(),
             ingest_time.timestamp_nanos(),
         )?
-        .v1_parse_lines_and_update_schema(lp, accept_partial)?
-        .convert_lines_to_buffer(ingest_time, self.wal_config.gen1_duration, precision);
+        .v1_parse_lines_and_update_schema(lp, accept_partial, ingest_time, precision)?
+        .convert_lines_to_buffer(self.wal_config.gen1_duration);
 
         // if there were catalog updates, ensure they get persisted to the wal, so they're
         // replayed on restart
@@ -267,8 +267,8 @@ impl WriteBufferImpl {
             self.catalog(),
             ingest_time.timestamp_nanos(),
         )?
-        .v3_parse_lines_and_update_schema(lp, accept_partial)?
-        .convert_lines_to_buffer(ingest_time, self.wal_config.gen1_duration, precision);
+        .v3_parse_lines_and_update_schema(lp, accept_partial, ingest_time, precision)?
+        .convert_lines_to_buffer(self.wal_config.gen1_duration);
 
         // if there were catalog updates, ensure they get persisted to the wal, so they're
         // replayed on restart
@@ -560,13 +560,14 @@ mod tests {
         let lp = "cpu,region=west user=23.2 100\nfoo f1=1i";
         WriteValidator::initialize(db_name, Arc::clone(&catalog), 0)
             .unwrap()
-            .v1_parse_lines_and_update_schema(lp, false)
-            .unwrap()
-            .convert_lines_to_buffer(
+            .v1_parse_lines_and_update_schema(
+                lp,
+                false,
                 Time::from_timestamp_nanos(0),
-                Gen1Duration::new_5m(),
                 Precision::Nanosecond,
-            );
+            )
+            .unwrap()
+            .convert_lines_to_buffer(Gen1Duration::new_5m());
 
         let db = catalog.db_schema_by_id(DbId::from(0)).unwrap();