DEL: Remove decoding and usage of record_count

Author: threecgreen

CHANGELOG.md

@@ -3,6 +3,7 @@
 ## 0.6.0 - TBD
 - Added support for imbalance schema
 - Added support for decoding `ts_out` field
+- Removed `record_count` from `Metadata`
 - Changed `Historical::BatchDownload` to return the paths of the downloaded files
 - Added flags `kSnapshot` and `kMaybeBadBook`
 

include/databento/dbn.hpp

@@ -40,8 +40,6 @@ struct Metadata {
   UnixNanos end;
   // The maximum number of records for the query.
   std::uint64_t limit;
-  // The total number of records.
-  std::uint64_t record_count;
   // The input symbology type.
   SType stype_in;
   // The output symbology type.
@@ -71,10 +69,9 @@ inline bool operator==(const Metadata& lhs, const Metadata& rhs) {
   return lhs.version == rhs.version && lhs.dataset == rhs.dataset &&
          lhs.schema == rhs.schema && lhs.start == rhs.start &&
          lhs.end == rhs.end && lhs.limit == rhs.limit &&
-         lhs.record_count == rhs.record_count && lhs.stype_in == rhs.stype_in &&
-         lhs.stype_out == rhs.stype_out && lhs.symbols == rhs.symbols &&
-         lhs.partial == rhs.partial && lhs.not_found == rhs.not_found &&
-         lhs.mappings == rhs.mappings;
+         lhs.stype_in == rhs.stype_in && lhs.stype_out == rhs.stype_out &&
+         lhs.symbols == rhs.symbols && lhs.partial == rhs.partial &&
+         lhs.not_found == rhs.not_found && lhs.mappings == rhs.mappings;
 }
 
 std::string ToString(const Metadata& metadata);

include/databento/dbn_decoder.hpp

@@ -29,8 +29,9 @@ class DbnDecoder {
 
   // Should only be called once
   Metadata DecodeMetadata();
-  // Lifetime of returned Record is until next call to ParseRecord.
-  Record DecodeRecord();
+  // Lifetime of returned Record is until next call to DecodeRecord. Returns
+  // nullptr once the end of the input has been reached.
+  const Record* DecodeRecord();
 
  private:
   static std::string DecodeSymbol(
@@ -51,5 +52,6 @@ class DbnDecoder {
   std::unique_ptr<IReadable> input_;
   std::vector<std::uint8_t> buffer_;
   std::size_t buffer_idx_{};
+  Record current_record_{nullptr};
 };
 }  // namespace databento

src/dbn.cpp

@@ -19,7 +19,6 @@ std::ostream& operator<<(std::ostream& stream, const Metadata& metadata) {
                     .AddField("start", metadata.start)
                     .AddField("end", metadata.end)
                     .AddField("limit", metadata.limit)
-                    .AddField("record_count", metadata.record_count)
                     .AddField("stype_in", metadata.stype_in)
                     .AddField("stype_out", metadata.stype_out);
 

src/dbn_decoder.cpp

@@ -105,7 +105,8 @@ databento::Metadata DbnDecoder::DecodeMetadataFields(
   res.end =
       UnixNanos{std::chrono::nanoseconds{Consume<std::uint64_t>(buffer_it)}};
   res.limit = Consume<std::uint64_t>(buffer_it);
-  res.record_count = Consume<std::uint64_t>(buffer_it);
+  // skip deprecated record_count
+  buffer_it += 8;
   res.stype_in = static_cast<SType>(Consume<std::uint8_t>(buffer_it));
   res.stype_out = static_cast<SType>(Consume<std::uint8_t>(buffer_it));
   // skip reserved
@@ -136,23 +137,23 @@ databento::Metadata DbnDecoder::DecodeMetadata() {
 }
 
 // assumes ParseMetadata has been called
-databento::Record DbnDecoder::DecodeRecord() {
+const databento::Record* DbnDecoder::DecodeRecord() {
   // need some unread unread_bytes
   const auto unread_bytes = buffer_.size() - buffer_idx_;
   if (unread_bytes == 0) {
     if (FillBuffer() == 0) {
-      throw DbnResponseError{"Reached end of DBN stream"};
+      return nullptr;
     }
   }
   // check length
   while (buffer_.size() - buffer_idx_ < BufferRecordHeader()->Size()) {
     if (FillBuffer() == 0) {
-      throw DbnResponseError{"Reached end of DBN stream"};
+      return nullptr;
     }
   }
-  Record res{BufferRecordHeader()};
-  buffer_idx_ += res.Size();
-  return res;
+  current_record_ = Record{BufferRecordHeader()};
+  buffer_idx_ += current_record_.Size();
+  return &current_record_;
 }
 
 size_t DbnDecoder::FillBuffer() {

src/dbn_file_store.cpp

@@ -12,13 +12,12 @@ DbnFileStore::DbnFileStore(const std::string& file_path)
 void DbnFileStore::Replay(const MetadataCallback& metadata_callback,
                           const RecordCallback& record_callback) {
   auto metadata = parser_.DecodeMetadata();
-  const auto record_count = metadata.record_count;
   if (metadata_callback) {
     metadata_callback(std::move(metadata));
   }
-  for (std::size_t i = 0; i < record_count; ++i) {
-    const auto record = parser_.DecodeRecord();
-    if (record_callback(record) == KeepGoing::Stop) {
+  const databento::Record* record;
+  while ((record = parser_.DecodeRecord()) != nullptr) {
+    if (record_callback(*record) == KeepGoing::Stop) {
       break;
     }
   }

src/historical.cpp

@@ -1071,14 +1071,12 @@ void Historical::TimeseriesGetRange(const HttplibParams& params,
   try {
     DbnDecoder dbn_decoder{channel};
     Metadata metadata = dbn_decoder.DecodeMetadata();
-    const auto record_count = metadata.record_count;
     if (metadata_callback) {
       metadata_callback(std::move(metadata));
     }
-    for (auto i = 0UL; i < record_count; ++i) {
-      const bool should_stop =
-          record_callback(dbn_decoder.DecodeRecord()) == KeepGoing::Stop;
-      if (should_stop) {
+    const Record* record;
+    while ((record = dbn_decoder.DecodeRecord()) != nullptr) {
+      if (record_callback(*record) == KeepGoing::Stop) {
         should_continue = false;
         break;
       }