OPT: Cache whether upgrade may be required in C++

Author: threecgreen

CHANGELOG.md

@@ -3,8 +3,11 @@
 ## 0.55.0 - TBD
 
 ### Enhancements
-- Added new publisher values for Cboe Titanium Cboe Global Indices Feed
+- Improved `DbnDecoder` throughput on current-version data and `AsIs` workloads by
+  caching whether the upgrade policy-version combination requires upgrading, skipping
+  the per-record `DecodeRecordCompat` dispatch on the fast path
 - Made `detail::Buffer` shifts explicit to avoid redundant moves during record decoding
+- Added new publisher values for Cboe Titanium Cboe Global Indices Feed
 - Added `Year` to `SplitDuration` enum for yearly historical batch job submissions
 
 ## 0.54.0 - 2026-04-21

include/databento/dbn_decoder.hpp

@@ -35,6 +35,9 @@ class DbnDecoder {
                                    VersionUpgradePolicy upgrade_policy, bool ts_out,
                                    std::array<std::byte, kMaxRecordLen>* compat_buffer,
                                    Record rec);
+  // Returns whether a record from `version`-formatted data requires runtime
+  // upgrade dispatch under `upgrade_policy`.
+  static bool NeedsUpgrade(VersionUpgradePolicy upgrade_policy, std::uint8_t version);
 
   // Should be called exactly once.
   Metadata DecodeMetadata();
@@ -61,6 +64,7 @@ class DbnDecoder {
   ILogReceiver* log_receiver_;
   std::uint8_t version_{};
   VersionUpgradePolicy upgrade_policy_;
+  bool needs_upgrade_{true};
   bool ts_out_{};
   std::unique_ptr<IReadable> input_;
   detail::Buffer buffer_{};

include/databento/detail/dbn_buffer_decoder.hpp

@@ -61,6 +61,7 @@ class DbnBufferDecoder {
   alignas(RecordHeader) std::array<std::byte, kMaxRecordLen> compat_buffer_{};
   std::uint8_t input_version_{};
   bool ts_out_{};
+  bool needs_upgrade_{true};
   DecoderState state_{DecoderState::Init};
 };
 }  // namespace databento::detail

src/dbn_decoder.cpp

@@ -173,6 +173,7 @@ databento::Metadata DbnDecoder::DecodeMetadata() {
       buffer_.ReadBegin(), kMetadataPreludeSize);
   buffer_.Consume(kMetadataPreludeSize);
   version_ = version;
+  needs_upgrade_ = NeedsUpgrade(upgrade_policy_, version_);
   buffer_.Reserve(size);
   input_->ReadExact(buffer_.WriteBegin(), size);
   buffer_.Fill(size);
@@ -273,6 +274,19 @@ databento::Record DbnDecoder::DecodeRecordCompat(
   return rec;
 }
 
+bool DbnDecoder::NeedsUpgrade(VersionUpgradePolicy upgrade_policy,
+                              std::uint8_t version) {
+  switch (upgrade_policy) {
+    case VersionUpgradePolicy::UpgradeToV2:
+      return version < 2;
+    case VersionUpgradePolicy::UpgradeToV3:
+      return version < 3;
+    case VersionUpgradePolicy::AsIs:
+    default:
+      return false;
+  }
+}
+
 // assumes DecodeMetadata has been called
 const databento::Record* DbnDecoder::DecodeRecord() {
   // need some unread unread_bytes
@@ -294,8 +308,10 @@ const databento::Record* DbnDecoder::DecodeRecord() {
   }
   current_record_ = Record{BufferRecordHeader()};
   buffer_.Consume(current_record_.Size());
-  current_record_ = DbnDecoder::DecodeRecordCompat(version_, upgrade_policy_, ts_out_,
-                                                   &compat_buffer_, current_record_);
+  if (needs_upgrade_) {
+    current_record_ = DbnDecoder::DecodeRecordCompat(version_, upgrade_policy_, ts_out_,
+                                                     &compat_buffer_, current_record_);
+  }
   return &current_record_;
 }
 

src/detail/dbn_buffer_decoder.cpp

@@ -25,6 +25,7 @@ databento::KeepGoing DbnBufferDecoder::Process(const char* data, std::size_t len
         std::tie(input_version_, bytes_needed_) =
             DbnDecoder::DecodeMetadataVersionAndSize(dbn_buffer_.ReadBegin(),
                                                      dbn_buffer_.ReadCapacity());
+        needs_upgrade_ = DbnDecoder::NeedsUpgrade(upgrade_policy_, input_version_);
         dbn_buffer_.Consume(kMetadataPreludeSize);
         dbn_buffer_.Reserve(bytes_needed_);
         state_ = DecoderState::Metadata;
@@ -56,8 +57,10 @@ databento::KeepGoing DbnBufferDecoder::Process(const char* data, std::size_t len
           if (dbn_buffer_.ReadCapacity() < bytes_needed_) {
             break;
           }
-          record = DbnDecoder::DecodeRecordCompat(input_version_, upgrade_policy_,
-                                                  ts_out_, &compat_buffer_, record);
+          if (needs_upgrade_) {
+            record = DbnDecoder::DecodeRecordCompat(input_version_, upgrade_policy_,
+                                                    ts_out_, &compat_buffer_, record);
+          }
           if (record_callback_(record) == KeepGoing::Stop) {
             return KeepGoing::Stop;
           }
@@ -78,6 +81,7 @@ std::ostream& operator<<(std::ostream& stream, const DbnBufferDecoder& buffer) {
       .AddField("bytes_needed_", buffer.bytes_needed_)
       .AddField("input_version_", buffer.input_version_)
       .AddField("ts_out_", buffer.ts_out_)
+      .AddField("needs_upgrade_", buffer.needs_upgrade_)
       .AddField("state_", buffer.state_)
       .Finish();
 }

tests/src/dbn_decoder_tests.cpp

@@ -217,6 +217,29 @@ TEST_F(DbnDecoderTests, TestUpgradeMbp1WithTsOut) {
   ASSERT_EQ(&orig, &upgraded);
 }
 
+TEST_F(DbnDecoderTests, TestNeedsUpgrade) {
+  struct Case {
+    VersionUpgradePolicy policy;
+    std::uint8_t version;
+    bool expected;
+  };
+  constexpr Case kCases[] = {
+      {VersionUpgradePolicy::AsIs, 1, false},
+      {VersionUpgradePolicy::AsIs, 2, false},
+      {VersionUpgradePolicy::AsIs, 3, false},
+      {VersionUpgradePolicy::UpgradeToV2, 1, true},
+      {VersionUpgradePolicy::UpgradeToV2, 2, false},
+      {VersionUpgradePolicy::UpgradeToV3, 1, true},
+      {VersionUpgradePolicy::UpgradeToV3, 2, true},
+      {VersionUpgradePolicy::UpgradeToV3, 3, false},
+  };
+  for (const auto& c : kCases) {
+    EXPECT_EQ(DbnDecoder::NeedsUpgrade(c.policy, c.version), c.expected)
+        << "policy=" << static_cast<int>(c.policy)
+        << " version=" << static_cast<int>(c.version);
+  }
+}
+
 class DbnDecoderSchemaTests
     : public DbnDecoderTests,
       public testing::WithParamInterface<std::pair<const char*, std::uint8_t>> {};