ADD: Add DBNv2 decoding to C++ client

Author: threecgreen

CHANGELOG.md

@@ -1,8 +1,33 @@
 # Changelog
 
-## 0.13.2 - TBD
+## 0.14.0 - TBD
 ### Enhancements
+- Added support for DBN encoding version 2 (DBNv2), affecting `SymbolMappingMsg`,
+  `InstrumentDefMsg`, and `Metadata`
+  - Version 1 structs can be converted to version 2 structs with the `ToV2()` method
+- Added `symbol_cstr_len` field to `Metadata` to indicate the length of fixed symbol
+  strings
+- Added `stype_in` and `stype_out` fields to `SymbolMappingMsg` to provide more context
+  with live symbology updates
+- Added `IndexTs` methods to every record type which returns the primary timestamp
+- Added `VersionUpgradePolicy` enum to allow specifying how to handle decoding records
+  from prior DBN versions
+- Added `InstrumentDefMsgV2` and `SymbolMappingMsgV2` type aliases
+- Added `kDbnVersion` constant for current DBN version
+- Added `kSymbolCstrLen`, `kSymbolCstrLenV1`, and `kSymbolCstrLenV2` constants for the
+  length of fixed-length symbol strings in different DBN versions
 - Added new publisher values in preparation for IFEU.IMPACT and NDEX.IMPACT datasets
+- Added `kMaxRecordLen` constant for the the length of the largest record type
+
+### Breaking changes
+- The old `InstrumentDefMsg` is now `InstrumentDefMsgV1` in `compat.hpp`
+- The old `SymbolMappingMsg` is now `SymbolMappingMsgV1` in `compat.hpp`
+- Converted the following enums to enum classes to allow safely adding new variants:
+  `SecurityUpdateAction` and `SType`
+- Renamed `dummy` to `reserved` in `InstrumentDefMsg`
+- Removed `reserved2`, `reserved3`, `reserved4`, and `reserved5` from `InstrumentDefMsg`
+- Moved position of `strike_price` within `InstrumentDefMsg`
+- Removed deprecated `SecurityUpdateAction::Invalid` variant
 
 ## 0.13.1 - 2023-10-23
 ### Enhancements

cmake/SourcesAndHeaders.cmake

@@ -1,5 +1,6 @@
 set(headers
   include/databento/batch.hpp
+  include/databento/compat.hpp
   include/databento/constants.hpp
   include/databento/datetime.hpp
   include/databento/dbn.hpp
@@ -34,6 +35,7 @@ set(headers
 
 set(sources
   src/batch.cpp
+  src/compat.cpp
   src/datetime.cpp
   src/dbn.cpp
   src/dbn_decoder.cpp

include/databento/compat.hpp

@@ -0,0 +1,156 @@
+// Record definitions from previous DBN versions and helper functions.
+#pragma once
+
+#include <cstddef>  // size_t
+#include <cstdint>
+
+#include "databento/constants.hpp"  // kSymbolCstrLen
+#include "databento/datetime.hpp"   // UnixNanos
+#include "databento/enums.hpp"
+#include "databento/record.hpp"
+
+namespace databento {
+static constexpr std::size_t kSymbolCstrLenV1 = 22;
+static constexpr std::size_t kSymbolCstrLenV2 = kSymbolCstrLen;
+
+constexpr std::size_t VersionSymbolCstrLen(std::uint8_t version) {
+  if (version < 2) {
+    return kSymbolCstrLenV1;
+  }
+  return kSymbolCstrLenV2;
+}
+
+using InstrumentDefMsgV2 = InstrumentDefMsg;
+using SymbolMappingMsgV2 = SymbolMappingMsg;
+
+// DBN version 1 instrument definition.
+struct InstrumentDefMsgV1 {
+  static bool HasRType(RType rtype) { return rtype == RType::InstrumentDef; }
+
+  InstrumentDefMsgV2 ToV2() const;
+  const char* Currency() const { return currency.data(); }
+  const char* SettlCurrency() const { return settl_currency.data(); }
+  const char* SecSubType() const { return secsubtype.data(); }
+  const char* RawSymbol() const { return raw_symbol.data(); }
+  const char* Group() const { return group.data(); }
+  const char* Exchange() const { return exchange.data(); }
+  const char* Asset() const { return asset.data(); }
+  const char* Cfi() const { return cfi.data(); }
+  const char* SecurityType() const { return security_type.data(); }
+  const char* UnitOfMeasure() const { return unit_of_measure.data(); }
+  const char* Underlying() const { return underlying.data(); }
+  const char* StrikePriceCurrency() const {
+    return strike_price_currency.data();
+  }
+
+  RecordHeader hd;
+  UnixNanos ts_recv;
+  std::int64_t min_price_increment;
+  std::int64_t display_factor;
+  UnixNanos expiration;
+  UnixNanos activation;
+  std::int64_t high_limit_price;
+  std::int64_t low_limit_price;
+  std::int64_t max_price_variation;
+  std::int64_t trading_reference_price;
+  std::int64_t unit_of_measure_qty;
+  std::int64_t min_price_increment_amount;
+  std::int64_t price_ratio;
+  std::int32_t inst_attrib_value;
+  std::uint32_t underlying_id;
+  std::uint32_t raw_instrument_id;
+  std::int32_t market_depth_implied;
+  std::int32_t market_depth;
+  std::uint32_t market_segment_id;
+  std::uint32_t max_trade_vol;
+  std::int32_t min_lot_size;
+  std::int32_t min_lot_size_block;
+  std::int32_t min_lot_size_round_lot;
+  std::uint32_t min_trade_vol;
+  std::array<char, 4> _reserved2;
+  std::int32_t contract_multiplier;
+  std::int32_t decay_quantity;
+  std::int32_t original_contract_size;
+  std::array<char, 4> _reserved3;
+  std::uint16_t trading_reference_date;
+  std::int16_t appl_id;
+  std::uint16_t maturity_year;
+  std::uint16_t decay_start_date;
+  std::uint16_t channel_id;
+  std::array<char, 4> currency;
+  std::array<char, 4> settl_currency;
+  std::array<char, 6> secsubtype;
+  std::array<char, kSymbolCstrLenV1> raw_symbol;
+  std::array<char, 21> group;
+  std::array<char, 5> exchange;
+  std::array<char, 7> asset;
+  std::array<char, 7> cfi;
+  std::array<char, 7> security_type;
+  std::array<char, 31> unit_of_measure;
+  std::array<char, 21> underlying;
+  std::array<char, 4> strike_price_currency;
+  InstrumentClass instrument_class;
+  std::array<char, 2> _reserved4;
+  std::int64_t strike_price;
+  std::array<char, 6> _reserved5;
+  MatchAlgorithm match_algorithm;
+  std::uint8_t md_security_trading_status;
+  std::uint8_t main_fraction;
+  std::uint8_t price_display_format;
+  std::uint8_t settl_price_type;
+  std::uint8_t sub_fraction;
+  std::uint8_t underlying_product;
+  SecurityUpdateAction security_update_action;
+  std::uint8_t maturity_month;
+  std::uint8_t maturity_day;
+  std::uint8_t maturity_week;
+  UserDefinedInstrument user_defined_instrument;
+  std::int8_t contract_multiplier_unit;
+  std::int8_t flow_schedule_type;
+  std::uint8_t tick_rule;
+  // padding for alignment
+  std::array<char, 3> dummy;
+};
+static_assert(sizeof(InstrumentDefMsgV1) == 360);
+
+/// A symbol mapping message.
+struct SymbolMappingMsgV1 {
+  static bool HasRType(RType rtype) { return rtype == RType::SymbolMapping; }
+
+  SymbolMappingMsgV2 ToV2() const;
+  const char* STypeInSymbol() const { return stype_in_symbol.data(); }
+  const char* STypeOutSymbol() const { return stype_out_symbol.data(); }
+
+  RecordHeader hd;
+  std::array<char, kSymbolCstrLenV1> stype_in_symbol;
+  std::array<char, kSymbolCstrLenV1> stype_out_symbol;
+  // padding for alignment
+  std::array<char, 4> dummy;
+  UnixNanos start_ts;
+  UnixNanos end_ts;
+};
+static_assert(sizeof(SymbolMappingMsgV1) == 80);
+
+bool operator==(const InstrumentDefMsgV1& lhs, const InstrumentDefMsgV1& rhs);
+inline bool operator!=(const InstrumentDefMsgV1& lhs,
+                       const InstrumentDefMsgV1& rhs) {
+  return !(lhs == rhs);
+}
+inline bool operator==(const SymbolMappingMsgV1& lhs,
+                       const SymbolMappingMsgV1& rhs) {
+  return std::tie(lhs.hd, lhs.stype_in_symbol, lhs.stype_out_symbol,
+                  lhs.start_ts, lhs.end_ts) ==
+         std::tie(rhs.hd, rhs.stype_in_symbol, rhs.stype_out_symbol,
+                  rhs.start_ts, rhs.end_ts);
+}
+inline bool operator!=(const SymbolMappingMsgV1& lhs,
+                       const SymbolMappingMsgV1& rhs) {
+  return !(lhs == rhs);
+}
+std::string ToString(const InstrumentDefMsgV1& instr_def_msg);
+std::ostream& operator<<(std::ostream& stream,
+                         const InstrumentDefMsgV1& instr_def_msg);
+std::string ToString(const SymbolMappingMsgV1& symbol_mapping_msg);
+std::ostream& operator<<(std::ostream& stream,
+                         const SymbolMappingMsgV1& symbol_mapping_msg);
+}  // namespace databento

include/databento/constants.hpp

@@ -20,6 +20,10 @@ static constexpr auto kUndefStatQuantity =
 // The sentinel value for an unset or null timestamp.
 static constexpr auto kUndefTimestamp =
     std::numeric_limits<std::uint64_t>::max();
+// The current version of the DBN encoding.
+static constexpr auto kDbnVersion = 2;
+// The length of fixed-length symbol strings.
+static constexpr auto kSymbolCstrLen = 71;
 
 // This is not necessarily a comprehensive list of available datasets. Please
 // use `Historical.MetadataListDatasets` to retrieve an up-to-date list.

include/databento/dbn.hpp

@@ -56,6 +56,9 @@ struct Metadata {
   SType stype_out;
   // Whether the records contain an appended send timestamp.
   bool ts_out;
+  // The length in bytes of fixed-length symbol strings, including a null
+  // terminator byte.
+  std::size_t symbol_cstr_len;
   // The original query input symbols from the request.
   std::vector<std::string> symbols;
   // Symbols that did not resolve for _at least one day_ in the query time
@@ -91,6 +94,7 @@ inline bool operator==(const Metadata& lhs, const Metadata& rhs) {
          (lhs.has_mixed_stype_in ? rhs.has_mixed_stype_in
                                  : lhs.stype_in == rhs.stype_in) &&
          lhs.stype_out == rhs.stype_out && lhs.ts_out == rhs.ts_out &&
+         lhs.symbol_cstr_len == rhs.symbol_cstr_len &&
          lhs.symbols == rhs.symbols && lhs.partial == rhs.partial &&
          lhs.not_found == rhs.not_found && lhs.mappings == rhs.mappings;
 }

include/databento/dbn_decoder.hpp

@@ -8,50 +8,67 @@
 #include "databento/dbn.hpp"
 #include "databento/detail/file_stream.hpp"
 #include "databento/detail/shared_channel.hpp"
+#include "databento/enums.hpp"  // Upgrade Policy
 #include "databento/ireadable.hpp"
 #include "databento/record.hpp"
 
 namespace databento {
-// DBN decoder. Use either the DbnChannelDecoder or DbnFileDecoder
-// specialization.
+// DBN decoder. Set upgrade_policy to control how DBN version 1 data should be
+// handled. Currently it defaults to returning this data as-is, but this default
+// will change in a future version.
 class DbnDecoder {
  public:
   explicit DbnDecoder(detail::SharedChannel channel);
   explicit DbnDecoder(detail::FileStream file_stream);
   explicit DbnDecoder(std::unique_ptr<IReadable> input);
+  DbnDecoder(std::unique_ptr<IReadable> input,
+             VersionUpgradePolicy upgrade_policy);
 
   // Decode metadata from the given buffer.
   static Metadata DecodeMetadata(const std::vector<std::uint8_t>& buffer);
   static std::pair<std::uint8_t, std::size_t> DecodeMetadataVersionAndSize(
       const std::uint8_t* buffer, std::size_t size);
   static Metadata DecodeMetadataFields(std::uint8_t version,
                                        const std::vector<std::uint8_t>& buffer);
+  // Decodes a record possibly applying upgrading the data according to the
+  // given version and upgrade policy. If an upgrade is applied,
+  // compat_buffer is modified.
+  static Record DecodeRecordCompat(
+      std::uint8_t version, VersionUpgradePolicy upgrade_policy,
+      std::array<std::uint8_t, kMaxRecordLen>* compat_buffer, Record rec);
 
-  // Should only be called once
+  // Should be called exactly once.
   Metadata DecodeMetadata();
   // 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(
+      std::size_t symbol_cstr_len,
       std::vector<std::uint8_t>::const_iterator& buffer_it);
   static std::vector<std::string> DecodeRepeatedSymbol(
+      std::size_t symbol_cstr_len,
       std::vector<std::uint8_t>::const_iterator& buffer_it,
       std::vector<std::uint8_t>::const_iterator buffer_end_it);
   static std::vector<SymbolMapping> DecodeSymbolMappings(
+      std::size_t symbol_cstr_len,
       std::vector<std::uint8_t>::const_iterator& buffer_it,
       std::vector<std::uint8_t>::const_iterator buffer_end_it);
   static SymbolMapping DecodeSymbolMapping(
+      std::size_t symbol_cstr_len,
       std::vector<std::uint8_t>::const_iterator& buffer_it,
       std::vector<std::uint8_t>::const_iterator buffer_end_it);
   bool DetectCompression();
   std::size_t FillBuffer();
   RecordHeader* BufferRecordHeader();
 
+  std::uint8_t version_{};
+  VersionUpgradePolicy upgrade_policy_;
   std::unique_ptr<IReadable> input_;
-  std::vector<std::uint8_t> buffer_;
+  std::vector<std::uint8_t> read_buffer_;
   std::size_t buffer_idx_{};
+  std::array<std::uint8_t, kMaxRecordLen> compat_buffer_{};
   Record current_record_{nullptr};
 };
 }  // namespace databento

include/databento/dbn_file_store.hpp

@@ -1,17 +1,19 @@
 #pragma once
 
-#include <memory>  // unique_ptr
 #include <string>
 
 #include "databento/dbn.hpp"          // Metadata
 #include "databento/dbn_decoder.hpp"  // DbnDecoder
+#include "databento/enums.hpp"        // VersionUpgradePolicy
 #include "databento/timeseries.hpp"   // MetadataCallback, RecordCallback
 
 namespace databento {
 // A reader for DBN files.
 class DbnFileStore {
  public:
   explicit DbnFileStore(const std::string& file_path);
+  DbnFileStore(const std::string& file_path,
+               VersionUpgradePolicy upgrade_policy);
 
   void Replay(const MetadataCallback& metadata_callback,
               const RecordCallback& record_callback);