impl(bigquery): add internal proto custom stream writers

[haphungw]

Add internal protobuf custom stream writers (CommittedWriter, PendingWriter, BufferedWriter) and stream creation methods on WriterBuilder.

Towards #6598

[gemini-code-assist]

Code Review

This pull request introduces support for buffered, committed, and pending stream writers in the BigQuery protobuf write client. It extracts shared logic into a new internal BaseWriter and implements BufferedWriter, CommittedWriter, and PendingWriter, while updating WriterBuilder to support constructing them. The feedback recommends avoiding an unnecessary string allocation in PendingWriter::commit by passing a borrowed &str directly to set_parent instead of calling .to_string(), which aligns with the repository's style guide.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 96.95%. Comparing base (22ce0cc) to head (7a3abc8).

Additional details and impacted files

@@           Coverage Diff            @@
##             main    #6748    +/-   ##
========================================
  Coverage   96.95%   96.95%            
========================================
  Files         319      323     +4     
  Lines      105346   105757   +411     
========================================
+ Hits       102137   102541   +404     
- Misses       3209     3216     +7     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[dbolduc]

Feel free to continue down this path. I am not sure if it is easier to refactor now or later.

As an idea, we can make these types generic on the schema and rows:

use crate::write::generated::gapic_storage::model::append_rows_request;

pub(crate) struct BaseWriter<T>
where T: DataFormat {
  pub(crate) runner: Runner,
  pub(crate) write_stream: String,
  pub(crate) schema: T::Schema,
  pub(crate) client: BigQueryWrite,
}

impl<T> BaseWriter {
  pub(crate) fn append_request(&self, rows: T::Rows) -> AppendRowsRequest { ... }
}

trait DataFormat {
  type Schema;
  type Rows;

  fn to_rows(schema: Self::Schema, rows: Self::Rows) -> crate::write::generated::gapic_storage::model::append_rows_request::Rows;
}

struct Proto;
impl DataFormat for Proto {
  type Schema: ProtoSchema,
  type Rows: ProtoRows,

  fn to_rows(schema: Self::Schema, rows: Self::Rows) -> append_rows_request::Rows {
    append_rows_request::Rows::ProtoRows(Box::new(
        ProtoData::new().set_writer_schema(schema).set_rows(rows)))
  }
}

// and same for Arrow

then we test that implementation once.

And define types like:

pub mod arrow {
  pub use DefaultWriter = common::DefaultWriter<Arrow>;
  // etc.
}
pub mod proto {
  pub use DefaultWriter = common::DefaultWriter<Proto>;
  // etc.
}

The one thing to look out for is if the common:: types leak in our docs. If so that would be unfortunate and an argument against this kind of unification