From 790b96a90ef2c50a0efdce8f804bb90aa6618a05 Mon Sep 17 00:00:00 2001 From: Shivam Dabas Date: Fri, 24 Jul 2026 11:17:10 +0530 Subject: [PATCH 1/5] improve(bq_driver): SQLtables improve --- .../odbc/bq_client_interface/datasets.cc | 3 ++ .../odbc/bq_client_interface/projects.cc | 4 ++ .../cloud/odbc/bq_client_interface/tables.cc | 1 + google/cloud/odbc/bq_client_interface/utils.h | 6 +++ .../bq_driver/internal/odbc_sql_tables.cc | 47 ++++++++++++++----- 5 files changed, 48 insertions(+), 13 deletions(-) diff --git a/google/cloud/odbc/bq_client_interface/datasets.cc b/google/cloud/odbc/bq_client_interface/datasets.cc index 834403ef66..7111f39dea 100644 --- a/google/cloud/odbc/bq_client_interface/datasets.cc +++ b/google/cloud/odbc/bq_client_interface/datasets.cc @@ -13,6 +13,7 @@ // limitations under the License. #include "google/cloud/odbc/bq_client_interface/datasets.h" +#include "google/cloud/odbc/bq_client_interface/utils.h" #include "google/cloud/odbc/internal/status_record_or.h" #include "google/cloud/bigquery/v2/minimal/internal/dataset_client.h" #include "absl/log/log.h" @@ -58,6 +59,7 @@ StatusRecordOr> ListAllDatasets( ListDatasetsRequest request; request.set_project_id(project_id); request.set_all_datasets(true); + request.set_max_results(kMetadataPageSize); LOG(INFO) << "ListAllDatasets:: Request body: " << request.DebugString(""); StreamRange datasets_response = dataset_client.ListDatasets(request, options); @@ -86,6 +88,7 @@ StatusRecordOr> FilterDatasets( request.set_project_id(project_id); request.set_all_datasets(dataset_filter.all); request.set_filter(dataset_filter.filter); + request.set_max_results(kMetadataPageSize); LOG(INFO) << "FilterDatasets:: Request body: " << request.DebugString(""); StreamRange datasets_response = diff --git a/google/cloud/odbc/bq_client_interface/projects.cc b/google/cloud/odbc/bq_client_interface/projects.cc index d21cdc2422..9a11afb6f6 100644 --- a/google/cloud/odbc/bq_client_interface/projects.cc +++ b/google/cloud/odbc/bq_client_interface/projects.cc @@ -14,6 +14,7 @@ #include "google/cloud/odbc/bq_client_interface/projects.h" #include "google/cloud/odbc/bq_client_interface/datasets.h" +#include "google/cloud/odbc/bq_client_interface/utils.h" #include "google/cloud/odbc/internal/sql_state_constants.h" #include "google/cloud/odbc/internal/status_record_or.h" #include "google/cloud/resourcemanager/v3/projects.pb.h" @@ -92,6 +93,7 @@ bool IsProjectBQEnabled(std::string const& bq_project_id, StatusRecordOr> ListAllProjects( ProjectClient& project_client, Options const& options) { ListProjectsRequest request; + request.set_max_results(kMetadataPageSize); StreamRange projects_response = project_client.ListProjects(request, options); @@ -112,6 +114,7 @@ StatusRecordOr GetProject(ProjectClient& project_client, std::string const& project_id, Options const& options) { ListProjectsRequest request; + request.set_max_results(kMetadataPageSize); StreamRange projects_response = project_client.ListProjects(request, options); @@ -252,6 +255,7 @@ StatusRecordOr> FilterProjects( ProjectClient& project_client, std::vector const& project_ids, Options const& options) { ListProjectsRequest request; + request.set_max_results(kMetadataPageSize); StreamRange projects_response = project_client.ListProjects(request, options); diff --git a/google/cloud/odbc/bq_client_interface/tables.cc b/google/cloud/odbc/bq_client_interface/tables.cc index d51903cb0c..654881b8f9 100644 --- a/google/cloud/odbc/bq_client_interface/tables.cc +++ b/google/cloud/odbc/bq_client_interface/tables.cc @@ -69,6 +69,7 @@ StatusRecordOr> ListAllTables( ListTablesRequest request; request.set_project_id(project_id); request.set_dataset_id(dataset_id); + request.set_max_results(kMetadataPageSize); LOG(INFO) << "ListAllTables:: Request body: " << request.DebugString(""); StreamRange tables_response = diff --git a/google/cloud/odbc/bq_client_interface/utils.h b/google/cloud/odbc/bq_client_interface/utils.h index 60ae235e9c..a7d30483e5 100644 --- a/google/cloud/odbc/bq_client_interface/utils.h +++ b/google/cloud/odbc/bq_client_interface/utils.h @@ -17,6 +17,7 @@ #include "absl/log/log.h" #include +#include #include #include #include @@ -27,6 +28,11 @@ struct MaxRetriesOption { using Type = int; }; +// Page size requested from paginated metadata list APIs (projects.list, +// datasets.list, tables.list). When maxResults is unset the BigQuery REST API +// returns small pages (50), which multiplies sequential HTTP round trips +constexpr std::int32_t kMetadataPageSize = 1000; + // URL-encodes a value for safe use as a single path segment. inline std::string UrlEncodeSegment(std::string const& value) { std::ostringstream os; diff --git a/google/cloud/odbc/bq_driver/internal/odbc_sql_tables.cc b/google/cloud/odbc/bq_driver/internal/odbc_sql_tables.cc index 88e7f6db84..05614540d2 100644 --- a/google/cloud/odbc/bq_driver/internal/odbc_sql_tables.cc +++ b/google/cloud/odbc/bq_driver/internal/odbc_sql_tables.cc @@ -444,23 +444,46 @@ StatusRecordOr GetResultSetForTables( }; std::vector tasks; - for (auto const& project_id : project_list) { + std::shared_ptr trace_option = TraceOptions::GetTraceOption(); + int max_threads = trace_option->max_threads; + + if (metadata_id == SQL_TRUE) { // When SQL_ATTR_METADATA_ID is true, the schema argument is an exact // dataset identifier (not a pattern). Skip listing every dataset in the // project via datasets.list - if (metadata_id == SQL_TRUE) { + for (auto const& project_id : project_list) { tasks.push_back({project_id, dataset_filter}); - continue; } - auto datasets_status_record_or = GetFilteredDatasetIds( - bq_client, project_id, dataset_filter, metadata_id); - if (!datasets_status_record_or) { - LOG(ERROR) << "GetResultSetForTables::GetFilteredDatasetIds:: " - << datasets_status_record_or.GetStatusRecord().message; - return datasets_status_record_or.GetStatusRecord(); + } else { + // Enumerate datasets for each project in parallel. Previously this was a + // serial loop issuing one datasets.list REST call per project, which + // dominated SQLTables latency when the catalog pattern matches many + // projects (e.g. catalog = "%"). + auto dataset_task = [&](std::string const& project_id) + -> StatusRecordOr> { + auto datasets_status_record_or = GetFilteredDatasetIds( + bq_client, project_id, dataset_filter, metadata_id); + if (!datasets_status_record_or) { + LOG(ERROR) << "GetResultSetForTables::GetFilteredDatasetIds:: " + << datasets_status_record_or.GetStatusRecord().message; + return datasets_status_record_or.GetStatusRecord(); + } + std::vector project_tasks; + project_tasks.reserve(datasets_status_record_or->size()); + for (auto& dataset_id : *datasets_status_record_or) { + project_tasks.push_back({project_id, std::move(dataset_id)}); + } + return project_tasks; + }; + auto dataset_results_or = + ExecuteParallelTasks>( + max_threads, project_list, dataset_task); + if (!dataset_results_or) { + return dataset_results_or.GetStatusRecord(); } - for (auto const& dataset_id : *datasets_status_record_or) { - tasks.push_back({project_id, dataset_id}); + for (auto& project_tasks : *dataset_results_or) { + tasks.insert(tasks.end(), std::make_move_iterator(project_tasks.begin()), + std::make_move_iterator(project_tasks.end())); } } @@ -491,8 +514,6 @@ StatusRecordOr GetResultSetForTables( }; // 3. Execute tasks using the generic utility - std::shared_ptr trace_option = TraceOptions::GetTraceOption(); - int max_threads = trace_option->max_threads; auto parallel_results_or = ExecuteParallelTasks( max_threads, tasks, parallel_func); From 8f7240c1b9c01f796e2b2ce3ea1f6f3c51e842c2 Mon Sep 17 00:00:00 2001 From: Shivam Dabas Date: Mon, 27 Jul 2026 11:07:36 +0530 Subject: [PATCH 2/5] revert sqltables change --- google/cloud/odbc/bq_client_interface/utils.h | 2 +- .../bq_driver/internal/odbc_sql_tables.cc | 47 +++++-------------- 2 files changed, 14 insertions(+), 35 deletions(-) diff --git a/google/cloud/odbc/bq_client_interface/utils.h b/google/cloud/odbc/bq_client_interface/utils.h index a7d30483e5..227d3bed54 100644 --- a/google/cloud/odbc/bq_client_interface/utils.h +++ b/google/cloud/odbc/bq_client_interface/utils.h @@ -31,7 +31,7 @@ struct MaxRetriesOption { // Page size requested from paginated metadata list APIs (projects.list, // datasets.list, tables.list). When maxResults is unset the BigQuery REST API // returns small pages (50), which multiplies sequential HTTP round trips -constexpr std::int32_t kMetadataPageSize = 1000; +constexpr std::int32_t kMetadataPageSize = 10000; // URL-encodes a value for safe use as a single path segment. inline std::string UrlEncodeSegment(std::string const& value) { diff --git a/google/cloud/odbc/bq_driver/internal/odbc_sql_tables.cc b/google/cloud/odbc/bq_driver/internal/odbc_sql_tables.cc index 05614540d2..88e7f6db84 100644 --- a/google/cloud/odbc/bq_driver/internal/odbc_sql_tables.cc +++ b/google/cloud/odbc/bq_driver/internal/odbc_sql_tables.cc @@ -444,46 +444,23 @@ StatusRecordOr GetResultSetForTables( }; std::vector tasks; - std::shared_ptr trace_option = TraceOptions::GetTraceOption(); - int max_threads = trace_option->max_threads; - - if (metadata_id == SQL_TRUE) { + for (auto const& project_id : project_list) { // When SQL_ATTR_METADATA_ID is true, the schema argument is an exact // dataset identifier (not a pattern). Skip listing every dataset in the // project via datasets.list - for (auto const& project_id : project_list) { + if (metadata_id == SQL_TRUE) { tasks.push_back({project_id, dataset_filter}); + continue; } - } else { - // Enumerate datasets for each project in parallel. Previously this was a - // serial loop issuing one datasets.list REST call per project, which - // dominated SQLTables latency when the catalog pattern matches many - // projects (e.g. catalog = "%"). - auto dataset_task = [&](std::string const& project_id) - -> StatusRecordOr> { - auto datasets_status_record_or = GetFilteredDatasetIds( - bq_client, project_id, dataset_filter, metadata_id); - if (!datasets_status_record_or) { - LOG(ERROR) << "GetResultSetForTables::GetFilteredDatasetIds:: " - << datasets_status_record_or.GetStatusRecord().message; - return datasets_status_record_or.GetStatusRecord(); - } - std::vector project_tasks; - project_tasks.reserve(datasets_status_record_or->size()); - for (auto& dataset_id : *datasets_status_record_or) { - project_tasks.push_back({project_id, std::move(dataset_id)}); - } - return project_tasks; - }; - auto dataset_results_or = - ExecuteParallelTasks>( - max_threads, project_list, dataset_task); - if (!dataset_results_or) { - return dataset_results_or.GetStatusRecord(); + auto datasets_status_record_or = GetFilteredDatasetIds( + bq_client, project_id, dataset_filter, metadata_id); + if (!datasets_status_record_or) { + LOG(ERROR) << "GetResultSetForTables::GetFilteredDatasetIds:: " + << datasets_status_record_or.GetStatusRecord().message; + return datasets_status_record_or.GetStatusRecord(); } - for (auto& project_tasks : *dataset_results_or) { - tasks.insert(tasks.end(), std::make_move_iterator(project_tasks.begin()), - std::make_move_iterator(project_tasks.end())); + for (auto const& dataset_id : *datasets_status_record_or) { + tasks.push_back({project_id, dataset_id}); } } @@ -514,6 +491,8 @@ StatusRecordOr GetResultSetForTables( }; // 3. Execute tasks using the generic utility + std::shared_ptr trace_option = TraceOptions::GetTraceOption(); + int max_threads = trace_option->max_threads; auto parallel_results_or = ExecuteParallelTasks( max_threads, tasks, parallel_func); From 71d8af6c85f110021a7056333df47daa983025ce Mon Sep 17 00:00:00 2001 From: Shivam Dabas Date: Tue, 28 Jul 2026 18:22:18 +0530 Subject: [PATCH 3/5] imrove parallel --- google/cloud/odbc/bq_driver/internal/utils.h | 128 ++++++++++-------- .../odbc/bq_driver/internal/utils_test.cc | 66 +++++++++ 2 files changed, 140 insertions(+), 54 deletions(-) diff --git a/google/cloud/odbc/bq_driver/internal/utils.h b/google/cloud/odbc/bq_driver/internal/utils.h index a557ce6725..fdc77c527f 100644 --- a/google/cloud/odbc/bq_driver/internal/utils.h +++ b/google/cloud/odbc/bq_driver/internal/utils.h @@ -33,6 +33,7 @@ extern HINSTANCE g_hDllInstance; #include "google/cloud/status_or.h" #include "re2/re2.h" #include +#include #include #include #include @@ -42,9 +43,12 @@ extern HINSTANCE g_hDllInstance; #include #include #include +#include +#include #include #include #include +#include #include namespace google::cloud::odbc_bq_driver_internal { @@ -102,75 +106,91 @@ size_t BufferSizeForType(SQLSMALLINT type, size_t requested); // TaskInput: The type of a single item in the input vector. // TaskResult: The type of data returned by the function on success. // -// Returns: A vector containing the results of all successful tasks, or the -// StatusRecord of the first error encountered. +// Returns: A vector containing the results of all successful tasks in input +// order, or the StatusRecord of the first (in input order) error encountered. +// After a task fails, tasks not yet started are skipped, but tasks already +// in flight run to completion before this function returns. +// +// Implemented as a fixed pool of max_threads workers pulling the next input +// from a shared atomic index. Each worker moves on to the next input the +// moment its current task finishes, so one slow task (a straggler REST call) +// never stalls the dispatch of the remaining work -- a previous +// dispatcher-based implementation blocked on the oldest in-flight task to +// free a slot, which serialized the whole batch behind stragglers and made +// catalog enumeration latency vary heavily from run to run. template odbc_internal::StatusRecordOr> ExecuteParallelTasks( std::uint32_t max_threads, std::vector const& inputs, std::function(TaskInput const&)> task_func) { - using FutureType = std::future>; - - std::vector active_futures; - std::vector aggregated_results; - odbc_internal::StatusRecord error_status = odbc_internal::StatusRecord::Ok(); - bool error_occurred = false; - - // Helper to collect result from a finished future - auto process_future = [&](FutureType& f) { - auto result = f.get(); - if (!result) { - // Record the first error that occurs, but keep draining threads - if (!error_occurred) { - error_status = result.GetStatusRecord(); - error_occurred = true; + if (inputs.empty()) { + return std::vector{}; + } + // A misconfigured MaxThreads of 0 previously spun the dispatch loop forever; + // treat it as serial execution. Never spawn more workers than inputs. + std::size_t const num_workers = (std::min)( + static_cast((std::max)(max_threads, std::uint32_t{1})), + inputs.size()); + + // One pre-sized slot per input: workers write disjoint indices, so no + // synchronization is needed, and results come back in input order. + std::vector>> slots( + inputs.size()); + std::atomic next_index{0}; + std::atomic error_occurred{false}; + std::exception_ptr first_exception; + std::mutex exception_mutex; + + auto worker = [&] { + while (!error_occurred.load(std::memory_order_relaxed)) { + std::size_t const index = + next_index.fetch_add(1, std::memory_order_relaxed); + if (index >= inputs.size()) { + return; } - } else if (!error_occurred) { - // Only store results if we are still in a success state - aggregated_results.push_back(std::move(*result)); - } - }; - - for (auto const& input : inputs) { - // If we have hit the thread limit, wait for at least one thread to finish - while (active_futures.size() >= max_threads) { - bool slot_freed = false; - for (auto it = active_futures.begin(); it != active_futures.end();) { - // Check if ready without blocking - if (it->wait_for(std::chrono::milliseconds(1)) == - std::future_status::ready) { - process_future(*it); - it = active_futures.erase(it); - slot_freed = true; - break; // We freed a slot, proceed to launch next task + try { + auto result = task_func(inputs[index]); + if (!result) { + error_occurred.store(true, std::memory_order_relaxed); } - ++it; - } - - // If no threads finished yet, block on the oldest one to prevent spinning - if (!slot_freed && !active_futures.empty()) { - process_future(active_futures.front()); - active_futures.erase(active_futures.begin()); + slots[index] = std::move(result); + } catch (...) { + // Propagate the exception on the calling thread (std::async did this + // via future::get); letting it escape a worker would call terminate. + std::lock_guard lock(exception_mutex); + if (!first_exception) { + first_exception = std::current_exception(); + } + error_occurred.store(true, std::memory_order_relaxed); + return; } } + }; - // If an error occurred previously, we stop spawning new tasks, - // but the loop continues to ensure we drain existing futures safely. - if (!error_occurred) { - active_futures.push_back( - std::async(std::launch::async, task_func, input)); - } + std::vector workers; + workers.reserve(num_workers); + for (std::size_t i = 0; i < num_workers; ++i) { + workers.emplace_back(worker); } - - // Wait for and collect all remaining threads - for (auto& f : active_futures) { - process_future(f); + for (auto& thread : workers) { + thread.join(); } - if (error_occurred) { - return error_status; + if (first_exception) { + std::rethrow_exception(first_exception); } + std::vector aggregated_results; + aggregated_results.reserve(inputs.size()); + for (auto& slot : slots) { + if (!slot.has_value()) { + continue; // Skipped after an error; an errored slot below reports it. + } + if (!*slot) { + return slot->GetStatusRecord(); + } + aggregated_results.push_back(std::move(**slot)); + } return aggregated_results; } diff --git a/google/cloud/odbc/bq_driver/internal/utils_test.cc b/google/cloud/odbc/bq_driver/internal/utils_test.cc index f134d0a0ca..41fe818c6b 100644 --- a/google/cloud/odbc/bq_driver/internal/utils_test.cc +++ b/google/cloud/odbc/bq_driver/internal/utils_test.cc @@ -27,6 +27,7 @@ using ::google::cloud::odbc_internal::SQLStates; using ::google::cloud::odbc_internal::StatusRecord; using ::google::cloud::odbc_internal::StatusRecordOr; using google::cloud::odbc_testing_utils::StatusRecordIs; +using ::testing::ElementsAre; using ::testing::HasSubstr; using ::testing::IsEmpty; using ::testing::UnorderedElementsAre; @@ -1016,4 +1017,69 @@ TEST(ExecuteParallelTasksTest, RespectsSlidingWindow) { EXPECT_GE(duration, (task_count / max_threads) * min_sleep_ms); } +TEST(ExecuteParallelTasksTest, ResultsPreserveInputOrder) { + std::vector inputs = {5, 1, 4, 2, 3}; + + // Sleep inversely to the input so tasks complete in the reverse of their + // submission order; the results must still come back in input order. + auto task = [](int input) -> StatusRecordOr { + std::this_thread::sleep_for(std::chrono::milliseconds(input * 10)); + return input; + }; + + auto result = ExecuteParallelTasks(5, inputs, task); + + ASSERT_STATUS_RECORD_OK(result); + EXPECT_THAT(*result, ElementsAre(5, 1, 4, 2, 3)); +} + +TEST(ExecuteParallelTasksTest, ZeroMaxThreadsRunsSerially) { + // A misconfigured MaxThreads of 0 must not hang; it degrades to serial + // execution. + std::vector inputs = {1, 2, 3}; + auto task = [](int input) -> StatusRecordOr { return input * 2; }; + + auto result = ExecuteParallelTasks(0, inputs, task); + + ASSERT_STATUS_RECORD_OK(result); + EXPECT_THAT(*result, ElementsAre(2, 4, 6)); +} + +TEST(ExecuteParallelTasksTest, StragglerDoesNotStallDispatch) { + // One slow task at the head of the queue must not prevent the other worker + // from chewing through the remaining short tasks. The previous + // dispatcher-based implementation blocked on the oldest in-flight future, + // so this workload took ~straggler + sum(short tasks) instead of + // ~max(straggler, sum(short tasks)). + int const straggler_ms = 300; + int const short_task_ms = 20; + int const short_task_count = 8; + + std::vector inputs; + inputs.push_back(straggler_ms); + for (int i = 0; i < short_task_count; ++i) { + inputs.push_back(short_task_ms); + } + + auto sleeping_task = [](int sleep_ms) -> StatusRecordOr { + std::this_thread::sleep_for(std::chrono::milliseconds(sleep_ms)); + return sleep_ms; + }; + + auto start_time = std::chrono::steady_clock::now(); + auto result = + ExecuteParallelTasks(2, inputs, sleeping_task); + auto duration = std::chrono::duration_cast( + std::chrono::steady_clock::now() - start_time) + .count(); + + ASSERT_STATUS_RECORD_OK(result); + EXPECT_EQ(result->size(), inputs.size()); + // Ideal: worker 1 takes the straggler (300ms) while worker 2 runs the eight + // 20ms tasks (160ms) => ~300ms total. The old scheduler needed ~460ms + // (straggler + all short tasks serialized behind it). Use a generous bound + // to stay robust on loaded CI machines while still distinguishing the two. + EXPECT_LT(duration, straggler_ms + short_task_count * short_task_ms - 60); +} + } // namespace google::cloud::odbc_bq_driver_internal From e89bd7cce30dc80d49d6f07a437a81b02ae789b2 Mon Sep 17 00:00:00 2001 From: Shivam Dabas Date: Tue, 28 Jul 2026 18:22:33 +0530 Subject: [PATCH 4/5] Revert "revert sqltables change" This reverts commit 8f7240c1b9c01f796e2b2ce3ea1f6f3c51e842c2. --- google/cloud/odbc/bq_client_interface/utils.h | 2 +- .../bq_driver/internal/odbc_sql_tables.cc | 47 ++++++++++++++----- 2 files changed, 35 insertions(+), 14 deletions(-) diff --git a/google/cloud/odbc/bq_client_interface/utils.h b/google/cloud/odbc/bq_client_interface/utils.h index 227d3bed54..a7d30483e5 100644 --- a/google/cloud/odbc/bq_client_interface/utils.h +++ b/google/cloud/odbc/bq_client_interface/utils.h @@ -31,7 +31,7 @@ struct MaxRetriesOption { // Page size requested from paginated metadata list APIs (projects.list, // datasets.list, tables.list). When maxResults is unset the BigQuery REST API // returns small pages (50), which multiplies sequential HTTP round trips -constexpr std::int32_t kMetadataPageSize = 10000; +constexpr std::int32_t kMetadataPageSize = 1000; // URL-encodes a value for safe use as a single path segment. inline std::string UrlEncodeSegment(std::string const& value) { diff --git a/google/cloud/odbc/bq_driver/internal/odbc_sql_tables.cc b/google/cloud/odbc/bq_driver/internal/odbc_sql_tables.cc index 88e7f6db84..05614540d2 100644 --- a/google/cloud/odbc/bq_driver/internal/odbc_sql_tables.cc +++ b/google/cloud/odbc/bq_driver/internal/odbc_sql_tables.cc @@ -444,23 +444,46 @@ StatusRecordOr GetResultSetForTables( }; std::vector tasks; - for (auto const& project_id : project_list) { + std::shared_ptr trace_option = TraceOptions::GetTraceOption(); + int max_threads = trace_option->max_threads; + + if (metadata_id == SQL_TRUE) { // When SQL_ATTR_METADATA_ID is true, the schema argument is an exact // dataset identifier (not a pattern). Skip listing every dataset in the // project via datasets.list - if (metadata_id == SQL_TRUE) { + for (auto const& project_id : project_list) { tasks.push_back({project_id, dataset_filter}); - continue; } - auto datasets_status_record_or = GetFilteredDatasetIds( - bq_client, project_id, dataset_filter, metadata_id); - if (!datasets_status_record_or) { - LOG(ERROR) << "GetResultSetForTables::GetFilteredDatasetIds:: " - << datasets_status_record_or.GetStatusRecord().message; - return datasets_status_record_or.GetStatusRecord(); + } else { + // Enumerate datasets for each project in parallel. Previously this was a + // serial loop issuing one datasets.list REST call per project, which + // dominated SQLTables latency when the catalog pattern matches many + // projects (e.g. catalog = "%"). + auto dataset_task = [&](std::string const& project_id) + -> StatusRecordOr> { + auto datasets_status_record_or = GetFilteredDatasetIds( + bq_client, project_id, dataset_filter, metadata_id); + if (!datasets_status_record_or) { + LOG(ERROR) << "GetResultSetForTables::GetFilteredDatasetIds:: " + << datasets_status_record_or.GetStatusRecord().message; + return datasets_status_record_or.GetStatusRecord(); + } + std::vector project_tasks; + project_tasks.reserve(datasets_status_record_or->size()); + for (auto& dataset_id : *datasets_status_record_or) { + project_tasks.push_back({project_id, std::move(dataset_id)}); + } + return project_tasks; + }; + auto dataset_results_or = + ExecuteParallelTasks>( + max_threads, project_list, dataset_task); + if (!dataset_results_or) { + return dataset_results_or.GetStatusRecord(); } - for (auto const& dataset_id : *datasets_status_record_or) { - tasks.push_back({project_id, dataset_id}); + for (auto& project_tasks : *dataset_results_or) { + tasks.insert(tasks.end(), std::make_move_iterator(project_tasks.begin()), + std::make_move_iterator(project_tasks.end())); } } @@ -491,8 +514,6 @@ StatusRecordOr GetResultSetForTables( }; // 3. Execute tasks using the generic utility - std::shared_ptr trace_option = TraceOptions::GetTraceOption(); - int max_threads = trace_option->max_threads; auto parallel_results_or = ExecuteParallelTasks( max_threads, tasks, parallel_func); From 608538cbb1c7122798ce5de2e5d834bcd2a965ac Mon Sep 17 00:00:00 2001 From: Shivam Dabas Date: Tue, 28 Jul 2026 18:22:35 +0530 Subject: [PATCH 5/5] Revert "improve(bq_driver): SQLtables improve" This reverts commit 790b96a90ef2c50a0efdce8f804bb90aa6618a05. --- .../odbc/bq_client_interface/datasets.cc | 3 -- .../odbc/bq_client_interface/projects.cc | 4 -- .../cloud/odbc/bq_client_interface/tables.cc | 1 - google/cloud/odbc/bq_client_interface/utils.h | 6 --- .../bq_driver/internal/odbc_sql_tables.cc | 47 +++++-------------- 5 files changed, 13 insertions(+), 48 deletions(-) diff --git a/google/cloud/odbc/bq_client_interface/datasets.cc b/google/cloud/odbc/bq_client_interface/datasets.cc index 7111f39dea..834403ef66 100644 --- a/google/cloud/odbc/bq_client_interface/datasets.cc +++ b/google/cloud/odbc/bq_client_interface/datasets.cc @@ -13,7 +13,6 @@ // limitations under the License. #include "google/cloud/odbc/bq_client_interface/datasets.h" -#include "google/cloud/odbc/bq_client_interface/utils.h" #include "google/cloud/odbc/internal/status_record_or.h" #include "google/cloud/bigquery/v2/minimal/internal/dataset_client.h" #include "absl/log/log.h" @@ -59,7 +58,6 @@ StatusRecordOr> ListAllDatasets( ListDatasetsRequest request; request.set_project_id(project_id); request.set_all_datasets(true); - request.set_max_results(kMetadataPageSize); LOG(INFO) << "ListAllDatasets:: Request body: " << request.DebugString(""); StreamRange datasets_response = dataset_client.ListDatasets(request, options); @@ -88,7 +86,6 @@ StatusRecordOr> FilterDatasets( request.set_project_id(project_id); request.set_all_datasets(dataset_filter.all); request.set_filter(dataset_filter.filter); - request.set_max_results(kMetadataPageSize); LOG(INFO) << "FilterDatasets:: Request body: " << request.DebugString(""); StreamRange datasets_response = diff --git a/google/cloud/odbc/bq_client_interface/projects.cc b/google/cloud/odbc/bq_client_interface/projects.cc index 9a11afb6f6..d21cdc2422 100644 --- a/google/cloud/odbc/bq_client_interface/projects.cc +++ b/google/cloud/odbc/bq_client_interface/projects.cc @@ -14,7 +14,6 @@ #include "google/cloud/odbc/bq_client_interface/projects.h" #include "google/cloud/odbc/bq_client_interface/datasets.h" -#include "google/cloud/odbc/bq_client_interface/utils.h" #include "google/cloud/odbc/internal/sql_state_constants.h" #include "google/cloud/odbc/internal/status_record_or.h" #include "google/cloud/resourcemanager/v3/projects.pb.h" @@ -93,7 +92,6 @@ bool IsProjectBQEnabled(std::string const& bq_project_id, StatusRecordOr> ListAllProjects( ProjectClient& project_client, Options const& options) { ListProjectsRequest request; - request.set_max_results(kMetadataPageSize); StreamRange projects_response = project_client.ListProjects(request, options); @@ -114,7 +112,6 @@ StatusRecordOr GetProject(ProjectClient& project_client, std::string const& project_id, Options const& options) { ListProjectsRequest request; - request.set_max_results(kMetadataPageSize); StreamRange projects_response = project_client.ListProjects(request, options); @@ -255,7 +252,6 @@ StatusRecordOr> FilterProjects( ProjectClient& project_client, std::vector const& project_ids, Options const& options) { ListProjectsRequest request; - request.set_max_results(kMetadataPageSize); StreamRange projects_response = project_client.ListProjects(request, options); diff --git a/google/cloud/odbc/bq_client_interface/tables.cc b/google/cloud/odbc/bq_client_interface/tables.cc index 654881b8f9..d51903cb0c 100644 --- a/google/cloud/odbc/bq_client_interface/tables.cc +++ b/google/cloud/odbc/bq_client_interface/tables.cc @@ -69,7 +69,6 @@ StatusRecordOr> ListAllTables( ListTablesRequest request; request.set_project_id(project_id); request.set_dataset_id(dataset_id); - request.set_max_results(kMetadataPageSize); LOG(INFO) << "ListAllTables:: Request body: " << request.DebugString(""); StreamRange tables_response = diff --git a/google/cloud/odbc/bq_client_interface/utils.h b/google/cloud/odbc/bq_client_interface/utils.h index a7d30483e5..60ae235e9c 100644 --- a/google/cloud/odbc/bq_client_interface/utils.h +++ b/google/cloud/odbc/bq_client_interface/utils.h @@ -17,7 +17,6 @@ #include "absl/log/log.h" #include -#include #include #include #include @@ -28,11 +27,6 @@ struct MaxRetriesOption { using Type = int; }; -// Page size requested from paginated metadata list APIs (projects.list, -// datasets.list, tables.list). When maxResults is unset the BigQuery REST API -// returns small pages (50), which multiplies sequential HTTP round trips -constexpr std::int32_t kMetadataPageSize = 1000; - // URL-encodes a value for safe use as a single path segment. inline std::string UrlEncodeSegment(std::string const& value) { std::ostringstream os; diff --git a/google/cloud/odbc/bq_driver/internal/odbc_sql_tables.cc b/google/cloud/odbc/bq_driver/internal/odbc_sql_tables.cc index 05614540d2..88e7f6db84 100644 --- a/google/cloud/odbc/bq_driver/internal/odbc_sql_tables.cc +++ b/google/cloud/odbc/bq_driver/internal/odbc_sql_tables.cc @@ -444,46 +444,23 @@ StatusRecordOr GetResultSetForTables( }; std::vector tasks; - std::shared_ptr trace_option = TraceOptions::GetTraceOption(); - int max_threads = trace_option->max_threads; - - if (metadata_id == SQL_TRUE) { + for (auto const& project_id : project_list) { // When SQL_ATTR_METADATA_ID is true, the schema argument is an exact // dataset identifier (not a pattern). Skip listing every dataset in the // project via datasets.list - for (auto const& project_id : project_list) { + if (metadata_id == SQL_TRUE) { tasks.push_back({project_id, dataset_filter}); + continue; } - } else { - // Enumerate datasets for each project in parallel. Previously this was a - // serial loop issuing one datasets.list REST call per project, which - // dominated SQLTables latency when the catalog pattern matches many - // projects (e.g. catalog = "%"). - auto dataset_task = [&](std::string const& project_id) - -> StatusRecordOr> { - auto datasets_status_record_or = GetFilteredDatasetIds( - bq_client, project_id, dataset_filter, metadata_id); - if (!datasets_status_record_or) { - LOG(ERROR) << "GetResultSetForTables::GetFilteredDatasetIds:: " - << datasets_status_record_or.GetStatusRecord().message; - return datasets_status_record_or.GetStatusRecord(); - } - std::vector project_tasks; - project_tasks.reserve(datasets_status_record_or->size()); - for (auto& dataset_id : *datasets_status_record_or) { - project_tasks.push_back({project_id, std::move(dataset_id)}); - } - return project_tasks; - }; - auto dataset_results_or = - ExecuteParallelTasks>( - max_threads, project_list, dataset_task); - if (!dataset_results_or) { - return dataset_results_or.GetStatusRecord(); + auto datasets_status_record_or = GetFilteredDatasetIds( + bq_client, project_id, dataset_filter, metadata_id); + if (!datasets_status_record_or) { + LOG(ERROR) << "GetResultSetForTables::GetFilteredDatasetIds:: " + << datasets_status_record_or.GetStatusRecord().message; + return datasets_status_record_or.GetStatusRecord(); } - for (auto& project_tasks : *dataset_results_or) { - tasks.insert(tasks.end(), std::make_move_iterator(project_tasks.begin()), - std::make_move_iterator(project_tasks.end())); + for (auto const& dataset_id : *datasets_status_record_or) { + tasks.push_back({project_id, dataset_id}); } } @@ -514,6 +491,8 @@ StatusRecordOr GetResultSetForTables( }; // 3. Execute tasks using the generic utility + std::shared_ptr trace_option = TraceOptions::GetTraceOption(); + int max_threads = trace_option->max_threads; auto parallel_results_or = ExecuteParallelTasks( max_threads, tasks, parallel_func);