fix: address unknown type review feedback

Enforce unknown as a v3-only optional type in schema validation and update paths, allow schema-update promotion from unknown to primitive types, reject unknown-to-nested projection, and reject unsupported Parquet writes for unknown list or map leaves.

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

Author: manuzhang

src/iceberg/parquet/parquet_writer.cc

@@ -20,7 +20,10 @@
 #include "iceberg/parquet/parquet_writer.h"
 
 #include <memory>
+#include <optional>
 #include <string_view>
+#include <utility>
+#include <vector>
 
 #include <arrow/c/bridge.h>
 #include <arrow/record_batch.h>
@@ -33,7 +36,11 @@
 
 #include "iceberg/arrow/arrow_io_internal.h"
 #include "iceberg/arrow/arrow_status_internal.h"
+#include "iceberg/parquet/parquet_data_util_internal.h"
+#include "iceberg/schema.h"
 #include "iceberg/schema_internal.h"
+#include "iceberg/schema_util.h"
+#include "iceberg/type.h"
 #include "iceberg/util/macros.h"
 
 namespace iceberg::parquet {
@@ -81,6 +88,75 @@ Result<std::optional<int32_t>> ParseCodecLevel(const WriterProperties& propertie
   return level;
 }
 
+Result<std::shared_ptr<Type>> PruneUnknownType(const std::shared_ptr<Type>& type);
+
+Result<std::optional<SchemaField>> PruneUnknownField(const SchemaField& field,
+                                                     bool allow_skip_unknown) {
+  if (field.type()->type_id() == TypeId::kUnknown) {
+    if (!allow_skip_unknown) {
+      return NotSupported(
+          "Cannot write unknown field '{}' inside a list or map to Parquet",
+          field.name());
+    }
+    return std::nullopt;
+  }
+
+  ICEBERG_ASSIGN_OR_RAISE(auto pruned_type, PruneUnknownType(field.type()));
+  return SchemaField(field.field_id(), field.name(), std::move(pruned_type),
+                     field.optional(), field.doc());
+}
+
+Result<std::shared_ptr<Type>> PruneUnknownType(const std::shared_ptr<Type>& type) {
+  switch (type->type_id()) {
+    case TypeId::kUnknown:
+      return NotSupported("Cannot write unknown field to Parquet");
+    case TypeId::kStruct: {
+      const auto& struct_type = static_cast<const StructType&>(*type);
+      std::vector<SchemaField> fields;
+      for (const auto& field : struct_type.fields()) {
+        ICEBERG_ASSIGN_OR_RAISE(auto pruned_field,
+                                PruneUnknownField(field, /*allow_skip_unknown=*/true));
+        if (pruned_field.has_value()) {
+          fields.emplace_back(std::move(pruned_field.value()));
+        }
+      }
+      return std::make_shared<StructType>(std::move(fields));
+    }
+    case TypeId::kList: {
+      const auto& list_type = static_cast<const ListType&>(*type);
+      ICEBERG_ASSIGN_OR_RAISE(
+          auto pruned_element,
+          PruneUnknownField(list_type.element(), /*allow_skip_unknown=*/false));
+      return std::make_shared<ListType>(std::move(pruned_element.value()));
+    }
+    case TypeId::kMap: {
+      const auto& map_type = static_cast<const MapType&>(*type);
+      ICEBERG_ASSIGN_OR_RAISE(
+          auto pruned_key,
+          PruneUnknownField(map_type.key(), /*allow_skip_unknown=*/false));
+      ICEBERG_ASSIGN_OR_RAISE(
+          auto pruned_value,
+          PruneUnknownField(map_type.value(), /*allow_skip_unknown=*/false));
+      return std::make_shared<MapType>(std::move(pruned_key.value()),
+                                       std::move(pruned_value.value()));
+    }
+    default:
+      return type;
+  }
+}
+
+Result<std::shared_ptr<Schema>> PruneUnknownFields(const Schema& schema) {
+  std::vector<SchemaField> fields;
+  for (const auto& field : schema.fields()) {
+    ICEBERG_ASSIGN_OR_RAISE(auto pruned_field,
+                            PruneUnknownField(field, /*allow_skip_unknown=*/true));
+    if (pruned_field.has_value()) {
+      fields.emplace_back(std::move(pruned_field.value()));
+    }
+  }
+  return std::make_shared<Schema>(std::move(fields), schema.schema_id());
+}
+
 }  // namespace
 
 class ParquetWriter::Impl {
@@ -97,8 +173,17 @@ class ParquetWriter::Impl {
     auto writer_properties = properties_builder.memory_pool(pool_)->build();
     auto arrow_writer_properties = ::parquet::default_arrow_writer_properties();
 
+    ArrowSchema input_c_schema;
+    ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(*options.schema, &input_c_schema));
+    ICEBERG_ARROW_ASSIGN_OR_RETURN(input_arrow_schema_,
+                                   ::arrow::ImportSchema(&input_c_schema));
+
+    ICEBERG_ASSIGN_OR_RAISE(write_schema_, PruneUnknownFields(*options.schema));
+    ICEBERG_ASSIGN_OR_RAISE(write_projection_, Project(*write_schema_, *options.schema,
+                                                       /*prune_source=*/false));
+
     ArrowSchema c_schema;
-    ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(*options.schema, &c_schema));
+    ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(*write_schema_, &c_schema));
     ICEBERG_ARROW_ASSIGN_OR_RETURN(arrow_schema_, ::arrow::ImportSchema(&c_schema));
 
     std::shared_ptr<::parquet::SchemaDescriptor> schema_descriptor;
@@ -123,8 +208,12 @@ class ParquetWriter::Impl {
   }
 
   Status Write(ArrowArray* array) {
-    ICEBERG_ARROW_ASSIGN_OR_RETURN(auto batch,
-                                   ::arrow::ImportRecordBatch(array, arrow_schema_));
+    ICEBERG_ARROW_ASSIGN_OR_RETURN(
+        auto input_batch, ::arrow::ImportRecordBatch(array, input_arrow_schema_));
+    ICEBERG_ASSIGN_OR_RAISE(
+        auto batch,
+        ProjectRecordBatch(std::move(input_batch), arrow_schema_, *write_schema_,
+                           write_projection_, arrow::MetadataColumnContext{}, pool_));
 
     ICEBERG_ARROW_RETURN_NOT_OK(writer_->WriteRecordBatch(*batch));
 
@@ -168,6 +257,11 @@ class ParquetWriter::Impl {
  private:
   // TODO(gangwu): make memory pool configurable
   ::arrow::MemoryPool* pool_ = ::arrow::default_memory_pool();
+  // The schema accepted from callers.
+  std::shared_ptr<::arrow::Schema> input_arrow_schema_;
+  // The Iceberg schema that has v3 unknown fields removed for physical writes.
+  std::shared_ptr<Schema> write_schema_;
+  SchemaProjection write_projection_;
   // Schema to write from the Parquet file.
   std::shared_ptr<::arrow::Schema> arrow_schema_;
   // The output stream to write Parquet file.

src/iceberg/schema.cc

@@ -46,6 +46,9 @@ Result<std::unique_ptr<Schema>> Schema::Make(std::vector<SchemaField> fields,
                                              int32_t schema_id,
                                              std::vector<int32_t> identifier_field_ids) {
   auto schema = std::make_unique<Schema>(std::move(fields), schema_id);
+  for (const auto& field : schema->fields()) {
+    ICEBERG_RETURN_UNEXPECTED(field.Validate());
+  }
 
   if (!identifier_field_ids.empty()) {
     auto id_to_parent = IndexParents(*schema);
@@ -63,6 +66,9 @@ Result<std::unique_ptr<Schema>> Schema::Make(
     std::vector<SchemaField> fields, int32_t schema_id,
     const std::vector<std::string>& identifier_field_names) {
   auto schema = std::make_unique<Schema>(std::move(fields), schema_id);
+  for (const auto& field : schema->fields()) {
+    ICEBERG_RETURN_UNEXPECTED(field.Validate());
+  }
 
   std::vector<int32_t> fresh_identifier_ids;
   for (const auto& name : identifier_field_names) {
@@ -288,6 +294,7 @@ Status Schema::Validate(int32_t format_version) const {
   // Check each field's type and defaults
   for (const auto& [field_id, field_ref] : id_to_field.get()) {
     const auto& field = field_ref.get();
+    ICEBERG_RETURN_UNEXPECTED(field.Validate());
 
     // Check if the field's type requires a minimum format version
     if (auto it = TableMetadata::kMinFormatVersions.find(field.type()->type_id());

src/iceberg/schema_field.cc

@@ -24,9 +24,26 @@
 
 #include "iceberg/type.h"
 #include "iceberg/util/formatter.h"  // IWYU pragma: keep
+#include "iceberg/util/macros.h"
 
 namespace iceberg {
 
+namespace {
+
+Status ValidateNestedFields(const Type& type) {
+  if (!type.is_nested()) {
+    return {};
+  }
+
+  const auto& nested_type = static_cast<const NestedType&>(type);
+  for (const auto& field : nested_type.fields()) {
+    ICEBERG_RETURN_UNEXPECTED(field.Validate());
+  }
+  return {};
+}
+
+}  // namespace
+
 SchemaField::SchemaField(int32_t field_id, std::string_view name,
                          std::shared_ptr<Type> type, bool optional, std::string_view doc)
     : field_id_(field_id),
@@ -62,6 +79,10 @@ Status SchemaField::Validate() const {
   if (type_ == nullptr) [[unlikely]] {
     return InvalidSchema("SchemaField cannot have null type");
   }
+  if (type_->type_id() == TypeId::kUnknown && !optional_) [[unlikely]] {
+    return InvalidSchema("Unknown type field '{}' must be optional", name_);
+  }
+  ICEBERG_RETURN_UNEXPECTED(ValidateNestedFields(*type_));
   return {};
 }
 

src/iceberg/schema_util.cc

@@ -101,12 +101,8 @@ Result<FieldProjection> ProjectField(const SchemaField& expected_field,
 
   if (source_field.type()->type_id() == TypeId::kUnknown &&
       expected_field.type()->is_nested()) {
-    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;
+    return InvalidSchema("Cannot project unknown field with id {} as nested type {}",
+                         expected_field.field_id(), *expected_field.type());
   }
 
   if (source_field.type()->type_id() == TypeId::kUnknown && !expected_field.optional()) {

src/iceberg/table_metadata.h

@@ -77,7 +77,9 @@ struct ICEBERG_EXPORT TableMetadata {
   static constexpr int64_t kInitialSequenceNumber = 0;
   static constexpr int64_t kInitialRowId = 0;
 
-  static inline const std::unordered_map<TypeId, int8_t> kMinFormatVersions = {};
+  static inline const std::unordered_map<TypeId, int8_t> kMinFormatVersions = {
+      {TypeId::kUnknown, 3},
+  };
 
   /// An integer version number for the format
   int8_t format_version;

src/iceberg/test/parquet_test.cc

@@ -533,6 +533,88 @@ TEST_F(ParquetReaderTest, ReadNestedUnknownProjection) {
   ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader));
 }
 
+TEST_F(ParquetReaderTest, WriteSkipsUnknownFields) {
+  temp_parquet_file_ = "write_skips_unknown.parquet";
+  auto schema = std::make_shared<Schema>(std::vector<SchemaField>{
+      SchemaField::MakeRequired(1, "id", int32()),
+      SchemaField::MakeOptional(2, "mystery", unknown()),
+      SchemaField::MakeOptional(3, "profile",
+                                std::make_shared<StructType>(std::vector<SchemaField>{
+                                    SchemaField::MakeOptional(4, "name", string()),
+                                    SchemaField::MakeOptional(5, "mystery", unknown()),
+                                })),
+  });
+
+  ArrowSchema arrow_c_schema;
+  ASSERT_THAT(ToArrowSchema(*schema, &arrow_c_schema), IsOk());
+  auto arrow_type = ::arrow::ImportType(&arrow_c_schema).ValueOrDie();
+  auto array = ::arrow::json::ArrayFromJSONString(arrow_type,
+                                                  R"([
+                     {"id": 1, "mystery": null, "profile": {"name": "Alice", "mystery": null}},
+                     {"id": 2, "mystery": null, "profile": {"name": "Bob", "mystery": null}}
+                   ])")
+                   .ValueOrDie();
+
+  WriterProperties writer_properties;
+  writer_properties.Set(WriterProperties::kParquetCompression,
+                        std::string("uncompressed"));
+  ASSERT_THAT(WriteArray(array, {.path = temp_parquet_file_,
+                                 .schema = schema,
+                                 .io = file_io_,
+                                 .properties = std::move(writer_properties)}),
+              IsOk());
+
+  auto io = internal::checked_cast<arrow::ArrowFileSystemFileIO&>(*file_io_);
+  auto input_stream = io.fs()->OpenInputFile(temp_parquet_file_).ValueOrDie();
+  auto metadata = ::parquet::ReadMetaData(input_stream);
+  ASSERT_EQ(metadata->schema()->num_columns(), 2);
+  EXPECT_THAT(metadata->schema()->ToString(), ::testing::HasSubstr("id"));
+  EXPECT_THAT(metadata->schema()->ToString(), ::testing::HasSubstr("name"));
+  EXPECT_THAT(metadata->schema()->ToString(),
+              ::testing::Not(::testing::HasSubstr("mystery")));
+
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto reader,
+      ReaderFactoryRegistry::Open(
+          FileFormatType::kParquet,
+          {.path = temp_parquet_file_, .io = file_io_, .projection = schema}));
+  ASSERT_NO_FATAL_FAILURE(VerifyNextBatch(*reader,
+                                          R"([
+        {"id": 1, "mystery": null, "profile": {"name": "Alice", "mystery": null}},
+        {"id": 2, "mystery": null, "profile": {"name": "Bob", "mystery": null}}
+      ])"));
+  ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader));
+}
+
+TEST_F(ParquetReaderTest, WriteRejectsUnknownListOrMapFields) {
+  auto list_schema = std::make_shared<Schema>(std::vector<SchemaField>{
+      SchemaField::MakeOptional(
+          1, "mysteries",
+          std::make_shared<ListType>(SchemaField::MakeOptional(2, "element", unknown()))),
+  });
+  auto list_writer_result = WriterFactoryRegistry::Open(
+      FileFormatType::kParquet,
+      {.path = "unknown_list.parquet", .schema = list_schema, .io = file_io_});
+  ASSERT_THAT(list_writer_result, IsError(ErrorKind::kNotSupported));
+  ASSERT_THAT(list_writer_result,
+              HasErrorMessage("Cannot write unknown field 'element' inside a list or map "
+                              "to Parquet"));
+
+  auto map_schema = std::make_shared<Schema>(std::vector<SchemaField>{
+      SchemaField::MakeOptional(
+          1, "properties",
+          std::make_shared<MapType>(SchemaField::MakeRequired(2, "key", string()),
+                                    SchemaField::MakeOptional(3, "value", unknown()))),
+  });
+  auto map_writer_result = WriterFactoryRegistry::Open(
+      FileFormatType::kParquet,
+      {.path = "unknown_map.parquet", .schema = map_schema, .io = file_io_});
+  ASSERT_THAT(map_writer_result, IsError(ErrorKind::kNotSupported));
+  ASSERT_THAT(map_writer_result,
+              HasErrorMessage("Cannot write unknown field 'value' inside a list or map "
+                              "to Parquet"));
+}
+
 class ParquetReadWrite : public ::testing::Test {
  protected:
   static void SetUpTestSuite() { parquet::RegisterAll(); }

src/iceberg/test/schema_test.cc

@@ -96,6 +96,53 @@ TEST(SchemaTest, Equality) {
   ASSERT_EQ(schema5, schema1);
 }
 
+TEST(SchemaTest, UnknownTypeRequiresFormatV3) {
+  iceberg::Schema schema({
+      iceberg::SchemaField::MakeOptional(1, "mystery", iceberg::unknown()),
+      iceberg::SchemaField::MakeOptional(
+          2, "profile",
+          iceberg::struct_({
+              iceberg::SchemaField::MakeOptional(3, "nested", iceberg::unknown()),
+          })),
+  });
+
+  EXPECT_THAT(schema.Validate(2), iceberg::IsError(iceberg::ErrorKind::kInvalidSchema));
+  EXPECT_THAT(schema.Validate(2),
+              iceberg::HasErrorMessage("unknown is not supported until v3"));
+  EXPECT_THAT(schema.Validate(3), iceberg::IsOk());
+}
+
+TEST(SchemaTest, RequiredUnknownFieldIsInvalid) {
+  iceberg::Schema direct_schema(
+      {iceberg::SchemaField::MakeRequired(1, "mystery", iceberg::unknown())});
+  EXPECT_THAT(direct_schema.Validate(3),
+              iceberg::IsError(iceberg::ErrorKind::kInvalidSchema));
+  EXPECT_THAT(direct_schema.Validate(3),
+              iceberg::HasErrorMessage("Unknown type field 'mystery' must be optional"));
+
+  auto schema = iceberg::Schema::Make(
+      {iceberg::SchemaField::MakeRequired(1, "mystery", iceberg::unknown())}, 100,
+      std::vector<int32_t>{});
+
+  EXPECT_THAT(schema, iceberg::IsError(iceberg::ErrorKind::kInvalidSchema));
+  EXPECT_THAT(schema,
+              iceberg::HasErrorMessage("Unknown type field 'mystery' must be optional"));
+}
+
+TEST(SchemaTest, RequiredNestedUnknownFieldIsInvalid) {
+  auto schema = iceberg::Schema::Make(
+      {iceberg::SchemaField::MakeOptional(
+          1, "profile",
+          iceberg::struct_({
+              iceberg::SchemaField::MakeRequired(2, "mystery", iceberg::unknown()),
+          }))},
+      100, std::vector<int32_t>{});
+
+  EXPECT_THAT(schema, iceberg::IsError(iceberg::ErrorKind::kInvalidSchema));
+  EXPECT_THAT(schema,
+              iceberg::HasErrorMessage("Unknown type field 'mystery' must be optional"));
+}
+
 TEST(SchemaTest, IdentifierFields) {
   using iceberg::ErrorKind;
   using iceberg::Schema;