Method AdjustFieldIdForMigratedPartitionedTables

be/src/exec/file-metadata-utils.cc:306–355

}

Status FileMetadataUtils::AdjustFieldIdForMigratedPartitionedTables(int *fieldID) const {
  DCHECK(file_desc_ != nullptr);
  DCHECK(scan_node_->hdfs_table()->IsIcebergTable());
  DCHECK(fieldID != nullptr);

  using namespace org::apache::impala::fb;

  auto transforms = IcebergPartitionTransforms();
  if (transforms == nullptr || transforms->size() == 0) return Status::OK();

  vector<int> sourceIDs;
  sourceIDs.reserve(transforms->size());
  for (auto transform : *transforms) {
    if (transform->transform_type() !=
        FbIcebergTransformType::FbIcebergTransformType_IDENTITY) {
      return Status(Substitute("$0 has invalid partition transform: $1",
          file_desc_->filename, transform->transform_type()));
    }
    sourceIDs.push_back(transform->source_id());
  }

  DCHECK_EQ(sourceIDs.size(), transforms->size());
  sort(sourceIDs.begin(), sourceIDs.end());

  string errorMsg = Substitute(
      "Migrated file $0 has unexpected schema or partitioning.", file_desc_->filename);

  // Field IDs must be consecutive.
  for (int i = 0; i < sourceIDs.size() - 1; ++i) {
    if (sourceIDs[i+1] - sourceIDs[i] != 1) {
      return Status(errorMsg);
    }
  }

  int fileFieldID = *fieldID;

  if (sourceIDs.front() == fileFieldID) {
    // Partition columns are not stored in the data file (as expected).
    // In this case we need to adjust fieldID (as it was calculated based on
    // the file schema only).
    *fieldID += sourceIDs.size();
  } else if (sourceIDs.back() == fileFieldID - 1) {
    // Partition columns are stored in the data file, no need to adjust fieldID.
  } else {
    // Some partitions are stored, some aren't let's raise an error for this mess.
    return Status(errorMsg);
  }

  return Status::OK();
}

bool FileMetadataUtils::NeedDataInFile(const SlotDescriptor* slot_desc) {
  if (IsValuePartitionCol(slot_desc)) return false;