| 266 | } |
| 267 | |
| 268 | Status DataSourceScanNode::GetNextInputBatch() { |
| 269 | next_row_idx_ = 0; |
| 270 | Ubsan::MemSet(cols_next_val_idx_.data(), 0, sizeof(int) * cols_next_val_idx_.size()); |
| 271 | |
| 272 | input_batch_.reset(new TGetNextResult()); |
| 273 | // Short-circuit: another scanner thread already reached EOS on the shared stream. |
| 274 | if (shared_conn_->eos()) { |
| 275 | input_batch_->__set_eos(true); |
| 276 | return Status::OK(); |
| 277 | } |
| 278 | // FetchBatch() serializes the JNI call via the Java-side fetchLock; this scan node |
| 279 | // then materializes the returned batch in parallel while others may be fetching. |
| 280 | COUNTER_ADD(num_ext_data_source_get_next_, 1); |
| 281 | COUNTER_ADD(jdbc_jni_call_count_, 1); |
| 282 | { |
| 283 | SCOPED_TIMER(jdbc_jni_wait_timer_); |
| 284 | RETURN_IF_ERROR(shared_conn_->FetchBatch(input_batch_.get())); |
| 285 | } |
| 286 | if (input_batch_->__isset.cursor_fetch_time_ns) { |
| 287 | COUNTER_ADD(jdbc_cursor_fetch_timer_, input_batch_->cursor_fetch_time_ns); |
| 288 | } |
| 289 | if (input_batch_->__isset.lock_wait_time_ns) { |
| 290 | COUNTER_ADD(jdbc_lock_wait_timer_, input_batch_->lock_wait_time_ns); |
| 291 | } |
| 292 | |
| 293 | RETURN_IF_ERROR(ValidateRowBatchSize()); |
| 294 | if (!InputBatchHasNext() && !input_batch_->eos) { |
| 295 | // The data source should have set eos, but if it didn't we should just log a |
| 296 | // warning and continue as if it had. |
| 297 | VLOG_QUERY << "Data source " << data_src_node_.data_source.name << " returned no " |
| 298 | << "rows but did not set 'eos'. No more rows will be fetched from the " |
| 299 | << "data source."; |
| 300 | input_batch_->eos = true; |
| 301 | } |
| 302 | return Status::OK(); |
| 303 | } |
| 304 | |
| 305 | // Sets the decimal value in the slot. Inline method to avoid nested switch statements. |
| 306 | inline Status SetDecimalVal(const ColumnType& type, char* bytes, int len, |