This is an automated email from the ASF dual-hosted git repository.
wgtmac pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 942e9af0 feat: apply write defaults in DataWriter (#942)
942e9af0 is described below
commit 942e9af077cf5dacc7daf0eefe55fbe74597b4cf
Author: Xin Huang <[email protected]>
AuthorDate: Fri Oct 30 01:29:47 2026 -0700
feat: apply write defaults in DataWriter (#942)
Writing a batch with an older schema should fill omitted columns with
Iceberg
write-default values. This PR adds optional
`DataWriterOptions::input_schema`;
for a required column with initial-default 42 and write-default 7, input
`[[1], [2]]` writes `[[1, 7], [2, 7]]`.
DataWriter prepares a reusable `WriteProjection` once in `Make` and applies
it
to each batch. The projection matches fields by ID recursively, preserves
supplied values and explicit nulls, fills missing optional fields with typed
nulls, and materializes missing fields from their write-default. When no
`input_schema` is provided, or the schemas are identical, the original Arrow
batch is passed through without projection.
Incompatible schemas fail before opening the file, including extra input
IDs,
missing required fields without a write-default, optional-to-required
fields,
and concrete-to-unknown writes. Existing primitive types must match; read
projection behavior is unchanged.
Parquet and Avro writers remain unchanged and continue to expect batches
matching their complete target schema. The shared projection is implemented
in
core using nanoarrow shared buffers without adding an Arrow C++ dependency.
Addresses the write-path portion of #637.
---
src/iceberg/arrow_c_data_util.cc | 179 ++++++++++++++++++
src/iceberg/arrow_c_data_util_internal.h | 21 +++
src/iceberg/data/data_writer.cc | 24 ++-
src/iceberg/data/data_writer.h | 2 +
src/iceberg/schema_util.cc | 64 +++++--
src/iceberg/schema_util.h | 15 +-
src/iceberg/test/arrow_c_data_util_test.cc | 283 +++++++++++++++++++++++++++++
src/iceberg/test/data_writer_test.cc | 37 ++++
src/iceberg/test/default_value_test.cc | 121 ++++++++++++
src/iceberg/test/schema_util_test.cc | 130 +++++++++++++
10 files changed, 856 insertions(+), 20 deletions(-)
diff --git a/src/iceberg/arrow_c_data_util.cc b/src/iceberg/arrow_c_data_util.cc
index a77012a6..eb7e90b7 100644
--- a/src/iceberg/arrow_c_data_util.cc
+++ b/src/iceberg/arrow_c_data_util.cc
@@ -17,25 +17,36 @@
* under the License.
*/
+#include <cerrno>
#include <cstdint>
#include <limits>
#include <memory>
#include <mutex>
#include <span>
+#include <string>
+#include <string_view>
+#include <type_traits>
#include <utility>
+#include <variant>
#include <vector>
#include <nanoarrow/nanoarrow.h>
#include "iceberg/arrow_c_data_guard_internal.h"
#include "iceberg/arrow_c_data_util_internal.h"
+#include "iceberg/expression/literal.h"
#include "iceberg/file_reader.h"
#include "iceberg/nanoarrow_status_internal.h"
#include "iceberg/result.h"
#include "iceberg/schema.h"
#include "iceberg/schema_field.h"
#include "iceberg/schema_internal.h"
+#include "iceberg/schema_util.h"
+#include "iceberg/type.h"
+#include "iceberg/util/checked_cast.h"
+#include "iceberg/util/decimal.h"
#include "iceberg/util/macros.h"
+#include "iceberg/util/uuid.h"
namespace iceberg {
@@ -394,4 +405,172 @@ Result<ArrowArray> ProjectBatch(ArrowArray* input_batch,
projection.selected_field_indices());
}
+namespace {
+
+int AppendWriteDefault(ArrowArray* array, const Literal& literal) {
+ return std::visit(
+ [&](const auto& value) -> int {
+ using Value = std::decay_t<decltype(value)>;
+ if constexpr (std::is_same_v<Value, std::monostate>) {
+ return ArrowArrayAppendNull(array, 1);
+ } else if constexpr (std::is_integral_v<Value>) {
+ return ArrowArrayAppendInt(array, value);
+ } else if constexpr (std::is_floating_point_v<Value>) {
+ return ArrowArrayAppendDouble(array, value);
+ } else if constexpr (std::is_same_v<Value, std::string>) {
+ return ArrowArrayAppendString(array,
+ ArrowStringView(value.data(),
value.size()));
+ } else if constexpr (std::is_same_v<Value, std::vector<uint8_t>>) {
+ ArrowBufferView bytes;
+ bytes.data.as_uint8 = value.data();
+ bytes.size_bytes = static_cast<int64_t>(value.size());
+ return ArrowArrayAppendBytes(array, bytes);
+ } else if constexpr (std::is_same_v<Value, Uuid>) {
+ ArrowBufferView bytes;
+ bytes.data.as_uint8 = value.bytes().data();
+ bytes.size_bytes = Uuid::kLength;
+ return ArrowArrayAppendBytes(array, bytes);
+ } else if constexpr (std::is_same_v<Value, Decimal>) {
+ const auto& type = internal::checked_cast<const
DecimalType&>(*literal.type());
+ ArrowDecimal decimal;
+ ArrowDecimalInit(&decimal, Decimal::kBitWidth, type.precision(),
type.scale());
+ ArrowDecimalSetBytes(&decimal, value.native_endian_bytes());
+ return ArrowArrayAppendDecimal(array, &decimal);
+ } else {
+ return EINVAL;
+ }
+ },
+ literal.value());
+}
+
+Result<ArrowArray> MaterializeWriteField(const FieldProjection& projection,
+ const ArrowSchema& schema, int64_t
length) {
+ ArrowArray result{};
+ ArrowError error;
+ ICEBERG_NANOARROW_RETURN_UNEXPECTED_WITH_ERROR(
+ ArrowArrayInitFromSchema(&result, &schema, &error), error);
+ internal::ArrowArrayGuard guard(&result);
+ ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowArrayStartAppending(&result));
+ if (projection.kind == FieldProjection::Kind::kNull) {
+ ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowArrayAppendNull(&result, length));
+ } else {
+ const auto& literal = std::get<Literal>(projection.from);
+ for (int64_t row = 0; row < length; ++row) {
+ ICEBERG_NANOARROW_RETURN_UNEXPECTED(AppendWriteDefault(&result,
literal));
+ }
+ }
+ ICEBERG_NANOARROW_RETURN_UNEXPECTED_WITH_ERROR(
+ ArrowArrayFinishBuildingDefault(&result, &error), error);
+ return std::exchange(result, ArrowArray{});
+}
+
+Result<ArrowArray> MaterializeWriteProjection(
+ ArrowArray& input, const ArrowSchema& schema,
+ std::span<const FieldProjection> projections) {
+ ArrowArray result{};
+ ICEBERG_NANOARROW_RETURN_UNEXPECTED(
+ ArrowArrayInitFromType(&result, NANOARROW_TYPE_UNINITIALIZED));
+ internal::ArrowArrayGuard guard(&result);
+ ICEBERG_NANOARROW_RETURN_UNEXPECTED(
+ ArrowArrayAllocateChildren(&result, schema.n_children));
+ result.n_buffers = input.n_buffers;
+ result.length = input.length;
+ result.offset = input.offset;
+ result.null_count = input.null_count;
+ // MoveShared makes every input buffer independently reference-counted.
Clone only
+ // the parent buffers; existing leaves can then move directly into the new
tree.
+ for (int64_t index = 0; index < input.n_buffers; ++index) {
+ ArrowBuffer buffer;
+ ArrowBufferInit(&buffer);
+ internal::ArrowArrayBufferGuard buffer_guard(&buffer);
+ ICEBERG_NANOARROW_RETURN_UNEXPECTED(
+ ArrowSharedBufferClone(ArrowArrayBuffer(&input, index), &buffer));
+ ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowArraySetBuffer(&result, index,
&buffer));
+ }
+ if (std::string_view(schema.format) == "+m") {
+ // Maps add an entries struct between the map and its key/value projection.
+ ICEBERG_ASSIGN_OR_RAISE(
+ *result.children[0],
+ MaterializeWriteProjection(*input.children[0], *schema.children[0],
projections));
+ } else {
+ const int64_t length = std::string_view(schema.format) == "+s"
+ ? input.offset + input.length
+ : input.children[0]->length;
+ for (size_t index = 0; index < projections.size(); ++index) {
+ const auto& projection = projections[index];
+ const auto& child_schema = *schema.children[index];
+ auto* output_child = result.children[index];
+ if (projection.kind == FieldProjection::Kind::kProjected) {
+ auto* child = input.children[std::get<size_t>(projection.from)];
+ if (projection.children.empty()) {
+ ArrowArrayMove(child, output_child);
+ } else {
+ ICEBERG_ASSIGN_OR_RAISE(
+ *output_child,
+ MaterializeWriteProjection(*child, child_schema,
projection.children));
+ }
+ } else {
+ ICEBERG_ASSIGN_OR_RAISE(*output_child,
+ MaterializeWriteField(projection,
child_schema, length));
+ }
+ }
+ }
+ return std::exchange(result, ArrowArray{});
+}
+
+} // namespace
+
+class WriteProjection::Impl {
+ public:
+ bool identical = false;
+ SchemaProjection projection;
+ ArrowSchema write_schema{};
+ ArrowArrayView input_view{};
+
+ ~Impl() {
+ ArrowArrayViewReset(&input_view);
+ if (write_schema.release != nullptr) {
+ ArrowSchemaRelease(&write_schema);
+ }
+ }
+};
+
+WriteProjection::WriteProjection(std::unique_ptr<Impl> impl) :
impl_(std::move(impl)) {}
+WriteProjection::~WriteProjection() = default;
+
+Result<std::unique_ptr<WriteProjection>> WriteProjection::Make(
+ const Schema& input_schema, const Schema& write_schema) {
+ auto impl = std::make_unique<Impl>();
+ impl->identical = input_schema == write_schema;
+ if (!impl->identical) {
+ ICEBERG_ASSIGN_OR_RAISE(impl->projection,
+ Project(write_schema, input_schema,
/*prune_source=*/false,
+ ProjectionMode::kWrite));
+ ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(write_schema,
&impl->write_schema));
+ ArrowSchema input_arrow_schema{};
+ ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(input_schema,
&input_arrow_schema));
+ internal::ArrowSchemaGuard guard(&input_arrow_schema);
+ ArrowError error;
+ ICEBERG_NANOARROW_RETURN_UNEXPECTED_WITH_ERROR(
+ ArrowArrayViewInitFromSchema(&impl->input_view, &input_arrow_schema,
&error),
+ error);
+ }
+ return std::unique_ptr<WriteProjection>(new
WriteProjection(std::move(impl)));
+}
+
+Result<ArrowArray> WriteProjection::AlignBatch(ArrowArray* input_batch) const {
+ if (impl_->identical) {
+ return std::exchange(*input_batch, ArrowArray{});
+ }
+ internal::ArrowArrayGuard input_guard(input_batch);
+ ArrowError error;
+ ICEBERG_NANOARROW_RETURN_UNEXPECTED_WITH_ERROR(
+ ArrowArrayViewSetArray(&impl_->input_view, input_batch, &error), error);
+ ArrowArray shared{};
+ internal::ArrowArrayGuard shared_guard(&shared);
+ ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowArrayMoveShared(input_batch,
&shared));
+ return MaterializeWriteProjection(shared, impl_->write_schema,
+ impl_->projection.fields);
+}
+
} // namespace iceberg
diff --git a/src/iceberg/arrow_c_data_util_internal.h
b/src/iceberg/arrow_c_data_util_internal.h
index e02db29a..899695b6 100644
--- a/src/iceberg/arrow_c_data_util_internal.h
+++ b/src/iceberg/arrow_c_data_util_internal.h
@@ -242,4 +242,25 @@ ICEBERG_EXPORT Result<ArrowArray> ProjectBatch(ArrowArray*
input_batch,
std::span<const int32_t>
row_indices,
ProjectionContext& projection);
+/// \brief A write projection prepared once for a stable input/write schema
pair.
+///
+/// Matches fields by id, fills missing fields with write-defaults or typed
nulls,
+/// and rejects incompatible schemas at Make. Existing values, including
explicit
+/// nulls, are preserved. Do not share a projection across concurrent calls.
+class ICEBERG_EXPORT WriteProjection {
+ public:
+ static Result<std::unique_ptr<WriteProjection>> Make(const Schema&
input_schema,
+ const Schema&
write_schema);
+ ~WriteProjection();
+
+ /// \brief Align a batch, consuming it even on failure.
+ /// Identical schemas transfer the original batch without conversion.
+ Result<ArrowArray> AlignBatch(ArrowArray* input_batch) const;
+
+ private:
+ class Impl;
+ explicit WriteProjection(std::unique_ptr<Impl> impl);
+ std::unique_ptr<Impl> impl_;
+};
+
} // namespace iceberg
diff --git a/src/iceberg/data/data_writer.cc b/src/iceberg/data/data_writer.cc
index 512f77cc..53fc7e4e 100644
--- a/src/iceberg/data/data_writer.cc
+++ b/src/iceberg/data/data_writer.cc
@@ -22,6 +22,8 @@
#include <map>
#include <utility>
+#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/arrow_c_data_util_internal.h"
#include "iceberg/data/writer.h" // IWYU pragma: keep
#include "iceberg/file_writer.h"
#include "iceberg/manifest/manifest_entry.h"
@@ -33,6 +35,11 @@ namespace iceberg {
class DataWriter::Impl {
public:
static Result<std::unique_ptr<Impl>> Make(DataWriterOptions options) {
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto projection,
+ WriteProjection::Make(
+ options.input_schema ? *options.input_schema : *options.schema,
+ *options.schema));
WriterOptions writer_options{
.path = options.path,
.schema = options.schema,
@@ -43,10 +50,15 @@ class DataWriter::Impl {
ICEBERG_ASSIGN_OR_RAISE(auto writer,
WriterFactoryRegistry::Open(options.format,
writer_options));
- return std::unique_ptr<Impl>(new Impl(std::move(options),
std::move(writer)));
+ return std::unique_ptr<Impl>(
+ new Impl(std::move(options), std::move(writer),
std::move(projection)));
}
- Status Write(ArrowArray* data) { return writer_->Write(data); }
+ Status Write(ArrowArray* data) {
+ ICEBERG_ASSIGN_OR_RAISE(auto aligned, projection_->AlignBatch(data));
+ internal::ArrowArrayGuard guard(&aligned);
+ return writer_->Write(&aligned);
+ }
Result<int64_t> Length() const { return writer_->length(); }
@@ -106,11 +118,15 @@ class DataWriter::Impl {
}
private:
- Impl(DataWriterOptions options, std::unique_ptr<Writer> writer)
- : options_(std::move(options)), writer_(std::move(writer)) {}
+ Impl(DataWriterOptions options, std::unique_ptr<Writer> writer,
+ std::unique_ptr<WriteProjection> projection)
+ : options_(std::move(options)),
+ writer_(std::move(writer)),
+ projection_(std::move(projection)) {}
DataWriterOptions options_;
std::unique_ptr<Writer> writer_;
+ std::unique_ptr<WriteProjection> projection_;
bool closed_ = false;
};
diff --git a/src/iceberg/data/data_writer.h b/src/iceberg/data/data_writer.h
index 9a2002f0..39d9f508 100644
--- a/src/iceberg/data/data_writer.h
+++ b/src/iceberg/data/data_writer.h
@@ -48,6 +48,8 @@ struct ICEBERG_DATA_EXPORT DataWriterOptions {
std::shared_ptr<FileIO> io;
std::optional<int32_t> sort_order_id;
std::unordered_map<std::string, std::string> properties;
+ /// \brief Schema of input batches; when omitted, batches use the target
schema.
+ std::shared_ptr<Schema> input_schema;
};
/// \brief Writer for Iceberg data files.
diff --git a/src/iceberg/schema_util.cc b/src/iceberg/schema_util.cc
index b9f32346..d7d26175 100644
--- a/src/iceberg/schema_util.cc
+++ b/src/iceberg/schema_util.cc
@@ -83,14 +83,25 @@ Status ValidateSchemaEvolution(const Type& expected_type,
const Type& source_typ
}
Result<FieldProjection> ProjectNested(const Type& expected_type, const Type&
source_type,
- bool prune_source);
+ bool prune_source, ProjectionMode mode);
Result<FieldProjection> ProjectField(const SchemaField& expected_field,
const SchemaField& source_field, size_t
source_index,
- bool prune_source) {
+ bool prune_source, ProjectionMode mode) {
FieldProjection projection;
+ if (mode == ProjectionMode::kWrite && source_field.optional() &&
+ !expected_field.optional()) {
+ return InvalidSchema("Field {} should be required, but is optional",
+ expected_field.ToString());
+ }
+
if (expected_field.type()->type_id() == TypeId::kUnknown) {
+ if (mode == ProjectionMode::kWrite &&
+ source_field.type()->type_id() != TypeId::kUnknown) {
+ return NotSupported("Cannot project {} as {}: type promotion is not
allowed",
+ *source_field.type(), *expected_field.type());
+ }
if (!expected_field.optional()) {
return InvalidSchema("Cannot project required field with id {} as null",
expected_field.field_id());
@@ -104,7 +115,9 @@ Result<FieldProjection> ProjectField(const SchemaField&
expected_field,
expected_field.field_id());
}
if (source_field.type()->type_id() == TypeId::kUnknown &&
- expected_field.type()->is_nested()) {
+ (expected_field.type()->is_nested() || mode == ProjectionMode::kWrite)) {
+ // The write path must materialize nulls with the expected type rather
than copy
+ // the source's unknown array into a differently typed column.
projection.kind = FieldProjection::Kind::kNull;
return projection;
}
@@ -112,7 +125,12 @@ Result<FieldProjection> ProjectField(const SchemaField&
expected_field,
if (expected_field.type()->is_nested()) {
ICEBERG_ASSIGN_OR_RAISE(
projection,
- ProjectNested(*expected_field.type(), *source_field.type(),
prune_source));
+ ProjectNested(*expected_field.type(), *source_field.type(),
prune_source, mode));
+ } else if (mode == ProjectionMode::kWrite) {
+ if (*expected_field.type() != *source_field.type()) {
+ return NotSupported("Cannot project {} as {}: type promotion is not
allowed",
+ *source_field.type(), *expected_field.type());
+ }
} else {
ICEBERG_RETURN_UNEXPECTED(
ValidateSchemaEvolution(*expected_field.type(), *source_field.type()));
@@ -127,7 +145,7 @@ Result<FieldProjection> ProjectField(const SchemaField&
expected_field,
}
Result<FieldProjection> ProjectNested(const Type& expected_type, const Type&
source_type,
- bool prune_source) {
+ bool prune_source, ProjectionMode mode) {
if (!expected_type.is_nested()) {
return InvalidSchema("Expected a nested type, but got {}", expected_type);
}
@@ -169,22 +187,40 @@ Result<FieldProjection> ProjectNested(const Type&
expected_type, const Type& sou
if (auto iter = source_field_map.find(field_id); iter !=
source_field_map.cend()) {
ICEBERG_ASSIGN_OR_RAISE(child_projection,
ProjectField(expected_field, *iter->second.field,
- iter->second.local_index,
prune_source));
- } else if (MetadataColumns::IsMetadataColumn(field_id)) {
+ iter->second.local_index,
prune_source, mode));
+ if (mode == ProjectionMode::kWrite) {
+ source_field_map.erase(iter);
+ }
+ } else if (mode == ProjectionMode::kRead &&
+ MetadataColumns::IsMetadataColumn(field_id)) {
child_projection.kind = FieldProjection::Kind::kMetadata;
- } else if (expected_field.initial_default() != nullptr) {
- // Rows written before the field existed assume its `initial-default`
value.
+ } else if (const auto& default_value = mode == ProjectionMode::kWrite
+ ? expected_field.write_default()
+ :
expected_field.initial_default();
+ default_value != nullptr) {
+ if (mode == ProjectionMode::kWrite) {
+ ICEBERG_RETURN_UNEXPECTED(expected_field.Validate());
+ }
child_projection.kind = FieldProjection::Kind::kDefault;
- child_projection.from = *expected_field.initial_default();
+ child_projection.from = *default_value;
} else if (expected_field.optional()) {
child_projection.kind = FieldProjection::Kind::kNull;
} else {
// TODO(gangwu): support constant value
+ if (mode == ProjectionMode::kWrite) {
+ return InvalidSchema("Missing required field: {} without a
write-default",
+ expected_field.ToString());
+ }
return InvalidSchema("Missing required field: {}",
expected_field.ToString());
}
result.children.emplace_back(std::move(child_projection));
}
+ if (mode == ProjectionMode::kWrite && !source_field_map.empty()) {
+ return InvalidSchema("Input field id {} is not present in write schema",
+ source_field_map.begin()->first);
+ }
+
if (prune_source) {
PruneFieldProjection(result);
}
@@ -195,9 +231,11 @@ Result<FieldProjection> ProjectNested(const Type&
expected_type, const Type& sou
} // namespace
Result<SchemaProjection> Project(const Schema& expected_schema,
- const Schema& source_schema, bool
prune_source) {
- ICEBERG_ASSIGN_OR_RAISE(auto field_projection,
- ProjectNested(expected_schema, source_schema,
prune_source));
+ const Schema& source_schema, bool
prune_source,
+ ProjectionMode mode) {
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto field_projection,
+ ProjectNested(expected_schema, source_schema, prune_source, mode));
return SchemaProjection{std::move(field_projection.children)};
}
diff --git a/src/iceberg/schema_util.h b/src/iceberg/schema_util.h
index 421d00e0..397013f4 100644
--- a/src/iceberg/schema_util.h
+++ b/src/iceberg/schema_util.h
@@ -81,6 +81,14 @@ struct ICEBERG_EXPORT SchemaProjection {
std::vector<FieldProjection> fields;
};
+/// \brief Whether schema projection is used for reading or writing.
+enum class ProjectionMode {
+ /// \brief Use initial-default and allow type promotion.
+ kRead,
+ /// \brief Use write-default, require matching types, and reject
optional-to-required.
+ kWrite,
+};
+
/// \brief Project the expected schema on top of the source schema.
///
/// \param expected_schema The expected schema.
@@ -90,10 +98,11 @@ struct ICEBERG_EXPORT SchemaProjection {
/// column pruning, so `prune_source` is set to true in this case such that
the `from`
/// field in `FieldProjection` exactly reflects the position (relative to its
nesting
/// level) to get the column value from the reader.
+/// \param mode Whether to project for reading or writing.
/// \return The projection result.
-ICEBERG_EXPORT Result<SchemaProjection> Project(const Schema& expected_schema,
- const Schema& source_schema,
- bool prune_source);
+ICEBERG_EXPORT Result<SchemaProjection> Project(
+ const Schema& expected_schema, const Schema& source_schema, bool
prune_source,
+ ProjectionMode mode = ProjectionMode::kRead);
ICEBERG_EXPORT std::string_view ToString(FieldProjection::Kind kind);
ICEBERG_EXPORT std::string ToString(const FieldProjection& projection);
diff --git a/src/iceberg/test/arrow_c_data_util_test.cc
b/src/iceberg/test/arrow_c_data_util_test.cc
index c93a928a..35bb30fd 100644
--- a/src/iceberg/test/arrow_c_data_util_test.cc
+++ b/src/iceberg/test/arrow_c_data_util_test.cc
@@ -21,24 +21,33 @@
/// Verifies ProjectBatch behavior across registered implementations.
#include <cstdint>
+#include <memory>
+#include <string>
#include <string_view>
#include <utility>
#include <vector>
+#include <arrow/array.h>
#include <arrow/c/bridge.h>
+#include <arrow/extension_type.h>
#include <arrow/json/from_string.h>
#include <arrow/record_batch.h>
#include <arrow/type.h>
#include <gtest/gtest.h>
#include "iceberg/arrow/arrow_register.h"
+#include "iceberg/arrow/arrow_status_internal.h"
#include "iceberg/arrow_c_data_guard_internal.h"
#include "iceberg/arrow_c_data_util_internal.h"
+#include "iceberg/expression/literal.h"
+#include "iceberg/result.h"
#include "iceberg/schema.h"
#include "iceberg/schema_field.h"
#include "iceberg/schema_internal.h"
#include "iceberg/test/matchers.h"
#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+#include "iceberg/util/uuid.h"
namespace iceberg::internal {
@@ -119,6 +128,37 @@ std::shared_ptr<Schema> MakeFullSchema() {
SchemaField::MakeOptional(3, "score",
float64())});
}
+Result<std::shared_ptr<::arrow::RecordBatch>> AlignForWrite(
+ const Schema& input_schema, const Schema& write_schema,
+ const ::arrow::RecordBatch& input) {
+ ICEBERG_ASSIGN_OR_RAISE(auto projection,
+ WriteProjection::Make(input_schema, write_schema));
+ ArrowArray c_array;
+ ICEBERG_ARROW_RETURN_NOT_OK(::arrow::ExportRecordBatch(input, &c_array));
+ ICEBERG_ASSIGN_OR_RAISE(auto aligned, projection->AlignBatch(&c_array));
+ ArrowArrayGuard aligned_guard(&aligned);
+
+ ArrowSchema c_schema;
+ ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(write_schema, &c_schema));
+ ICEBERG_ARROW_ASSIGN_OR_RETURN(auto arrow_schema,
::arrow::ImportSchema(&c_schema));
+ ICEBERG_ARROW_ASSIGN_OR_RETURN(auto result,
+ ::arrow::ImportRecordBatch(&aligned,
arrow_schema));
+ ICEBERG_ARROW_RETURN_NOT_OK(result->ValidateFull());
+ return result;
+}
+
+SchemaField DefaultedField(int32_t id, std::string name, std::shared_ptr<Type>
type,
+ bool optional, Literal initial_default,
+ Literal write_default) {
+ return {id,
+ std::move(name),
+ std::move(type),
+ optional,
+ {},
+ std::make_shared<const Literal>(std::move(initial_default)),
+ std::make_shared<const Literal>(std::move(write_default))};
+}
+
} // namespace
TEST(ProjectBatchTest, ProjectSelectedRowsWithoutColumnProjection) {
@@ -191,4 +231,247 @@ TEST(ProjectBatchTest, ProjectionRejectsNestedPruning) {
EXPECT_THAT(projection, IsError(ErrorKind::kInvalidArgument));
}
+TEST(WriteProjectionTest, IdenticalSchemasTransferBatchWithoutConversion) {
+ auto input_schema = MakeFullSchema();
+ auto write_schema = MakeFullSchema();
+ auto input =
+ MakeBatch(*input_schema,
R"([[1,"alice",1.0],[2,null,2.0],[3,"carol",3.0]])")
+ ->Slice(1, 2);
+ ICEBERG_UNWRAP_OR_FAIL(auto projection,
+ WriteProjection::Make(*input_schema, *write_schema));
+ ArrowArray c_array;
+ ASSERT_TRUE(::arrow::ExportRecordBatch(*input, &c_array).ok());
+ auto original = c_array;
+
+ ICEBERG_UNWRAP_OR_FAIL(auto aligned, projection->AlignBatch(&c_array));
+ ArrowArrayGuard aligned_guard(&aligned);
+ EXPECT_EQ(c_array.release, nullptr);
+ EXPECT_EQ(aligned.private_data, original.private_data);
+ EXPECT_EQ(aligned.release, original.release);
+ EXPECT_EQ(aligned.buffers, original.buffers);
+ EXPECT_EQ(aligned.children, original.children);
+
+ auto actual = ::arrow::ImportRecordBatch(&aligned,
input->schema()).ValueOrDie();
+ ASSERT_TRUE(actual->ValidateFull().ok());
+ EXPECT_TRUE(actual->Equals(*input));
+}
+
+TEST(WriteProjectionTest, MaterializesPrimitiveDefaults) {
+ ICEBERG_UNWRAP_OR_FAIL(auto uuid,
+
Uuid::FromString("30313233-3435-3637-3839-616263646566"));
+ struct TestColumn {
+ std::string name;
+ Literal value;
+ std::string_view expected_json;
+ };
+ const std::vector<TestColumn> columns = {
+ {.name = "decimal",
+ .value = Literal::Decimal(-12345, 9, 2),
+ .expected_json = R"(["-123.45","-123.45"])"},
+ {.name = "uuid",
+ .value = Literal::UUID(uuid),
+ .expected_json = R"(["0123456789abcdef","0123456789abcdef"])"},
+ {.name = "binary",
+ .value = Literal::Binary({'a', 0, 'b'}),
+ .expected_json = R"(["a\u0000b","a\u0000b"])"},
+ {.name = "fixed",
+ .value = Literal::Fixed({0, 'x', 0}),
+ .expected_json = R"(["\u0000x\u0000","\u0000x\u0000"])"},
+ {.name = "float", .value = Literal::Float(1.25F), .expected_json =
"[1.25,1.25]"},
+ {.name = "double", .value = Literal::Double(-2.5), .expected_json =
"[-2.5,-2.5]"},
+ {.name = "date", .value = Literal::Date(-1), .expected_json = "[-1,-1]"},
+ {.name = "timestamp",
+ .value = Literal::Timestamp(-1000001),
+ .expected_json = "[-1000001,-1000001]"},
+ };
+ Schema input_schema({SchemaField::MakeRequired(1, "id", int32())});
+ std::vector<SchemaField> write_fields(input_schema.fields().begin(),
+ input_schema.fields().end());
+ for (const auto& column : columns) {
+ write_fields.push_back(
+ SchemaField::MakeRequired(static_cast<int32_t>(write_fields.size() +
1),
+ column.name, column.value.type())
+ .WithWriteDefault(std::make_shared<const Literal>(column.value)));
+ }
+ Schema write_schema(std::move(write_fields));
+ auto input = MakeBatch(input_schema, "[[1],[2]]");
+ ICEBERG_UNWRAP_OR_FAIL(auto actual, AlignForWrite(input_schema,
write_schema, *input));
+
+ for (size_t index = 0; index < columns.size(); ++index) {
+ SCOPED_TRACE(columns[index].name);
+ auto actual_column = actual->column(static_cast<int>(index + 1));
+ if (actual_column->type_id() == ::arrow::Type::EXTENSION) {
+ actual_column =
+
std::static_pointer_cast<::arrow::ExtensionArray>(actual_column)->storage();
+ }
+ auto expected = ::arrow::json::ArrayFromJSONString(
+ actual_column->type(),
std::string(columns[index].expected_json))
+ .ValueOrDie();
+ EXPECT_TRUE(actual_column->Equals(expected)) << actual_column->ToString();
+ }
+}
+
+TEST(WriteProjectionTest, DetachedChildRetainsInputAndReleasesItOnce) {
+ Schema input_schema({SchemaField::MakeRequired(1, "id", int32())});
+ Schema write_schema({SchemaField::MakeRequired(1, "id", int32()),
+ SchemaField::MakeOptional(2, "name", string())});
+ ICEBERG_UNWRAP_OR_FAIL(auto projection,
+ WriteProjection::Make(input_schema, write_schema));
+ auto input = MakeBatch(input_schema, "[[1],[2]]");
+ struct ReleaseState {
+ ArrowArray original{};
+ int count = 0;
+ } state;
+ ArrowArray c_array;
+ ASSERT_TRUE(::arrow::ExportRecordBatch(*input, &c_array).ok());
+ input.reset();
+ state.original = c_array;
+ c_array.private_data = &state;
+ c_array.release = [](ArrowArray* array) {
+ auto* state = static_cast<ReleaseState*>(array->private_data);
+ ++state->count;
+ *array = state->original;
+ array->release(array);
+ };
+ ArrowArray retained_child{};
+ ArrowArrayGuard child_guard(&retained_child);
+ {
+ ICEBERG_UNWRAP_OR_FAIL(auto aligned, projection->AlignBatch(&c_array));
+ ArrowArrayGuard aligned_guard(&aligned);
+ EXPECT_EQ(c_array.release, nullptr);
+ retained_child = std::exchange(*aligned.children[0], ArrowArray{});
+ }
+ auto actual = ::arrow::ImportArray(&retained_child,
::arrow::int32()).ValueOrDie();
+ auto expected =
+ ::arrow::json::ArrayFromJSONString(::arrow::int32(),
"[1,2]").ValueOrDie();
+ EXPECT_TRUE(actual->Equals(expected));
+ actual.reset();
+ EXPECT_EQ(state.count, 1);
+}
+
+struct WriteProjectionParam {
+ std::string name;
+ std::vector<SchemaField> input_fields;
+ std::vector<SchemaField> write_fields;
+ std::string_view input_json;
+ std::string_view expected_json;
+ int64_t slice_offset = 0;
+};
+
+class WriteProjectionParamTest : public
::testing::TestWithParam<WriteProjectionParam> {};
+
+TEST_P(WriteProjectionParamTest, AlignBatch) {
+ const auto& param = GetParam();
+ Schema input_schema(param.input_fields);
+ Schema write_schema(param.write_fields);
+ auto input = MakeBatch(input_schema,
param.input_json)->Slice(param.slice_offset);
+
+ ICEBERG_UNWRAP_OR_FAIL(auto actual, AlignForWrite(input_schema,
write_schema, *input));
+ auto expected = MakeBatch(write_schema, param.expected_json);
+ EXPECT_TRUE(actual->Equals(*expected)) << actual->ToString();
+}
+
+std::vector<WriteProjectionParam> WriteAlignmentCases() {
+ std::vector<WriteProjectionParam> cases;
+
+ {
+ auto input_item = std::make_shared<StructType>(std::vector<SchemaField>{
+ SchemaField::MakeRequired(4, "name", string()),
+ SchemaField::MakeRequired(5, "score", int32()),
+ });
+ auto write_item = std::make_shared<StructType>(std::vector<SchemaField>{
+ SchemaField::MakeRequired(5, "rating", int32()),
+ SchemaField::MakeRequired(4, "renamed", string()),
+ DefaultedField(11, "age", int32(), /*optional=*/false,
Literal::Int(10),
+ Literal::Int(18)),
+ });
+ auto input_value = std::make_shared<StructType>(std::vector<SchemaField>{
+ SchemaField::MakeRequired(9, "label", string()),
+ SchemaField::MakeRequired(10, "code", int32()),
+ });
+ auto write_value = std::make_shared<StructType>(std::vector<SchemaField>{
+ SchemaField::MakeRequired(10, "code_new", int32()),
+ SchemaField::MakeRequired(9, "title", string()),
+ DefaultedField(12, "flag", boolean(), /*optional=*/false,
Literal::Boolean(false),
+ Literal::Boolean(true)),
+ });
+ auto nested_list = [](std::shared_ptr<Type> element) {
+ return list(SchemaField::MakeOptional(
+ 2, "element",
+ list(SchemaField::MakeOptional(3, "element", std::move(element)))));
+ };
+ auto nested_map = [](std::shared_ptr<Type> value) {
+ return map(SchemaField::MakeRequired(7, "key", string()),
+ SchemaField::MakeOptional(8, "value", std::move(value)));
+ };
+ cases.push_back({
+ .name = "StructsInsideSlicedNestedCollections",
+ .input_fields = {SchemaField::MakeOptional(1, "items",
nested_list(input_item)),
+ SchemaField::MakeOptional(6, "mapping",
+ nested_map(input_value))},
+ .write_fields = {SchemaField::MakeOptional(1, "items",
nested_list(write_item)),
+ SchemaField::MakeOptional(6, "mapping",
+ nested_map(write_value))},
+ .input_json = R"([
+ [[[]],[]],
+ [[[{"name":"alice","score":1},null],[]],
+ [["a",{"label":"bob","code":2}],["b",null]]],
+ [null,null],
+ [[],[]]
+ ])",
+ .expected_json = R"([
+ [[[{"rating":1,"renamed":"alice","age":18},null],[]],
+ [["a",{"code_new":2,"title":"bob","flag":true}],["b",null]]],
+ [null,null],
+ [[],[]]
+ ])",
+ .slice_offset = 1,
+ });
+ }
+
+ {
+ auto input_info = std::make_shared<StructType>(
+ std::vector<SchemaField>{SchemaField::MakeRequired(4, "age",
int32())});
+ auto write_info = std::make_shared<StructType>(std::vector<SchemaField>{
+ SchemaField::MakeRequired(4, "age", int32()),
+ DefaultedField(5, "score", int32(), /*optional=*/false,
Literal::Int(0),
+ Literal::Int(99)),
+ });
+ auto input_person = std::make_shared<StructType>(std::vector<SchemaField>{
+ SchemaField::MakeRequired(2, "name", string()),
+ SchemaField::MakeOptional(3, "info", input_info),
+ });
+ auto write_person = std::make_shared<StructType>(std::vector<SchemaField>{
+ SchemaField::MakeOptional(3, "info", write_info),
+ SchemaField::MakeRequired(2, "username", string()),
+ SchemaField::MakeOptional(6, "comment", string()),
+ });
+ cases.push_back({
+ .name = "SlicedStructOffsetsAndParentNulls",
+ .input_fields = {SchemaField::MakeOptional(1, "person", input_person)},
+ .write_fields = {SchemaField::MakeOptional(1, "person", write_person)},
+ .input_json = R"([
+ [{"name":"first","info":{"age":10}}],
+ [null],
+ [{"name":"bob","info":null}],
+ [{"name":"carol","info":{"age":30}}]
+ ])",
+ .expected_json = R"([
+ [null],
+ [{"info":null,"username":"bob","comment":null}],
+ [{"info":{"age":30,"score":99},"username":"carol","comment":null}]
+ ])",
+ .slice_offset = 1,
+ });
+ }
+
+ return cases;
+}
+
+INSTANTIATE_TEST_SUITE_P(WriteAlignment, WriteProjectionParamTest,
+ ::testing::ValuesIn(WriteAlignmentCases()),
+ [](const
::testing::TestParamInfo<WriteProjectionParam>& info) {
+ return info.param.name;
+ });
+
} // namespace iceberg::internal
diff --git a/src/iceberg/test/data_writer_test.cc
b/src/iceberg/test/data_writer_test.cc
index d9536167..ea923c05 100644
--- a/src/iceberg/test/data_writer_test.cc
+++ b/src/iceberg/test/data_writer_test.cc
@@ -196,6 +196,43 @@ INSTANTIATE_TEST_SUITE_P(
::testing::Values(std::make_pair(FileFormatType::kParquet,
"test_data.parquet"),
std::make_pair(FileFormatType::kAvro,
"test_data.avro")));
+TEST(DataWriterSchemaTest, RejectsIncompatibleInputBeforeOpeningFile) {
+ const auto id = SchemaField::MakeRequired(1, "id", int32());
+ const auto added = SchemaField::MakeRequired(2, "added", int32());
+ struct TestCase {
+ std::vector<SchemaField> input_fields;
+ std::vector<SchemaField> write_fields;
+ std::string_view expected_error;
+ ErrorKind error_kind = ErrorKind::kInvalidSchema;
+ };
+ const std::vector<TestCase> cases = {
+ {.input_fields = {id},
+ .write_fields = {id, added},
+ .expected_error = "Missing required field"},
+ {.input_fields = {id, added},
+ .write_fields = {id},
+ .expected_error = "Input field id 2"},
+ {.input_fields = {SchemaField::MakeOptional(1, "id", int32())},
+ .write_fields = {id},
+ .expected_error = "should be required, but is optional"},
+ {.input_fields = {SchemaField::MakeOptional(1, "id", int32())},
+ .write_fields = {SchemaField::MakeOptional(1, "id", unknown())},
+ .expected_error = "Cannot project",
+ .error_kind = ErrorKind::kNotSupported},
+ };
+ for (const auto& test_case : cases) {
+ SCOPED_TRACE(test_case.expected_error);
+ // Schema errors must be reported before a file or format writer is opened.
+ DataWriterOptions options{
+ .schema = std::make_shared<Schema>(test_case.write_fields),
+ .input_schema = std::make_shared<Schema>(test_case.input_fields),
+ };
+ auto result = DataWriter::Make(options);
+ EXPECT_THAT(result, IsError(test_case.error_kind));
+ EXPECT_THAT(result, HasErrorMessage(test_case.expected_error));
+ }
+}
+
TEST_F(DataWriterTest, WriteAndClose) {
auto writer_result = DataWriter::Make(MakeDefaultOptions());
ASSERT_THAT(writer_result, IsOk());
diff --git a/src/iceberg/test/default_value_test.cc
b/src/iceberg/test/default_value_test.cc
index f85f2e0d..302745cf 100644
--- a/src/iceberg/test/default_value_test.cc
+++ b/src/iceberg/test/default_value_test.cc
@@ -23,6 +23,8 @@
#include <memory>
#include <string>
#include <string_view>
+#include <unordered_map>
+#include <utility>
#include <vector>
#include <arrow/array.h>
@@ -38,6 +40,7 @@
#include "iceberg/arrow/arrow_io_internal.h"
#include "iceberg/avro/avro_register.h"
+#include "iceberg/data/data_writer.h"
#include "iceberg/expression/literal.h"
#include "iceberg/file_format.h"
#include "iceberg/file_reader.h"
@@ -72,6 +75,12 @@ struct DefaultValueEndToEndParam {
bool avro_skip_datum = true;
};
+struct WriteDefaultEndToEndParam {
+ std::string name;
+ FileFormatType format;
+ std::string path;
+};
+
class DefaultValueEndToEndTest
: public UpdateTestBase,
public ::testing::WithParamInterface<DefaultValueEndToEndParam> {
@@ -86,6 +95,82 @@ class DefaultValueEndToEndTest
}
};
+class WriteDefaultEndToEndTest
+ : public ::testing::TestWithParam<WriteDefaultEndToEndParam> {
+ protected:
+ static void SetUpTestSuite() {
+ parquet::RegisterAll();
+ avro::RegisterAll();
+ }
+
+ void SetUp() override { file_io_ =
arrow::ArrowFileSystemFileIO::MakeMockFileIO(); }
+
+ std::shared_ptr<::arrow::Array> CreateArray(const Schema& schema,
+ std::string_view json) {
+ ArrowSchema arrow_c_schema;
+ ICEBERG_THROW_NOT_OK(ToArrowSchema(schema, &arrow_c_schema));
+ auto arrow_schema = ::arrow::ImportType(&arrow_c_schema).ValueOrDie();
+ return
::arrow::json::ArrayFromJSONString(::arrow::struct_(arrow_schema->fields()),
+ std::string(json))
+ .ValueOrDie();
+ }
+
+ static std::unordered_map<std::string, std::string> FormatProperties(
+ const WriteDefaultEndToEndParam& param) {
+ if (param.format == FileFormatType::kParquet) {
+ return {{"write.parquet.compression-codec", "uncompressed"}};
+ }
+ return {};
+ }
+
+ void WriteAndRead(const std::shared_ptr<Schema>& input_schema,
+ const std::shared_ptr<Schema>& write_schema,
+ const std::vector<std::string_view>& input_batches,
+ std::string_view expected_json) {
+ const auto& param = GetParam();
+ DataWriterOptions options{
+ .path = param.path,
+ .schema = write_schema,
+ .format = param.format,
+ .io = file_io_,
+ .properties = FormatProperties(param),
+ .input_schema = input_schema,
+ };
+ ICEBERG_UNWRAP_OR_FAIL(auto writer, DataWriter::Make(options));
+ for (auto input_json : input_batches) {
+ auto input = CreateArray(*input_schema, input_json);
+ ArrowArray arrow_array;
+ ASSERT_TRUE(::arrow::ExportArray(*input, &arrow_array).ok());
+ ASSERT_THAT(writer->Write(&arrow_array), IsOk());
+ }
+ ASSERT_THAT(writer->Close(), IsOk());
+
+ ICEBERG_UNWRAP_OR_FAIL(auto reader, ReaderFactoryRegistry::Open(
+ param.format, {.path = param.path,
+ .io = file_io_,
+ .projection =
write_schema}));
+ auto expected = CreateArray(*write_schema, expected_json);
+ int64_t row_offset = 0;
+ while (true) {
+ ICEBERG_UNWRAP_OR_FAIL(auto batch, reader->Next());
+ if (!batch.has_value()) {
+ break;
+ }
+ auto actual = ::arrow::ImportArray(&batch.value(),
expected->type()).ValueOrDie();
+ ASSERT_LE(row_offset + actual->length(), expected->length());
+ auto expected_batch = expected->Slice(row_offset, actual->length());
+ ASSERT_TRUE(actual->Equals(expected_batch))
+ << "actual: " << actual->ToString()
+ << "\nexpected: " << expected_batch->ToString();
+ row_offset += actual->length();
+ }
+ EXPECT_EQ(row_offset, expected->length());
+ ASSERT_THAT(reader->Close(), IsOk());
+ }
+
+ std::shared_ptr<FileIO> file_io_;
+};
+
TEST_P(DefaultValueEndToEndTest, WriteEvolveReadFillsInitialDefault) {
const auto& param = GetParam();
ICEBERG_UNWRAP_OR_FAIL(auto original_schema, table_->schema());
@@ -186,6 +271,32 @@ TEST_P(DefaultValueEndToEndTest,
WriteEvolveReadFillsInitialDefault) {
ASSERT_FALSE(next_batch.has_value());
}
+TEST_P(WriteDefaultEndToEndTest, AlignsBatchesAndFillsWriteDefaults) {
+ auto input_schema = std::make_shared<Schema>(std::vector<SchemaField>{
+ SchemaField::MakeOptional(2, "name", string()),
+ SchemaField::MakeRequired(1, "id", int32()),
+ SchemaField::MakeOptional(5, "score", unknown()),
+ });
+ auto write_schema = std::make_shared<Schema>(std::vector<SchemaField>{
+ SchemaField::MakeRequired(1, "id", int32()),
+ SchemaField::MakeOptional(2, "name", string())
+ .WithWriteDefault(std::make_shared<const
Literal>(Literal::String("default"))),
+ SchemaField(3, "added", int32(), /*optional=*/false, /*doc=*/{},
+ std::make_shared<const Literal>(Literal::Int(42)),
+ std::make_shared<const Literal>(Literal::Int(7))),
+ SchemaField::MakeOptional(4, "comment", string()),
+ SchemaField::MakeOptional(5, "score", int32()),
+ });
+ // This file is written after the column exists, so the missing input column
uses
+ // write-default (7), not initial-default (42). Existing nulls remain null;
the
+ // unknown input column and missing optional column become target-typed
nulls.
+ WriteAndRead(input_schema, write_schema,
+ {R"([["alice", 1, null], [null, 2, null]])",
+ R"([["bob", 3, null], [null, 4, null]])"},
+ R"([[1, "alice", 7, null, null], [2, null, 7, null, null],
+ [3, "bob", 7, null, null], [4, null, 7, null, null]])");
+}
+
namespace {
// Two-row column of `json` values at `type`, for the simple cases.
@@ -261,6 +372,16 @@ INSTANTIATE_TEST_SUITE_P(
return info.param.name;
});
+INSTANTIATE_TEST_SUITE_P(
+ Formats, WriteDefaultEndToEndTest,
+ ::testing::Values(WriteDefaultEndToEndParam{"parquet",
FileFormatType::kParquet,
+ "write-default.parquet"},
+ WriteDefaultEndToEndParam{"avro", FileFormatType::kAvro,
+ "write-default.avro"}),
+ [](const ::testing::TestParamInfo<WriteDefaultEndToEndParam>& info) {
+ return info.param.name;
+ });
+
} // namespace
} // namespace iceberg
diff --git a/src/iceberg/test/schema_util_test.cc
b/src/iceberg/test/schema_util_test.cc
index 9a3eff88..785a9035 100644
--- a/src/iceberg/test/schema_util_test.cc
+++ b/src/iceberg/test/schema_util_test.cc
@@ -232,6 +232,131 @@ TEST(SchemaUtilTest,
ProjectPresentFieldIgnoresInitialDefault) {
AssertProjectedField(projection_result->fields[0], 0);
}
+TEST(SchemaUtilTest, ProjectWritePolicyIgnoresInitialDefault) {
+ // With the write default source, a field having only `initial-default`
behaves as if
+ // it had no default: optional becomes null and required is rejected.
+ Schema source_schema(
+ {SchemaField::MakeRequired(/*field_id=*/1, "id", iceberg::int64())});
+ Schema expected_schema({
+ SchemaField::MakeRequired(/*field_id=*/1, "id", iceberg::int64()),
+ SchemaField(/*field_id=*/10, "extra", iceberg::string(),
/*optional=*/true,
+ /*doc=*/{}, std::make_shared<const
Literal>(Literal::String("n/a"))),
+ });
+
+ auto projection_result = Project(expected_schema, source_schema,
/*prune_source=*/false,
+ ProjectionMode::kWrite);
+ ASSERT_THAT(projection_result, IsOk());
+ ASSERT_EQ(projection_result->fields[1].kind, FieldProjection::Kind::kNull);
+}
+
+TEST(SchemaUtilTest, ProjectWritePolicyUnknownSourceProducesNull) {
+ Schema source_schema(
+ {SchemaField::MakeOptional(/*field_id=*/1, "value",
iceberg::unknown())});
+ Schema expected_schema({
+ SchemaField(/*field_id=*/1, "value", iceberg::int32(), /*optional=*/true,
+ /*doc=*/{}, /*initial_default=*/nullptr,
+ std::make_shared<const Literal>(Literal::Int(7))),
+ });
+
+ auto write_projection = Project(expected_schema, source_schema,
/*prune_source=*/false,
+ ProjectionMode::kWrite);
+ ASSERT_THAT(write_projection, IsOk());
+ ASSERT_EQ(write_projection->fields.size(), 1);
+ // A supplied null stays null, even when the field has a write-default.
+ ASSERT_EQ(write_projection->fields[0].kind, FieldProjection::Kind::kNull);
+
+ auto read_projection = Project(expected_schema, source_schema,
/*prune_source=*/false);
+ ASSERT_THAT(read_projection, IsOk());
+ ASSERT_EQ(read_projection->fields.size(), 1);
+ AssertProjectedField(read_projection->fields[0], 0);
+}
+
+TEST(SchemaUtilTest, ProjectWriteRejectsTypePromotion) {
+ Schema source_schema(
+ {SchemaField::MakeRequired(/*field_id=*/1, "id", iceberg::int32())});
+ Schema expected_schema(
+ {SchemaField::MakeRequired(/*field_id=*/1, "id", iceberg::int64())});
+
+ auto projection_result = Project(expected_schema, source_schema,
/*prune_source=*/false,
+ ProjectionMode::kWrite);
+ ASSERT_THAT(projection_result, IsError(ErrorKind::kNotSupported));
+ ASSERT_THAT(projection_result, HasErrorMessage("type promotion is not
allowed"));
+}
+
+class ProjectWriteCompatibilityTest : public ::testing::TestWithParam<TypeId> {
+ protected:
+ Schema MakeSchema(bool optional, std::shared_ptr<Type> type = int32()) const
{
+ auto value = SchemaField(1, "value", std::move(type), optional);
+ std::shared_ptr<Type> nested;
+ switch (GetParam()) {
+ case TypeId::kStruct:
+ nested = std::make_shared<StructType>(std::vector<SchemaField>{value});
+ break;
+ case TypeId::kList:
+ nested = list(value.WithName("element"));
+ break;
+ case TypeId::kMap:
+ nested = map(SchemaField::MakeRequired(2, "key", string()), value);
+ break;
+ default:
+ return Schema({value});
+ }
+ return Schema({SchemaField::MakeOptional(3, "nested", nested)});
+ }
+};
+
+TEST_P(ProjectWriteCompatibilityTest, RejectsOptionalToRequired) {
+ auto source_schema = MakeSchema(/*optional=*/true);
+ auto expected_schema = MakeSchema(/*optional=*/false);
+
+ auto result = Project(expected_schema, source_schema, /*prune_source=*/false,
+ ProjectionMode::kWrite);
+ ASSERT_THAT(result, IsError(ErrorKind::kInvalidSchema));
+ EXPECT_THAT(result, HasErrorMessage("should be required, but is optional"));
+ EXPECT_THAT(Project(source_schema, expected_schema, /*prune_source=*/false,
+ ProjectionMode::kWrite),
+ IsOk());
+ EXPECT_THAT(Project(expected_schema, source_schema, /*prune_source=*/false),
IsOk());
+}
+
+TEST_P(ProjectWriteCompatibilityTest, RejectsConcreteToUnknown) {
+ auto source_schema = MakeSchema(/*optional=*/true);
+ auto expected_schema = MakeSchema(/*optional=*/true, unknown());
+
+ auto result = Project(expected_schema, source_schema, /*prune_source=*/false,
+ ProjectionMode::kWrite);
+ ASSERT_THAT(result, IsError(ErrorKind::kNotSupported));
+ EXPECT_THAT(result, HasErrorMessage("type promotion is not allowed"));
+ EXPECT_THAT(Project(expected_schema, source_schema, /*prune_source=*/false),
IsOk());
+}
+
+TEST_P(ProjectWriteCompatibilityTest, RejectsExtraInputFields) {
+ std::vector<SchemaField> fields{
+ SchemaField::MakeOptional(10, "kept", int32()),
+ SchemaField::MakeOptional(11, "extra", int32()),
+ };
+ auto source_schema =
+ GetParam() == TypeId::kInt
+ ? Schema(fields)
+ : MakeSchema(/*optional=*/true,
std::make_shared<StructType>(fields));
+ fields.pop_back();
+ auto expected_schema =
+ GetParam() == TypeId::kInt
+ ? Schema(fields)
+ : MakeSchema(/*optional=*/true,
std::make_shared<StructType>(fields));
+
+ auto result = Project(expected_schema, source_schema, /*prune_source=*/false,
+ ProjectionMode::kWrite);
+ ASSERT_THAT(result, IsError(ErrorKind::kInvalidSchema));
+ EXPECT_THAT(result,
+ HasErrorMessage("Input field id 11 is not present in write
schema"));
+ EXPECT_THAT(Project(expected_schema, source_schema, /*prune_source=*/false),
IsOk());
+}
+
+INSTANTIATE_TEST_SUITE_P(Fields, ProjectWriteCompatibilityTest,
+ ::testing::Values(TypeId::kInt, TypeId::kStruct,
TypeId::kList,
+ TypeId::kMap));
+
TEST(SchemaUtilTest, ProjectMetadataColumn) {
Schema source_schema = CreateFlatSchema();
Schema expected_schema({
@@ -247,6 +372,11 @@ TEST(SchemaUtilTest, ProjectMetadataColumn) {
ASSERT_EQ(projection.fields.size(), 2);
AssertProjectedField(projection.fields[0], 0);
ASSERT_EQ(projection.fields[1].kind, FieldProjection::Kind::kMetadata);
+
+ auto write_projection = Project(expected_schema, source_schema,
/*prune_source=*/false,
+ ProjectionMode::kWrite);
+ EXPECT_THAT(write_projection, IsError(ErrorKind::kInvalidSchema));
+ EXPECT_THAT(write_projection, HasErrorMessage("Missing required field"));
}
TEST(SchemaUtilTest, ProjectSchemaEvolutionIntToLong) {
