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-unsubscr...@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For additional commands, e-mail: issues-h...@iceberg.apache.org
