[PATCHv4 00/10] promoted index for reading partial partitions

Nadav Har'El

<nyh@scylladb.com>

unread,

Aug 7, 2016, 7:49:19 AM8/7/16

to scylladb-dev@googlegroups.com, Nadav Har'El

<nyh@scylladb.com>

unread,

Aug 7, 2016, 7:49:29 AM8/7/16

to scylladb-dev@googlegroups.com, Nadav Har'El

Our sstable reading code is currently hard-coded to read entire partitions,
even if we know that only a subset of the columns are requested.
This patch introduces find_disk_ranges(), a function to find the ranges of
bytes we need to read from the sstable data file to guarantee that the
desired columns from the desired partition are read.

The returned range may be the entire byte range of the given partition -
as found using the summary and index files - but if the index contains a
"promoted index" (basically a sample of column positions for each key)
we may return a smaller range. The "disk_read_range" type introduced in
the previous patch is extended here to support reading a partial partition -
by including additional information which would be missed when reading only
part of a partition (viz., the partition key and the partition's tombstone).

This function isn't used in this patch - we will wire its use in the next
patch, which will complete the read-side support for the promoted index.

Signed-off-by: Nadav Har'El <n...@scylladb.com>
---

sstables/partition.cc | 162 ++++++++++++++++++++++++++++++++++++++++
sstables/sstables.hh | 38 +++++++++
2 files changed, 200 insertions(+)

--- .before/sstables/sstables.hh 2016-08-07 14:46:39.189291526 +0300
+++ .after/sstables/sstables.hh 2016-08-07 14:46:39.190291534 +0300
@@ -155,15 +155,40 @@ public:

// object lives until then (e.g., using the do_with() idiom).
future<> data_consume_rows_at_once(row_consumer& consumer, uint64_t pos, uint64_t end);

+ // disk_read_range describes a byte ranges covering part of an sstable
+ // row that we need to read from disk. Usually this is the whole byte
+ // range covering a single sstable row, but in very large rows we might
+ // want to only read a subset of the atoms which we know contains the
+ // columns we are looking for. When the range to be read does NOT include
+ // the entire row, the caller needs to supply the optional "row_info"
+ // containing information about the entire row (key and deletion time)
+ // which is normally read from the beginning of the row.
struct disk_read_range {
+ // TODO: this should become a vector of ranges
uint64_t start;
uint64_t end;
+ // When the range above does not cover the beginning of the sstable
+ // row, we need to supply information which is only available at the
+ // beginning of the row - the row's key and its tombstone if any.
+ struct row_info {
+ key k;
+ deletion_time deltime;
+ };
+ std::experimental::optional<row_info> ri;
disk_read_range() : start(0), end(0) {}
disk_read_range(uint64_t start, uint64_t end) :
start(start), end(end) { }
+ disk_read_range(uint64_t start, uint64_t end, const key& key, const deletion_time& deltime) :
+ start(start), end(end), ri(row_info{key, deltime}) { }
explicit operator bool() const {
return start != end;
}
+ // found_row() is true if the row was found. This is not the same as
+ // operator bool(): It is possible that found_row() but the promoted
+ // index ruled out anything to read (in this case "ri" was set).
+ bool found_row() const {
+ return start != end || ri;

+ }
};

// data_consume_rows() iterates over rows in the data file from

@@ -481,6 +506,19 @@ private:
// The ring_position doesn't have to survive deferring.
future<uint64_t> upper_bound(schema_ptr, const dht::ring_position&, const io_priority_class& pc);

+ // find_disk_ranges finds the ranges of bytes we need to read from the
+ // sstable to read the desired columns out of the given key. This range
+ // may be the entire byte range of the given partition - as found using
+ // the summary and index files - but if the index contains a "promoted
+ // index" (a sample of column positions for each key) it may be a smaller
+ // range. The returned range may contain columns beyond those requested
+ // in ck_filtering, so it is the reader's duty to use ck_filtering again
+ // when parsing the data read from the returned range.
+ future<disk_read_range> find_disk_ranges(schema_ptr schema,
+ const sstables::key& key,
+ query::clustering_key_filtering_context ck_filtering,
+ const io_priority_class& pc);
+
future<summary_entry&> read_summary_entry(size_t i);

// FIXME: pending on Bloom filter implementation
--- .before/sstables/partition.cc 2016-08-07 14:46:39.196291584 +0300
+++ .after/sstables/partition.cc 2016-08-07 14:46:39.197291593 +0300
@@ -28,6 +28,7 @@
#include "unimplemented.hh"
#include "utils/move.hh"
#include "dht/i_partitioner.hh"
+#include <seastar/core/byteorder.hh>

namespace sstables {

@@ -771,6 +772,167 @@ sstables::sstable::read_row(schema_ptr s
});
});
}
+
+template <typename T>
+static inline T read_be(const signed char* p) {
+ return ::read_be<T>(reinterpret_cast<const char*>(p));
+}
+
+template<typename T>
+static inline T consume_be(bytes_view& p) {
+ T i = read_be<T>(p.data());
+ p.remove_prefix(sizeof(T));
+ return i;
+}
+
+static inline bytes_view consume_bytes(bytes_view& p, size_t len) {
+ auto ret = bytes_view(p.data(), len);
+ p.remove_prefix(len);
+ return ret;
+}
+
+static inline clustering_key_prefix get_clustering_key(
+ const schema& schema, bytes_view col_name) {
+ mp_row_consumer::column col(schema, std::move(col_name), api::max_timestamp);
+ return std::move(col.clustering);
+}
+
+future<sstable::disk_read_range>
+sstables::sstable::find_disk_ranges(
+ schema_ptr schema, const sstables::key& key,
+ query::clustering_key_filtering_context ck_filtering,
+ const io_priority_class& pc) {
+ auto& partitioner = dht::global_partitioner();
+ auto token = partitioner.get_token(key_view(key));
+
+ if (token < partitioner.get_token(key_view(_summary.first_key.value))
+ || token > partitioner.get_token(key_view(_summary.last_key.value))) {
+ return make_ready_future<disk_read_range>();
+ }
+ auto summary_idx = adjust_binary_search_index(binary_search(_summary.entries, key, token));
+ if (summary_idx < 0) {
+ return make_ready_future<disk_read_range>();
+ }
+
+ return read_indexes(summary_idx, pc).then([this, schema, ck_filtering, &key, token, summary_idx, &pc] (auto index_list) {
+ auto index_idx = this->binary_search(index_list, key, token);
+ if (index_idx < 0) {
+ return make_ready_future<disk_read_range>();
+ }
+ index_entry& ie = index_list[index_idx];
+ if (ie.get_promoted_index_bytes().size() >= 16) {
+ auto& ck_ranges = ck_filtering.get_ranges(
+ partition_key::from_exploded(*schema, key.explode(*schema)));
+ if (ck_ranges.size() == 1 && ck_ranges[0].is_full()) {
+ // When no clustering filter is given to sstable::read_row(),
+ // we get here one range unbounded on both sides. This is fine
+ // (the code below will work with an unbounded range), but
+ // let's drop this range to revert to the classic behavior of
+ // reading entire sstable row without using the promoted index
+ } else if (ck_ranges.size() == 1) {
+ auto data = ie.get_promoted_index_bytes();
+ // note we already verified above that data.size >= 16
+ sstables::deletion_time deltime;
+ deltime.local_deletion_time = consume_be<uint32_t>(data);
+ deltime.marked_for_delete_at = consume_be<uint64_t>(data);
+ uint32_t num_blocks = consume_be<uint32_t>(data);
+ // We do a linear search on the promoted index. If we were to
+ // look in the same promoted index several times it might have
+ // made sense to build an array of key starts so we can do a
+ // binary search. We could do this once we have a key cache.
+ bool has_range_start = bool(ck_ranges[0].start());
+ auto range_start = ck_ranges[0].start()->value();
+ bool found_range_start = false;
+ uint64_t range_start_pos;
+ bool has_range_end = bool(ck_ranges[0].end());
+ auto range_end = ck_ranges[0].end()->value();
+
+ auto cmp = clustering_key_prefix::tri_compare(*schema);
+ while (num_blocks--) {
+ if (data.size() < 2) {
+ // When we break out of this loop, we give up on
+ // using the promoted index, and fall back to
+ // reading the entire partition.
+ // FIXME: this and all other "break" cases below,
+ // are errors. Log them (with rate limit) and count.
+ break;
+ }
+ uint16_t len = consume_be<uint16_t>(data);
+ if (data.size() < len) {
+ break;
+ }
+ // The promoted index contains ranges of full column
+ // names, which may include a clustering key and column.
+ // But we only need to match the clustering key, because
+ // we got a clustering key range to search for.
+ auto start_ck = get_clustering_key(*schema,
+ consume_bytes(data, len));
+ if (data.size() < 2) {
+ break;
+ }
+ len = consume_be<uint16_t>(data);
+ if (data.size() < len) {
+ break;
+ }
+ auto end_ck = get_clustering_key(*schema,
+ consume_bytes(data, len));
+ if (data.size() < 16) {
+ break;
+ }
+ uint64_t offset = consume_be<uint64_t>(data);
+ uint64_t width = consume_be<uint64_t>(data);
+ if (!found_range_start) {
+ if (!has_range_start || cmp(range_start, end_ck) <= 0) {
+ range_start_pos = ie.position() + offset;
+ found_range_start = true;
+ }
+ }
+ bool found_range_end = false;
+ uint64_t range_end_pos;
+ if (has_range_end) {
+ if (cmp(range_end, start_ck) < 0) {
+ // this block is already past the range_end
+ found_range_end = true;
+ range_end_pos = ie.position() + offset;
+ } else if (cmp(range_end, end_ck) < 0 || num_blocks == 0) {
+ // range_end is in the middle of this block.
+ // Note the strict inequality above is important:
+ // if range_end==end_ck the next block may contain
+ // still more items matching range_end.
+ found_range_end = true;
+ range_end_pos = ie.position() + offset + width;
+ }
+ } else if (num_blocks == 0) {
+ // When !has_range_end, read until the last block.
+ // In this case we could have also found the end of
+ // the partition using the index.
+ found_range_end = true;
+ range_end_pos = ie.position() + offset + width;
+ }
+ if (found_range_end) {
+ if (!found_range_start) {
+ // return empty range
+ range_start_pos = range_end_pos = 0;
+ }
+ return make_ready_future<disk_read_range>(
+ disk_read_range(range_start_pos, range_end_pos,
+ key, deltime));
+ }
+ }
+ }
+ // Else, if more than one clustering-key range needs to be read,
+ // fall back to reading the entire partition.
+ // FIXME: support multiple ranges, and do not fall back to reading
+ // the entire partition.
+ }
+ // If we're still here there is no promoted index, or we had problems
+ // using it, so just just find the entire partition's range.
+ auto start = ie.position();
+ return this->data_end_position(summary_idx, index_idx, index_list, pc).then([start] (uint64_t end) {
+ return disk_read_range(start, end);

+ });
+ });
+}

class mutation_reader::impl {
private:

Nadav Har'El

<nyh@scylladb.com>

unread,

Aug 7, 2016, 7:49:30 AM8/7/16

to scylladb-dev@googlegroups.com, Nadav Har'El

This patch adds support more efficiently reading small parts of a large
partition, without reading the entire partition as we had to do so far.
This is done using the "promoted index".

The "promoted index" is stored in the sstable index file, and provides
for each large sstable row ("partition" in CQL nomenclature) a sample of
the column names at (for example) 64KB intervals. This means that when we
read a slice of columns (e.g., cql rows), or page through a large partition,
we do not have to read the entire partition from disk.

This patch only implements the read side of promoted index - a later patch
will add the write-side support (i.e., writing the promoted index to the
index file while saving the sstable). Nevertheless this patch can already
be tested by reading existing sstables from Cassandra which include a
promoted index - such as the one included in the test in the previous patch.

The use of the promoted index currently has two limitations:

1. It is only used when reading a single partition with sstable::read_row(),
not when scanning through many partitions with sstable::read_range_rows()
or sstable::read_rows().

2. It is only used when filtering a single clustering-key range, rather
than a list of disjoint ranges. A single range is the common case.

These two issues will be improved later. In the meantime, in those
unsupported cases we simply continue to read entire partitions, so we're not
worse-off than before.

Also note that this patch only helps when sstable::read_row() is used with
a clustering-key prefix (i.e., a slice). Our higher-level request handling
code may decide to read an entire partition into the cache, and not use
a clustering-key prefix at all when reading. We will need to indepdently
improve the high-level code to use read_row()'s slicing capabilities
when paging through large partitions, for example.

Signed-off-by: Nadav Har'El <n...@scylladb.com>
---

sstables/partition.cc | 33 +++++-------------------------
sstables/row.cc | 44 ++++++++++++++++++++++++++++++++++------
2 files changed, 44 insertions(+), 33 deletions(-)

--- .before/sstables/row.cc 2016-08-07 14:46:39.268292187 +0300
+++ .after/sstables/row.cc 2016-08-07 14:46:39.268292187 +0300
@@ -63,6 +63,7 @@ private:
bool _deleted;
uint32_t _ttl, _expiration;

+ bool _read_partial_row = false;

public:
bool non_consuming() const {
@@ -332,12 +333,42 @@ public:
}

data_consume_rows_context(row_consumer& consumer,
- input_stream<char> && input, uint64_t maxlen) :
- continuous_data_consumer(std::move(input), maxlen)
- , _consumer(consumer) {
+ input_stream<char> && input, uint64_t maxlen,
+ std::experimental::optional<sstable::disk_read_range::row_info> ri = {})
+ : continuous_data_consumer(std::move(input), maxlen)
+ , _consumer(consumer) {
+ // If the "ri" option is given, we are reading a partition from the
+ // middle (in the beginning of an atom), as would happen when we use
+ // the "promoted index" to skip closer to where a particular column
+ // starts. When we start in the middle of the partition, we will not
+ // read the key nor the tombstone from the disk, so the caller needs
+ // to provide them (the tombstone is provided in the promoted index
+ // exactly for that reason).
+ if (ri) {
+ _read_partial_row = true;
+ auto ret = _consumer.consume_row_start(ri->k, ri->deltime);
+ if (ret == row_consumer::proceed::yes) {

+ _state = state::ATOM_START;

+ } else {
+ // If we were asked to stop parsing after consuming the row
+ // start, we can't go to ATOM_START, need to use a new state
+ // which stops parsing, and continues at ATOM_START later.
+ _state = state::STOP_THEN_ATOM_START;
+ }
+ }
}

void verify_end_state() {
+ if (_read_partial_row) {
+ // If reading a partial row (i.e., when we have a clustering row
+ // filter and using a promoted index), we may be in ATOM_START
+ // state instead of ROW_START. In that case we did not read the
+ // end-of-row marker and consume_row_end() was never called.
+ if (_state == state::ATOM_START) {
+ _consumer.consume_row_end();
+ return;
+ }
+ }
if (_state != state::ROW_START || _prestate != prestate::NONE) {
throw malformed_sstable_exception("end of input, but not end of row");
}
@@ -354,9 +385,10 @@ private:
shared_sstable _sst;
std::unique_ptr<data_consume_rows_context> _ctx;
public:
- impl(shared_sstable sst, row_consumer& consumer, input_stream<char>&& input, uint64_t maxlen)
+ impl(shared_sstable sst, row_consumer& consumer, input_stream<char>&& input, uint64_t maxlen,
+ std::experimental::optional<sstable::disk_read_range::row_info> ri)
: _sst(std::move(sst))
- , _ctx(new data_consume_rows_context(consumer, std::move(input), maxlen))
+ , _ctx(new data_consume_rows_context(consumer, std::move(input), maxlen, ri))
{ }
~impl() {
if (_ctx) {
@@ -391,7 +423,7 @@ data_consume_context sstable::data_consu

// to maintain its own byte count.
return std::make_unique<data_consume_context::impl>(shared_from_this(),

consumer, data_stream(toread.start, toread.end - toread.start,

- consumer.io_priority()), toread.end - toread.start);
+ consumer.io_priority()), toread.end - toread.start, toread.ri);
}

data_consume_context sstable::data_consume_rows(row_consumer& consumer) {
--- .before/sstables/partition.cc 2016-08-07 14:46:39.269292196 +0300
+++ .after/sstables/partition.cc 2016-08-07 14:46:39.271292213 +0300
@@ -738,37 +738,16 @@ sstables::sstable::read_row(schema_ptr s
if (!filter_has_key(key)) {
return make_ready_future<streamed_mutation_opt>();
}
-
- auto& partitioner = dht::global_partitioner();
- auto token = partitioner.get_token(key_view(key));
-
- auto& summary = _summary;
-
- if (token < partitioner.get_token(key_view(summary.first_key.value))
- || token > partitioner.get_token(key_view(summary.last_key.value))) {
- _filter_tracker.add_false_positive();
- return make_ready_future<streamed_mutation_opt>();
- }
-
- auto summary_idx = adjust_binary_search_index(binary_search(summary.entries, key, token));
- if (summary_idx < 0) {
- _filter_tracker.add_false_positive();
- return make_ready_future<streamed_mutation_opt>();
- }
-
- return read_indexes(summary_idx, pc).then([this, schema, ck_filtering, &key, token, summary_idx, &pc] (auto index_list) {
- auto index_idx = this->binary_search(index_list, key, token);
- if (index_idx < 0) {
+ return find_disk_ranges(schema, key, ck_filtering, pc).then([this, &key, ck_filtering, &pc, schema] (disk_read_range toread) {
+ if (!toread.found_row()) {
_filter_tracker.add_false_positive();
+ }
+ if (!toread) {
return make_ready_future<streamed_mutation_opt>();
}
_filter_tracker.add_true_positive();
-
- auto position = index_list[index_idx].position();
- return this->data_end_position(summary_idx, index_idx, index_list, pc).then([&key, schema, ck_filtering, this, position, &pc] (uint64_t end) {

- return sstable_streamed_mutation::create(schema, this->shared_from_this(), key, ck_filtering, pc, {position, end}).then([] (auto sm) {

- return streamed_mutation_opt(std::move(sm));
- });
+ return sstable_streamed_mutation::create(schema, this->shared_from_this(), key, ck_filtering, pc, std::move(toread)).then([] (auto sm) {
+ return streamed_mutation_opt(std::move(sm));
});
});
}

Nadav Har'El

<nyh@scylladb.com>

unread,

Aug 7, 2016, 7:49:31 AM8/7/16

to scylladb-dev@googlegroups.com, Nadav Har'El

Add to the range_tombstone_accumulator a range_tombstones_for_row(ck)
method.

Just like the existing tombstone_for_row(ck), this function drops from
the accumulator tombstones that end before ck. But while the existing
function returned just a single tombstone affecting the given row (the
most recent tombstone), the new function range_tombstones_for_row(ck)
returns all the accumulated range tombstones which cover ck.

This function will be useful for the promoted-index writing code later,
which divides a partition into blocks which may be read independently,
so each block needs to start with a repeat of the earlier tombstones
which still cover the first row in the new block.

Signed-off-by: Nadav Har'El <n...@scylladb.com>
---

range_tombstone.hh | 5 +++++

1 file changed, 5 insertions(+)

--- .before/range_tombstone.hh 2016-08-07 14:46:39.333292732 +0300
+++ .after/range_tombstone.hh 2016-08-07 14:46:39.333292732 +0300
@@ -294,6 +294,11 @@ public:
return _current_tombstone;
}

+ const std::deque<range_tombstone>& range_tombstones_for_row(const clustering_key_prefix& ck) {
+ drop_unneeded_tombstones(ck);
+ return _range_tombstones;
+ }
+
void apply(const range_tombstone& rt);

void clear();

Nadav Har'El

<nyh@scylladb.com>

unread,

Aug 7, 2016, 7:49:33 AM8/7/16

to scylladb-dev@googlegroups.com, Nadav Har'El

This patch adds writing of promoted index to sstables.

The promoted index is basically a sample of columns and their positions
for large partitions: The promoted index appears in the sstable's index
file for partitions which are larger than 64 KB, and divides the partition
to 64 KB blocks (as in Cassandra, this interval is configurable through
the column_index_size_in_kb config parameter). Beyond modifying the index
file, having a promoted index may also modify the data file: Since each
of blocks may be read independently, we need to add in the beginning of
each block the list of range tombstones that are still open at that
position.

See also https://github.com/scylladb/scylla/wiki/SSTables-Index-File

Fixes #959

Signed-off-by: Nadav Har'El <n...@scylladb.com>
---

sstables/sstables.cc | 191 +++++++++++++++++++++++++++++++++++++++--
sstables/sstables.hh | 21 ++++
2 files changed, 207 insertions(+), 5 deletions(-)

--- .before/sstables/sstables.hh 2016-08-07 14:46:39.412293393 +0300
+++ .after/sstables/sstables.hh 2016-08-07 14:46:39.413293402 +0300
@@ -407,6 +407,27 @@ private:
uint64_t _filter_file_size = 0;
uint64_t _bytes_on_disk = 0;

+ // _pi_write is used temporarily for building the promoted
+ // index (column sample) of one partition when writing a new sstable.
+ struct {
+ // Unfortunately we cannot output the promoted index directly to the
+ // index file because it needs to be prepended by its size.
+ bytes_ostream data;
+ uint32_t numblocks;
+ deletion_time deltime;
+ uint64_t block_start_offset;
+ uint64_t block_next_start_offset;
+ bytes block_first_colname;
+ bytes block_last_colname;
+ std::experimental::optional<range_tombstone_accumulator> tombstone_accumulator;
+ const schema* schemap;
+ size_t desired_block_size;
+ } _pi_write;
+
+ void maybe_flush_pi_block(file_writer& out,
+ const composite& clustering_key,
+ const std::vector<bytes_view>& column_names);
+
sstring _ks;
sstring _cf;
sstring _dir;
--- .before/sstables/sstables.cc 2016-08-07 14:46:39.418293444 +0300
+++ .after/sstables/sstables.cc 2016-08-07 14:46:39.421293469 +0300
@@ -31,6 +31,7 @@
#include "core/do_with.hh"
#include "core/thread.hh"
#include <seastar/core/shared_future.hh>
+#include <seastar/core/byteorder.hh>
#include <iterator>

#include "types.hh"
@@ -56,6 +57,7 @@

#include "checked-file-impl.hh"
#include "disk-error-handler.hh"
+#include "service/storage_service.hh"

thread_local disk_error_signal_type sstable_read_error;
thread_local disk_error_signal_type sstable_write_error;
@@ -296,6 +298,12 @@ inline void write(file_writer& out, byte
out.write(reinterpret_cast<const char*>(s.data()), s.size()).get();
}

+inline void write(file_writer& out, bytes_ostream s) {
+ for (bytes_view fragment : s) {
+ write(out, fragment);
+ }
+}
+
// All composite parsers must come after this
template<typename First, typename... Rest>
future<> parse(random_access_reader& in, First& first, Rest&&... rest) {
@@ -1066,6 +1074,108 @@ future<> sstable::load() {
});
}

+static void output_promoted_index_entry(bytes_ostream& promoted_index,
+ const bytes& first_col,
+ const bytes& last_col,
+ uint64_t offset, uint64_t width) {
+ char s[2];
+ write_be(s, uint16_t(first_col.size()));
+ promoted_index.write(s, 2);
+ promoted_index.write(first_col);
+ write_be(s, uint16_t(last_col.size()));
+ promoted_index.write(s, 2);
+ promoted_index.write(last_col);
+ char q[8];
+ write_be(q, uint64_t(offset));
+ promoted_index.write(q, 8);
+ write_be(q, uint64_t(width));
+ promoted_index.write(q, 8);
+}
+
+// FIXME: use this in write_column_name() instead of repeating the code
+static bytes serialize_colname(const composite& clustering_key,
+ const std::vector<bytes_view>& column_names, composite::eoc marker) {
+ auto c = composite::from_exploded(column_names, marker);
+ auto ck_bview = bytes_view(clustering_key);
+ // The marker is not a component, so if the last component is empty (IOW,
+ // only serializes to the marker), then we just replace the key's last byte
+ // with the marker. If the component however it is not empty, then the
+ // marker should be in the end of it, and we just join them together as we
+ // do for any normal component
+ if (c.size() == 1) {
+ ck_bview.remove_suffix(1);
+ }
+ size_t sz = ck_bview.size() + c.size();
+ if (sz > std::numeric_limits<uint16_t>::max()) {
+ throw std::runtime_error(sprint("Column name too large (%d > %d)", sz, std::numeric_limits<uint16_t>::max()));
+ }
+ bytes colname(bytes::initialized_later(), sz);
+ std::copy(ck_bview.begin(), ck_bview.end(), colname.begin());
+ std::copy(c.get_bytes().begin(), c.get_bytes().end(), colname.begin() + ck_bview.size());
+ return colname;
+}
+
+// Call maybe_flush_pi_block() before writing the given sstable atom to the
+// output. This may start a new promoted-index block depending on how much
+// data we've already written since the start of the current block. Starting
+// a new block involves both outputting the range of the old block to the
+// index file, and outputting again the currently-open range tombstones to
+// the data file.
+// TODO: currently, maybe_flush_pi_block serializes the column name on every
+// call, saving it in _pi_write.block_last_colname which we need for closing
+// each block, as well as for closing the last block. We could instead save
+// just the unprocessed arguments, and serialize them only when needed at the
+// end of the block. For this we would need this function to take rvalue
+// references (so data is moved in), and need not to use vector of byte_view
+// (which might be gone later).
+void sstable::maybe_flush_pi_block(file_writer& out,
+ const composite& clustering_key,
+ const std::vector<bytes_view>& column_names) {
+ bytes colname = serialize_colname(clustering_key, column_names, composite::eoc::none);
+ if (_pi_write.block_first_colname.empty()) {
+ // This is the first column in the partition, or first column since we
+ // closed a promoted-index block. Remember its name and position -
+ // we'll need to write it to the promoted index.
+ _pi_write.block_start_offset = out.offset();
+ _pi_write.block_next_start_offset = out.offset() + _pi_write.desired_block_size;
+ _pi_write.block_first_colname = colname;
+ _pi_write.block_last_colname = std::move(colname);
+ } else if (out.offset() >= _pi_write.block_next_start_offset) {
+ // If we wrote enough bytes to the partition since we output a sample
+ // to the promoted index, output one now and start a new one.
+ output_promoted_index_entry(_pi_write.data,
+ _pi_write.block_first_colname,
+ _pi_write.block_last_colname,
+ _pi_write.block_start_offset - _c_stats.start_offset,
+ out.offset() - _pi_write.block_start_offset);
+ _pi_write.numblocks++;
+ _pi_write.block_start_offset = out.offset();
+ // Because the new block can be read without the previous blocks, we
+ // need to repeat the range tombstones which are still open.
+ // Note that block_start_offset is before outputting those (so the new
+ // block includes them), but we set block_next_start_offset after - so
+ // even if we wrote a lot of open tombstones, we still get a full
+ // block size of new data.
+ if (!clustering_key.empty()) {
+ auto& rts = _pi_write.tombstone_accumulator->range_tombstones_for_row(
+ clustering_key_prefix(clustering_key.values()));
+ for (const auto& rt : rts) {
+ auto start = composite::from_clustering_element(*_pi_write.schemap, rt.start);
+ auto end = composite::from_clustering_element(*_pi_write.schemap, rt.end);
+ write_range_tombstone(out,
+ start, rt.start_kind, end, rt.end_kind, {}, rt.tomb);
+ }
+ }
+ _pi_write.block_next_start_offset = out.offset() + _pi_write.desired_block_size;
+ _pi_write.block_first_colname = colname;
+ _pi_write.block_last_colname = std::move(colname);
+ } else {
+ // Keep track of the last column in the partition - we'll need it to close
+ // the last block in the promoted index, unfortunately.
+ _pi_write.block_last_colname = std::move(colname);
+ }
+}
+
// @clustering_key: it's expected that clustering key is already in its composite form.
// NOTE: empty clustering key means that there is no clustering key.
void sstable::write_column_name(file_writer& out, const composite& clustering_key, const std::vector<bytes_view>& column_names, composite::eoc marker) {
@@ -1223,6 +1333,7 @@ void sstable::write_collection(file_writ
const bytes& column_name = cdef.name();
write_range_tombstone(out, clustering_key, clustering_key, { bytes_view(column_name) }, mview.tomb);
for (auto& cp: mview.cells) {
+ maybe_flush_pi_block(out, clustering_key, { column_name, cp.first });
write_column_name(out, clustering_key, { column_name, cp.first });
write_cell(out, cp.second);
}
@@ -1234,11 +1345,22 @@ void sstable::write_clustered_row(file_w
auto clustering_key = composite::from_clustering_element(schema, clustered_row.key());

if (schema.is_compound() && !schema.is_dense()) {
+ maybe_flush_pi_block(out, clustering_key, { bytes_view() });
write_row_marker(out, clustered_row.marker(), clustering_key);
}
// Before writing cells, range tombstone must be written if the row has any (deletable_row::t).
if (clustered_row.tomb()) {
+ maybe_flush_pi_block(out, clustering_key, {});
write_range_tombstone(out, clustering_key, clustering_key, {}, clustered_row.tomb());
+ // Because we currently may break a partition to promoted-index blocks
+ // in the middle of a clustered row, we also need to track the current
+ // row's tombstone - not just range tombstones - which may effect the
+ // beginning of a new block.
+ // TODO: consider starting a new block only between rows, so the
+ // following code can be dropped:
+ _pi_write.tombstone_accumulator->apply(range_tombstone(
+ clustered_row.key(), bound_kind::incl_start,
+ clustered_row.key(), bound_kind::incl_end, clustered_row.tomb()));
}

// Write all cells of a partition's row.
@@ -1256,14 +1378,18 @@ void sstable::write_clustered_row(file_w

if (schema.is_compound()) {
if (schema.is_dense()) {
+ maybe_flush_pi_block(out, composite(), { bytes_view(clustering_key) });
write_column_name(out, bytes_view(clustering_key));
} else {
+ maybe_flush_pi_block(out, clustering_key, { bytes_view(column_name) });
write_column_name(out, clustering_key, { bytes_view(column_name) });
}
} else {
if (schema.is_dense()) {
+ maybe_flush_pi_block(out, composite(), { bytes_view(clustered_row.key().get_component(schema, 0)) });
write_column_name(out, bytes_view(clustered_row.key().get_component(schema, 0)));
} else {
+ maybe_flush_pi_block(out, composite(), { bytes_view(column_name) });
write_column_name(out, bytes_view(column_name));
}
}
@@ -1282,16 +1408,25 @@ void sstable::write_static_row(file_writ
assert(column_definition.is_static());
atomic_cell_view cell = c.as_atomic_cell();
auto sp = composite::static_prefix(schema);
+ maybe_flush_pi_block(out, sp, { bytes_view(column_definition.name()) });
write_column_name(out, sp, { bytes_view(column_definition.name()) });
write_cell(out, cell);
});
}

-static void write_index_entry(file_writer& out, disk_string_view<uint16_t>& key, uint64_t pos) {
- // FIXME: support promoted indexes.
- uint32_t promoted_index_size = 0;
+static void write_index_header(file_writer& out, disk_string_view<uint16_t>& key, uint64_t pos) {
+ write(out, key, pos);
+}

- write(out, key, pos, promoted_index_size);
+static void write_index_promoted(file_writer& out, bytes_ostream& promoted_index,
+ deletion_time deltime, uint32_t numblocks) {
+ uint32_t promoted_index_size = promoted_index.size();
+ if (promoted_index_size) {
+ promoted_index_size += 16 /* deltime + numblocks */;
+ write(out, promoted_index_size, deltime, numblocks, promoted_index);
+ } else {
+ write(out, promoted_index_size);
+ }
}

static void prepare_summary(summary& s, uint64_t expected_partition_count, uint32_t min_index_interval) {
@@ -1405,6 +1540,17 @@ file_writer components_writer::index_fil
return file_writer(sst._index_file, std::move(options));
}

+// Get the currently loaded configuration, or the default configuration in
+// case none has been loaded (this happens, for example, in unit tests).
+static const db::config& get_config() {
+ if (service::get_storage_service().local_is_initialized()) {
+ return service::get_local_storage_service().db().local().get_config();
+ } else {
+ static db::config default_config;
+ return default_config;
+ }
+}
+
components_writer::components_writer(sstable& sst, const schema& s, file_writer& out,
uint64_t estimated_partitions, uint64_t max_sstable_size,
const io_priority_class& pc)
@@ -1415,6 +1561,7 @@ components_writer::components_writer(sst
, _max_sstable_size(max_sstable_size)
{
_sst._filter = utils::i_filter::get_filter(estimated_partitions, _schema.bloom_filter_fp_chance());
+ _sst._pi_write.desired_block_size = get_config().column_index_size_in_kb() * 1024;

prepare_summary(_sst._summary, estimated_partitions, _schema.min_index_interval());

@@ -1435,7 +1582,17 @@ void components_writer::consume_new_part
p_key.value = bytes_view(*_partition_key);

// Write index file entry from partition key into index file.
- write_index_entry(_index, p_key, _out.offset());
+ // Write an index entry minus the "promoted index" (sample of columns)
+ // part. We can only write that after processing the entire partition
+ // and collecting the sample of columns.
+ write_index_header(_index, p_key, _out.offset());
+ _sst._pi_write.data = {};
+ _sst._pi_write.numblocks = 0;
+ _sst._pi_write.deltime.local_deletion_time = std::numeric_limits<int32_t>::max();
+ _sst._pi_write.deltime.marked_for_delete_at = std::numeric_limits<int64_t>::min();
+ _sst._pi_write.block_start_offset = _out.offset();
+ _sst._pi_write.tombstone_accumulator = range_tombstone_accumulator(_schema, false);
+ _sst._pi_write.schemap = &_schema; // sadly we need this

// Write partition key into data file.
write(_out, p_key);
@@ -1461,6 +1618,8 @@ void components_writer::consume(tombston
}
write(_out, d);
_tombstone_written = true;
+ // TODO: need to verify we don't do this twice?
+ _sst._pi_write.deltime = d;
}

stop_iteration components_writer::consume(static_row&& sr) {
@@ -1477,13 +1636,35 @@ stop_iteration components_writer::consum

stop_iteration components_writer::consume(range_tombstone&& rt) {
ensure_tombstone_is_written();
+ // Remember the range tombstone so when we need to open a new promoted
+ // index block, we can figure out which ranges are still open and need
+ // to be repeated in the data file. Note that apply() also drops ranges
+ // already closed by rt.start, so the accumulator doesn't grow boundless.
+ _sst._pi_write.tombstone_accumulator->apply(rt);
auto start = composite::from_clustering_element(_schema, std::move(rt.start));
auto end = composite::from_clustering_element(_schema, std::move(rt.end));
+ _sst.maybe_flush_pi_block(_out, start, {});
_sst.write_range_tombstone(_out, std::move(start), rt.start_kind, std::move(end), rt.end_kind, {}, rt.tomb);
return stop_iteration::no;
}

stop_iteration components_writer::consume_end_of_partition() {
+ // If there is an incomplete block in the promoted index, write it too.
+ // However, if the _promoted_index is still empty, don't add a single
+ // chunk - better not output a promoted index at all in this case.
+ if (!_sst._pi_write.data.empty() && !_sst._pi_write.block_first_colname.empty()) {
+ output_promoted_index_entry(_sst._pi_write.data,
+ _sst._pi_write.block_first_colname,
+ _sst._pi_write.block_last_colname,
+ _sst._pi_write.block_start_offset - _sst._c_stats.start_offset,
+ _out.offset() - _sst._pi_write.block_start_offset);
+ _sst._pi_write.numblocks++;
+ _sst._pi_write.block_first_colname = {};
+ }
+ write_index_promoted(_index, _sst._pi_write.data, _sst._pi_write.deltime,
+ _sst._pi_write.numblocks);
+ _sst._pi_write.data = {};
+
ensure_tombstone_is_written();
int16_t end_of_row = 0;
write(_out, end_of_row);

Nadav Har'El

<nyh@scylladb.com>

unread,

Aug 7, 2016, 7:49:35 AM8/7/16

to scylladb-dev@googlegroups.com, Nadav Har'El

In this unit test, we create using Scylla C++ code, the same large
partition with 13520 CQL rows as we previously imported from Cassandra
for the large partition test. We then verify that the sstable index file
we just wrote is byte-for-byte identical to the one previously created by
Cassandra. They should indeed be identical, because the data file has the
same layout (even if timestamps are different) and our default promoted-
index block size is the same (64K) so the sample of columns should be
identical.

Signed-off-by: Nadav Har'El <n...@scylladb.com>
---

tests/sstable_test.cc | 54 ++++++++++++++++++++++++++++++++++++++++
1 file changed, 54 insertions(+)

--- .before/tests/sstable_test.cc 2016-08-07 14:46:39.498294114 +0300
+++ .after/tests/sstable_test.cc 2016-08-07 14:46:39.499294122 +0300
@@ -38,6 +38,7 @@
#include "sstable_test.hh"
#include "tmpdir.hh"
#include "partition_slice_builder.hh"
+#include "tests/test_services.hh"

#include "disk-error-handler.hh"

@@ -1166,3 +1167,56 @@ SEASTAR_TEST_CASE(sub_partitions_read) {
});
});
}
+
+// A silly, inefficient but effective, way to compare two files by reading
+// them entirely into memory.
+static future<> compare_files(sstring file1, sstring file2) {
+ return read_file(file1).then([file2] (auto in1) {
+ return read_file(file2).then([in1 = std::move(in1)] (auto in2) {
+ // assert that both files have the same size.
+ BOOST_REQUIRE(in1.second == in2.second);
+ // assert that both files have the same content.
+ BOOST_REQUIRE(::memcmp(in1.first.get(), in2.first.get(), in1.second) == 0);

+ });
+ });
+

+}
+
+// This test creates the same data as we previously created with Cassandra
+// in the tests/sstables/large_partition directory (which we read in the
+// promoted_index_read test above). The index file in both sstables - which
+// includes the promoted index - should be bit-for-bit identical, otherwise
+// we have a problem in our promoted index writing code (or in the data
+// writing code, because the promoted index points to offsets in the data).
+SEASTAR_TEST_CASE(promoted_index_write) {
+ return test_setup::do_with_test_directory([] {
+ auto s = large_partition_schema();
+ auto mtp = make_lw_shared<memtable>(s);
+ auto key = partition_key::from_exploded(*s, {to_bytes("v1")});
+ mutation m(key, s);
+ auto col = s->get_column_definition("t3");
+ BOOST_REQUIRE(col && !col->is_static());
+ for (char i = 'a'; i <= 'z'; i++) {
+ for (char j = 'A'; j <= 'Z'; j++) {
+ for (int k = 0; k < 20; k++) {
+ auto& row = m.partition().clustered_row(
+ clustering_key::from_exploded(
+ *s, {to_bytes(sprint("%d%c%c", k, i, j))}));
+ row.cells().apply(*col,
+ atomic_cell::make_live(2345,
+ col->type->decompose(sstring(sprint("z%c",i)))));
+ row.apply(row_marker(1234));
+ }
+ }
+ }
+ mtp->apply(std::move(m));
+ auto sst = make_lw_shared<sstable>("try1", "data",
+ "tests/sstables/tests-temporary", 100,

+ sstables::sstable::version_types::ka, big);

+ return sst->write_components(*mtp).then([s] {
+ return compare_files(
+ "tests/sstables/large_partition/try1-data-ka-3-Index.db",
+ "tests/sstables/tests-temporary/try1-data-ka-100-Index.db");
+ }).then([sst, mtp] {});
+ });
+}

Avi Kivity

<avi@scylladb.com>

unread,

Aug 7, 2016, 9:16:49 AM8/7/16

to Nadav Har'El, scylladb-dev@googlegroups.com

Can you summarize the changes relative to the previous version, to guide
review?

Nadav Har'El

<nyh@scylladb.com>

unread,

Aug 7, 2016, 9:28:58 AM8/7/16

to Avi Kivity, scylladb-dev

On Sun, Aug 7, 2016 at 4:16 PM, Avi Kivity <a...@scylladb.com> wrote:

Can you summarize the changes relative to the previous version, to guide review?

Not much, basically:

1. Rebase to current master (there were a bunch of changes, unfortunately, like using shared_sstable instead of sstable& in a bunch of places).

2. Fix the FIXME that bothered you. I'm not 100% sure I did this correctly in every case because I didn't know how to test all these cases in practice.

3. Maybe a couple of other tiny fixes I don't remember.

Avi Kivity

<avi@scylladb.com>

unread,

Aug 7, 2016, 10:14:07 AM8/7/16

to Nadav Har'El, scylladb-dev@googlegroups.com

Please post a git url I can pull this from.

On 08/07/2016 02:49 PM, Nadav Har'El wrote:

Nadav Har'El

<nyh@scylladb.com>

unread,

Aug 7, 2016, 10:49:40 AM8/7/16

to Avi Kivity, scylladb-dev

g...@github.com:nyh/scylla.git

branch "promoted-index-v4"

--
Nadav Har'El

n...@scylladb.com

Commit Bot

<bot@cloudius-systems.com>

unread,

Aug 7, 2016, 10:54:05 AM8/7/16

to scylladb-dev@googlegroups.com, Nadav Har'El

From: Nadav Har'El <n...@scylladb.com>
Committer: Nadav Har'El <n...@scylladb.com>
Branch: master

tests: add test for reading parts of a large partition

This patch adds a test that takes an sstable with one partition of 13,520
clustering rows (spanning 700 KB in the data file), and attempts to read
various slices CQL rows, counting that we got back the expected number
of rows.

The sstable included here was generated by Cassandra, and includes a
promoted index. Promoted index reading is not supported yet (we will
add it in the next patch), so for now the code will always read the
entire partition from disk; But still the clustering-key filtering is
already functional, and will drop some of the rows as requested,
so this test will pass.

Later, when we add promoted index support, we should check that this test
still passes - promoted index will make the reads in this test more
efficient (which the test cannot verify), but the important thing to check
is that it doesn't break any of these tests.

Signed-off-by: Nadav Har'El <n...@scylladb.com>

---

diff --git a/tests/sstable_test.cc b/tests/sstable_test.cc
--- a/tests/sstable_test.cc
+++ b/tests/sstable_test.cc

+ auto sst = make_lw_shared<sstable>(

+ "try1", "data", "tests/sstables/large_partition", 3,

+ sstables::sstable::version_types::ka, big);

+ auto fut = sst->load();
+ return std::move(fut).then([sst = std::move(sst)] {
+ return std::move(sst);
+ });
+}
+
+// This is a rudimentary test that reads an sstable exported from Cassandra
+// which contains a promoted index. It just checks that the promoted index
+// is read from disk, as an unparsed array, and doesn't actually use it to
+// search for anything.
+SEASTAR_TEST_CASE(promoted_index_read) {
+ return load_large_partition_sst().then([] (auto sstp) {
+ schema_ptr s = large_partition_schema();
+ return sstables::test(sstp).read_indexes(0).then([sstp]
(index_list vec) {
+ BOOST_REQUIRE(vec.size() == 1);
+ index_entry &e = vec[0];
+ BOOST_REQUIRE(e.get_promoted_index_bytes().size() == 468);

+ });
+ });
+}
+

+// Use an empty string for ck1, ck2, or both, for unbounded ranges.
+static query::partition_slice make_partition_slice(const schema& s,
sstring ck1, sstring ck2) {
+ std::experimental::optional<query::clustering_range::bound> b1;
+ if (!ck1.empty()) {
+ b1.emplace(clustering_key_prefix::from_single_value(
+ s, utf8_type->decompose(ck1)));

+ }

+}
+

+// This test reads, using sstable::read_row(), a slice (a range of
clustering
+// rows) from one large partition in an sstable written in Cassandra.
+// This large partition includes 13520 clustering rows, and spans about
+// 700 KB on disk. When we ask to read only a part of it, the promoted
index

+// (included in this sstable) may be used to allow reading only a part of
the

+ BOOST_REQUIRE(nrows == 0);
+ });

+ }).then([sstp, s] () {
+ // range that is outside (before) the actual range of the data.
+ // No rows should match.
+ return count_rows(sstp, s, "v1", "_a", "_b").then([] (int
nrows) {

+ BOOST_REQUIRE(nrows == 0);
+ });

+ }).then([sstp, s] () {
+ // half-infinite range
+ return count_rows(sstp, s, "v1", "", "10aA").then([] (int
nrows) {
+ BOOST_REQUIRE(nrows == (1*26*26 + 1));
+ });
+ }).then([sstp, s] () {
+ // half-infinite range
+ return count_rows(sstp, s, "v1", "10aA", "").then([] (int
nrows) {
+ BOOST_REQUIRE(nrows == 19*26*26);
+ });
+ }).then([sstp, s] () {
+ // count all rows, but giving an explicit all-encompasing
filter
+ return count_rows(sstp, s, "v1", "", "").then([] (int nrows) {
+ BOOST_REQUIRE(nrows == 20*26*26);
+ });
+ }).then([sstp, s] () {
+ // count all rows, without a filter
+ return count_rows(sstp, s, "v1").then([] (int nrows) {
+ BOOST_REQUIRE(nrows == 20*26*26);

+ });
+ });

+ });
+}
+
+// Same as previous test, just using read_range_rows instead of read_row
+// to read parts of potentially more than one partition (in this particular
+// sstable, there is actually just one partition).
+SEASTAR_TEST_CASE(sub_partitions_read) {
+ schema_ptr s = large_partition_schema();
+ return load_large_partition_sst().then([s] (auto sstp) {
+ return count_rows(sstp, s, "18wX", "18xB").then([] (int nrows) {
+ BOOST_REQUIRE(nrows == 5);

+ });
+ });
+}

diff --git
a/tests/sstables/large_partition/try1-data-ka-3-CompressionInfo.db
b/tests/sstables/large_partition/try1-data-ka-3-CompressionInfo.db
--- a/tests/sstables/large_partition/try1-data-ka-3-CompressionInfo.db
+++ b/tests/sstables/large_partition/try1-data-ka-3-CompressionInfo.db
null
diff --git a/tests/sstables/large_partition/try1-data-ka-3-Data.db
b/tests/sstables/large_partition/try1-data-ka-3-Data.db
--- a/tests/sstables/large_partition/try1-data-ka-3-Data.db
+++ b/tests/sstables/large_partition/try1-data-ka-3-Data.db
null
diff --git a/tests/sstables/large_partition/try1-data-ka-3-Digest.sha1
b/tests/sstables/large_partition/try1-data-ka-3-Digest.sha1
--- a/tests/sstables/large_partition/try1-data-ka-3-Digest.sha1

+++ b/tests/sstables/large_partition/try1-data-ka-3-Digest.sha1
@@ -0,0 +1 @@
+2833048369
\ No newline at end of file
diff --git a/tests/sstables/large_partition/try1-data-ka-3-Filter.db
b/tests/sstables/large_partition/try1-data-ka-3-Filter.db

--- a/tests/sstables/large_partition/try1-data-ka-3-Filter.db
+++ b/tests/sstables/large_partition/try1-data-ka-3-Filter.db
null
diff --git a/tests/sstables/large_partition/try1-data-ka-3-Index.db
b/tests/sstables/large_partition/try1-data-ka-3-Index.db
--- a/tests/sstables/large_partition/try1-data-ka-3-Index.db
+++ b/tests/sstables/large_partition/try1-data-ka-3-Index.db
null
diff --git a/tests/sstables/large_partition/try1-data-ka-3-Statistics.db
b/tests/sstables/large_partition/try1-data-ka-3-Statistics.db
--- a/tests/sstables/large_partition/try1-data-ka-3-Statistics.db
+++ b/tests/sstables/large_partition/try1-data-ka-3-Statistics.db
null
diff --git a/tests/sstables/large_partition/try1-data-ka-3-Summary.db
b/tests/sstables/large_partition/try1-data-ka-3-Summary.db
--- a/tests/sstables/large_partition/try1-data-ka-3-Summary.db
+++ b/tests/sstables/large_partition/try1-data-ka-3-Summary.db
null
diff --git a/tests/sstables/large_partition/try1-data-ka-3-TOC.txt
b/tests/sstables/large_partition/try1-data-ka-3-TOC.txt
--- a/tests/sstables/large_partition/try1-data-ka-3-TOC.txt

Commit Bot

<bot@cloudius-systems.com>

unread,

Aug 7, 2016, 10:54:06 AM8/7/16

to scylladb-dev@googlegroups.com, Nadav Har'El

From: Nadav Har'El <n...@scylladb.com>
Committer: Nadav Har'El <n...@scylladb.com>
Branch: master

sstables: STOP_THEN_ATOM_START parser state

In a later patch adding "promoted index" read support, we would like to
parse only part of an sstable row. In that case, the parser should start
not at the usual ROW_START state, but rather at the ATOM_START state.

But there's a problem: The sstable parser consumer currently assumes that
the parser stops after the start of the row, before reading any atoms.
So in the partial row case too, we must stop parsing before reading the
first atom.

For this, this patch adds the new "STOP_THEN_ATOM_START" parser state.
When starting in this state, the parser stops immediately (with
row_consumer::proceed::no), and when restarted again it will be in the
ATOM_START case.

Signed-off-by: Nadav Har'El <n...@scylladb.com>

---

diff --git a/sstables/row.cc b/sstables/row.cc
--- a/sstables/row.cc
+++ b/sstables/row.cc
@@ -51,6 +51,7 @@ class data_consume_rows_context : public
data_consumer::continuous_data_consumer

RANGE_TOMBSTONE_3,
RANGE_TOMBSTONE_4,
RANGE_TOMBSTONE_5,
+ STOP_THEN_ATOM_START,
} _state = state::ROW_START;

row_consumer& _consumer;

@@ -69,6 +70,7 @@ class data_consume_rows_context : public
data_consumer::continuous_data_consumer

|| (_state == state::CELL_VALUE_BYTES_2)
|| (_state == state::ATOM_START_2)
|| (_state == state::ATOM_MASK_2)
+ || (_state == state::STOP_THEN_ATOM_START)
|| (_state == state::EXPIRING_CELL_3)) && (_prestate ==
prestate::NONE));
}

@@ -319,6 +321,9 @@ class data_consume_rows_context : public
data_consumer::continuous_data_consumer

}
break;
}
+ case state::STOP_THEN_ATOM_START:

+ _state = state::ATOM_START;

Commit Bot

<bot@cloudius-systems.com>

unread,

Aug 7, 2016, 10:54:07 AM8/7/16

to scylladb-dev@googlegroups.com, Nadav Har'El

From: Nadav Har'El <n...@scylladb.com>
Committer: Nadav Har'El <n...@scylladb.com>
Branch: master

sstables: disk_read_range

Currently, the main sstable data parsing entry point data_consume_rows()
takes a contiguous range of bytes to read from disk and parse. This range
is supposed to be an entire partition or contiguous group of partitions.
and is self contained (can be parsed without extra information about the
identity of these partitions).

For the promoted index feature (which we will add in a following patch)
we will want the range to span only a part of a partition, and will need
the caller to provide some information not available to the parser (such
as the partition's key). In the future, we will also want to support a
vector of byte ranges, instead of just one.

So in preparation for this, this patch simply replaces the start/end pair
by a new class disk_read_range, which can be easily extended in later
patches. No new functionality is introduced in this patch.

Signed-off-by: Nadav Har'El <n...@scylladb.com>

---

diff --git a/sstables/partition.cc b/sstables/partition.cc
--- a/sstables/partition.cc
+++ b/sstables/partition.cc

@@ -590,17 +590,17 @@ struct sstable_data_source {
, _context(_sst->data_consume_rows(_consumer))
{ }

- sstable_data_source(shared_sstable sst, mp_row_consumer&& consumer,
uint64_t start, uint64_t end)
+ sstable_data_source(shared_sstable sst, mp_row_consumer&& consumer,
sstable::disk_read_range toread)
: _sst(std::move(sst))
, _consumer(std::move(consumer))
- , _context(_sst->data_consume_rows(_consumer, start, end))
+ , _context(_sst->data_consume_rows(_consumer, std::move(toread)))
{ }

sstable_data_source(schema_ptr s, shared_sstable sst, const
sstables::key& k, const io_priority_class& pc,
- query::clustering_key_filtering_context ck_filtering, uint64_t
start, uint64_t end)
+ query::clustering_key_filtering_context ck_filtering,
sstable::disk_read_range toread)
: _sst(std::move(sst))
, _consumer(k, s, ck_filtering, pc)
- , _context(_sst->data_consume_rows(_consumer, start, end))
+ , _context(_sst->data_consume_rows(_consumer, std::move(toread)))
{ }
};

@@ -680,9 +680,9 @@ class sstable_streamed_mutation : public
streamed_mutation::impl {

static future<streamed_mutation> create(schema_ptr s, shared_sstable
sst, const sstables::key& k,

query::clustering_key_filtering_context ck_filtering,
- const io_priority_class& pc,
uint64_t start, uint64_t end)
+ const io_priority_class& pc,
sstable::disk_read_range toread)
{
- auto ds = make_lw_shared<sstable_data_source>(s, sst, k, pc,
ck_filtering, start, end);
+ auto ds = make_lw_shared<sstable_data_source>(s, sst, k, pc,
ck_filtering, std::move(toread));
return ds->_context.read().then([s, ds] {
auto mut = ds->_consumer.get_mutation();
assert(mut);

@@ -763,7 +763,7 @@ sstables::sstable::read_row(schema_ptr schema,

auto position = index_list[index_idx].position();

return this->data_end_position(summary_idx, index_idx, index_list,
pc).then([&key, schema, ck_filtering, this, position, &pc] (uint64_t end) {
- return sstable_streamed_mutation::create(schema,
this->shared_from_this(), key, ck_filtering, pc, position, end).then([]
(auto sm) {

+ return sstable_streamed_mutation::create(schema,

this->shared_from_this(), key, ck_filtering, pc, {position, end}).then([]
(auto sm) {

return streamed_mutation_opt(std::move(sm));
});
});

@@ -781,12 +781,12 @@ class mutation_reader::impl {

mp_row_consumer _consumer;
std::function<future<lw_shared_ptr<sstable_data_source>> ()>
_get_data_source;
public:
- impl(shared_sstable sst, schema_ptr schema, uint64_t start, uint64_t
end,
+ impl(shared_sstable sst, schema_ptr schema, sstable::disk_read_range
toread,
const io_priority_class &pc)
: _schema(schema)
, _consumer(schema, query::no_clustering_key_filtering, pc)
- , _get_data_source([this, sst = std::move(sst), start, end] {
- auto ds = make_lw_shared<sstable_data_source>(std::move(sst),
std::move(_consumer), start, end);
+ , _get_data_source([this, sst = std::move(sst), toread] {
+ auto ds = make_lw_shared<sstable_data_source>(std::move(sst),
std::move(_consumer), std::move(toread));
return
make_ready_future<lw_shared_ptr<sstable_data_source>>(std::move(ds));
}) { }
impl(shared_sstable sst, schema_ptr schema,

@@ -808,7 +808,7 @@ class mutation_reader::impl {

, _get_data_source([this, sst = std::move(sst), start =
std::move(start), end = std::move(end)] () mutable {
return start().then([this, sst = std::move(sst), end =
std::move(end)] (uint64_t start) mutable {
return end().then([this, sst = std::move(sst), start]
(uint64_t end) mutable {
- return
make_lw_shared<sstable_data_source>(std::move(sst), std::move(_consumer),
start, end);
+ return
make_lw_shared<sstable_data_source>(std::move(sst), std::move(_consumer),
sstable::disk_read_range{start, end});
});
});
}) { }

diff --git a/sstables/row.cc b/sstables/row.cc
--- a/sstables/row.cc
+++ b/sstables/row.cc

@@ -383,18 +383,19 @@ future<> data_consume_context::read() {
}

data_consume_context sstable::data_consume_rows(
- row_consumer& consumer, uint64_t start, uint64_t end) {
+ row_consumer& consumer, sstable::disk_read_range toread) {
// TODO: The second "end - start" below is redundant: The first one
tells
// data_stream() to stop at the "end" byte, which allows optimal read-
// ahead and avoiding over-read at the end. The second one tells the
// consumer to stop at exactly the same place, and forces the consumer

// to maintain its own byte count.
return std::make_unique<data_consume_context::impl>(shared_from_this(),

- consumer, data_stream(start, end - start,
consumer.io_priority()), end - start);
+ consumer, data_stream(toread.start, toread.end - toread.start,
+ consumer.io_priority()), toread.end - toread.start);
}

data_consume_context sstable::data_consume_rows(row_consumer& consumer) {
- return data_consume_rows(consumer, 0, data_size());
+ return data_consume_rows(consumer, {0, data_size()});
}

future<> sstable::data_consume_rows_at_once(row_consumer& consumer,

diff --git a/sstables/sstables.hh b/sstables/sstables.hh
--- a/sstables/sstables.hh
+++ b/sstables/sstables.hh
@@ -155,6 +155,16 @@ public:

// object lives until then (e.g., using the do_with() idiom).
future<> data_consume_rows_at_once(row_consumer& consumer, uint64_t
pos, uint64_t end);

+ struct disk_read_range {
+ uint64_t start;
+ uint64_t end;
+ disk_read_range() : start(0), end(0) {}
+ disk_read_range(uint64_t start, uint64_t end) :
+ start(start), end(end) { }

+ explicit operator bool() const {
+ return start != end;
+ }

+ };

// data_consume_rows() iterates over rows in the data file from

// a particular range, feeding them into the consumer. The iteration is
@@ -172,7 +182,7 @@ public:
// The caller must ensure (e.g., using do_with()) that the context
object,
// as well as the sstable, remains alive as long as a read() is in
// progress (i.e., returned a future which hasn't completed yet).
- data_consume_context data_consume_rows(row_consumer& consumer,
uint64_t start, uint64_t end);
+ data_consume_context data_consume_rows(row_consumer& consumer,
disk_read_range toread);

// Like data_consume_rows() with bounds, but iterates over whole range
data_consume_context data_consume_rows(row_consumer& consumer);

diff --git a/tests/sstable_test.cc b/tests/sstable_test.cc
--- a/tests/sstable_test.cc
+++ b/tests/sstable_test.cc

@@ -448,7 +448,7 @@ SEASTAR_TEST_CASE(compressed_row_read_at_once) {

SEASTAR_TEST_CASE(uncompressed_rows_read_one) {
return reusable_sst("tests/sstables/uncompressed", 1).then([] (auto
sstp) {
return do_with(test_row_consumer(1418656871665302), [sstp] (auto&
c) {
- auto context = sstp->data_consume_rows(c, 0, 95);
+ auto context = sstp->data_consume_rows(c, {0, 95});
auto fut = context.read();
return fut.then([sstp, &c, context = std::move(context)] {
BOOST_REQUIRE(c.count_row_start == 1);

@@ -465,7 +465,7 @@ SEASTAR_TEST_CASE(uncompressed_rows_read_one) {

Commit Bot

<bot@cloudius-systems.com>

unread,

Aug 7, 2016, 10:54:09 AM8/7/16

to scylladb-dev@googlegroups.com, Nadav Har'El

From: Nadav Har'El <n...@scylladb.com>
Committer: Nadav Har'El <n...@scylladb.com>
Branch: master

make column-name parsing code public

The "struct column" code in partition.cc is generally useful code for
parsing serialized column names from the sstable. It is currently private
inside the "mp_row_consumer" class. But in a next patch we'll also want
to use it in the "sstable" class, for the promoted-index parsing code,
which among other things also needs to deserialize column names.

The trivial fix, in this patch, is to make this code "public". However,
for now it is still available only in partition.cc.

Signed-off-by: Nadav Har'El <n...@scylladb.com>

---
diff --git a/sstables/partition.cc b/sstables/partition.cc
--- a/sstables/partition.cc
+++ b/sstables/partition.cc

@@ -125,6 +125,7 @@ class mp_row_consumer : public row_consumer {

stdx::optional<new_mutation> _mutation;
bool _is_mutation_end;

+public:
struct column {
bool is_static;
bytes_view col_name;

@@ -200,6 +201,7 @@ class mp_row_consumer : public row_consumer {

Commit Bot

<bot@cloudius-systems.com>

unread,

Aug 7, 2016, 10:54:11 AM8/7/16

to scylladb-dev@googlegroups.com, Nadav Har'El

From: Nadav Har'El <n...@scylladb.com>
Committer: Nadav Har'El <n...@scylladb.com>
Branch: master

sstables: expose promoted index in index entry

Our index_entry type, holding one partition's entry that we read from the
index file, already contained the "_promoted_index" which we read from
disk - as an unparsed byte buffer. But there wasn't any API to access
this buffer after it was read. This patch adds a trivial getter, to get
a read-only view of this buffer.

Signed-off-by: Nadav Har'El <n...@scylladb.com>

---

diff --git a/sstables/types.hh b/sstables/types.hh
--- a/sstables/types.hh
+++ b/sstables/types.hh

Commit Bot

<bot@cloudius-systems.com>

unread,

Aug 7, 2016, 10:54:13 AM8/7/16

to scylladb-dev@googlegroups.com, Nadav Har'El

From: Nadav Har'El <n...@scylladb.com>
Committer: Nadav Har'El <n...@scylladb.com>
Branch: master

sstables: promoted index read support

Signed-off-by: Nadav Har'El <n...@scylladb.com>

---

diff --git a/sstables/partition.cc b/sstables/partition.cc
--- a/sstables/partition.cc
+++ b/sstables/partition.cc

@@ -738,37 +738,16 @@ sstables::sstable::read_row(schema_ptr schema,

- auto position = index_list[index_idx].position();
- return this->data_end_position(summary_idx, index_idx, index_list,

pc).then([&key, schema, ck_filtering, this, position, &pc] (uint64_t end) {
- return sstable_streamed_mutation::create(schema,

this->shared_from_this(), key, ck_filtering, pc, {position, end}).then([]
(auto sm) {
- return streamed_mutation_opt(std::move(sm));
- });

+ return sstable_streamed_mutation::create(schema,

this->shared_from_this(), key, ck_filtering, pc, std::move(toread)).then([]
(auto sm) {

+ return streamed_mutation_opt(std::move(sm));
});
});
}

diff --git a/sstables/row.cc b/sstables/row.cc
--- a/sstables/row.cc
+++ b/sstables/row.cc

@@ -63,6 +63,7 @@ class data_consume_rows_context : public
data_consumer::continuous_data_consumer

bool _deleted;
uint32_t _ttl, _expiration;

+ bool _read_partial_row = false;

public:
bool non_consuming() const {

@@ -332,12 +333,42 @@ class data_consume_rows_context : public
data_consumer::continuous_data_consumer
}

data_consume_rows_context(row_consumer& consumer,
- input_stream<char> && input, uint64_t maxlen) :
- continuous_data_consumer(std::move(input), maxlen)
- , _consumer(consumer) {
+ input_stream<char> && input, uint64_t maxlen,
+
std::experimental::optional<sstable::disk_read_range::row_info> ri = {})
+ : continuous_data_consumer(std::move(input), maxlen)
+ , _consumer(consumer) {
+ // If the "ri" option is given, we are reading a partition from the
+ // middle (in the beginning of an atom), as would happen when we
use
+ // the "promoted index" to skip closer to where a particular column
+ // starts. When we start in the middle of the partition, we will
not
+ // read the key nor the tombstone from the disk, so the caller
needs
+ // to provide them (the tombstone is provided in the promoted index
+ // exactly for that reason).
+ if (ri) {
+ _read_partial_row = true;
+ auto ret = _consumer.consume_row_start(ri->k, ri->deltime);
+ if (ret == row_consumer::proceed::yes) {

+ _state = state::ATOM_START;

+ } else {
+ // If we were asked to stop parsing after consuming the row
+ // start, we can't go to ATOM_START, need to use a new
state
+ // which stops parsing, and continues at ATOM_START later.
+ _state = state::STOP_THEN_ATOM_START;
+ }
+ }
}

void verify_end_state() {
+ if (_read_partial_row) {
+ // If reading a partial row (i.e., when we have a clustering
row
+ // filter and using a promoted index), we may be in ATOM_START
+ // state instead of ROW_START. In that case we did not read the
+ // end-of-row marker and consume_row_end() was never called.
+ if (_state == state::ATOM_START) {
+ _consumer.consume_row_end();
+ return;
+ }
+ }
if (_state != state::ROW_START || _prestate != prestate::NONE) {
throw malformed_sstable_exception("end of input, but not end
of row");
}

@@ -354,9 +385,10 @@ class data_consume_context::impl {

shared_sstable _sst;
std::unique_ptr<data_consume_rows_context> _ctx;
public:
- impl(shared_sstable sst, row_consumer& consumer, input_stream<char>&&
input, uint64_t maxlen)
+ impl(shared_sstable sst, row_consumer& consumer, input_stream<char>&&
input, uint64_t maxlen,
+
std::experimental::optional<sstable::disk_read_range::row_info> ri)
: _sst(std::move(sst))
- , _ctx(new data_consume_rows_context(consumer, std::move(input),
maxlen))
+ , _ctx(new data_consume_rows_context(consumer, std::move(input),
maxlen, ri))
{ }
~impl() {
if (_ctx) {

@@ -391,7 +423,7 @@ data_consume_context sstable::data_consume_rows(

// to maintain its own byte count.
return std::make_unique<data_consume_context::impl>(shared_from_this(),

consumer, data_stream(toread.start, toread.end - toread.start,

Commit Bot

<bot@cloudius-systems.com>

unread,

Aug 7, 2016, 10:54:13 AM8/7/16

to scylladb-dev@googlegroups.com, Nadav Har'El

From: Nadav Har'El <n...@scylladb.com>
Committer: Nadav Har'El <n...@scylladb.com>
Branch: master

sstables: introduce find_disk_ranges()

Our sstable reading code is currently hard-coded to read entire partitions,
even if we know that only a subset of the columns are requested.
This patch introduces find_disk_ranges(), a function to find the ranges of
bytes we need to read from the sstable data file to guarantee that the
desired columns from the desired partition are read.

The returned range may be the entire byte range of the given partition -
as found using the summary and index files - but if the index contains a
"promoted index" (basically a sample of column positions for each key)
we may return a smaller range. The "disk_read_range" type introduced in
the previous patch is extended here to support reading a partial partition -
by including additional information which would be missed when reading only
part of a partition (viz., the partition key and the partition's tombstone).

This function isn't used in this patch - we will wire its use in the next
patch, which will complete the read-side support for the promoted index.

Signed-off-by: Nadav Har'El <n...@scylladb.com>

---
diff --git a/sstables/partition.cc b/sstables/partition.cc
--- a/sstables/partition.cc
+++ b/sstables/partition.cc

@@ -28,6 +28,7 @@
#include "unimplemented.hh"
#include "utils/move.hh"
#include "dht/i_partitioner.hh"
+#include <seastar/core/byteorder.hh>

namespace sstables {

@@ -772,6 +773,167 @@ sstables::sstable::read_row(schema_ptr schema,
});

+ return read_indexes(summary_idx, pc).then([this, schema, ck_filtering,

&key, token, summary_idx, &pc] (auto index_list) {

+ });
+ });
+}
+

class mutation_reader::impl {
private:
schema_ptr _schema;

diff --git a/sstables/sstables.hh b/sstables/sstables.hh
--- a/sstables/sstables.hh
+++ b/sstables/sstables.hh

@@ -155,15 +155,40 @@ public:

// object lives until then (e.g., using the do_with() idiom).
future<> data_consume_rows_at_once(row_consumer& consumer, uint64_t
pos, uint64_t end);

+ // disk_read_range describes a byte ranges covering part of an sstable
+ // row that we need to read from disk. Usually this is the whole byte
+ // range covering a single sstable row, but in very large rows we might
+ // want to only read a subset of the atoms which we know contains the
+ // columns we are looking for. When the range to be read does NOT
include
+ // the entire row, the caller needs to supply the optional "row_info"
+ // containing information about the entire row (key and deletion time)
+ // which is normally read from the beginning of the row.
struct disk_read_range {
+ // TODO: this should become a vector of ranges
uint64_t start;
uint64_t end;
+ // When the range above does not cover the beginning of the sstable
+ // row, we need to supply information which is only available at

the

+ // beginning of the row - the row's key and its tombstone if any.
+ struct row_info {
+ key k;
+ deletion_time deltime;

+ };

+ std::experimental::optional<row_info> ri;
disk_read_range() : start(0), end(0) {}
disk_read_range(uint64_t start, uint64_t end) :
start(start), end(end) { }
+ disk_read_range(uint64_t start, uint64_t end, const key& key,
const deletion_time& deltime) :
+ start(start), end(end), ri(row_info{key, deltime}) { }
explicit operator bool() const {

return start != end;
}

+ // found_row() is true if the row was found. This is not the same
as

+ // operator bool(): It is possible that found_row() but the
promoted

+ // index ruled out anything to read (in this case "ri" was set).

+ bool found_row() const {
+ return start != end || ri;

+ }
};

// data_consume_rows() iterates over rows in the data file from

Commit Bot

<bot@cloudius-systems.com>

unread,

Aug 7, 2016, 10:54:14 AM8/7/16

to scylladb-dev@googlegroups.com, Nadav Har'El

From: Nadav Har'El <n...@scylladb.com>
Committer: Nadav Har'El <n...@scylladb.com>
Branch: master

range tombstone accumulator: add method

Add to the range_tombstone_accumulator a range_tombstones_for_row(ck)
method.

Just like the existing tombstone_for_row(ck), this function drops from
the accumulator tombstones that end before ck. But while the existing
function returned just a single tombstone affecting the given row (the
most recent tombstone), the new function range_tombstones_for_row(ck)
returns all the accumulated range tombstones which cover ck.

This function will be useful for the promoted-index writing code later,
which divides a partition into blocks which may be read independently,
so each block needs to start with a repeat of the earlier tombstones
which still cover the first row in the new block.

Signed-off-by: Nadav Har'El <n...@scylladb.com>

---

diff --git a/range_tombstone.hh b/range_tombstone.hh
--- a/range_tombstone.hh
+++ b/range_tombstone.hh

Commit Bot

<bot@cloudius-systems.com>

unread,

Aug 7, 2016, 10:54:16 AM8/7/16

to scylladb-dev@googlegroups.com, Nadav Har'El

From: Nadav Har'El <n...@scylladb.com>
Committer: Nadav Har'El <n...@scylladb.com>
Branch: master

sstables: promoted index write support

This patch adds writing of promoted index to sstables.

The promoted index is basically a sample of columns and their positions
for large partitions: The promoted index appears in the sstable's index
file for partitions which are larger than 64 KB, and divides the partition
to 64 KB blocks (as in Cassandra, this interval is configurable through
the column_index_size_in_kb config parameter). Beyond modifying the index
file, having a promoted index may also modify the data file: Since each
of blocks may be read independently, we need to add in the beginning of
each block the list of range tombstones that are still open at that
position.

See also https://github.com/scylladb/scylla/wiki/SSTables-Index-File

Fixes #959

Signed-off-by: Nadav Har'El <n...@scylladb.com>

---

diff --git a/sstables/sstables.cc b/sstables/sstables.cc
--- a/sstables/sstables.cc
+++ b/sstables/sstables.cc

@@ -31,6 +31,7 @@
#include "core/do_with.hh"
#include "core/thread.hh"
#include <seastar/core/shared_future.hh>
+#include <seastar/core/byteorder.hh>
#include <iterator>

#include "types.hh"
@@ -56,6 +57,7 @@

#include "checked-file-impl.hh"
#include "disk-error-handler.hh"
+#include "service/storage_service.hh"

thread_local disk_error_signal_type sstable_read_error;
thread_local disk_error_signal_type sstable_write_error;

@@ -296,6 +298,12 @@ inline void write(file_writer& out, bytes_view s) {

@@ -1223,6 +1333,7 @@ void sstable::write_collection(file_writer& out,
const composite& clustering_key

const bytes& column_name = cdef.name();
write_range_tombstone(out, clustering_key, clustering_key, {
bytes_view(column_name) }, mview.tomb);
for (auto& cp: mview.cells) {
+ maybe_flush_pi_block(out, clustering_key, { column_name, cp.first
});
write_column_name(out, clustering_key, { column_name, cp.first });
write_cell(out, cp.second);
}

@@ -1234,11 +1345,22 @@ void sstable::write_clustered_row(file_writer& out,
const schema& schema, const

auto clustering_key = composite::from_clustering_element(schema,
clustered_row.key());

if (schema.is_compound() && !schema.is_dense()) {
+ maybe_flush_pi_block(out, clustering_key, { bytes_view() });
write_row_marker(out, clustered_row.marker(), clustering_key);
}
// Before writing cells, range tombstone must be written if the row
has any (deletable_row::t).
if (clustered_row.tomb()) {
+ maybe_flush_pi_block(out, clustering_key, {});
write_range_tombstone(out, clustering_key, clustering_key, {},
clustered_row.tomb());
+ // Because we currently may break a partition to promoted-index
blocks
+ // in the middle of a clustered row, we also need to track the
current
+ // row's tombstone - not just range tombstones - which may effect

the

+ // beginning of a new block.
+ // TODO: consider starting a new block only between rows, so the
+ // following code can be dropped:
+ _pi_write.tombstone_accumulator->apply(range_tombstone(
+ clustered_row.key(), bound_kind::incl_start,
+ clustered_row.key(), bound_kind::incl_end,
clustered_row.tomb()));
}

// Write all cells of a partition's row.

@@ -1256,14 +1378,18 @@ void sstable::write_clustered_row(file_writer& out,
const schema& schema, const

if (schema.is_compound()) {
if (schema.is_dense()) {
+ maybe_flush_pi_block(out, composite(), {
bytes_view(clustering_key) });
write_column_name(out, bytes_view(clustering_key));
} else {
+ maybe_flush_pi_block(out, clustering_key, {
bytes_view(column_name) });
write_column_name(out, clustering_key, {
bytes_view(column_name) });
}
} else {
if (schema.is_dense()) {
+ maybe_flush_pi_block(out, composite(), {
bytes_view(clustered_row.key().get_component(schema, 0)) });
write_column_name(out,
bytes_view(clustered_row.key().get_component(schema, 0)));
} else {
+ maybe_flush_pi_block(out, composite(), {
bytes_view(column_name) });
write_column_name(out, bytes_view(column_name));
}
}

@@ -1282,16 +1408,25 @@ void sstable::write_static_row(file_writer& out,
const schema& schema, const row

components_writer::index_file_writer(sstable& sst, const io_priority

return file_writer(sst._index_file, std::move(options));
}

+// Get the currently loaded configuration, or the default configuration in
+// case none has been loaded (this happens, for example, in unit tests).
+static const db::config& get_config() {
+ if (service::get_storage_service().local_is_initialized()) {
+ return
service::get_local_storage_service().db().local().get_config();
+ } else {
+ static db::config default_config;
+ return default_config;
+ }
+}
+
components_writer::components_writer(sstable& sst, const schema& s,
file_writer& out,
uint64_t estimated_partitions,
uint64_t max_sstable_size,
const io_priority_class& pc)

@@ -1415,6 +1561,7 @@ components_writer::components_writer(sstable& sst,
const schema& s, file_writer&

, _max_sstable_size(max_sstable_size)
{
_sst._filter = utils::i_filter::get_filter(estimated_partitions,
_schema.bloom_filter_fp_chance());
+ _sst._pi_write.desired_block_size =
get_config().column_index_size_in_kb() * 1024;

prepare_summary(_sst._summary, estimated_partitions,
_schema.min_index_interval());

@@ -1435,7 +1582,17 @@ void components_writer::consume_new_partition(const
dht::decorated_key& dk) {

p_key.value = bytes_view(*_partition_key);

// Write index file entry from partition key into index file.
- write_index_entry(_index, p_key, _out.offset());
+ // Write an index entry minus the "promoted index" (sample of columns)
+ // part. We can only write that after processing the entire partition
+ // and collecting the sample of columns.
+ write_index_header(_index, p_key, _out.offset());
+ _sst._pi_write.data = {};
+ _sst._pi_write.numblocks = 0;
+ _sst._pi_write.deltime.local_deletion_time =
std::numeric_limits<int32_t>::max();
+ _sst._pi_write.deltime.marked_for_delete_at =
std::numeric_limits<int64_t>::min();
+ _sst._pi_write.block_start_offset = _out.offset();
+ _sst._pi_write.tombstone_accumulator =
range_tombstone_accumulator(_schema, false);
+ _sst._pi_write.schemap = &_schema; // sadly we need this

// Write partition key into data file.
write(_out, p_key);

@@ -1461,6 +1618,8 @@ void components_writer::consume(tombstone t) {

}
write(_out, d);
_tombstone_written = true;
+ // TODO: need to verify we don't do this twice?
+ _sst._pi_write.deltime = d;
}

stop_iteration components_writer::consume(static_row&& sr) {
@@ -1477,13 +1636,35 @@ stop_iteration

components_writer::consume(clustering_row&& cr) {

diff --git a/sstables/sstables.hh b/sstables/sstables.hh
--- a/sstables/sstables.hh
+++ b/sstables/sstables.hh

@@ -407,6 +407,27 @@ private:
uint64_t _filter_file_size = 0;
uint64_t _bytes_on_disk = 0;

+ // _pi_write is used temporarily for building the promoted

+ // index (column sample) of one partition when writing a new sstable.
+ struct {

Commit Bot

<bot@cloudius-systems.com>

unread,

Aug 7, 2016, 10:54:17 AM8/7/16

to scylladb-dev@googlegroups.com, Nadav Har'El

From: Nadav Har'El <n...@scylladb.com>
Committer: Nadav Har'El <n...@scylladb.com>
Branch: master

tests: add test for promoted index writing

In this unit test, we create using Scylla C++ code, the same large
partition with 13520 CQL rows as we previously imported from Cassandra
for the large partition test. We then verify that the sstable index file
we just wrote is byte-for-byte identical to the one previously created by
Cassandra. They should indeed be identical, because the data file has the
same layout (even if timestamps are different) and our default promoted-
index block size is the same (64K) so the sample of columns should be
identical.

Signed-off-by: Nadav Har'El <n...@scylladb.com>

---

diff --git a/tests/sstable_test.cc b/tests/sstable_test.cc
--- a/tests/sstable_test.cc
+++ b/tests/sstable_test.cc

@@ -38,6 +38,7 @@
#include "sstable_test.hh"
#include "tmpdir.hh"
#include "partition_slice_builder.hh"
+#include "tests/test_services.hh"

#include "disk-error-handler.hh"

@@ -1166,3 +1167,56 @@ SEASTAR_TEST_CASE(sub_partitions_read) {
});
});
}
+
+// A silly, inefficient but effective, way to compare two files by reading
+// them entirely into memory.
+static future<> compare_files(sstring file1, sstring file2) {
+ return read_file(file1).then([file2] (auto in1) {
+ return read_file(file2).then([in1 = std::move(in1)] (auto in2) {
+ // assert that both files have the same size.
+ BOOST_REQUIRE(in1.second == in2.second);
+ // assert that both files have the same content.
+ BOOST_REQUIRE(::memcmp(in1.first.get(), in2.first.get(),
in1.second) == 0);

+ });
+ });
+
+}

+ auto sst = make_lw_shared<sstable>("try1", "data",
+ "tests/sstables/tests-temporary", 100,

+ sstables::sstable::version_types::ka, big);

Commit Bot

<bot@cloudius-systems.com>

unread,

Aug 9, 2016, 9:58:36 AM8/9/16

to scylladb-dev@googlegroups.com, Nadav Har'El

From: Nadav Har'El <n...@scylladb.com>

Committer: Avi Kivity <a...@scylladb.com>
Branch: branch-1.3

sstables: promoted index write support

This patch adds writing of promoted index to sstables.

The promoted index is basically a sample of columns and their positions
for large partitions: The promoted index appears in the sstable's index
file for partitions which are larger than 64 KB, and divides the partition
to 64 KB blocks (as in Cassandra, this interval is configurable through
the column_index_size_in_kb config parameter). Beyond modifying the index
file, having a promoted index may also modify the data file: Since each
of blocks may be read independently, we need to add in the beginning of
each block the list of range tombstones that are still open at that
position.

See also https://github.com/scylladb/scylla/wiki/SSTables-Index-File

Fixes #959

Signed-off-by: Nadav Har'El <n...@scylladb.com>

(cherry picked from commit 0d8463aba580cb3c84cb9cafa825e0e2eb16ebc1)

@@ -372,6 +372,27 @@ private:

Tomasz Grabiec

<tgrabiec@scylladb.com>

unread,

Aug 10, 2016, 8:06:50 AM8/10/16

to Nadav Har'El, scylladb-dev

Paging queries will use a slice with start bound set but no end bound. I think we already have bounds for the whole partition from the index, we could set found_range_end right away if end bound is not set so that the scanning of promoted index stops as soon as we find the start bound.

--
You received this message because you are subscribed to the Google Groups "ScyllaDB development" group.
To unsubscribe from this group and stop receiving emails from it, send an email to scylladb-dev+unsubscribe@googlegroups.com.
To post to this group, send email to scylla...@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/scylladb-dev/94eb2c060008f7869105397c7863%40google.com.
For more options, visit https://groups.google.com/d/optout.

Tomasz Grabiec

<tgrabiec@scylladb.com>

unread,

Aug 10, 2016, 9:33:50 AM8/10/16

to Nadav Har'El, scylladb-dev

On Sun, Aug 7, 2016 at 4:54 PM, Commit Bot <b...@cloudius-systems.com> wrote:

From: Nadav Har'El <n...@scylladb.com>
Committer: Nadav Har'El <n...@scylladb.com>
Branch: master

sstables: promoted index read support

This patch adds support more efficiently reading small parts of a large
partition, without reading the entire partition as we had to do so far.
This is done using the "promoted index".

The "promoted index" is stored in the sstable index file, and provides
for each large sstable row ("partition" in CQL nomenclature) a sample of
the column names at (for example) 64KB intervals. This means that when we
read a slice of columns (e.g., cql rows), or page through a large partition,
we do not have to read the entire partition from disk.

This patch only implements the read side of promoted index - a later patch
will add the write-side support (i.e., writing the promoted index to the
index file while saving the sstable). Nevertheless this patch can already
be tested by reading existing sstables from Cassandra which include a
promoted index - such as the one included in the test in the previous patch.

The use of the promoted index currently has two limitations:

1. It is only used when reading a single partition with sstable::read_row(),
not when scanning through many partitions with sstable::read_range_rows()
or sstable::read_rows().

2. It is only used when filtering a single clustering-key range, rather
than a list of disjoint ranges. A single range is the common case.

This will affect performance of CQL statements with "IN" restriction on clustering columns. We should open a github issue about this so that we don't forget it.

We should still read the static row.

https://github.com/scylladb/scylla/issues/1568

--
You received this message because you are subscribed to the Google Groups "ScyllaDB development" group.
To unsubscribe from this group and stop receiving emails from it, send an email to scylladb-dev+unsubscribe@googlegroups.com.
To post to this group, send email to scylla...@googlegroups.com.

To view this discussion on the web visit https://groups.google.com/d/msgid/scylladb-dev/001a1147597408ee0e05397c7998%40google.com.

Tomasz Grabiec

<tgrabiec@scylladb.com>

unread,

Aug 10, 2016, 12:20:28 PM8/10/16

to Nadav Har'El, scylladb-dev

On Sun, Aug 7, 2016 at 4:54 PM, Commit Bot <b...@cloudius-systems.com> wrote:

From: Nadav Har'El <n...@scylladb.com>
Committer: Nadav Har'El <n...@scylladb.com>
Branch: master

<snip>

diff --git a/sstables/sstables.hh b/sstables/sstables.hh
--- a/sstables/sstables.hh
+++ b/sstables/sstables.hh
@@ -407,6 +407,27 @@ private:
uint64_t _filter_file_size = 0;
uint64_t _bytes_on_disk = 0;

+ // _pi_write is used temporarily for building the promoted
+ // index (column sample) of one partition when writing a new sstable.
+ struct {
+ // Unfortunately we cannot output the promoted index directly to the
+ // index file because it needs to be prepended by its size.

Does it mean that sstable compaction memory footprint is proportional to partition size? This may be a problem with large partitions.

Nadav Har'El

<nyh@scylladb.com>

unread,

Aug 10, 2016, 12:51:36 PM8/10/16

to Tomasz Grabiec, scylladb-dev

On Wed, Aug 10, 2016 at 7:20 PM, Tomasz Grabiec <tgra...@scylladb.com> wrote:

+ // _pi_write is used temporarily for building the promoted
+ // index (column sample) of one partition when writing a new sstable.
+ struct {
+ // Unfortunately we cannot output the promoted index directly to the
+ // index file because it needs to be prepended by its size.

Does it mean that sstable compaction memory footprint is proportional to partition size? This may be a problem with large partitions.

Yes, but not just compaction - also read of the resulting sstable - because reading the index entry (to find the partition) always reads the entire "promoted index" saved in this index entry - it was like that even before my patch.

I think the idea is that for large but not super-super-huge partitions, the divisor (64K) is supposed to be so high you won't care about the proportionality. Of course, we know how sucky this is in theory as the partitions grow towards infinity.

Avi Kivity

<avi@scylladb.com>

unread,

Aug 11, 2016, 4:00:29 AM8/11/16

to Nadav Har'El, Tomasz Grabiec, scylladb-dev

Or in practice. Let's say each promoted index entry is 100 bytes. Then we have a 640:1 ratio between partition size and index size. A 64MB translates to a 100kB index entry, so you're limited to ~100MB partitions if you want reasonable performance.

If we could guess the partition size up front, we could make the block size sqrt((partition size in bytes) / (estimated promoted index entry size in bytes)). This would make the block size and the index entry size equal, minimizing bandwidth. But it's better to spend that effort on a proper O(log N) index.

Pekka Enberg

<penberg@scylladb.com>

unread,

Aug 24, 2016, 5:45:00 AM8/24/16

to Tomasz Grabiec, Nadav Har'El, scylladb-dev

So did anyone open an issue about it?

Tomasz Grabiec

<tgrabiec@scylladb.com>

unread,

Aug 29, 2016, 11:18:02 AM8/29/16

to Pekka Enberg, Nadav Har'El, scylladb-dev

https://github.com/scylladb/scylla/issues/1621

Reply all

Reply to author

Forward