Skip to content

Commit 959fda6

Browse files
zhjwpkuzhjwpku
authored
feat: add Iceberg v3 type definitions (#752)
Introduce the Iceberg v3 types (variant, geometry, geography), including their schema/JSON serialization and type-system integration (visitors, schema projection, etc.). Reading and writing data of these types is not implemented yet: conversion to/from Arrow, Avro, and Parquet returns an error, as do identity transform binding and scalar validation for them.
1 parent f38090e commit 959fda6

28 files changed

Lines changed: 859 additions & 126 deletions

‎src/iceberg/avro/avro_schema_util.cc‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -248,6 +248,18 @@ Status ToAvroNodeVisitor::Visit(const UnknownType&, ::avro::NodePtr* node) {
248248
return {};
249249
}
250250

251+
Status ToAvroNodeVisitor::Visit(const VariantType&, ::avro::NodePtr*) {
252+
return NotSupported("Writing Iceberg variant type to Avro is not supported");
253+
}
254+
255+
Status ToAvroNodeVisitor::Visit(const GeometryType&, ::avro::NodePtr*) {
256+
return NotSupported("Writing Iceberg geometry type to Avro is not supported");
257+
}
258+
259+
Status ToAvroNodeVisitor::Visit(const GeographyType&, ::avro::NodePtr*) {
260+
return NotSupported("Writing Iceberg geography type to Avro is not supported");
261+
}
262+
251263
Status ToAvroNodeVisitor::Visit(const StructType& type, ::avro::NodePtr* node) {
252264
*node = std::make_shared<::avro::NodeRecord>();
253265

@@ -631,6 +643,11 @@ Status ValidateAvroSchemaEvolution(const Type& expected_type,
631643
break;
632644
case TypeId::kUnknown:
633645
return {};
646+
case TypeId::kVariant:
647+
case TypeId::kGeometry:
648+
case TypeId::kGeography:
649+
return NotSupported("Reading Iceberg type {} from Avro is not supported",
650+
expected_type);
634651
default:
635652
break;
636653
}

‎src/iceberg/avro/avro_schema_util_internal.h‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,9 @@ class ToAvroNodeVisitor {
5959
Status Visit(const FixedType& type, ::avro::NodePtr* node);
6060
Status Visit(const BinaryType& type, ::avro::NodePtr* node);
6161
Status Visit(const UnknownType&, ::avro::NodePtr*);
62+
Status Visit(const VariantType&, ::avro::NodePtr*);
63+
Status Visit(const GeometryType&, ::avro::NodePtr*);
64+
Status Visit(const GeographyType&, ::avro::NodePtr*);
6265
Status Visit(const StructType& type, ::avro::NodePtr* node);
6366
Status Visit(const ListType& type, ::avro::NodePtr* node);
6467
Status Visit(const MapType& type, ::avro::NodePtr* node);

‎src/iceberg/delete_file_index.cc‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,7 @@ Status EqualityDeleteFile::ConvertBoundsIfNeeded() const {
5656
}
5757

5858
const auto& schema_field = field.value().get();
59-
if (schema_field.type()->is_nested()) {
59+
if (!schema_field.type()->is_primitive()) {
6060
continue;
6161
}
6262

@@ -103,7 +103,7 @@ Result<bool> CanContainEqDeletesForFile(const DataFile& data_file,
103103
}
104104

105105
const auto& field = found_field.value().get();
106-
if (field.type()->is_nested()) {
106+
if (!field.type()->is_primitive()) {
107107
continue;
108108
}
109109

‎src/iceberg/json_serde.cc‎

Lines changed: 88 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -389,6 +389,12 @@ nlohmann::json ToJson(const Type& type) {
389389
return "uuid";
390390
case TypeId::kUnknown:
391391
return "unknown";
392+
case TypeId::kVariant:
393+
return "variant";
394+
case TypeId::kGeometry:
395+
return type.ToString();
396+
case TypeId::kGeography:
397+
return type.ToString();
392398
}
393399
std::unreachable();
394400
}
@@ -459,9 +465,10 @@ Result<std::unique_ptr<Type>> ListTypeFromJson(const nlohmann::json& json) {
459465
ICEBERG_ASSIGN_OR_RAISE(auto element_required,
460466
GetJsonValue<bool>(json, kElementRequired));
461467

462-
return std::make_unique<ListType>(
463-
SchemaField(element_id, std::string(ListType::kElementName),
464-
std::move(element_type), !element_required));
468+
ICEBERG_ASSIGN_OR_RAISE(auto type, ListType::Make(SchemaField(
469+
element_id, std::string(ListType::kElementName),
470+
std::move(element_type), !element_required)));
471+
return std::unique_ptr<Type>(std::move(type));
465472
}
466473

467474
Result<std::unique_ptr<Type>> MapTypeFromJson(const nlohmann::json& json) {
@@ -478,79 +485,126 @@ Result<std::unique_ptr<Type>> MapTypeFromJson(const nlohmann::json& json) {
478485
/*optional=*/false);
479486
SchemaField value_field(value_id, std::string(MapType::kValueName),
480487
std::move(value_type), !value_required);
481-
return std::make_unique<MapType>(std::move(key_field), std::move(value_field));
488+
ICEBERG_ASSIGN_OR_RAISE(auto type,
489+
MapType::Make(std::move(key_field), std::move(value_field)));
490+
return std::unique_ptr<Type>(std::move(type));
482491
}
483492

484493
} // namespace
485494

486495
Result<std::unique_ptr<Type>> TypeFromJson(const nlohmann::json& json) {
487496
if (json.is_string()) {
488-
std::string type_str = json.get<std::string>();
489-
if (type_str == "boolean") {
497+
const auto type_name = json.get<std::string>();
498+
const auto normalized_type_name = StringUtils::ToLower(type_name);
499+
if (normalized_type_name == "boolean") {
490500
return std::make_unique<BooleanType>();
491-
} else if (type_str == "int") {
501+
} else if (normalized_type_name == "int") {
492502
return std::make_unique<IntType>();
493-
} else if (type_str == "long") {
503+
} else if (normalized_type_name == "long") {
494504
return std::make_unique<LongType>();
495-
} else if (type_str == "float") {
505+
} else if (normalized_type_name == "float") {
496506
return std::make_unique<FloatType>();
497-
} else if (type_str == "double") {
507+
} else if (normalized_type_name == "double") {
498508
return std::make_unique<DoubleType>();
499-
} else if (type_str == "date") {
509+
} else if (normalized_type_name == "date") {
500510
return std::make_unique<DateType>();
501-
} else if (type_str == "time") {
511+
} else if (normalized_type_name == "time") {
502512
return std::make_unique<TimeType>();
503-
} else if (type_str == "timestamp") {
513+
} else if (normalized_type_name == "timestamp") {
504514
return std::make_unique<TimestampType>();
505-
} else if (type_str == "timestamptz") {
515+
} else if (normalized_type_name == "timestamptz") {
506516
return std::make_unique<TimestampTzType>();
507-
} else if (type_str == "timestamp_ns") {
517+
} else if (normalized_type_name == "timestamp_ns") {
508518
return std::make_unique<TimestampNsType>();
509-
} else if (type_str == "timestamptz_ns") {
519+
} else if (normalized_type_name == "timestamptz_ns") {
510520
return std::make_unique<TimestampTzNsType>();
511-
} else if (type_str == "string") {
521+
} else if (normalized_type_name == "string") {
512522
return std::make_unique<StringType>();
513-
} else if (type_str == "binary") {
523+
} else if (normalized_type_name == "binary") {
514524
return std::make_unique<BinaryType>();
515-
} else if (type_str == "uuid") {
525+
} else if (normalized_type_name == "uuid") {
516526
return std::make_unique<UuidType>();
517-
} else if (type_str == "unknown") {
527+
} else if (normalized_type_name == "unknown") {
518528
return std::make_unique<UnknownType>();
519-
} else if (type_str.starts_with("fixed")) {
520-
std::regex fixed_regex(R"(fixed\[\s*(\d+)\s*\])");
529+
} else if (normalized_type_name == "variant") {
530+
return std::make_unique<VariantType>();
531+
} else if (normalized_type_name.starts_with("fixed")) {
532+
static const std::regex kFixedRegex(R"(fixed\[\s*(\d+)\s*\])");
521533
std::smatch match;
522-
if (std::regex_match(type_str, match, fixed_regex)) {
534+
if (std::regex_match(normalized_type_name, match, kFixedRegex)) {
523535
ICEBERG_ASSIGN_OR_RAISE(auto length,
524536
StringUtils::ParseNumber<int32_t>(match[1].str()));
525537
return std::make_unique<FixedType>(length);
526538
}
527-
return JsonParseError("Invalid fixed type: {}", type_str);
528-
} else if (type_str.starts_with("decimal")) {
529-
std::regex decimal_regex(R"(decimal\(\s*(\d+)\s*,\s*(\d+)\s*\))");
539+
return JsonParseError("Invalid fixed type: {}", type_name);
540+
} else if (normalized_type_name.starts_with("decimal")) {
541+
static const std::regex kDecimalRegex(R"(decimal\(\s*(\d+)\s*,\s*(\d+)\s*\))");
530542
std::smatch match;
531-
if (std::regex_match(type_str, match, decimal_regex)) {
543+
if (std::regex_match(normalized_type_name, match, kDecimalRegex)) {
532544
ICEBERG_ASSIGN_OR_RAISE(auto precision,
533545
StringUtils::ParseNumber<int32_t>(match[1].str()));
534546
ICEBERG_ASSIGN_OR_RAISE(auto scale,
535547
StringUtils::ParseNumber<int32_t>(match[2].str()));
536548
return std::make_unique<DecimalType>(precision, scale);
537549
}
538-
return JsonParseError("Invalid decimal type: {}", type_str);
550+
return JsonParseError("Invalid decimal type: {}", type_name);
551+
} else if (normalized_type_name.starts_with("geometry")) {
552+
static const std::regex kGeometryRegex(R"(geometry\s*(?:\(\s*([^)]*?)\s*\))?)",
553+
std::regex_constants::icase);
554+
std::smatch match;
555+
if (std::regex_match(type_name, match, kGeometryRegex)) {
556+
if (match[1].matched) {
557+
auto crs = match[1].str();
558+
if (crs.empty()) {
559+
return JsonParseError("Invalid geometry type: {}", type_name);
560+
}
561+
ICEBERG_ASSIGN_OR_RAISE(auto type, GeometryType::Make(std::move(crs)));
562+
return std::unique_ptr<Type>(std::move(type));
563+
}
564+
ICEBERG_ASSIGN_OR_RAISE(auto type, GeometryType::Make());
565+
return std::unique_ptr<Type>(std::move(type));
566+
}
567+
return JsonParseError("Invalid geometry type: {}", type_name);
568+
} else if (normalized_type_name.starts_with("geography")) {
569+
static const std::regex kGeographyRegex(
570+
R"(geography\s*(?:\(\s*([^,]*?)\s*(?:,\s*(\w*)\s*)?\))?)",
571+
std::regex_constants::icase);
572+
std::smatch match;
573+
if (std::regex_match(type_name, match, kGeographyRegex)) {
574+
auto crs = match[1].str();
575+
if (match[1].matched && crs.empty()) {
576+
return JsonParseError("Invalid geography type: {}", type_name);
577+
}
578+
if (match[2].matched) {
579+
ICEBERG_ASSIGN_OR_RAISE(auto algorithm,
580+
EdgeAlgorithmFromString(match[2].str()));
581+
ICEBERG_ASSIGN_OR_RAISE(auto type,
582+
GeographyType::Make(std::move(crs), algorithm));
583+
return std::unique_ptr<Type>(std::move(type));
584+
}
585+
if (match[1].matched) {
586+
ICEBERG_ASSIGN_OR_RAISE(auto type, GeographyType::Make(std::move(crs)));
587+
return std::unique_ptr<Type>(std::move(type));
588+
}
589+
ICEBERG_ASSIGN_OR_RAISE(auto type, GeographyType::Make());
590+
return std::unique_ptr<Type>(std::move(type));
591+
}
592+
return JsonParseError("Invalid geography type: {}", type_name);
539593
} else {
540-
return JsonParseError("Unknown primitive type: {}", type_str);
594+
return JsonParseError("Cannot parse type string: {}", type_name);
541595
}
542596
}
543597

544598
// For complex types like struct, list, and map
545-
ICEBERG_ASSIGN_OR_RAISE(auto type_str, GetJsonValue<std::string>(json, kType));
546-
if (type_str == kStruct) {
599+
ICEBERG_ASSIGN_OR_RAISE(auto complex_type_name, GetJsonValue<std::string>(json, kType));
600+
if (complex_type_name == kStruct) {
547601
return StructTypeFromJson(json);
548-
} else if (type_str == kList) {
602+
} else if (complex_type_name == kList) {
549603
return ListTypeFromJson(json);
550-
} else if (type_str == kMap) {
604+
} else if (complex_type_name == kMap) {
551605
return MapTypeFromJson(json);
552606
} else {
553-
return JsonParseError("Unknown complex type: {}", type_str);
607+
return JsonParseError("Unknown complex type: {}", complex_type_name);
554608
}
555609
}
556610

‎src/iceberg/metrics_config.cc‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -197,6 +197,8 @@ Result<std::unordered_set<int32_t>> MetricsConfig::LimitFieldIds(const Schema& s
197197
Status Visit(const Type& type) {
198198
if (type.is_nested()) {
199199
return VisitNested(internal::checked_cast<const NestedType&>(type));
200+
} else if (type.is_variant()) {
201+
return {};
200202
} else {
201203
return VisitPrimitive(internal::checked_cast<const PrimitiveType&>(type));
202204
}
@@ -207,8 +209,7 @@ Result<std::unordered_set<int32_t>> MetricsConfig::LimitFieldIds(const Schema& s
207209
if (!ShouldContinue()) {
208210
break;
209211
}
210-
// TODO(zhuo.wang): variant type should also be handled here
211-
if (field.type()->is_primitive()) {
212+
if (!field.type()->is_nested()) {
212213
ids_.insert(field.field_id());
213214
}
214215
}

‎src/iceberg/parquet/parquet_metrics.cc‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -423,6 +423,10 @@ class CollectMetricsVisitor {
423423

424424
Status VisitMap(const MapType& /*type*/, const std::string& /*prefix*/) { return {}; }
425425

426+
Status VisitVariant(const VariantType& /*type*/, const std::string& /*prefix*/) {
427+
return {};
428+
}
429+
426430
Status VisitPrimitive(const PrimitiveType& /*type*/, const std::string& /*prefix*/) {
427431
return {};
428432
}

‎src/iceberg/parquet/parquet_schema_util.cc‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -239,6 +239,11 @@ Status ValidateParquetSchemaEvolution(
239239
break;
240240
case TypeId::kUnknown:
241241
return {};
242+
case TypeId::kVariant:
243+
case TypeId::kGeometry:
244+
case TypeId::kGeography:
245+
return NotSupported("Reading Iceberg type {} from Parquet is not supported",
246+
expected_type);
242247
case TypeId::kStruct:
243248
if (arrow_type->id() == ::arrow::Type::STRUCT) {
244249
return {};

‎src/iceberg/parquet/parquet_writer.cc‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -178,6 +178,10 @@ class FieldMetricsCollector {
178178

179179
Status VisitMap(const MapType& /*type*/, const ::arrow::Array& /*array*/) { return {}; }
180180

181+
Status VisitVariant(const VariantType& /*type*/, const ::arrow::Array& /*array*/) {
182+
return {};
183+
}
184+
181185
Status VisitPrimitive(const PrimitiveType& type, const ::arrow::Array& array) {
182186
switch (type.type_id()) {
183187
case TypeId::kFloat:

‎src/iceberg/schema_internal.cc‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919

2020
#include "iceberg/schema_internal.h"
2121

22+
#include <cerrno>
2223
#include <charconv>
2324
#include <cstring>
2425
#include <optional>
@@ -39,6 +40,33 @@ constexpr const char* kArrowExtensionMetadata = "ARROW:extension:metadata";
3940
constexpr const char* kArrowUuidExtensionName = "arrow.uuid";
4041
constexpr int32_t kUnknownFieldId = -1;
4142

43+
Status CheckArrowCompatible(const Type& type) {
44+
switch (type.type_id()) {
45+
case TypeId::kVariant:
46+
case TypeId::kGeometry:
47+
case TypeId::kGeography:
48+
return NotSupported("Iceberg type {} is not supported by Arrow conversion",
49+
type.ToString());
50+
case TypeId::kStruct:
51+
for (const auto& field : static_cast<const StructType&>(type).fields()) {
52+
ICEBERG_RETURN_UNEXPECTED(CheckArrowCompatible(*field.type()));
53+
}
54+
break;
55+
case TypeId::kList:
56+
ICEBERG_RETURN_UNEXPECTED(
57+
CheckArrowCompatible(*static_cast<const ListType&>(type).element().type()));
58+
break;
59+
case TypeId::kMap: {
60+
const auto& map_type = static_cast<const MapType&>(type);
61+
ICEBERG_RETURN_UNEXPECTED(CheckArrowCompatible(*map_type.key().type()));
62+
ICEBERG_RETURN_UNEXPECTED(CheckArrowCompatible(*map_type.value().type()));
63+
} break;
64+
default:
65+
break;
66+
}
67+
return {};
68+
}
69+
4270
// Convert an Iceberg type to Arrow schema. Return value is Nanoarrow error code.
4371
ArrowErrorCode ToArrowSchema(const Type& type, bool optional, std::string_view name,
4472
std::optional<int32_t> field_id, ArrowSchema* schema) {
@@ -153,6 +181,11 @@ ArrowErrorCode ToArrowSchema(const Type& type, bool optional, std::string_view n
153181
case TypeId::kUnknown:
154182
NANOARROW_RETURN_NOT_OK(ArrowSchemaSetType(schema, NANOARROW_TYPE_NA));
155183
break;
184+
case TypeId::kVariant:
185+
case TypeId::kGeometry:
186+
case TypeId::kGeography:
187+
ArrowBufferReset(&metadata_buffer);
188+
return EINVAL;
156189
}
157190

158191
if (!name.empty()) {
@@ -179,6 +212,8 @@ Status ToArrowSchema(const Schema& schema, ArrowSchema* out) {
179212
return InvalidArgument("Output Arrow schema cannot be null");
180213
}
181214

215+
ICEBERG_RETURN_UNEXPECTED(CheckArrowCompatible(schema));
216+
182217
ArrowSchemaInit(out);
183218

184219
if (ArrowErrorCode errorCode = ToArrowSchema(schema, /*optional=*/false, /*name=*/"",

‎src/iceberg/table_metadata.h‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -79,9 +79,8 @@ struct ICEBERG_EXPORT TableMetadata {
7979
static constexpr int64_t kInitialRowId = 0;
8080

8181
static inline const std::unordered_map<TypeId, int8_t> kMinFormatVersions = {
82-
{TypeId::kTimestampNs, 3},
83-
{TypeId::kTimestampTzNs, 3},
84-
{TypeId::kUnknown, 3},
82+
{TypeId::kTimestampNs, 3}, {TypeId::kTimestampTzNs, 3}, {TypeId::kUnknown, 3},
83+
{TypeId::kVariant, 3}, {TypeId::kGeometry, 3}, {TypeId::kGeography, 3},
8584
};
8685

8786
/// An integer version number for the format

0 commit comments

Comments
 (0)