Skip to content

Commit 2b09f55

Browse files
zhjwpkuzhjwpku
committed
feat: add Iceberg v3 type definitions
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 fc9781f commit 2b09f55

28 files changed

Lines changed: 696 additions & 135 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: 67 additions & 22 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
}
@@ -486,58 +492,97 @@ Result<std::unique_ptr<Type>> MapTypeFromJson(const nlohmann::json& json) {
486492
Result<std::unique_ptr<Type>> TypeFromJson(const nlohmann::json& json) {
487493
if (json.is_string()) {
488494
std::string type_str = json.get<std::string>();
489-
if (type_str == "boolean") {
495+
std::string lower_type_str = StringUtils::ToLower(type_str);
496+
if (lower_type_str == "boolean") {
490497
return std::make_unique<BooleanType>();
491-
} else if (type_str == "int") {
498+
} else if (lower_type_str == "int") {
492499
return std::make_unique<IntType>();
493-
} else if (type_str == "long") {
500+
} else if (lower_type_str == "long") {
494501
return std::make_unique<LongType>();
495-
} else if (type_str == "float") {
502+
} else if (lower_type_str == "float") {
496503
return std::make_unique<FloatType>();
497-
} else if (type_str == "double") {
504+
} else if (lower_type_str == "double") {
498505
return std::make_unique<DoubleType>();
499-
} else if (type_str == "date") {
506+
} else if (lower_type_str == "date") {
500507
return std::make_unique<DateType>();
501-
} else if (type_str == "time") {
508+
} else if (lower_type_str == "time") {
502509
return std::make_unique<TimeType>();
503-
} else if (type_str == "timestamp") {
510+
} else if (lower_type_str == "timestamp") {
504511
return std::make_unique<TimestampType>();
505-
} else if (type_str == "timestamptz") {
512+
} else if (lower_type_str == "timestamptz") {
506513
return std::make_unique<TimestampTzType>();
507-
} else if (type_str == "timestamp_ns") {
514+
} else if (lower_type_str == "timestamp_ns") {
508515
return std::make_unique<TimestampNsType>();
509-
} else if (type_str == "timestamptz_ns") {
516+
} else if (lower_type_str == "timestamptz_ns") {
510517
return std::make_unique<TimestampTzNsType>();
511-
} else if (type_str == "string") {
518+
} else if (lower_type_str == "string") {
512519
return std::make_unique<StringType>();
513-
} else if (type_str == "binary") {
520+
} else if (lower_type_str == "binary") {
514521
return std::make_unique<BinaryType>();
515-
} else if (type_str == "uuid") {
522+
} else if (lower_type_str == "uuid") {
516523
return std::make_unique<UuidType>();
517-
} else if (type_str == "unknown") {
524+
} else if (lower_type_str == "unknown") {
518525
return std::make_unique<UnknownType>();
519-
} else if (type_str.starts_with("fixed")) {
520-
std::regex fixed_regex(R"(fixed\[\s*(\d+)\s*\])");
526+
} else if (lower_type_str == "variant") {
527+
return std::make_unique<VariantType>();
528+
} else if (lower_type_str.starts_with("fixed")) {
529+
static const std::regex fixed_regex(R"(fixed\[\s*(\d+)\s*\])");
521530
std::smatch match;
522-
if (std::regex_match(type_str, match, fixed_regex)) {
531+
if (std::regex_match(lower_type_str, match, fixed_regex)) {
523532
ICEBERG_ASSIGN_OR_RAISE(auto length,
524533
StringUtils::ParseNumber<int32_t>(match[1].str()));
525534
return std::make_unique<FixedType>(length);
526535
}
527536
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*\))");
537+
} else if (lower_type_str.starts_with("decimal")) {
538+
static const std::regex decimal_regex(R"(decimal\(\s*(\d+)\s*,\s*(\d+)\s*\))");
530539
std::smatch match;
531-
if (std::regex_match(type_str, match, decimal_regex)) {
540+
if (std::regex_match(lower_type_str, match, decimal_regex)) {
532541
ICEBERG_ASSIGN_OR_RAISE(auto precision,
533542
StringUtils::ParseNumber<int32_t>(match[1].str()));
534543
ICEBERG_ASSIGN_OR_RAISE(auto scale,
535544
StringUtils::ParseNumber<int32_t>(match[2].str()));
536545
return std::make_unique<DecimalType>(precision, scale);
537546
}
538547
return JsonParseError("Invalid decimal type: {}", type_str);
548+
} else if (lower_type_str.starts_with("geometry")) {
549+
static const std::regex geometry_regex(R"(\s*(?:\(\s*([^)]*?)\s*\))?)");
550+
std::smatch match;
551+
const auto type_params = type_str.substr(std::string_view("geometry").size());
552+
if (std::regex_match(type_params, match, geometry_regex)) {
553+
if (match[1].matched) {
554+
auto crs = match[1].str();
555+
if (crs.empty()) {
556+
return JsonParseError("Invalid geometry type: {}", type_str);
557+
}
558+
return std::make_unique<GeometryType>(std::move(crs));
559+
}
560+
return std::make_unique<GeometryType>();
561+
}
562+
return JsonParseError("Invalid geometry type: {}", type_str);
563+
} else if (lower_type_str.starts_with("geography")) {
564+
static const std::regex geography_regex(
565+
R"(\s*(?:\(\s*([^,]*?)\s*(?:,\s*(\w*)\s*)?\))?)");
566+
std::smatch match;
567+
const auto type_params = type_str.substr(std::string_view("geography").size());
568+
if (std::regex_match(type_params, match, geography_regex)) {
569+
auto crs = match[1].str();
570+
if (match[1].matched && crs.empty()) {
571+
return JsonParseError("Invalid geography type: {}", type_str);
572+
}
573+
if (match[2].matched) {
574+
ICEBERG_ASSIGN_OR_RAISE(auto algorithm,
575+
EdgeAlgorithmFromString(match[2].str()));
576+
return std::make_unique<GeographyType>(std::move(crs), algorithm);
577+
}
578+
if (match[1].matched) {
579+
return std::make_unique<GeographyType>(std::move(crs));
580+
}
581+
return std::make_unique<GeographyType>();
582+
}
583+
return JsonParseError("Invalid geography type: {}", type_str);
539584
} else {
540-
return JsonParseError("Unknown primitive type: {}", type_str);
585+
return JsonParseError("Cannot parse type string: {}", type_str);
541586
}
542587
}
543588

‎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.type_id() == TypeId::kVariant) {
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:

0 commit comments

Comments
 (0)