zhjwpku commented on code in PR #1055:
URL: https://github.com/apache/iceberg-cpp/pull/1055#discussion_r4436203078
##########
src/iceberg/avro/avro_schema_util.cc:
##########
@@ -903,6 +903,166 @@ Result<SchemaProjection> Project(const Schema&
expected_schema,
namespace {
+Result<std::shared_ptr<Type>> FromAvroNode(const ::avro::NodePtr& node);
+
+std::optional<std::string> GetIcebergFieldName(const ::avro::NodePtr& node,
+ size_t field_idx) {
+ if (field_idx >= node->customAttributes()) {
+ return std::nullopt;
+ }
+ auto name = node->customAttributesAt(field_idx).getAttribute(
+ std::string(kIcebergFieldNameProp));
+ if (name.has_value() && name->size() >= 2 && name->front() == '"' &&
+ name->back() == '"') {
+ return name->substr(1, name->size() - 2);
+ }
+ return name;
+}
+
+// Converts an Avro field node to an Iceberg field. A union with null is read
as an
+// optional field and a bare null node as an optional field of unknown type.
+Result<SchemaField> FromAvroField(int32_t field_id, std::string_view name,
+ const ::avro::NodePtr& node) {
+ if (node->type() == ::avro::AVRO_NULL) {
+ return SchemaField::MakeOptional(field_id, name, iceberg::unknown());
+ }
+ ::avro::NodePtr value_node;
+ ICEBERG_RETURN_UNEXPECTED(UnwrapUnion(node, &value_node));
+ ICEBERG_ASSIGN_OR_RAISE(auto type, FromAvroNode(value_node));
+ bool optional = node->type() == ::avro::AVRO_UNION;
+ return SchemaField(field_id, name, std::move(type), optional);
+}
+
+Result<std::vector<SchemaField>> FromAvroRecordFields(const ::avro::NodePtr&
node) {
+ if (node->type() != ::avro::AVRO_RECORD) {
+ return InvalidSchema("Expected AVRO_RECORD type, but got {}",
ToString(node));
+ }
+ std::vector<SchemaField> fields;
+ fields.reserve(node->leaves());
+ for (size_t i = 0; i < node->leaves(); ++i) {
+ ICEBERG_ASSIGN_OR_RAISE(int32_t field_id, GetFieldId(node, i));
+ std::string name = GetIcebergFieldName(node, i).value_or(node->nameAt(i));
+ ICEBERG_ASSIGN_OR_RAISE(auto field, FromAvroField(field_id, name,
node->leafAt(i)));
+ fields.emplace_back(std::move(field));
+ }
+ return fields;
+}
+
+Result<std::shared_ptr<Type>> FromAvroTimestamp(const ::avro::NodePtr& node,
bool nanos) {
+ bool adjust_to_utc = GetAdjustToUtc(node).value_or("false") == "true";
+ if (nanos) {
+ return adjust_to_utc ? std::shared_ptr<Type>(iceberg::timestamptz_ns())
+ : std::shared_ptr<Type>(iceberg::timestamp_ns());
+ }
+ return adjust_to_utc ? std::shared_ptr<Type>(iceberg::timestamp_tz())
+ : std::shared_ptr<Type>(iceberg::timestamp());
+}
+
+Result<std::shared_ptr<Type>> FromAvroNode(const ::avro::NodePtr& node) {
+ switch (node->type()) {
+ case ::avro::AVRO_RECORD: {
+ ICEBERG_ASSIGN_OR_RAISE(auto fields, FromAvroRecordFields(node));
+ return iceberg::struct_(std::move(fields));
+ }
+ case ::avro::AVRO_ARRAY: {
+ if (node->leaves() != 1) {
+ return InvalidSchema("Array type must have exactly one node, got {}",
+ node->leaves());
+ }
+ if (HasMapLogicalType(node)) {
+ const auto& entry_node = node->leafAt(0);
+ ICEBERG_ASSIGN_OR_RAISE(auto entry_fields,
FromAvroRecordFields(entry_node));
+ if (entry_fields.size() != 2) {
+ return InvalidSchema(
+ "Array-backed map type must have a record node with two fields");
+ }
+ SchemaField key(entry_fields[0].field_id(), MapType::kKeyName,
Review Comment:
Do we need to validate the key is not optional? We had the following in
`ToAvroNodeVisitor::Visit(const MapType& type, ::avro::NodePtr* node)`, it
would be better to mirror that here.
```
if (key_field.optional()) [[unlikely]] {
return InvalidArgument("Map key `{}` must be required",
key_field.name());
}
```
##########
src/iceberg/parquet/parquet_schema_util.cc:
##########
@@ -826,6 +882,27 @@ void ApplyNameMapping(const NameMapping& mapping,
}
}
+Result<std::unique_ptr<Schema>> FromParquetSchema(
+ const ::parquet::arrow::SchemaManifest& parquet_schema) {
+ ::arrow::FieldVector fields;
+ fields.reserve(parquet_schema.schema_fields.size());
+ for (const auto& parquet_field : parquet_schema.schema_fields) {
+ if (auto field = ToArrowFieldWithIds(parquet_field); field != nullptr) {
Review Comment:
Fields without an ID are skipped here, but Avro fails on missing IDs, should
we made them consistent?
##########
src/iceberg/parquet/parquet_reader.cc:
##########
@@ -304,6 +320,16 @@ class ParquetReader::Impl {
reader_, ::parquet::arrow::FileReader::Make(pool_,
std::move(file_reader),
arrow_reader_properties));
+ ::parquet::arrow::SchemaManifest schema_manifest;
Review Comment:
SchemaManifest is built twice, once here, once in BuildProjection, I think
we can reduce one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
