Sets up the query table by generating and executing the necessary DML statements. System tables are external. The columns on the table will be all columns with a schema version less than or equal to the specified target_version. Non-system tables are partitioned on cluster_id and hour(start_time_utc). Equivalent to: CREATE [EXTERNAL] TABLE IF NOT EXISTS {{table}} (...) PARTITIONED BY SPEC (cluster
| 194 | /// PARTITIONED BY SPEC (cluster_id, HOUR(start_time_utc)) |
| 195 | /// STORED AS ICEBERG |
| 196 | static Status _createTable(CatalogServiceIf* svc, const string& ip_addr, |
| 197 | const string& table_name, const Version& target_version, bool is_system_table) { |
| 198 | LOG(INFO) << "Creating workload management table '" << table_name |
| 199 | << "' on schema version '" << target_version.ToString() << "'"; |
| 200 | |
| 201 | TDdlExecResponse resp; |
| 202 | TDdlExecRequest req; |
| 203 | TCreateTableParams t_create_tbl; |
| 204 | |
| 205 | // General table properties. |
| 206 | t_create_tbl.__set_if_not_exists(true); |
| 207 | t_create_tbl.__set_owner(FLAGS_workload_mgmt_maintenance_user); |
| 208 | map<string, string> table_props = { |
| 209 | make_pair(WM_SCHEMA_VER_PROP_NAME_1_0_0, VERSION_1_0_0.ToString()), |
| 210 | make_pair(WM_SCHEMA_VER_PROP_NAME, target_version.ToString()), |
| 211 | make_pair("format-version", "2"), make_pair("OBJCAPABILITIES", "EXTREAD,EXTWRITE")}; |
| 212 | |
| 213 | // User provided properties. |
| 214 | if (!FLAGS_query_log_table_props.empty()) { |
| 215 | for (const auto& prop : strings::Split(FLAGS_query_log_table_props, ",")) { |
| 216 | vector<string> prop_parts = strings::Split(prop, "="); |
| 217 | if (prop_parts.size() != 2) { |
| 218 | return Status(StrCat("property '", prop, |
| 219 | "' is not in the expected 'key=value' format")); |
| 220 | } |
| 221 | table_props.insert(make_pair(trim_copy(prop_parts[0]), trim_copy(prop_parts[1]))); |
| 222 | } |
| 223 | } |
| 224 | |
| 225 | // Table name and database. |
| 226 | TTableName t_table_name; |
| 227 | t_table_name.__set_db_name(WM_DB); |
| 228 | t_table_name.__set_table_name(table_name); |
| 229 | t_create_tbl.__set_table_name(t_table_name); |
| 230 | |
| 231 | // Table columns. |
| 232 | t_create_tbl.__set_columns(_buildCols([&target_version](const FieldDefinition& f) { |
| 233 | return f.schema_version <= target_version; |
| 234 | })); |
| 235 | |
| 236 | if (is_system_table) { |
| 237 | // Table properties unique to system tables. |
| 238 | t_create_tbl.__set_is_external(true); |
| 239 | table_props.insert( |
| 240 | make_pair(g_CatalogObjects_constants.TBL_PROP_SYSTEM_TABLE, "true")); |
| 241 | } else { |
| 242 | // Table properties unique to non-system tables. |
| 243 | t_create_tbl.__set_is_external(false); |
| 244 | t_create_tbl.__set_file_format(THdfsFileFormat::type::ICEBERG); |
| 245 | table_props.insert(make_pair("engine.hive.enabled", "true")); |
| 246 | table_props.insert(make_pair("write.delete.mode", "merge-on-read")); |
| 247 | table_props.insert(make_pair("write.format.default", "parquet")); |
| 248 | table_props.insert(make_pair("write.merge.mode", "merge-on-read")); |
| 249 | table_props.insert(make_pair("write.parquet.compression-codec", "snappy")); |
| 250 | table_props.insert(make_pair("write.update.mode", "merge-on-read")); |
| 251 | |
| 252 | // Table partitioning. |
| 253 | TIcebergPartitionSpec t_partion_spec; |
no test coverage detected