github-actions[bot] commented on code in PR #65935:
URL: https://github.com/apache/doris/pull/65935#discussion_r3635392756
##########
be/src/format_v2/column_mapper.cpp:
##########
@@ -875,7 +1162,7 @@ static VExprSPtr rewrite_table_expr_to_file_expr(
}
DORIS_CHECK(rewrite_context != nullptr);
DORIS_CHECK(can_localize != nullptr);
- if (auto* runtime_filter =
dynamic_cast<VRuntimeFilterWrapper*>(expr.get());
+ if (auto* runtime_filter = dynamic_cast<RuntimeFilterExpr*>(expr.get());
Review Comment:
[P1] Keep localizing the branch-4.1 runtime-filter wrapper
Production `RuntimeFilterConsumer` in this branch still builds
`VRuntimeFilterWrapper`, not `RuntimeFilterExpr`, so this special case misses
every real runtime filter. Generic recursion cannot repair it: `children()`
delegates to `_impl`, while inherited `set_children()` writes the wrapper's
unused base child vector, leaving the executed impl with table-global slot
refs. Once a global slot maps to a different file-local position (for example
because a partition or missing column precedes it), the file-local filter reads
the wrong/out-of-range column and can discard matching rows before residual
evaluation. Please keep the `VRuntimeFilterWrapper` branch here, or handle both
wrapper types through the wrapped impl.
##########
be/src/format_v2/table_reader.cpp:
##########
@@ -875,12 +1051,18 @@ Status TableReader::prepare_split(const
SplitReadOptions& options) {
// active and no predicate can arrive later. The metadata path can return
several batches for
// one split; after its first synthetic batch there is no way to recover
the real rows if a
// runtime filter arrives before the next scheduler turn.
- if (_push_down_agg_type == TPushAggOp::type::COUNT &&
options.all_runtime_filters_applied &&
+ // Table-level metadata only contains the number of rows; it cannot
evaluate an expression or
+ // the NULL state of a COUNT argument. Require the new FE's explicit empty
argument list, which
+ // means COUNT(*)/COUNT(1). A non-empty list means COUNT(col), while
nullopt comes from an old FE
+ // whose COUNT semantics are unknown during a BE-first rolling upgrade.
+ if (_push_down_agg_type == TPushAggOp::type::COUNT &&
_push_down_count_columns.has_value() &&
Review Comment:
[P1] Preserve old-FE table counts after split reduction
An old FE has already reduced this plan before the new argument field
exists: Iceberg retains only one/few representative tasks and Paimon similarly
sublists its count splits, then both attach the full total as
`table_level_row_count`. During a supported BE-first upgrade,
`_push_down_count_columns` is therefore `nullopt`; this condition ignores that
total and physically scans only the retained files, so ordinary `COUNT(*)`
undercounts. V1 has the same regression because its effective aggregate becomes
`NONE`. The absent field should remain conservative for ordinary per-file
plans, but a nonnegative table-level count cannot fall back after split
reduction; preserve the legacy shortcut (or reject the plan) on both scanner
paths.
##########
be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp:
##########
@@ -0,0 +1,1785 @@
+// 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.dictionary_index_only) {
+ if (context.encoding != ParquetValueEncoding::DICTIONARY) {
+ return Status::IOError("Dictionary filter requested for a
non-dictionary page");
+ }
+ auto* output = check_and_get_column<ColumnInt32>(&column);
+ if (output == nullptr) {
+ return Status::InternalError("Dictionary indices require an INT32
output column");
+ }
+ // Dictionary IDs have the same RLE/bit-packed representation for
every physical type.
+ // Decode them before dispatching to a typed SerDe; otherwise a
fixed-width SerDe treats
+ // the IDs as values and the row filter indexes its bitmap with
materialized data.
+ RETURN_IF_ERROR(source.decode_dictionary_indices(num_values,
&state.dictionary_indices));
+ DORIS_CHECK_EQ(state.dictionary_indices.size(), num_values);
+ const size_t old_size = output->size();
+ auto& indices = output->get_data();
+ indices.resize(old_size + num_values);
+ for (size_t row = 0; row < num_values; ++row) {
+ if (UNLIKELY(state.dictionary_indices[row] >
+
static_cast<uint32_t>(std::numeric_limits<int32_t>::max()))) {
+ indices.resize(old_size);
+ return Status::Corruption("Parquet dictionary index {} exceeds
INT32",
+ state.dictionary_indices[row]);
+ }
+ indices[old_size + row] =
static_cast<int32_t>(state.dictionary_indices[row]);
+ }
+ return Status::OK();
+ }
+ 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 constexpr (IN_COLLECTION && OFFSET_INDEX) {
+ if (!_page_reader->is_header_v2() &&
_page_reader->has_active_offset_index()) {
+ // V1 nested pages do not declare their logical row count. An
OffsetIndex span cannot
+ // be trusted until repetition levels are decoded, so keep the
sequential cursor path.
+ _page_reader->discard_offset_index();
+ _offset_index = nullptr;
+ }
+ }
+ const bool active_offset_index = _page_reader->has_active_offset_index();
+ if (page_num_values < 0 || page_num_values > _metadata.num_values ||
+ (!active_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) {
+ const size_t page_start_row = _page_reader->start_row();
+ const size_t page_end_row = _page_reader->end_row();
+ if (UNLIKELY(page_end_row < page_start_row ||
+ static_cast<size_t>(page_num_values) != page_end_row -
page_start_row)) {
+ // Flat columns have exactly one physical value slot per logical
row. Rejecting a
+ // divergent header/OffsetIndex span prevents every later page
from shifting rows.
+ return Status::Corruption(
+ "Parquet flat data page has {} values for logical row
range [{}, {})",
+ page_num_values, page_start_row, page_end_row);
+ }
+ } else if (_page_reader->is_header_v2() && active_offset_index) {
+ const size_t page_start_row = _page_reader->start_row();
+ const size_t page_end_row = _page_reader->end_row();
+ if (UNLIKELY(page_end_row < page_start_row ||
+ static_cast<size_t>(header->data_page_header_v2.num_rows)
!=
+ page_end_row - page_start_row)) {
+ // V2 is the only repeated-page format that states its logical row
count in the page
+ // header, so it must agree with the optional index before indexed
seeking is allowed.
+ return Status::Corruption(
+ "Parquet nested data page has {} rows for indexed row
range [{}, {})",
+ header->data_page_header_v2.num_rows, page_start_row,
page_end_row);
+ }
+ }
+ _remaining_rep_nums = page_num_values;
+ _remaining_def_nums = page_num_values;
+ _remaining_num_values = page_num_values;
+
+ // no offset will parse all header.
+ if (!active_offset_index) {
+ _chunk_parsed_values += _remaining_num_values;
+ }
+ _state = HEADER_PARSED;
+ return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::next_page() {
+ _state = INITIALIZED;
+ RETURN_IF_ERROR(_page_reader->next_page());
+ return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::_get_uncompressed_levels(
+ const tparquet::DataPageHeaderV2& page_v2, Slice& page_data) {
+ const size_t rl = page_v2.repetition_levels_byte_length;
+ const size_t dl = page_v2.definition_levels_byte_length;
+ if (UNLIKELY(rl > page_data.size || dl > page_data.size - rl)) {
+ // Validate the physical slice again because a cached entry may itself
be truncated.
+ return Status::Corruption("Parquet data page v2 level bytes exceed
available payload");
+ }
+ _v2_rep_levels = Slice(page_data.data, rl);
+ _v2_def_levels = Slice(page_data.data + rl, dl);
+ page_data.data += dl + rl;
+ page_data.size -= dl + rl;
+ return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::load_page_data() {
+ if (_state == DATA_LOADED) {
+ return Status::OK();
+ }
+ if (UNLIKELY(_state != HEADER_PARSED)) {
+ return Status::Corruption("Should parse page header");
+ }
+
+ const tparquet::PageHeader* header = nullptr;
+ RETURN_IF_ERROR(_page_reader->get_page_header(&header));
+ RETURN_IF_ERROR(validate_uncompressed_page_sizes(
+ *header, _metadata.codec,
_page_read_ctx.data_page_v2_always_compressed));
+ int32_t uncompressed_size = header->uncompressed_page_size;
+ bool page_loaded = false;
+
+ // First, try to reuse a cache handle previously discovered by PageReader
+ // (header-only lookup) to avoid a second lookup here.
+ if (_page_read_ctx.enable_parquet_file_page_cache &&
!config::disable_storage_page_cache &&
+ StoragePageCache::instance() != nullptr) {
+ if (_page_reader->has_page_cache_handle()) {
+ const PageCacheHandle& handle = _page_reader->page_cache_handle();
+ Slice cached = handle.data();
+ size_t header_size = _page_reader->header_bytes().size();
+ size_t levels_size = 0;
+ if (header->__isset.data_page_header_v2) {
+ const tparquet::DataPageHeaderV2& header_v2 =
header->data_page_header_v2;
+ size_t rl = header_v2.repetition_levels_byte_length;
+ size_t dl = header_v2.definition_levels_byte_length;
+ levels_size = rl + dl;
+ if (UNLIKELY(header_size > cached.size ||
+ levels_size > cached.size - header_size)) {
+ return Status::Corruption("Cached Parquet page is shorter
than its v2 levels");
+ }
+ _v2_rep_levels =
+ Slice(reinterpret_cast<const uint8_t*>(cached.data) +
header_size, rl);
+ _v2_def_levels =
+ Slice(reinterpret_cast<const uint8_t*>(cached.data) +
header_size + rl, dl);
+ }
+ // payload_slice points to the bytes after header and levels
+ if (UNLIKELY(header_size + levels_size > cached.size)) {
+ return Status::Corruption("Cached Parquet page is shorter than
its header");
+ }
+ Slice payload_slice(cached.data + header_size + levels_size,
+ cached.size - header_size - levels_size);
+
+ bool cache_payload_is_decompressed =
_page_reader->is_cache_payload_decompressed();
+ const size_t expected_payload_size =
+ cache_payload_is_decompressed
+ ?
static_cast<size_t>(header->uncompressed_page_size) - levels_size
+ :
static_cast<size_t>(header->compressed_page_size) - levels_size;
+ if (UNLIKELY(payload_slice.size != expected_payload_size)) {
+ return Status::Corruption("Cached Parquet page payload has
size {}, expected {}",
+ payload_slice.size,
expected_payload_size);
+ }
+
+ if (cache_payload_is_decompressed) {
+ // Cached payload is already uncompressed
+ _page_data = payload_slice;
+ } else {
+ CHECK(_block_compress_codec);
+ // Decompress cached payload into _decompress_buf for decoding
+ size_t uncompressed_payload_size =
+ header->__isset.data_page_header_v2
+ ?
static_cast<size_t>(header->uncompressed_page_size) - levels_size
+ :
static_cast<size_t>(header->uncompressed_page_size);
+ _reserve_decompress_buf(uncompressed_payload_size);
+ _page_data = Slice(_decompress_buf.get(),
uncompressed_payload_size);
+ SCOPED_RAW_TIMER(&_chunk_statistics.decompress_time);
+ _chunk_statistics.decompress_cnt++;
+
RETURN_IF_ERROR(_block_compress_codec->decompress(payload_slice, &_page_data));
+ if (UNLIKELY(_page_data.size != uncompressed_payload_size)) {
+ return Status::Corruption("Parquet page decompressed to {}
bytes, expected {}",
+ _page_data.size,
uncompressed_payload_size);
+ }
+ }
+ // page cache counters were incremented when PageReader did the
header-only
+ // cache lookup. Do not increment again to avoid double-counting.
+ page_loaded = true;
+ }
+ }
+
+ if (!page_loaded) {
+ const bool prepare_cache_payload = can_prepare_page_cache_payload(
+ _page_read_ctx.enable_parquet_file_page_cache,
config::disable_storage_page_cache,
+ StoragePageCache::instance() != nullptr,
!_page_reader->header_bytes().empty());
+ if (_block_compress_codec != nullptr) {
+ Slice compressed_data;
+ RETURN_IF_ERROR(_page_reader->get_page_data(compressed_data));
+ std::vector<uint8_t> level_bytes;
+ if (header->__isset.data_page_header_v2) {
+ const tparquet::DataPageHeaderV2& header_v2 =
header->data_page_header_v2;
+ // uncompressed_size = rl + dl + uncompressed_data_size
+ // compressed_size = rl + dl + compressed_data_size
+ uncompressed_size -= header_v2.repetition_levels_byte_length +
+ header_v2.definition_levels_byte_length;
+ // copy level bytes (rl + dl) so that we can cache header +
levels + uncompressed payload
+ size_t rl = header_v2.repetition_levels_byte_length;
+ size_t dl = header_v2.definition_levels_byte_length;
+ size_t level_sz = rl + dl;
+ if (prepare_cache_payload && level_sz > 0) {
+ level_bytes.resize(level_sz);
+ memcpy(level_bytes.data(), compressed_data.data, level_sz);
+ }
+ // now remove levels from compressed_data for decompression
+ RETURN_IF_ERROR(_get_uncompressed_levels(header_v2,
compressed_data));
+ }
+ bool is_v2_compressed = header->__isset.data_page_header_v2 &&
+ (header->data_page_header_v2.is_compressed
||
+
_page_read_ctx.data_page_v2_always_compressed);
+ bool page_has_compression = header->__isset.data_page_header ||
is_v2_compressed;
+
+ if (page_has_compression) {
+ // Decompress payload for immediate decoding
+ _reserve_decompress_buf(uncompressed_size);
Review Comment:
[P1] Validate the decompressed size before allocating it
`uncompressed_page_size` is file-controlled, but this reserves it before the
codec inspects the bounded compressed payload. For example, a tiny Snappy page
can advertise `INT32_MAX`; `_reserve_decompress_buf()` then attempts a roughly
2 GiB tracked allocation (and the dictionary path does the same at line 1170)
before Snappy can report that its encoded output length does not match. A
single malformed external file can therefore consume the scan's memory budget
or force cancellation before the cheap corruption check. Please preflight
codecs that expose their output length and enforce a checked page-output limit
before allocating in both paths.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/AggregateStrategies.java:
##########
@@ -714,10 +737,21 @@ 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)) {
+ if (logicalScan instanceof LogicalFileScan && mergeOp ==
PushDownAggOp.COUNT
+ && containsCountStar && column.isAllowNull() &&
checkNullSlots.contains(slot)) {
+ // One metadata cardinality cannot represent both COUNT(*)
and the smaller
+ // COUNT(nullable_col); synthetic rows would make one
upper aggregate wrong.
+ return canNotPush;
+ }
+ // 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] Gate nullable COUNT on backend capability
`enableFileScannerV2` selects the scanner but does not prove that every
scheduled BE understands the new optional COUNT-argument field. During a smooth
upgrade, an old BE ignores field 55 but still sees the COUNT opcode; its
pre-patch v2 reader treats a nullable primitive column as COUNT(*), so `[1,
NULL, 2]` produces three synthetic rows and `COUNT(col)` returns 3 instead of
2. Old smooth-upgrade-source BEs remain eligible for external scans (this patch
adds a separate explicit guard for the new position-deletes path). Please
capability-gate this optimization or keep it disabled while any selected BE
lacks the argument contract.
##########
be/src/format/table/iceberg_position_delete_sys_table_reader.cpp:
##########
@@ -0,0 +1,614 @@
+// 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/table/iceberg_position_delete_sys_table_reader.h"
+
+#include <gen_cpp/ExternalTableSchema_types.h>
+#include <gen_cpp/PlanNodes_types.h>
+
+#include <algorithm>
+#include <memory>
+#include <utility>
+
+#include "common/cast_set.h"
+#include "core/block/block.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_string.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_string.h"
+#include "core/data_type_serde/data_type_serde.h"
+#include "core/types.h"
+#include "format/orc/vorc_reader.h"
+#include "format/parquet/schema_desc.h"
+#include "format/parquet/vparquet_reader.h"
+#include "format/table/parquet_utils.h"
+#include "format/table/table_format_reader.h"
+#include "runtime/runtime_state.h"
+
+namespace doris {
+
+namespace {
+
+constexpr const char* kFilePathColumn = "file_path";
+constexpr const char* kPosColumn = "pos";
+constexpr const char* kRowColumn = "row";
+constexpr const char* kPartitionColumn = "partition";
+constexpr const char* kSpecIdColumn = "spec_id";
+constexpr const char* kDeleteFilePathColumn = "delete_file_path";
+constexpr const char* kContentOffsetColumn = "content_offset";
+constexpr const char* kContentSizeInBytesColumn = "content_size_in_bytes";
+constexpr const char* kIcebergOrcAttribute = "iceberg.id";
+constexpr int kPositionDeleteContent = 1;
+
+bool block_has_row(const Block& block, size_t row) {
+ return block.columns() > 0 && row < block.rows();
+}
+
+void insert_int64_nullable(MutableColumnPtr& column, const int64_t* value) {
+ if (value == nullptr) {
+ parquet_utils::insert_null(column);
+ } else {
+ parquet_utils::insert_int64(column, *value);
+ }
+}
+
+// Fail loudly if the filled output block is malformed. Each output column
must be produced by a
+// file slot; a projected system-table column that is not backed by a file
slot would be skipped
+// during fill and left shorter than the rest, producing a block with
inconsistent column lengths.
+void check_output_columns_aligned(const MutableColumns& columns) {
+ if (columns.empty()) {
+ return;
+ }
+ const size_t expected_rows = columns.front()->size();
+ for (const auto& column : columns) {
+ DORIS_CHECK(column->size() == expected_rows)
+ << "Iceberg position delete system table output block has
inconsistent column "
+ "sizes; a projected column is not backed by a file slot";
+ }
+}
+
+const ColumnString* get_string_column(const Block& block, const std::string&
name) {
+ auto pos = block.get_position_by_name(name);
+ if (pos < 0) {
+ return nullptr;
+ }
+ return
check_and_get_column<ColumnString>(block.get_by_position(pos).column.get());
+}
+
+const ColumnInt64* get_int64_column(const Block& block, const std::string&
name) {
+ auto pos = block.get_position_by_name(name);
+ if (pos < 0) {
+ return nullptr;
+ }
+ return
check_and_get_column<ColumnInt64>(block.get_by_position(pos).column.get());
+}
+
+const schema::external::TField* get_field_ptr(const
schema::external::TFieldPtr& field_ptr) {
+ if (!field_ptr.__isset.field_ptr || field_ptr.field_ptr == nullptr) {
+ return nullptr;
+ }
+ return field_ptr.field_ptr.get();
+}
+
+const schema::external::TField* find_current_schema_field(const
TFileScanRangeParams* params,
+ const std::string&
name) {
+ if (params == nullptr || !params->__isset.history_schema_info ||
+ params->history_schema_info.empty()) {
+ return nullptr;
+ }
+ const schema::external::TSchema* schema =
¶ms->history_schema_info.front();
+ if (params->__isset.current_schema_id) {
+ for (const auto& candidate : params->history_schema_info) {
+ if (candidate.__isset.schema_id && candidate.schema_id ==
params->current_schema_id) {
+ schema = &candidate;
+ break;
+ }
+ }
+ }
+ if (!schema->__isset.root_field || !schema->root_field.__isset.fields) {
+ return nullptr;
+ }
+ for (const auto& field_ptr : schema->root_field.fields) {
+ const auto* field = get_field_ptr(field_ptr);
+ if (field != nullptr && field->__isset.name && field->name == name) {
+ return field;
+ }
+ }
+ return nullptr;
+}
+
+template <typename ReadColumns>
+std::shared_ptr<TableSchemaChangeHelper::StructNode>
create_position_delete_root_node(
+ const ReadColumns& read_columns) {
+ auto root_node = std::make_shared<TableSchemaChangeHelper::StructNode>();
+ for (const auto& column : read_columns) {
+ if (column.name == kRowColumn) {
+ continue;
+ }
+ root_node->add_children(column.name, column.name,
+
TableSchemaChangeHelper::ConstNode::get_instance());
+ }
+ return root_node;
+}
+
+} // namespace
+
+IcebergPositionDeleteSysTableReader::IcebergPositionDeleteSysTableReader(
+ const std::vector<SlotDescriptor*>& file_slot_descs, RuntimeState*
state,
+ RuntimeProfile* profile, const TFileRangeDesc& range,
+ const TFileScanRangeParams* range_params,
std::shared_ptr<io::IOContext> io_ctx,
+ FileMetaCache* meta_cache)
+ : _file_slot_descs(file_slot_descs),
+ _state(state),
+ _profile(profile),
+ _range(range),
+ _range_params(range_params),
+ _io_ctx(std::move(io_ctx)) {
+ _meta_cache = meta_cache;
+}
+
+IcebergPositionDeleteSysTableReader::~IcebergPositionDeleteSysTableReader() =
default;
+
+Status IcebergPositionDeleteSysTableReader::init_reader() {
+ if (_state == nullptr || _profile == nullptr || _range_params == nullptr ||
+ _io_ctx == nullptr) {
+ return Status::InvalidArgument(
+ "invalid Iceberg position delete system table reader context");
+ }
+ if (!_range.__isset.table_format_params ||
!_range.table_format_params.__isset.iceberg_params) {
+ return Status::InternalError("Iceberg position delete system table
range misses params");
+ }
+
+ _iceberg_file_desc = &_range.table_format_params.iceberg_params;
+ if (!_iceberg_file_desc->__isset.delete_files ||
_iceberg_file_desc->delete_files.size() != 1) {
+ return Status::InternalError(
+ "Iceberg position delete system table range should contain
exactly one delete "
+ "file");
+ }
+ _delete_file_desc = _iceberg_file_desc->delete_files.data();
+ if (is_iceberg_deletion_vector(*_delete_file_desc)) {
+ _delete_file_kind = DeleteFileKind::DELETION_VECTOR;
+ } else if (_delete_file_desc->__isset.content &&
+ _delete_file_desc->content == kPositionDeleteContent) {
+ _delete_file_kind = DeleteFileKind::POSITION_DELETE;
+ } else if (!_delete_file_desc->__isset.content) {
+ return Status::InternalError(
+ "Iceberg position delete system table delete file misses
content");
+ } else {
+ return Status::InternalError(
+ "Iceberg position delete system table does not support delete
file content {}",
+ _delete_file_desc->content);
+ }
+ _batch_size = _state->batch_size();
+
+ if (_delete_file_kind == DeleteFileKind::DELETION_VECTOR) {
+ return _init_deletion_vector_reader();
+ }
+ return _init_position_delete_reader();
+}
+
+Status IcebergPositionDeleteSysTableReader::get_columns(
+ std::unordered_map<std::string, DataTypePtr>* name_to_type,
+ std::unordered_set<std::string>* missing_cols) {
+ missing_cols->clear();
+ for (const auto* slot : _file_slot_descs) {
+ name_to_type->emplace(slot->col_name(), slot->get_data_type_ptr());
+ }
+ return Status::OK();
+}
+
+bool IcebergPositionDeleteSysTableReader::count_read_rows() {
+ return _delete_file_kind == DeleteFileKind::POSITION_DELETE;
+}
+
+void IcebergPositionDeleteSysTableReader::_collect_profile_before_close() {
+ if (_position_reader != nullptr) {
+ _position_reader->collect_profile_before_close();
+ }
+}
+
+Status IcebergPositionDeleteSysTableReader::close() {
+ if (_position_reader != nullptr) {
+ RETURN_IF_ERROR(_position_reader->close());
+ }
+ _partition_value.reset();
+ _next_dv_position.reset();
+ _dv_positions = roaring::Roaring64Map();
+ return Status::OK();
+}
+
+Status IcebergPositionDeleteSysTableReader::_init_position_delete_reader() {
+ if (!_delete_file_desc->__isset.file_format) {
+ return Status::InternalError("Iceberg position delete file misses file
format");
+ }
+
+ // `row` is optional in position delete files and expensive to read, so
only read it when the
+ // query actually projects it. Whether the delete file physically stores
`row` is decided from
+ // the reader's own footer/type below, reusing the same reader instance
that is initialized
+ // afterwards so the delete file is opened and its footer parsed only once.
+ const bool row_requested = _output_column_requested(kRowColumn);
+
+ if (_delete_file_desc->file_format == TFileFormatType::FORMAT_PARQUET) {
+ auto parquet_reader =
+ ParquetReader::create_unique(_profile, *_range_params, _range,
_batch_size,
+
const_cast<cctz::time_zone*>(&_state->timezone_obj()),
+ _io_ctx, _state, _meta_cache);
+
+ const FieldDescriptor* schema = nullptr;
+ int row_index = -1;
+ if (row_requested) {
+ RETURN_IF_ERROR(parquet_reader->get_file_metadata_schema(&schema));
+ DORIS_CHECK(schema != nullptr);
+ row_index = schema->get_column_index(kRowColumn);
+ }
+ const bool read_row = row_requested && row_index >= 0;
+ _init_read_columns(read_row);
+ std::vector<std::string> read_column_names;
+ read_column_names.reserve(_read_columns.size());
+ for (const auto& column : _read_columns) {
+ read_column_names.push_back(column.name);
+ }
+
+ std::shared_ptr<TableSchemaChangeHelper::Node> table_info_node =
+ TableSchemaChangeHelper::ConstNode::get_instance();
+ if (read_row) {
+ const auto* table_row_field =
find_current_schema_field(_range_params, kRowColumn);
+ if (table_row_field == nullptr) {
+ return Status::InternalError(
+ "Iceberg position delete system table row schema is
missing");
+ }
+ const auto* file_row_field =
schema->get_column(static_cast<size_t>(row_index));
+ std::shared_ptr<TableSchemaChangeHelper::Node> row_node;
+ // The branch-4.1 helper selects ID or name-mapping mode up front
instead of doing so
+ // inside each nested node, so seed that mode from the projected
row field.
+ const bool exist_field_id = file_row_field->field_id != -1;
+
RETURN_IF_ERROR(TableSchemaChangeHelper::BuildTableInfoUtil::by_parquet_field_id(
+ *table_row_field, *file_row_field, exist_field_id,
row_node));
+ auto root_node = create_position_delete_root_node(_read_columns);
+ root_node->add_children(kRowColumn, file_row_field->name,
row_node);
+ table_info_node = std::move(root_node);
+ }
+ // branch-4.1's legacy reader API predates ReaderInitContext.
Position-delete scans have no
+ // predicates, so pass empty filter state while preserving the schema
mapping built above.
+ VExprContextSPtrs conjuncts;
+ phmap::flat_hash_map<int,
std::vector<std::shared_ptr<ColumnPredicate>>>
+ slot_id_to_predicates;
+ RETURN_IF_ERROR(parquet_reader->init_reader(
+ read_column_names, &_read_col_name_to_block_idx, conjuncts,
slot_id_to_predicates,
+ nullptr, nullptr, nullptr, nullptr, nullptr,
std::move(table_info_node), false));
+ _position_reader = std::move(parquet_reader);
+ return Status::OK();
+ }
+
+ if (_delete_file_desc->file_format == TFileFormatType::FORMAT_ORC) {
+ auto orc_reader =
+ OrcReader::create_unique(_profile, _state, *_range_params,
_range, _batch_size,
+ _state->timezone(), _io_ctx,
_meta_cache);
+
+ const orc::Type* row_type = nullptr;
+ if (row_requested) {
+ const orc::Type* root_type = nullptr;
+ RETURN_IF_ERROR(orc_reader->get_file_type(&root_type));
+ DORIS_CHECK(root_type != nullptr);
+ for (uint64_t i = 0; i < root_type->getSubtypeCount(); ++i) {
+ if (root_type->getFieldName(i) == kRowColumn) {
+ row_type = root_type->getSubtype(i);
+ break;
+ }
+ }
+ }
+ const bool read_row = row_requested && row_type != nullptr;
+ _init_read_columns(read_row);
+ std::vector<std::string> read_column_names;
+ read_column_names.reserve(_read_columns.size());
+ for (const auto& column : _read_columns) {
+ read_column_names.push_back(column.name);
+ }
+
+ std::shared_ptr<TableSchemaChangeHelper::Node> table_info_node =
+ TableSchemaChangeHelper::ConstNode::get_instance();
+ if (read_row) {
+ const auto* table_row_field =
find_current_schema_field(_range_params, kRowColumn);
+ if (table_row_field == nullptr) {
+ return Status::InternalError(
+ "Iceberg position delete system table row schema is
missing");
+ }
+ std::shared_ptr<TableSchemaChangeHelper::Node> row_node;
+ const bool exist_field_id =
row_type->hasAttributeKey(kIcebergOrcAttribute);
+
RETURN_IF_ERROR(TableSchemaChangeHelper::BuildTableInfoUtil::by_orc_field_id(
+ *table_row_field, row_type, kIcebergOrcAttribute,
exist_field_id, row_node));
+ auto root_node = create_position_delete_root_node(_read_columns);
+ root_node->add_children(kRowColumn, kRowColumn, row_node);
+ table_info_node = std::move(root_node);
+ }
+ VExprContextSPtrs conjuncts;
+ RETURN_IF_ERROR(orc_reader->init_reader(&read_column_names,
&_read_col_name_to_block_idx,
+ conjuncts, false, nullptr,
nullptr, nullptr,
+ nullptr,
std::move(table_info_node)));
+ _position_reader = std::move(orc_reader);
+ return Status::OK();
+ }
+
+ return Status::NotSupported("Unsupported Iceberg position delete file
format {}",
+ _delete_file_desc->file_format);
+}
+
+Status IcebergPositionDeleteSysTableReader::_init_deletion_vector_reader() {
+ if (!_delete_file_desc->__isset.referenced_data_file_path ||
+ _delete_file_desc->referenced_data_file_path.empty()) {
+ return Status::InternalError("Iceberg deletion vector misses
referenced data file path");
+ }
+
+ IcebergDeleteFileReaderOptions options;
+ options.state = _state;
+ options.profile = _profile;
+ options.scan_params = _range_params;
+ options.io_ctx = _io_ctx.get();
+ options.meta_cache = _meta_cache;
+ if (_range.__isset.fs_name) {
+ options.fs_name = &_range.fs_name;
+ }
+
+ _dv_positions = roaring::Roaring64Map();
+ RETURN_IF_ERROR(read_iceberg_deletion_vector(*_delete_file_desc, options,
&_dv_positions));
+ _next_dv_position.emplace(_dv_positions.begin());
+ return Status::OK();
+}
+
+Status IcebergPositionDeleteSysTableReader::get_next_block(Block* block,
size_t* read_rows,
+ bool* eof) {
+ DORIS_CHECK(_io_ctx != nullptr);
+ if (_io_ctx->should_stop) {
+ *read_rows = 0;
+ *eof = true;
+ return Status::OK();
+ }
+
+ if (_delete_file_kind == DeleteFileKind::DELETION_VECTOR) {
+ return _append_deletion_vector_block(block, read_rows, eof);
+ }
+
+ if (_position_reader == nullptr) {
+ return Status::InternalError("Iceberg position delete reader is not
initialized");
+ }
+
+ while (true) {
+ Block delete_block = _create_delete_block();
+ size_t delete_rows = 0;
+ bool position_reader_eof = false;
+ RETURN_IF_ERROR(_position_reader->get_next_block(&delete_block,
&delete_rows,
+
&position_reader_eof));
+ if (delete_rows > 0) {
+ RETURN_IF_ERROR(
+ _append_position_delete_block(block, delete_block,
delete_rows, read_rows));
+ *eof = position_reader_eof;
+ return Status::OK();
+ }
+ if (position_reader_eof) {
+ *read_rows = 0;
+ *eof = true;
+ return Status::OK();
+ }
+ }
+}
+
+Status
IcebergPositionDeleteSysTableReader::_append_position_delete_block(Block*
output_block,
+
const Block& delete_block,
+
size_t delete_rows,
+
size_t* appended_rows) {
+ auto columns_guard = output_block->mutate_columns_scoped();
+ auto& columns = columns_guard.mutable_columns();
+ auto name_to_pos = output_block->get_name_to_pos_map();
+
+ for (size_t row = 0; row < delete_rows; ++row) {
+ for (const auto* slot : _file_slot_descs) {
+ auto it = name_to_pos.find(slot->col_name());
+ if (it == name_to_pos.end()) {
+ continue;
+ }
+ RETURN_IF_ERROR(_append_sys_column(columns[it->second], *slot,
&delete_block, row, 0));
+ }
+ }
+ // Every output column must be produced by a file slot above; a projected
system-table column
+ // that is not backed by a file slot would be left short and yield a
malformed block with
+ // inconsistent column lengths.
+ check_output_columns_aligned(columns);
+ *appended_rows = delete_rows;
+ return Status::OK();
+}
+
+Status
IcebergPositionDeleteSysTableReader::_append_deletion_vector_block(Block* block,
+
size_t* read_rows,
+
bool* eof) {
+ if (!_next_dv_position.has_value() || *_next_dv_position ==
_dv_positions.end()) {
+ *read_rows = 0;
+ *eof = true;
+ return Status::OK();
+ }
+ const size_t rows_limit = std::max<size_t>(_batch_size, 1);
+
+ auto columns_guard = block->mutate_columns_scoped();
+ auto& columns = columns_guard.mutable_columns();
+ auto name_to_pos = block->get_name_to_pos_map();
+
+ size_t rows = 0;
+ while (rows < rows_limit && *_next_dv_position != _dv_positions.end()) {
+ const uint64_t dv_pos = **_next_dv_position;
+ for (const auto* slot : _file_slot_descs) {
+ auto it = name_to_pos.find(slot->col_name());
+ if (it == name_to_pos.end()) {
+ continue;
+ }
+ RETURN_IF_ERROR(_append_sys_column(columns[it->second], *slot,
nullptr, 0, dv_pos));
+ }
+ ++(*_next_dv_position);
+ ++rows;
+ }
+ check_output_columns_aligned(columns);
+ *read_rows = rows;
+ *eof = *_next_dv_position == _dv_positions.end();
+ return Status::OK();
+}
+
+Status
IcebergPositionDeleteSysTableReader::_append_sys_column(MutableColumnPtr&
column,
+ const
SlotDescriptor& slot,
+ const Block*
delete_block,
+ size_t
source_row, uint64_t dv_pos) {
+ const std::string& name = slot.col_name();
+ if (name == kFilePathColumn) {
+ if (_delete_file_kind == DeleteFileKind::DELETION_VECTOR) {
+ parquet_utils::insert_string(column,
_delete_file_desc->referenced_data_file_path);
+ return Status::OK();
+ }
+ const auto* path_column = get_string_column(*delete_block,
kFilePathColumn);
+ if (path_column == nullptr || !block_has_row(*delete_block,
source_row)) {
+ return Status::InternalError("Iceberg position delete file_path
column is missing");
+ }
+ parquet_utils::insert_string(column,
path_column->get_data_at(source_row).to_string());
+ return Status::OK();
+ }
+
+ if (name == kPosColumn) {
+ if (_delete_file_kind == DeleteFileKind::DELETION_VECTOR) {
+ parquet_utils::insert_int64(column, cast_set<Int64>(dv_pos));
+ return Status::OK();
+ }
+ const auto* pos_column = get_int64_column(*delete_block, kPosColumn);
+ if (pos_column == nullptr || !block_has_row(*delete_block,
source_row)) {
+ return Status::InternalError("Iceberg position delete pos column
is missing");
+ }
+ parquet_utils::insert_int64(column,
pos_column->get_element(source_row));
+ return Status::OK();
+ }
+
+ if (name == kRowColumn) {
+ if (delete_block != nullptr) {
+ auto row_pos = delete_block->get_position_by_name(kRowColumn);
+ if (row_pos >= 0) {
+ const auto& row_column =
*delete_block->get_by_position(row_pos).column;
+ if (source_row < row_column.size()) {
+ column->insert_from(row_column, source_row);
+ return Status::OK();
+ }
+ }
+ }
+ parquet_utils::insert_null(column);
+ return Status::OK();
+ }
+
+ if (name == kPartitionColumn) {
+ return _append_partition_column(column, slot);
+ }
+
+ if (name == kSpecIdColumn) {
+ if (_iceberg_file_desc->__isset.partition_spec_id) {
+ parquet_utils::insert_int32(column,
+
cast_set<Int32>(_iceberg_file_desc->partition_spec_id));
+ } else {
+ parquet_utils::insert_null(column);
+ }
+ return Status::OK();
+ }
+
+ if (name == kDeleteFilePathColumn) {
+ parquet_utils::insert_string(column, _delete_file_output_path());
+ return Status::OK();
+ }
+
+ if (name == kContentOffsetColumn) {
+ const int64_t* value = _delete_file_desc->__isset.content_offset
+ ? &_delete_file_desc->content_offset
+ : nullptr;
+ insert_int64_nullable(column, value);
+ return Status::OK();
+ }
+
+ if (name == kContentSizeInBytesColumn) {
+ const int64_t* value = _delete_file_desc->__isset.content_size_in_bytes
+ ?
&_delete_file_desc->content_size_in_bytes
+ : nullptr;
+ insert_int64_nullable(column, value);
+ return Status::OK();
+ }
+
+ return Status::InternalError("Unknown Iceberg position delete system table
column: {}", name);
+}
+
+Status
IcebergPositionDeleteSysTableReader::_append_partition_column(MutableColumnPtr&
column,
+ const
SlotDescriptor& slot) {
+ if (!_iceberg_file_desc->__isset.partition_data_json ||
+ _iceberg_file_desc->partition_data_json.empty()) {
+ parquet_utils::insert_null(column);
+ return Status::OK();
+ }
+
+ if (!_partition_value) {
+ auto partition_value = slot.get_data_type_ptr()->create_column();
+ auto serde = slot.get_data_type_ptr()->get_serde();
+ StringRef
partition_data(_iceberg_file_desc->partition_data_json.data(),
+
_iceberg_file_desc->partition_data_json.size());
+ DataTypeSerDe::FormatOptions options =
DataTypeSerDe::get_default_format_options();
+ RETURN_IF_ERROR(serde->from_string(partition_data, *partition_value,
options));
+ DORIS_CHECK(partition_value->size() == 1);
+ _partition_value = std::move(partition_value);
+ }
+ column->insert_from(*_partition_value, 0);
+ return Status::OK();
+}
+
+Block IcebergPositionDeleteSysTableReader::_create_delete_block() const {
+ Block block;
+ for (const auto& column : _read_columns) {
+ block.insert(ColumnWithTypeAndName(column.type->create_column(),
column.type, column.name));
+ }
+ return block;
+}
+
+bool IcebergPositionDeleteSysTableReader::_output_column_requested(const
std::string& name) const {
+ return std::any_of(_file_slot_descs.begin(), _file_slot_descs.end(),
+ [&name](const SlotDescriptor* slot) { return
slot->col_name() == name; });
+}
+
+void IcebergPositionDeleteSysTableReader::_init_read_columns(bool read_row) {
+ _read_columns.clear();
+ _read_columns.push_back({kFilePathColumn,
std::make_shared<DataTypeString>()});
Review Comment:
[P1] Preserve optional delete-column schemas on the v1 path
This patch adds support for Parquet position-delete files whose `file_path`
and `pos` fields are optional but contain no nulls, and updates the shared
helper and v2 reader to materialize nullable columns before rejecting actual
null values. This new v1 system-table reader instead requests non-nullable
columns, so `VParquetColumnReader` rejects the optional physical schema
(`max_def_level() > 0`) during the first `get_next_block()`, before any rows
can be exposed. Please read these two source fields as nullable and
unwrap/check their null maps as the helper and v2 path do.
##########
be/src/format/table/iceberg_position_delete_sys_table_reader.cpp:
##########
@@ -0,0 +1,614 @@
+// 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/table/iceberg_position_delete_sys_table_reader.h"
+
+#include <gen_cpp/ExternalTableSchema_types.h>
+#include <gen_cpp/PlanNodes_types.h>
+
+#include <algorithm>
+#include <memory>
+#include <utility>
+
+#include "common/cast_set.h"
+#include "core/block/block.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_string.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_string.h"
+#include "core/data_type_serde/data_type_serde.h"
+#include "core/types.h"
+#include "format/orc/vorc_reader.h"
+#include "format/parquet/schema_desc.h"
+#include "format/parquet/vparquet_reader.h"
+#include "format/table/parquet_utils.h"
+#include "format/table/table_format_reader.h"
+#include "runtime/runtime_state.h"
+
+namespace doris {
+
+namespace {
+
+constexpr const char* kFilePathColumn = "file_path";
+constexpr const char* kPosColumn = "pos";
+constexpr const char* kRowColumn = "row";
+constexpr const char* kPartitionColumn = "partition";
+constexpr const char* kSpecIdColumn = "spec_id";
+constexpr const char* kDeleteFilePathColumn = "delete_file_path";
+constexpr const char* kContentOffsetColumn = "content_offset";
+constexpr const char* kContentSizeInBytesColumn = "content_size_in_bytes";
+constexpr const char* kIcebergOrcAttribute = "iceberg.id";
+constexpr int kPositionDeleteContent = 1;
+
+bool block_has_row(const Block& block, size_t row) {
+ return block.columns() > 0 && row < block.rows();
+}
+
+void insert_int64_nullable(MutableColumnPtr& column, const int64_t* value) {
+ if (value == nullptr) {
+ parquet_utils::insert_null(column);
+ } else {
+ parquet_utils::insert_int64(column, *value);
+ }
+}
+
+// Fail loudly if the filled output block is malformed. Each output column
must be produced by a
+// file slot; a projected system-table column that is not backed by a file
slot would be skipped
+// during fill and left shorter than the rest, producing a block with
inconsistent column lengths.
+void check_output_columns_aligned(const MutableColumns& columns) {
+ if (columns.empty()) {
+ return;
+ }
+ const size_t expected_rows = columns.front()->size();
+ for (const auto& column : columns) {
+ DORIS_CHECK(column->size() == expected_rows)
+ << "Iceberg position delete system table output block has
inconsistent column "
+ "sizes; a projected column is not backed by a file slot";
+ }
+}
+
+const ColumnString* get_string_column(const Block& block, const std::string&
name) {
+ auto pos = block.get_position_by_name(name);
+ if (pos < 0) {
+ return nullptr;
+ }
+ return
check_and_get_column<ColumnString>(block.get_by_position(pos).column.get());
+}
+
+const ColumnInt64* get_int64_column(const Block& block, const std::string&
name) {
+ auto pos = block.get_position_by_name(name);
+ if (pos < 0) {
+ return nullptr;
+ }
+ return
check_and_get_column<ColumnInt64>(block.get_by_position(pos).column.get());
+}
+
+const schema::external::TField* get_field_ptr(const
schema::external::TFieldPtr& field_ptr) {
+ if (!field_ptr.__isset.field_ptr || field_ptr.field_ptr == nullptr) {
+ return nullptr;
+ }
+ return field_ptr.field_ptr.get();
+}
+
+const schema::external::TField* find_current_schema_field(const
TFileScanRangeParams* params,
+ const std::string&
name) {
+ if (params == nullptr || !params->__isset.history_schema_info ||
+ params->history_schema_info.empty()) {
+ return nullptr;
+ }
+ const schema::external::TSchema* schema =
¶ms->history_schema_info.front();
+ if (params->__isset.current_schema_id) {
+ for (const auto& candidate : params->history_schema_info) {
+ if (candidate.__isset.schema_id && candidate.schema_id ==
params->current_schema_id) {
+ schema = &candidate;
+ break;
+ }
+ }
+ }
+ if (!schema->__isset.root_field || !schema->root_field.__isset.fields) {
+ return nullptr;
+ }
+ for (const auto& field_ptr : schema->root_field.fields) {
+ const auto* field = get_field_ptr(field_ptr);
+ if (field != nullptr && field->__isset.name && field->name == name) {
+ return field;
+ }
+ }
+ return nullptr;
+}
+
+template <typename ReadColumns>
+std::shared_ptr<TableSchemaChangeHelper::StructNode>
create_position_delete_root_node(
+ const ReadColumns& read_columns) {
+ auto root_node = std::make_shared<TableSchemaChangeHelper::StructNode>();
+ for (const auto& column : read_columns) {
+ if (column.name == kRowColumn) {
+ continue;
+ }
+ root_node->add_children(column.name, column.name,
+
TableSchemaChangeHelper::ConstNode::get_instance());
+ }
+ return root_node;
+}
+
+} // namespace
+
+IcebergPositionDeleteSysTableReader::IcebergPositionDeleteSysTableReader(
+ const std::vector<SlotDescriptor*>& file_slot_descs, RuntimeState*
state,
+ RuntimeProfile* profile, const TFileRangeDesc& range,
+ const TFileScanRangeParams* range_params,
std::shared_ptr<io::IOContext> io_ctx,
+ FileMetaCache* meta_cache)
+ : _file_slot_descs(file_slot_descs),
+ _state(state),
+ _profile(profile),
+ _range(range),
+ _range_params(range_params),
+ _io_ctx(std::move(io_ctx)) {
+ _meta_cache = meta_cache;
+}
+
+IcebergPositionDeleteSysTableReader::~IcebergPositionDeleteSysTableReader() =
default;
+
+Status IcebergPositionDeleteSysTableReader::init_reader() {
+ if (_state == nullptr || _profile == nullptr || _range_params == nullptr ||
+ _io_ctx == nullptr) {
+ return Status::InvalidArgument(
+ "invalid Iceberg position delete system table reader context");
+ }
+ if (!_range.__isset.table_format_params ||
!_range.table_format_params.__isset.iceberg_params) {
+ return Status::InternalError("Iceberg position delete system table
range misses params");
+ }
+
+ _iceberg_file_desc = &_range.table_format_params.iceberg_params;
+ if (!_iceberg_file_desc->__isset.delete_files ||
_iceberg_file_desc->delete_files.size() != 1) {
+ return Status::InternalError(
+ "Iceberg position delete system table range should contain
exactly one delete "
+ "file");
+ }
+ _delete_file_desc = _iceberg_file_desc->delete_files.data();
+ if (is_iceberg_deletion_vector(*_delete_file_desc)) {
+ _delete_file_kind = DeleteFileKind::DELETION_VECTOR;
+ } else if (_delete_file_desc->__isset.content &&
+ _delete_file_desc->content == kPositionDeleteContent) {
+ _delete_file_kind = DeleteFileKind::POSITION_DELETE;
+ } else if (!_delete_file_desc->__isset.content) {
+ return Status::InternalError(
+ "Iceberg position delete system table delete file misses
content");
+ } else {
+ return Status::InternalError(
+ "Iceberg position delete system table does not support delete
file content {}",
+ _delete_file_desc->content);
+ }
+ _batch_size = _state->batch_size();
+
+ if (_delete_file_kind == DeleteFileKind::DELETION_VECTOR) {
+ return _init_deletion_vector_reader();
+ }
+ return _init_position_delete_reader();
+}
+
+Status IcebergPositionDeleteSysTableReader::get_columns(
+ std::unordered_map<std::string, DataTypePtr>* name_to_type,
+ std::unordered_set<std::string>* missing_cols) {
+ missing_cols->clear();
+ for (const auto* slot : _file_slot_descs) {
+ name_to_type->emplace(slot->col_name(), slot->get_data_type_ptr());
+ }
+ return Status::OK();
+}
+
+bool IcebergPositionDeleteSysTableReader::count_read_rows() {
+ return _delete_file_kind == DeleteFileKind::POSITION_DELETE;
+}
+
+void IcebergPositionDeleteSysTableReader::_collect_profile_before_close() {
+ if (_position_reader != nullptr) {
+ _position_reader->collect_profile_before_close();
+ }
+}
+
+Status IcebergPositionDeleteSysTableReader::close() {
+ if (_position_reader != nullptr) {
+ RETURN_IF_ERROR(_position_reader->close());
+ }
+ _partition_value.reset();
+ _next_dv_position.reset();
+ _dv_positions = roaring::Roaring64Map();
+ return Status::OK();
+}
+
+Status IcebergPositionDeleteSysTableReader::_init_position_delete_reader() {
+ if (!_delete_file_desc->__isset.file_format) {
+ return Status::InternalError("Iceberg position delete file misses file
format");
+ }
+
+ // `row` is optional in position delete files and expensive to read, so
only read it when the
+ // query actually projects it. Whether the delete file physically stores
`row` is decided from
+ // the reader's own footer/type below, reusing the same reader instance
that is initialized
+ // afterwards so the delete file is opened and its footer parsed only once.
+ const bool row_requested = _output_column_requested(kRowColumn);
+
+ if (_delete_file_desc->file_format == TFileFormatType::FORMAT_PARQUET) {
+ auto parquet_reader =
+ ParquetReader::create_unique(_profile, *_range_params, _range,
_batch_size,
+
const_cast<cctz::time_zone*>(&_state->timezone_obj()),
+ _io_ctx, _state, _meta_cache);
+
+ const FieldDescriptor* schema = nullptr;
+ int row_index = -1;
+ if (row_requested) {
+ RETURN_IF_ERROR(parquet_reader->get_file_metadata_schema(&schema));
+ DORIS_CHECK(schema != nullptr);
+ row_index = schema->get_column_index(kRowColumn);
+ }
+ const bool read_row = row_requested && row_index >= 0;
+ _init_read_columns(read_row);
+ std::vector<std::string> read_column_names;
+ read_column_names.reserve(_read_columns.size());
+ for (const auto& column : _read_columns) {
+ read_column_names.push_back(column.name);
+ }
+
+ std::shared_ptr<TableSchemaChangeHelper::Node> table_info_node =
+ TableSchemaChangeHelper::ConstNode::get_instance();
+ if (read_row) {
+ const auto* table_row_field =
find_current_schema_field(_range_params, kRowColumn);
+ if (table_row_field == nullptr) {
+ return Status::InternalError(
+ "Iceberg position delete system table row schema is
missing");
+ }
+ const auto* file_row_field =
schema->get_column(static_cast<size_t>(row_index));
+ std::shared_ptr<TableSchemaChangeHelper::Node> row_node;
+ // The branch-4.1 helper selects ID or name-mapping mode up front
instead of doing so
+ // inside each nested node, so seed that mode from the projected
row field.
+ const bool exist_field_id = file_row_field->field_id != -1;
+
RETURN_IF_ERROR(TableSchemaChangeHelper::BuildTableInfoUtil::by_parquet_field_id(
+ *table_row_field, *file_row_field, exist_field_id,
row_node));
+ auto root_node = create_position_delete_root_node(_read_columns);
+ root_node->add_children(kRowColumn, file_row_field->name,
row_node);
+ table_info_node = std::move(root_node);
+ }
+ // branch-4.1's legacy reader API predates ReaderInitContext.
Position-delete scans have no
+ // predicates, so pass empty filter state while preserving the schema
mapping built above.
+ VExprContextSPtrs conjuncts;
+ phmap::flat_hash_map<int,
std::vector<std::shared_ptr<ColumnPredicate>>>
+ slot_id_to_predicates;
+ RETURN_IF_ERROR(parquet_reader->init_reader(
Review Comment:
[P1] Finalize the nested Parquet projection before reading
On branch-4.1, `init_reader()` does not populate the legacy reader's
`all_read_columns`; that happens only in `set_fill_columns()`. This path
installs the nested reader without that second step, so it returns `(0 rows,
eof=false)` and the wrapper's `while (true)` immediately retries without
rechecking cancellation, hanging every v1 Parquet `$position_deletes` scan. The
sibling helper changed in this patch explicitly calls `set_fill_columns({},
{})` and notes that omitting it yields empty batches forever. Please make the
same call here before storing `_position_reader`.
--
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]
