Method flush

bolt/serializers/ArrowSerializer.cpp:537–709

  }

  void flush(OutputStream* stream) override {
    ::arrow::ipc::IpcWriteOptions wopts =
        ::arrow::ipc::IpcWriteOptions::Defaults();
    wopts.metadata_version = ::arrow::ipc::MetadataVersion::V4;

    // schema
    std::shared_ptr<::arrow::Schema> schema;
    if (!batches_.empty()) {
      schema = batches_.front()->schema();
    } else {
      ArrowSchema cSchema{};
      ArrowSchemaReleaser schemaRel(&cSchema);
      auto* pool = streamArena_->pool();
      auto empty = BaseVector::create<RowVector>(rowType_, 0, pool);
      exportToArrow(empty, cSchema, arrowBridgeOptions_, {});
      auto schRes = ::arrow::ImportSchema(&cSchema);
      BOLT_USER_CHECK(
          schRes.ok(),
          "ImportSchema(empty) failed: {}",
          schRes.status().ToString());
      schema = *schRes;
      schemaRel.disarm();
    }

#ifndef NDEBUG
    for (size_t i = 0; i < batches_.size(); ++i) {
      BOLT_CHECK(
          batches_[i]->schema()->Equals(*schema),
          "Schema mismatch at batch {}:\nexpected: {}\n   found: {}",
          i,
          schema->ToString(),
          batches_[i]->schema()->ToString());
    }
#endif

    const int32_t rowsToWrite = totalRows_;

    // header = rows | codec | uncompressed size | compressed size | checksum
    writeInt32(stream, rowsToWrite);
    const std::streampos codec_pos = stream->tellp();
    const char codecMask = 0;

    auto writeCodec = [&](char codecMask,
                          int32_t uncompressedSize,
                          int32_t compressedSize,
                          bool first = false) {
      auto pos = stream->tellp();
      stream->seekp(codec_pos);
      stream->write(&codecMask, 1);
      writeInt32(stream, uncompressedSize);
      writeInt32(stream, compressedSize);
#ifdef BOLT_ENABLE_CRC
      writeInt64(stream, 0);
#endif
      if (!first) {
        stream->seekp(pos);
        VLOG(1) << "[Flush] wrote codec header: "
                << " rows=" << rowsToWrite