Copilot commented on code in PR #50252:
URL: https://github.com/apache/arrow/pull/50252#discussion_r4409624364
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -594,6 +596,26 @@ class ExtensionReader : public ColumnReaderImpl {
std::unique_ptr<ColumnReaderImpl> storage_reader_;
};
+class VariantReader : public ExtensionReader {
+ public:
+ VariantReader(std::shared_ptr<ReaderContext> ctx, std::shared_ptr<Field>
field,
+ std::unique_ptr<ColumnReaderImpl> storage_reader)
+ : ExtensionReader(std::move(field), std::move(storage_reader)),
+ ctx_(std::move(ctx)) {}
+
+ Status BuildArray(int64_t length_upper_bound,
+ std::shared_ptr<ChunkedArray>* out) override {
+ RETURN_NOT_OK(ExtensionReader::BuildArray(length_upper_bound, out));
+ if (ctx_->reader_properties->variant_validation_enabled()) {
+ PARQUET_CATCH_NOT_OK(variant::ValidateVariants<false>(**out,
ctx_->pool));
Review Comment:
This validates a nested Variant without its ancestors' visibility masks. For
a required Variant child inside a nullable struct, a null parent leaves the
child without a validity bitmap, so validation tries to parse hidden metadata
and rejects an otherwise valid file. Validate after the enclosing column is
assembled, or propagate ancestor visibility into this call.
##########
cpp/src/parquet/arrow/schema.cc:
##########
@@ -131,21 +369,40 @@ Status MapToNode(const std::shared_ptr<::arrow::MapType>&
type, const std::strin
return Status::OK();
}
-Status VariantToNode(
- const std::shared_ptr<::arrow::extension::VariantExtensionType>& type,
- const std::string& name, bool nullable, int field_id,
- const WriterProperties& properties, const ArrowWriterProperties&
arrow_properties,
- NodePtr* out) {
- NodePtr metadata_node;
- RETURN_NOT_OK(FieldToNode("metadata", type->metadata(), properties,
arrow_properties,
- &metadata_node));
+Status VariantToNode(const std::shared_ptr<VariantExtensionType>& type,
+ const std::string& name, bool nullable, int field_id,
+ const WriterProperties& properties,
+ const ArrowWriterProperties& arrow_properties, NodePtr*
out) {
+ // `VariantExtensionType` already validates the general storage shape.
Writing is
+ // stricter than reading: we keep accepting typed-only Variant schemas on
read,
+ // but must not write them.
+ if (type->value() == nullptr) {
+ return Status::Invalid("Invalid Variant write schema: missing top-level
value field");
+ }
- NodePtr value_node;
- RETURN_NOT_OK(
- FieldToNode("value", type->value(), properties, arrow_properties,
&value_node));
+ std::vector<NodePtr> children;
+ children.reserve(type->storage_type()->num_fields());
+
+ auto AddChild = [&](const std::shared_ptr<::arrow::Field>& field) {
+ if (field == nullptr) {
+ return Status::OK();
+ }
+ NodePtr child;
+ if (field->name() == "typed_value") {
+ RETURN_NOT_OK(VariantTypedValueToNode(field, properties,
arrow_properties, &child));
+ } else {
+ RETURN_NOT_OK(
+ FieldToNode(field->name(), field, properties, arrow_properties,
&child));
+ }
+ children.emplace_back(std::move(child));
+ return Status::OK();
+ };
+
+ RETURN_NOT_OK(AddChild(type->metadata()));
+ RETURN_NOT_OK(AddChild(type->value()));
+ RETURN_NOT_OK(AddChild(type->typed_value()));
Review Comment:
These calls force Parquet children into metadata/value/typed_value order,
but the writer traverses Arrow storage fields positionally. A supported array
with `typed_value` first therefore sends its typed leaf to the metadata column,
causing a write failure or incorrect output. Preserve storage-field order when
constructing these nodes.
##########
cpp/src/parquet/arrow/schema.cc:
##########
@@ -112,6 +114,242 @@ Status ListToNode(const
std::shared_ptr<::arrow::BaseListType>& type,
return Status::OK();
}
+static constexpr char FIELD_ID_KEY[] = "PARQUET:field_id";
+
+int FieldIdFromMetadata(
+ const std::shared_ptr<const ::arrow::KeyValueMetadata>& metadata) {
+ if (!metadata) {
+ return -1;
+ }
+ int key = metadata->FindKey(FIELD_ID_KEY);
+ if (key < 0) {
+ return -1;
+ }
+ const std::string& field_id_str = metadata->value(key);
+ int field_id;
+ if (::arrow::internal::ParseValue<::arrow::Int32Type>(
+ field_id_str.c_str(), field_id_str.length(), &field_id)) {
+ if (field_id < 0) {
+ // Thrift should convert any negative value to null but normalize to -1
here in
+ // case we later check this in logic.
+ return -1;
+ }
+ return field_id;
+ } else {
+ return -1;
+ }
+}
+
+Status VariantTypedValueToNode(const std::shared_ptr<Field>& field,
+ const WriterProperties& properties,
+ const ArrowWriterProperties& arrow_properties,
+ NodePtr* out);
+
+Status VariantFieldGroupToNode(const std::shared_ptr<Field>& field,
+ const WriterProperties& properties,
+ const ArrowWriterProperties& arrow_properties,
+ NodePtr* out) {
+ const int field_id = FieldIdFromMetadata(field->metadata());
+ const auto& type = field->type();
+ if (type->id() != ArrowTypeId::STRUCT) {
+ return Status::Invalid("Invalid Variant shredded field group: ",
type->ToString());
+ }
+
+ std::vector<NodePtr> children;
+ children.reserve(type->num_fields());
+ for (const auto& child : type->fields()) {
+ NodePtr node;
+ if (child->name() == "value") {
+ RETURN_NOT_OK(
+ FieldToNode(child->name(), child, properties, arrow_properties,
&node));
+ } else if (child->name() == "typed_value") {
+ RETURN_NOT_OK(VariantTypedValueToNode(child, properties,
arrow_properties, &node));
+ } else {
+ return Status::Invalid("Invalid Variant shredded field group child: ",
+ child->name());
+ }
+ children.emplace_back(std::move(node));
+ }
+
+ *out = GroupNode::Make(field->name(),
RepetitionFromNullable(field->nullable()),
+ std::move(children), nullptr, field_id);
+ return Status::OK();
+}
+
+Status VariantObjectToNode(const std::shared_ptr<Field>& field,
+ const WriterProperties& properties,
+ const ArrowWriterProperties& arrow_properties,
NodePtr* out) {
+ const int field_id = FieldIdFromMetadata(field->metadata());
+ const auto& type = field->type();
+ if (type->num_fields() == 0) {
+ return Status::Invalid("Invalid Variant object typed_value: expected at
least one ",
+ "shredded field");
+ }
+
+ std::vector<NodePtr> children(type->num_fields());
+ for (int i = 0; i < type->num_fields(); ++i) {
+ RETURN_NOT_OK(VariantFieldGroupToNode(type->field(i), properties,
arrow_properties,
+ &children[i]));
+ }
+
+ *out = GroupNode::Make(field->name(),
RepetitionFromNullable(field->nullable()),
+ std::move(children), nullptr, field_id);
+ return Status::OK();
+}
+
+Status VariantListToNode(const std::shared_ptr<Field>& field,
+ const WriterProperties& properties,
+ const ArrowWriterProperties& arrow_properties,
NodePtr* out) {
+ const int field_id = FieldIdFromMetadata(field->metadata());
+ const auto list_type =
std::static_pointer_cast<::arrow::BaseListType>(field->type());
+
+ NodePtr element;
+
RETURN_NOT_OK(VariantFieldGroupToNode(list_type->value_field()->WithName("element"),
+ properties, arrow_properties,
&element));
+
+ NodePtr list = GroupNode::Make("list", Repetition::REPEATED, {element});
+ *out = GroupNode::Make(field->name(),
RepetitionFromNullable(field->nullable()), {list},
+ LogicalType::List(), field_id);
+ return Status::OK();
+}
+
+Status VariantPrimitiveToNode(const std::shared_ptr<Field>& field, NodePtr*
out) {
+ std::shared_ptr<const LogicalType> logical_type = LogicalType::None();
+ ParquetType::type type = ParquetType::UNDEFINED;
+ const int field_id = FieldIdFromMetadata(field->metadata());
+
+ int length = -1;
+ int precision = -1;
+ int scale = -1;
+
+ switch (field->type()->id()) {
+ case ArrowTypeId::BOOL:
+ type = ParquetType::BOOLEAN;
+ break;
+ case ArrowTypeId::INT8:
+ type = ParquetType::INT32;
+ logical_type = LogicalType::Int(8, true);
+ break;
+ case ArrowTypeId::INT16:
+ type = ParquetType::INT32;
+ logical_type = LogicalType::Int(16, true);
+ break;
+ case ArrowTypeId::INT32:
+ type = ParquetType::INT32;
+ break;
+ case ArrowTypeId::INT64:
+ type = ParquetType::INT64;
+ break;
+ case ArrowTypeId::FLOAT:
+ type = ParquetType::FLOAT;
+ break;
+ case ArrowTypeId::DOUBLE:
+ type = ParquetType::DOUBLE;
+ break;
+ case ArrowTypeId::DECIMAL32:
+ case ArrowTypeId::DECIMAL64:
+ case ArrowTypeId::DECIMAL128: {
+ const auto& decimal = checked_cast<const
::arrow::DecimalType&>(*field->type());
+ precision = decimal.precision();
+ scale = decimal.scale();
+ if (scale < 0 || scale > precision) {
+ return Status::Invalid("Invalid Variant decimal type: ",
decimal.ToString());
+ }
+ if (decimal.id() == ArrowTypeId::DECIMAL32 && precision >= 1 &&
precision <= 9) {
+ type = ParquetType::INT32;
+ } else if (decimal.id() == ArrowTypeId::DECIMAL64 && precision > 9 &&
+ precision <= 18) {
+ type = ParquetType::INT64;
+ } else if (decimal.id() == ArrowTypeId::DECIMAL128 && precision > 18 &&
+ precision <= 38) {
+ type = ParquetType::FIXED_LEN_BYTE_ARRAY;
+ length = ::arrow::DecimalType::DecimalSize(precision);
+ } else {
+ return Status::Invalid("Invalid Variant decimal type: ",
decimal.ToString());
+ }
+ PARQUET_CATCH_NOT_OK(logical_type = LogicalType::Decimal(precision,
scale));
+ break;
+ }
+ case ArrowTypeId::DATE32:
+ type = ParquetType::INT32;
+ logical_type = LogicalType::Date();
+ break;
+ case ArrowTypeId::TIME64:
+ type = ParquetType::INT64;
+ logical_type =
+ LogicalType::Time(/*is_adjusted_to_utc=*/false,
LogicalType::TimeUnit::MICROS);
+ break;
+ case ArrowTypeId::TIMESTAMP: {
+ type = ParquetType::INT64;
+ const auto& timestamp = checked_cast<const
::arrow::TimestampType&>(*field->type());
+ const bool utc = !timestamp.timezone().empty();
+ switch (timestamp.unit()) {
+ case ::arrow::TimeUnit::MICRO:
+ logical_type = LogicalType::Timestamp(utc,
LogicalType::TimeUnit::MICROS,
+
/*is_from_converted_type=*/false,
+
/*force_set_converted_type=*/true);
+ break;
+ case ::arrow::TimeUnit::NANO:
+ logical_type = LogicalType::Timestamp(utc,
LogicalType::TimeUnit::NANOS,
+
/*is_from_converted_type=*/false,
+
/*force_set_converted_type=*/false);
Review Comment:
The timestamp annotation always retains the input unit, whereas
`TypedColumnWriterImpl<Int64Type>::WriteArrowTimestamps` still applies the
global timestamp-coercion options (and automatic nanosecond-to-microsecond
conversion for Parquet 1.0/2.4). For example, a nanosecond `typed_value` of
1000 with microsecond coercion is written as 1 but annotated NANOS, silently
changing the timestamp. Keep the Variant schema and value conversion
consistent, or reject incompatible coercion options before writing.
##########
cpp/src/parquet/arrow/schema.cc:
##########
@@ -131,21 +369,40 @@ Status MapToNode(const std::shared_ptr<::arrow::MapType>&
type, const std::strin
return Status::OK();
}
-Status VariantToNode(
- const std::shared_ptr<::arrow::extension::VariantExtensionType>& type,
- const std::string& name, bool nullable, int field_id,
- const WriterProperties& properties, const ArrowWriterProperties&
arrow_properties,
- NodePtr* out) {
- NodePtr metadata_node;
- RETURN_NOT_OK(FieldToNode("metadata", type->metadata(), properties,
arrow_properties,
- &metadata_node));
+Status VariantToNode(const std::shared_ptr<VariantExtensionType>& type,
+ const std::string& name, bool nullable, int field_id,
+ const WriterProperties& properties,
+ const ArrowWriterProperties& arrow_properties, NodePtr*
out) {
+ // `VariantExtensionType` already validates the general storage shape.
Writing is
+ // stricter than reading: we keep accepting typed-only Variant schemas on
read,
+ // but must not write them.
+ if (type->value() == nullptr) {
+ return Status::Invalid("Invalid Variant write schema: missing top-level
value field");
+ }
Review Comment:
Checking only the top-level value field still allows read-compatible schemas
to be written, including optional object field groups and missing object value
fields hidden by null typed objects. Row validation does not enforce these
schema requirements, so writing can succeed with a layout rejected by the
strict factory. Check the complete storage shape here, independently of which
rows are populated.
##########
cpp/src/parquet/variant/decoding.cc:
##########
@@ -0,0 +1,530 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "parquet/variant/decoding.h"
+
+#include <algorithm>
+#include <cstring>
+#include <limits>
+#include <utility>
+
+#include "arrow/util/endian.h"
+#include "arrow/util/logging_internal.h"
+#include "arrow/util/small_vector.h"
+#include "parquet/exception.h"
+#include "parquet/variant/format_internal.h"
+
+namespace parquet::variant {
+
+namespace bit_util = ::arrow::bit_util;
+
+namespace internal {
+
+class ViewAccess {
+ public:
+ template <typename View, typename... Args>
+ static View Make(Args&&... args) {
+ return View(std::forward<Args>(args)...);
+ }
+};
+
+} // namespace internal
+
+namespace {
+
+using ::arrow::internal::SmallVector;
+using internal::ViewAccess;
+
+constexpr size_t kInlineElements = 16;
+
+uint32_t ReadLittleEndian(std::string_view data, size_t offset, size_t width) {
+ DCHECK_LE(width, sizeof(uint32_t));
+ uint32_t value = 0;
+ std::memcpy(&value, data.data() + offset, width);
+ return bit_util::FromLittleEndian(value);
+}
+
+void CheckAvailable(std::string_view data, size_t offset, size_t size,
+ std::string_view context) {
+ if (offset > data.size() || data.size() - offset < size) {
+ throw ParquetInvalidOrCorruptedFileException("Invalid Variant encoding:
truncated ",
+ context);
+ }
+}
+
+size_t OffsetCount(uint32_t count) {
+ const auto converted_count = static_cast<size_t>(count);
+ if (converted_count == std::numeric_limits<size_t>::max()) {
+ throw ParquetInvalidOrCorruptedFileException(
+ "Invalid Variant encoding: offset count overflow");
+ }
+ return converted_count + 1;
+}
+
+size_t OffsetBytes(size_t offset_count, uint8_t offset_size) {
+ if (offset_count >
+ std::numeric_limits<size_t>::max() / static_cast<size_t>(offset_size)) {
+ throw ParquetInvalidOrCorruptedFileException(
+ "Invalid Variant encoding: offset table size overflow");
+ }
+ return offset_count * static_cast<size_t>(offset_size);
+}
+
+size_t PrimitivePayloadSize(std::string_view value, size_t offset,
+ VariantPrimitiveType primitive) {
+ switch (primitive) {
+ case VariantPrimitiveType::kNull:
+ case VariantPrimitiveType::kBooleanTrue:
+ case VariantPrimitiveType::kBooleanFalse:
+ return 0;
+ case VariantPrimitiveType::kInt8:
+ return 1;
+ case VariantPrimitiveType::kInt16:
+ return 2;
+ case VariantPrimitiveType::kInt32:
+ case VariantPrimitiveType::kDate:
+ case VariantPrimitiveType::kFloat:
+ return 4;
+ case VariantPrimitiveType::kInt64:
+ case VariantPrimitiveType::kDouble:
+ case VariantPrimitiveType::kTimestampMicros:
+ case VariantPrimitiveType::kTimestampNTZMicros:
+ case VariantPrimitiveType::kTimeNTZMicros:
+ case VariantPrimitiveType::kTimestampNanos:
+ case VariantPrimitiveType::kTimestampNTZNanos:
+ return 8;
+ case VariantPrimitiveType::kDecimal4:
+ return 5;
+ case VariantPrimitiveType::kDecimal8:
+ return 9;
+ case VariantPrimitiveType::kDecimal16:
+ return 17;
+ case VariantPrimitiveType::kUuid:
+ return 16;
+ case VariantPrimitiveType::kBinary:
+ case VariantPrimitiveType::kString: {
+ CheckAvailable(value, offset, 4, "variable-length size");
+ const uint32_t length = ReadLittleEndian(value, offset, 4);
+ return 4 + static_cast<size_t>(length);
+ }
+ }
+ throw ParquetInvalidOrCorruptedFileException(
+ "Invalid Variant encoding: unknown primitive type");
+}
+
+size_t ParsePrimitive(std::string_view value, size_t offset,
+ VariantPrimitiveType primitive) {
+ if (!internal::IsKnownVariantPrimitive(primitive)) {
+ throw ParquetInvalidOrCorruptedFileException(
+ "Invalid Variant encoding: unknown primitive type ",
static_cast<int>(primitive));
+ }
+
+ const size_t payload_size = PrimitivePayloadSize(value, offset, primitive);
+ CheckAvailable(value, offset, payload_size, "primitive value");
+
+ if (internal::IsDecimalVariantPrimitive(primitive)) {
+ const auto scale = static_cast<uint8_t>(value[offset]);
+ internal::ValidateDecimalScale(scale);
+ }
+
+ if (primitive == VariantPrimitiveType::kString) {
+ const uint32_t length = ReadLittleEndian(value, offset, 4);
+ internal::ValidateUtf8(value.substr(offset + 4, length), "primitive string
value");
+ }
+
+ return payload_size;
+}
+
+template <bool validate_children>
+size_t ParseValue(std::string_view value, const VariantMetadataView& metadata,
+ VariantValueView* out);
+
+template <bool validate_children>
+size_t ParseArray(std::string_view value, const VariantMetadataView& metadata,
+ uint8_t header, VariantValueView* out) {
+ const auto offset_size = static_cast<uint8_t>((header & 0x03) + 1);
+ const bool is_large = (header & 0x04) != 0;
+ const size_t count_size = is_large ? 4 : 1;
+
+ size_t offset = 1;
+ CheckAvailable(value, offset, count_size, "array size");
+ const uint32_t num_elements = ReadLittleEndian(value, offset, count_size);
+ offset += count_size;
+ const size_t offset_count = OffsetCount(num_elements);
+
+ CheckAvailable(value, offset, OffsetBytes(offset_count, offset_size), "array
offsets");
+
+ SmallVector<uint32_t, kInlineElements + 1> offsets(offset_count);
+ for (size_t i = 0; i < offset_count; ++i) {
+ offsets[i] = ReadLittleEndian(value, offset, offset_size);
+ offset += offset_size;
+ }
+
+ if (offsets[0] != 0) {
+ throw ParquetInvalidOrCorruptedFileException(
+ "Invalid Variant encoding: first array offset must be 0");
+ }
+ for (uint32_t i = 0; i < num_elements; ++i) {
+ if (offsets[i] > offsets[i + 1]) {
+ throw ParquetInvalidOrCorruptedFileException(
+ "Invalid Variant encoding: array offsets must be monotonic");
+ }
+ }
+
+ const size_t values_start = offset;
+ const size_t total_value_size = offsets[num_elements];
+ CheckAvailable(value, values_start, total_value_size, "array values");
+
+ if constexpr (validate_children) {
+ for (uint32_t i = 0; i < num_elements; ++i) {
+ const size_t child_consumed = ParseValue<true>(
+ value.substr(values_start + offsets[i]), metadata, /*out=*/nullptr);
Review Comment:
Recursive validation has no nesting limit, so deeply nested arrays in
file-supplied Variant bytes can exhaust the native stack during default-enabled
read validation. The Parquet schema-depth limit does not protect nesting inside
a binary value. Bound the depth in `ParseValue` or use an explicit traversal
stack, covering object recursion at lines 296–297 as well.
##########
cpp/src/parquet/variant/meson.build:
##########
@@ -0,0 +1,44 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+exc = executable(
+ 'parquet-variant-test',
+ sources: [
+ 'builder_test.cc',
+ 'decoding_test.cc',
+ 'parquet_testing_test.cc',
+ 'shred_test.cc',
+ 'test_util_internal.cc',
+ 'type_test.cc',
+ 'unshred_test.cc',
+ 'validate_test.cc',
+ ],
+ dependencies: parquet_test_dep,
Review Comment:
`parquet_test_dep` is undefined when this subdirectory is evaluated: the
parent enters `subdir('variant')` at line 140 but only defines this dependency
at line 301. This makes Meson configuration fail whenever Parquet is enabled.
Register this executable after the dependency is initialized, for example
through the parent's `parquet_tests` table.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
