Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions src/iceberg/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -244,6 +244,7 @@ if(ICEBERG_BUILD_BUNDLE)
arrow/arrow_io.cc
arrow/s3/arrow_s3_file_io.cc
arrow/arrow_register.cc
arrow/literal_util.cc
arrow/metadata_column_util.cc
avro/avro_data_util.cc
avro/avro_direct_decoder.cc
Expand Down
195 changes: 195 additions & 0 deletions src/iceberg/arrow/literal_util.cc
Original file line number Diff line number Diff line change
@@ -0,0 +1,195 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/

#include <cstring>
#include <string>
#include <utility>
#include <vector>

#include <arrow/array.h>
#include <arrow/array/builder_base.h>
#include <arrow/array/util.h>
#include <arrow/buffer.h>
#include <arrow/compute/api.h>
#include <arrow/scalar.h>
#include <arrow/type.h>

#include "iceberg/arrow/arrow_status_internal.h"
#include "iceberg/arrow/literal_util_internal.h"
#include "iceberg/type.h"
#include "iceberg/util/checked_cast.h"
#include "iceberg/util/formatter.h" // IWYU pragma: keep
#include "iceberg/util/macros.h"

namespace iceberg::arrow {

namespace {

Result<std::shared_ptr<::arrow::DataType>> ToArrowType(const PrimitiveType& type) {
switch (type.type_id()) {
case TypeId::kBoolean:
return ::arrow::boolean();
case TypeId::kInt:
return ::arrow::int32();
case TypeId::kLong:
return ::arrow::int64();
case TypeId::kFloat:
return ::arrow::float32();
case TypeId::kDouble:
return ::arrow::float64();
case TypeId::kDecimal: {
const DecimalType& decimal_type = internal::checked_cast<const DecimalType&>(type);
return ::arrow::decimal128(decimal_type.precision(), decimal_type.scale());
}
case TypeId::kDate:
return ::arrow::date32();
case TypeId::kTime:
return ::arrow::time64(::arrow::TimeUnit::MICRO);
case TypeId::kTimestamp:
return ::arrow::timestamp(::arrow::TimeUnit::MICRO);
case TypeId::kTimestampTz:
return ::arrow::timestamp(::arrow::TimeUnit::MICRO, "UTC");
case TypeId::kTimestampNs:
return ::arrow::timestamp(::arrow::TimeUnit::NANO);
case TypeId::kTimestampTzNs:
return ::arrow::timestamp(::arrow::TimeUnit::NANO, "UTC");
case TypeId::kString:
return ::arrow::utf8();
case TypeId::kBinary:
return ::arrow::binary();
case TypeId::kFixed: {
const FixedType& fixed_type = internal::checked_cast<const FixedType&>(type);
return ::arrow::fixed_size_binary(static_cast<int32_t>(fixed_type.length()));
}
case TypeId::kUuid:
return ::arrow::fixed_size_binary(16);
default:
return NotSupported("Cannot convert {} to an Arrow type", type);
}
}

Result<std::shared_ptr<::arrow::Buffer>> ToArrowBuffer(
const std::vector<uint8_t>& bytes) {
ICEBERG_ARROW_ASSIGN_OR_RETURN(std::unique_ptr<::arrow::Buffer> buffer,
::arrow::AllocateBuffer(bytes.size()));
std::memcpy(buffer->mutable_data(), bytes.data(), bytes.size());
return std::shared_ptr<::arrow::Buffer>(std::move(buffer));
}

} // namespace

Result<std::shared_ptr<::arrow::Scalar>> ToArrowScalar(const Literal& literal) {
if (literal.type() == nullptr) {
return InvalidArgument("Cannot convert a literal without type to an Arrow scalar");
}

if (literal.IsAboveMax() || literal.IsBelowMin()) {
return NotSupported("Cannot convert {} to an Arrow scalar", literal);
}

ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<::arrow::DataType> arrow_type,
ToArrowType(*literal.type()));
if (literal.IsNull()) {
return ::arrow::MakeNullScalar(std::move(arrow_type));
}

const Literal::Value& value = literal.value();
switch (literal.type()->type_id()) {
case TypeId::kBoolean:
return std::make_shared<::arrow::BooleanScalar>(std::get<bool>(value));
case TypeId::kInt:
return std::make_shared<::arrow::Int32Scalar>(std::get<int32_t>(value));
case TypeId::kLong:
return std::make_shared<::arrow::Int64Scalar>(std::get<int64_t>(value));
case TypeId::kFloat:
return std::make_shared<::arrow::FloatScalar>(std::get<float>(value));
case TypeId::kDouble:
return std::make_shared<::arrow::DoubleScalar>(std::get<double>(value));
case TypeId::kDecimal: {
const Decimal& decimal = std::get<Decimal>(value);
::arrow::Decimal128 arrow_decimal(
static_cast<int64_t>(decimal.value() >> 64),
static_cast<uint64_t>(decimal.value() & ~uint64_t{0}));
return std::make_shared<::arrow::Decimal128Scalar>(arrow_decimal,
std::move(arrow_type));
}
case TypeId::kDate:
return std::make_shared<::arrow::Date32Scalar>(std::get<int32_t>(value));
case TypeId::kTime:
return std::make_shared<::arrow::Time64Scalar>(std::get<int64_t>(value),
std::move(arrow_type));
case TypeId::kTimestamp:
case TypeId::kTimestampTz:
case TypeId::kTimestampNs:
case TypeId::kTimestampTzNs:
return std::make_shared<::arrow::TimestampScalar>(std::get<int64_t>(value),
std::move(arrow_type));
case TypeId::kString:
return std::make_shared<::arrow::StringScalar>(std::get<std::string>(value));
case TypeId::kBinary: {
ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<::arrow::Buffer> buffer,
ToArrowBuffer(std::get<std::vector<uint8_t>>(value)));
return std::make_shared<::arrow::BinaryScalar>(std::move(buffer));
}
case TypeId::kFixed: {
ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<::arrow::Buffer> buffer,
ToArrowBuffer(std::get<std::vector<uint8_t>>(value)));
return std::make_shared<::arrow::FixedSizeBinaryScalar>(std::move(buffer),
std::move(arrow_type));
}
case TypeId::kUuid: {
const Uuid& uuid = std::get<Uuid>(value);
ICEBERG_ASSIGN_OR_RAISE(
std::shared_ptr<::arrow::Buffer> buffer,
ToArrowBuffer(std::vector<uint8_t>(uuid.bytes().begin(), uuid.bytes().end())));
return std::make_shared<::arrow::FixedSizeBinaryScalar>(std::move(buffer),
std::move(arrow_type));
}
default:
return NotSupported("Cannot convert {} literal to an Arrow scalar",
*literal.type());
}
}

Result<std::shared_ptr<::arrow::Array>> MakeDefaultArray(
const Literal& literal, const std::shared_ptr<::arrow::DataType>& type,
int64_t num_rows, ::arrow::MemoryPool* pool) {
ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<::arrow::Scalar> scalar,
ToArrowScalar(literal));
ICEBERG_ARROW_ASSIGN_OR_RETURN(std::shared_ptr<::arrow::Array> array,
::arrow::MakeArrayFromScalar(*scalar, num_rows, pool));
if (!array->type()->Equals(*type)) {
ICEBERG_ARROW_ASSIGN_OR_RETURN(::arrow::Datum cast_result,
::arrow::compute::Cast(array, type));
return cast_result.make_array();
}
return array;
}

Status AppendDefaultToBuilder(const Literal& literal, ::arrow::ArrayBuilder* builder) {
ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<::arrow::Scalar> scalar,
ToArrowScalar(literal));
if (!scalar->type->Equals(*builder->type())) {
ICEBERG_ARROW_ASSIGN_OR_RETURN(scalar, scalar->CastTo(builder->type()));
}
ICEBERG_ARROW_RETURN_NOT_OK(builder->AppendScalar(*scalar));
return {};
}

} // namespace iceberg::arrow
49 changes: 49 additions & 0 deletions src/iceberg/arrow/literal_util_internal.h
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/

#pragma once

#include <cstdint>
#include <memory>

#include <arrow/type_fwd.h>

#include "iceberg/expression/literal.h"
#include "iceberg/result.h"

namespace iceberg::arrow {

/// \brief Convert a primitive literal to an Arrow scalar of its canonical Arrow type.
///
/// A null literal converts to a null scalar of the corresponding Arrow type.
Result<std::shared_ptr<::arrow::Scalar>> ToArrowScalar(const Literal& literal);

/// \brief Create an Arrow array of `num_rows` rows where every row holds the literal
/// value, e.g. to materialize a missing column with a default value.
///
/// The array is cast to `type` when the literal's canonical Arrow type differs.
Result<std::shared_ptr<::arrow::Array>> MakeDefaultArray(
const Literal& literal, const std::shared_ptr<::arrow::DataType>& type,
int64_t num_rows, ::arrow::MemoryPool* pool);

/// \brief Append the literal value once to `builder`, e.g. to materialize a missing
/// field with a default value while building rows.
Status AppendDefaultToBuilder(const Literal& literal, ::arrow::ArrayBuilder* builder);

} // namespace iceberg::arrow
9 changes: 9 additions & 0 deletions src/iceberg/avro/avro_data_util.cc
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
#include <avro/Types.hh>

#include "iceberg/arrow/arrow_status_internal.h"
#include "iceberg/arrow/literal_util_internal.h"
#include "iceberg/avro/avro_data_util_internal.h"
#include "iceberg/avro/avro_schema_util_internal.h"
#include "iceberg/metadata_columns.h"
Expand Down Expand Up @@ -87,6 +88,9 @@ Status AppendStructToBuilder(const ::avro::NodePtr& avro_node,
metadata_context, field_builder));
} else if (field_projection.kind == FieldProjection::Kind::kNull) {
ICEBERG_ARROW_RETURN_NOT_OK(field_builder->AppendNull());
} else if (field_projection.kind == FieldProjection::Kind::kDefault) {
ICEBERG_RETURN_UNEXPECTED(arrow::AppendDefaultToBuilder(
std::get<Literal>(field_projection.from), field_builder));
} else if (field_projection.kind == FieldProjection::Kind::kMetadata) {
int32_t field_id = expected_field.field_id();
if (field_id == MetadataColumns::kFilePathColumnId) {
Expand Down Expand Up @@ -462,6 +466,11 @@ Status AppendFieldToBuilder(const ::avro::NodePtr& avro_node,
return {};
}

if (projection.kind == FieldProjection::Kind::kDefault) {
return arrow::AppendDefaultToBuilder(std::get<Literal>(projection.from),
array_builder);
}

if (avro_node->type() == ::avro::AVRO_UNION) {
size_t branch = avro_datum.unionBranch();
if (avro_node->leafAt(branch)->type() == ::avro::AVRO_NULL) {
Expand Down
4 changes: 4 additions & 0 deletions src/iceberg/avro/avro_direct_decoder.cc
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
#include <avro/Types.hh>

#include "iceberg/arrow/arrow_status_internal.h"
#include "iceberg/arrow/literal_util_internal.h"
#include "iceberg/avro/avro_direct_decoder_internal.h"
#include "iceberg/avro/avro_schema_util_internal.h"
#include "iceberg/metadata_columns.h"
Expand Down Expand Up @@ -209,6 +210,9 @@ Status DecodeStructToBuilder(const ::avro::NodePtr& avro_node, ::avro::Decoder&
auto* field_builder = struct_builder->field_builder(static_cast<int>(proj_idx));
if (field_projection.kind == FieldProjection::Kind::kNull) {
ICEBERG_ARROW_RETURN_NOT_OK(field_builder->AppendNull());
} else if (field_projection.kind == FieldProjection::Kind::kDefault) {
ICEBERG_RETURN_UNEXPECTED(arrow::AppendDefaultToBuilder(
std::get<Literal>(field_projection.from), field_builder));
} else if (field_projection.kind == FieldProjection::Kind::kMetadata) {
int32_t field_id = expected_field.field_id();
if (field_id == MetadataColumns::kFilePathColumnId) {
Expand Down
4 changes: 4 additions & 0 deletions src/iceberg/avro/avro_schema_util.cc
Original file line number Diff line number Diff line change
Expand Up @@ -730,6 +730,10 @@ Result<FieldProjection> ProjectStruct(const StructType& struct_type,
iter->second.local_index, prune_source));
} else if (MetadataColumns::IsMetadataColumn(field_id)) {
child_projection.kind = FieldProjection::Kind::kMetadata;
} else if (expected_field.initial_default().has_value()) {
// Rows written before the field existed assume its `initial-default` value.
child_projection.kind = FieldProjection::Kind::kDefault;
child_projection.from = expected_field.initial_default()->get();
} else if (expected_field.optional()) {
child_projection.kind = FieldProjection::Kind::kNull;
} else {
Expand Down
5 changes: 2 additions & 3 deletions src/iceberg/data/delete_filter.cc
Original file line number Diff line number Diff line change
Expand Up @@ -368,9 +368,8 @@ Result<bool> MergeField(SchemaField& existing, const SchemaField& required) {
if (!changed) {
return false;
}
existing = SchemaField(existing.field_id(), std::string(existing.name()),
std::make_shared<StructType>(std::move(fields)),
existing.optional(), std::string(existing.doc()));
existing = existing.WithIdAndType(existing.field_id(),
std::make_shared<StructType>(std::move(fields)));
return true;
}

Expand Down
32 changes: 28 additions & 4 deletions src/iceberg/json_serde.cc
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,8 @@
#include <nlohmann/json.hpp>

#include "iceberg/constants.h"
#include "iceberg/expression/json_serde_internal.h"
#include "iceberg/expression/literal.h"
#include "iceberg/json_serde_internal.h"
#include "iceberg/name_mapping.h"
#include "iceberg/partition_field.h"
Expand Down Expand Up @@ -298,6 +300,15 @@ nlohmann::json ToJson(const SchemaField& field) {
if (!field.doc().empty()) {
json[kDoc] = field.doc();
}
// Defaults are validated to be primitive literals matching the field type, so
// single-value serialization cannot fail here.
if (field.initial_default().has_value()) {
ICEBERG_ASSIGN_OR_THROW(json[kInitialDefault],
ToJson(field.initial_default()->get()));
}
if (field.write_default().has_value()) {
ICEBERG_ASSIGN_OR_THROW(json[kWriteDefault], ToJson(field.write_default()->get()));
}
return json;
}

Expand All @@ -310,7 +321,6 @@ nlohmann::json ToJson(const Type& type) {
nlohmann::json fields_json = nlohmann::json::array();
for (const auto& field : struct_type.fields()) {
fields_json.push_back(ToJson(field));
// TODO(gangwu): add default values
}
json[kFields] = fields_json;
return json;
Expand Down Expand Up @@ -552,9 +562,23 @@ Result<std::unique_ptr<SchemaField>> FieldFromJson(const nlohmann::json& json) {
ICEBERG_ASSIGN_OR_RAISE(auto name, GetJsonValue<std::string>(json, kName));
ICEBERG_ASSIGN_OR_RAISE(auto required, GetJsonValue<bool>(json, kRequired));
ICEBERG_ASSIGN_OR_RAISE(auto doc, GetJsonValueOrDefault<std::string>(json, kDoc));

return std::make_unique<SchemaField>(field_id, std::move(name), std::move(type),
!required, doc);
ICEBERG_ASSIGN_OR_RAISE(std::optional<nlohmann::json> initial_default_json,
GetJsonValueOptional<nlohmann::json>(json, kInitialDefault));
ICEBERG_ASSIGN_OR_RAISE(std::optional<nlohmann::json> write_default_json,
GetJsonValueOptional<nlohmann::json>(json, kWriteDefault));

SchemaField field(field_id, std::move(name), std::move(type), !required, doc);
if (initial_default_json.has_value()) {
ICEBERG_ASSIGN_OR_RAISE(Literal literal,
LiteralFromJson(*initial_default_json, field.type().get()));
field = field.WithInitialDefault(std::move(literal));
}
if (write_default_json.has_value()) {
ICEBERG_ASSIGN_OR_RAISE(Literal literal,
LiteralFromJson(*write_default_json, field.type().get()));
field = field.WithWriteDefault(std::move(literal));
}
return std::make_unique<SchemaField>(std::move(field));
}

Result<std::unique_ptr<Schema>> SchemaFromJson(const nlohmann::json& json) {
Expand Down
Loading