diff --git a/.gitignore b/.gitignore
index db05f9bb79..3b3a0c7305 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,5 +1,7 @@
 build/
 build-perf/
+build-baseline/
+build-perf-baseline/
 cmake-out/
 cmake-build-debug/
 build-out/
diff --git a/google/cloud/odbc/bq_driver/internal/odbc_desc_handle.h b/google/cloud/odbc/bq_driver/internal/odbc_desc_handle.h
index 9844bd7b62..08c8e93f6d 100644
--- a/google/cloud/odbc/bq_driver/internal/odbc_desc_handle.h
+++ b/google/cloud/odbc/bq_driver/internal/odbc_desc_handle.h
@@ -44,6 +44,7 @@ class DescriptorHandle : public Handle {
   DescriptorType GetType() { return type_; }
 
   HeaderRecord& GetHeaderRecord() { return header_record_; }
+  HeaderRecord const& GetHeaderRecord() const { return header_record_; }
 
   inline bool HasDescriptorRecord(int index) const {
     return descriptor_records_.count(index) > 0;
@@ -54,6 +55,9 @@ class DescriptorHandle : public Handle {
   DescriptorRecord& GetDescriptorRecord(int index) {
     return descriptor_records_[index];
   }
+  DescriptorRecord const& GetDescriptorRecord(int index) const {
+    return descriptor_records_.at(index);
+  }
 
   inline void ClearDescriptorRecordsMap() { descriptor_records_.clear(); }
 
diff --git a/google/cloud/odbc/bq_driver/internal/odbc_sql_execute_utils.cc b/google/cloud/odbc/bq_driver/internal/odbc_sql_execute_utils.cc
index e304737658..67fe7e028b 100644
--- a/google/cloud/odbc/bq_driver/internal/odbc_sql_execute_utils.cc
+++ b/google/cloud/odbc/bq_driver/internal/odbc_sql_execute_utils.cc
@@ -620,67 +620,6 @@ StatusRecord ProcessRecordBatch(
   return StatusRecord::Ok();
 }
 
-StatusRecord ReadNextResultsFromStream(StatementHandle& stmt_handle) {
-  std::optional<StreamRange<ReadRowsResponse>>& optional_stream =
-      stmt_handle.GetReadRowsStream();
-  if (!optional_stream.has_value()) {
-    return StatusRecord{SQLStates::k_HY000(),
-                        "Internal Error: No HTAPI read stream found!!"};
-  }
-  auto& read_rows_stream = *optional_stream;
-  auto& optional_it = stmt_handle.GetReadRowsIterator();
-  if (!optional_it.has_value()) {
-    // Initialize iterator to begin() and store it.
-    auto it = read_rows_stream.begin();
-    optional_it = std::move(it);
-  } else {
-    // Advance the stored iterator.
-    // The previous call processed the element at the iterator,
-    // so we increment it to move to the next element.
-    ++(*optional_it);
-    // Irrespective of any errors, if the iterator hasn't reached the end, we
-    // want to cache it
-    stmt_handle.SetReadRowsIterator(*optional_it);
-  }
-  auto& it = *optional_it;
-  if (it != read_rows_stream.end()) {
-    auto const& read_row_status = *it;
-    if (!read_row_status) {
-      return StatusRecord::ConvertFrom(read_row_status.status());
-    }
-    ReadRowsResponse row = *read_row_status;
-    if (row.has_arrow_record_batch()) {
-      // The schema is coming from ResultSet cached in the statement handle.
-      // We don't want to generate the schema again for every batch since it
-      // will remain the same.
-      std::shared_ptr<arrow::Schema> schema = stmt_handle.GetArrowSchema();
-      StatusRecordOr<std::shared_ptr<arrow::RecordBatch>> record_batch_status =
-          GetArrowRecordBatch(row.arrow_record_batch(), schema);
-      if (!record_batch_status) {
-        return record_batch_status.GetStatusRecord();
-      }
-      // We are reading ResultSet from the stmt_handle because we want to
-      // preserve the previous state like `num_rows_fetched_yet`.
-      ResultSet& result_set = stmt_handle.GetResultSet();
-      // To have SQLFetch read the rows from the start, we are setting the
-      // cursor to default.
-      result_set.cursor = -1;
-      return ProcessRecordBatch(schema, *record_batch_status, result_set);
-    } else {
-      return StatusRecord{
-          SQLStates::k_HY000(),
-          "Internal Error: cannot find arrow record batch to process!"};
-    }
-  } else {
-    stmt_handle.ClearReadRowsStream();
-    stmt_handle.ClearReadRowsIterator();
-    // Empty result set.
-    stmt_handle.GetResultSet().rows.clear();
-    stmt_handle.GetResultSet().cursor = -1;
-    LOG(INFO) << "FetchBQDataReadArrow:: Read stream ended.";
-  }
-  return StatusRecord::Ok();
-}
 
 StatusRecord FetchBQDataReadArrow(StatementHandle& stmt_handle,
                                   TableReference& table_ref) {
@@ -692,7 +631,7 @@ StatusRecord FetchBQDataReadArrow(StatementHandle& stmt_handle,
 
   CreateReadSessionRequest create_read_session_request;
   create_read_session_request.set_parent("projects/" + project_id);
-  create_read_session_request.set_max_stream_count(1);
+  create_read_session_request.set_max_stream_count(4);
   auto* read_session = create_read_session_request.mutable_read_session();
   read_session->set_table(table_path);
   read_session->set_data_format(ARROW);
@@ -710,8 +649,6 @@ StatusRecord FetchBQDataReadArrow(StatementHandle& stmt_handle,
   auto session = *read_session_status;
 
   if (!session.streams().empty()) {
-    std::string read_stream_name = session.streams(0).name();
-
     ResultSet result_set;
     StatusRecordOr<std::shared_ptr<arrow::Schema>> schema_status =
         GetArrowSchema(session.arrow_schema(), result_set.row_schema);
@@ -723,16 +660,22 @@ StatusRecord FetchBQDataReadArrow(StatementHandle& stmt_handle,
     std::shared_ptr<arrow::Schema> schema = *schema_status;
     stmt_handle.SetArrowSchema(schema);
 
-    // Create a ReadRowsRequest.
-    ReadRowsRequest read_rows_request;
-    read_rows_request.set_read_stream(read_stream_name);
-
-    // Before we call ReadNextResultsFromStream, we are caching the stream
-    StreamRange<google::cloud::bigquery::storage::v1::ReadRowsResponse>
-        read_rows_stream =
-            bq_client->GetReadRowsStream(read_rows_request, options);
-    stmt_handle.SetReadRowsStream(std::move(read_rows_stream));
-    return ReadNextResultsFromStream(stmt_handle);
+    std::vector<std::unique_ptr<ReadStreamInfo>> stream_infos;
+    for (auto const& stream : session.streams()) {
+      ReadRowsRequest read_rows_request;
+      read_rows_request.set_read_stream(stream.name());
+      
+      StreamRange<google::cloud::bigquery::storage::v1::ReadRowsResponse>
+          read_rows_stream =
+              bq_client->GetReadRowsStream(read_rows_request, options);
+              
+      auto stream_info = std::make_unique<ReadStreamInfo>();
+      stream_info->stream = std::move(read_rows_stream);
+      stream_info->iterator = stream_info->stream.begin();
+      stream_infos.push_back(std::move(stream_info));
+    }
+    stmt_handle.SetReadRowsStreams(std::move(stream_infos));
+    return StatusRecord::Ok();
   }
 
   return StatusRecord{SQLStates::k_HY000(),
diff --git a/google/cloud/odbc/bq_driver/internal/odbc_sql_execute_utils.h b/google/cloud/odbc/bq_driver/internal/odbc_sql_execute_utils.h
index 287854b41f..78ef94b392 100644
--- a/google/cloud/odbc/bq_driver/internal/odbc_sql_execute_utils.h
+++ b/google/cloud/odbc/bq_driver/internal/odbc_sql_execute_utils.h
@@ -19,6 +19,17 @@
 #include "google/cloud/odbc/bq_driver/internal/odbc_desc_handle.h"
 #include "google/cloud/odbc/bq_driver/internal/odbc_stmt_handle.h"
 
+#if (!defined(_WIN32) || defined(_WIN64)) && !defined(NO_ARROW)
+namespace arrow {
+class Schema;
+class RecordBatch;
+}  // namespace arrow
+
+namespace google::cloud::bigquery::storage::v1 {
+class ArrowRecordBatch;
+}  // namespace google::cloud::bigquery::storage::v1
+#endif
+
 namespace google::cloud::odbc_bq_driver_internal {
 
 // Updates the list of `QueryParameter`s with the value for those parameters
@@ -62,11 +73,16 @@ odbc_internal::StatusRecordOr<DSResults> ExecuteScript(
         post_query_request);
 
 #if (!defined(_WIN32) || defined(_WIN64)) && !defined(NO_ARROW)
-/*
- * @brief Reads the next set of rows from the stream cached in the statement
- * handle
- */
-StatusRecord ReadNextResultsFromStream(StatementHandle& stmt_handle);
+
+odbc_internal::StatusRecordOr<std::shared_ptr<arrow::RecordBatch>>
+GetArrowRecordBatch(
+    ::google::cloud::bigquery::storage::v1::ArrowRecordBatch const&
+        record_batch_in,
+    std::shared_ptr<arrow::Schema> schema);
+
+odbc_internal::StatusRecord ProcessRecordBatch(
+    std::shared_ptr<arrow::Schema> schema,
+    std::shared_ptr<arrow::RecordBatch> record_batch, ResultSet& result_set);
 
 #endif  // (!defined(_WIN32) || defined(_WIN64)) && !defined(NO_ARROW)
 
diff --git a/google/cloud/odbc/bq_driver/internal/odbc_sql_fetch.cc b/google/cloud/odbc/bq_driver/internal/odbc_sql_fetch.cc
index 69618d66f8..36f03771e1 100644
--- a/google/cloud/odbc/bq_driver/internal/odbc_sql_fetch.cc
+++ b/google/cloud/odbc/bq_driver/internal/odbc_sql_fetch.cc
@@ -16,6 +16,13 @@
 #include "google/cloud/odbc/bq_driver/internal/data_translation.h"
 #include "google/cloud/odbc/bq_driver/internal/odbc_sql_execute_utils.h"
 #include "google/cloud/odbc/bq_driver/internal/trace_utils.h"
+#include <atomic>
+#include <mutex>
+#include <thread>
+#include <vector>
+#if (!defined(_WIN32) || defined(_WIN64)) && !defined(NO_ARROW)
+#include <arrow/record_batch.h>
+#endif
 
 namespace google::cloud::odbc_bq_driver_internal {
 
@@ -24,7 +31,7 @@ using google::cloud::odbc_internal::StatusRecord;
 
 StatusRecord WriteToApplicationBuffer(DSValue const& ds_val,
                                       BQDataType bq_data_type,
-                                      DescriptorRecord& app_desc_rec,
+                                      DescriptorRecord const& app_desc_rec,
                                       SQLLEN bind_offset,
                                       SQLLEN bind_offset_ind) {
   SQLSMALLINT target_c_type = app_desc_rec.concise_type;
@@ -104,7 +111,7 @@ StatusRecord WriteToApplicationBuffer(DSValue const& ds_val,
 
 // This is according to the spec:
 // https://learn.microsoft.com/en-us/sql/odbc/reference/syntax/sqlbindcol-function?view=sql-server-ver16#buffer-addresses
-SQLLEN GetElemSize(DescriptorRecord& app_desc_rec) {
+SQLLEN GetElemSize(DescriptorRecord const& app_desc_rec) {
   SQLSMALLINT target_c_type = app_desc_rec.concise_type;
   SQLLEN app_buffer_len = app_desc_rec.octet_length;
   switch (target_c_type) {
@@ -148,7 +155,7 @@ SQLLEN GetElemSize(DescriptorRecord& app_desc_rec) {
 }
 
 StatusRecord WriteDSRow(DSRow const& ds_row, RowSchema const& schema,
-                        DescriptorHandle& ard, int row_num) {
+                        DescriptorHandle const& ard, int row_num) {
   SQLLEN* bind_offset_ptr = ard.GetHeaderRecord().bind_offset_ptr;
   SQLLEN bind_offset = 0;
   if (bind_offset_ptr) {
@@ -162,7 +169,7 @@ StatusRecord WriteDSRow(DSRow const& ds_row, RowSchema const& schema,
     if (!ard.HasDescriptorRecord(col_index + 1)) {
       continue;
     }
-    DescriptorRecord& col_desc = ard.GetDescriptorRecord(col_index + 1);
+    DescriptorRecord const& col_desc = ard.GetDescriptorRecord(col_index + 1);
 
     SQLLEN elem_size, elem_size_ind;
     SQLINTEGER bind_type = ard.GetHeaderRecord().bind_type;
@@ -194,7 +201,7 @@ StatusRecord WriteDSRow(DSRow const& ds_row, RowSchema const& schema,
 }
 
 StatusRecord WriteRowset(ResultSet const& result_set, int const rowset_size,
-                         DescriptorHandle& ard, DescriptorHandle& ird) {
+                         DescriptorHandle const& ard, DescriptorHandle& ird) {
   if (rowset_size <= 0) {
     LOG(ERROR) << "WriteRowset:: rowset_size should not be <= 0";
     StatusRecord status_record = {SQLStates::k_HY000(),
@@ -256,26 +263,17 @@ StatusRecord FetchNextResultSet(StatementHandle& stmt_handle) {
   }
 #if (!defined(_WIN32) || defined(_WIN64)) && !defined(NO_ARROW)
   if (stmt_handle.WasHtapiEnabled()) {
-    StatusRecord read_status = ReadNextResultsFromStream(stmt_handle);
-    if (!read_status.ok()) {
-      LOG(ERROR) << "ReadNextResultsFromStream:: " << read_status.message;
-      return read_status;
-    }
-  } else {
-    StatusRecord read_status = FetchNextPageResultSet(stmt_handle);
-    if (!read_status.ok()) {
-      LOG(ERROR) << "FetchNextPageResultSet:: " << read_status.message;
-      return read_status;
-    }
+    LOG(ERROR) << "FetchNextResultSet:: HTAPI is enabled, this function should not be called.";
+    return StatusRecord{SQLStates::k_HY000(),
+                        "Internal Error: FetchNextResultSet called when HTAPI is enabled"};
   }
-#else
+#endif
 
   StatusRecord read_status = FetchNextPageResultSet(stmt_handle);
   if (!read_status.ok()) {
     LOG(ERROR) << "FetchNextPageResultSet:: " << read_status.message;
     return read_status;
   }
-#endif  // (!defined(_WIN32) || defined(_WIN64)) && !defined(NO_ARROW)
   auto& rs_rows = result_set.rows;
   if (rs_rows.empty()) {
     LOG(INFO) << "FetchNextResultSet:: Empty result set fetched.";
@@ -288,5 +286,192 @@ StatusRecord FetchNextResultSet(StatementHandle& stmt_handle) {
   stmt_handle.SetStmtState(StmtStates::kStatementExecutedWithRs);
   return StatusRecord::Ok();
 }
+#if (!defined(_WIN32) || defined(_WIN64)) && !defined(NO_ARROW)
+
+using ::google::cloud::bigquery::storage::v1::ReadRowsResponse;
+
+struct ParallelFetchContext {
+  std::atomic<size_t> next_write_row_index{0};
+  std::atomic<bool> cancelled{false};
+  std::mutex error_mutex;
+  StatusRecord status;
+
+  void SetError(StatusRecord const& err) {
+    std::lock_guard<std::mutex> lock(error_mutex);
+    if (status.ok()) {
+      status = err;
+    }
+    cancelled = true;
+  }
+};
+
+StatusRecord WriteRowsetAtOffset(ResultSet const& result_set,
+                                 int const rowset_size_to_write,
+                                 int starting_row_offset,
+                                 DescriptorHandle const& ard,
+                                 DescriptorHandle& ird) {
+  SQLUSMALLINT* row_status_ptr = ird.GetHeaderRecord().array_status_ptr;
+  for (int i = 0; i < rowset_size_to_write && i < result_set.rows.size(); i++) {
+    StatusRecord status_record =
+        WriteDSRow(result_set.rows[i], result_set.row_schema, ard,
+                   starting_row_offset + i);
+    if (!status_record.ok()) {
+      return status_record;
+    }
+    if (row_status_ptr) {
+      row_status_ptr[starting_row_offset + i] = SQL_ROW_SUCCESS;
+    }
+  }
+  return StatusRecord::Ok();
+}
+
+void FetchStreamWorker(int stream_index, ParallelFetchContext& ctx,
+                       StatementHandle& stmt_handle, int rowset_size,
+                       DescriptorHandle const& ard, DescriptorHandle& ird) {
+  auto& streams = stmt_handle.GetReadRowsStreams();
+  if (stream_index >= streams.size()) {
+    return;
+  }
+  auto& stream_info = *streams[stream_index];
+  std::shared_ptr<arrow::Schema> schema = stmt_handle.GetArrowSchema();
+
+  if (!stream_info.iterator.has_value()) {
+    return;
+  }
+  auto& it = *stream_info.iterator;
+  auto& read_rows_stream = stream_info.stream;
+
+  while (true) {
+    if (ctx.cancelled) {
+      break;
+    }
+    if (it == read_rows_stream.end()) {
+      break;
+    }
+
+    auto const& read_row_status = *it;
+    if (!read_row_status) {
+      ctx.SetError(StatusRecord::ConvertFrom(read_row_status.status()));
+      break;
+    }
+    ReadRowsResponse row = *read_row_status;
+    ++it;
+
+    if (!row.has_arrow_record_batch()) {
+      ctx.SetError({SQLStates::k_HY000(),
+                    "Internal Error: cannot find arrow record batch"});
+      break;
+    }
+
+    auto record_batch_status =
+        GetArrowRecordBatch(row.arrow_record_batch(), schema);
+    if (!record_batch_status) {
+      ctx.SetError(record_batch_status.GetStatusRecord());
+      break;
+    }
+
+    auto record_batch = *record_batch_status;
+    int num_rows = record_batch->num_rows();
+    if (num_rows == 0) {
+      continue;
+    }
+
+    size_t start_row = ctx.next_write_row_index.fetch_add(num_rows);
+    if (start_row >= rowset_size) {
+      ctx.cancelled = true;
+      break;
+    }
+
+    size_t rows_to_write = num_rows;
+    if (start_row + num_rows > rowset_size) {
+      rows_to_write = rowset_size - start_row;
+      ctx.cancelled = true;
+    }
+
+    std::shared_ptr<arrow::RecordBatch> sliced_batch = record_batch;
+    if (rows_to_write < num_rows) {
+      sliced_batch = record_batch->Slice(0, rows_to_write);
+    }
+
+    ResultSet local_result_set;
+    local_result_set.row_schema = stmt_handle.GetResultSet().row_schema;
+
+    StatusRecord status =
+        ProcessRecordBatch(schema, sliced_batch, local_result_set);
+    if (!status.ok()) {
+      ctx.SetError(status);
+      break;
+    }
+
+    status = WriteRowsetAtOffset(local_result_set, rows_to_write, start_row,
+                                 ard, ird);
+    if (!status.ok()) {
+      ctx.SetError(status);
+      break;
+    }
+  }
+}
+
+StatusRecord ParallelFetchAndWrite(StatementHandle& stmt_handle,
+                                   int rowset_size, DescriptorHandle const& ard,
+                                   DescriptorHandle& ird) {
+  ParallelFetchContext ctx;
+  auto& streams = stmt_handle.GetReadRowsStreams();
+  int num_streams = streams.size();
+
+  std::vector<std::thread> threads;
+  threads.reserve(num_streams);
+
+  for (int i = 0; i < num_streams; ++i) {
+    threads.emplace_back(FetchStreamWorker, i, std::ref(ctx),
+                         std::ref(stmt_handle), rowset_size, std::ref(ard),
+                         std::ref(ird));
+  }
+
+  for (auto& t : threads) {
+    if (t.joinable()) {
+      t.join();
+    }
+  }
+
+  if (!ctx.status.ok()) {
+    return ctx.status;
+  }
+
+  size_t total_rows_written = ctx.next_write_row_index.load();
+  if (total_rows_written > rowset_size) {
+    total_rows_written = rowset_size;
+  }
+
+  // Update unused status array
+  SQLUSMALLINT* row_status_ptr = ird.GetHeaderRecord().array_status_ptr;
+  if (row_status_ptr) {
+    for (size_t i = total_rows_written; i < rowset_size; i++) {
+      row_status_ptr[i] = SQL_ROW_NOROW;
+    }
+  }
+
+  SQLULEN* rows_processed_ptr = ird.GetHeaderRecord().rows_processed_ptr;
+  if (rows_processed_ptr) {
+    *rows_processed_ptr = total_rows_written;
+  }
+
+  stmt_handle.GetResultSet().num_rows_fetched_yet += total_rows_written;
+
+  if (total_rows_written == 0) {
+    return StatusRecord({SQLStates::k_SQL_NO_DATA(), "No data found"});
+  }
+
+  return StatusRecord::Ok();
+}
+
+#else
+
+StatusRecord ParallelFetchAndWrite(StatementHandle&, int,
+                                   DescriptorHandle const&, DescriptorHandle&) {
+  return StatusRecord{SQLStates::k_HYC00(), "HTAPI not supported"};
+}
+
+#endif  // (!defined(_WIN32) || defined(_WIN64)) && !defined(NO_ARROW)
 
 }  // namespace google::cloud::odbc_bq_driver_internal
diff --git a/google/cloud/odbc/bq_driver/internal/odbc_sql_fetch.h b/google/cloud/odbc/bq_driver/internal/odbc_sql_fetch.h
index 1a01b551a0..8c2e300b3b 100644
--- a/google/cloud/odbc/bq_driver/internal/odbc_sql_fetch.h
+++ b/google/cloud/odbc/bq_driver/internal/odbc_sql_fetch.h
@@ -23,13 +23,18 @@ namespace google::cloud::odbc_bq_driver_internal {
 
 // Writes rowset_size number of rows to the columns bound by the application
 google::cloud::odbc_internal::StatusRecord WriteRowset(
-    ResultSet const& result_set, int rowset_size, DescriptorHandle& ard,
+    ResultSet const& result_set, int rowset_size, DescriptorHandle const& ard,
     DescriptorHandle& ird);
 
 // Fetches the next batch of ResultSet rows
 google::cloud::odbc_internal::StatusRecord FetchNextResultSet(
     StatementHandle& stmt_handle);
 
+// Parallelly fetches data from multiple streams and writes to application buffers
+google::cloud::odbc_internal::StatusRecord ParallelFetchAndWrite(
+    StatementHandle& stmt_handle, int rowset_size,
+    DescriptorHandle const& ard, DescriptorHandle& ird);
+
 }  // namespace google::cloud::odbc_bq_driver_internal
 
 #endif  // CPP_BIGQUERY_ODBC_GOOGLE_CLOUD_ODBC_BQ_DRIVER_INTERNAL_ODBC_SQL_FETCH_H
diff --git a/google/cloud/odbc/bq_driver/internal/odbc_stmt_handle.h b/google/cloud/odbc/bq_driver/internal/odbc_stmt_handle.h
index 08e70a3d5e..2c93d63045 100644
--- a/google/cloud/odbc/bq_driver/internal/odbc_stmt_handle.h
+++ b/google/cloud/odbc/bq_driver/internal/odbc_stmt_handle.h
@@ -51,6 +51,11 @@ struct PagingInfo {
   std::string job_id;
 };
 
+struct ReadStreamInfo {
+  StreamRange<::google::cloud::bigquery::storage::v1::ReadRowsResponse> stream;
+  std::optional<StreamRange<::google::cloud::bigquery::storage::v1::ReadRowsResponse>::iterator> iterator;
+};
+
 class ConnectionHandle;
 
 class StatementHandle : public Handle {
@@ -117,40 +122,19 @@ class StatementHandle : public Handle {
     ds_results_ = ds_results;
   }
 
-  // Getter for the iterator
-  inline std::optional<StreamRange<
-      ::google::cloud::bigquery::storage::v1::ReadRowsResponse>::iterator>&
-  GetReadRowsIterator() {
-    return read_rows_iterator_;
-  }
-
-  // Setter for the iterator
-  inline void SetReadRowsIterator(
-      StreamRange<::google::cloud::bigquery::storage::v1::ReadRowsResponse>::
-          iterator it) {
-    read_rows_iterator_ = std::move(it);
-  }
-
-  // Clearer for the iterator
-  inline void ClearReadRowsIterator() { read_rows_iterator_.reset(); }
-
-  // Returns true if read_rows_stream_ is NOT null (i.e., has a value).
-  inline bool WasHtapiEnabled() const { return read_rows_stream_.has_value(); }
+  // Returns true if we have HTAPI streams.
+  inline bool WasHtapiEnabled() const { return !read_rows_streams_.empty(); }
 
-  void SetReadRowsStream(
-      StreamRange<::google::cloud::bigquery::storage::v1::ReadRowsResponse>
-          stream_range) {
-    read_rows_stream_ = std::move(stream_range);
+  inline void SetReadRowsStreams(
+      std::vector<std::unique_ptr<ReadStreamInfo>> streams) {
+    read_rows_streams_ = std::move(streams);
   }
 
-  std::optional<
-      StreamRange<::google::cloud::bigquery::storage::v1::ReadRowsResponse>>&
-  GetReadRowsStream() {
-    return read_rows_stream_;
+  inline std::vector<std::unique_ptr<ReadStreamInfo>>& GetReadRowsStreams() {
+    return read_rows_streams_;
   }
 
-  // Clearer: Provides the mechanism to "clear" the stream.
-  inline void ClearReadRowsStream() { read_rows_stream_.reset(); }
+  inline void ClearReadRowsStreams() { read_rows_streams_.clear(); }
 
 #if (!defined(_WIN32) || defined(_WIN64)) && !defined(NO_ARROW)
   inline void SetArrowSchema(std::shared_ptr<arrow::Schema> arrow_schema) {
@@ -339,12 +323,7 @@ class StatementHandle : public Handle {
   std::string cursor_name_;
   DSResults ds_results_;
 
-  std::optional<StreamRange<
-      ::google::cloud::bigquery::storage::v1::ReadRowsResponse>::iterator>
-      read_rows_iterator_;
-  std::optional<
-      StreamRange<::google::cloud::bigquery::storage::v1::ReadRowsResponse>>
-      read_rows_stream_;
+  std::vector<std::unique_ptr<ReadStreamInfo>> read_rows_streams_;
   mutable std::mutex statement_handle_mutex_;
   ::google::cloud::bigquery_v2_minimal_internal::PostQueryRequest
       post_query_request_;
diff --git a/google/cloud/odbc/bq_driver/odbc_sql_results.cc b/google/cloud/odbc/bq_driver/odbc_sql_results.cc
index c9f1bd4e4f..d882d76ddb 100644
--- a/google/cloud/odbc/bq_driver/odbc_sql_results.cc
+++ b/google/cloud/odbc/bq_driver/odbc_sql_results.cc
@@ -39,6 +39,7 @@ using google::cloud::odbc_bq_driver_internal::DescriptorType;
 using google::cloud::odbc_bq_driver_internal::DSRow;
 using google::cloud::odbc_bq_driver_internal::DSValue;
 using google::cloud::odbc_bq_driver_internal::FetchNextResultSet;
+using google::cloud::odbc_bq_driver_internal::ParallelFetchAndWrite;
 using google::cloud::odbc_bq_driver_internal::GetColumnData;
 using google::cloud::odbc_bq_driver_internal::IntValueToOutputBufferResponse;
 using google::cloud::odbc_bq_driver_internal::IsLengthSensitiveType;
@@ -198,6 +199,20 @@ SQLRETURN SQLFetchInternal(SQLHSTMT statement_handle) {
   if (result_set.cursor >= result_set.rows.size()) {
     LOG(INFO) << "SQLFetch:: cursor: " << result_set.cursor
               << " is >= result set size: " << result_set.rows.size();
+    if (handle.WasHtapiEnabled()) {
+      int rowset_size = ard.GetHeaderRecord().array_size;
+      if (!rowset_size) {
+        rowset_size = 1;
+      }
+      DescriptorHandle& ird = handle.GetDescriptorHandle(DescriptorType::kIRD);
+      StatusRecord status = ParallelFetchAndWrite(handle, rowset_size, ard, ird);
+
+      // Clear cached rows and reset cursor since parallel write directly populated the buffers
+      handle.GetResultSet().rows.clear();
+      handle.GetResultSet().cursor = -1;
+
+      return LogAndReturnCode(handle, status);
+    }
     StatusRecord next_page_status = FetchNextResultSet(handle);
     if (!next_page_status.ok()) {
       LOG(ERROR) << "SQLFetch:: " << next_page_status.message;