You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "callum-ryan (via GitHub)" <gi...@apache.org> on 2024/08/09 19:36:27 UTC
[PR] feat: SQL Catalog - namespaces [iceberg-rust]
callum-ryan opened a new pull request, #534:
URL: https://github.com/apache/iceberg-rust/pull/534
Carrying on from [PR #524 ](https://github.com/apache/iceberg-rust/pull/524), this is the implementation of the namespace functionality
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "callum-ryan (via GitHub)" <gi...@apache.org>.
callum-ryan commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1730004206
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -141,21 +142,30 @@ impl SqlCatalog {
}
/// SQLX Any does not implement PostgresSQL bindings, so we have to do this.
- pub async fn execute_statement(
- &self,
- query: &String,
- args: Vec<Option<&String>>,
- ) -> Result<Vec<AnyRow>> {
- let query_with_placeholders: Cow<str> =
- if self.sql_bind_style == SqlBindStyle::DollarNumeric {
- let mut query = query.clone();
- for i in 0..args.len() {
- query = query.replacen("?", &format!("${}", i + 1), 1);
- }
- Cow::Owned(query)
- } else {
- Cow::Borrowed(query)
- };
+ fn replace_placeholders(&self, query: &str) -> String {
+ match self.sql_bind_style {
Review Comment:
@Xuanwo @liurenjie1024 thoughts on this? Keen to close this PR out.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1731193964
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -141,21 +142,30 @@ impl SqlCatalog {
}
/// SQLX Any does not implement PostgresSQL bindings, so we have to do this.
- pub async fn execute_statement(
- &self,
- query: &String,
- args: Vec<Option<&String>>,
- ) -> Result<Vec<AnyRow>> {
- let query_with_placeholders: Cow<str> =
- if self.sql_bind_style == SqlBindStyle::DollarNumeric {
- let mut query = query.clone();
- for i in 0..args.len() {
- query = query.replacen("?", &format!("${}", i + 1), 1);
- }
- Cow::Owned(query)
- } else {
- Cow::Borrowed(query)
- };
+ fn replace_placeholders(&self, query: &str) -> String {
+ match self.sql_bind_style {
Review Comment:
I think it's kind of weird to ask user to pass this as this may lead to unclear error. I think this may be not a blocker issue, but at least please add some comments to help user to understand it.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "sdd (via GitHub)" <gi...@apache.org>.
sdd commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1714277890
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,344 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ pub async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ let table_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+ let namespaces_stmt = format!(
+ "SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ match parent {
+ Some(parent) => match self.namespace_exists(parent).await? {
+ true => {
+ let parent_str = parent.join(".");
+ let parent_table_namespaces_stmt = format!(
+ "{table_namespaces_stmt} AND {CATALOG_FIELD_TABLE_NAMESPACE} LIKE CONCAT(?, '%')"
+ );
+ let parent_namespaces_stmt =
+ format!("{namespaces_stmt} AND {NAMESPACE_FIELD_NAME} LIKE CONCAT(?, '%')");
+
+ let namespace_rows = self
+ .fetch_rows(
+ &format!(
+ "{parent_namespaces_stmt} UNION {parent_table_namespaces_stmt}"
+ ),
+ vec![
+ Some(&self.name),
+ Some(&parent_str),
+ Some(&self.name),
+ Some(&parent_str),
+ ],
+ )
+ .await?;
+
+ Ok(namespace_rows
+ .iter()
+ .filter_map(|r| {
+ let nsp = r.try_get::<String, _>(0).ok();
Review Comment:
Is this the behaviour we want? using `.ok()` suppresses any error that might get raised in `r.try_get()`. Would we not want to bail out of `list_namespaces` with the error instead?
Using `filter_map` can make this harder to do. We could make the error-handling behaviour easier to implement by using a for loop instead as we stay inside the context of `list_namespaces` rather than traversing into a closure where we can't use `?`.
If you initialize a `Vec` for the results using `Vec::with_capacity(namespace_rows.len())` then you still only have one heap allocation, as you would have here from the `collect()` call anyway.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "sdd (via GitHub)" <gi...@apache.org>.
sdd commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1725609116
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -276,4 +579,385 @@ mod tests {
new_sql_catalog(warehouse_loc.clone()).await;
new_sql_catalog(warehouse_loc.clone()).await;
}
+
+ #[tokio::test]
+ async fn test_list_namespaces_returns_empty_vector() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+
+ assert_eq!(catalog.list_namespaces(None).await.unwrap(), vec![]);
+ }
+
+ #[tokio::test]
+ async fn test_list_namespaces_returns_multiple_namespaces() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_1 = NamespaceIdent::new("a".into());
+ let namespace_ident_2 = NamespaceIdent::new("b".into());
+ create_namespaces(&catalog, &vec![&namespace_ident_1, &namespace_ident_2]).await;
+
+ assert_eq!(
+ to_set(catalog.list_namespaces(None).await.unwrap()),
+ to_set(vec![namespace_ident_1, namespace_ident_2])
+ );
+ }
+
+ #[tokio::test]
+ async fn test_list_namespaces_returns_only_top_level_namespaces() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_1 = NamespaceIdent::new("a".into());
+ let namespace_ident_2 = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+ let namespace_ident_3 = NamespaceIdent::new("b".into());
+ create_namespaces(&catalog, &vec![
+ &namespace_ident_1,
+ &namespace_ident_2,
+ &namespace_ident_3,
+ ])
+ .await;
+
+ assert_eq!(
+ to_set(catalog.list_namespaces(None).await.unwrap()),
+ to_set(vec![namespace_ident_1, namespace_ident_3])
+ );
+ }
+
+ #[tokio::test]
+ async fn test_list_namespaces_returns_no_namespaces_under_parent() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_1 = NamespaceIdent::new("a".into());
+ let namespace_ident_2 = NamespaceIdent::new("b".into());
+ create_namespaces(&catalog, &vec![&namespace_ident_1, &namespace_ident_2]).await;
+
+ assert_eq!(
+ catalog
+ .list_namespaces(Some(&namespace_ident_1))
+ .await
+ .unwrap(),
+ vec![]
+ );
+ }
+
+ #[tokio::test]
+ async fn test_list_namespaces_returns_namespace_under_parent() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_1 = NamespaceIdent::new("a".into());
+ let namespace_ident_2 = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+ let namespace_ident_3 = NamespaceIdent::new("c".into());
+ create_namespaces(&catalog, &vec![
+ &namespace_ident_1,
+ &namespace_ident_2,
+ &namespace_ident_3,
+ ])
+ .await;
+
+ assert_eq!(
+ to_set(catalog.list_namespaces(None).await.unwrap()),
+ to_set(vec![namespace_ident_1.clone(), namespace_ident_3])
+ );
+
+ assert_eq!(
+ catalog
+ .list_namespaces(Some(&namespace_ident_1))
+ .await
+ .unwrap(),
+ vec![NamespaceIdent::from_strs(vec!["a", "b"]).unwrap()]
+ );
+ }
+
+ #[tokio::test]
+ async fn test_list_namespaces_returns_multiple_namespaces_under_parent() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_1 = NamespaceIdent::new("a".to_string());
+ let namespace_ident_2 = NamespaceIdent::from_strs(vec!["a", "a"]).unwrap();
+ let namespace_ident_3 = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+ let namespace_ident_4 = NamespaceIdent::from_strs(vec!["a", "c"]).unwrap();
+ let namespace_ident_5 = NamespaceIdent::new("b".into());
+ create_namespaces(&catalog, &vec![
+ &namespace_ident_1,
+ &namespace_ident_2,
+ &namespace_ident_3,
+ &namespace_ident_4,
+ &namespace_ident_5,
+ ])
+ .await;
+
+ assert_eq!(
+ to_set(
+ catalog
+ .list_namespaces(Some(&namespace_ident_1))
+ .await
+ .unwrap()
+ ),
+ to_set(vec![
+ NamespaceIdent::from_strs(vec!["a", "a"]).unwrap(),
+ NamespaceIdent::from_strs(vec!["a", "b"]).unwrap(),
+ NamespaceIdent::from_strs(vec!["a", "c"]).unwrap(),
+ ])
+ );
+ }
+
+ #[tokio::test]
+ async fn test_namespace_exists_returns_false() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident = NamespaceIdent::new("a".into());
+ create_namespace(&catalog, &namespace_ident).await;
+
+ assert!(!catalog
+ .namespace_exists(&NamespaceIdent::new("b".into()))
+ .await
+ .unwrap());
+ }
+
+ #[tokio::test]
+ async fn test_namespace_exists_returns_true() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident = NamespaceIdent::new("a".into());
+ create_namespace(&catalog, &namespace_ident).await;
+
+ assert!(catalog.namespace_exists(&namespace_ident).await.unwrap());
+ }
+
+ #[tokio::test]
+ async fn test_create_namespace_with_properties() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident = NamespaceIdent::new("abc".into());
+
+ let mut properties: HashMap<String, String> = HashMap::new();
+ properties.insert("k".into(), "v".into());
+
+ assert_eq!(
+ catalog
+ .create_namespace(&namespace_ident, properties.clone())
+ .await
+ .unwrap(),
+ Namespace::with_properties(namespace_ident.clone(), properties.clone())
+ );
+
+ assert_eq!(
+ catalog.get_namespace(&namespace_ident).await.unwrap(),
+ Namespace::with_properties(namespace_ident, properties)
+ );
+ }
+
+ #[tokio::test]
+ async fn test_create_namespace_throws_error_if_namespace_already_exists() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident = NamespaceIdent::new("a".into());
+ create_namespace(&catalog, &namespace_ident).await;
+
+ assert_eq!(
+ catalog
+ .create_namespace(&namespace_ident, HashMap::new())
+ .await
+ .unwrap_err()
+ .to_string(),
+ format!(
+ "Unexpected => Namespace {:?} already exists",
+ &namespace_ident
+ )
+ );
+
+ assert_eq!(
+ catalog.get_namespace(&namespace_ident).await.unwrap(),
+ Namespace::with_properties(namespace_ident, default_properties())
+ );
+ }
+
+ #[tokio::test]
+ async fn test_create_nested_namespace() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let parent_namespace_ident = NamespaceIdent::new("a".into());
+ create_namespace(&catalog, &parent_namespace_ident).await;
+
+ let child_namespace_ident = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+
+ assert_eq!(
+ catalog
+ .create_namespace(&child_namespace_ident, HashMap::new())
+ .await
+ .unwrap(),
+ Namespace::new(child_namespace_ident.clone())
+ );
+
+ assert_eq!(
+ catalog.get_namespace(&child_namespace_ident).await.unwrap(),
+ Namespace::with_properties(child_namespace_ident, default_properties())
+ );
+ }
+
+ #[tokio::test]
+ async fn test_create_deeply_nested_namespace() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_a = NamespaceIdent::new("a".into());
+ let namespace_ident_a_b = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+ create_namespaces(&catalog, &vec![&namespace_ident_a, &namespace_ident_a_b]).await;
+
+ let namespace_ident_a_b_c = NamespaceIdent::from_strs(vec!["a", "b", "c"]).unwrap();
+
+ assert_eq!(
+ catalog
+ .create_namespace(&namespace_ident_a_b_c, HashMap::new())
+ .await
+ .unwrap(),
+ Namespace::new(namespace_ident_a_b_c.clone())
+ );
+
+ assert_eq!(
+ catalog.get_namespace(&namespace_ident_a_b_c).await.unwrap(),
+ Namespace::with_properties(namespace_ident_a_b_c, default_properties())
+ );
+ }
+
+ #[tokio::test]
+ async fn test_create_nested_namespace_throws_error_if_top_level_namespace_doesnt_exist() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+
+ let nested_namespace_ident = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+
+ assert_eq!(
+ catalog
+ .create_namespace(&nested_namespace_ident, HashMap::new())
+ .await
+ .unwrap_err()
+ .to_string(),
+ format!(
+ "Unexpected => No such namespace: {:?}",
+ NamespaceIdent::new("a".into())
+ )
+ );
+
+ assert_eq!(catalog.list_namespaces(None).await.unwrap(), vec![]);
+ }
+
+ #[tokio::test]
+ async fn test_drop_namespace() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident = NamespaceIdent::new("abc".into());
+ create_namespace(&catalog, &namespace_ident).await;
+
+ catalog.drop_namespace(&namespace_ident).await.unwrap();
+
+ assert!(!catalog.namespace_exists(&namespace_ident).await.unwrap())
+ }
+
+ #[tokio::test]
+ async fn test_drop_nested_namespace() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_a = NamespaceIdent::new("a".into());
+ let namespace_ident_a_b = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+ create_namespaces(&catalog, &vec![&namespace_ident_a, &namespace_ident_a_b]).await;
+
+ catalog.drop_namespace(&namespace_ident_a_b).await.unwrap();
+
+ assert!(!catalog
+ .namespace_exists(&namespace_ident_a_b)
+ .await
+ .unwrap());
+
+ assert!(catalog.namespace_exists(&namespace_ident_a).await.unwrap());
+ }
+
+ #[tokio::test]
+ async fn test_drop_deeply_nested_namespace() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_a = NamespaceIdent::new("a".into());
+ let namespace_ident_a_b = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+ let namespace_ident_a_b_c = NamespaceIdent::from_strs(vec!["a", "b", "c"]).unwrap();
+ create_namespaces(&catalog, &vec![
+ &namespace_ident_a,
+ &namespace_ident_a_b,
+ &namespace_ident_a_b_c,
+ ])
+ .await;
+
+ catalog
+ .drop_namespace(&namespace_ident_a_b_c)
+ .await
+ .unwrap();
+
+ assert!(!catalog
+ .namespace_exists(&namespace_ident_a_b_c)
+ .await
+ .unwrap());
+
+ assert!(catalog
+ .namespace_exists(&namespace_ident_a_b)
+ .await
+ .unwrap());
+
+ assert!(catalog.namespace_exists(&namespace_ident_a).await.unwrap());
+ }
+
+ #[tokio::test]
+ async fn test_drop_namespace_throws_error_if_namespace_doesnt_exist() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+
+ let non_existent_namespace_ident = NamespaceIdent::new("abc".into());
+ assert_eq!(
+ catalog
+ .drop_namespace(&non_existent_namespace_ident)
+ .await
+ .unwrap_err()
+ .to_string(),
+ format!(
+ "Unexpected => No such namespace: {:?}",
+ non_existent_namespace_ident
+ )
+ )
+ }
+
+ #[tokio::test]
+ async fn test_drop_namespace_throws_error_if_nested_namespace_doesnt_exist() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ create_namespace(&catalog, &NamespaceIdent::new("a".into())).await;
+
+ let non_existent_namespace_ident =
+ NamespaceIdent::from_vec(vec!["a".into(), "b".into()]).unwrap();
+ assert_eq!(
+ catalog
+ .drop_namespace(&non_existent_namespace_ident)
+ .await
+ .unwrap_err()
+ .to_string(),
+ format!(
+ "Unexpected => No such namespace: {:?}",
+ non_existent_namespace_ident
+ )
+ )
+ }
+
+ #[tokio::test]
+ #[ignore = "Java/Python do not drop nested namespaces?"]
+ async fn test_dropping_a_namespace_also_drops_namespaces_nested_under_that_one() {
Review Comment:
I think your second suggestion, only allowing dropping a namespace when it is empty, makes sense. Or how about adding a boolean flag, `drop_children`? If true, drop any children present in the namespace as well as the namespace. If false, only drop the namespace if it is empty, and return an `Err` otherwise.
All of these approaches seem more sensible than the Python / Java approach of dropping a namespace but leaving any tables that were within it with a missing parent. That feels buggy.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "callum-ryan (via GitHub)" <gi...@apache.org>.
callum-ryan commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1725539258
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,312 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ // UNION will remove duplicates.
+ let all_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
+ UNION
+ SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ let namespace_rows = self
+ .fetch_rows(&all_namespaces_stmt, vec![
+ Some(&self.name),
+ Some(&self.name),
+ ])
+ .await?;
+
+ let mut namespaces = HashSet::<NamespaceIdent>::with_capacity(namespace_rows.len());
+
+ if let Some(parent) = parent {
+ if self.namespace_exists(parent).await? {
+ let parent_str = parent.join(".");
+
+ for row in namespace_rows.iter() {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ // if parent = a, then we only want to see a.b, a.c returned.
+ if nsp != parent_str && nsp.starts_with(&parent_str) {
+ namespaces.insert(NamespaceIdent::from_strs(nsp.split("."))?);
+ }
+ }
+
+ Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
+ } else {
+ no_such_namespace_err(parent)
+ }
+ } else {
+ for row in namespace_rows.iter() {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ let mut levels = nsp.split(".").collect::<Vec<&str>>();
+ if !levels.is_empty() {
+ let first_level = levels.drain(..1).collect::<Vec<&str>>();
+ namespaces.insert(NamespaceIdent::from_strs(first_level)?);
+ }
+ }
+
+ Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
+ }
}
async fn create_namespace(
&self,
- _namespace: &NamespaceIdent,
- _properties: HashMap<String, String>,
+ namespace: &NamespaceIdent,
+ properties: HashMap<String, String>,
) -> Result<Namespace> {
- todo!()
+ let exists = self.namespace_exists(namespace).await?;
+
+ if exists {
+ return Err(Error::new(
+ iceberg::ErrorKind::Unexpected,
+ format!("Namespace {:?} already exists", namespace),
+ ));
+ }
+
+ for i in 1..namespace.len() {
+ let parent_namespace = NamespaceIdent::from_vec(namespace[..i].to_vec())?;
+ let parent_exists = self.namespace_exists(&parent_namespace).await?;
+ if !parent_exists {
+ return no_such_namespace_err(&parent_namespace);
+ }
+ }
+
+ let namespace_str = namespace.join(".");
+ let insert = format!(
+ "INSERT INTO {NAMESPACE_TABLE_NAME} ({CATALOG_FIELD_CATALOG_NAME}, {NAMESPACE_FIELD_NAME}, {NAMESPACE_FIELD_PROPERTY_KEY}, {NAMESPACE_FIELD_PROPERTY_VALUE})
+ VALUES (?, ?, ?, ?)");
+ if !properties.is_empty() {
+ let mut query_args = Vec::with_capacity(properties.len() * 4);
+ let mut properties_insert = insert.clone();
+ for (index, (key, value)) in properties.iter().enumerate() {
+ query_args.extend_from_slice(&[
+ Some(self.name.as_str()),
+ Some(namespace_str.as_str()),
+ Some(key.as_str()),
+ Some(value.as_str()),
+ ]);
+ if index > 0 {
+ properties_insert = format!("{properties_insert}, (?, ?, ?, ?)");
+ }
+ }
+
+ self.execute(&properties_insert, query_args, None).await?;
+
+ Ok(Namespace::with_properties(namespace.clone(), properties))
+ } else {
+ // set a default property of exists = true
+ // up for debate if this is worthwhile
+ self.execute(
+ &insert,
+ vec![
+ Some(&self.name),
+ Some(&namespace_str),
+ Some("exists"),
+ Some("true"),
+ ],
+ None,
+ )
+ .await?;
+ Ok(Namespace::with_properties(namespace.clone(), properties))
+ }
}
- async fn get_namespace(&self, _namespace: &NamespaceIdent) -> Result<Namespace> {
- todo!()
+ async fn get_namespace(&self, namespace: &NamespaceIdent) -> Result<Namespace> {
+ let exists = self.namespace_exists(namespace).await?;
+ if exists {
+ let namespace_props = self
+ .fetch_rows(
+ &format!(
+ "SELECT
+ {NAMESPACE_FIELD_NAME},
+ {NAMESPACE_FIELD_PROPERTY_KEY},
+ {NAMESPACE_FIELD_PROPERTY_VALUE}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
+ AND {NAMESPACE_FIELD_NAME} = ?"
+ ),
+ vec![Some(&self.name), Some(&namespace.join("."))],
+ )
+ .await?;
+
+ let mut properties = HashMap::with_capacity(namespace_props.len());
+
+ for row in namespace_props {
+ let key = row
+ .try_get::<String, _>(NAMESPACE_FIELD_PROPERTY_KEY)
+ .map_err(from_sqlx_error)?;
+ let value = row
+ .try_get::<String, _>(NAMESPACE_FIELD_PROPERTY_VALUE)
+ .map_err(from_sqlx_error)?;
+
+ properties.insert(key, value);
+ }
+
+ Ok(Namespace::with_properties(namespace.clone(), properties))
+ } else {
+ no_such_namespace_err(namespace)
+ }
}
- async fn namespace_exists(&self, _namespace: &NamespaceIdent) -> Result<bool> {
- todo!()
+ async fn namespace_exists(&self, namespace: &NamespaceIdent) -> Result<bool> {
+ let namespace_str = namespace.join(".");
+
+ let table_namespaces = self
+ .fetch_rows(
+ &format!(
+ "SELECT 1 FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
+ AND {CATALOG_FIELD_TABLE_NAMESPACE} = ?
+ LIMIT 1"
+ ),
+ vec![Some(&self.name), Some(&namespace_str)],
+ )
+ .await?;
+
+ if !table_namespaces.is_empty() {
+ Ok(true)
+ } else {
+ let namespaces = self
+ .fetch_rows(
+ &format!(
+ "SELECT 1 FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
+ AND {NAMESPACE_FIELD_NAME} = ?
+ LIMIT 1"
+ ),
+ vec![Some(&self.name), Some(&namespace_str)],
+ )
+ .await?;
+ if !namespaces.is_empty() {
+ Ok(true)
+ } else {
+ Ok(false)
+ }
+ }
}
async fn update_namespace(
&self,
- _namespace: &NamespaceIdent,
- _properties: HashMap<String, String>,
+ namespace: &NamespaceIdent,
+ properties: HashMap<String, String>,
) -> Result<()> {
- todo!()
+ let exists = self.namespace_exists(namespace).await?;
+ if exists {
+ let existing_properties = self.get_namespace(namespace).await?.properties().clone();
+ let namespace_str = namespace.join(".");
+
+ let mut updates = vec![];
+ let mut inserts = vec![];
+
+ for (key, value) in properties.iter() {
+ if existing_properties.contains_key(key) {
+ if existing_properties.get(key) != Some(value) {
+ updates.push((key, value));
+ }
+ } else {
+ inserts.push((key, value));
+ }
+ }
+
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let update_stmt = format!(
+ "UPDATE {NAMESPACE_TABLE_NAME} SET {NAMESPACE_FIELD_PROPERTY_VALUE} = ?
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
+ AND {NAMESPACE_FIELD_NAME} = ?
+ AND {NAMESPACE_FIELD_PROPERTY_KEY} = ?"
+ );
+
+ let insert_stmt = format!(
+ "INSERT INTO {NAMESPACE_TABLE_NAME} ({CATALOG_FIELD_CATALOG_NAME}, {NAMESPACE_FIELD_NAME}, {NAMESPACE_FIELD_PROPERTY_KEY}, {NAMESPACE_FIELD_PROPERTY_VALUE})
+ VALUES (?, ?, ?, ?)"
+ );
+
+ for (key, value) in updates {
+ self.execute(
+ &update_stmt,
+ vec![
+ Some(value),
+ Some(&self.name),
+ Some(&namespace_str),
+ Some(key),
+ ],
+ Some(&mut tx),
+ )
+ .await?;
+ }
+
+ for (key, value) in inserts {
+ self.execute(
+ &insert_stmt,
+ vec![
+ Some(&self.name),
+ Some(&namespace_str),
+ Some(key),
+ Some(value),
+ ],
+ Some(&mut tx),
+ )
+ .await?;
+ }
+
+ let _ = tx.commit().await.map_err(from_sqlx_error)?;
+
+ Ok(())
+ } else {
+ no_such_namespace_err(namespace)
+ }
}
- async fn drop_namespace(&self, _namespace: &NamespaceIdent) -> Result<()> {
- todo!()
+ async fn drop_namespace(&self, namespace: &NamespaceIdent) -> Result<()> {
+ let exists = self.namespace_exists(namespace).await?;
+ if exists {
+ // TODO: check that the namespace is empty
Review Comment:
Yes, this functionality will be added with the `list_tables` method I am going to implement in the next phase of the SQL Catalog. I'll remove this for now.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "callum-ryan (via GitHub)" <gi...@apache.org>.
callum-ryan commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1720765041
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,335 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ pub async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ let table_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+ let namespaces_stmt = format!(
+ "SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ match parent {
+ Some(parent) => {
+ if self.namespace_exists(parent).await? {
+ let parent_str = parent.join(".");
+ let parent_table_namespaces_stmt = format!(
+ "{table_namespaces_stmt} AND {CATALOG_FIELD_TABLE_NAMESPACE} LIKE CONCAT(?, '%')"
+ );
+ let parent_namespaces_stmt =
+ format!("{namespaces_stmt} AND {NAMESPACE_FIELD_NAME} LIKE CONCAT(?, '%')");
+
+ let namespace_rows = self
+ .fetch_rows(
+ &format!(
+ "{parent_namespaces_stmt} UNION {parent_table_namespaces_stmt}"
+ ),
+ vec![
+ Some(&self.name),
+ Some(&parent_str),
+ Some(&self.name),
+ Some(&parent_str),
+ ],
+ )
+ .await?;
+
+ let mut namespaces = Vec::<NamespaceIdent>::with_capacity(namespace_rows.len());
+
+ for row in namespace_rows {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ if nsp != parent_str {
+ namespaces.push(NamespaceIdent::from_strs(nsp.split("."))?);
+ }
+ }
+
+ Ok(namespaces)
+ } else {
+ no_such_namespace_err(parent)
+ }
+ }
+ None => {
+ let namespace_rows = self
+ .fetch_rows(
+ &format!("{namespaces_stmt} UNION {table_namespaces_stmt}"),
+ vec![Some(&self.name), Some(&self.name)],
+ )
+ .await?;
+
+ let mut namespaces = HashSet::<NamespaceIdent>::with_capacity(namespace_rows.len());
+
+ for row in namespace_rows.iter() {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ let mut levels = nsp.split(".").collect::<Vec<&str>>();
+ if !levels.is_empty() {
+ let first_level = levels.drain(..1).collect::<Vec<&str>>();
+ namespaces.insert(NamespaceIdent::from_strs(first_level)?);
+ }
+ }
+
+ Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
+ }
+ }
}
async fn create_namespace(
&self,
- _namespace: &NamespaceIdent,
- _properties: HashMap<String, String>,
+ namespace: &NamespaceIdent,
+ properties: HashMap<String, String>,
) -> Result<Namespace> {
- todo!()
+ let exists = self.namespace_exists(namespace).await?;
+
+ if exists {
+ return Err(Error::new(
+ iceberg::ErrorKind::Unexpected,
+ format!("Namespace {:?} already exists", namespace),
+ ));
+ }
+
+ for i in 1..namespace.len() {
+ let parent_namespace = NamespaceIdent::from_vec(namespace[..i].to_vec())?;
+ let parent_exists = self.namespace_exists(&parent_namespace).await?;
+ if !parent_exists {
+ return no_such_namespace_err(&parent_namespace);
+ }
+ }
+
+ let namespace_str = namespace.join(".");
+ let insert = format!(
+ "INSERT INTO {NAMESPACE_TABLE_NAME} ({CATALOG_FIELD_CATALOG_NAME}, {NAMESPACE_FIELD_NAME}, {NAMESPACE_FIELD_PROPERTY_KEY}, {NAMESPACE_FIELD_PROPERTY_VALUE})
+ VALUES (?, ?, ?, ?)");
+ if !properties.is_empty() {
+ let mut query_args = Vec::with_capacity(properties.len() * 4);
+ let mut properties_insert = insert.clone();
+ for (index, (key, value)) in properties.iter().enumerate() {
+ query_args.extend_from_slice(&[
+ Some(self.name.as_str()),
+ Some(namespace_str.as_str()),
+ Some(key.as_str()),
+ Some(value.as_str()),
+ ]);
+ if index > 0 {
+ properties_insert = format!("{properties_insert}, (?, ?, ?, ?)");
+ }
+ }
+
+ self.execute(&properties_insert, query_args, None).await?;
+
+ Ok(Namespace::with_properties(namespace.clone(), properties))
+ } else {
+ // set a default property of exists = true
+ // up for debate if this is worthwhile
+ self.execute(
+ &insert,
+ vec![
+ Some(&self.name),
+ Some(&namespace_str),
+ Some("exists"),
+ Some("true"),
+ ],
+ None,
+ )
+ .await?;
+ Ok(Namespace::with_properties(namespace.clone(), properties))
+ }
}
- async fn get_namespace(&self, _namespace: &NamespaceIdent) -> Result<Namespace> {
- todo!()
+ async fn get_namespace(&self, namespace: &NamespaceIdent) -> Result<Namespace> {
+ let exists = self.namespace_exists(namespace).await?;
+ if exists {
+ let namespace_props = self
+ .fetch_rows(
+ &format!(
+ "SELECT
+ {NAMESPACE_FIELD_NAME},
+ {NAMESPACE_FIELD_PROPERTY_KEY},
+ {NAMESPACE_FIELD_PROPERTY_VALUE}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
+ AND {NAMESPACE_FIELD_NAME} = ?"
+ ),
+ vec![Some(&self.name), Some(&namespace.join("."))],
+ )
+ .await?;
+
+ let mut properties = HashMap::with_capacity(namespace_props.len());
+
+ for row in namespace_props {
+ let key = row
+ .try_get::<String, _>(NAMESPACE_FIELD_PROPERTY_KEY)
+ .map_err(from_sqlx_error)?;
+ let value = row
+ .try_get::<String, _>(NAMESPACE_FIELD_PROPERTY_VALUE)
+ .map_err(from_sqlx_error)?;
+
+ properties.insert(key, value);
+ }
+
+ Ok(Namespace::with_properties(namespace.clone(), properties))
+ } else {
+ no_such_namespace_err(namespace)
+ }
}
- async fn namespace_exists(&self, _namespace: &NamespaceIdent) -> Result<bool> {
- todo!()
+ async fn namespace_exists(&self, namespace: &NamespaceIdent) -> Result<bool> {
+ let namespace_str = namespace.join(".");
+
+ let table_namespaces = self
+ .fetch_rows(
+ &format!(
+ "SELECT 1 FROM {CATALOG_TABLE_NAME}
Review Comment:
This is handled in the if/else on L401 - If there is no table in the table's table, we check the namespace properties table.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "callum-ryan (via GitHub)" <gi...@apache.org>.
callum-ryan commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1720764099
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,335 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ pub async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ let table_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
Review Comment:
The DISTINCT is left out on purpose as the UNION later on removes duplicates across both of the queries. I'll see if this is easier to do client-side when I make the whole function easier to read.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "callum-ryan (via GitHub)" <gi...@apache.org>.
callum-ryan commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1722173257
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,335 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ pub async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ let table_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+ let namespaces_stmt = format!(
+ "SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ match parent {
+ Some(parent) => {
+ if self.namespace_exists(parent).await? {
+ let parent_str = parent.join(".");
+ let parent_table_namespaces_stmt = format!(
+ "{table_namespaces_stmt} AND {CATALOG_FIELD_TABLE_NAMESPACE} LIKE CONCAT(?, '%')"
+ );
+ let parent_namespaces_stmt =
+ format!("{namespaces_stmt} AND {NAMESPACE_FIELD_NAME} LIKE CONCAT(?, '%')");
+
+ let namespace_rows = self
+ .fetch_rows(
+ &format!(
+ "{parent_namespaces_stmt} UNION {parent_table_namespaces_stmt}"
+ ),
+ vec![
+ Some(&self.name),
+ Some(&parent_str),
+ Some(&self.name),
+ Some(&parent_str),
+ ],
+ )
+ .await?;
+
+ let mut namespaces = Vec::<NamespaceIdent>::with_capacity(namespace_rows.len());
+
+ for row in namespace_rows {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ if nsp != parent_str {
+ namespaces.push(NamespaceIdent::from_strs(nsp.split("."))?);
+ }
+ }
+
+ Ok(namespaces)
+ } else {
+ no_such_namespace_err(parent)
+ }
+ }
+ None => {
+ let namespace_rows = self
+ .fetch_rows(
+ &format!("{namespaces_stmt} UNION {table_namespaces_stmt}"),
+ vec![Some(&self.name), Some(&self.name)],
+ )
+ .await?;
+
+ let mut namespaces = HashSet::<NamespaceIdent>::with_capacity(namespace_rows.len());
Review Comment:
This already is a set, is there something else that you are ditto'ing that I have missed?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "callum-ryan (via GitHub)" <gi...@apache.org>.
callum-ryan commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1720764034
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,335 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ pub async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ let table_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+ let namespaces_stmt = format!(
+ "SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ match parent {
+ Some(parent) => {
+ if self.namespace_exists(parent).await? {
+ let parent_str = parent.join(".");
+ let parent_table_namespaces_stmt = format!(
+ "{table_namespaces_stmt} AND {CATALOG_FIELD_TABLE_NAMESPACE} LIKE CONCAT(?, '%')"
Review Comment:
I'll take a look at refactoring this to make it easier to read.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "sdd (via GitHub)" <gi...@apache.org>.
sdd commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1714265481
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,344 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ pub async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ let table_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+ let namespaces_stmt = format!(
+ "SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ match parent {
+ Some(parent) => match self.namespace_exists(parent).await? {
Review Comment:
`match true false` is a bit unorthodox, and I always find nested `match` to add to cognitive load a bit. You could just:
```rust
match parent {
Some(parent) => if self.namespace_exists(parent).await? {
// ...
} else {
// ...
}
None => {
// ...
}
}
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "callum-ryan (via GitHub)" <gi...@apache.org>.
callum-ryan commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1731169469
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -141,21 +142,30 @@ impl SqlCatalog {
}
/// SQLX Any does not implement PostgresSQL bindings, so we have to do this.
- pub async fn execute_statement(
- &self,
- query: &String,
- args: Vec<Option<&String>>,
- ) -> Result<Vec<AnyRow>> {
- let query_with_placeholders: Cow<str> =
- if self.sql_bind_style == SqlBindStyle::DollarNumeric {
- let mut query = query.clone();
- for i in 0..args.len() {
- query = query.replacen("?", &format!("${}", i + 1), 1);
- }
- Cow::Owned(query)
- } else {
- Cow::Borrowed(query)
- };
+ fn replace_placeholders(&self, query: &str) -> String {
+ match self.sql_bind_style {
Review Comment:
the dollar-numeric bind style is postgres specific - the ? syntax is for dbs like mysql, sqlite. I am pretty sure only sqlite supports both styles.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "callum-ryan (via GitHub)" <gi...@apache.org>.
callum-ryan commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1714153256
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -141,21 +142,24 @@ impl SqlCatalog {
}
/// SQLX Any does not implement PostgresSQL bindings, so we have to do this.
- pub async fn execute_statement(
- &self,
- query: &String,
- args: Vec<Option<&String>>,
- ) -> Result<Vec<AnyRow>> {
- let query_with_placeholders: Cow<str> =
- if self.sql_bind_style == SqlBindStyle::DollarNumeric {
- let mut query = query.clone();
- for i in 0..args.len() {
- query = query.replacen("?", &format!("${}", i + 1), 1);
+ pub fn build_query(&self, query: &str) -> String {
Review Comment:
corrected
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "callum-ryan (via GitHub)" <gi...@apache.org>.
callum-ryan commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1715744495
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,344 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ pub async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ let table_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+ let namespaces_stmt = format!(
+ "SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ match parent {
+ Some(parent) => match self.namespace_exists(parent).await? {
+ true => {
+ let parent_str = parent.join(".");
+ let parent_table_namespaces_stmt = format!(
+ "{table_namespaces_stmt} AND {CATALOG_FIELD_TABLE_NAMESPACE} LIKE CONCAT(?, '%')"
+ );
+ let parent_namespaces_stmt =
+ format!("{namespaces_stmt} AND {NAMESPACE_FIELD_NAME} LIKE CONCAT(?, '%')");
+
+ let namespace_rows = self
+ .fetch_rows(
+ &format!(
+ "{parent_namespaces_stmt} UNION {parent_table_namespaces_stmt}"
+ ),
+ vec![
+ Some(&self.name),
+ Some(&parent_str),
+ Some(&self.name),
+ Some(&parent_str),
+ ],
+ )
+ .await?;
+
+ Ok(namespace_rows
+ .iter()
+ .filter_map(|r| {
+ let nsp = r.try_get::<String, _>(0).ok();
Review Comment:
I've changed this to initialise the vec outside, then loop through - which allowed me to use `?` and map the errors accordingly.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "callum-ryan (via GitHub)" <gi...@apache.org>.
callum-ryan commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1722175695
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,335 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ pub async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ let table_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
Review Comment:
I've had another look through, I could add some more comments to clear this up that the UNION will remove duplicates. I think it's far easier to have the database do the uniqueness check as opposed to code to either run 2 queries, then union on the client, or force the query to return all rows (UNION ALL) then do the unique check on the client.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "callum-ryan (via GitHub)" <gi...@apache.org>.
callum-ryan commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1730004295
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -276,4 +579,385 @@ mod tests {
new_sql_catalog(warehouse_loc.clone()).await;
new_sql_catalog(warehouse_loc.clone()).await;
}
+
+ #[tokio::test]
+ async fn test_list_namespaces_returns_empty_vector() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+
+ assert_eq!(catalog.list_namespaces(None).await.unwrap(), vec![]);
+ }
+
+ #[tokio::test]
+ async fn test_list_namespaces_returns_multiple_namespaces() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_1 = NamespaceIdent::new("a".into());
+ let namespace_ident_2 = NamespaceIdent::new("b".into());
+ create_namespaces(&catalog, &vec![&namespace_ident_1, &namespace_ident_2]).await;
+
+ assert_eq!(
+ to_set(catalog.list_namespaces(None).await.unwrap()),
+ to_set(vec![namespace_ident_1, namespace_ident_2])
+ );
+ }
+
+ #[tokio::test]
+ async fn test_list_namespaces_returns_only_top_level_namespaces() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_1 = NamespaceIdent::new("a".into());
+ let namespace_ident_2 = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+ let namespace_ident_3 = NamespaceIdent::new("b".into());
+ create_namespaces(&catalog, &vec![
+ &namespace_ident_1,
+ &namespace_ident_2,
+ &namespace_ident_3,
+ ])
+ .await;
+
+ assert_eq!(
+ to_set(catalog.list_namespaces(None).await.unwrap()),
+ to_set(vec![namespace_ident_1, namespace_ident_3])
+ );
+ }
+
+ #[tokio::test]
+ async fn test_list_namespaces_returns_no_namespaces_under_parent() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_1 = NamespaceIdent::new("a".into());
+ let namespace_ident_2 = NamespaceIdent::new("b".into());
+ create_namespaces(&catalog, &vec![&namespace_ident_1, &namespace_ident_2]).await;
+
+ assert_eq!(
+ catalog
+ .list_namespaces(Some(&namespace_ident_1))
+ .await
+ .unwrap(),
+ vec![]
+ );
+ }
+
+ #[tokio::test]
+ async fn test_list_namespaces_returns_namespace_under_parent() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_1 = NamespaceIdent::new("a".into());
+ let namespace_ident_2 = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+ let namespace_ident_3 = NamespaceIdent::new("c".into());
+ create_namespaces(&catalog, &vec![
+ &namespace_ident_1,
+ &namespace_ident_2,
+ &namespace_ident_3,
+ ])
+ .await;
+
+ assert_eq!(
+ to_set(catalog.list_namespaces(None).await.unwrap()),
+ to_set(vec![namespace_ident_1.clone(), namespace_ident_3])
+ );
+
+ assert_eq!(
+ catalog
+ .list_namespaces(Some(&namespace_ident_1))
+ .await
+ .unwrap(),
+ vec![NamespaceIdent::from_strs(vec!["a", "b"]).unwrap()]
+ );
+ }
+
+ #[tokio::test]
+ async fn test_list_namespaces_returns_multiple_namespaces_under_parent() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_1 = NamespaceIdent::new("a".to_string());
+ let namespace_ident_2 = NamespaceIdent::from_strs(vec!["a", "a"]).unwrap();
+ let namespace_ident_3 = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+ let namespace_ident_4 = NamespaceIdent::from_strs(vec!["a", "c"]).unwrap();
+ let namespace_ident_5 = NamespaceIdent::new("b".into());
+ create_namespaces(&catalog, &vec![
+ &namespace_ident_1,
+ &namespace_ident_2,
+ &namespace_ident_3,
+ &namespace_ident_4,
+ &namespace_ident_5,
+ ])
+ .await;
+
+ assert_eq!(
+ to_set(
+ catalog
+ .list_namespaces(Some(&namespace_ident_1))
+ .await
+ .unwrap()
+ ),
+ to_set(vec![
+ NamespaceIdent::from_strs(vec!["a", "a"]).unwrap(),
+ NamespaceIdent::from_strs(vec!["a", "b"]).unwrap(),
+ NamespaceIdent::from_strs(vec!["a", "c"]).unwrap(),
+ ])
+ );
+ }
+
+ #[tokio::test]
+ async fn test_namespace_exists_returns_false() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident = NamespaceIdent::new("a".into());
+ create_namespace(&catalog, &namespace_ident).await;
+
+ assert!(!catalog
+ .namespace_exists(&NamespaceIdent::new("b".into()))
+ .await
+ .unwrap());
+ }
+
+ #[tokio::test]
+ async fn test_namespace_exists_returns_true() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident = NamespaceIdent::new("a".into());
+ create_namespace(&catalog, &namespace_ident).await;
+
+ assert!(catalog.namespace_exists(&namespace_ident).await.unwrap());
+ }
+
+ #[tokio::test]
+ async fn test_create_namespace_with_properties() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident = NamespaceIdent::new("abc".into());
+
+ let mut properties: HashMap<String, String> = HashMap::new();
+ properties.insert("k".into(), "v".into());
+
+ assert_eq!(
+ catalog
+ .create_namespace(&namespace_ident, properties.clone())
+ .await
+ .unwrap(),
+ Namespace::with_properties(namespace_ident.clone(), properties.clone())
+ );
+
+ assert_eq!(
+ catalog.get_namespace(&namespace_ident).await.unwrap(),
+ Namespace::with_properties(namespace_ident, properties)
+ );
+ }
+
+ #[tokio::test]
+ async fn test_create_namespace_throws_error_if_namespace_already_exists() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident = NamespaceIdent::new("a".into());
+ create_namespace(&catalog, &namespace_ident).await;
+
+ assert_eq!(
+ catalog
+ .create_namespace(&namespace_ident, HashMap::new())
+ .await
+ .unwrap_err()
+ .to_string(),
+ format!(
+ "Unexpected => Namespace {:?} already exists",
+ &namespace_ident
+ )
+ );
+
+ assert_eq!(
+ catalog.get_namespace(&namespace_ident).await.unwrap(),
+ Namespace::with_properties(namespace_ident, default_properties())
+ );
+ }
+
+ #[tokio::test]
+ async fn test_create_nested_namespace() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let parent_namespace_ident = NamespaceIdent::new("a".into());
+ create_namespace(&catalog, &parent_namespace_ident).await;
+
+ let child_namespace_ident = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+
+ assert_eq!(
+ catalog
+ .create_namespace(&child_namespace_ident, HashMap::new())
+ .await
+ .unwrap(),
+ Namespace::new(child_namespace_ident.clone())
+ );
+
+ assert_eq!(
+ catalog.get_namespace(&child_namespace_ident).await.unwrap(),
+ Namespace::with_properties(child_namespace_ident, default_properties())
+ );
+ }
+
+ #[tokio::test]
+ async fn test_create_deeply_nested_namespace() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_a = NamespaceIdent::new("a".into());
+ let namespace_ident_a_b = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+ create_namespaces(&catalog, &vec![&namespace_ident_a, &namespace_ident_a_b]).await;
+
+ let namespace_ident_a_b_c = NamespaceIdent::from_strs(vec!["a", "b", "c"]).unwrap();
+
+ assert_eq!(
+ catalog
+ .create_namespace(&namespace_ident_a_b_c, HashMap::new())
+ .await
+ .unwrap(),
+ Namespace::new(namespace_ident_a_b_c.clone())
+ );
+
+ assert_eq!(
+ catalog.get_namespace(&namespace_ident_a_b_c).await.unwrap(),
+ Namespace::with_properties(namespace_ident_a_b_c, default_properties())
+ );
+ }
+
+ #[tokio::test]
+ async fn test_create_nested_namespace_throws_error_if_top_level_namespace_doesnt_exist() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+
+ let nested_namespace_ident = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+
+ assert_eq!(
+ catalog
+ .create_namespace(&nested_namespace_ident, HashMap::new())
+ .await
+ .unwrap_err()
+ .to_string(),
+ format!(
+ "Unexpected => No such namespace: {:?}",
+ NamespaceIdent::new("a".into())
+ )
+ );
+
+ assert_eq!(catalog.list_namespaces(None).await.unwrap(), vec![]);
+ }
+
+ #[tokio::test]
+ async fn test_drop_namespace() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident = NamespaceIdent::new("abc".into());
+ create_namespace(&catalog, &namespace_ident).await;
+
+ catalog.drop_namespace(&namespace_ident).await.unwrap();
+
+ assert!(!catalog.namespace_exists(&namespace_ident).await.unwrap())
+ }
+
+ #[tokio::test]
+ async fn test_drop_nested_namespace() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_a = NamespaceIdent::new("a".into());
+ let namespace_ident_a_b = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+ create_namespaces(&catalog, &vec![&namespace_ident_a, &namespace_ident_a_b]).await;
+
+ catalog.drop_namespace(&namespace_ident_a_b).await.unwrap();
+
+ assert!(!catalog
+ .namespace_exists(&namespace_ident_a_b)
+ .await
+ .unwrap());
+
+ assert!(catalog.namespace_exists(&namespace_ident_a).await.unwrap());
+ }
+
+ #[tokio::test]
+ async fn test_drop_deeply_nested_namespace() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_a = NamespaceIdent::new("a".into());
+ let namespace_ident_a_b = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+ let namespace_ident_a_b_c = NamespaceIdent::from_strs(vec!["a", "b", "c"]).unwrap();
+ create_namespaces(&catalog, &vec![
+ &namespace_ident_a,
+ &namespace_ident_a_b,
+ &namespace_ident_a_b_c,
+ ])
+ .await;
+
+ catalog
+ .drop_namespace(&namespace_ident_a_b_c)
+ .await
+ .unwrap();
+
+ assert!(!catalog
+ .namespace_exists(&namespace_ident_a_b_c)
+ .await
+ .unwrap());
+
+ assert!(catalog
+ .namespace_exists(&namespace_ident_a_b)
+ .await
+ .unwrap());
+
+ assert!(catalog.namespace_exists(&namespace_ident_a).await.unwrap());
+ }
+
+ #[tokio::test]
+ async fn test_drop_namespace_throws_error_if_namespace_doesnt_exist() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+
+ let non_existent_namespace_ident = NamespaceIdent::new("abc".into());
+ assert_eq!(
+ catalog
+ .drop_namespace(&non_existent_namespace_ident)
+ .await
+ .unwrap_err()
+ .to_string(),
+ format!(
+ "Unexpected => No such namespace: {:?}",
+ non_existent_namespace_ident
+ )
+ )
+ }
+
+ #[tokio::test]
+ async fn test_drop_namespace_throws_error_if_nested_namespace_doesnt_exist() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ create_namespace(&catalog, &NamespaceIdent::new("a".into())).await;
+
+ let non_existent_namespace_ident =
+ NamespaceIdent::from_vec(vec!["a".into(), "b".into()]).unwrap();
+ assert_eq!(
+ catalog
+ .drop_namespace(&non_existent_namespace_ident)
+ .await
+ .unwrap_err()
+ .to_string(),
+ format!(
+ "Unexpected => No such namespace: {:?}",
+ non_existent_namespace_ident
+ )
+ )
+ }
+
+ #[tokio::test]
+ #[ignore = "Java/Python do not drop nested namespaces?"]
+ async fn test_dropping_a_namespace_also_drops_namespaces_nested_under_that_one() {
Review Comment:
I will add this when I re-add the drop_namespaces function in the next PR for tables. I'll check if the namespace is empty for both tables, and child namespaces.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "callum-ryan (via GitHub)" <gi...@apache.org>.
callum-ryan commented on PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#issuecomment-2284572219
> Hi, I need more time to review the SQL individually. Here are some additional comments that I believe should take priority.
>
> By the way, have you considered using `diesel` instead? I find it easier to maintain compared to raw SQL.
Personal preference but I tend to avoid ORMs - I can see that Java has gone the raw SQL way, and Python uses sqlalchemy's ORM. Happy to put this up for discussion.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "callum-ryan (via GitHub)" <gi...@apache.org>.
callum-ryan commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1740062315
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,312 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ // UNION will remove duplicates.
+ let all_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
+ UNION
+ SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ let namespace_rows = self
+ .fetch_rows(&all_namespaces_stmt, vec![
+ Some(&self.name),
+ Some(&self.name),
+ ])
+ .await?;
+
+ let mut namespaces = HashSet::<NamespaceIdent>::with_capacity(namespace_rows.len());
+
+ if let Some(parent) = parent {
+ if self.namespace_exists(parent).await? {
+ let parent_str = parent.join(".");
+
+ for row in namespace_rows.iter() {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ // if parent = a, then we only want to see a.b, a.c returned.
+ if nsp != parent_str && nsp.starts_with(&parent_str) {
+ namespaces.insert(NamespaceIdent::from_strs(nsp.split("."))?);
+ }
+ }
+
+ Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
+ } else {
+ no_such_namespace_err(parent)
+ }
+ } else {
+ for row in namespace_rows.iter() {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ let mut levels = nsp.split(".").collect::<Vec<&str>>();
+ if !levels.is_empty() {
+ let first_level = levels.drain(..1).collect::<Vec<&str>>();
+ namespaces.insert(NamespaceIdent::from_strs(first_level)?);
+ }
+ }
+
+ Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
+ }
}
async fn create_namespace(
&self,
- _namespace: &NamespaceIdent,
- _properties: HashMap<String, String>,
+ namespace: &NamespaceIdent,
+ properties: HashMap<String, String>,
) -> Result<Namespace> {
- todo!()
+ let exists = self.namespace_exists(namespace).await?;
+
+ if exists {
+ return Err(Error::new(
+ iceberg::ErrorKind::Unexpected,
+ format!("Namespace {:?} already exists", namespace),
+ ));
+ }
+
+ for i in 1..namespace.len() {
Review Comment:
@liurenjie1024 I asked in the dev slack chat and you are correct - we should support the reference implementation method. So I have removed this check.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#issuecomment-2330489500
> I wonder if its possible to expose this as Python API and also test against the python unit tests
Hi, @kevinjqliu I'm fine with adding it in pyiceberg_core module. But instead of adding sql catalog only, I think it's better to add a wrapper around `Catalog` trait? What do you think?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "callum-ryan (via GitHub)" <gi...@apache.org>.
callum-ryan commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1722169292
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -141,21 +142,30 @@ impl SqlCatalog {
}
/// SQLX Any does not implement PostgresSQL bindings, so we have to do this.
- pub async fn execute_statement(
- &self,
- query: &String,
- args: Vec<Option<&String>>,
- ) -> Result<Vec<AnyRow>> {
- let query_with_placeholders: Cow<str> =
- if self.sql_bind_style == SqlBindStyle::DollarNumeric {
- let mut query = query.clone();
- for i in 0..args.len() {
- query = query.replacen("?", &format!("${}", i + 1), 1);
- }
- Cow::Owned(query)
- } else {
- Cow::Borrowed(query)
- };
+ fn replace_placeholders(&self, query: &str) -> String {
+ match self.sql_bind_style {
+ SqlBindStyle::DollarNumeric => {
+ let mut count = 1;
+ query
+ .chars()
+ .fold(String::with_capacity(query.len()), |mut acc, c| {
+ if c == '?' {
+ acc.push('$');
+ acc.push_str(&count.to_string());
+ count += 1;
+ } else {
+ acc.push(c);
+ }
+ acc
+ })
+ }
+ _ => query.to_owned(),
+ }
+ }
+
+ /// Fetch a vec of AnyRows from a given query
+ pub async fn fetch_rows(&self, query: &str, args: Vec<Option<&str>>) -> Result<Vec<AnyRow>> {
Review Comment:
removed!
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,335 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ pub async fn execute(
Review Comment:
removed
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "sdd (via GitHub)" <gi...@apache.org>.
sdd commented on PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#issuecomment-2284722772
I'm also ok with this being SQL rather than Diesel. Diesel feels like a large dependency to bring in for a small number of quite straightforward queries.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "callum-ryan (via GitHub)" <gi...@apache.org>.
callum-ryan commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1715743769
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,344 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ pub async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ let table_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+ let namespaces_stmt = format!(
+ "SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ match parent {
+ Some(parent) => match self.namespace_exists(parent).await? {
Review Comment:
thanks, not entirely sure why I did this considering you had a similar comment on the big old PR!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "callum-ryan (via GitHub)" <gi...@apache.org>.
callum-ryan commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1725559714
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,312 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ // UNION will remove duplicates.
+ let all_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
+ UNION
+ SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ let namespace_rows = self
+ .fetch_rows(&all_namespaces_stmt, vec![
+ Some(&self.name),
+ Some(&self.name),
+ ])
+ .await?;
+
+ let mut namespaces = HashSet::<NamespaceIdent>::with_capacity(namespace_rows.len());
+
+ if let Some(parent) = parent {
+ if self.namespace_exists(parent).await? {
+ let parent_str = parent.join(".");
+
+ for row in namespace_rows.iter() {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ // if parent = a, then we only want to see a.b, a.c returned.
+ if nsp != parent_str && nsp.starts_with(&parent_str) {
+ namespaces.insert(NamespaceIdent::from_strs(nsp.split("."))?);
+ }
+ }
+
+ Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
+ } else {
+ no_such_namespace_err(parent)
+ }
+ } else {
+ for row in namespace_rows.iter() {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ let mut levels = nsp.split(".").collect::<Vec<&str>>();
+ if !levels.is_empty() {
+ let first_level = levels.drain(..1).collect::<Vec<&str>>();
+ namespaces.insert(NamespaceIdent::from_strs(first_level)?);
+ }
+ }
+
+ Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
+ }
}
async fn create_namespace(
&self,
- _namespace: &NamespaceIdent,
- _properties: HashMap<String, String>,
+ namespace: &NamespaceIdent,
+ properties: HashMap<String, String>,
) -> Result<Namespace> {
- todo!()
+ let exists = self.namespace_exists(namespace).await?;
+
+ if exists {
+ return Err(Error::new(
+ iceberg::ErrorKind::Unexpected,
+ format!("Namespace {:?} already exists", namespace),
+ ));
+ }
+
+ for i in 1..namespace.len() {
+ let parent_namespace = NamespaceIdent::from_vec(namespace[..i].to_vec())?;
+ let parent_exists = self.namespace_exists(&parent_namespace).await?;
+ if !parent_exists {
+ return no_such_namespace_err(&parent_namespace);
+ }
+ }
+
+ let namespace_str = namespace.join(".");
+ let insert = format!(
+ "INSERT INTO {NAMESPACE_TABLE_NAME} ({CATALOG_FIELD_CATALOG_NAME}, {NAMESPACE_FIELD_NAME}, {NAMESPACE_FIELD_PROPERTY_KEY}, {NAMESPACE_FIELD_PROPERTY_VALUE})
+ VALUES (?, ?, ?, ?)");
+ if !properties.is_empty() {
+ let mut query_args = Vec::with_capacity(properties.len() * 4);
+ let mut properties_insert = insert.clone();
+ for (index, (key, value)) in properties.iter().enumerate() {
+ query_args.extend_from_slice(&[
+ Some(self.name.as_str()),
+ Some(namespace_str.as_str()),
+ Some(key.as_str()),
+ Some(value.as_str()),
+ ]);
+ if index > 0 {
+ properties_insert = format!("{properties_insert}, (?, ?, ?, ?)");
Review Comment:
I have implemented this as a `push_str` which will extend the original allocation of `insert_stmt` instead of allocating again and again.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "callum-ryan (via GitHub)" <gi...@apache.org>.
callum-ryan commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1725533714
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,312 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ // UNION will remove duplicates.
+ let all_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
+ UNION
+ SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ let namespace_rows = self
+ .fetch_rows(&all_namespaces_stmt, vec![
+ Some(&self.name),
+ Some(&self.name),
+ ])
+ .await?;
+
+ let mut namespaces = HashSet::<NamespaceIdent>::with_capacity(namespace_rows.len());
+
+ if let Some(parent) = parent {
+ if self.namespace_exists(parent).await? {
+ let parent_str = parent.join(".");
+
+ for row in namespace_rows.iter() {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ // if parent = a, then we only want to see a.b, a.c returned.
+ if nsp != parent_str && nsp.starts_with(&parent_str) {
+ namespaces.insert(NamespaceIdent::from_strs(nsp.split("."))?);
+ }
+ }
+
+ Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
+ } else {
+ no_such_namespace_err(parent)
+ }
+ } else {
+ for row in namespace_rows.iter() {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ let mut levels = nsp.split(".").collect::<Vec<&str>>();
+ if !levels.is_empty() {
+ let first_level = levels.drain(..1).collect::<Vec<&str>>();
+ namespaces.insert(NamespaceIdent::from_strs(first_level)?);
+ }
+ }
+
+ Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
+ }
}
async fn create_namespace(
&self,
- _namespace: &NamespaceIdent,
- _properties: HashMap<String, String>,
+ namespace: &NamespaceIdent,
+ properties: HashMap<String, String>,
) -> Result<Namespace> {
- todo!()
+ let exists = self.namespace_exists(namespace).await?;
+
+ if exists {
+ return Err(Error::new(
+ iceberg::ErrorKind::Unexpected,
+ format!("Namespace {:?} already exists", namespace),
+ ));
+ }
+
+ for i in 1..namespace.len() {
Review Comment:
I agree that Java/Python do not implement this, but it seems nonsensical to me to be able to create a namespace under one that does not already exist? If you want to create a namespace `a.b` then surely we should be checking if namespace `a` actually exists?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1730525891
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,312 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ // UNION will remove duplicates.
+ let all_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
+ UNION
+ SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ let namespace_rows = self
+ .fetch_rows(&all_namespaces_stmt, vec![
+ Some(&self.name),
+ Some(&self.name),
+ ])
+ .await?;
+
+ let mut namespaces = HashSet::<NamespaceIdent>::with_capacity(namespace_rows.len());
+
+ if let Some(parent) = parent {
+ if self.namespace_exists(parent).await? {
+ let parent_str = parent.join(".");
+
+ for row in namespace_rows.iter() {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ // if parent = a, then we only want to see a.b, a.c returned.
+ if nsp != parent_str && nsp.starts_with(&parent_str) {
+ namespaces.insert(NamespaceIdent::from_strs(nsp.split("."))?);
+ }
+ }
+
+ Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
+ } else {
+ no_such_namespace_err(parent)
+ }
+ } else {
+ for row in namespace_rows.iter() {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ let mut levels = nsp.split(".").collect::<Vec<&str>>();
+ if !levels.is_empty() {
+ let first_level = levels.drain(..1).collect::<Vec<&str>>();
+ namespaces.insert(NamespaceIdent::from_strs(first_level)?);
+ }
+ }
+
+ Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
+ }
}
async fn create_namespace(
&self,
- _namespace: &NamespaceIdent,
- _properties: HashMap<String, String>,
+ namespace: &NamespaceIdent,
+ properties: HashMap<String, String>,
) -> Result<Namespace> {
- todo!()
+ let exists = self.namespace_exists(namespace).await?;
+
+ if exists {
+ return Err(Error::new(
+ iceberg::ErrorKind::Unexpected,
+ format!("Namespace {:?} already exists", namespace),
+ ));
+ }
+
+ for i in 1..namespace.len() {
Review Comment:
Sorry, but I'm not convinced. When we create namespace `a.b` without `a` already created, we could treat `a` as an empty namespace without any property, and this is also how [namespaceExists](https://github.com/apache/iceberg/blob/2b21020aedb63c26295005d150c05f0a5a5f0eb2/core/src/main/java/org/apache/iceberg/jdbc/JdbcUtil.java#L784) work.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1720598113
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,335 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ pub async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ let table_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+ let namespaces_stmt = format!(
+ "SELECT {NAMESPACE_FIELD_NAME}
Review Comment:
Ditto.
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,335 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ pub async fn execute(
Review Comment:
Ditto.
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,335 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ pub async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ let table_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
Review Comment:
We need to add `DISTINCT` here.
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,335 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ pub async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ let table_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+ let namespaces_stmt = format!(
+ "SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ match parent {
+ Some(parent) => {
+ if self.namespace_exists(parent).await? {
+ let parent_str = parent.join(".");
+ let parent_table_namespaces_stmt = format!(
+ "{table_namespaces_stmt} AND {CATALOG_FIELD_TABLE_NAMESPACE} LIKE CONCAT(?, '%')"
+ );
+ let parent_namespaces_stmt =
+ format!("{namespaces_stmt} AND {NAMESPACE_FIELD_NAME} LIKE CONCAT(?, '%')");
+
+ let namespace_rows = self
+ .fetch_rows(
+ &format!(
+ "{parent_namespaces_stmt} UNION {parent_table_namespaces_stmt}"
+ ),
+ vec![
+ Some(&self.name),
+ Some(&parent_str),
+ Some(&self.name),
+ Some(&parent_str),
+ ],
+ )
+ .await?;
+
+ let mut namespaces = Vec::<NamespaceIdent>::with_capacity(namespace_rows.len());
Review Comment:
This should be a set.
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,335 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ pub async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ let table_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+ let namespaces_stmt = format!(
+ "SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ match parent {
+ Some(parent) => {
+ if self.namespace_exists(parent).await? {
+ let parent_str = parent.join(".");
+ let parent_table_namespaces_stmt = format!(
+ "{table_namespaces_stmt} AND {CATALOG_FIELD_TABLE_NAMESPACE} LIKE CONCAT(?, '%')"
+ );
+ let parent_namespaces_stmt =
+ format!("{namespaces_stmt} AND {NAMESPACE_FIELD_NAME} LIKE CONCAT(?, '%')");
+
+ let namespace_rows = self
+ .fetch_rows(
+ &format!(
+ "{parent_namespaces_stmt} UNION {parent_table_namespaces_stmt}"
+ ),
+ vec![
+ Some(&self.name),
+ Some(&parent_str),
+ Some(&self.name),
+ Some(&parent_str),
+ ],
+ )
+ .await?;
+
+ let mut namespaces = Vec::<NamespaceIdent>::with_capacity(namespace_rows.len());
+
+ for row in namespace_rows {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ if nsp != parent_str {
+ namespaces.push(NamespaceIdent::from_strs(nsp.split("."))?);
+ }
+ }
+
+ Ok(namespaces)
+ } else {
+ no_such_namespace_err(parent)
+ }
+ }
+ None => {
+ let namespace_rows = self
+ .fetch_rows(
+ &format!("{namespaces_stmt} UNION {table_namespaces_stmt}"),
+ vec![Some(&self.name), Some(&self.name)],
+ )
+ .await?;
+
+ let mut namespaces = HashSet::<NamespaceIdent>::with_capacity(namespace_rows.len());
+
+ for row in namespace_rows.iter() {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ let mut levels = nsp.split(".").collect::<Vec<&str>>();
+ if !levels.is_empty() {
+ let first_level = levels.drain(..1).collect::<Vec<&str>>();
+ namespaces.insert(NamespaceIdent::from_strs(first_level)?);
+ }
+ }
+
+ Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
+ }
+ }
}
async fn create_namespace(
&self,
- _namespace: &NamespaceIdent,
- _properties: HashMap<String, String>,
+ namespace: &NamespaceIdent,
+ properties: HashMap<String, String>,
) -> Result<Namespace> {
- todo!()
+ let exists = self.namespace_exists(namespace).await?;
+
+ if exists {
+ return Err(Error::new(
+ iceberg::ErrorKind::Unexpected,
+ format!("Namespace {:?} already exists", namespace),
+ ));
+ }
+
+ for i in 1..namespace.len() {
+ let parent_namespace = NamespaceIdent::from_vec(namespace[..i].to_vec())?;
+ let parent_exists = self.namespace_exists(&parent_namespace).await?;
+ if !parent_exists {
+ return no_such_namespace_err(&parent_namespace);
+ }
+ }
+
+ let namespace_str = namespace.join(".");
+ let insert = format!(
+ "INSERT INTO {NAMESPACE_TABLE_NAME} ({CATALOG_FIELD_CATALOG_NAME}, {NAMESPACE_FIELD_NAME}, {NAMESPACE_FIELD_PROPERTY_KEY}, {NAMESPACE_FIELD_PROPERTY_VALUE})
+ VALUES (?, ?, ?, ?)");
+ if !properties.is_empty() {
+ let mut query_args = Vec::with_capacity(properties.len() * 4);
+ let mut properties_insert = insert.clone();
+ for (index, (key, value)) in properties.iter().enumerate() {
+ query_args.extend_from_slice(&[
+ Some(self.name.as_str()),
+ Some(namespace_str.as_str()),
+ Some(key.as_str()),
+ Some(value.as_str()),
+ ]);
+ if index > 0 {
+ properties_insert = format!("{properties_insert}, (?, ?, ?, ?)");
+ }
+ }
+
+ self.execute(&properties_insert, query_args, None).await?;
+
+ Ok(Namespace::with_properties(namespace.clone(), properties))
+ } else {
+ // set a default property of exists = true
+ // up for debate if this is worthwhile
+ self.execute(
+ &insert,
+ vec![
+ Some(&self.name),
+ Some(&namespace_str),
+ Some("exists"),
+ Some("true"),
+ ],
+ None,
+ )
+ .await?;
+ Ok(Namespace::with_properties(namespace.clone(), properties))
+ }
}
- async fn get_namespace(&self, _namespace: &NamespaceIdent) -> Result<Namespace> {
- todo!()
+ async fn get_namespace(&self, namespace: &NamespaceIdent) -> Result<Namespace> {
+ let exists = self.namespace_exists(namespace).await?;
+ if exists {
+ let namespace_props = self
+ .fetch_rows(
+ &format!(
+ "SELECT
+ {NAMESPACE_FIELD_NAME},
+ {NAMESPACE_FIELD_PROPERTY_KEY},
+ {NAMESPACE_FIELD_PROPERTY_VALUE}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
+ AND {NAMESPACE_FIELD_NAME} = ?"
+ ),
+ vec![Some(&self.name), Some(&namespace.join("."))],
+ )
+ .await?;
+
+ let mut properties = HashMap::with_capacity(namespace_props.len());
+
+ for row in namespace_props {
+ let key = row
+ .try_get::<String, _>(NAMESPACE_FIELD_PROPERTY_KEY)
+ .map_err(from_sqlx_error)?;
+ let value = row
+ .try_get::<String, _>(NAMESPACE_FIELD_PROPERTY_VALUE)
+ .map_err(from_sqlx_error)?;
+
+ properties.insert(key, value);
+ }
+
+ Ok(Namespace::with_properties(namespace.clone(), properties))
+ } else {
+ no_such_namespace_err(namespace)
+ }
}
- async fn namespace_exists(&self, _namespace: &NamespaceIdent) -> Result<bool> {
- todo!()
+ async fn namespace_exists(&self, namespace: &NamespaceIdent) -> Result<bool> {
+ let namespace_str = namespace.join(".");
+
+ let table_namespaces = self
+ .fetch_rows(
+ &format!(
+ "SELECT 1 FROM {CATALOG_TABLE_NAME}
Review Comment:
I think we also need to check namespace property table? It's possible that a namespace has property but has not table created yet.
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,335 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ pub async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ let table_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+ let namespaces_stmt = format!(
+ "SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ match parent {
+ Some(parent) => {
+ if self.namespace_exists(parent).await? {
+ let parent_str = parent.join(".");
+ let parent_table_namespaces_stmt = format!(
+ "{table_namespaces_stmt} AND {CATALOG_FIELD_TABLE_NAMESPACE} LIKE CONCAT(?, '%')"
Review Comment:
To be honest, I feel this is a little difficult to read. I can understand that you are trying to remove duplicates, but sometimes a little duplicate is worth it if it improves readability.
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -141,21 +142,30 @@ impl SqlCatalog {
}
/// SQLX Any does not implement PostgresSQL bindings, so we have to do this.
- pub async fn execute_statement(
- &self,
- query: &String,
- args: Vec<Option<&String>>,
- ) -> Result<Vec<AnyRow>> {
- let query_with_placeholders: Cow<str> =
- if self.sql_bind_style == SqlBindStyle::DollarNumeric {
- let mut query = query.clone();
- for i in 0..args.len() {
- query = query.replacen("?", &format!("${}", i + 1), 1);
- }
- Cow::Owned(query)
- } else {
- Cow::Borrowed(query)
- };
+ fn replace_placeholders(&self, query: &str) -> String {
+ match self.sql_bind_style {
+ SqlBindStyle::DollarNumeric => {
+ let mut count = 1;
+ query
+ .chars()
+ .fold(String::with_capacity(query.len()), |mut acc, c| {
+ if c == '?' {
+ acc.push('$');
+ acc.push_str(&count.to_string());
+ count += 1;
+ } else {
+ acc.push(c);
+ }
+ acc
+ })
+ }
+ _ => query.to_owned(),
+ }
+ }
+
+ /// Fetch a vec of AnyRows from a given query
+ pub async fn fetch_rows(&self, query: &str, args: Vec<Option<&str>>) -> Result<Vec<AnyRow>> {
Review Comment:
Do we need to make this public?
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,335 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ pub async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ let table_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+ let namespaces_stmt = format!(
+ "SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ match parent {
+ Some(parent) => {
+ if self.namespace_exists(parent).await? {
+ let parent_str = parent.join(".");
+ let parent_table_namespaces_stmt = format!(
+ "{table_namespaces_stmt} AND {CATALOG_FIELD_TABLE_NAMESPACE} LIKE CONCAT(?, '%')"
+ );
+ let parent_namespaces_stmt =
+ format!("{namespaces_stmt} AND {NAMESPACE_FIELD_NAME} LIKE CONCAT(?, '%')");
+
+ let namespace_rows = self
+ .fetch_rows(
+ &format!(
+ "{parent_namespaces_stmt} UNION {parent_table_namespaces_stmt}"
+ ),
+ vec![
+ Some(&self.name),
+ Some(&parent_str),
+ Some(&self.name),
+ Some(&parent_str),
+ ],
+ )
+ .await?;
+
+ let mut namespaces = Vec::<NamespaceIdent>::with_capacity(namespace_rows.len());
+
+ for row in namespace_rows {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ if nsp != parent_str {
Review Comment:
Should this be `starts_with`?
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,335 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ pub async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ let table_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+ let namespaces_stmt = format!(
+ "SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ match parent {
+ Some(parent) => {
+ if self.namespace_exists(parent).await? {
+ let parent_str = parent.join(".");
+ let parent_table_namespaces_stmt = format!(
+ "{table_namespaces_stmt} AND {CATALOG_FIELD_TABLE_NAMESPACE} LIKE CONCAT(?, '%')"
+ );
+ let parent_namespaces_stmt =
+ format!("{namespaces_stmt} AND {NAMESPACE_FIELD_NAME} LIKE CONCAT(?, '%')");
+
+ let namespace_rows = self
+ .fetch_rows(
+ &format!(
+ "{parent_namespaces_stmt} UNION {parent_table_namespaces_stmt}"
+ ),
+ vec![
+ Some(&self.name),
+ Some(&parent_str),
+ Some(&self.name),
+ Some(&parent_str),
+ ],
+ )
+ .await?;
+
+ let mut namespaces = Vec::<NamespaceIdent>::with_capacity(namespace_rows.len());
+
+ for row in namespace_rows {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ if nsp != parent_str {
+ namespaces.push(NamespaceIdent::from_strs(nsp.split("."))?);
+ }
+ }
+
+ Ok(namespaces)
+ } else {
+ no_such_namespace_err(parent)
+ }
+ }
+ None => {
+ let namespace_rows = self
+ .fetch_rows(
+ &format!("{namespaces_stmt} UNION {table_namespaces_stmt}"),
+ vec![Some(&self.name), Some(&self.name)],
+ )
+ .await?;
+
+ let mut namespaces = HashSet::<NamespaceIdent>::with_capacity(namespace_rows.len());
Review Comment:
Ditto.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "sdd (via GitHub)" <gi...@apache.org>.
sdd commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1714292148
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,344 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ pub async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ let table_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+ let namespaces_stmt = format!(
+ "SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ match parent {
+ Some(parent) => match self.namespace_exists(parent).await? {
+ true => {
+ let parent_str = parent.join(".");
+ let parent_table_namespaces_stmt = format!(
+ "{table_namespaces_stmt} AND {CATALOG_FIELD_TABLE_NAMESPACE} LIKE CONCAT(?, '%')"
+ );
+ let parent_namespaces_stmt =
+ format!("{namespaces_stmt} AND {NAMESPACE_FIELD_NAME} LIKE CONCAT(?, '%')");
+
+ let namespace_rows = self
+ .fetch_rows(
+ &format!(
+ "{parent_namespaces_stmt} UNION {parent_table_namespaces_stmt}"
+ ),
+ vec![
+ Some(&self.name),
+ Some(&parent_str),
+ Some(&self.name),
+ Some(&parent_str),
+ ],
+ )
+ .await?;
+
+ Ok(namespace_rows
+ .iter()
+ .filter_map(|r| {
+ let nsp = r.try_get::<String, _>(0).ok();
+ nsp.and_then(|n| {
+ if n == parent_str {
+ // Filter out itself
+ None
+ } else {
+ NamespaceIdent::from_strs(n.split(".")).ok()
+ }
+ })
+ })
+ .collect())
+ }
+ false => no_such_namespace_err(parent),
+ },
+ None => {
+ let namespace_rows = self
+ .fetch_rows(
+ &format!("{namespaces_stmt} UNION {table_namespaces_stmt}"),
+ vec![Some(&self.name), Some(&self.name)],
+ )
+ .await?;
+
+ Ok(namespace_rows
+ .iter()
+ .filter_map(|r| {
+ let nsp = r.try_get::<String, _>(0).ok();
+ nsp.and_then(|n| {
+ // for each row, split a.b.c into a, b, c levels
+ let mut levels = n.split(".").collect::<Vec<&str>>();
+ if !levels.is_empty() {
+ // only return first-level idents
+ let first_level = levels.drain(..1).collect::<Vec<&str>>();
+ NamespaceIdent::from_strs(first_level).ok()
+ } else {
+ None
+ }
+ })
+ })
+ .collect::<HashSet<NamespaceIdent>>()
+ .into_iter()
+ .collect::<Vec<NamespaceIdent>>())
+ }
+ }
}
async fn create_namespace(
&self,
- _namespace: &NamespaceIdent,
- _properties: HashMap<String, String>,
+ namespace: &NamespaceIdent,
+ properties: HashMap<String, String>,
) -> Result<Namespace> {
- todo!()
+ let exists = self.namespace_exists(namespace).await?;
+
+ if exists {
+ return Err(Error::new(
+ iceberg::ErrorKind::Unexpected,
+ format!("Namespace {:?} already exists", namespace),
+ ));
+ }
+
+ for i in 1..namespace.len() {
+ let parent_namespace = NamespaceIdent::from_vec(namespace[..i].to_vec())?;
+ let parent_exists = self.namespace_exists(&parent_namespace).await?;
+ if !parent_exists {
+ return no_such_namespace_err(&parent_namespace);
+ }
+ }
+
+ let namespace_str = namespace.join(".");
+ let insert = format!(
+ "INSERT INTO {NAMESPACE_TABLE_NAME} ({CATALOG_FIELD_CATALOG_NAME}, {NAMESPACE_FIELD_NAME}, {NAMESPACE_FIELD_PROPERTY_KEY}, {NAMESPACE_FIELD_PROPERTY_VALUE})
+ VALUES (?, ?, ?, ?)");
+ if !properties.is_empty() {
+ let mut query_args = Vec::with_capacity(properties.len() * 4);
+ let mut properties_insert = insert.clone();
+ for (index, (key, value)) in properties.iter().enumerate() {
+ query_args.extend_from_slice(&[
+ Some(self.name.as_str()),
+ Some(namespace_str.as_str()),
+ Some(key.as_str()),
+ Some(value.as_str()),
+ ]);
+ if index > 0 {
+ properties_insert = format!("{properties_insert}, (?, ?, ?, ?)");
+ }
+ }
+
+ self.execute(&properties_insert, query_args, None).await?;
+
+ Ok(Namespace::with_properties(namespace.clone(), properties))
+ } else {
+ // set a default property of exists = true
+ // up for debate if this is worthwhile
+ self.execute(
+ &insert,
+ vec![
+ Some(&self.name),
+ Some(&namespace_str),
+ Some("exists"),
+ Some("true"),
+ ],
+ None,
+ )
+ .await?;
+ Ok(Namespace::with_properties(namespace.clone(), properties))
+ }
}
- async fn get_namespace(&self, _namespace: &NamespaceIdent) -> Result<Namespace> {
- todo!()
+ async fn get_namespace(&self, namespace: &NamespaceIdent) -> Result<Namespace> {
+ let exists = self.namespace_exists(namespace).await?;
+ if exists {
+ let namespace_props = self
+ .fetch_rows(
+ &format!(
+ "SELECT
+ {NAMESPACE_FIELD_NAME},
+ {NAMESPACE_FIELD_PROPERTY_KEY},
+ {NAMESPACE_FIELD_PROPERTY_VALUE}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
+ AND {NAMESPACE_FIELD_NAME} = ?"
+ ),
+ vec![Some(&self.name), Some(&namespace.join("."))],
+ )
+ .await?;
+
+ let properties: HashMap<String, String> = namespace_props
+ .iter()
+ .filter_map(|r| {
+ let key = r.try_get(NAMESPACE_FIELD_PROPERTY_KEY).ok();
Review Comment:
similarly here?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "callum-ryan (via GitHub)" <gi...@apache.org>.
callum-ryan commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1740065220
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -141,21 +142,30 @@ impl SqlCatalog {
}
/// SQLX Any does not implement PostgresSQL bindings, so we have to do this.
- pub async fn execute_statement(
- &self,
- query: &String,
- args: Vec<Option<&String>>,
- ) -> Result<Vec<AnyRow>> {
- let query_with_placeholders: Cow<str> =
- if self.sql_bind_style == SqlBindStyle::DollarNumeric {
- let mut query = query.clone();
- for i in 0..args.len() {
- query = query.replacen("?", &format!("${}", i + 1), 1);
- }
- Cow::Owned(query)
- } else {
- Cow::Borrowed(query)
- };
+ fn replace_placeholders(&self, query: &str) -> String {
+ match self.sql_bind_style {
Review Comment:
Ok - I've added some comments to the config to explain the parameter's options. It might be worth a rethink when I do the Table functionality in this coming week. Maybe as an Option that defaults to looking at the db type if not provided.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "callum-ryan (via GitHub)" <gi...@apache.org>.
callum-ryan commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1722169865
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,335 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ pub async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ let table_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+ let namespaces_stmt = format!(
+ "SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ match parent {
+ Some(parent) => {
+ if self.namespace_exists(parent).await? {
+ let parent_str = parent.join(".");
+ let parent_table_namespaces_stmt = format!(
+ "{table_namespaces_stmt} AND {CATALOG_FIELD_TABLE_NAMESPACE} LIKE CONCAT(?, '%')"
+ );
+ let parent_namespaces_stmt =
+ format!("{namespaces_stmt} AND {NAMESPACE_FIELD_NAME} LIKE CONCAT(?, '%')");
+
+ let namespace_rows = self
+ .fetch_rows(
+ &format!(
+ "{parent_namespaces_stmt} UNION {parent_table_namespaces_stmt}"
+ ),
+ vec![
+ Some(&self.name),
+ Some(&parent_str),
+ Some(&self.name),
+ Some(&parent_str),
+ ],
+ )
+ .await?;
+
+ let mut namespaces = Vec::<NamespaceIdent>::with_capacity(namespace_rows.len());
Review Comment:
fixed
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "callum-ryan (via GitHub)" <gi...@apache.org>.
callum-ryan commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1722260283
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,335 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ pub async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ let table_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
Review Comment:
Ok nevermind, I have changed this - we can do the parent filtering with the query return instead of pushing it to SQL. See https://github.com/apache/iceberg-rust/pull/534/commits/b0b731c2ed2cab8ffd3df381b17f989e8b8f6380
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,335 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ pub async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ let table_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+ let namespaces_stmt = format!(
+ "SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ match parent {
+ Some(parent) => {
+ if self.namespace_exists(parent).await? {
+ let parent_str = parent.join(".");
+ let parent_table_namespaces_stmt = format!(
+ "{table_namespaces_stmt} AND {CATALOG_FIELD_TABLE_NAMESPACE} LIKE CONCAT(?, '%')"
Review Comment:
https://github.com/apache/iceberg-rust/pull/534/commits/b0b731c2ed2cab8ffd3df381b17f989e8b8f6380
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1725114189
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,312 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ // UNION will remove duplicates.
+ let all_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
+ UNION
+ SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ let namespace_rows = self
+ .fetch_rows(&all_namespaces_stmt, vec![
+ Some(&self.name),
+ Some(&self.name),
+ ])
+ .await?;
+
+ let mut namespaces = HashSet::<NamespaceIdent>::with_capacity(namespace_rows.len());
+
+ if let Some(parent) = parent {
+ if self.namespace_exists(parent).await? {
+ let parent_str = parent.join(".");
+
+ for row in namespace_rows.iter() {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ // if parent = a, then we only want to see a.b, a.c returned.
+ if nsp != parent_str && nsp.starts_with(&parent_str) {
+ namespaces.insert(NamespaceIdent::from_strs(nsp.split("."))?);
+ }
+ }
+
+ Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
+ } else {
+ no_such_namespace_err(parent)
+ }
+ } else {
+ for row in namespace_rows.iter() {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ let mut levels = nsp.split(".").collect::<Vec<&str>>();
+ if !levels.is_empty() {
+ let first_level = levels.drain(..1).collect::<Vec<&str>>();
+ namespaces.insert(NamespaceIdent::from_strs(first_level)?);
+ }
+ }
+
+ Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
+ }
}
async fn create_namespace(
&self,
- _namespace: &NamespaceIdent,
- _properties: HashMap<String, String>,
+ namespace: &NamespaceIdent,
+ properties: HashMap<String, String>,
) -> Result<Namespace> {
- todo!()
+ let exists = self.namespace_exists(namespace).await?;
+
+ if exists {
+ return Err(Error::new(
+ iceberg::ErrorKind::Unexpected,
+ format!("Namespace {:?} already exists", namespace),
+ ));
+ }
+
+ for i in 1..namespace.len() {
Review Comment:
Do we really need this check? I don't see java implementation does this.
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,312 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ // UNION will remove duplicates.
+ let all_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
+ UNION
+ SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ let namespace_rows = self
+ .fetch_rows(&all_namespaces_stmt, vec![
+ Some(&self.name),
+ Some(&self.name),
+ ])
+ .await?;
+
+ let mut namespaces = HashSet::<NamespaceIdent>::with_capacity(namespace_rows.len());
+
+ if let Some(parent) = parent {
+ if self.namespace_exists(parent).await? {
+ let parent_str = parent.join(".");
+
+ for row in namespace_rows.iter() {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ // if parent = a, then we only want to see a.b, a.c returned.
+ if nsp != parent_str && nsp.starts_with(&parent_str) {
+ namespaces.insert(NamespaceIdent::from_strs(nsp.split("."))?);
+ }
+ }
+
+ Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
+ } else {
+ no_such_namespace_err(parent)
+ }
+ } else {
+ for row in namespace_rows.iter() {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ let mut levels = nsp.split(".").collect::<Vec<&str>>();
+ if !levels.is_empty() {
+ let first_level = levels.drain(..1).collect::<Vec<&str>>();
+ namespaces.insert(NamespaceIdent::from_strs(first_level)?);
+ }
+ }
+
+ Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
+ }
}
async fn create_namespace(
&self,
- _namespace: &NamespaceIdent,
- _properties: HashMap<String, String>,
+ namespace: &NamespaceIdent,
+ properties: HashMap<String, String>,
) -> Result<Namespace> {
- todo!()
+ let exists = self.namespace_exists(namespace).await?;
+
+ if exists {
+ return Err(Error::new(
+ iceberg::ErrorKind::Unexpected,
+ format!("Namespace {:?} already exists", namespace),
+ ));
+ }
+
+ for i in 1..namespace.len() {
+ let parent_namespace = NamespaceIdent::from_vec(namespace[..i].to_vec())?;
+ let parent_exists = self.namespace_exists(&parent_namespace).await?;
+ if !parent_exists {
+ return no_such_namespace_err(&parent_namespace);
+ }
+ }
+
+ let namespace_str = namespace.join(".");
+ let insert = format!(
+ "INSERT INTO {NAMESPACE_TABLE_NAME} ({CATALOG_FIELD_CATALOG_NAME}, {NAMESPACE_FIELD_NAME}, {NAMESPACE_FIELD_PROPERTY_KEY}, {NAMESPACE_FIELD_PROPERTY_VALUE})
+ VALUES (?, ?, ?, ?)");
+ if !properties.is_empty() {
+ let mut query_args = Vec::with_capacity(properties.len() * 4);
+ let mut properties_insert = insert.clone();
+ for (index, (key, value)) in properties.iter().enumerate() {
+ query_args.extend_from_slice(&[
+ Some(self.name.as_str()),
+ Some(namespace_str.as_str()),
+ Some(key.as_str()),
+ Some(value.as_str()),
+ ]);
+ if index > 0 {
+ properties_insert = format!("{properties_insert}, (?, ?, ?, ?)");
+ }
+ }
+
+ self.execute(&properties_insert, query_args, None).await?;
+
+ Ok(Namespace::with_properties(namespace.clone(), properties))
+ } else {
+ // set a default property of exists = true
+ // up for debate if this is worthwhile
+ self.execute(
+ &insert,
+ vec![
+ Some(&self.name),
+ Some(&namespace_str),
+ Some("exists"),
+ Some("true"),
+ ],
+ None,
+ )
+ .await?;
+ Ok(Namespace::with_properties(namespace.clone(), properties))
+ }
}
- async fn get_namespace(&self, _namespace: &NamespaceIdent) -> Result<Namespace> {
- todo!()
+ async fn get_namespace(&self, namespace: &NamespaceIdent) -> Result<Namespace> {
+ let exists = self.namespace_exists(namespace).await?;
+ if exists {
+ let namespace_props = self
+ .fetch_rows(
+ &format!(
+ "SELECT
+ {NAMESPACE_FIELD_NAME},
+ {NAMESPACE_FIELD_PROPERTY_KEY},
+ {NAMESPACE_FIELD_PROPERTY_VALUE}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
+ AND {NAMESPACE_FIELD_NAME} = ?"
+ ),
+ vec![Some(&self.name), Some(&namespace.join("."))],
+ )
+ .await?;
+
+ let mut properties = HashMap::with_capacity(namespace_props.len());
+
+ for row in namespace_props {
+ let key = row
+ .try_get::<String, _>(NAMESPACE_FIELD_PROPERTY_KEY)
+ .map_err(from_sqlx_error)?;
+ let value = row
+ .try_get::<String, _>(NAMESPACE_FIELD_PROPERTY_VALUE)
+ .map_err(from_sqlx_error)?;
+
+ properties.insert(key, value);
+ }
+
+ Ok(Namespace::with_properties(namespace.clone(), properties))
+ } else {
+ no_such_namespace_err(namespace)
+ }
}
- async fn namespace_exists(&self, _namespace: &NamespaceIdent) -> Result<bool> {
- todo!()
+ async fn namespace_exists(&self, namespace: &NamespaceIdent) -> Result<bool> {
+ let namespace_str = namespace.join(".");
+
+ let table_namespaces = self
+ .fetch_rows(
+ &format!(
+ "SELECT 1 FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
+ AND {CATALOG_FIELD_TABLE_NAMESPACE} = ?
+ LIMIT 1"
+ ),
+ vec![Some(&self.name), Some(&namespace_str)],
+ )
+ .await?;
+
+ if !table_namespaces.is_empty() {
+ Ok(true)
+ } else {
+ let namespaces = self
+ .fetch_rows(
+ &format!(
+ "SELECT 1 FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
+ AND {NAMESPACE_FIELD_NAME} = ?
+ LIMIT 1"
+ ),
+ vec![Some(&self.name), Some(&namespace_str)],
+ )
+ .await?;
+ if !namespaces.is_empty() {
+ Ok(true)
+ } else {
+ Ok(false)
+ }
+ }
}
async fn update_namespace(
&self,
- _namespace: &NamespaceIdent,
- _properties: HashMap<String, String>,
+ namespace: &NamespaceIdent,
+ properties: HashMap<String, String>,
) -> Result<()> {
- todo!()
+ let exists = self.namespace_exists(namespace).await?;
+ if exists {
+ let existing_properties = self.get_namespace(namespace).await?.properties().clone();
+ let namespace_str = namespace.join(".");
+
+ let mut updates = vec![];
+ let mut inserts = vec![];
+
+ for (key, value) in properties.iter() {
+ if existing_properties.contains_key(key) {
+ if existing_properties.get(key) != Some(value) {
+ updates.push((key, value));
+ }
+ } else {
+ inserts.push((key, value));
+ }
+ }
+
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let update_stmt = format!(
+ "UPDATE {NAMESPACE_TABLE_NAME} SET {NAMESPACE_FIELD_PROPERTY_VALUE} = ?
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
+ AND {NAMESPACE_FIELD_NAME} = ?
+ AND {NAMESPACE_FIELD_PROPERTY_KEY} = ?"
+ );
+
+ let insert_stmt = format!(
+ "INSERT INTO {NAMESPACE_TABLE_NAME} ({CATALOG_FIELD_CATALOG_NAME}, {NAMESPACE_FIELD_NAME}, {NAMESPACE_FIELD_PROPERTY_KEY}, {NAMESPACE_FIELD_PROPERTY_VALUE})
+ VALUES (?, ?, ?, ?)"
+ );
+
+ for (key, value) in updates {
+ self.execute(
+ &update_stmt,
+ vec![
+ Some(value),
+ Some(&self.name),
+ Some(&namespace_str),
+ Some(key),
+ ],
+ Some(&mut tx),
+ )
+ .await?;
+ }
+
+ for (key, value) in inserts {
+ self.execute(
+ &insert_stmt,
+ vec![
+ Some(&self.name),
+ Some(&namespace_str),
+ Some(key),
+ Some(value),
+ ],
+ Some(&mut tx),
+ )
+ .await?;
+ }
+
+ let _ = tx.commit().await.map_err(from_sqlx_error)?;
+
+ Ok(())
+ } else {
+ no_such_namespace_err(namespace)
+ }
}
- async fn drop_namespace(&self, _namespace: &NamespaceIdent) -> Result<()> {
- todo!()
+ async fn drop_namespace(&self, namespace: &NamespaceIdent) -> Result<()> {
+ let exists = self.namespace_exists(namespace).await?;
+ if exists {
+ // TODO: check that the namespace is empty
Review Comment:
If we decided to implement checking in following pr, I would sugges to move the implementation of this method in following pr. Or please create an issue to track this.
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -141,21 +142,30 @@ impl SqlCatalog {
}
/// SQLX Any does not implement PostgresSQL bindings, so we have to do this.
- pub async fn execute_statement(
- &self,
- query: &String,
- args: Vec<Option<&String>>,
- ) -> Result<Vec<AnyRow>> {
- let query_with_placeholders: Cow<str> =
- if self.sql_bind_style == SqlBindStyle::DollarNumeric {
- let mut query = query.clone();
- for i in 0..args.len() {
- query = query.replacen("?", &format!("${}", i + 1), 1);
- }
- Cow::Owned(query)
- } else {
- Cow::Borrowed(query)
- };
+ fn replace_placeholders(&self, query: &str) -> String {
+ match self.sql_bind_style {
Review Comment:
Asking user to pass `sql_bind_style` would be error prone, could we infer this from url?
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,312 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ // UNION will remove duplicates.
+ let all_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
+ UNION
+ SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ let namespace_rows = self
+ .fetch_rows(&all_namespaces_stmt, vec![
+ Some(&self.name),
+ Some(&self.name),
+ ])
+ .await?;
+
+ let mut namespaces = HashSet::<NamespaceIdent>::with_capacity(namespace_rows.len());
+
+ if let Some(parent) = parent {
+ if self.namespace_exists(parent).await? {
+ let parent_str = parent.join(".");
+
+ for row in namespace_rows.iter() {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ // if parent = a, then we only want to see a.b, a.c returned.
+ if nsp != parent_str && nsp.starts_with(&parent_str) {
+ namespaces.insert(NamespaceIdent::from_strs(nsp.split("."))?);
+ }
+ }
+
+ Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
+ } else {
+ no_such_namespace_err(parent)
+ }
+ } else {
+ for row in namespace_rows.iter() {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ let mut levels = nsp.split(".").collect::<Vec<&str>>();
+ if !levels.is_empty() {
+ let first_level = levels.drain(..1).collect::<Vec<&str>>();
+ namespaces.insert(NamespaceIdent::from_strs(first_level)?);
+ }
+ }
+
+ Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
+ }
}
async fn create_namespace(
&self,
- _namespace: &NamespaceIdent,
- _properties: HashMap<String, String>,
+ namespace: &NamespaceIdent,
+ properties: HashMap<String, String>,
) -> Result<Namespace> {
- todo!()
+ let exists = self.namespace_exists(namespace).await?;
+
+ if exists {
+ return Err(Error::new(
+ iceberg::ErrorKind::Unexpected,
+ format!("Namespace {:?} already exists", namespace),
+ ));
+ }
+
+ for i in 1..namespace.len() {
+ let parent_namespace = NamespaceIdent::from_vec(namespace[..i].to_vec())?;
+ let parent_exists = self.namespace_exists(&parent_namespace).await?;
+ if !parent_exists {
+ return no_such_namespace_err(&parent_namespace);
+ }
+ }
+
+ let namespace_str = namespace.join(".");
+ let insert = format!(
+ "INSERT INTO {NAMESPACE_TABLE_NAME} ({CATALOG_FIELD_CATALOG_NAME}, {NAMESPACE_FIELD_NAME}, {NAMESPACE_FIELD_PROPERTY_KEY}, {NAMESPACE_FIELD_PROPERTY_VALUE})
Review Comment:
I think this is incorrect? We don't need table name field?
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,312 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ // UNION will remove duplicates.
+ let all_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
+ UNION
+ SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ let namespace_rows = self
+ .fetch_rows(&all_namespaces_stmt, vec![
+ Some(&self.name),
+ Some(&self.name),
+ ])
+ .await?;
+
+ let mut namespaces = HashSet::<NamespaceIdent>::with_capacity(namespace_rows.len());
+
+ if let Some(parent) = parent {
+ if self.namespace_exists(parent).await? {
+ let parent_str = parent.join(".");
+
+ for row in namespace_rows.iter() {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ // if parent = a, then we only want to see a.b, a.c returned.
+ if nsp != parent_str && nsp.starts_with(&parent_str) {
+ namespaces.insert(NamespaceIdent::from_strs(nsp.split("."))?);
+ }
+ }
+
+ Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
+ } else {
+ no_such_namespace_err(parent)
+ }
+ } else {
+ for row in namespace_rows.iter() {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ let mut levels = nsp.split(".").collect::<Vec<&str>>();
+ if !levels.is_empty() {
+ let first_level = levels.drain(..1).collect::<Vec<&str>>();
+ namespaces.insert(NamespaceIdent::from_strs(first_level)?);
+ }
+ }
+
+ Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
+ }
}
async fn create_namespace(
&self,
- _namespace: &NamespaceIdent,
- _properties: HashMap<String, String>,
+ namespace: &NamespaceIdent,
+ properties: HashMap<String, String>,
) -> Result<Namespace> {
- todo!()
+ let exists = self.namespace_exists(namespace).await?;
+
+ if exists {
+ return Err(Error::new(
+ iceberg::ErrorKind::Unexpected,
+ format!("Namespace {:?} already exists", namespace),
+ ));
+ }
+
+ for i in 1..namespace.len() {
+ let parent_namespace = NamespaceIdent::from_vec(namespace[..i].to_vec())?;
+ let parent_exists = self.namespace_exists(&parent_namespace).await?;
+ if !parent_exists {
+ return no_such_namespace_err(&parent_namespace);
+ }
+ }
+
+ let namespace_str = namespace.join(".");
+ let insert = format!(
+ "INSERT INTO {NAMESPACE_TABLE_NAME} ({CATALOG_FIELD_CATALOG_NAME}, {NAMESPACE_FIELD_NAME}, {NAMESPACE_FIELD_PROPERTY_KEY}, {NAMESPACE_FIELD_PROPERTY_VALUE})
+ VALUES (?, ?, ?, ?)");
+ if !properties.is_empty() {
+ let mut query_args = Vec::with_capacity(properties.len() * 4);
+ let mut properties_insert = insert.clone();
+ for (index, (key, value)) in properties.iter().enumerate() {
+ query_args.extend_from_slice(&[
+ Some(self.name.as_str()),
+ Some(namespace_str.as_str()),
+ Some(key.as_str()),
+ Some(value.as_str()),
+ ]);
+ if index > 0 {
+ properties_insert = format!("{properties_insert}, (?, ?, ?, ?)");
Review Comment:
We still need to insert `(exists, true)` for non empty properties
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -276,4 +579,385 @@ mod tests {
new_sql_catalog(warehouse_loc.clone()).await;
new_sql_catalog(warehouse_loc.clone()).await;
}
+
+ #[tokio::test]
+ async fn test_list_namespaces_returns_empty_vector() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+
+ assert_eq!(catalog.list_namespaces(None).await.unwrap(), vec![]);
+ }
+
+ #[tokio::test]
+ async fn test_list_namespaces_returns_multiple_namespaces() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_1 = NamespaceIdent::new("a".into());
+ let namespace_ident_2 = NamespaceIdent::new("b".into());
+ create_namespaces(&catalog, &vec![&namespace_ident_1, &namespace_ident_2]).await;
+
+ assert_eq!(
+ to_set(catalog.list_namespaces(None).await.unwrap()),
+ to_set(vec![namespace_ident_1, namespace_ident_2])
+ );
+ }
+
+ #[tokio::test]
+ async fn test_list_namespaces_returns_only_top_level_namespaces() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_1 = NamespaceIdent::new("a".into());
+ let namespace_ident_2 = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+ let namespace_ident_3 = NamespaceIdent::new("b".into());
+ create_namespaces(&catalog, &vec![
+ &namespace_ident_1,
+ &namespace_ident_2,
+ &namespace_ident_3,
+ ])
+ .await;
+
+ assert_eq!(
+ to_set(catalog.list_namespaces(None).await.unwrap()),
+ to_set(vec![namespace_ident_1, namespace_ident_3])
+ );
+ }
+
+ #[tokio::test]
+ async fn test_list_namespaces_returns_no_namespaces_under_parent() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_1 = NamespaceIdent::new("a".into());
+ let namespace_ident_2 = NamespaceIdent::new("b".into());
+ create_namespaces(&catalog, &vec![&namespace_ident_1, &namespace_ident_2]).await;
+
+ assert_eq!(
+ catalog
+ .list_namespaces(Some(&namespace_ident_1))
+ .await
+ .unwrap(),
+ vec![]
+ );
+ }
+
+ #[tokio::test]
+ async fn test_list_namespaces_returns_namespace_under_parent() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_1 = NamespaceIdent::new("a".into());
+ let namespace_ident_2 = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+ let namespace_ident_3 = NamespaceIdent::new("c".into());
+ create_namespaces(&catalog, &vec![
+ &namespace_ident_1,
+ &namespace_ident_2,
+ &namespace_ident_3,
+ ])
+ .await;
+
+ assert_eq!(
+ to_set(catalog.list_namespaces(None).await.unwrap()),
+ to_set(vec![namespace_ident_1.clone(), namespace_ident_3])
+ );
+
+ assert_eq!(
+ catalog
+ .list_namespaces(Some(&namespace_ident_1))
+ .await
+ .unwrap(),
+ vec![NamespaceIdent::from_strs(vec!["a", "b"]).unwrap()]
+ );
+ }
+
+ #[tokio::test]
+ async fn test_list_namespaces_returns_multiple_namespaces_under_parent() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_1 = NamespaceIdent::new("a".to_string());
+ let namespace_ident_2 = NamespaceIdent::from_strs(vec!["a", "a"]).unwrap();
+ let namespace_ident_3 = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+ let namespace_ident_4 = NamespaceIdent::from_strs(vec!["a", "c"]).unwrap();
+ let namespace_ident_5 = NamespaceIdent::new("b".into());
+ create_namespaces(&catalog, &vec![
+ &namespace_ident_1,
+ &namespace_ident_2,
+ &namespace_ident_3,
+ &namespace_ident_4,
+ &namespace_ident_5,
+ ])
+ .await;
+
+ assert_eq!(
+ to_set(
+ catalog
+ .list_namespaces(Some(&namespace_ident_1))
+ .await
+ .unwrap()
+ ),
+ to_set(vec![
+ NamespaceIdent::from_strs(vec!["a", "a"]).unwrap(),
+ NamespaceIdent::from_strs(vec!["a", "b"]).unwrap(),
+ NamespaceIdent::from_strs(vec!["a", "c"]).unwrap(),
+ ])
+ );
+ }
+
+ #[tokio::test]
+ async fn test_namespace_exists_returns_false() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident = NamespaceIdent::new("a".into());
+ create_namespace(&catalog, &namespace_ident).await;
+
+ assert!(!catalog
+ .namespace_exists(&NamespaceIdent::new("b".into()))
+ .await
+ .unwrap());
+ }
+
+ #[tokio::test]
+ async fn test_namespace_exists_returns_true() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident = NamespaceIdent::new("a".into());
+ create_namespace(&catalog, &namespace_ident).await;
+
+ assert!(catalog.namespace_exists(&namespace_ident).await.unwrap());
+ }
+
+ #[tokio::test]
+ async fn test_create_namespace_with_properties() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident = NamespaceIdent::new("abc".into());
+
+ let mut properties: HashMap<String, String> = HashMap::new();
+ properties.insert("k".into(), "v".into());
+
+ assert_eq!(
+ catalog
+ .create_namespace(&namespace_ident, properties.clone())
+ .await
+ .unwrap(),
+ Namespace::with_properties(namespace_ident.clone(), properties.clone())
+ );
+
+ assert_eq!(
+ catalog.get_namespace(&namespace_ident).await.unwrap(),
+ Namespace::with_properties(namespace_ident, properties)
+ );
+ }
+
+ #[tokio::test]
+ async fn test_create_namespace_throws_error_if_namespace_already_exists() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident = NamespaceIdent::new("a".into());
+ create_namespace(&catalog, &namespace_ident).await;
+
+ assert_eq!(
+ catalog
+ .create_namespace(&namespace_ident, HashMap::new())
+ .await
+ .unwrap_err()
+ .to_string(),
+ format!(
+ "Unexpected => Namespace {:?} already exists",
+ &namespace_ident
+ )
+ );
+
+ assert_eq!(
+ catalog.get_namespace(&namespace_ident).await.unwrap(),
+ Namespace::with_properties(namespace_ident, default_properties())
+ );
+ }
+
+ #[tokio::test]
+ async fn test_create_nested_namespace() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let parent_namespace_ident = NamespaceIdent::new("a".into());
+ create_namespace(&catalog, &parent_namespace_ident).await;
+
+ let child_namespace_ident = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+
+ assert_eq!(
+ catalog
+ .create_namespace(&child_namespace_ident, HashMap::new())
+ .await
+ .unwrap(),
+ Namespace::new(child_namespace_ident.clone())
+ );
+
+ assert_eq!(
+ catalog.get_namespace(&child_namespace_ident).await.unwrap(),
+ Namespace::with_properties(child_namespace_ident, default_properties())
+ );
+ }
+
+ #[tokio::test]
+ async fn test_create_deeply_nested_namespace() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_a = NamespaceIdent::new("a".into());
+ let namespace_ident_a_b = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+ create_namespaces(&catalog, &vec![&namespace_ident_a, &namespace_ident_a_b]).await;
+
+ let namespace_ident_a_b_c = NamespaceIdent::from_strs(vec!["a", "b", "c"]).unwrap();
+
+ assert_eq!(
+ catalog
+ .create_namespace(&namespace_ident_a_b_c, HashMap::new())
+ .await
+ .unwrap(),
+ Namespace::new(namespace_ident_a_b_c.clone())
+ );
+
+ assert_eq!(
+ catalog.get_namespace(&namespace_ident_a_b_c).await.unwrap(),
+ Namespace::with_properties(namespace_ident_a_b_c, default_properties())
+ );
+ }
+
+ #[tokio::test]
+ async fn test_create_nested_namespace_throws_error_if_top_level_namespace_doesnt_exist() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+
+ let nested_namespace_ident = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+
+ assert_eq!(
+ catalog
+ .create_namespace(&nested_namespace_ident, HashMap::new())
+ .await
+ .unwrap_err()
+ .to_string(),
+ format!(
+ "Unexpected => No such namespace: {:?}",
+ NamespaceIdent::new("a".into())
+ )
+ );
+
+ assert_eq!(catalog.list_namespaces(None).await.unwrap(), vec![]);
+ }
+
+ #[tokio::test]
+ async fn test_drop_namespace() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident = NamespaceIdent::new("abc".into());
+ create_namespace(&catalog, &namespace_ident).await;
+
+ catalog.drop_namespace(&namespace_ident).await.unwrap();
+
+ assert!(!catalog.namespace_exists(&namespace_ident).await.unwrap())
+ }
+
+ #[tokio::test]
+ async fn test_drop_nested_namespace() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_a = NamespaceIdent::new("a".into());
+ let namespace_ident_a_b = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+ create_namespaces(&catalog, &vec![&namespace_ident_a, &namespace_ident_a_b]).await;
+
+ catalog.drop_namespace(&namespace_ident_a_b).await.unwrap();
+
+ assert!(!catalog
+ .namespace_exists(&namespace_ident_a_b)
+ .await
+ .unwrap());
+
+ assert!(catalog.namespace_exists(&namespace_ident_a).await.unwrap());
+ }
+
+ #[tokio::test]
+ async fn test_drop_deeply_nested_namespace() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_a = NamespaceIdent::new("a".into());
+ let namespace_ident_a_b = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+ let namespace_ident_a_b_c = NamespaceIdent::from_strs(vec!["a", "b", "c"]).unwrap();
+ create_namespaces(&catalog, &vec![
+ &namespace_ident_a,
+ &namespace_ident_a_b,
+ &namespace_ident_a_b_c,
+ ])
+ .await;
+
+ catalog
+ .drop_namespace(&namespace_ident_a_b_c)
+ .await
+ .unwrap();
+
+ assert!(!catalog
+ .namespace_exists(&namespace_ident_a_b_c)
+ .await
+ .unwrap());
+
+ assert!(catalog
+ .namespace_exists(&namespace_ident_a_b)
+ .await
+ .unwrap());
+
+ assert!(catalog.namespace_exists(&namespace_ident_a).await.unwrap());
+ }
+
+ #[tokio::test]
+ async fn test_drop_namespace_throws_error_if_namespace_doesnt_exist() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+
+ let non_existent_namespace_ident = NamespaceIdent::new("abc".into());
+ assert_eq!(
+ catalog
+ .drop_namespace(&non_existent_namespace_ident)
+ .await
+ .unwrap_err()
+ .to_string(),
+ format!(
+ "Unexpected => No such namespace: {:?}",
+ non_existent_namespace_ident
+ )
+ )
+ }
+
+ #[tokio::test]
+ async fn test_drop_namespace_throws_error_if_nested_namespace_doesnt_exist() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ create_namespace(&catalog, &NamespaceIdent::new("a".into())).await;
+
+ let non_existent_namespace_ident =
+ NamespaceIdent::from_vec(vec!["a".into(), "b".into()]).unwrap();
+ assert_eq!(
+ catalog
+ .drop_namespace(&non_existent_namespace_ident)
+ .await
+ .unwrap_err()
+ .to_string(),
+ format!(
+ "Unexpected => No such namespace: {:?}",
+ non_existent_namespace_ident
+ )
+ )
+ }
+
+ #[tokio::test]
+ #[ignore = "Java/Python do not drop nested namespaces?"]
+ async fn test_dropping_a_namespace_also_drops_namespaces_nested_under_that_one() {
Review Comment:
Hmm, should we really take this behavior?
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,312 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ // UNION will remove duplicates.
+ let all_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
+ UNION
+ SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ let namespace_rows = self
+ .fetch_rows(&all_namespaces_stmt, vec![
+ Some(&self.name),
+ Some(&self.name),
+ ])
+ .await?;
+
+ let mut namespaces = HashSet::<NamespaceIdent>::with_capacity(namespace_rows.len());
+
+ if let Some(parent) = parent {
+ if self.namespace_exists(parent).await? {
+ let parent_str = parent.join(".");
+
+ for row in namespace_rows.iter() {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ // if parent = a, then we only want to see a.b, a.c returned.
+ if nsp != parent_str && nsp.starts_with(&parent_str) {
+ namespaces.insert(NamespaceIdent::from_strs(nsp.split("."))?);
+ }
+ }
+
+ Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
+ } else {
+ no_such_namespace_err(parent)
+ }
+ } else {
+ for row in namespace_rows.iter() {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ let mut levels = nsp.split(".").collect::<Vec<&str>>();
+ if !levels.is_empty() {
+ let first_level = levels.drain(..1).collect::<Vec<&str>>();
+ namespaces.insert(NamespaceIdent::from_strs(first_level)?);
+ }
+ }
+
+ Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
+ }
}
async fn create_namespace(
&self,
- _namespace: &NamespaceIdent,
- _properties: HashMap<String, String>,
+ namespace: &NamespaceIdent,
+ properties: HashMap<String, String>,
) -> Result<Namespace> {
- todo!()
+ let exists = self.namespace_exists(namespace).await?;
+
+ if exists {
+ return Err(Error::new(
+ iceberg::ErrorKind::Unexpected,
+ format!("Namespace {:?} already exists", namespace),
+ ));
+ }
+
+ for i in 1..namespace.len() {
+ let parent_namespace = NamespaceIdent::from_vec(namespace[..i].to_vec())?;
+ let parent_exists = self.namespace_exists(&parent_namespace).await?;
+ if !parent_exists {
+ return no_such_namespace_err(&parent_namespace);
+ }
+ }
+
+ let namespace_str = namespace.join(".");
+ let insert = format!(
+ "INSERT INTO {NAMESPACE_TABLE_NAME} ({CATALOG_FIELD_CATALOG_NAME}, {NAMESPACE_FIELD_NAME}, {NAMESPACE_FIELD_PROPERTY_KEY}, {NAMESPACE_FIELD_PROPERTY_VALUE})
+ VALUES (?, ?, ?, ?)");
+ if !properties.is_empty() {
+ let mut query_args = Vec::with_capacity(properties.len() * 4);
+ let mut properties_insert = insert.clone();
+ for (index, (key, value)) in properties.iter().enumerate() {
+ query_args.extend_from_slice(&[
+ Some(self.name.as_str()),
+ Some(namespace_str.as_str()),
+ Some(key.as_str()),
+ Some(value.as_str()),
+ ]);
+ if index > 0 {
+ properties_insert = format!("{properties_insert}, (?, ?, ?, ?)");
Review Comment:
Though this is correct, I feel this maybe slow since it involves many allocations. How about `iter::repeat("(?, ?, ?, ?)").take(properties.len()).join(", ")`
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "callum-ryan (via GitHub)" <gi...@apache.org>.
callum-ryan commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1725537409
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -276,4 +579,385 @@ mod tests {
new_sql_catalog(warehouse_loc.clone()).await;
new_sql_catalog(warehouse_loc.clone()).await;
}
+
+ #[tokio::test]
+ async fn test_list_namespaces_returns_empty_vector() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+
+ assert_eq!(catalog.list_namespaces(None).await.unwrap(), vec![]);
+ }
+
+ #[tokio::test]
+ async fn test_list_namespaces_returns_multiple_namespaces() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_1 = NamespaceIdent::new("a".into());
+ let namespace_ident_2 = NamespaceIdent::new("b".into());
+ create_namespaces(&catalog, &vec![&namespace_ident_1, &namespace_ident_2]).await;
+
+ assert_eq!(
+ to_set(catalog.list_namespaces(None).await.unwrap()),
+ to_set(vec![namespace_ident_1, namespace_ident_2])
+ );
+ }
+
+ #[tokio::test]
+ async fn test_list_namespaces_returns_only_top_level_namespaces() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_1 = NamespaceIdent::new("a".into());
+ let namespace_ident_2 = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+ let namespace_ident_3 = NamespaceIdent::new("b".into());
+ create_namespaces(&catalog, &vec![
+ &namespace_ident_1,
+ &namespace_ident_2,
+ &namespace_ident_3,
+ ])
+ .await;
+
+ assert_eq!(
+ to_set(catalog.list_namespaces(None).await.unwrap()),
+ to_set(vec![namespace_ident_1, namespace_ident_3])
+ );
+ }
+
+ #[tokio::test]
+ async fn test_list_namespaces_returns_no_namespaces_under_parent() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_1 = NamespaceIdent::new("a".into());
+ let namespace_ident_2 = NamespaceIdent::new("b".into());
+ create_namespaces(&catalog, &vec![&namespace_ident_1, &namespace_ident_2]).await;
+
+ assert_eq!(
+ catalog
+ .list_namespaces(Some(&namespace_ident_1))
+ .await
+ .unwrap(),
+ vec![]
+ );
+ }
+
+ #[tokio::test]
+ async fn test_list_namespaces_returns_namespace_under_parent() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_1 = NamespaceIdent::new("a".into());
+ let namespace_ident_2 = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+ let namespace_ident_3 = NamespaceIdent::new("c".into());
+ create_namespaces(&catalog, &vec![
+ &namespace_ident_1,
+ &namespace_ident_2,
+ &namespace_ident_3,
+ ])
+ .await;
+
+ assert_eq!(
+ to_set(catalog.list_namespaces(None).await.unwrap()),
+ to_set(vec![namespace_ident_1.clone(), namespace_ident_3])
+ );
+
+ assert_eq!(
+ catalog
+ .list_namespaces(Some(&namespace_ident_1))
+ .await
+ .unwrap(),
+ vec![NamespaceIdent::from_strs(vec!["a", "b"]).unwrap()]
+ );
+ }
+
+ #[tokio::test]
+ async fn test_list_namespaces_returns_multiple_namespaces_under_parent() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_1 = NamespaceIdent::new("a".to_string());
+ let namespace_ident_2 = NamespaceIdent::from_strs(vec!["a", "a"]).unwrap();
+ let namespace_ident_3 = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+ let namespace_ident_4 = NamespaceIdent::from_strs(vec!["a", "c"]).unwrap();
+ let namespace_ident_5 = NamespaceIdent::new("b".into());
+ create_namespaces(&catalog, &vec![
+ &namespace_ident_1,
+ &namespace_ident_2,
+ &namespace_ident_3,
+ &namespace_ident_4,
+ &namespace_ident_5,
+ ])
+ .await;
+
+ assert_eq!(
+ to_set(
+ catalog
+ .list_namespaces(Some(&namespace_ident_1))
+ .await
+ .unwrap()
+ ),
+ to_set(vec![
+ NamespaceIdent::from_strs(vec!["a", "a"]).unwrap(),
+ NamespaceIdent::from_strs(vec!["a", "b"]).unwrap(),
+ NamespaceIdent::from_strs(vec!["a", "c"]).unwrap(),
+ ])
+ );
+ }
+
+ #[tokio::test]
+ async fn test_namespace_exists_returns_false() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident = NamespaceIdent::new("a".into());
+ create_namespace(&catalog, &namespace_ident).await;
+
+ assert!(!catalog
+ .namespace_exists(&NamespaceIdent::new("b".into()))
+ .await
+ .unwrap());
+ }
+
+ #[tokio::test]
+ async fn test_namespace_exists_returns_true() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident = NamespaceIdent::new("a".into());
+ create_namespace(&catalog, &namespace_ident).await;
+
+ assert!(catalog.namespace_exists(&namespace_ident).await.unwrap());
+ }
+
+ #[tokio::test]
+ async fn test_create_namespace_with_properties() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident = NamespaceIdent::new("abc".into());
+
+ let mut properties: HashMap<String, String> = HashMap::new();
+ properties.insert("k".into(), "v".into());
+
+ assert_eq!(
+ catalog
+ .create_namespace(&namespace_ident, properties.clone())
+ .await
+ .unwrap(),
+ Namespace::with_properties(namespace_ident.clone(), properties.clone())
+ );
+
+ assert_eq!(
+ catalog.get_namespace(&namespace_ident).await.unwrap(),
+ Namespace::with_properties(namespace_ident, properties)
+ );
+ }
+
+ #[tokio::test]
+ async fn test_create_namespace_throws_error_if_namespace_already_exists() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident = NamespaceIdent::new("a".into());
+ create_namespace(&catalog, &namespace_ident).await;
+
+ assert_eq!(
+ catalog
+ .create_namespace(&namespace_ident, HashMap::new())
+ .await
+ .unwrap_err()
+ .to_string(),
+ format!(
+ "Unexpected => Namespace {:?} already exists",
+ &namespace_ident
+ )
+ );
+
+ assert_eq!(
+ catalog.get_namespace(&namespace_ident).await.unwrap(),
+ Namespace::with_properties(namespace_ident, default_properties())
+ );
+ }
+
+ #[tokio::test]
+ async fn test_create_nested_namespace() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let parent_namespace_ident = NamespaceIdent::new("a".into());
+ create_namespace(&catalog, &parent_namespace_ident).await;
+
+ let child_namespace_ident = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+
+ assert_eq!(
+ catalog
+ .create_namespace(&child_namespace_ident, HashMap::new())
+ .await
+ .unwrap(),
+ Namespace::new(child_namespace_ident.clone())
+ );
+
+ assert_eq!(
+ catalog.get_namespace(&child_namespace_ident).await.unwrap(),
+ Namespace::with_properties(child_namespace_ident, default_properties())
+ );
+ }
+
+ #[tokio::test]
+ async fn test_create_deeply_nested_namespace() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_a = NamespaceIdent::new("a".into());
+ let namespace_ident_a_b = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+ create_namespaces(&catalog, &vec![&namespace_ident_a, &namespace_ident_a_b]).await;
+
+ let namespace_ident_a_b_c = NamespaceIdent::from_strs(vec!["a", "b", "c"]).unwrap();
+
+ assert_eq!(
+ catalog
+ .create_namespace(&namespace_ident_a_b_c, HashMap::new())
+ .await
+ .unwrap(),
+ Namespace::new(namespace_ident_a_b_c.clone())
+ );
+
+ assert_eq!(
+ catalog.get_namespace(&namespace_ident_a_b_c).await.unwrap(),
+ Namespace::with_properties(namespace_ident_a_b_c, default_properties())
+ );
+ }
+
+ #[tokio::test]
+ async fn test_create_nested_namespace_throws_error_if_top_level_namespace_doesnt_exist() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+
+ let nested_namespace_ident = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+
+ assert_eq!(
+ catalog
+ .create_namespace(&nested_namespace_ident, HashMap::new())
+ .await
+ .unwrap_err()
+ .to_string(),
+ format!(
+ "Unexpected => No such namespace: {:?}",
+ NamespaceIdent::new("a".into())
+ )
+ );
+
+ assert_eq!(catalog.list_namespaces(None).await.unwrap(), vec![]);
+ }
+
+ #[tokio::test]
+ async fn test_drop_namespace() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident = NamespaceIdent::new("abc".into());
+ create_namespace(&catalog, &namespace_ident).await;
+
+ catalog.drop_namespace(&namespace_ident).await.unwrap();
+
+ assert!(!catalog.namespace_exists(&namespace_ident).await.unwrap())
+ }
+
+ #[tokio::test]
+ async fn test_drop_nested_namespace() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_a = NamespaceIdent::new("a".into());
+ let namespace_ident_a_b = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+ create_namespaces(&catalog, &vec![&namespace_ident_a, &namespace_ident_a_b]).await;
+
+ catalog.drop_namespace(&namespace_ident_a_b).await.unwrap();
+
+ assert!(!catalog
+ .namespace_exists(&namespace_ident_a_b)
+ .await
+ .unwrap());
+
+ assert!(catalog.namespace_exists(&namespace_ident_a).await.unwrap());
+ }
+
+ #[tokio::test]
+ async fn test_drop_deeply_nested_namespace() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ let namespace_ident_a = NamespaceIdent::new("a".into());
+ let namespace_ident_a_b = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
+ let namespace_ident_a_b_c = NamespaceIdent::from_strs(vec!["a", "b", "c"]).unwrap();
+ create_namespaces(&catalog, &vec![
+ &namespace_ident_a,
+ &namespace_ident_a_b,
+ &namespace_ident_a_b_c,
+ ])
+ .await;
+
+ catalog
+ .drop_namespace(&namespace_ident_a_b_c)
+ .await
+ .unwrap();
+
+ assert!(!catalog
+ .namespace_exists(&namespace_ident_a_b_c)
+ .await
+ .unwrap());
+
+ assert!(catalog
+ .namespace_exists(&namespace_ident_a_b)
+ .await
+ .unwrap());
+
+ assert!(catalog.namespace_exists(&namespace_ident_a).await.unwrap());
+ }
+
+ #[tokio::test]
+ async fn test_drop_namespace_throws_error_if_namespace_doesnt_exist() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+
+ let non_existent_namespace_ident = NamespaceIdent::new("abc".into());
+ assert_eq!(
+ catalog
+ .drop_namespace(&non_existent_namespace_ident)
+ .await
+ .unwrap_err()
+ .to_string(),
+ format!(
+ "Unexpected => No such namespace: {:?}",
+ non_existent_namespace_ident
+ )
+ )
+ }
+
+ #[tokio::test]
+ async fn test_drop_namespace_throws_error_if_nested_namespace_doesnt_exist() {
+ let warehouse_loc = temp_path();
+ let catalog = new_sql_catalog(warehouse_loc).await;
+ create_namespace(&catalog, &NamespaceIdent::new("a".into())).await;
+
+ let non_existent_namespace_ident =
+ NamespaceIdent::from_vec(vec!["a".into(), "b".into()]).unwrap();
+ assert_eq!(
+ catalog
+ .drop_namespace(&non_existent_namespace_ident)
+ .await
+ .unwrap_err()
+ .to_string(),
+ format!(
+ "Unexpected => No such namespace: {:?}",
+ non_existent_namespace_ident
+ )
+ )
+ }
+
+ #[tokio::test]
+ #[ignore = "Java/Python do not drop nested namespaces?"]
+ async fn test_dropping_a_namespace_also_drops_namespaces_nested_under_that_one() {
Review Comment:
I left this as an ignored test because I think this is up for debate. It's technically correct that we should not allow dropping a namespace with child namespaces, but the other implementations do not have this.
Maybe as an alternative I could implement a check to fail when there are child namespaces in a call to `drop_namespace`? What do you think?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "callum-ryan (via GitHub)" <gi...@apache.org>.
callum-ryan commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1722170737
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,335 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ pub async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ let table_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+ let namespaces_stmt = format!(
+ "SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ match parent {
+ Some(parent) => {
+ if self.namespace_exists(parent).await? {
+ let parent_str = parent.join(".");
+ let parent_table_namespaces_stmt = format!(
+ "{table_namespaces_stmt} AND {CATALOG_FIELD_TABLE_NAMESPACE} LIKE CONCAT(?, '%')"
+ );
+ let parent_namespaces_stmt =
+ format!("{namespaces_stmt} AND {NAMESPACE_FIELD_NAME} LIKE CONCAT(?, '%')");
+
+ let namespace_rows = self
+ .fetch_rows(
+ &format!(
+ "{parent_namespaces_stmt} UNION {parent_table_namespaces_stmt}"
+ ),
+ vec![
+ Some(&self.name),
+ Some(&parent_str),
+ Some(&self.name),
+ Some(&parent_str),
+ ],
+ )
+ .await?;
+
+ let mut namespaces = Vec::<NamespaceIdent>::with_capacity(namespace_rows.len());
+
+ for row in namespace_rows {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ if nsp != parent_str {
Review Comment:
I've added a comment here to the loop - all of the returned rows from the query will start with parent_str, this is to exclude the parent from the list of returned namespaceident.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1730514726
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -141,21 +142,30 @@ impl SqlCatalog {
}
/// SQLX Any does not implement PostgresSQL bindings, so we have to do this.
- pub async fn execute_statement(
- &self,
- query: &String,
- args: Vec<Option<&String>>,
- ) -> Result<Vec<AnyRow>> {
- let query_with_placeholders: Cow<str> =
- if self.sql_bind_style == SqlBindStyle::DollarNumeric {
- let mut query = query.clone();
- for i in 0..args.len() {
- query = query.replacen("?", &format!("${}", i + 1), 1);
- }
- Cow::Owned(query)
- } else {
- Cow::Borrowed(query)
- };
+ fn replace_placeholders(&self, query: &str) -> String {
+ match self.sql_bind_style {
Review Comment:
Sorry, I'm not sure if I understand correctly, does dollar bind style work with non postgres database backend?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1713021439
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -141,21 +142,24 @@ impl SqlCatalog {
}
/// SQLX Any does not implement PostgresSQL bindings, so we have to do this.
- pub async fn execute_statement(
- &self,
- query: &String,
- args: Vec<Option<&String>>,
- ) -> Result<Vec<AnyRow>> {
- let query_with_placeholders: Cow<str> =
- if self.sql_bind_style == SqlBindStyle::DollarNumeric {
- let mut query = query.clone();
- for i in 0..args.len() {
- query = query.replacen("?", &format!("${}", i + 1), 1);
+ pub fn build_query(&self, query: &str) -> String {
Review Comment:
Maybe this function doesn't need to be public?
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -141,21 +142,24 @@ impl SqlCatalog {
}
/// SQLX Any does not implement PostgresSQL bindings, so we have to do this.
- pub async fn execute_statement(
- &self,
- query: &String,
- args: Vec<Option<&String>>,
- ) -> Result<Vec<AnyRow>> {
- let query_with_placeholders: Cow<str> =
- if self.sql_bind_style == SqlBindStyle::DollarNumeric {
- let mut query = query.clone();
- for i in 0..args.len() {
- query = query.replacen("?", &format!("${}", i + 1), 1);
+ pub fn build_query(&self, query: &str) -> String {
+ match self.sql_bind_style {
+ SqlBindStyle::DollarNumeric => {
+ let mut query = query.to_owned();
+ let mut i = 1;
+ while let Some(pos) = query.find('?') {
Review Comment:
Do you think it's a good idea to use `chars`, for example:
```rust
fn replace_placeholders(query: &str) -> String {
let mut result = String::with_capacity(query.len());
let mut count = 1;
for c in query.chars() {
if c == '?' {
result.push_str(&format!("${count}"));
count += 1;
} else {
result.push(c);
}
}
result
}
```
In this way, we can replace all `?` in `O(n)` instead of `O(n^2)`.
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -141,21 +142,24 @@ impl SqlCatalog {
}
/// SQLX Any does not implement PostgresSQL bindings, so we have to do this.
- pub async fn execute_statement(
- &self,
- query: &String,
- args: Vec<Option<&String>>,
- ) -> Result<Vec<AnyRow>> {
- let query_with_placeholders: Cow<str> =
- if self.sql_bind_style == SqlBindStyle::DollarNumeric {
- let mut query = query.clone();
- for i in 0..args.len() {
- query = query.replacen("?", &format!("${}", i + 1), 1);
+ pub fn build_query(&self, query: &str) -> String {
+ match self.sql_bind_style {
+ SqlBindStyle::DollarNumeric => {
+ let mut query = query.to_owned();
+ let mut i = 1;
+ while let Some(pos) = query.find('?') {
+ query.replace_range(pos..pos + 1, &format!("${}", i));
+ i += 1;
}
- Cow::Owned(query)
- } else {
- Cow::Borrowed(query)
- };
+ query
+ }
+ _ => query.to_owned(),
+ }
+ }
+
+ /// Fetch a vec of AnyRows from a given query
+ pub async fn fetch_rows(&self, query: &str, args: Vec<Option<&String>>) -> Result<Vec<AnyRow>> {
Review Comment:
`&String` isn't useful in Rust; how about using `&str` instead?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "callum-ryan (via GitHub)" <gi...@apache.org>.
callum-ryan commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1714153827
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -141,21 +142,24 @@ impl SqlCatalog {
}
/// SQLX Any does not implement PostgresSQL bindings, so we have to do this.
- pub async fn execute_statement(
- &self,
- query: &String,
- args: Vec<Option<&String>>,
- ) -> Result<Vec<AnyRow>> {
- let query_with_placeholders: Cow<str> =
- if self.sql_bind_style == SqlBindStyle::DollarNumeric {
- let mut query = query.clone();
- for i in 0..args.len() {
- query = query.replacen("?", &format!("${}", i + 1), 1);
+ pub fn build_query(&self, query: &str) -> String {
+ match self.sql_bind_style {
+ SqlBindStyle::DollarNumeric => {
+ let mut query = query.to_owned();
+ let mut i = 1;
+ while let Some(pos) = query.find('?') {
Review Comment:
thanks! I have implemented it as a fold()
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -141,21 +142,24 @@ impl SqlCatalog {
}
/// SQLX Any does not implement PostgresSQL bindings, so we have to do this.
- pub async fn execute_statement(
- &self,
- query: &String,
- args: Vec<Option<&String>>,
- ) -> Result<Vec<AnyRow>> {
- let query_with_placeholders: Cow<str> =
- if self.sql_bind_style == SqlBindStyle::DollarNumeric {
- let mut query = query.clone();
- for i in 0..args.len() {
- query = query.replacen("?", &format!("${}", i + 1), 1);
+ pub fn build_query(&self, query: &str) -> String {
+ match self.sql_bind_style {
+ SqlBindStyle::DollarNumeric => {
+ let mut query = query.to_owned();
+ let mut i = 1;
+ while let Some(pos) = query.find('?') {
+ query.replace_range(pos..pos + 1, &format!("${}", i));
+ i += 1;
}
- Cow::Owned(query)
- } else {
- Cow::Borrowed(query)
- };
+ query
+ }
+ _ => query.to_owned(),
+ }
+ }
+
+ /// Fetch a vec of AnyRows from a given query
+ pub async fn fetch_rows(&self, query: &str, args: Vec<Option<&String>>) -> Result<Vec<AnyRow>> {
Review Comment:
fixed in a few places
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "sdd (via GitHub)" <gi...@apache.org>.
sdd commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1714279852
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,344 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ pub async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ let table_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+ let namespaces_stmt = format!(
+ "SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ match parent {
+ Some(parent) => match self.namespace_exists(parent).await? {
+ true => {
+ let parent_str = parent.join(".");
+ let parent_table_namespaces_stmt = format!(
+ "{table_namespaces_stmt} AND {CATALOG_FIELD_TABLE_NAMESPACE} LIKE CONCAT(?, '%')"
+ );
+ let parent_namespaces_stmt =
+ format!("{namespaces_stmt} AND {NAMESPACE_FIELD_NAME} LIKE CONCAT(?, '%')");
+
+ let namespace_rows = self
+ .fetch_rows(
+ &format!(
+ "{parent_namespaces_stmt} UNION {parent_table_namespaces_stmt}"
+ ),
+ vec![
+ Some(&self.name),
+ Some(&parent_str),
+ Some(&self.name),
+ Some(&parent_str),
+ ],
+ )
+ .await?;
+
+ Ok(namespace_rows
+ .iter()
+ .filter_map(|r| {
+ let nsp = r.try_get::<String, _>(0).ok();
+ nsp.and_then(|n| {
+ if n == parent_str {
+ // Filter out itself
+ None
+ } else {
+ NamespaceIdent::from_strs(n.split(".")).ok()
+ }
+ })
+ })
+ .collect())
+ }
+ false => no_such_namespace_err(parent),
+ },
+ None => {
+ let namespace_rows = self
+ .fetch_rows(
+ &format!("{namespaces_stmt} UNION {table_namespaces_stmt}"),
+ vec![Some(&self.name), Some(&self.name)],
+ )
+ .await?;
+
+ Ok(namespace_rows
+ .iter()
+ .filter_map(|r| {
+ let nsp = r.try_get::<String, _>(0).ok();
Review Comment:
ditto
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "callum-ryan (via GitHub)" <gi...@apache.org>.
callum-ryan commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1725530833
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -141,21 +142,30 @@ impl SqlCatalog {
}
/// SQLX Any does not implement PostgresSQL bindings, so we have to do this.
- pub async fn execute_statement(
- &self,
- query: &String,
- args: Vec<Option<&String>>,
- ) -> Result<Vec<AnyRow>> {
- let query_with_placeholders: Cow<str> =
- if self.sql_bind_style == SqlBindStyle::DollarNumeric {
- let mut query = query.clone();
- for i in 0..args.len() {
- query = query.replacen("?", &format!("${}", i + 1), 1);
- }
- Cow::Owned(query)
- } else {
- Cow::Borrowed(query)
- };
+ fn replace_placeholders(&self, query: &str) -> String {
+ match self.sql_bind_style {
Review Comment:
I originally had this behaviour, grabbing the style from the SQL pool based on the type of database, I changed it due to @Xuanwo's comment here https://github.com/apache/iceberg-rust/pull/524#discussion_r1704892006.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 merged PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: SQL Catalog - namespaces [iceberg-rust]
Posted by "callum-ryan (via GitHub)" <gi...@apache.org>.
callum-ryan commented on code in PR #534:
URL: https://github.com/apache/iceberg-rust/pull/534#discussion_r1725535153
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -167,43 +177,312 @@ impl SqlCatalog {
.await
.map_err(from_sqlx_error)
}
+
+ /// Execute statements in a transaction, provided or not
+ async fn execute(
+ &self,
+ query: &str,
+ args: Vec<Option<&str>>,
+ transaction: Option<&mut Transaction<'_, Any>>,
+ ) -> Result<AnyQueryResult> {
+ let query_with_placeholders = self.replace_placeholders(query);
+
+ let mut sqlx_query = sqlx::query(&query_with_placeholders);
+ for arg in args {
+ sqlx_query = sqlx_query.bind(arg);
+ }
+
+ match transaction {
+ Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
+ None => {
+ let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
+ let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
+ let _ = tx.commit().await.map_err(from_sqlx_error);
+ result
+ }
+ }
+ }
}
#[async_trait]
impl Catalog for SqlCatalog {
async fn list_namespaces(
&self,
- _parent: Option<&NamespaceIdent>,
+ parent: Option<&NamespaceIdent>,
) -> Result<Vec<NamespaceIdent>> {
- todo!()
+ // UNION will remove duplicates.
+ let all_namespaces_stmt = format!(
+ "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
+ FROM {CATALOG_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
+ UNION
+ SELECT {NAMESPACE_FIELD_NAME}
+ FROM {NAMESPACE_TABLE_NAME}
+ WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
+ );
+
+ let namespace_rows = self
+ .fetch_rows(&all_namespaces_stmt, vec![
+ Some(&self.name),
+ Some(&self.name),
+ ])
+ .await?;
+
+ let mut namespaces = HashSet::<NamespaceIdent>::with_capacity(namespace_rows.len());
+
+ if let Some(parent) = parent {
+ if self.namespace_exists(parent).await? {
+ let parent_str = parent.join(".");
+
+ for row in namespace_rows.iter() {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ // if parent = a, then we only want to see a.b, a.c returned.
+ if nsp != parent_str && nsp.starts_with(&parent_str) {
+ namespaces.insert(NamespaceIdent::from_strs(nsp.split("."))?);
+ }
+ }
+
+ Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
+ } else {
+ no_such_namespace_err(parent)
+ }
+ } else {
+ for row in namespace_rows.iter() {
+ let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
+ let mut levels = nsp.split(".").collect::<Vec<&str>>();
+ if !levels.is_empty() {
+ let first_level = levels.drain(..1).collect::<Vec<&str>>();
+ namespaces.insert(NamespaceIdent::from_strs(first_level)?);
+ }
+ }
+
+ Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
+ }
}
async fn create_namespace(
&self,
- _namespace: &NamespaceIdent,
- _properties: HashMap<String, String>,
+ namespace: &NamespaceIdent,
+ properties: HashMap<String, String>,
) -> Result<Namespace> {
- todo!()
+ let exists = self.namespace_exists(namespace).await?;
+
+ if exists {
+ return Err(Error::new(
+ iceberg::ErrorKind::Unexpected,
+ format!("Namespace {:?} already exists", namespace),
+ ));
+ }
+
+ for i in 1..namespace.len() {
+ let parent_namespace = NamespaceIdent::from_vec(namespace[..i].to_vec())?;
+ let parent_exists = self.namespace_exists(&parent_namespace).await?;
+ if !parent_exists {
+ return no_such_namespace_err(&parent_namespace);
+ }
+ }
+
+ let namespace_str = namespace.join(".");
+ let insert = format!(
+ "INSERT INTO {NAMESPACE_TABLE_NAME} ({CATALOG_FIELD_CATALOG_NAME}, {NAMESPACE_FIELD_NAME}, {NAMESPACE_FIELD_PROPERTY_KEY}, {NAMESPACE_FIELD_PROPERTY_VALUE})
Review Comment:
`NAMESPACE_TABLE_NAME` is the variable name of the `iceberg_namespace_properties` table.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org