github-actions[bot] commented on code in PR #65674:
URL: https://github.com/apache/doris/pull/65674#discussion_r3608716910
##########
be/src/format_v2/parquet/parquet_reader.cpp:
##########
@@ -119,22 +119,23 @@ void collect_request_leaf_column_ids(
}
std::vector<ParquetPageCacheRange> build_page_cache_ranges(
- const ::parquet::FileMetaData& metadata,
+ const tparquet::FileMetaData& metadata,
const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
const format::FileScanRequest& request, const RowGroupScanPlan&
row_group_plan) {
std::unordered_set<int> leaf_column_ids;
collect_request_leaf_column_ids(file_schema, request, &leaf_column_ids);
std::vector<ParquetPageCacheRange> ranges;
ranges.reserve(row_group_plan.row_groups.size() * leaf_column_ids.size());
for (const auto& row_group_plan_item : row_group_plan.row_groups) {
- auto row_group_metadata =
metadata.RowGroup(row_group_plan_item.row_group_id);
- DORIS_CHECK(row_group_metadata != nullptr);
+ const auto& row_group_metadata =
metadata.row_groups[row_group_plan_item.row_group_id];
for (const auto leaf_column_id : leaf_column_ids) {
- DORIS_CHECK(leaf_column_id >= 0 && leaf_column_id <
row_group_metadata->num_columns());
- auto column_metadata =
row_group_metadata->ColumnChunk(leaf_column_id);
- DORIS_CHECK(column_metadata != nullptr);
- const int64_t offset = column_chunk_start_offset(*column_metadata);
- const int64_t size = column_metadata->total_compressed_size();
+ DORIS_CHECK(leaf_column_id >= 0 &&
+ leaf_column_id <
static_cast<int>(row_group_metadata.columns.size()));
+ const auto& column_metadata =
row_group_metadata.columns[leaf_column_id].meta_data;
+ const int64_t offset =
column_metadata.__isset.dictionary_page_offset
Review Comment:
[P1] Remove or validate this obsolete page-cache range traversal The new
metadata-safety test deliberately permits an optional `dictionary_page_offset =
-1` by falling back to the valid data-page offset in
`build_native_prefetch_ranges()`. This separate open-time helper instead
selects the set dictionary offset unconditionally and `DORIS_CHECK`s it, so
enabling the Parquet page cache turns that same file into a BE abort.
`register_page_cache_ranges()` now discards the result because native readers
cache exact pages, so remove this traversal or reuse the checked chunk-range
helper and propagate a `Status` rather than asserting on file metadata.
##########
be/src/format_v2/parquet/native_schema_desc.cpp:
##########
@@ -0,0 +1,779 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "format_v2/parquet/native_schema_desc.h"
+
+#include <ctype.h>
+
+#include <algorithm>
+#include <ostream>
+#include <utility>
+
+#include "common/cast_set.h"
+#include "common/exception.h"
+#include "common/logging.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_factory.hpp"
+#include "core/data_type/data_type_map.h"
+#include "core/data_type/data_type_struct.h"
+#include "core/data_type/define_primitive_type.h"
+#include "util/slice.h"
+#include "util/string_util.h"
+
+namespace doris::format::parquet {
+
+static bool is_group_node(const tparquet::SchemaElement& schema) {
+ return schema.num_children > 0;
+}
+
+static bool is_list_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.converted_type && schema.converted_type ==
tparquet::ConvertedType::LIST;
+}
+
+static bool is_map_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.converted_type &&
+ (schema.converted_type == tparquet::ConvertedType::MAP ||
+ schema.converted_type == tparquet::ConvertedType::MAP_KEY_VALUE);
+}
+
+static bool is_repeated_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.repetition_type &&
+ schema.repetition_type == tparquet::FieldRepetitionType::REPEATED;
+}
+
+static bool is_required_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.repetition_type &&
+ schema.repetition_type == tparquet::FieldRepetitionType::REQUIRED;
+}
+
+static bool is_optional_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.repetition_type &&
+ schema.repetition_type == tparquet::FieldRepetitionType::OPTIONAL;
+}
+
+static int num_children_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.num_children ? schema.num_children : 0;
+}
+
+static Status validate_native_schema_structure(
+ const std::vector<tparquet::SchemaElement>& schemas) {
+ if (schemas.empty() || !is_group_node(schemas[0])) {
+ return Status::InvalidArgument("Wrong parquet root schema element");
+ }
+
+ struct PendingGroup {
+ size_t remaining_children;
+ size_t depth;
+ };
+ const auto root_children = num_children_node(schemas[0]);
+ if (root_children < 0 || static_cast<size_t>(root_children) >
schemas.size() - 1) {
+ return Status::InvalidArgument("Invalid parquet root child count {}",
root_children);
+ }
+ std::vector<PendingGroup> pending {{static_cast<size_t>(root_children),
0}};
+ for (size_t pos = 1; pos < schemas.size(); ++pos) {
+ while (!pending.empty() && pending.back().remaining_children == 0) {
+ pending.pop_back();
+ }
+ if (pending.empty()) {
+ return Status::InvalidArgument("Schema element {} is not reachable
from the root", pos);
+ }
+
+ const size_t depth = pending.back().depth + 1;
+ --pending.back().remaining_children;
+ const int children = num_children_node(schemas[pos]);
+ if (children < 0 || static_cast<size_t>(children) > schemas.size() -
pos - 1) {
+ return Status::InvalidArgument("Invalid child count {} at schema
element {}", children,
+ pos);
+ }
+ if (children > 0) {
+ // Bound the tree before any resize or recursion so an untrusted
footer cannot
+ // turn a tiny schema into an unbounded allocation or parser stack.
+ if (depth > MAX_NATIVE_SCHEMA_DEPTH) {
+ return Status::InvalidArgument("Parquet schema depth {}
exceeds limit {}", depth,
+ MAX_NATIVE_SCHEMA_DEPTH);
+ }
+ pending.push_back({static_cast<size_t>(children), depth});
+ }
+ }
+ while (!pending.empty() && pending.back().remaining_children == 0) {
+ pending.pop_back();
+ }
+ if (!pending.empty()) {
+ return Status::InvalidArgument("Parquet schema ended before all
children were parsed");
+ }
+ return Status::OK();
+}
+
+/**
+ * `repeated_parent_def_level` is the definition level of the first ancestor
node whose repetition_type equals REPEATED.
+ * Empty array/map values are not stored in doris columns, so have to use
`repeated_parent_def_level` to skip the
+ * empty or null values in ancestor node.
+ *
+ * For instance, considering an array of strings with 3 rows like the
following:
+ * null, [], [a, b, c]
+ * We can store four elements in data column: null, a, b, c
+ * and the offsets column is: 1, 1, 4
+ * and the null map is: 1, 0, 0
+ * For the i-th row in array column: range from `offsets[i - 1]` until
`offsets[i]` represents the elements in this row,
+ * so we can't store empty array/map values in doris data column.
+ * As a comparison, spark does not require `repeated_parent_def_level`,
+ * because the spark column stores empty array/map values , and use anther
length column to indicate empty values.
+ * Please reference:
https://github.com/apache/spark/blob/master/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetColumnVector.java
+ *
+ * Furthermore, we can also avoid store null array/map values in doris data
column.
+ * The same three rows as above, We can only store three elements in data
column: a, b, c
+ * and the offsets column is: 0, 0, 3
+ * and the null map is: 1, 0, 0
+ *
+ * Inherit the repetition and definition level from parent node, if the parent
node is repeated,
+ * we should set repeated_parent_def_level = definition_level, otherwise as
repeated_parent_def_level.
+ * @param parent parent node
+ * @param repeated_parent_def_level the first ancestor node whose
repetition_type equals REPEATED
+ */
+static void set_child_node_level(NativeFieldSchema* parent, int16_t
repeated_parent_def_level) {
+ for (auto& child : parent->children) {
+ child.repetition_level = parent->repetition_level;
+ child.definition_level = parent->definition_level;
+ child.repeated_parent_def_level = repeated_parent_def_level;
+ }
+}
+
+static bool is_struct_list_node(const tparquet::SchemaElement& schema) {
+ const std::string& name = schema.name;
+ static const Slice array_slice("array", 5);
+ static const Slice tuple_slice("_tuple", 6);
+ Slice slice(name);
+ return slice == array_slice || slice.ends_with(tuple_slice);
+}
+
+std::string NativeFieldSchema::debug_string() const {
+ std::stringstream ss;
+ ss << "NativeFieldSchema(name=" << name << ", R=" << repetition_level
+ << ", D=" << definition_level;
+ if (children.size() > 0) {
+ ss << ", type=" << data_type->get_name() << ", children=[";
+ for (int i = 0; i < children.size(); ++i) {
+ if (i != 0) {
+ ss << ", ";
+ }
+ ss << children[i].debug_string();
+ }
+ ss << "]";
+ } else {
+ ss << ", physical_type=" << physical_type;
+ ss << " , doris_type=" << data_type->get_name();
+ }
+ ss << ")";
+ return ss.str();
+}
+
+Status NativeFieldDescriptor::parse_from_thrift(
+ const std::vector<tparquet::SchemaElement>& t_schemas) {
+ _fields.clear();
+ _physical_fields.clear();
+ _name_to_field.clear();
+ RETURN_IF_ERROR(validate_native_schema_structure(t_schemas));
+ const auto& root_schema = t_schemas[0];
+ _fields.resize(root_schema.num_children);
+ _next_schema_pos = 1;
+
+ for (int i = 0; i < root_schema.num_children; ++i) {
+ RETURN_IF_ERROR(parse_node_field(t_schemas, _next_schema_pos,
&_fields[i]));
+ if (_name_to_field.find(_fields[i].name) != _name_to_field.end()) {
+ return Status::InvalidArgument("Duplicated field name: {}",
_fields[i].name);
+ }
+ _name_to_field.emplace(_fields[i].name, &_fields[i]);
+ }
+
+ if (_next_schema_pos != t_schemas.size()) {
+ return Status::InvalidArgument("Remaining {} unparsed schema elements",
+ t_schemas.size() - _next_schema_pos);
+ }
+
+ return Status::OK();
+}
+
+Status NativeFieldDescriptor::parse_node_field(
+ const std::vector<tparquet::SchemaElement>& t_schemas, size_t curr_pos,
+ NativeFieldSchema* node_field) {
+ if (curr_pos >= t_schemas.size()) {
+ return Status::InvalidArgument("Out-of-bounds index of schema
elements");
+ }
+ auto& t_schema = t_schemas[curr_pos];
+ if (is_group_node(t_schema)) {
+ // nested structure or nullable list
+ return parse_group_field(t_schemas, curr_pos, node_field);
+ }
+ if (is_repeated_node(t_schema)) {
+ // repeated <primitive-type> <name> (LIST)
+ // produce required list<element>
+ node_field->repetition_level++;
+ node_field->definition_level++;
+ node_field->children.resize(1);
+ set_child_node_level(node_field, node_field->definition_level);
+ auto child = &node_field->children[0];
+ parse_physical_field(t_schema, false, child);
+
+ node_field->name = t_schema.name;
+ node_field->lower_case_name = to_lower(t_schema.name);
+ node_field->data_type =
std::make_shared<DataTypeArray>(make_nullable(child->data_type));
+ _next_schema_pos = curr_pos + 1;
+ node_field->field_id = t_schema.__isset.field_id ? t_schema.field_id :
-1;
+ } else {
+ bool is_optional = is_optional_node(t_schema);
+ if (is_optional) {
+ node_field->definition_level++;
+ }
+ parse_physical_field(t_schema, is_optional, node_field);
+ _next_schema_pos = curr_pos + 1;
+ }
+ return Status::OK();
+}
+
+void NativeFieldDescriptor::parse_physical_field(const
tparquet::SchemaElement& physical_schema,
+ bool is_nullable,
+ NativeFieldSchema*
physical_field) {
+ physical_field->name = physical_schema.name;
+ physical_field->lower_case_name = to_lower(physical_field->name);
+ physical_field->parquet_schema = physical_schema;
+ physical_field->physical_type = physical_schema.type;
+ physical_field->column_id = NATIVE_UNASSIGNED_COLUMN_ID; // Initialize
column_id
+ _physical_fields.push_back(physical_field);
+ physical_field->physical_column_index =
cast_set<int>(_physical_fields.size() - 1);
+ auto type = get_doris_type(physical_schema, is_nullable);
+ physical_field->data_type = type.first;
+ physical_field->is_type_compatibility = type.second;
+ physical_field->field_id = physical_schema.__isset.field_id ?
physical_schema.field_id : -1;
+}
+
+std::pair<DataTypePtr, bool> NativeFieldDescriptor::get_doris_type(
+ const tparquet::SchemaElement& physical_schema, bool nullable) {
+ std::pair<DataTypePtr, bool> ans = {std::make_shared<DataTypeNothing>(),
false};
+ try {
+ if (physical_schema.__isset.logicalType) {
+ ans = convert_to_doris_type(physical_schema.logicalType, nullable);
+ } else if (physical_schema.__isset.converted_type) {
+ ans = convert_to_doris_type(physical_schema, nullable);
+ }
+ } catch (...) {
+ // now the Not supported exception are ignored
+ // so those byte_array maybe be treated as varbinary(now) :
string(before)
+ }
+ if (ans.first->get_primitive_type() == PrimitiveType::INVALID_TYPE) {
+ switch (physical_schema.type) {
+ case tparquet::Type::BOOLEAN:
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_BOOLEAN, nullable);
+ break;
+ case tparquet::Type::INT32:
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_INT,
nullable);
+ break;
+ case tparquet::Type::INT64:
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_BIGINT, nullable);
+ break;
+ case tparquet::Type::INT96:
+ if (_enable_mapping_timestamp_tz) {
+ // treat INT96 as TIMESTAMPTZ
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_TIMESTAMPTZ, nullable,
+ 0, 6);
+ } else {
+ // in most cases, it's a nano timestamp
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_DATETIMEV2, nullable,
+ 0, 6);
+ }
+ break;
+ case tparquet::Type::FLOAT:
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_FLOAT, nullable);
+ break;
+ case tparquet::Type::DOUBLE:
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_DOUBLE, nullable);
+ break;
+ case tparquet::Type::BYTE_ARRAY:
+ if (_enable_mapping_varbinary) {
+ // if physical_schema not set logicalType and converted_type,
+ // we treat BYTE_ARRAY as VARBINARY by default, so that we can
read all data directly.
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_VARBINARY, nullable);
+ } else {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_STRING, nullable);
+ }
+ break;
+ case tparquet::Type::FIXED_LEN_BYTE_ARRAY:
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_STRING, nullable);
+ break;
+ default:
+ throw Exception(Status::InternalError("Not supported parquet
logicalType{}",
+ physical_schema.type));
+ break;
+ }
+ }
+ return ans;
+}
+
+std::pair<DataTypePtr, bool> NativeFieldDescriptor::convert_to_doris_type(
+ tparquet::LogicalType logicalType, bool nullable) {
+ std::pair<DataTypePtr, bool> ans = {std::make_shared<DataTypeNothing>(),
false};
+ bool& is_type_compatibility = ans.second;
+ if (logicalType.__isset.STRING) {
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_STRING,
nullable);
+ } else if (logicalType.__isset.DECIMAL) {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_DECIMAL128I, nullable,
+
logicalType.DECIMAL.precision,
+
logicalType.DECIMAL.scale);
+ } else if (logicalType.__isset.DATE) {
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_DATEV2,
nullable);
+ } else if (logicalType.__isset.INTEGER) {
+ if (logicalType.INTEGER.isSigned) {
+ if (logicalType.INTEGER.bitWidth <= 8) {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_TINYINT, nullable);
+ } else if (logicalType.INTEGER.bitWidth <= 16) {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_SMALLINT, nullable);
+ } else if (logicalType.INTEGER.bitWidth <= 32) {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_INT, nullable);
+ } else {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_BIGINT, nullable);
+ }
+ } else {
+ is_type_compatibility = true;
+ if (logicalType.INTEGER.bitWidth <= 8) {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_SMALLINT, nullable);
+ } else if (logicalType.INTEGER.bitWidth <= 16) {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_INT, nullable);
+ } else if (logicalType.INTEGER.bitWidth <= 32) {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_BIGINT, nullable);
+ } else {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_LARGEINT, nullable);
+ }
+ }
+ } else if (logicalType.__isset.TIME) {
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_TIMEV2,
nullable);
+ } else if (logicalType.__isset.TIMESTAMP) {
+ if (_enable_mapping_timestamp_tz) {
+ if (logicalType.TIMESTAMP.isAdjustedToUTC) {
+ // treat TIMESTAMP with isAdjustedToUTC as TIMESTAMPTZ
+ ans.first = DataTypeFactory::instance().create_data_type(
+ TYPE_TIMESTAMPTZ, nullable, 0,
+ logicalType.TIMESTAMP.unit.__isset.MILLIS ? 3 : 6);
+ return ans;
+ }
+ }
+ ans.first = DataTypeFactory::instance().create_data_type(
+ TYPE_DATETIMEV2, nullable, 0,
logicalType.TIMESTAMP.unit.__isset.MILLIS ? 3 : 6);
+ } else if (logicalType.__isset.JSON) {
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_STRING,
nullable);
+ } else if (logicalType.__isset.UUID) {
+ if (_enable_mapping_varbinary) {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_VARBINARY, nullable, -1,
+ -1, 16);
+ } else {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_STRING, nullable);
+ }
+ } else if (logicalType.__isset.FLOAT16) {
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_FLOAT,
nullable);
+ } else {
Review Comment:
[P1] Preserve STRING inference for ENUM and BSON leaves This mapper omits
logical ENUM/BSON, and the converted-type overload omits their legacy
annotations too. Both cases therefore throw into `get_doris_type()`'s
BYTE_ARRAY fallback; V2 enables varbinary mapping, so valid ENUM/BSON columns
are now exposed as VARBINARY. The established `parquet_type.cpp` resolver and
this native decoder both classify these annotations as STRING. Handle ENUM/BSON
alongside STRING/JSON in both overloads and cover valid annotated leaves, not
only the invalid ENUM-group case.
##########
be/src/format_v2/parquet/native_schema_desc.cpp:
##########
@@ -0,0 +1,779 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "format_v2/parquet/native_schema_desc.h"
+
+#include <ctype.h>
+
+#include <algorithm>
+#include <ostream>
+#include <utility>
+
+#include "common/cast_set.h"
+#include "common/exception.h"
+#include "common/logging.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_factory.hpp"
+#include "core/data_type/data_type_map.h"
+#include "core/data_type/data_type_struct.h"
+#include "core/data_type/define_primitive_type.h"
+#include "util/slice.h"
+#include "util/string_util.h"
+
+namespace doris::format::parquet {
+
+static bool is_group_node(const tparquet::SchemaElement& schema) {
+ return schema.num_children > 0;
+}
+
+static bool is_list_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.converted_type && schema.converted_type ==
tparquet::ConvertedType::LIST;
+}
+
+static bool is_map_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.converted_type &&
+ (schema.converted_type == tparquet::ConvertedType::MAP ||
+ schema.converted_type == tparquet::ConvertedType::MAP_KEY_VALUE);
+}
+
+static bool is_repeated_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.repetition_type &&
+ schema.repetition_type == tparquet::FieldRepetitionType::REPEATED;
+}
+
+static bool is_required_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.repetition_type &&
+ schema.repetition_type == tparquet::FieldRepetitionType::REQUIRED;
+}
+
+static bool is_optional_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.repetition_type &&
+ schema.repetition_type == tparquet::FieldRepetitionType::OPTIONAL;
+}
+
+static int num_children_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.num_children ? schema.num_children : 0;
+}
+
+static Status validate_native_schema_structure(
+ const std::vector<tparquet::SchemaElement>& schemas) {
+ if (schemas.empty() || !is_group_node(schemas[0])) {
+ return Status::InvalidArgument("Wrong parquet root schema element");
+ }
+
+ struct PendingGroup {
+ size_t remaining_children;
+ size_t depth;
+ };
+ const auto root_children = num_children_node(schemas[0]);
+ if (root_children < 0 || static_cast<size_t>(root_children) >
schemas.size() - 1) {
+ return Status::InvalidArgument("Invalid parquet root child count {}",
root_children);
+ }
+ std::vector<PendingGroup> pending {{static_cast<size_t>(root_children),
0}};
+ for (size_t pos = 1; pos < schemas.size(); ++pos) {
+ while (!pending.empty() && pending.back().remaining_children == 0) {
+ pending.pop_back();
+ }
+ if (pending.empty()) {
+ return Status::InvalidArgument("Schema element {} is not reachable
from the root", pos);
+ }
+
+ const size_t depth = pending.back().depth + 1;
+ --pending.back().remaining_children;
+ const int children = num_children_node(schemas[pos]);
Review Comment:
[P1] Validate SchemaElement kind and repetition before parsing This pass
bounds `num_children`, but it never enforces the Parquet contract that groups
set `num_children` and no `type`, primitives set `type` and no `num_children`,
and every non-root node sets `repetition_type`. With neither kind field, the
node is treated as a leaf and the generated default type (BOOLEAN) is consumed;
with no repetition bit it is treated as REQUIRED. For an optional Page V1 leaf
that can leave definition-level bytes at the value cursor instead of returning
corruption. Validate these isset invariants here before building the tree and
add missing/dual-kind plus missing-repetition footer tests.
##########
be/src/format_v2/parquet/reader/native/page_reader.cpp:
##########
@@ -0,0 +1,270 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "format_v2/parquet/reader/native/page_reader.h"
+
+#include <fmt/format.h>
+#include <gen_cpp/parquet_types.h>
+#include <stddef.h>
+#include <stdint.h>
+
+#include <algorithm>
+
+#include "common/compiler_util.h" // IWYU pragma: keep
+#include "common/config.h"
+#include "io/fs/buffered_reader.h"
+#include "runtime/runtime_profile.h"
+#include "storage/cache/page_cache.h"
+#include "util/slice.h"
+#include "util/thrift_util.h"
+
+namespace doris {
+namespace io {
+struct IOContext;
+} // namespace io
+} // namespace doris
+
+namespace doris::format::parquet::native {
+static constexpr size_t INIT_PAGE_HEADER_SIZE = 128;
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status PageReader<IN_COLLECTION, OFFSET_INDEX>::_validate_page_header(uint32_t
header_size) const {
+ if (UNLIKELY(_cur_page_header.compressed_page_size < 0 ||
+ _cur_page_header.uncompressed_page_size < 0)) {
+ return Status::Corruption("Parquet page has a negative compressed or
uncompressed size");
+ }
+ if (UNLIKELY(header_size > _end_offset - _offset ||
+ static_cast<uint64_t>(_cur_page_header.compressed_page_size) >
+ _end_offset - _offset - header_size)) {
+ // Sizes are untrusted signed Thrift fields and must fit the column
chunk before arithmetic.
+ return Status::Corruption("Parquet page payload exceeds its column
chunk");
+ }
+ if (_cur_page_header.__isset.data_page_header_v2) {
Review Comment:
[P1] Validate the page-type discriminant before choosing a layout
`PageHeader.type` is the contract that identifies which single page-specific
header is set, but this validation and `is_header_v2()` choose V2 solely from
`__isset.data_page_header_v2`. A DATA_PAGE_V2 carrying only a V1 member is
decoded as V1, a DATA_PAGE carrying a V2 member is decoded as V2, and
absent/dual members are not rejected before level, compression, and payload
cursors are selected. Require the matching member (and reject competing
members) for each page type, with malformed swapped/absent/dual-header tests.
##########
be/src/format_v2/parquet/native_schema_desc.cpp:
##########
@@ -0,0 +1,779 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "format_v2/parquet/native_schema_desc.h"
+
+#include <ctype.h>
+
+#include <algorithm>
+#include <ostream>
+#include <utility>
+
+#include "common/cast_set.h"
+#include "common/exception.h"
+#include "common/logging.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_factory.hpp"
+#include "core/data_type/data_type_map.h"
+#include "core/data_type/data_type_struct.h"
+#include "core/data_type/define_primitive_type.h"
+#include "util/slice.h"
+#include "util/string_util.h"
+
+namespace doris::format::parquet {
+
+static bool is_group_node(const tparquet::SchemaElement& schema) {
+ return schema.num_children > 0;
+}
+
+static bool is_list_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.converted_type && schema.converted_type ==
tparquet::ConvertedType::LIST;
+}
+
+static bool is_map_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.converted_type &&
+ (schema.converted_type == tparquet::ConvertedType::MAP ||
+ schema.converted_type == tparquet::ConvertedType::MAP_KEY_VALUE);
+}
+
+static bool is_repeated_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.repetition_type &&
+ schema.repetition_type == tparquet::FieldRepetitionType::REPEATED;
+}
+
+static bool is_required_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.repetition_type &&
+ schema.repetition_type == tparquet::FieldRepetitionType::REQUIRED;
+}
+
+static bool is_optional_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.repetition_type &&
+ schema.repetition_type == tparquet::FieldRepetitionType::OPTIONAL;
+}
+
+static int num_children_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.num_children ? schema.num_children : 0;
+}
+
+static Status validate_native_schema_structure(
+ const std::vector<tparquet::SchemaElement>& schemas) {
+ if (schemas.empty() || !is_group_node(schemas[0])) {
+ return Status::InvalidArgument("Wrong parquet root schema element");
+ }
+
+ struct PendingGroup {
+ size_t remaining_children;
+ size_t depth;
+ };
+ const auto root_children = num_children_node(schemas[0]);
+ if (root_children < 0 || static_cast<size_t>(root_children) >
schemas.size() - 1) {
+ return Status::InvalidArgument("Invalid parquet root child count {}",
root_children);
+ }
+ std::vector<PendingGroup> pending {{static_cast<size_t>(root_children),
0}};
+ for (size_t pos = 1; pos < schemas.size(); ++pos) {
+ while (!pending.empty() && pending.back().remaining_children == 0) {
+ pending.pop_back();
+ }
+ if (pending.empty()) {
+ return Status::InvalidArgument("Schema element {} is not reachable
from the root", pos);
+ }
+
+ const size_t depth = pending.back().depth + 1;
+ --pending.back().remaining_children;
+ const int children = num_children_node(schemas[pos]);
+ if (children < 0 || static_cast<size_t>(children) > schemas.size() -
pos - 1) {
+ return Status::InvalidArgument("Invalid child count {} at schema
element {}", children,
+ pos);
+ }
+ if (children > 0) {
+ // Bound the tree before any resize or recursion so an untrusted
footer cannot
+ // turn a tiny schema into an unbounded allocation or parser stack.
+ if (depth > MAX_NATIVE_SCHEMA_DEPTH) {
+ return Status::InvalidArgument("Parquet schema depth {}
exceeds limit {}", depth,
+ MAX_NATIVE_SCHEMA_DEPTH);
+ }
+ pending.push_back({static_cast<size_t>(children), depth});
+ }
+ }
+ while (!pending.empty() && pending.back().remaining_children == 0) {
+ pending.pop_back();
+ }
+ if (!pending.empty()) {
+ return Status::InvalidArgument("Parquet schema ended before all
children were parsed");
+ }
+ return Status::OK();
+}
+
+/**
+ * `repeated_parent_def_level` is the definition level of the first ancestor
node whose repetition_type equals REPEATED.
+ * Empty array/map values are not stored in doris columns, so have to use
`repeated_parent_def_level` to skip the
+ * empty or null values in ancestor node.
+ *
+ * For instance, considering an array of strings with 3 rows like the
following:
+ * null, [], [a, b, c]
+ * We can store four elements in data column: null, a, b, c
+ * and the offsets column is: 1, 1, 4
+ * and the null map is: 1, 0, 0
+ * For the i-th row in array column: range from `offsets[i - 1]` until
`offsets[i]` represents the elements in this row,
+ * so we can't store empty array/map values in doris data column.
+ * As a comparison, spark does not require `repeated_parent_def_level`,
+ * because the spark column stores empty array/map values , and use anther
length column to indicate empty values.
+ * Please reference:
https://github.com/apache/spark/blob/master/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetColumnVector.java
+ *
+ * Furthermore, we can also avoid store null array/map values in doris data
column.
+ * The same three rows as above, We can only store three elements in data
column: a, b, c
+ * and the offsets column is: 0, 0, 3
+ * and the null map is: 1, 0, 0
+ *
+ * Inherit the repetition and definition level from parent node, if the parent
node is repeated,
+ * we should set repeated_parent_def_level = definition_level, otherwise as
repeated_parent_def_level.
+ * @param parent parent node
+ * @param repeated_parent_def_level the first ancestor node whose
repetition_type equals REPEATED
+ */
+static void set_child_node_level(NativeFieldSchema* parent, int16_t
repeated_parent_def_level) {
+ for (auto& child : parent->children) {
+ child.repetition_level = parent->repetition_level;
+ child.definition_level = parent->definition_level;
+ child.repeated_parent_def_level = repeated_parent_def_level;
+ }
+}
+
+static bool is_struct_list_node(const tparquet::SchemaElement& schema) {
+ const std::string& name = schema.name;
+ static const Slice array_slice("array", 5);
+ static const Slice tuple_slice("_tuple", 6);
+ Slice slice(name);
+ return slice == array_slice || slice.ends_with(tuple_slice);
+}
+
+std::string NativeFieldSchema::debug_string() const {
+ std::stringstream ss;
+ ss << "NativeFieldSchema(name=" << name << ", R=" << repetition_level
+ << ", D=" << definition_level;
+ if (children.size() > 0) {
+ ss << ", type=" << data_type->get_name() << ", children=[";
+ for (int i = 0; i < children.size(); ++i) {
+ if (i != 0) {
+ ss << ", ";
+ }
+ ss << children[i].debug_string();
+ }
+ ss << "]";
+ } else {
+ ss << ", physical_type=" << physical_type;
+ ss << " , doris_type=" << data_type->get_name();
+ }
+ ss << ")";
+ return ss.str();
+}
+
+Status NativeFieldDescriptor::parse_from_thrift(
+ const std::vector<tparquet::SchemaElement>& t_schemas) {
+ _fields.clear();
+ _physical_fields.clear();
+ _name_to_field.clear();
+ RETURN_IF_ERROR(validate_native_schema_structure(t_schemas));
+ const auto& root_schema = t_schemas[0];
+ _fields.resize(root_schema.num_children);
+ _next_schema_pos = 1;
+
+ for (int i = 0; i < root_schema.num_children; ++i) {
+ RETURN_IF_ERROR(parse_node_field(t_schemas, _next_schema_pos,
&_fields[i]));
+ if (_name_to_field.find(_fields[i].name) != _name_to_field.end()) {
+ return Status::InvalidArgument("Duplicated field name: {}",
_fields[i].name);
+ }
+ _name_to_field.emplace(_fields[i].name, &_fields[i]);
+ }
+
+ if (_next_schema_pos != t_schemas.size()) {
+ return Status::InvalidArgument("Remaining {} unparsed schema elements",
+ t_schemas.size() - _next_schema_pos);
+ }
+
+ return Status::OK();
+}
+
+Status NativeFieldDescriptor::parse_node_field(
+ const std::vector<tparquet::SchemaElement>& t_schemas, size_t curr_pos,
+ NativeFieldSchema* node_field) {
+ if (curr_pos >= t_schemas.size()) {
+ return Status::InvalidArgument("Out-of-bounds index of schema
elements");
+ }
+ auto& t_schema = t_schemas[curr_pos];
+ if (is_group_node(t_schema)) {
+ // nested structure or nullable list
+ return parse_group_field(t_schemas, curr_pos, node_field);
+ }
+ if (is_repeated_node(t_schema)) {
+ // repeated <primitive-type> <name> (LIST)
+ // produce required list<element>
+ node_field->repetition_level++;
+ node_field->definition_level++;
+ node_field->children.resize(1);
+ set_child_node_level(node_field, node_field->definition_level);
+ auto child = &node_field->children[0];
+ parse_physical_field(t_schema, false, child);
+
+ node_field->name = t_schema.name;
+ node_field->lower_case_name = to_lower(t_schema.name);
+ node_field->data_type =
std::make_shared<DataTypeArray>(make_nullable(child->data_type));
+ _next_schema_pos = curr_pos + 1;
+ node_field->field_id = t_schema.__isset.field_id ? t_schema.field_id :
-1;
+ } else {
+ bool is_optional = is_optional_node(t_schema);
+ if (is_optional) {
+ node_field->definition_level++;
+ }
+ parse_physical_field(t_schema, is_optional, node_field);
+ _next_schema_pos = curr_pos + 1;
+ }
+ return Status::OK();
+}
+
+void NativeFieldDescriptor::parse_physical_field(const
tparquet::SchemaElement& physical_schema,
+ bool is_nullable,
+ NativeFieldSchema*
physical_field) {
+ physical_field->name = physical_schema.name;
+ physical_field->lower_case_name = to_lower(physical_field->name);
+ physical_field->parquet_schema = physical_schema;
+ physical_field->physical_type = physical_schema.type;
+ physical_field->column_id = NATIVE_UNASSIGNED_COLUMN_ID; // Initialize
column_id
+ _physical_fields.push_back(physical_field);
+ physical_field->physical_column_index =
cast_set<int>(_physical_fields.size() - 1);
+ auto type = get_doris_type(physical_schema, is_nullable);
+ physical_field->data_type = type.first;
+ physical_field->is_type_compatibility = type.second;
+ physical_field->field_id = physical_schema.__isset.field_id ?
physical_schema.field_id : -1;
+}
+
+std::pair<DataTypePtr, bool> NativeFieldDescriptor::get_doris_type(
+ const tparquet::SchemaElement& physical_schema, bool nullable) {
+ std::pair<DataTypePtr, bool> ans = {std::make_shared<DataTypeNothing>(),
false};
+ try {
+ if (physical_schema.__isset.logicalType) {
+ ans = convert_to_doris_type(physical_schema.logicalType, nullable);
+ } else if (physical_schema.__isset.converted_type) {
+ ans = convert_to_doris_type(physical_schema, nullable);
+ }
+ } catch (...) {
+ // now the Not supported exception are ignored
+ // so those byte_array maybe be treated as varbinary(now) :
string(before)
+ }
+ if (ans.first->get_primitive_type() == PrimitiveType::INVALID_TYPE) {
+ switch (physical_schema.type) {
+ case tparquet::Type::BOOLEAN:
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_BOOLEAN, nullable);
+ break;
+ case tparquet::Type::INT32:
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_INT,
nullable);
+ break;
+ case tparquet::Type::INT64:
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_BIGINT, nullable);
+ break;
+ case tparquet::Type::INT96:
+ if (_enable_mapping_timestamp_tz) {
+ // treat INT96 as TIMESTAMPTZ
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_TIMESTAMPTZ, nullable,
+ 0, 6);
+ } else {
+ // in most cases, it's a nano timestamp
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_DATETIMEV2, nullable,
+ 0, 6);
+ }
+ break;
+ case tparquet::Type::FLOAT:
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_FLOAT, nullable);
+ break;
+ case tparquet::Type::DOUBLE:
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_DOUBLE, nullable);
+ break;
+ case tparquet::Type::BYTE_ARRAY:
+ if (_enable_mapping_varbinary) {
+ // if physical_schema not set logicalType and converted_type,
+ // we treat BYTE_ARRAY as VARBINARY by default, so that we can
read all data directly.
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_VARBINARY, nullable);
+ } else {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_STRING, nullable);
+ }
+ break;
+ case tparquet::Type::FIXED_LEN_BYTE_ARRAY:
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_STRING, nullable);
+ break;
+ default:
+ throw Exception(Status::InternalError("Not supported parquet
logicalType{}",
+ physical_schema.type));
+ break;
+ }
+ }
+ return ans;
+}
+
+std::pair<DataTypePtr, bool> NativeFieldDescriptor::convert_to_doris_type(
+ tparquet::LogicalType logicalType, bool nullable) {
+ std::pair<DataTypePtr, bool> ans = {std::make_shared<DataTypeNothing>(),
false};
+ bool& is_type_compatibility = ans.second;
+ if (logicalType.__isset.STRING) {
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_STRING,
nullable);
+ } else if (logicalType.__isset.DECIMAL) {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_DECIMAL128I, nullable,
+
logicalType.DECIMAL.precision,
+
logicalType.DECIMAL.scale);
+ } else if (logicalType.__isset.DATE) {
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_DATEV2,
nullable);
+ } else if (logicalType.__isset.INTEGER) {
+ if (logicalType.INTEGER.isSigned) {
+ if (logicalType.INTEGER.bitWidth <= 8) {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_TINYINT, nullable);
+ } else if (logicalType.INTEGER.bitWidth <= 16) {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_SMALLINT, nullable);
+ } else if (logicalType.INTEGER.bitWidth <= 32) {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_INT, nullable);
+ } else {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_BIGINT, nullable);
+ }
+ } else {
+ is_type_compatibility = true;
+ if (logicalType.INTEGER.bitWidth <= 8) {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_SMALLINT, nullable);
+ } else if (logicalType.INTEGER.bitWidth <= 16) {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_INT, nullable);
+ } else if (logicalType.INTEGER.bitWidth <= 32) {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_BIGINT, nullable);
+ } else {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_LARGEINT, nullable);
+ }
+ }
+ } else if (logicalType.__isset.TIME) {
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_TIMEV2,
nullable);
Review Comment:
[P1] Derive TIMEV2 scale from the Parquet time unit The factory default here
is scale 0, while valid logical TIME(MILLIS/MICROS) carries fractional
precision and the native decoder preserves that unit and materializes
microseconds. The established `parquet_type.cpp` resolver reports TIMEV2(3) and
TIMEV2(6); returning TIMEV2(0) from the new inferred schema suppresses those
fractional digits at serialization. Map the unit to the same scale and add
inferred-schema/value coverage for both units.
##########
be/src/format_v2/parquet/reader/native_column_reader.cpp:
##########
@@ -0,0 +1,678 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "format_v2/parquet/reader/native_column_reader.h"
+
+#include <algorithm>
+#include <cstddef>
+#include <limits>
+#include <ranges>
+#include <string>
+#include <utility>
+
+#include "common/cast_set.h"
+#include "common/config.h"
+#include "core/assert_cast.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_array.h"
+#include "core/data_type/data_type_map.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_struct.h"
+#include "format_v2/column_data.h"
+#include "format_v2/parquet/parquet_column_schema.h"
+#include "format_v2/parquet/parquet_file_context.h"
+#include "runtime/runtime_state.h"
+
+namespace doris::format::parquet {
+namespace {
+
+constexpr size_t MAX_RETAINED_BATCH_SCRATCH_BYTES = 4UL << 20;
+
+DataTypePtr projected_type(const ParquetColumnSchema& schema,
+ const format::LocalColumnIndex* projection) {
+ if (!format::is_partial_projection(projection)) {
+ return schema.type;
+ }
+ switch (schema.kind) {
+ case ParquetColumnSchemaKind::PRIMITIVE:
+ return schema.type;
+ case ParquetColumnSchemaKind::STRUCT: {
+ DataTypes child_types;
+ Strings child_names;
+ child_types.reserve(projection->children.size());
+ child_names.reserve(projection->children.size());
+ for (const auto& child_projection : projection->children) {
+ const auto child_it = std::ranges::find_if(schema.children,
[&](const auto& child) {
+ return child->local_id == child_projection.local_id();
+ });
+ DORIS_CHECK(child_it != schema.children.end());
+ child_types.push_back(make_nullable(projected_type(**child_it,
&child_projection)));
+ child_names.push_back((*child_it)->name);
+ }
+ DataTypePtr type = std::make_shared<DataTypeStruct>(child_types,
child_names);
+ return schema.type->is_nullable() ? make_nullable(type) : type;
+ }
+ case ParquetColumnSchemaKind::LIST: {
+ DORIS_CHECK(schema.children.size() == 1);
+ const auto* child_projection =
+ format::find_child_projection(projection,
schema.children[0]->local_id);
+ DORIS_CHECK(child_projection != nullptr);
+ DataTypePtr type = std::make_shared<DataTypeArray>(
+ projected_type(*schema.children[0], child_projection));
+ return schema.type->is_nullable() ? make_nullable(type) : type;
+ }
+ case ParquetColumnSchemaKind::MAP: {
+ DORIS_CHECK(schema.children.size() == 2);
+ const auto* value_projection =
+ format::find_child_projection(projection,
schema.children[1]->local_id);
+ DORIS_CHECK(value_projection != nullptr);
+ DataTypePtr type = std::make_shared<DataTypeMap>(
+ make_nullable(schema.children[0]->type),
+ make_nullable(projected_type(*schema.children[1],
value_projection)));
+ return schema.type->is_nullable() ? make_nullable(type) : type;
+ }
+ }
+ DORIS_CHECK(false);
+ return nullptr;
+}
+
+const NativeFieldSchema* find_child_field(const NativeFieldSchema& parent,
+ const ParquetColumnSchema& child) {
+ auto field_it = std::ranges::find_if(parent.children, [&](const
NativeFieldSchema& field) {
+ return (child.parquet_field_id >= 0 && field.field_id ==
child.parquet_field_id) ||
+ field.name == child.name;
+ });
+ return field_it == parent.children.end() ? nullptr : &*field_it;
+}
+
+void collect_physical_subtree_ids(const NativeFieldSchema& field,
std::set<uint64_t>* ids) {
+ DORIS_CHECK(ids != nullptr);
+ ids->insert(field.get_column_id());
+ for (const auto& child : field.children) {
+ collect_physical_subtree_ids(child, ids);
+ }
+}
+
+void collect_projected_ids(const ParquetColumnSchema& schema,
+ const format::LocalColumnIndex* projection,
+ const NativeFieldSchema& native_field,
std::set<uint64_t>* ids) {
+ DORIS_CHECK(ids != nullptr);
+ if (!format::is_partial_projection(projection)) {
+ return;
+ }
+ for (const auto& child_projection : projection->children) {
+ const auto schema_it = std::ranges::find_if(schema.children, [&](const
auto& child) {
+ return child->local_id == child_projection.local_id();
+ });
+ DORIS_CHECK(schema_it != schema.children.end());
+ const NativeFieldSchema* child_field = find_child_field(native_field,
**schema_it);
+ DORIS_CHECK(child_field != nullptr);
+ if (format::is_full_projection(&child_projection)) {
+ // A full child path is a request for its complete physical
subtree. Keeping only the
+ // child group id makes its grandchildren SkipReadingReaders and
silently defaults
+ // STRUCT fields (or breaks ARRAY/MAP shape invariants).
+ collect_physical_subtree_ids(*child_field, ids);
+ } else {
+ ids->insert(child_field->get_column_id());
+ collect_projected_ids(**schema_it, &child_projection,
*child_field, ids);
+ }
+ }
+ if (schema.kind == ParquetColumnSchemaKind::MAP) {
+ DORIS_CHECK(!native_field.children.empty());
+ // MAP entry existence and offsets are owned by the key stream even
for value-only
+ // projections. Keep the key reader live so the native complex reader
can validate
+ // key/value entry alignment while constructing offsets.
+ ids->insert(native_field.children[0].get_column_id());
+ }
+}
+
+Status append_non_null_dictionary_values(MutableColumnPtr& target,
MutableColumnPtr values) {
+ DORIS_CHECK(target);
+ DORIS_CHECK(values);
+ const size_t value_count = values->size();
+ if (auto* nullable = check_and_get_column<ColumnNullable>(*target);
nullable != nullptr) {
+ nullable->get_nested_column().insert_range_from(*values, 0,
value_count);
+ auto& null_map = nullable->get_null_map_data();
+ null_map.resize_fill(null_map.size() + value_count, 0);
+ return Status::OK();
+ }
+ target->insert_range_from(*values, 0, value_count);
+ return Status::OK();
+}
+
+} // namespace
+
+NativeColumnReader::NativeColumnReader(const ParquetColumnSchema& schema,
+ DataTypePtr projected_type,
+ ParquetColumnReaderProfile profile)
+ : ParquetColumnReader(schema, std::move(projected_type), profile),
+ _nested(schema.kind != ParquetColumnSchemaKind::PRIMITIVE) {}
+
+NativeColumnReader::~NativeColumnReader() {
+ (void)sync_native_profile();
+}
+
+Status NativeColumnReader::create(
+ const ParquetColumnSchema& column_schema, const
format::LocalColumnIndex* projection,
+ io::FileReaderSPtr file, const NativeParquetMetadata* metadata, int
row_group_id,
+ const std::vector<RowRange>& selected_ranges,
+ const std::unordered_map<int, tparquet::OffsetIndex>& offset_indexes,
+ const cctz::time_zone* timezone, io::IOContext* io_ctx, RuntimeState*
runtime_state,
+ bool enable_page_cache, const std::string& page_cache_file_key,
+ bool enable_dictionary_filter, ParquetColumnReaderProfile profile,
+ std::unique_ptr<ParquetColumnReader>* reader) {
+ if (reader == nullptr) {
+ return Status::InvalidArgument("Native parquet reader result is null");
+ }
+ if (file == nullptr || metadata == nullptr) {
+ return Status::InvalidArgument("Native parquet file context is not
initialized");
+ }
+ if (row_group_id < 0 ||
+ row_group_id >=
static_cast<int>(metadata->to_thrift().row_groups.size())) {
+ return Status::InvalidArgument("Invalid native parquet row group {}",
row_group_id);
+ }
+ const auto& native_schema = metadata->schema();
+ if (column_schema.local_id < 0 || column_schema.local_id >=
native_schema.size()) {
+ return Status::InvalidArgument("Invalid native parquet top-level
column id {} for {}",
+ column_schema.local_id,
column_schema.name);
+ }
+ auto* field =
const_cast<NativeFieldSchema*>(native_schema.get_column(column_schema.local_id));
+ DORIS_CHECK(field != nullptr);
+ if (field->name != column_schema.name &&
+ !(field->field_id >= 0 && field->field_id ==
column_schema.parquet_field_id)) {
+ return Status::Corruption(
+ "Native/metadata parquet schema mismatch at column {}:
native={}, arrow={}",
+ column_schema.local_id, field->name, column_schema.name);
+ }
+
+ auto type = projected_type(column_schema, projection);
+ std::shared_ptr<NativeSchemaNode> schema_node;
+ RETURN_IF_ERROR(build_native_schema_node(type, column_schema,
&schema_node));
+ std::set<uint64_t> projected_ids;
+ collect_projected_ids(column_schema, projection, *field, &projected_ids);
+
+ auto native_reader = std::unique_ptr<NativeColumnReader>(
+ new NativeColumnReader(column_schema, std::move(type), profile));
+ RETURN_IF_ERROR(native_reader->init(
+ std::move(file), metadata, row_group_id, field,
std::move(schema_node),
+ std::move(projected_ids), selected_ranges, offset_indexes,
timezone, io_ctx,
+ runtime_state, enable_page_cache, page_cache_file_key,
enable_dictionary_filter));
+ *reader = std::move(native_reader);
+ return Status::OK();
+}
+
+Status NativeColumnReader::init(
+ io::FileReaderSPtr file, const NativeParquetMetadata* metadata, int
row_group_id,
+ NativeFieldSchema* field, std::shared_ptr<NativeSchemaNode>
schema_node,
+ std::set<uint64_t> projected_column_ids, const std::vector<RowRange>&
selected_ranges,
+ const std::unordered_map<int, tparquet::OffsetIndex>& offset_indexes,
+ const cctz::time_zone* timezone, io::IOContext* io_ctx, RuntimeState*
runtime_state,
+ bool enable_page_cache, const std::string& page_cache_file_key,
+ bool enable_dictionary_filter) {
+ DORIS_CHECK(file != nullptr);
+ DORIS_CHECK(metadata != nullptr);
+ DORIS_CHECK(field != nullptr);
+ DORIS_CHECK(schema_node != nullptr);
+ const auto& row_group = metadata->to_thrift().row_groups[row_group_id];
+ DORIS_CHECK(row_group.num_rows > 0);
+ _row_group_rows = row_group.num_rows;
+ _selected_ranges = selected_ranges;
+ DORIS_CHECK(!_selected_ranges.empty());
+ for (const auto& range : _selected_ranges) {
+ DORIS_CHECK(range.start >= 0);
+ DORIS_CHECK(range.length > 0);
+ DORIS_CHECK(range.start + range.length <= _row_group_rows);
+ _row_ranges.add(::doris::RowRange(range.start, range.start +
range.length));
+ }
+ // Offset indexes are immutable row-group metadata owned by
ParquetScanScheduler. Sharing them
+ // avoids retaining the full N-column page-location map once per projected
reader.
+ _offset_indexes = &offset_indexes;
+ _schema_node = std::move(schema_node);
+ _projected_column_ids = std::move(projected_column_ids);
+ _dictionary_filter_enabled = enable_dictionary_filter;
+
+ const size_t max_group_buffer = config::parquet_rowgroup_max_buffer_mb <<
20;
+ const size_t max_column_buffer = config::parquet_column_max_buffer_mb <<
20;
+ const size_t max_buffer_size = std::min(max_group_buffer,
max_column_buffer);
+ RuntimeState* native_runtime_state = runtime_state;
+ const bool runtime_page_cache_enabled =
+ runtime_state == nullptr ||
+ runtime_state->query_options().enable_parquet_file_page_cache;
+ if (runtime_page_cache_enabled != enable_page_cache) {
+ TQueryOptions query_options =
+ runtime_state == nullptr ? TQueryOptions() :
runtime_state->query_options();
+ query_options.__set_enable_parquet_file_page_cache(enable_page_cache);
+ _page_cache_runtime_state = RuntimeState::create_unique(query_options,
TQueryGlobals());
+ native_runtime_state = _page_cache_runtime_state.get();
+ }
+ const auto& thrift_metadata = metadata->to_thrift();
+ const auto compat = native::parquet_reader_compat(
+ thrift_metadata.__isset.created_by ? thrift_metadata.created_by :
"");
+ RETURN_IF_ERROR(native::ColumnReader::create(
+ std::move(file), field, row_group, _row_ranges, timezone, io_ctx,
_native_reader,
+ max_buffer_size, *_offset_indexes, native_runtime_state, false,
_projected_column_ids,
+ _filter_column_ids, page_cache_file_key, compat,
+ runtime_state != nullptr && runtime_state->enable_strict_mode()));
+ DORIS_CHECK(_native_reader != nullptr);
+ _skip_column = _type->create_column();
+ return Status::OK();
+}
+
+Status NativeColumnReader::read_with_filter(int64_t rows, const uint8_t*
filter_data,
+ bool filter_all, MutableColumnPtr&
column,
+ const DataTypePtr& output_type,
bool dictionary_ids,
+ int64_t* rows_read) {
+ DORIS_CHECK(rows >= 0);
+ DORIS_CHECK(column);
+ DORIS_CHECK(output_type != nullptr);
+ DORIS_CHECK(rows_read != nullptr);
+ *rows_read = 0;
+ if (rows == 0) {
+ return Status::OK();
+ }
+
+ native::FilterMap filter;
+ RETURN_IF_ERROR(filter.init(filter_data, static_cast<size_t>(rows),
filter_all));
+ _native_reader->reset_filter_map_index();
+ ColumnPtr native_column(std::move(column));
+ bool eof = false;
+ int64_t native_calls = 0;
+ int64_t consecutive_empty_calls = 0;
+ while (*rows_read < rows && !eof) {
+ ++native_calls;
+ size_t loop_rows = 0;
+ RETURN_IF_ERROR(_native_reader->read_column_data(
+ native_column, output_type, _schema_node, filter,
+ static_cast<size_t>(rows - *rows_read), &loop_rows, &eof,
dictionary_ids));
+ if (loop_rows == 0 && !eof) {
+ // A selected RowRanges plan may reject the current data page
completely. V1 advances
+ // the page cursor and deliberately returns zero rows so the
caller can request the
+ // next page. Bound consecutive empty transitions by the Row Group
row count to retain
+ // a deterministic corruption exit if a decoder ever stops
advancing.
+ if (++consecutive_empty_calls > _row_group_rows + 1) {
+ column = IColumn::mutate(std::move(native_column));
+ return Status::Corruption("Native parquet reader made no
progress for column {}",
+ _name);
+ }
+ continue;
+ }
+ consecutive_empty_calls = 0;
+ *rows_read += static_cast<int64_t>(loop_rows);
+ }
+ column = IColumn::mutate(std::move(native_column));
+ if (_profile.native_read_calls != nullptr) {
+ COUNTER_UPDATE(_profile.native_read_calls, native_calls);
+ }
+ if (_nested && _profile.nested_batches != nullptr) {
+ COUNTER_UPDATE(_profile.nested_batches, 1);
+ }
+ // Retained-capacity inspection walks the native reader tree. Check it
periodically instead of
+ // on every small batch; row-group destruction is still the hard lifetime
bound for scratch.
+ constexpr size_t SCRATCH_CHECK_BATCH_INTERVAL = 16;
+ if (++_batches_since_scratch_check >= SCRATCH_CHECK_BATCH_INTERVAL) {
+
_native_reader->release_batch_scratch(MAX_RETAINED_BATCH_SCRATCH_BYTES);
+ _batches_since_scratch_check = 0;
+ }
+ if (*rows_read != rows) {
+ return Status::Corruption("Native parquet reader returned {} rows,
expected {} for {}",
+ *rows_read, rows, _name);
+ }
+ return Status::OK();
+}
+
+Status NativeColumnReader::validate_selected_span(int64_t rows) {
+ DORIS_CHECK(rows >= 0);
+ while (_selected_range_idx < _selected_ranges.size()) {
+ const auto& range = _selected_ranges[_selected_range_idx];
+ const int64_t range_end = range.start + range.length;
+ if (_logical_row_position < range_end) {
+ break;
+ }
+ ++_selected_range_idx;
+ }
+ if (_selected_range_idx >= _selected_ranges.size()) {
+ return Status::Corruption("Native parquet read past selected ranges
for column {}", _name);
+ }
+ const auto& range = _selected_ranges[_selected_range_idx];
+ if (_logical_row_position < range.start ||
+ rows > range.start + range.length - _logical_row_position) {
+ return Status::Corruption(
+ "Native parquet read [{}, {}) crosses selected range [{}, {})
for column {}",
+ _logical_row_position, _logical_row_position + rows,
range.start,
+ range.start + range.length, _name);
+ }
+ return Status::OK();
+}
+
+void NativeColumnReader::advance_selected_span(int64_t rows) {
+ _logical_row_position += rows;
+ while (_selected_range_idx < _selected_ranges.size() &&
+ _logical_row_position >=
_selected_ranges[_selected_range_idx].start +
+
_selected_ranges[_selected_range_idx].length) {
+ ++_selected_range_idx;
+ }
+}
+
+Status NativeColumnReader::read(int64_t rows, MutableColumnPtr& column,
int64_t* rows_read) {
+ RETURN_IF_ERROR(validate_selected_span(rows));
+ RETURN_IF_ERROR(read_with_filter(rows, nullptr, false, column, _type,
false, rows_read));
+ advance_selected_span(*rows_read);
+ update_reader_read_rows(*rows_read);
+ return Status::OK();
+}
+
+Status NativeColumnReader::skip(int64_t rows) {
+ if (rows <= 0) {
+ return Status::OK();
+ }
+ DORIS_CHECK(_logical_row_position <= _row_group_rows - rows);
+ int64_t remaining = rows;
+ int64_t native_skipped_rows = 0;
+ while (remaining > 0) {
+ while (_selected_range_idx < _selected_ranges.size() &&
+ _logical_row_position >=
_selected_ranges[_selected_range_idx].start +
+
_selected_ranges[_selected_range_idx].length) {
+ ++_selected_range_idx;
+ }
+ if (_selected_range_idx >= _selected_ranges.size()) {
+ _logical_row_position += remaining;
+ break;
+ }
+ const auto& range = _selected_ranges[_selected_range_idx];
+ if (_logical_row_position < range.start) {
+ const int64_t gap = std::min(remaining, range.start -
_logical_row_position);
+ _logical_row_position += gap;
+ remaining -= gap;
+ continue;
+ }
+ const int64_t selected_rows = detail::bounded_native_lazy_skip_rows(
+ std::min(remaining, range.start + range.length -
_logical_row_position));
+ _skip_column->clear();
+ // Pending skips can span many filtered batches and are replayed for
every lazy column.
+ // Chunking here bounds each dense bitmap while preserving one logical
scheduler skip.
+ _filter_scratch.assign(static_cast<size_t>(selected_rows), 0);
+ int64_t rows_read = 0;
+ RETURN_IF_ERROR(read_with_filter(selected_rows,
_filter_scratch.data(), true, _skip_column,
+ _type, false, &rows_read));
+ DORIS_CHECK(_skip_column->empty());
+ DORIS_CHECK(rows_read == selected_rows);
+ _logical_row_position += rows_read;
+ native_skipped_rows += rows_read;
+ remaining -= rows_read;
+ }
+ update_reader_skip_rows(native_skipped_rows);
+ return Status::OK();
+}
+
+Status NativeColumnReader::select(const SelectionVector& selection, uint16_t
selected_rows,
+ int64_t batch_rows, MutableColumnPtr&
column) {
+ RETURN_IF_ERROR(validate_selected_span(batch_rows));
+ const uint8_t* filter_data = nullptr;
+ RETURN_IF_ERROR(selection.materialize_filter(selected_rows, batch_rows,
&filter_data));
+ const size_t old_size = column->size();
+ int64_t rows_read = 0;
+ RETURN_IF_ERROR(read_with_filter(batch_rows, filter_data, selected_rows ==
0, column, _type,
+ false, &rows_read));
+ advance_selected_span(rows_read);
+ if (column->size() != old_size + selected_rows) {
+ return Status::Corruption(
+ "Native parquet selection appended {} rows, expected {} for
column {}",
+ column->size() - old_size, selected_rows, _name);
+ }
+ if (_profile.reader_select_rows != nullptr) {
+ COUNTER_UPDATE(_profile.reader_select_rows, selected_rows);
+ }
+ update_reader_read_rows(selected_rows);
+ update_reader_skip_rows(batch_rows - selected_rows);
+ return Status::OK();
+}
+
+Status NativeColumnReader::select_with_dictionary_filter(const
SelectionVector& selection,
+ uint16_t
selected_rows, int64_t batch_rows,
+ const
IColumn::Filter& dictionary_filter,
+ MutableColumnPtr&
column,
+ IColumn::Filter*
row_filter,
+ bool* used_filter) {
+ DORIS_CHECK(row_filter != nullptr);
+ DORIS_CHECK(used_filter != nullptr);
+ RETURN_IF_ERROR(validate_selected_span(batch_rows));
+ *used_filter = false;
+ row_filter->clear();
+ if (!_dictionary_filter_enabled) {
+ return Status::OK();
+ }
+ *used_filter = true;
+
+ const uint8_t* filter_data = nullptr;
+ RETURN_IF_ERROR(selection.materialize_filter(selected_rows, batch_rows,
&filter_data));
+ const bool nullable = _type->is_nullable();
+ DataTypePtr id_type = std::make_shared<DataTypeInt32>();
+ if (nullable) {
+ id_type = make_nullable(id_type);
+ }
+ if (!_dictionary_id_column) {
+ _dictionary_id_column = id_type->create_column();
+ }
+ _dictionary_id_column->clear();
+ int64_t rows_read = 0;
+ RETURN_IF_ERROR(read_with_filter(batch_rows, filter_data, selected_rows ==
0,
+ _dictionary_id_column, id_type, true,
&rows_read));
+ advance_selected_span(rows_read);
+ if (_dictionary_id_column->size() != selected_rows) {
+ return Status::Corruption(
+ "Native parquet dictionary reader appended {} rows, expected
{} for {}",
+ _dictionary_id_column->size(), selected_rows, _name);
+ }
+
+ const ColumnInt32* ids = nullptr;
+ const NullMap* null_map = nullptr;
+ if (const auto* nullable_ids =
check_and_get_column<ColumnNullable>(*_dictionary_id_column);
+ nullable_ids != nullptr) {
+ ids =
check_and_get_column<ColumnInt32>(nullable_ids->get_nested_column());
+ null_map = &nullable_ids->get_null_map_data();
+ } else {
+ ids = check_and_get_column<ColumnInt32>(*_dictionary_id_column);
+ }
+ DORIS_CHECK(ids != nullptr);
+
+ if (!_matched_dictionary_ids) {
+ _matched_dictionary_ids = ColumnInt32::create();
+ }
+ _matched_dictionary_ids->clear();
+ auto& matched_ids =
assert_cast<ColumnInt32&>(*_matched_dictionary_ids).get_data();
+ row_filter->reserve(selected_rows);
+ const auto& id_data = ids->get_data();
+ for (size_t row = 0; row < selected_rows; ++row) {
+ bool keep = false;
+ if (null_map == nullptr || (*null_map)[row] == 0) {
+ const int32_t dictionary_id = id_data[row];
+ if (dictionary_id < 0 ||
+ static_cast<size_t>(dictionary_id) >=
dictionary_filter.size()) {
+ return Status::Corruption(
+ "Invalid parquet dictionary id {} for column {} with
{} entries",
+ dictionary_id, _name, dictionary_filter.size());
+ }
+ keep = dictionary_filter[static_cast<size_t>(dictionary_id)] != 0;
+ if (keep) {
+ matched_ids.push_back(dictionary_id);
+ }
+ }
+ row_filter->push_back(keep ? 1 : 0);
+ }
+
+ const auto* matched_id_column =
check_and_get_column<ColumnInt32>(*_matched_dictionary_ids);
+ DORIS_CHECK(matched_id_column != nullptr);
+ auto string_values =
+
DORIS_TRY(_native_reader->convert_dict_column_to_string_column(matched_id_column));
+ RETURN_IF_ERROR(append_non_null_dictionary_values(column,
std::move(string_values)));
+ if (_profile.reader_select_rows != nullptr) {
+ COUNTER_UPDATE(_profile.reader_select_rows, selected_rows);
+ }
+ update_reader_read_rows(cast_set<int64_t>(matched_ids.size()));
+ update_reader_skip_rows(batch_rows -
cast_set<int64_t>(matched_ids.size()));
+ return Status::OK();
+}
+
+void NativeColumnReader::flush_profile() {
+ int64_t max_leaf_page_reads = 0;
+ record_page_fragments(sync_native_profile(&max_leaf_page_reads),
max_leaf_page_reads);
+}
+
+Result<MutableColumnPtr> NativeColumnReader::dictionary_values() {
+ DORIS_CHECK(_native_reader != nullptr);
+ return _native_reader->dictionary_values();
+}
+
+void NativeColumnReader::record_page_fragments(int64_t page_fragments,
+ int64_t max_leaf_page_reads) {
+ if (_profile.native_page_fragments != nullptr) {
+ COUNTER_UPDATE(_profile.native_page_fragments, page_fragments);
+ }
+ // Summed fragments from MAP/STRUCT siblings do not mean any individual
leaf crossed a page.
+ if (max_leaf_page_reads > 1 && _profile.page_crossing_batches != nullptr) {
Review Comment:
[P2] Count page crossings at the scheduler batch boundary `flush_profile()`
runs only every 16 scheduler batches (or at the row-group tail), so
`max_leaf_page_reads > 1` describes an amortized interval, not one batch.
Sixteen batches that each open one page become one false crossing, sixteen real
crossings become one, and each projected top-level column can increment the
same shared counter. This makes `PageCrossingBatches` incomparable with
per-batch `TotalBatches` as documented. Record/OR a crossing flag for each
scheduler batch and keep the periodic flush only for additive cumulative
statistics.
##########
be/src/format_v2/parquet/reader/native/delta_bit_pack_decoder.h:
##########
@@ -0,0 +1,669 @@
+// 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 <gen_cpp/parquet_types.h>
+#include <glog/logging.h>
+
+#include <cstddef>
+#include <cstdint>
+#include <cstring>
+#include <limits>
+#include <memory>
+#include <ostream>
+#include <string>
+#include <type_traits>
+#include <utility>
+#include <vector>
+
+#include "common/status.h"
+#include "core/column/column_fixed_length_object.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type.h"
+#include "exec/common/arithmetic_overflow.h"
+#include "format_v2/parquet/reader/native/decoder.h"
+#include "util/bit_stream_utils.h"
+#include "util/bit_stream_utils.inline.h"
+#include "util/slice.h"
+
+namespace doris::format::parquet::native {
+class DeltaDecoder : public Decoder {
+public:
+ DeltaDecoder() = default;
+ ~DeltaDecoder() override = default;
+};
+
+/**
+ * Format
+ * [header] [block 1] [block 2] ... [block N]
+ * Header
+ * [block size] [_mini_blocks_per_block] [_total_value_count] [first
value]
+ * Block
+ * [min delta] [list of bitwidths of the mini blocks] [miniblocks]
+ */
+template <typename T>
+class DeltaBitPackDecoder final : public DeltaDecoder {
+public:
+ using UT = std::make_unsigned_t<T>;
+
+ DeltaBitPackDecoder() = default;
+ ~DeltaBitPackDecoder() override = default;
+
+ Status skip_values(size_t num_values) override {
+ constexpr size_t kSkipBatchSize = 4096;
+ _values.resize(std::min(num_values, kSkipBatchSize));
+ size_t skipped = 0;
+ while (skipped < num_values) {
+ const size_t batch_size = std::min(num_values - skipped,
kSkipBatchSize);
+ uint32_t num_valid_values = 0;
+ RETURN_IF_ERROR(_get_internal(_values.data(),
static_cast<uint32_t>(batch_size),
+ &num_valid_values));
+ // Skips retain exact validation without allocating in proportion
to a sparse gap.
+ if (UNLIKELY(num_valid_values != batch_size)) {
+ return Status::IOError("Expected to skip {} Parquet delta
values, skipped {}",
+ num_values, skipped + num_valid_values);
+ }
+ skipped += batch_size;
+ }
+ return Status::OK();
+ }
+
+ Status decode_fixed_values(size_t num_values, ParquetFixedValueConsumer&
consumer) override {
+ _values.resize(num_values);
+ uint32_t decoded_count = 0;
+ RETURN_IF_ERROR(
+ _get_internal(_values.data(), cast_set<uint32_t>(num_values),
&decoded_count));
+ if (UNLIKELY(decoded_count != num_values)) {
+ return Status::IOError("Expected {} Parquet delta values, decoded
{}", num_values,
+ decoded_count);
+ }
+ return consumer.consume(reinterpret_cast<const
uint8_t*>(_values.data()), num_values,
+ sizeof(T));
+ }
+
+ Status decode_selected_fixed_values(const ParquetSelection& selection,
+ ParquetFixedValueConsumer& consumer)
override {
+ _values.resize(selection.total_values);
+ uint32_t decoded_count = 0;
+ RETURN_IF_ERROR(_get_internal(_values.data(),
cast_set<uint32_t>(selection.total_values),
+ &decoded_count));
+ if (UNLIKELY(decoded_count != selection.total_values)) {
+ return Status::IOError("Expected {} Parquet delta values, decoded
{}",
+ selection.total_values, decoded_count);
+ }
+ size_t output = 0;
+ for (const auto& range : selection.ranges) {
+ memmove(_values.data() + output, _values.data() + range.first,
range.count * sizeof(T));
+ output += range.count;
+ }
+ DORIS_CHECK_EQ(output, selection.selected_values);
+ return consumer.consume(reinterpret_cast<const
uint8_t*>(_values.data()), output,
+ sizeof(T));
+ }
+
+ Status decode(T* buffer, uint32_t num_values, uint32_t* out_num_values) {
+ return _get_internal(buffer, num_values, out_num_values);
+ }
+
+ uint32_t valid_values_count() {
+ // _total_value_count in header ignores of null values
+ return _total_values_remaining;
+ }
+
+ void release_scratch(size_t max_retained_bytes) override {
+ release_vector_if_oversized(&_values, max_retained_bytes);
+ release_vector_if_oversized(&_delta_bit_widths, max_retained_bytes);
+ }
+ size_t retained_scratch_bytes() const override {
+ return _values.capacity() * sizeof(T) + _delta_bit_widths.capacity() *
sizeof(uint8_t);
+ }
+ size_t active_scratch_bytes() const override {
+ return _values.size() * sizeof(T) + _delta_bit_widths.size() *
sizeof(uint8_t);
+ }
+
+ Status set_data(Slice* slice) override {
+ _bit_reader.reset(
+ new BitReader((const uint8_t*)slice->data,
cast_set<uint32_t>(slice->size)));
+ RETURN_IF_ERROR(_init_header());
+ _data = slice;
+ _offset = 0;
+ return Status::OK();
+ }
+
+ // Set BitReader which is already initialized by
DeltaLengthByteArrayDecoder or
+ // DeltaByteArrayDecoder
+ Status set_bit_reader(std::shared_ptr<BitReader> bit_reader) {
+ _bit_reader = std::move(bit_reader);
+ RETURN_IF_ERROR(_init_header());
+ return Status::OK();
+ }
+
+private:
+ static constexpr int kMaxDeltaBitWidth = static_cast<int>(sizeof(T) * 8);
+ Status _init_header();
+ Status _init_block();
+ Status _init_mini_block(int bit_width);
+ Status _get_internal(T* buffer, uint32_t max_values, uint32_t*
out_num_values);
+
+ std::vector<T> _values;
+
+ std::shared_ptr<BitReader> _bit_reader;
+ uint32_t _values_per_block;
+ uint32_t _mini_blocks_per_block;
+ uint32_t _values_per_mini_block;
+ uint32_t _total_value_count;
+
+ T _min_delta;
+ T _last_value;
+
+ uint32_t _mini_block_idx;
+ std::vector<uint8_t> _delta_bit_widths;
+ int _delta_bit_width;
+ // If the page doesn't contain any block, `_block_initialized` will
+ // always be false. Otherwise, it will be true when first block
initialized.
+ bool _block_initialized;
+
+ uint32_t _total_values_remaining;
+ // Remaining values in current mini block. If the current block is the
last mini block,
+ // _values_remaining_current_mini_block may greater than
_total_values_remaining.
+ uint32_t _values_remaining_current_mini_block;
+};
+
+class DeltaLengthByteArrayDecoder final : public DeltaDecoder {
+public:
+ explicit DeltaLengthByteArrayDecoder()
+ : _len_decoder(), _buffered_length(0), _buffered_data(0) {}
+
+ void set_expected_values(size_t expected_values) override {
+ Decoder::set_expected_values(expected_values);
+ _len_decoder.set_expected_values(expected_values);
+ }
+
+ Status skip_values(size_t num_values) override {
+ constexpr size_t kSkipBatchSize = 4096;
+ _values.resize(std::min(num_values, kSkipBatchSize));
+ size_t skipped = 0;
+ while (skipped < num_values) {
+ const size_t batch_size = std::min(num_values - skipped,
kSkipBatchSize);
+ int num_valid_values = 0;
+ RETURN_IF_ERROR(
+ _get_internal(_values.data(),
static_cast<int>(batch_size), &num_valid_values));
+ if (UNLIKELY(num_valid_values != batch_size)) {
+ return Status::IOError(
+ "Expected to skip {} Parquet delta-length values,
skipped {}", num_values,
+ skipped + num_valid_values);
+ }
+ skipped += batch_size;
+ }
+ return Status::OK();
+ }
+
+ Status decode_binary_values(size_t num_values, ParquetBinaryValueConsumer&
consumer) override {
+ _values.resize(num_values);
+ int decoded_count = 0;
+ RETURN_IF_ERROR(
+ _get_internal(_values.data(), cast_set<int32_t>(num_values),
&decoded_count));
+ if (UNLIKELY(decoded_count != num_values)) {
+ return Status::IOError("Expected {} Parquet delta-length values,
decoded {}",
+ num_values, decoded_count);
+ }
+ _string_refs.resize(num_values);
+ for (size_t row = 0; row < num_values; ++row) {
+ _string_refs[row] = StringRef(_values[row].data,
_values[row].size);
+ }
+ return consumer.consume(_string_refs.data(), _string_refs.size());
+ }
+
+ Status decode_selected_binary_values(const ParquetSelection& selection,
+ ParquetBinaryValueConsumer& consumer)
override {
+ _values.resize(selection.total_values);
+ int decoded_count = 0;
+ RETURN_IF_ERROR(_get_internal(_values.data(),
cast_set<int32_t>(selection.total_values),
+ &decoded_count));
+ if (UNLIKELY(decoded_count != selection.total_values)) {
+ return Status::IOError("Expected {} Parquet delta-length values,
decoded {}",
+ selection.total_values, decoded_count);
+ }
+ _string_refs.resize(selection.selected_values);
+ size_t output = 0;
+ for (const auto& range : selection.ranges) {
+ for (size_t row = 0; row < range.count; ++row) {
+ const auto& value = _values[range.first + row];
+ _string_refs[output++] = StringRef(value.data, value.size);
+ }
+ }
+ DORIS_CHECK_EQ(output, selection.selected_values);
+ return consumer.consume(_string_refs.data(), _string_refs.size());
+ }
+
+ Status decode(Slice* buffer, int num_values, int* out_num_values) {
+ return _get_internal(buffer, num_values, out_num_values);
+ }
+
+ int valid_values_count() const { return _num_valid_values; }
+
+ void release_scratch(size_t max_retained_bytes) override {
+ release_vector_if_oversized(&_values, max_retained_bytes);
+ release_vector_if_oversized(&_string_refs, max_retained_bytes);
+ release_vector_if_oversized(&_buffered_length, max_retained_bytes);
+ release_vector_if_oversized(&_buffered_data, max_retained_bytes);
+ _len_decoder.release_scratch(max_retained_bytes);
+ }
+ size_t retained_scratch_bytes() const override {
+ return _values.capacity() * sizeof(Slice) + _string_refs.capacity() *
sizeof(StringRef) +
+ _buffered_length.capacity() * sizeof(int32_t) +
+ _buffered_data.capacity() * sizeof(char) +
_len_decoder.retained_scratch_bytes();
+ }
+ size_t active_scratch_bytes() const override {
+ return _values.size() * sizeof(Slice) + _string_refs.size() *
sizeof(StringRef) +
+ _buffered_length.size() * sizeof(int32_t) +
_buffered_data.size() * sizeof(char) +
+ _len_decoder.active_scratch_bytes();
+ }
+
+ Status set_data(Slice* slice) override {
+ if (slice == nullptr || slice->size == 0) {
+ // Reused decoders must never retain lengths or payload pointers
from the prior page.
+ _bit_reader.reset();
+ _data = nullptr;
+ _offset = 0;
+ _num_valid_values = 0;
+ _buffered_length.clear();
+ _buffered_data.clear();
+ return Status::Corruption("Parquet delta-length page is empty");
+ }
+ _bit_reader = std::make_shared<BitReader>((const uint8_t*)slice->data,
slice->size);
+ _data = slice;
+ _offset = 0;
+ RETURN_IF_ERROR(_init_lengths());
+ return Status::OK();
+ }
+
+ Status set_bit_reader(std::shared_ptr<BitReader> bit_reader) {
+ _bit_reader = std::move(bit_reader);
+ RETURN_IF_ERROR(_init_lengths());
+ return Status::OK();
+ }
+
+private:
+ Status _init_lengths();
+ Status _get_internal(Slice* buffer, int max_values, int* out_num_values);
+
+ std::vector<Slice> _values;
+ std::vector<StringRef> _string_refs;
+ std::shared_ptr<BitReader> _bit_reader;
+ DeltaBitPackDecoder<int32_t> _len_decoder;
+
+ int _num_valid_values;
+ std::vector<int32_t> _buffered_length;
+ std::vector<char> _buffered_data;
+};
+
+class DeltaByteArrayDecoder : public DeltaDecoder {
+public:
+ explicit DeltaByteArrayDecoder() : _buffered_prefix_length(0),
_buffered_data(0) {}
+
+ void set_expected_values(size_t expected_values) override {
+ Decoder::set_expected_values(expected_values);
+ _prefix_len_decoder.set_expected_values(expected_values);
+ _suffix_decoder.set_expected_values(expected_values);
+ }
+
+ Status skip_values(size_t num_values) override {
+ constexpr size_t kSkipBatchSize = 4096;
+ size_t skipped = 0;
+ while (skipped < num_values) {
+ const size_t batch_size = std::min(num_values - skipped,
kSkipBatchSize);
+ RETURN_IF_ERROR(_decode_slices(batch_size));
+ RETURN_IF_ERROR(_validate_fixed_width_values());
+ skipped += batch_size;
+ }
+ return Status::OK();
+ }
+
+ Status decode_binary_values(size_t num_values, ParquetBinaryValueConsumer&
consumer) override {
+ RETURN_IF_ERROR(_decode_slices(num_values));
+ _string_refs.resize(num_values);
+ for (size_t row = 0; row < num_values; ++row) {
+ _string_refs[row] = StringRef(_values[row].data,
_values[row].size);
+ }
+ return consumer.consume(_string_refs.data(), _string_refs.size());
+ }
+
+ Status decode_selected_binary_values(const ParquetSelection& selection,
+ ParquetBinaryValueConsumer& consumer)
override {
+ RETURN_IF_ERROR(_decode_slices(selection.total_values));
+ _string_refs.resize(selection.selected_values);
+ size_t output = 0;
+ for (const auto& range : selection.ranges) {
+ for (size_t row = 0; row < range.count; ++row) {
+ const auto& value = _values[range.first + row];
+ _string_refs[output++] = StringRef(value.data, value.size);
+ }
+ }
+ DORIS_CHECK_EQ(output, selection.selected_values);
+ return consumer.consume(_string_refs.data(), _string_refs.size());
+ }
+
+ Status decode_fixed_values(size_t num_values, ParquetFixedValueConsumer&
consumer) override {
+ RETURN_IF_ERROR(_decode_slices(num_values));
+ RETURN_IF_ERROR(_validate_fixed_width_values());
+ const size_t byte_size = num_values *
static_cast<size_t>(_type_length);
+ _fixed_values.resize(byte_size);
+ for (size_t row = 0; row < num_values; ++row) {
+ memcpy(_fixed_values.data() + row * _type_length,
_values[row].data, _type_length);
+ }
+ return consumer.consume(_fixed_values.data(), num_values,
+ static_cast<size_t>(_type_length));
+ }
+
+ Status decode_selected_fixed_values(const ParquetSelection& selection,
+ ParquetFixedValueConsumer& consumer)
override {
+ RETURN_IF_ERROR(_decode_slices(selection.total_values));
+ // Validate every consumed value, including filtered ranges, so
selection cannot hide a
+ // malformed FIXED_LEN_BYTE_ARRAY width that a later cursor advance
has already committed.
+ RETURN_IF_ERROR(_validate_fixed_width_values());
+ DORIS_CHECK(_type_length > 0);
+ const size_t value_width = static_cast<size_t>(_type_length);
+ if (UNLIKELY(selection.selected_values >
+ std::numeric_limits<size_t>::max() / value_width)) {
+ return Status::IOError("Parquet delta-byte-array selection byte
size overflows");
+ }
+ _fixed_values.resize(selection.selected_values * value_width);
+ size_t output = 0;
+ for (const auto& range : selection.ranges) {
+ for (size_t row = 0; row < range.count; ++row) {
+ const auto& value = _values[range.first + row];
+ memcpy(_fixed_values.data() + output * value_width,
value.data, value_width);
+ ++output;
+ }
+ }
+ DORIS_CHECK_EQ(output, selection.selected_values);
+ return consumer.consume(_fixed_values.data(), output, value_width);
+ }
+
+ Status set_data(Slice* slice) override {
+ _bit_reader = std::make_shared<BitReader>((const uint8_t*)slice->data,
slice->size);
+ auto prefix_reader = std::make_shared<BitReader>(*_bit_reader);
+ RETURN_IF_ERROR(_prefix_len_decoder.set_bit_reader(prefix_reader));
+
+ // get the number of encoded prefix lengths
+ int num_prefix = _prefix_len_decoder.valid_values_count();
+ DeltaBitPackDecoder<int32_t> prefix_locator;
+ prefix_locator.set_expected_values(_expected_values);
+ RETURN_IF_ERROR(prefix_locator.set_bit_reader(_bit_reader));
+ constexpr size_t kLocateBatchSize = 4096;
+ std::vector<int32_t> ignored_prefixes(
+ std::min<size_t>(static_cast<size_t>(num_prefix),
kLocateBatchSize));
+ size_t located = 0;
+ while (located < static_cast<size_t>(num_prefix)) {
+ const size_t batch_size =
+ std::min(static_cast<size_t>(num_prefix) - located,
kLocateBatchSize);
+ uint32_t decoded = 0;
+ RETURN_IF_ERROR(prefix_locator.decode(ignored_prefixes.data(),
+
static_cast<uint32_t>(batch_size), &decoded));
+ if (UNLIKELY(decoded != batch_size)) {
+ return Status::Corruption("Parquet delta prefix stream ended
early");
+ }
+ located += batch_size;
+ }
+ _num_valid_values = num_prefix;
+
+ // at this time, the decoder_ will be at the start of the encoded
suffix data.
+ RETURN_IF_ERROR(_suffix_decoder.set_bit_reader(_bit_reader));
+ if (UNLIKELY(_suffix_decoder.valid_values_count() != num_prefix)) {
+ return Status::Corruption("Parquet delta prefix and suffix counts
differ");
+ }
+
+ // TODO: read corrupted files written with bug(PARQUET-246).
_last_value should be set
+ // to _last_value_in_previous_page when decoding a new page(except the
first page)
+ _last_value = "";
+ return Status::OK();
+ }
+
+ Status decode(Slice* buffer, int num_values, int* out_num_values) {
+ return _get_internal(buffer, num_values, out_num_values);
+ }
+
+ void release_scratch(size_t max_retained_bytes) override {
+ release_vector_if_oversized(&_values, max_retained_bytes);
+ release_vector_if_oversized(&_string_refs, max_retained_bytes);
+ release_vector_if_oversized(&_fixed_values, max_retained_bytes);
+ release_vector_if_oversized(&_buffered_prefix_length,
max_retained_bytes);
+ release_vector_if_oversized(&_buffered_data, max_retained_bytes);
+ _prefix_len_decoder.release_scratch(max_retained_bytes);
+ _suffix_decoder.release_scratch(max_retained_bytes);
+ if (_last_value.capacity() > max_retained_bytes)
std::string().swap(_last_value);
+ if (_last_value_in_previous_page.capacity() > max_retained_bytes) {
+ std::string().swap(_last_value_in_previous_page);
+ }
+ }
+ size_t retained_scratch_bytes() const override {
+ return _values.capacity() * sizeof(Slice) + _string_refs.capacity() *
sizeof(StringRef) +
+ _fixed_values.capacity() * sizeof(uint8_t) +
+ _buffered_prefix_length.capacity() * sizeof(int32_t) +
+ _buffered_data.capacity() * sizeof(char) +
_last_value.capacity() +
+ _last_value_in_previous_page.capacity() +
+ _prefix_len_decoder.retained_scratch_bytes() +
+ _suffix_decoder.retained_scratch_bytes();
+ }
+ size_t active_scratch_bytes() const override {
+ return _values.size() * sizeof(Slice) + _string_refs.size() *
sizeof(StringRef) +
+ _fixed_values.size() * sizeof(uint8_t) +
+ _buffered_prefix_length.size() * sizeof(int32_t) +
+ _buffered_data.size() * sizeof(char) + _last_value.size() +
+ _last_value_in_previous_page.size() +
_prefix_len_decoder.active_scratch_bytes() +
+ _suffix_decoder.active_scratch_bytes();
+ }
+
+private:
+ Status _validate_fixed_width_values() const {
+ if (_type_length <= 0) {
+ return Status::OK();
+ }
+ const size_t value_width = static_cast<size_t>(_type_length);
+ for (const auto& value : _values) {
+ if (UNLIKELY(value.size != value_width)) {
+ return Status::Corruption("Parquet fixed value has length {},
expected {}",
+ value.size, value_width);
+ }
+ }
+ return Status::OK();
+ }
+
+ Status _decode_slices(size_t num_values) {
+ _values.resize(num_values);
+ int decoded_count = 0;
+ RETURN_IF_ERROR(
+ _get_internal(_values.data(), cast_set<int32_t>(num_values),
&decoded_count));
+ if (UNLIKELY(decoded_count != num_values)) {
+ return Status::IOError("Expected {} Parquet delta-byte-array
values, decoded {}",
+ num_values, decoded_count);
+ }
+ return Status::OK();
+ }
+
+ Status _get_internal(Slice* buffer, int max_values, int* out_num_values);
+
+ std::vector<Slice> _values;
+ std::vector<StringRef> _string_refs;
+ std::vector<uint8_t> _fixed_values;
+ std::shared_ptr<BitReader> _bit_reader;
+ DeltaBitPackDecoder<int32_t> _prefix_len_decoder;
+ DeltaLengthByteArrayDecoder _suffix_decoder;
+ std::string _last_value;
+ // string buffer for last value in previous page
+ std::string _last_value_in_previous_page;
+ int _num_valid_values;
+ std::vector<int32_t> _buffered_prefix_length;
+ std::vector<char> _buffered_data;
+};
+} // namespace doris::format::parquet::native
+
+namespace doris::format::parquet::native {
+
+template <typename T>
+Status DeltaBitPackDecoder<T>::_init_header() {
+ if (!_bit_reader->GetVlqInt(&_values_per_block) ||
+ !_bit_reader->GetVlqInt(&_mini_blocks_per_block) ||
+ !_bit_reader->GetVlqInt(&_total_value_count) ||
+ !_bit_reader->GetZigZagVlqInt(&_last_value)) {
+ return Status::IOError("Init header eof");
+ }
+ if (_values_per_block == 0) {
+ return Status::InvalidArgument("Cannot have zero value per block");
+ }
+ if (_values_per_block % 128 != 0) {
+ return Status::InvalidArgument(
+ "the number of values in a block must be multiple of 128, but
it's " +
+ std::to_string(_values_per_block));
+ }
+ if (_mini_blocks_per_block == 0) {
+ return Status::InvalidArgument("Cannot have zero miniblock per block");
+ }
+ _values_per_mini_block = _values_per_block / _mini_blocks_per_block;
+ if (_values_per_mini_block == 0) {
+ return Status::InvalidArgument("Cannot have zero value per miniblock");
+ }
+ if (_values_per_mini_block % 32 != 0) {
+ return Status::InvalidArgument(
+ "The number of values in a miniblock must be multiple of 32,
but it's " +
+ std::to_string(_values_per_mini_block));
+ }
+ // Encoded counts are external ULEB32 values. Bound them by the page's
advertised logical
+ // values before any vector uses the count; optional levels may only
reduce this upper bound.
+ if (UNLIKELY(_total_value_count > _expected_values)) {
+ return Status::Corruption("Parquet delta header advertises {} values,
page allows {}",
+ _total_value_count, _expected_values);
+ }
+ _total_values_remaining = _total_value_count;
+ _delta_bit_widths.clear();
+ // init as empty property
+ _block_initialized = false;
+ _delta_bit_width = 0;
+ _values_remaining_current_mini_block = 0;
+ return Status::OK();
+}
+
+template <typename T>
+Status DeltaBitPackDecoder<T>::_init_block() {
+ DCHECK_GT(_total_values_remaining, 0) << "InitBlock called at EOF";
+ if (!_bit_reader->GetZigZagVlqInt(&_min_delta)) {
+ return Status::IOError("Init block eof");
+ }
+
+ // One byte follows for each miniblock. Defer allocation until a block is
actually consumed so
+ // a one-value page cannot turn an irrelevant malicious miniblock count
into a large resize.
+ if (UNLIKELY(_mini_blocks_per_block >
static_cast<uint32_t>(_bit_reader->bytes_left()))) {
+ return Status::Corruption("Parquet delta miniblock count {} exceeds
remaining {} bytes",
+ _mini_blocks_per_block,
_bit_reader->bytes_left());
+ }
+ _delta_bit_widths.resize(_mini_blocks_per_block);
+
+ // read the bitwidth of each miniblock
+ uint8_t* bit_width_data = _delta_bit_widths.data();
+ for (uint32_t i = 0; i < _mini_blocks_per_block; ++i) {
+ if (!_bit_reader->GetAligned<uint8_t>(1, bit_width_data + i)) {
+ return Status::IOError("Decode bit-width EOF");
+ }
+ // Note that non-conformant bitwidth entries are allowed by the
Parquet spec
+ // for extraneous miniblocks in the last block (GH-14923), so we check
+ // the bitwidths when actually using them (see InitMiniBlock()).
+ }
+ _mini_block_idx = 0;
+ _block_initialized = true;
+ RETURN_IF_ERROR(_init_mini_block(bit_width_data[0]));
+ return Status::OK();
+}
+
+template <typename T>
+Status DeltaBitPackDecoder<T>::_init_mini_block(int bit_width) {
+ if (bit_width > kMaxDeltaBitWidth) [[unlikely]] {
+ return Status::InvalidArgument("delta bit width larger than integer
bit width");
+ }
+ _delta_bit_width = bit_width;
+ _values_remaining_current_mini_block = _values_per_mini_block;
+ return Status::OK();
+}
+
+template <typename T>
+Status DeltaBitPackDecoder<T>::_get_internal(T* buffer, uint32_t num_values,
+ uint32_t* out_num_values) {
+ num_values = std::min<uint32_t>(num_values, _total_values_remaining);
+ if (num_values == 0) {
+ *out_num_values = 0;
+ return Status::OK();
+ }
+ uint32_t i = 0;
+ while (i < num_values) {
+ if (_values_remaining_current_mini_block == 0) [[unlikely]] {
+ if (!_block_initialized) [[unlikely]] {
+ buffer[i++] = _last_value;
+ DCHECK_EQ(i, 1); // we're at the beginning of the page
+ if (i == num_values) {
+ // When block is uninitialized and i reaches num_values we
have two
+ // different possibilities:
+ // 1. _total_value_count == 1, which means that the page
may have only
+ // one value (encoded in the header), and we should not
initialize
+ // any block.
+ // 2. _total_value_count != 1, which means we should
initialize the
+ // incoming block for subsequent reads.
+ if (_total_value_count != 1) {
+ RETURN_IF_ERROR(_init_block());
+ }
+ break;
+ }
+ RETURN_IF_ERROR(_init_block());
+ } else {
+ ++_mini_block_idx;
+ if (_mini_block_idx < _mini_blocks_per_block) {
+
RETURN_IF_ERROR(_init_mini_block(_delta_bit_widths.data()[_mini_block_idx]));
+ } else {
+ RETURN_IF_ERROR(_init_block());
+ }
+ }
+ }
+
+ uint32_t values_decode =
std::min(_values_remaining_current_mini_block, num_values - i);
+ for (uint32_t j = 0; j < values_decode; ++j) {
+ if (!_bit_reader->GetValue(_delta_bit_width, buffer + i + j)) {
+ return Status::IOError("Get batch EOF");
+ }
+ }
+ for (int j = 0; j < values_decode; ++j) {
+ // Addition between min_delta, packed int and last_value should be
treated as
+ // unsigned addition. Overflow is as expected.
+ buffer[i + j] = static_cast<UT>(_min_delta) +
static_cast<UT>(buffer[i + j]) +
+ static_cast<UT>(_last_value);
+ _last_value = buffer[i + j];
+ }
+ _values_remaining_current_mini_block -= values_decode;
+ i += values_decode;
+ }
+ _total_values_remaining -= num_values;
+
+ if (_total_values_remaining == 0) [[unlikely]] {
+ if (!_bit_reader->Advance(_delta_bit_width *
_values_remaining_current_mini_block)) {
Review Comment:
[P1] Widen the final miniblock padding calculation The multiplication occurs
in `uint32_t` before the result reaches `Advance(int64_t)`. For example, block
size 134217856 with one miniblock, two values, and width 32 requires skipping
4,294,971,360 bits after the second value, but this expression wraps to 4,064
bits. That block still fits the signed page-size domain; in
DELTA_LENGTH_BYTE_ARRAY the shared locator is then 512 MiB early and returns
padding as string payload. Use checked uint64/int64 multiplication, verify it
fits the remaining stream, and cover the overflow boundary plus the byte-array
cursor handoff.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
