Method InitNewRange

be/src/exec/text/hdfs-text-scanner.cc:220–262

}

Status HdfsTextScanner::InitNewRange() {
  DCHECK_EQ(scan_state_, CONSTRUCTED);

  auto compression_type = stream_ ->file_desc()->file_compression;
  // Update the decompressor based on the compression type of the file in the context.
  DCHECK(compression_type != THdfsCompression::SNAPPY)
      << "FE should have generated SNAPPY_BLOCKED instead.";
  // In Hadoop, text files compressed into .DEFLATE files contain
  // deflate with zlib wrappings as opposed to raw deflate, which
  // is what THdfsCompression::DEFLATE implies. Since deflate is
  // the default compression algorithm used in Hadoop, it makes
  // sense to map it to type DEFAULT in Impala instead
  if (compression_type == THdfsCompression::DEFLATE) {
    compression_type = THdfsCompression::DEFAULT;
  }
  RETURN_IF_ERROR(UpdateDecompressor(compression_type));

  HdfsPartitionDescriptor* hdfs_partition = context_->partition_descriptor();
  char field_delim = hdfs_partition->field_delim();
  char collection_delim = hdfs_partition->collection_delim();
  if (scan_node_->materialized_slots().size() == 0) {
    field_delim = '\0';
    collection_delim = '\0';
  }

  delimited_text_parser_.reset(new TupleDelimitedTextParser(
      scan_node_->hdfs_table()->num_cols(), scan_node_->num_partition_keys(),
      scan_node_->is_materialized_col(), hdfs_partition->line_delim(),
      field_delim, collection_delim, hdfs_partition->escape_char()));
  text_converter_.reset(new TextConverter(hdfs_partition->escape_char(),
      scan_node_->hdfs_table()->null_column_value(), true,
      state_->strict_mode()));

  const auto& encoding = hdfs_partition->encoding_value();
  if (!encoding.empty() && encoding != "UTF-8") {
    decoder_.reset(new CharCodec(data_buffer_pool_.get(), encoding,
        hdfs_partition->line_delim(), scan_node_->tuple_desc()->string_slots().empty()));
  }

  RETURN_IF_ERROR(ResetScanner());
  scan_state_ = SCAN_RANGE_INITIALIZED;
  return Status::OK();
}

Status HdfsTextScanner::ResetScanner() {
  // Assumes that N partition keys occupy entries 0 through N-1 in materialized_slots_.