// Copyright 2017 Google LLC

//

// Licensed 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

//

// https://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.

#include "google/cloud/bigtable/row_range.h"

namespace google {

namespace cloud {

namespace bigtable {

GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN

namespace btproto = ::google::bigtable::v2;

RowRange::RowRange(::google::bigtable::v2::RowRange rhs)

: row_range_(std::move(rhs)) {

// The service treats an empty end key as end of table. Some of our

// intersection logic does not, though. So we are best off sanitizing the

// input, by clearing the end key if it is empty.

if (row_range_.has_end_key_closed()) {

if (internal::IsEmptyRowKey(row_range_.end_key_closed())) {

row_range_.clear_end_key_closed();

}

}

if (row_range_.has_end_key_open()) {

if (internal::IsEmptyRowKey(row_range_.end_key_open())) {

row_range_.clear_end_key_open();

}

}

}

bool RowRange::IsEmpty() const {

RowKeyType unused;

// We do not want to copy the strings unnecessarily, so initialize a reference

// pointing to *_key_closed() or *_key_open(), as needed.

auto const* start = &unused;

bool start_open = false;

switch (row_range_.start_key_case()) {

case btproto::RowRange::kStartKeyClosed:

start = &row_range_.start_key_closed();

break;

case btproto::RowRange::kStartKeyOpen:

start = &row_range_.start_key_open();

start_open = true;

break;

case btproto::RowRange::START_KEY_NOT_SET:

break;

}

// We need to initialize this to something to make g++ happy, but it cannot

// be a value that is discarded in all switch() cases to make Clang happy.

auto const* end = &row_range_.end_key_closed();

bool end_open = false;

switch (row_range_.end_key_case()) {

case btproto::RowRange::kEndKeyClosed:

// Already initialized.

break;

case btproto::RowRange::kEndKeyOpen:

end = &row_range_.end_key_open();

end_open = true;

break;

case btproto::RowRange::END_KEY_NOT_SET:

// A range ending at +infinity is never empty.

return false;

}

// Special case of an open interval of two consecutive strings.

if (start_open && end_open && internal::ConsecutiveRowKeys(*start, *end)) {

return true;

}

// Compare the strings as byte vectors (careful with unsigned chars).

int cmp = internal::CompareRowKey(*start, *end);

if (cmp == 0) {

return start_open || end_open;

}

return cmp > 0;

}

bool RowRange::BelowStart(RowKeyType const& key) const {

switch (row_range_.start_key_case()) {

case btproto::RowRange::START_KEY_NOT_SET:

break;

case btproto::RowRange::kStartKeyClosed:

return key < row_range_.start_key_closed();

case btproto::RowRange::kStartKeyOpen:

return key <= row_range_.start_key_open();

}

return false;

}

bool RowRange::AboveEnd(RowKeyType const& key) const {

switch (row_range_.end_key_case()) {

case btproto::RowRange::END_KEY_NOT_SET:

break;

case btproto::RowRange::kEndKeyClosed:

return key > row_range_.end_key_closed();

case btproto::RowRange::kEndKeyOpen:

return key >= row_range_.end_key_open();

}

return false;

}

std::pair<bool, RowRange> RowRange::Intersect(RowRange const& range) const {

if (range.IsEmpty()) {

return std::make_pair(false, RowRange::Empty());

}

std::string empty;

// The algorithm is simple: start with *this as a the resulting range. Update

// both endpoints based on the value of @p range. If the resulting range is

// empty there is no intersection.

RowRange intersection(*this);

switch (range.row_range_.start_key_case()) {

case btproto::RowRange::START_KEY_NOT_SET:

break;

case btproto::RowRange::kStartKeyClosed: {

auto const& start = range.row_range_.start_key_closed();

// If `range` starts above the current range then there is no

// intersection.

if (intersection.AboveEnd(start)) {

return std::make_pair(false, Empty());

}

// If `start` is inside the intersection (as computed so far), then the

// intersection must start at `start`, and it would be closed if `range`

// is closed at the start.

if (intersection.Contains(start)) {

intersection.row_range_.set_start_key_closed(start);

}

} break;

case btproto::RowRange::kStartKeyOpen: {

// The case where `range` is open on the start point is analogous.

auto const& start = range.row_range_.start_key_open();

if (intersection.AboveEnd(start)) {

return std::make_pair(false, Empty());

}

if (intersection.Contains(start)) {

intersection.row_range_.set_start_key_open(start);

}

} break;

}

// Then check if the end limit of @p range is below *this.

switch (range.row_range_.end_key_case()) {

case btproto::RowRange::END_KEY_NOT_SET:

break;

case btproto::RowRange::kEndKeyClosed: {

// If `range` ends before the start of the intersection there is no

// intersection and we can return immediately.

auto const& end = range.row_range_.end_key_closed();

if (intersection.BelowStart(end)) {

return std::make_pair(false, Empty());

}

// If `end` is inside the intersection as computed so far, then the

// intersection must end at `end` and it is closed if `range` is closed

// at the end.

if (intersection.Contains(end)) {

intersection.row_range_.set_end_key_closed(end);

}

} break;

case btproto::RowRange::kEndKeyOpen: {

// Do the analogous thing for `end` being a open endpoint.

auto const& end = range.row_range_.end_key_open();

if (intersection.BelowStart(end)) {

return std::make_pair(false, Empty());

}

if (intersection.Contains(end)) {

intersection.row_range_.set_end_key_open(end);

}

} break;

}

bool is_empty = intersection.IsEmpty();

return std::make_pair(!is_empty, std::move(intersection));

}

bool operator==(RowRange const& lhs, RowRange const& rhs) {

if (lhs.as_proto().start_key_case() != rhs.as_proto().start_key_case()) {

return false;

}

switch (lhs.as_proto().start_key_case()) {

case btproto::RowRange::START_KEY_NOT_SET:

break;

case btproto::RowRange::kStartKeyClosed:

if (lhs.as_proto().start_key_closed() !=

rhs.as_proto().start_key_closed()) {

return false;

}

break;

case btproto::RowRange::kStartKeyOpen:

if (lhs.as_proto().start_key_open() != rhs.as_proto().start_key_open()) {

return false;

}

break;

}

if (lhs.as_proto().end_key_case() != rhs.as_proto().end_key_case()) {

return false;

}

switch (lhs.as_proto().end_key_case()) {

case btproto::RowRange::END_KEY_NOT_SET:

break;

case btproto::RowRange::kEndKeyClosed:

if (lhs.as_proto().end_key_closed() != rhs.as_proto().end_key_closed()) {

return false;

}

break;

case btproto::RowRange::kEndKeyOpen:

if (lhs.as_proto().end_key_open() != rhs.as_proto().end_key_open()) {

return false;

}

break;

}

return true;

}

std::ostream& operator<<(std::ostream& os, RowRange const& x) {

switch (x.as_proto().start_key_case()) {

case btproto::RowRange::START_KEY_NOT_SET:

os << "['', ";

break;

case btproto::RowRange::kStartKeyClosed:

os << "['" << x.as_proto().start_key_closed() << "', ";

break;

case btproto::RowRange::kStartKeyOpen:

os << "('" << x.as_proto().start_key_open() << "', ";

}

switch (x.as_proto().end_key_case()) {

case btproto::RowRange::END_KEY_NOT_SET:

os << "'')";

break;

case btproto::RowRange::kEndKeyClosed:

os << "'" << x.as_proto().end_key_closed() << "']";

break;

case btproto::RowRange::kEndKeyOpen:

os << "'" << x.as_proto().end_key_open() << "')";

break;

}

return os;

}

GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END

} // namespace bigtable

} // namespace cloud

} // namespace google