| 98 | } |
| 99 | |
| 100 | bool CacheMachine::addHostFrameToCache(std::unique_ptr<ral::frame::BlazingHostTable> host_table, std::string message_id) { |
| 101 | |
| 102 | // we dont want to add empty tables to a cache, unless we have never added anything |
| 103 | if (!this->something_added || host_table->num_rows() > 0){ |
| 104 | CodeTimer cacheEventTimer; |
| 105 | cacheEventTimer.start(); |
| 106 | |
| 107 | num_rows_added += host_table->num_rows(); |
| 108 | num_bytes_added += host_table->sizeInBytes(); |
| 109 | |
| 110 | if (message_id == ""){ |
| 111 | message_id = this->cache_machine_name; |
| 112 | } |
| 113 | |
| 114 | auto cache_data = std::make_unique<CPUCacheData>(std::move(host_table)); |
| 115 | auto item = std::make_unique<message>(std::move(cache_data), message_id); |
| 116 | this->waitingCache->put(std::move(item)); |
| 117 | this->something_added = true; |
| 118 | |
| 119 | cacheEventTimer.stop(); |
| 120 | if(cache_events_logger) { |
| 121 | cache_events_logger->info("{ral_id}|{query_id}|{message_id}|{cache_id}|{num_rows}|{num_bytes}|{event_type}|{timestamp_begin}|{timestamp_end}|{description}", |
| 122 | "ral_id"_a=(ctx ? ctx->getNodeIndex(ral::communication::CommunicationData::getInstance().getSelfNode()) : -1), |
| 123 | "query_id"_a=(ctx ? ctx->getContextToken() : -1), |
| 124 | "message_id"_a=cache_machine_name, |
| 125 | "cache_id"_a=cache_id, |
| 126 | "num_rows"_a=num_rows_added, |
| 127 | "num_bytes"_a=num_bytes_added, |
| 128 | "event_type"_a="AddHostFrameToCache", |
| 129 | "timestamp_begin"_a=cacheEventTimer.start_time(), |
| 130 | "timestamp_end"_a=cacheEventTimer.end_time(), |
| 131 | "description"_a="Add to CacheMachine"); |
| 132 | } |
| 133 | |
| 134 | return true; |
| 135 | } |
| 136 | |
| 137 | return false; |
| 138 | } |
| 139 | |
| 140 | void CacheMachine::put(size_t index, std::unique_ptr<ral::frame::BlazingTable> table) { |
| 141 | this->addToCache(std::move(table), this->cache_machine_name + "_" + std::to_string(index), true); |
no test coverage detected