Repository: influxdata/influxdb
File: influxdb3_write/src/write_buffer/mod.rs (Rust)

Code before:
    ) -> Result<BufferedWriteRequest> {
        debug!("write_lp to {} in writebuffer", db_name);

        // validated lines will update the in-memory catalog, ensuring that all write operations
        // past this point will be infallible
        let result = WriteValidator::initialize(
            db_name.clone(),
            self.catalog(),
            ingest_time.timestamp_nanos(),
        )?
        .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
        let mut ops = Vec::with_capacity(2);
        if let Some(catalog_batch) = result.catalog_updates {
            ops.push(WalOp::Catalog(catalog_batch));
        }
        ops.push(WalOp::Write(result.valid_data));

        if no_sync {
            let wal = Arc::clone(&self.wal);
            tokio::spawn(async move {
                if let Err(err) = wal.write_ops(ops).await {
                    error!("Failed to persist data to the wal: {err}");
                }
            });
        } else {
            // write to the wal. Behind the scenes the ops get buffered in memory and once a second (or
            // whatever the configured wal flush interval is set to) the buffer is flushed and all the
            // data is persisted into a single wal file in the configured object store. Then the
            // contents are sent to the configured notifier, which in this case is the queryable buffer.
            // Thus, after this returns, the data is both durable and queryable.
            self.wal.write_ops(ops).await?;
        }

        // record metrics for lines written, rejected, and bytes written
        self.metrics
            .record_lines(&db_name, result.line_count as u64);
        self.metrics
            .record_lines_rejected(&db_name, result.errors.len() as u64);
        self.metrics
            .record_bytes(&db_name, result.valid_bytes_count);

        Ok(BufferedWriteRequest {
            db_name,
            invalid_lines: result.errors,
            line_count: result.line_count,
            field_count: result.field_count,
            index_count: result.index_count,

Review comment (suggestion):
We shouldn't require a spawned task here. There's already a function on the WAL called `buffer_op_unconfirmed` (see https://github.com/influxdata/influxdb/blob/main/influxdb3_wal/src/lib.rs#L70). Just update this to instead take a vec and you'll be able to do this without a spawned task. It puts it into a mutex'd vec that gets used when the flush happens.

Code after:
        precision: Precision,
        no_sync: bool,
    ) -> Result<BufferedWriteRequest> {
        debug!("write_lp to {} in writebuffer", db_name);

        // validated lines will update the in-memory catalog, ensuring that all write operations
        // past this point will be infallible
        let result = WriteValidator::initialize(
            db_name.clone(),
            self.catalog(),
            ingest_time.timestamp_nanos(),
        )?
        .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
        let mut ops = Vec::with_capacity(2);
        if let Some(catalog_batch) = result.catalog_updates {
            ops.push(WalOp::Catalog(catalog_batch));
        }
        ops.push(WalOp::Write(result.valid_data));

        if no_sync {
            self.wal.write_ops_unconfirmed(ops).await?;
        } else {
            // write to the wal. Behind the scenes the ops get buffered in memory and once a second (or
            // whatever the configured wal flush interval is set to) the buffer is flushed and all the
            // data is persisted into a single wal file in the configured object store. Then the
            // contents are sent to the configured notifier, which in this case is the queryable buffer.
            // Thus, after this returns, the data is both durable and queryable.
            self.wal.write_ops(ops).await?;
        }

        // record metrics for lines written, rejected, and bytes written
        self.metrics
            .record_lines(&db_name, result.line_count as u64);
        self.metrics
            .record_lines_rejected(&db_name, result.errors.len() as u64);
        self.metrics
            .record_bytes(&db_name, result.valid_bytes_count);

        Ok(BufferedWriteRequest {
            db_name,
            invalid_lines: result.errors,
            line_count: result.line_count,
            field_count: result.field_count,
            index_count: result.index_count,
        })
    }