Skip to content

Commit 0e4c6a4

Browse files
committed
Address review: cache default Arrow scalars for Avro reads
Precompute cast scalars in FieldProjection attributes via PrepareDefaultScalars at reader init so per-row default fills only call AppendScalar.
1 parent 9e1605f commit 0e4c6a4

5 files changed

Lines changed: 163 additions & 7 deletions

File tree

src/iceberg/avro/avro_data_util.cc

Lines changed: 105 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@
1717
* under the License.
1818
*/
1919

20+
#include <span>
21+
2022
#include <arrow/array/builder_binary.h>
2123
#include <arrow/array/builder_decimal.h>
2224
#include <arrow/array/builder_nested.h>
@@ -91,8 +93,7 @@ Status AppendStructToBuilder(const ::avro::NodePtr& avro_node,
9193
} else if (field_projection.kind == FieldProjection::Kind::kNull) {
9294
ICEBERG_ARROW_RETURN_NOT_OK(field_builder->AppendNull());
9395
} else if (field_projection.kind == FieldProjection::Kind::kDefault) {
94-
ICEBERG_RETURN_UNEXPECTED(AppendDefaultToBuilder(
95-
std::get<Literal>(field_projection.from), field_builder));
96+
ICEBERG_RETURN_UNEXPECTED(AppendDefaultToBuilder(field_projection, field_builder));
9697
} else if (field_projection.kind == FieldProjection::Kind::kMetadata) {
9798
int32_t field_id = expected_field.field_id();
9899
if (field_id == MetadataColumns::kFilePathColumnId) {
@@ -472,7 +473,7 @@ Status AppendFieldToBuilder(const ::avro::NodePtr& avro_node,
472473
}
473474

474475
if (projection.kind == FieldProjection::Kind::kDefault) {
475-
return AppendDefaultToBuilder(std::get<Literal>(projection.from), array_builder);
476+
return AppendDefaultToBuilder(projection, array_builder);
476477
}
477478

478479
const bool is_row_lineage =
@@ -506,7 +507,10 @@ Status AppendFieldToBuilder(const ::avro::NodePtr& avro_node,
506507

507508
} // namespace
508509

509-
Status AppendDefaultToBuilder(const Literal& literal, ::arrow::ArrayBuilder* builder) {
510+
namespace {
511+
512+
Result<std::shared_ptr<::arrow::Scalar>> MakeDefaultScalar(
513+
const Literal& literal, const std::shared_ptr<::arrow::DataType>& builder_type) {
510514
// The builder's own memory pool is not exposed, so the small scalar buffer uses the
511515
// default pool.
512516
ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<::arrow::Scalar> scalar,
@@ -516,7 +520,7 @@ Status AppendDefaultToBuilder(const Literal& literal, ::arrow::ArrayBuilder* bui
516520
// yields the storage scalar (fixed_size_binary(16) for uuid) and Scalar::CastTo has no
517521
// kernel that targets an extension type. This mirrors MakeDefaultArray's extension
518522
// handling.
519-
std::shared_ptr<::arrow::DataType> target_type = builder->type();
523+
std::shared_ptr<::arrow::DataType> target_type = builder_type;
520524
if (target_type->id() == ::arrow::Type::EXTENSION) {
521525
target_type = internal::checked_cast<const ::arrow::ExtensionType&>(*target_type)
522526
.storage_type();
@@ -525,10 +529,106 @@ Status AppendDefaultToBuilder(const Literal& literal, ::arrow::ArrayBuilder* bui
525529
if (!scalar->type->Equals(*target_type)) {
526530
ICEBERG_ARROW_ASSIGN_OR_RETURN(scalar, scalar->CastTo(target_type));
527531
}
532+
return scalar;
533+
}
534+
535+
Status PrepareDefaultScalars(std::span<FieldProjection> projections,
536+
::arrow::ArrayBuilder* builder) {
537+
auto* struct_builder = internal::checked_cast<::arrow::StructBuilder*>(builder);
538+
if (static_cast<size_t>(struct_builder->num_fields()) != projections.size()) {
539+
return InvalidArgument(
540+
"Inconsistent number of struct builder fields ({}) and projections ({})",
541+
struct_builder->num_fields(), projections.size());
542+
}
543+
544+
for (size_t i = 0; i < projections.size(); ++i) {
545+
auto& field_projection = projections[i];
546+
auto* field_builder = struct_builder->field_builder(static_cast<int>(i));
547+
548+
if (field_projection.kind == FieldProjection::Kind::kDefault) {
549+
if (dynamic_cast<const AvroDefaultAttributes*>(field_projection.attributes.get()) !=
550+
nullptr) {
551+
continue;
552+
}
553+
auto attrs = std::make_shared<AvroDefaultAttributes>();
554+
ICEBERG_ASSIGN_OR_RAISE(attrs->scalar,
555+
MakeDefaultScalar(std::get<Literal>(field_projection.from),
556+
field_builder->type()));
557+
field_projection.attributes = std::move(attrs);
558+
continue;
559+
}
560+
561+
if (field_projection.kind != FieldProjection::Kind::kProjected ||
562+
field_projection.children.empty()) {
563+
continue;
564+
}
565+
566+
switch (field_builder->type()->id()) {
567+
case ::arrow::Type::STRUCT:
568+
ICEBERG_RETURN_UNEXPECTED(
569+
PrepareDefaultScalars(field_projection.children, field_builder));
570+
break;
571+
case ::arrow::Type::LIST: {
572+
// List projections store a single child for the element. Defaults only appear
573+
// on nested struct fields of that element.
574+
auto* list_builder = internal::checked_cast<::arrow::ListBuilder*>(field_builder);
575+
auto& element_projection = field_projection.children[0];
576+
if (element_projection.kind == FieldProjection::Kind::kProjected &&
577+
!element_projection.children.empty() &&
578+
list_builder->value_builder()->type()->id() == ::arrow::Type::STRUCT) {
579+
ICEBERG_RETURN_UNEXPECTED(PrepareDefaultScalars(element_projection.children,
580+
list_builder->value_builder()));
581+
}
582+
break;
583+
}
584+
case ::arrow::Type::MAP: {
585+
auto* map_builder = internal::checked_cast<::arrow::MapBuilder*>(field_builder);
586+
if (field_projection.children.size() >= 1 &&
587+
!field_projection.children[0].children.empty() &&
588+
map_builder->key_builder()->type()->id() == ::arrow::Type::STRUCT) {
589+
ICEBERG_RETURN_UNEXPECTED(PrepareDefaultScalars(
590+
field_projection.children[0].children, map_builder->key_builder()));
591+
}
592+
if (field_projection.children.size() >= 2 &&
593+
!field_projection.children[1].children.empty() &&
594+
map_builder->item_builder()->type()->id() == ::arrow::Type::STRUCT) {
595+
ICEBERG_RETURN_UNEXPECTED(PrepareDefaultScalars(
596+
field_projection.children[1].children, map_builder->item_builder()));
597+
}
598+
break;
599+
}
600+
default:
601+
break;
602+
}
603+
}
604+
return {};
605+
}
606+
607+
} // namespace
608+
609+
Status PrepareDefaultScalars(SchemaProjection& projection,
610+
::arrow::ArrayBuilder* root_builder) {
611+
return PrepareDefaultScalars(projection.fields, root_builder);
612+
}
613+
614+
Status AppendDefaultToBuilder(const Literal& literal, ::arrow::ArrayBuilder* builder) {
615+
ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<::arrow::Scalar> scalar,
616+
MakeDefaultScalar(literal, builder->type()));
528617
ICEBERG_ARROW_RETURN_NOT_OK(builder->AppendScalar(*scalar));
529618
return {};
530619
}
531620

621+
Status AppendDefaultToBuilder(const FieldProjection& projection,
622+
::arrow::ArrayBuilder* builder) {
623+
if (const auto* attrs =
624+
dynamic_cast<const AvroDefaultAttributes*>(projection.attributes.get());
625+
attrs != nullptr && attrs->scalar != nullptr) {
626+
ICEBERG_ARROW_RETURN_NOT_OK(builder->AppendScalar(*attrs->scalar));
627+
return {};
628+
}
629+
return AppendDefaultToBuilder(std::get<Literal>(projection.from), builder);
630+
}
631+
532632
Status AppendDatumToBuilder(const ::avro::NodePtr& avro_node,
533633
const ::avro::GenericDatum& avro_datum,
534634
const SchemaProjection& projection,

src/iceberg/avro/avro_data_util_internal.h

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

2020
#pragma once
2121

22+
#include <memory>
23+
2224
#include <arrow/array/builder_base.h>
25+
#include <arrow/scalar.h>
2326
#include <avro/GenericDatum.hh>
2427

2528
#include "iceberg/arrow/metadata_column_util_internal.h"
@@ -28,13 +31,36 @@
2831

2932
namespace iceberg::avro {
3033

34+
/// \brief Cached Arrow scalar for a `kDefault` field projection.
35+
///
36+
/// Materialized once (see `PrepareDefaultScalars`) so row-by-row Avro decode only
37+
/// needs `AppendScalar` instead of repeating `ToArrowScalar` / `CastTo` per row.
38+
struct AvroDefaultAttributes : FieldProjection::ExtraAttributes {
39+
std::shared_ptr<::arrow::Scalar> scalar;
40+
};
41+
42+
/// \brief Precompute cast Arrow scalars for every `kDefault` field under `projection`.
43+
///
44+
/// Walks `root_builder` in lockstep with the projection so each default is cast to the
45+
/// builder's Arrow type once per scan. Safe to call repeatedly; existing
46+
/// `AvroDefaultAttributes` entries are left unchanged.
47+
Status PrepareDefaultScalars(SchemaProjection& projection,
48+
::arrow::ArrayBuilder* root_builder);
49+
3150
/// \brief Append a literal once to `builder` while decoding Avro row-by-row.
3251
///
3352
/// Used to materialize `FieldProjection::Kind::kDefault`. Shares `ToArrowScalar` with
3453
/// Parquet's batch path (`MakeDefaultArray`); the append shape stays Avro-local because
3554
/// Avro builds Arrow arrays via per-row `ArrayBuilder`s rather than whole-column arrays.
55+
/// Prefer the `FieldProjection` overload after `PrepareDefaultScalars` so the scalar is
56+
/// reused across rows.
3657
Status AppendDefaultToBuilder(const Literal& literal, ::arrow::ArrayBuilder* builder);
3758

59+
/// \brief Append a `kDefault` projection, reusing a scalar cached on
60+
/// `projection.attributes` when present.
61+
Status AppendDefaultToBuilder(const FieldProjection& projection,
62+
::arrow::ArrayBuilder* builder);
63+
3864
/// \brief Append an Avro datum to an Arrow array builder.
3965
///
4066
/// This function handles schema evolution by using the provided projection to map

src/iceberg/avro/avro_direct_decoder.cc

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -211,8 +211,7 @@ Status DecodeStructToBuilder(const ::avro::NodePtr& avro_node, ::avro::Decoder&
211211
if (field_projection.kind == FieldProjection::Kind::kNull) {
212212
ICEBERG_ARROW_RETURN_NOT_OK(field_builder->AppendNull());
213213
} else if (field_projection.kind == FieldProjection::Kind::kDefault) {
214-
ICEBERG_RETURN_UNEXPECTED(AppendDefaultToBuilder(
215-
std::get<Literal>(field_projection.from), field_builder));
214+
ICEBERG_RETURN_UNEXPECTED(AppendDefaultToBuilder(field_projection, field_builder));
216215
} else if (field_projection.kind == FieldProjection::Kind::kMetadata) {
217216
int32_t field_id = expected_field.field_id();
218217
if (field_id == MetadataColumns::kFilePathColumnId) {

src/iceberg/avro/avro_reader.cc

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -434,6 +434,8 @@ class AvroReader::Impl {
434434
builder_result.status().message());
435435
}
436436
context_->builder_ = builder_result.MoveValueUnsafe();
437+
ICEBERG_RETURN_UNEXPECTED(
438+
PrepareDefaultScalars(projection_, context_->builder_.get()));
437439
backend_->InitReadContext(backend_->GetReaderSchema());
438440

439441
return {};

src/iceberg/test/avro_data_test.cc

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -726,6 +726,35 @@ TEST(AppendDefaultToBuilderTest, CastsToBuilderType) {
726726
ASSERT_EQ(static_cast<const ::arrow::Int64Array&>(*array).Value(0), 7);
727727
}
728728

729+
TEST(AppendDefaultToBuilderTest, ReusesPreparedScalar) {
730+
auto pool = ::arrow::default_memory_pool();
731+
auto child = std::make_shared<::arrow::Int64Builder>(pool);
732+
::arrow::StructBuilder struct_builder(
733+
::arrow::struct_({::arrow::field("d", ::arrow::int64())}), pool, {child});
734+
735+
FieldProjection projection;
736+
projection.kind = FieldProjection::Kind::kDefault;
737+
projection.from = Literal::Long(42);
738+
739+
SchemaProjection schema_projection;
740+
schema_projection.fields.push_back(projection);
741+
ASSERT_THAT(PrepareDefaultScalars(schema_projection, &struct_builder), IsOk());
742+
ASSERT_NE(dynamic_cast<const AvroDefaultAttributes*>(
743+
schema_projection.fields[0].attributes.get()),
744+
nullptr);
745+
746+
auto* field_builder = struct_builder.field_builder(0);
747+
ASSERT_THAT(AppendDefaultToBuilder(schema_projection.fields[0], field_builder), IsOk());
748+
ASSERT_THAT(AppendDefaultToBuilder(schema_projection.fields[0], field_builder), IsOk());
749+
750+
std::shared_ptr<::arrow::Array> array;
751+
ASSERT_TRUE(field_builder->Finish(&array).ok());
752+
ASSERT_EQ(array->length(), 2);
753+
const auto& long_array = static_cast<const ::arrow::Int64Array&>(*array);
754+
ASSERT_EQ(long_array.Value(0), 42);
755+
ASSERT_EQ(long_array.Value(1), 42);
756+
}
757+
729758
TEST(AppendDatumToBuilderTest, NestedStructWithMissingOptionalFields) {
730759
Schema iceberg_schema({
731760
SchemaField::MakeRequired(1, "id", iceberg::int32()),

0 commit comments

Comments
 (0)