|
19 | 19 |
|
20 | 20 | #include <algorithm> |
21 | 21 | #include <cstring> |
| 22 | +#include <iterator> |
22 | 23 | #include <memory> |
23 | | -#include <random> |
| 24 | +#include <span> |
24 | 25 | #include <unordered_set> |
25 | 26 | #include <utility> |
26 | 27 | #include <vector> |
27 | 28 |
|
28 | | -#include "arrow/array.h" |
| 29 | +#include "arrow/array.h" // IWYU pragma: keep |
| 30 | +#include "arrow/array/concatenate.h" |
29 | 31 | #include "arrow/buffer.h" |
30 | 32 | #include "arrow/extension_type.h" |
31 | 33 | #include "arrow/io/memory.h" |
|
35 | 37 | #include "arrow/type.h" |
36 | 38 | #include "arrow/type_traits.h" |
37 | 39 | #include "arrow/util/async_generator.h" |
| 40 | +#include "arrow/util/bit_run_reader.h" |
38 | 41 | #include "arrow/util/bit_util.h" |
39 | 42 | #include "arrow/util/future.h" |
40 | 43 | #include "arrow/util/iterator.h" |
|
45 | 48 | #include "arrow/util/type_traits.h" |
46 | 49 |
|
47 | 50 | #include "parquet/arrow/reader_internal.h" |
48 | | -#include "parquet/bloom_filter.h" |
49 | | -#include "parquet/bloom_filter_reader.h" |
50 | 51 | #include "parquet/column_reader.h" |
51 | 52 | #include "parquet/exception.h" |
52 | 53 | #include "parquet/file_reader.h" |
53 | 54 | #include "parquet/metadata.h" |
54 | | -#include "parquet/page_index.h" |
55 | 55 | #include "parquet/properties.h" |
56 | 56 | #include "parquet/schema.h" |
57 | 57 |
|
@@ -725,13 +725,66 @@ class PARQUET_NO_EXPORT FixedSizeListReader : public ListReader<int32_t> { |
725 | 725 | DCHECK_EQ(data->buffers.size(), 2); |
726 | 726 | DCHECK_EQ(field()->type()->id(), ::arrow::Type::FIXED_SIZE_LIST); |
727 | 727 | const auto& type = checked_cast<::arrow::FixedSizeListType&>(*field()->type()); |
728 | | - const int32_t* offsets = reinterpret_cast<const int32_t*>(data->buffers[1]->data()); |
729 | | - for (int x = 1; x <= data->length; x++) { |
730 | | - int32_t size = offsets[x] - offsets[x - 1]; |
731 | | - if (size != type.list_size()) { |
732 | | - return Status::Invalid("Expected all lists to be of size=", type.list_size(), |
733 | | - " but index ", x, " had size=", size); |
| 728 | + const auto* offsets = reinterpret_cast<const int32_t*>(data->buffers[1]->data()); |
| 729 | + const int32_t list_size = type.list_size(); |
| 730 | + const uint8_t* valid_bits = |
| 731 | + data->buffers[0] != nullptr ? data->buffers[0]->data() : nullptr; |
| 732 | + auto validate_offsets = [&](int64_t start, int64_t length, |
| 733 | + int32_t expected_size) -> Status { |
| 734 | + std::span<const int32_t> run_offsets(offsets + start, |
| 735 | + static_cast<size_t>(length + 1)); |
| 736 | + const auto first_invalid_offset = std::ranges::adjacent_find( |
| 737 | + run_offsets, |
| 738 | + [&](int32_t left, int32_t right) { return right - left != expected_size; }); |
| 739 | + if (first_invalid_offset != run_offsets.end()) { |
| 740 | + const int64_t x = |
| 741 | + start + std::ranges::distance(run_offsets.begin(), first_invalid_offset); |
| 742 | + return Status::Invalid("Expected offset at index ", x + 1, " to be ", |
| 743 | + offsets[x] + expected_size, " but got ", offsets[x + 1]); |
734 | 744 | } |
| 745 | + return Status::OK(); |
| 746 | + }; |
| 747 | + if (valid_bits != nullptr) { |
| 748 | + bool needs_padding = false; |
| 749 | + ::arrow::ArrayVector child_arrays; |
| 750 | + |
| 751 | + auto append_child_run = [&](int64_t start, int64_t length, bool valid) -> Status { |
| 752 | + const int64_t child_length = length * list_size; |
| 753 | + if (!valid) { |
| 754 | + ARROW_ASSIGN_OR_RAISE( |
| 755 | + auto null_array, |
| 756 | + ::arrow::MakeArrayOfNull(type.value_type(), child_length, ctx_->pool)); |
| 757 | + child_arrays.push_back(std::move(null_array)); |
| 758 | + return Status::OK(); |
| 759 | + } |
| 760 | + child_arrays.push_back( |
| 761 | + ::arrow::MakeArray(data->child_data[0]->Slice(offsets[start], child_length))); |
| 762 | + return Status::OK(); |
| 763 | + }; |
| 764 | + |
| 765 | + auto visit_run = [&](int64_t start, int64_t length, bool valid) -> Status { |
| 766 | + RETURN_NOT_OK(validate_offsets(start, length, valid ? list_size : 0)); |
| 767 | + if (valid && !needs_padding) { |
| 768 | + return Status::OK(); |
| 769 | + } |
| 770 | + if (!needs_padding) { |
| 771 | + needs_padding = true; |
| 772 | + if (start > 0) { |
| 773 | + RETURN_NOT_OK(append_child_run(/*start=*/0, start, /*valid=*/true)); |
| 774 | + } |
| 775 | + } |
| 776 | + return append_child_run(start, length, valid); |
| 777 | + }; |
| 778 | + |
| 779 | + RETURN_NOT_OK(::arrow::internal::VisitBitRuns(valid_bits, data->offset, |
| 780 | + data->length, visit_run)); |
| 781 | + if (needs_padding) { |
| 782 | + ARROW_ASSIGN_OR_RAISE(auto child_array_with_padding, |
| 783 | + ::arrow::Concatenate(child_arrays, ctx_->pool)); |
| 784 | + data->child_data[0] = child_array_with_padding->data(); |
| 785 | + } |
| 786 | + } else { |
| 787 | + RETURN_NOT_OK(validate_offsets(/*start=*/0, data->length, list_size)); |
735 | 788 | } |
736 | 789 | data->buffers.resize(1); |
737 | 790 | std::shared_ptr<Array> result = ::arrow::MakeArray(data); |
|
0 commit comments