Method CalculateDataFiles

be/src/exec/iceberg-delete-builder.cc:117–208

IcebergDeleteBuilder::~IcebergDeleteBuilder() {}

Status IcebergDeleteBuilder::CalculateDataFiles() {
  auto& fragment_state_map = runtime_state_->query_state()->FragmentStateMap();
  auto fragment_it = fragment_state_map.end();
  PlanNode* delete_scan_node = nullptr;
  bool found = false;
  std::queue<const PlanNode*> q;
  for (auto it = fragment_state_map.begin(); !found && it != fragment_state_map.end();
       it++) {
    q.push(it->second->plan_tree());
    while (!q.empty()) {
      auto* current = q.front();
      q.pop();
      if (current->tnode_->node_id == join_node_id_) {
        fragment_it = it;
        // Tuple caching can place a TupleCacheNode above the scan node. Look past
        // a TupleCacheNode to get to the scan node.
        delete_scan_node = PlanNode::LookPastTupleCache(current->children_[0]);
        DCHECK_EQ(delete_scan_node->tnode_->node_type, TPlanNodeType::HDFS_SCAN_NODE)
          << "Failed to calculate delete files: "
          << Substitute("Unexpected type for plan node $0: $1",
                 delete_scan_node->tnode_->node_id, delete_scan_node->tnode_->node_type);
        found = true;
        while (!q.empty()) q.pop();
        break;
      }
      for (auto* child : current->children_) {
        q.push(child);
      }
    }
  }

  const vector<const PlanFragmentInstanceCtxPB*>& instance_ctx_pbs =
      fragment_it->second->instance_ctx_pbs();
  for (auto ctx : instance_ctx_pbs) {
    auto ranges = ctx->per_node_scan_ranges().find(delete_scan_node->tnode_->node_id);
    if (ranges == ctx->per_node_scan_ranges().end()) continue;

    auto tuple_id = delete_scan_node->tnode_->hdfs_scan_node.tuple_id;
    auto tuple_desc = runtime_state_->desc_tbl().GetTupleDescriptor(tuple_id);
    DCHECK(tuple_desc->table_desc() != nullptr);
    auto hdfs_table = static_cast<const HdfsTableDescriptor*>(tuple_desc->table_desc());
    DCHECK(hdfs_table->IsIcebergTable());

    for (const ScanRangeParamsPB& params : ranges->second.scan_ranges()) {
      DCHECK(params.scan_range().has_hdfs_file_split());
      const HdfsFileSplitPB& split = params.scan_range().hdfs_file_split();

      HdfsPartitionDescriptor* partition_desc =
          hdfs_table->GetPartition(split.partition_id());

      std::filesystem::path file_path;
      if (split.relative_path().empty()) {
        file_path.append(split.absolute_path());
      } else {
        file_path.append(partition_desc->location()).append(split.relative_path());
      }
      auto& file_path_str = file_path.native();
      char* ptr_copy =