From 5ac14bc616a84e27462fefc80d5c18538ca086e1 Mon Sep 17 00:00:00 2001 From: Rok Mihevc Date: Thu, 4 Jun 2026 15:56:22 +0200 Subject: [PATCH 1/6] add FixedSizeListLogicalType --- cpp/src/generated/parquet_types.cpp | 67 ++++++++++++++++++++ cpp/src/generated/parquet_types.h | 42 +++++++++++- cpp/src/generated/parquet_types.tcc | 74 ++++++++++++++++++++++ cpp/src/parquet/arrow/arrow_schema_test.cc | 24 ++++++- cpp/src/parquet/arrow/schema.cc | 35 +++++++--- cpp/src/parquet/parquet.thrift | 12 ++++ cpp/src/parquet/schema.cc | 10 ++- cpp/src/parquet/schema_test.cc | 34 ++++++++++ cpp/src/parquet/types.cc | 74 +++++++++++++++++++++- cpp/src/parquet/types.h | 14 ++++ python/pyarrow/_parquet.pyx | 1 + python/pyarrow/includes/libparquet.pxd | 1 + 12 files changed, 372 insertions(+), 16 deletions(-) diff --git a/cpp/src/generated/parquet_types.cpp b/cpp/src/generated/parquet_types.cpp index 0ee973f2a2d6..bb643a0b2ad3 100644 --- a/cpp/src/generated/parquet_types.cpp +++ b/cpp/src/generated/parquet_types.cpp @@ -2255,6 +2255,58 @@ void GeographyType::printTo(std::ostream& out) const { } + +FixedSizeListType::~FixedSizeListType() noexcept { +} + +FixedSizeListType::FixedSizeListType() noexcept + : num_values(0) { +} + +void FixedSizeListType::__set_num_values(const int32_t val) { + this->num_values = val; +} +std::ostream& operator<<(std::ostream& out, const FixedSizeListType& obj) +{ + obj.printTo(out); + return out; +} + + +void swap(FixedSizeListType &a, FixedSizeListType &b) { + using ::std::swap; + swap(a.num_values, b.num_values); +} + +bool FixedSizeListType::operator==(const FixedSizeListType & rhs) const +{ + if (!(num_values == rhs.num_values)) + return false; + return true; +} + +FixedSizeListType::FixedSizeListType(const FixedSizeListType& other110) noexcept { + num_values = other110.num_values; +} +FixedSizeListType::FixedSizeListType(FixedSizeListType&& other111) noexcept { + num_values = other111.num_values; +} +FixedSizeListType& FixedSizeListType::operator=(const FixedSizeListType& other112) noexcept { + num_values = other112.num_values; + return *this; +} +FixedSizeListType& FixedSizeListType::operator=(FixedSizeListType&& other113) noexcept { + num_values = other113.num_values; + return *this; +} +void FixedSizeListType::printTo(std::ostream& out) const { + using ::apache::thrift::to_string; + out << "FixedSizeListType("; + out << "num_values=" << to_string(num_values); + out << ")"; +} + + LogicalType::~LogicalType() noexcept { } @@ -2345,6 +2397,11 @@ void LogicalType::__set_GEOGRAPHY(const GeographyType& val) { this->GEOGRAPHY = val; __isset.GEOGRAPHY = true; } + +void LogicalType::__set_FIXED_SIZE_LIST(const FixedSizeListType& val) { + this->FIXED_SIZE_LIST = val; +__isset.FIXED_SIZE_LIST = true; +} std::ostream& operator<<(std::ostream& out, const LogicalType& obj) { obj.printTo(out); @@ -2371,6 +2428,7 @@ void swap(LogicalType &a, LogicalType &b) { swap(a.VARIANT, b.VARIANT); swap(a.GEOMETRY, b.GEOMETRY); swap(a.GEOGRAPHY, b.GEOGRAPHY); + swap(a.FIXED_SIZE_LIST, b.FIXED_SIZE_LIST); swap(a.__isset, b.__isset); } @@ -2444,6 +2502,10 @@ bool LogicalType::operator==(const LogicalType & rhs) const return false; else if (__isset.GEOGRAPHY && !(GEOGRAPHY == rhs.GEOGRAPHY)) return false; + if (__isset.FIXED_SIZE_LIST != rhs.__isset.FIXED_SIZE_LIST) + return false; + else if (__isset.FIXED_SIZE_LIST && !(FIXED_SIZE_LIST == rhs.FIXED_SIZE_LIST)) + return false; return true; } @@ -2465,6 +2527,7 @@ LogicalType::LogicalType(const LogicalType& other119) { VARIANT = other119.VARIANT; GEOMETRY = other119.GEOMETRY; GEOGRAPHY = other119.GEOGRAPHY; + FIXED_SIZE_LIST = other119.FIXED_SIZE_LIST; __isset = other119.__isset; } LogicalType::LogicalType(LogicalType&& other120) noexcept { @@ -2485,6 +2548,7 @@ LogicalType::LogicalType(LogicalType&& other120) noexcept { VARIANT = std::move(other120.VARIANT); GEOMETRY = std::move(other120.GEOMETRY); GEOGRAPHY = std::move(other120.GEOGRAPHY); + FIXED_SIZE_LIST = std::move(other120.FIXED_SIZE_LIST); __isset = other120.__isset; } LogicalType& LogicalType::operator=(const LogicalType& other121) { @@ -2505,6 +2569,7 @@ LogicalType& LogicalType::operator=(const LogicalType& other121) { VARIANT = other121.VARIANT; GEOMETRY = other121.GEOMETRY; GEOGRAPHY = other121.GEOGRAPHY; + FIXED_SIZE_LIST = other121.FIXED_SIZE_LIST; __isset = other121.__isset; return *this; } @@ -2526,6 +2591,7 @@ LogicalType& LogicalType::operator=(LogicalType&& other122) noexcept { VARIANT = std::move(other122.VARIANT); GEOMETRY = std::move(other122.GEOMETRY); GEOGRAPHY = std::move(other122.GEOGRAPHY); + FIXED_SIZE_LIST = std::move(other122.FIXED_SIZE_LIST); __isset = other122.__isset; return *this; } @@ -2549,6 +2615,7 @@ void LogicalType::printTo(std::ostream& out) const { out << ", " << "VARIANT="; (__isset.VARIANT ? (out << to_string(VARIANT)) : (out << "")); out << ", " << "GEOMETRY="; (__isset.GEOMETRY ? (out << to_string(GEOMETRY)) : (out << "")); out << ", " << "GEOGRAPHY="; (__isset.GEOGRAPHY ? (out << to_string(GEOGRAPHY)) : (out << "")); + out << ", " << "FIXED_SIZE_LIST="; (__isset.FIXED_SIZE_LIST ? (out << to_string(FIXED_SIZE_LIST)) : (out << "")); out << ")"; } diff --git a/cpp/src/generated/parquet_types.h b/cpp/src/generated/parquet_types.h index 1f1e254f5cf2..31800465157c 100644 --- a/cpp/src/generated/parquet_types.h +++ b/cpp/src/generated/parquet_types.h @@ -1620,8 +1620,44 @@ void swap(GeographyType &a, GeographyType &b); std::ostream& operator<<(std::ostream& out, const GeographyType& obj); +/** + * Fixed-size list logical type annotation + */ +class FixedSizeListType { + public: + + FixedSizeListType(const FixedSizeListType&) noexcept; + FixedSizeListType(FixedSizeListType&&) noexcept; + FixedSizeListType& operator=(const FixedSizeListType&) noexcept; + FixedSizeListType& operator=(FixedSizeListType&&) noexcept; + FixedSizeListType() noexcept; + + virtual ~FixedSizeListType() noexcept; + int32_t num_values; + + void __set_num_values(const int32_t val); + + bool operator == (const FixedSizeListType & rhs) const; + bool operator != (const FixedSizeListType &rhs) const { + return !(*this == rhs); + } + + bool operator < (const FixedSizeListType & ) const; + + template + uint32_t read(Protocol_* iprot); + template + uint32_t write(Protocol_* oprot) const; + + virtual void printTo(std::ostream& out) const; +}; + +void swap(FixedSizeListType &a, FixedSizeListType &b); + +std::ostream& operator<<(std::ostream& out, const FixedSizeListType& obj); + typedef struct _LogicalType__isset { - _LogicalType__isset() : STRING(false), MAP(false), LIST(false), ENUM(false), DECIMAL(false), DATE(false), TIME(false), TIMESTAMP(false), INTEGER(false), UNKNOWN(false), JSON(false), BSON(false), UUID(false), FLOAT16(false), VARIANT(false), GEOMETRY(false), GEOGRAPHY(false) {} + _LogicalType__isset() : STRING(false), MAP(false), LIST(false), ENUM(false), DECIMAL(false), DATE(false), TIME(false), TIMESTAMP(false), INTEGER(false), UNKNOWN(false), JSON(false), BSON(false), UUID(false), FLOAT16(false), VARIANT(false), GEOMETRY(false), GEOGRAPHY(false), FIXED_SIZE_LIST(false) {} bool STRING :1; bool MAP :1; bool LIST :1; @@ -1639,6 +1675,7 @@ typedef struct _LogicalType__isset { bool VARIANT :1; bool GEOMETRY :1; bool GEOGRAPHY :1; + bool FIXED_SIZE_LIST :1; } _LogicalType__isset; /** @@ -1675,6 +1712,7 @@ class LogicalType { VariantType VARIANT; GeometryType GEOMETRY; GeographyType GEOGRAPHY; + FixedSizeListType FIXED_SIZE_LIST; _LogicalType__isset __isset; @@ -1712,6 +1750,8 @@ class LogicalType { void __set_GEOGRAPHY(const GeographyType& val); + void __set_FIXED_SIZE_LIST(const FixedSizeListType& val); + bool operator == (const LogicalType & rhs) const; bool operator != (const LogicalType &rhs) const { return !(*this == rhs); diff --git a/cpp/src/generated/parquet_types.tcc b/cpp/src/generated/parquet_types.tcc index 78e3e2549394..7e77d23c362a 100644 --- a/cpp/src/generated/parquet_types.tcc +++ b/cpp/src/generated/parquet_types.tcc @@ -1625,6 +1625,67 @@ uint32_t GeographyType::write(Protocol_* oprot) const { return xfer; } + +template +uint32_t FixedSizeListType::read(Protocol_* iprot) { + + ::apache::thrift::protocol::TInputRecursionTracker tracker(*iprot); + uint32_t xfer = 0; + std::string fname; + ::apache::thrift::protocol::TType ftype; + int16_t fid; + + xfer += iprot->readStructBegin(fname); + + using ::apache::thrift::protocol::TProtocolException; + + bool isset_num_values = false; + + while (true) + { + xfer += iprot->readFieldBegin(fname, ftype, fid); + if (ftype == ::apache::thrift::protocol::T_STOP) { + break; + } + switch (fid) + { + case 1: + if (ftype == ::apache::thrift::protocol::T_I32) { + xfer += iprot->readI32(this->num_values); + isset_num_values = true; + } else { + xfer += iprot->skip(ftype); + } + break; + default: + xfer += iprot->skip(ftype); + break; + } + xfer += iprot->readFieldEnd(); + } + + xfer += iprot->readStructEnd(); + + if (!isset_num_values) + throw TProtocolException(TProtocolException::INVALID_DATA); + return xfer; +} + +template +uint32_t FixedSizeListType::write(Protocol_* oprot) const { + uint32_t xfer = 0; + ::apache::thrift::protocol::TOutputRecursionTracker tracker(*oprot); + xfer += oprot->writeStructBegin("FixedSizeListType"); + + xfer += oprot->writeFieldBegin("num_values", ::apache::thrift::protocol::T_I32, 1); + xfer += oprot->writeI32(this->num_values); + xfer += oprot->writeFieldEnd(); + + xfer += oprot->writeFieldStop(); + xfer += oprot->writeStructEnd(); + return xfer; +} + template uint32_t LogicalType::read(Protocol_* iprot) { @@ -1783,6 +1844,14 @@ uint32_t LogicalType::read(Protocol_* iprot) { xfer += iprot->skip(ftype); } break; + case 19: + if (ftype == ::apache::thrift::protocol::T_STRUCT) { + xfer += this->FIXED_SIZE_LIST.read(iprot); + this->__isset.FIXED_SIZE_LIST = true; + } else { + xfer += iprot->skip(ftype); + } + break; default: xfer += iprot->skip(ftype); break; @@ -1886,6 +1955,11 @@ uint32_t LogicalType::write(Protocol_* oprot) const { xfer += this->GEOGRAPHY.write(oprot); xfer += oprot->writeFieldEnd(); } + if (this->__isset.FIXED_SIZE_LIST) { + xfer += oprot->writeFieldBegin("FIXED_SIZE_LIST", ::apache::thrift::protocol::T_STRUCT, 19); + xfer += this->FIXED_SIZE_LIST.write(oprot); + xfer += oprot->writeFieldEnd(); + } xfer += oprot->writeFieldStop(); xfer += oprot->writeStructEnd(); return xfer; diff --git a/cpp/src/parquet/arrow/arrow_schema_test.cc b/cpp/src/parquet/arrow/arrow_schema_test.cc index 7d9ecb5e6449..baed6caf0062 100644 --- a/cpp/src/parquet/arrow/arrow_schema_test.cc +++ b/cpp/src/parquet/arrow/arrow_schema_test.cc @@ -748,6 +748,24 @@ TEST_F(TestConvertParquetSchema, ParquetLists) { } } +TEST_F(TestConvertParquetSchema, ParquetFixedSizeList) { + std::vector parquet_fields; + std::vector> arrow_fields; + + auto element = PrimitiveNode::Make("string", Repetition::OPTIONAL, + ParquetType::BYTE_ARRAY, ConvertedType::UTF8); + auto list = GroupNode::Make("list", Repetition::REPEATED, {element}); + parquet_fields.push_back(GroupNode::Make("my_fixed_size_list", Repetition::OPTIONAL, + {list}, LogicalType::FixedSizeList(10))); + + auto arrow_element = ::arrow::field("string", UTF8, true); + auto arrow_list = ::arrow::fixed_size_list(arrow_element, 10); + arrow_fields.push_back(::arrow::field("my_fixed_size_list", arrow_list, true)); + + ASSERT_OK(ConvertSchema(parquet_fields)); + ASSERT_NO_FATAL_FAILURE(CheckFlatSchema(arrow_fields)); +} + TEST_F(TestConvertParquetSchema, UnsupportedThings) { std::vector unsupported_nodes; @@ -1747,7 +1765,7 @@ TEST_F(TestConvertArrowSchema, ParquetOtherLists) { arrow_fields.push_back(::arrow::field("my_list", arrow_list, false)); } // // FixedSizeList[10] (list-like non-null, elements nullable) - // required group my_list (LIST) { + // required group my_list (FIXED_SIZE_LIST(10)) { // repeated group list { // optional binary element (UTF8); // } @@ -1756,8 +1774,8 @@ TEST_F(TestConvertArrowSchema, ParquetOtherLists) { auto element = PrimitiveNode::Make("element", Repetition::OPTIONAL, ParquetType::BYTE_ARRAY, ConvertedType::UTF8); auto list = GroupNode::Make("list", Repetition::REPEATED, {element}); - parquet_fields.push_back( - GroupNode::Make("my_list", Repetition::REQUIRED, {list}, ConvertedType::LIST)); + parquet_fields.push_back(GroupNode::Make("my_list", Repetition::REQUIRED, {list}, + LogicalType::FixedSizeList(10))); auto arrow_element = ::arrow::field("string", UTF8, true); auto arrow_list = ::arrow::fixed_size_list(arrow_element, 10); arrow_fields.push_back(::arrow::field("my_list", arrow_list, false)); diff --git a/cpp/src/parquet/arrow/schema.cc b/cpp/src/parquet/arrow/schema.cc index 9c4c462c6b8c..0558a239db9c 100644 --- a/cpp/src/parquet/arrow/schema.cc +++ b/cpp/src/parquet/arrow/schema.cc @@ -106,8 +106,14 @@ Status ListToNode(const std::shared_ptr<::arrow::BaseListType>& type, &element)); NodePtr list = GroupNode::Make("list", Repetition::REPEATED, {element}); - *out = GroupNode::Make(name, RepetitionFromNullable(nullable), {list}, - LogicalType::List(), field_id); + std::shared_ptr logical_type = LogicalType::List(); + if (type->id() == ::arrow::Type::FIXED_SIZE_LIST) { + const auto& fixed_size_list_type = + checked_cast(*type); + logical_type = LogicalType::FixedSizeList(fixed_size_list_type.list_size()); + } + *out = GroupNode::Make(name, RepetitionFromNullable(nullable), {list}, logical_type, + field_id); return Status::OK(); } @@ -626,8 +632,9 @@ Status MapToSchemaField(const GroupNode& group, LevelInfo current_levels, } // MAP-annotated groups cannot be repeated unless served as an element within a // LIST-annotated group with 2-level structure. - if (group.is_repeated() && - (group.parent() == nullptr || !group.parent()->logical_type()->is_list())) { + if (group.is_repeated() && (group.parent() == nullptr || + !(group.parent()->logical_type()->is_list() || + group.parent()->logical_type()->is_fixed_size_list()))) { return Status::Invalid("MAP-annotated groups must not be repeated."); } @@ -709,7 +716,9 @@ Status ResolveList(const GroupNode& group, LevelInfo current_levels, // When it is repeated, the LIST-annotated 2-level structure can only serve as an // element within another LIST-annotated 2-level structure. if (group.is_repeated() && - (group.parent() == nullptr || !group.parent()->logical_type()->is_list())) { + (group.parent() == nullptr || + !(group.parent()->logical_type()->is_list() || + group.parent()->logical_type()->is_fixed_size_list()))) { return Status::Invalid("LIST-annotated groups must not be repeated."); } return {}; @@ -754,7 +763,8 @@ Status ResolveList(const GroupNode& group, LevelInfo current_levels, const auto& repeated_field = list_group.field(0); if (!list_group.logical_type()->is_none() || repeated_field->is_repeated()) { RETURN_NOT_OK(check_two_level_list_repetition(group)); - if (list_group.logical_type()->is_list()) { + if (list_group.logical_type()->is_list() || + list_group.logical_type()->is_fixed_size_list()) { // Special case where the inner type might be a list with two-level encoding // like below: // @@ -828,8 +838,15 @@ Status ListToSchemaField(const GroupNode& group, LevelInfo current_levels, int16_t repeated_ancestor_def_level = current_levels.IncrementRepeated(); RETURN_NOT_OK(ResolveList(group, current_levels, ctx, out)); - ARROW_ASSIGN_OR_RAISE(auto list_type, - MakeArrowList(child_field->field, ctx->properties)); + std::shared_ptr<::arrow::DataType> list_type; + if (group.logical_type()->is_fixed_size_list()) { + const auto& fixed_size_list_type = + checked_cast(*group.logical_type()); + list_type = + ::arrow::fixed_size_list(child_field->field, fixed_size_list_type.num_values()); + } else { + ARROW_ASSIGN_OR_RAISE(list_type, MakeArrowList(child_field->field, ctx->properties)); + } out->field = ::arrow::field(group.name(), std::move(list_type), group.is_optional(), FieldIdMetadata(group.field_id())); out->level_info = current_levels; @@ -842,7 +859,7 @@ Status ListToSchemaField(const GroupNode& group, LevelInfo current_levels, Status GroupToSchemaField(const GroupNode& node, LevelInfo current_levels, SchemaTreeContext* ctx, const SchemaField* parent, SchemaField* out) { - if (node.logical_type()->is_list()) { + if (node.logical_type()->is_list() || node.logical_type()->is_fixed_size_list()) { return ListToSchemaField(node, current_levels, ctx, parent, out); } else if (node.logical_type()->is_map()) { return MapToSchemaField(node, current_levels, ctx, parent, out); diff --git a/cpp/src/parquet/parquet.thrift b/cpp/src/parquet/parquet.thrift index e3cc5adb9648..abb677816169 100644 --- a/cpp/src/parquet/parquet.thrift +++ b/cpp/src/parquet/parquet.thrift @@ -462,6 +462,17 @@ struct GeographyType { 2: optional EdgeInterpolationAlgorithm algorithm; } +/** + * Fixed-size list logical type annotation + * + * Annotates a standard list-shaped column whose non-null parent values contain + * exactly num_values children written contiguously. Readers that do not + * understand this annotation can interpret the column as a regular LIST. + */ +struct FixedSizeListType { + 1: required i32 num_values +} + /** * LogicalType annotations to replace ConvertedType. * @@ -495,6 +506,7 @@ union LogicalType { 16: VariantType VARIANT // no compatible ConvertedType 17: GeometryType GEOMETRY // no compatible ConvertedType 18: GeographyType GEOGRAPHY // no compatible ConvertedType + 19: FixedSizeListType FIXED_SIZE_LIST // no compatible ConvertedType } /** diff --git a/cpp/src/parquet/schema.cc b/cpp/src/parquet/schema.cc index 0cfa49c21c16..7b199de02437 100644 --- a/cpp/src/parquet/schema.cc +++ b/cpp/src/parquet/schema.cc @@ -339,6 +339,13 @@ GroupNode::GroupNode(const std::string& name, Repetition::type repetition, if (logical_type_) { // Check for logical type <=> node type consistency if (logical_type_->is_nested()) { + if (logical_type_->is_fixed_size_list()) { + if (fields_.size() != 1 || !fields_[0]->is_repeated()) { + throw ParquetException( + "FIXED_SIZE_LIST logical type must annotate a group with a single " + "repeated child"); + } + } // For backward compatibility, assign equivalent legacy converted type (if possible) converted_type_ = logical_type_->ToConvertedType(nullptr); } else { @@ -716,8 +723,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_test.cc b/cpp/src/parquet/schema_test.cc index 2950a7df70f8..5ebec70c977b 100644 --- a/cpp/src/parquet/schema_test.cc +++ b/cpp/src/parquet/schema_test.cc @@ -1582,6 +1582,8 @@ TEST(TestLogicalTypeOperation, LogicalTypeRepresentation) { R"({"Type": "Geography", "crs": "srid:1234", "algorithm": "karney"})"}, {LogicalType::Variant(), "Variant(1)", R"({"Type": "Variant", "SpecVersion": 1})"}, {LogicalType::Variant(2), "Variant(2)", R"({"Type": "Variant", "SpecVersion": 2})"}, + {LogicalType::FixedSizeList(10), "FixedSizeList(10)", + R"({"Type": "FixedSizeList", "NumValues": 10})"}, {LogicalType::None(), "None", R"({"Type": "None"})"}, }; @@ -1635,6 +1637,7 @@ TEST(TestLogicalTypeOperation, LogicalTypeSortOrder) { {LogicalType::Geometry(), SortOrder::UNKNOWN}, {LogicalType::Geography(), SortOrder::UNKNOWN}, {LogicalType::Variant(), SortOrder::UNKNOWN}, + {LogicalType::FixedSizeList(10), SortOrder::UNKNOWN}, {LogicalType::None(), SortOrder::UNKNOWN}}; for (const ExpectedSortOrder& c : cases) { @@ -2354,9 +2357,40 @@ TEST(TestLogicalTypeSerialization, Roundtrips) { // Group nodes ... ConfirmGroupNodeRoundtrip("map", LogicalType::Map()); ConfirmGroupNodeRoundtrip("list", LogicalType::List()); + ConfirmGroupNodeRoundtrip("fixed_size_list", LogicalType::FixedSizeList(10)); ConfirmGroupNodeRoundtrip("variant", LogicalType::Variant()); } +TEST(TestLogicalTypeSerialization, FixedSizeListNumValues) { + constexpr int32_t num_values = 10; + auto element = PrimitiveNode::Make("element", Repetition::REQUIRED, Type::INT32); + auto list = GroupNode::Make("list", Repetition::REPEATED, {element}); + NodePtr fixed_size_list_node = + GroupNode::Make("fixed_size_list", Repetition::OPTIONAL, {list}, + LogicalType::FixedSizeList(num_values)); + + auto logical_type = fixed_size_list_node->logical_type(); + ASSERT_TRUE(logical_type->is_fixed_size_list()); + const auto& fixed_size_list_type = + checked_cast(*logical_type); + ASSERT_EQ(fixed_size_list_type.num_values(), num_values); + ASSERT_EQ(fixed_size_list_node->converted_type(), ConvertedType::LIST); + + std::vector elements; + ToParquet(reinterpret_cast(fixed_size_list_node.get()), &elements); + ASSERT_TRUE(elements[0].__isset.logicalType); + ASSERT_TRUE(elements[0].logicalType.__isset.FIXED_SIZE_LIST); + ASSERT_EQ(elements[0].logicalType.FIXED_SIZE_LIST.num_values, num_values); + ASSERT_TRUE(elements[0].__isset.converted_type); + ASSERT_EQ(elements[0].converted_type, format::ConvertedType::LIST); + + auto roundtripped = LogicalType::FromThrift(elements[0].logicalType); + ASSERT_TRUE(roundtripped->is_fixed_size_list()); + const auto& roundtripped_fixed_size_list_type = + checked_cast(*roundtripped); + ASSERT_EQ(roundtripped_fixed_size_list_type.num_values(), num_values); +} + TEST(TestLogicalTypeSerialization, VariantSpecificationVersion) { // Confirm that Variant logical type sets specification_version to expected value in // thrift serialization diff --git a/cpp/src/parquet/types.cc b/cpp/src/parquet/types.cc index fb4eb92a7544..ef6b0099a47a 100644 --- a/cpp/src/parquet/types.cc +++ b/cpp/src/parquet/types.cc @@ -602,6 +602,8 @@ std::shared_ptr LogicalType::FromThrift( } return VariantLogicalType::Make(spec_version); + } else if (type.__isset.FIXED_SIZE_LIST) { + return FixedSizeListLogicalType::Make(type.FIXED_SIZE_LIST.num_values); } else { // Sentinel type for one we do not recognize return UndefinedLogicalType::Make(); @@ -673,6 +675,10 @@ std::shared_ptr LogicalType::Variant(int8_t spec_version) { return VariantLogicalType::Make(spec_version); } +std::shared_ptr LogicalType::FixedSizeList(int32_t num_values) { + return FixedSizeListLogicalType::Make(num_values); +} + std::shared_ptr LogicalType::None() { return NoLogicalType::Make(); } /* @@ -758,6 +764,7 @@ class LogicalType::Impl { class Geometry; class Geography; class Variant; + class FixedSizeList; class No; class Undefined; @@ -839,6 +846,9 @@ bool LogicalType::is_geography() const { bool LogicalType::is_variant() const { return impl_->type() == LogicalType::Type::VARIANT; } +bool LogicalType::is_fixed_size_list() const { + return impl_->type() == LogicalType::Type::FIXED_SIZE_LIST; +} bool LogicalType::is_none() const { return impl_->type() == LogicalType::Type::NONE; } bool LogicalType::is_valid() const { return impl_->type() != LogicalType::Type::UNDEFINED; @@ -847,7 +857,8 @@ bool LogicalType::is_invalid() const { return !is_valid(); } bool LogicalType::is_nested() const { return impl_->type() == LogicalType::Type::LIST || impl_->type() == LogicalType::Type::MAP || - impl_->type() == LogicalType::Type::VARIANT; + impl_->type() == LogicalType::Type::VARIANT || + impl_->type() == LogicalType::Type::FIXED_SIZE_LIST; } bool LogicalType::is_nonnested() const { return !is_nested(); } bool LogicalType::is_serialized() const { return impl_->is_serialized(); } @@ -2016,6 +2027,67 @@ std::shared_ptr VariantLogicalType::Make(const int8_t spec_ve return logical_type; } +class LogicalType::Impl::FixedSizeList final : public LogicalType::Impl::SimpleCompatible, + public LogicalType::Impl::Inapplicable { + public: + friend class FixedSizeListLogicalType; + + std::string ToString() const override; + std::string ToJSON() const override; + format::LogicalType ToThrift() const override; + bool Equals(const LogicalType& other) const override; + + int32_t num_values() const { return num_values_; } + + private: + explicit FixedSizeList(int32_t num_values) + : LogicalType::Impl(LogicalType::Type::FIXED_SIZE_LIST, SortOrder::UNKNOWN), + LogicalType::Impl::SimpleCompatible(ConvertedType::LIST), + LogicalType::Impl::Inapplicable(), + num_values_(num_values) {} + + int32_t num_values_; +}; + +int32_t FixedSizeListLogicalType::num_values() const { + return (dynamic_cast(*impl_)).num_values(); +} + +std::string LogicalType::Impl::FixedSizeList::ToString() const { + std::stringstream type; + type << "FixedSizeList(" << num_values_ << ")"; + return type.str(); +} + +std::string LogicalType::Impl::FixedSizeList::ToJSON() const { + std::stringstream json; + json << R"({"Type": "FixedSizeList", "NumValues": )" << num_values_ << "}"; + return json.str(); +} + +format::LogicalType LogicalType::Impl::FixedSizeList::ToThrift() const { + format::LogicalType type; + format::FixedSizeListType fixed_size_list_type; + fixed_size_list_type.__set_num_values(num_values_); + type.__set_FIXED_SIZE_LIST(fixed_size_list_type); + return type; +} + +bool LogicalType::Impl::FixedSizeList::Equals(const LogicalType& other) const { + return other.type() == LogicalType::Type::FIXED_SIZE_LIST && + dynamic_cast(other).num_values() == num_values_; +} + +std::shared_ptr FixedSizeListLogicalType::Make(int32_t num_values) { + if (num_values < 0) { + throw ParquetException("FixedSizeList logical type requires non-negative num_values"); + } + auto logical_type = + std::shared_ptr(new FixedSizeListLogicalType()); + logical_type->impl_.reset(new LogicalType::Impl::FixedSizeList(num_values)); + return logical_type; +} + class LogicalType::Impl::No final : public LogicalType::Impl::SimpleCompatible, public LogicalType::Impl::UniversalApplicable { public: diff --git a/cpp/src/parquet/types.h b/cpp/src/parquet/types.h index ad4df5119e75..54dd5fd3dc03 100644 --- a/cpp/src/parquet/types.h +++ b/cpp/src/parquet/types.h @@ -162,6 +162,7 @@ class PARQUET_EXPORT LogicalType { GEOMETRY, GEOGRAPHY, VARIANT, + FIXED_SIZE_LIST, NONE // Not a real logical type; should always be last element }; }; @@ -230,6 +231,7 @@ class PARQUET_EXPORT LogicalType { static std::shared_ptr Float16(); static std::shared_ptr Variant( int8_t specVersion = kVariantSpecVersion); + static std::shared_ptr FixedSizeList(int32_t num_values); static std::shared_ptr Geometry(std::string crs = ""); @@ -293,6 +295,7 @@ class PARQUET_EXPORT LogicalType { bool is_geometry() const; bool is_geography() const; bool is_variant() const; + bool is_fixed_size_list() const; bool is_none() const; /// \brief Return true if this logical type is of a known type. bool is_valid() const; @@ -509,6 +512,17 @@ class PARQUET_EXPORT VariantLogicalType : public LogicalType { VariantLogicalType() = default; }; +/// \brief Allowed for group nodes only, annotating a list-shaped column. +class PARQUET_EXPORT FixedSizeListLogicalType : public LogicalType { + public: + static std::shared_ptr Make(int32_t num_values); + + int32_t num_values() const; + + private: + FixedSizeListLogicalType() = default; +}; + /// \brief Allowed for any physical type. class PARQUET_EXPORT NoLogicalType : public LogicalType { public: diff --git a/python/pyarrow/_parquet.pyx b/python/pyarrow/_parquet.pyx index 2358a961ebd9..d8b898718c2e 100644 --- a/python/pyarrow/_parquet.pyx +++ b/python/pyarrow/_parquet.pyx @@ -1464,6 +1464,7 @@ cdef logical_type_name_from_enum(ParquetLogicalTypeId type_): ParquetLogicalType_JSON: 'JSON', ParquetLogicalType_BSON: 'BSON', ParquetLogicalType_UUID: 'UUID', + ParquetLogicalType_FIXED_SIZE_LIST: 'FIXED_SIZE_LIST', ParquetLogicalType_NONE: 'NONE', }.get(type_, 'UNKNOWN') diff --git a/python/pyarrow/includes/libparquet.pxd b/python/pyarrow/includes/libparquet.pxd index df353cc7805f..7590ecf1108d 100644 --- a/python/pyarrow/includes/libparquet.pxd +++ b/python/pyarrow/includes/libparquet.pxd @@ -69,6 +69,7 @@ cdef extern from "parquet/api/schema.h" namespace "parquet" nogil: ParquetLogicalType_UUID" parquet::LogicalType::Type::UUID" ParquetLogicalType_GEOMETRY" parquet::LogicalType::Type::GEOMETRY" ParquetLogicalType_GEOGRAPHY" parquet::LogicalType::Type::GEOGRAPHY" + ParquetLogicalType_FIXED_SIZE_LIST" parquet::LogicalType::Type::FIXED_SIZE_LIST" ParquetLogicalType_NONE" parquet::LogicalType::Type::NONE" enum ParquetTimeUnit" parquet::LogicalType::TimeUnit::unit": From 3f7b1e3b2e1469ab9e6e842ca0bbe193e0a88b1e Mon Sep 17 00:00:00 2001 From: Rok Mihevc Date: Thu, 4 Jun 2026 16:15:35 +0200 Subject: [PATCH 2/6] fast read path --- cpp/src/parquet/arrow/arrow_schema_test.cc | 26 ++++- cpp/src/parquet/arrow/reader.cc | 99 +++++++++++++++++-- cpp/src/parquet/arrow/schema.cc | 4 +- cpp/src/parquet/arrow/schema.h | 3 + cpp/src/parquet/properties.h | 30 +++++- python/pyarrow/_dataset_parquet.pyx | 5 + python/pyarrow/_parquet.pxd | 1 + python/pyarrow/_parquet.pyx | 12 ++- python/pyarrow/includes/libparquet.pxd | 2 + python/pyarrow/parquet/core.py | 8 ++ .../pyarrow/tests/parquet/test_data_types.py | 19 ++++ 11 files changed, 197 insertions(+), 12 deletions(-) diff --git a/cpp/src/parquet/arrow/arrow_schema_test.cc b/cpp/src/parquet/arrow/arrow_schema_test.cc index baed6caf0062..5044cf88c4db 100644 --- a/cpp/src/parquet/arrow/arrow_schema_test.cc +++ b/cpp/src/parquet/arrow/arrow_schema_test.cc @@ -1765,7 +1765,7 @@ TEST_F(TestConvertArrowSchema, ParquetOtherLists) { arrow_fields.push_back(::arrow::field("my_list", arrow_list, false)); } // // FixedSizeList[10] (list-like non-null, elements nullable) - // required group my_list (FIXED_SIZE_LIST(10)) { + // required group my_list (LIST) { // repeated group list { // optional binary element (UTF8); // } @@ -1774,8 +1774,8 @@ TEST_F(TestConvertArrowSchema, ParquetOtherLists) { auto element = PrimitiveNode::Make("element", Repetition::OPTIONAL, ParquetType::BYTE_ARRAY, ConvertedType::UTF8); auto list = GroupNode::Make("list", Repetition::REPEATED, {element}); - parquet_fields.push_back(GroupNode::Make("my_list", Repetition::REQUIRED, {list}, - LogicalType::FixedSizeList(10))); + parquet_fields.push_back( + GroupNode::Make("my_list", Repetition::REQUIRED, {list}, ConvertedType::LIST)); auto arrow_element = ::arrow::field("string", UTF8, true); auto arrow_list = ::arrow::fixed_size_list(arrow_element, 10); arrow_fields.push_back(::arrow::field("my_list", arrow_list, false)); @@ -1786,6 +1786,26 @@ TEST_F(TestConvertArrowSchema, ParquetOtherLists) { ASSERT_NO_FATAL_FAILURE(CheckFlatSchema(parquet_fields)); } +TEST_F(TestConvertArrowSchema, ParquetFixedSizeListLogicalTypeEnabled) { + std::vector parquet_fields; + std::vector> arrow_fields; + + auto element = PrimitiveNode::Make("element", Repetition::OPTIONAL, + ParquetType::BYTE_ARRAY, ConvertedType::UTF8); + auto list = GroupNode::Make("list", Repetition::REPEATED, {element}); + parquet_fields.push_back(GroupNode::Make("my_list", Repetition::REQUIRED, {list}, + LogicalType::FixedSizeList(10))); + auto arrow_element = ::arrow::field("string", UTF8, true); + auto arrow_list = ::arrow::fixed_size_list(arrow_element, 10); + arrow_fields.push_back(::arrow::field("my_list", arrow_list, false)); + + ArrowWriterProperties::Builder builder; + builder.enable_fixed_size_list_logical_type(); + ASSERT_OK(ConvertSchema(arrow_fields, builder.build())); + + ASSERT_NO_FATAL_FAILURE(CheckFlatSchema(parquet_fields)); +} + TEST_F(TestConvertArrowSchema, ParquetNestedComplianceEnabledNullable) { std::vector parquet_fields; std::vector> arrow_fields; diff --git a/cpp/src/parquet/arrow/reader.cc b/cpp/src/parquet/arrow/reader.cc index a60af69aec9f..407fe6541a3a 100644 --- a/cpp/src/parquet/arrow/reader.cc +++ b/cpp/src/parquet/arrow/reader.cc @@ -603,7 +603,7 @@ class ListReader : public ColumnReaderImpl { bool IsOrHasRepeatedChild() const final { return true; } - Status LoadBatch(int64_t number_of_records) final { + Status LoadBatch(int64_t number_of_records) override { return item_reader_->LoadBatch(number_of_records); } @@ -673,7 +673,7 @@ class ListReader : public ColumnReaderImpl { const std::shared_ptr field() override { return field_; } - private: + protected: std::shared_ptr ctx_; std::shared_ptr field_; ::parquet::internal::LevelInfo level_info_; @@ -684,9 +684,92 @@ class PARQUET_NO_EXPORT FixedSizeListReader : public ListReader { public: FixedSizeListReader(std::shared_ptr ctx, std::shared_ptr field, ::parquet::internal::LevelInfo level_info, - std::unique_ptr child_reader) - : ListReader(std::move(ctx), std::move(field), level_info, - std::move(child_reader)) {} + std::unique_ptr child_reader, + bool use_fixed_size_list_fast_path) + : ListReader(std::move(ctx), std::move(field), level_info, std::move(child_reader)), + use_fixed_size_list_fast_path_(use_fixed_size_list_fast_path) {} + + Status BuildArray(int64_t length_upper_bound, + std::shared_ptr* out) override { + DCHECK_EQ(field_->type()->id(), ::arrow::Type::FIXED_SIZE_LIST); + const auto& type = checked_cast(*field_->type()); + const int32_t list_size = type.list_size(); + + // The fast path applies to dense fixed-size lists with the Parquet + // FIXED_SIZE_LIST annotation. Nullable elements and zero-length lists still + // use the regular LIST reconstruction path because they require interpreting + // inner definition levels or empty-list sentinels. + if (!use_fixed_size_list_fast_path_ || list_size == 0 || + type.value_field()->nullable()) { + return ListReader::BuildArray(length_upper_bound, out); + } + + const int16_t* def_levels = nullptr; + int64_t num_levels = 0; + RETURN_NOT_OK(item_reader_->GetDefLevels(&def_levels, &num_levels)); + + std::shared_ptr validity_buffer; + int64_t values_read = 0; + int64_t null_count = 0; + int64_t child_values = 0; + + if (field_->nullable()) { + ARROW_ASSIGN_OR_RAISE(validity_buffer, + AllocateResizableBuffer( + bit_util::BytesForBits(length_upper_bound), ctx_->pool)); + uint8_t* valid_bits = validity_buffer->mutable_data(); + + const int16_t null_list_def_level = level_info_.def_level - 1; + int64_t level_index = 0; + while (level_index < num_levels && values_read < length_upper_bound) { + if (def_levels[level_index] < null_list_def_level) { + bit_util::ClearBit(valid_bits, values_read); + ++null_count; + ++values_read; + ++level_index; + } else if (def_levels[level_index] < level_info_.def_level) { + return Status::Invalid( + "Encountered non-null FIXED_SIZE_LIST value with fewer than ", list_size, + " children"); + } else { + bit_util::SetBit(valid_bits, values_read); + ++values_read; + child_values += list_size; + level_index += list_size; + } + } + + if (level_index != num_levels) { + return Status::Invalid("FIXED_SIZE_LIST column ended in the middle of a list"); + } + RETURN_NOT_OK(validity_buffer->Resize(bit_util::BytesForBits(values_read))); + validity_buffer->ZeroPadding(); + } else { + values_read = std::min(length_upper_bound, num_levels / list_size); + child_values = values_read * list_size; + if (num_levels % list_size != 0) { + return Status::Invalid("FIXED_SIZE_LIST column ended in the middle of a list"); + } + } + + RETURN_NOT_OK(item_reader_->BuildArray(child_values, out)); + ARROW_ASSIGN_OR_RAISE(std::shared_ptr item_chunk, ChunksToSingle(**out)); + if (item_chunk->length != child_values) { + return Status::Invalid("Expected ", child_values, + " child values for FIXED_SIZE_LIST but got ", + item_chunk->length); + } + + std::vector> buffers{null_count > 0 ? validity_buffer + : nullptr}; + auto data = std::make_shared( + field_->type(), /*length=*/values_read, std::move(buffers), + std::vector>{std::move(item_chunk)}, null_count); + std::shared_ptr result = ::arrow::MakeArray(data); + *out = std::make_shared(result); + return Status::OK(); + } + ::arrow::Result> AssembleArray( std::shared_ptr data) final { DCHECK_EQ(data->buffers.size(), 2); @@ -704,6 +787,9 @@ class PARQUET_NO_EXPORT FixedSizeListReader : public ListReader { std::shared_ptr result = ::arrow::MakeArray(data); return std::make_shared(result); } + + private: + bool use_fixed_size_list_fast_path_; }; class PARQUET_NO_EXPORT StructReader : public ColumnReaderImpl { @@ -951,7 +1037,8 @@ Status GetReader(const SchemaField& field, const std::shared_ptr& arrow_f } *out = std::make_unique(ctx, list_field, field.level_info, - std::move(child_reader)); + std::move(child_reader), + field.is_fixed_size_list); } else { return Status::UnknownError("Unknown list type: ", field.field->ToString()); } diff --git a/cpp/src/parquet/arrow/schema.cc b/cpp/src/parquet/arrow/schema.cc index 0558a239db9c..a9c1bf92f845 100644 --- a/cpp/src/parquet/arrow/schema.cc +++ b/cpp/src/parquet/arrow/schema.cc @@ -107,7 +107,8 @@ Status ListToNode(const std::shared_ptr<::arrow::BaseListType>& type, NodePtr list = GroupNode::Make("list", Repetition::REPEATED, {element}); std::shared_ptr logical_type = LogicalType::List(); - if (type->id() == ::arrow::Type::FIXED_SIZE_LIST) { + if (type->id() == ::arrow::Type::FIXED_SIZE_LIST && + arrow_properties.write_fixed_size_list_logical_type()) { const auto& fixed_size_list_type = checked_cast(*type); logical_type = LogicalType::FixedSizeList(fixed_size_list_type.list_size()); @@ -849,6 +850,7 @@ Status ListToSchemaField(const GroupNode& group, LevelInfo current_levels, } out->field = ::arrow::field(group.name(), std::move(list_type), group.is_optional(), FieldIdMetadata(group.field_id())); + out->is_fixed_size_list = group.logical_type()->is_fixed_size_list(); out->level_info = current_levels; // At this point current levels contains the def level for this list, // we need to reset to the prior parent. diff --git a/cpp/src/parquet/arrow/schema.h b/cpp/src/parquet/arrow/schema.h index dd60fde43422..6bbf461db66b 100644 --- a/cpp/src/parquet/arrow/schema.h +++ b/cpp/src/parquet/arrow/schema.h @@ -96,6 +96,9 @@ struct PARQUET_EXPORT SchemaField { parquet::internal::LevelInfo level_info; + // True when the Parquet node was annotated with FIXED_SIZE_LIST. + bool is_fixed_size_list = false; + bool is_leaf() const { return column_index != -1; } }; diff --git a/cpp/src/parquet/properties.h b/cpp/src/parquet/properties.h index 6634bac4f684..fe897b6e28d9 100644 --- a/cpp/src/parquet/properties.h +++ b/cpp/src/parquet/properties.h @@ -1298,6 +1298,7 @@ class PARQUET_EXPORT ArrowWriterProperties { truncated_timestamps_allowed_(false), store_schema_(false), compliant_nested_types_(true), + write_fixed_size_list_logical_type_(false), engine_version_(V2), use_threads_(kArrowDefaultUseThreads), executor_(NULLPTR), @@ -1367,6 +1368,22 @@ class PARQUET_EXPORT ArrowWriterProperties { return this; } + /// \brief EXPERIMENTAL: Write Arrow FixedSizeList as a Parquet + /// FIXED_SIZE_LIST logical type annotation. + /// + /// When disabled, Arrow FixedSizeList is written as a regular Parquet LIST. + /// This is disabled by default while the logical type is experimental. + Builder* enable_fixed_size_list_logical_type() { + write_fixed_size_list_logical_type_ = true; + return this; + } + + /// \brief Disable writing the experimental FIXED_SIZE_LIST logical type. + Builder* disable_fixed_size_list_logical_type() { + write_fixed_size_list_logical_type_ = false; + return this; + } + /// Set the version of the Parquet writer engine. Builder* set_engine_version(EngineVersion version) { engine_version_ = version; @@ -1409,7 +1426,8 @@ class PARQUET_EXPORT ArrowWriterProperties { return std::shared_ptr(new ArrowWriterProperties( write_timestamps_as_int96_, coerce_timestamps_enabled_, coerce_timestamps_unit_, truncated_timestamps_allowed_, store_schema_, compliant_nested_types_, - engine_version_, use_threads_, executor_, write_time_adjusted_to_utc_)); + write_fixed_size_list_logical_type_, engine_version_, use_threads_, executor_, + write_time_adjusted_to_utc_)); } private: @@ -1421,6 +1439,7 @@ class PARQUET_EXPORT ArrowWriterProperties { bool store_schema_; bool compliant_nested_types_; + bool write_fixed_size_list_logical_type_; EngineVersion engine_version_; bool use_threads_; @@ -1447,6 +1466,12 @@ class PARQUET_EXPORT ArrowWriterProperties { /// "element". bool compliant_nested_types() const { return compliant_nested_types_; } + /// \brief Whether to write Arrow FixedSizeList using the experimental Parquet + /// FIXED_SIZE_LIST logical type annotation. + bool write_fixed_size_list_logical_type() const { + return write_fixed_size_list_logical_type_; + } + /// \brief The underlying engine version to use when writing Arrow data. /// /// V2 is currently the latest V1 is considered deprecated but left in @@ -1471,6 +1496,7 @@ class PARQUET_EXPORT ArrowWriterProperties { ::arrow::TimeUnit::type coerce_timestamps_unit, bool truncated_timestamps_allowed, bool store_schema, bool compliant_nested_types, + bool write_fixed_size_list_logical_type, EngineVersion engine_version, bool use_threads, ::arrow::internal::Executor* executor, bool write_time_adjusted_to_utc) @@ -1480,6 +1506,7 @@ class PARQUET_EXPORT ArrowWriterProperties { truncated_timestamps_allowed_(truncated_timestamps_allowed), store_schema_(store_schema), compliant_nested_types_(compliant_nested_types), + write_fixed_size_list_logical_type_(write_fixed_size_list_logical_type), engine_version_(engine_version), use_threads_(use_threads), executor_(executor), @@ -1491,6 +1518,7 @@ class PARQUET_EXPORT ArrowWriterProperties { const bool truncated_timestamps_allowed_; const bool store_schema_; const bool compliant_nested_types_; + const bool write_fixed_size_list_logical_type_; const EngineVersion engine_version_; const bool use_threads_; ::arrow::internal::Executor* executor_; diff --git a/python/pyarrow/_dataset_parquet.pyx b/python/pyarrow/_dataset_parquet.pyx index 534f7790923a..f1eb63fa2e3e 100644 --- a/python/pyarrow/_dataset_parquet.pyx +++ b/python/pyarrow/_dataset_parquet.pyx @@ -620,6 +620,7 @@ cdef class ParquetFileWriteOptions(FileWriteOptions): "coerce_timestamps", "allow_truncated_timestamps", "use_compliant_nested_type", + "write_fixed_size_list_logical_type", } setters = set() @@ -676,6 +677,9 @@ cdef class ParquetFileWriteOptions(FileWriteOptions): writer_engine_version="V2", use_compliant_nested_type=( self._properties["use_compliant_nested_type"] + ), + write_fixed_size_list_logical_type=( + self._properties["write_fixed_size_list_logical_type"] ) ) @@ -705,6 +709,7 @@ cdef class ParquetFileWriteOptions(FileWriteOptions): coerce_timestamps=None, allow_truncated_timestamps=False, use_compliant_nested_type=True, + write_fixed_size_list_logical_type=False, encryption_properties=None, write_batch_size=None, dictionary_pagesize_limit=None, diff --git a/python/pyarrow/_parquet.pxd b/python/pyarrow/_parquet.pxd index 36fc2ccf2f33..50ab1d854412 100644 --- a/python/pyarrow/_parquet.pxd +++ b/python/pyarrow/_parquet.pxd @@ -69,6 +69,7 @@ cdef shared_ptr[ArrowWriterProperties] _create_arrow_writer_properties( use_compliant_nested_type=*, store_schema=*, write_time_adjusted_to_utc=*, + write_fixed_size_list_logical_type=*, ) except * diff --git a/python/pyarrow/_parquet.pyx b/python/pyarrow/_parquet.pyx index d8b898718c2e..d6172ee7d14a 100644 --- a/python/pyarrow/_parquet.pyx +++ b/python/pyarrow/_parquet.pyx @@ -2282,7 +2282,8 @@ cdef shared_ptr[ArrowWriterProperties] _create_arrow_writer_properties( writer_engine_version=None, use_compliant_nested_type=True, store_schema=True, - write_time_adjusted_to_utc=False) except *: + write_time_adjusted_to_utc=False, + write_fixed_size_list_logical_type=False) except *: """Arrow writer properties""" cdef: shared_ptr[ArrowWriterProperties] arrow_properties @@ -2323,6 +2324,13 @@ cdef shared_ptr[ArrowWriterProperties] _create_arrow_writer_properties( else: arrow_props.disable_compliant_nested_types() + # write_fixed_size_list_logical_type + + if write_fixed_size_list_logical_type: + arrow_props.enable_fixed_size_list_logical_type() + else: + arrow_props.disable_fixed_size_list_logical_type() + # writer_engine_version if writer_engine_version == "V1": @@ -2397,6 +2405,7 @@ cdef class ParquetWriter(_Weakrefable): store_decimal_as_integer=False, use_content_defined_chunking=False, write_time_adjusted_to_utc=False, + write_fixed_size_list_logical_type=False, bloom_filter_options=None): cdef: shared_ptr[WriterProperties] properties @@ -2444,6 +2453,7 @@ cdef class ParquetWriter(_Weakrefable): use_compliant_nested_type=use_compliant_nested_type, store_schema=store_schema, write_time_adjusted_to_utc=write_time_adjusted_to_utc, + write_fixed_size_list_logical_type=write_fixed_size_list_logical_type, ) pool = maybe_unbox_memory_pool(memory_pool) diff --git a/python/pyarrow/includes/libparquet.pxd b/python/pyarrow/includes/libparquet.pxd index 7590ecf1108d..d64c4be7583f 100644 --- a/python/pyarrow/includes/libparquet.pxd +++ b/python/pyarrow/includes/libparquet.pxd @@ -530,6 +530,8 @@ cdef extern from "parquet/api/writer.h" namespace "parquet" nogil: Builder* store_schema() Builder* enable_compliant_nested_types() Builder* disable_compliant_nested_types() + Builder* enable_fixed_size_list_logical_type() + Builder* disable_fixed_size_list_logical_type() Builder* set_engine_version(ArrowWriterEngineVersion version) Builder* set_time_adjusted_to_utc(c_bool adjusted) shared_ptr[ArrowWriterProperties] build() diff --git a/python/pyarrow/parquet/core.py b/python/pyarrow/parquet/core.py index ff880fdcf52c..319f4dc23f98 100644 --- a/python/pyarrow/parquet/core.py +++ b/python/pyarrow/parquet/core.py @@ -884,6 +884,10 @@ def _sanitize_table(table, new_schema, flavor): it will restore the timezone (Parquet only stores the UTC values without timezone), or columns with duration type will be restored from the int64 Parquet column. +write_fixed_size_list_logical_type : bool, default False + EXPERIMENTAL: If True, write Arrow FixedSizeList columns as Parquet + FIXED_SIZE_LIST logical types. If False, write them as regular Parquet + LIST columns for maximum compatibility. write_page_index : bool, default False Whether to write a page index in general for all columns. Writing statistics to the page index disables the old method of writing @@ -1077,6 +1081,7 @@ def __init__(self, where, schema, filesystem=None, write_batch_size=None, dictionary_pagesize_limit=None, store_schema=True, + write_fixed_size_list_logical_type=False, write_page_index=False, write_page_checksum=False, sorting_columns=None, @@ -1132,6 +1137,7 @@ def __init__(self, where, schema, filesystem=None, write_batch_size=write_batch_size, dictionary_pagesize_limit=dictionary_pagesize_limit, store_schema=store_schema, + write_fixed_size_list_logical_type=write_fixed_size_list_logical_type, write_page_index=write_page_index, write_page_checksum=write_page_checksum, sorting_columns=sorting_columns, @@ -2010,6 +2016,7 @@ def write_table(table, where, row_group_size=None, version='2.6', write_batch_size=None, dictionary_pagesize_limit=None, store_schema=True, + write_fixed_size_list_logical_type=False, write_page_index=False, write_page_checksum=False, sorting_columns=None, @@ -2044,6 +2051,7 @@ def write_table(table, where, row_group_size=None, version='2.6', write_batch_size=write_batch_size, dictionary_pagesize_limit=dictionary_pagesize_limit, store_schema=store_schema, + write_fixed_size_list_logical_type=write_fixed_size_list_logical_type, write_page_index=write_page_index, write_page_checksum=write_page_checksum, sorting_columns=sorting_columns, diff --git a/python/pyarrow/tests/parquet/test_data_types.py b/python/pyarrow/tests/parquet/test_data_types.py index c546bc1532ac..1a9ac59fadf2 100644 --- a/python/pyarrow/tests/parquet/test_data_types.py +++ b/python/pyarrow/tests/parquet/test_data_types.py @@ -614,3 +614,22 @@ def test_undefined_logical_type(parquet_test_datadir): b"unknown string 2", b"unknown string 3" ] + + +def test_fixed_size_list_logical_type_write_flag(tempdir): + values = pa.array([1, 2, 3, 4, 5, 6], type=pa.int32()) + list_type = pa.list_(pa.field("element", pa.int32(), nullable=False), 3) + table = pa.table({"fsl": pa.FixedSizeListArray.from_arrays(values, type=list_type)}) + + legacy_path = tempdir / "fixed_size_list_legacy.parquet" + pq.write_table(table, legacy_path, store_schema=False) + legacy_read = pq.read_table(legacy_path) + assert pa.types.is_list(legacy_read.schema.field("fsl").type) + + annotated_path = tempdir / "fixed_size_list_annotated.parquet" + pq.write_table(table, annotated_path, store_schema=False, + write_fixed_size_list_logical_type=True) + annotated_read = pq.read_table(annotated_path) + assert annotated_read.schema.field("fsl").type == list_type + assert annotated_read.column("fsl").combine_chunks().equals( + table.column("fsl").combine_chunks()) From 4cee287fdc26e66bb471d06aff407174dd4d0c77 Mon Sep 17 00:00:00 2001 From: Rok Mihevc Date: Mon, 8 Jun 2026 20:48:09 +0200 Subject: [PATCH 3/6] GH-50000: [Parquet] Speed up dense fixed size list reads --- cpp/src/parquet/arrow/reader.cc | 105 ++++++++++++++++++++++++++++--- cpp/src/parquet/column_reader.cc | 48 ++++++++++++++ cpp/src/parquet/column_reader.h | 16 +++++ 3 files changed, 161 insertions(+), 8 deletions(-) diff --git a/cpp/src/parquet/arrow/reader.cc b/cpp/src/parquet/arrow/reader.cc index 407fe6541a3a..7fba1df3bda4 100644 --- a/cpp/src/parquet/arrow/reader.cc +++ b/cpp/src/parquet/arrow/reader.cc @@ -19,6 +19,7 @@ #include #include +#include #include #include #include @@ -123,6 +124,16 @@ class ColumnReaderImpl : public ColumnReader { virtual ::arrow::Status LoadBatch(int64_t num_records) = 0; + virtual ::arrow::Status LoadBatchWithValueHint(int64_t num_records, + int64_t values_per_record) { + return LoadBatch(num_records); + } + + virtual ::arrow::Status LoadDenseFixedSizeListBatch(int64_t num_records, + int64_t list_size) { + return LoadBatchWithValueHint(num_records, list_size); + } + virtual ::arrow::Status BuildArray(int64_t length_upper_bound, std::shared_ptr<::arrow::ChunkedArray>* out) = 0; virtual bool IsOrHasRepeatedChild() const = 0; @@ -488,11 +499,28 @@ class LeafReader : public ColumnReaderImpl { bool IsOrHasRepeatedChild() const final { return false; } Status LoadBatch(int64_t records_to_read) final { + return LoadBatchWithValueHint(records_to_read, /*values_per_record=*/1); + } + + Status LoadBatchWithValueHint(int64_t records_to_read, + int64_t values_per_record) final { BEGIN_PARQUET_CATCH_EXCEPTIONS out_ = nullptr; record_reader_->Reset(); - // Pre-allocation gives much better performance for flat columns - record_reader_->Reserve(records_to_read); + record_reader_->SetReadBatchSizeMultiplier(values_per_record); + + int64_t values_to_reserve = records_to_read; + if (values_per_record > 1) { + const int64_t max_safe_records = + std::numeric_limits::max() / values_per_record; + if (records_to_read <= max_safe_records) { + values_to_reserve = records_to_read * values_per_record; + } + } + // Pre-allocation gives much better performance for flat columns and dense + // fixed-size repeated columns. + record_reader_->Reserve(values_to_reserve); + const bool should_load_statistics = ctx_->reader_properties->should_load_statistics(); int64_t num_target_row_groups = 0; while (records_to_read > 0) { @@ -522,6 +550,37 @@ class LeafReader : public ColumnReaderImpl { END_PARQUET_CATCH_EXCEPTIONS } + Status LoadDenseFixedSizeListBatch(int64_t records_to_read, + int64_t list_size) final { + BEGIN_PARQUET_CATCH_EXCEPTIONS + out_ = nullptr; + record_reader_->Reset(); + const bool should_load_statistics = ctx_->reader_properties->should_load_statistics(); + int64_t num_target_row_groups = 0; + while (records_to_read > 0) { + if (!record_reader_->HasMoreData()) { + break; + } + int64_t records_read = + record_reader_->ReadDenseFixedSizeListRecords(records_to_read, list_size); + records_to_read -= records_read; + if (records_read == 0) { + NextRowGroup(); + } else { + num_target_row_groups++; + if (should_load_statistics) { + break; + } + } + } + RETURN_NOT_OK(TransferColumnData( + record_reader_.get(), + num_target_row_groups == 1 ? input_->column_chunk_metadata() : nullptr, field_, + descr_, ctx_.get(), &out_)); + return Status::OK(); + END_PARQUET_CATCH_EXCEPTIONS + } + ::arrow::Status BuildArray(int64_t length_upper_bound, std::shared_ptr<::arrow::ChunkedArray>* out) final { *out = out_; @@ -689,6 +748,22 @@ class PARQUET_NO_EXPORT FixedSizeListReader : public ListReader { : ListReader(std::move(ctx), std::move(field), level_info, std::move(child_reader)), use_fixed_size_list_fast_path_(use_fixed_size_list_fast_path) {} + Status LoadBatch(int64_t number_of_records) override { + DCHECK_EQ(field_->type()->id(), ::arrow::Type::FIXED_SIZE_LIST); + const auto& type = checked_cast(*field_->type()); + const int32_t list_size = type.list_size(); + + if (!use_fixed_size_list_fast_path_ || list_size <= 0 || + type.value_field()->nullable()) { + return ListReader::LoadBatch(number_of_records); + } + + if (!field_->nullable()) { + return item_reader_->LoadDenseFixedSizeListBatch(number_of_records, list_size); + } + return item_reader_->LoadBatchWithValueHint(number_of_records, list_size); + } + Status BuildArray(int64_t length_upper_bound, std::shared_ptr* out) override { DCHECK_EQ(field_->type()->id(), ::arrow::Type::FIXED_SIZE_LIST); @@ -704,6 +779,26 @@ class PARQUET_NO_EXPORT FixedSizeListReader : public ListReader { return ListReader::BuildArray(length_upper_bound, out); } + if (!field_->nullable()) { + RETURN_NOT_OK(item_reader_->BuildArray(length_upper_bound * list_size, out)); + ARROW_ASSIGN_OR_RAISE(std::shared_ptr item_chunk, ChunksToSingle(**out)); + if (item_chunk->length % list_size != 0) { + return Status::Invalid("FIXED_SIZE_LIST column ended in the middle of a list"); + } + const int64_t values_read = item_chunk->length / list_size; + if (values_read > length_upper_bound) { + return Status::Invalid("Read more FIXED_SIZE_LIST values than requested"); + } + std::vector> buffers{nullptr}; + auto data = std::make_shared( + field_->type(), /*length=*/values_read, std::move(buffers), + std::vector>{std::move(item_chunk)}, + /*null_count=*/0); + std::shared_ptr result = ::arrow::MakeArray(data); + *out = std::make_shared(result); + return Status::OK(); + } + const int16_t* def_levels = nullptr; int64_t num_levels = 0; RETURN_NOT_OK(item_reader_->GetDefLevels(&def_levels, &num_levels)); @@ -744,12 +839,6 @@ class PARQUET_NO_EXPORT FixedSizeListReader : public ListReader { } RETURN_NOT_OK(validity_buffer->Resize(bit_util::BytesForBits(values_read))); validity_buffer->ZeroPadding(); - } else { - values_read = std::min(length_upper_bound, num_levels / list_size); - child_values = values_read * list_size; - if (num_levels % list_size != 0) { - return Status::Invalid("FIXED_SIZE_LIST column ended in the middle of a list"); - } } RETURN_NOT_OK(item_reader_->BuildArray(child_values, out)); diff --git a/cpp/src/parquet/column_reader.cc b/cpp/src/parquet/column_reader.cc index 79b837f755c3..7ce9512cd9e0 100644 --- a/cpp/src/parquet/column_reader.cc +++ b/cpp/src/parquet/column_reader.cc @@ -22,6 +22,7 @@ #include #include #include +#include #include #include #include @@ -1281,6 +1282,15 @@ class TypedRecordReader : public TypedColumnReaderImpl, } int64_t level_batch_size = std::max(kMinLevelBatchSize, num_records); + if (this->max_rep_level_ > 0 && read_batch_size_multiplier_ > 1) { + const int64_t max_safe_records = + std::numeric_limits::max() / read_batch_size_multiplier_; + if (num_records <= max_safe_records) { + level_batch_size = + std::max(level_batch_size, + num_records * read_batch_size_multiplier_); + } + } // If we are in the middle of a record, we continue until reaching the // desired number of records or the end of the current record if we've found @@ -1646,6 +1656,44 @@ class TypedRecordReader : public TypedColumnReaderImpl, ReserveValues(capacity); } + void SetReadBatchSizeMultiplier(int64_t multiplier) override { + read_batch_size_multiplier_ = std::max(1, multiplier); + } + + int64_t ReadDenseFixedSizeListRecords(int64_t num_records, + int64_t list_size) override { + if (num_records == 0) return 0; + if (list_size <= 0) { + throw ParquetException("Invalid fixed-size-list size"); + } + const int64_t max_safe_records = std::numeric_limits::max() / list_size; + if (num_records > max_safe_records) { + throw ParquetException("Total size of fixed-size-list values too large"); + } + + int64_t values_remaining = num_records * list_size; + int64_t values_read_total = 0; + ReserveValues(values_remaining); + + while (values_remaining > 0) { + if (!this->HasNextInternal()) { + break; + } + const int64_t batch_size = + std::min(values_remaining, this->available_values_current_page()); + if (batch_size == 0) { + break; + } + ReadValuesDense(batch_size); + values_written_ += batch_size; + this->ConsumeBufferedValues(batch_size); + values_remaining -= batch_size; + values_read_total += batch_size; + } + + return values_read_total / list_size; + } + int64_t UpdateCapacity(int64_t capacity, int64_t size, int64_t extra_size) { if (extra_size < 0) { throw ParquetException("Negative size (corrupt file?)"); diff --git a/cpp/src/parquet/column_reader.h b/cpp/src/parquet/column_reader.h index ac4469b1904f..17b3058d7653 100644 --- a/cpp/src/parquet/column_reader.h +++ b/cpp/src/parquet/column_reader.h @@ -293,6 +293,20 @@ class PARQUET_EXPORT RecordReader { /// \brief Pre-allocate space for data. Results in better flat read performance virtual void Reserve(int64_t num_values) = 0; + /// \brief Set a multiplier used to size repeated-field level batches. + /// + /// Dense fixed-size repeated fields may contain many physical values per + /// logical record. Hinting this ratio avoids repeatedly decoding tiny level + /// batches when reading such columns. + virtual void SetReadBatchSizeMultiplier(int64_t multiplier) = 0; + + /// \brief Read dense fixed-size-list records without materializing levels. + /// + /// This is only valid when the caller has established that every logical + /// record contains exactly list_size non-null physical values. + virtual int64_t ReadDenseFixedSizeListRecords(int64_t num_records, + int64_t list_size) = 0; + /// \brief Clear consumed values and repetition/definition levels as the /// result of calling ReadRecords /// For FLBA and ByteArray types, call GetBuilderChunks() to reset them. @@ -430,6 +444,8 @@ class PARQUET_EXPORT RecordReader { // If read_dense_for_nullable_ is true, the BinaryRecordReader/DictionaryRecordReader // might still populate the validity bitmap buffer. bool read_dense_for_nullable_ = false; + + int64_t read_batch_size_multiplier_ = 1; }; class BinaryRecordReader : virtual public RecordReader { From bcf56aeca1142038ccedb7cd9f2bac9c7496954d Mon Sep 17 00:00:00 2001 From: Rok Mihevc Date: Tue, 9 Jun 2026 00:14:51 +0200 Subject: [PATCH 4/6] GH-50000: [Parquet] Fix fixed size list schema test --- cpp/src/parquet/arrow/arrow_schema_test.cc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cpp/src/parquet/arrow/arrow_schema_test.cc b/cpp/src/parquet/arrow/arrow_schema_test.cc index 5044cf88c4db..e0dcf8f3068c 100644 --- a/cpp/src/parquet/arrow/arrow_schema_test.cc +++ b/cpp/src/parquet/arrow/arrow_schema_test.cc @@ -763,7 +763,7 @@ TEST_F(TestConvertParquetSchema, ParquetFixedSizeList) { arrow_fields.push_back(::arrow::field("my_fixed_size_list", arrow_list, true)); ASSERT_OK(ConvertSchema(parquet_fields)); - ASSERT_NO_FATAL_FAILURE(CheckFlatSchema(arrow_fields)); + ASSERT_NO_FATAL_FAILURE(CheckFlatSchema(::arrow::schema(arrow_fields))); } TEST_F(TestConvertParquetSchema, UnsupportedThings) { From 12a30ff6f384f0219a391c135dbcac68fd67f197 Mon Sep 17 00:00:00 2001 From: Rok Mihevc Date: Tue, 9 Jun 2026 15:54:38 +0200 Subject: [PATCH 5/6] GH-50000: [Parquet] Skip repetition levels for nullable dense fixed size lists --- cpp/src/parquet/arrow/reader.cc | 18 ++++++++- cpp/src/parquet/column_reader.cc | 68 +++++++++++++++++++++++++++++++- cpp/src/parquet/column_reader.h | 16 ++++++++ 3 files changed, 99 insertions(+), 3 deletions(-) diff --git a/cpp/src/parquet/arrow/reader.cc b/cpp/src/parquet/arrow/reader.cc index 7fba1df3bda4..68c362e35a05 100644 --- a/cpp/src/parquet/arrow/reader.cc +++ b/cpp/src/parquet/arrow/reader.cc @@ -134,6 +134,14 @@ class ColumnReaderImpl : public ColumnReader { return LoadBatchWithValueHint(num_records, list_size); } + // Load a nullable dense FIXED_SIZE_LIST. Definition levels are still needed to + // locate null rows, but repetition levels are redundant for top-level fixed + // multiplicity, so leaf readers may skip decoding them. + virtual ::arrow::Status LoadNullableFixedSizeListBatch(int64_t num_records, + int64_t list_size) { + return LoadBatchWithValueHint(num_records, list_size); + } + virtual ::arrow::Status BuildArray(int64_t length_upper_bound, std::shared_ptr<::arrow::ChunkedArray>* out) = 0; virtual bool IsOrHasRepeatedChild() const = 0; @@ -550,6 +558,14 @@ class LeafReader : public ColumnReaderImpl { END_PARQUET_CATCH_EXCEPTIONS } + Status LoadNullableFixedSizeListBatch(int64_t records_to_read, + int64_t list_size) final { + record_reader_->SetFixedSizeListRepSkipLength(list_size); + Status status = LoadBatchWithValueHint(records_to_read, list_size); + record_reader_->SetFixedSizeListRepSkipLength(0); + return status; + } + Status LoadDenseFixedSizeListBatch(int64_t records_to_read, int64_t list_size) final { BEGIN_PARQUET_CATCH_EXCEPTIONS @@ -761,7 +777,7 @@ class PARQUET_NO_EXPORT FixedSizeListReader : public ListReader { if (!field_->nullable()) { return item_reader_->LoadDenseFixedSizeListBatch(number_of_records, list_size); } - return item_reader_->LoadBatchWithValueHint(number_of_records, list_size); + return item_reader_->LoadNullableFixedSizeListBatch(number_of_records, list_size); } Status BuildArray(int64_t length_upper_bound, diff --git a/cpp/src/parquet/column_reader.cc b/cpp/src/parquet/column_reader.cc index 7ce9512cd9e0..51e0a6bbbcf6 100644 --- a/cpp/src/parquet/column_reader.cc +++ b/cpp/src/parquet/column_reader.cc @@ -1328,7 +1328,7 @@ class TypedRecordReader : public TypedColumnReaderImpl, batch_size)) { throw ParquetException(kErrorRepDefLevelNotMatchesNumValues); } - if (this->max_rep_level_ > 0) { + if (this->max_rep_level_ > 0 && fixed_size_list_rep_skip_length_ <= 0) { int64_t rep_levels_read = this->ReadRepetitionLevels(batch_size, rep_levels); if (ARROW_PREDICT_FALSE(rep_levels_read != batch_size)) { throw ParquetException(kErrorRepDefLevelNotMatchesNumValues); @@ -1660,6 +1660,11 @@ class TypedRecordReader : public TypedColumnReaderImpl, read_batch_size_multiplier_ = std::max(1, multiplier); } + void SetFixedSizeListRepSkipLength(int64_t list_size) override { + fixed_size_list_rep_skip_length_ = std::max(0, list_size); + fixed_size_list_levels_remaining_ = 0; + } + int64_t ReadDenseFixedSizeListRecords(int64_t num_records, int64_t list_size) override { if (num_records == 0) return 0; @@ -1771,6 +1776,7 @@ class TypedRecordReader : public TypedColumnReaderImpl, void SetPageReader(std::unique_ptr reader) override { at_record_start_ = true; + fixed_size_list_levels_remaining_ = 0; this->pager_ = std::move(reader); ResetDecoders(); } @@ -1810,7 +1816,10 @@ class TypedRecordReader : public TypedColumnReaderImpl, // nullable. Even if they are required, we may have to read ahead and // delimit the records to get the right number of values and they will // have associated levels. - int64_t records_read = DelimitRecords(num_records, values_to_read); + int64_t records_read = + fixed_size_list_rep_skip_length_ > 0 + ? DelimitFixedSizeListRecords(num_records, values_to_read) + : DelimitRecords(num_records, values_to_read); if (!nullable_values() || read_dense_for_nullable_) { ReadValuesDense(*values_to_read); // null_count is always 0 for required. @@ -1821,6 +1830,61 @@ class TypedRecordReader : public TypedColumnReaderImpl, return records_read; } + // Delimit records of an annotated dense FIXED_SIZE_LIST column from + // definition levels + fixed multiplicity, without consulting repetition + // levels. For valid annotated data this yields the same records_read, + // values_seen, and levels_position_ as DelimitRecords: + // - a present parent row has exactly `list_size` max-def-level entries; + // - a null parent row is represented by a single sub-max-def-level entry. + // fixed_size_list_levels_remaining_ carries an unfinished present record + // across level batches/pages, mirroring at_record_start_ in the + // repetition-level path. + int64_t DelimitFixedSizeListRecords(int64_t num_records, int64_t* values_seen) { + const int16_t* const def_levels = this->def_levels(); + const int16_t max_def_level = this->max_def_level_; + const int64_t list_size = fixed_size_list_rep_skip_length_; + ARROW_DCHECK_GT(list_size, 0); + int64_t records_read = 0; + int64_t values = 0; + + if (!at_record_start_) { + const int64_t take = std::min(fixed_size_list_levels_remaining_, + levels_written_ - levels_position_); + levels_position_ += take; + values += take; + fixed_size_list_levels_remaining_ -= take; + if (fixed_size_list_levels_remaining_ > 0) { + *values_seen = values; + return records_read; + } + at_record_start_ = true; + ++records_read; + } + + while (levels_position_ < levels_written_ && records_read < num_records) { + if (def_levels[levels_position_] == max_def_level) { + const int64_t available = levels_written_ - levels_position_; + if (available >= list_size) { + levels_position_ += list_size; + values += list_size; + ++records_read; + } else { + levels_position_ += available; + values += available; + fixed_size_list_levels_remaining_ = list_size - available; + at_record_start_ = false; + break; + } + } else { + ++levels_position_; + ++records_read; + } + } + + *values_seen = values; + return records_read; + } + // Reads optional records and returns number of records read. Fills in // values_to_read and null_count. int64_t ReadOptionalRecords(int64_t num_records, int64_t* values_to_read, diff --git a/cpp/src/parquet/column_reader.h b/cpp/src/parquet/column_reader.h index 17b3058d7653..33fe5f408a24 100644 --- a/cpp/src/parquet/column_reader.h +++ b/cpp/src/parquet/column_reader.h @@ -307,6 +307,15 @@ class PARQUET_EXPORT RecordReader { virtual int64_t ReadDenseFixedSizeListRecords(int64_t num_records, int64_t list_size) = 0; + /// \brief Enable/disable the nullable FIXED_SIZE_LIST repetition-level-skip + /// path for subsequent ReadRecords calls. + /// + /// When enabled with list_size > 0, repeated records are delimited from + /// definition levels and the fixed multiplicity instead of decoded repetition + /// levels. This is only valid for dense fixed-size lists where the caller does + /// not need to read child repetition levels back. + virtual void SetFixedSizeListRepSkipLength(int64_t list_size) = 0; + /// \brief Clear consumed values and repetition/definition levels as the /// result of calling ReadRecords /// For FLBA and ByteArray types, call GetBuilderChunks() to reset them. @@ -446,6 +455,13 @@ class PARQUET_EXPORT RecordReader { bool read_dense_for_nullable_ = false; int64_t read_batch_size_multiplier_ = 1; + + // When > 0, ReadRecords delimits FIXED_SIZE_LIST records from definition + // levels + this fixed multiplicity and skips repetition-level decoding. + int64_t fixed_size_list_rep_skip_length_ = 0; + // Remaining child entries of a present FIXED_SIZE_LIST record split across + // level batches/pages. + int64_t fixed_size_list_levels_remaining_ = 0; }; class BinaryRecordReader : virtual public RecordReader { From 4b5ce71b84cce58b50455c8e55c504e93928c64b Mon Sep 17 00:00:00 2001 From: Rok Mihevc Date: Tue, 9 Jun 2026 17:36:13 +0200 Subject: [PATCH 6/6] Fix dataset writer properties --- python/pyarrow/_dataset_parquet.pyx | 2 ++ 1 file changed, 2 insertions(+) diff --git a/python/pyarrow/_dataset_parquet.pyx b/python/pyarrow/_dataset_parquet.pyx index f1eb63fa2e3e..a25cfc210774 100644 --- a/python/pyarrow/_dataset_parquet.pyx +++ b/python/pyarrow/_dataset_parquet.pyx @@ -678,6 +678,8 @@ cdef class ParquetFileWriteOptions(FileWriteOptions): use_compliant_nested_type=( self._properties["use_compliant_nested_type"] ), + store_schema=True, + write_time_adjusted_to_utc=False, write_fixed_size_list_logical_type=( self._properties["write_fixed_size_list_logical_type"] )