paging_state implementation

Note: serial format blob is different compared to origin, due to scyllas
different internal architecture. I.e. we query actual rows.
But drivers etc ignore the content of the blob, it is opaque.

configure.py

@@ -374,6 +374,7 @@ scylla_core = (['database.cc',
                  'service/storage_service.cc',
                  'service/pending_range_calculator_service.cc',
                  'service/load_broadcaster.cc',
+                 'service/pager/paging_state.cc',
                  'streaming/streaming.cc',
                  'streaming/stream_task.cc',
                  'streaming/stream_session.cc',

service/pager/paging_state.cc

@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/*
+ * Copyright 2015 Cloudius Systems
+ *
+ * Modified by Cloudius Systems
+ */
+
+/*
+ * This file is part of Scylla.
+ *
+ * Scylla is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * Scylla is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with Scylla.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "bytes.hh"
+#include "keys.hh"
+#include "utils/data_input.hh"
+#include "utils/data_output.hh"
+#include "db/serializer.hh"
+#include "paging_state.hh"
+
+service::pager::paging_state::paging_state(partition_key pk, clustering_key ck,
+        uint32_t rem)
+        : _partition_key(std::move(pk)), _clustering_key(ck), _remaining(rem) {
+}
+
+::shared_ptr<service::pager::paging_state> service::pager::paging_state::deserialize(
+        bytes_opt data) {
+    if (!data) {
+        return nullptr;
+    }
+
+    data_input in(*data);
+
+    try {
+        auto pk = db::serializer<partition_key_view>::read(in);
+        auto ck = db::serializer<clustering_key_view>::read(in);
+        auto rem(in.read<uint32_t>());
+
+        return ::make_shared<paging_state>(std::move(pk), std::move(ck), rem);
+    } catch (...) {
+        std::throw_with_nested(
+                exceptions::protocol_exception(
+                        "Invalid value for the paging state"));
+    }
+}
+
+bytes_opt service::pager::paging_state::serialize() const {
+    auto pkv = _partition_key.view();
+    auto ckv = _clustering_key.view();
+
+    db::serializer<partition_key_view> pks(pkv);
+    db::serializer<clustering_key_view> cks(ckv);
+
+    auto size = pks.size() + cks.size() + sizeof(uint32_t);
+
+    bytes res{bytes::initialized_later(), size};
+    data_output out(res);
+
+    pks.write(out);
+    cks.write(out);
+    out.write(_remaining);
+
+    return {res};
+}

service/pager/paging_state.hh

@@ -41,80 +41,43 @@
 
 #pragma once
 
-#include "unimplemented.hh"
 #include "bytes.hh"
+#include "keys.hh"
 
 namespace service {
 
 namespace pager {
 
 class paging_state final {
-#if 0
-    public final ByteBuffer partitionKey;
-    public final ByteBuffer cellName;
-    public final int remaining;
-
-    public PagingState(ByteBuffer partitionKey, ByteBuffer cellName, int remaining)
-    {
-        this.partitionKey = partitionKey == null ? ByteBufferUtil.EMPTY_BYTE_BUFFER : partitionKey;
-        this.cellName = cellName == null ? ByteBufferUtil.EMPTY_BYTE_BUFFER : cellName;
-        this.remaining = remaining;
-    }
-
-    public static PagingState deserialize(ByteBuffer bytes)
-    {
-        if (bytes == null)
-            return null;
-
-        try
-        {
-            DataInputStream in = new DataInputStream(ByteBufferUtil.inputStream(bytes));
-            ByteBuffer pk = ByteBufferUtil.readWithShortLength(in);
-            ByteBuffer cn = ByteBufferUtil.readWithShortLength(in);
-            int remaining = in.readInt();
-            return new PagingState(pk, cn, remaining);
-        }
-        catch (IOException e)
-        {
-            throw new ProtocolException("Invalid value for the paging state");
-        }
-    }
-#endif
+    partition_key _partition_key;
+    clustering_key _clustering_key;
+    uint32_t _remaining;
 
 public:
+    paging_state(partition_key pk, clustering_key ck, uint32_t rem);
 
-    bytes_opt serialize() {
-        fail(unimplemented::cause::PAGING);
-#if 0
-        try
-        {
-            DataOutputBuffer out = new DataOutputBuffer(serializedSize());
-            ByteBufferUtil.writeWithShortLength(partitionKey, out);
-            ByteBufferUtil.writeWithShortLength(cellName, out);
-            out.writeInt(remaining);
-            return out.asByteBuffer();
-        }
-        catch (IOException e)
-        {
-            throw new RuntimeException(e);
-        }
-#endif
+    /**
+     * Last processed key, i.e. where to start from in next paging round
+     */
+    const partition_key& get_partition_key() const {
+        return _partition_key;
+    }
+    /**
+     * Clustering key in last partition. I.e. first, next, row
+     */
+    const clustering_key& get_clustering_key() const {
+        return _clustering_key;
+    }
+    /**
+     * Max remaining rows to fetch in total.
+     * I.e. initial row_limit - #rows returned so far.
+     */
+    uint32_t get_remaining() const {
+        return _remaining;
     }
 
-#if 0
-    private int serializedSize()
-    {
-        return 2 + partitionKey.remaining()
-             + 2 + cellName.remaining()
-             + 4;
-    }
-
-    @Override
-    public String toString()
-    {
-        return String.format("PagingState(key=%s, cellname=%s, remaining=%d", ByteBufferUtil.bytesToHex(partitionKey), ByteBufferUtil.bytesToHex(cellName), remaining);
-    }
-#endif
+    static ::shared_ptr<paging_state> deserialize(bytes_opt bytes);
+    bytes_opt serialize() const;
 };
 
 }