Gabriel39 commented on code in PR #65674:
URL: https://github.com/apache/doris/pull/65674#discussion_r3601325414
##########
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:
Fixed in f3e1d84f753. INT96 now validates nanos-of-day, uses widened day
arithmetic, enforces the Doris 0001-9999 range, and routes
plain/dictionary/decoded failures through the existing strict/non-strict state.
Added valid-boundary, corrupt-Julian/nanos, dictionary-ID, and decoded-value
tests.
##########
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:
Fixed in f3e1d84f753. DATETIMEV2 and TIMESTAMPTZ now share checked unit
scaling and target-range validation; TIMESTAMPTZ plain, typed-dictionary, and
decoded/statistics paths preserve strict rollback and non-strict NULL
propagation. Tests cover INT64_MAX milliseconds, year 10000, years 0001/9999,
decoded values, and dictionary IDs.
##########
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:
Fixed in f3e1d84f753. OffsetIndex validation now receives the owning
ColumnMetaData data_page_offset and rejects the whole optional index unless
location zero matches it exactly. Added a shifted-first-location test while
retaining the sequential-reader fallback.
--
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]
