Method GetSpaced

cpp/src/arrow/util/rle_encoding_internal.h:1054–1133

template <typename T>
template <typename Converter>
auto RleBitPackedDecoder<T>::GetSpaced(Converter converter,
                                       typename Converter::out_type* out,
                                       rle_size_t batch_size,
                                       const uint8_t* validity_bits,
                                       int64_t validity_bits_offset,
                                       rle_size_t null_count) -> rle_size_t {
  using ControlFlow = RleBitPackedParser::ControlFlow;

  ARROW_DCHECK_GT(batch_size, 0);

  auto batch = internal::BatchCounter::FromBatchSizeAndNulls(batch_size, null_count);

  if (ARROW_PREDICT_FALSE(batch.is_fully_null())) {
    converter.WriteZero(out, out + batch.null_remaining());
    return batch.null_remaining();
  }

  arrow::internal::BitRunReader validity_reader(validity_bits, validity_bits_offset,
                                                /*length=*/batch.total_remaining());
  arrow::internal::BitRun validity_run = validity_reader.NextRun();

  const auto check_and_handle_fully_null_remaining = [&]() {
    if (batch.is_fully_null()) {
      ARROW_DCHECK(validity_run.length == 0 || !validity_run.set);
      ARROW_DCHECK_GE(validity_run.length, batch.null_remaining());

      converter.WriteZero(out, out + batch.null_remaining());
      out += batch.null_remaining();
      batch.AccrueReadNulls(batch.null_remaining());
    }
  };

  // Remaining from a previous call that would have left some unread data from a run.
  if (ARROW_PREDICT_FALSE(run_remaining() > 0)) {
    const auto read = internal::RunGetSpaced(&converter, out, batch.total_remaining(),
                                             batch.null_remaining(), value_bit_width_,
                                             &validity_reader, &validity_run, &decoder_);

    batch.AccrueReadNulls(read.null_read);
    batch.AccrueReadValues(read.values_read);
    out += read.values_read + read.null_read;

    // Either we fulfilled all the batch values to be read
    if (ARROW_PREDICT_FALSE(batch.values_remaining() == 0)) {
      // There may be remaining null if they are not greedily filled
      check_and_handle_fully_null_remaining();
      return batch.total_read();
    }

    // We finished the remaining run
    ARROW_DCHECK(run_remaining() == 0);
  }

  ParseWithCallable([&](auto run) {
    using RunDecoder = typename decltype(run)::template DecoderType<value_type>;

    RunDecoder decoder(run, value_bit_width_);