Retrieves the schema version of the specified table by reading its table properties.
| 362 | |
| 363 | /// Retrieves the schema version of the specified table by reading its table properties. |
| 364 | static Status _getTableSchemaVersion(Catalog* catalog, CatalogServiceIf* svc, |
| 365 | const string& ip_addr, const string& db, const string& tbl, Version* table_version) { |
| 366 | DCHECK_NE(nullptr, table_version); |
| 367 | |
| 368 | TGetTablesResult get_table_results; |
| 369 | RETURN_IF_ERROR(catalog->GetTableNames(db, &tbl, &get_table_results)); |
| 370 | if (get_table_results.tables.empty()) { |
| 371 | *table_version = NO_TABLE_EXISTS; |
| 372 | return Status::OK(); |
| 373 | } |
| 374 | |
| 375 | // Reset the table metadata to get the full metadata so its catalog object is fully |
| 376 | // populated enabling its properties to be read. |
| 377 | TResetMetadataRequest reset_req; |
| 378 | TResetMetadataResponse reset_resp; |
| 379 | TTableName table_name; |
| 380 | table_name.__set_db_name(db); |
| 381 | table_name.__set_table_name(tbl); |
| 382 | reset_req.__set_table_name(table_name); |
| 383 | reset_req.__set_is_refresh(true); |
| 384 | reset_req.__set_header(_getHeader(ip_addr)); |
| 385 | reset_req.__set_sync_ddl(true); |
| 386 | |
| 387 | svc->ResetMetadata(reset_resp, reset_req); |
| 388 | if (reset_resp.result.status.status_code != TErrorCode::type::OK) { |
| 389 | return StatusFromThrift(reset_resp.result.status); |
| 390 | } |
| 391 | |
| 392 | // Read the table's catalog object to retrieve its table properties. |
| 393 | TGetCatalogObjectResponse o_resp; |
| 394 | TGetCatalogObjectRequest o_req; |
| 395 | TCatalogObject catalog_obj; |
| 396 | TTable tbl_obj; |
| 397 | |
| 398 | tbl_obj.__set_db_name(db); |
| 399 | tbl_obj.__set_tbl_name(tbl); |
| 400 | |
| 401 | catalog_obj.__set_type(TCatalogObjectType::type::TABLE); |
| 402 | catalog_obj.__set_table(tbl_obj); |
| 403 | |
| 404 | o_req.__set_header(_getHeader(ip_addr)); |
| 405 | o_req.__set_object_desc(catalog_obj); |
| 406 | |
| 407 | svc->GetCatalogObject(o_resp, o_req); |
| 408 | if (o_resp.status.status_code != TErrorCode::type::OK) { |
| 409 | return StatusFromThrift(o_resp.status); |
| 410 | } |
| 411 | |
| 412 | // Determine the actual table schema version. The "wm_schema_version" table property |
| 413 | // takes precendence over the former "schema_version" table property. |
| 414 | optional<string> schema_ver_value; |
| 415 | optional<string> wm_schema_ver_value; |
| 416 | |
| 417 | // To maintain backwards compatibility, a new table schema version property had to be |
| 418 | // added, otherwise Impala daemons running older code will error on startup when |
| 419 | // workload management is enabled. |
| 420 | for (const auto& iter : o_resp.catalog_object.table.metastore_table.parameters) { |
| 421 | if (iter.first == WM_SCHEMA_VER_PROP_NAME_1_0_0) { |
no test coverage detected