| 389 | } |
| 390 | |
| 391 | int64_t WriteDataPage(const DataPage& page) override { |
| 392 | const int64_t uncompressed_size = page.uncompressed_size(); |
| 393 | if (uncompressed_size > std::numeric_limits<int32_t>::max()) { |
| 394 | throw ParquetException("Uncompressed data page size overflows INT32_MAX. Size:", |
| 395 | uncompressed_size); |
| 396 | } |
| 397 | |
| 398 | std::shared_ptr<Buffer> compressed_data = page.buffer(); |
| 399 | const uint8_t* output_data_buffer = compressed_data->data(); |
| 400 | int64_t output_data_len = compressed_data->size(); |
| 401 | |
| 402 | if (output_data_len > std::numeric_limits<int32_t>::max()) { |
| 403 | throw ParquetException("Compressed data page size overflows INT32_MAX. Size:", |
| 404 | output_data_len); |
| 405 | } |
| 406 | |
| 407 | if (data_encryptor_.get()) { |
| 408 | PARQUET_THROW_NOT_OK(encryption_buffer_->Resize( |
| 409 | data_encryptor_->CiphertextLength(output_data_len), false)); |
| 410 | UpdateEncryption(encryption::kDataPage); |
| 411 | output_data_len = |
| 412 | data_encryptor_->Encrypt(compressed_data->span_as<uint8_t>(), |
| 413 | encryption_buffer_->mutable_span_as<uint8_t>()); |
| 414 | output_data_buffer = encryption_buffer_->data(); |
| 415 | } |
| 416 | |
| 417 | format::PageHeader page_header; |
| 418 | page_header.__set_uncompressed_page_size(static_cast<int32_t>(uncompressed_size)); |
| 419 | page_header.__set_compressed_page_size(static_cast<int32_t>(output_data_len)); |
| 420 | |
| 421 | if (page_checksum_verification_) { |
| 422 | uint32_t crc32 = |
| 423 | ::arrow::internal::crc32(/* prev */ 0, output_data_buffer, output_data_len); |
| 424 | page_header.__set_crc(static_cast<int32_t>(crc32)); |
| 425 | } |
| 426 | |
| 427 | if (page.type() == PageType::DATA_PAGE) { |
| 428 | const DataPageV1& v1_page = checked_cast<const DataPageV1&>(page); |
| 429 | SetDataPageHeader(page_header, v1_page); |
| 430 | } else if (page.type() == PageType::DATA_PAGE_V2) { |
| 431 | const DataPageV2& v2_page = checked_cast<const DataPageV2&>(page); |
| 432 | SetDataPageV2Header(page_header, v2_page); |
| 433 | } else { |
| 434 | throw ParquetException("Unexpected page type"); |
| 435 | } |
| 436 | |
| 437 | PARQUET_ASSIGN_OR_THROW(int64_t start_pos, sink_->Tell()); |
| 438 | if (page_ordinal_ == 0) { |
| 439 | data_page_offset_ = start_pos; |
| 440 | } |
| 441 | |
| 442 | if (meta_encryptor_) { |
| 443 | UpdateEncryption(encryption::kDataPageHeader); |
| 444 | } |
| 445 | const int64_t header_size = |
| 446 | thrift_serializer_->Serialize(&page_header, sink_.get(), meta_encryptor_.get()); |
| 447 | PARQUET_THROW_NOT_OK(sink_->Write(output_data_buffer, output_data_len)); |
| 448 |
nothing calls this directly
no test coverage detected