github-actions[bot] commented on code in PR #65674:
URL: https://github.com/apache/doris/pull/65674#discussion_r3601093671
##########
be/src/core/data_type_serde/data_type_timestamptz_serde.cpp:
##########
@@ -74,6 +76,46 @@ int64_t decoded_timestamp_micros(const DecodedColumnView&
view, int64_t value) {
return value;
}
+class TimestampTzParquetConsumer final : public ParquetFixedValueConsumer {
+public:
+ TimestampTzParquetConsumer(IColumn& column, const ParquetDecodeContext&
context)
+ : _data(assert_cast<ColumnTimeStampTz&>(column).get_data()),
_context(context) {}
+
+ Status consume(const uint8_t* values, size_t num_values, size_t
value_width) override {
+ for (size_t row = 0; row < num_values; ++row) {
+ int64_t timestamp_micros;
+ if (_context.physical_type == ParquetPhysicalType::INT96) {
+ DORIS_CHECK_EQ(value_width, sizeof(DecodedInt96Timestamp));
+ timestamp_micros = unaligned_load<DecodedInt96Timestamp>(
+ values + row *
sizeof(DecodedInt96Timestamp))
+ .to_timestamp_micros();
+ } else {
+ DORIS_CHECK(_context.physical_type ==
ParquetPhysicalType::INT64);
+ DORIS_CHECK_EQ(value_width, sizeof(int64_t));
+ timestamp_micros = unaligned_load<int64_t>(values + row *
sizeof(int64_t));
+ if (_context.time_unit == ParquetTimeUnit::MILLIS) {
+ timestamp_micros *= 1000;
Review Comment:
[P1] Check TIMESTAMPTZ unit conversion and target range before appending.
This INT64 is file data: `INT64_MAX` milliseconds overflows at `*= 1000`, after
which a wrapped value can be appended as a plausible unrelated timestamp. Even
the non-overflowing `253402300800000` milliseconds represents year 10000, but
unchecked `from_unixtime()` appends it outside TIMESTAMPTZ's documented
0001-9999 range and still returns OK. The adjacent DATETIMEV2 consumer already
has checked unit conversion and failure state, while this consumer ignores the
passed state; its decoded/statistics path repeats the same omissions. Reuse a
checked unit-and-target-range helper for plain, dictionary, and statistics
paths, and cover strict/non-strict overflow plus year-boundary cases.
##########
be/src/core/data_type_serde/data_type_datetimev2_serde.cpp:
##########
@@ -133,6 +136,89 @@ int64_t decoded_timestamp_micros(const DecodedColumnView&
view, int64_t value) {
return value;
}
+Status parquet_timestamp_micros(const ParquetDecodeContext& context, int64_t
value,
+ int64_t* result) {
+ if (context.time_unit == ParquetTimeUnit::MILLIS) {
+ if (value > std::numeric_limits<int64_t>::max() / 1000 ||
+ value < std::numeric_limits<int64_t>::min() / 1000) {
+ return Status::DataQualityError("Parquet timestamp overflows
microseconds");
+ }
+ *result = value * 1000;
+ return Status::OK();
+ }
+ if (context.time_unit == ParquetTimeUnit::NANOS) {
+ *result = value / 1000;
+ return Status::OK();
+ }
+ *result = value;
+ return Status::OK();
+}
+
+class DateTimeV2ParquetConsumer final : public ParquetFixedValueConsumer {
+public:
+ DateTimeV2ParquetConsumer(IColumn& column, const ParquetDecodeContext&
context,
+ ParquetMaterializationState* state = nullptr)
+ : _data(assert_cast<ColumnDateTimeV2&>(column).get_data()),
+ _context(context),
+ _state(state) {}
+
+ Status consume(const uint8_t* values, size_t num_values, size_t
value_width) override {
+ const size_t old_size = _data.size();
+ static const auto utc_timezone = cctz::utc_time_zone();
+ const auto& timezone = _context.timezone == nullptr ? utc_timezone :
*_context.timezone;
+ for (size_t row = 0; row < num_values; ++row) {
+ int64_t timestamp_micros;
+ if (_context.physical_type == ParquetPhysicalType::INT96) {
+ DORIS_CHECK_EQ(value_width, sizeof(DecodedInt96Timestamp));
+ timestamp_micros = unaligned_load<DecodedInt96Timestamp>(
+ values + row *
sizeof(DecodedInt96Timestamp))
+ .to_timestamp_micros();
Review Comment:
[P1] Validate INT96 before materializing it. `to_timestamp_micros()`
subtracts the Julian epoch and multiplies by 86,400,000,000 in signed types,
and it never bounds nanos-of-day. For `{0, INT32_MAX}` this overflows before
this added call returns; the result is passed to unchecked `from_unixtime()`,
an invalid year is appended, and `consume()` still returns OK. This branch also
bypasses the conversion-failure state used just below, and TIMESTAMPTZ repeats
it. Use checked/widened arithmetic, enforce the INT96 day-nanos and Doris year
ranges, route failures through materialization state for plain/dictionary
pages, and add boundary/corruption tests.
##########
be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp:
##########
@@ -0,0 +1,1119 @@
+// 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 <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/custom_allocator.h"
+#include "core/data_type_serde/data_type_serde.h"
+#include "format/parquet/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"
+
+namespace cctz {
+class time_zone;
+} // namespace cctz
+namespace doris {
+namespace io {
+class BufferedStreamReader;
+struct IOContext;
+} // namespace io
+} // namespace doris
+
+namespace doris::format::parquet::native {
+
+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) {
+ 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 row_count) {
+ if (index.page_locations.empty() || row_count < 0 ||
Review Comment:
[P1] Bind the first OffsetIndex location to `data_page_offset`. The
monotonic/in-chunk checks still accept an index shifted by one page: with
actual P0/P1/P2 at 100/120/140, `{(120,row0),(140,row10)}` passes validation.
This reader parses P0 sequentially, but a selected range starting at row 10
makes `next_page()` use location 1 and jump to P2, returning P2 under P1's row
range while never reading P1. This is the missing coherence case beyond the
existing backward/overlap thread. Pass the owning data-page offset into
validation, fall back unless location zero matches it, and add a
shifted-first-location pruning test.
--
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]
