[SPARK-55250][SQL] Reduce Hive client calls on CREATE NAMESPACE

[pan3793]

What changes were proposed in this pull request?

This PR reduces Hive client calls by eliminating unnecessary catalog.databaseExists in CreateNamespaceExec. Now the Hive client calls of CREATE NAMESPACE [IF NOT EXISTS] foo.bar decreased from 3 to 1.

Why are the changes needed?

Improve perf by reducing RPC.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

UT is added.

Was this patch authored or co-authored using generative AI tooling?

No.

[github-actions]

JIRA Issue Information

=== Improvement SPARK-55250 ===
Summary: Reduce Hive client calls on CREATE NAMESPACE
Assignee: None
Status: Open
Affected: ["4.2.0"]

---

This comment was automatically generated by GitHub Actions

[pan3793]

we don't need this check because createNamespace will throw NamespaceAlreadyExistsException if the namespace already exists, by contract.

spark/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsNamespaces.java

Lines 121 to 131 in 9ce067c

	/**
	* Create a namespace in the catalog.
	*
	* @param namespace a multi-part namespace
	* @param metadata a string map of properties for the given namespace
	* @throws NamespaceAlreadyExistsException If the namespace already exists
	* @throws UnsupportedOperationException If create is not a supported operation
	*/
	void createNamespace(
	String[] namespace,
	Map<String, String> metadata) throws NamespaceAlreadyExistsException;

this also makes it atomic, previously, there are chance that the namespace is created by another request between catalog.namespaceExists and catalog.createNamespace, we should delegate it to the Connector to handle that.

[cloud-fan]

can you pint out where Spark throws NamespaceAlreadyExistsException? In HiveExternalCatalog or HiveClient?

[pan3793]

happens here

spark/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala

Lines 345 to 355 in 589fedc

	override def createDatabase(
	database: CatalogDatabase,
	ignoreIfExists: Boolean): Unit = withHiveState {
	val hiveDb = toHiveDatabase(database, Some(userName))
	try {
	shim.createDatabase(client, hiveDb, ignoreIfExists)
	} catch {
	case _: AlreadyExistsException =>
	throw new DatabaseAlreadyExistsException(database.name)
	}
	}

spark/sql/api/src/main/scala/org/apache/spark/sql/catalyst/analysis/alreadyExistException.scala

Lines 31 to 32 in 589fedc

	class DatabaseAlreadyExistsException(db: String)
	extends NamespaceAlreadyExistsException(Array(db))