| 1420 | } |
| 1421 | |
| 1422 | Status WriteArrow(const int16_t* def_levels, const int16_t* rep_levels, |
| 1423 | int64_t num_levels, const ::arrow::Array& leaf_array, |
| 1424 | ArrowWriteContext* ctx, bool leaf_field_nullable) override { |
| 1425 | BEGIN_PARQUET_CATCH_EXCEPTIONS |
| 1426 | // Leaf nulls are canonical when there is only a single null element after a list |
| 1427 | // and it is at the leaf. |
| 1428 | bool single_nullable_element = |
| 1429 | (level_info_.def_level == level_info_.repeated_ancestor_def_level + 1) && |
| 1430 | leaf_field_nullable; |
| 1431 | if (!leaf_field_nullable && leaf_array.null_count() != 0) { |
| 1432 | return Status::Invalid("Column '", descr_->name(), |
| 1433 | "' is declared non-nullable but contains nulls"); |
| 1434 | } |
| 1435 | bool maybe_parent_nulls = level_info_.HasNullableValues() && !single_nullable_element; |
| 1436 | if (maybe_parent_nulls) { |
| 1437 | ARROW_ASSIGN_OR_RAISE( |
| 1438 | bits_buffer_, |
| 1439 | ::arrow::AllocateResizableBuffer( |
| 1440 | bit_util::BytesForBits(properties_->write_batch_size()), ctx->memory_pool)); |
| 1441 | bits_buffer_->ZeroPadding(); |
| 1442 | } |
| 1443 | |
| 1444 | if (ARROW_PREDICT_FALSE(properties_->content_defined_chunking_enabled())) { |
| 1445 | DCHECK(content_defined_chunker_.has_value()); |
| 1446 | auto chunks = content_defined_chunker_->GetChunks(def_levels, rep_levels, |
| 1447 | num_levels, leaf_array); |
| 1448 | for (size_t i = 0; i < chunks.size(); i++) { |
| 1449 | auto chunk = chunks[i]; |
| 1450 | auto chunk_array = leaf_array.Slice(chunk.value_offset); |
| 1451 | auto chunk_def_levels = AddIfNotNull(def_levels, chunk.level_offset); |
| 1452 | auto chunk_rep_levels = AddIfNotNull(rep_levels, chunk.level_offset); |
| 1453 | if (leaf_array.type()->id() == ::arrow::Type::DICTIONARY) { |
| 1454 | ARROW_CHECK_OK(WriteArrowDictionary(chunk_def_levels, chunk_rep_levels, |
| 1455 | chunk.levels_to_write, *chunk_array, ctx, |
| 1456 | maybe_parent_nulls)); |
| 1457 | } else { |
| 1458 | ARROW_CHECK_OK(WriteArrowDense(chunk_def_levels, chunk_rep_levels, |
| 1459 | chunk.levels_to_write, *chunk_array, ctx, |
| 1460 | maybe_parent_nulls)); |
| 1461 | } |
| 1462 | bool is_last_chunk = i == (chunks.size() - 1); |
| 1463 | if (num_buffered_values_ > 0 && !is_last_chunk) { |
| 1464 | // Explicitly add a new data page according to the content-defined chunk |
| 1465 | // boundaries. This way the same chunks will have the same byte-sequence |
| 1466 | // in the resulting file, which can be identified by content addressible |
| 1467 | // storage. |
| 1468 | // Note that the last chunk doesn't trigger a new data page in order to |
| 1469 | // allow subsequent WriteArrow() calls to continue writing to the same |
| 1470 | // data page, the chunker's state is not being reset after the last chunk. |
| 1471 | AddDataPage(); |
| 1472 | } |
| 1473 | } |
| 1474 | return Status::OK(); |
| 1475 | } else { |
| 1476 | if (leaf_array.type()->id() == ::arrow::Type::DICTIONARY) { |
| 1477 | return WriteArrowDictionary(def_levels, rep_levels, num_levels, leaf_array, ctx, |
| 1478 | maybe_parent_nulls); |
| 1479 | } else { |