| 2391 | } |
| 2392 | |
| 2393 | Status ClientRequestState::LogLineageRecord() { |
| 2394 | const TExecRequest& request = exec_request(); |
| 2395 | if (request.stmt_type == TStmtType::EXPLAIN || (!request.__isset.query_exec_request && |
| 2396 | !request.__isset.catalog_op_request)) { |
| 2397 | return Status::OK(); |
| 2398 | } |
| 2399 | TLineageGraph lineage_graph; |
| 2400 | if (request.__isset.query_exec_request && |
| 2401 | request.query_exec_request.__isset.lineage_graph) { |
| 2402 | lineage_graph = request.query_exec_request.lineage_graph; |
| 2403 | } else if (request.__isset.catalog_op_request && |
| 2404 | request.catalog_op_request.__isset.lineage_graph) { |
| 2405 | lineage_graph = request.catalog_op_request.lineage_graph; |
| 2406 | } else { |
| 2407 | return Status::OK(); |
| 2408 | } |
| 2409 | |
| 2410 | if (catalog_op_executor_ != nullptr && catalog_op_type() == TCatalogOpType::DDL) { |
| 2411 | const TDdlExecResponse* response = ddl_exec_response(); |
| 2412 | //Set table location in the lineage graph. Currently, this is only set for external |
| 2413 | // tables in frontend. |
| 2414 | if (response->__isset.table_location) { |
| 2415 | lineage_graph.__set_table_location(response->table_location); |
| 2416 | } |
| 2417 | // Update vertices that have -1 table_create_time for a newly created table/view. |
| 2418 | if (response->__isset.table_name && |
| 2419 | response->__isset.table_create_time) { |
| 2420 | for (auto &vertex: lineage_graph.vertices) { |
| 2421 | if (!vertex.__isset.metadata) continue; |
| 2422 | if (vertex.metadata.table_name == response->table_name && |
| 2423 | vertex.metadata.table_create_time == -1) { |
| 2424 | vertex.metadata.__set_table_create_time(response->table_create_time); |
| 2425 | } |
| 2426 | } |
| 2427 | } |
| 2428 | } |
| 2429 | |
| 2430 | // Set the query end time in TLineageGraph. Must use UNIX time directly rather than |
| 2431 | // e.g. converting from end_time() (IMPALA-4440). |
| 2432 | lineage_graph.__set_ended(UnixMillis() / 1000); |
| 2433 | |
| 2434 | string lineage_record; |
| 2435 | LineageUtil::TLineageToJSON(lineage_graph, &lineage_record); |
| 2436 | |
| 2437 | if (parent_server_->AreQueryHooksEnabled()) { |
| 2438 | // invoke QueryEventHooks |
| 2439 | TQueryCompleteContext query_complete_context; |
| 2440 | query_complete_context.__set_lineage_string(lineage_record); |
| 2441 | const Status& status = ExecEnv::GetInstance()->frontend()->CallQueryCompleteHooks( |
| 2442 | query_complete_context); |
| 2443 | |
| 2444 | if (!status.ok()) { |
| 2445 | LOG(ERROR) << "Failed to send query lineage info to FE CallQueryCompleteHooks" |
| 2446 | << status.GetDetail(); |
| 2447 | if (FLAGS_abort_on_failed_lineage_event) { |
| 2448 | CLEAN_EXIT_WITH_ERROR("Shutting down Impala Server due to " |
| 2449 | "abort_on_failed_lineage_event=true"); |
| 2450 | } |
nothing calls this directly
no test coverage detected