adamreeve commented on code in PR #50823:
URL: https://github.com/apache/arrow/pull/50823#discussion_r4407675469
##########
cpp/src/parquet/decoder.cc:
##########
@@ -2085,93 +2102,201 @@ class DeltaByteArrayDecoderImpl : public
TypedDecoderImpl<DType> {
}
protected:
- template <bool is_first_run>
- static void BuildBufferInternal(const int32_t* prefix_len_ptr, int i,
ByteArray* buffer,
- std::string_view* prefix, uint8_t**
data_ptr) {
- if (ARROW_PREDICT_FALSE(static_cast<size_t>(prefix_len_ptr[i]) >
prefix->length())) {
- throw ParquetException("prefix length too large in DELTA_BYTE_ARRAY");
- }
- // For now, `buffer` points to string suffixes, and the suffix decoder
- // ensures that the suffix data has sufficient lifetime.
- if (prefix_len_ptr[i] == 0) {
- // prefix is empty: buffer[i] already points to the suffix.
- *prefix = std::string_view{buffer[i]};
- return;
+ // Output that produces one `T` (`ByteArray` or `FixedLenByteArray`) per
value.
+ // A value is only copied (into `buffered_data_`) when both its prefix and
its
+ // suffix are non-empty; otherwise it references the suffix in the page or
the
+ // previous value.
+ struct ValueOutput {
+ T* buffer;
+ ResizableBuffer* buffered_data;
+ uint8_t* data_ptr = nullptr;
+ const uint8_t* data_end = nullptr;
+
+ void Prepare(const int32_t* prefix_len_ptr, const int32_t* suffix_lengths,
+ int max_values) {
+ int64_t data_size = 0;
+ for (int i = 0; i < max_values; ++i) {
+ if (prefix_len_ptr[i] == 0) {
+ // We don't need to copy the suffix if the prefix length is 0.
+ continue;
+ }
+ if (ARROW_PREDICT_FALSE(prefix_len_ptr[i] < 0)) {
+ throw ParquetException("negative prefix length in DELTA_BYTE_ARRAY");
+ }
+ if (suffix_lengths[i] == 0 && i != 0) {
+ // We don't need to copy the prefix if the suffix length is 0
+ // and this is not the first run (that is, the prefix doesn't point
+ // to the mutable `last_value_`).
+ continue;
+ }
+ if (ARROW_PREDICT_FALSE(
+ AddWithOverflow(data_size, prefix_len_ptr[i], &data_size) ||
+ AddWithOverflow(data_size, suffix_lengths[i], &data_size))) {
+ throw ParquetException("excess expansion in DELTA_BYTE_ARRAY");
+ }
+ }
+ PARQUET_THROW_NOT_OK(buffered_data->Resize(data_size));
+ data_ptr = buffered_data->mutable_data();
+ data_end = data_ptr + data_size;
+ }
+
+ std::string_view Emit(int i, std::string_view prefix, std::string_view
suffix) {
+ std::string_view value;
+ if (prefix.empty()) {
+ value = suffix;
+ } else if (suffix.empty() && i != 0) {
+ // This is not possible for the first run since the prefix would point
to
+ // the mutable `last_value_`.
+ value = prefix;
+ } else {
+ DCHECK_LE(data_ptr + prefix.size() + suffix.size(), data_end);
+ memcpy(data_ptr, prefix.data(), prefix.size());
+ memcpy(data_ptr + prefix.size(), suffix.data(), suffix.size());
+ value = {reinterpret_cast<const char*>(data_ptr), prefix.size() +
suffix.size()};
+ data_ptr += value.size();
+ }
+ buffer[i].ptr = reinterpret_cast<const uint8_t*>(value.data());
+ if constexpr (std::is_same_v<T, ByteArray>) {
+ buffer[i].len = static_cast<uint32_t>(value.size());
+ }
+ return value;
}
- DCHECK_EQ(is_first_run, i == 0);
- if constexpr (!is_first_run) {
- if (buffer[i].len == 0) {
- // suffix is empty: buffer[i] can simply point to the prefix.
- // This is not possible for the first run since the prefix
- // would point to the mutable `last_value_`.
- *prefix = prefix->substr(0, prefix_len_ptr[i]);
- buffer[i] = ByteArray(*prefix);
- return;
+ };
+
+ // Output that writes fixed-length values contiguously into a caller-provided
+ // buffer. Used by the FLBA dense decode path; avoids `buffered_data_` and
any
+ // temporary per-value storage.
+ struct DenseOutput {
+ uint8_t* out;
+ int type_length;
+
+ void Prepare(const int32_t* prefix_len_ptr, const int32_t* suffix_lengths,
+ int max_values) {}
+
+ std::string_view Emit(int i, std::string_view prefix, std::string_view
suffix) {
+ uint8_t* dst = out + static_cast<int64_t>(i) * type_length;
+ if (!prefix.empty()) {
+ memcpy(dst, prefix.data(), prefix.size());
+ }
+ if (!suffix.empty()) {
+ memcpy(dst + prefix.size(), suffix.data(), suffix.size());
}
+ // The next value's prefix references this value, which now lives in the
+ // caller's buffer.
+ return std::string_view{reinterpret_cast<const char*>(dst),
+ static_cast<size_t>(type_length)};
+ }
+ };
+
+ // Output that appends each value to an Arrow binary builder through
`helper`.
+ // A value is assembled in `scratch` only when both its prefix and its
suffix are
+ // non-empty; otherwise it is appended straight from the page or the
previous value.
+ template <typename Helper>
+ struct ArrowOutput {
+ Helper* helper;
+ std::string* scratch;
+ // Number of slots (values and nulls) the builder receives.
+ int num_slots;
+ // Builder data still to be appended, reserved by `Prepare`.
+ int64_t remaining_data_length = 0;
+ // Set when there are nulls: the runs of valid slots. The nulls before a
run are
+ // appended when its first value is; `position` is the end of the current
run.
Review Comment:
It might be worth commenting that the trailing nulls are not handled here
and must be added by the caller.
##########
cpp/src/parquet/decoder.cc:
##########
@@ -2085,93 +2102,201 @@ class DeltaByteArrayDecoderImpl : public
TypedDecoderImpl<DType> {
}
protected:
- template <bool is_first_run>
- static void BuildBufferInternal(const int32_t* prefix_len_ptr, int i,
ByteArray* buffer,
- std::string_view* prefix, uint8_t**
data_ptr) {
- if (ARROW_PREDICT_FALSE(static_cast<size_t>(prefix_len_ptr[i]) >
prefix->length())) {
- throw ParquetException("prefix length too large in DELTA_BYTE_ARRAY");
- }
- // For now, `buffer` points to string suffixes, and the suffix decoder
- // ensures that the suffix data has sufficient lifetime.
- if (prefix_len_ptr[i] == 0) {
- // prefix is empty: buffer[i] already points to the suffix.
- *prefix = std::string_view{buffer[i]};
- return;
+ // Output that produces one `T` (`ByteArray` or `FixedLenByteArray`) per
value.
+ // A value is only copied (into `buffered_data_`) when both its prefix and
its
+ // suffix are non-empty; otherwise it references the suffix in the page or
the
+ // previous value.
+ struct ValueOutput {
+ T* buffer;
+ ResizableBuffer* buffered_data;
+ uint8_t* data_ptr = nullptr;
+ const uint8_t* data_end = nullptr;
+
+ void Prepare(const int32_t* prefix_len_ptr, const int32_t* suffix_lengths,
+ int max_values) {
+ int64_t data_size = 0;
+ for (int i = 0; i < max_values; ++i) {
+ if (prefix_len_ptr[i] == 0) {
+ // We don't need to copy the suffix if the prefix length is 0.
+ continue;
+ }
+ if (ARROW_PREDICT_FALSE(prefix_len_ptr[i] < 0)) {
+ throw ParquetException("negative prefix length in DELTA_BYTE_ARRAY");
+ }
+ if (suffix_lengths[i] == 0 && i != 0) {
+ // We don't need to copy the prefix if the suffix length is 0
+ // and this is not the first run (that is, the prefix doesn't point
+ // to the mutable `last_value_`).
+ continue;
+ }
+ if (ARROW_PREDICT_FALSE(
+ AddWithOverflow(data_size, prefix_len_ptr[i], &data_size) ||
+ AddWithOverflow(data_size, suffix_lengths[i], &data_size))) {
+ throw ParquetException("excess expansion in DELTA_BYTE_ARRAY");
+ }
+ }
+ PARQUET_THROW_NOT_OK(buffered_data->Resize(data_size));
+ data_ptr = buffered_data->mutable_data();
+ data_end = data_ptr + data_size;
+ }
+
+ std::string_view Emit(int i, std::string_view prefix, std::string_view
suffix) {
+ std::string_view value;
+ if (prefix.empty()) {
+ value = suffix;
+ } else if (suffix.empty() && i != 0) {
+ // This is not possible for the first run since the prefix would point
to
+ // the mutable `last_value_`.
+ value = prefix;
+ } else {
+ DCHECK_LE(data_ptr + prefix.size() + suffix.size(), data_end);
+ memcpy(data_ptr, prefix.data(), prefix.size());
+ memcpy(data_ptr + prefix.size(), suffix.data(), suffix.size());
+ value = {reinterpret_cast<const char*>(data_ptr), prefix.size() +
suffix.size()};
+ data_ptr += value.size();
+ }
+ buffer[i].ptr = reinterpret_cast<const uint8_t*>(value.data());
+ if constexpr (std::is_same_v<T, ByteArray>) {
+ buffer[i].len = static_cast<uint32_t>(value.size());
+ }
+ return value;
}
- DCHECK_EQ(is_first_run, i == 0);
- if constexpr (!is_first_run) {
- if (buffer[i].len == 0) {
- // suffix is empty: buffer[i] can simply point to the prefix.
- // This is not possible for the first run since the prefix
- // would point to the mutable `last_value_`.
- *prefix = prefix->substr(0, prefix_len_ptr[i]);
- buffer[i] = ByteArray(*prefix);
- return;
+ };
+
+ // Output that writes fixed-length values contiguously into a caller-provided
+ // buffer. Used by the FLBA dense decode path; avoids `buffered_data_` and
any
+ // temporary per-value storage.
+ struct DenseOutput {
+ uint8_t* out;
+ int type_length;
+
+ void Prepare(const int32_t* prefix_len_ptr, const int32_t* suffix_lengths,
+ int max_values) {}
+
+ std::string_view Emit(int i, std::string_view prefix, std::string_view
suffix) {
+ uint8_t* dst = out + static_cast<int64_t>(i) * type_length;
+ if (!prefix.empty()) {
+ memcpy(dst, prefix.data(), prefix.size());
+ }
+ if (!suffix.empty()) {
+ memcpy(dst + prefix.size(), suffix.data(), suffix.size());
}
+ // The next value's prefix references this value, which now lives in the
+ // caller's buffer.
+ return std::string_view{reinterpret_cast<const char*>(dst),
+ static_cast<size_t>(type_length)};
+ }
+ };
+
+ // Output that appends each value to an Arrow binary builder through
`helper`.
+ // A value is assembled in `scratch` only when both its prefix and its
suffix are
+ // non-empty; otherwise it is appended straight from the page or the
previous value.
+ template <typename Helper>
+ struct ArrowOutput {
+ Helper* helper;
+ std::string* scratch;
+ // Number of slots (values and nulls) the builder receives.
+ int num_slots;
+ // Builder data still to be appended, reserved by `Prepare`.
+ int64_t remaining_data_length = 0;
+ // Set when there are nulls: the runs of valid slots. The nulls before a
run are
+ // appended when its first value is; `position` is the end of the current
run.
+ ::arrow::internal::SetBitRunReader* valid_runs = nullptr;
+ int64_t position = 0;
+ int64_t run_remaining = 0;
+
+ void Prepare(const int32_t* prefix_len_ptr, const int32_t* suffix_lengths,
+ int max_values) {
+ // Each value is exactly `prefix_len + suffix_len` bytes, so reserve
exactly the
+ // data the batch appends; `Emit` can then append without capacity
checks.
+ for (int i = 0; i < max_values; ++i) {
+ if (ARROW_PREDICT_FALSE(prefix_len_ptr[i] < 0)) {
+ throw ParquetException("negative prefix length in DELTA_BYTE_ARRAY");
+ }
+ remaining_data_length +=
+ static_cast<int64_t>(prefix_len_ptr[i]) + suffix_lengths[i];
+ }
+ PARQUET_THROW_NOT_OK(helper->Prepare(num_slots, remaining_data_length));
}
- // Both prefix and suffix are non-empty, so we need to decode the string
- // into `data_ptr`.
- // 1. Copy the prefix
- memcpy(*data_ptr, prefix->data(), prefix_len_ptr[i]);
- // 2. Copy the suffix.
- memcpy(*data_ptr + prefix_len_ptr[i], buffer[i].ptr, buffer[i].len);
- // 3. Point buffer[i] to the decoded string.
- buffer[i].ptr = *data_ptr;
- buffer[i].len += prefix_len_ptr[i];
- *data_ptr += buffer[i].len;
- *prefix = std::string_view{buffer[i]};
- }
- int GetInternal(ByteArray* buffer, int max_values) {
- // Decode up to `max_values` strings into an internal buffer
- // and reference them into `buffer`.
+ std::string_view Emit(int i, std::string_view prefix, std::string_view
suffix) {
+ std::string_view value;
+ if (prefix.empty()) {
+ value = suffix;
+ } else if (suffix.empty()) {
+ // For the first value, `prefix` is `last_value_`, which is not
modified
+ // until the whole batch is decoded.
+ value = prefix;
+ } else {
+ if (prefix.data() == scratch->data()) {
+ // The previous value is in `scratch`: keep its first
`prefix.size()` bytes.
+ scratch->resize(prefix.size());
+ } else {
+ scratch->assign(prefix);
+ }
+ scratch->append(suffix);
+ value = *scratch;
+ }
+ if (valid_runs != nullptr) {
+ if (run_remaining == 0) {
+ const auto run = valid_runs->NextRun();
+ PARQUET_THROW_NOT_OK(helper->AppendNulls(run.position - position));
+ position = run.position + run.length;
Review Comment:
It's probably worth a debug check here that `run.length` is > 0.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
