refactor(state table): use memcpmarble's max value to encoding instead hack with NULL byte+1

[st1page]

risingwave/src/stream/src/common/table/state_table.rs

Lines 1033 to 1065 in 615f873

	if prefix_serializer.as_ref().unwrap().get_order_types().first().unwrap().is_ascending() {
	// We either serialize null into `0u8`, data into `(1u8 || scalar)`, or serialize null
	// into `1u8`, data into `(0u8 || scalar)`. We do not want to delete null
	// here, so `range_begin_suffix` cannot be `vec![]` when null is represented as `0u8`.
	let range_begin_suffix = vec![*first_byte];
	for vnode in self.vnodes.iter_vnodes() {
	let mut range_begin = vnode.to_be_bytes().to_vec();
	let mut range_end = range_begin.clone();
	range_begin.extend(&range_begin_suffix);
	range_end.extend(&watermark_suffix);
	delete_ranges.push((
	Bound::Included(Bytes::from(range_begin)),
	Bound::Excluded(Bytes::from(range_end)),
	));
	}
	} else {
	assert_ne!(*first_byte, u8::MAX);
	let following_bytes = next_key(&watermark_suffix[1..]);
	if !following_bytes.is_empty() {
	for vnode in self.vnodes.iter_vnodes() {
	let mut range_begin = vnode.to_be_bytes().to_vec();
	let mut range_end = range_begin.clone();
	range_begin.push(*first_byte);
	range_begin.extend(&following_bytes);
	range_end.push(first_byte + 1);
	delete_ranges.push((
	Bound::Included(Bytes::from(range_begin)),
	Bound::Excluded(Bytes::from(range_end)),
	));
	}
	}
	}
	}

[fuyufjh]

?

[st1page]

close after #14526