About Partition reshuffle function

[JasonLi-cn]

Why does the reshuffle function not use the consistent Hash algorithm to ensure cache stability?

databend/src/query/catalog/src/plan/partition.rs

Lines 115 to 188 in ea5eaf0

	pub fn reshuffle(&self, executors: Vec<String>) -> Result<HashMap<String, Partitions>> {
	let mut executors_sorted = executors;
	executors_sorted.sort();
	let num_executors = executors_sorted.len();
	let partitions = match self.kind {
	PartitionsShuffleKind::Seq => self.partitions.clone(),
	PartitionsShuffleKind::Mod => {
	// Sort by hash%executor_nums.
	let mut parts = self
	.partitions
	.iter()
	.map(|p| (p.hash() % num_executors as u64, p.clone()))
	.collect::<Vec<_>>();
	parts.sort_by(|a, b| a.0.cmp(&b.0));
	parts.into_iter().map(|x| x.1).collect()
	}
	PartitionsShuffleKind::Rand => {
	let mut rng = thread_rng();
	let mut parts = self.partitions.clone();
	parts.shuffle(&mut rng);
	parts
	}
	PartitionsShuffleKind::Broadcast => {
	let mut executor_part = HashMap::default();
	for executor in executors_sorted.iter() {
	executor_part.insert(
	executor.clone(),
	Partitions::create(
	PartitionsShuffleKind::Seq,
	self.partitions.clone(),
	self.is_lazy,
	),
	);
	}
	return Ok(executor_part);
	}
	};
	// parts_per_executor = num_parts / num_executors
	// remain = num_parts % num_executors
	// part distribution:
	// executor number | Part number of each executor
	// ------------------------------------------------------
	// num_executors - remain | parts_per_executor
	// remain | parts_per_executor + 1
	let num_parts = partitions.len();
	let mut executor_part = HashMap::default();
	for (idx, executor) in executors_sorted.into_iter().enumerate() {
	let begin = num_parts * idx / num_executors;
	let end = num_parts * (idx + 1) / num_executors;
	let parts = if begin == end {
	// reach here only when num_executors > num_parts
	vec![]
	} else {
	partitions[begin..end].to_vec()
	};
	executor_part.insert(
	executor,
	Partitions::create(PartitionsShuffleKind::Seq, parts, self.is_lazy),
	);
	}
	Ok(executor_part)
	}
	pub fn compute_sha256(&self) -> Result<String> {
	let buf = serde_json::to_vec(&self.partitions)?;
	let sha = sha2::Sha256::digest(buf);
	Ok(format!("{:x}", sha))
	}
	}

[sundy-li]

That's right, it could be an improvement.