github-actions[bot] commented on code in PR #65674:
URL: https://github.com/apache/doris/pull/65674#discussion_r3619976963
##########
be/src/format_v2/parquet/parquet_scan.cpp:
##########
@@ -1351,54 +1840,82 @@ Status
ParquetScanScheduler::read_filter_columns(int64_t batch_rows,
return Status::OK();
};
+ auto compact_predicate_columns_with_profile =
+ [&](bool discard_predicate_only_payload) -> Status {
+ const int64_t start_ns = MonotonicNanos();
+ auto status =
compact_predicate_columns(discard_predicate_only_payload);
+ update_counter_if_not_null(_scan_profile.predicate_compaction_time,
+ MonotonicNanos() - start_ns);
+ return status;
+ };
+
RETURN_IF_ERROR(read_round_by_round());
+ // Single-column expressions only touch the just-read column, so earlier
columns can retain
+ // their own row mappings. Compact only when a later expression needs a
shared coordinate
+ // space; otherwise the final boundary can discard hidden predicate
payloads without scanning
+ // them again.
+ if (!schedule.remaining_conjuncts.empty()) {
+ RETURN_IF_ERROR(compact_predicate_columns_with_profile(false));
+ }
RETURN_IF_ERROR(execute_scheduled_conjuncts_with_profile(schedule.remaining_conjuncts));
- if (_scan_profile.predicate_filter_time == nullptr) {
- return execute_scheduled_delete_conjuncts();
+ if (!request.delete_conjuncts.empty()) {
+ RETURN_IF_ERROR(compact_predicate_columns_with_profile(false));
}
- SCOPED_TIMER(_scan_profile.predicate_filter_time);
- return execute_scheduled_delete_conjuncts();
-}
-
-bool ParquetScanScheduler::prepare_current_row_group_reader(
- ParquetFileContext& file_context,
- const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
- const format::FileScanRequest& request, int row_group_idx) {
- if (file_context.metadata == nullptr) {
- return false;
+ if (_scan_profile.predicate_filter_time == nullptr) {
+ RETURN_IF_ERROR(execute_scheduled_delete_conjuncts());
+ } else {
+ SCOPED_TIMER(_scan_profile.predicate_filter_time);
+ RETURN_IF_ERROR(execute_scheduled_delete_conjuncts());
}
- const auto ranges = build_row_group_prefetch_ranges(
- *file_context.metadata, file_schema,
request_scan_columns(request), row_group_idx);
- const size_t avg_io_size = detail::average_prefetch_range_size(ranges);
- return file_context.set_random_access_ranges(ranges, avg_io_size, _profile,
- _merge_read_slice_size);
+ return compact_predicate_columns_with_profile(true);
}
-void ParquetScanScheduler::prefetch_current_row_group_columns(
+Status ParquetScanScheduler::prefetch_current_row_group_columns(
ParquetFileContext& file_context,
const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
const std::vector<format::LocalColumnIndex>& scan_columns, bool*
prefetched) {
DORIS_CHECK(prefetched != nullptr);
if (_current_merge_range_active || *prefetched || scan_columns.empty() ||
- _current_row_group_id < 0 || file_context.metadata == nullptr) {
- return;
+ _current_row_group_id < 0 || file_context.native_metadata == nullptr) {
+ return Status::OK();
}
*prefetched = true;
// The scanner request separates predicate and non-predicate columns so
Parquet can read
// predicate columns first and lazily materialize the rest. Keep the same
contract for
// prefetch: callers decide which side to warm, and this helper only
translates that selected
// projection into physical column-chunk byte ranges for the current row
group.
- file_context.prefetch_ranges(
- build_row_group_prefetch_ranges(*file_context.metadata,
file_schema, scan_columns,
- _current_row_group_id),
- nullptr);
+ const auto& metadata = file_context.native_metadata->to_thrift();
+ const auto compat = native::parquet_reader_compat(
+ metadata.__isset.created_by ? metadata.created_by : std::string
{});
+ std::vector<ParquetPageCacheRange> ranges;
+ RETURN_IF_ERROR(detail::build_native_prefetch_ranges(
+ metadata, file_schema, scan_columns, _current_row_group_id,
+ file_context.native_file->size(), compat.parquet_816_padding,
&ranges));
+ file_context.prefetch_ranges(ranges, nullptr);
+ return Status::OK();
}
Status ParquetScanScheduler::read_current_row_group_batch(
ParquetFileContext& file_context,
const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
int64_t batch_rows,
const format::FileScanRequest& request, int64_t batch_first_file_row,
Block* file_block,
size_t* rows) {
+ // Reader statistics are cumulative plain integers. Publishing their delta
recursively for
+ // every tiny batch is measurable on wide/nested scans, so flush
periodically and force the
+ // tail at row-group reset/close.
+ Defer profile_flush {[this, batch_rows]() {
Review Comment:
[P2] Keep this batch's page-crossing sample open until its pending lazy
slices are drained. When `selected_rows > _batch_size`, this scope materializes
only the first output slice, but the `Defer` immediately calls
`finish_current_reader_batch_profiles()` and advances every native reader's
page-count snapshot. Later `read_next_batch()` calls materialize the remaining
lazy slices without another boundary sample. A lazy column that reads page A in
the first slice and page B in a later slice therefore contributes two separate
one-page deltas instead of one crossing; the crossing is lost at row-group EOF
or attributed to the next physical batch. Please finalize `PageCrossingBatches`
once after the pending selection is empty, and cover a pending drain whose
later slice enters the next lazy page.
##########
be/src/format_v2/parquet/reader/native/delta_bit_pack_decoder.cpp:
##########
@@ -0,0 +1,180 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "format_v2/parquet/reader/native/delta_bit_pack_decoder.h"
+
+namespace doris::format::parquet::native {
+Status DeltaLengthByteArrayDecoder::_init_lengths() {
+ auto length_reader = std::make_shared<BitReader>(*_bit_reader);
+ RETURN_IF_ERROR(_len_decoder.set_bit_reader(length_reader));
+ const uint32_t num_lengths = _len_decoder.valid_values_count();
+
+ DeltaBitPackDecoder<int32_t> length_locator;
+ length_locator.set_expected_values(_expected_values);
+ RETURN_IF_ERROR(length_locator.set_bit_reader(_bit_reader));
+ constexpr size_t kLocateBatchSize = 4096;
+ std::vector<int32_t> lengths(std::min<size_t>(num_lengths,
kLocateBatchSize));
+ size_t located = 0;
+ int64_t payload_size = 0;
+ while (located < num_lengths) {
+ const size_t batch_size = std::min<size_t>(num_lengths - located,
kLocateBatchSize);
+ uint32_t decoded = 0;
+ RETURN_IF_ERROR(
+ length_locator.decode(lengths.data(),
static_cast<uint32_t>(batch_size), &decoded));
+ if (UNLIKELY(decoded != batch_size)) {
+ return Status::Corruption("Parquet delta length stream ended
early");
+ }
+ for (size_t i = 0; i < batch_size; ++i) {
+ if (UNLIKELY(lengths[i] < 0) ||
+ common::add_overflow(payload_size,
static_cast<int64_t>(lengths[i]),
+ payload_size)) {
+ return Status::Corruption("Invalid Parquet delta byte-array
length");
+ }
+ }
+ located += batch_size;
+ }
+ // The locator leaves the shared reader at the first payload byte without
retaining every
+ // length; the independent decoder supplies bounded batches during
materialization or skip.
+ if (UNLIKELY(payload_size > _bit_reader->bytes_left())) {
+ return Status::Corruption("Parquet delta lengths require {} bytes,
only {} remain",
+ payload_size, _bit_reader->bytes_left());
+ }
+ _num_valid_values = num_lengths;
+ return Status::OK();
+}
+
+Status DeltaLengthByteArrayDecoder::_get_internal(Slice* buffer, int
max_values,
+ int* out_num_values) {
+ // Decode up to `max_values` strings into an internal buffer
+ // and reference them into `buffer`.
+ max_values = std::min(max_values, _num_valid_values);
+ if (max_values == 0) {
+ *out_num_values = 0;
+ return Status::OK();
+ }
+
+ int64_t data_size = 0;
+ _buffered_length.resize(max_values);
+ uint32_t lengths_decoded = 0;
+ RETURN_IF_ERROR(_len_decoder.decode(_buffered_length.data(),
static_cast<uint32_t>(max_values),
+ &lengths_decoded));
+ if (UNLIKELY(lengths_decoded != max_values)) {
+ return Status::Corruption("Parquet delta length stream ended early");
+ }
+ const int32_t* length_ptr = _buffered_length.data();
+ for (int i = 0; i < max_values; ++i) {
+ int32_t len = length_ptr[i];
+ if (len < 0) [[unlikely]] {
+ return Status::InvalidArgument("Negative string delta length");
+ }
+ buffer[i].size = len;
+ if (common::add_overflow(data_size, static_cast<int64_t>(len),
data_size)) {
+ return Status::InvalidArgument("Excess expansion in
DELTA_(LENGTH_)BYTE_ARRAY");
+ }
+ }
+ // Every declared byte must exist in this page. Check before resize so a
tiny malformed stream
+ // cannot reserve memory based only on attacker-controlled decoded lengths.
+ if (UNLIKELY(data_size > _bit_reader->bytes_left())) {
+ return Status::Corruption("Parquet delta lengths require {} bytes,
only {} remain",
+ data_size, _bit_reader->bytes_left());
+ }
+ _buffered_data.resize(data_size);
+ char* data_ptr = _buffered_data.data();
+ for (int64_t j = 0; j < data_size; j++) {
+ if (!_bit_reader->GetValue(8, data_ptr + j)) {
+ return Status::IOError("Get length bytes EOF");
+ }
+ }
+
+ for (int i = 0; i < max_values; ++i) {
+ buffer[i].data = data_ptr;
+ data_ptr += buffer[i].size;
+ }
+ // this->num_values_ -= max_values;
+ _num_valid_values -= max_values;
+ *out_num_values = max_values;
+ return Status::OK();
+}
+
+Status DeltaByteArrayDecoder::_get_internal(Slice* buffer, int max_values,
int* out_num_values) {
+ // Decode up to `max_values` strings into an internal buffer
+ // and reference them into `buffer`.
+ max_values = std::min(max_values, _num_valid_values);
+ if (max_values == 0) {
+ *out_num_values = max_values;
+ return Status::OK();
+ }
+
+ int suffix_read;
+ RETURN_IF_ERROR(_suffix_decoder.decode(buffer, max_values, &suffix_read));
+ if (suffix_read != max_values) [[unlikely]] {
+ return Status::IOError("Read {}, expecting {} from suffix decoder",
+ std::to_string(suffix_read),
std::to_string(max_values));
+ }
+
+ int64_t data_size = 0;
+ _buffered_prefix_length.resize(max_values);
+ uint32_t prefixes_decoded = 0;
+ RETURN_IF_ERROR(_prefix_len_decoder.decode(
+ _buffered_prefix_length.data(), static_cast<uint32_t>(max_values),
&prefixes_decoded));
+ if (UNLIKELY(prefixes_decoded != max_values)) {
+ return Status::Corruption("Parquet delta prefix stream ended early");
+ }
+ const int32_t* prefix_len_ptr = _buffered_prefix_length.data();
+ size_t preceding_value_size = _last_value.size();
+ for (int i = 0; i < max_values; ++i) {
+ if (prefix_len_ptr[i] < 0) [[unlikely]] {
+ return Status::InvalidArgument("negative prefix length in
DELTA_BYTE_ARRAY");
+ }
+ const size_t prefix_size = static_cast<size_t>(prefix_len_ptr[i]);
+ if (prefix_size > preceding_value_size) [[unlikely]] {
+ // Prefixes form a dependency chain, so validate each one before
aggregate allocation.
+ return Status::InvalidArgument("prefix length too large in
DELTA_BYTE_ARRAY");
+ }
+ size_t reconstructed_size = 0;
+ if (common::add_overflow(prefix_size, buffer[i].size,
reconstructed_size) ||
+ reconstructed_size >
static_cast<size_t>(std::numeric_limits<int64_t>::max()) ||
+ common::add_overflow(data_size,
static_cast<int64_t>(reconstructed_size), data_size))
+ [[unlikely]] {
+ return Status::InvalidArgument("excess expansion in
DELTA_BYTE_ARRAY");
+ }
+ preceding_value_size = reconstructed_size;
+ }
+ _buffered_data.resize(data_size);
Review Comment:
[P1] Bound this reconstruction by bytes, not just value count.
`_get_internal()` allocates the sum of every reconstructed prefix plus suffix
before filtering or skipping is applied. A roughly 4 MiB valid page can encode
4096 identical 4 MiB strings as one suffix followed by compact full-prefix
references, so the 4096-row skip chunk (and a sparse selection keeping zero or
one row) attempts about 16 GiB here. Level-only aggregate reads also reach this
path while discarding payloads. Please reconstruct incrementally or use a
byte-bounded chunk while retaining only the previous value and selected output,
and cover long repeated prefixes in filter-all/sparse/count paths.
##########
be/src/format_v2/parquet/native_schema_desc.cpp:
##########
@@ -0,0 +1,825 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "format_v2/parquet/native_schema_desc.h"
+
+#include <ctype.h>
+
+#include <algorithm>
+#include <ostream>
+#include <utility>
+
+#include "common/cast_set.h"
+#include "common/exception.h"
+#include "common/logging.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_factory.hpp"
+#include "core/data_type/data_type_map.h"
+#include "core/data_type/data_type_struct.h"
+#include "core/data_type/define_primitive_type.h"
+#include "util/slice.h"
+#include "util/string_util.h"
+
+namespace doris::format::parquet {
+
+static bool is_group_node(const tparquet::SchemaElement& schema) {
+ return schema.num_children > 0;
+}
+
+static bool is_list_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.converted_type && schema.converted_type ==
tparquet::ConvertedType::LIST;
Review Comment:
[P1] Preserve logicalType-only LIST/MAP schemas here. The removed
Arrow-backed parser recognized both the converted annotation and
`logical_type->is_list()` / `is_map()`, but these new helpers consult only
`converted_type`. A valid group encoded only with
`SchemaElement.logicalType.LIST` or `.MAP` now falls through to generic STRUCT
parsing, so the file exposes the wrong nested type and table mapping rejects it
(or builds the wrong shape). Please recognize `schema.logicalType.__isset.LIST`
/ `.MAP` as well and add logicalType-only LIST and MAP schema cases; the new
tests currently set only `converted_type`.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
