| 84 | } |
| 85 | |
| 86 | Status TupleCacheNode::Open(RuntimeState* state) { |
| 87 | SCOPED_TIMER(runtime_profile()->total_time_counter()); |
| 88 | ScopedOpenEventAdder ea(this); |
| 89 | RETURN_IF_ERROR(ExecNode::Open(state)); |
| 90 | |
| 91 | // The frontend cannot create a TupleCacheNode if enable_tuple_cache=false |
| 92 | // Fail the query if we see this. |
| 93 | if (!state->query_options().enable_tuple_cache) { |
| 94 | return Status("Invalid tuple caching configuration: enable_tuple_cache=false"); |
| 95 | } |
| 96 | |
| 97 | TupleCacheMgr* tuple_cache_mgr = ExecEnv::GetInstance()->tuple_cache_mgr(); |
| 98 | handle_ = tuple_cache_mgr->Lookup(combined_key_, true); |
| 99 | if (tuple_cache_mgr->IsAvailableForRead(handle_)) { |
| 100 | if (tuple_cache_mgr->DebugDumpEnabled() && TupleCacheVerificationEnabled(state)) { |
| 101 | // If the node is marked to skip correctness verification, we don't want to read |
| 102 | // from the cache as that would prevent its children from executing. |
| 103 | if (!skip_correctness_verification_) { |
| 104 | VLOG_FILE << "Tuple Cache: correctness verification for " << combined_key_; |
| 105 | DCHECK(num_correctness_verification_counter_ != nullptr); |
| 106 | COUNTER_ADD(num_correctness_verification_counter_, 1); |
| 107 | // We need the original fragment id to construct the path for the reference debug |
| 108 | // cache file. If it's missing from the metadata, we return an error status |
| 109 | // immediately. |
| 110 | string org_fragment_id = |
| 111 | tuple_cache_mgr->GetFragmentIdForTupleCache(combined_key_); |
| 112 | if (org_fragment_id.empty()) { |
| 113 | return Status(TErrorCode::TUPLE_CACHE_INCONSISTENCY, |
| 114 | Substitute("Metadata of tuple cache '$0' is missing for correctness check", |
| 115 | combined_key_)); |
| 116 | } |
| 117 | string ref_sub_dir; |
| 118 | string sub_dir; |
| 119 | string ref_file_path = GetDebugDumpPath(state, org_fragment_id, &ref_sub_dir); |
| 120 | string file_path = GetDebugDumpPath(state, string(), &sub_dir); |
| 121 | DCHECK_EQ(ref_sub_dir, sub_dir); |
| 122 | DCHECK(!ref_sub_dir.empty()); |
| 123 | DCHECK(!ref_file_path.empty()); |
| 124 | DCHECK(!file_path.empty()); |
| 125 | // Create the subdirectory for the debug caches if needed. |
| 126 | RETURN_IF_ERROR(tuple_cache_mgr->CreateDebugDumpSubdir(ref_sub_dir)); |
| 127 | // Open the writer for writing the tuple data from the cache entries to be |
| 128 | // the reference cache data. |
| 129 | debug_dump_text_writer_ref_ = make_unique<TupleTextFileWriter>(ref_file_path); |
| 130 | RETURN_IF_ERROR(debug_dump_text_writer_ref_->Open()); |
| 131 | // Open the writer for writing the tuple data from children in GetNext() to |
| 132 | // compare with the reference debug cache file. |
| 133 | debug_dump_text_writer_ = make_unique<TupleTextFileWriter>(file_path); |
| 134 | RETURN_IF_ERROR(debug_dump_text_writer_->Open()); |
| 135 | } |
| 136 | } else { |
| 137 | reader_ = make_unique<TupleFileReader>( |
| 138 | tuple_cache_mgr->GetPath(handle_), mem_tracker(), runtime_profile()); |
| 139 | Status status = reader_->Open(state); |
| 140 | // Clear reader if it's not usable |
| 141 | if (!status.ok()) { |
| 142 | LOG(WARNING) << "Could not read cache entry for " |
| 143 | << tuple_cache_mgr->GetPath(handle_); |
no test coverage detected