This is an automated email from the ASF dual-hosted git repository.
pitrou pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow.git
The following commit(s) were added to refs/heads/main by this push:
new df40dd400bb GH-51238: [C++][Python][Parquet] Limit schema nesting
depth when reading (#51239)
df40dd400bb is described below
commit df40dd400bbaea59e621123ead6c41a51f0fc654
Author: Antoine Pitrou <[email protected]>
AuthorDate: Wed Sep 9 18:06:11 2026 +0200
GH-51238: [C++][Python][Parquet] Limit schema nesting depth when reading
(#51239)
### Rationale for this change
Reconstructing a nested Schema from the Parquet Thrift metadata implies a
recursive call that can blow up the stack on pathologically-nested schemas
(with thousands of nesting levels or more).
By adding a limit on the schema nesting depth, we turn a stack
overflow-induced crash into a regular Parquet error.
### Are these changes tested?
By additional unit tests; also privately with a proof-of-concept reproducer
that induces a stack overflow exhaustion.
### Are there any user-facing changes?
In the unlikely case where a legitimate Parquet file has a deeper schema
than the default schema nesting limit in this PR (100), an error will be raised
when reading where it used to succeed. The user can bump the limit to
circumvent the error.
**This PR contains a "Critical Fix".** It fixes a crash on a deeply nested
Parquet schema that would provoke a stack overflow. It is not an exploitable
vulnerability except through denial of service.
Thanks to "1K0CT" for the initial report.
* GitHub Issue: #51238
Authored-by: Antoine Pitrou <[email protected]>
Signed-off-by: Antoine Pitrou <[email protected]>
---
cpp/src/arrow/dataset/file_parquet.cc | 6 ++-
cpp/src/parquet/arrow/arrow_schema_test.cc | 6 ++-
cpp/src/parquet/metadata.cc | 27 ++++++++----
cpp/src/parquet/metadata.h | 7 +--
cpp/src/parquet/properties.h | 14 ++++++
cpp/src/parquet/reader_test.cc | 24 ++++++++++-
cpp/src/parquet/schema.cc | 63 +++++++++++++++++++--------
cpp/src/parquet/schema_internal.h | 10 +++--
cpp/src/parquet/schema_test.cc | 69 ++++++++++++++++++++++++------
python/pyarrow/_dataset_parquet.pyx | 25 +++++++++--
python/pyarrow/_parquet.pyx | 7 +++
python/pyarrow/includes/libparquet.pxd | 3 ++
python/pyarrow/parquet/core.py | 23 ++++++++--
python/pyarrow/tests/parquet/test_basic.py | 22 ++++++++++
python/pyarrow/tests/test_dataset.py | 6 +++
15 files changed, 255 insertions(+), 57 deletions(-)
diff --git a/cpp/src/arrow/dataset/file_parquet.cc
b/cpp/src/arrow/dataset/file_parquet.cc
index ba0e93f09d4..a1fcfe70904 100644
--- a/cpp/src/arrow/dataset/file_parquet.cc
+++ b/cpp/src/arrow/dataset/file_parquet.cc
@@ -68,7 +68,7 @@ parquet::ReaderProperties MakeReaderProperties(
const ParquetFileFormat& format, ParquetFragmentScanOptions*
parquet_scan_options,
const std::string& path = "", std::shared_ptr<fs::FileSystem> filesystem =
nullptr,
MemoryPool* pool = default_memory_pool()) {
- // Can't mutate pool after construction
+ // FIXME (GH-51264): Can't mutate pool after ReaderProperties construction.
parquet::ReaderProperties properties(pool);
if (parquet_scan_options->reader_properties->is_buffered_stream_enabled()) {
properties.enable_buffered_stream();
@@ -76,6 +76,8 @@ parquet::ReaderProperties MakeReaderProperties(
properties.disable_buffered_stream();
}
properties.set_buffer_size(parquet_scan_options->reader_properties->buffer_size());
+ properties.set_footer_read_size(
+ parquet_scan_options->reader_properties->footer_read_size());
auto file_decryption_prop =
parquet_scan_options->reader_properties->file_decryption_properties();
@@ -101,6 +103,8 @@ parquet::ReaderProperties MakeReaderProperties(
parquet_scan_options->reader_properties->thrift_string_size_limit());
properties.set_thrift_container_size_limit(
parquet_scan_options->reader_properties->thrift_container_size_limit());
+ properties.set_schema_depth_limit(
+ parquet_scan_options->reader_properties->schema_depth_limit());
properties.set_page_checksum_verification(
parquet_scan_options->reader_properties->page_checksum_verification());
diff --git a/cpp/src/parquet/arrow/arrow_schema_test.cc
b/cpp/src/parquet/arrow/arrow_schema_test.cc
index 894f6890028..27c302fe0d4 100644
--- a/cpp/src/parquet/arrow/arrow_schema_test.cc
+++ b/cpp/src/parquet/arrow/arrow_schema_test.cc
@@ -1902,8 +1902,10 @@ class TestConvertRoundTrip : public ::testing::Test {
::parquet::default_writer_properties();
RETURN_NOT_OK(ToParquetSchema(arrow_schema_.get(), *properties.get(),
*arrow_properties, &parquet_schema_));
- ::parquet::schema::ToParquet(parquet_schema_->group_node(),
&parquet_format_schema_);
- auto parquet_schema =
::parquet::schema::FromParquet(parquet_format_schema_);
+ ::parquet::schema::SchemaToThrift(parquet_schema_->group_node(),
+ &parquet_format_schema_);
+ auto parquet_schema =
+ ::parquet::schema::SchemaFromThrift(parquet_format_schema_,
/*max_depth=*/100);
return FromParquetSchema(parquet_schema.get(), &result_schema_);
}
diff --git a/cpp/src/parquet/metadata.cc b/cpp/src/parquet/metadata.cc
index 183fcc82b1d..61a111fc0c3 100644
--- a/cpp/src/parquet/metadata.cc
+++ b/cpp/src/parquet/metadata.cc
@@ -778,9 +778,12 @@ class FileMetaData::FileMetaDataImpl {
public:
FileMetaDataImpl() = default;
- explicit FileMetaDataImpl(
- const void* metadata, int64_t metadata_len, ReaderProperties properties,
- std::shared_ptr<InternalFileDecryptor> file_decryptor = nullptr)
+ explicit FileMetaDataImpl(ReaderProperties properties)
+ : properties_(std::move(properties)) {}
+
+ FileMetaDataImpl(const void* metadata, int64_t metadata_len,
+ ReaderProperties properties,
+ std::shared_ptr<InternalFileDecryptor> file_decryptor =
nullptr)
: properties_(std::move(properties)),
file_decryptor_(std::move(file_decryptor)) {
metadata_ = std::make_unique<format::FileMetaData>();
@@ -1022,8 +1025,8 @@ class FileMetaData::FileMetaDataImpl {
if (metadata_->schema.empty()) {
throw ParquetException("Empty file schema (no root)");
}
- schema_.Init(schema::Unflatten(&metadata_->schema[0],
-
static_cast<int>(metadata_->schema.size())));
+ schema_.Init(schema::Unflatten(metadata_->schema,
+
/*max_depth=*/properties_.schema_depth_limit()));
}
void InitColumnOrders() {
@@ -1074,6 +1077,9 @@ FileMetaData::FileMetaData(const void* metadata, int64_t
metadata_len,
: impl_(new FileMetaDataImpl(metadata, metadata_len, properties,
std::move(file_decryptor))) {}
+FileMetaData::FileMetaData(ReaderProperties properties)
+ : impl_(new FileMetaDataImpl(std::move(properties))) {}
+
FileMetaData::FileMetaData() : impl_(new FileMetaDataImpl()) {}
FileMetaData::~FileMetaData() = default;
@@ -2147,10 +2153,15 @@ class FileMetaDataBuilder::FileMetaDataBuilderImpl {
}
}
-
ToParquet(static_cast<parquet::schema::GroupNode*>(schema_->schema_root().get()),
- &metadata_->schema);
- auto file_meta_data = std::unique_ptr<FileMetaData>(new FileMetaData());
+
SchemaToThrift(static_cast<parquet::schema::GroupNode*>(schema_->schema_root().get()),
+ &metadata_->schema);
+ ReaderProperties properties;
+ // Disable schema nesting depth for schema restruction in InitSchema below.
+ properties.set_schema_depth_limit(std::numeric_limits<int32_t>::max());
+ auto file_meta_data =
+ std::unique_ptr<FileMetaData>(new FileMetaData(std::move(properties)));
file_meta_data->impl_->metadata_ = std::move(metadata_);
+ // XXX Why are we reconstructing the schema from the flattened Thrift
structures?
file_meta_data->impl_->InitSchema();
file_meta_data->impl_->InitKeyValueMetadata();
return file_meta_data;
diff --git a/cpp/src/parquet/metadata.h b/cpp/src/parquet/metadata.h
index bab6bba1587..a79f790dfba 100644
--- a/cpp/src/parquet/metadata.h
+++ b/cpp/src/parquet/metadata.h
@@ -386,9 +386,10 @@ class PARQUET_EXPORT FileMetaData {
friend class SerializedFile;
friend class SerializedRowGroup;
- explicit FileMetaData(const void* serialized_metadata, int64_t metadata_len,
- const ReaderProperties& properties,
- std::shared_ptr<InternalFileDecryptor> file_decryptor
= NULLPTR);
+ explicit FileMetaData(ReaderProperties properties);
+ FileMetaData(const void* serialized_metadata, int64_t metadata_len,
+ const ReaderProperties& properties,
+ std::shared_ptr<InternalFileDecryptor> file_decryptor =
NULLPTR);
void set_file_decryptor(std::shared_ptr<InternalFileDecryptor>
file_decryptor);
const std::shared_ptr<InternalFileDecryptor>& file_decryptor() const;
diff --git a/cpp/src/parquet/properties.h b/cpp/src/parquet/properties.h
index e2244a1176e..f135faa0183 100644
--- a/cpp/src/parquet/properties.h
+++ b/cpp/src/parquet/properties.h
@@ -68,6 +68,10 @@ constexpr int32_t kDefaultThriftStringSizeLimit = 100 * 1000
* 1000;
// kDefaultStringSizeLimit.
constexpr int32_t kDefaultThriftContainerSizeLimit = 1000 * 1000;
+// Maximum schema nesting depth. This default value is conservatively small as
+// some systems may not set a very large stack size.
+constexpr int32_t kDefaultSchemaDepthLimit = 100;
+
// PARQUET-978: Minimize footer reads by reading 64 KB from the end of the file
constexpr int64_t kDefaultFooterReadSize = 64 * 1024;
@@ -121,6 +125,15 @@ class PARQUET_EXPORT ReaderProperties {
thrift_container_size_limit_ = size;
}
+ /// \brief Return the schema nesting depth limit.
+ ///
+ /// This limit helps prevent denial of service through excessive recursion
+ /// (stack overflow) when reconstructing the Parquet schema from the file
metadata.
+ /// The default value is conservative enough for most use cases.
+ int32_t schema_depth_limit() const { return schema_depth_limit_; }
+ /// Set the schema nesting depth limit.
+ void set_schema_depth_limit(int32_t size) { schema_depth_limit_ = size; }
+
/// Set the decryption properties.
void file_decryption_properties(std::shared_ptr<FileDecryptionProperties>
decryption) {
file_decryption_properties_ = std::move(decryption);
@@ -146,6 +159,7 @@ class PARQUET_EXPORT ReaderProperties {
int64_t buffer_size_ = kDefaultBufferSize;
int32_t thrift_string_size_limit_ = kDefaultThriftStringSizeLimit;
int32_t thrift_container_size_limit_ = kDefaultThriftContainerSizeLimit;
+ int32_t schema_depth_limit_ = kDefaultSchemaDepthLimit;
bool buffered_stream_enabled_ = false;
bool page_checksum_verification_ = false;
// Used with a RecordReader.
diff --git a/cpp/src/parquet/reader_test.cc b/cpp/src/parquet/reader_test.cc
index eeb839e71fe..d223ce0db64 100644
--- a/cpp/src/parquet/reader_test.cc
+++ b/cpp/src/parquet/reader_test.cc
@@ -138,6 +138,8 @@ std::string byte_stream_split_extended() {
return data_file("byte_stream_split_extended.gzip.parquet");
}
+std::string nested_lists() { return data_file("nested_lists.snappy.parquet"); }
+
template <typename DType, typename ValueType = typename DType::c_type>
std::vector<ValueType> ReadColumnValues(ParquetFileReader* file_reader, int
row_group,
int column, int64_t
expected_values_read) {
@@ -705,14 +707,32 @@ TEST(TestFileReader, RecordReaderWithExposingDictionary) {
}
}
+TEST(TestFileReader, SchemaDepthLimit) {
+#ifndef ARROW_WITH_SNAPPY
+ GTEST_SKIP() << "Test requires Snappy compression";
+#endif
+ ReaderProperties reader_props;
+ // File has a column "a.list.element.list.element.list.element"
+ // (nesting depth 8 including the root)
+ reader_props.set_schema_depth_limit(8);
+ std::unique_ptr<ParquetFileReader> file_reader =
+ ParquetFileReader::OpenFile(nested_lists(), /*memory_map=*/false,
reader_props);
+ reader_props.set_schema_depth_limit(7);
+ EXPECT_THAT(
+ [&] {
+ ParquetFileReader::OpenFile(nested_lists(), /*memory_map=*/false,
reader_props);
+ },
+ ::testing::ThrowsMessage<ParquetException>(
+ ::testing::HasSubstr("Parquet schema too deeply nested")));
+}
+
class TestLocalFile : public ::testing::Test {
public:
void SetUp() {
std::string dir_string(test::get_data_dir());
std::stringstream ss;
- ss << dir_string << "/"
- << "alltypes_plain.parquet";
+ ss << dir_string << "/" << "alltypes_plain.parquet";
PARQUET_ASSIGN_OR_THROW(handle, ReadableFile::Open(ss.str()));
fileno = handle->file_descriptor();
diff --git a/cpp/src/parquet/schema.cc b/cpp/src/parquet/schema.cc
index 0cfa49c21c1..3cb91f9a84e 100644
--- a/cpp/src/parquet/schema.cc
+++ b/cpp/src/parquet/schema.cc
@@ -20,6 +20,7 @@
#include <algorithm>
#include <cstring>
#include <memory>
+#include <sstream>
#include <string>
#include <type_traits>
#include <utility>
@@ -544,11 +545,15 @@ void PrimitiveNode::ToParquet(void* opaque_element) const
{
// ----------------------------------------------------------------------
// Schema converters
-std::unique_ptr<Node> Unflatten(const format::SchemaElement* elements, int
length) {
+std::unique_ptr<Node> Unflatten(std::span<const format::SchemaElement>
elements,
+ int max_depth) {
+ if (elements.empty()) {
+ throw ParquetException("Empty Parquet schema (no root)");
+ }
if (elements[0].num_children == 0) {
- if (length == 1) {
+ if (elements.size() == 1) {
// Degenerate case of Parquet file with no columns
- return GroupNode::FromParquet(elements, {});
+ return GroupNode::FromParquet(&elements[0], {});
} else {
throw ParquetException(
"Parquet schema had multiple nodes but root had no children");
@@ -558,11 +563,12 @@ std::unique_ptr<Node> Unflatten(const
format::SchemaElement* elements, int lengt
// We don't check that the root node is repeated since this is not
// consistently set by implementations
- int pos = 0;
+ size_t pos = 0;
+ size_t num_reserved = 0;
- std::function<std::unique_ptr<Node>()> NextNode = [&]() {
- if (pos == length) {
- throw ParquetException("Malformed schema: not enough elements");
+ std::function<std::unique_ptr<Node>(int depth)> NextNode = [&](int depth) {
+ if (pos == elements.size()) {
+ throw ParquetException("Malformed Parquet schema: not enough elements");
}
const SchemaElement& element = elements[pos++];
const void* opaque_element = static_cast<const void*>(&element);
@@ -572,22 +578,42 @@ std::unique_ptr<Node> Unflatten(const
format::SchemaElement* elements, int lengt
return PrimitiveNode::FromParquet(opaque_element);
} else {
// Group node (may have 0 children, but cannot have a type)
- NodeVector fields;
+ // Protect against denial-of-service through stack exhaustion when
parsing
+ // deeply nested schemas.
+ if (depth >= max_depth) {
+ std::stringstream ss;
+ ss << "Parquet schema too deeply nested, consider increasing schema
depth limit "
+ "(current limit is "
+ << max_depth << ")";
+ throw ParquetException(ss.str());
+ }
+ if (element.num_children < 0) {
+ throw ParquetException("Malformed Parquet schema: negative number of
children");
+ }
+ // Guard against excessive pre-reservation by an invalid schema.
+ // For example, a sequence of group nodes advertising N, N-1, etc.
children
+ // could lead to quadratic preallocation.
+ num_reserved += static_cast<size_t>(element.num_children);
+ if (num_reserved > elements.size()) {
+ throw ParquetException("Malformed Parquet schema: not enough
elements");
+ }
+ NodeVector fields(element.num_children);
for (int i = 0; i < element.num_children; ++i) {
- std::unique_ptr<Node> field = NextNode();
- fields.push_back(NodePtr(field.release()));
+ fields[i] = NextNode(depth + 1);
}
return GroupNode::FromParquet(opaque_element, std::move(fields));
}
};
- return NextNode();
+ auto root = NextNode(/*depth=*/1);
+ if (pos != elements.size()) {
+ throw ParquetException("Malformed Parquet schema: too many elements");
+ }
+ return root;
}
-std::shared_ptr<SchemaDescriptor> FromParquet(const
std::vector<SchemaElement>& schema) {
- if (schema.empty()) {
- throw ParquetException("Empty file schema (no root)");
- }
- std::unique_ptr<Node> root = Unflatten(&schema[0],
static_cast<int>(schema.size()));
+std::shared_ptr<SchemaDescriptor> SchemaFromThrift(std::span<const
SchemaElement> schema,
+ int max_depth) {
+ std::unique_ptr<Node> root = Unflatten(schema, max_depth);
std::shared_ptr<SchemaDescriptor> descr =
std::make_shared<SchemaDescriptor>();
descr->Init(std::shared_ptr<GroupNode>(static_cast<GroupNode*>(root.release())));
return descr;
@@ -615,7 +641,7 @@ class SchemaVisitor : public Node::ConstVisitor {
std::vector<format::SchemaElement>* elements_;
};
-void ToParquet(const GroupNode* schema, std::vector<format::SchemaElement>*
out) {
+void SchemaToThrift(const GroupNode* schema,
std::vector<format::SchemaElement>* out) {
SchemaVisitor visitor(out);
schema->VisitConst(&visitor);
}
@@ -716,8 +742,7 @@ struct SchemaPrinter : public Node::ConstVisitor {
void Visit(const GroupNode* node) {
PrintRepLevel(node->repetition(), stream_);
- stream_ << " group "
- << "field_id=" << node->field_id() << " " << node->name();
+ stream_ << " group " << "field_id=" << node->field_id() << " " <<
node->name();
auto lt = node->converted_type();
const auto& la = node->logical_type();
if (la && la->is_valid() && !la->is_none()) {
diff --git a/cpp/src/parquet/schema_internal.h
b/cpp/src/parquet/schema_internal.h
index c0cfffc87e2..56b6dc1bf24 100644
--- a/cpp/src/parquet/schema_internal.h
+++ b/cpp/src/parquet/schema_internal.h
@@ -20,6 +20,7 @@
#pragma once
#include <memory>
+#include <span>
#include <vector>
#include "parquet/platform.h"
@@ -38,17 +39,18 @@ namespace schema {
// Conversion from Parquet Thrift metadata
PARQUET_EXPORT
-std::shared_ptr<SchemaDescriptor> FromParquet(
- const std::vector<format::SchemaElement>& schema);
+std::shared_ptr<SchemaDescriptor> SchemaFromThrift(
+ std::span<const format::SchemaElement> schema, int max_depth);
PARQUET_EXPORT
-std::unique_ptr<Node> Unflatten(const format::SchemaElement* elements, int
length);
+std::unique_ptr<Node> Unflatten(std::span<const format::SchemaElement> schema,
+ int max_depth);
// ----------------------------------------------------------------------
// Conversion to Parquet Thrift metadata
PARQUET_EXPORT
-void ToParquet(const GroupNode* schema, std::vector<format::SchemaElement>*
out);
+void SchemaToThrift(const GroupNode* schema,
std::vector<format::SchemaElement>* out);
} // namespace schema
} // namespace parquet
diff --git a/cpp/src/parquet/schema_test.cc b/cpp/src/parquet/schema_test.cc
index 3888e4f8d95..6c8e6366adf 100644
--- a/cpp/src/parquet/schema_test.cc
+++ b/cpp/src/parquet/schema_test.cc
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+#include <gmock/gmock.h>
#include <gtest/gtest.h>
#include <cstdlib>
@@ -29,6 +30,7 @@
#include "parquet/exception.h"
#include "parquet/schema.h"
#include "parquet/schema_internal.h"
+#include "parquet/test_util.h"
#include "parquet/thrift_internal.h"
#include "parquet/types.h"
@@ -417,8 +419,8 @@ class TestSchemaConverter : public ::testing::Test {
public:
void setUp() { name_ = "parquet_schema"; }
- void Convert(const parquet::format::SchemaElement* elements, int length) {
- node_ = Unflatten(elements, length);
+ void Convert(std::span<const parquet::format::SchemaElement> elements) {
+ node_ = Unflatten(elements, max_depth_);
ASSERT_TRUE(node_->is_group());
group_ = static_cast<const GroupNode*>(node_.get());
}
@@ -427,6 +429,7 @@ class TestSchemaConverter : public ::testing::Test {
std::string name_;
const GroupNode* group_;
std::unique_ptr<Node> node_;
+ int max_depth_ = 10;
};
bool check_for_parent_consistency(const GroupNode* node) {
@@ -464,7 +467,7 @@ TEST_F(TestSchemaConverter, NestedExample) {
elements.push_back(elt);
elements.push_back(NewPrimitive("item", FieldRepetitionType::OPTIONAL,
Type::INT64, 4));
- ASSERT_NO_FATAL_FAILURE(Convert(&elements[0],
static_cast<int>(elements.size())));
+ ASSERT_NO_FATAL_FAILURE(Convert(elements));
// Construct the expected schema
NodeVector fields;
@@ -492,7 +495,7 @@ TEST_F(TestSchemaConverter, ZeroColumns) {
// ARROW-3843
SchemaElement elements[1];
elements[0] = NewGroup("schema", FieldRepetitionType::REPEATED, 0, 0);
- ASSERT_NO_THROW(Convert(elements, 1));
+ ASSERT_NO_THROW(Convert(elements));
}
TEST_F(TestSchemaConverter, InvalidRoot) {
@@ -504,7 +507,7 @@ TEST_F(TestSchemaConverter, InvalidRoot) {
SchemaElement elements[2];
elements[0] =
NewPrimitive("not-a-group", FieldRepetitionType::REQUIRED, Type::INT32,
0);
- ASSERT_THROW(Convert(elements, 2), ParquetException);
+ ASSERT_THROW(Convert(elements), ParquetException);
// While the Parquet spec indicates that the root group should have REPEATED
// repetition type, some implementations may return REQUIRED or OPTIONAL
@@ -512,10 +515,10 @@ TEST_F(TestSchemaConverter, InvalidRoot) {
// practicality matter.
elements[0] = NewGroup("not-repeated", FieldRepetitionType::REQUIRED, 1, 0);
elements[1] = NewPrimitive("a", FieldRepetitionType::REQUIRED, Type::INT32,
1);
- ASSERT_NO_FATAL_FAILURE(Convert(elements, 2));
+ ASSERT_NO_FATAL_FAILURE(Convert(elements));
elements[0] = NewGroup("not-repeated", FieldRepetitionType::OPTIONAL, 1, 0);
- ASSERT_NO_FATAL_FAILURE(Convert(elements, 2));
+ ASSERT_NO_FATAL_FAILURE(Convert(elements));
}
TEST_F(TestSchemaConverter, NotEnoughChildren) {
@@ -523,7 +526,49 @@ TEST_F(TestSchemaConverter, NotEnoughChildren) {
SchemaElement elt;
std::vector<SchemaElement> elements;
elements.push_back(NewGroup(name_, FieldRepetitionType::REPEATED, 2, 0));
- ASSERT_THROW(Convert(&elements[0], 1), ParquetException);
+ EXPECT_THAT([&] { Convert(elements); },
+ ::testing::ThrowsMessage<ParquetException>(
+ ::testing::HasSubstr("not enough elements")));
+}
+
+TEST_F(TestSchemaConverter, TooManyElements) {
+ SchemaElement elt;
+ std::vector<SchemaElement> elements;
+ elements.push_back(NewGroup(name_, FieldRepetitionType::REPEATED,
/*num_children=*/2));
+ elements.push_back(NewPrimitive("int1", FieldRepetitionType::REQUIRED,
Type::INT32));
+ elements.push_back(NewPrimitive("int2", FieldRepetitionType::REQUIRED,
Type::INT32));
+ // Unexpected supplementary node
+ elements.push_back(NewPrimitive("int3", FieldRepetitionType::REQUIRED,
Type::INT32));
+ EXPECT_THAT([&] { Convert(elements); },
::testing::ThrowsMessage<ParquetException>(
+ ::testing::HasSubstr("too many
elements")));
+}
+
+TEST_F(TestSchemaConverter, MaxDepth) {
+ this->max_depth_ = 5;
+
+ std::vector<SchemaElement> wide_schema;
+ std::vector<SchemaElement> deep_schema;
+
+ // Max depth doesn't limit breadth of schema
+ wide_schema.push_back(NewGroup("root", FieldRepetitionType::REQUIRED,
+ /*num_children=*/this->max_depth_ + 1));
+ for (int i = 0; i < this->max_depth_ + 1; ++i) {
+ wide_schema.push_back(NewPrimitive("int" + std::to_string(i),
+ FieldRepetitionType::REQUIRED,
Type::INT32));
+ }
+ ASSERT_NO_FATAL_FAILURE(Convert(wide_schema));
+
+ // Max depth prevents excessive recursion
+ for (int i = 0; i < this->max_depth_; ++i) {
+ deep_schema.push_back(NewGroup("group" + std::to_string(i),
+ FieldRepetitionType::REQUIRED,
/*num_children=*/1));
+ }
+ deep_schema.push_back(NewPrimitive("int", FieldRepetitionType::REQUIRED,
Type::INT32));
+ EXPECT_THAT([&] { Convert(deep_schema); },
+ ::testing::ThrowsMessage<ParquetException>(
+ ::testing::HasSubstr("Parquet schema too deeply nested")));
+ ++this->max_depth_;
+ ASSERT_NO_FATAL_FAILURE(Convert(deep_schema));
}
// ----------------------------------------------------------------------
@@ -533,7 +578,7 @@ class TestSchemaFlatten : public ::testing::Test {
public:
void setUp() { name_ = "parquet_schema"; }
- void Flatten(const GroupNode* schema) { ToParquet(schema, &elements_); }
+ void Flatten(const GroupNode* schema) { SchemaToThrift(schema, &elements_); }
protected:
std::string name_;
@@ -2286,7 +2331,7 @@ TEST(TestLogicalTypeSerialization,
SchemaElementNestedCases) {
timestamp_node, int_node, decimal_node},
ListLogicalType::Make());
std::vector<format::SchemaElement> list_elements;
- ToParquet(reinterpret_cast<GroupNode*>(list_node.get()), &list_elements);
+ SchemaToThrift(reinterpret_cast<GroupNode*>(list_node.get()),
&list_elements);
ASSERT_EQ(list_elements[0].name, "list");
ASSERT_TRUE(list_elements[0].__isset.converted_type);
ASSERT_TRUE(list_elements[0].__isset.logicalType);
@@ -2303,7 +2348,7 @@ TEST(TestLogicalTypeSerialization,
SchemaElementNestedCases) {
NodePtr map_node =
GroupNode::Make("map", Repetition::REQUIRED, {}, MapLogicalType::Make());
std::vector<format::SchemaElement> map_elements;
- ToParquet(reinterpret_cast<GroupNode*>(map_node.get()), &map_elements);
+ SchemaToThrift(reinterpret_cast<GroupNode*>(map_node.get()), &map_elements);
ASSERT_EQ(map_elements[0].name, "map");
ASSERT_TRUE(map_elements[0].__isset.converted_type);
ASSERT_TRUE(map_elements[0].__isset.logicalType);
@@ -2401,7 +2446,7 @@ TEST(TestLogicalTypeSerialization,
VariantSpecificationVersion) {
// Verify thrift serialization
std::vector<format::SchemaElement> elements;
- ToParquet(reinterpret_cast<GroupNode*>(variant_node.get()), &elements);
+ SchemaToThrift(reinterpret_cast<GroupNode*>(variant_node.get()), &elements);
// Verify that logicalType is set and is VARIANT
ASSERT_EQ(elements[0].name, "variant");
diff --git a/python/pyarrow/_dataset_parquet.pyx
b/python/pyarrow/_dataset_parquet.pyx
index 534f7790923..6ac9383a022 100644
--- a/python/pyarrow/_dataset_parquet.pyx
+++ b/python/pyarrow/_dataset_parquet.pyx
@@ -758,6 +758,10 @@ cdef class ParquetFragmentScanOptions(FragmentScanOptions):
If not None, override the maximum total size of containers allocated
when decoding Thrift structures. The default limit should be
sufficient for most Parquet files.
+ schema_depth_limit : int, default None
+ If not None, override the maximum nesting depth of the Parquet file
schema.
+ This guards against recursion overflow on invalid schemas.
+ The default limit should be sufficient for most Parquet files.
decryption_config : pyarrow.dataset.ParquetDecryptionConfig, default None
If not None, use the provided ParquetDecryptionConfig to decrypt the
Parquet file.
@@ -781,6 +785,7 @@ cdef class ParquetFragmentScanOptions(FragmentScanOptions):
cache_options=None,
thrift_string_size_limit=None,
thrift_container_size_limit=None,
+ schema_depth_limit=None,
decryption_config=None,
decryption_properties=None,
bint page_checksum_verification=False,
@@ -798,6 +803,8 @@ cdef class ParquetFragmentScanOptions(FragmentScanOptions):
self.thrift_string_size_limit = thrift_string_size_limit
if thrift_container_size_limit is not None:
self.thrift_container_size_limit = thrift_container_size_limit
+ if schema_depth_limit is not None:
+ self.schema_depth_limit = schema_depth_limit
if decryption_config is not None:
self.parquet_decryption_config = decryption_config
if decryption_properties is not None:
@@ -874,6 +881,16 @@ cdef class ParquetFragmentScanOptions(FragmentScanOptions):
raise ValueError("size must be larger than zero")
self.reader_properties().set_thrift_container_size_limit(size)
+ @property
+ def schema_depth_limit(self):
+ return self.reader_properties().schema_depth_limit()
+
+ @schema_depth_limit.setter
+ def schema_depth_limit(self, limit):
+ if limit <= 0:
+ raise ValueError("limit must be larger than zero")
+ self.reader_properties().set_schema_depth_limit(limit)
+
@property
def decryption_properties(self):
if not parquet_encryption_enabled:
@@ -941,11 +958,12 @@ cdef class
ParquetFragmentScanOptions(FragmentScanOptions):
attrs = (
self.use_buffered_stream, self.buffer_size, self.pre_buffer,
self.cache_options,
self.thrift_string_size_limit, self.thrift_container_size_limit,
- self.page_checksum_verification, self.arrow_extensions_enabled)
+ self.schema_depth_limit, self.page_checksum_verification,
+ self.arrow_extensions_enabled)
other_attrs = (
other.use_buffered_stream, other.buffer_size, other.pre_buffer,
other.cache_options,
- other.thrift_string_size_limit,
- other.thrift_container_size_limit,
other.page_checksum_verification,
+ other.thrift_string_size_limit, other.thrift_container_size_limit,
+ other.schema_depth_limit, other.page_checksum_verification,
other.arrow_extensions_enabled)
return attrs == other_attrs
@@ -963,6 +981,7 @@ cdef class ParquetFragmentScanOptions(FragmentScanOptions):
cache_options=self.cache_options,
thrift_string_size_limit=self.thrift_string_size_limit,
thrift_container_size_limit=self.thrift_container_size_limit,
+ schema_depth_limit=self.schema_depth_limit,
page_checksum_verification=self.page_checksum_verification,
arrow_extensions_enabled=self.arrow_extensions_enabled
)
diff --git a/python/pyarrow/_parquet.pyx b/python/pyarrow/_parquet.pyx
index 932632a5041..2621c3bd6d6 100644
--- a/python/pyarrow/_parquet.pyx
+++ b/python/pyarrow/_parquet.pyx
@@ -1592,6 +1592,7 @@ cdef class ParquetReader(_Weakrefable):
FileDecryptionProperties decryption_properties=None,
thrift_string_size_limit=None,
thrift_container_size_limit=None,
+ schema_depth_limit=None,
page_checksum_verification=False,
arrow_extensions_enabled=False):
"""
@@ -1611,6 +1612,7 @@ cdef class ParquetReader(_Weakrefable):
decryption_properties : FileDecryptionProperties, optional
thrift_string_size_limit : int, optional
thrift_container_size_limit : int, optional
+ schema_depth_limit : int, optional
page_checksum_verification : bool, default False
arrow_extensions_enabled : bool, default False
"""
@@ -1646,6 +1648,11 @@ cdef class ParquetReader(_Weakrefable):
"must be larger than zero")
properties.set_thrift_container_size_limit(
thrift_container_size_limit)
+ if schema_depth_limit is not None:
+ if schema_depth_limit <= 0:
+ raise ValueError("schema_depth_limit "
+ "must be larger than zero")
+ properties.set_schema_depth_limit(schema_depth_limit)
if decryption_properties is not None:
properties.file_decryption_properties(
diff --git a/python/pyarrow/includes/libparquet.pxd
b/python/pyarrow/includes/libparquet.pxd
index df353cc7805..915b1d6dd36 100644
--- a/python/pyarrow/includes/libparquet.pxd
+++ b/python/pyarrow/includes/libparquet.pxd
@@ -431,6 +431,9 @@ cdef extern from "parquet/api/reader.h" namespace "parquet"
nogil:
void set_thrift_container_size_limit(int32_t size)
int32_t thrift_container_size_limit() const
+ void set_schema_depth_limit(int32_t limit)
+ int32_t schema_depth_limit() const
+
void file_decryption_properties(shared_ptr[CFileDecryptionProperties]
decryption)
shared_ptr[CFileDecryptionProperties] file_decryption_properties() \
diff --git a/python/pyarrow/parquet/core.py b/python/pyarrow/parquet/core.py
index 4acfa4f6e22..8e49a96a8f8 100644
--- a/python/pyarrow/parquet/core.py
+++ b/python/pyarrow/parquet/core.py
@@ -258,6 +258,10 @@ class ParquetFile:
If not None, override the maximum total size of containers allocated
when decoding Thrift structures. The default limit should be
sufficient for most Parquet files.
+ schema_depth_limit : int, default None
+ If not None, override the maximum nesting depth of the Parquet file
schema.
+ This guards against recursion overflow on invalid schemas.
+ The default limit should be sufficient for most Parquet files.
filesystem : FileSystem, default None
If nothing passed, will be inferred based on path.
Path will try to be found in the local on-disk filesystem otherwise
@@ -316,7 +320,8 @@ class ParquetFile:
memory_map=False, buffer_size=0, pre_buffer=True,
coerce_int96_timestamp_unit=None,
decryption_properties=None, thrift_string_size_limit=None,
- thrift_container_size_limit=None, filesystem=None,
+ thrift_container_size_limit=None, schema_depth_limit=None,
+ filesystem=None,
page_checksum_verification=False,
arrow_extensions_enabled=True):
self._close_source = getattr(source, 'closed', True)
@@ -337,6 +342,7 @@ class ParquetFile:
decryption_properties=decryption_properties,
thrift_string_size_limit=thrift_string_size_limit,
thrift_container_size_limit=thrift_container_size_limit,
+ schema_depth_limit=schema_depth_limit,
page_checksum_verification=page_checksum_verification,
arrow_extensions_enabled=arrow_extensions_enabled,
)
@@ -1372,6 +1378,10 @@ thrift_container_size_limit : int, default None
If not None, override the maximum total size of containers allocated
when decoding Thrift structures. The default limit should be
sufficient for most Parquet files.
+schema_depth_limit : int, default None
+ If not None, override the maximum nesting depth of the Parquet file schema.
+ This guards against recursion overflow on invalid schemas.
+ The default limit should be sufficient for most Parquet files.
page_checksum_verification : bool, default False
If True, verify the page checksum for each page read from the file.
arrow_extensions_enabled : bool, default True
@@ -1390,7 +1400,7 @@ Examples
ignore_prefixes=None,
pre_buffer=True, coerce_int96_timestamp_unit=None,
decryption_properties=None, thrift_string_size_limit=None,
- thrift_container_size_limit=None,
+ thrift_container_size_limit=None, schema_depth_limit=None,
page_checksum_verification=False,
arrow_extensions_enabled=True):
import pyarrow.dataset as ds
@@ -1401,6 +1411,7 @@ Examples
"coerce_int96_timestamp_unit": coerce_int96_timestamp_unit,
"thrift_string_size_limit": thrift_string_size_limit,
"thrift_container_size_limit": thrift_container_size_limit,
+ "schema_depth_limit": schema_depth_limit,
"page_checksum_verification": page_checksum_verification,
"arrow_extensions_enabled": arrow_extensions_enabled,
"binary_type": binary_type,
@@ -1788,6 +1799,10 @@ thrift_container_size_limit : int, default None
If not None, override the maximum total size of containers allocated
when decoding Thrift structures. The default limit should be
sufficient for most Parquet files.
+schema_depth_limit : int, default None
+ If not None, override the maximum nesting depth of the Parquet file schema.
+ This guards against recursion overflow on invalid schemas.
+ The default limit should be sufficient for most Parquet files.
page_checksum_verification : bool, default False
If True, verify the checksum for each page read from the file.
arrow_extensions_enabled : bool, default True
@@ -1888,7 +1903,7 @@ def read_table(source, *, columns=None, use_threads=True,
ignore_prefixes=None, pre_buffer=True,
coerce_int96_timestamp_unit=None,
decryption_properties=None, thrift_string_size_limit=None,
- thrift_container_size_limit=None,
+ thrift_container_size_limit=None, schema_depth_limit=None,
page_checksum_verification=False,
arrow_extensions_enabled=True):
@@ -1910,6 +1925,7 @@ def read_table(source, *, columns=None, use_threads=True,
decryption_properties=decryption_properties,
thrift_string_size_limit=thrift_string_size_limit,
thrift_container_size_limit=thrift_container_size_limit,
+ schema_depth_limit=schema_depth_limit,
page_checksum_verification=page_checksum_verification,
arrow_extensions_enabled=arrow_extensions_enabled,
)
@@ -1958,6 +1974,7 @@ def read_table(source, *, columns=None, use_threads=True,
decryption_properties=decryption_properties,
thrift_string_size_limit=thrift_string_size_limit,
thrift_container_size_limit=thrift_container_size_limit,
+ schema_depth_limit=schema_depth_limit,
page_checksum_verification=page_checksum_verification,
)
diff --git a/python/pyarrow/tests/parquet/test_basic.py
b/python/pyarrow/tests/parquet/test_basic.py
index 20e3f51bb67..8b91090989a 100644
--- a/python/pyarrow/tests/parquet/test_basic.py
+++ b/python/pyarrow/tests/parquet/test_basic.py
@@ -934,6 +934,28 @@ def test_thrift_size_limits(tempdir):
assert got == table
+def test_schema_depth_limit(tempdir):
+ path = tempdir / 'nested_schema.parquet'
+
+ # A 10-level nested list. The Parquet schema nesting depth will be 22:
+ # - two levels of nesting for each Arrow list
+ # - one level for the list leaf
+ # - one level for the schema root
+ array = pa.array([[[[[[[[[[[42]]]]]]]]]]])
+ table = pa.table([array], names=['nested_list'])
+ pq.write_table(table, path)
+
+ with pytest.raises(
+ OSError,
+ match="Parquet schema too deeply nested"):
+ pq.read_table(path, schema_depth_limit=21)
+
+ got = pq.read_table(path, schema_depth_limit=22)
+ assert got == table
+ got = pq.read_table(path)
+ assert got == table
+
+
def test_page_checksum_verification_write_table(tempdir):
"""Check that checksum verification works for datasets created with
pq.write_table()"""
diff --git a/python/pyarrow/tests/test_dataset.py
b/python/pyarrow/tests/test_dataset.py
index 09d7cfb9d9d..1e0fa7c2e24 100644
--- a/python/pyarrow/tests/test_dataset.py
+++ b/python/pyarrow/tests/test_dataset.py
@@ -1022,6 +1022,7 @@ def test_parquet_scan_options():
cache_opts = pa.CacheOptions(
hole_size_limit=2**10, range_size_limit=8*2**10, lazy=True)
opts7 = ds.ParquetFragmentScanOptions(pre_buffer=True,
cache_options=cache_opts)
+ opts8 = ds.ParquetFragmentScanOptions(schema_depth_limit=42)
assert opts1.use_buffered_stream is False
assert opts1.buffer_size == 2**13
@@ -1030,6 +1031,7 @@ def test_parquet_scan_options():
assert opts1.thrift_string_size_limit == 100_000_000 # default in C++
assert opts1.thrift_container_size_limit == 1_000_000 # default in C++
assert opts1.page_checksum_verification is False
+ assert opts1.schema_depth_limit == 100 # default in C++
assert opts2.use_buffered_stream is False
assert opts2.buffer_size == 2**12
@@ -1056,6 +1058,8 @@ def test_parquet_scan_options():
assert opts7.cache_options == cache_opts
assert opts7.cache_options != opts1.cache_options
+ assert opts8.schema_depth_limit == 42
+
assert opts1 == opts1
assert opts1 != opts2
assert opts2 != opts3
@@ -1063,6 +1067,7 @@ def test_parquet_scan_options():
assert opts5 != opts1
assert opts6 != opts1
assert opts7 != opts1
+ assert opts8 != opts1
def test_file_format_pickling(pickle_module):
@@ -1097,6 +1102,7 @@ def test_file_format_pickling(pickle_module):
buffer_size=4096,
thrift_string_size_limit=123,
thrift_container_size_limit=456,
+ schema_depth_limit=42,
),
])
