Skip to content

Commit b3f9ce3

Browse files
manuzhangcodex
manuzhang
and
codex
committed
Reject null projections for required fields
Keep Avro and Parquet null physical fields from being coerced into null projections for required Iceberg fields, including required nested list elements. This preserves required-column semantics while still allowing optional unknown/null projections to materialize nulls. Co-authored-by: Codex <codex@openai.com>
1 parent 54a5dc8 commit b3f9ce3

4 files changed

Lines changed: 122 additions & 13 deletions

File tree

‎src/iceberg/avro/avro_schema_util.cc‎

Lines changed: 12 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -629,15 +629,20 @@ Result<FieldProjection> ProjectNested(const Type& expected_type,
629629
const ::avro::NodePtr& avro_node,
630630
bool prune_source);
631631

632-
Result<FieldProjection> ProjectField(const Type& expected_type,
632+
Result<FieldProjection> ProjectField(const SchemaField& expected_field,
633633
const ::avro::NodePtr& avro_node,
634634
size_t source_index, bool prune_source) {
635+
const Type& expected_type = *expected_field.type();
635636
::avro::NodePtr field_node;
636637
ICEBERG_RETURN_UNEXPECTED(UnwrapUnion(avro_node, &field_node));
637638

638639
FieldProjection projection;
639640
if (expected_type.type_id() == TypeId::kUnknown ||
640641
field_node->type() == ::avro::AVRO_NULL) {
642+
if (!expected_field.optional()) {
643+
return InvalidSchema("Cannot project required field with ID: {} as null",
644+
expected_field.field_id());
645+
}
641646
projection.kind = FieldProjection::Kind::kNull;
642647
return projection;
643648
}
@@ -688,9 +693,9 @@ Result<FieldProjection> ProjectStruct(const StructType& struct_type,
688693
FieldProjection child_projection;
689694

690695
if (auto iter = node_info_map.find(field_id); iter != node_info_map.cend()) {
691-
ICEBERG_ASSIGN_OR_RAISE(
692-
child_projection, ProjectField(*expected_field.type(), iter->second.field_node,
693-
iter->second.local_index, prune_source));
696+
ICEBERG_ASSIGN_OR_RAISE(child_projection,
697+
ProjectField(expected_field, iter->second.field_node,
698+
iter->second.local_index, prune_source));
694699
} else if (MetadataColumns::IsMetadataColumn(field_id)) {
695700
child_projection.kind = FieldProjection::Kind::kMetadata;
696701
} else if (expected_field.optional()) {
@@ -728,8 +733,8 @@ Result<FieldProjection> ProjectList(const ListType& list_type,
728733

729734
FieldProjection element_projection;
730735
ICEBERG_ASSIGN_OR_RAISE(element_projection,
731-
ProjectField(*expected_element_field.type(),
732-
avro_node->leafAt(0), size_t{0}, prune_source));
736+
ProjectField(expected_element_field, avro_node->leafAt(0),
737+
size_t{0}, prune_source));
733738

734739
FieldProjection result;
735740
result.children.emplace_back(std::move(element_projection));
@@ -788,7 +793,7 @@ Result<FieldProjection> ProjectMap(const MapType& map_type,
788793
const auto& expected_sub_field = map_type.fields()[i];
789794
ICEBERG_ASSIGN_OR_RAISE(
790795
sub_projection,
791-
ProjectField(*expected_sub_field.type(), map_node->leafAt(i), i, prune_source));
796+
ProjectField(expected_sub_field, map_node->leafAt(i), i, prune_source));
792797
result.children.emplace_back(std::move(sub_projection));
793798
}
794799

‎src/iceberg/parquet/parquet_schema_util.cc‎

Lines changed: 10 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -208,13 +208,18 @@ Result<FieldProjection> ProjectNested(
208208
const Type& nested_type,
209209
const std::vector<::parquet::arrow::SchemaField>& parquet_fields);
210210

211-
Result<FieldProjection> ProjectField(const Type& expected_type,
211+
Result<FieldProjection> ProjectField(const SchemaField& expected_field,
212212
const ::parquet::arrow::SchemaField& parquet_field,
213213
size_t source_index) {
214+
const Type& expected_type = *expected_field.type();
214215
ICEBERG_RETURN_UNEXPECTED(ValidateParquetSchemaEvolution(expected_type, parquet_field));
215216

216217
FieldProjection projection;
217218
if (expected_type.type_id() == TypeId::kUnknown || IsNullPhysicalField(parquet_field)) {
219+
if (!expected_field.optional()) {
220+
return InvalidSchema("Cannot project required field with id {} as null",
221+
expected_field.field_id());
222+
}
218223
projection.kind = FieldProjection::Kind::kNull;
219224
return projection;
220225
}
@@ -265,8 +270,8 @@ Result<FieldProjection> ProjectStruct(
265270

266271
if (auto iter = field_context_map.find(field_id); iter != field_context_map.cend()) {
267272
const auto& parquet_field = iter->second.parquet_field;
268-
ICEBERG_ASSIGN_OR_RAISE(child_projection, ProjectField(*field.type(), parquet_field,
269-
iter->second.local_index));
273+
ICEBERG_ASSIGN_OR_RAISE(
274+
child_projection, ProjectField(field, parquet_field, iter->second.local_index));
270275
} else if (MetadataColumns::IsMetadataColumn(field_id)) {
271276
child_projection.kind = FieldProjection::Kind::kMetadata;
272277
} else if (field.optional()) {
@@ -303,7 +308,7 @@ Result<FieldProjection> ProjectList(
303308
}
304309

305310
ICEBERG_ASSIGN_OR_RAISE(auto element_projection,
306-
ProjectField(*element_field.type(), parquet_field, size_t{0}));
311+
ProjectField(element_field, parquet_field, size_t{0}));
307312

308313
FieldProjection result;
309314
result.children.emplace_back(std::move(element_projection));
@@ -344,8 +349,7 @@ Result<FieldProjection> ProjectMap(
344349
for (size_t i = 0; i < parquet_fields.size(); ++i) {
345350
const auto& sub_node = parquet_fields[i];
346351
const auto& sub_field = map_type.fields()[i];
347-
ICEBERG_ASSIGN_OR_RAISE(auto sub_projection,
348-
ProjectField(*sub_field.type(), sub_node, i));
352+
ICEBERG_ASSIGN_OR_RAISE(auto sub_projection, ProjectField(sub_field, sub_node, i));
349353
result.children.emplace_back(std::move(sub_projection));
350354
}
351355

‎src/iceberg/test/avro_schema_test.cc‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -928,6 +928,55 @@ TEST(AvroSchemaProjectionTest, ProjectUnknownExpectedFieldAsNull) {
928928
ASSERT_EQ(projection.fields[0].kind, FieldProjection::Kind::kNull);
929929
}
930930

931+
TEST(AvroSchemaProjectionTest, RejectNullLeafForRequiredField) {
932+
Schema expected_schema({
933+
SchemaField::MakeRequired(/*field_id=*/1, "value", iceberg::int32()),
934+
});
935+
936+
std::string avro_schema_json = R"({
937+
"type": "record",
938+
"name": "iceberg_schema",
939+
"fields": [
940+
{"name": "value", "type": "null", "field-id": 1}
941+
]
942+
})";
943+
auto avro_schema = ::avro::compileJsonSchemaFromString(avro_schema_json);
944+
945+
auto projection_result =
946+
Project(expected_schema, avro_schema.root(), /*prune_source=*/false);
947+
ASSERT_THAT(projection_result, IsError(ErrorKind::kInvalidSchema));
948+
ASSERT_THAT(projection_result,
949+
HasErrorMessage("Cannot project required field with ID: 1 as null"));
950+
}
951+
952+
TEST(AvroSchemaProjectionTest, RejectNullListElementForRequiredElement) {
953+
Schema expected_schema({
954+
SchemaField::MakeOptional(
955+
/*field_id=*/1, "numbers",
956+
std::make_shared<ListType>(SchemaField::MakeRequired(
957+
/*field_id=*/101, "element", iceberg::int32()))),
958+
});
959+
960+
std::string avro_schema_json = R"({
961+
"type": "record",
962+
"name": "iceberg_schema",
963+
"fields": [
964+
{"name": "numbers", "type": ["null", {
965+
"type": "array",
966+
"items": "null",
967+
"element-id": 101
968+
}], "field-id": 1}
969+
]
970+
})";
971+
auto avro_schema = ::avro::compileJsonSchemaFromString(avro_schema_json);
972+
973+
auto projection_result =
974+
Project(expected_schema, avro_schema.root(), /*prune_source=*/false);
975+
ASSERT_THAT(projection_result, IsError(ErrorKind::kInvalidSchema));
976+
ASSERT_THAT(projection_result,
977+
HasErrorMessage("Cannot project required field with ID: 101 as null"));
978+
}
979+
931980
TEST(AvroSchemaProjectionTest, ProjectSchemaEvolutionIncompatibleTypes) {
932981
// Create iceberg schema expecting an int
933982
Schema expected_schema({

‎src/iceberg/test/parquet_schema_test.cc‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -137,6 +137,21 @@ ::parquet::arrow::SchemaField MakeNullSchemaField(const std::string& name, int f
137137
return schema_field;
138138
}
139139

140+
::parquet::arrow::SchemaField MakeListSchemaFieldWithNullElement(const std::string& name,
141+
int field_id,
142+
int element_field_id) {
143+
::parquet::arrow::SchemaField element_field =
144+
MakeNullSchemaField("element", element_field_id);
145+
146+
::parquet::arrow::SchemaField schema_field;
147+
schema_field.field =
148+
::arrow::field(name, ::arrow::list(element_field.field))
149+
->WithMetadata(::arrow::key_value_metadata({std::string(kParquetFieldIdKey)},
150+
{std::to_string(field_id)}));
151+
schema_field.children = {std::move(element_field)};
152+
return schema_field;
153+
}
154+
140155
#define ASSERT_PROJECTED_FIELD(field_projection, index) \
141156
ASSERT_EQ(field_projection.kind, FieldProjection::Kind::kProjected); \
142157
ASSERT_EQ(std::get<1>(field_projection.from), index);
@@ -365,6 +380,42 @@ TEST(ParquetSchemaProjectionTest, ProjectNullPhysicalFieldsAsNull) {
365380
ASSERT_TRUE(SelectedColumnIndices(projection).empty());
366381
}
367382

383+
TEST(ParquetSchemaProjectionTest, RejectNullPhysicalFieldForRequiredField) {
384+
Schema expected_schema({
385+
SchemaField::MakeRequired(/*field_id=*/1, "age", iceberg::int32()),
386+
});
387+
388+
::parquet::arrow::SchemaManifest schema_manifest;
389+
schema_manifest.schema_fields = {
390+
MakeNullSchemaField("age", /*field_id=*/1),
391+
};
392+
393+
auto projection_result = Project(expected_schema, schema_manifest);
394+
ASSERT_THAT(projection_result, IsError(ErrorKind::kInvalidSchema));
395+
ASSERT_THAT(projection_result,
396+
HasErrorMessage("Cannot project required field with id 1 as null"));
397+
}
398+
399+
TEST(ParquetSchemaProjectionTest, RejectNullPhysicalListElementForRequiredElement) {
400+
Schema expected_schema({
401+
SchemaField::MakeOptional(
402+
/*field_id=*/1, "numbers",
403+
std::make_shared<ListType>(SchemaField::MakeRequired(
404+
/*field_id=*/101, "element", iceberg::int32()))),
405+
});
406+
407+
::parquet::arrow::SchemaManifest schema_manifest;
408+
schema_manifest.schema_fields = {
409+
MakeListSchemaFieldWithNullElement("numbers", /*field_id=*/1,
410+
/*element_field_id=*/101),
411+
};
412+
413+
auto projection_result = Project(expected_schema, schema_manifest);
414+
ASSERT_THAT(projection_result, IsError(ErrorKind::kInvalidSchema));
415+
ASSERT_THAT(projection_result,
416+
HasErrorMessage("Cannot project required field with id 101 as null"));
417+
}
418+
368419
TEST(ParquetSchemaProjectionTest, ProjectUnknownExpectedFieldAsNull) {
369420
Schema expected_schema({
370421
SchemaField::MakeOptional(/*field_id=*/1, "mystery", iceberg::unknown()),

0 commit comments

Comments
 (0)