github-actions[bot] commented on code in PR #65674:
URL: https://github.com/apache/doris/pull/65674#discussion_r3624033850
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/AggregateStrategies.java:
##########
@@ -740,10 +740,15 @@ private LogicalAggregate<? extends Plan>
storageLayerAggregate(
}
}
if (mergeOp == PushDownAggOp.COUNT || mergeOp ==
PushDownAggOp.MIX) {
- // NULL value behavior in `count` function is zero, so
- // we should not use row_count to speed up query. the col
- // must be not null
- if (column.isAllowNull() && checkNullSlots.contains(slot)) {
+ // Nullable file COUNT is exact only when this query is routed
to FileScannerV2,
+ // which carries the semantic argument and counts definition
levels. Gating on the
+ // session switch keeps V1 on its original full-column
evaluation path.
+ boolean supportsNullableFileCount = logicalScan instanceof
LogicalFileScan
Review Comment:
[P1] Keep nullable COUNT pushdown away from row-count-only text readers.
This makes every `LogicalFileScan` eligible whenever V2 is on, not just
Parquet/ORC. For CSV/TEXT, `TableReader` builds a non-empty count request for
`COUNT(nullable_col)`, but `DelimitedTextReader::get_aggregate_result` ignores
`request.columns` and increments once per valid record. A file with `[NULL, x]`
therefore returns 2 instead of 1, and the reader is closed as successfully
pushed down. Please make these formats return `NotSupported` for non-empty
COUNT columns, or restrict eligibility to readers that can count nullability.
##########
be/src/format_v2/table_reader.cpp:
##########
@@ -588,16 +616,17 @@ Status TableReader::_build_table_filters_from_conjuncts()
{
// `_table_filters` omits expressions without slot references, but
such an expression still
// occupies a position in the row-level conjunct order. Record how
many localized filters
// precede the first unsafe original conjunct so constant pruning
cannot jump over a
- // slotless non-deterministic/error-preserving barrier.
+ // slotless non-deterministic/error-preserving barrier. Unsafe
predicates must still reach
Review Comment:
[P1] Do not localize nondeterministic predicates that `Scanner`
re-evaluates. Removing the unsafe-prefix barrier sends a slot-bearing predicate
such as `x > rand()` into Parquet/ORC: the file reader evaluates the localized
clone, while `Scanner::_filter_output_block` still evaluates the original
conjunct on returned rows. The clones have separate random states, so a row
must pass two independent samples and query results change. Keep
nondeterministic unsafe predicates above `TableReader`, or ensure `Scanner`
does not re-evaluate a consumed predicate with equivalent ordering and cache
semantics.
##########
be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp:
##########
@@ -0,0 +1,1728 @@
+// 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/column_chunk_reader.h"
+
+#include <cctz/time_zone.h>
+#include <gen_cpp/parquet_types.h>
+#include <glog/logging.h>
+#include <parquet/metadata.h>
+#include <string.h>
+
+#include <algorithm>
+#include <cstdint>
+#include <limits>
+#include <memory>
+#include <utility>
+
+#include "common/compiler_util.h" // IWYU pragma: keep
+#include "core/column/column.h"
+#include "core/column/column_decimal.h"
+#include "core/column/column_dictionary.h"
+#include "core/column/column_varbinary.h"
+#include "core/column/column_vector.h"
+#include "core/custom_allocator.h"
+#include "core/data_type_serde/data_type_serde.h"
+#include "core/data_type_serde/parquet_timestamp.h"
+#include "exprs/vexpr.h"
+#include "format_v2/parquet/native_schema_desc.h"
+#include "format_v2/parquet/reader/native/decoder.h"
+#include "format_v2/parquet/reader/native/level_decoder.h"
+#include "format_v2/parquet/reader/native/page_reader.h"
+#include "io/fs/buffered_reader.h"
+#include "runtime/runtime_profile.h"
+#include "storage/cache/page_cache.h"
+#include "util/bit_util.h"
+#include "util/block_compression.h"
+#include "util/cpu_info.h"
+#include "util/unaligned.h"
+
+namespace cctz {
+class time_zone;
+} // namespace cctz
+namespace doris {
+namespace io {
+class BufferedStreamReader;
+struct IOContext;
+} // namespace io
+} // namespace doris
+
+namespace doris::format::parquet::native {
+
+bool can_prepare_page_cache_payload(bool session_cache_enabled, bool
storage_cache_disabled,
+ bool cache_available, bool
header_available) {
+ return session_cache_enabled && !storage_cache_disabled && cache_available
&& header_available;
+}
+
+Status validate_uncompressed_page_sizes(const tparquet::PageHeader& header,
+ tparquet::CompressionCodec::type codec,
+ bool data_page_v2_always_compressed) {
+ const bool is_v2 = header.__isset.data_page_header_v2;
+ const bool page_is_compressed =
+ codec != tparquet::CompressionCodec::UNCOMPRESSED &&
+ (!is_v2 || header.data_page_header_v2.is_compressed ||
data_page_v2_always_compressed);
+ if (!page_is_compressed &&
+ UNLIKELY(header.compressed_page_size !=
header.uncompressed_page_size)) {
+ // An uncompressed payload has one physical representation, so
accepting two lengths makes
+ // cold reads and decompressed cache hits consume different byte
boundaries.
+ return Status::Corruption(
+ "Uncompressed Parquet page sizes differ: compressed={},
uncompressed={}",
+ header.compressed_page_size, header.uncompressed_page_size);
+ }
+ return Status::OK();
+}
+
+ParquetReaderCompat parquet_reader_compat(const std::string& created_by) {
+ if (created_by.empty()) {
+ return {};
+ }
+ const ::parquet::ApplicationVersion version(created_by);
+ return {.parquet_816_padding =
+
version.VersionLt(::parquet::ApplicationVersion::PARQUET_816_FIXED_VERSION()),
+ .data_page_v2_always_compressed = version.VersionLt(
+
::parquet::ApplicationVersion::PARQUET_CPP_10353_FIXED_VERSION())};
+}
+
+Status compute_column_chunk_range(const tparquet::ColumnMetaData& metadata,
size_t file_size,
+ bool parquet_816_padding, ColumnChunkRange*
range) {
+ DORIS_CHECK(range != nullptr);
+ int64_t start = metadata.data_page_offset;
+ if (metadata.__isset.dictionary_page_offset &&
metadata.dictionary_page_offset > 0 &&
+ metadata.dictionary_page_offset < start) {
+ // Some writers use dictionary_page_offset=0 as an absence sentinel.
Range validation must
+ // follow has_dict_page() or the native reader starts at the Parquet
magic bytes.
+ start = metadata.dictionary_page_offset;
+ }
+ const int64_t length = metadata.total_compressed_size;
+ if (UNLIKELY(start < 0 || length < 0)) {
+ return Status::Corruption("Parquet column chunk has a negative offset
or length");
+ }
+ const uint64_t unsigned_start = static_cast<uint64_t>(start);
+ const uint64_t unsigned_length = static_cast<uint64_t>(length);
+ if (UNLIKELY(unsigned_start > file_size || unsigned_length > file_size -
unsigned_start)) {
+ // Thrift range fields are signed and untrusted; validate before
converting them to the
+ // unsigned stream-reader coordinates so overflow cannot wrap back
into the file.
+ return Status::Corruption("Parquet column chunk [{}, {}) exceeds file
size {}", start,
+ unsigned_start + unsigned_length, file_size);
+ }
+ size_t bounded_length = static_cast<size_t>(unsigned_length);
+ if (parquet_816_padding) {
+ // parquet-mr before PARQUET-816 under-reported the chunk by up to 100
bytes. Padding stays
+ // file-bounded and is only enabled for the affected writer versions.
+ bounded_length += std::min<size_t>(100, file_size - unsigned_start -
unsigned_length);
+ }
+ range->offset = static_cast<size_t>(unsigned_start);
+ range->length = bounded_length;
+ return Status::OK();
+}
+
+bool validate_offset_index(const tparquet::OffsetIndex& index, const
ColumnChunkRange& chunk_range,
+ int64_t data_page_offset, int64_t row_count) {
+ if (index.page_locations.empty() || data_page_offset < 0 || row_count < 0
||
+ index.page_locations.front().first_row_index != 0 ||
+ index.page_locations.front().offset != data_page_offset ||
+ chunk_range.length > std::numeric_limits<size_t>::max() -
chunk_range.offset) {
+ return false;
+ }
+ // Row indexes alone cannot detect a uniformly shifted OffsetIndex. Anchor
its first location
+ // to the owning metadata so page-to-row mapping cannot silently move by
one physical page.
+ const uint64_t chunk_begin = chunk_range.offset;
+ const uint64_t chunk_end = chunk_begin + chunk_range.length;
+ uint64_t previous_end = chunk_begin;
+ int64_t previous_row = -1;
+ for (const auto& location : index.page_locations) {
+ if (location.first_row_index <= previous_row ||
location.first_row_index >= row_count ||
+ location.offset < 0 || location.compressed_page_size <= 0) {
+ return false;
+ }
+ const uint64_t begin = static_cast<uint64_t>(location.offset);
+ const uint64_t size =
static_cast<uint64_t>(location.compressed_page_size);
+ if (begin < chunk_begin || begin < previous_end || begin > chunk_end ||
+ size > chunk_end - begin) {
+ return false;
+ }
+ previous_row = location.first_row_index;
+ previous_end = begin + size;
+ }
+ return true;
+}
+
+namespace {
+
+class EmptyValueSectionDecoder final : public Decoder {
+public:
+ Status skip_values(size_t num_values) override {
+ if (UNLIKELY(num_values != 0)) {
+ return Status::Corruption(
+ "Parquet definition levels require {} values from an empty
value section",
+ num_values);
+ }
+ return Status::OK();
+ }
+};
+
+Status append_v2_int96_datetime(ColumnDateTimeV2::Container& data,
+ const ParquetInt96Timestamp& value,
+ const cctz::time_zone& timezone) {
+ static constexpr int32_t JULIAN_EPOCH_OFFSET_DAYS = 2440588;
+ static constexpr int64_t MICROS_PER_DAY = 86400000000LL;
+ static constexpr int64_t MICROS_PER_SECOND = 1000000LL;
+
+ // Arrow normalized out-of-day INT96 nanos before the native V2 path
replaced it. Preserve that
+ // legacy-writer compatibility here; rejecting the carrier loses valid
pre-epoch/year-0 values
+ // already accepted by Doris external-table scans.
+ const __int128 days = static_cast<__int128>(value.julian_day) -
JULIAN_EPOCH_OFFSET_DAYS;
+ // Truncate the signed nanos field before day normalization. Flooring a
negative value first
+ // changes the historical result by one microsecond.
+ const __int128 timestamp_micros = days * MICROS_PER_DAY +
value.nanos_of_day / 1000;
+ if (timestamp_micros < std::numeric_limits<int64_t>::min() ||
+ timestamp_micros > std::numeric_limits<int64_t>::max()) {
+ return Status::DataQualityError("Parquet INT96 timestamp overflows
microseconds");
+ }
+
+ const int64_t micros = static_cast<int64_t>(timestamp_micros);
+ int64_t epoch_seconds = micros / MICROS_PER_SECOND;
+ int64_t micros_of_second = micros % MICROS_PER_SECOND;
+ if (micros_of_second < 0) {
+ micros_of_second += MICROS_PER_SECOND;
+ --epoch_seconds;
+ }
+ DateV2Value<DateTimeV2ValueType> datetime;
+ datetime.from_unixtime(epoch_seconds, timezone);
+ datetime.set_microsecond(static_cast<uint32_t>(micros_of_second));
+ if (!datetime.is_valid_date()) {
+ return Status::DataQualityError("Parquet INT96 timestamp is outside
the Doris range");
+ }
+ data.push_back(datetime);
+ return Status::OK();
+}
+
+class V2Int96DateTimeConsumer final : public ParquetFixedValueConsumer {
+public:
+ V2Int96DateTimeConsumer(IColumn& column, const ParquetDecodeContext&
context,
+ ParquetMaterializationState* state)
+ : _data(assert_cast<ColumnDateTimeV2&>(column).get_data()),
_state(state) {
+ static const auto utc = cctz::utc_time_zone();
+ _timezone = context.timezone == nullptr ? &utc : context.timezone;
+ }
+
+ Status consume(const uint8_t* values, size_t num_values, size_t
value_width) override {
+ DORIS_CHECK_EQ(value_width, sizeof(ParquetInt96Timestamp));
+ const size_t old_size = _data.size();
+ for (size_t row = 0; row < num_values; ++row) {
+ const auto value = unaligned_load<ParquetInt96Timestamp>(
+ values + row * sizeof(ParquetInt96Timestamp));
+ const auto status = append_v2_int96_datetime(_data, value,
*_timezone);
+ if (!status.ok()) {
+ if (_state != nullptr &&
_state->mark_conversion_failure(_data.size())) {
+ _data.emplace_back();
+ continue;
+ }
+ _data.resize(old_size);
+ return status;
+ }
+ }
+ return Status::OK();
+ }
+
+private:
+ ColumnDateTimeV2::Container& _data;
+ ParquetMaterializationState* _state;
+ const cctz::time_zone* _timezone = nullptr;
+};
+
+class RejectV2Int96BinaryConsumer final : public ParquetBinaryValueConsumer {
+public:
+ Status consume(const StringRef*, size_t) override {
+ return Status::NotSupported("INT96 cannot be decoded from binary
Parquet values");
+ }
+};
+
+Status read_v2_int96_datetime(IColumn& column, ParquetDecodeSource& source,
+ const ParquetDecodeContext& context, size_t
num_values,
+ ParquetMaterializationState& state) {
+ V2Int96DateTimeConsumer consumer(column, context, &state);
+ if (context.encoding != ParquetValueEncoding::DICTIONARY) {
+ return source.decode_fixed_values(num_values, consumer);
+ }
+ if (state.dictionary_generation != source.dictionary_generation()) {
+ state.typed_dictionary = column.clone_empty();
+ auto* output_null_map =
state.begin_dictionary_conversion(source.dictionary_size());
+ V2Int96DateTimeConsumer dictionary_consumer(*state.typed_dictionary,
context, &state);
+ RejectV2Int96BinaryConsumer binary_consumer;
+ const auto dictionary_status =
+ source.decode_dictionary(dictionary_consumer, binary_consumer);
+ state.end_dictionary_conversion(output_null_map);
+ RETURN_IF_ERROR(dictionary_status);
+ DORIS_CHECK_EQ(state.typed_dictionary->size(),
source.dictionary_size());
+ state.dictionary_generation = source.dictionary_generation();
+ }
+ RETURN_IF_ERROR(source.decode_dictionary_indices(num_values,
&state.dictionary_indices));
+ DORIS_CHECK_EQ(state.dictionary_indices.size(), num_values);
+ return state.materialize_dictionary(column);
+}
+
+Status read_native_or_serde(IColumn& column, const DataTypeSerDe& serde,
+ ParquetDecodeSource& source, const
ParquetDecodeContext& context,
+ size_t num_values, ParquetMaterializationState&
state) {
+ if (context.physical_type == ParquetPhysicalType::INT96 &&
+ check_and_get_column<ColumnDateTimeV2>(&column) != nullptr) {
+ return read_v2_int96_datetime(column, source, context, num_values,
state);
+ }
+ return serde.read_column_from_parquet(column, source, context, num_values,
state);
+}
+
+Status translate_value_encoding(tparquet::Encoding::type encoding,
+ ParquetValueEncoding* translated) {
+ DORIS_CHECK(translated != nullptr);
+ switch (encoding) {
+ case tparquet::Encoding::PLAIN:
+ *translated = ParquetValueEncoding::PLAIN;
+ return Status::OK();
+ case tparquet::Encoding::RLE_DICTIONARY:
+ case tparquet::Encoding::PLAIN_DICTIONARY:
+ *translated = ParquetValueEncoding::DICTIONARY;
+ return Status::OK();
+ case tparquet::Encoding::RLE:
+ *translated = ParquetValueEncoding::RLE;
+ return Status::OK();
+ case tparquet::Encoding::BIT_PACKED:
+ *translated = ParquetValueEncoding::BIT_PACKED;
+ return Status::OK();
+ case tparquet::Encoding::DELTA_BINARY_PACKED:
+ *translated = ParquetValueEncoding::DELTA_BINARY_PACKED;
+ return Status::OK();
+ case tparquet::Encoding::DELTA_LENGTH_BYTE_ARRAY:
+ *translated = ParquetValueEncoding::DELTA_LENGTH_BYTE_ARRAY;
+ return Status::OK();
+ case tparquet::Encoding::DELTA_BYTE_ARRAY:
+ *translated = ParquetValueEncoding::DELTA_BYTE_ARRAY;
+ return Status::OK();
+ case tparquet::Encoding::BYTE_STREAM_SPLIT:
+ *translated = ParquetValueEncoding::BYTE_STREAM_SPLIT;
+ return Status::OK();
+ default:
+ return Status::NotSupported("Unsupported Parquet encoding {}",
+ tparquet::to_string(encoding));
+ }
+}
+
+template <bool HAS_FILTER>
+Status decode_selected_values(IColumn& column, const DataTypeSerDe& serde,
Decoder& decoder,
+ const ParquetDecodeContext& context,
+ ParquetMaterializationState& state,
ColumnSelectVector& select_vector,
+ int64_t* materialization_time) {
+ SCOPED_RAW_TIMER(materialization_time);
+ ColumnSelectVector::DataReadType read_type;
+ while (const size_t run_length =
select_vector.get_next_run<HAS_FILTER>(&read_type)) {
+ switch (read_type) {
+ case ColumnSelectVector::CONTENT:
+ RETURN_IF_ERROR(
+ read_native_or_serde(column, serde, decoder, context,
run_length, state));
+ break;
+ case ColumnSelectVector::NULL_DATA:
+ column.insert_many_defaults(run_length);
+ break;
+ case ColumnSelectVector::FILTERED_CONTENT:
+ RETURN_IF_ERROR(decoder.skip_values(run_length));
+ break;
+ case ColumnSelectVector::FILTERED_NULL:
+ break;
+ }
+ }
+ return Status::OK();
+}
+
+// Presents one sparse page request as an ordinary sequential source to
DataTypeSerDe. SerDe is
+// entered once per page fragment; the concrete decoder decides whether to
gather selected spans,
+// batch-decode and compact, or use the cursor-preserving range fallback.
+class SelectedDecodeSource final : public ParquetDecodeSource {
+public:
+ SelectedDecodeSource(Decoder& decoder, const ParquetSelection& selection)
+ : _decoder(decoder), _selection(selection) {}
+
+ Status decode_fixed_values(size_t num_values, ParquetFixedValueConsumer&
consumer) override {
+ DORIS_CHECK_EQ(num_values, _selection.selected_values);
+ return _decoder.decode_selected_fixed_values(_selection, consumer);
+ }
+
+ Status decode_binary_values(size_t num_values, ParquetBinaryValueConsumer&
consumer) override {
+ DORIS_CHECK_EQ(num_values, _selection.selected_values);
+ return _decoder.decode_selected_binary_values(_selection, consumer);
+ }
+
+ Status skip_values(size_t num_values) override {
+ return Status::NotSupported("Selected Parquet source cannot be
skipped, values={}",
+ num_values);
+ }
+
+ bool has_dictionary() const override { return _decoder.has_dictionary(); }
+ uint64_t dictionary_generation() const override { return
_decoder.dictionary_generation(); }
+ size_t dictionary_size() const override { return
_decoder.dictionary_size(); }
+
+ Status decode_dictionary(ParquetFixedValueConsumer& fixed_consumer,
+ ParquetBinaryValueConsumer& binary_consumer)
override {
+ return _decoder.decode_dictionary(fixed_consumer, binary_consumer);
+ }
+
+ Status decode_dictionary_indices(size_t num_values, std::vector<uint32_t>*
indices) override {
+ DORIS_CHECK_EQ(num_values, _selection.selected_values);
+ return _decoder.decode_selected_dictionary_indices(_selection,
indices);
+ }
+
+ Status decode_dictionary_values(size_t num_values,
+ ParquetDictionaryValueConsumer& consumer)
override {
+ DORIS_CHECK_EQ(num_values, _selection.selected_values);
+ return _decoder.decode_selected_dictionary_values(_selection,
consumer);
+ }
+
+ bool prefer_dictionary_index_materialization(size_t dictionary_bytes)
const override {
+ // Avoid touching a dictionary larger than L2 for rows already removed
by sparse selection.
+ // Cache-resident or dense batches instead fuse RLE decode and gather,
eliminating the
+ // page-sized intermediate ID vector.
+ return _selection.selected_values < _selection.total_values &&
+ dictionary_bytes >
static_cast<size_t>(CpuInfo::get_l2_cache_size());
+ }
+
+private:
+ Decoder& _decoder;
+ const ParquetSelection& _selection;
+};
+
+Status decode_selected_non_null_values(IColumn& column, const DataTypeSerDe&
serde,
+ Decoder& decoder, const
ParquetDecodeContext& context,
+ ParquetMaterializationState& state,
+ ColumnSelectVector& select_vector,
+ int64_t* materialization_time) {
+ auto& selection = state.selection;
+ selection.ranges.clear();
+ selection.total_values = select_vector.num_values();
+ selection.selected_values = 0;
+
+ size_t cursor = 0;
+ ColumnSelectVector::DataReadType read_type;
+ while (const size_t run_length =
select_vector.get_next_run<true>(&read_type)) {
+ DORIS_CHECK(read_type == ColumnSelectVector::CONTENT ||
+ read_type == ColumnSelectVector::FILTERED_CONTENT);
+ if (read_type == ColumnSelectVector::CONTENT) {
+ selection.ranges.push_back({.first = cursor, .count = run_length});
+ selection.selected_values += run_length;
+ }
+ cursor += run_length;
+ }
+ DORIS_CHECK_EQ(cursor, selection.total_values);
+ if (selection.selected_values == 0) {
+ return decoder.skip_values(selection.total_values);
+ }
+
+ SCOPED_RAW_TIMER(materialization_time);
+ SelectedDecodeSource selected_source(decoder, selection);
+ return read_native_or_serde(column, serde, selected_source, context,
selection.selected_values,
+ state);
+}
+
+template <typename Visitor>
+bool visit_nullable_expandable_column(IColumn& column, Visitor&& visitor) {
+#define VISIT_COLUMN(TYPE) \
+ if (auto* typed = check_and_get_column<TYPE>(&column)) { \
+ visitor(*typed); \
+ return true; \
+ }
+ VISIT_COLUMN(ColumnUInt8)
+ VISIT_COLUMN(ColumnInt8)
+ VISIT_COLUMN(ColumnInt16)
+ VISIT_COLUMN(ColumnInt32)
+ VISIT_COLUMN(ColumnInt64)
+ VISIT_COLUMN(ColumnInt128)
+ VISIT_COLUMN(ColumnDate)
+ VISIT_COLUMN(ColumnDateTime)
+ VISIT_COLUMN(ColumnDateV2)
+ VISIT_COLUMN(ColumnDateTimeV2)
+ VISIT_COLUMN(ColumnFloat32)
+ VISIT_COLUMN(ColumnFloat64)
+ VISIT_COLUMN(ColumnIPv4)
+ VISIT_COLUMN(ColumnIPv6)
+ VISIT_COLUMN(ColumnTimeV2)
+ VISIT_COLUMN(ColumnTimeStampTz)
+ VISIT_COLUMN(ColumnOffset32)
+ VISIT_COLUMN(ColumnOffset64)
+ VISIT_COLUMN(ColumnDecimal32)
+ VISIT_COLUMN(ColumnDecimal64)
+ VISIT_COLUMN(ColumnDecimal128V2)
+ VISIT_COLUMN(ColumnDecimal128V3)
+ VISIT_COLUMN(ColumnDecimal256)
+ VISIT_COLUMN(ColumnString)
+ VISIT_COLUMN(ColumnString64)
+ VISIT_COLUMN(ColumnVarbinary)
+ VISIT_COLUMN(ColumnDictI32)
+#undef VISIT_COLUMN
+ return false;
+}
+
+template <typename ColumnType>
+void expand_nullable_pod_values(ColumnType& column, size_t old_size, size_t
compact_values,
+ const NullMap& selected_nulls) {
+ auto& data = column.get_data();
+ DORIS_CHECK_EQ(data.size(), old_size + compact_values);
+ data.resize(old_size + selected_nulls.size());
+ size_t source = compact_values;
+ for (size_t output = selected_nulls.size(); output > 0;) {
+ --output;
+ if (selected_nulls[output] != 0) {
+ data[old_size + output] = typename ColumnType::value_type {};
+ } else {
+ DORIS_CHECK(source > 0);
+ --source;
+ data[old_size + output] = std::move(data[old_size + source]);
+ }
+ }
+ DORIS_CHECK_EQ(source, 0);
+}
+
+template <typename Offset>
+void expand_nullable_string_values(ColumnStr<Offset>& column, size_t old_size,
+ size_t compact_values, const NullMap&
selected_nulls) {
+ auto& offsets = column.get_offsets();
+ DORIS_CHECK_EQ(offsets.size(), old_size + compact_values);
+ const Offset prefix_end = old_size == 0 ? 0 : offsets[old_size - 1];
+ offsets.resize(old_size + selected_nulls.size());
+ size_t source = compact_values;
+ for (size_t output = selected_nulls.size(); output > 0;) {
+ --output;
+ if (selected_nulls[output] == 0) {
+ DORIS_CHECK(source > 0);
+ --source;
+ offsets[old_size + output] = offsets[old_size + source];
+ } else {
+ offsets[old_size + output] = source == 0 ? prefix_end :
offsets[old_size + source - 1];
+ }
+ }
+ DORIS_CHECK_EQ(source, 0);
+}
+
+template <typename ColumnType>
+void expand_nullable_values(ColumnType& column, size_t old_size, size_t
compact_values,
+ const NullMap& selected_nulls) {
+ expand_nullable_pod_values(column, old_size, compact_values,
selected_nulls);
+}
+
+template <typename Offset>
+void expand_nullable_values(ColumnStr<Offset>& column, size_t old_size, size_t
compact_values,
+ const NullMap& selected_nulls) {
+ expand_nullable_string_values(column, old_size, compact_values,
selected_nulls);
+}
+
+void remap_nullable_conversion_failures(IColumn::Filter*
conversion_failure_null_map,
+ size_t old_size, size_t compact_values,
+ const NullMap& selected_nulls) {
+ if (conversion_failure_null_map == nullptr) {
+ return;
+ }
+ DORIS_CHECK(conversion_failure_null_map->size() >= old_size +
selected_nulls.size());
+ size_t source = compact_values;
+ // Walk backwards so writing an expanded row cannot overwrite an unread
compact failure bit.
+ for (size_t output = selected_nulls.size(); output > 0;) {
+ --output;
+ if (selected_nulls[output] != 0) {
+ (*conversion_failure_null_map)[old_size + output] = 1;
+ } else {
+ DORIS_CHECK(source > 0);
+ --source;
+ (*conversion_failure_null_map)[old_size + output] =
+ (*conversion_failure_null_map)[old_size + source];
+ }
+ }
+ DORIS_CHECK_EQ(source, 0);
+}
+
+Status decode_selected_nullable_values(IColumn& column, const DataTypeSerDe&
serde,
+ Decoder& decoder, const
ParquetDecodeContext& context,
+ ParquetMaterializationState& state,
+ ColumnSelectVector& select_vector,
NullMap& selected_nulls,
+ int64_t* materialization_time) {
+ auto& selection = state.selection;
+ selection.ranges.clear();
+ selection.total_values = 0;
+ selection.selected_values = 0;
+ selected_nulls.clear();
+ selected_nulls.reserve(select_vector.num_values() -
select_vector.num_filtered());
+
+ size_t physical_cursor = 0;
+ ColumnSelectVector::DataReadType read_type;
+ while (const size_t run_length =
select_vector.get_next_run<true>(&read_type)) {
+ switch (read_type) {
+ case ColumnSelectVector::CONTENT:
+ if (!selection.ranges.empty() &&
+ selection.ranges.back().first + selection.ranges.back().count
== physical_cursor) {
+ selection.ranges.back().count += run_length;
+ } else {
+ selection.ranges.push_back({.first = physical_cursor, .count =
run_length});
+ }
+ selection.selected_values += run_length;
+ selected_nulls.resize_fill(selected_nulls.size() + run_length, 0);
+ physical_cursor += run_length;
+ break;
+ case ColumnSelectVector::NULL_DATA:
+ selected_nulls.resize_fill(selected_nulls.size() + run_length, 1);
+ break;
+ case ColumnSelectVector::FILTERED_CONTENT:
+ physical_cursor += run_length;
+ break;
+ case ColumnSelectVector::FILTERED_NULL:
+ break;
+ }
+ }
+ selection.total_values = physical_cursor;
+ DORIS_CHECK_EQ(selection.total_values, select_vector.num_values() -
select_vector.num_nulls());
+ DORIS_CHECK_EQ(selected_nulls.size(),
+ select_vector.num_values() - select_vector.num_filtered());
+
+ const size_t old_size = column.size();
+ SCOPED_RAW_TIMER(materialization_time);
+ if (selection.selected_values == 0) {
+ RETURN_IF_ERROR(decoder.skip_values(selection.total_values));
+ column.insert_many_defaults(selected_nulls.size());
+ return Status::OK();
+ }
+
+ if (state.conversion_failure_null_map != nullptr) {
+ DORIS_CHECK(state.conversion_failure_null_map->size() >= old_size +
selected_nulls.size());
+ memset(state.conversion_failure_null_map->data() + old_size, 0,
selection.selected_values);
+ }
+ SelectedDecodeSource selected_source(decoder, selection);
+ const auto status = read_native_or_serde(column, serde, selected_source,
context,
+ selection.selected_values, state);
+ if (!status.ok()) {
+ if (state.conversion_failure_null_map != nullptr) {
+ memcpy(state.conversion_failure_null_map->data() + old_size,
selected_nulls.data(),
+ selected_nulls.size());
+ }
+ return status;
+ }
+ DORIS_CHECK_EQ(column.size(), old_size + selection.selected_values);
+
+ remap_nullable_conversion_failures(state.conversion_failure_null_map,
old_size,
+ selection.selected_values,
selected_nulls);
+ const bool expanded = visit_nullable_expandable_column(column, [&](auto&
typed_column) {
+ // Decode into the final nested column compactly, then expand in
place. This preserves the
+ // V2 no-intermediate-column invariant while restoring the nullable
sparse row layout.
+ expand_nullable_values(typed_column, old_size,
selection.selected_values, selected_nulls);
+ });
+ DORIS_CHECK(expanded);
+ return Status::OK();
+}
+
+class PlainPredicateConsumer final : public ParquetFixedValueConsumer {
+public:
+ PlainPredicateConsumer(const VExprSPtrs& conjuncts, DataTypePtr data_type,
int column_id,
+ IColumn::Filter* matches)
+ : _conjuncts(conjuncts),
+ _data_type(std::move(data_type)),
+ _column_id(column_id),
+ _matches(matches) {
+ DORIS_CHECK(_matches != nullptr);
+ }
+
+ Status consume(const uint8_t* values, size_t num_values, size_t
value_width) override {
+ const size_t old_size = _matches->size();
+ _matches->resize_fill(old_size + num_values, 1);
+ for (const auto& conjunct : _conjuncts) {
+ RETURN_IF_ERROR(conjunct->execute_on_raw_fixed_values(values,
num_values, value_width,
+ _data_type,
_column_id,
+
_matches->data() + old_size));
+ }
+ return Status::OK();
+ }
+
+private:
+ const VExprSPtrs& _conjuncts;
+ DataTypePtr _data_type;
+ int _column_id;
+ IColumn::Filter* _matches;
+};
+
+} // namespace
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::ColumnChunkReader(
+ io::BufferedStreamReader* reader, tparquet::ColumnChunk* column_chunk,
+ NativeFieldSchema* field_schema, const tparquet::OffsetIndex*
offset_index,
+ size_t total_rows, io::IOContext* io_ctx, const
ParquetPageReadContext& page_read_ctx,
+ const ColumnChunkRange* chunk_range)
+ : _field_schema(field_schema),
+ _max_rep_level(field_schema->repetition_level),
+ _max_def_level(field_schema->definition_level),
+ _stream_reader(reader),
+ _metadata(column_chunk->meta_data),
+ _offset_index(offset_index),
+ _total_rows(total_rows),
+ _io_ctx(io_ctx),
+ _page_read_ctx(page_read_ctx) {
+ if (chunk_range != nullptr) {
+ _chunk_range = *chunk_range;
+ _has_validated_chunk_range = true;
+ }
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::init() {
+ size_t start_offset = _has_validated_chunk_range
+ ? _chunk_range.offset
+ : (has_dict_page(_metadata) ?
_metadata.dictionary_page_offset
+ :
_metadata.data_page_offset);
+ size_t chunk_size =
+ _has_validated_chunk_range ? _chunk_range.length :
_metadata.total_compressed_size;
+ // create page reader
+ _page_reader = create_page_reader<IN_COLLECTION, OFFSET_INDEX>(
+ _stream_reader, _io_ctx, start_offset, chunk_size, _total_rows,
_metadata,
+ _page_read_ctx, _offset_index);
+ // get the block compression codec
+ RETURN_IF_ERROR(get_block_compression_codec(_metadata.codec,
&_block_compress_codec));
+ _state = INITIALIZED;
+ RETURN_IF_ERROR(_parse_first_page_header());
+ return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::skip_nested_values(
+ const std::vector<level_t>& def_levels, size_t start_index) {
+ size_t no_value_cnt = 0;
+ size_t value_cnt = 0;
+
+ DORIS_CHECK(start_index <= def_levels.size());
+ for (size_t idx = start_index; idx < def_levels.size(); idx++) {
+ level_t def_level = def_levels[idx];
+ if (IN_COLLECTION && def_level <
_field_schema->repeated_parent_def_level) {
+ no_value_cnt++;
+ } else if (def_level < _field_schema->definition_level) {
+ no_value_cnt++;
+ } else {
+ value_cnt++;
+ }
+ }
+
+ RETURN_IF_ERROR(skip_values(value_cnt, true));
+ RETURN_IF_ERROR(skip_values(no_value_cnt, false));
+ return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::read_levels(
+ size_t num_values, std::vector<level_t>* rep_levels,
std::vector<level_t>* def_levels) {
+ DORIS_CHECK(rep_levels != nullptr);
+ DORIS_CHECK(def_levels != nullptr);
+ if (_remaining_num_values < num_values || _remaining_rep_nums < num_values
||
+ _remaining_def_nums < num_values) {
+ return Status::Corruption(
+ "Parquet level reader requested {} slots with only {}/{}/{}
remaining", num_values,
+ _remaining_num_values, _remaining_rep_nums,
_remaining_def_nums);
+ }
+
+ const size_t start_index = def_levels->size();
+ rep_levels->resize(rep_levels->size() + num_values, 0);
+ def_levels->resize(def_levels->size() + num_values, 0);
+ if (_max_rep_level > 0) {
+ const size_t decoded = _rep_level_decoder.get_levels(
+ rep_levels->data() + rep_levels->size() - num_values,
num_values);
+ if (decoded != num_values) {
+ return Status::Corruption("Parquet repetition level stream ended
after {} of {} slots",
+ decoded, num_values);
+ }
+ }
+ if (_max_def_level > 0) {
+ const size_t decoded = _def_level_decoder.get_levels(
+ def_levels->data() + def_levels->size() - num_values,
num_values);
+ if (decoded != num_values) {
+ return Status::Corruption("Parquet definition level stream ended
after {} of {} slots",
+ decoded, num_values);
+ }
+ }
+ _remaining_rep_nums -= num_values;
+ _remaining_def_nums -= num_values;
+ return skip_nested_values(*def_levels, start_index);
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::_parse_first_page_header() {
+ while (true) {
+ RETURN_IF_ERROR(_page_reader->parse_page_header());
+ const tparquet::PageHeader* header = nullptr;
+ RETURN_IF_ERROR(_page_reader->get_page_header(&header));
+ if (header->type == tparquet::PageType::DATA_PAGE ||
+ header->type == tparquet::PageType::DATA_PAGE_V2) {
+ _state = INITIALIZED;
+ return parse_page_header();
+ }
+ if (header->type != tparquet::PageType::DICTIONARY_PAGE) {
+ RETURN_IF_ERROR(_page_reader->skip_auxiliary_page());
+ _state = INITIALIZED;
+ continue;
+ }
+ // the first page maybe directory page even if
_metadata.__isset.dictionary_page_offset == false,
+ // so we should parse the directory page in next_page()
+ RETURN_IF_ERROR(_decode_dict_page());
+ // parse the real first data page
+ RETURN_IF_ERROR(_page_reader->dict_next_page());
+ _state = INITIALIZED;
+ // A dictionary is the only non-data page with decoder state. Any
following index or
+ // extension pages are skipped by the same pre-data loop.
+ }
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::parse_page_header() {
+ if (_state == HEADER_PARSED || _state == DATA_LOADED) {
+ return Status::OK();
+ }
+ const tparquet::PageHeader* header = nullptr;
+ while (true) {
+ RETURN_IF_ERROR(_page_reader->parse_page_header());
+ RETURN_IF_ERROR(_page_reader->get_page_header(&header));
+ if (header->type == tparquet::PageType::DATA_PAGE ||
+ header->type == tparquet::PageType::DATA_PAGE_V2) {
+ break;
+ }
+ if (header->type == tparquet::PageType::DICTIONARY_PAGE) {
+ return Status::Corruption("Parquet dictionary page appears after
data pages");
+ }
+ RETURN_IF_ERROR(_page_reader->skip_auxiliary_page());
+ }
+ int32_t page_num_values = _page_reader->is_header_v2() ?
header->data_page_header_v2.num_values
+ :
header->data_page_header.num_values;
+ if (page_num_values < 0 || page_num_values > _metadata.num_values ||
+ (!OFFSET_INDEX &&
+ static_cast<uint64_t>(page_num_values) >
+ static_cast<uint64_t>(_metadata.num_values) -
_chunk_parsed_values)) {
+ // Page counts are untrusted and feed both level decoders and scratch
sizing. Bound each
+ // page by the column metadata before converting to unsigned counters.
+ return Status::Corruption("Parquet data page value count {} exceeds
column total {}",
+ page_num_values, _metadata.num_values);
+ }
+ if constexpr (!IN_COLLECTION) {
Review Comment:
[P1] Reconcile indexed row spans for repeated pages too. OffsetIndexes are
transferred to projected nested leaves, but this check is flat-only. With exact
physical page rectangles and forged `first_row_index` values 0 and 120 for two
V2 pages that each contain 100 row starts, a sparse selection at row 120 seeks
directly to page 2 and returns its actual row 100; reading one row returns
before the exhaustion check can detect the mismatch. At least require V2
`num_rows` to match the indexed span for repeated pages, and use a conservative
sequential path for V1 unless repetition-level row starts can be validated.
##########
be/src/format_v2/parquet/parquet_statistics.cpp:
##########
@@ -1388,127 +1052,150 @@ 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) ||
Review Comment:
[P1] Reject NULL results from non-NULL statistic bounds. This generic decode
uses the nullable file type while leaving decoded conversion non-strict. For an
OPTIONAL DECIMAL bound outside its declared precision, the nested SerDe marks
the output NULL and returns OK; NULL/NULL then looks ordered, sets
`has_min_max`, and later hits the ZoneMap TYPE_NULL-versus-DECIMAL
`DORIS_CHECK`. Make metadata-bound conversion strict, or explicitly reject a
NULL `Field` here, so malformed optional statistics fall back conservatively.
##########
be/src/core/data_type_serde/data_type_number_serde.cpp:
##########
@@ -179,6 +215,212 @@ Status read_integer_decoded_values(IColumn& column, const
DecodedColumnView& vie
}
}
+template <typename DorisCppType, typename SourceType>
+Status append_parquet_number(PaddedPODArray<DorisCppType>& data, const
uint8_t* values,
+ size_t num_values, const ParquetDecodeContext&
context,
+ ParquetMaterializationState* state) {
+ const size_t old_size = data.size();
+ data.resize(old_size + num_values);
+ if constexpr (std::is_same_v<DorisCppType, SourceType>) {
+ // Identical fixed-width physical/logical types need no validation or
conversion. Parquet
+ // PLAIN values and Doris POD columns share the byte representation on
supported targets,
+ // so preserve one dense vector-at-a-time memcpy instead of converting
value by value.
+ memcpy(data.data() + old_size, values, num_values *
sizeof(SourceType));
+ return Status::OK();
+ }
+ if constexpr (parquet_number_conversion_always_fits<DorisCppType,
SourceType>()) {
+ // A widening conversion cannot fail, so keeping range checks in the
row loop only blocks
+ // auto-vectorization. Input can be unaligned at a Parquet page
boundary; load explicitly.
+ for (size_t row = 0; row < num_values; ++row) {
+ data[old_size + row] = static_cast<DorisCppType>(
+ unaligned_load<SourceType>(values + row *
sizeof(SourceType)));
+ }
+ return Status::OK();
+ }
+ for (size_t row = 0; row < num_values; ++row) {
+ const auto value = unaligned_load<SourceType>(values + row *
sizeof(SourceType));
+ if (!decoded_number_value_fits<DorisCppType>(value)) {
+ if (state != nullptr &&
state->can_insert_null_on_conversion_failure()) {
+ data[old_size + row] = DorisCppType();
+ DORIS_CHECK(state->mark_conversion_failure(old_size + row));
+ continue;
+ }
+ data.resize(old_size);
+ return Status::DataQualityError("Parquet value is out of range at
row {}", row);
+ }
+ data[old_size + row] = static_cast<DorisCppType>(value);
+ }
+ return Status::OK();
+}
+
+template <typename DorisCppType, typename SourceType, typename LogicalType>
+Status append_parquet_logical_integers(PaddedPODArray<DorisCppType>& data,
const uint8_t* values,
+ size_t num_values,
ParquetMaterializationState* state) {
+ const size_t old_size = data.size();
+ data.resize(old_size + num_values);
+ if constexpr (parquet_number_conversion_always_fits<DorisCppType,
LogicalType>()) {
+ for (size_t row = 0; row < num_values; ++row) {
+ const auto physical_value =
+ unaligned_load<SourceType>(values + row *
sizeof(SourceType));
+ data[old_size + row] =
Review Comment:
[P1] Validate sub-word logical carriers before narrowing them. Parquet
`INT(8,true)` is physically INT32, but a malformed carrier such as 1000 is cast
to Int8 here and silently becomes -24; this always-fits branch returns it
directly, and the fallback below also checks only the already-narrowed value.
Signed 16-bit and unsigned 8/16 carriers wrap similarly across plain, selected,
and dictionary reads, while the metadata helper repeats the pattern and can
publish wrong pruning bounds. Check the physical carrier against the declared
logical domain before casting, then route failures through the existing
strict/non-strict state.
##########
be/src/format_v2/parquet/parquet_statistics.cpp:
##########
@@ -1388,127 +1052,150 @@ 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,
+ int64_t page_rows, 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];
+ const bool all_null = column_index.null_pages[page_idx];
+ if (page_rows < 0 || null_count < 0 || null_count > page_rows ||
+ all_null != (null_count == page_rows)) {
+ // The caller supplies the exact flat page or row-group span.
Contradictory optional null
+ // metadata must disable pruning instead of turning a partial span
into an all-null proof.
return false;
}
-
- const auto page_count = offset_index->page_locations().size();
- page_skip_plan->leaf_column_id = column_schema.leaf_column_id;
- page_skip_plan->skipped_pages.resize(page_count);
- page_skip_plan->skipped_page_compressed_sizes.resize(page_count);
- const auto& page_locations = offset_index->page_locations();
- for (size_t page_idx = 0; page_idx < page_count; ++page_idx) {
- const RowRange row_range = page_row_range(*offset_index, page_idx,
row_group_rows);
- if (row_range.length == 0 || ranges_intersect(selected_ranges,
row_range)) {
- continue;
+ page_statistics->has_null_count = true;
+ page_statistics->has_null = null_count > 0;
+ page_statistics->has_not_null = !all_null;
+ if (!page_statistics->has_not_null) {
+ return true;
+ }
+ switch (column_schema.type_descriptor.physical_type) {
+ case tparquet::Type::BOOLEAN:
+ return set_native_page_scalar_min_max<uint8_t>(column_index,
column_schema, page_idx,
+ DecodedValueKind::BOOL,
page_statistics,
+ timezone);
+ case tparquet::Type::INT32:
+ return set_native_page_scalar_min_max<int32_t>(
Review Comment:
[P1] Reject out-of-range DATE bounds before publishing this ZoneMap. This
new generic path routes DATE metadata through
`DataTypeDateV2SerDe::read_column_from_decoded_values`, which adds the epoch
offset in `int32_t`, ignores a failed `get_date_from_daynr()`, and appends the
default zero date. Two out-of-range bounds can therefore return OK, look
ordered, and prune valid DATE rows as false negatives. Reuse the checked
`int64_t` conversion used by `DateV2ParquetConsumer` so invalid optional
footer/page bounds make the statistics unusable.
--
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]
