Skip to content

Commit 976f4e9

Browse files
zjw1111zjw1111
committed
fix2
1 parent f2c649a commit 976f4e9

3 files changed

Lines changed: 55 additions & 54 deletions

File tree

‎src/paimon/format/avro/avro_direct_decoder.cpp‎

Lines changed: 51 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -49,46 +49,46 @@ bool HasMapLogicalType(const ::avro::NodePtr& node) {
4949
/// Forward declaration for mutual recursion.
5050
Status DecodeFieldToBuilder(const ::avro::NodePtr& avro_node,
5151
const std::optional<std::set<size_t>>& projection,
52-
::avro::Decoder& decoder, arrow::ArrayBuilder* array_builder,
53-
AvroDirectDecoder::DecodeContext& ctx);
52+
::avro::Decoder* decoder, arrow::ArrayBuilder* array_builder,
53+
AvroDirectDecoder::DecodeContext* ctx);
5454

5555
/// \brief Skip an Avro value based on its schema without decoding
56-
Status SkipAvroValue(const ::avro::NodePtr& avro_node, ::avro::Decoder& decoder) {
56+
Status SkipAvroValue(const ::avro::NodePtr& avro_node, ::avro::Decoder* decoder) {
5757
switch (avro_node->type()) {
5858
case ::avro::AVRO_NULL:
59-
decoder.decodeNull();
59+
decoder->decodeNull();
6060
return Status::OK();
6161

6262
case ::avro::AVRO_BOOL:
63-
decoder.decodeBool();
63+
decoder->decodeBool();
6464
return Status::OK();
6565

6666
case ::avro::AVRO_INT:
67-
decoder.decodeInt();
67+
decoder->decodeInt();
6868
return Status::OK();
6969

7070
case ::avro::AVRO_LONG:
71-
decoder.decodeLong();
71+
decoder->decodeLong();
7272
return Status::OK();
7373

7474
case ::avro::AVRO_FLOAT:
75-
decoder.decodeFloat();
75+
decoder->decodeFloat();
7676
return Status::OK();
7777

7878
case ::avro::AVRO_DOUBLE:
79-
decoder.decodeDouble();
79+
decoder->decodeDouble();
8080
return Status::OK();
8181

8282
case ::avro::AVRO_STRING:
83-
decoder.skipString();
83+
decoder->skipString();
8484
return Status::OK();
8585

8686
case ::avro::AVRO_BYTES:
87-
decoder.skipBytes();
87+
decoder->skipBytes();
8888
return Status::OK();
8989

9090
case ::avro::AVRO_FIXED:
91-
decoder.skipFixed(avro_node->fixedSize());
91+
decoder->skipFixed(avro_node->fixedSize());
9292
return Status::OK();
9393

9494
case ::avro::AVRO_RECORD: {
@@ -100,38 +100,38 @@ Status SkipAvroValue(const ::avro::NodePtr& avro_node, ::avro::Decoder& decoder)
100100
}
101101

102102
case ::avro::AVRO_ENUM:
103-
decoder.decodeEnum();
103+
decoder->decodeEnum();
104104
return Status::OK();
105105

106106
case ::avro::AVRO_ARRAY: {
107107
const auto& element_node = avro_node->leafAt(0);
108108
// skipArray() returns count like arrayStart(), must handle all blocks
109-
int64_t block_count = decoder.skipArray();
109+
int64_t block_count = decoder->skipArray();
110110
while (block_count > 0) {
111111
for (int64_t i = 0; i < block_count; ++i) {
112112
PAIMON_RETURN_NOT_OK(SkipAvroValue(element_node, decoder));
113113
}
114-
block_count = decoder.arrayNext();
114+
block_count = decoder->arrayNext();
115115
}
116116
return Status::OK();
117117
}
118118

119119
case ::avro::AVRO_MAP: {
120120
const auto& value_node = avro_node->leafAt(1);
121121
// skipMap() returns count like mapStart(), must handle all blocks
122-
int64_t block_count = decoder.skipMap();
122+
int64_t block_count = decoder->skipMap();
123123
while (block_count > 0) {
124124
for (int64_t i = 0; i < block_count; ++i) {
125-
decoder.skipString(); // Skip key (always string in Avro maps)
125+
decoder->skipString(); // Skip key (always string in Avro maps)
126126
PAIMON_RETURN_NOT_OK(SkipAvroValue(value_node, decoder));
127127
}
128-
block_count = decoder.mapNext();
128+
block_count = decoder->mapNext();
129129
}
130130
return Status::OK();
131131
}
132132

133133
case ::avro::AVRO_UNION: {
134-
const size_t branch_index = decoder.decodeUnionIndex();
134+
const size_t branch_index = decoder->decodeUnionIndex();
135135
// Validate branch index
136136
const size_t num_branches = avro_node->leaves();
137137
if (branch_index >= num_branches) {
@@ -150,8 +150,8 @@ Status SkipAvroValue(const ::avro::NodePtr& avro_node, ::avro::Decoder& decoder)
150150
/// Decode Avro record directly to Arrow struct builder.
151151
Status DecodeStructToBuilder(const ::avro::NodePtr& avro_node,
152152
const std::optional<std::set<size_t>>& projection,
153-
::avro::Decoder& decoder, arrow::ArrayBuilder* array_builder,
154-
AvroDirectDecoder::DecodeContext& ctx) {
153+
::avro::Decoder* decoder, arrow::ArrayBuilder* array_builder,
154+
AvroDirectDecoder::DecodeContext* ctx) {
155155
if (avro_node->type() != ::avro::AVRO_RECORD) {
156156
return Status::Invalid(
157157
fmt::format("Expected Avro record, got type: {}", ToString(avro_node)));
@@ -179,9 +179,9 @@ Status DecodeStructToBuilder(const ::avro::NodePtr& avro_node,
179179
}
180180

181181
/// Decode Avro array directly to Arrow list builder.
182-
Status DecodeListToBuilder(const ::avro::NodePtr& avro_node, ::avro::Decoder& decoder,
182+
Status DecodeListToBuilder(const ::avro::NodePtr& avro_node, ::avro::Decoder* decoder,
183183
arrow::ArrayBuilder* array_builder,
184-
AvroDirectDecoder::DecodeContext& ctx) {
184+
AvroDirectDecoder::DecodeContext* ctx) {
185185
if (avro_node->type() != ::avro::AVRO_ARRAY) {
186186
return Status::Invalid(
187187
fmt::format("Expected Avro array, got type: {}", ToString(avro_node)));
@@ -194,22 +194,22 @@ Status DecodeListToBuilder(const ::avro::NodePtr& avro_node, ::avro::Decoder& de
194194
const auto& element_node = avro_node->leafAt(0);
195195

196196
// Read array block count
197-
int64_t block_count = decoder.arrayStart();
197+
int64_t block_count = decoder->arrayStart();
198198
while (block_count != 0) {
199199
for (int64_t i = 0; i < block_count; ++i) {
200200
PAIMON_RETURN_NOT_OK(DecodeFieldToBuilder(element_node, /*projection=*/std::nullopt,
201201
decoder, value_builder, ctx));
202202
}
203-
block_count = decoder.arrayNext();
203+
block_count = decoder->arrayNext();
204204
}
205205

206206
return Status::OK();
207207
}
208208

209209
/// Decode Avro map directly to Arrow map builder.
210-
Status DecodeMapToBuilder(const ::avro::NodePtr& avro_node, ::avro::Decoder& decoder,
210+
Status DecodeMapToBuilder(const ::avro::NodePtr& avro_node, ::avro::Decoder* decoder,
211211
arrow::ArrayBuilder* array_builder,
212-
AvroDirectDecoder::DecodeContext& ctx) {
212+
AvroDirectDecoder::DecodeContext* ctx) {
213213
auto* map_builder = arrow::internal::checked_cast<arrow::MapBuilder*>(array_builder);
214214

215215
if (avro_node->type() == ::avro::AVRO_MAP) {
@@ -222,15 +222,15 @@ Status DecodeMapToBuilder(const ::avro::NodePtr& avro_node, ::avro::Decoder& dec
222222
auto* item_builder = map_builder->item_builder();
223223

224224
// Read map block count
225-
int64_t block_count = decoder.mapStart();
225+
int64_t block_count = decoder->mapStart();
226226
while (block_count != 0) {
227227
for (int64_t i = 0; i < block_count; ++i) {
228228
PAIMON_RETURN_NOT_OK(DecodeFieldToBuilder(key_node, /*projection=*/std::nullopt,
229229
decoder, key_builder, ctx));
230230
PAIMON_RETURN_NOT_OK(DecodeFieldToBuilder(value_node, /*projection=*/std::nullopt,
231231
decoder, item_builder, ctx));
232232
}
233-
block_count = decoder.mapNext();
233+
block_count = decoder->mapNext();
234234
}
235235
return Status::OK();
236236
} else if (avro_node->type() == ::avro::AVRO_ARRAY && HasMapLogicalType(avro_node)) {
@@ -249,15 +249,15 @@ Status DecodeMapToBuilder(const ::avro::NodePtr& avro_node, ::avro::Decoder& dec
249249
const auto& value_node = record_node->leafAt(1);
250250

251251
// Read array block count
252-
int64_t block_count = decoder.arrayStart();
252+
int64_t block_count = decoder->arrayStart();
253253
while (block_count != 0) {
254254
for (int64_t i = 0; i < block_count; ++i) {
255255
PAIMON_RETURN_NOT_OK(DecodeFieldToBuilder(key_node, /*projection=*/std::nullopt,
256256
decoder, key_builder, ctx));
257257
PAIMON_RETURN_NOT_OK(DecodeFieldToBuilder(value_node, /*projection=*/std::nullopt,
258258
decoder, item_builder, ctx));
259259
}
260-
block_count = decoder.arrayNext();
260+
block_count = decoder->arrayNext();
261261
}
262262
return Status::OK();
263263
} else {
@@ -269,21 +269,21 @@ Status DecodeMapToBuilder(const ::avro::NodePtr& avro_node, ::avro::Decoder& dec
269269
/// Decode Avro data directly to Arrow array builder.
270270
Status DecodeAvroValueToBuilder(const ::avro::NodePtr& avro_node,
271271
const std::optional<std::set<size_t>>& projection,
272-
::avro::Decoder& decoder, arrow::ArrayBuilder* array_builder,
273-
AvroDirectDecoder::DecodeContext& ctx) {
272+
::avro::Decoder* decoder, arrow::ArrayBuilder* array_builder,
273+
AvroDirectDecoder::DecodeContext* ctx) {
274274
auto type = avro_node->type();
275275
auto logical_type = avro_node->logicalType();
276276

277277
switch (type) {
278278
case ::avro::AVRO_BOOL: {
279279
auto* builder = arrow::internal::checked_cast<arrow::BooleanBuilder*>(array_builder);
280-
bool value = decoder.decodeBool();
280+
bool value = decoder->decodeBool();
281281
PAIMON_RETURN_NOT_OK_FROM_ARROW(builder->Append(value));
282282
return Status::OK();
283283
}
284284

285285
case ::avro::AVRO_INT: {
286-
int32_t value = decoder.decodeInt();
286+
int32_t value = decoder->decodeInt();
287287
auto arrow_type = array_builder->type();
288288
switch (arrow_type->id()) {
289289
case arrow::Type::INT8: {
@@ -323,7 +323,7 @@ Status DecodeAvroValueToBuilder(const ::avro::NodePtr& avro_node,
323323
}
324324

325325
case ::avro::AVRO_LONG: {
326-
int64_t value = decoder.decodeLong();
326+
int64_t value = decoder->decodeLong();
327327
switch (logical_type.type()) {
328328
case ::avro::LogicalType::Type::NONE: {
329329
auto* builder =
@@ -357,40 +357,41 @@ Status DecodeAvroValueToBuilder(const ::avro::NodePtr& avro_node,
357357

358358
case ::avro::AVRO_FLOAT: {
359359
auto* builder = arrow::internal::checked_cast<arrow::FloatBuilder*>(array_builder);
360-
float value = decoder.decodeFloat();
360+
float value = decoder->decodeFloat();
361361
PAIMON_RETURN_NOT_OK_FROM_ARROW(builder->Append(value));
362362
return Status::OK();
363363
}
364364
case ::avro::AVRO_DOUBLE: {
365365
auto* builder = arrow::internal::checked_cast<arrow::DoubleBuilder*>(array_builder);
366-
double value = decoder.decodeDouble();
366+
double value = decoder->decodeDouble();
367367
PAIMON_RETURN_NOT_OK_FROM_ARROW(builder->Append(value));
368368
return Status::OK();
369369
}
370370
case ::avro::AVRO_STRING: {
371371
auto* builder = arrow::internal::checked_cast<arrow::StringBuilder*>(array_builder);
372-
decoder.decodeString(ctx.string_scratch);
373-
PAIMON_RETURN_NOT_OK_FROM_ARROW(builder->Append(ctx.string_scratch));
372+
decoder->decodeString(ctx->string_scratch);
373+
PAIMON_RETURN_NOT_OK_FROM_ARROW(builder->Append(ctx->string_scratch));
374374
return Status::OK();
375375
}
376376

377377
case ::avro::AVRO_BYTES: {
378-
decoder.decodeBytes(ctx.bytes_scratch);
378+
decoder->decodeBytes(ctx->bytes_scratch);
379379
switch (logical_type.type()) {
380380
case ::avro::LogicalType::Type::NONE: {
381381
auto* builder =
382382
arrow::internal::checked_cast<arrow::BinaryBuilder*>(array_builder);
383-
PAIMON_RETURN_NOT_OK_FROM_ARROW(builder->Append(
384-
ctx.bytes_scratch.data(), static_cast<int32_t>(ctx.bytes_scratch.size())));
383+
PAIMON_RETURN_NOT_OK_FROM_ARROW(
384+
builder->Append(ctx->bytes_scratch.data(),
385+
static_cast<int32_t>(ctx->bytes_scratch.size())));
385386
return Status::OK();
386387
}
387388
case ::avro::LogicalType::Type::DECIMAL: {
388389
auto* builder =
389390
arrow::internal::checked_cast<arrow::Decimal128Builder*>(array_builder);
390391
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
391392
arrow::Decimal128 decimal,
392-
arrow::Decimal128::FromBigEndian(ctx.bytes_scratch.data(),
393-
ctx.bytes_scratch.size()));
393+
arrow::Decimal128::FromBigEndian(ctx->bytes_scratch.data(),
394+
ctx->bytes_scratch.size()));
394395
PAIMON_RETURN_NOT_OK_FROM_ARROW(builder->Append(decimal));
395396
return Status::OK();
396397
}
@@ -422,10 +423,10 @@ Status DecodeAvroValueToBuilder(const ::avro::NodePtr& avro_node,
422423

423424
Status DecodeFieldToBuilder(const ::avro::NodePtr& avro_node,
424425
const std::optional<std::set<size_t>>& projection,
425-
::avro::Decoder& decoder, arrow::ArrayBuilder* array_builder,
426-
AvroDirectDecoder::DecodeContext& ctx) {
426+
::avro::Decoder* decoder, arrow::ArrayBuilder* array_builder,
427+
AvroDirectDecoder::DecodeContext* ctx) {
427428
if (avro_node->type() == ::avro::AVRO_UNION) {
428-
const size_t branch_index = decoder.decodeUnionIndex();
429+
const size_t branch_index = decoder->decodeUnionIndex();
429430

430431
// Validate branch index
431432
const size_t num_branches = avro_node->leaves();
@@ -450,9 +451,9 @@ Status DecodeFieldToBuilder(const ::avro::NodePtr& avro_node,
450451

451452
Status AvroDirectDecoder::DecodeAvroToBuilder(const ::avro::NodePtr& avro_node,
452453
const std::optional<std::set<size_t>>& projection,
453-
::avro::Decoder& decoder,
454+
::avro::Decoder* decoder,
454455
arrow::ArrayBuilder* array_builder,
455-
DecodeContext& ctx) {
456+
DecodeContext* ctx) {
456457
return DecodeFieldToBuilder(avro_node, projection, decoder, array_builder, ctx);
457458
}
458459

‎src/paimon/format/avro/avro_direct_decoder.h‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -56,8 +56,8 @@ class AvroDirectDecoder {
5656
/// @return Status indicating success, or an error status
5757
static Status DecodeAvroToBuilder(const ::avro::NodePtr& avro_node,
5858
const std::optional<std::set<size_t>>& projection,
59-
::avro::Decoder& decoder, arrow::ArrayBuilder* array_builder,
60-
DecodeContext& ctx);
59+
::avro::Decoder* decoder, arrow::ArrayBuilder* array_builder,
60+
DecodeContext* ctx);
6161
};
6262

6363
} // namespace paimon::avro

‎src/paimon/format/avro/avro_file_batch_reader.cpp‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -102,8 +102,8 @@ Result<BatchReader::ReadBatch> AvroFileBatchReader::NextBatch() {
102102
}
103103
reader_->decr();
104104
PAIMON_RETURN_NOT_OK(AvroDirectDecoder::DecodeAvroToBuilder(
105-
reader_->dataSchema().root(), read_fields_projection_, reader_->decoder(),
106-
array_builder_.get(), decode_context_));
105+
reader_->dataSchema().root(), read_fields_projection_, &reader_->decoder(),
106+
array_builder_.get(), &decode_context_));
107107
}
108108
previous_first_row_ = next_row_to_read_;
109109
next_row_to_read_ += array_builder_->length();

0 commit comments

Comments
 (0)