| 1704 | } |
| 1705 | |
| 1706 | Status ClientRequestState::UpdateCatalog() { |
| 1707 | const TExecRequest& exec_req = exec_request(); |
| 1708 | if (!exec_req.__isset.query_exec_request || |
| 1709 | exec_req.query_exec_request.stmt_type != TStmtType::DML) { |
| 1710 | return Status::OK(); |
| 1711 | } |
| 1712 | |
| 1713 | query_events_->MarkEvent("DML data written"); |
| 1714 | SCOPED_TIMER(ADD_TIMER(server_profile_, "MetastoreUpdateTimer")); |
| 1715 | |
| 1716 | const TQueryExecRequest& query_exec_request = exec_req.query_exec_request; |
| 1717 | if (query_exec_request.__isset.finalize_params) { |
| 1718 | const TFinalizeParams& finalize_params = query_exec_request.finalize_params; |
| 1719 | TUpdateCatalogRequest catalog_update; |
| 1720 | catalog_update.__set_sync_ddl(exec_req.query_options.sync_ddl); |
| 1721 | catalog_update.__set_header(GetCatalogServiceRequestHeader()); |
| 1722 | if (exec_req.query_options.__isset.debug_action) { |
| 1723 | catalog_update.__set_debug_action(exec_req.query_options.debug_action); |
| 1724 | } |
| 1725 | DmlExecState* dml_exec_state = GetCoordinator()->dml_exec_state(); |
| 1726 | if (!dml_exec_state->PrepareCatalogUpdate(&catalog_update, finalize_params)) { |
| 1727 | VLOG_QUERY << "No partitions altered, not updating metastore (query id: " |
| 1728 | << PrintId(query_id()) << ")"; |
| 1729 | } else { |
| 1730 | // TODO: We track partitions written to, not created, which means |
| 1731 | // that we do more work than is necessary, because written-to |
| 1732 | // partitions don't always require a metastore change. |
| 1733 | if (VLOG_IS_ON(1)) { |
| 1734 | vector<string> part_list; |
| 1735 | for (auto it : catalog_update.updated_partitions) part_list.push_back(it.first); |
| 1736 | VLOG_QUERY << "Updating metastore with " |
| 1737 | << catalog_update.updated_partitions.size() |
| 1738 | << " altered partitions (" |
| 1739 | << join (part_list, ", ") << ")"; |
| 1740 | } |
| 1741 | |
| 1742 | catalog_update.target_table = finalize_params.table_name; |
| 1743 | catalog_update.db_name = finalize_params.table_db; |
| 1744 | catalog_update.is_overwrite = finalize_params.is_overwrite; |
| 1745 | if (InTransaction()) { |
| 1746 | catalog_update.__set_transaction_id(finalize_params.transaction_id); |
| 1747 | catalog_update.__set_write_id(finalize_params.write_id); |
| 1748 | } |
| 1749 | if (finalize_params.__isset.iceberg_params) { |
| 1750 | TIcebergOperationParam& cat_ice_op = catalog_update.iceberg_operation; |
| 1751 | catalog_update.__isset.iceberg_operation = true; |
| 1752 | if (!CreateIcebergCatalogOps(finalize_params, &cat_ice_op)) { |
| 1753 | VLOG_QUERY << "No Iceberg partitions altered, not updating metastore " |
| 1754 | << "(query id: " << PrintId(query_id()) << ")"; |
| 1755 | return Status::OK(); |
| 1756 | } |
| 1757 | } |
| 1758 | |
| 1759 | Status cnxn_status; |
| 1760 | CatalogServiceConnection client(ExecEnv::GetInstance()->catalogd_client_cache(), |
| 1761 | *ExecEnv::GetInstance()->GetCatalogdAddress().get(), &cnxn_status); |
| 1762 | RETURN_IF_ERROR(cnxn_status); |
| 1763 |
nothing calls this directly
no test coverage detected