From 87b0f11be3a973fcbdee94c88ade76ffde95b667 Mon Sep 17 00:00:00 2001
From: Piotr Jastrzebski <piotr@scylladb.com>
Date: Wed, 31 May 2017 16:30:16 +0200
Subject: [PATCH] partition_snapshot: Add getters for static row and partition
 tombstone

[tgrabiec:
  - Extracted from a different patch
  - Renamed concept names to more familiar Map and Reduce
  - Renamed aggregate() to squashed() to match the existing nomenclature
  - Uncommented the concepts
  ]
---
 partition_version.cc | 54 ++++++++++++++++++++++++++++++++++++++++++++
 partition_version.hh |  2 ++
 2 files changed, 56 insertions(+)

diff --git a/partition_version.cc b/partition_version.cc
index 331b914a10..0a37dc61c4 100644
--- a/partition_version.cc
+++ b/partition_version.cc
@@ -62,6 +62,60 @@ partition_version::~partition_version()
     }
 }
 
+namespace {
+
+GCC6_CONCEPT(
+
+// A functor which transforms objects from Domain into objects from CoDomain
+template<typename U, typename Domain, typename CoDomain>
+concept bool Mapper() {
+    return requires(U obj, const Domain& src) {
+        { obj(src) } -> const CoDomain&
+    };
+}
+
+// A functor which merges two objects from Domain into one. The result is stored in the first argument.
+template<typename U, typename Domain>
+concept bool Reducer() {
+    return requires(U obj, Domain& dst, const Domain& src) {
+        { obj(dst, src) } -> void;
+    };
+}
+
+)
+
+// Calculates the value of particular part of mutation_partition represented by
+// the version chain starting from v.
+// |map| extracts the part from each version.
+// |reduce| Combines parts from the two versions.
+template <typename Result, typename Map, typename Reduce>
+GCC6_CONCEPT(
+requires Mapper<Map, mutation_partition, Result>() && Reducer<Reduce, Result>()
+)
+inline Result squashed(const partition_version_ref& v, Map&& map, Reduce&& reduce) {
+    Result r = map(v->partition());
+    auto it = v->next();
+    while (it) {
+        reduce(r, map(it->partition()));
+        it = it->next();
+    }
+    return r;
+}
+
+}
+
+row partition_snapshot::static_row() const {
+    return ::squashed<row>(version(),
+                         [] (const mutation_partition& mp) -> const row& { return mp.static_row(); },
+                         [this] (row& a, const row& b) { a.apply(*_schema, column_kind::static_column, b); });
+}
+
+tombstone partition_snapshot::partition_tombstone() const {
+    return ::squashed<tombstone>(version(),
+                               [] (const mutation_partition& mp) { return mp.partition_tombstone(); },
+                               [] (tombstone& a, tombstone b) { a.apply(b); });
+}
+
 partition_snapshot::~partition_snapshot() {
     if (_version && _version.is_unique_owner()) {
         auto v = &*_version;
diff --git a/partition_version.hh b/partition_version.hh
index b3a8d61f69..9e68c00057 100644
--- a/partition_version.hh
+++ b/partition_version.hh
@@ -221,6 +221,8 @@ public:
     }
 
     unsigned version_count();
+    tombstone partition_tombstone() const;
+    row static_row() const;
 };
 
 class partition_entry {