feat: support Iceberg v3 unknown type

Add an Iceberg unknown primitive type and JSON, Arrow, Avro, Parquet, projection, and data path support for null-only unknown fields. Enforce optionality invariants so required projections cannot be materialized from unknown/null-only fields.

Co-authored-by: Codex <codex@openai.com>

test: cover forbidden nested type promotions

Assert that promotion helpers reject nested type targets for unknown and regular primitive source types.

Co-authored-by: Codex <codex@openai.com>

Author: manuzhang

src/iceberg/avro/avro_data_util.cc

@@ -457,6 +457,11 @@ Status AppendFieldToBuilder(const ::avro::NodePtr& avro_node,
                             const SchemaField& projected_field,
                             const arrow::MetadataColumnContext& metadata_context,
                             ::arrow::ArrayBuilder* array_builder) {
+  if (projection.kind == FieldProjection::Kind::kNull) {
+    ICEBERG_ARROW_RETURN_NOT_OK(array_builder->AppendNull());
+    return {};
+  }
+
   if (avro_node->type() == ::avro::AVRO_UNION) {
     size_t branch = avro_datum.unionBranch();
     if (avro_node->leafAt(branch)->type() == ::avro::AVRO_NULL) {
@@ -507,6 +512,9 @@ Status ExtractDatumFromArray(const ::arrow::Array& array, int64_t index,
   }
 
   if (array.IsNull(index)) {
+    if (datum->type() == ::avro::AVRO_NULL) {
+      return {};
+    }
     if (!datum->isUnion()) [[unlikely]] {
       return InvalidSchema("Cannot extract null to non-union type: {}",
                            ::avro::toString(datum->type()));

src/iceberg/avro/avro_direct_decoder.cc

@@ -588,6 +588,12 @@ Status DecodeFieldToBuilder(const ::avro::NodePtr& avro_node, ::avro::Decoder& d
                             const SchemaField& projected_field,
                             const arrow::MetadataColumnContext& metadata_context,
                             ::arrow::ArrayBuilder* array_builder, DecodeContext& ctx) {
+  if (projection.kind == FieldProjection::Kind::kNull) {
+    ICEBERG_RETURN_UNEXPECTED(SkipAvroValue(avro_node, decoder));
+    ICEBERG_ARROW_RETURN_NOT_OK(array_builder->AppendNull());
+    return {};
+  }
+
   if (avro_node->type() == ::avro::AVRO_UNION) {
     const size_t branch_index = decoder.decodeUnionIndex();
 

src/iceberg/avro/avro_direct_encoder.cc

@@ -80,7 +80,7 @@ Status EncodeArrowToAvro(const ::avro::NodePtr& avro_node, ::avro::Encoder& enco
     return EncodeArrowToAvro(branches.value_node, encoder, type, array, row_index, ctx);
   }
 
-  if (is_null) {
+  if (is_null && avro_node->type() != ::avro::AVRO_NULL) {
     return InvalidArgument("Null value in non-nullable field");
   }
 

src/iceberg/avro/avro_schema_util.cc

@@ -237,6 +237,11 @@ Status ToAvroNodeVisitor::Visit(const BinaryType& type, ::avro::NodePtr* node) {
   return {};
 }
 
+Status ToAvroNodeVisitor::Visit(const UnknownType&, ::avro::NodePtr* node) {
+  *node = std::make_shared<::avro::NodePrimitive>(::avro::AVRO_NULL);
+  return {};
+}
+
 Status ToAvroNodeVisitor::Visit(const StructType& type, ::avro::NodePtr* node) {
   *node = std::make_shared<::avro::NodeRecord>();
 
@@ -338,7 +343,7 @@ Status ToAvroNodeVisitor::Visit(const SchemaField& field, ::avro::NodePtr* node)
   field_ids_.push(field.field_id());
   ICEBERG_RETURN_UNEXPECTED(VisitTypeInline(*field.type(), /*visitor=*/this, node));
 
-  if (field.optional()) {
+  if (field.optional() && (*node)->type() != ::avro::AVRO_NULL) {
     ::avro::MultiLeaves union_types;
     union_types.add(std::make_shared<::avro::NodePrimitive>(::avro::AVRO_NULL));
     union_types.add(std::move(*node));
@@ -383,8 +388,8 @@ Status HasIdVisitor::Visit(const ::avro::NodePtr& node) {
     case ::avro::AVRO_STRING:
     case ::avro::AVRO_BYTES:
     case ::avro::AVRO_FIXED:
-      return {};
     case ::avro::AVRO_NULL:
+      return {};
     case ::avro::AVRO_ENUM:
     default:
       return InvalidSchema("Unsupported Avro type: {}", static_cast<int>(node->type()));
@@ -512,6 +517,10 @@ Result<int32_t> GetFieldId(const ::avro::NodePtr& node, size_t field_idx) {
 
 Status ValidateAvroSchemaEvolution(const Type& expected_type,
                                    const ::avro::NodePtr& avro_node) {
+  if (avro_node->type() == ::avro::AVRO_NULL) {
+    return {};
+  }
+
   switch (expected_type.type_id()) {
     case TypeId::kBoolean:
       if (avro_node->type() == ::avro::AVRO_BOOL) {
@@ -615,6 +624,8 @@ Status ValidateAvroSchemaEvolution(const Type& expected_type,
         return {};
       }
       break;
+    case TypeId::kUnknown:
+      return {};
     default:
       break;
   }
@@ -650,6 +661,35 @@ Result<FieldProjection> ProjectNested(const Type& expected_type,
                                       const ::avro::NodePtr& avro_node,
                                       bool prune_source);
 
+Result<FieldProjection> ProjectField(const SchemaField& expected_field,
+                                     const ::avro::NodePtr& avro_node,
+                                     size_t source_index, bool prune_source) {
+  const Type& expected_type = *expected_field.type();
+  ::avro::NodePtr field_node;
+  ICEBERG_RETURN_UNEXPECTED(UnwrapUnion(avro_node, &field_node));
+
+  FieldProjection projection;
+  if (expected_type.type_id() == TypeId::kUnknown ||
+      field_node->type() == ::avro::AVRO_NULL) {
+    if (!expected_field.optional()) {
+      return InvalidSchema("Cannot project required field with ID: {} as null",
+                           expected_field.field_id());
+    }
+    projection.kind = FieldProjection::Kind::kNull;
+    return projection;
+  }
+
+  if (expected_type.is_nested()) {
+    ICEBERG_ASSIGN_OR_RAISE(projection,
+                            ProjectNested(expected_type, field_node, prune_source));
+  } else {
+    ICEBERG_RETURN_UNEXPECTED(ValidateAvroSchemaEvolution(expected_type, field_node));
+  }
+  projection.from = source_index;
+  projection.kind = FieldProjection::Kind::kProjected;
+  return projection;
+}
+
 Result<FieldProjection> ProjectStruct(const StructType& struct_type,
                                       const ::avro::NodePtr& avro_node,
                                       bool prune_source) {
@@ -685,18 +725,9 @@ Result<FieldProjection> ProjectStruct(const StructType& struct_type,
     FieldProjection child_projection;
 
     if (auto iter = node_info_map.find(field_id); iter != node_info_map.cend()) {
-      ::avro::NodePtr field_node;
-      ICEBERG_RETURN_UNEXPECTED(UnwrapUnion(iter->second.field_node, &field_node));
-      if (expected_field.type()->is_nested()) {
-        ICEBERG_ASSIGN_OR_RAISE(
-            child_projection,
-            ProjectNested(*expected_field.type(), field_node, prune_source));
-      } else {
-        ICEBERG_RETURN_UNEXPECTED(
-            ValidateAvroSchemaEvolution(*expected_field.type(), field_node));
-      }
-      child_projection.from = iter->second.local_index;
-      child_projection.kind = FieldProjection::Kind::kProjected;
+      ICEBERG_ASSIGN_OR_RAISE(child_projection,
+                              ProjectField(expected_field, iter->second.field_node,
+                                           iter->second.local_index, prune_source));
     } else if (MetadataColumns::IsMetadataColumn(field_id)) {
       child_projection.kind = FieldProjection::Kind::kMetadata;
     } else if (expected_field.optional()) {
@@ -733,20 +764,9 @@ Result<FieldProjection> ProjectList(const ListType& list_type,
   }
 
   FieldProjection element_projection;
-  ::avro::NodePtr element_node;
-  ICEBERG_RETURN_UNEXPECTED(UnwrapUnion(avro_node->leafAt(0), &element_node));
-  if (expected_element_field.type()->is_nested()) {
-    ICEBERG_ASSIGN_OR_RAISE(
-        element_projection,
-        ProjectNested(*expected_element_field.type(), element_node, prune_source));
-  } else {
-    ICEBERG_RETURN_UNEXPECTED(
-        ValidateAvroSchemaEvolution(*expected_element_field.type(), element_node));
-  }
-
-  // Set the element projection metadata but preserve its children
-  element_projection.kind = FieldProjection::Kind::kProjected;
-  element_projection.from = size_t{0};
+  ICEBERG_ASSIGN_OR_RAISE(element_projection,
+                          ProjectField(expected_element_field, avro_node->leafAt(0),
+                                       size_t{0}, prune_source));
 
   FieldProjection result;
   result.children.emplace_back(std::move(element_projection));
@@ -802,18 +822,10 @@ Result<FieldProjection> ProjectMap(const MapType& map_type,
 
   for (size_t i = 0; i < map_node->leaves(); ++i) {
     FieldProjection sub_projection;
-    ::avro::NodePtr sub_node;
-    ICEBERG_RETURN_UNEXPECTED(UnwrapUnion(map_node->leafAt(i), &sub_node));
     const auto& expected_sub_field = map_type.fields()[i];
-    if (expected_sub_field.type()->is_nested()) {
-      ICEBERG_ASSIGN_OR_RAISE(sub_projection, ProjectNested(*expected_sub_field.type(),
-                                                            sub_node, prune_source));
-    } else {
-      ICEBERG_RETURN_UNEXPECTED(
-          ValidateAvroSchemaEvolution(*expected_sub_field.type(), sub_node));
-    }
-    sub_projection.kind = FieldProjection::Kind::kProjected;
-    sub_projection.from = i;
+    ICEBERG_ASSIGN_OR_RAISE(
+        sub_projection,
+        ProjectField(expected_sub_field, map_node->leafAt(i), i, prune_source));
     result.children.emplace_back(std::move(sub_projection));
   }
 
@@ -1049,9 +1061,9 @@ Result<::avro::NodePtr> MakeAvroNodeWithFieldIds(const ::avro::NodePtr& original
     case ::avro::AVRO_STRING:
     case ::avro::AVRO_BYTES:
     case ::avro::AVRO_FIXED:
+    case ::avro::AVRO_NULL:
       // For primitive types, just return a copy
       return original_node;
-    case ::avro::AVRO_NULL:
     case ::avro::AVRO_ENUM:
     default:
       return InvalidSchema("Unsupported Avro type for field ID application: {}",

src/iceberg/avro/avro_schema_util_internal.h

@@ -58,6 +58,7 @@ class ToAvroNodeVisitor {
   Status Visit(const UuidType& type, ::avro::NodePtr* node);
   Status Visit(const FixedType& type, ::avro::NodePtr* node);
   Status Visit(const BinaryType& type, ::avro::NodePtr* node);
+  Status Visit(const UnknownType&, ::avro::NodePtr*);
   Status Visit(const StructType& type, ::avro::NodePtr* node);
   Status Visit(const ListType& type, ::avro::NodePtr* node);
   Status Visit(const MapType& type, ::avro::NodePtr* node);

src/iceberg/json_serde.cc

@@ -377,6 +377,8 @@ nlohmann::json ToJson(const Type& type) {
     }
     case TypeId::kUuid:
       return "uuid";
+    case TypeId::kUnknown:
+      return "unknown";
   }
   std::unreachable();
 }
@@ -441,12 +443,22 @@ Result<std::unique_ptr<Type>> StructTypeFromJson(const nlohmann::json& json) {
   return std::make_unique<StructType>(std::move(fields));
 }
 
+Status ValidateUnknownFieldOptional(const Type& type, bool optional,
+                                    std::string_view field_name) {
+  if (type.type_id() == TypeId::kUnknown && !optional) {
+    return JsonParseError("Unknown type field '{}' must be optional", field_name);
+  }
+  return {};
+}
+
 Result<std::unique_ptr<Type>> ListTypeFromJson(const nlohmann::json& json) {
   ICEBERG_ASSIGN_OR_RAISE(auto element_type, TypeFromJson(json[kElement]));
   ICEBERG_ASSIGN_OR_RAISE(auto element_id, GetJsonValue<int32_t>(json, kElementId));
   ICEBERG_ASSIGN_OR_RAISE(auto element_required,
                           GetJsonValue<bool>(json, kElementRequired));
 
+  ICEBERG_RETURN_UNEXPECTED(ValidateUnknownFieldOptional(*element_type, !element_required,
+                                                         ListType::kElementName));
   return std::make_unique<ListType>(
       SchemaField(element_id, std::string(ListType::kElementName),
                   std::move(element_type), !element_required));
@@ -462,6 +474,11 @@ Result<std::unique_ptr<Type>> MapTypeFromJson(const nlohmann::json& json) {
   ICEBERG_ASSIGN_OR_RAISE(auto value_id, GetJsonValue<int32_t>(json, kValueId));
   ICEBERG_ASSIGN_OR_RAISE(auto value_required, GetJsonValue<bool>(json, kValueRequired));
 
+  if (key_type->type_id() == TypeId::kUnknown) {
+    return JsonParseError("Map 'key' cannot be unknown type");
+  }
+  ICEBERG_RETURN_UNEXPECTED(
+      ValidateUnknownFieldOptional(*value_type, !value_required, MapType::kValueName));
   SchemaField key_field(key_id, std::string(MapType::kKeyName), std::move(key_type),
                         /*optional=*/false);
   SchemaField value_field(value_id, std::string(MapType::kValueName),
@@ -502,6 +519,8 @@ Result<std::unique_ptr<Type>> TypeFromJson(const nlohmann::json& json) {
       return std::make_unique<BinaryType>();
     } else if (type_str == "uuid") {
       return std::make_unique<UuidType>();
+    } else if (type_str == "unknown") {
+      return std::make_unique<UnknownType>();
     } else if (type_str.starts_with("fixed")) {
       std::regex fixed_regex(R"(fixed\[\s*(\d+)\s*\])");
       std::smatch match;
@@ -548,6 +567,7 @@ Result<std::unique_ptr<SchemaField>> FieldFromJson(const nlohmann::json& json) {
   ICEBERG_ASSIGN_OR_RAISE(auto required, GetJsonValue<bool>(json, kRequired));
   ICEBERG_ASSIGN_OR_RAISE(auto doc, GetJsonValueOrDefault<std::string>(json, kDoc));
 
+  ICEBERG_RETURN_UNEXPECTED(ValidateUnknownFieldOptional(*type, !required, name));
   return std::make_unique<SchemaField>(field_id, std::move(name), std::move(type),
                                        !required, doc);
 }
@@ -949,6 +969,7 @@ Result<std::shared_ptr<Schema>> ParseSchemas(
     for (const auto& schema_json : schema_array) {
       ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<Schema> schema,
                               SchemaFromJson(schema_json));
+      ICEBERG_RETURN_UNEXPECTED(schema->Validate(format_version));
       if (schema->schema_id() == current_schema_id) {
         current_schema = schema;
       }
@@ -965,6 +986,7 @@ Result<std::shared_ptr<Schema>> ParseSchemas(
     ICEBERG_ASSIGN_OR_RAISE(auto schema_json,
                             GetJsonValue<nlohmann::json>(json, kSchema));
     ICEBERG_ASSIGN_OR_RAISE(current_schema, SchemaFromJson(schema_json));
+    ICEBERG_RETURN_UNEXPECTED(current_schema->Validate(format_version));
     current_schema_id = current_schema->schema_id();
     schemas.push_back(current_schema);
   }

src/iceberg/parquet/parquet_data_util.cc

@@ -166,7 +166,14 @@ Result<std::shared_ptr<::arrow::Array>> ProjectListArrayImpl(
   const auto& output_element_type = output_list_type->value_type();
 
   std::shared_ptr<::arrow::Array> projected_values;
-  if (element_field.type()->is_nested()) {
+  if (element_projection.kind == FieldProjection::Kind::kNull) {
+    ICEBERG_ASSIGN_OR_RAISE(
+        projected_values,
+        MakeNullArray(output_element_type, list_array->values()->length(), pool));
+  } else if (element_projection.kind != FieldProjection::Kind::kProjected) {
+    return NotImplemented("Unsupported list element projection kind: {}",
+                          ToString(element_projection.kind));
+  } else if (element_field.type()->is_nested()) {
     const auto& nested_type =
         internal::checked_cast<const NestedType&>(*element_field.type());
     ICEBERG_ASSIGN_OR_RAISE(
@@ -219,7 +226,14 @@ Result<std::shared_ptr<::arrow::Array>> ProjectMapArray(
 
   // Project keys
   std::shared_ptr<::arrow::Array> projected_keys;
-  if (key_type->is_nested()) {
+  if (key_projection.kind == FieldProjection::Kind::kNull) {
+    ICEBERG_ASSIGN_OR_RAISE(
+        projected_keys,
+        MakeNullArray(output_map_type->key_type(), map_array->keys()->length(), pool));
+  } else if (key_projection.kind != FieldProjection::Kind::kProjected) {
+    return NotImplemented("Unsupported map key projection kind: {}",
+                          ToString(key_projection.kind));
+  } else if (key_type->is_nested()) {
     const auto& nested_type = internal::checked_cast<const NestedType&>(*key_type);
     ICEBERG_ASSIGN_OR_RAISE(
         projected_keys,
@@ -233,7 +247,14 @@ Result<std::shared_ptr<::arrow::Array>> ProjectMapArray(
 
   // Project values
   std::shared_ptr<::arrow::Array> projected_items;
-  if (value_type->is_nested()) {
+  if (value_projection.kind == FieldProjection::Kind::kNull) {
+    ICEBERG_ASSIGN_OR_RAISE(
+        projected_items,
+        MakeNullArray(output_map_type->item_type(), map_array->items()->length(), pool));
+  } else if (value_projection.kind != FieldProjection::Kind::kProjected) {
+    return NotImplemented("Unsupported map value projection kind: {}",
+                          ToString(value_projection.kind));
+  } else if (value_type->is_nested()) {
     const auto& nested_type = internal::checked_cast<const NestedType&>(*value_type);
     ICEBERG_ASSIGN_OR_RAISE(
         projected_items,