| 827 | } |
| 828 | |
| 829 | void ClientRequestState::ExecDdlRequestImpl(bool exec_in_worker_thread) { |
| 830 | bool is_CTAS = (catalog_op_type() == TCatalogOpType::DDL |
| 831 | && ddl_type() == TDdlType::CREATE_TABLE_AS_SELECT); |
| 832 | const TExecRequest& exec_req = exec_request(); |
| 833 | |
| 834 | catalog_op_executor_.reset( |
| 835 | new CatalogOpExecutor(ExecEnv::GetInstance(), frontend_, server_profile_)); |
| 836 | |
| 837 | // Indirectly check if running in thread async_exec_thread_. |
| 838 | if (exec_in_worker_thread) { |
| 839 | VLOG_QUERY << "Running in worker thread"; |
| 840 | DCHECK(exec_state() == ExecState::PENDING); |
| 841 | |
| 842 | // 1. For any non-CTAS DDLs, transition to RUNNING |
| 843 | // 2. For CTAS DDLs, transition to RUNNING during FinishExecQueryOrDmlRequest() |
| 844 | // called by ExecQueryOrDmlRequest(). |
| 845 | if (!is_CTAS) UpdateNonErrorExecState(ExecState::RUNNING); |
| 846 | } |
| 847 | |
| 848 | // Optionally wait with a debug action before Exec() below. |
| 849 | DebugActionNoFail(exec_req.query_options, "CRS_DELAY_BEFORE_CATALOG_OP_EXEC"); |
| 850 | |
| 851 | Status status = catalog_op_executor_->Exec(exec_req.catalog_op_request); |
| 852 | query_events_->MarkEvent("CatalogDdlRequest finished"); |
| 853 | if (otel_trace_query()) { |
| 854 | otel_trace_manager_->AddChildSpanEvent("UpdateCatalogFinished"); |
| 855 | } |
| 856 | AddCatalogTimeline(); |
| 857 | { |
| 858 | lock_guard<mutex> l(lock_); |
| 859 | RETURN_VOID_IF_ERROR(UpdateQueryStatus(status, exec_in_worker_thread)); |
| 860 | } |
| 861 | |
| 862 | // If this is a CTAS request, there will usually be more work to do |
| 863 | // after executing the CREATE TABLE statement (the INSERT portion of the operation). |
| 864 | // The exception is if the user specified IF NOT EXISTS and the table already |
| 865 | // existed, in which case we do not execute the INSERT. |
| 866 | if (catalog_op_type() == TCatalogOpType::DDL && |
| 867 | ddl_type() == TDdlType::CREATE_TABLE_AS_SELECT && |
| 868 | !catalog_op_executor_->ddl_exec_response()->new_table_created) { |
| 869 | DCHECK(exec_req.catalog_op_request. |
| 870 | ddl_params.create_table_params.if_not_exists); |
| 871 | return; |
| 872 | } |
| 873 | |
| 874 | // Add newly created table to catalog cache. |
| 875 | status = parent_server_->ProcessCatalogUpdateResult( |
| 876 | *catalog_op_executor_->update_catalog_result(), |
| 877 | exec_req.query_options.sync_ddl, query_options(), query_events_); |
| 878 | { |
| 879 | lock_guard<mutex> l(lock_); |
| 880 | RETURN_VOID_IF_ERROR(UpdateQueryStatus(status, exec_in_worker_thread)); |
| 881 | } |
| 882 | |
| 883 | if (is_CTAS) { |
| 884 | // At this point, the remainder of the CTAS request executes |
| 885 | // like a normal DML request. As with other DML requests, it will |
| 886 | // wait for another catalog update if any partitions were altered as a result |
nothing calls this directly
no test coverage detected