In ParseInternalKey(), include corrupt key info in Status (#7515)

Summary:
Fixes Issue https://github.com/facebook/rocksdb/issues/7497

When allow_data_in_errors db_options is set, log error key details in `ParseInternalKey()`

Have fixed most of the calls. Have few TODOs still pending - because have to make more deeper changes to pass in the allow_data_in_errors flag. Will do those in a separate PR later.

Tests:
- make check
- some of the existing tests that exercise the "internal key too small" condition are: dbformat_test, cuckoo_table_builder_test
- some of the existing tests that exercise the corrupted key path are: corruption_test, merge_helper_test, compaction_iterator_test

Example of new status returns:
- Key too small - `Corrupted Key: Internal Key too small. Size=5`
- Corrupt key with allow_data_in_errors option set to false: `Corrupted Key: '<redacted>' seq:3, type:3`
- Corrupt key with allow_data_in_errors option set to true: `Corrupted Key: '61' seq:3, type:3`

Pull Request resolved: https://github.com/facebook/rocksdb/pull/7515

Reviewed By: ajkr

Differential Revision: D24240264

Pulled By: ramvadiv

fbshipit-source-id: bc48f5d4475ac19d7713e16df37505b31aac42e7
main
Ramkumar Vadivelu 4 years ago committed by Facebook GitHub Bot
parent 6c2c0635c9
commit 9a690a74e1
  1. 13
      db/column_family.cc
  2. 3
      db/column_family.h
  3. 44
      db/compaction/compaction_iterator.cc
  4. 2
      db/compaction/compaction_iterator_test.cc
  5. 5
      db/compaction/compaction_job_test.cc
  6. 6
      db/db_compaction_filter_test.cc
  7. 4
      db/db_impl/db_impl_compaction_flush.cc
  8. 9
      db/db_iter.cc
  9. 9
      db/db_iter_test.cc
  10. 9
      db/db_test_util.cc
  11. 6
      db/db_with_timestamp_basic_test.cc
  12. 19
      db/dbformat.cc
  13. 27
      db/dbformat.h
  14. 6
      db/dbformat_test.cc
  15. 29
      db/external_sst_file_ingestion_job.cc
  16. 14
      db/import_column_family_job.cc
  17. 20
      db/merge_helper.cc
  18. 5
      db/merge_helper.h
  19. 14
      db/range_del_aggregator.cc
  20. 7
      db/range_del_aggregator.h
  21. 8
      db/repair.cc
  22. 4
      db/table_properties_collector.cc
  23. 3
      db/version_set.cc
  24. 2
      db/write_batch_test.cc
  25. 8
      java/rocksjni/write_batch_test.cc
  26. 12
      table/block_based/block_based_table_reader.cc
  27. 7
      table/cuckoo/cuckoo_table_builder.cc
  28. 5
      table/cuckoo/cuckoo_table_builder_test.cc
  29. 3
      table/cuckoo/cuckoo_table_reader.cc
  30. 11
      table/mock_table.cc
  31. 3
      table/plain/plain_table_builder.cc
  32. 13
      table/plain/plain_table_key_coding.cc
  33. 8
      table/plain/plain_table_reader.cc
  34. 14
      table/sst_file_dumper.cc
  35. 21
      table/table_test.cc
  36. 3
      tools/block_cache_analyzer/block_cache_trace_analyzer.h
  37. 9
      tools/sst_dump_tool.cc
  38. 10
      utilities/blob_db/blob_compaction_filter.cc
  39. 7
      utilities/debug.cc

@ -1088,7 +1088,7 @@ bool ColumnFamilyData::RangeOverlapWithCompaction(
Status ColumnFamilyData::RangesOverlapWithMemtables(
const autovector<Range>& ranges, SuperVersion* super_version,
bool* overlap) {
bool allow_data_in_errors, bool* overlap) {
assert(overlap != nullptr);
*overlap = false;
// Create an InternalIterator over all unflushed memtables
@ -1121,13 +1121,12 @@ Status ColumnFamilyData::RangesOverlapWithMemtables(
memtable_iter->Seek(range_start.Encode());
status = memtable_iter->status();
ParsedInternalKey seek_result;
if (status.ok()) {
if (memtable_iter->Valid() &&
ParseInternalKey(memtable_iter->key(), &seek_result) !=
Status::OK()) {
status = Status::Corruption("DB have corrupted keys");
}
if (status.ok() && memtable_iter->Valid()) {
status = ParseInternalKey(memtable_iter->key(), &seek_result,
allow_data_in_errors);
}
if (status.ok()) {
if (memtable_iter->Valid() &&
ucmp->Compare(seek_result.user_key, ranges[i].limit) <= 0) {

@ -406,7 +406,8 @@ class ColumnFamilyData {
//
// Thread-safe
Status RangesOverlapWithMemtables(const autovector<Range>& ranges,
SuperVersion* super_version, bool* overlap);
SuperVersion* super_version,
bool allow_data_in_errors, bool* overlap);
// A flag to tell a manual compaction is to compact all levels together
// instead of a specific level.

@ -158,13 +158,13 @@ void CompactionIterator::Next() {
if (merge_out_iter_.Valid()) {
key_ = merge_out_iter_.key();
value_ = merge_out_iter_.value();
Status s = ParseInternalKey(key_, &ikey_);
Status s = ParseInternalKey(key_, &ikey_, allow_data_in_errors_);
// MergeUntil stops when it encounters a corrupt key and does not
// include them in the result, so we expect the keys here to be valid.
assert(s.ok());
if (!s.ok()) {
ROCKS_LOG_FATAL(info_log_, "Invalid key (%s) in compaction",
key_.ToString(true).c_str());
ROCKS_LOG_FATAL(info_log_, "Invalid key in compaction. %s",
s.getState());
}
// Keep current_key_ in sync.
@ -277,22 +277,14 @@ void CompactionIterator::NextFromInput() {
value_ = input_->value();
iter_stats_.num_input_records++;
Status pikStatus = ParseInternalKey(key_, &ikey_);
if (!pikStatus.ok()) {
Status pik_status = ParseInternalKey(key_, &ikey_, allow_data_in_errors_);
if (!pik_status.ok()) {
iter_stats_.num_input_corrupt_records++;
// If `expect_valid_internal_key_` is false, return the corrupted key
// and let the caller decide what to do with it.
// TODO(noetzli): We should have a more elegant solution for this.
if (expect_valid_internal_key_) {
std::string msg("Corrupted internal key not expected.");
if (allow_data_in_errors_) {
msg.append(" Corrupt key: " + ikey_.user_key.ToString(/*hex=*/true) +
". ");
msg.append("key type: " + std::to_string(ikey_.type) + ".");
msg.append("seq: " + std::to_string(ikey_.sequence) + ".");
}
status_ = Status::Corruption(msg.c_str());
status_ = pik_status;
return;
}
key_ = current_key_.SetInternalKey(key_);
@ -469,7 +461,8 @@ void CompactionIterator::NextFromInput() {
// Check whether the next key exists, is not corrupt, and is the same key
// as the single delete.
if (input_->Valid() &&
ParseInternalKey(input_->key(), &next_ikey) == Status::OK() &&
ParseInternalKey(input_->key(), &next_ikey, allow_data_in_errors_)
.ok() &&
cmp_->Equal(ikey_.user_key, next_ikey.user_key)) {
// Check whether the next key belongs to the same snapshot as the
// SingleDelete.
@ -630,7 +623,8 @@ void CompactionIterator::NextFromInput() {
// than *full_history_ts_low_.
while (!IsPausingManualCompaction() && !IsShuttingDown() &&
input_->Valid() &&
(ParseInternalKey(input_->key(), &next_ikey) == Status::OK()) &&
(ParseInternalKey(input_->key(), &next_ikey, allow_data_in_errors_)
.ok()) &&
0 == cmp_->CompareWithoutTimestamp(ikey_.user_key,
next_ikey.user_key) &&
(prev_snapshot == 0 ||
@ -640,7 +634,8 @@ void CompactionIterator::NextFromInput() {
// If you find you still need to output a row with this key, we need to output the
// delete too
if (input_->Valid() &&
(ParseInternalKey(input_->key(), &next_ikey) == Status::OK()) &&
(ParseInternalKey(input_->key(), &next_ikey, allow_data_in_errors_)
.ok()) &&
0 == cmp_->CompareWithoutTimestamp(ikey_.user_key,
next_ikey.user_key)) {
valid_ = true;
@ -658,8 +653,9 @@ void CompactionIterator::NextFromInput() {
// have hit (A)
// We encapsulate the merge related state machine in a different
// object to minimize change to the existing flow.
Status s = merge_helper_->MergeUntil(input_, range_del_agg_,
prev_snapshot, bottommost_level_);
Status s =
merge_helper_->MergeUntil(input_, range_del_agg_, prev_snapshot,
bottommost_level_, allow_data_in_errors_);
merge_out_iter_.SeekToFirst();
if (!s.ok() && !s.IsMergeInProgress()) {
@ -670,13 +666,13 @@ void CompactionIterator::NextFromInput() {
// These will be correctly set below.
key_ = merge_out_iter_.key();
value_ = merge_out_iter_.value();
pikStatus = ParseInternalKey(key_, &ikey_);
pik_status = ParseInternalKey(key_, &ikey_, allow_data_in_errors_);
// MergeUntil stops when it encounters a corrupt key and does not
// include them in the result, so we expect the keys here to valid.
assert(pikStatus.ok());
if (!pikStatus.ok()) {
ROCKS_LOG_FATAL(info_log_, "Invalid key (%s) in compaction",
key_.ToString(true).c_str());
assert(pik_status.ok());
if (!pik_status.ok()) {
ROCKS_LOG_FATAL(info_log_, "Invalid key in compaction. %s",
pik_status.getState());
}
// Keep current_key_ in sync.
current_key_.UpdateInternalKey(ikey_.sequence, ikey_.type);

@ -271,7 +271,7 @@ class CompactionIteratorTest : public testing::TestWithParam<bool> {
earliest_write_conflict_snapshot, snapshot_checker_.get(),
Env::Default(), false /* report_detailed_time */, false,
range_del_agg_.get(), nullptr /* blob_file_builder */,
false /*allow_data_in_errors*/, std::move(compaction), filter,
true /*allow_data_in_errors*/, std::move(compaction), filter,
&shutting_down_, /*preserve_deletes_seqnum=*/0,
/*manual_compaction_paused=*/nullptr, /*info_log=*/nullptr,
full_history_ts_low));

@ -144,7 +144,8 @@ class CompactionJobTest : public testing::Test {
std::string skey;
std::string value;
std::tie(skey, value) = kv;
const Status pikStatus = ParseInternalKey(skey, &key);
const Status pik_status =
ParseInternalKey(skey, &key, true /* log_err_key */);
smallest_seqno = std::min(smallest_seqno, key.sequence);
largest_seqno = std::max(largest_seqno, key.sequence);
@ -162,7 +163,7 @@ class CompactionJobTest : public testing::Test {
first_key = false;
if (pikStatus.ok() && key.type == kTypeBlobIndex) {
if (pik_status.ok() && key.type == kTypeBlobIndex) {
BlobIndex blob_index;
const Status s = blob_index.DecodeFrom(value);
if (!s.ok()) {

@ -314,7 +314,7 @@ TEST_F(DBTestCompactionFilter, CompactionFilter) {
ASSERT_OK(iter->status());
while (iter->Valid()) {
ParsedInternalKey ikey(Slice(), 0, kTypeValue);
ASSERT_OK(ParseInternalKey(iter->key(), &ikey));
ASSERT_OK(ParseInternalKey(iter->key(), &ikey, true /* log_err_key */));
total++;
if (ikey.sequence != 0) {
count++;
@ -405,7 +405,7 @@ TEST_F(DBTestCompactionFilter, CompactionFilter) {
ASSERT_OK(iter->status());
while (iter->Valid()) {
ParsedInternalKey ikey(Slice(), 0, kTypeValue);
ASSERT_OK(ParseInternalKey(iter->key(), &ikey));
ASSERT_OK(ParseInternalKey(iter->key(), &ikey, true /* log_err_key */));
ASSERT_NE(ikey.sequence, (unsigned)0);
count++;
iter->Next();
@ -624,7 +624,7 @@ TEST_F(DBTestCompactionFilter, CompactionFilterContextManual) {
ASSERT_OK(iter->status());
while (iter->Valid()) {
ParsedInternalKey ikey(Slice(), 0, kTypeValue);
ASSERT_OK(ParseInternalKey(iter->key(), &ikey));
ASSERT_OK(ParseInternalKey(iter->key(), &ikey, true /* log_err_key */));
total++;
if (ikey.sequence != 0) {
count++;

@ -792,7 +792,9 @@ Status DBImpl::CompactRange(const CompactRangeOptions& options,
// changes to RangesOverlapWithMemtables.
Range range(*begin, *end);
SuperVersion* super_version = cfd->GetReferencedSuperVersion(this);
cfd->RangesOverlapWithMemtables({range}, super_version, &flush_needed);
cfd->RangesOverlapWithMemtables({range}, super_version,
immutable_db_options_.allow_data_in_errors,
&flush_needed);
CleanupSuperVersion(super_version);
}

@ -107,11 +107,12 @@ Status DBIter::GetProperty(std::string prop_name, std::string* prop) {
}
bool DBIter::ParseKey(ParsedInternalKey* ikey) {
if (ParseInternalKey(iter_.key(), ikey) != Status::OK()) {
status_ = Status::Corruption("corrupted internal key in DBIter");
Status s =
ParseInternalKey(iter_.key(), ikey, false /* log_err_key */); // TODO
if (!s.ok()) {
status_ = Status::Corruption("In DBIter: ", s.getState());
valid_ = false;
ROCKS_LOG_ERROR(logger_, "corrupted internal key in DBIter: %s",
iter_.key().ToString(true).c_str());
ROCKS_LOG_ERROR(logger_, "In DBIter: %s", status_.getState());
return false;
} else {
return true;

@ -99,10 +99,11 @@ class TestIterator : public InternalIterator {
}
for (auto it = data_.begin(); it != data_.end(); ++it) {
ParsedInternalKey ikey;
Status pikStatus = ParseInternalKey(it->first, &ikey);
pikStatus.PermitUncheckedError();
assert(pikStatus.ok());
if (!pikStatus.ok() || ikey.user_key != _key) {
Status pik_status =
ParseInternalKey(it->first, &ikey, true /* log_err_key */);
pik_status.PermitUncheckedError();
assert(pik_status.ok());
if (!pik_status.ok() || ikey.user_key != _key) {
continue;
}
if (valid_ && data_.begin() + iter_ > it) {

@ -964,7 +964,8 @@ std::string DBTestBase::AllEntriesFor(const Slice& user_key, int cf) {
bool first = true;
while (iter->Valid()) {
ParsedInternalKey ikey(Slice(), 0, kTypeValue);
if (ParseInternalKey(iter->key(), &ikey) != Status::OK()) {
if (ParseInternalKey(iter->key(), &ikey, true /* log_err_key */) !=
Status::OK()) {
result += "CORRUPTED";
} else {
if (!last_options_.comparator->Equal(ikey.user_key, user_key)) {
@ -1371,12 +1372,12 @@ void DBTestBase::validateNumberOfEntries(int numValues, int cf) {
kMaxSequenceNumber));
}
iter->SeekToFirst();
ASSERT_EQ(iter->status().ok(), true);
ASSERT_OK(iter->status());
int seq = numValues;
while (iter->Valid()) {
ParsedInternalKey ikey;
ikey.clear();
ASSERT_OK(ParseInternalKey(iter->key(), &ikey));
ASSERT_OK(ParseInternalKey(iter->key(), &ikey, true /* log_err_key */));
// checks sequence number for updates
ASSERT_EQ(ikey.sequence, (unsigned)seq--);
@ -1581,7 +1582,7 @@ void DBTestBase::VerifyDBInternal(
for (auto p : true_data) {
ASSERT_TRUE(iter->Valid());
ParsedInternalKey ikey;
ASSERT_OK(ParseInternalKey(iter->key(), &ikey));
ASSERT_OK(ParseInternalKey(iter->key(), &ikey, true /* log_err_key */));
ASSERT_EQ(p.first, ikey.user_key);
ASSERT_EQ(p.second, iter->value());
iter->Next();

@ -141,7 +141,8 @@ class DBBasicTestWithTimestampBase : public DBTestBase {
ukey_and_ts.assign(expected_ukey.data(), expected_ukey.size());
ukey_and_ts.append(expected_ts.data(), expected_ts.size());
ParsedInternalKey parsed_ikey;
ASSERT_OK(ParseInternalKey(it->key(), &parsed_ikey));
ASSERT_OK(
ParseInternalKey(it->key(), &parsed_ikey, true /* log_err_key */));
ASSERT_EQ(ukey_and_ts, parsed_ikey.user_key);
ASSERT_EQ(expected_val_type, parsed_ikey.type);
ASSERT_EQ(expected_seq, parsed_ikey.sequence);
@ -161,7 +162,8 @@ class DBBasicTestWithTimestampBase : public DBTestBase {
ukey_and_ts.append(expected_ts.data(), expected_ts.size());
ParsedInternalKey parsed_ikey;
ASSERT_OK(ParseInternalKey(it->key(), &parsed_ikey));
ASSERT_OK(
ParseInternalKey(it->key(), &parsed_ikey, true /* log_err_key */));
ASSERT_EQ(expected_val_type, parsed_ikey.type);
ASSERT_EQ(Slice(ukey_and_ts), parsed_ikey.user_key);
if (expected_val_type == kTypeValue) {

@ -49,7 +49,8 @@ EntryType GetEntryType(ValueType value_type) {
bool ParseFullKey(const Slice& internal_key, FullKey* fkey) {
ParsedInternalKey ikey;
if (ParseInternalKey(internal_key, &ikey) != Status::OK()) {
if (!ParseInternalKey(internal_key, &ikey, false /*log_err_key */)
.ok()) { // TODO
return false;
}
fkey->user_key = ikey.user_key;
@ -77,12 +78,18 @@ void AppendInternalKeyFooter(std::string* result, SequenceNumber s,
PutFixed64(result, PackSequenceAndType(s, t));
}
std::string ParsedInternalKey::DebugString(bool hex) const {
std::string ParsedInternalKey::DebugString(bool log_err_key, bool hex) const {
std::string result = "'";
if (log_err_key) {
result += user_key.ToString(hex);
} else {
result += "<redacted>";
}
char buf[50];
snprintf(buf, sizeof(buf), "' seq:%" PRIu64 ", type:%d", sequence,
static_cast<int>(type));
std::string result = "'";
result += user_key.ToString(hex);
result += buf;
return result;
}
@ -90,8 +97,8 @@ std::string ParsedInternalKey::DebugString(bool hex) const {
std::string InternalKey::DebugString(bool hex) const {
std::string result;
ParsedInternalKey parsed;
if (ParseInternalKey(rep_, &parsed) == Status::OK()) {
result = parsed.DebugString(hex);
if (ParseInternalKey(rep_, &parsed, false /* log_err_key */).ok()) {
result = parsed.DebugString(true /* log_err_key */, hex); // TODO
} else {
result = "(bad)";
result.append(EscapeString(rep_));

@ -112,7 +112,7 @@ struct ParsedInternalKey {
// u contains timestamp if user timestamp feature is enabled.
ParsedInternalKey(const Slice& u, const SequenceNumber& seq, ValueType t)
: user_key(u), sequence(seq), type(t) {}
std::string DebugString(bool hex = false) const;
std::string DebugString(bool log_err_key, bool hex) const;
void clear() {
user_key.clear();
@ -172,7 +172,7 @@ extern void AppendInternalKeyFooter(std::string* result, SequenceNumber s,
//
// On error, returns false, leaves "*result" in an undefined state.
extern Status ParseInternalKey(const Slice& internal_key,
ParsedInternalKey* result);
ParsedInternalKey* result, bool log_err_key);
// Returns the user key portion of an internal key.
inline Slice ExtractUserKey(const Slice& internal_key) {
@ -291,8 +291,8 @@ class InternalKey {
bool Valid() const {
ParsedInternalKey parsed;
return (ParseInternalKey(Slice(rep_), &parsed) == Status::OK()) ? true
: false;
return (ParseInternalKey(Slice(rep_), &parsed, false /* log_err_key */)
.ok()); // TODO
}
void DecodeFrom(const Slice& s) { rep_.assign(s.data(), s.size()); }
@ -325,7 +325,7 @@ class InternalKey {
AppendInternalKeyFooter(&rep_, s, t);
}
std::string DebugString(bool hex = false) const;
std::string DebugString(bool hex) const;
};
inline int InternalKeyComparator::Compare(const InternalKey& a,
@ -334,20 +334,27 @@ inline int InternalKeyComparator::Compare(const InternalKey& a,
}
inline Status ParseInternalKey(const Slice& internal_key,
ParsedInternalKey* result) {
ParsedInternalKey* result, bool log_err_key) {
const size_t n = internal_key.size();
if (n < kNumInternalBytes) {
return Status::Corruption("Internal Key too small");
return Status::Corruption("Corrupted Key: Internal Key too small. Size=" +
std::to_string(n) + ". ");
}
uint64_t num = DecodeFixed64(internal_key.data() + n - kNumInternalBytes);
unsigned char c = num & 0xff;
result->sequence = num >> 8;
result->type = static_cast<ValueType>(c);
assert(result->type <= ValueType::kMaxValue);
result->user_key = Slice(internal_key.data(), n - kNumInternalBytes);
return IsExtendedValueType(result->type)
? Status::OK()
: Status::Corruption("Invalid Key Type");
if (IsExtendedValueType(result->type)) {
return Status::OK();
} else {
return Status::Corruption("Corrupted Key",
result->DebugString(log_err_key, true));
}
}
// Update the sequence number in the internal key.

@ -41,12 +41,12 @@ static void TestKey(const std::string& key,
Slice in(encoded);
ParsedInternalKey decoded("", 0, kTypeValue);
ASSERT_OK(ParseInternalKey(in, &decoded));
ASSERT_OK(ParseInternalKey(in, &decoded, true /* log_err_key */));
ASSERT_EQ(key, decoded.user_key.ToString());
ASSERT_EQ(seq, decoded.sequence);
ASSERT_EQ(vt, decoded.type);
ASSERT_NOK(ParseInternalKey(Slice("bar"), &decoded));
ASSERT_NOK(ParseInternalKey(Slice("bar"), &decoded, true /* log_err_key */));
}
class FormatTest : public testing::Test {};
@ -186,7 +186,7 @@ TEST_F(FormatTest, UpdateInternalKey) {
Slice in(ikey);
ParsedInternalKey decoded;
ASSERT_OK(ParseInternalKey(in, &decoded));
ASSERT_OK(ParseInternalKey(in, &decoded, true /* log_err_key */));
ASSERT_EQ(user_key, decoded.user_key.ToString());
ASSERT_EQ(new_seq, decoded.sequence);
ASSERT_EQ(new_val_type, decoded.type);

@ -313,8 +313,8 @@ Status ExternalSstFileIngestionJob::NeedsFlush(bool* flush_needed,
ranges.emplace_back(file_to_ingest.smallest_internal_key.user_key(),
file_to_ingest.largest_internal_key.user_key());
}
Status status =
cfd_->RangesOverlapWithMemtables(ranges, super_version, flush_needed);
Status status = cfd_->RangesOverlapWithMemtables(
ranges, super_version, db_options_.allow_data_in_errors, flush_needed);
if (status.ok() && *flush_needed &&
!ingestion_options_.allow_blocking_flush) {
status = Status::InvalidArgument("External file requires flush");
@ -599,22 +599,28 @@ Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
file_to_ingest->largest_internal_key =
InternalKey("", 0, ValueType::kTypeValue);
bool bounds_set = false;
bool allow_data_in_errors = db_options_.allow_data_in_errors;
iter->SeekToFirst();
if (iter->Valid()) {
if (ParseInternalKey(iter->key(), &key) != Status::OK()) {
return Status::Corruption("external file have corrupted keys");
Status pik_status =
ParseInternalKey(iter->key(), &key, allow_data_in_errors);
if (!pik_status.ok()) {
return Status::Corruption("Corrupted key in external file. ",
pik_status.getState());
}
if (key.sequence != 0) {
return Status::Corruption("external file have non zero sequence number");
return Status::Corruption("External file has non zero sequence number");
}
file_to_ingest->smallest_internal_key.SetFrom(key);
iter->SeekToLast();
if (ParseInternalKey(iter->key(), &key) != Status::OK()) {
return Status::Corruption("external file have corrupted keys");
pik_status = ParseInternalKey(iter->key(), &key, allow_data_in_errors);
if (!pik_status.ok()) {
return Status::Corruption("Corrupted key in external file. ",
pik_status.getState());
}
if (key.sequence != 0) {
return Status::Corruption("external file have non zero sequence number");
return Status::Corruption("External file has non zero sequence number");
}
file_to_ingest->largest_internal_key.SetFrom(key);
@ -627,8 +633,11 @@ Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
if (range_del_iter != nullptr) {
for (range_del_iter->SeekToFirst(); range_del_iter->Valid();
range_del_iter->Next()) {
if (ParseInternalKey(range_del_iter->key(), &key) != Status::OK()) {
return Status::Corruption("external file have corrupted keys");
Status pik_status =
ParseInternalKey(range_del_iter->key(), &key, allow_data_in_errors);
if (!pik_status.ok()) {
return Status::Corruption("Corrupted key in external file. ",
pik_status.getState());
}
RangeTombstone tombstone(key, range_del_iter->value());

@ -252,15 +252,21 @@ Status ImportColumnFamilyJob::GetIngestedFileInfo(
// Get first (smallest) key from file
iter->SeekToFirst();
if (ParseInternalKey(iter->key(), &key) != Status::OK()) {
return Status::Corruption("external file have corrupted keys");
Status pik_status =
ParseInternalKey(iter->key(), &key, db_options_.allow_data_in_errors);
if (!pik_status.ok()) {
return Status::Corruption("Corrupted Key in external file. ",
pik_status.getState());
}
file_to_import->smallest_internal_key.SetFrom(key);
// Get last (largest) key from file
iter->SeekToLast();
if (ParseInternalKey(iter->key(), &key) != Status::OK()) {
return Status::Corruption("external file have corrupted keys");
pik_status =
ParseInternalKey(iter->key(), &key, db_options_.allow_data_in_errors);
if (!pik_status.ok()) {
return Status::Corruption("Corrupted Key in external file. ",
pik_status.getState());
}
file_to_import->largest_internal_key.SetFrom(key);

@ -116,7 +116,8 @@ Status MergeHelper::TimedFullMerge(const MergeOperator* merge_operator,
Status MergeHelper::MergeUntil(InternalIterator* iter,
CompactionRangeDelAggregator* range_del_agg,
const SequenceNumber stop_before,
const bool at_bottom) {
const bool at_bottom,
const bool allow_data_in_errors) {
// Get a copy of the internal key, before it's invalidated by iter->Next()
// Also maintain the list of merge operands seen.
assert(HasOperator());
@ -139,7 +140,7 @@ Status MergeHelper::MergeUntil(InternalIterator* iter,
// orig_ikey is backed by keys_.back() if !keys_.empty()
ParsedInternalKey orig_ikey;
Status s = ParseInternalKey(original_key, &orig_ikey);
Status s = ParseInternalKey(original_key, &orig_ikey, allow_data_in_errors);
assert(s.ok());
if (!s.ok()) return s;
@ -153,12 +154,12 @@ Status MergeHelper::MergeUntil(InternalIterator* iter,
ParsedInternalKey ikey;
assert(keys_.size() == merge_context_.GetNumOperands());
if (ParseInternalKey(iter->key(), &ikey) != Status::OK()) {
Status pik_status =
ParseInternalKey(iter->key(), &ikey, allow_data_in_errors);
if (!pik_status.ok()) {
// stop at corrupted key
if (assert_valid_internal_key_) {
assert(!"Corrupted internal key not expected.");
s = Status::Corruption("Corrupted internal key not expected.");
return s;
return pik_status;
}
break;
} else if (first_key) {
@ -268,9 +269,10 @@ Status MergeHelper::MergeUntil(InternalIterator* iter,
if (keys_.size() == 1) {
// we need to re-anchor the orig_ikey because it was anchored by
// original_key before
Status pikStatus = ParseInternalKey(keys_.back(), &orig_ikey);
pikStatus.PermitUncheckedError();
assert(pikStatus.ok());
pik_status =
ParseInternalKey(keys_.back(), &orig_ikey, allow_data_in_errors);
pik_status.PermitUncheckedError();
assert(pik_status.ok());
}
if (filter == CompactionFilter::Decision::kKeep) {
merge_context_.PushOperand(

@ -66,6 +66,8 @@ class MergeHelper {
// 0 means no restriction
// at_bottom: (IN) true if the iterator covers the bottem level, which means
// we could reach the start of the history of this user key.
// allow_data_in_errors: (IN) if true, data details will be displayed in
// error/log messages.
//
// Returns one of the following statuses:
// - OK: Entries were successfully merged.
@ -80,7 +82,8 @@ class MergeHelper {
Status MergeUntil(InternalIterator* iter,
CompactionRangeDelAggregator* range_del_agg = nullptr,
const SequenceNumber stop_before = 0,
const bool at_bottom = false);
const bool at_bottom = false,
const bool allow_data_in_errors = false);
// Filters a merge operand using the compaction filter specified
// in the constructor. Returns the decision that the filter made.

@ -33,9 +33,10 @@ TruncatedRangeDelIterator::TruncatedRangeDelIterator(
if (smallest != nullptr) {
pinned_bounds_.emplace_back();
auto& parsed_smallest = pinned_bounds_.back();
Status pikStatus = ParseInternalKey(smallest->Encode(), &parsed_smallest);
pikStatus.PermitUncheckedError();
assert(pikStatus.ok());
Status pik_status = ParseInternalKey(smallest->Encode(), &parsed_smallest,
false /* log_err_key */); // TODO
pik_status.PermitUncheckedError();
assert(pik_status.ok());
smallest_ = &parsed_smallest;
}
@ -43,9 +44,10 @@ TruncatedRangeDelIterator::TruncatedRangeDelIterator(
pinned_bounds_.emplace_back();
auto& parsed_largest = pinned_bounds_.back();
Status pikStatus = ParseInternalKey(largest->Encode(), &parsed_largest);
pikStatus.PermitUncheckedError();
assert(pikStatus.ok());
Status pik_status = ParseInternalKey(largest->Encode(), &parsed_largest,
false /* log_err_key */); // TODO
pik_status.PermitUncheckedError();
assert(pik_status.ok());
if (parsed_largest.type == kTypeRangeDeletion &&
parsed_largest.sequence == kMaxSequenceNumber) {

@ -284,9 +284,10 @@ class RangeDelAggregator {
bool ShouldDelete(const Slice& key, RangeDelPositioningMode mode) {
ParsedInternalKey parsed;
Status pikStatus = ParseInternalKey(key, &parsed);
assert(pikStatus.ok());
if (!pikStatus.ok()) {
Status pik_status =
ParseInternalKey(key, &parsed, false /* log_err_key */); // TODO
assert(pik_status.ok());
if (!pik_status.ok()) {
return false;
}

@ -554,10 +554,12 @@ class Repairer {
ParsedInternalKey parsed;
for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
Slice key = iter->key();
if (ParseInternalKey(key, &parsed) != Status::OK()) {
Status pik_status =
ParseInternalKey(key, &parsed, db_options_.allow_data_in_errors);
if (!pik_status.ok()) {
ROCKS_LOG_ERROR(db_options_.info_log,
"Table #%" PRIu64 ": unparsable key %s",
t->meta.fd.GetNumber(), EscapeString(key).c_str());
"Table #%" PRIu64 ": unparsable key - %s",
t->meta.fd.GetNumber(), pik_status.getState());
continue;
}

@ -33,8 +33,8 @@ Status UserKeyTablePropertiesCollector::InternalAdd(const Slice& key,
const Slice& value,
uint64_t file_size) {
ParsedInternalKey ikey;
Status s = ParseInternalKey(key, &ikey);
if (s != Status::OK()) {
Status s = ParseInternalKey(key, &ikey, false /* log_err_key */); // TODO
if (!s.ok()) {
return s;
}

@ -95,7 +95,8 @@ Status OverlapWithIterator(const Comparator* ucmp,
*overlap = false;
if (iter->Valid()) {
ParsedInternalKey seek_result;
Status s = ParseInternalKey(iter->key(), &seek_result);
Status s = ParseInternalKey(iter->key(), &seek_result,
false /* log_err_key */); // TODO
if (!s.ok()) return s;
if (ucmp->CompareWithoutTimestamp(seek_result.user_key, largest_user_key) <=

@ -62,7 +62,7 @@ static std::string PrintContents(WriteBatch* b) {
for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
ParsedInternalKey ikey;
ikey.clear();
EXPECT_OK(ParseInternalKey(iter->key(), &ikey));
EXPECT_OK(ParseInternalKey(iter->key(), &ikey, true /* log_err_key */));
switch (ikey.type) {
case kTypeValue:
state.append("Put(");

@ -63,10 +63,10 @@ jbyteArray Java_org_rocksdb_WriteBatchTest_getContents(JNIEnv* env,
for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
ROCKSDB_NAMESPACE::ParsedInternalKey ikey;
ikey.clear();
ROCKSDB_NAMESPACE::Status pikStatus =
ROCKSDB_NAMESPACE::ParseInternalKey(iter->key(), &ikey);
pikStatus.PermitUncheckedError();
assert(pikStatus.ok());
ROCKSDB_NAMESPACE::Status pik_status = ROCKSDB_NAMESPACE::ParseInternalKey(
iter->key(), &ikey, true /* log_err_key */);
pik_status.PermitUncheckedError();
assert(pik_status.ok());
switch (ikey.type) {
case ROCKSDB_NAMESPACE::kTypeValue:
state.append("Put(");

@ -2394,8 +2394,10 @@ Status BlockBasedTable::Get(const ReadOptions& read_options, const Slice& key,
// Call the *saver function on each entry/block until it returns false
for (; biter.Valid(); biter.Next()) {
ParsedInternalKey parsed_key;
if (ParseInternalKey(biter.key(), &parsed_key) != Status::OK()) {
s = Status::Corruption(Slice());
Status pik_status = ParseInternalKey(
biter.key(), &parsed_key, false /* log_err_key */); // TODO
if (!pik_status.ok()) {
s = pik_status;
}
if (!get_context->SaveValue(
@ -2718,8 +2720,10 @@ void BlockBasedTable::MultiGet(const ReadOptions& read_options,
ParsedInternalKey parsed_key;
Cleanable dummy;
Cleanable* value_pinner = nullptr;
if (ParseInternalKey(biter->key(), &parsed_key) != Status::OK()) {
s = Status::Corruption(Slice());
Status pik_status = ParseInternalKey(
biter->key(), &parsed_key, false /* log_err_key */); // TODO
if (!pik_status.ok()) {
s = pik_status;
}
if (biter->IsValuePinned()) {
if (reusing_block) {

@ -90,8 +90,11 @@ void CuckooTableBuilder::Add(const Slice& key, const Slice& value) {
return;
}
ParsedInternalKey ikey;
if (ParseInternalKey(key, &ikey) != Status::OK()) {
status_ = Status::Corruption("Unable to parse key into inernal key.");
Status pik_status =
ParseInternalKey(key, &ikey, false /* log_err_key */); // TODO
if (!pik_status.ok()) {
status_ = Status::Corruption("Unable to parse key into internal key. ",
pik_status.getState());
return;
}
if (ikey.type != kTypeDeletion && ikey.type != kTypeValue) {

@ -47,8 +47,9 @@ class CuckooBuilderTest : public testing::Test {
uint64_t num_deletions = 0;
for (const auto& key : keys) {
ParsedInternalKey parsed;
if (ParseInternalKey(key, &parsed) == Status::OK() &&
parsed.type == kTypeDeletion) {
Status pik_status =
ParseInternalKey(key, &parsed, true /* log_err_key */);
if (pik_status.ok() && parsed.type == kTypeDeletion) {
num_deletions++;
}
}

@ -172,7 +172,8 @@ Status CuckooTableReader::Get(const ReadOptions& /*readOptions*/,
} else {
Slice full_key(bucket, key_length_);
ParsedInternalKey found_ikey;
Status s = ParseInternalKey(full_key, &found_ikey);
Status s = ParseInternalKey(full_key, &found_ikey,
false /* log_err_key */); // TODO
if (!s.ok()) return s;
bool dont_care __attribute__((__unused__));
get_context->SaveValue(found_ikey, value, &dont_care);

@ -207,8 +207,10 @@ Status MockTableReader::Get(const ReadOptions&, const Slice& key,
std::unique_ptr<MockTableIterator> iter(new MockTableIterator(table_));
for (iter->Seek(key); iter->Valid(); iter->Next()) {
ParsedInternalKey parsed_key;
if (ParseInternalKey(iter->key(), &parsed_key) != Status::OK()) {
return Status::Corruption(Slice());
Status pik_status =
ParseInternalKey(iter->key(), &parsed_key, true /* log_err_key */);
if (!pik_status.ok()) {
return pik_status;
}
bool dont_care __attribute__((__unused__));
@ -303,8 +305,9 @@ void MockTableFactory::AssertLatestFile(const KVVector& file_contents) {
ParsedInternalKey ikey;
std::string key, value;
std::tie(key, value) = kv;
ASSERT_OK(ParseInternalKey(Slice(key), &ikey));
std::cout << ikey.DebugString(false) << " -> " << value << std::endl;
ASSERT_OK(ParseInternalKey(Slice(key), &ikey, true /* log_err_key */));
std::cout << ikey.DebugString(true, false) << " -> " << value
<< std::endl;
}
FAIL();
}

@ -132,7 +132,8 @@ void PlainTableBuilder::Add(const Slice& key, const Slice& value) {
size_t meta_bytes_buf_size = 0;
ParsedInternalKey internal_key;
if (ParseInternalKey(key, &internal_key) != Status::OK()) {
if (!ParseInternalKey(key, &internal_key, false /* log_err_key */)
.ok()) { // TODO
assert(false);
return;
}

@ -85,8 +85,10 @@ IOStatus PlainTableKeyEncoder::AppendKey(const Slice& key,
uint64_t* offset, char* meta_bytes_buf,
size_t* meta_bytes_buf_size) {
ParsedInternalKey parsed_key;
if (ParseInternalKey(key, &parsed_key) != Status::OK()) {
return IOStatus::Corruption(Slice());
Status pik_status =
ParseInternalKey(key, &parsed_key, false /* log_err_key */); // TODO
if (!pik_status.ok()) {
return IOStatus::Corruption(pik_status.getState());
}
Slice key_to_write = key; // Portion of internal key to write out.
@ -279,9 +281,12 @@ Status PlainTableKeyDecoder::ReadInternalKey(
return file_reader_.status();
}
*internal_key_valid = true;
if (ParseInternalKey(*internal_key, parsed_key) != Status::OK()) {
Status pik_status = ParseInternalKey(*internal_key, parsed_key,
false /* log_err_key */); // TODO
if (!pik_status.ok()) {
return Status::Corruption(
Slice("Incorrect value type found when reading the next key"));
Slice("Corrupted key found during next key read. "),
pik_status.getState());
}
*bytes_read += user_key_size + 8;
}

@ -457,7 +457,8 @@ Status PlainTableReader::GetOffset(PlainTableKeyDecoder* decoder,
uint32_t high = upper_bound;
ParsedInternalKey mid_key;
ParsedInternalKey parsed_target;
Status s = ParseInternalKey(target, &parsed_target);
Status s = ParseInternalKey(target, &parsed_target,
false /* log_err_key */); // TODO
if (!s.ok()) return s;
// The key is between [low, high). Do a binary search between it.
@ -593,8 +594,9 @@ Status PlainTableReader::Get(const ReadOptions& /*ro*/, const Slice& target,
}
ParsedInternalKey found_key;
ParsedInternalKey parsed_target;
s = ParseInternalKey(target, &parsed_target);
if (!s.ok()) return Status::Corruption(Slice());
s = ParseInternalKey(target, &parsed_target,
false /* log_err_key */); // TODO
if (!s.ok()) return s;
Slice found_value;
while (offset < file_info_.data_end_offset) {

@ -441,9 +441,9 @@ Status SstFileDumper::ReadSequential(bool print_kv, uint64_t read_num,
if (read_num > 0 && i > read_num) break;
ParsedInternalKey ikey;
if (ParseInternalKey(key, &ikey) != Status::OK()) {
std::cerr << "Internal Key [" << key.ToString(true /* in hex*/)
<< "] parse error!\n";
Status pik_status = ParseInternalKey(key, &ikey, true /* log_err_key */);
if (!pik_status.ok()) {
std::cerr << pik_status.getState() << "\n";
continue;
}
@ -459,7 +459,8 @@ Status SstFileDumper::ReadSequential(bool print_kv, uint64_t read_num,
if (print_kv) {
if (!decode_blob_index_ || ikey.type != kTypeBlobIndex) {
fprintf(stdout, "%s => %s\n", ikey.DebugString(output_hex_).c_str(),
fprintf(stdout, "%s => %s\n",
ikey.DebugString(true, output_hex_).c_str(),
value.ToString(output_hex_).c_str());
} else {
BlobIndex blob_index;
@ -467,11 +468,12 @@ Status SstFileDumper::ReadSequential(bool print_kv, uint64_t read_num,
const Status s = blob_index.DecodeFrom(value);
if (!s.ok()) {
fprintf(stderr, "%s => error decoding blob index\n",
ikey.DebugString(output_hex_).c_str());
ikey.DebugString(true, output_hex_).c_str());
continue;
}
fprintf(stdout, "%s => %s\n", ikey.DebugString(output_hex_).c_str(),
fprintf(stdout, "%s => %s\n",
ikey.DebugString(true, output_hex_).c_str(),
blob_index.DebugString(output_hex_).c_str());
}
}

@ -252,9 +252,11 @@ class KeyConvertingIterator : public InternalIterator {
Slice key() const override {
assert(Valid());
ParsedInternalKey parsed_key;
if (ParseInternalKey(iter_->key(), &parsed_key) != Status::OK()) {
status_ = Status::Corruption("malformed internal key");
return Slice("corrupted key");
Status pik_status =
ParseInternalKey(iter_->key(), &parsed_key, true /* log_err_key */);
if (!pik_status.ok()) {
status_ = pik_status;
return Slice(status_.getState());
}
return parsed_key.user_key;
}
@ -1571,7 +1573,8 @@ TEST_P(BlockBasedTableTest, RangeDelBlock) {
for (size_t i = 0; i < expected_tombstones.size(); i++) {
ASSERT_TRUE(iter->Valid());
ParsedInternalKey parsed_key;
ASSERT_OK(ParseInternalKey(iter->key(), &parsed_key));
ASSERT_OK(
ParseInternalKey(iter->key(), &parsed_key, true /* log_err_key */));
RangeTombstone t(parsed_key, iter->value());
const auto& expected_t = expected_tombstones[i];
ASSERT_EQ(t.start_key_, expected_t.start_key_);
@ -4138,7 +4141,7 @@ TEST_P(BlockBasedTableTest, DISABLED_TableWithGlobalSeqno) {
char current_c = 'a';
for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
ParsedInternalKey pik;
ASSERT_OK(ParseInternalKey(iter->key(), &pik));
ASSERT_OK(ParseInternalKey(iter->key(), &pik, true /* log_err_key */));
ASSERT_EQ(pik.type, ValueType::kTypeValue);
ASSERT_EQ(pik.sequence, 0);
@ -4159,7 +4162,7 @@ TEST_P(BlockBasedTableTest, DISABLED_TableWithGlobalSeqno) {
current_c = 'a';
for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
ParsedInternalKey pik;
ASSERT_OK(ParseInternalKey(iter->key(), &pik));
ASSERT_OK(ParseInternalKey(iter->key(), &pik, true /* log_err_key */));
ASSERT_EQ(pik.type, ValueType::kTypeValue);
ASSERT_EQ(pik.sequence, 10);
@ -4177,7 +4180,7 @@ TEST_P(BlockBasedTableTest, DISABLED_TableWithGlobalSeqno) {
ASSERT_TRUE(iter->Valid());
ParsedInternalKey pik;
ASSERT_OK(ParseInternalKey(iter->key(), &pik));
ASSERT_OK(ParseInternalKey(iter->key(), &pik, true /* log_err_key */));
ASSERT_EQ(pik.type, ValueType::kTypeValue);
ASSERT_EQ(pik.sequence, 10);
@ -4196,7 +4199,7 @@ TEST_P(BlockBasedTableTest, DISABLED_TableWithGlobalSeqno) {
current_c = 'a';
for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
ParsedInternalKey pik;
ASSERT_OK(ParseInternalKey(iter->key(), &pik));
ASSERT_OK(ParseInternalKey(iter->key(), &pik, true /* log_err_key */));
ASSERT_EQ(pik.type, ValueType::kTypeValue);
ASSERT_EQ(pik.sequence, 3);
@ -4215,7 +4218,7 @@ TEST_P(BlockBasedTableTest, DISABLED_TableWithGlobalSeqno) {
ASSERT_TRUE(iter->Valid());
ParsedInternalKey pik;
ASSERT_OK(ParseInternalKey(iter->key(), &pik));
ASSERT_OK(ParseInternalKey(iter->key(), &pik, true /* log_err_key */));
ASSERT_EQ(pik.type, ValueType::kTypeValue);
ASSERT_EQ(pik.sequence, 3);

@ -103,7 +103,8 @@ struct BlockAccessInfo {
num_referenced_key_exist_in_block++;
if (referenced_data_size > block_size && block_size != 0) {
ParsedInternalKey internal_key;
Status s = ParseInternalKey(access.referenced_key, &internal_key);
Status s = ParseInternalKey(access.referenced_key, &internal_key,
false /* log_err_key */); // TODO
assert(s.ok()); // TODO
}
} else {

@ -238,12 +238,13 @@ int SSTDumpTool::Run(int argc, char const* const* argv, Options options) {
Slice sl_key = ROCKSDB_NAMESPACE::Slice(in_key);
ParsedInternalKey ikey;
int retc = 0;
if (ParseInternalKey(sl_key, &ikey) != Status::OK()) {
std::cerr << "Internal Key [" << sl_key.ToString(true /* in hex*/)
<< "] parse error!\n";
Status pik_status =
ParseInternalKey(sl_key, &ikey, true /* log_err_key */);
if (!pik_status.ok()) {
std::cerr << pik_status.getState() << "\n";
retc = -1;
}
fprintf(stdout, "key=%s\n", ikey.DebugString(true).c_str());
fprintf(stdout, "key=%s\n", ikey.DebugString(true, true).c_str());
return retc;
} else if (ParseIntArg(argv[i], "--compression_level_from=",
"compression_level_from must be numeric",

@ -66,7 +66,10 @@ CompactionFilter::Decision BlobIndexCompactionFilterBase::FilterV2(
// Hack: Internal key is passed to BlobIndexCompactionFilter for it to
// get sequence number.
ParsedInternalKey ikey;
if (ParseInternalKey(key, &ikey) != Status::OK()) {
if (!ParseInternalKey(
key, &ikey,
context_.blob_db_impl->db_options_.allow_data_in_errors)
.ok()) {
assert(false);
return Decision::kKeep;
}
@ -83,7 +86,10 @@ CompactionFilter::Decision BlobIndexCompactionFilterBase::FilterV2(
// Hack: Internal key is passed to BlobIndexCompactionFilter for it to
// get sequence number.
ParsedInternalKey ikey;
if (ParseInternalKey(key, &ikey) != Status::OK()) {
if (!ParseInternalKey(
key, &ikey,
context_.blob_db_impl->db_options_.allow_data_in_errors)
.ok()) {
assert(false);
return Decision::kKeep;
}

@ -55,9 +55,10 @@ Status GetAllKeyVersions(DB* db, ColumnFamilyHandle* cfh, Slice begin_key,
size_t num_keys = 0;
for (; iter->Valid(); iter->Next()) {
ParsedInternalKey ikey;
if (ParseInternalKey(iter->key(), &ikey) != Status::OK()) {
return Status::Corruption("Internal Key [" + iter->key().ToString() +
"] parse error!");
Status pik_status =
ParseInternalKey(iter->key(), &ikey, true /* log_err_key */); // TODO
if (!pik_status.ok()) {
return pik_status;
}
if (!end_key.empty() &&

Loading…
Cancel
Save