github-actions[bot] commented on code in PR #65674:
URL: https://github.com/apache/doris/pull/65674#discussion_r3618732887
##########
be/src/format_v2/parquet/parquet_statistics.cpp:
##########
@@ -1388,127 +1051,148 @@ void collect_request_leaf_schemas(
}
}
-bool build_page_skip_plan_for_leaf(
- const std::shared_ptr<::parquet::RowGroupPageIndexReader>& row_group,
- const ParquetColumnSchema& column_schema, const std::vector<RowRange>&
selected_ranges,
- int64_t row_group_rows, ParquetPageSkipPlan* page_skip_plan) {
- DORIS_CHECK(page_skip_plan != nullptr);
- *page_skip_plan = ParquetPageSkipPlan {};
- if (column_schema.kind != ParquetColumnSchemaKind::PRIMITIVE ||
- column_schema.descriptor == nullptr || column_schema.leaf_column_id <
0 ||
- column_schema.descriptor->max_repetition_level() != 0) {
+template <typename ValueType>
+bool set_native_page_scalar_min_max(const tparquet::ColumnIndex& column_index,
+ const ParquetColumnSchema& column_schema,
size_t page_idx,
+ DecodedValueKind kind,
ParquetColumnStatistics* page_statistics,
+ const cctz::time_zone* timezone) {
+ if (page_idx >= column_index.min_values.size() || page_idx >=
column_index.max_values.size() ||
+ column_index.min_values[page_idx].size() != sizeof(ValueType) ||
+ column_index.max_values[page_idx].size() != sizeof(ValueType)) {
return false;
}
-
- std::shared_ptr<::parquet::OffsetIndex> offset_index;
- try {
- offset_index = row_group->GetOffsetIndex(column_schema.leaf_column_id);
- } catch (const ::parquet::ParquetException&) {
+ const auto min_value =
unaligned_load<ValueType>(column_index.min_values[page_idx].data());
+ const auto max_value =
unaligned_load<ValueType>(column_index.max_values[page_idx].data());
+ if constexpr (std::is_same_v<ValueType, int64_t>) {
+ if (!timestamp_min_max_is_safe(column_schema, min_value, max_value,
timezone)) {
+ return false;
+ }
+ }
+ if (!valid_min_max(min_value, max_value)) {
+ return true;
+ }
+ if (!set_decoded_field(column_schema, kind, min_value,
&page_statistics->min_value, timezone) ||
+ !set_decoded_field(column_schema, kind, max_value,
&page_statistics->max_value, timezone)) {
return false;
- } catch (const std::exception&) {
+ }
+ if (decoded_min_max_is_ordered(*page_statistics)) {
+ page_statistics->has_min_max = true;
+ }
+ return true;
+}
+
+bool build_native_page_statistics(const tparquet::ColumnIndex& column_index,
+ const ParquetColumnSchema& column_schema,
size_t page_idx,
+ ParquetColumnStatistics* page_statistics,
+ const cctz::time_zone* timezone) {
+ DORIS_CHECK(page_statistics != nullptr);
+ *page_statistics = {};
+ if (!column_index.__isset.null_counts || page_idx >=
column_index.null_pages.size() ||
+ page_idx >= column_index.null_counts.size()) {
return false;
}
- if (offset_index == nullptr) {
+ const int64_t null_count = column_index.null_counts[page_idx];
+ if (null_count < 0 || (column_index.null_pages[page_idx] && null_count ==
0)) {
Review Comment:
[P1] Validate null_count against this page's OffsetIndex row span before
using null_pages as an all-null proof. Page-index pruning only reaches flat
primitive leaves, so a ten-row page cannot legitimately report null_pages=true
with null_count=5 or 11; both currently set has_not_null=false, and an ordinary
comparison then returns kNoMatch and skips any real matching values without
reading the page. Bound every null count by the span and require an all-null
page's count to equal it, falling back from contradictory optional index
metadata.
##########
be/src/format_v2/parquet/parquet_scan.cpp:
##########
@@ -1585,6 +2120,10 @@ Status ParquetScanScheduler::read_next_batch(
_current_range_rows_read = 0;
}
if (*rows == 0) {
+ // Output-width feedback has no sample for a fully rejected batch.
Grow only the
+ // predicate-side physical read so a 32-row probe cannot pin a
long filtered prefix;
+ // any eventual output still stays below RuntimeState's row cap.
+ predicate_batch_rows =
grow_empty_predicate_batch(predicate_batch_rows);
Review Comment:
[P1] Keep the empty-prefix accelerator from widening the output batch.
FileScannerV2 passes a row cap derived from the preferred block bytes (32 for a
sufficiently wide projection), but after one rejected probe this grows the next
physical batch to 256. If those rows survive, read_current_row_group_batch
materializes and returns all 256; after further empty spans it can return 4096
rows, 128x the caller cap, so wide lazy columns can produce a multi-gigabyte
block or OOM. Please separate predicate-only probing from output
materialization, or otherwise ensure every returned/materialized batch remains
capped by _batch_size.
##########
be/src/format_v2/parquet/reader/native/page_reader.cpp:
##########
@@ -0,0 +1,306 @@
+// 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/page_reader.h"
+
+#include <fmt/format.h>
+#include <gen_cpp/parquet_types.h>
+#include <stddef.h>
+#include <stdint.h>
+
+#include <algorithm>
+
+#include "common/compiler_util.h" // IWYU pragma: keep
+#include "common/config.h"
+#include "io/fs/buffered_reader.h"
+#include "runtime/runtime_profile.h"
+#include "storage/cache/page_cache.h"
+#include "util/slice.h"
+#include "util/thrift_util.h"
+
+namespace doris {
+namespace io {
+struct IOContext;
+} // namespace io
+} // namespace doris
+
+namespace doris::format::parquet::native {
+static constexpr size_t INIT_PAGE_HEADER_SIZE = 128;
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status PageReader<IN_COLLECTION, OFFSET_INDEX>::_validate_page_header(uint32_t
header_size) const {
+ if (UNLIKELY(_cur_page_header.compressed_page_size < 0 ||
+ _cur_page_header.uncompressed_page_size < 0)) {
+ return Status::Corruption("Parquet page has a negative compressed or
uncompressed size");
+ }
+ if (UNLIKELY(header_size > _end_offset - _offset ||
Review Comment:
[P1] Reconcile each indexed PageLocation with the data page actually parsed
here. validate_offset_index only proves that the declared location rectangles
do not overlap, while this check only proves that the parsed header plus
payload fits somewhere in the whole chunk. A page can therefore cross the next
location; alternatively, this reader can skip an auxiliary page at the declared
offset and accept a shifted data page before next_page seeks backward into it.
Require both the parsed data page's header start to equal PageLocation.offset
and its serialized header plus payload size to equal
PageLocation.compressed_page_size (including on cache hits), or discard the
optional index and fall back.
##########
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;
+}
+
+static bool is_map_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.converted_type &&
+ (schema.converted_type == tparquet::ConvertedType::MAP ||
+ schema.converted_type == tparquet::ConvertedType::MAP_KEY_VALUE);
+}
+
+static bool is_repeated_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.repetition_type &&
+ schema.repetition_type == tparquet::FieldRepetitionType::REPEATED;
+}
+
+static bool is_required_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.repetition_type &&
+ schema.repetition_type == tparquet::FieldRepetitionType::REQUIRED;
+}
+
+static bool is_optional_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.repetition_type &&
+ schema.repetition_type == tparquet::FieldRepetitionType::OPTIONAL;
+}
+
+static int num_children_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.num_children ? schema.num_children : 0;
+}
+
+static Status validate_native_schema_structure(
+ const std::vector<tparquet::SchemaElement>& schemas) {
+ if (schemas.empty()) {
+ return Status::InvalidArgument("Wrong parquet root schema element");
+ }
+
+ const auto& root = schemas[0];
+ if (root.__isset.type || !root.__isset.num_children || root.num_children
<= 0 ||
Review Comment:
[P2] Preserve the previous empty-schema contract. The replaced schema path
had an explicit positive test for a required root with zero children, and the
Parquet SchemaElement contract does not require num_children to be positive.
The new downstream path can already represent zero physical fields and answer
metadata-only COUNT(*), but this condition now rejects the file before reaching
it. Please allow num_children == 0 when the flattened schema contains only the
root and restore the compatibility test.
##########
be/src/format_v2/parquet/reader/native/level_reader.cpp:
##########
@@ -0,0 +1,216 @@
+// 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/level_reader.h"
+
+#include <algorithm>
+#include <utility>
+
+#include "format_v2/parquet/native_schema_desc.h"
+#include "format_v2/parquet/reader/native/column_chunk_reader.h"
+#include "io/fs/buffered_reader.h"
+#include "io/fs/tracing_file_reader.h"
+
+namespace doris::format::parquet::native {
+
+class LevelReader::Impl {
+public:
+ virtual ~Impl() = default;
+
+ virtual Status init() = 0;
+ virtual Status read_rows(size_t rows, std::vector<level_t>*
repetition_levels,
+ std::vector<level_t>* definition_levels, size_t*
rows_read) = 0;
+ virtual ColumnChunkReaderStatistics statistics() = 0;
+};
+
+template <bool IN_COLLECTION>
+class LevelReaderImpl final : public LevelReader::Impl {
+public:
+ LevelReaderImpl(io::FileReaderSPtr file, tparquet::ColumnChunk
column_chunk,
+ NativeFieldSchema* field, size_t total_rows, size_t
max_buffer_size,
+ io::IOContext* io_ctx, bool enable_page_cache, std::string
page_cache_file_key,
+ ParquetReaderCompat compat)
+ : _file(std::move(file)),
+ _column_chunk(std::move(column_chunk)),
+ _field(field),
+ _total_rows(total_rows),
+ _max_buffer_size(max_buffer_size),
+ _io_ctx(io_ctx),
+ _enable_page_cache(enable_page_cache),
+ _page_cache_file_key(std::move(page_cache_file_key)),
+ _compat(compat) {}
+
+ Status init() override {
+ DORIS_CHECK(_file != nullptr);
+ DORIS_CHECK(_field != nullptr);
+ const auto& metadata = _column_chunk.meta_data;
+ ColumnChunkRange chunk_range;
+ RETURN_IF_ERROR(compute_column_chunk_range(metadata, _file->size(),
+
_compat.parquet_816_padding, &chunk_range));
+ const size_t chunk_start = chunk_range.offset;
+ const size_t chunk_size = chunk_range.length;
+ size_t prefetch_buffer_size = std::min(chunk_size, _max_buffer_size);
+ auto* tracing_reader =
typeid_cast<io::TracingFileReader*>(_file.get());
+ if ((tracing_reader != nullptr &&
+
typeid_cast<io::MergeRangeFileReader*>(tracing_reader->inner_reader().get()) !=
+ nullptr) ||
+ typeid_cast<io::MergeRangeFileReader*>(_file.get()) != nullptr) {
+ prefetch_buffer_size = 0;
+ }
+ _stream = std::make_unique<io::BufferedFileStreamReader>(_file,
chunk_start, chunk_size,
+
prefetch_buffer_size);
+ _chunk_reader = std::make_unique<ColumnChunkReader<IN_COLLECTION,
false>>(
+ _stream.get(), &_column_chunk, _field, nullptr, _total_rows,
_io_ctx,
+ ParquetPageReadContext(_enable_page_cache,
_page_cache_file_key,
+ _compat.data_page_v2_always_compressed),
+ &chunk_range);
+ return _chunk_reader->init();
+ }
+
+ Status read_rows(size_t rows, std::vector<level_t>* repetition_levels,
+ std::vector<level_t>* definition_levels, size_t*
rows_read) override {
+ DORIS_CHECK(repetition_levels != nullptr);
+ DORIS_CHECK(definition_levels != nullptr);
+ DORIS_CHECK(rows_read != nullptr);
+ repetition_levels->clear();
+ definition_levels->clear();
+ *rows_read = 0;
+ if (_current_row > _total_rows || rows > _total_rows - _current_row) {
+ return Status::Corruption("Parquet level reader requested rows
[{}, {}) of {}",
+ _current_row, _current_row + rows,
_total_rows);
+ }
+ if constexpr (IN_COLLECTION) {
+ RETURN_IF_ERROR(
+ read_nested_rows(rows, repetition_levels,
definition_levels, rows_read));
+ } else {
+ RETURN_IF_ERROR(read_flat_rows(rows, repetition_levels,
definition_levels, rows_read));
+ }
+ _current_row += *rows_read;
+ return Status::OK();
+ }
+
+ ColumnChunkReaderStatistics statistics() override { return
_chunk_reader->statistics(); }
+
+private:
+ Status read_flat_rows(size_t rows, std::vector<level_t>* repetition_levels,
+ std::vector<level_t>* definition_levels, size_t*
rows_read) {
+ while (*rows_read < rows) {
+ RETURN_IF_ERROR(_chunk_reader->parse_page_header());
+ RETURN_IF_ERROR(_chunk_reader->load_page_data_idempotent());
+ const size_t read_now = std::min(
+ rows - *rows_read,
static_cast<size_t>(_chunk_reader->remaining_num_values()));
+ if (read_now == 0) {
Review Comment:
[P2] Advance a zero-value data page before applying the no-progress error. A
leading V1/V2 page with num_values=0 and a later nonempty page is accepted by
the page/chunk validation and by the ordinary scalar reader, which observes an
empty row range and calls next_page. COUNT(nullable_col) reaches this
level-only path instead and returns corruption before its identical next-page
branch. Please skip bounded zero-value pages when another page exists and add
scalar-versus-COUNT parity coverage.
--
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]
