From c4dfefe3d199f9af16bb2678a60eb24fe6b43831 Mon Sep 17 00:00:00 2001
From: Glauber Costa <glommer@cloudius-systems.com>
Date: Wed, 6 May 2015 18:14:35 -0400
Subject: [PATCH] sstable key: accept markers other than zero

Composites have an extra byte at the end of each element. The middle bytes
are expected to be zero, but the last one may not always be. In cases like
those of a range tombstone, we will have a marker with special significance.

Since we are exploding the composite into its components, we don't actually
need to retrieve it. We merely need to make sure that the marker is the one
we expect.

Signed-off-by: Glauber Costa <glommer@cloudius-systems.com>
---
 sstables/key.cc |  5 +++--
 sstables/key.hh | 14 ++++++++++++++
 2 files changed, 17 insertions(+), 2 deletions(-)

diff --git a/sstables/key.cc b/sstables/key.cc
index d35f670f1f..19f3b5fbc8 100644
--- a/sstables/key.cc
+++ b/sstables/key.cc
@@ -132,8 +132,9 @@ std::vector<bytes> explode_composite(bytes_view _bytes) {
         auto b = in.read_view_to_blob<uint16_t>();
         ret.push_back(to_bytes(b));
         auto marker = in.read<uint8_t>();
-        if (marker != 0) {
-            throw runtime_exception(sprint("non-zero marker found (%d). Can't handle that yet.", marker));
+        // The components will be separated by a null byte, but the last one has special significance.
+        if (in.has_next() && (marker != 0)) {
+            throw runtime_exception(sprint("non-zero component divider found (%d) mid", marker));
         }
     }
     return ret;
diff --git a/sstables/key.hh b/sstables/key.hh
index 7397d2f25f..a99d6215b6 100644
--- a/sstables/key.hh
+++ b/sstables/key.hh
@@ -25,6 +25,19 @@ public:
     }
 };
 
+enum class composite_marker : bytes::value_type {
+    start_range = -1,
+    none = 0,
+    end_range = 1,
+};
+
+inline void check_marker(bytes_view component, composite_marker expected) {
+    auto found = composite_marker(component.back());
+    if (found != expected) {
+        throw runtime_exception(sprint("Unexpected marker. Found %d, expected %d\n", uint8_t(found), uint8_t(expected)));
+    }
+}
+
 // Our internal representation differs slightly (in the way it serializes) from Origin.
 // In order to be able to achieve read and write compatibility for sstables - so they can
 // be imported and exported - we need to always convert a key to this representation.
@@ -57,6 +70,7 @@ public:
     composite_view(bytes_view b) : _bytes(b) {}
 
     std::vector<bytes> explode() const;
+
     explicit operator bytes_view() const {
         return _bytes;
     }