case class CreateTableCommand(table: CatalogTable, ifNotExists: Boolean) extends RunnableCommand {

  override def run(sparkSession: SparkSession): Seq[Row] = {

    sparkSession.sessionState.catalog.createTable(table, ifNotExists)

    Seq.empty[Row]

  }

}

  /**

   * Internal catalog for managing table and database states.

   */

  lazy val catalog = new SessionCatalog(

    sparkSession.sharedState.externalCatalog,

    sparkSession.sharedState.globalTempViewManager,

    functionResourceLoader,

    functionRegistry,

    conf,

    newHadoopConf())

  /**

   * A catalog that interacts with external systems.

   */

  val externalCatalog: ExternalCatalog =

    SharedState.reflect[ExternalCatalog, SparkConf, Configuration](

      SharedState.externalCatalogClassName(sparkContext.conf),

      sparkContext.conf,

      sparkContext.hadoopConfiguration)

...

  private val HIVE_EXTERNAL_CATALOG_CLASS_NAME = "org.apache.spark.sql.hive.HiveExternalCatalog"

  private def externalCatalogClassName(conf: SparkConf): String = {

    conf.get(CATALOG_IMPLEMENTATION) match {

      case "hive" => HIVE_EXTERNAL_CATALOG_CLASS_NAME

      case "in-memory" => classOf[InMemoryCatalog].getCanonicalName

    }

  }

  /**

   * A Hive client used to interact with the metastore.

   */

  val client: HiveClient = {

    HiveUtils.newClientForMetadata(conf, hadoopConf)

  }

  private def withClient[T](body: => T): T = synchronized {

    try {

      body

    } catch {

      case NonFatal(exception) if isClientException(exception) =>

        val e = exception match {

          // Since we are using shim, the exceptions thrown by the underlying method of

          // Method.invoke() are wrapped by InvocationTargetException

          case i: InvocationTargetException => i.getCause

          case o => o

        }

        throw new AnalysisException(

          e.getClass.getCanonicalName + ": " + e.getMessage, cause = Some(e))

    }

  }

  override def createDatabase(

      dbDefinition: CatalogDatabase,

      ignoreIfExists: Boolean): Unit = withClient {

    client.createDatabase(dbDefinition, ignoreIfExists)

  }

  /** The isolated client interface to Hive. */

  private[hive] def createClient(): HiveClient = {

    if (!isolationOn) {

      return new HiveClientImpl(version, sparkConf, hadoopConf, config, baseClassLoader, this)

    }

    // Pre-reflective instantiation setup.

    logDebug("Initializing the logger to avoid disaster...")

    val origLoader = Thread.currentThread().getContextClassLoader

    Thread.currentThread.setContextClassLoader(classLoader)

    try {

      classLoader

        .loadClass(classOf[HiveClientImpl].getName)

        .getConstructors.head

        .newInstance(version, sparkConf, hadoopConf, config, classLoader, this)

        .asInstanceOf[HiveClient]

    } catch {

  def withHiveState[A](f: => A): A = retryLocked {

    val original = Thread.currentThread().getContextClassLoader

    // Set the thread local metastore client to the client associated with this HiveClientImpl.

    Hive.set(client)

    // The classloader in clientLoader could be changed after addJar, always use the latest

    // classloader

    state.getConf.setClassLoader(clientLoader.classLoader)

    // setCurrentSessionState will use the classLoader associated

    // with the HiveConf in `state` to override the context class loader of the current

    // thread.

    shim.setCurrentSessionState(state)

    val ret = try f finally {

      Thread.currentThread().setContextClassLoader(original)

      HiveCatalogMetrics.incrementHiveClientCalls(1)

    }

    ret

  }

  private def retryLocked[A](f: => A): A = clientLoader.synchronized {

...

  override def createDatabase(

      database: CatalogDatabase,

      ignoreIfExists: Boolean): Unit = withHiveState {

    client.createDatabase(

      new HiveDatabase(

        database.name,

        database.description,

        database.locationUri,

        Option(database.properties).map(_.asJava).orNull),

        ignoreIfExists)

  }