| 250 | } |
| 251 | |
| 252 | Status DmlExecState::FinalizeHdfsInsert(const TFinalizeParams& params, |
| 253 | bool s3_skip_insert_staging, HdfsTableDescriptor* hdfs_table, |
| 254 | RuntimeProfile* profile) { |
| 255 | lock_guard<mutex> l(lock_); |
| 256 | PermissionCache permissions_cache; |
| 257 | HdfsFsCache::HdfsFsMap filesystem_connection_cache; |
| 258 | HdfsOperationSet partition_create_ops(&filesystem_connection_cache); |
| 259 | |
| 260 | // INSERT finalization happens in the five following steps |
| 261 | // 1. If OVERWRITE, remove all the files in the target directory |
| 262 | // 2. Create all the necessary partition directories. |
| 263 | |
| 264 | // Loop over all partitions that were updated by this insert, and create the set of |
| 265 | // filesystem operations required to create the correct partition structure on disk. |
| 266 | for (const PartitionStatusMap::value_type& partition : per_partition_status_) { |
| 267 | SCOPED_TIMER(ADD_CHILD_TIMER(profile, "Overwrite/PartitionCreationTimer", |
| 268 | "FinalizationTimer")); |
| 269 | // INSERT allows writes to tables that have partitions on multiple filesystems. |
| 270 | // So we need to open connections to different filesystems as necessary. We use a |
| 271 | // local connection cache and populate it with one connection per filesystem that the |
| 272 | // partitions are on. |
| 273 | hdfsFS partition_fs_connection; |
| 274 | RETURN_IF_ERROR(HdfsFsCache::instance()->GetConnection( |
| 275 | partition.second.partition_base_dir(), &partition_fs_connection, |
| 276 | &filesystem_connection_cache)); |
| 277 | |
| 278 | // Look up the partition in the descriptor table. |
| 279 | stringstream part_path_ss; |
| 280 | if (partition.second.id() == -1) { |
| 281 | // If this is a non-existant partition, use the default partition location of |
| 282 | // <base_dir>/part_key_1=val/part_key_2=val/... |
| 283 | part_path_ss << params.hdfs_base_dir << "/" << partition.first; |
| 284 | } else { |
| 285 | HdfsPartitionDescriptor* part = hdfs_table->GetPartition(partition.second.id()); |
| 286 | DCHECK(part != nullptr) |
| 287 | << "table_id=" << hdfs_table->id() << " partition_id=" << partition.second.id(); |
| 288 | part_path_ss << part->location(); |
| 289 | } |
| 290 | const string& part_path = part_path_ss.str(); |
| 291 | bool is_s3_path = IsS3APath(part_path.c_str()); |
| 292 | |
| 293 | // If this is an overwrite insert, we will need to delete any updated partitions |
| 294 | if (params.is_overwrite) { |
| 295 | if (partition.first.empty()) { |
| 296 | // If the root directory is written to, then the table must not be partitioned |
| 297 | DCHECK(per_partition_status_.size() == 1); |
| 298 | |
| 299 | RETURN_IF_ERROR(DeleteUnpartitionedDirData( |
| 300 | partition_fs_connection, part_path, &partition_create_ops)); |
| 301 | } else { |
| 302 | // This is a partition directory, not the root directory; we can delete |
| 303 | // recursively with abandon, after checking that it ever existed. |
| 304 | // TODO: There's a potential race here between checking for the directory |
| 305 | // and a third-party deleting it. |
| 306 | if (FLAGS_insert_inherit_permissions && !is_s3_path) { |
| 307 | // There is no directory structure in S3, so "inheriting" permissions is not |
| 308 | // possible. |
| 309 | // TODO: Try to mimic inheriting permissions for S3. |
no test coverage detected