leekeiabstraction commented on code in PR #208: URL: https://github.com/apache/fluss-rust/pull/208#discussion_r2724469362
########## crates/fluss/src/metadata/partition.rs: ########## @@ -0,0 +1,469 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use crate::PartitionId; +use crate::error::{Error, Result}; +use crate::proto::{PbKeyValue, PbPartitionInfo, PbPartitionSpec}; +use std::collections::HashMap; +use std::fmt::{Display, Formatter}; + +/// Represents a partition spec in fluss. Partition columns and values are NOT of strict order, and +/// they need to be re-arranged to the correct order by comparing with a list of strictly ordered +/// partition keys. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PartitionSpec { + partition_spec: HashMap<String, String>, +} + +impl PartitionSpec { + pub fn new(partition_spec: HashMap<String, String>) -> Self { + Self { partition_spec } + } + + pub fn get_spec_map(&self) -> &HashMap<String, String> { + &self.partition_spec + } + + pub fn to_pb(&self) -> PbPartitionSpec { + PbPartitionSpec { + partition_key_values: self + .partition_spec + .iter() + .map(|(k, v)| PbKeyValue { + key: k.clone(), + value: v.clone(), + }) + .collect(), + } + } + + pub fn from_pb(pb: &PbPartitionSpec) -> Self { + let partition_spec = pb + .partition_key_values + .iter() + .map(|kv| (kv.key.clone(), kv.value.clone())) + .collect(); + Self { partition_spec } + } +} + +impl Display for PartitionSpec { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "PartitionSpec{{{:?}}}", self.partition_spec) + } +} + +/// Represents a partition, which is the resolved version of PartitionSpec. The partition +/// spec is re-arranged into the correct order by comparing it with a list of strictly ordered +/// partition keys. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct ResolvedPartitionSpec { + partition_keys: Vec<String>, + partition_values: Vec<String>, +} + +pub const PARTITION_SPEC_SEPARATOR: &str = "$"; + +impl ResolvedPartitionSpec { + pub fn new(partition_keys: Vec<String>, partition_values: Vec<String>) -> Result<Self> { + if partition_keys.len() != partition_values.len() { + return Err(Error::IllegalArgument { + message: "The number of partition keys and partition values should be the same." + .to_string(), + }); + } + Ok(Self { + partition_keys, + partition_values, + }) + } + + pub fn from_partition_spec( + partition_keys: Vec<String>, + partition_spec: &PartitionSpec, + ) -> Self { + let partition_values = + Self::get_reordered_partition_values(&partition_keys, partition_spec); + Self { + partition_keys, + partition_values, + } + } + + pub fn from_partition_value(partition_key: String, partition_value: String) -> Self { + Self { + partition_keys: vec![partition_key], + partition_values: vec![partition_value], + } + } + + pub fn from_partition_name(partition_keys: Vec<String>, partition_name: &str) -> Self { + let partition_values: Vec<String> = partition_name + .split(PARTITION_SPEC_SEPARATOR) + .map(|s| s.to_string()) + .collect(); + Self { + partition_keys, + partition_values, + } + } + + pub fn from_partition_qualified_name(qualified_partition_name: &str) -> Result<Self> { + let mut keys = Vec::new(); + let mut values = Vec::new(); + + for pair in qualified_partition_name.split('/') { + let parts: Vec<&str> = pair.splitn(2, '=').collect(); + if parts.len() != 2 { + return Err(Error::IllegalArgument { + message: format!( + "Invalid partition name format. Expected key=value, got: {}", + pair + ), + }); + } + keys.push(parts[0].to_string()); + values.push(parts[1].to_string()); + } + + Ok(Self { + partition_keys: keys, + partition_values: values, + }) + } + + pub fn get_partition_keys(&self) -> &[String] { + &self.partition_keys + } + + pub fn get_partition_values(&self) -> &[String] { + &self.partition_values + } + + pub fn to_partition_spec(&self) -> PartitionSpec { + let mut spec_map = HashMap::new(); + for (i, key) in self.partition_keys.iter().enumerate() { + spec_map.insert(key.clone(), self.partition_values[i].clone()); + } + PartitionSpec::new(spec_map) + } + + /// Generate the partition name for a partition table with specified partition values. + /// + /// The partition name is in the following format: value1$value2$...$valueN + pub fn get_partition_name(&self) -> String { + self.partition_values.join(PARTITION_SPEC_SEPARATOR) + } + + /// Returns the qualified partition name for a partition spec. + /// The format is: key1=value1/key2=value2/.../keyN=valueN + pub fn get_partition_qualified_name(&self) -> String { + let mut sb = String::new(); + for (i, key) in self.partition_keys.iter().enumerate() { + sb.push_str(key); + sb.push('='); + sb.push_str(&self.partition_values[i]); + if i != self.partition_keys.len() - 1 { + sb.push('/'); + } + } + sb + } + + pub fn contains(&self, other: &ResolvedPartitionSpec) -> Result<bool> { + let other_partition_keys = other.get_partition_keys(); + let other_partition_values = other.get_partition_values(); + + let mut expected_partition_values = Vec::new(); + for other_partition_key in other_partition_keys { + let key_index = self + .partition_keys + .iter() + .position(|k| k == other_partition_key); + match key_index { + Some(idx) => expected_partition_values.push(self.partition_values[idx].clone()), + None => { + return Err(Error::IllegalArgument { + message: format!( + "table does not contain partitionKey: {}", + other_partition_key + ), + }); + } + } + } + + let expected_partition_name = expected_partition_values.join(PARTITION_SPEC_SEPARATOR); + let other_partition_name = other_partition_values.join(PARTITION_SPEC_SEPARATOR); + + Ok(expected_partition_name == other_partition_name) + } + + pub fn to_pb(&self) -> PbPartitionSpec { + PbPartitionSpec { + partition_key_values: self + .partition_keys + .iter() + .zip(self.partition_values.iter()) + .map(|(k, v)| PbKeyValue { + key: k.clone(), + value: v.clone(), + }) + .collect(), + } + } + + pub fn from_pb(pb: &PbPartitionSpec) -> Self { + let partition_keys = pb + .partition_key_values + .iter() + .map(|kv| kv.key.clone()) + .collect(); + let partition_values = pb + .partition_key_values + .iter() + .map(|kv| kv.value.clone()) + .collect(); + Self { + partition_keys, + partition_values, + } + } + + fn get_reordered_partition_values( + partition_keys: &[String], + partition_spec: &PartitionSpec, + ) -> Vec<String> { + let partition_spec_map = partition_spec.get_spec_map(); + partition_keys + .iter() + .map(|key| partition_spec_map.get(key).cloned().unwrap_or_default()) + .collect() + } +} + +impl Display for ResolvedPartitionSpec { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.get_partition_qualified_name()) + } +} + +/// Information of a partition metadata, includes the partition's name and the partition id that +/// represents the unique identifier of the partition. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct PartitionInfo { + partition_id: i64, Review Comment: Addressed in latest commit ########## crates/fluss/src/metadata/partition.rs: ########## @@ -0,0 +1,469 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use crate::PartitionId; +use crate::error::{Error, Result}; +use crate::proto::{PbKeyValue, PbPartitionInfo, PbPartitionSpec}; +use std::collections::HashMap; +use std::fmt::{Display, Formatter}; + +/// Represents a partition spec in fluss. Partition columns and values are NOT of strict order, and +/// they need to be re-arranged to the correct order by comparing with a list of strictly ordered +/// partition keys. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PartitionSpec { + partition_spec: HashMap<String, String>, +} + +impl PartitionSpec { + pub fn new(partition_spec: HashMap<String, String>) -> Self { + Self { partition_spec } + } + + pub fn get_spec_map(&self) -> &HashMap<String, String> { + &self.partition_spec + } + + pub fn to_pb(&self) -> PbPartitionSpec { + PbPartitionSpec { + partition_key_values: self + .partition_spec + .iter() + .map(|(k, v)| PbKeyValue { + key: k.clone(), + value: v.clone(), + }) + .collect(), + } + } + + pub fn from_pb(pb: &PbPartitionSpec) -> Self { + let partition_spec = pb + .partition_key_values + .iter() + .map(|kv| (kv.key.clone(), kv.value.clone())) + .collect(); + Self { partition_spec } + } +} + +impl Display for PartitionSpec { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "PartitionSpec{{{:?}}}", self.partition_spec) + } +} + +/// Represents a partition, which is the resolved version of PartitionSpec. The partition +/// spec is re-arranged into the correct order by comparing it with a list of strictly ordered +/// partition keys. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct ResolvedPartitionSpec { + partition_keys: Vec<String>, + partition_values: Vec<String>, +} + +pub const PARTITION_SPEC_SEPARATOR: &str = "$"; + +impl ResolvedPartitionSpec { + pub fn new(partition_keys: Vec<String>, partition_values: Vec<String>) -> Result<Self> { + if partition_keys.len() != partition_values.len() { + return Err(Error::IllegalArgument { + message: "The number of partition keys and partition values should be the same." + .to_string(), + }); + } + Ok(Self { + partition_keys, + partition_values, + }) + } + + pub fn from_partition_spec( + partition_keys: Vec<String>, + partition_spec: &PartitionSpec, + ) -> Self { + let partition_values = + Self::get_reordered_partition_values(&partition_keys, partition_spec); + Self { + partition_keys, + partition_values, + } + } + + pub fn from_partition_value(partition_key: String, partition_value: String) -> Self { + Self { + partition_keys: vec![partition_key], + partition_values: vec![partition_value], + } + } + + pub fn from_partition_name(partition_keys: Vec<String>, partition_name: &str) -> Self { + let partition_values: Vec<String> = partition_name + .split(PARTITION_SPEC_SEPARATOR) + .map(|s| s.to_string()) + .collect(); + Self { + partition_keys, + partition_values, + } + } + + pub fn from_partition_qualified_name(qualified_partition_name: &str) -> Result<Self> { + let mut keys = Vec::new(); + let mut values = Vec::new(); + + for pair in qualified_partition_name.split('/') { + let parts: Vec<&str> = pair.splitn(2, '=').collect(); + if parts.len() != 2 { + return Err(Error::IllegalArgument { + message: format!( + "Invalid partition name format. Expected key=value, got: {}", + pair + ), + }); + } + keys.push(parts[0].to_string()); + values.push(parts[1].to_string()); + } + + Ok(Self { + partition_keys: keys, + partition_values: values, + }) + } + + pub fn get_partition_keys(&self) -> &[String] { + &self.partition_keys + } + + pub fn get_partition_values(&self) -> &[String] { + &self.partition_values + } + + pub fn to_partition_spec(&self) -> PartitionSpec { + let mut spec_map = HashMap::new(); + for (i, key) in self.partition_keys.iter().enumerate() { + spec_map.insert(key.clone(), self.partition_values[i].clone()); + } + PartitionSpec::new(spec_map) + } + + /// Generate the partition name for a partition table with specified partition values. + /// + /// The partition name is in the following format: value1$value2$...$valueN + pub fn get_partition_name(&self) -> String { + self.partition_values.join(PARTITION_SPEC_SEPARATOR) + } + + /// Returns the qualified partition name for a partition spec. + /// The format is: key1=value1/key2=value2/.../keyN=valueN + pub fn get_partition_qualified_name(&self) -> String { + let mut sb = String::new(); + for (i, key) in self.partition_keys.iter().enumerate() { + sb.push_str(key); + sb.push('='); + sb.push_str(&self.partition_values[i]); + if i != self.partition_keys.len() - 1 { + sb.push('/'); + } + } + sb + } + + pub fn contains(&self, other: &ResolvedPartitionSpec) -> Result<bool> { + let other_partition_keys = other.get_partition_keys(); + let other_partition_values = other.get_partition_values(); + + let mut expected_partition_values = Vec::new(); + for other_partition_key in other_partition_keys { + let key_index = self + .partition_keys + .iter() + .position(|k| k == other_partition_key); + match key_index { + Some(idx) => expected_partition_values.push(self.partition_values[idx].clone()), + None => { + return Err(Error::IllegalArgument { + message: format!( + "table does not contain partitionKey: {}", + other_partition_key + ), + }); + } + } + } + + let expected_partition_name = expected_partition_values.join(PARTITION_SPEC_SEPARATOR); + let other_partition_name = other_partition_values.join(PARTITION_SPEC_SEPARATOR); + + Ok(expected_partition_name == other_partition_name) + } + + pub fn to_pb(&self) -> PbPartitionSpec { + PbPartitionSpec { + partition_key_values: self + .partition_keys + .iter() + .zip(self.partition_values.iter()) + .map(|(k, v)| PbKeyValue { + key: k.clone(), + value: v.clone(), + }) + .collect(), + } + } + + pub fn from_pb(pb: &PbPartitionSpec) -> Self { + let partition_keys = pb + .partition_key_values + .iter() + .map(|kv| kv.key.clone()) + .collect(); + let partition_values = pb + .partition_key_values + .iter() + .map(|kv| kv.value.clone()) + .collect(); + Self { + partition_keys, + partition_values, + } + } + + fn get_reordered_partition_values( + partition_keys: &[String], + partition_spec: &PartitionSpec, + ) -> Vec<String> { + let partition_spec_map = partition_spec.get_spec_map(); + partition_keys + .iter() + .map(|key| partition_spec_map.get(key).cloned().unwrap_or_default()) + .collect() + } +} + +impl Display for ResolvedPartitionSpec { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.get_partition_qualified_name()) + } +} + +/// Information of a partition metadata, includes the partition's name and the partition id that +/// represents the unique identifier of the partition. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct PartitionInfo { + partition_id: i64, + partition_spec: ResolvedPartitionSpec, +} + +impl PartitionInfo { + pub fn new(partition_id: i64, partition_spec: ResolvedPartitionSpec) -> Self { + Self { + partition_id, + partition_spec, + } + } + + /// Get the partition id. The id is globally unique in the Fluss cluster. + pub fn get_partition_id(&self) -> i64 { Review Comment: Addressed in latest commit ########## crates/fluss/src/metadata/partition.rs: ########## @@ -0,0 +1,469 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use crate::PartitionId; +use crate::error::{Error, Result}; +use crate::proto::{PbKeyValue, PbPartitionInfo, PbPartitionSpec}; +use std::collections::HashMap; +use std::fmt::{Display, Formatter}; + +/// Represents a partition spec in fluss. Partition columns and values are NOT of strict order, and +/// they need to be re-arranged to the correct order by comparing with a list of strictly ordered +/// partition keys. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PartitionSpec { + partition_spec: HashMap<String, String>, +} + +impl PartitionSpec { + pub fn new(partition_spec: HashMap<String, String>) -> Self { + Self { partition_spec } + } + + pub fn get_spec_map(&self) -> &HashMap<String, String> { + &self.partition_spec + } + + pub fn to_pb(&self) -> PbPartitionSpec { + PbPartitionSpec { + partition_key_values: self + .partition_spec + .iter() + .map(|(k, v)| PbKeyValue { + key: k.clone(), + value: v.clone(), + }) + .collect(), + } + } + + pub fn from_pb(pb: &PbPartitionSpec) -> Self { + let partition_spec = pb + .partition_key_values + .iter() + .map(|kv| (kv.key.clone(), kv.value.clone())) + .collect(); + Self { partition_spec } + } +} + +impl Display for PartitionSpec { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "PartitionSpec{{{:?}}}", self.partition_spec) + } +} + +/// Represents a partition, which is the resolved version of PartitionSpec. The partition +/// spec is re-arranged into the correct order by comparing it with a list of strictly ordered +/// partition keys. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct ResolvedPartitionSpec { + partition_keys: Vec<String>, + partition_values: Vec<String>, +} + +pub const PARTITION_SPEC_SEPARATOR: &str = "$"; + +impl ResolvedPartitionSpec { + pub fn new(partition_keys: Vec<String>, partition_values: Vec<String>) -> Result<Self> { + if partition_keys.len() != partition_values.len() { + return Err(Error::IllegalArgument { + message: "The number of partition keys and partition values should be the same." + .to_string(), + }); + } + Ok(Self { + partition_keys, + partition_values, + }) + } + + pub fn from_partition_spec( + partition_keys: Vec<String>, + partition_spec: &PartitionSpec, + ) -> Self { + let partition_values = + Self::get_reordered_partition_values(&partition_keys, partition_spec); + Self { + partition_keys, + partition_values, + } + } + + pub fn from_partition_value(partition_key: String, partition_value: String) -> Self { + Self { + partition_keys: vec![partition_key], + partition_values: vec![partition_value], + } + } + + pub fn from_partition_name(partition_keys: Vec<String>, partition_name: &str) -> Self { + let partition_values: Vec<String> = partition_name + .split(PARTITION_SPEC_SEPARATOR) + .map(|s| s.to_string()) + .collect(); + Self { + partition_keys, + partition_values, + } + } + + pub fn from_partition_qualified_name(qualified_partition_name: &str) -> Result<Self> { + let mut keys = Vec::new(); + let mut values = Vec::new(); + + for pair in qualified_partition_name.split('/') { + let parts: Vec<&str> = pair.splitn(2, '=').collect(); + if parts.len() != 2 { + return Err(Error::IllegalArgument { + message: format!( + "Invalid partition name format. Expected key=value, got: {}", + pair + ), + }); + } + keys.push(parts[0].to_string()); + values.push(parts[1].to_string()); + } + + Ok(Self { + partition_keys: keys, + partition_values: values, + }) + } + + pub fn get_partition_keys(&self) -> &[String] { + &self.partition_keys + } + + pub fn get_partition_values(&self) -> &[String] { + &self.partition_values + } + + pub fn to_partition_spec(&self) -> PartitionSpec { + let mut spec_map = HashMap::new(); + for (i, key) in self.partition_keys.iter().enumerate() { + spec_map.insert(key.clone(), self.partition_values[i].clone()); + } + PartitionSpec::new(spec_map) + } + + /// Generate the partition name for a partition table with specified partition values. + /// + /// The partition name is in the following format: value1$value2$...$valueN + pub fn get_partition_name(&self) -> String { + self.partition_values.join(PARTITION_SPEC_SEPARATOR) + } + + /// Returns the qualified partition name for a partition spec. + /// The format is: key1=value1/key2=value2/.../keyN=valueN + pub fn get_partition_qualified_name(&self) -> String { + let mut sb = String::new(); + for (i, key) in self.partition_keys.iter().enumerate() { + sb.push_str(key); + sb.push('='); + sb.push_str(&self.partition_values[i]); + if i != self.partition_keys.len() - 1 { + sb.push('/'); + } + } + sb + } + + pub fn contains(&self, other: &ResolvedPartitionSpec) -> Result<bool> { + let other_partition_keys = other.get_partition_keys(); + let other_partition_values = other.get_partition_values(); + + let mut expected_partition_values = Vec::new(); + for other_partition_key in other_partition_keys { + let key_index = self + .partition_keys + .iter() + .position(|k| k == other_partition_key); + match key_index { + Some(idx) => expected_partition_values.push(self.partition_values[idx].clone()), + None => { + return Err(Error::IllegalArgument { + message: format!( + "table does not contain partitionKey: {}", + other_partition_key + ), + }); + } + } + } + + let expected_partition_name = expected_partition_values.join(PARTITION_SPEC_SEPARATOR); + let other_partition_name = other_partition_values.join(PARTITION_SPEC_SEPARATOR); + + Ok(expected_partition_name == other_partition_name) + } + + pub fn to_pb(&self) -> PbPartitionSpec { + PbPartitionSpec { + partition_key_values: self + .partition_keys + .iter() + .zip(self.partition_values.iter()) + .map(|(k, v)| PbKeyValue { + key: k.clone(), + value: v.clone(), + }) + .collect(), + } + } + + pub fn from_pb(pb: &PbPartitionSpec) -> Self { + let partition_keys = pb + .partition_key_values + .iter() + .map(|kv| kv.key.clone()) + .collect(); + let partition_values = pb + .partition_key_values + .iter() + .map(|kv| kv.value.clone()) + .collect(); + Self { + partition_keys, + partition_values, + } + } + + fn get_reordered_partition_values( + partition_keys: &[String], + partition_spec: &PartitionSpec, + ) -> Vec<String> { + let partition_spec_map = partition_spec.get_spec_map(); + partition_keys + .iter() + .map(|key| partition_spec_map.get(key).cloned().unwrap_or_default()) + .collect() + } +} + +impl Display for ResolvedPartitionSpec { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.get_partition_qualified_name()) + } +} + +/// Information of a partition metadata, includes the partition's name and the partition id that +/// represents the unique identifier of the partition. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct PartitionInfo { + partition_id: i64, + partition_spec: ResolvedPartitionSpec, +} + +impl PartitionInfo { + pub fn new(partition_id: i64, partition_spec: ResolvedPartitionSpec) -> Self { Review Comment: Addressed in latest commit -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
