diff options
author | Andrew Or <andrew@databricks.com> | 2016-04-27 14:17:36 -0700 |
---|---|---|
committer | Reynold Xin <rxin@databricks.com> | 2016-04-27 14:17:36 -0700 |
commit | 37575115b98fdc9ebadb2ebcbcd9907a3af1076c (patch) | |
tree | 10cc2116017dbda327b8e714bd3c193f870b7a30 /sql/catalyst/src/main | |
parent | 4672e9838b130d006965efeba2665676aa995ebc (diff) | |
download | spark-37575115b98fdc9ebadb2ebcbcd9907a3af1076c.tar.gz spark-37575115b98fdc9ebadb2ebcbcd9907a3af1076c.tar.bz2 spark-37575115b98fdc9ebadb2ebcbcd9907a3af1076c.zip |
[SPARK-14940][SQL] Move ExternalCatalog to own file
## What changes were proposed in this pull request?
`interfaces.scala` was getting big. This just moves the biggest class in there to a new file for cleanliness.
## How was this patch tested?
Just moving things around.
Author: Andrew Or <andrew@databricks.com>
Closes #12721 from andrewor14/move-external-catalog.
Diffstat (limited to 'sql/catalyst/src/main')
5 files changed, 190 insertions, 170 deletions
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/NoSuchItemException.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/NoSuchItemException.scala index 5e18316c94..11ef9e1160 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/NoSuchItemException.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/NoSuchItemException.scala @@ -18,7 +18,7 @@ package org.apache.spark.sql.catalyst.analysis import org.apache.spark.sql.AnalysisException -import org.apache.spark.sql.catalyst.catalog.ExternalCatalog.TablePartitionSpec +import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec /** diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/ExternalCatalog.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/ExternalCatalog.scala new file mode 100644 index 0000000000..178ae6d7c2 --- /dev/null +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/ExternalCatalog.scala @@ -0,0 +1,185 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.catalyst.catalog + +import org.apache.spark.sql.AnalysisException + + +/** + * Interface for the system catalog (of columns, partitions, tables, and databases). + * + * This is only used for non-temporary items, and implementations must be thread-safe as they + * can be accessed in multiple threads. This is an external catalog because it is expected to + * interact with external systems. + * + * Implementations should throw [[AnalysisException]] when table or database don't exist. + */ +abstract class ExternalCatalog { + import CatalogTypes.TablePartitionSpec + + protected def requireDbExists(db: String): Unit = { + if (!databaseExists(db)) { + throw new AnalysisException(s"Database '$db' does not exist") + } + } + + // -------------------------------------------------------------------------- + // Databases + // -------------------------------------------------------------------------- + + def createDatabase(dbDefinition: CatalogDatabase, ignoreIfExists: Boolean): Unit + + def dropDatabase(db: String, ignoreIfNotExists: Boolean, cascade: Boolean): Unit + + /** + * Alter a database whose name matches the one specified in `dbDefinition`, + * assuming the database exists. + * + * Note: If the underlying implementation does not support altering a certain field, + * this becomes a no-op. + */ + def alterDatabase(dbDefinition: CatalogDatabase): Unit + + def getDatabase(db: String): CatalogDatabase + + def databaseExists(db: String): Boolean + + def listDatabases(): Seq[String] + + def listDatabases(pattern: String): Seq[String] + + def setCurrentDatabase(db: String): Unit + + // -------------------------------------------------------------------------- + // Tables + // -------------------------------------------------------------------------- + + def createTable(db: String, tableDefinition: CatalogTable, ignoreIfExists: Boolean): Unit + + def dropTable(db: String, table: String, ignoreIfNotExists: Boolean): Unit + + def renameTable(db: String, oldName: String, newName: String): Unit + + /** + * Alter a table whose name that matches the one specified in `tableDefinition`, + * assuming the table exists. + * + * Note: If the underlying implementation does not support altering a certain field, + * this becomes a no-op. + */ + def alterTable(db: String, tableDefinition: CatalogTable): Unit + + def getTable(db: String, table: String): CatalogTable + + def getTableOption(db: String, table: String): Option[CatalogTable] + + def tableExists(db: String, table: String): Boolean + + def listTables(db: String): Seq[String] + + def listTables(db: String, pattern: String): Seq[String] + + def loadTable( + db: String, + table: String, + loadPath: String, + isOverwrite: Boolean, + holdDDLTime: Boolean): Unit + + def loadPartition( + db: String, + table: String, + loadPath: String, + partition: TablePartitionSpec, + isOverwrite: Boolean, + holdDDLTime: Boolean, + inheritTableSpecs: Boolean, + isSkewedStoreAsSubdir: Boolean): Unit + + // -------------------------------------------------------------------------- + // Partitions + // -------------------------------------------------------------------------- + + def createPartitions( + db: String, + table: String, + parts: Seq[CatalogTablePartition], + ignoreIfExists: Boolean): Unit + + def dropPartitions( + db: String, + table: String, + parts: Seq[TablePartitionSpec], + ignoreIfNotExists: Boolean): Unit + + /** + * Override the specs of one or many existing table partitions, assuming they exist. + * This assumes index i of `specs` corresponds to index i of `newSpecs`. + */ + def renamePartitions( + db: String, + table: String, + specs: Seq[TablePartitionSpec], + newSpecs: Seq[TablePartitionSpec]): Unit + + /** + * Alter one or many table partitions whose specs that match those specified in `parts`, + * assuming the partitions exist. + * + * Note: If the underlying implementation does not support altering a certain field, + * this becomes a no-op. + */ + def alterPartitions( + db: String, + table: String, + parts: Seq[CatalogTablePartition]): Unit + + def getPartition(db: String, table: String, spec: TablePartitionSpec): CatalogTablePartition + + /** + * List the metadata of all partitions that belong to the specified table, assuming it exists. + * + * A partial partition spec may optionally be provided to filter the partitions returned. + * For instance, if there exist partitions (a='1', b='2'), (a='1', b='3') and (a='2', b='4'), + * then a partial spec of (a='1') will return the first two only. + * @param db database name + * @param table table name + * @param partialSpec partition spec + */ + def listPartitions( + db: String, + table: String, + partialSpec: Option[TablePartitionSpec] = None): Seq[CatalogTablePartition] + + // -------------------------------------------------------------------------- + // Functions + // -------------------------------------------------------------------------- + + def createFunction(db: String, funcDefinition: CatalogFunction): Unit + + def dropFunction(db: String, funcName: String): Unit + + def renameFunction(db: String, oldName: String, newName: String): Unit + + def getFunction(db: String, funcName: String): CatalogFunction + + def functionExists(db: String, funcName: String): Boolean + + def listFunctions(db: String, pattern: String): Seq[String] + +} diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/InMemoryCatalog.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/InMemoryCatalog.scala index 28a67067d0..60eb7329f7 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/InMemoryCatalog.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/InMemoryCatalog.scala @@ -33,7 +33,7 @@ import org.apache.spark.sql.catalyst.util.StringUtils * All public methods should be synchronized for thread-safety. */ class InMemoryCatalog extends ExternalCatalog { - import ExternalCatalog._ + import CatalogTypes.TablePartitionSpec private class TableDesc(var table: CatalogTable) { val partitions = new mutable.HashMap[TablePartitionSpec, CatalogTablePartition] diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala index 91d35de790..d7fd54308a 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala @@ -45,7 +45,7 @@ class SessionCatalog( functionResourceLoader: FunctionResourceLoader, functionRegistry: FunctionRegistry, conf: CatalystConf) extends Logging { - import ExternalCatalog._ + import CatalogTypes.TablePartitionSpec def this( externalCatalog: ExternalCatalog, diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala index d1e2b3f664..5efaf8f201 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala @@ -27,171 +27,6 @@ import org.apache.spark.sql.catalyst.plans.logical.{LeafNode, LogicalPlan} /** - * Interface for the system catalog (of columns, partitions, tables, and databases). - * - * This is only used for non-temporary items, and implementations must be thread-safe as they - * can be accessed in multiple threads. This is an external catalog because it is expected to - * interact with external systems. - * - * Implementations should throw [[AnalysisException]] when table or database don't exist. - */ -abstract class ExternalCatalog { - import ExternalCatalog._ - - protected def requireDbExists(db: String): Unit = { - if (!databaseExists(db)) { - throw new AnalysisException(s"Database '$db' does not exist") - } - } - - // -------------------------------------------------------------------------- - // Databases - // -------------------------------------------------------------------------- - - def createDatabase(dbDefinition: CatalogDatabase, ignoreIfExists: Boolean): Unit - - def dropDatabase(db: String, ignoreIfNotExists: Boolean, cascade: Boolean): Unit - - /** - * Alter a database whose name matches the one specified in `dbDefinition`, - * assuming the database exists. - * - * Note: If the underlying implementation does not support altering a certain field, - * this becomes a no-op. - */ - def alterDatabase(dbDefinition: CatalogDatabase): Unit - - def getDatabase(db: String): CatalogDatabase - - def databaseExists(db: String): Boolean - - def listDatabases(): Seq[String] - - def listDatabases(pattern: String): Seq[String] - - def setCurrentDatabase(db: String): Unit - - // -------------------------------------------------------------------------- - // Tables - // -------------------------------------------------------------------------- - - def createTable(db: String, tableDefinition: CatalogTable, ignoreIfExists: Boolean): Unit - - def dropTable(db: String, table: String, ignoreIfNotExists: Boolean): Unit - - def renameTable(db: String, oldName: String, newName: String): Unit - - /** - * Alter a table whose name that matches the one specified in `tableDefinition`, - * assuming the table exists. - * - * Note: If the underlying implementation does not support altering a certain field, - * this becomes a no-op. - */ - def alterTable(db: String, tableDefinition: CatalogTable): Unit - - def getTable(db: String, table: String): CatalogTable - - def getTableOption(db: String, table: String): Option[CatalogTable] - - def tableExists(db: String, table: String): Boolean - - def listTables(db: String): Seq[String] - - def listTables(db: String, pattern: String): Seq[String] - - def loadTable( - db: String, - table: String, - loadPath: String, - isOverwrite: Boolean, - holdDDLTime: Boolean): Unit - - def loadPartition( - db: String, - table: String, - loadPath: String, - partition: TablePartitionSpec, - isOverwrite: Boolean, - holdDDLTime: Boolean, - inheritTableSpecs: Boolean, - isSkewedStoreAsSubdir: Boolean): Unit - - // -------------------------------------------------------------------------- - // Partitions - // -------------------------------------------------------------------------- - - def createPartitions( - db: String, - table: String, - parts: Seq[CatalogTablePartition], - ignoreIfExists: Boolean): Unit - - def dropPartitions( - db: String, - table: String, - parts: Seq[TablePartitionSpec], - ignoreIfNotExists: Boolean): Unit - - /** - * Override the specs of one or many existing table partitions, assuming they exist. - * This assumes index i of `specs` corresponds to index i of `newSpecs`. - */ - def renamePartitions( - db: String, - table: String, - specs: Seq[TablePartitionSpec], - newSpecs: Seq[TablePartitionSpec]): Unit - - /** - * Alter one or many table partitions whose specs that match those specified in `parts`, - * assuming the partitions exist. - * - * Note: If the underlying implementation does not support altering a certain field, - * this becomes a no-op. - */ - def alterPartitions( - db: String, - table: String, - parts: Seq[CatalogTablePartition]): Unit - - def getPartition(db: String, table: String, spec: TablePartitionSpec): CatalogTablePartition - - /** - * List the metadata of all partitions that belong to the specified table, assuming it exists. - * - * A partial partition spec may optionally be provided to filter the partitions returned. - * For instance, if there exist partitions (a='1', b='2'), (a='1', b='3') and (a='2', b='4'), - * then a partial spec of (a='1') will return the first two only. - * @param db database name - * @param table table name - * @param partialSpec partition spec - */ - def listPartitions( - db: String, - table: String, - partialSpec: Option[TablePartitionSpec] = None): Seq[CatalogTablePartition] - - // -------------------------------------------------------------------------- - // Functions - // -------------------------------------------------------------------------- - - def createFunction(db: String, funcDefinition: CatalogFunction): Unit - - def dropFunction(db: String, funcName: String): Unit - - def renameFunction(db: String, oldName: String, newName: String): Unit - - def getFunction(db: String, funcName: String): CatalogFunction - - def functionExists(db: String, funcName: String): Boolean - - def listFunctions(db: String, pattern: String): Seq[String] - -} - - -/** * A function defined in the catalog. * * @param identifier name of the function @@ -235,7 +70,7 @@ case class CatalogColumn( * @param storage storage format of the partition */ case class CatalogTablePartition( - spec: ExternalCatalog.TablePartitionSpec, + spec: CatalogTypes.TablePartitionSpec, storage: CatalogStorageFormat) @@ -316,7 +151,7 @@ case class CatalogDatabase( properties: Map[String, String]) -object ExternalCatalog { +object CatalogTypes { /** * Specifications of a table partition. Mapping column name to column value. */ |