You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by ga...@apache.org on 2026/05/26 09:45:51 UTC
(doris) branch refact_reader_branch updated: Complete basic parquet reader (#63659)
This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch refact_reader_branch
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/refact_reader_branch by this push:
new 3dbfc4c596d Complete basic parquet reader (#63659)
3dbfc4c596d is described below
commit 3dbfc4c596dc9997db4f1211765c02872ae5252d
Author: Gabriel <li...@selectdb.com>
AuthorDate: Tue May 26 17:45:41 2026 +0800
Complete basic parquet reader (#63659)
Co-authored-by: Socrates <su...@selectdb.com>
---
.../data_type_serde/data_type_datetimev2_serde.cpp | 30 +-
.../data_type_serde/data_type_datetimev2_serde.h | 2 +
.../data_type_serde/data_type_datev2_serde.cpp | 20 +-
.../core/data_type_serde/data_type_datev2_serde.h | 2 +
.../data_type_serde/data_type_decimal_serde.cpp | 64 +++
.../core/data_type_serde/data_type_decimal_serde.h | 2 +
.../data_type_serde/data_type_nullable_serde.cpp | 20 +-
.../data_type_serde/data_type_nullable_serde.h | 2 +
.../data_type_serde/data_type_number_serde.cpp | 52 ++
.../core/data_type_serde/data_type_number_serde.h | 3 +
be/src/core/data_type_serde/data_type_serde.cpp | 6 +
be/src/core/data_type_serde/data_type_serde.h | 7 +
.../data_type_serde/data_type_string_serde.cpp | 28 +
.../core/data_type_serde/data_type_string_serde.h | 3 +
.../core/data_type_serde/data_type_time_serde.cpp | 44 ++
be/src/core/data_type_serde/data_type_time_serde.h | 2 +
be/src/core/data_type_serde/decoded_column_view.h | 62 +++
be/src/exprs/vslot_ref.h | 2 +-
be/src/format/new_parquet/column_reader.cpp | 562 +++++++++++++++++++
be/src/format/new_parquet/column_reader.h | 116 ++++
.../format/new_parquet/parquet_column_schema.cpp | 158 ++++++
be/src/format/new_parquet/parquet_column_schema.h | 67 +++
be/src/format/new_parquet/parquet_reader.cpp | 601 +++++++++++++++++++++
be/src/format/new_parquet/parquet_reader.h | 138 +++++
be/src/format/new_parquet/parquet_statistics.cpp | 254 +++++++++
be/src/format/new_parquet/parquet_statistics.h | 88 +++
be/src/format/new_parquet/parquet_type.cpp | 349 ++++++++++++
be/src/format/new_parquet/parquet_type.h | 84 +++
be/src/format/new_parquet/selection_vector.h | 116 ++++
be/src/format/parquet/parquet_reader.h | 77 ---
be/src/format/reader/column_mapper.cpp | 7 +-
be/src/format/reader/expr/delete_predicate.cpp | 4 +-
be/src/format/reader/file_reader.cpp | 42 ++
be/src/format/reader/file_reader.h | 118 ++--
be/src/format/reader/table_reader.h | 22 +-
.../data_type_serde_decoded_values_test.cpp | 278 ++++++++++
.../new_parquet/parquet_column_reader_test.cpp | 562 +++++++++++++++++++
be/test/format/new_parquet/parquet_reader_test.cpp | 341 ++++++++++++
docs/doris-arrow-parquet-reader-implementation.md | 291 ++++++++++
39 files changed, 4485 insertions(+), 141 deletions(-)
diff --git a/be/src/core/data_type_serde/data_type_datetimev2_serde.cpp b/be/src/core/data_type_serde/data_type_datetimev2_serde.cpp
index 92a5106b481..fc2c14d1829 100644
--- a/be/src/core/data_type_serde/data_type_datetimev2_serde.cpp
+++ b/be/src/core/data_type_serde/data_type_datetimev2_serde.cpp
@@ -22,9 +22,9 @@
#include <chrono> // IWYU pragma: keep
#include <cstdint>
-
#include "common/status.h"
#include "core/column/column_const.h"
+#include "core/data_type_serde/decoded_column_view.h"
#include "core/data_type/data_type_decimal.h"
#include "core/data_type/data_type_number.h"
#include "core/data_type/primitive_type.h"
@@ -451,6 +451,34 @@ Status DataTypeDateTimeV2SerDe::read_column_from_arrow(IColumn& column,
return Status::OK();
}
+Status DataTypeDateTimeV2SerDe::read_column_from_decoded_values(
+ IColumn& column, const DecodedColumnView& view) const {
+ if (view.value_kind != DecodedValueKind::INT64) {
+ return Status::NotSupported("DATETIMEV2 decoded reader expects INT64 source");
+ }
+ if (view.values == nullptr && view.row_count > 0) {
+ return Status::Corruption("Decoded value buffer is null for {}", column.get_name());
+ }
+ auto& data = assert_cast<ColumnDateTimeV2&>(column).get_data();
+ const auto* values = reinterpret_cast<const int64_t*>(view.values);
+ static const cctz::time_zone utc_time_zone = cctz::utc_time_zone();
+ const int64_t second_mask = view.time_unit == DecodedTimeUnit::MILLIS ? 1000 : 1000000;
+ for (int64_t row = 0; row < view.row_count; ++row) {
+ int64_t epoch_seconds = values[row] / second_mask;
+ int64_t sub_second = values[row] % second_mask;
+ if (sub_second < 0) {
+ sub_second += second_mask;
+ --epoch_seconds;
+ }
+ const int32_t microsecond = static_cast<int32_t>(sub_second * (1000000 / second_mask));
+ DateV2Value<DateTimeV2ValueType> datetime_value;
+ datetime_value.from_unixtime(epoch_seconds, utc_time_zone);
+ datetime_value.set_microsecond(static_cast<uint64_t>(microsecond));
+ data.push_back(datetime_value);
+ }
+ return Status::OK();
+}
+
Status DataTypeDateTimeV2SerDe::write_column_to_mysql_binary(const IColumn& column,
MysqlRowBinaryBuffer& result,
int64_t row_idx, bool col_const,
diff --git a/be/src/core/data_type_serde/data_type_datetimev2_serde.h b/be/src/core/data_type_serde/data_type_datetimev2_serde.h
index 0389432a621..34d0373eba1 100644
--- a/be/src/core/data_type_serde/data_type_datetimev2_serde.h
+++ b/be/src/core/data_type_serde/data_type_datetimev2_serde.h
@@ -88,6 +88,8 @@ public:
const cctz::time_zone& ctz) const override;
Status read_column_from_arrow(IColumn& column, const arrow::Array* arrow_array, int64_t start,
int64_t end, const cctz::time_zone& ctz) const override;
+ Status read_column_from_decoded_values(IColumn& column,
+ const DecodedColumnView& view) const override;
Status write_column_to_mysql_binary(const IColumn& column, MysqlRowBinaryBuffer& row_buffer,
int64_t row_idx, bool col_const,
diff --git a/be/src/core/data_type_serde/data_type_datev2_serde.cpp b/be/src/core/data_type_serde/data_type_datev2_serde.cpp
index d72f3664ae6..9410df86eaa 100644
--- a/be/src/core/data_type_serde/data_type_datev2_serde.cpp
+++ b/be/src/core/data_type_serde/data_type_datev2_serde.cpp
@@ -22,8 +22,8 @@
#include <fmt/core.h>
#include <cstdint>
-
#include "core/column/column_const.h"
+#include "core/data_type_serde/decoded_column_view.h"
#include "core/data_type/data_type_decimal.h"
#include "core/data_type/data_type_number.h"
#include "core/data_type/define_primitive_type.h"
@@ -125,6 +125,24 @@ Status DataTypeDateV2SerDe::read_column_from_arrow(IColumn& column, const arrow:
return Status::OK();
}
+Status DataTypeDateV2SerDe::read_column_from_decoded_values(
+ IColumn& column, const DecodedColumnView& view) const {
+ if (view.value_kind != DecodedValueKind::INT32) {
+ return Status::NotSupported("DATEV2 decoded reader expects INT32 source");
+ }
+ if (view.values == nullptr && view.row_count > 0) {
+ return Status::Corruption("Decoded value buffer is null for {}", column.get_name());
+ }
+ auto& data = assert_cast<ColumnDateV2&>(column).get_data();
+ const auto* values = reinterpret_cast<const int32_t*>(view.values);
+ for (int64_t row = 0; row < view.row_count; ++row) {
+ DateV2Value<DateV2ValueType> date_v2;
+ date_v2.get_date_from_daynr(values[row] + date_threshold);
+ data.push_back(date_v2);
+ }
+ return Status::OK();
+}
+
Status DataTypeDateV2SerDe::write_column_to_mysql_binary(const IColumn& column,
MysqlRowBinaryBuffer& result,
int64_t row_idx, bool col_const,
diff --git a/be/src/core/data_type_serde/data_type_datev2_serde.h b/be/src/core/data_type_serde/data_type_datev2_serde.h
index 0375f9be4b4..ff985d61345 100644
--- a/be/src/core/data_type_serde/data_type_datev2_serde.h
+++ b/be/src/core/data_type_serde/data_type_datev2_serde.h
@@ -86,6 +86,8 @@ public:
const cctz::time_zone& ctz) const override;
Status read_column_from_arrow(IColumn& column, const arrow::Array* arrow_array, int64_t start,
int64_t end, const cctz::time_zone& ctz) const override;
+ Status read_column_from_decoded_values(IColumn& column,
+ const DecodedColumnView& view) const override;
Status write_column_to_mysql_binary(const IColumn& column, MysqlRowBinaryBuffer& row_buffer,
int64_t row_idx, bool col_const,
const FormatOptions& options) const override;
diff --git a/be/src/core/data_type_serde/data_type_decimal_serde.cpp b/be/src/core/data_type_serde/data_type_decimal_serde.cpp
index 084f9c75df3..c744cdb8a2f 100644
--- a/be/src/core/data_type_serde/data_type_decimal_serde.cpp
+++ b/be/src/core/data_type_serde/data_type_decimal_serde.cpp
@@ -31,6 +31,7 @@
#include "core/column/column_decimal.h"
#include "core/data_type/data_type_decimal.h"
#include "core/data_type/define_primitive_type.h"
+#include "core/data_type_serde/decoded_column_view.h"
#include "core/types.h"
#include "exec/common/arithmetic_overflow.h"
#include "exprs/function/cast/cast_to_decimal.h"
@@ -43,6 +44,53 @@
#include "util/string_parser.hpp"
namespace doris {
+namespace {
+
+template <typename NativeType>
+NativeType decode_big_endian_signed_integer(const uint8_t* data, int length) {
+ using UnsignedNativeType =
+ std::conditional_t<std::is_same_v<NativeType, Int128>, unsigned __int128,
+ std::make_unsigned_t<NativeType>>;
+ UnsignedNativeType value = data != nullptr && length > 0 && (data[0] & 0x80) != 0
+ ? static_cast<UnsignedNativeType>(-1)
+ : 0;
+ for (int i = 0; i < length; ++i) {
+ value = static_cast<UnsignedNativeType>((value << 8) | data[i]);
+ }
+ return static_cast<NativeType>(value);
+}
+
+template <PrimitiveType T>
+typename PrimitiveTypeTraits<T>::CppType read_decimal_decoded_value(
+ const DecodedColumnView& view, int64_t row) {
+ using FieldType = typename PrimitiveTypeTraits<T>::CppType;
+ if (view.value_kind == DecodedValueKind::INT32) {
+ const auto* values = reinterpret_cast<const int32_t*>(view.values);
+ return FieldType {static_cast<typename FieldType::NativeType>(values[row])};
+ }
+ if (view.value_kind == DecodedValueKind::INT64) {
+ const auto* values = reinterpret_cast<const int64_t*>(view.values);
+ return FieldType {static_cast<typename FieldType::NativeType>(values[row])};
+ }
+ const auto& value = (*view.binary_values)[row];
+ const auto length = view.value_kind == DecodedValueKind::FIXED_BINARY
+ ? view.fixed_length
+ : cast_set<int, size_t, false>(value.size);
+ return FieldType {static_cast<typename FieldType::NativeType>(
+ decode_big_endian_signed_integer<Int128>(reinterpret_cast<const uint8_t*>(value.data),
+ length))};
+}
+
+template <PrimitiveType T>
+Status read_decimal_decoded_values(IColumn& column, const DecodedColumnView& view) {
+ auto& data = assert_cast<ColumnDecimal<T>&>(column).get_data();
+ for (int64_t row = 0; row < view.row_count; ++row) {
+ data.push_back(read_decimal_decoded_value<T>(view, row));
+ }
+ return Status::OK();
+}
+
+} // namespace
template <PrimitiveType T>
Status DataTypeDecimalSerDe<T>::from_string_batch(const ColumnString& str, ColumnNullable& column,
@@ -381,6 +429,22 @@ Status DataTypeDecimalSerDe<T>::read_column_from_arrow(IColumn& column,
return Status::OK();
}
+template <PrimitiveType T>
+Status DataTypeDecimalSerDe<T>::read_column_from_decoded_values(
+ IColumn& column, const DecodedColumnView& view) const {
+ if constexpr (T == TYPE_DECIMAL32 || T == TYPE_DECIMAL64 || T == TYPE_DECIMAL128I ||
+ T == TYPE_DECIMAL256) {
+ if (view.value_kind == DecodedValueKind::INT32 ||
+ view.value_kind == DecodedValueKind::INT64 ||
+ view.value_kind == DecodedValueKind::BINARY ||
+ view.value_kind == DecodedValueKind::FIXED_BINARY) {
+ return read_decimal_decoded_values<T>(column, view);
+ }
+ }
+ return Status::NotSupported("Unsupported decoded values for {} from source kind {}",
+ get_name(), static_cast<int>(view.value_kind));
+}
+
template <PrimitiveType T>
Status DataTypeDecimalSerDe<T>::write_column_to_mysql_binary(const IColumn& column,
MysqlRowBinaryBuffer& result,
diff --git a/be/src/core/data_type_serde/data_type_decimal_serde.h b/be/src/core/data_type_serde/data_type_decimal_serde.h
index 0185672e024..089835a21be 100644
--- a/be/src/core/data_type_serde/data_type_decimal_serde.h
+++ b/be/src/core/data_type_serde/data_type_decimal_serde.h
@@ -107,6 +107,8 @@ public:
const cctz::time_zone& ctz) const override;
Status read_column_from_arrow(IColumn& column, const arrow::Array* arrow_array, int64_t start,
int64_t end, const cctz::time_zone& ctz) const override;
+ Status read_column_from_decoded_values(IColumn& column,
+ const DecodedColumnView& view) const override;
Status write_column_to_mysql_binary(const IColumn& column, MysqlRowBinaryBuffer& row_buffer,
int64_t row_idx, bool col_const,
const FormatOptions& options) const override;
diff --git a/be/src/core/data_type_serde/data_type_nullable_serde.cpp b/be/src/core/data_type_serde/data_type_nullable_serde.cpp
index a93f8d6126c..6b15b29c63a 100644
--- a/be/src/core/data_type_serde/data_type_nullable_serde.cpp
+++ b/be/src/core/data_type_serde/data_type_nullable_serde.cpp
@@ -22,13 +22,14 @@
#include <algorithm>
#include <boost/iterator/iterator_facade.hpp>
-#include <memory>
+#include <vector>
#include "core/assert_cast.h"
#include "core/column/column.h"
#include "core/column/column_const.h"
#include "core/column/column_nullable.h"
#include "core/column/column_vector.h"
+#include "core/data_type_serde/decoded_column_view.h"
#include "core/data_type_serde/data_type_serde.h"
#include "core/data_type_serde/data_type_string_serde.h"
#include "exprs/function/cast/cast_base.h"
@@ -350,6 +351,23 @@ Status DataTypeNullableSerDe::read_column_from_arrow(IColumn& column,
ctz);
}
+Status DataTypeNullableSerDe::read_column_from_decoded_values(
+ IColumn& column, const DecodedColumnView& view) const {
+ auto& nullable_column = assert_cast<ColumnNullable&>(column);
+ auto& null_map = nullable_column.get_null_map_data();
+ const auto old_size = null_map.size();
+ null_map.resize(null_map.size() + view.row_count);
+ if (view.null_map != nullptr) {
+ // TODO: skip if no null in map
+ auto* dst = null_map.data() + old_size;
+ memcpy(dst, view.null_map, view.row_count);
+ }
+ DecodedColumnView nested_view = view;
+ nested_view.null_map = nullptr;
+ return nested_serde->read_column_from_decoded_values(nullable_column.get_nested_column(),
+ nested_view);
+}
+
bool DataTypeNullableSerDe::write_column_to_mysql_text(const IColumn& column, BufferWritable& bw,
int64_t row_idx,
const FormatOptions& options) const {
diff --git a/be/src/core/data_type_serde/data_type_nullable_serde.h b/be/src/core/data_type_serde/data_type_nullable_serde.h
index cfb4e1e3bca..376f3692dc1 100644
--- a/be/src/core/data_type_serde/data_type_nullable_serde.h
+++ b/be/src/core/data_type_serde/data_type_nullable_serde.h
@@ -86,6 +86,8 @@ public:
const cctz::time_zone& ctz) const override;
Status read_column_from_arrow(IColumn& column, const arrow::Array* arrow_array, int64_t start,
int64_t end, const cctz::time_zone& ctz) const override;
+ Status read_column_from_decoded_values(IColumn& column,
+ const DecodedColumnView& view) const override;
Status write_column_to_mysql_binary(const IColumn& column, MysqlRowBinaryBuffer& row_buffer,
int64_t row_idx, bool col_const,
const FormatOptions& options) const override;
diff --git a/be/src/core/data_type_serde/data_type_number_serde.cpp b/be/src/core/data_type_serde/data_type_number_serde.cpp
index 39e9c0726c4..131e6d05941 100644
--- a/be/src/core/data_type_serde/data_type_number_serde.cpp
+++ b/be/src/core/data_type_serde/data_type_number_serde.cpp
@@ -26,6 +26,7 @@
#include "core/column/column_nullable.h"
#include "core/data_type/define_primitive_type.h"
#include "core/data_type/primitive_type.h"
+#include "core/data_type_serde/decoded_column_view.h"
#include "core/data_type_serde/data_type_serde.h"
#include "core/packed_int128.h"
#include "core/types.h"
@@ -42,6 +43,29 @@
#include "util/to_string.h"
namespace doris {
+namespace {
+
+template <typename NativeType>
+const NativeType* decoded_values_as(const DecodedColumnView& view) {
+ return reinterpret_cast<const NativeType*>(view.values);
+}
+
+template <PrimitiveType DorisType, typename SourceType>
+Status read_number_decoded_values(IColumn& column, const DecodedColumnView& view) {
+ if (view.values == nullptr && view.row_count > 0) {
+ return Status::Corruption("Decoded value buffer is null for {}", column.get_name());
+ }
+ auto& data = assert_cast<typename PrimitiveTypeTraits<DorisType>::ColumnType&>(column)
+ .get_data();
+ const auto* values = decoded_values_as<SourceType>(view);
+ for (int64_t row = 0; row < view.row_count; ++row) {
+ using DorisCppType = typename PrimitiveTypeTraits<DorisType>::CppType;
+ data.push_back(static_cast<DorisCppType>(values[row]));
+ }
+ return Status::OK();
+}
+
+} // namespace
// Type map的基本结构
template <typename Key, typename Value, typename... Rest>
struct TypeMap {
@@ -156,6 +180,34 @@ Status DataTypeNumberSerDe<T>::write_column_to_arrow(const IColumn& column, cons
return Status::OK();
}
+template <PrimitiveType T>
+Status DataTypeNumberSerDe<T>::read_column_from_decoded_values(
+ IColumn& column, const DecodedColumnView& view) const {
+ if constexpr (T == TYPE_BOOLEAN) {
+ if (view.value_kind == DecodedValueKind::BOOL) {
+ return read_number_decoded_values<TYPE_BOOLEAN, bool>(column, view);
+ }
+ } else if constexpr (T == TYPE_INT) {
+ if (view.value_kind == DecodedValueKind::INT32) {
+ return read_number_decoded_values<TYPE_INT, int32_t>(column, view);
+ }
+ } else if constexpr (T == TYPE_BIGINT) {
+ if (view.value_kind == DecodedValueKind::INT64) {
+ return read_number_decoded_values<TYPE_BIGINT, int64_t>(column, view);
+ }
+ } else if constexpr (T == TYPE_FLOAT) {
+ if (view.value_kind == DecodedValueKind::FLOAT) {
+ return read_number_decoded_values<TYPE_FLOAT, float>(column, view);
+ }
+ } else if constexpr (T == TYPE_DOUBLE) {
+ if (view.value_kind == DecodedValueKind::DOUBLE) {
+ return read_number_decoded_values<TYPE_DOUBLE, double>(column, view);
+ }
+ }
+ return Status::NotSupported("Unsupported decoded values for {} from source kind {}",
+ get_name(), static_cast<int>(view.value_kind));
+}
+
template <PrimitiveType T>
Status DataTypeNumberSerDe<T>::deserialize_one_cell_from_json(IColumn& column, Slice& slice,
const FormatOptions& options) const {
diff --git a/be/src/core/data_type_serde/data_type_number_serde.h b/be/src/core/data_type_serde/data_type_number_serde.h
index b57f9f9d212..0e0a3acfc1a 100644
--- a/be/src/core/data_type_serde/data_type_number_serde.h
+++ b/be/src/core/data_type_serde/data_type_number_serde.h
@@ -117,6 +117,9 @@ public:
Status read_column_from_arrow(IColumn& column, const arrow::Array* arrow_array, int64_t start,
int64_t end, const cctz::time_zone& ctz) const override;
+ Status read_column_from_decoded_values(IColumn& column,
+ const DecodedColumnView& view) const override;
+
Status write_column_to_mysql_binary(const IColumn& column, MysqlRowBinaryBuffer& row_buffer,
int64_t row_idx, bool col_const,
const FormatOptions& options) const override;
diff --git a/be/src/core/data_type_serde/data_type_serde.cpp b/be/src/core/data_type_serde/data_type_serde.cpp
index ac688ae6c30..b6a49524887 100644
--- a/be/src/core/data_type_serde/data_type_serde.cpp
+++ b/be/src/core/data_type_serde/data_type_serde.cpp
@@ -34,6 +34,12 @@
namespace doris {
DataTypeSerDe::~DataTypeSerDe() = default;
+Status DataTypeSerDe::read_column_from_decoded_values(IColumn& column,
+ const DecodedColumnView& view) const {
+ return Status::NotSupported("read_column_from_decoded_values is not supported for {}",
+ get_name());
+}
+
DataTypeSerDeSPtrs create_data_type_serdes(const DataTypes& types) {
DataTypeSerDeSPtrs serdes;
serdes.reserve(types.size());
diff --git a/be/src/core/data_type_serde/data_type_serde.h b/be/src/core/data_type_serde/data_type_serde.h
index 7c007c6558d..07ad5d5d1b0 100644
--- a/be/src/core/data_type_serde/data_type_serde.h
+++ b/be/src/core/data_type_serde/data_type_serde.h
@@ -27,6 +27,7 @@
#include "common/cast_set.h"
#include "common/status.h"
#include "core/column/column_nullable.h"
+#include "core/data_type_serde/decoded_column_view.h"
#include "core/field.h"
#include "core/string_buffer.hpp"
#include "core/types.h"
@@ -485,6 +486,12 @@ public:
int64_t start, int64_t end,
const cctz::time_zone& ctz) const = 0;
+ // Read already decoded column values into a Doris column. The input view is format-neutral:
+ // file readers translate their decoder output into DecodedColumnView, while SerDe owns
+ // the Doris-type-specific materialization into IColumn.
+ virtual Status read_column_from_decoded_values(IColumn& column,
+ const DecodedColumnView& view) const;
+
// ORC serializer
virtual Status write_column_to_orc(const std::string& timezone, const IColumn& column,
const NullMap* null_map,
diff --git a/be/src/core/data_type_serde/data_type_string_serde.cpp b/be/src/core/data_type_serde/data_type_string_serde.cpp
index b7a59b3c07e..478cdf3b5e6 100644
--- a/be/src/core/data_type_serde/data_type_string_serde.cpp
+++ b/be/src/core/data_type_serde/data_type_string_serde.cpp
@@ -18,12 +18,29 @@
#include "core/data_type_serde/data_type_string_serde.h"
#include "core/column/column_string.h"
+#include "core/data_type_serde/decoded_column_view.h"
#include "core/data_type/define_primitive_type.h"
#include "util/jsonb_document_cast.h"
#include "util/jsonb_utils.h"
#include "util/jsonb_writer.h"
namespace doris {
+namespace {
+
+template <typename ColumnType>
+Status read_string_decoded_values(IColumn& column, const DecodedColumnView& view) {
+ if (view.binary_values == nullptr && view.row_count > 0) {
+ return Status::Corruption("Decoded binary values are null for {}", column.get_name());
+ }
+ auto& string_column = assert_cast<ColumnType&>(column);
+ for (int64_t row = 0; row < view.row_count; ++row) {
+ const auto& value = (*view.binary_values)[row];
+ string_column.insert_data(value.data, value.size);
+ }
+ return Status::OK();
+}
+
+} // namespace
template <typename ColumnType>
Status DataTypeStringSerDeBase<ColumnType>::serialize_column_to_json(const IColumn& column,
@@ -313,6 +330,17 @@ Status DataTypeStringSerDeBase<ColumnType>::read_column_from_arrow(
return Status::OK();
}
+template <typename ColumnType>
+Status DataTypeStringSerDeBase<ColumnType>::read_column_from_decoded_values(
+ IColumn& column, const DecodedColumnView& view) const {
+ if (view.value_kind != DecodedValueKind::BINARY &&
+ view.value_kind != DecodedValueKind::FIXED_BINARY) {
+ return Status::NotSupported("Unsupported decoded values for {} from source kind {}",
+ get_name(), static_cast<int>(view.value_kind));
+ }
+ return read_string_decoded_values<ColumnType>(column, view);
+}
+
template <typename ColumnType>
Status DataTypeStringSerDeBase<ColumnType>::write_column_to_orc(
const std::string& timezone, const IColumn& column, const NullMap* null_map,
diff --git a/be/src/core/data_type_serde/data_type_string_serde.h b/be/src/core/data_type_serde/data_type_string_serde.h
index 79c8450835d..81b80eab4a5 100644
--- a/be/src/core/data_type_serde/data_type_string_serde.h
+++ b/be/src/core/data_type_serde/data_type_string_serde.h
@@ -203,6 +203,9 @@ public:
Status read_column_from_arrow(IColumn& column, const arrow::Array* arrow_array, int64_t start,
int64_t end, const cctz::time_zone& ctz) const override;
+ Status read_column_from_decoded_values(IColumn& column,
+ const DecodedColumnView& view) const override;
+
Status write_column_to_mysql_binary(const IColumn& column, MysqlRowBinaryBuffer& result,
int64_t row_idx, bool col_const,
const FormatOptions& options) const override {
diff --git a/be/src/core/data_type_serde/data_type_time_serde.cpp b/be/src/core/data_type_serde/data_type_time_serde.cpp
index e57fd08a271..65e1afa577d 100644
--- a/be/src/core/data_type_serde/data_type_time_serde.cpp
+++ b/be/src/core/data_type_serde/data_type_time_serde.cpp
@@ -17,6 +17,7 @@
#include "core/data_type_serde/data_type_time_serde.h"
+#include "core/data_type_serde/decoded_column_view.h"
#include "core/data_type/data_type_decimal.h"
#include "core/data_type/data_type_number.h"
#include "core/data_type/primitive_type.h"
@@ -25,6 +26,33 @@
#include "exprs/function/cast/cast_to_time_impl.hpp"
namespace doris {
+namespace {
+
+TimeValue::TimeType read_time_decoded_value(const DecodedColumnView& view, int64_t row) {
+ int64_t micros = 0;
+ if (view.value_kind == DecodedValueKind::INT32) {
+ const auto* values = reinterpret_cast<const int32_t*>(view.values);
+ micros = static_cast<int64_t>(values[row]) * 1000;
+ } else {
+ const auto* values = reinterpret_cast<const int64_t*>(view.values);
+ micros = values[row];
+ if (view.time_unit == DecodedTimeUnit::MILLIS) {
+ micros *= 1000;
+ } else if (view.time_unit == DecodedTimeUnit::NANOS) {
+ micros /= 1000;
+ }
+ }
+ const bool negative = micros < 0;
+ const int64_t abs_micros = std::abs(micros);
+ return TimeValue::make_time(abs_micros / TimeValue::ONE_HOUR_MICROSECONDS,
+ (abs_micros % TimeValue::ONE_HOUR_MICROSECONDS) /
+ TimeValue::ONE_MINUTE_MICROSECONDS,
+ (abs_micros % TimeValue::ONE_MINUTE_MICROSECONDS) /
+ TimeValue::ONE_SECOND_MICROSECONDS,
+ abs_micros % TimeValue::ONE_SECOND_MICROSECONDS, negative);
+}
+
+} // namespace
Status DataTypeTimeV2SerDe::write_column_to_mysql_binary(const IColumn& column,
MysqlRowBinaryBuffer& result,
@@ -145,6 +173,22 @@ Status DataTypeTimeV2SerDe::from_string_strict_mode(StringRef& str, IColumn& col
return Status::OK();
}
+Status DataTypeTimeV2SerDe::read_column_from_decoded_values(
+ IColumn& column, const DecodedColumnView& view) const {
+ if (view.value_kind != DecodedValueKind::INT32 &&
+ view.value_kind != DecodedValueKind::INT64) {
+ return Status::NotSupported("TIMEV2 decoded reader expects INT32 or INT64 source");
+ }
+ if (view.values == nullptr && view.row_count > 0) {
+ return Status::Corruption("Decoded value buffer is null for {}", column.get_name());
+ }
+ auto& data = assert_cast<ColumnTimeV2&>(column).get_data();
+ for (int64_t row = 0; row < view.row_count; ++row) {
+ data.push_back(read_time_decoded_value(view, row));
+ }
+ return Status::OK();
+}
+
template <typename IntDataType>
Status DataTypeTimeV2SerDe::from_int_batch(const typename IntDataType::ColumnType& int_col,
ColumnNullable& target_col) const {
diff --git a/be/src/core/data_type_serde/data_type_time_serde.h b/be/src/core/data_type_serde/data_type_time_serde.h
index db703616b49..e3fccf379c9 100644
--- a/be/src/core/data_type_serde/data_type_time_serde.h
+++ b/be/src/core/data_type_serde/data_type_time_serde.h
@@ -67,6 +67,8 @@ public:
template <typename DecimalDataType>
Status from_decimal_strict_mode_batch(const typename DecimalDataType::ColumnType& decimal_col,
IColumn& target_col) const;
+ Status read_column_from_decoded_values(IColumn& column,
+ const DecodedColumnView& view) const override;
int get_scale() const override { return _scale; }
protected:
diff --git a/be/src/core/data_type_serde/decoded_column_view.h b/be/src/core/data_type_serde/decoded_column_view.h
new file mode 100644
index 00000000000..9b0b14b17c7
--- /dev/null
+++ b/be/src/core/data_type_serde/decoded_column_view.h
@@ -0,0 +1,62 @@
+// 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.
+
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+#include <vector>
+
+#include "common/status.h"
+#include "core/string_ref.h"
+
+namespace doris {
+
+class IColumn;
+
+// 已解码 column batch 的物理值来源类型。
+// 该枚举只描述通用内存布局，不包含 Parquet/ORC/Arrow 等格式专有类型。
+enum class DecodedValueKind {
+ BOOL,
+ INT32,
+ INT64,
+ FLOAT,
+ DOUBLE,
+ BINARY,
+ FIXED_BINARY,
+};
+
+enum class DecodedTimeUnit {
+ UNKNOWN,
+ MILLIS,
+ MICROS,
+ NANOS,
+};
+
+struct DecodedColumnView {
+ DecodedValueKind value_kind = DecodedValueKind::INT32;
+ DecodedTimeUnit time_unit = DecodedTimeUnit::UNKNOWN;
+ int64_t row_count = 0;
+ int decimal_precision = -1;
+ int decimal_scale = -1;
+ int fixed_length = -1;
+ const uint8_t* values = nullptr;
+ const uint8_t* null_map = nullptr;
+ const std::vector<StringRef>* binary_values = nullptr;
+};
+
+} // namespace doris
diff --git a/be/src/exprs/vslot_ref.h b/be/src/exprs/vslot_ref.h
index ceb702728eb..6e7197f4cf6 100644
--- a/be/src/exprs/vslot_ref.h
+++ b/be/src/exprs/vslot_ref.h
@@ -67,7 +67,7 @@ public:
column_ids.insert(_column_id);
}
- MOCK_FUNCTION const std::string& column_name() const { return *_column_name; }
+ virtual const std::string& column_name() const { return *_column_name; }
uint64_t get_digest(uint64_t seed) const override;
diff --git a/be/src/format/new_parquet/column_reader.cpp b/be/src/format/new_parquet/column_reader.cpp
new file mode 100644
index 00000000000..bb101b5c5fc
--- /dev/null
+++ b/be/src/format/new_parquet/column_reader.cpp
@@ -0,0 +1,562 @@
+// 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/new_parquet/column_reader.h"
+
+#include <arrow/array/array_binary.h>
+#include <parquet/api/reader.h>
+#include <parquet/api/schema.h>
+
+#include <algorithm>
+#include <cstddef>
+#include <cstdint>
+#include <exception>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "core/column/column.h"
+#include "core/column/column_struct.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type_serde/decoded_column_view.h"
+#include "format/new_parquet/parquet_column_schema.h"
+
+namespace doris::parquet {
+namespace {
+
+class ScalarColumnReader final : public ParquetColumnReader {
+public:
+ ScalarColumnReader(int parquet_leaf_column_id, const ::parquet::ColumnDescriptor* descriptor,
+ ParquetTypeDescriptor type_descriptor, DataTypePtr type, std::string name,
+ std::shared_ptr<::parquet::internal::RecordReader> record_reader)
+ : _file_column_id(parquet_leaf_column_id),
+ _parquet_leaf_column_id(parquet_leaf_column_id),
+ _descriptor(descriptor),
+ _type_descriptor(std::move(type_descriptor)),
+ _type(std::move(type)),
+ _name(std::move(name)),
+ _record_reader(std::move(record_reader)) {}
+
+ int file_column_id() const override { return _file_column_id; }
+ int parquet_leaf_column_id() const override { return _parquet_leaf_column_id; }
+ const DataTypePtr& type() const override { return _type; }
+ const std::string& name() const override { return _name; }
+
+ Status read(int64_t rows, MutableColumnPtr& column, int64_t* rows_read) override;
+ Status skip(int64_t rows) override;
+
+ const ::parquet::ColumnDescriptor* descriptor() const { return _descriptor; }
+ const std::shared_ptr<::parquet::internal::RecordReader>& record_reader() const {
+ return _record_reader;
+ }
+
+private:
+ int _file_column_id = -1;
+ int _parquet_leaf_column_id = -1;
+ const ::parquet::ColumnDescriptor* _descriptor = nullptr;
+ ParquetTypeDescriptor _type_descriptor;
+ DataTypePtr _type;
+ std::string _name;
+ std::shared_ptr<::parquet::internal::RecordReader> _record_reader;
+};
+
+class StructColumnReader final : public ParquetColumnReader {
+public:
+ StructColumnReader(const ParquetColumnSchema& schema,
+ std::vector<std::unique_ptr<ParquetColumnReader>> children)
+ : _field_id(schema.field_id),
+ _type(schema.type),
+ _name(schema.name),
+ _children(std::move(children)) {}
+
+ int file_column_id() const override { return _field_id; }
+ int parquet_leaf_column_id() const override { return -1; }
+ const DataTypePtr& type() const override { return _type; }
+ const std::string& name() const override { return _name; }
+
+ Status read(int64_t rows, MutableColumnPtr& column, int64_t* rows_read) override;
+ Status skip(int64_t rows) override;
+
+private:
+ int _field_id = -1;
+ DataTypePtr _type;
+ std::string _name;
+ std::vector<std::unique_ptr<ParquetColumnReader>> _children;
+};
+
+Status read_records(ScalarColumnReader& column_reader, int64_t batch_rows,
+ ::parquet::internal::RecordReader** record_reader, int64_t* rows_read) {
+ auto reader = column_reader.record_reader();
+ if (reader == nullptr) {
+ return Status::InternalError("Parquet record reader is not initialized for column {}",
+ column_reader.name());
+ }
+
+ int64_t records_read = 0;
+ try {
+ reader->Reset();
+ reader->Reserve(batch_rows);
+ records_read = reader->ReadRecords(batch_rows);
+ } catch (const ::parquet::ParquetException& e) {
+ return Status::Corruption("Failed to read parquet records for column {}: {}",
+ column_reader.name(), e.what());
+ } catch (const std::exception& e) {
+ return Status::InternalError("Failed to read parquet records for column {}: {}",
+ column_reader.name(), e.what());
+ }
+ if (records_read < 0 || records_read > batch_rows) {
+ return Status::Corruption("Invalid parquet record read result for column {}: {}",
+ column_reader.name(), records_read);
+ }
+ *record_reader = reader.get();
+ *rows_read = records_read;
+ return Status::OK();
+}
+
+struct RowRange {
+ int64_t start = 0;
+ int64_t length = 0;
+};
+
+std::vector<RowRange> selection_to_ranges(const SelectionVector& selection,
+ uint16_t selected_rows) {
+ std::vector<RowRange> ranges;
+ if (selected_rows == 0) {
+ return ranges;
+ }
+
+ int64_t range_start = selection.get_index(0);
+ int64_t previous = selection.get_index(0);
+ for (uint16_t selection_idx = 1; selection_idx < selected_rows; ++selection_idx) {
+ const int64_t current = selection.get_index(selection_idx);
+ DCHECK_GT(current, previous);
+ if (current == previous + 1) {
+ previous = current;
+ continue;
+ }
+ ranges.push_back(RowRange {range_start, previous - range_start + 1});
+ range_start = current;
+ previous = current;
+ }
+ ranges.push_back(RowRange {range_start, previous - range_start + 1});
+ return ranges;
+}
+
+DecodedTimeUnit decoded_time_unit(ParquetTimeUnit time_unit) {
+ switch (time_unit) {
+ case ParquetTimeUnit::MILLIS:
+ return DecodedTimeUnit::MILLIS;
+ case ParquetTimeUnit::MICROS:
+ return DecodedTimeUnit::MICROS;
+ case ParquetTimeUnit::NANOS:
+ return DecodedTimeUnit::NANOS;
+ case ParquetTimeUnit::UNKNOWN:
+ default:
+ return DecodedTimeUnit::UNKNOWN;
+ }
+}
+
+DecodedValueKind decoded_value_kind(const ParquetTypeDescriptor& type_descriptor) {
+ switch (type_descriptor.physical_type) {
+ case ::parquet::Type::BOOLEAN:
+ return DecodedValueKind::BOOL;
+ case ::parquet::Type::INT32:
+ return DecodedValueKind::INT32;
+ case ::parquet::Type::INT64:
+ return DecodedValueKind::INT64;
+ case ::parquet::Type::FLOAT:
+ return DecodedValueKind::FLOAT;
+ case ::parquet::Type::DOUBLE:
+ return DecodedValueKind::DOUBLE;
+ case ::parquet::Type::FIXED_LEN_BYTE_ARRAY:
+ return DecodedValueKind::FIXED_BINARY;
+ case ::parquet::Type::BYTE_ARRAY:
+ default:
+ return DecodedValueKind::BINARY;
+ }
+}
+
+Status build_null_map(const ScalarColumnReader& column_reader,
+ ::parquet::internal::RecordReader& record_reader, int64_t records_read,
+ NullMap* null_map) {
+ if (column_reader.descriptor()->max_definition_level() == 0) {
+ return Status::OK();
+ }
+ if (record_reader.read_dense_for_nullable()) {
+ return Status::NotSupported(
+ "Dense nullable parquet record reader is not supported for column {}",
+ column_reader.name());
+ }
+ auto* def_levels = record_reader.def_levels();
+ if (def_levels == nullptr && records_read > 0) {
+ return Status::Corruption(
+ "Parquet record reader returned null definition levels for nullable column {}",
+ column_reader.name());
+ }
+ const int16_t max_definition_level = column_reader.descriptor()->max_definition_level();
+ null_map->resize(records_read);
+ auto* __restrict dst = null_map->data();
+ const auto* __restrict src = def_levels;
+ for (int64_t record_idx = 0; record_idx < records_read; ++record_idx) {
+ dst[record_idx] = src[record_idx] != max_definition_level;
+ }
+ return Status::OK();
+}
+
+Status get_binary_chunks(const ScalarColumnReader& column_reader,
+ ::parquet::internal::RecordReader& record_reader,
+ std::vector<std::shared_ptr<::arrow::Array>>* chunks) {
+ auto* binary_reader = dynamic_cast<::parquet::internal::BinaryRecordReader*>(&record_reader);
+ if (binary_reader == nullptr) {
+ return Status::InternalError("Parquet binary record reader is not available for column {}",
+ column_reader.name());
+ }
+ *chunks = binary_reader->GetBuilderChunks();
+ return Status::OK();
+}
+
+Status build_binary_values(const ScalarColumnReader& column_reader,
+ const std::vector<std::shared_ptr<::arrow::Array>>& chunks,
+ int64_t records_read, std::vector<StringRef>* binary_values) {
+ binary_values->reserve(records_read);
+ for (const auto& chunk : chunks) {
+ if (chunk == nullptr) {
+ return Status::Corruption(
+ "Parquet binary record reader returned null chunk for column {}",
+ column_reader.name());
+ }
+ if (auto* binary_array = dynamic_cast<::arrow::BinaryArray*>(chunk.get())) {
+ for (int64_t row_idx = 0; row_idx < binary_array->length(); ++row_idx) {
+ if (binary_array->IsNull(row_idx)) {
+ binary_values->emplace_back(static_cast<const char*>(nullptr), 0);
+ continue;
+ }
+ int32_t length = 0;
+ const uint8_t* value = binary_array->GetValue(row_idx, &length);
+ binary_values->emplace_back(reinterpret_cast<const char*>(value), length);
+ }
+ } else if (auto* fixed_array = dynamic_cast<::arrow::FixedSizeBinaryArray*>(chunk.get())) {
+ for (int64_t row_idx = 0; row_idx < fixed_array->length(); ++row_idx) {
+ if (fixed_array->IsNull(row_idx)) {
+ binary_values->emplace_back(static_cast<const char*>(nullptr), 0);
+ continue;
+ }
+ binary_values->emplace_back(
+ reinterpret_cast<const char*>(fixed_array->GetValue(row_idx)),
+ fixed_array->byte_width());
+ }
+ } else {
+ return Status::InternalError("Unexpected Arrow binary array type for column {}",
+ column_reader.name());
+ }
+ }
+ if (binary_values->size() != static_cast<size_t>(records_read)) {
+ return Status::Corruption(
+ "Invalid parquet binary record read result for column {}: rows={}, records={}",
+ column_reader.name(), binary_values->size(), records_read);
+ }
+ return Status::OK();
+}
+
+} // namespace
+
+Status ScalarColumnReader::read(int64_t rows, MutableColumnPtr& column, int64_t* rows_read) {
+ if (column.get() == nullptr || rows_read == nullptr) {
+ return Status::InvalidArgument("Invalid parquet column read result pointer for column {}",
+ _name);
+ }
+ if (_record_reader == nullptr) {
+ return Status::InternalError("Parquet record reader is not initialized for column {}",
+ _name);
+ }
+ ::parquet::internal::RecordReader* record_reader = nullptr;
+ RETURN_IF_ERROR(read_records(*this, rows, &record_reader, rows_read));
+ if (record_reader->values_written() != *rows_read) {
+ return Status::Corruption(
+ "Invalid parquet record read result for column {}: values={}, records={}", _name,
+ record_reader->values_written(), *rows_read);
+ }
+
+ NullMap null_map;
+ RETURN_IF_ERROR(build_null_map(*this, *record_reader, *rows_read, &null_map));
+
+ std::vector<StringRef> binary_values;
+ std::vector<std::shared_ptr<::arrow::Array>> binary_chunks;
+ DecodedColumnView view;
+ view.value_kind = decoded_value_kind(_type_descriptor);
+ view.time_unit = decoded_time_unit(_type_descriptor.time_unit);
+ view.row_count = *rows_read;
+ view.decimal_precision = _type_descriptor.decimal_precision;
+ view.decimal_scale = _type_descriptor.decimal_scale;
+ view.fixed_length = _type_descriptor.fixed_length;
+ view.null_map = null_map.empty() ? nullptr : null_map.data();
+ if (view.value_kind == DecodedValueKind::BINARY ||
+ view.value_kind == DecodedValueKind::FIXED_BINARY) {
+ RETURN_IF_ERROR(get_binary_chunks(*this, *record_reader, &binary_chunks));
+ RETURN_IF_ERROR(build_binary_values(*this, binary_chunks, *rows_read, &binary_values));
+ view.binary_values = &binary_values;
+ } else {
+ view.values = record_reader->values();
+ }
+
+ RETURN_IF_ERROR(_type->get_serde()->read_column_from_decoded_values(*column, view));
+ return Status::OK();
+}
+
+Status ScalarColumnReader::skip(int64_t rows) {
+ if (rows <= 0) {
+ return Status::OK();
+ }
+
+ if (_record_reader == nullptr) {
+ return Status::InternalError("Parquet record reader is not initialized for column {}",
+ _name);
+ }
+ int64_t skipped_rows = 0;
+ try {
+ _record_reader->Reset();
+ while (skipped_rows < rows) {
+ const int64_t skipped = _record_reader->SkipRecords(rows - skipped_rows);
+ if (skipped <= 0) {
+ return Status::Corruption(
+ "Failed to skip parquet records for column {}: skipped {} of {} rows",
+ _name, skipped_rows, rows);
+ }
+ skipped_rows += skipped;
+ }
+ } catch (const ::parquet::ParquetException& e) {
+ return Status::Corruption("Failed to skip parquet records for column {}: {}", _name,
+ e.what());
+ } catch (const std::exception& e) {
+ return Status::InternalError("Failed to skip parquet records for column {}: {}", _name,
+ e.what());
+ }
+ return Status::OK();
+}
+
+Status StructColumnReader::read(int64_t rows, MutableColumnPtr& column, int64_t* rows_read) {
+ if (column.get() == nullptr || rows_read == nullptr) {
+ return Status::InvalidArgument("Invalid parquet struct read result pointer for column {}",
+ _name);
+ }
+ if (_children.empty()) {
+ column->resize(static_cast<size_t>(rows));
+ *rows_read = rows;
+ return Status::OK();
+ }
+
+ int64_t expected_rows = -1;
+ size_t child_idx = 0;
+ DCHECK_EQ(assert_cast<ColumnStruct&>(*column).get_columns().size(), _children.size());
+ for (auto& child_reader : _children) {
+ int64_t child_rows = 0;
+ auto child_column =
+ assert_cast<ColumnStruct&>(*column).get_column_ptr(child_idx)->assume_mutable();
+ RETURN_IF_ERROR(child_reader->read(rows, child_column, &child_rows));
+ if (expected_rows < 0) {
+ expected_rows = child_rows;
+ } else if (child_rows != expected_rows) {
+ return Status::Corruption(
+ "Parquet struct children returned different row counts in column {}: {} vs {}",
+ _name, expected_rows, child_rows);
+ }
+ child_idx++;
+ }
+
+ *rows_read = std::max<int64_t>(expected_rows, 0);
+ return Status::OK();
+}
+
+Status StructColumnReader::skip(int64_t rows) {
+ if (rows <= 0) {
+ return Status::OK();
+ }
+ for (auto& child_reader : _children) {
+ RETURN_IF_ERROR(child_reader->skip(rows));
+ }
+ return Status::OK();
+}
+
+Status ParquetColumnReader::skip(int64_t rows) {
+ return Status::NotSupported("Parquet column skip is not implemented, rows={}", rows);
+}
+
+Status ParquetColumnReader::select(const SelectionVector& sel, uint16_t selected_rows,
+ int64_t batch_rows, MutableColumnPtr& column) {
+ if (column.get() == nullptr) {
+ return Status::InvalidArgument("Parquet selected read result is null for column {}",
+ name());
+ }
+ RETURN_IF_ERROR(sel.verify(selected_rows, batch_rows));
+
+ const auto ranges = selection_to_ranges(sel, selected_rows);
+ int64_t cursor = 0;
+ for (const auto& range : ranges) {
+ if (range.start < cursor || range.start + range.length > batch_rows) {
+ return Status::InvalidArgument("Invalid parquet selection range [{}, {}) for column {}",
+ range.start, range.start + range.length, name());
+ }
+ RETURN_IF_ERROR(skip(range.start - cursor));
+
+ int64_t range_rows_read = 0;
+ RETURN_IF_ERROR(read(range.length, column, &range_rows_read));
+ if (range_rows_read != range.length) {
+ return Status::Corruption(
+ "Parquet selected read returned {} rows, expected {} rows for column {}",
+ range_rows_read, range.length, name());
+ }
+ cursor = range.start + range.length;
+ }
+ RETURN_IF_ERROR(skip(batch_rows - cursor));
+ return Status::OK();
+}
+
+ParquetColumnReaderFactory::ParquetColumnReaderFactory(
+ std::shared_ptr<::parquet::RowGroupReader> row_group, int num_leaf_columns)
+ : _row_group(std::move(row_group)),
+ _record_readers(static_cast<size_t>(num_leaf_columns)) {}
+
+Status ParquetColumnReaderFactory::create_scalar_reader(
+ int parquet_leaf_column_id, const ParquetTypeDescriptor& type_descriptor,
+ const ::parquet::ColumnDescriptor* descriptor, DataTypePtr type, std::string name,
+ std::shared_ptr<::parquet::internal::RecordReader> record_reader,
+ std::unique_ptr<ParquetColumnReader>* reader) const {
+ if (reader == nullptr) {
+ return Status::InvalidArgument("reader is null");
+ }
+ if (descriptor == nullptr || type == nullptr || record_reader == nullptr) {
+ return Status::InvalidArgument("Invalid parquet column reader arguments for column {}",
+ name);
+ }
+ *reader = std::make_unique<ScalarColumnReader>(parquet_leaf_column_id, descriptor,
+ type_descriptor, std::move(type),
+ std::move(name), std::move(record_reader));
+ return Status::OK();
+}
+
+Status ParquetColumnReaderFactory::create_scalar_column_reader(
+ const ParquetColumnSchema& column_schema,
+ std::unique_ptr<ParquetColumnReader>* reader) const {
+ if (reader == nullptr) {
+ return Status::InvalidArgument("reader is null");
+ }
+ if (column_schema.leaf_column_id < 0 ||
+ column_schema.leaf_column_id >= static_cast<int>(_record_readers.size())) {
+ return Status::InvalidArgument("Invalid parquet leaf column id {} for column {}",
+ column_schema.leaf_column_id, column_schema.name);
+ }
+ if (!supports_record_reader(column_schema.type_descriptor)) {
+ return Status::NotSupported(
+ "Current parquet reader only supports primitive columns without repetition; "
+ "column {} is not supported",
+ column_schema.name);
+ }
+ std::shared_ptr<::parquet::internal::RecordReader> record_reader;
+ RETURN_IF_ERROR(get_record_reader(column_schema.leaf_column_id, column_schema.descriptor,
+ column_schema.name, &record_reader));
+ return create_scalar_reader(column_schema.leaf_column_id, column_schema.type_descriptor,
+ column_schema.descriptor, column_schema.type, column_schema.name,
+ std::move(record_reader), reader);
+}
+
+Status ParquetColumnReaderFactory::get_record_reader(
+ int leaf_column_id, const ::parquet::ColumnDescriptor* descriptor, const std::string& name,
+ std::shared_ptr<::parquet::internal::RecordReader>* reader) const {
+ if (reader == nullptr) {
+ return Status::InvalidArgument("reader is null");
+ }
+ if (_row_group == nullptr) {
+ return Status::InternalError("Parquet row group reader is not initialized for column {}",
+ name);
+ }
+ if (leaf_column_id < 0 || leaf_column_id >= static_cast<int>(_record_readers.size())) {
+ return Status::InvalidArgument("Invalid parquet leaf column id {} for column {}",
+ leaf_column_id, name);
+ }
+ if (descriptor == nullptr) {
+ return Status::InvalidArgument("Parquet column descriptor is null for column {}", name);
+ }
+ if (descriptor->max_repetition_level() != 0 || descriptor->max_definition_level() > 1) {
+ return Status::NotSupported(
+ "Current parquet reader only supports RecordReader-backed columns; column {} is "
+ "not supported",
+ name);
+ }
+ if (_record_readers[leaf_column_id] == nullptr) {
+ try {
+ _record_readers[leaf_column_id] =
+ _row_group->RecordReader(leaf_column_id, /*read_dictionary=*/false);
+ } catch (const ::parquet::ParquetException& e) {
+ return Status::Corruption("Failed to create parquet record reader for column {}: {}",
+ name, e.what());
+ } catch (const std::exception& e) {
+ return Status::InternalError("Failed to create parquet record reader for column {}: {}",
+ name, e.what());
+ }
+ }
+ if (_record_readers[leaf_column_id] == nullptr) {
+ return Status::Corruption("Failed to create parquet record reader for column {}", name);
+ }
+ *reader = _record_readers[leaf_column_id];
+ return Status::OK();
+}
+
+Status ParquetColumnReaderFactory::create_struct_column_reader(
+ const ParquetColumnSchema& column_schema,
+ std::unique_ptr<ParquetColumnReader>* reader) const {
+ if (reader == nullptr) {
+ return Status::InvalidArgument("reader is null");
+ }
+ if (column_schema.type != nullptr && column_schema.type->is_nullable()) {
+ return Status::NotSupported(
+ "Nullable parquet STRUCT reader is not implemented for column {}",
+ column_schema.name);
+ }
+ std::vector<std::unique_ptr<ParquetColumnReader>> child_readers;
+ child_readers.reserve(column_schema.children.size());
+ for (const auto& child_schema : column_schema.children) {
+ std::unique_ptr<ParquetColumnReader> child_reader;
+ RETURN_IF_ERROR(create(*child_schema, &child_reader));
+ child_readers.push_back(std::move(child_reader));
+ }
+ *reader = std::make_unique<StructColumnReader>(column_schema, std::move(child_readers));
+ return Status::OK();
+}
+
+Status ParquetColumnReaderFactory::create(const ParquetColumnSchema& column_schema,
+ std::unique_ptr<ParquetColumnReader>* reader) const {
+ if (reader == nullptr) {
+ return Status::InvalidArgument("reader is null");
+ }
+ switch (column_schema.kind) {
+ case ParquetColumnSchemaKind::PRIMITIVE:
+ return create_scalar_column_reader(column_schema, reader);
+ case ParquetColumnSchemaKind::STRUCT:
+ return create_struct_column_reader(column_schema, reader);
+ case ParquetColumnSchemaKind::LIST:
+ return Status::NotSupported("Parquet LIST reader is not implemented for column {}",
+ column_schema.name);
+ case ParquetColumnSchemaKind::MAP:
+ return Status::NotSupported("Parquet MAP reader is not implemented for column {}",
+ column_schema.name);
+ }
+ return Status::NotSupported("Unsupported parquet column schema kind for column {}",
+ column_schema.name);
+}
+
+} // namespace doris::parquet
diff --git a/be/src/format/new_parquet/column_reader.h b/be/src/format/new_parquet/column_reader.h
new file mode 100644
index 00000000000..cd59fc3960a
--- /dev/null
+++ b/be/src/format/new_parquet/column_reader.h
@@ -0,0 +1,116 @@
+// 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.
+
+#pragma once
+
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "common/status.h"
+#include "core/data_type/data_type.h"
+#include "format/new_parquet/parquet_type.h"
+#include "format/new_parquet/selection_vector.h"
+
+namespace parquet {
+class ColumnDescriptor;
+class RowGroupReader;
+
+namespace internal {
+class RecordReader;
+} // namespace internal
+} // namespace parquet
+
+namespace doris {
+class IColumn;
+
+namespace parquet {
+struct ParquetColumnSchema;
+
+// Doris 的 Parquet column reader 抽象。
+// 该类包装 Arrow Parquet RecordReader，负责将 file-local Parquet leaf column 读取成
+// Doris-owned column。它不理解 Iceberg/global schema，也不处理 table-level
+// cast/default/generated/partition 语义。
+class ParquetColumnReader {
+public:
+ virtual ~ParquetColumnReader() = default;
+
+ // FileReader 暴露给上层 scan request 的 file-local column id。
+ // 对 top-level primitive 列，它通常等于 Parquet leaf column id；对 struct/list/map
+ // 这类复杂列，它表示 file schema tree 中的逻辑字段 id。
+ virtual int file_column_id() const = 0;
+
+ // Parquet 文件内部的 leaf column id，用于访问 RowGroupReader::RecordReader、
+ // ColumnChunk metadata、statistics/page index 等 Parquet 物理列结构。
+ // 只有 primitive leaf reader 有有效值；复杂列 reader 没有单一 leaf column，返回 -1。
+ virtual int parquet_leaf_column_id() const = 0;
+
+ virtual const DataTypePtr& type() const = 0;
+ virtual const std::string& name() const = 0;
+
+ // 读取一个 file-local column batch。
+ virtual Status read(int64_t rows, MutableColumnPtr& column, int64_t* rows_read) = 0;
+
+ // 跳过指定行数。这里必须使用 row-level skip，不能退回到 value-level Skip。
+ virtual Status skip(int64_t rows);
+
+ // 按 selection 读取当前 batch 中需要输出的行，并在末尾跳过 batch 内剩余行。
+ // 该方法只允许通过 skip + read 推进 reader 游标，不允许退化为整批 read + filter。
+ virtual Status select(const SelectionVector& sel, uint16_t selected_rows, int64_t batch_rows,
+ MutableColumnPtr& column);
+};
+
+// Parquet column reader 工厂。
+// 工厂绑定当前 row group，并根据 file-local schema tree 创建 Doris 自己的 column
+// reader。Arrow internal RecordReader 的创建和缓存必须封装在这里，避免泄露到
+// ParquetReader 主流程。后续 reader options、Dremel assembler、延时物化 cache/skip
+// 策略都应挂在该工厂上下文里，而不是继续扩展自由函数参数。
+class ParquetColumnReaderFactory {
+public:
+ ParquetColumnReaderFactory(std::shared_ptr<::parquet::RowGroupReader> row_group,
+ int num_leaf_columns);
+
+ // 根据 file-local schema tree 创建 column reader。复杂类型会在这里递归创建
+ // children。该入口只理解 Parquet file schema，不处理 table/global schema。
+ Status create(const ParquetColumnSchema& column_schema,
+ std::unique_ptr<ParquetColumnReader>* reader) const;
+
+private:
+ Status create_scalar_column_reader(const ParquetColumnSchema& column_schema,
+ std::unique_ptr<ParquetColumnReader>* reader) const;
+
+ Status create_struct_column_reader(const ParquetColumnSchema& column_schema,
+ std::unique_ptr<ParquetColumnReader>* reader) const;
+
+ Status get_record_reader(int leaf_column_id, const ::parquet::ColumnDescriptor* descriptor,
+ const std::string& name,
+ std::shared_ptr<::parquet::internal::RecordReader>* reader) const;
+
+ Status create_scalar_reader(int parquet_leaf_column_id,
+ const ParquetTypeDescriptor& type_descriptor,
+ const ::parquet::ColumnDescriptor* descriptor, DataTypePtr type,
+ std::string name,
+ std::shared_ptr<::parquet::internal::RecordReader> record_reader,
+ std::unique_ptr<ParquetColumnReader>* reader) const;
+
+ std::shared_ptr<::parquet::RowGroupReader> _row_group;
+ mutable std::vector<std::shared_ptr<::parquet::internal::RecordReader>> _record_readers;
+};
+
+} // namespace parquet
+} // namespace doris
diff --git a/be/src/format/new_parquet/parquet_column_schema.cpp b/be/src/format/new_parquet/parquet_column_schema.cpp
new file mode 100644
index 00000000000..131bf9f22c0
--- /dev/null
+++ b/be/src/format/new_parquet/parquet_column_schema.cpp
@@ -0,0 +1,158 @@
+// 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/new_parquet/parquet_column_schema.h"
+
+#include <parquet/api/schema.h>
+
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_map.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_struct.h"
+#include "format/new_parquet/parquet_type.h"
+
+namespace doris::parquet {
+namespace {
+
+bool is_list_node(const ::parquet::schema::Node& node) {
+ const auto& logical_type = node.logical_type();
+ return node.converted_type() == ::parquet::ConvertedType::LIST ||
+ (logical_type != nullptr && logical_type->is_valid() && logical_type->is_list());
+}
+
+bool is_map_node(const ::parquet::schema::Node& node) {
+ const auto& logical_type = node.logical_type();
+ return node.converted_type() == ::parquet::ConvertedType::MAP ||
+ node.converted_type() == ::parquet::ConvertedType::MAP_KEY_VALUE ||
+ (logical_type != nullptr && logical_type->is_valid() && logical_type->is_map());
+}
+
+DataTypePtr nullable_if_needed(DataTypePtr type, const ::parquet::schema::Node& node) {
+ return node.is_optional() ? make_nullable(type) : type;
+}
+
+Status build_node_schema(const ::parquet::SchemaDescriptor& schema,
+ const ::parquet::schema::Node& node,
+ std::unique_ptr<ParquetColumnSchema>* result) {
+ if (result == nullptr) {
+ return Status::InvalidArgument("result is null");
+ }
+ auto column_schema = std::make_unique<ParquetColumnSchema>();
+ column_schema->field_id = node.field_id();
+ column_schema->name = node.name();
+ column_schema->node = &node;
+
+ if (node.is_primitive()) {
+ const int leaf_column_id = schema.ColumnIndex(node);
+ if (leaf_column_id < 0) {
+ return Status::InvalidArgument("Cannot find leaf column id for parquet column {}",
+ node.name());
+ }
+ column_schema->kind = ParquetColumnSchemaKind::PRIMITIVE;
+ column_schema->leaf_column_id = leaf_column_id;
+ column_schema->descriptor = schema.Column(leaf_column_id);
+ column_schema->type_descriptor = resolve_parquet_type(column_schema->descriptor);
+ column_schema->type = column_schema->type_descriptor.doris_type;
+ if (column_schema->type == nullptr) {
+ return Status::NotSupported("Unsupported parquet column type for column {}",
+ node.name());
+ }
+ *result = std::move(column_schema);
+ return Status::OK();
+ }
+
+ const auto& group = static_cast<const ::parquet::schema::GroupNode&>(node);
+ if (is_list_node(node)) {
+ column_schema->kind = ParquetColumnSchemaKind::LIST;
+ if (group.field_count() != 1) {
+ return Status::NotSupported("Unsupported parquet LIST encoding for column {}",
+ node.name());
+ }
+ std::unique_ptr<ParquetColumnSchema> child;
+ RETURN_IF_ERROR(build_node_schema(schema, *group.field(0), &child));
+ column_schema->type =
+ nullable_if_needed(std::make_shared<DataTypeArray>(child->type), node);
+ column_schema->children.push_back(std::move(child));
+ *result = std::move(column_schema);
+ return Status::OK();
+ }
+
+ if (is_map_node(node)) {
+ column_schema->kind = ParquetColumnSchemaKind::MAP;
+ if (group.field_count() != 1) {
+ return Status::NotSupported("Unsupported parquet MAP encoding for column {}",
+ node.name());
+ }
+ std::unique_ptr<ParquetColumnSchema> key_value;
+ RETURN_IF_ERROR(build_node_schema(schema, *group.field(0), &key_value));
+ if (key_value->children.size() != 2) {
+ return Status::NotSupported("Unsupported parquet MAP key_value layout for column {}",
+ node.name());
+ }
+ auto key_type = key_value->children[0]->type;
+ auto value_type = key_value->children[1]->type;
+ column_schema->type =
+ nullable_if_needed(std::make_shared<DataTypeMap>(key_type, value_type), node);
+ column_schema->children.push_back(std::move(key_value));
+ *result = std::move(column_schema);
+ return Status::OK();
+ }
+
+ column_schema->kind = ParquetColumnSchemaKind::STRUCT;
+ DataTypes child_types;
+ Strings child_names;
+ child_types.reserve(group.field_count());
+ child_names.reserve(group.field_count());
+ for (int child_idx = 0; child_idx < group.field_count(); ++child_idx) {
+ std::unique_ptr<ParquetColumnSchema> child;
+ RETURN_IF_ERROR(build_node_schema(schema, *group.field(child_idx), &child));
+ child_types.push_back(child->type);
+ child_names.push_back(child->name);
+ column_schema->children.push_back(std::move(child));
+ }
+ column_schema->type =
+ nullable_if_needed(std::make_shared<DataTypeStruct>(child_types, child_names), node);
+ *result = std::move(column_schema);
+ return Status::OK();
+}
+
+} // namespace
+
+Status build_parquet_column_schema(const ::parquet::SchemaDescriptor& schema,
+ std::vector<std::unique_ptr<ParquetColumnSchema>>* fields) {
+ if (fields == nullptr) {
+ return Status::InvalidArgument("fields is null");
+ }
+ fields->clear();
+ const auto* root = schema.group_node();
+ if (root == nullptr) {
+ return Status::InvalidArgument("Parquet schema root is null");
+ }
+ fields->reserve(root->field_count());
+ for (int field_idx = 0; field_idx < root->field_count(); ++field_idx) {
+ std::unique_ptr<ParquetColumnSchema> field;
+ RETURN_IF_ERROR(build_node_schema(schema, *root->field(field_idx), &field));
+ fields->push_back(std::move(field));
+ }
+ return Status::OK();
+}
+
+} // namespace doris::parquet
diff --git a/be/src/format/new_parquet/parquet_column_schema.h b/be/src/format/new_parquet/parquet_column_schema.h
new file mode 100644
index 00000000000..0d089a0f9cd
--- /dev/null
+++ b/be/src/format/new_parquet/parquet_column_schema.h
@@ -0,0 +1,67 @@
+// 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.
+
+#pragma once
+
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "common/status.h"
+#include "core/data_type/data_type.h"
+#include "format/new_parquet/parquet_type.h"
+
+namespace parquet {
+class ColumnDescriptor;
+class SchemaDescriptor;
+
+namespace schema {
+class Node;
+} // namespace schema
+} // namespace parquet
+
+namespace doris::parquet {
+
+enum class ParquetColumnSchemaKind {
+ PRIMITIVE,
+ STRUCT,
+ LIST,
+ MAP,
+};
+
+// 新 Parquet reader 的 file-local schema tree。
+// 它描述 Parquet 逻辑字段到 leaf column ordinal 的关系，不包含 table/global schema 语义。
+struct ParquetColumnSchema {
+ int field_id = -1;
+ // Parquet schema 中的 primitive leaf column ordinal。
+ // 该 id 用于访问 ColumnDescriptor、RowGroupReader::RecordReader、ColumnChunk
+ // metadata 和 statistics。复杂类型节点本身没有单一 leaf column，因此为 -1。
+ int leaf_column_id = -1;
+ std::string name;
+ DataTypePtr type;
+ ParquetTypeDescriptor type_descriptor;
+ ParquetColumnSchemaKind kind = ParquetColumnSchemaKind::PRIMITIVE;
+ const ::parquet::schema::Node* node = nullptr;
+ const ::parquet::ColumnDescriptor* descriptor = nullptr;
+ std::vector<std::unique_ptr<ParquetColumnSchema>> children;
+};
+
+// 从 Arrow Parquet core schema 构造 file-local schema tree。
+Status build_parquet_column_schema(const ::parquet::SchemaDescriptor& schema,
+ std::vector<std::unique_ptr<ParquetColumnSchema>>* fields);
+
+} // namespace doris::parquet
diff --git a/be/src/format/new_parquet/parquet_reader.cpp b/be/src/format/new_parquet/parquet_reader.cpp
new file mode 100644
index 00000000000..fc00484758e
--- /dev/null
+++ b/be/src/format/new_parquet/parquet_reader.cpp
@@ -0,0 +1,601 @@
+// 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/new_parquet/parquet_reader.h"
+
+#include <arrow/buffer.h>
+#include <arrow/io/interfaces.h>
+#include <arrow/result.h>
+#include <parquet/api/reader.h>
+
+#include <algorithm>
+#include <memory>
+#include <string_view>
+#include <utility>
+#include <vector>
+
+#include "common/exception.h"
+#include "core/block/block.h"
+#include "core/data_type/data_type_nullable.h"
+#include "format/new_parquet/column_reader.h"
+#include "format/new_parquet/parquet_column_schema.h"
+#include "format/new_parquet/parquet_statistics.h"
+#include "format/new_parquet/selection_vector.h"
+#include "io/fs/file_reader.h"
+#include "storage/predicate/column_predicate.h"
+#include "util/slice.h"
+
+namespace doris::parquet {
+
+constexpr int64_t DEFAULT_PARQUET_READ_BATCH_SIZE = 4096;
+
+Status arrow_status_to_doris_status(const arrow::Status& status) {
+ if (status.ok()) {
+ return Status::OK();
+ }
+ if (status.IsIOError()) {
+ return Status::IOError(status.ToString());
+ }
+ if (status.IsInvalid()) {
+ return Status::InvalidArgument(status.ToString());
+ }
+ return Status::InternalError(status.ToString());
+}
+
+class DorisRandomAccessFile final : public arrow::io::RandomAccessFile {
+public:
+ DorisRandomAccessFile(io::FileReaderSPtr file_reader, io::IOContext* io_ctx)
+ : _file_reader(std::move(file_reader)), _io_ctx(io_ctx) {
+ set_mode(arrow::io::FileMode::READ);
+ }
+
+ arrow::Status Close() override {
+ _closed = true;
+ return arrow::Status::OK();
+ }
+
+ bool closed() const override { return _closed; }
+
+ arrow::Result<int64_t> Tell() const override { return _pos; }
+
+ arrow::Status Seek(int64_t position) override {
+ if (position < 0) {
+ return arrow::Status::Invalid("negative seek position");
+ }
+ _pos = position;
+ return arrow::Status::OK();
+ }
+
+ arrow::Result<int64_t> GetSize() override {
+ if (!_file_reader) {
+ return arrow::Status::IOError("Doris file reader is not open");
+ }
+ return static_cast<int64_t>(_file_reader->size());
+ }
+
+ arrow::Result<int64_t> Read(int64_t nbytes, void* out) override {
+ ARROW_ASSIGN_OR_RAISE(auto bytes_read, ReadAt(_pos, nbytes, out));
+ _pos += bytes_read;
+ return bytes_read;
+ }
+
+ arrow::Result<std::shared_ptr<arrow::Buffer>> Read(int64_t nbytes) override {
+ ARROW_ASSIGN_OR_RAISE(auto buffer, arrow::AllocateResizableBuffer(nbytes));
+ ARROW_ASSIGN_OR_RAISE(auto bytes_read, Read(nbytes, buffer->mutable_data()));
+ ARROW_RETURN_NOT_OK(buffer->Resize(bytes_read, false));
+ buffer->ZeroPadding();
+ return buffer;
+ }
+
+ arrow::Result<int64_t> ReadAt(int64_t position, int64_t nbytes, void* out) override {
+ if (!_file_reader) {
+ return arrow::Status::IOError("Doris file reader is not open");
+ }
+ if (position < 0 || nbytes < 0) {
+ return arrow::Status::Invalid("negative read position or length");
+ }
+ size_t bytes_read = 0;
+ Status st = _file_reader->read_at(
+ static_cast<size_t>(position),
+ Slice(static_cast<uint8_t*>(out), static_cast<size_t>(nbytes)), &bytes_read,
+ _io_ctx);
+ if (!st.ok()) {
+ return arrow::Status::IOError(st.to_string_no_stack());
+ }
+ return static_cast<int64_t>(bytes_read);
+ }
+
+ arrow::Result<std::shared_ptr<arrow::Buffer>> ReadAt(int64_t position,
+ int64_t nbytes) override {
+ ARROW_ASSIGN_OR_RAISE(auto buffer, arrow::AllocateResizableBuffer(nbytes));
+ ARROW_ASSIGN_OR_RAISE(auto bytes_read, ReadAt(position, nbytes, buffer->mutable_data()));
+ ARROW_RETURN_NOT_OK(buffer->Resize(bytes_read, false));
+ buffer->ZeroPadding();
+ return buffer;
+ }
+
+private:
+ io::FileReaderSPtr _file_reader;
+ io::IOContext* _io_ctx = nullptr;
+ int64_t _pos = 0;
+ bool _closed = false;
+};
+
+struct ParquetReaderScanState {
+ // Doris 文件句柄适配成 Arrow RandomAccessFile。该对象只处理随机读，不携带
+ // table/global schema 语义。
+ std::shared_ptr<arrow::io::RandomAccessFile> arrow_file;
+
+ // Arrow Parquet core reader 和 footer metadata。ParquetReader 只依赖 core API，
+ // 不使用 parquet::arrow reader，也不输出 Arrow Array/RecordBatch。
+ std::unique_ptr<::parquet::ParquetFileReader> file_reader;
+ std::shared_ptr<::parquet::FileMetaData> metadata;
+ const ::parquet::SchemaDescriptor* schema = nullptr;
+ std::vector<std::unique_ptr<ParquetColumnSchema>> file_schema;
+
+ // 当前 scan 的 top-level file-local projection 和 row group 列表。projected_fields
+ // 决定输出 block；具体 leaf column reader 由 ParquetColumnReaderFactory 按需创建。
+ std::vector<int> predicate_fields;
+ std::vector<int> non_predicate_fields;
+ std::vector<int> selected_row_groups;
+ size_t next_row_group_idx = 0;
+ std::shared_ptr<::parquet::RowGroupReader> current_row_group;
+ std::vector<std::unique_ptr<ParquetColumnReader>> current_predicate_columns;
+ std::vector<std::unique_ptr<ParquetColumnReader>> current_non_predicate_columns;
+ int64_t current_row_group_rows = 0;
+ int64_t current_row_group_rows_read = 0;
+};
+
+Status ParquetReader::_reset_reader_position() {
+ _state->next_row_group_idx = 0;
+ _state->current_row_group.reset();
+ _state->current_predicate_columns.clear();
+ _state->current_non_predicate_columns.clear();
+ _state->current_row_group_rows = 0;
+ _state->current_row_group_rows_read = 0;
+ return Status::OK();
+}
+
+void ParquetReader::_reset_current_row_group() {
+ _state->current_row_group.reset();
+ _state->current_predicate_columns.clear();
+ _state->current_non_predicate_columns.clear();
+ _state->current_row_group_rows = 0;
+ _state->current_row_group_rows_read = 0;
+}
+
+void ParquetReader::_fill_schema_field(const ParquetColumnSchema& column_schema,
+ reader::SchemaField* field) const {
+ field->id = column_schema.leaf_column_id >= 0 ? column_schema.leaf_column_id
+ : column_schema.field_id;
+ field->name = column_schema.name;
+ field->type = column_schema.type;
+ field->children.clear();
+ field->children.reserve(column_schema.children.size());
+ for (const auto& child : column_schema.children) {
+ reader::SchemaField child_field;
+ _fill_schema_field(*child, &child_field);
+ field->children.push_back(std::move(child_field));
+ }
+}
+
+bool ParquetReader::_has_structured_filter(const reader::FileLocalFilter& local_filter) {
+ for (const auto& predicate : local_filter.predicates) {
+ if (predicate != nullptr) {
+ return true;
+ }
+ }
+ return false;
+}
+
+Status ParquetReader::_read_filter_columns(int64_t batch_rows, Block* file_block,
+ SelectionVector* selection, uint16_t* selected_rows) {
+ selection->resize(static_cast<size_t>(batch_rows));
+ for (size_t filter_idx = 0; filter_idx < _request->predicate_columns.size(); ++filter_idx) {
+ const int file_field_id = _request->predicate_columns[filter_idx];
+ auto& column_reader = _state->current_predicate_columns[filter_idx];
+ auto column = file_block->get_by_position(column_reader->file_column_id())
+ .column->assume_mutable();
+ DCHECK_EQ(file_block->get_by_position(column_reader->file_column_id())
+ .type->get_primitive_type(),
+ column_reader->type()->get_primitive_type());
+ int64_t column_rows = 0;
+ RETURN_IF_ERROR(column_reader->read(batch_rows, column, &column_rows));
+ if (column_rows != batch_rows) {
+ return Status::Corruption("Parquet filter column {} returned {} rows, expected {} rows",
+ column_reader->name(), column_rows, batch_rows);
+ }
+
+ for (const auto& local_filter : _request->local_filters) {
+ if (local_filter.file_column_id != file_field_id ||
+ !_has_structured_filter(local_filter)) {
+ continue;
+ }
+ if (*selected_rows == 0) {
+ break;
+ }
+ for (const auto& predicate : local_filter.predicates) {
+ *selected_rows = predicate->evaluate(*column, selection->data(), *selected_rows);
+ if (*selected_rows == 0) {
+ break;
+ }
+ }
+ break;
+ }
+ file_block->replace_by_position(file_field_id, std::move(column));
+ if (*selected_rows == 0) {
+ break;
+ }
+ }
+ return Status::OK();
+}
+
+Status ParquetReader::_validate_supported_local_filters(
+ const std::vector<reader::FileLocalFilter>& local_filters) {
+ for (const auto& local_filter : local_filters) {
+ if (local_filter.conjunct != nullptr) {
+ return Status::NotSupported(
+ "Parquet expression filter fallback is not implemented for field {}",
+ local_filter.file_column_id);
+ }
+ }
+ return Status::OK();
+}
+
+IColumn::Filter ParquetReader::_selection_to_filter(const SelectionVector& selection,
+ uint16_t selected_rows, int64_t batch_rows) {
+ IColumn::Filter filter(static_cast<size_t>(batch_rows), 0);
+ for (uint16_t selection_idx = 0; selection_idx < selected_rows; ++selection_idx) {
+ filter[selection.get_index(selection_idx)] = 1;
+ }
+ return filter;
+}
+
+Status ParquetReader::_open_next_row_group(bool* has_row_group) {
+ *has_row_group = false;
+ while (_state->next_row_group_idx < _state->selected_row_groups.size()) {
+ const int row_group_idx = _state->selected_row_groups[_state->next_row_group_idx++];
+ try {
+ _state->current_row_group = _state->file_reader->RowGroup(row_group_idx);
+ } catch (const ::parquet::ParquetException& e) {
+ return Status::Corruption("Failed to open parquet row group {}: {}", row_group_idx,
+ e.what());
+ } catch (const std::exception& e) {
+ return Status::InternalError("Failed to open parquet row group {}: {}", row_group_idx,
+ e.what());
+ }
+
+ auto row_group_metadata = _state->metadata->RowGroup(row_group_idx);
+ _state->current_row_group_rows =
+ row_group_metadata == nullptr ? 0 : row_group_metadata->num_rows();
+ if (_state->current_row_group_rows < 0) {
+ return Status::Corruption("Invalid negative row count in parquet row group {}",
+ row_group_idx);
+ } else if (_state->current_row_group_rows == 0) {
+ _reset_current_row_group();
+ continue;
+ }
+ _state->current_row_group_rows_read = 0;
+ _state->current_predicate_columns.clear();
+ _state->current_non_predicate_columns.clear();
+
+ ParquetColumnReaderFactory column_reader_factory(_state->current_row_group,
+ _state->schema->num_columns());
+ for (const auto file_field_id : _request->predicate_columns) {
+ const auto& column_schema = _state->file_schema[file_field_id];
+ std::unique_ptr<ParquetColumnReader> column_reader;
+ RETURN_IF_ERROR(column_reader_factory.create(*column_schema, &column_reader));
+ _state->current_predicate_columns.push_back(std::move(column_reader));
+ }
+ for (const auto file_field_id : _request->non_predicate_columns) {
+ const auto& column_schema = _state->file_schema[file_field_id];
+ std::unique_ptr<ParquetColumnReader> column_reader;
+ RETURN_IF_ERROR(column_reader_factory.create(*column_schema, &column_reader));
+ _state->current_non_predicate_columns.push_back(std::move(column_reader));
+ }
+ *has_row_group = true;
+ break;
+ }
+ return Status::OK();
+}
+
+// `file_block` has a complete struct derived from the file's schema.
+Status ParquetReader::_read_current_row_group_batch(int64_t batch_rows, Block* file_block,
+ size_t* rows) {
+ if (_state->current_predicate_columns.empty() &&
+ _state->current_non_predicate_columns.empty()) {
+ *rows = static_cast<size_t>(batch_rows);
+ return Status::OK();
+ }
+ SelectionVector selection;
+ uint16_t selected_rows = batch_rows;
+ // 1. Read all predicate columns and evaluate selection vector.
+ RETURN_IF_ERROR(_read_filter_columns(batch_rows, file_block, &selection, &selected_rows));
+
+ // 2. Materialize all predicate columns after filtering.
+ const bool need_filter_output = selected_rows != batch_rows;
+ if (need_filter_output) {
+ IColumn::Filter output_filter = _selection_to_filter(selection, selected_rows, batch_rows);
+ for (const auto file_field_id : _request->predicate_columns) {
+ RETURN_IF_CATCH_EXCEPTION(file_block->replace_by_position(
+ file_field_id, file_block->get_by_position(file_field_id)
+ .column->filter(output_filter, selected_rows)));
+ }
+ }
+
+ // 3. Materialize all non-predicate columns with selection.
+ for (size_t output_idx = 0; output_idx < _state->current_non_predicate_columns.size();
+ ++output_idx) {
+ auto& column_reader = _state->current_non_predicate_columns[output_idx];
+ auto col = file_block->get_columns()[column_reader->file_column_id()]->assume_mutable();
+ DCHECK_EQ(file_block->get_by_position(column_reader->file_column_id())
+ .type->get_primitive_type(),
+ column_reader->type()->get_primitive_type());
+ if (need_filter_output) {
+ [[maybe_unused]] auto old_size = col->size();
+ RETURN_IF_ERROR(column_reader->select(selection, selected_rows, batch_rows, col));
+ if (col->size() != old_size + selected_rows) {
+ return Status::Corruption(
+ "Parquet selected output column {} returned {} rows, expected {} rows",
+ column_reader->name(), col->size(), old_size + selected_rows);
+ }
+ } else {
+ int64_t column_rows = 0;
+ RETURN_IF_ERROR(column_reader->read(batch_rows, col, &column_rows));
+ if (column_rows != batch_rows) {
+ return Status::Corruption(
+ "Parquet output column {} returned {} rows, expected {} rows",
+ column_reader->name(), column_rows, batch_rows);
+ }
+ }
+ }
+
+ *rows = static_cast<size_t>(selected_rows);
+ return Status::OK();
+}
+
+ParquetReader::ParquetReader(std::unique_ptr<io::FileSystemProperties>& system_properties,
+ std::unique_ptr<io::FileDescription>& file_description,
+ std::shared_ptr<io::IOContext> io_ctx, RuntimeProfile* profile)
+ : FileReader(system_properties, file_description, io_ctx, profile) {}
+
+ParquetReader::~ParquetReader() = default;
+
+Status ParquetReader::init(RuntimeState* state) {
+ RETURN_IF_ERROR(reader::FileReader::init(state));
+ _state = std::make_unique<ParquetReaderScanState>();
+ _state->arrow_file =
+ std::make_shared<DorisRandomAccessFile>(_tracing_file_reader, _io_ctx.get());
+
+ try {
+ _state->file_reader = ::parquet::ParquetFileReader::Open(
+ _state->arrow_file, ::parquet::default_reader_properties());
+ _state->metadata = _state->file_reader->metadata();
+ _state->schema = _state->metadata != nullptr ? _state->metadata->schema() : nullptr;
+ } catch (const ::parquet::ParquetException& e) {
+ return Status::Corruption("Failed to open parquet file: {}", e.what());
+ } catch (const std::exception& e) {
+ return Status::InternalError("Failed to open parquet file: {}", e.what());
+ }
+
+ if (_state->metadata == nullptr || _state->schema == nullptr) {
+ return Status::Corruption("Failed to read parquet metadata");
+ }
+ RETURN_IF_ERROR(build_parquet_column_schema(*_state->schema, &_state->file_schema));
+ return Status::OK();
+}
+
+Status ParquetReader::get_schema(std::vector<reader::SchemaField>* file_schema) const {
+ if (file_schema == nullptr) {
+ return Status::InvalidArgument("file_schema is null");
+ }
+ file_schema->clear();
+ if (_state == nullptr || _state->schema == nullptr) {
+ return Status::Uninitialized("ParquetReader is not open");
+ }
+
+ file_schema->reserve(_state->file_schema.size());
+ for (size_t column_idx = 0; column_idx < _state->file_schema.size(); ++column_idx) {
+ reader::SchemaField field;
+ _fill_schema_field(*_state->file_schema[column_idx], &field);
+ field.id = static_cast<reader::ColumnId>(column_idx);
+ file_schema->push_back(std::move(field));
+ }
+ return Status::OK();
+}
+
+Status ParquetReader::open(std::unique_ptr<reader::FileScanRequest>& request) {
+ if (_state == nullptr || _state->metadata == nullptr || _state->schema == nullptr) {
+ return Status::Uninitialized("ParquetReader is not open");
+ }
+ RETURN_IF_ERROR(reader::FileReader::open(request));
+
+ const int num_fields = static_cast<int>(_state->file_schema.size());
+ for (const auto& local_filter : _request->local_filters) {
+ if (local_filter.file_column_id < 0 || local_filter.file_column_id >= num_fields) {
+ return Status::InvalidArgument("Invalid parquet filter top-level field id {}",
+ local_filter.file_column_id);
+ }
+ }
+ RETURN_IF_ERROR(_validate_supported_local_filters(_request->local_filters));
+
+ RETURN_IF_ERROR(select_row_groups_by_statistics(*_state->metadata, _state->file_schema,
+ *_request, &_state->selected_row_groups));
+ RETURN_IF_ERROR(_reset_reader_position());
+ _eof = _state->selected_row_groups.empty();
+ return Status::OK();
+}
+
+Status ParquetReader::get_block(Block* file_block, size_t* rows, bool* eof) {
+ if (_state == nullptr || _state->file_reader == nullptr || _state->schema == nullptr) {
+ return Status::Uninitialized("ParquetReader is not open");
+ }
+ *rows = 0;
+ if (_eof) {
+ *eof = true;
+ return Status::OK();
+ }
+
+ while (true) {
+ if (_state->current_row_group == nullptr) {
+ bool has_row_group = false;
+ RETURN_IF_ERROR(_open_next_row_group(&has_row_group));
+ if (!has_row_group) {
+ _eof = true;
+ *eof = true;
+ return Status::OK();
+ }
+ }
+
+ const int64_t remaining_rows =
+ _state->current_row_group_rows - _state->current_row_group_rows_read;
+ if (remaining_rows <= 0) {
+ _reset_current_row_group();
+ continue;
+ }
+
+ const int64_t batch_rows =
+ std::min<int64_t>(DEFAULT_PARQUET_READ_BATCH_SIZE, remaining_rows);
+ const int64_t physical_rows_read = batch_rows;
+ RETURN_IF_ERROR(_read_current_row_group_batch(batch_rows, file_block, rows));
+ _state->current_row_group_rows_read += physical_rows_read;
+ if (_state->current_row_group_rows_read >= _state->current_row_group_rows) {
+ _reset_current_row_group();
+ }
+ if (*rows == 0) {
+ continue;
+ }
+ *eof = false;
+ // TODO: Compute _request->reader_expression_map to filter file_block
+ return Status::OK();
+ }
+}
+
+Status ParquetReader::close() {
+ if (_state != nullptr) {
+ if (_state->file_reader != nullptr) {
+ try {
+ _state->file_reader->Close();
+ } catch (const std::exception&) {
+ // close 需要保持幂等；这里不覆盖此前 scan 路径上的真实错误。
+ }
+ }
+ if (_state->arrow_file != nullptr) {
+ static_cast<void>(arrow_status_to_doris_status(_state->arrow_file->Close()));
+ }
+ _state = std::make_unique<ParquetReaderScanState>();
+ }
+ return FileReader::close();
+}
+
+void ParquetReader::_init_profile() {
+ if (_profile != nullptr) {
+ static const char* parquet_profile = "ParquetReader";
+ ADD_TIMER_WITH_LEVEL(_profile, parquet_profile, 1);
+
+ _parquet_profile.filtered_row_groups = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "RowGroupsFiltered", TUnit::UNIT, parquet_profile, 1);
+ _parquet_profile.filtered_row_groups_by_min_max = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "RowGroupsFilteredByMinMax", TUnit::UNIT, parquet_profile, 1);
+ _parquet_profile.filtered_row_groups_by_bloom_filter = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "RowGroupsFilteredByBloomFilter", TUnit::UNIT, parquet_profile, 1);
+ _parquet_profile.to_read_row_groups = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "RowGroupsReadNum", TUnit::UNIT, parquet_profile, 1);
+ _parquet_profile.total_row_groups = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "RowGroupsTotalNum", TUnit::UNIT, parquet_profile, 1);
+ _parquet_profile.filtered_group_rows = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "FilteredRowsByGroup", TUnit::UNIT, parquet_profile, 1);
+ _parquet_profile.filtered_page_rows = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "FilteredRowsByPage", TUnit::UNIT, parquet_profile, 1);
+ _parquet_profile.lazy_read_filtered_rows = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "FilteredRowsByLazyRead", TUnit::UNIT, parquet_profile, 1);
+ _parquet_profile.filtered_bytes = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "FilteredBytes", TUnit::BYTES, parquet_profile, 1);
+ _parquet_profile.raw_rows_read = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "RawRowsRead", TUnit::UNIT, parquet_profile, 1);
+ _parquet_profile.column_read_time =
+ ADD_CHILD_TIMER_WITH_LEVEL(_profile, "ColumnReadTime", parquet_profile, 1);
+ _parquet_profile.parse_meta_time =
+ ADD_CHILD_TIMER_WITH_LEVEL(_profile, "ParseMetaTime", parquet_profile, 1);
+ _parquet_profile.parse_footer_time =
+ ADD_CHILD_TIMER_WITH_LEVEL(_profile, "ParseFooterTime", parquet_profile, 1);
+ _parquet_profile.file_reader_create_time =
+ ADD_CHILD_TIMER_WITH_LEVEL(_profile, "FileReaderCreateTime", parquet_profile, 1);
+ _parquet_profile.open_file_num =
+ ADD_CHILD_COUNTER_WITH_LEVEL(_profile, "FileNum", TUnit::UNIT, parquet_profile, 1);
+ _parquet_profile.page_index_read_calls =
+ ADD_COUNTER_WITH_LEVEL(_profile, "PageIndexReadCalls", TUnit::UNIT, 1);
+ _parquet_profile.page_index_filter_time =
+ ADD_CHILD_TIMER_WITH_LEVEL(_profile, "PageIndexFilterTime", parquet_profile, 1);
+ _parquet_profile.read_page_index_time =
+ ADD_CHILD_TIMER_WITH_LEVEL(_profile, "PageIndexReadTime", parquet_profile, 1);
+ _parquet_profile.parse_page_index_time =
+ ADD_CHILD_TIMER_WITH_LEVEL(_profile, "PageIndexParseTime", parquet_profile, 1);
+ _parquet_profile.row_group_filter_time =
+ ADD_CHILD_TIMER_WITH_LEVEL(_profile, "RowGroupFilterTime", parquet_profile, 1);
+ _parquet_profile.file_footer_read_calls =
+ ADD_COUNTER_WITH_LEVEL(_profile, "FileFooterReadCalls", TUnit::UNIT, 1);
+ _parquet_profile.file_footer_hit_cache =
+ ADD_COUNTER_WITH_LEVEL(_profile, "FileFooterHitCache", TUnit::UNIT, 1);
+ _parquet_profile.decompress_time =
+ ADD_CHILD_TIMER_WITH_LEVEL(_profile, "DecompressTime", parquet_profile, 1);
+ _parquet_profile.decompress_cnt = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "DecompressCount", TUnit::UNIT, parquet_profile, 1);
+ _parquet_profile.page_read_counter = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "PageReadCount", TUnit::UNIT, parquet_profile, 1);
+ _parquet_profile.page_cache_write_counter = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "PageCacheWriteCount", TUnit::UNIT, parquet_profile, 1);
+ _parquet_profile.page_cache_compressed_write_counter = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "PageCacheCompressedWriteCount", TUnit::UNIT, parquet_profile, 1);
+ _parquet_profile.page_cache_decompressed_write_counter = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "PageCacheDecompressedWriteCount", TUnit::UNIT, parquet_profile, 1);
+ _parquet_profile.page_cache_hit_counter = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "PageCacheHitCount", TUnit::UNIT, parquet_profile, 1);
+ _parquet_profile.page_cache_missing_counter = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "PageCacheMissingCount", TUnit::UNIT, parquet_profile, 1);
+ _parquet_profile.page_cache_compressed_hit_counter = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "PageCacheCompressedHitCount", TUnit::UNIT, parquet_profile, 1);
+ _parquet_profile.page_cache_decompressed_hit_counter = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "PageCacheDecompressedHitCount", TUnit::UNIT, parquet_profile, 1);
+ _parquet_profile.decode_header_time =
+ ADD_CHILD_TIMER_WITH_LEVEL(_profile, "PageHeaderDecodeTime", parquet_profile, 1);
+ _parquet_profile.read_page_header_time =
+ ADD_CHILD_TIMER_WITH_LEVEL(_profile, "PageHeaderReadTime", parquet_profile, 1);
+ _parquet_profile.decode_value_time =
+ ADD_CHILD_TIMER_WITH_LEVEL(_profile, "DecodeValueTime", parquet_profile, 1);
+ _parquet_profile.decode_dict_time =
+ ADD_CHILD_TIMER_WITH_LEVEL(_profile, "DecodeDictTime", parquet_profile, 1);
+ _parquet_profile.decode_level_time =
+ ADD_CHILD_TIMER_WITH_LEVEL(_profile, "DecodeLevelTime", parquet_profile, 1);
+ _parquet_profile.decode_null_map_time =
+ ADD_CHILD_TIMER_WITH_LEVEL(_profile, "DecodeNullMapTime", parquet_profile, 1);
+ _parquet_profile.skip_page_header_num = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "SkipPageHeaderNum", TUnit::UNIT, parquet_profile, 1);
+ _parquet_profile.parse_page_header_num = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "ParsePageHeaderNum", TUnit::UNIT, parquet_profile, 1);
+ _parquet_profile.predicate_filter_time =
+ ADD_CHILD_TIMER_WITH_LEVEL(_profile, "PredicateFilterTime", parquet_profile, 1);
+ _parquet_profile.dict_filter_rewrite_time =
+ ADD_CHILD_TIMER_WITH_LEVEL(_profile, "DictFilterRewriteTime", parquet_profile, 1);
+ _parquet_profile.convert_time =
+ ADD_CHILD_TIMER_WITH_LEVEL(_profile, "ConvertTime", parquet_profile, 1);
+ _parquet_profile.bloom_filter_read_time =
+ ADD_CHILD_TIMER_WITH_LEVEL(_profile, "BloomFilterReadTime", parquet_profile, 1);
+ }
+}
+
+} // namespace doris::parquet
diff --git a/be/src/format/new_parquet/parquet_reader.h b/be/src/format/new_parquet/parquet_reader.h
new file mode 100644
index 00000000000..426960a4dfd
--- /dev/null
+++ b/be/src/format/new_parquet/parquet_reader.h
@@ -0,0 +1,138 @@
+// 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.
+
+#pragma once
+
+#include <memory>
+#include <vector>
+
+#include "common/status.h"
+#include "format/reader/file_reader.h"
+#include "parquet_column_schema.h"
+#include "selection_vector.h"
+
+namespace doris {
+namespace io {
+struct IOContext;
+} // namespace io
+} // namespace doris
+
+namespace doris::parquet {
+
+struct ParquetReaderScanState;
+
+// ParquetReader 的 file-local scan 请求。
+// 当前没有新增 Parquet-only 字段，但保留独立类型，便于后续加入 row group/page index
+// 等 Parquet 专属选项。
+struct ParquetScanRequest : public reader::FileScanRequest {};
+
+// Parquet 文件物理读取层。
+// 该类只理解 Parquet file-local schema 和 ParquetScanRequest，不理解 Iceberg/global
+// schema，不处理 table-level cast/default/generated/partition 语义。
+class ParquetReader : public reader::FileReader {
+public:
+ ParquetReader(std::unique_ptr<io::FileSystemProperties>& system_properties,
+ std::unique_ptr<io::FileDescription>& file_description,
+ std::shared_ptr<io::IOContext> io_ctx, RuntimeProfile* profile);
+ ~ParquetReader() override;
+
+ // 打开 Parquet 文件并解析 footer metadata。
+ // init 成功后可以调用 get_schema() 获取 Parquet file-local schema。
+ Status init(RuntimeState* state) override;
+
+ // 解析 Parquet footer 并返回 Parquet 文件自身的 schema。
+ // 该方法只能在 open() 成功后调用，不要求 init() 已经执行。
+ // 这里不做 Iceberg schema evolution，也不把字段转换成 table/global schema。
+ Status get_schema(std::vector<reader::SchemaField>* file_schema) const override;
+
+ Status open(std::unique_ptr<reader::FileScanRequest>& request) override;
+ // 读取下一批 Parquet file-local block。
+ // 该方法只能在 init() 成功后调用。
+ // 返回列必须保持 file-local 语义，不能在这里补 default/generated/partition 列。
+ Status get_block(Block* file_block, size_t* rows, bool* eof) override;
+
+ Status close() override;
+
+protected:
+ void _init_profile() override;
+
+private:
+ struct ParquetProfile {
+ RuntimeProfile::Counter* filtered_row_groups = nullptr;
+ RuntimeProfile::Counter* filtered_row_groups_by_min_max = nullptr;
+ RuntimeProfile::Counter* filtered_row_groups_by_bloom_filter = nullptr;
+ RuntimeProfile::Counter* to_read_row_groups = nullptr;
+ RuntimeProfile::Counter* total_row_groups = nullptr;
+ RuntimeProfile::Counter* filtered_group_rows = nullptr;
+ RuntimeProfile::Counter* filtered_page_rows = nullptr;
+ RuntimeProfile::Counter* lazy_read_filtered_rows = nullptr;
+ RuntimeProfile::Counter* filtered_bytes = nullptr;
+ RuntimeProfile::Counter* raw_rows_read = nullptr;
+ RuntimeProfile::Counter* column_read_time = nullptr;
+ RuntimeProfile::Counter* parse_meta_time = nullptr;
+ RuntimeProfile::Counter* parse_footer_time = nullptr;
+ RuntimeProfile::Counter* file_reader_create_time = nullptr;
+ RuntimeProfile::Counter* open_file_num = nullptr;
+ RuntimeProfile::Counter* row_group_filter_time = nullptr;
+ RuntimeProfile::Counter* page_index_read_calls = nullptr;
+ RuntimeProfile::Counter* page_index_filter_time = nullptr;
+ RuntimeProfile::Counter* read_page_index_time = nullptr;
+ RuntimeProfile::Counter* parse_page_index_time = nullptr;
+ RuntimeProfile::Counter* file_footer_read_calls = nullptr;
+ RuntimeProfile::Counter* file_footer_hit_cache = nullptr;
+ RuntimeProfile::Counter* decompress_time = nullptr;
+ RuntimeProfile::Counter* decompress_cnt = nullptr;
+ RuntimeProfile::Counter* page_read_counter = nullptr;
+ RuntimeProfile::Counter* page_cache_write_counter = nullptr;
+ RuntimeProfile::Counter* page_cache_compressed_write_counter = nullptr;
+ RuntimeProfile::Counter* page_cache_decompressed_write_counter = nullptr;
+ RuntimeProfile::Counter* page_cache_hit_counter = nullptr;
+ RuntimeProfile::Counter* page_cache_missing_counter = nullptr;
+ RuntimeProfile::Counter* page_cache_compressed_hit_counter = nullptr;
+ RuntimeProfile::Counter* page_cache_decompressed_hit_counter = nullptr;
+ RuntimeProfile::Counter* decode_header_time = nullptr;
+ RuntimeProfile::Counter* read_page_header_time = nullptr;
+ RuntimeProfile::Counter* decode_value_time = nullptr;
+ RuntimeProfile::Counter* decode_dict_time = nullptr;
+ RuntimeProfile::Counter* decode_level_time = nullptr;
+ RuntimeProfile::Counter* decode_null_map_time = nullptr;
+ RuntimeProfile::Counter* skip_page_header_num = nullptr;
+ RuntimeProfile::Counter* parse_page_header_num = nullptr;
+ RuntimeProfile::Counter* predicate_filter_time = nullptr;
+ RuntimeProfile::Counter* dict_filter_rewrite_time = nullptr;
+ RuntimeProfile::Counter* convert_time = nullptr;
+ RuntimeProfile::Counter* bloom_filter_read_time = nullptr;
+ };
+ Status _reset_reader_position();
+ void _reset_current_row_group();
+ void _fill_schema_field(const ParquetColumnSchema& column_schema,
+ reader::SchemaField* field) const;
+ bool _has_structured_filter(const reader::FileLocalFilter& local_filter);
+ Status _read_filter_columns(int64_t batch_rows, Block* file_block, SelectionVector* selection,
+ uint16_t* selected_rows);
+ Status _validate_supported_local_filters(
+ const std::vector<reader::FileLocalFilter>& local_filters);
+ IColumn::Filter _selection_to_filter(const SelectionVector& selection, uint16_t selected_rows,
+ int64_t batch_rows);
+ Status _open_next_row_group(bool* has_row_group);
+ Status _read_current_row_group_batch(int64_t batch_rows, Block* file_block, size_t* rows);
+
+ std::unique_ptr<ParquetReaderScanState> _state;
+ ParquetProfile _parquet_profile;
+};
+
+} // namespace doris::parquet
diff --git a/be/src/format/new_parquet/parquet_statistics.cpp b/be/src/format/new_parquet/parquet_statistics.cpp
new file mode 100644
index 00000000000..aebc6d4e04d
--- /dev/null
+++ b/be/src/format/new_parquet/parquet_statistics.cpp
@@ -0,0 +1,254 @@
+// 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/new_parquet/parquet_statistics.h"
+
+#include <parquet/api/reader.h>
+#include <parquet/api/schema.h>
+#include <parquet/statistics.h>
+#include <parquet/types.h>
+
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "core/data_type/data_type.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/primitive_type.h"
+#include "core/field.h"
+#include "format/new_parquet/parquet_column_schema.h"
+#include "storage/index/zone_map/zone_map_index.h"
+#include "storage/predicate/column_predicate.h"
+
+namespace doris::parquet {
+namespace {
+
+PrimitiveType physical_filter_type(const ParquetColumnSchema& column_schema) {
+ if (column_schema.type == nullptr) {
+ return INVALID_TYPE;
+ }
+ switch (remove_nullable(column_schema.type)->get_primitive_type()) {
+ case TYPE_BOOLEAN:
+ case TYPE_INT:
+ case TYPE_BIGINT:
+ case TYPE_FLOAT:
+ case TYPE_DOUBLE:
+ case TYPE_STRING:
+ return remove_nullable(column_schema.type)->get_primitive_type();
+ default:
+ return INVALID_TYPE;
+ }
+}
+
+template <typename ParquetDType, PrimitiveType DorisType, typename ConvertFn>
+bool set_typed_min_max(const std::shared_ptr<::parquet::Statistics>& statistics, ConvertFn convert,
+ ParquetColumnStatistics* column_statistics) {
+ auto typed_statistics =
+ std::static_pointer_cast<::parquet::TypedStatistics<ParquetDType>>(statistics);
+ column_statistics->min_value = Field::create_field<DorisType>(convert(typed_statistics->min()));
+ column_statistics->max_value = Field::create_field<DorisType>(convert(typed_statistics->max()));
+ return true;
+}
+
+bool set_string_min_max(const std::shared_ptr<::parquet::Statistics>& statistics,
+ const ::parquet::ColumnDescriptor* descriptor,
+ ParquetColumnStatistics* column_statistics) {
+ switch (statistics->physical_type()) {
+ case ::parquet::Type::BYTE_ARRAY: {
+ auto typed_statistics =
+ std::static_pointer_cast<::parquet::TypedStatistics<::parquet::ByteArrayType>>(
+ statistics);
+ column_statistics->min_value = Field::create_field<TYPE_STRING>(
+ ::parquet::ByteArrayToString(typed_statistics->min()));
+ column_statistics->max_value = Field::create_field<TYPE_STRING>(
+ ::parquet::ByteArrayToString(typed_statistics->max()));
+ return true;
+ }
+ case ::parquet::Type::FIXED_LEN_BYTE_ARRAY: {
+ if (descriptor == nullptr || descriptor->type_length() <= 0) {
+ return false;
+ }
+ auto typed_statistics =
+ std::static_pointer_cast<::parquet::TypedStatistics<::parquet::FLBAType>>(
+ statistics);
+ const int type_length = descriptor->type_length();
+ column_statistics->min_value = Field::create_field<TYPE_STRING>(std::string(
+ reinterpret_cast<const char*>(typed_statistics->min().ptr), type_length));
+ column_statistics->max_value = Field::create_field<TYPE_STRING>(std::string(
+ reinterpret_cast<const char*>(typed_statistics->max().ptr), type_length));
+ return true;
+ }
+ default:
+ return false;
+ }
+}
+
+bool is_null_only_predicate(const ColumnPredicate& predicate) {
+ return predicate.type() == PredicateType::IS_NULL ||
+ predicate.type() == PredicateType::IS_NOT_NULL;
+}
+
+segment_v2::ZoneMap to_column_predicate_statistics(const ParquetColumnStatistics& statistics) {
+ segment_v2::ZoneMap predicate_statistics;
+ predicate_statistics.min_value = statistics.min_value;
+ predicate_statistics.max_value = statistics.max_value;
+ predicate_statistics.has_null = statistics.has_null;
+ predicate_statistics.has_not_null = statistics.has_not_null;
+ return predicate_statistics;
+}
+
+} // namespace
+
+ParquetColumnStatistics ParquetStatisticsUtils::TransformColumnStatistics(
+ const ParquetColumnSchema& column_schema,
+ const std::shared_ptr<::parquet::Statistics>& statistics) {
+ ParquetColumnStatistics result;
+ if (statistics == nullptr) {
+ return result;
+ }
+
+ result.has_null = statistics->HasNullCount() && statistics->null_count() > 0;
+ result.has_not_null = statistics->num_values() > 0 || statistics->HasMinMax();
+ result.has_null_count = statistics->HasNullCount();
+ if (!result.has_not_null || !statistics->HasMinMax()) {
+ return result;
+ }
+
+ switch (statistics->physical_type()) {
+ case ::parquet::Type::BOOLEAN:
+ result.has_min_max = set_typed_min_max<::parquet::BooleanType, TYPE_BOOLEAN>(
+ statistics, [](bool value) { return static_cast<UInt8>(value); }, &result);
+ return result;
+ case ::parquet::Type::INT32:
+ result.has_min_max = set_typed_min_max<::parquet::Int32Type, TYPE_INT>(
+ statistics, [](int32_t value) { return value; }, &result);
+ return result;
+ case ::parquet::Type::INT64:
+ result.has_min_max = set_typed_min_max<::parquet::Int64Type, TYPE_BIGINT>(
+ statistics, [](int64_t value) { return value; }, &result);
+ return result;
+ case ::parquet::Type::FLOAT:
+ result.has_min_max = set_typed_min_max<::parquet::FloatType, TYPE_FLOAT>(
+ statistics, [](float value) { return value; }, &result);
+ return result;
+ case ::parquet::Type::DOUBLE:
+ result.has_min_max = set_typed_min_max<::parquet::DoubleType, TYPE_DOUBLE>(
+ statistics, [](double value) { return value; }, &result);
+ return result;
+ case ::parquet::Type::BYTE_ARRAY:
+ case ::parquet::Type::FIXED_LEN_BYTE_ARRAY:
+ result.has_min_max = set_string_min_max(statistics, column_schema.descriptor, &result);
+ return result;
+ default:
+ return result;
+ }
+}
+
+bool ParquetStatisticsUtils::CheckStatistics(const reader::FileLocalFilter& local_filter,
+ const ParquetColumnStatistics& statistics) {
+ if (!statistics.has_any_statistics()) {
+ return false;
+ }
+
+ // TODO: replace local_filter.predicates by local_filter.conjuncts
+ for (const auto& column_predicate : local_filter.predicates) {
+ if (is_null_only_predicate(*column_predicate)) {
+ if (!statistics.has_null_count) {
+ continue;
+ }
+ } else if (!statistics.has_any_statistics()) {
+ continue;
+ }
+ if (!column_predicate->evaluate_and(to_column_predicate_statistics(statistics))) {
+ return true;
+ }
+ }
+ return false;
+}
+
+bool ParquetStatisticsUtils::RowGroupExcludes(
+ const ::parquet::RowGroupMetaData& row_group,
+ const std::vector<std::unique_ptr<ParquetColumnSchema>>& schema,
+ const reader::FileLocalFilter& local_filter) {
+ DCHECK(local_filter.file_column_id >= 0 &&
+ local_filter.file_column_id < row_group.num_columns());
+ DCHECK_LT(local_filter.file_column_id, schema.size());
+ auto column_chunk = row_group.ColumnChunk(local_filter.file_column_id);
+ if (column_chunk == nullptr) {
+ return false;
+ }
+ return CheckStatistics(local_filter,
+ TransformColumnStatistics(*schema[local_filter.file_column_id],
+ column_chunk->statistics()));
+}
+
+Status ParquetStatisticsUtils::SelectRowGroups(
+ const ::parquet::FileMetaData& metadata,
+ const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
+ const reader::FileScanRequest& request, std::vector<int>* selected_row_groups) {
+ if (selected_row_groups == nullptr) {
+ return Status::InvalidArgument("selected_row_groups is null");
+ }
+ selected_row_groups->clear();
+
+ const int num_row_groups = metadata.num_row_groups();
+ selected_row_groups->reserve(num_row_groups);
+ for (int row_group_idx = 0; row_group_idx < num_row_groups; ++row_group_idx) {
+ auto row_group = metadata.RowGroup(row_group_idx);
+ if (row_group == nullptr) {
+ selected_row_groups->push_back(row_group_idx);
+ continue;
+ }
+ bool drop = false;
+ for (const auto& local_filter : request.local_filters) {
+ if (RowGroupExcludes(*row_group, file_schema, local_filter)) {
+ drop = true;
+ break;
+ }
+ }
+ if (drop) {
+ continue;
+ }
+ selected_row_groups->push_back(row_group_idx);
+ }
+ return Status::OK();
+}
+
+bool ParquetStatisticsUtils::BloomFilterSupported(const ParquetColumnSchema& column_schema) {
+ switch (physical_filter_type(column_schema)) {
+ case TYPE_BOOLEAN:
+ case TYPE_INT:
+ case TYPE_BIGINT:
+ case TYPE_FLOAT:
+ case TYPE_DOUBLE:
+ case TYPE_STRING:
+ return true;
+ default:
+ return false;
+ }
+}
+
+Status select_row_groups_by_statistics(
+ const ::parquet::FileMetaData& metadata,
+ const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
+ const reader::FileScanRequest& request, std::vector<int>* selected_row_groups) {
+ return ParquetStatisticsUtils::SelectRowGroups(metadata, file_schema, request,
+ selected_row_groups);
+}
+
+} // namespace doris::parquet
diff --git a/be/src/format/new_parquet/parquet_statistics.h b/be/src/format/new_parquet/parquet_statistics.h
new file mode 100644
index 00000000000..0def08d4b08
--- /dev/null
+++ b/be/src/format/new_parquet/parquet_statistics.h
@@ -0,0 +1,88 @@
+// 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.
+
+#pragma once
+
+#include <memory>
+#include <vector>
+
+#include "common/status.h"
+#include "core/field.h"
+#include "format/reader/file_reader.h"
+
+namespace parquet {
+class FileMetaData;
+class RowGroupMetaData;
+class Statistics;
+} // namespace parquet
+
+namespace doris {
+class ColumnPredicate;
+} // namespace doris
+
+namespace doris::parquet {
+
+struct ParquetColumnSchema;
+
+// Parquet row group column statistics 转换后的 Doris 统计视图。
+// DuckDB 会把 Parquet stats 转换成 BaseStatistics，然后让 TableFilter 自己判断；
+// Doris 新 reader 先保存 file-local min/max/null 信息，再交给 ColumnPredicate 判断。
+struct ParquetColumnStatistics {
+ Field min_value;
+ Field max_value;
+ bool has_null = false;
+ bool has_not_null = false;
+ bool has_null_count = false;
+ bool has_min_max = false;
+
+ bool has_any_statistics() const { return has_null_count || has_min_max; }
+};
+
+// Parquet file-local statistics/page index/bloom filter 工具类。
+// 结构参考 DuckDB ParquetStatisticsUtils：先把 Parquet metadata 转成统一统计对象，
+// 再由 filter/predicate 判断是否可以裁剪。这里不理解 table/global schema。
+struct ParquetStatisticsUtils {
+ static ParquetColumnStatistics TransformColumnStatistics(
+ const ParquetColumnSchema& column_schema,
+ const std::shared_ptr<::parquet::Statistics>& statistics);
+
+ // Return true if the statistics indicate that the row group can be safely skipped according to the local filter.
+ static bool CheckStatistics(const reader::FileLocalFilter& local_filter,
+ const ParquetColumnStatistics& statistics);
+
+ static bool RowGroupExcludes(const ::parquet::RowGroupMetaData& row_group,
+ const std::vector<std::unique_ptr<ParquetColumnSchema>>& schema,
+ const reader::FileLocalFilter& local_filter);
+
+ static Status SelectRowGroups(
+ const ::parquet::FileMetaData& metadata,
+ const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
+ const reader::FileScanRequest& request, std::vector<int>* selected_row_groups);
+
+ static bool BloomFilterSupported(const ParquetColumnSchema& column_schema);
+};
+
+// Parquet file-local statistics/page index/bloom filter 裁剪入口。
+// 这里只消费已经 localize 到 file schema 的 FileScanRequest，不理解 table/global schema。
+// 后续 page index、dictionary、bloom filter 等文件格式优化也应继续收敛在这一层，避免污染
+// ParquetReader 的 scan 调度代码。
+Status select_row_groups_by_statistics(
+ const ::parquet::FileMetaData& metadata,
+ const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
+ const reader::FileScanRequest& request, std::vector<int>* selected_row_groups);
+
+} // namespace doris::parquet
diff --git a/be/src/format/new_parquet/parquet_type.cpp b/be/src/format/new_parquet/parquet_type.cpp
new file mode 100644
index 00000000000..53c7b4f2ed9
--- /dev/null
+++ b/be/src/format/new_parquet/parquet_type.cpp
@@ -0,0 +1,349 @@
+// 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/new_parquet/parquet_type.h"
+
+#include <parquet/api/schema.h>
+
+#include <memory>
+#include <string>
+
+#include "core/data_type/data_type_factory.hpp"
+#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/primitive_type.h"
+
+namespace doris::parquet {
+namespace {
+
+DataTypePtr create_type(PrimitiveType type, bool nullable, int precision = 0, int scale = 0) {
+ return DataTypeFactory::instance().create_data_type(type, nullable, precision, scale);
+}
+
+PrimitiveType decimal_primitive_type(int precision) {
+ return precision > 38 ? TYPE_DECIMAL256 : TYPE_DECIMAL128I;
+}
+
+bool has_non_physical_annotation(const ::parquet::ColumnDescriptor* column) {
+ if (column == nullptr) {
+ return false;
+ }
+ const auto& logical_type = column->logical_type();
+ return column->converted_type() != ::parquet::ConvertedType::NONE ||
+ (logical_type != nullptr && logical_type->is_valid() && !logical_type->is_none());
+}
+
+void mark_decimal(const ::parquet::ColumnDescriptor* column, int precision, int scale,
+ ParquetTypeDescriptor* result) {
+ result->is_decimal = true;
+ result->decimal_precision = precision;
+ result->decimal_scale = scale;
+ switch (column->physical_type()) {
+ case ::parquet::Type::INT32:
+ result->extra_type_info = ParquetExtraTypeInfo::DECIMAL_INT32;
+ break;
+ case ::parquet::Type::INT64:
+ result->extra_type_info = ParquetExtraTypeInfo::DECIMAL_INT64;
+ break;
+ case ::parquet::Type::BYTE_ARRAY:
+ case ::parquet::Type::FIXED_LEN_BYTE_ARRAY:
+ result->extra_type_info = ParquetExtraTypeInfo::DECIMAL_BYTE_ARRAY;
+ break;
+ default:
+ result->extra_type_info = ParquetExtraTypeInfo::NONE;
+ break;
+ }
+}
+
+DataTypePtr converted_type_to_doris_type(const ::parquet::ColumnDescriptor* column,
+ ParquetTypeDescriptor* result) {
+ const bool nullable = column->max_definition_level() > 0;
+ switch (column->converted_type()) {
+ case ::parquet::ConvertedType::UTF8:
+ case ::parquet::ConvertedType::ENUM:
+ case ::parquet::ConvertedType::JSON:
+ case ::parquet::ConvertedType::BSON:
+ return create_type(TYPE_STRING, nullable);
+ case ::parquet::ConvertedType::DECIMAL:
+ mark_decimal(column, column->type_precision(), column->type_scale(), result);
+ return create_type(decimal_primitive_type(column->type_precision()), nullable,
+ column->type_precision(), column->type_scale());
+ case ::parquet::ConvertedType::DATE:
+ return create_type(TYPE_DATEV2, nullable);
+ case ::parquet::ConvertedType::TIME_MILLIS:
+ result->time_unit = ParquetTimeUnit::MILLIS;
+ result->extra_type_info = ParquetExtraTypeInfo::UNIT_MS;
+ return create_type(TYPE_TIMEV2, nullable, 0, 3);
+ case ::parquet::ConvertedType::TIME_MICROS:
+ result->time_unit = ParquetTimeUnit::MICROS;
+ result->extra_type_info = ParquetExtraTypeInfo::UNIT_MICROS;
+ return create_type(TYPE_TIMEV2, nullable, 0, 6);
+ case ::parquet::ConvertedType::TIMESTAMP_MILLIS:
+ result->is_timestamp = true;
+ result->time_unit = ParquetTimeUnit::MILLIS;
+ result->extra_type_info = ParquetExtraTypeInfo::UNIT_MS;
+ return create_type(TYPE_DATETIMEV2, nullable, 0, 3);
+ case ::parquet::ConvertedType::TIMESTAMP_MICROS:
+ result->is_timestamp = true;
+ result->time_unit = ParquetTimeUnit::MICROS;
+ result->extra_type_info = ParquetExtraTypeInfo::UNIT_MICROS;
+ return create_type(TYPE_DATETIMEV2, nullable, 0, 6);
+ case ::parquet::ConvertedType::INT_8:
+ return create_type(TYPE_TINYINT, nullable);
+ case ::parquet::ConvertedType::UINT_8:
+ case ::parquet::ConvertedType::INT_16:
+ return create_type(TYPE_SMALLINT, nullable);
+ case ::parquet::ConvertedType::UINT_16:
+ case ::parquet::ConvertedType::INT_32:
+ return create_type(TYPE_INT, nullable);
+ case ::parquet::ConvertedType::UINT_32:
+ case ::parquet::ConvertedType::INT_64:
+ return create_type(TYPE_BIGINT, nullable);
+ case ::parquet::ConvertedType::UINT_64:
+ return create_type(TYPE_LARGEINT, nullable);
+ case ::parquet::ConvertedType::NONE:
+ default:
+ return nullptr;
+ }
+}
+
+DataTypePtr logical_type_to_doris_type(const ::parquet::ColumnDescriptor* column,
+ ParquetTypeDescriptor* result) {
+ const auto& logical_type = column->logical_type();
+ if (logical_type == nullptr || !logical_type->is_valid() || logical_type->is_none()) {
+ return nullptr;
+ }
+ const bool nullable = column->max_definition_level() > 0;
+ if (logical_type->is_string() || logical_type->is_enum() || logical_type->is_JSON() ||
+ logical_type->is_BSON() || logical_type->is_UUID()) {
+ return create_type(TYPE_STRING, nullable);
+ }
+ if (logical_type->is_decimal()) {
+ const auto& decimal_type = static_cast<const ::parquet::DecimalLogicalType&>(*logical_type);
+ mark_decimal(column, decimal_type.precision(), decimal_type.scale(), result);
+ return create_type(decimal_primitive_type(decimal_type.precision()), nullable,
+ decimal_type.precision(), decimal_type.scale());
+ }
+ if (logical_type->is_date()) {
+ return create_type(TYPE_DATEV2, nullable);
+ }
+ if (logical_type->is_time()) {
+ const auto& time_type = static_cast<const ::parquet::TimeLogicalType&>(*logical_type);
+ int scale = 0;
+ if (time_type.time_unit() == ::parquet::LogicalType::TimeUnit::MILLIS) {
+ scale = 3;
+ result->time_unit = ParquetTimeUnit::MILLIS;
+ result->extra_type_info = ParquetExtraTypeInfo::UNIT_MS;
+ } else if (time_type.time_unit() == ::parquet::LogicalType::TimeUnit::MICROS) {
+ scale = 6;
+ result->time_unit = ParquetTimeUnit::MICROS;
+ result->extra_type_info = ParquetExtraTypeInfo::UNIT_MICROS;
+ } else {
+ return nullptr;
+ }
+ return create_type(TYPE_TIMEV2, nullable, 0, scale);
+ }
+ if (logical_type->is_timestamp()) {
+ const auto& timestamp_type =
+ static_cast<const ::parquet::TimestampLogicalType&>(*logical_type);
+ int scale = 0;
+ if (timestamp_type.time_unit() == ::parquet::LogicalType::TimeUnit::MILLIS) {
+ scale = 3;
+ result->time_unit = ParquetTimeUnit::MILLIS;
+ result->extra_type_info = ParquetExtraTypeInfo::UNIT_MS;
+ } else if (timestamp_type.time_unit() == ::parquet::LogicalType::TimeUnit::MICROS) {
+ scale = 6;
+ result->time_unit = ParquetTimeUnit::MICROS;
+ result->extra_type_info = ParquetExtraTypeInfo::UNIT_MICROS;
+ } else {
+ return nullptr;
+ }
+ result->is_timestamp = true;
+ return create_type(TYPE_DATETIMEV2, nullable, 0, scale);
+ }
+ if (logical_type->is_int()) {
+ const auto& int_type = static_cast<const ::parquet::IntLogicalType&>(*logical_type);
+ switch (int_type.bit_width()) {
+ case 8:
+ return create_type(int_type.is_signed() ? TYPE_TINYINT : TYPE_SMALLINT, nullable);
+ case 16:
+ return create_type(int_type.is_signed() ? TYPE_SMALLINT : TYPE_INT, nullable);
+ case 32:
+ return create_type(int_type.is_signed() ? TYPE_INT : TYPE_BIGINT, nullable);
+ case 64:
+ return create_type(int_type.is_signed() ? TYPE_BIGINT : TYPE_LARGEINT, nullable);
+ default:
+ return nullptr;
+ }
+ }
+ return nullptr;
+}
+
+DataTypePtr physical_type_to_doris_type(const ::parquet::ColumnDescriptor* column) {
+ const bool nullable = column->max_definition_level() > 0;
+ DataTypePtr type;
+ switch (column->physical_type()) {
+ case ::parquet::Type::BOOLEAN:
+ type = std::make_shared<DataTypeBool>();
+ break;
+ case ::parquet::Type::INT32:
+ type = std::make_shared<DataTypeInt32>();
+ break;
+ case ::parquet::Type::INT64:
+ type = std::make_shared<DataTypeInt64>();
+ break;
+ case ::parquet::Type::FLOAT:
+ type = std::make_shared<DataTypeFloat32>();
+ break;
+ case ::parquet::Type::DOUBLE:
+ type = std::make_shared<DataTypeFloat64>();
+ break;
+ case ::parquet::Type::BYTE_ARRAY:
+ case ::parquet::Type::FIXED_LEN_BYTE_ARRAY:
+ type = std::make_shared<DataTypeString>();
+ break;
+ case ::parquet::Type::INT96:
+ type = std::make_shared<DataTypeString>();
+ break;
+ default:
+ return nullptr;
+ }
+ return nullable ? make_nullable(type) : type;
+}
+
+DataTypePtr direct_flat_primitive_doris_type(const ::parquet::ColumnDescriptor* column) {
+ if (column == nullptr || column->max_repetition_level() != 0 ||
+ column->max_definition_level() > 1 || has_non_physical_annotation(column)) {
+ return nullptr;
+ }
+
+ const bool nullable = column->max_definition_level() > 0;
+ switch (column->physical_type()) {
+ case ::parquet::Type::BOOLEAN:
+ return create_type(TYPE_BOOLEAN, nullable);
+ case ::parquet::Type::INT32:
+ return create_type(TYPE_INT, nullable);
+ case ::parquet::Type::INT64:
+ return create_type(TYPE_BIGINT, nullable);
+ case ::parquet::Type::FLOAT:
+ return create_type(TYPE_FLOAT, nullable);
+ case ::parquet::Type::DOUBLE:
+ return create_type(TYPE_DOUBLE, nullable);
+ default:
+ return nullptr;
+ }
+}
+
+bool record_reader_physical_type_supported(::parquet::Type::type physical_type) {
+ switch (physical_type) {
+ case ::parquet::Type::BOOLEAN:
+ case ::parquet::Type::INT32:
+ case ::parquet::Type::INT64:
+ case ::parquet::Type::FLOAT:
+ case ::parquet::Type::DOUBLE:
+ case ::parquet::Type::BYTE_ARRAY:
+ case ::parquet::Type::FIXED_LEN_BYTE_ARRAY:
+ return true;
+ default:
+ return false;
+ }
+}
+
+bool record_reader_integer_annotation_supported(const ::parquet::ColumnDescriptor* column,
+ const DataTypePtr& doris_type) {
+ const auto& logical_type = column->logical_type();
+ const bool has_int_logical_type =
+ logical_type != nullptr && logical_type->is_valid() && logical_type->is_int();
+ const bool has_int_converted_type =
+ column->converted_type() == ::parquet::ConvertedType::INT_8 ||
+ column->converted_type() == ::parquet::ConvertedType::UINT_8 ||
+ column->converted_type() == ::parquet::ConvertedType::INT_16 ||
+ column->converted_type() == ::parquet::ConvertedType::UINT_16 ||
+ column->converted_type() == ::parquet::ConvertedType::INT_32 ||
+ column->converted_type() == ::parquet::ConvertedType::UINT_32 ||
+ column->converted_type() == ::parquet::ConvertedType::INT_64 ||
+ column->converted_type() == ::parquet::ConvertedType::UINT_64;
+ auto primitive_type = remove_nullable(doris_type)->get_primitive_type();
+ return (has_int_logical_type || has_int_converted_type) &&
+ (primitive_type == TYPE_TINYINT || primitive_type == TYPE_SMALLINT ||
+ primitive_type == TYPE_INT || primitive_type == TYPE_BIGINT);
+}
+
+} // namespace
+
+std::string parquet_column_name(const ::parquet::ColumnDescriptor* column) {
+ if (column == nullptr) {
+ return {};
+ }
+ auto path = column->path();
+ if (path) {
+ return path->ToDotString();
+ }
+ return column->name();
+}
+
+ParquetTypeDescriptor resolve_parquet_type(const ::parquet::ColumnDescriptor* column) {
+ ParquetTypeDescriptor result;
+ if (column == nullptr) {
+ return result;
+ }
+
+ result.physical_type = column->physical_type();
+ result.converted_type = column->converted_type();
+ result.fixed_length = column->type_length();
+
+ if (auto logical_type = logical_type_to_doris_type(column, &result); logical_type != nullptr) {
+ result.doris_type = logical_type;
+ } else if (auto converted_type = converted_type_to_doris_type(column, &result);
+ converted_type != nullptr) {
+ result.doris_type = converted_type;
+ } else {
+ result.doris_type = physical_type_to_doris_type(column);
+ if (result.physical_type == ::parquet::Type::INT96) {
+ result.extra_type_info = ParquetExtraTypeInfo::IMPALA_TIMESTAMP;
+ }
+ }
+
+ result.is_string_like =
+ !result.is_decimal && (result.physical_type == ::parquet::Type::BYTE_ARRAY ||
+ result.physical_type == ::parquet::Type::FIXED_LEN_BYTE_ARRAY);
+
+ if (column->max_repetition_level() != 0 || column->max_definition_level() > 1) {
+ result.supports_record_reader = false;
+ return result;
+ }
+ if (!record_reader_physical_type_supported(result.physical_type)) {
+ result.supports_record_reader = false;
+ return result;
+ }
+ if (direct_flat_primitive_doris_type(column) != nullptr || result.is_string_like ||
+ (result.is_decimal && result.decimal_precision <= 38) ||
+ (result.is_timestamp && result.physical_type == ::parquet::Type::INT64) ||
+ record_reader_integer_annotation_supported(column, result.doris_type) ||
+ remove_nullable(result.doris_type)->get_primitive_type() == TYPE_DATEV2 ||
+ remove_nullable(result.doris_type)->get_primitive_type() == TYPE_TIMEV2) {
+ result.supports_record_reader = true;
+ }
+ return result;
+}
+
+bool supports_record_reader(const ParquetTypeDescriptor& type_descriptor) {
+ return type_descriptor.supports_record_reader;
+}
+
+} // namespace doris::parquet
diff --git a/be/src/format/new_parquet/parquet_type.h b/be/src/format/new_parquet/parquet_type.h
new file mode 100644
index 00000000000..1404f84bc36
--- /dev/null
+++ b/be/src/format/new_parquet/parquet_type.h
@@ -0,0 +1,84 @@
+// 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.
+
+#pragma once
+
+#include <parquet/types.h>
+
+#include <string>
+
+#include "core/data_type/data_type.h"
+
+namespace parquet {
+class ColumnDescriptor;
+} // namespace parquet
+
+namespace doris::parquet {
+
+// Parquet logical/converted annotation 解析后留下的额外编码信息。
+// 这对应 DuckDB ParquetColumnSchema::type_info：Doris type 只能表达最终展示类型，
+// 读值时还需要知道 decimal/timestamp/time 在 Parquet 中的物理编码方式。
+enum class ParquetExtraTypeInfo {
+ NONE,
+ DECIMAL_INT32,
+ DECIMAL_INT64,
+ DECIMAL_BYTE_ARRAY,
+ UNIT_MS,
+ UNIT_MICROS,
+ UNIT_NS,
+ IMPALA_TIMESTAMP,
+};
+
+enum class ParquetTimeUnit {
+ UNKNOWN,
+ MILLIS,
+ MICROS,
+ NANOS,
+};
+
+// Parquet file-local column descriptor 的类型解析结果。
+// 该结构只解释 Parquet physical/logical/converted type，不包含 table/global schema
+// evolution，也不依赖 Arrow internal RecordReader API。
+struct ParquetTypeDescriptor {
+ DataTypePtr doris_type;
+ ParquetExtraTypeInfo extra_type_info = ParquetExtraTypeInfo::NONE;
+ ParquetTimeUnit time_unit = ParquetTimeUnit::UNKNOWN;
+ ::parquet::Type::type physical_type = ::parquet::Type::UNDEFINED;
+ ::parquet::ConvertedType::type converted_type = ::parquet::ConvertedType::UNDEFINED;
+ int decimal_precision = -1;
+ int decimal_scale = -1;
+ int fixed_length = -1;
+ bool is_decimal = false;
+ bool is_timestamp = false;
+ bool is_string_like = false;
+ bool supports_record_reader = false;
+};
+
+// 返回 Parquet leaf column 的 file-local 展示名。
+std::string parquet_column_name(const ::parquet::ColumnDescriptor* column);
+
+// 将 Parquet file-local column descriptor 解析成 Doris file-local 类型和读值所需的
+// 编码信息。这里不做 table schema evolution；类型提升和 default/generated/partition
+// 列由 table 层处理。
+ParquetTypeDescriptor resolve_parquet_type(const ::parquet::ColumnDescriptor* column);
+
+// 判断当前阶段是否可以通过 Arrow Parquet RecordReader 读取该列。
+// 当前支持 flat primitive/string/decimal/timestamp。复杂 nested column 仍通过 children
+// 递归组合，list/map assembler 后续补齐。
+bool supports_record_reader(const ParquetTypeDescriptor& type_descriptor);
+
+} // namespace doris::parquet
diff --git a/be/src/format/new_parquet/selection_vector.h b/be/src/format/new_parquet/selection_vector.h
new file mode 100644
index 00000000000..22a9d3507e2
--- /dev/null
+++ b/be/src/format/new_parquet/selection_vector.h
@@ -0,0 +1,116 @@
+// 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.
+
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+#include <vector>
+
+#include "common/status.h"
+
+namespace doris::parquet {
+
+// 类似 DuckDB SelectionVector 的轻量行号视图。
+// 它只表达一个 batch 内被选中的 row offset，不持有 table/global schema 语义。
+// 未绑定 data 时表示 identity selection：get_index(i) == i。
+class SelectionVector {
+public:
+ using Index = uint16_t;
+
+ SelectionVector() = default;
+
+ explicit SelectionVector(size_t count) { resize(count); }
+
+ SelectionVector(Index* data, size_t count) { initialize(data, count); }
+
+ void initialize(Index* data, size_t count) {
+ _owned.clear();
+ _data = data;
+ _size = count;
+ }
+
+ void resize(size_t count) {
+ _owned.resize(count);
+ _data = _owned.data();
+ _size = count;
+ for (size_t idx = 0; idx < count; ++idx) {
+ _data[idx] = static_cast<Index>(idx);
+ }
+ }
+
+ void clear() {
+ _owned.clear();
+ _data = nullptr;
+ _size = 0;
+ }
+
+ size_t size() const { return _size; }
+
+ bool is_set() const { return _data != nullptr; }
+
+ Index* data() { return _data; }
+
+ const Index* data() const { return _data; }
+
+ size_t get_index(size_t idx) const {
+ if (_data == nullptr) {
+ return idx;
+ }
+ return _data[idx];
+ }
+
+ void set_index(size_t idx, Index value) { _data[idx] = value; }
+
+ Status verify(size_t count, int64_t batch_rows) const {
+ if (batch_rows < 0) {
+ return Status::InvalidArgument("Negative parquet selection batch rows {}", batch_rows);
+ }
+ if (count > static_cast<size_t>(batch_rows)) {
+ return Status::InvalidArgument("Parquet selection count {} exceeds batch rows {}",
+ count, batch_rows);
+ }
+ if (_data != nullptr && count > _size) {
+ return Status::InvalidArgument("Parquet selection count {} exceeds vector size {}",
+ count, _size);
+ }
+ size_t previous = 0;
+ for (size_t idx = 0; idx < count; ++idx) {
+ const size_t current = get_index(idx);
+ if (current >= static_cast<size_t>(batch_rows)) {
+ return Status::InvalidArgument(
+ "Parquet selection index {} out of range [0, {}) at position {}", current,
+ batch_rows, idx);
+ }
+ if (idx > 0 && current <= previous) {
+ return Status::InvalidArgument(
+ "Parquet selection index {} is not strictly greater than previous {} at "
+ "position {}",
+ current, previous, idx);
+ }
+ previous = current;
+ }
+ return Status::OK();
+ }
+
+private:
+ std::vector<Index> _owned;
+ Index* _data = nullptr;
+ size_t _size = 0;
+};
+
+} // namespace doris::parquet
diff --git a/be/src/format/parquet/parquet_reader.h b/be/src/format/parquet/parquet_reader.h
deleted file mode 100644
index 65227aba04c..00000000000
--- a/be/src/format/parquet/parquet_reader.h
+++ /dev/null
@@ -1,77 +0,0 @@
-// 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.
-
-#pragma once
-
-#include <vector>
-
-#include "common/status.h"
-#include "format/reader/file_reader.h"
-
-namespace doris {
-namespace io {
-struct IOContext;
-} // namespace io
-} // namespace doris
-
-namespace doris::parquet {
-
-// ParquetReader 的 file-local scan 请求。
-// 当前没有新增 Parquet-only 字段，但保留独立类型，便于后续加入 row group/page index
-// 等 Parquet 专属选项。
-struct ParquetScanRequest : public reader::FileScanRequest {};
-
-// Parquet 文件物理读取层。
-// 该类只理解 Parquet file-local schema 和 ParquetScanRequest，不理解 Iceberg/global
-// schema，不处理 table-level cast/default/generated/partition 语义。
-class ParquetReader : public reader::FileReader {
-public:
- virtual ~ParquetReader() = default;
-
- // 解析 Parquet footer 并返回 Parquet 文件自身的 schema。
- // 这里不做 Iceberg schema evolution，也不把字段转换成 table/global schema。
- Status get_schema(std::vector<reader::SchemaField>* file_schema) const override {
- // 真实实现会从 Parquet footer / schema descriptor 展开 file-local schema。
- file_schema->clear();
- return Status::OK();
- }
-
- // 初始化 Parquet 专属 scan。
- // 后续可以在 ParquetScanRequest 中扩展 row group、page index、bloom filter 等
- // Parquet-only 选项；table-level 语义仍然必须由 TableColumnMapper 提前转换。
- Status init(const ParquetScanRequest& request) {
- // 真实实现会根据 projected_file_columns、local_filters 和 reader_expression_map
- // 初始化 row group、column chunk、page reader 以及延时物化计划。
- return reader::FileReader::init(request);
- }
-
- // 读取下一批 Parquet file-local block。
- // 返回列必须保持 file-local 语义，不能在这里补 default/generated/partition 列。
- Status next(Block* file_block, size_t* rows, bool* eof) override {
- // 真实实现会输出 file-local block。stub 默认立即 EOF。
- return reader::FileReader::next(file_block, rows, eof);
- }
-
- // 通用 FileReader 初始化入口。
- // 当上层只持有 reader::FileReader 指针时会走该接口；Parquet 专属参数通过
- // ParquetScanRequest 重载表达。
- Status init(const reader::FileScanRequest& request) override {
- return reader::FileReader::init(request);
- }
-};
-
-} // namespace doris::parquet
diff --git a/be/src/format/reader/column_mapper.cpp b/be/src/format/reader/column_mapper.cpp
index 7510413d07f..0eed9d3e566 100644
--- a/be/src/format/reader/column_mapper.cpp
+++ b/be/src/format/reader/column_mapper.cpp
@@ -95,9 +95,10 @@ Status TableColumnMapper::create_scan_request(const std::map<int32_t, TableFilte
file_request->reader_expression_map.clear();
for (const auto& table_column : projected_columns) {
const auto* mapping = _find_mapping(table_column.id);
- if (mapping != nullptr && mapping->file_column_id.has_value() &&
- table_filters.count(table_column.id) == 0) {
- file_request->non_predicate_columns.push_back(*mapping->file_column_id);
+ if (mapping != nullptr && mapping->file_column_id.has_value()) {
+ if (table_filters.count(table_column.id) == 0) {
+ file_request->non_predicate_columns.push_back(*mapping->file_column_id);
+ }
}
}
RETURN_IF_ERROR(localize_filters(table_filters, file_request));
diff --git a/be/src/format/reader/expr/delete_predicate.cpp b/be/src/format/reader/expr/delete_predicate.cpp
index d1ca03a5201..01844fa8a07 100644
--- a/be/src/format/reader/expr/delete_predicate.cpp
+++ b/be/src/format/reader/expr/delete_predicate.cpp
@@ -86,7 +86,7 @@ Status DeletePredicate::execute(VExprContext* context, Block* block, int* result
DCHECK_EQ(row_ids.size(), count);
if (_deleted_rows.empty()) {
block->insert({std::move(res_col), std::make_shared<DataTypeBool>(), expr_name()});
- *result_column_id = block->get_columns().size() - 1;
+ *result_column_id = static_cast<int>(block->get_columns().size() - 1);
return Status::OK();
}
const int64_t* delete_rows = _deleted_rows.data();
@@ -106,7 +106,7 @@ Status DeletePredicate::execute(VExprContext* context, Block* block, int* result
++start_index;
}
block->insert({std::move(res_col), std::make_shared<DataTypeBool>(), expr_name()});
- *result_column_id = block->get_columns().size() - 1;
+ *result_column_id = static_cast<int>(block->get_columns().size() - 1);
return Status::OK();
}
diff --git a/be/src/format/reader/file_reader.cpp b/be/src/format/reader/file_reader.cpp
new file mode 100644
index 00000000000..daf9e2cf4f8
--- /dev/null
+++ b/be/src/format/reader/file_reader.cpp
@@ -0,0 +1,42 @@
+// 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/reader/file_reader.h"
+
+#include "io/fs/buffered_reader.h"
+#include "io/fs/tracing_file_reader.h"
+#include "runtime/runtime_state.h"
+
+namespace doris::reader {
+
+Status FileReader::init(RuntimeState* state) {
+ _init_profile();
+ SCOPED_RAW_TIMER(&_reader_statistics.file_reader_create_time);
+ ++_reader_statistics.open_file_num;
+ io::FileReaderOptions reader_options =
+ FileFactory::get_reader_options(state->query_options(), *_file_description);
+ _file_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
+ _profile, *_system_properties, *_file_description, reader_options,
+ io::DelegateReader::AccessMode::RANDOM, _io_ctx));
+ _tracing_file_reader = _io_ctx ? std::make_shared<io::TracingFileReader>(
+ _file_reader, _io_ctx->file_reader_stats)
+ : _file_reader;
+ _eof = false;
+ return Status::OK();
+}
+
+} // namespace doris::reader
diff --git a/be/src/format/reader/file_reader.h b/be/src/format/reader/file_reader.h
index 96ace67d8de..7fbdb9b576c 100644
--- a/be/src/format/reader/file_reader.h
+++ b/be/src/format/reader/file_reader.h
@@ -19,6 +19,7 @@
#include <cstddef>
#include <cstdint>
+#include <map>
#include <memory>
#include <string>
#include <utility>
@@ -27,6 +28,7 @@
#include "common/status.h"
#include "core/data_type/data_type.h"
#include "exprs/vexpr_fwd.h"
+#include "io/file_factory.h"
#include "io/fs/file_reader_writer_fwd.h"
namespace doris {
@@ -96,64 +98,75 @@ struct FileScanRequest {
std::vector<FileLocalFilter> local_filters;
// fallback path if filters cannot be localized to file-local predicates. The expression can reference projected_file_columns and partition columns.
std::vector<std::pair<ColumnId, VExprContextSPtr>> reader_expression_map;
- // partition key -> value
- std::map<std::string, Field> partition_values;
-
- // projected_columns' id is file-local column id, and they are all from file schema.
- // For example,
- // file schema: [0: id (int), 1: name (string), 2: age (int)]
- // predicate: age > 30
- // table-level projection: [name, id]
- // predicate_columns: [2]
- // non_predicate_columns: [1, 0]
- // projected_columns are columns in blocks returned to table reader: [1, 0] means only name and id are projected,
- std::vector<ColumnId> projected_columns;
};
// 文件物理读取层通用接口。
// 该接口只描述 file-local schema、file-local scan request 和 file-local block。
// TableReader/IcebergTableReader 可以通过它组合不同文件格式 reader。
+/**
+ * +-----> get_schema() -----------------+
+ * FileReader() -----> init() ----| -----> close()
+ * +-----> open() -----> get_block() ----+
+ */
class FileReader {
public:
+ struct ReaderStatistics {
+ int32_t filtered_row_groups = 0;
+ int32_t filtered_row_groups_by_min_max = 0;
+ int32_t filtered_row_groups_by_bloom_filter = 0;
+ int32_t read_row_groups = 0;
+ int64_t filtered_group_rows = 0;
+ int64_t filtered_page_rows = 0;
+ int64_t lazy_read_filtered_rows = 0;
+ int64_t read_rows = 0;
+ int64_t filtered_bytes = 0;
+ int64_t column_read_time = 0;
+ int64_t parse_meta_time = 0;
+ int64_t parse_footer_time = 0;
+ int64_t file_footer_read_calls = 0;
+ int64_t file_footer_hit_cache = 0;
+ int64_t file_reader_create_time = 0;
+ int64_t open_file_num = 0;
+ int64_t row_group_filter_time = 0;
+ int64_t page_index_filter_time = 0;
+ int64_t read_page_index_time = 0;
+ int64_t parse_page_index_time = 0;
+ int64_t predicate_filter_time = 0;
+ int64_t dict_filter_rewrite_time = 0;
+ int64_t bloom_filter_read_time = 0;
+ };
+
+ FileReader(std::unique_ptr<io::FileSystemProperties>& system_properties,
+ std::unique_ptr<io::FileDescription>& file_description,
+ std::shared_ptr<io::IOContext> io_ctx, RuntimeProfile* profile)
+ : _system_properties(std::move(system_properties)),
+ _file_description(std::move(file_description)),
+ _io_ctx(io_ctx),
+ _profile(profile) {}
virtual ~FileReader() = default;
- // 打开一个物理文件并加载文件级元数据。
- // 该方法只建立 file-local reader 状态，不接收 table schema，也不做 projection/filter
- // 规划；这些输入由 init(FileScanRequest) 提供。
- virtual Status open(io::FileReaderSPtr file, io::IOContext* io_ctx = nullptr) {
- // 真实实现会保存文件句柄、IO 上下文并读取文件元数据。
- _file = std::move(file);
- _io_ctx = io_ctx;
- _eof = false;
- return Status::OK();
- }
+ // Initialize file reader and parse file metadata.
+ virtual Status init(RuntimeState* state);
- // 返回文件自己的 schema 视图。
- // 返回结果必须是 file-local schema：列 id、类型和 children 都按文件格式展开，
- // 不在这里解释 Iceberg field id、缺失列、默认值或 generated column。
- virtual Status get_schema(std::vector<SchemaField>* file_schema) const {
- // 真实实现会展开文件格式自己的 file-local schema。
- file_schema->clear();
- return Status::OK();
- }
+ // Get file-local schema from file metadata. The file schema is determined by file format and file content, and does not contain table/global schema semantics. For example, Iceberg field id, name mapping, default/generated/partition columns are not interpreted in file reader. This method can only be called after init() successfully, but does not require open() to be called.
+ virtual Status get_schema(std::vector<SchemaField>* file_schema) const = 0;
- // 初始化一次 file-local scan。
- // request 由 TableColumnMapper 生成，只包含文件列投影、本地过滤条件和 reader
- // expression。FileReader 可以基于它初始化 row group/page/stripe 等文件格式计划。
- virtual Status init(const FileScanRequest& request) {
- // 真实实现会根据 projected columns、local filters 和 reader expressions
- // 初始化文件格式自己的物理读取计划。
- // _request.projected_file_columns = request.projected_file_columns;
- _request.local_filters = request.local_filters;
- _request.reader_expression_map = request.reader_expression_map;
+ // Open the file reader with file-local scan request. The file reader should initialize its internal state according to the request, but does not need to interpret table/global schema semantics. For example, all schema change, filter localization, default/generated/partition columns should be handled in table reader layer. This method can only be called after init() successfully.
+ virtual Status open(std::unique_ptr<FileScanRequest>& request) {
+ _request = std::move(request);
return Status::OK();
}
// 读取下一批 file-local block。
+ // 该方法只能在 init(FileScanRequest) 成功后调用。
// file_block 的列顺序和类型必须遵守 FileScanRequest，而不是 table/global schema。
- // eof 表示当前文件 reader 是否读完；多文件切换由 TableReader 负责。
- virtual Status get_block(Block* file_block, bool* eof) {
+ // rows 返回当前批次输出行数；eof 表示当前文件 reader 是否读完；多文件切换由
+ // TableReader 负责。
+ virtual Status get_block(Block* file_block, size_t* rows, bool* eof) {
// stub 默认立即 EOF。
+ if (rows != nullptr) {
+ *rows = 0;
+ }
if (eof != nullptr) {
*eof = true;
}
@@ -164,18 +177,29 @@ public:
// 关闭当前物理文件 reader 并释放文件层状态。
// 该方法不处理 table-level delete/finalize 状态，后者由 TableReader 子类管理。
virtual Status close() {
- _file.reset();
- _io_ctx = nullptr;
- _request = FileScanRequest {};
+ _file_reader.reset();
+ _tracing_file_reader.reset();
+ _io_ctx.reset();
+ _request.reset();
_eof = true;
return Status::OK();
}
protected:
- io::FileReaderSPtr _file;
- io::IOContext* _io_ctx = nullptr;
- FileScanRequest _request;
+ virtual void _init_profile() {}
+ io::FileReaderSPtr _file_reader;
+ // _tracing_file_reader wraps _file_reader.
+ // _file_reader is original file reader.
+ // _tracing_file_reader is tracing file reader with io context.
+ // If io_ctx is null, _tracing_file_reader will be the same as file_reader.
+ io::FileReaderSPtr _tracing_file_reader = nullptr;
+ std::unique_ptr<FileScanRequest> _request;
bool _eof = true;
+ ReaderStatistics _reader_statistics;
+ std::unique_ptr<io::FileSystemProperties> _system_properties;
+ std::unique_ptr<io::FileDescription> _file_description;
+ std::shared_ptr<io::IOContext> _io_ctx;
+ RuntimeProfile* _profile = nullptr;
};
} // namespace doris::reader
diff --git a/be/src/format/reader/table_reader.h b/be/src/format/reader/table_reader.h
index c3744427aa0..7572383b8ad 100644
--- a/be/src/format/reader/table_reader.h
+++ b/be/src/format/reader/table_reader.h
@@ -181,7 +181,12 @@ public:
// TODO: reuse column's memory
current_block.insert({field.type->create_column(), field.type, field.name});
}
- RETURN_IF_ERROR(_data_reader.reader->get_block(¤t_block, ¤t_eof));
+ size_t current_rows = 0;
+ RETURN_IF_ERROR(
+ _data_reader.reader->get_block(¤t_block, ¤t_rows, ¤t_eof));
+ if (current_rows == 0 && !current_eof) {
+ continue;
+ }
size_t idx = 0;
for (const auto& mapping : _data_reader.column_mapper.mappings()) {
@@ -232,15 +237,16 @@ protected:
// 打开当前具体 reader。
// 子类在这里基于当前 split/task 初始化底层 FileReader。
virtual Status open_reader() {
- _data_reader.block_schema.clear();
- RETURN_IF_ERROR(_data_reader.reader->get_schema(&_data_reader.block_schema));
- RETURN_IF_ERROR(_data_reader.column_mapper.create_mapping(
- _options.projected_columns, _partition_values, _data_reader.block_schema));
+ std::vector<SchemaField> file_schema;
+ RETURN_IF_ERROR(_data_reader.reader->get_schema(&file_schema));
+ _data_reader.block_schema = file_schema;
+ RETURN_IF_ERROR(_data_reader.column_mapper.create_mapping(_options.projected_columns,
+ _partition_values, file_schema));
- FileScanRequest file_request;
+ auto file_request = std::make_unique<FileScanRequest>();
RETURN_IF_ERROR(_data_reader.column_mapper.create_scan_request(
- _table_filters, _options.projected_columns, &file_request));
- RETURN_IF_ERROR(_data_reader.reader->init(file_request));
+ _table_filters, _options.projected_columns, file_request.get()));
+ RETURN_IF_ERROR(_data_reader.reader->open(file_request));
return Status::OK();
}
diff --git a/be/test/core/data_type_serde/data_type_serde_decoded_values_test.cpp b/be/test/core/data_type_serde/data_type_serde_decoded_values_test.cpp
new file mode 100644
index 00000000000..10f15bb28b1
--- /dev/null
+++ b/be/test/core/data_type_serde/data_type_serde_decoded_values_test.cpp
@@ -0,0 +1,278 @@
+// 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 <gtest/gtest.h>
+
+#include <cstdint>
+#include <memory>
+#include <vector>
+
+#include "core/assert_cast.h"
+#include "core/column/column_decimal.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_date_or_datetime_v2.h"
+#include "core/data_type/data_type_decimal.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/decoded_column_view.h"
+#include "core/string_ref.h"
+
+namespace doris {
+
+TEST(DataTypeSerDeDecodedValuesTest, ReadInt32Values) {
+ auto type = std::make_shared<DataTypeInt32>();
+ auto column = type->create_column();
+ const int32_t values[] = {10, -20, 30};
+
+ DecodedColumnView view;
+ view.value_kind = DecodedValueKind::INT32;
+ view.row_count = 3;
+ view.values = reinterpret_cast<const uint8_t*>(values);
+
+ auto st = type->get_serde()->read_column_from_decoded_values(*column, view);
+ ASSERT_TRUE(st.ok()) << st;
+
+ const auto& int_column = assert_cast<const ColumnInt32&>(*column);
+ ASSERT_EQ(int_column.size(), 3);
+ EXPECT_EQ(int_column.get_element(0), 10);
+ EXPECT_EQ(int_column.get_element(1), -20);
+ EXPECT_EQ(int_column.get_element(2), 30);
+}
+
+TEST(DataTypeSerDeDecodedValuesTest, ReadPrimitiveNumberValues) {
+ {
+ auto type = std::make_shared<DataTypeBool>();
+ auto column = type->create_column();
+ const bool values[] = {true, false, true};
+
+ DecodedColumnView view;
+ view.value_kind = DecodedValueKind::BOOL;
+ view.row_count = 3;
+ view.values = reinterpret_cast<const uint8_t*>(values);
+
+ auto st = type->get_serde()->read_column_from_decoded_values(*column, view);
+ ASSERT_TRUE(st.ok()) << st;
+
+ const auto& bool_column = assert_cast<const ColumnBool&>(*column);
+ ASSERT_EQ(bool_column.size(), 3);
+ EXPECT_EQ(bool_column.get_element(0), 1);
+ EXPECT_EQ(bool_column.get_element(1), 0);
+ EXPECT_EQ(bool_column.get_element(2), 1);
+ }
+ {
+ auto type = std::make_shared<DataTypeInt64>();
+ auto column = type->create_column();
+ const int64_t values[] = {10000000000L, -9L, 42L};
+
+ DecodedColumnView view;
+ view.value_kind = DecodedValueKind::INT64;
+ view.row_count = 3;
+ view.values = reinterpret_cast<const uint8_t*>(values);
+
+ auto st = type->get_serde()->read_column_from_decoded_values(*column, view);
+ ASSERT_TRUE(st.ok()) << st;
+
+ const auto& int_column = assert_cast<const ColumnInt64&>(*column);
+ ASSERT_EQ(int_column.size(), 3);
+ EXPECT_EQ(int_column.get_element(0), 10000000000L);
+ EXPECT_EQ(int_column.get_element(1), -9L);
+ EXPECT_EQ(int_column.get_element(2), 42L);
+ }
+ {
+ auto type = std::make_shared<DataTypeFloat32>();
+ auto column = type->create_column();
+ const float values[] = {1.5F, -2.25F};
+
+ DecodedColumnView view;
+ view.value_kind = DecodedValueKind::FLOAT;
+ view.row_count = 2;
+ view.values = reinterpret_cast<const uint8_t*>(values);
+
+ auto st = type->get_serde()->read_column_from_decoded_values(*column, view);
+ ASSERT_TRUE(st.ok()) << st;
+
+ const auto& float_column = assert_cast<const ColumnFloat32&>(*column);
+ ASSERT_EQ(float_column.size(), 2);
+ EXPECT_FLOAT_EQ(float_column.get_element(0), 1.5F);
+ EXPECT_FLOAT_EQ(float_column.get_element(1), -2.25F);
+ }
+ {
+ auto type = std::make_shared<DataTypeFloat64>();
+ auto column = type->create_column();
+ const double values[] = {3.5, -4.75};
+
+ DecodedColumnView view;
+ view.value_kind = DecodedValueKind::DOUBLE;
+ view.row_count = 2;
+ view.values = reinterpret_cast<const uint8_t*>(values);
+
+ auto st = type->get_serde()->read_column_from_decoded_values(*column, view);
+ ASSERT_TRUE(st.ok()) << st;
+
+ const auto& double_column = assert_cast<const ColumnFloat64&>(*column);
+ ASSERT_EQ(double_column.size(), 2);
+ EXPECT_DOUBLE_EQ(double_column.get_element(0), 3.5);
+ EXPECT_DOUBLE_EQ(double_column.get_element(1), -4.75);
+ }
+}
+
+TEST(DataTypeSerDeDecodedValuesTest, ReadStringValues) {
+ auto type = std::make_shared<DataTypeString>();
+ auto column = type->create_column();
+ std::vector<StringRef> values = {
+ StringRef("alpha", 5),
+ StringRef("beta", 4),
+ StringRef("gamma", 5),
+ };
+
+ DecodedColumnView view;
+ view.value_kind = DecodedValueKind::BINARY;
+ view.row_count = values.size();
+ view.binary_values = &values;
+
+ auto st = type->get_serde()->read_column_from_decoded_values(*column, view);
+ ASSERT_TRUE(st.ok()) << st;
+
+ const auto& string_column = assert_cast<const ColumnString&>(*column);
+ ASSERT_EQ(string_column.size(), 3);
+ EXPECT_EQ(string_column.get_data_at(0).to_string(), "alpha");
+ EXPECT_EQ(string_column.get_data_at(1).to_string(), "beta");
+ EXPECT_EQ(string_column.get_data_at(2).to_string(), "gamma");
+}
+
+TEST(DataTypeSerDeDecodedValuesTest, ReadDateAndDateTimeValues) {
+ {
+ auto type = std::make_shared<DataTypeDateV2>();
+ auto column = type->create_column();
+ const int32_t values[] = {0, 1, 18628};
+
+ DecodedColumnView view;
+ view.value_kind = DecodedValueKind::INT32;
+ view.row_count = 3;
+ view.values = reinterpret_cast<const uint8_t*>(values);
+
+ auto st = type->get_serde()->read_column_from_decoded_values(*column, view);
+ ASSERT_TRUE(st.ok()) << st;
+
+ ASSERT_EQ(column->size(), 3);
+ EXPECT_EQ(type->to_string(*column, 0), "1970-01-01");
+ EXPECT_EQ(type->to_string(*column, 1), "1970-01-02");
+ EXPECT_EQ(type->to_string(*column, 2), "2021-01-01");
+ }
+ {
+ auto type = std::make_shared<DataTypeDateTimeV2>(6);
+ auto column = type->create_column();
+ const int64_t values[] = {0, 1234567, -1};
+
+ DecodedColumnView view;
+ view.value_kind = DecodedValueKind::INT64;
+ view.time_unit = DecodedTimeUnit::MICROS;
+ view.row_count = 3;
+ view.values = reinterpret_cast<const uint8_t*>(values);
+
+ auto st = type->get_serde()->read_column_from_decoded_values(*column, view);
+ ASSERT_TRUE(st.ok()) << st;
+
+ ASSERT_EQ(column->size(), 3);
+ EXPECT_EQ(type->to_string(*column, 0), "1970-01-01 00:00:00.000000");
+ EXPECT_EQ(type->to_string(*column, 1), "1970-01-01 00:00:01.234567");
+ EXPECT_EQ(type->to_string(*column, 2), "1969-12-31 23:59:59.999999");
+ }
+}
+
+TEST(DataTypeSerDeDecodedValuesTest, ReadDecimalValues) {
+ auto type = std::make_shared<DataTypeDecimal128>(18, 2);
+ auto column = type->create_column();
+ const int64_t values[] = {12345, -67, 0};
+
+ DecodedColumnView view;
+ view.value_kind = DecodedValueKind::INT64;
+ view.row_count = 3;
+ view.values = reinterpret_cast<const uint8_t*>(values);
+ view.decimal_precision = 18;
+ view.decimal_scale = 2;
+
+ auto st = type->get_serde()->read_column_from_decoded_values(*column, view);
+ ASSERT_TRUE(st.ok()) << st;
+
+ const auto& decimal_column = assert_cast<const ColumnDecimal128V3&>(*column);
+ ASSERT_EQ(decimal_column.size(), 3);
+ EXPECT_EQ(decimal_column.get_element(0), Decimal128V3(12345));
+ EXPECT_EQ(decimal_column.get_element(1), Decimal128V3(-67));
+ EXPECT_EQ(decimal_column.get_element(2), Decimal128V3(0));
+ EXPECT_EQ(type->to_string(*column, 0), "123.45");
+ EXPECT_EQ(type->to_string(*column, 1), "-0.67");
+}
+
+TEST(DataTypeSerDeDecodedValuesTest, ReadNullableInt32Values) {
+ auto type = std::make_shared<DataTypeNullable>(std::make_shared<DataTypeInt32>());
+ auto column = type->create_column();
+ const int32_t values[] = {1, 2, 3, 4};
+ const uint8_t null_map[] = {0, 1, 0, 1};
+
+ DecodedColumnView view;
+ view.value_kind = DecodedValueKind::INT32;
+ view.row_count = 4;
+ view.values = reinterpret_cast<const uint8_t*>(values);
+ view.null_map = null_map;
+
+ auto st = type->get_serde()->read_column_from_decoded_values(*column, view);
+ ASSERT_TRUE(st.ok()) << st;
+
+ const auto& nullable_column = assert_cast<const ColumnNullable&>(*column);
+ const auto& nested_column = assert_cast<const ColumnInt32&>(nullable_column.get_nested_column());
+ ASSERT_EQ(nullable_column.size(), 4);
+ EXPECT_FALSE(nullable_column.is_null_at(0));
+ EXPECT_TRUE(nullable_column.is_null_at(1));
+ EXPECT_FALSE(nullable_column.is_null_at(2));
+ EXPECT_TRUE(nullable_column.is_null_at(3));
+ EXPECT_EQ(nested_column.get_element(0), 1);
+ EXPECT_EQ(nested_column.get_element(1), 2);
+ EXPECT_EQ(nested_column.get_element(2), 3);
+ EXPECT_EQ(nested_column.get_element(3), 4);
+}
+
+TEST(DataTypeSerDeDecodedValuesTest, RejectMismatchedValueKind) {
+ auto type = std::make_shared<DataTypeInt32>();
+ auto column = type->create_column();
+ const int64_t values[] = {1};
+
+ DecodedColumnView view;
+ view.value_kind = DecodedValueKind::INT64;
+ view.row_count = 1;
+ view.values = reinterpret_cast<const uint8_t*>(values);
+
+ auto st = type->get_serde()->read_column_from_decoded_values(*column, view);
+ EXPECT_FALSE(st.ok());
+}
+
+TEST(DataTypeSerDeDecodedValuesTest, RejectMissingValueBuffer) {
+ auto type = std::make_shared<DataTypeInt32>();
+ auto column = type->create_column();
+
+ DecodedColumnView view;
+ view.value_kind = DecodedValueKind::INT32;
+ view.row_count = 1;
+
+ auto st = type->get_serde()->read_column_from_decoded_values(*column, view);
+ EXPECT_FALSE(st.ok());
+}
+
+} // namespace doris
diff --git a/be/test/format/new_parquet/parquet_column_reader_test.cpp b/be/test/format/new_parquet/parquet_column_reader_test.cpp
new file mode 100644
index 00000000000..e4a0841f5af
--- /dev/null
+++ b/be/test/format/new_parquet/parquet_column_reader_test.cpp
@@ -0,0 +1,562 @@
+// 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 <arrow/api.h>
+#include <arrow/io/api.h>
+#include <gtest/gtest.h>
+#include <parquet/api/reader.h>
+#include <parquet/arrow/writer.h>
+
+#include <filesystem>
+#include <functional>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "core/assert_cast.h"
+#include "core/column/column_decimal.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.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/types.h"
+#include "format/new_parquet/column_reader.h"
+#include "format/new_parquet/parquet_column_schema.h"
+#include "format/new_parquet/selection_vector.h"
+
+namespace doris::parquet {
+namespace {
+
+constexpr int64_t ROW_COUNT = 5;
+
+std::shared_ptr<arrow::Array> finish_array(arrow::ArrayBuilder* builder) {
+ std::shared_ptr<arrow::Array> array;
+ EXPECT_TRUE(builder->Finish(&array).ok());
+ return array;
+}
+
+class ParquetColumnReaderTest : public testing::Test {
+protected:
+ void SetUp() override {
+ _test_dir = std::filesystem::temp_directory_path() / "doris_parquet_column_reader_test";
+ std::filesystem::remove_all(_test_dir);
+ std::filesystem::create_directories(_test_dir);
+ _file_path = (_test_dir / "reader.parquet").string();
+ write_parquet_file();
+ _file_reader = ::parquet::ParquetFileReader::OpenFile(_file_path, false);
+ auto metadata = _file_reader->metadata();
+ ASSERT_EQ(metadata->num_row_groups(), 1);
+ _row_group = _file_reader->RowGroup(0);
+ ASSERT_NE(_row_group, nullptr);
+ auto schema_descriptor = _file_reader->metadata()->schema();
+ ASSERT_NE(schema_descriptor, nullptr);
+ auto st = build_parquet_column_schema(*schema_descriptor, &_fields);
+ ASSERT_TRUE(st.ok()) << st;
+ ASSERT_EQ(_fields.size(), _expected_by_field.size());
+ }
+
+ void TearDown() override { std::filesystem::remove_all(_test_dir); }
+
+ template <typename Builder, typename Value>
+ std::shared_ptr<arrow::Array> build_required_array(const std::vector<Value>& values) {
+ Builder builder;
+ for (const auto& value : values) {
+ EXPECT_TRUE(builder.Append(value).ok());
+ }
+ return finish_array(&builder);
+ }
+
+ std::shared_ptr<arrow::Array> build_string_array(const std::vector<std::string>& values) {
+ arrow::StringBuilder builder;
+ for (const auto& value : values) {
+ EXPECT_TRUE(builder.Append(value).ok());
+ }
+ return finish_array(&builder);
+ }
+
+ std::shared_ptr<arrow::Array> build_binary_array(const std::vector<std::string>& values) {
+ arrow::BinaryBuilder builder;
+ for (const auto& value : values) {
+ EXPECT_TRUE(builder.Append(reinterpret_cast<const uint8_t*>(value.data()),
+ static_cast<int32_t>(value.size()))
+ .ok());
+ }
+ return finish_array(&builder);
+ }
+
+ std::shared_ptr<arrow::Array> build_fixed_binary_array(
+ const std::shared_ptr<arrow::DataType>& type,
+ const std::vector<std::string>& values) {
+ arrow::FixedSizeBinaryBuilder builder(type, arrow::default_memory_pool());
+ for (const auto& value : values) {
+ EXPECT_TRUE(builder.Append(reinterpret_cast<const uint8_t*>(value.data())).ok());
+ }
+ return finish_array(&builder);
+ }
+
+ std::shared_ptr<arrow::Array> build_nullable_int32_array() {
+ arrow::Int32Builder builder;
+ EXPECT_TRUE(builder.Append(1).ok());
+ EXPECT_TRUE(builder.AppendNull().ok());
+ EXPECT_TRUE(builder.Append(3).ok());
+ EXPECT_TRUE(builder.AppendNull().ok());
+ EXPECT_TRUE(builder.Append(5).ok());
+ return finish_array(&builder);
+ }
+
+ std::shared_ptr<arrow::Array> build_time32_array(const std::shared_ptr<arrow::DataType>& type,
+ const std::vector<int32_t>& values) {
+ arrow::Time32Builder builder(type, arrow::default_memory_pool());
+ for (const auto value : values) {
+ EXPECT_TRUE(builder.Append(value).ok());
+ }
+ return finish_array(&builder);
+ }
+
+ std::shared_ptr<arrow::Array> build_time64_array(const std::shared_ptr<arrow::DataType>& type,
+ const std::vector<int64_t>& values) {
+ arrow::Time64Builder builder(type, arrow::default_memory_pool());
+ for (const auto value : values) {
+ EXPECT_TRUE(builder.Append(value).ok());
+ }
+ return finish_array(&builder);
+ }
+
+ std::shared_ptr<arrow::Array> build_timestamp_array(
+ const std::shared_ptr<arrow::DataType>& type,
+ const std::vector<int64_t>& values) {
+ arrow::TimestampBuilder builder(type, arrow::default_memory_pool());
+ for (const auto value : values) {
+ EXPECT_TRUE(builder.Append(value).ok());
+ }
+ return finish_array(&builder);
+ }
+
+ std::shared_ptr<arrow::Array> build_decimal_array(
+ const std::shared_ptr<arrow::DataType>& type,
+ const std::vector<int64_t>& values) {
+ arrow::Decimal128Builder builder(type, arrow::default_memory_pool());
+ for (const auto value : values) {
+ EXPECT_TRUE(builder.Append(arrow::Decimal128(value)).ok());
+ }
+ return finish_array(&builder);
+ }
+
+ void add_field(const std::shared_ptr<arrow::Field>& field, std::shared_ptr<arrow::Array> array,
+ std::function<void(const ParquetColumnSchema&, const IColumn&)> validator) {
+ _arrow_fields.push_back(field);
+ _arrays.push_back(std::move(array));
+ _expected_by_field.push_back(std::move(validator));
+ }
+
+ void write_parquet_file() {
+ add_field(arrow::field("bool_col", arrow::boolean(), false),
+ build_required_array<arrow::BooleanBuilder, bool>(
+ {true, false, true, false, true}),
+ [](const ParquetColumnSchema& schema, const IColumn& column) {
+ EXPECT_EQ(schema.type_descriptor.physical_type, ::parquet::Type::BOOLEAN);
+ const auto& values = assert_cast<const ColumnBool&>(column);
+ EXPECT_EQ(values.get_element(0), 1);
+ EXPECT_EQ(values.get_element(1), 0);
+ EXPECT_EQ(values.get_element(4), 1);
+ });
+ add_field(arrow::field("int32_col", arrow::int32(), false),
+ build_required_array<arrow::Int32Builder, int32_t>({10, 20, 30, 40, 50}),
+ [](const ParquetColumnSchema& schema, const IColumn& column) {
+ EXPECT_EQ(schema.type_descriptor.physical_type, ::parquet::Type::INT32);
+ const auto& values = assert_cast<const ColumnInt32&>(column);
+ EXPECT_EQ(values.get_element(0), 10);
+ EXPECT_EQ(values.get_element(4), 50);
+ });
+ add_field(arrow::field("int64_col", arrow::int64(), false),
+ build_required_array<arrow::Int64Builder, int64_t>(
+ {10000000000L, -9L, 42L, 77L, 123L}),
+ [](const ParquetColumnSchema& schema, const IColumn& column) {
+ EXPECT_EQ(schema.type_descriptor.physical_type, ::parquet::Type::INT64);
+ const auto& values = assert_cast<const ColumnInt64&>(column);
+ EXPECT_EQ(values.get_element(0), 10000000000L);
+ EXPECT_EQ(values.get_element(1), -9L);
+ });
+ add_field(arrow::field("float_col", arrow::float32(), false),
+ build_required_array<arrow::FloatBuilder, float>(
+ {1.5F, -2.25F, 3.0F, 4.5F, 5.75F}),
+ [](const ParquetColumnSchema& schema, const IColumn& column) {
+ EXPECT_EQ(schema.type_descriptor.physical_type, ::parquet::Type::FLOAT);
+ const auto& values = assert_cast<const ColumnFloat32&>(column);
+ EXPECT_FLOAT_EQ(values.get_element(0), 1.5F);
+ EXPECT_FLOAT_EQ(values.get_element(1), -2.25F);
+ });
+ add_field(arrow::field("double_col", arrow::float64(), false),
+ build_required_array<arrow::DoubleBuilder, double>(
+ {3.5, -4.75, 6.0, 7.25, 8.5}),
+ [](const ParquetColumnSchema& schema, const IColumn& column) {
+ EXPECT_EQ(schema.type_descriptor.physical_type, ::parquet::Type::DOUBLE);
+ const auto& values = assert_cast<const ColumnFloat64&>(column);
+ EXPECT_DOUBLE_EQ(values.get_element(0), 3.5);
+ EXPECT_DOUBLE_EQ(values.get_element(1), -4.75);
+ });
+ add_field(arrow::field("binary_col", arrow::binary(), false),
+ build_binary_array({"bin_a", "bin_b", "bin_c", "bin_d", "bin_e"}),
+ [](const ParquetColumnSchema& schema, const IColumn& column) {
+ EXPECT_EQ(schema.type_descriptor.physical_type, ::parquet::Type::BYTE_ARRAY);
+ const auto& values = assert_cast<const ColumnString&>(column);
+ EXPECT_EQ(values.get_data_at(0).to_string(), "bin_a");
+ EXPECT_EQ(values.get_data_at(3).to_string(), "bin_d");
+ });
+ add_field(arrow::field("string_col", arrow::utf8(), false),
+ build_string_array({"alpha", "beta", "gamma", "delta", "epsilon"}),
+ [](const ParquetColumnSchema& schema, const IColumn& column) {
+ EXPECT_TRUE(schema.type_descriptor.is_string_like);
+ const auto& values = assert_cast<const ColumnString&>(column);
+ EXPECT_EQ(values.get_data_at(0).to_string(), "alpha");
+ EXPECT_EQ(values.get_data_at(4).to_string(), "epsilon");
+ });
+ add_field(arrow::field("fixed_binary_col", arrow::fixed_size_binary(4), false),
+ build_fixed_binary_array(arrow::fixed_size_binary(4),
+ {"aaaa", "bbbb", "cccc", "dddd", "eeee"}),
+ [](const ParquetColumnSchema& schema, const IColumn& column) {
+ EXPECT_EQ(schema.type_descriptor.physical_type,
+ ::parquet::Type::FIXED_LEN_BYTE_ARRAY);
+ EXPECT_EQ(schema.type_descriptor.fixed_length, 4);
+ const auto& values = assert_cast<const ColumnString&>(column);
+ EXPECT_EQ(values.get_data_at(0).to_string(), "aaaa");
+ EXPECT_EQ(values.get_data_at(2).to_string(), "cccc");
+ });
+ add_field(arrow::field("date_col", arrow::date32(), false),
+ build_required_array<arrow::Date32Builder, int32_t>({0, 1, 18628, 18629, 18630}),
+ [](const ParquetColumnSchema& schema, const IColumn& column) {
+ EXPECT_EQ(schema.type_descriptor.physical_type, ::parquet::Type::INT32);
+ EXPECT_EQ(remove_nullable(schema.type)->get_primitive_type(), TYPE_DATEV2);
+ EXPECT_EQ(schema.type->to_string(column, 0), "1970-01-01");
+ EXPECT_EQ(schema.type->to_string(column, 2), "2021-01-01");
+ });
+ add_field(arrow::field("time_millis_col", arrow::time32(arrow::TimeUnit::MILLI), false),
+ build_time32_array(arrow::time32(arrow::TimeUnit::MILLI),
+ {0, 1000, 3723004, 43200000, 86399000}),
+ [](const ParquetColumnSchema& schema, const IColumn& column) {
+ EXPECT_EQ(schema.type_descriptor.physical_type, ::parquet::Type::INT32);
+ EXPECT_EQ(remove_nullable(schema.type)->get_primitive_type(), TYPE_TIMEV2);
+ EXPECT_EQ(schema.type->to_string(column, 1), "00:00:01.000");
+ EXPECT_EQ(schema.type->to_string(column, 2), "01:02:03.004");
+ });
+ add_field(arrow::field("time_micros_col", arrow::time64(arrow::TimeUnit::MICRO), false),
+ build_time64_array(arrow::time64(arrow::TimeUnit::MICRO),
+ {0, 1000000, 3723004567, 43200000000, 86399000000}),
+ [](const ParquetColumnSchema& schema, const IColumn& column) {
+ EXPECT_EQ(schema.type_descriptor.physical_type, ::parquet::Type::INT64);
+ EXPECT_EQ(remove_nullable(schema.type)->get_primitive_type(), TYPE_TIMEV2);
+ EXPECT_EQ(schema.type->to_string(column, 1), "00:00:01.000000");
+ EXPECT_EQ(schema.type->to_string(column, 2), "01:02:03.004567");
+ });
+ add_field(arrow::field("timestamp_millis_col",
+ arrow::timestamp(arrow::TimeUnit::MILLI), false),
+ build_timestamp_array(arrow::timestamp(arrow::TimeUnit::MILLI),
+ {0, 1234, 1609459200000, 1609459201000, -1}),
+ [](const ParquetColumnSchema& schema, const IColumn& column) {
+ EXPECT_EQ(schema.type_descriptor.physical_type, ::parquet::Type::INT64);
+ EXPECT_EQ(remove_nullable(schema.type)->get_primitive_type(),
+ TYPE_DATETIMEV2);
+ EXPECT_EQ(schema.type->to_string(column, 1),
+ "1970-01-01 00:00:01.234");
+ EXPECT_EQ(schema.type->to_string(column, 4),
+ "1969-12-31 23:59:59.999");
+ });
+ add_field(arrow::field("timestamp_micros_col",
+ arrow::timestamp(arrow::TimeUnit::MICRO), false),
+ build_timestamp_array(arrow::timestamp(arrow::TimeUnit::MICRO),
+ {0, 1234567, 1609459200000000, 1609459201000000, -1}),
+ [](const ParquetColumnSchema& schema, const IColumn& column) {
+ EXPECT_EQ(schema.type_descriptor.physical_type, ::parquet::Type::INT64);
+ EXPECT_EQ(remove_nullable(schema.type)->get_primitive_type(),
+ TYPE_DATETIMEV2);
+ EXPECT_EQ(schema.type->to_string(column, 1),
+ "1970-01-01 00:00:01.234567");
+ EXPECT_EQ(schema.type->to_string(column, 4),
+ "1969-12-31 23:59:59.999999");
+ });
+ add_field(arrow::field("decimal_fixed_binary_9_2_col", arrow::decimal128(9, 2), false),
+ build_decimal_array(arrow::decimal128(9, 2), {12345, -67, 0, 987, 1000}),
+ [](const ParquetColumnSchema& schema, const IColumn& column) {
+ EXPECT_EQ(schema.type_descriptor.physical_type,
+ ::parquet::Type::FIXED_LEN_BYTE_ARRAY);
+ EXPECT_TRUE(schema.type_descriptor.is_decimal);
+ EXPECT_EQ(remove_nullable(schema.type)->get_primitive_type(),
+ TYPE_DECIMAL32);
+ const auto& values = assert_cast<const ColumnDecimal32&>(column);
+ EXPECT_EQ(values.get_element(0), Decimal32(12345));
+ EXPECT_EQ(schema.type->to_string(column, 0), "123.45");
+ });
+ add_field(arrow::field("decimal_fixed_binary_18_6_col", arrow::decimal128(18, 6), false),
+ build_decimal_array(arrow::decimal128(18, 6),
+ {1234567, -670000, 0, 9870000, 1000000}),
+ [](const ParquetColumnSchema& schema, const IColumn& column) {
+ EXPECT_EQ(schema.type_descriptor.physical_type,
+ ::parquet::Type::FIXED_LEN_BYTE_ARRAY);
+ EXPECT_TRUE(schema.type_descriptor.is_decimal);
+ EXPECT_EQ(remove_nullable(schema.type)->get_primitive_type(),
+ TYPE_DECIMAL64);
+ const auto& values = assert_cast<const ColumnDecimal64&>(column);
+ EXPECT_EQ(values.get_element(0), Decimal64(1234567));
+ EXPECT_EQ(schema.type->to_string(column, 0), "1.234567");
+ });
+ add_field(arrow::field("nullable_int_col", arrow::int32(), true),
+ build_nullable_int32_array(),
+ [](const ParquetColumnSchema& schema, const IColumn& column) {
+ EXPECT_TRUE(schema.type->is_nullable());
+ const auto& nullable_column = assert_cast<const ColumnNullable&>(column);
+ const auto& nested_column =
+ assert_cast<const ColumnInt32&>(nullable_column.get_nested_column());
+ ASSERT_EQ(nullable_column.size(), ROW_COUNT);
+ EXPECT_FALSE(nullable_column.is_null_at(0));
+ EXPECT_TRUE(nullable_column.is_null_at(1));
+ EXPECT_FALSE(nullable_column.is_null_at(2));
+ EXPECT_TRUE(nullable_column.is_null_at(3));
+ EXPECT_EQ(nested_column.get_element(0), 1);
+ EXPECT_EQ(nested_column.get_element(2), 3);
+ });
+
+ auto schema = arrow::schema(_arrow_fields);
+ auto table = arrow::Table::Make(schema, _arrays);
+
+ auto file_result = arrow::io::FileOutputStream::Open(_file_path);
+ ASSERT_TRUE(file_result.ok()) << file_result.status();
+ std::shared_ptr<arrow::io::FileOutputStream> out = *file_result;
+
+ ::parquet::WriterProperties::Builder builder;
+ builder.version(::parquet::ParquetVersion::PARQUET_2_6);
+ builder.data_page_version(::parquet::ParquetDataPageVersion::V2);
+ builder.compression(::parquet::Compression::UNCOMPRESSED);
+ PARQUET_THROW_NOT_OK(
+ ::parquet::arrow::WriteTable(*table, arrow::default_memory_pool(), out, ROW_COUNT,
+ builder.build()));
+ }
+
+ std::unique_ptr<ParquetColumnReader> create_reader(size_t field_idx) const {
+ ParquetColumnReaderFactory factory(_row_group, _file_reader->metadata()->num_columns());
+ std::unique_ptr<ParquetColumnReader> reader;
+ auto st = factory.create(*_fields[field_idx], &reader);
+ EXPECT_TRUE(st.ok()) << st;
+ return reader;
+ }
+
+ void read_and_validate(size_t field_idx) const {
+ auto reader = create_reader(field_idx);
+ MutableColumnPtr column = reader->type()->create_column();
+ int64_t rows_read = 0;
+ auto st = reader->read(ROW_COUNT, column, &rows_read);
+ ASSERT_TRUE(st.ok()) << st;
+ ASSERT_EQ(rows_read, ROW_COUNT);
+ ASSERT_EQ(column->size(), ROW_COUNT);
+ _expected_by_field[field_idx](*_fields[field_idx], *column);
+ }
+
+ std::filesystem::path _test_dir;
+ std::string _file_path;
+ std::unique_ptr<::parquet::ParquetFileReader> _file_reader;
+ std::shared_ptr<::parquet::RowGroupReader> _row_group;
+ std::vector<std::unique_ptr<ParquetColumnSchema>> _fields;
+ std::vector<std::shared_ptr<arrow::Field>> _arrow_fields;
+ std::vector<std::shared_ptr<arrow::Array>> _arrays;
+ std::vector<std::function<void(const ParquetColumnSchema&, const IColumn&)>> _expected_by_field;
+};
+
+TEST_F(ParquetColumnReaderTest, ReadAllSupportedPhysicalAndLogicalTypes) {
+ for (size_t field_idx = 0; field_idx < _fields.size(); ++field_idx) {
+ SCOPED_TRACE(_fields[field_idx]->name);
+ ASSERT_TRUE(supports_record_reader(_fields[field_idx]->type_descriptor));
+ read_and_validate(field_idx);
+ }
+}
+
+TEST_F(ParquetColumnReaderTest, SkipThenRead) {
+ auto reader = create_reader(1);
+ auto st = reader->skip(2);
+ ASSERT_TRUE(st.ok()) << st;
+
+ MutableColumnPtr column = reader->type()->create_column();
+ int64_t rows_read = 0;
+ st = reader->read(2, column, &rows_read);
+ ASSERT_TRUE(st.ok()) << st;
+ ASSERT_EQ(rows_read, 2);
+
+ const auto& int_values = assert_cast<const ColumnInt32&>(*column);
+ ASSERT_EQ(int_values.size(), 2);
+ EXPECT_EQ(int_values.get_element(0), 30);
+ EXPECT_EQ(int_values.get_element(1), 40);
+}
+
+TEST_F(ParquetColumnReaderTest, SelectReadsOnlySelectedRanges) {
+ auto reader = create_reader(1);
+ SelectionVector selection(3);
+ selection.set_index(0, 0);
+ selection.set_index(1, 2);
+ selection.set_index(2, 4);
+
+ MutableColumnPtr column = reader->type()->create_column();
+ auto st = reader->select(selection, 3, ROW_COUNT, column);
+ ASSERT_TRUE(st.ok()) << st;
+
+ const auto& int_values = assert_cast<const ColumnInt32&>(*column);
+ ASSERT_EQ(int_values.size(), 3);
+ EXPECT_EQ(int_values.get_element(0), 10);
+ EXPECT_EQ(int_values.get_element(1), 30);
+ EXPECT_EQ(int_values.get_element(2), 50);
+}
+
+TEST_F(ParquetColumnReaderTest, ResolveSupportedPhysicalAndLogicalSchemas) {
+ std::vector<::parquet::schema::NodePtr> nodes = {
+ ::parquet::schema::PrimitiveNode::Make(
+ "required_bool", ::parquet::Repetition::REQUIRED, ::parquet::Type::BOOLEAN),
+ ::parquet::schema::PrimitiveNode::Make(
+ "required_int32", ::parquet::Repetition::REQUIRED, ::parquet::Type::INT32),
+ ::parquet::schema::PrimitiveNode::Make(
+ "required_int64", ::parquet::Repetition::REQUIRED, ::parquet::Type::INT64),
+ ::parquet::schema::PrimitiveNode::Make(
+ "required_float", ::parquet::Repetition::REQUIRED, ::parquet::Type::FLOAT),
+ ::parquet::schema::PrimitiveNode::Make(
+ "required_double", ::parquet::Repetition::REQUIRED, ::parquet::Type::DOUBLE),
+ ::parquet::schema::PrimitiveNode::Make(
+ "required_binary", ::parquet::Repetition::REQUIRED, ::parquet::Type::BYTE_ARRAY),
+ ::parquet::schema::PrimitiveNode::Make(
+ "required_fixed_binary", ::parquet::Repetition::REQUIRED,
+ ::parquet::Type::FIXED_LEN_BYTE_ARRAY, ::parquet::ConvertedType::NONE, 4),
+ ::parquet::schema::PrimitiveNode::Make(
+ "optional_int32", ::parquet::Repetition::OPTIONAL, ::parquet::Type::INT32),
+ ::parquet::schema::PrimitiveNode::Make(
+ "utf8_binary", ::parquet::Repetition::REQUIRED, ::parquet::Type::BYTE_ARRAY,
+ ::parquet::ConvertedType::UTF8),
+ ::parquet::schema::PrimitiveNode::Make(
+ "enum_binary", ::parquet::Repetition::REQUIRED, ::parquet::Type::BYTE_ARRAY,
+ ::parquet::ConvertedType::ENUM),
+ ::parquet::schema::PrimitiveNode::Make(
+ "json_binary", ::parquet::Repetition::REQUIRED, ::parquet::Type::BYTE_ARRAY,
+ ::parquet::ConvertedType::JSON),
+ ::parquet::schema::PrimitiveNode::Make(
+ "bson_binary", ::parquet::Repetition::REQUIRED, ::parquet::Type::BYTE_ARRAY,
+ ::parquet::ConvertedType::BSON),
+ ::parquet::schema::PrimitiveNode::Make(
+ "decimal_int32", ::parquet::Repetition::REQUIRED, ::parquet::Type::INT32,
+ ::parquet::ConvertedType::DECIMAL, -1, 9, 2),
+ ::parquet::schema::PrimitiveNode::Make(
+ "decimal_int64", ::parquet::Repetition::REQUIRED, ::parquet::Type::INT64,
+ ::parquet::ConvertedType::DECIMAL, -1, 18, 6),
+ ::parquet::schema::PrimitiveNode::Make(
+ "decimal_binary", ::parquet::Repetition::REQUIRED, ::parquet::Type::BYTE_ARRAY,
+ ::parquet::ConvertedType::DECIMAL, -1, 18, 6),
+ ::parquet::schema::PrimitiveNode::Make(
+ "decimal_fixed_binary", ::parquet::Repetition::REQUIRED,
+ ::parquet::Type::FIXED_LEN_BYTE_ARRAY, ::parquet::ConvertedType::DECIMAL, 8,
+ 18, 6),
+ ::parquet::schema::PrimitiveNode::Make(
+ "date_int32", ::parquet::Repetition::REQUIRED, ::parquet::Type::INT32,
+ ::parquet::ConvertedType::DATE),
+ ::parquet::schema::PrimitiveNode::Make(
+ "time_millis_int32", ::parquet::Repetition::REQUIRED, ::parquet::Type::INT32,
+ ::parquet::ConvertedType::TIME_MILLIS),
+ ::parquet::schema::PrimitiveNode::Make(
+ "time_micros_int64", ::parquet::Repetition::REQUIRED, ::parquet::Type::INT64,
+ ::parquet::ConvertedType::TIME_MICROS),
+ ::parquet::schema::PrimitiveNode::Make(
+ "timestamp_millis_int64", ::parquet::Repetition::REQUIRED,
+ ::parquet::Type::INT64, ::parquet::ConvertedType::TIMESTAMP_MILLIS),
+ ::parquet::schema::PrimitiveNode::Make(
+ "timestamp_micros_int64", ::parquet::Repetition::REQUIRED,
+ ::parquet::Type::INT64, ::parquet::ConvertedType::TIMESTAMP_MICROS),
+ ::parquet::schema::PrimitiveNode::Make(
+ "int8_int32", ::parquet::Repetition::REQUIRED, ::parquet::Type::INT32,
+ ::parquet::ConvertedType::INT_8),
+ ::parquet::schema::PrimitiveNode::Make(
+ "uint8_int32", ::parquet::Repetition::REQUIRED, ::parquet::Type::INT32,
+ ::parquet::ConvertedType::UINT_8),
+ ::parquet::schema::PrimitiveNode::Make(
+ "int16_int32", ::parquet::Repetition::REQUIRED, ::parquet::Type::INT32,
+ ::parquet::ConvertedType::INT_16),
+ ::parquet::schema::PrimitiveNode::Make(
+ "uint16_int32", ::parquet::Repetition::REQUIRED, ::parquet::Type::INT32,
+ ::parquet::ConvertedType::UINT_16),
+ ::parquet::schema::PrimitiveNode::Make(
+ "int32_int32", ::parquet::Repetition::REQUIRED, ::parquet::Type::INT32,
+ ::parquet::ConvertedType::INT_32),
+ ::parquet::schema::PrimitiveNode::Make(
+ "uint32_int32", ::parquet::Repetition::REQUIRED, ::parquet::Type::INT32,
+ ::parquet::ConvertedType::UINT_32),
+ ::parquet::schema::PrimitiveNode::Make(
+ "int64_int64", ::parquet::Repetition::REQUIRED, ::parquet::Type::INT64,
+ ::parquet::ConvertedType::INT_64),
+ };
+
+ auto schema =
+ ::parquet::schema::GroupNode::Make("schema", ::parquet::Repetition::REQUIRED, nodes);
+ ::parquet::SchemaDescriptor descriptor;
+ descriptor.Init(schema);
+
+ std::vector<std::unique_ptr<ParquetColumnSchema>> fields;
+ auto st = build_parquet_column_schema(descriptor, &fields);
+ ASSERT_TRUE(st.ok()) << st;
+ ASSERT_EQ(fields.size(), nodes.size());
+
+ for (const auto& field : fields) {
+ SCOPED_TRACE(field->name);
+ ASSERT_TRUE(supports_record_reader(field->type_descriptor));
+ ASSERT_NE(field->type, nullptr);
+ }
+}
+
+TEST_F(ParquetColumnReaderTest, RejectUnsupportedPhysicalAndLogicalTypes) {
+ auto schema = ::parquet::schema::GroupNode::Make(
+ "schema", ::parquet::Repetition::REQUIRED,
+ {
+ ::parquet::schema::PrimitiveNode::Make(
+ "int96_col", ::parquet::Repetition::REQUIRED, ::parquet::Type::INT96),
+ ::parquet::schema::PrimitiveNode::Make(
+ "repeated_int32_col", ::parquet::Repetition::REPEATED,
+ ::parquet::Type::INT32),
+ ::parquet::schema::PrimitiveNode::Make(
+ "decimal256_fixed_col", ::parquet::Repetition::REQUIRED,
+ ::parquet::Type::FIXED_LEN_BYTE_ARRAY, ::parquet::ConvertedType::DECIMAL,
+ 20, 39, 6),
+ ::parquet::schema::PrimitiveNode::Make(
+ "uint64_col", ::parquet::Repetition::REQUIRED, ::parquet::Type::INT64,
+ ::parquet::ConvertedType::UINT_64),
+ ::parquet::schema::PrimitiveNode::Make(
+ "time_nanos_col", ::parquet::Repetition::REQUIRED,
+ ::parquet::LogicalType::Time(false,
+ ::parquet::LogicalType::TimeUnit::NANOS),
+ ::parquet::Type::INT64),
+ ::parquet::schema::PrimitiveNode::Make(
+ "timestamp_nanos_col", ::parquet::Repetition::REQUIRED,
+ ::parquet::LogicalType::Timestamp(
+ false, ::parquet::LogicalType::TimeUnit::NANOS),
+ ::parquet::Type::INT64),
+ });
+ ::parquet::SchemaDescriptor descriptor;
+ descriptor.Init(schema);
+
+ std::vector<std::unique_ptr<ParquetColumnSchema>> fields;
+ auto st = build_parquet_column_schema(descriptor, &fields);
+ ASSERT_TRUE(st.ok()) << st;
+ ASSERT_EQ(fields.size(), 6);
+
+ for (const auto& field : fields) {
+ SCOPED_TRACE(field->name);
+ ASSERT_FALSE(supports_record_reader(field->type_descriptor));
+ }
+}
+
+} // namespace
+} // namespace doris::parquet
diff --git a/be/test/format/new_parquet/parquet_reader_test.cpp b/be/test/format/new_parquet/parquet_reader_test.cpp
new file mode 100644
index 00000000000..2086cd8cfb7
--- /dev/null
+++ b/be/test/format/new_parquet/parquet_reader_test.cpp
@@ -0,0 +1,341 @@
+// 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 <arrow/api.h>
+#include <arrow/io/api.h>
+#include <gtest/gtest.h>
+#include <parquet/api/reader.h>
+#include <parquet/arrow/writer.h>
+
+#include <filesystem>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "core/assert_cast.h"
+#include "core/block/block.h"
+#include "core/column/column_string.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/primitive_type.h"
+#include "core/field.h"
+#include "format/new_parquet/parquet_reader.h"
+#include "format/reader/file_reader.h"
+#include "gen_cpp/Types_types.h"
+#include "io/io_common.h"
+#include "runtime/runtime_state.h"
+#include "storage/predicate/predicate_creator.h"
+
+namespace doris {
+namespace {
+
+constexpr int64_t ROW_COUNT = 5;
+
+std::shared_ptr<arrow::Array> finish_array(arrow::ArrayBuilder* builder) {
+ std::shared_ptr<arrow::Array> array;
+ EXPECT_TRUE(builder->Finish(&array).ok());
+ return array;
+}
+
+std::shared_ptr<arrow::Array> build_int32_array(const std::vector<int32_t>& values) {
+ arrow::Int32Builder builder;
+ for (const auto value : values) {
+ EXPECT_TRUE(builder.Append(value).ok());
+ }
+ return finish_array(&builder);
+}
+
+std::shared_ptr<arrow::Array> build_string_array(const std::vector<std::string>& values) {
+ arrow::StringBuilder builder;
+ for (const auto& value : values) {
+ EXPECT_TRUE(builder.Append(value).ok());
+ }
+ return finish_array(&builder);
+}
+
+void write_parquet_file(const std::string& file_path, int64_t row_group_size = ROW_COUNT) {
+ auto schema = arrow::schema({
+ arrow::field("id", arrow::int32(), false),
+ arrow::field("value", arrow::utf8(), false),
+ });
+ auto table = arrow::Table::Make(
+ schema, {build_int32_array({1, 2, 3, 4, 5}),
+ build_string_array({"one", "two", "three", "four", "five"})});
+
+ auto file_result = arrow::io::FileOutputStream::Open(file_path);
+ ASSERT_TRUE(file_result.ok()) << file_result.status();
+ std::shared_ptr<arrow::io::FileOutputStream> out = *file_result;
+
+ ::parquet::WriterProperties::Builder builder;
+ builder.version(::parquet::ParquetVersion::PARQUET_2_6);
+ builder.data_page_version(::parquet::ParquetDataPageVersion::V2);
+ builder.compression(::parquet::Compression::UNCOMPRESSED);
+ PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(
+ *table, arrow::default_memory_pool(), out, row_group_size, builder.build()));
+}
+
+Block build_file_block(const std::vector<reader::SchemaField>& schema) {
+ Block block;
+ for (const auto& field : schema) {
+ block.insert({field.type->create_column(), field.type, field.name});
+ }
+ return block;
+}
+
+class TestFileReader final : public reader::FileReader {
+public:
+ TestFileReader(std::unique_ptr<io::FileSystemProperties>& system_properties,
+ std::unique_ptr<io::FileDescription>& file_description,
+ std::shared_ptr<io::IOContext> io_ctx)
+ : reader::FileReader(system_properties, file_description, std::move(io_ctx), nullptr) {}
+
+ Status get_schema(std::vector<reader::SchemaField>* file_schema) const override {
+ file_schema->clear();
+ reader::SchemaField field;
+ field.id = 0;
+ field.name = "id";
+ field.type = std::make_shared<DataTypeInt32>();
+ file_schema->push_back(std::move(field));
+ return Status::OK();
+ }
+
+ bool has_request() const { return _request != nullptr; }
+
+ bool eof() const { return _eof; }
+};
+
+TEST(FileReaderTest, OpenStoresRequestAndCloseClearsState) {
+ auto system_properties = std::make_unique<io::FileSystemProperties>();
+ system_properties->system_type = TFileType::FILE_LOCAL;
+ auto file_description = std::make_unique<io::FileDescription>();
+ auto io_ctx = std::make_shared<io::IOContext>();
+ TestFileReader reader(system_properties, file_description, io_ctx);
+
+ auto request = std::make_unique<reader::FileScanRequest>();
+ request->non_predicate_columns.push_back(0);
+ ASSERT_TRUE(reader.open(request).ok());
+ EXPECT_EQ(request, nullptr);
+ EXPECT_TRUE(reader.has_request());
+
+ ASSERT_TRUE(reader.close().ok());
+ EXPECT_FALSE(reader.has_request());
+ EXPECT_TRUE(reader.eof());
+}
+
+class NewParquetReaderTest : public testing::Test {
+protected:
+ void SetUp() override {
+ _test_dir = std::filesystem::temp_directory_path() / "doris_new_parquet_reader_test";
+ std::filesystem::remove_all(_test_dir);
+ std::filesystem::create_directories(_test_dir);
+ _file_path = (_test_dir / "reader.parquet").string();
+ write_parquet_file(_file_path);
+ }
+
+ void TearDown() override { std::filesystem::remove_all(_test_dir); }
+
+ std::unique_ptr<parquet::ParquetReader> create_reader() const {
+ auto system_properties = std::make_unique<io::FileSystemProperties>();
+ system_properties->system_type = TFileType::FILE_LOCAL;
+ auto file_description = std::make_unique<io::FileDescription>();
+ file_description->path = _file_path;
+ file_description->file_size = static_cast<int64_t>(std::filesystem::file_size(_file_path));
+ return std::make_unique<parquet::ParquetReader>(system_properties, file_description,
+ nullptr, nullptr);
+ }
+
+ std::filesystem::path _test_dir;
+ std::string _file_path;
+};
+
+TEST_F(NewParquetReaderTest, GetSchemaReturnsFileLocalColumns) {
+ auto reader = create_reader();
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<reader::SchemaField> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ ASSERT_EQ(schema.size(), 2);
+ EXPECT_EQ(schema[0].id, 0);
+ EXPECT_EQ(schema[0].name, "id");
+ EXPECT_EQ(schema[0].type->get_primitive_type(), TYPE_INT);
+ EXPECT_EQ(schema[1].id, 1);
+ EXPECT_EQ(schema[1].name, "value");
+ EXPECT_EQ(schema[1].type->get_primitive_type(), TYPE_STRING);
+}
+
+TEST_F(NewParquetReaderTest, ReadSingleRowGroupThenEof) {
+ auto reader = create_reader();
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<reader::SchemaField> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ Block block = build_file_block(schema);
+
+ auto request = std::make_unique<reader::FileScanRequest>();
+ request->non_predicate_columns = {0, 1};
+ ASSERT_TRUE(reader->open(request).ok());
+
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ EXPECT_FALSE(eof);
+ ASSERT_EQ(rows, ROW_COUNT);
+
+ const auto& ids = assert_cast<const ColumnInt32&>(*block.get_by_position(0).column);
+ const auto& values = assert_cast<const ColumnString&>(*block.get_by_position(1).column);
+ ASSERT_EQ(ids.size(), ROW_COUNT);
+ ASSERT_EQ(values.size(), ROW_COUNT);
+ EXPECT_EQ(ids.get_element(0), 1);
+ EXPECT_EQ(ids.get_element(4), 5);
+ EXPECT_EQ(values.get_data_at(0).to_string(), "one");
+ EXPECT_EQ(values.get_data_at(4).to_string(), "five");
+
+ rows = 0;
+ eof = false;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ EXPECT_TRUE(eof);
+ EXPECT_EQ(rows, 0);
+}
+
+TEST_F(NewParquetReaderTest, ReadMultipleRowGroups) {
+ write_parquet_file(_file_path, 2);
+ auto parquet_file_reader = ::parquet::ParquetFileReader::OpenFile(_file_path, false);
+ ASSERT_EQ(parquet_file_reader->metadata()->num_row_groups(), 3);
+
+ auto reader = create_reader();
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<reader::SchemaField> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_unique<reader::FileScanRequest>();
+ request->non_predicate_columns = {0, 1};
+ ASSERT_TRUE(reader->open(request).ok());
+
+ std::vector<int32_t> ids;
+ std::vector<std::string> values;
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ if (rows == 0) {
+ continue;
+ }
+ const auto& id_column = assert_cast<const ColumnInt32&>(*block.get_by_position(0).column);
+ const auto& value_column =
+ assert_cast<const ColumnString&>(*block.get_by_position(1).column);
+ for (size_t row = 0; row < rows; ++row) {
+ ids.push_back(id_column.get_element(row));
+ values.push_back(value_column.get_data_at(row).to_string());
+ }
+ }
+
+ EXPECT_EQ(ids, std::vector<int32_t>({1, 2, 3, 4, 5}));
+ EXPECT_EQ(values, std::vector<std::string>({"one", "two", "three", "four", "five"}));
+}
+
+TEST_F(NewParquetReaderTest, ReadPredicateAndNonPredicateColumnsWithSelection) {
+ auto reader = create_reader();
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<reader::SchemaField> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ Block block = build_file_block(schema);
+
+ auto request = std::make_unique<reader::FileScanRequest>();
+ request->predicate_columns = {0};
+ request->non_predicate_columns = {1};
+ reader::FileLocalFilter filter;
+ filter.file_column_id = 0;
+ filter.predicates.push_back(create_comparison_predicate<PredicateType::GT>(
+ 0, "id", schema[0].type, Field::create_field<TYPE_INT>(2), false));
+ request->local_filters.push_back(std::move(filter));
+ ASSERT_TRUE(reader->open(request).ok());
+
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ EXPECT_FALSE(eof);
+ ASSERT_EQ(rows, 3);
+
+ const auto& ids = assert_cast<const ColumnInt32&>(*block.get_by_position(0).column);
+ const auto& values = assert_cast<const ColumnString&>(*block.get_by_position(1).column);
+ ASSERT_EQ(ids.size(), 3);
+ ASSERT_EQ(values.size(), 3);
+ EXPECT_EQ(ids.get_element(0), 3);
+ EXPECT_EQ(ids.get_element(1), 4);
+ EXPECT_EQ(ids.get_element(2), 5);
+ EXPECT_EQ(values.get_data_at(0).to_string(), "three");
+ EXPECT_EQ(values.get_data_at(1).to_string(), "four");
+ EXPECT_EQ(values.get_data_at(2).to_string(), "five");
+
+ rows = 0;
+ eof = false;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ EXPECT_TRUE(eof);
+ EXPECT_EQ(rows, 0);
+}
+
+TEST_F(NewParquetReaderTest, PredicateFiltersRowGroupsByStatistics) {
+ write_parquet_file(_file_path, 2);
+ auto parquet_file_reader = ::parquet::ParquetFileReader::OpenFile(_file_path, false);
+ ASSERT_EQ(parquet_file_reader->metadata()->num_row_groups(), 3);
+
+ auto reader = create_reader();
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<reader::SchemaField> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_unique<reader::FileScanRequest>();
+ request->non_predicate_columns = {0, 1};
+ reader::FileLocalFilter filter;
+ filter.file_column_id = 0;
+ filter.predicates.push_back(create_comparison_predicate<PredicateType::GT>(
+ 0, "id", schema[0].type, Field::create_field<TYPE_INT>(2), false));
+ request->local_filters.push_back(std::move(filter));
+ ASSERT_TRUE(reader->open(request).ok());
+
+ std::vector<int32_t> ids;
+ std::vector<std::string> values;
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ if (rows == 0) {
+ continue;
+ }
+ const auto& id_column = assert_cast<const ColumnInt32&>(*block.get_by_position(0).column);
+ const auto& value_column =
+ assert_cast<const ColumnString&>(*block.get_by_position(1).column);
+ for (size_t row = 0; row < rows; ++row) {
+ ids.push_back(id_column.get_element(row));
+ values.push_back(value_column.get_data_at(row).to_string());
+ }
+ }
+
+ EXPECT_EQ(ids, std::vector<int32_t>({3, 4, 5}));
+ EXPECT_EQ(values, std::vector<std::string>({"three", "four", "five"}));
+}
+
+} // namespace
+} // namespace doris
diff --git a/docs/doris-arrow-parquet-reader-implementation.md b/docs/doris-arrow-parquet-reader-implementation.md
new file mode 100644
index 00000000000..b3bfe5c69e0
--- /dev/null
+++ b/docs/doris-arrow-parquet-reader-implementation.md
@@ -0,0 +1,291 @@
+# Doris New Parquet Reader Design And Status
+
+This document describes the design and current implementation status of the new
+Parquet reader under `be/src/format/new_parquet/`.
+
+The goal of this PR is to build a file-local Parquet reader based on Arrow C++
+Parquet core APIs while keeping Doris-owned `Block` and `Column` as the scan
+output. It does not replace the old `vparquet` path yet.
+
+## Design Goals
+
+- Use Arrow C++ Parquet core APIs for Parquet file metadata, row group and column
+ decoding.
+- Keep `doris::parquet::ParquetReader` as a file-local reader.
+- Do not use `parquet::arrow::FileReader`, `arrow::RecordBatch`,
+ `arrow::Table` or `arrow::Array` as the scan output path.
+- Do not put Iceberg table schema, schema evolution, default columns, generated
+ columns or partition columns into `ParquetReader`.
+- Keep table schema mapping and filter localization in the reader/table layer,
+ especially `TableColumnMapper`.
+- Let the new implementation live in `be/src/format/new_parquet/` so it can
+ evolve independently from the old `be/src/format/parquet/` implementation.
+
+## Layering
+
+```text
+TableReader / IcebergTableReader
+ -> TableColumnMapper
+ -> reader::FileScanRequest
+ -> doris::parquet::ParquetReader
+ -> DorisRandomAccessFile
+ -> parquet::ParquetFileReader
+ -> parquet::RowGroupReader
+ -> parquet::ColumnReader / parquet::internal::RecordReader
+ -> Doris Block / Column
+```
+
+`ParquetReader` only consumes file-local information:
+
+- file-local schema fields;
+- file-local projection columns;
+- file-local predicate columns;
+- file-local `ColumnPredicate` and `VExprContext` filters.
+
+Any table-level cast, default value, generated column, partition value, Iceberg
+field id mapping or schema evolution rule must be handled before or after the
+file reader layer.
+
+## Code Layout
+
+```text
+be/src/format/new_parquet/parquet_reader.h
+be/src/format/new_parquet/parquet_reader.cpp
+be/src/format/new_parquet/column_reader.h
+be/src/format/new_parquet/column_reader.cpp
+be/src/format/new_parquet/parquet_statistics.h
+be/src/format/new_parquet/parquet_statistics.cpp
+```
+
+`parquet_reader.*` owns file open, schema export, scan state, row group
+scheduling, predicate-first reading and output block assembly.
+
+`column_reader.*` owns Doris column assembly for one projected Parquet field. It
+wraps Arrow Parquet column-level APIs and converts decoded values into Doris
+columns.
+
+`parquet_statistics.*` owns row group statistics pruning. Future page index,
+bloom filter and dictionary pruning should also live there rather than being
+mixed into the main scan loop.
+
+## Main Components
+
+### DorisRandomAccessFile
+
+`DorisRandomAccessFile` adapts Doris `io::FileReader` to
+`arrow::io::RandomAccessFile`.
+
+It only handles random IO and file size lookup. It does not parse Parquet schema,
+does not evaluate filters, and does not carry table-level semantics.
+
+### ParquetReaderScanState
+
+`ParquetReaderScanState` is an internal scan state stored in
+`parquet_reader.cpp`. It tracks:
+
+- Arrow random access file;
+- Arrow Parquet file reader and metadata;
+- Parquet schema descriptor;
+- selected row groups;
+- current row group reader;
+- current row group row offset;
+- projected file columns;
+- predicate columns and non-predicate output columns;
+- current row group column readers.
+
+This state is intentionally private to the Parquet reader implementation.
+
+### ParquetColumnReader
+
+`ParquetColumnReader` is Doris's file-local column reader abstraction. It is not
+the same as Arrow's `parquet::ColumnReader`.
+
+Current implementations:
+
+- `PrimitiveColumnReader`
+- `StructColumnReader`
+
+`PrimitiveColumnReader` supports both the existing Arrow
+`parquet::TypedColumnReader` path and the new
+`parquet::internal::RecordReader` path for selected primitive reads.
+
+`StructColumnReader` currently supports basic required struct assembly by
+recursively reading child readers. Complex nested selective materialization is
+not complete.
+
+### ParquetColumnReaderFactory
+
+`ParquetColumnReaderFactory` creates Doris column readers from the current row
+group's Arrow Parquet readers and the file-local `ParquetColumnSchema`.
+
+The factory centralizes reader construction so later work can add reader
+options, Dremel assemblers, selected-read policies and cache state without
+passing those details through free functions.
+
+### ParquetStatisticsUtils
+
+`ParquetStatisticsUtils` compiles file-local predicates into Parquet column
+predicate plans and evaluates row group metadata conservatively.
+
+It only understands Parquet file-local schema and Doris `ColumnPredicate`. It
+does not know Iceberg schema, slot descriptors or table schema mapping.
+
+## Scan Request Semantics
+
+The new reader consumes `reader::FileScanRequest`.
+
+Important fields:
+
+- `predicate_columns`: file-local columns that must be read first to evaluate
+ filters.
+- `non_predicate_columns`: file-local projection columns that are only read
+ after selection is known.
+- `projected_columns`: file-local columns that should appear in the output block.
+- `local_filters`: file-local filters produced by the table layer.
+- `reader_expression_map`: fallback expressions for filters that cannot be
+ represented as direct file-local predicates.
+
+The output block is still file-local. It is not a table/global schema block.
+
+## Predicate Pushdown
+
+Doris new reader uses two existing filter representations:
+
+- `ColumnPredicate`: structured single-column predicates, used for row group
+ statistics pruning and decoded value filtering.
+- `VExprContext`: expression filters, used for fallback expression evaluation
+ and residual filters.
+
+Current implementation status:
+
+- row group min/max pruning is wired through `parquet_statistics.*`;
+- supported stats types include boolean, int32, int64, float, double and
+ string/binary;
+- unsupported stats, missing stats or unsafe cases keep the row group;
+- `IS NULL` and `IS NOT NULL` pruning use null count when available;
+- page index, bloom filter and dictionary pruning are not implemented yet.
+
+Correctness rule: pruning must be conservative. If the reader cannot prove that
+a row group cannot match, it must keep the row group.
+
+## Lazy Materialization
+
+The scan loop follows a predicate-first model:
+
+1. Read predicate columns.
+2. Evaluate `ColumnPredicate` and build a selection vector.
+3. If a predicate column is also projected, reuse the decoded predicate column.
+4. Read non-predicate output columns using the selection.
+5. Assemble the file-local output block in projected column order.
+
+The current selected-read implementation uses Arrow Parquet
+`parquet::internal::RecordReader` for supported primitive columns.
+
+Why this is needed:
+
+- `parquet::TypedColumnReader::Skip` skips physical values, not SQL rows.
+- For nullable columns, row count and physical value count differ.
+- For repeated/nested columns, a row can contain multiple physical values.
+
+`RecordReader::SkipRecords` and `RecordReader::ReadRecords` provide row-level
+movement. Doris compresses the selection vector into row ranges and alternates
+skip/read operations.
+
+Current support:
+
+- selected read for primitive boolean, int32, int64, float and double when the
+ RecordReader path is available;
+- fallback path reads the whole batch and filters it when selected read is not
+ supported;
+- output columns are skipped when the selection is empty;
+- predicate columns are reused when they are also projected.
+
+Limitations:
+
+- `parquet::internal::RecordReader` is an Arrow internal/experimental API, so it
+ must remain hidden behind Doris `ParquetColumnReader`;
+- string, decimal and timestamp selected reads still need broader validation;
+- nested selected materialization needs a dedicated Dremel assembler.
+
+## Type Coverage
+
+Currently implemented:
+
+- flat required and nullable boolean;
+- flat required and nullable int32 / int64;
+- flat required and nullable float / double;
+- BYTE_ARRAY and FIXED_LEN_BYTE_ARRAY string/binary with Doris-owned memory;
+- decimal with precision up to 38 for int32, int64, byte array and fixed-length
+ byte array physical encodings;
+- INT64 timestamp millis and micros into Doris `DateTimeV2`;
+- basic required struct assembly.
+
+Not implemented or incomplete:
+
+- INT96 timestamp;
+- nanosecond timestamp;
+- TIMESTAMPTZ semantics;
+- DECIMAL256;
+- nullable struct;
+- list and map;
+- complex column pruning;
+- complex column lazy materialization.
+
+## Current Implementation Status
+
+Implemented in this PR:
+
+- new `new_parquet` module;
+- Arrow-backed Parquet file open and metadata read;
+- file-local schema export;
+- row group scheduling;
+- projected leaf reader creation;
+- primitive column decoding into Doris columns;
+- string, decimal and INT64 timestamp decoding;
+- basic struct reader;
+- row group statistics pruning skeleton and initial implementation;
+- predicate-first scan flow;
+- primitive RecordReader-backed selected materialization;
+- Debug BE build fixes.
+
+Validated:
+
+- `git diff --check`;
+- `BUILD_TYPE=DEBUG ./build.sh --be` on
+ `fedora:/home/socrates/code/doris`.
+
+## Future Work
+
+Near term:
+
+- add unit tests for primitive required/nullable selected reads;
+- validate selection edge cases: empty selection, full selection, sparse
+ selection and highly fragmented ranges;
+- add a selection-rate policy so dense selections can fall back to whole-batch
+ read plus filter;
+- stabilize string, decimal and timestamp selected reads;
+- keep Arrow internal API usage isolated in `column_reader.*`.
+
+Mid term:
+
+- implement page index pruning in `parquet_statistics.*`;
+- implement bloom filter pruning for equality predicates;
+- add dictionary-aware filtering where Arrow exposes enough metadata safely;
+- expand complex type assembly for nullable struct, list and map;
+- add tests for row group pruning correctness and unsupported-type fallback.
+
+Long term:
+
+- support nested column pruning;
+- support nested lazy materialization;
+- support page-level row range selection;
+- integrate the new file-local reader with table readers after the API boundary
+ is stable;
+- keep old `vparquet` compatibility until the new path is functionally complete.
+
+## Key Rule
+
+`ParquetReader` must remain a file-local reader. If a feature requires table
+schema, Iceberg schema evolution, partition values, default/generated columns or
+final table output semantics, it belongs in `TableColumnMapper` or
+`TableReader`, not in `be/src/format/new_parquet/`.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org