perf: improve the performance of ValueRowDeserializer

[fuyufjh]

The current implementation of ValueRowDeserializer is not efficient, which didn't fully utilize the performance of our row encoding format.

risingwave/src/common/src/util/value_encoding/column_aware_row_encoding.rs

Lines 205 to 259 in d1b4612

	fn deserialize(&self, mut encoded_bytes: &[u8]) -> Result<Vec<Datum>> {
	let flag = Flag::from_bits(encoded_bytes.get_u8()).expect("should be a valid flag");
	let offset_bytes = match flag - Flag::EMPTY {
	Flag::OFFSET8 => 1,
	Flag::OFFSET16 => 2,
	Flag::OFFSET32 => 4,
	_ => return Err(ValueEncodingError::InvalidFlag(flag.bits())),
	};
	let datum_num = encoded_bytes.get_u32_le() as usize;
	let offsets_start_idx = 4 * datum_num;
	let data_start_idx = offsets_start_idx + datum_num * offset_bytes;
	let offsets = &encoded_bytes[offsets_start_idx..data_start_idx];
	let data = &encoded_bytes[data_start_idx..];
	let mut datums: Vec<Option<Datum>> = vec![None; self.schema.len()];
	let mut contained_indices = BTreeSet::new();
	for i in 0..datum_num {
	let this_id = encoded_bytes.get_i32_le();
	if let Some(&decoded_idx) = self.needed_column_ids.get(&this_id) {
	contained_indices.insert(decoded_idx);
	let this_offset_start_idx = i * offset_bytes;
	let mut this_offset_slice =
	&offsets[this_offset_start_idx..(this_offset_start_idx + offset_bytes)];
	let this_offset = deserialize_width(offset_bytes, &mut this_offset_slice);
	let data = if i + 1 < datum_num {
	let mut next_offset_slice = &offsets[(this_offset_start_idx + offset_bytes)
	..(this_offset_start_idx + 2 * offset_bytes)];
	let next_offset = deserialize_width(offset_bytes, &mut next_offset_slice);
	if this_offset == next_offset {
	None
	} else {
	let mut data_slice = &data[this_offset..next_offset];
	Some(deserialize_value(
	&self.schema[decoded_idx],
	&mut data_slice,
	)?)
	}
	} else if this_offset == data.len() {
	None
	} else {
	let mut data_slice = &data[this_offset..];
	Some(deserialize_value(
	&self.schema[decoded_idx],
	&mut data_slice,
	)?)
	};
	datums[decoded_idx] = Some(data);
	}
	}
	for (id, datum) in &self.default_column_values {
	if !contained_indices.contains(id) {
	datums[*id].get_or_insert(datum.clone());
	}
	}
	Ok(datums.into_iter().map(|d| d.unwrap_or(None)).collect())
	}

As shown in the figure below, the main time consumption is due to container operations.

This encoding is rather friendly for random access, isn't it? The sorted column_id is obviously intended to facilitate binary search.

https://github.com/risingwavelabs/rfcs/blob/75091f0c7f197f718b8343eb121932df5530ddb1/rfcs/0090-table-schema-change.md#column-aware-row-encoding

[BugenZhao]

Would you please share some backgrounds? For example, in which case does this become the bottleneck of the query or the job?

[zwang28]

Would you please share some backgrounds? For example, in which case does this become the bottleneck of the query or the job?

My case is count(*) a table with 100 int columns, 10 million rows.

The bottleneck is the StorageTable that calls ValueRowDeserializer always explicitly deserializes the full row, which can be optimized to deserialize needed columns only (maybe by leveraging the needed_column_ids).

I'll post a comparision after optimizing it later.
With this minor change, the query latency drops from 60sec to 15sec.