Method ComputeFragmentInstanceKey

be/src/exec/tuple-cache-node.cc:498–543

}

void TupleCacheNode::ComputeFragmentInstanceKey(const RuntimeState* state) {
  const PlanFragmentInstanceCtxPB& ctx = state->instance_ctx_pb();
  uint32_t hash = 0;
  // Collect the HdfsScanNodes below this point. The HdfsScanNodes have information about
  // the partitions that we need to include in the fragment instance key. Some locations
  // may have a large number of scan nodes below them, so construct a map from the node
  // id to the HdfsScanNodeBase.
  vector<ExecNode*> scan_nodes;
  CollectNodes(TPlanNodeType::HDFS_SCAN_NODE, &scan_nodes);
  unordered_map<int, const HdfsScanNodeBase*> id_to_scan_node_map;
  for (const ExecNode* exec_node : scan_nodes) {
    const HdfsScanNodeBase* scan_node =
        static_cast<const HdfsScanNodeBase*>(exec_node);
    int node_id = exec_node->plan_node().tnode_->node_id;
    DCHECK(id_to_scan_node_map.find(node_id) == id_to_scan_node_map.end())
        << "Duplicate scan node id: " << node_id;
    id_to_scan_node_map[node_id] = scan_node;
  }
  for (int32_t node_id : plan_node().tnode_->tuple_cache_node.input_scan_node_ids) {
    const HdfsTableDescriptor* hdfs_table = id_to_scan_node_map[node_id]->hdfs_table();
    DCHECK(hdfs_table != nullptr);
    auto ranges = ctx.per_node_scan_ranges().find(node_id);
    if (ranges == ctx.per_node_scan_ranges().end()) continue;
    for (const ScanRangeParamsPB& params : ranges->second.scan_ranges()) {
      // This only supports HDFS right now
      DCHECK(params.scan_range().has_hdfs_file_split());
      const HdfsFileSplitPB& split = params.scan_range().hdfs_file_split();
      // Information on the partition can influence how files are processed. For example,
      // for text files, the delimiter can be specified on the partition level. There
      // are several such attributes. We need to incorporate the partition information
      // into the hash for each file split.
      const HdfsPartitionDescriptor* partition_desc =
          hdfs_table->GetPartition(split.partition_id());
      DCHECK(partition_desc != nullptr);
      if (partition_desc != nullptr) {
        hash = HashHdfsPartitionDescriptor(partition_desc, hash);
      } else {
        LOG(WARNING) << "Partition id " << split.partition_id()
                     << " not found in table " << hdfs_table->fully_qualified_name()
                     << " split filename: " << split.relative_path();
      }
      hash = HashHdfsFileSplit(split, hash);
    }
  }
  fragment_instance_key_ = hash;
}

}