| 208 | } |
| 209 | |
| 210 | Status KuduScanner::OpenNextScanToken(const string& scan_token, bool* eos) { |
| 211 | DCHECK(scanner_ == nullptr); |
| 212 | kudu::client::KuduScanner* scanner; |
| 213 | KUDU_RETURN_IF_ERROR(kudu::client::KuduScanToken::DeserializeIntoScanner( |
| 214 | scan_node_->kudu_client(), scan_token, &scanner), |
| 215 | BuildErrorString("Unable to deserialize scan token")); |
| 216 | scanner_.reset(scanner); |
| 217 | |
| 218 | if (state_->query_options().kudu_replica_selection |
| 219 | == TKuduReplicaSelection::LEADER_ONLY) { |
| 220 | KUDU_RETURN_IF_ERROR(scanner_->SetSelection(kudu::client::KuduClient::LEADER_ONLY), |
| 221 | BuildErrorString("Could not set replica selection")); |
| 222 | } |
| 223 | kudu::client::KuduScanner::ReadMode mode; |
| 224 | RETURN_IF_ERROR(StringToKuduReadMode(FLAGS_kudu_read_mode, &mode)); |
| 225 | if (state_->query_options().kudu_read_mode != TKuduReadMode::DEFAULT) { |
| 226 | RETURN_IF_ERROR(StringToKuduReadMode( |
| 227 | PrintValue(state_->query_options().kudu_read_mode), &mode)); |
| 228 | } |
| 229 | KUDU_RETURN_IF_ERROR( |
| 230 | scanner_->SetReadMode(mode), BuildErrorString("Could not set scanner ReadMode")); |
| 231 | if (state_->query_options().kudu_snapshot_read_timestamp_micros > 0) { |
| 232 | KUDU_RETURN_IF_ERROR(scanner_->SetSnapshotMicros( |
| 233 | state_->query_options().kudu_snapshot_read_timestamp_micros), |
| 234 | BuildErrorString("Could not set snapshot timestamp")); |
| 235 | } |
| 236 | KUDU_RETURN_IF_ERROR(scanner_->SetTimeoutMillis(FLAGS_kudu_operation_timeout_ms), |
| 237 | BuildErrorString("Could not set scanner timeout")); |
| 238 | VLOG_ROW << "Starting KuduScanner with ReadMode=" << mode |
| 239 | << " timeout=" << FLAGS_kudu_operation_timeout_ms |
| 240 | << " node with id=" << scan_node_->id() |
| 241 | << " Kudu table=" << scan_node_->table_desc()->table_name(); |
| 242 | |
| 243 | if (!timestamp_slots_.empty()) { |
| 244 | uint64_t row_format_flags = |
| 245 | kudu::client::KuduScanner::PAD_UNIXTIME_MICROS_TO_16_BYTES; |
| 246 | scanner_->SetRowFormatFlags(row_format_flags); |
| 247 | } |
| 248 | |
| 249 | if (scan_node_->filter_ctxs_.size() > 0) { |
| 250 | for (const FilterContext& ctx : scan_node_->filter_ctxs_) { |
| 251 | if (!ctx.filter->HasFilter() || ctx.filter->AlwaysTrue()) { |
| 252 | // If it's always true, the filter won't actually remove any rows so we |
| 253 | // don't need to push it down to Kudu. |
| 254 | continue; |
| 255 | } else if (ctx.filter->AlwaysFalse()) { |
| 256 | // We can skip this entire scan if it's always false. |
| 257 | CloseCurrentClientScanner(); |
| 258 | *eos = true; |
| 259 | return Status::OK(); |
| 260 | } |
| 261 | |
| 262 | auto it = ctx.filter->filter_desc().planid_to_target_ndx.find(scan_node_->id()); |
| 263 | const TRuntimeFilterTargetDesc& target_desc = |
| 264 | ctx.filter->filter_desc().targets[it->second]; |
| 265 | const string& col_name = target_desc.kudu_col_name; |
| 266 | DCHECK(col_name != ""); |
| 267 |
no test coverage detected