Repository: risingwavelabs/risingwave
File: src/storage/src/hummock/sstable/block.rs (Rust)

Code before:
            }
        }
        min_len
    }

    let prev_table_key = prev_block_last.user_key.table_key.as_ref();
    let next_table_key = block_smallest.user_key.table_key.as_ref();

    let lcp = lcp_len(prev_table_key, next_table_key);

    // Shortened key = LCP prefix + first differing byte.
    // Only shorten if the result is strictly shorter than next_table_key.
    let shortened_len = lcp + 1;
    if shortened_len >= next_table_key.len() {
        return None;
    }

    // Build candidate key: take LCP + 1 bytes from block_smallest
    let cand = FullKey::new_with_gap_epoch(
        block_smallest.user_key.table_id,
        TableKey(&next_table_key[..shortened_len]),
        block_smallest.epoch_with_gap,
    );

    // Verify invariant: prev_block_last < cand <= block_smallest
    if prev_block_last.cmp(&cand).is_ge() || cand.cmp(block_smallest).is_gt() {
        return None;
    }

    Some(cand.copy_into())
}

#[cfg(test)]
mod tests {

    use risingwave_common::util::epoch::test_epoch;
    use risingwave_hummock_sdk::key::MAX_KEY_LEN;

    use super::*;
    use crate::hummock::{BlockHolder, BlockIterator};

    #[test]
    fn test_block_enc_dec() {
        let options = BlockBuilderOptions::default();
        let mut builder = BlockBuilder::new(options);
        builder.add_for_test(construct_full_key_struct_for_test(0, b"k1", 1), b"v01");
        builder.add_for_test(construct_full_key_struct_for_test(0, b"k2", 2), b"v02");
        builder.add_for_test(construct_full_key_struct_for_test(0, b"k3", 3), b"v03");
        builder.add_for_test(construct_full_key_struct_for_test(0, b"k4", 4), b"v04");
        let capacity = builder.uncompressed_block_size();
        assert_eq!(capacity, builder.approximate_len() - 9);

Review comment (bug):
Shouldn't this be an assertion? If we hit the condition here, that means the algorithm is wrongly implemented.

Code after:
/// Attempts to shorten `block_smallest` key while preserving block boundary correctness.
///
/// Returns a shortened key if `prev_block_last < shortened <= block_smallest` can be satisfied
/// with fewer bytes. This reduces block metadata size by exploiting common prefixes between
/// adjacent blocks.
///
/// Returns `None` if the key cannot be shortened (e.g., keys differ only in the last byte).
pub fn try_shorten_block_smallest_key(
    prev_block_last: &FullKey<&[u8]>,
    block_smallest: &FullKey<&[u8]>,
) -> Option<FullKey<Vec<u8>>> {
    /// Returns the length of the longest common prefix between two byte slices.
    fn lcp_len(a: &[u8], b: &[u8]) -> usize {
        let min_len = a.len().min(b.len());
        for i in 0..min_len {
            if a[i] != b[i] {
                return i;
            }
        }
        min_len
    }

    let prev_table_key = prev_block_last.user_key.table_key.as_ref();
    let next_table_key = block_smallest.user_key.table_key.as_ref();
    assert_eq!(
        prev_block_last.user_key.table_id, block_smallest.user_key.table_id,
        "table ids must match for shortening block smallest key"
    );

    let lcp = lcp_len(prev_table_key, next_table_key);

    // Shortened key = LCP prefix + first differing byte.
    // Only shorten if the result is strictly shorter than next_table_key.
    let shortened_len = lcp + 1;
    if shortened_len >= next_table_key.len() {
        return None;
    }

    // Build candidate key: take LCP + 1 bytes from block_smallest
    let cand = FullKey::new_with_gap_epoch(
        block_smallest.user_key.table_id,
        TableKey(&next_table_key[..shortened_len]),
        block_smallest.epoch_with_gap,
    );

    // Invariant: prev_block_last < cand <= block_smallest
    // This must hold by construction. If violated, it indicates a bug in the shortening logic.
    // Use assert! (not debug_assert!) to fail fast and prevent writing incorrect keys to storage.
    assert!(
        prev_block_last.cmp(&cand).is_lt(),
        "Invariant violated: prev_block_last >= cand. prev={:?}, cand={:?}",