@ -74,10 +74,12 @@ void PlainTableBuilder::Add(const Slice& key, const Slice& value) {
if ( ! IsFixedLength ( ) ) {
if ( ! IsFixedLength ( ) ) {
// Write key length
// Write key length
key_size_str_ . clear ( ) ;
char key_size_buf [ 5 ] ; // tmp buffer for key size as varint32
PutVarint32 ( & key_size_str_ , user_key_size ) ;
char * ptr = EncodeVarint32 ( key_size_buf , user_key_size ) ;
file_ - > Append ( key_size_str_ ) ;
assert ( ptr < = key_size_buf + sizeof ( key_size_buf ) ) ;
offset_ + = key_size_str_ . length ( ) ;
auto len = ptr - key_size_buf ;
file_ - > Append ( Slice ( key_size_buf , len ) ) ;
offset_ + = len ;
}
}
// Write key
// Write key
@ -86,25 +88,32 @@ void PlainTableBuilder::Add(const Slice& key, const Slice& value) {
status_ = Status : : Corruption ( Slice ( ) ) ;
status_ = Status : : Corruption ( Slice ( ) ) ;
return ;
return ;
}
}
// For value size as varint32 (up to 5 bytes).
// If the row is of value type with seqId 0, flush the special flag together
// in this buffer to safe one file append call, which takes 1 byte.
char value_size_buf [ 6 ] ;
size_t value_size_buf_size = 0 ;
if ( parsed_key . sequence = = 0 & & parsed_key . type = = kTypeValue ) {
if ( parsed_key . sequence = = 0 & & parsed_key . type = = kTypeValue ) {
file_ - > Append ( Slice ( key . data ( ) , user_key_size ) ) ;
file_ - > Append ( Slice ( key . data ( ) , user_key_size ) ) ;
char tmp_char = PlainTableFactory : : kValueTypeSeqId0 ;
offset_ + = user_key_size ;
file_ - > Append ( Slice ( & tmp_char , 1 ) ) ;
value_size_buf [ 0 ] = PlainTableFactory : : kValueTypeSeqId0 ;
offset_ + = key . size ( ) - 7 ;
value_size_buf_size = 1 ;
} else {
} else {
file_ - > Append ( key ) ;
file_ - > Append ( key ) ;
offset_ + = key . size ( ) ;
offset_ + = key . size ( ) ;
}
}
// Write value length
// Write value length
value_size_str_ . clear ( ) ;
int value_size = value . size ( ) ;
int value_size = value . size ( ) ;
PutVarint32 ( & value_size_str_ , value_size ) ;
char * end_ptr =
file_ - > Append ( value_size_str_ ) ;
EncodeVarint32 ( value_size_buf + value_size_buf_size , value_size ) ;
assert ( end_ptr < = value_size_buf + sizeof ( value_size_buf ) ) ;
value_size_buf_size = end_ptr - value_size_buf ;
file_ - > Append ( Slice ( value_size_buf , value_size_buf_size ) ) ;
// Write value
// Write value
file_ - > Append ( value ) ;
file_ - > Append ( value ) ;
offset_ + = value_size + value_size_str_ . length ( ) ;
offset_ + = value_size + value_size_buf_size ;
properties_ . num_entries + + ;
properties_ . num_entries + + ;
properties_ . raw_key_size + = key . size ( ) ;
properties_ . raw_key_size + = key . size ( ) ;