| 100 | // ----------------------------------------------------------------------- |
| 101 | |
| 102 | Status SharedJdbcConnection::Open(const string& jar_path, const string& class_name, |
| 103 | const string& api_version, const string& init_string, |
| 104 | const extdatasource::TOpenParams& params, extdatasource::TOpenResult* result) { |
| 105 | ref_count_.fetch_add(1, std::memory_order_relaxed); |
| 106 | |
| 107 | std::unique_lock<std::mutex> lk(open_mu_); |
| 108 | if (!open_done_) { |
| 109 | // First caller: initialize the executor and open the Java data source. |
| 110 | // All N C++ scanner threads share this single connection. |
| 111 | Status s = executor_.Init(jar_path, class_name, api_version, init_string); |
| 112 | if (s.ok()) s = executor_.Open(params, result); |
| 113 | if (s.ok()) s = StatusFromThrift(result->status); |
| 114 | if (s.ok()) scan_handle_ = result->scan_handle; |
| 115 | open_status_ = s; |
| 116 | open_done_ = true; |
| 117 | open_cv_.notify_all(); |
| 118 | return s; |
| 119 | } |
| 120 | |
| 121 | // Subsequent callers: wait for the first caller to finish, then reuse. |
| 122 | open_cv_.wait(lk, [this] { return open_done_; }); |
| 123 | result->__set_scan_handle(scan_handle_); |
| 124 | return open_status_; |
| 125 | } |
| 126 | |
| 127 | Status SharedJdbcConnection::FetchBatch(extdatasource::TGetNextResult* result) { |
| 128 | // Fast path: another thread already exhausted the stream. |
nothing calls this directly
no test coverage detected