| 301 | } |
| 302 | |
| 303 | Status IcebergBufferedDeleteSink::SetCurrentPartition(RuntimeState* state, |
| 304 | int32_t spec_id, const std::string& partition_encoded) { |
| 305 | current_partition_.reset(new OutputPartition()); |
| 306 | // Build the unique name for this partition from the partition keys, e.g. "j=1/f=foo/" |
| 307 | // etc. |
| 308 | RETURN_IF_ERROR(ConstructPartitionInfo( |
| 309 | spec_id, partition_encoded, current_partition_.get())); |
| 310 | Status status = InitOutputPartition(state, *prototype_partition_, |
| 311 | current_partition_.get(), false); |
| 312 | if (!status.ok()) { |
| 313 | // We failed to create the output partition successfully. Clean it up now. |
| 314 | if (current_partition_->writer != nullptr) { |
| 315 | current_partition_->writer->Close(); |
| 316 | } |
| 317 | return status; |
| 318 | } |
| 319 | |
| 320 | // With partition evolution it's possible that we have the same partition names |
| 321 | // with different spec ids. E.g. in case of TRUNCATE(1000, col) => TRUNCATE(500, col), |
| 322 | // we might need to delete rows from partition "col_trunc=1000" with both spec ids. In |
| 323 | // this case we might already have "col_trunc=1000" in dml_exec_state, so no need to |
| 324 | // add it. |
| 325 | if (!dml_exec_state_.PartitionExists(current_partition_->partition_name)) { |
| 326 | // Save the partition name so that the coordinator can create the partition |
| 327 | // directory structure if needed. |
| 328 | dml_exec_state_.AddPartition( |
| 329 | current_partition_->partition_name, prototype_partition_->id(), |
| 330 | &table_desc_->hdfs_base_dir(), |
| 331 | nullptr); |
| 332 | } |
| 333 | return Status::OK(); |
| 334 | } |
| 335 | |
| 336 | Status IcebergBufferedDeleteSink::TryAllocateUnalignedBuffer(int buffer_size, |
| 337 | uint8_t** buffer) { |
nothing calls this directly
no test coverage detected