| 258 | } |
| 259 | |
| 260 | Status TupleCacheNode::GetNext( |
| 261 | RuntimeState* state, RowBatch* output_row_batch, bool* eos) { |
| 262 | SCOPED_TIMER(runtime_profile()->total_time_counter()); |
| 263 | ScopedGetNextEventAdder ea(this, eos); |
| 264 | RETURN_IF_ERROR(ExecDebugAction(TExecNodePhase::GETNEXT, state)); |
| 265 | RETURN_IF_CANCELLED(state); |
| 266 | RETURN_IF_ERROR(QueryMaintenance(state)); |
| 267 | |
| 268 | // Save the number of rows in case GetNext() is called with a non-empty batch, |
| 269 | // which can happen in a subplan. |
| 270 | int num_rows_before = output_row_batch->num_rows(); |
| 271 | |
| 272 | // If we have a Reader, return the next batch from it. |
| 273 | // Else GetNext from child, write to Writer, and return the batch. |
| 274 | if (reader_) { |
| 275 | Status status = reader_->GetNext(state, buffer_pool_client(), output_row_batch, eos); |
| 276 | if (status.ok()) { |
| 277 | cached_rowbatch_returned_to_caller_ = true; |
| 278 | // Close the child now that there is no hope of recovery in case of failure. This |
| 279 | // allows it to tear down any state and notify any affected threads that the |
| 280 | // children won't ever reach Open(). |
| 281 | child(0)->Close(state); |
| 282 | } else { |
| 283 | // If we have returned a cached row batch to the caller, then it is not safe |
| 284 | // to try to get any rows from the child as they could be duplicates. Any |
| 285 | // error needs to end the query. |
| 286 | if (cached_rowbatch_returned_to_caller_) return status; |
| 287 | |
| 288 | // We haven't returned a RowBatch to the caller yet, so we can recover by aborting |
| 289 | // the read from the cache and fetching from the child. We won't try to write to |
| 290 | // the cache. |
| 291 | LOG(WARNING) << "Unable to read cache file: " << status.GetDetail() |
| 292 | << "Falling back to regular non-cached path."; |
| 293 | reader_.reset(); |
| 294 | // If reader_ is set, then the child was never opened and needs to be opened now |
| 295 | RETURN_IF_ERROR(child(0)->Open(state)); |
| 296 | RETURN_IF_ERROR(child(0)->GetNext(state, output_row_batch, eos)); |
| 297 | } |
| 298 | } else { |
| 299 | RETURN_IF_ERROR(child(0)->GetNext(state, output_row_batch, eos)); |
| 300 | if (writer_) { |
| 301 | Status status = writer_->Write(state, output_row_batch); |
| 302 | TupleCacheMgr* tuple_cache_mgr = ExecEnv::GetInstance()->tuple_cache_mgr(); |
| 303 | // If there was an error or we exceeded the file size limit, stop caching but |
| 304 | // continue reading from the child node. |
| 305 | if (!status.ok()) { |
| 306 | bool set_tombstone = false; |
| 307 | if (status.code() == TErrorCode::TUPLE_CACHE_ENTRY_SIZE_LIMIT_EXCEEDED) { |
| 308 | VLOG_FILE << "Tuple Cache entry for " << combined_key_ |
| 309 | << " hit the maximum file size: " << status.GetDetail(); |
| 310 | COUNTER_ADD(num_halted_counter_, 1); |
| 311 | set_tombstone = true; |
| 312 | } else if (status.code() == |
| 313 | TErrorCode::TUPLE_CACHE_OUTSTANDING_WRITE_LIMIT_EXCEEDED) { |
| 314 | VLOG_FILE << "Tuple Cache entry for " << combined_key_ |
| 315 | << " hit the outstanding writes limit: " << status.GetDetail(); |
| 316 | COUNTER_ADD(num_backpressure_halted_counter_, 1); |
| 317 | } else { |
nothing calls this directly
no test coverage detected