Function ReadDictionary

cpp/src/arrow/ipc/reader.cc:888–945

}

Status ReadDictionary(const Buffer& metadata, IpcReadContext context,
                      DictionaryKind* kind, io::RandomAccessFile* file) {
  const flatbuf::Message* message = nullptr;
  RETURN_NOT_OK(internal::VerifyMessage(metadata.data(), metadata.size(), &message));
  const auto dictionary_batch = message->header_as_DictionaryBatch();
  if (dictionary_batch == nullptr) {
    return Status::IOError(
        "Header-type of flatbuffer-encoded Message is not DictionaryBatch.");
  }

  // The dictionary is embedded in a record batch with a single column
  const auto batch_meta = dictionary_batch->data();

  CHECK_FLATBUFFERS_NOT_NULL(batch_meta, "DictionaryBatch.data");

  RETURN_NOT_OK(GetCompression(batch_meta, &context.compression));
  if (context.compression == Compression::UNCOMPRESSED &&
      message->version() == flatbuf::MetadataVersion::MetadataVersion_V4) {
    // Possibly obtain codec information from experimental serialization format
    // in 0.17.x
    RETURN_NOT_OK(GetCompressionExperimental(message, &context.compression));
  }

  const int64_t id = dictionary_batch->id();

  // Look up the dictionary value type, which must have been added to the
  // DictionaryMemo already prior to invoking this function
  ARROW_ASSIGN_OR_RAISE(auto value_type, context.dictionary_memo->GetDictionaryType(id));

  // Load the dictionary data from the dictionary batch
  ArrayLoader loader(batch_meta, internal::GetMetadataVersion(message->version()),
                     context.options, file);
  auto dict_data = std::make_shared<ArrayData>();
  const Field dummy_field("", value_type);
  RETURN_NOT_OK(loader.Load(&dummy_field, dict_data.get()));

  // Run post-load steps: buffer decompression, etc.
  RecordBatchLoader batch_loader{context, /*schema=*/nullptr, batch_meta->length(),
                                 /*inclusion_mask=*/std::vector<bool>{}};
  ARROW_ASSIGN_OR_RAISE(
      auto dict_columns,
      batch_loader.CreateColumns({dict_data}, /*resolve_dictionaries=*/false));
  DCHECK_EQ(dict_columns.size(), 1);
  dict_data = dict_columns[0];

  if (dictionary_batch->isDelta()) {
    if (kind != nullptr) {
      *kind = DictionaryKind::Delta;
    }
    return context.dictionary_memo->AddDictionaryDelta(id, dict_data);
  }
  ARROW_ASSIGN_OR_RAISE(bool inserted,
                        context.dictionary_memo->AddOrReplaceDictionary(id, dict_data));
  if (kind != nullptr) {
    *kind = inserted ? DictionaryKind::New : DictionaryKind::Replacement;
  }
  return Status::OK();
}