Function _upgradeTable

be/src/catalog/workload-management-init.cc:302–361

Upgrades a table by running alter table statements. Columns with a schema version greater than the current table schema version and less than or equal to the target version will be added to the table. Equivalent to: ALTER TABLE {{table}} ADD IF NOT EXISTS COLUMNS (...) ALTER TABLE {{table}} SET TBLPROPERTIES ('wm_schema_version'='X.X.X')

/// Equivalent to: ALTER TABLE {{table}} ADD IF NOT EXISTS COLUMNS (...)
///                ALTER TABLE {{table}} SET TBLPROPERTIES ('wm_schema_version'='X.X.X')
static Status _upgradeTable(CatalogServiceIf* svc, const string& ip_addr,
    const string& table_name, const Version& current_ver, const Version& target_ver) {
  DCHECK_NE(current_ver, target_ver);

  // Add new columns.
  TDdlExecResponse resp;
  TDdlExecRequest req;
  TAlterTableParams t_alter_params;
  TAlterTableAddColsParams t_add_cols;
  TTableName t_table_name;

  t_table_name.__set_db_name(WM_DB);
  t_table_name.__set_table_name(table_name);

  t_add_cols.__set_if_not_exists(true);
  t_add_cols.__set_columns(
      _buildCols([&current_ver, &target_ver](const FieldDefinition& f) {
        return f.schema_version > current_ver && f.schema_version <= target_ver;
      }));

  t_alter_params.__set_add_cols_params(t_add_cols);
  t_alter_params.__set_alter_type(TAlterTableType::type::ADD_COLUMNS);
  t_alter_params.__set_table_name(t_table_name);

  TDdlQueryOptions query_opts;
  query_opts.__set_sync_ddl(true);

  req.__set_query_options(query_opts);
  req.__set_header(_getHeader(ip_addr));
  req.__set_ddl_type(TDdlType::type::ALTER_TABLE);
  req.__set_alter_table_params(t_alter_params);

  svc->ExecDdl(resp, req);
  if (resp.result.status.status_code != TErrorCode::type::OK) {
    return StatusFromThrift(resp.result.status);
  }

  // Update table schema version.
  TAlterTableSetTblPropertiesParams t_set_props_params;
  TAlterTableParams t_alter_schema_ver;

  t_set_props_params.__set_properties(
      {make_pair(WM_SCHEMA_VER_PROP_NAME, target_ver.ToString())});
  t_set_props_params.__set_target(TTablePropertyType::type::TBL_PROPERTY);

  t_alter_schema_ver.__set_alter_type(TAlterTableType::type::SET_TBL_PROPERTIES);
  t_alter_schema_ver.__set_set_tbl_properties_params(t_set_props_params);
  t_alter_schema_ver.__set_table_name(t_table_name);

  resp = TDdlExecResponse();
  req = TDdlExecRequest();

  req.__set_header(_getHeader(ip_addr));
  req.__set_ddl_type(TDdlType::type::ALTER_TABLE);
  req.__set_query_options(query_opts);
  req.__set_alter_table_params(t_alter_schema_ver);

  svc->ExecDdl(resp, req);