From 598d685ac271bf318764d698bb4f1102a5786ec4 Mon Sep 17 00:00:00 2001
From: jingzhang <beyond1920@126.com>
Date: Fri, 17 Mar 2017 20:51:37 +0800
Subject: [PATCH 1/4] Introduce interface for external catalog, and provide an
 in-memory implementation for test or develop. Integrate with calcite catalog.

---
 flink-libraries/flink-table/pom.xml           |  20 ++-
 .../flink/table/annotation/TableType.java     |  45 +++++
 .../resources/tableSourceConverter.properties |  19 +++
 .../apache/flink/table/api/exceptions.scala   |  94 +++++++++++
 .../table/catalog/CRUDExternalCatalog.scala   | 110 +++++++++++++
 .../catalog/ExternalCatalogDatabase.scala     |  31 ++++
 .../table/catalog/ExternalCatalogSchema.scala | 155 ++++++++++++++++++
 .../table/catalog/ExternalCatalogTable.scala  |  76 +++++++++
 .../catalog/ExternalTableSourceUtil.scala     | 146 +++++++++++++++++
 .../catalog/InMemoryExternalCatalog.scala     | 155 ++++++++++++++++++
 .../catalog/ReadonlyExternalCatalog.scala     |  72 ++++++++
 .../table/catalog/TableSourceConverter.scala  |  48 ++++++
 .../sources/CsvTableSourceConverter.scala     |  73 +++++++++
 .../catalog/ExternalCatalogSchemaTest.scala   |  88 ++++++++++
 .../catalog/InMemoryExternalCatalogTest.scala | 145 ++++++++++++++++
 .../flink/table/utils/CommonTestData.scala    |  89 +++++++++-
 16 files changed, 1357 insertions(+), 9 deletions(-)
 create mode 100644 flink-libraries/flink-table/src/main/java/org/apache/flink/table/annotation/TableType.java
 create mode 100644 flink-libraries/flink-table/src/main/resources/tableSourceConverter.properties
 create mode 100644 flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/CRUDExternalCatalog.scala
 create mode 100644 flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalCatalogDatabase.scala
 create mode 100644 flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalCatalogSchema.scala
 create mode 100644 flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalCatalogTable.scala
 create mode 100644 flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalTableSourceUtil.scala
 create mode 100644 flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/InMemoryExternalCatalog.scala
 create mode 100644 flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ReadonlyExternalCatalog.scala
 create mode 100644 flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/TableSourceConverter.scala
 create mode 100644 flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/CsvTableSourceConverter.scala
 create mode 100644 flink-libraries/flink-table/src/test/scala/org/apache/flink/table/catalog/ExternalCatalogSchemaTest.scala
 create mode 100644 flink-libraries/flink-table/src/test/scala/org/apache/flink/table/catalog/InMemoryExternalCatalogTest.scala

diff --git a/flink-libraries/flink-table/pom.xml b/flink-libraries/flink-table/pom.xml
index b26fe542acd35..a2945e87d294e 100644
--- a/flink-libraries/flink-table/pom.xml
+++ b/flink-libraries/flink-table/pom.xml
@@ -92,6 +92,22 @@ under the License.
 			</exclusions>
 		</dependency>
 
+        <dependency>
+            <groupId>org.reflections</groupId>
+            <artifactId>reflections</artifactId>
+            <version>0.9.10</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.google.code.findbugs</groupId>
+                    <artifactId>annotations</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>com.google.guava</groupId>
+                    <artifactId>guava</artifactId>
+                </exclusion>
+            </exclusions>
+            <scope>compile</scope>
+        </dependency>
 
 		<!-- test dependencies -->
 
@@ -178,7 +194,8 @@ under the License.
 									<include>org.apache.calcite:*</include>
 									<include>org.apache.calcite.avatica:*</include>
 									<include>net.hydromatic:*</include>
-								</includes>
+                                    <include>org.reflections:*</include>
+                                </includes>
 							</artifactSet>
 							<relocations>
 								<!-- We currently don't relocate slf4j as we have "logger not found" 
@@ -216,5 +233,4 @@ under the License.
 
 		</plugins>
 	</build>
-
 </project>
diff --git a/flink-libraries/flink-table/src/main/java/org/apache/flink/table/annotation/TableType.java b/flink-libraries/flink-table/src/main/java/org/apache/flink/table/annotation/TableType.java
new file mode 100644
index 0000000000000..1cebe53fbf5cc
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/java/org/apache/flink/table/annotation/TableType.java
@@ -0,0 +1,45 @@
+/*
+ * 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.flink.table.annotation;
+
+import org.apache.flink.annotation.Public;
+import org.apache.flink.table.catalog.TableSourceConverter;
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * A {@link TableSourceConverter} with this annotation bind the converter with table type.
+ */
+@Documented
+@Target(ElementType.TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+@Public
+public @interface TableType {
+
+	/**
+	 * Specifies the external catalog table type of {@link TableSourceConverter}.
+	 *
+	 * @return the external catalog table type of {@link TableSourceConverter}.
+	 */
+	String value();
+
+}
diff --git a/flink-libraries/flink-table/src/main/resources/tableSourceConverter.properties b/flink-libraries/flink-table/src/main/resources/tableSourceConverter.properties
new file mode 100644
index 0000000000000..7e2bb962a54a3
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/resources/tableSourceConverter.properties
@@ -0,0 +1,19 @@
+################################################################################
+#  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.
+################################################################################
+
+scan.package=org.apache.flink.table.sources
\ No newline at end of file
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/exceptions.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/exceptions.scala
index 7ca45a6cfb45f..8632436d761e4 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/exceptions.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/exceptions.scala
@@ -18,6 +18,8 @@
 
 package org.apache.flink.table.api
 
+import org.apache.flink.table.catalog.TableSourceConverter
+
 /**
   * Exception for all errors occurring during expression parsing.
   */
@@ -71,3 +73,95 @@ object ValidationException {
   * Exception for unwanted method calling on unresolved expression.
   */
 case class UnresolvedException(msg: String) extends RuntimeException(msg)
+
+/**
+  * Exception for operation on a nonexistent table
+  *
+  * @param db    database name
+  * @param table table name
+  * @param cause
+  */
+case class TableNotExistException(
+    db: String,
+    table: String,
+    cause: Throwable)
+    extends RuntimeException(s"table $db.$table does not exist!", cause) {
+
+  def this(db: String, table: String) = this(db, table, null)
+
+}
+
+/**
+  * Exception for adding an already existed table
+  *
+  * @param db    database name
+  * @param table table name
+  * @param cause
+  */
+case class TableAlreadyExistException(
+    db: String,
+    table: String,
+    cause: Throwable)
+    extends RuntimeException(s"table $db.$table already exists!", cause) {
+
+  def this(db: String, table: String) = this(db, table, null)
+
+}
+
+/**
+  * Exception for operation on a nonexistent database
+  *
+  * @param db database name
+  * @param cause
+  */
+case class DatabaseNotExistException(
+    db: String,
+    cause: Throwable)
+    extends RuntimeException(s"database $db does not exist!", cause) {
+
+  def this(db: String) = this(db, null)
+}
+
+/**
+  * Exception for adding an already existed database
+  *
+  * @param db database name
+  * @param cause
+  */
+case class DatabaseAlreadyExistException(
+    db: String,
+    cause: Throwable)
+    extends RuntimeException(s"database $db already exists!", cause) {
+
+  def this(db: String) = this(db, null)
+}
+
+/**
+  * Exception for does not find any matched [[TableSourceConverter]] for a specified table type
+  *
+  * @param tableType table type
+  * @param cause
+  */
+case class NoMatchedTableSourceConverterException(
+    tableType: String,
+    cause: Throwable)
+    extends RuntimeException(s"find no table source converter matched table type $tableType!",
+      cause) {
+
+  def this(tableType: String) = this(tableType, null)
+}
+
+/**
+  * Exception for find more than one matched [[TableSourceConverter]] for a specified table type
+  *
+  * @param tableType table type
+  * @param cause
+  */
+case class AmbiguousTableSourceConverterException(
+    tableType: String,
+    cause: Throwable)
+    extends RuntimeException(s"more than one table source converter matched table type $tableType!",
+      cause) {
+
+  def this(tableType: String) = this(tableType, null)
+}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/CRUDExternalCatalog.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/CRUDExternalCatalog.scala
new file mode 100644
index 0000000000000..8f98256ed5bec
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/CRUDExternalCatalog.scala
@@ -0,0 +1,110 @@
+/*
+ * 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.flink.table.catalog
+
+import org.apache.flink.table.api._
+
+/**
+  * This class is responsible for interact with external catalog.
+  * Its main responsibilities including:
+  * <ul>
+  * <li> create/drop/alter database or tables for DDL operations
+  * <li> provide tables for calcite catalog, it looks up databases or tables in the external catalog
+  * </ul>
+  */
+trait CRUDExternalCatalog extends ReadonlyExternalCatalog {
+
+  /**
+    * Adds table into external Catalog
+    *
+    * @param table          description of table which to create
+    * @param ignoreIfExists if table already exists in the catalog, not throw exception and leave
+    *                       the existed table if ignoreIfExists is true;
+    *                       else throw a TableAlreadyExistException.
+    * @throws DatabaseNotExistException  if database does not exist in the catalog yet
+    * @throws TableAlreadyExistException if table already exists in the catalog and
+    *                                    ignoreIfExists is false
+    */
+  @throws[DatabaseNotExistException]
+  @throws[TableAlreadyExistException]
+  def createTable(table: ExternalCatalogTable, ignoreIfExists: Boolean): Unit
+
+  /**
+    * Deletes table from external Catalog
+    *
+    * @param dbName            database name
+    * @param tableName         table name
+    * @param ignoreIfNotExists if table not exist yet, not throw exception if ignoreIfNotExists is
+    *                          true; else throw TableNotExistException
+    * @throws DatabaseNotExistException if database does not exist in the catalog yet
+    * @throws TableNotExistException    if table does not exist in the catalog yet
+    */
+  @throws[DatabaseNotExistException]
+  @throws[TableNotExistException]
+  def dropTable(dbName: String, tableName: String, ignoreIfNotExists: Boolean): Unit
+
+  /**
+    * Modifies an existing table in the external catalog
+    *
+    * @param table             description of table which to modify
+    * @param ignoreIfNotExists if the table not exist yet, not throw exception if ignoreIfNotExists
+    *                          is true; else throw TableNotExistException
+    * @throws DatabaseNotExistException if database does not exist in the catalog yet
+    * @throws TableNotExistException    if table does not exist in the catalog yet
+    */
+  @throws[DatabaseNotExistException]
+  @throws[TableNotExistException]
+  def alterTable(table: ExternalCatalogTable, ignoreIfNotExists: Boolean): Unit
+
+  /**
+    * Adds database into external Catalog
+    *
+    * @param db             description of database which to create
+    * @param ignoreIfExists if database already exists in the catalog, not throw exception and leave
+    *                       the existed database if ignoreIfExists is true;
+    *                       else throw a DatabaseAlreadyExistException.
+    * @throws DatabaseAlreadyExistException if database already exists in the catalog and
+    *                                       ignoreIfExists is false
+    */
+  @throws[DatabaseAlreadyExistException]
+  def createDatabase(db: ExternalCatalogDatabase, ignoreIfExists: Boolean): Unit
+
+  /**
+    * Deletes database from external Catalog
+    *
+    * @param dbName            database name
+    * @param ignoreIfNotExists if database not exist yet, not throw exception if ignoreIfNotExists
+    *                          is true; else throw DatabaseNotExistException
+    * @throws DatabaseNotExistException if database does not exist in the catalog yet
+    */
+  @throws[DatabaseNotExistException]
+  def dropDatabase(dbName: String, ignoreIfNotExists: Boolean): Unit
+
+  /**
+    * Modifies existed database into external Catalog
+    *
+    * @param db                description of database which to modify
+    * @param ignoreIfNotExists if database not exist yet, not throw exception if ignoreIfNotExists
+    *                          is true; else throw DatabaseNotExistException
+    * @throws DatabaseNotExistException if database does not exist in the catalog yet
+    */
+  @throws[DatabaseNotExistException]
+  def alterDatabase(db: ExternalCatalogDatabase, ignoreIfNotExists: Boolean): Unit
+
+}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalCatalogDatabase.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalCatalogDatabase.scala
new file mode 100644
index 0000000000000..c2a4702d62cab
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalCatalogDatabase.scala
@@ -0,0 +1,31 @@
+/*
+ * 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.flink.table.catalog
+
+import java.util.{HashMap => JHashMap, Map => JMap}
+
+/**
+  * Database definition of the external catalog.
+  *
+  * @param dbName     database name
+  * @param properties database properties
+  */
+case class ExternalCatalogDatabase(
+    dbName: String,
+    properties: JMap[String, String] = new JHashMap())
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalCatalogSchema.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalCatalogSchema.scala
new file mode 100644
index 0000000000000..0357afbbecafb
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalCatalogSchema.scala
@@ -0,0 +1,155 @@
+/*
+ * 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.flink.table.catalog
+
+import java.util.{Collections => JCollections, Collection => JCollection, LinkedHashSet => JLinkedHashSet, Set => JSet}
+
+import org.apache.calcite.linq4j.tree.Expression
+import org.apache.calcite.schema._
+import org.apache.flink.table.api.{DatabaseNotExistException, TableNotExistException}
+import org.slf4j.{Logger, LoggerFactory}
+
+import scala.collection.JavaConverters._
+
+/**
+  * This class is responsible for connect external catalog to calcite catalog.
+  * In this way, it is possible to look-up and access tables in SQL queries
+  * without registering tables in advance.
+  * The databases in the external catalog registers as calcite sub-Schemas of current schema.
+  * The tables in a given database registers as calcite tables
+  * of the [[ExternalCatalogDatabaseSchema]].
+  *
+  * @param catalogIdentifier external catalog name
+  * @param catalog           external catalog
+  */
+class ExternalCatalogSchema(
+    catalogIdentifier: String,
+    catalog: ReadonlyExternalCatalog) extends Schema {
+
+  private val LOG: Logger = LoggerFactory.getLogger(this.getClass)
+
+  /**
+    * Looks up database by the given sub-schema name in the external catalog,
+    * returns it Wrapped in a [[ExternalCatalogDatabaseSchema]] with the given database name.
+    *
+    * @param name Sub-schema name
+    * @return Sub-schema with a given name, or null
+    */
+  override def getSubSchema(name: String): Schema = {
+    try {
+      val db = catalog.getDatabase(name)
+      new ExternalCatalogDatabaseSchema(db.dbName, catalog)
+    } catch {
+      case e: DatabaseNotExistException =>
+        LOG.warn(s"Database $name does not exist in externalCatalog $catalogIdentifier")
+        null
+    }
+  }
+
+  /**
+    * Lists the databases of the external catalog,
+    * returns the lists as the names of this schema's sub-schemas.
+    *
+    * @return names of this schema's child schemas
+    */
+  override def getSubSchemaNames: JSet[String] = new JLinkedHashSet(catalog.listDatabases())
+
+  override def getTable(name: String): Table = null
+
+  override def isMutable: Boolean = true
+
+  override def getFunctions(name: String): JCollection[Function] = JCollections.emptyList[Function]
+
+  override def getExpression(parentSchema: SchemaPlus, name: String): Expression =
+    Schemas.subSchemaExpression(parentSchema, name, getClass)
+
+  override def getFunctionNames: JSet[String] = JCollections.emptySet[String]
+
+  override def getTableNames: JSet[String] = JCollections.emptySet[String]
+
+  override def contentsHaveChangedSince(lastCheck: Long, now: Long): Boolean = true
+
+  /**
+    * Registers sub-Schemas to current schema plus
+    *
+    * @param plusOfThis
+    */
+  def registerSubSchemas(plusOfThis: SchemaPlus) {
+    catalog.listDatabases().asScala.foreach(db => plusOfThis.add(db, getSubSchema(db)))
+  }
+
+  private class ExternalCatalogDatabaseSchema(
+      schemaName: String,
+      flinkExternalCatalog: ReadonlyExternalCatalog) extends Schema {
+
+    override def getTable(name: String): Table = {
+      try {
+        val externalCatalogTable = flinkExternalCatalog.getTable(schemaName, name)
+        ExternalTableSourceUtil.fromExternalCatalogTable(externalCatalogTable)
+      } catch {
+        case TableNotExistException(db, table, cause) => {
+          LOG.warn(s"Table $db.$table does not exist in externalCatalog $catalogIdentifier")
+          null
+        }
+      }
+    }
+
+    override def getTableNames: JSet[String] =
+      new JLinkedHashSet(flinkExternalCatalog.listTables(schemaName))
+
+    override def getSubSchema(name: String): Schema = null
+
+    override def getSubSchemaNames: JSet[String] = JCollections.emptySet[String]
+
+    override def isMutable: Boolean = true
+
+    override def getFunctions(name: String): JCollection[Function] =
+      JCollections.emptyList[Function]
+
+    override def getExpression(parentSchema: SchemaPlus, name: String): Expression =
+      Schemas.subSchemaExpression(parentSchema, name, getClass)
+
+    override def getFunctionNames: JSet[String] = JCollections.emptySet[String]
+
+    override def contentsHaveChangedSince(lastCheck: Long, now: Long): Boolean = true
+
+  }
+
+}
+
+object ExternalCatalogSchema {
+
+  /**
+    * Creates a FlinkExternalCatalogSchema.
+    *
+    * @param parentSchema              Parent schema
+    * @param externalCatalogIdentifier External catalog identifier
+    * @param externalCatalog           External catalog object
+    * @return Created schema
+    */
+  def create(
+      parentSchema: SchemaPlus,
+      externalCatalogIdentifier: String,
+      externalCatalog: ReadonlyExternalCatalog): ExternalCatalogSchema = {
+    val newSchema = new ExternalCatalogSchema(externalCatalogIdentifier, externalCatalog)
+    val schemaPlusOfNewSchema = parentSchema.add(externalCatalogIdentifier, newSchema)
+    newSchema.registerSubSchemas(schemaPlusOfNewSchema)
+    newSchema
+  }
+}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalCatalogTable.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalCatalogTable.scala
new file mode 100644
index 0000000000000..91c90ce2e657a
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalCatalogTable.scala
@@ -0,0 +1,76 @@
+/*
+ * 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.flink.table.catalog
+
+import java.util.{HashMap => JHashMap, Map => JMap}
+import java.lang.{Long => JLong}
+
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.table.plan.stats.TableStats
+
+/**
+  * Table definition of the external catalog.
+  *
+  * @param identifier           identifier of external catalog table, including dbName and tableName
+  * @param tableType            type of external catalog table, e.g csv, hbase, kafka
+  * @param schema               schema of table data, including column names and column types
+  * @param properties           properties of external catalog table
+  * @param stats                statistics of external catalog table
+  * @param comment              comment of external catalog table
+  * @param createTime           create time of external catalog table
+  * @param lastAccessTime       last access time of of external catalog table
+  */
+case class ExternalCatalogTable(
+    identifier: TableIdentifier,
+    tableType: String,
+    schema: DataSchema,
+    properties: JMap[String, String] = new JHashMap(),
+    stats: TableStats = null,
+    comment: String = null,
+    createTime: JLong = System.currentTimeMillis,
+    lastAccessTime: JLong = -1L)
+
+/**
+  * Identifier of external catalog table
+  *
+  * @param database database name
+  * @param table    table name
+  */
+case class TableIdentifier(
+    database: String,
+    table: String) {
+
+  override def toString: String = s"$database.$table"
+
+}
+
+/**
+  * Schema of External catalog table's columns
+  *
+  * @param columnTypes types of each column
+  * @param columnNames names of each column
+  */
+case class DataSchema(
+    columnTypes: Array[TypeInformation[_]],
+    columnNames: Array[String]) {
+
+  override def toString: String =
+    columnNames.zip(columnTypes).map(x => s"${x._1}: ${x._2}").mkString(", ")
+
+}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalTableSourceUtil.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalTableSourceUtil.scala
new file mode 100644
index 0000000000000..c16a8af78aa95
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalTableSourceUtil.scala
@@ -0,0 +1,146 @@
+/*
+ * 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.flink.table.catalog
+
+import java.io.IOException
+import java.lang.reflect.Modifier
+import java.net.URL
+import java.util.Properties
+
+import org.apache.flink.table.annotation.TableType
+import org.apache.flink.table.api.{AmbiguousTableSourceConverterException, NoMatchedTableSourceConverterException}
+import org.apache.flink.table.plan.schema.TableSourceTable
+import org.apache.flink.table.plan.stats.FlinkStatistic
+import org.apache.flink.table.sources.TableSource
+import org.apache.flink.util.InstantiationUtil
+import org.reflections.Reflections
+import org.slf4j.{Logger, LoggerFactory}
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+
+/**
+  * The utility class is used to convert ExternalCatalogTable to TableSourceTable.
+  */
+object ExternalTableSourceUtil {
+
+  // config file to specify scan package to search TableSourceConverter
+  private val tableSourceConverterConfigFileName = "tableSourceConverter.properties"
+
+  private val LOG: Logger = LoggerFactory.getLogger(this.getClass)
+
+  // registered table type with the TableSourceConverter.
+  // Key is table type name, Value is set of converter class.
+  private val tableTypeToTableSourceConvertersClazz = {
+    val registeredConverters =
+      new mutable.HashMap[String, mutable.Set[Class[_ <: TableSourceConverter[_]]]]
+          with mutable.MultiMap[String, Class[_ <: TableSourceConverter[_]]]
+    // scan all config files to find TableSourceConverters which are annotationed with TableType.
+    val resourceUrls = getClass.getClassLoader.getResources(tableSourceConverterConfigFileName)
+    while (resourceUrls.hasMoreElements) {
+      val url = resourceUrls.nextElement()
+      parseScanPackageFromConfigFile(url) match {
+        case Some(scanPackage) =>
+          val clazzWithAnnotations = new Reflections(scanPackage)
+              .getTypesAnnotatedWith(classOf[TableType])
+          clazzWithAnnotations.asScala.foreach(clazz =>
+            if (classOf[TableSourceConverter[_]].isAssignableFrom(clazz)) {
+              if (Modifier.isAbstract(clazz.getModifiers()) ||
+                  Modifier.isInterface(clazz.getModifiers)) {
+                LOG.warn(s"Class ${clazz.getName} is annotated with TableType " +
+                    s"but an abstract class or interface.")
+              } else {
+                val tableTypeAnnotation: TableType =
+                  clazz.getAnnotation(classOf[TableType])
+                val tableType = tableTypeAnnotation.value()
+                val converterClazz = clazz.asInstanceOf[Class[_ <: TableSourceConverter[_]]]
+                registeredConverters.addBinding(tableType, converterClazz)
+                LOG.info(s"Registers the converter ${clazz.getName} to table type [$tableType]. ")
+              }
+            } else {
+              LOG.warn(
+                s"Class ${clazz.getName} is annotated with TableType, " +
+                    s"but does not implement the TableSourceConverter interface.")
+            }
+          )
+        case None =>
+          LOG.warn(s"Fail to get scan package from config file [$url].")
+      }
+    }
+    registeredConverters
+  }
+
+  /**
+    * Converts an [[ExternalCatalogTable]] instance to a [[TableSourceTable]] instance
+    *
+    * @param externalCatalogTable the [[ExternalCatalogTable]] instance which to convert
+    * @return converted [[TableSourceTable]] instance from the input catalog table
+    */
+  def fromExternalCatalogTable(externalCatalogTable: ExternalCatalogTable): TableSourceTable[_] = {
+    val tableType = externalCatalogTable.tableType
+    val propertyKeys = externalCatalogTable.properties.keySet()
+    tableTypeToTableSourceConvertersClazz.get(tableType) match {
+      case Some(converterClasses) =>
+        val matchedConverters = converterClasses.map(InstantiationUtil.instantiate(_))
+            .filter(converter => propertyKeys.containsAll(converter.requiredProperties))
+        if (matchedConverters.isEmpty) {
+          LOG.error(s"Cannot find any TableSourceConverter binded to table type [$tableType]. " +
+              s"Register TableSourceConverter via externalCatalogTable.properties file.")
+          throw new NoMatchedTableSourceConverterException(tableType)
+        } else if (matchedConverters.size > 1) {
+          LOG.error(s"Finds more than one matched TableSourceConverter for type [$tableType], " +
+              s"they are ${matchedConverters.map(_.getClass.getName)}")
+          throw new AmbiguousTableSourceConverterException(tableType)
+        } else {
+          val convertedTableSource: TableSource[_] = matchedConverters.head
+              .fromExternalCatalogTable(externalCatalogTable)
+              .asInstanceOf[TableSource[_]]
+          val flinkStatistic = if (externalCatalogTable.stats != null) {
+            FlinkStatistic.of(externalCatalogTable.stats)
+          } else {
+            FlinkStatistic.UNKNOWN
+          }
+          new TableSourceTable(convertedTableSource, flinkStatistic)
+        }
+      case None =>
+        LOG.error(s"Cannot find any TableSourceConverter binded to table type [$tableType]. " +
+            s"Register TableSourceConverter via externalCatalogTable.properties file.")
+        throw new NoMatchedTableSourceConverterException(tableType)
+    }
+  }
+
+  private def parseScanPackageFromConfigFile(url: URL): Option[String] = {
+    val properties = new Properties()
+    try {
+      properties.load(url.openStream())
+      val scanPackage = properties.getProperty("scan.package")
+      if (scanPackage == null || scanPackage.isEmpty) {
+        LOG.warn(s"Config file $url does not contain scan package!")
+        None
+      } else {
+        Some(scanPackage)
+      }
+    } catch {
+      case e: IOException =>
+        LOG.warn(s"Fail to open config file [$url]", e)
+        None
+    }
+  }
+
+}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/InMemoryExternalCatalog.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/InMemoryExternalCatalog.scala
new file mode 100644
index 0000000000000..a3d019dcc797e
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/InMemoryExternalCatalog.scala
@@ -0,0 +1,155 @@
+/*
+ * 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.flink.table.catalog
+
+import org.apache.flink.table.api.{DatabaseAlreadyExistException, DatabaseNotExistException, TableAlreadyExistException, TableNotExistException}
+import java.util.{List => JList}
+
+import scala.collection.mutable.HashMap
+import scala.collection.JavaConverters._
+
+/**
+  * This class is an in-memory implementation of [[ReadonlyExternalCatalog]].
+  *
+  * It could be used for testing or developing instead of used in production environment.
+  */
+class InMemoryExternalCatalog extends CRUDExternalCatalog {
+
+  private val databases = new HashMap[String, Database]
+
+  @throws[DatabaseNotExistException]
+  @throws[TableAlreadyExistException]
+  override def createTable(
+      table: ExternalCatalogTable,
+      ignoreIfExists: Boolean): Unit = synchronized {
+    val dbName = table.identifier.database
+    val tables = getTables(dbName)
+    val tableName = table.identifier.table
+    if (tables.contains(tableName)) {
+      if (!ignoreIfExists) {
+        throw new TableAlreadyExistException(dbName, tableName)
+      }
+    } else {
+      tables.put(tableName, table)
+    }
+  }
+
+  @throws[DatabaseNotExistException]
+  @throws[TableNotExistException]
+  override def dropTable(
+      dbName: String,
+      tableName: String,
+      ignoreIfNotExists: Boolean): Unit = synchronized {
+    val tables = getTables(dbName)
+    if (tables.remove(tableName).isEmpty && !ignoreIfNotExists) {
+      throw new TableNotExistException(dbName, tableName)
+    }
+  }
+
+  @throws[DatabaseNotExistException]
+  @throws[TableNotExistException]
+  override def alterTable(
+      table: ExternalCatalogTable,
+      ignoreIfNotExists: Boolean): Unit = synchronized {
+    val dbName = table.identifier.database
+    val tables = getTables(dbName)
+    val tableName = table.identifier.table
+    if (tables.contains(tableName)) {
+      tables.put(tableName, table)
+    } else if (!ignoreIfNotExists) {
+      throw new TableNotExistException(dbName, tableName)
+    }
+  }
+
+  @throws[DatabaseNotExistException]
+  override def listTables(dbName: String): JList[String] = synchronized {
+    val tables = getTables(dbName)
+    tables.keys.toList.asJava
+  }
+
+  @throws[DatabaseNotExistException]
+  @throws[TableNotExistException]
+  override def getTable(dbName: String, tableName: String): ExternalCatalogTable = synchronized {
+    val tables = getTables(dbName)
+    tables.get(tableName) match {
+      case Some(table) => table
+      case None => throw new TableNotExistException(dbName, tableName)
+    }
+  }
+
+  @throws[DatabaseAlreadyExistException]
+  override def createDatabase(
+      db: ExternalCatalogDatabase,
+      ignoreIfExists: Boolean): Unit = synchronized {
+    val dbName = db.dbName
+    if (databases.contains(dbName)) {
+      if (!ignoreIfExists) {
+        throw new DatabaseAlreadyExistException(dbName)
+      }
+    } else {
+      databases.put(dbName, new Database(db))
+    }
+  }
+
+  @throws[DatabaseNotExistException]
+  override def alterDatabase(
+      db: ExternalCatalogDatabase,
+      ignoreIfNotExists: Boolean): Unit = synchronized {
+    val dbName = db.dbName
+    databases.get(dbName) match {
+      case Some(database) => database.db = db
+      case None =>
+        if (!ignoreIfNotExists) {
+          throw new DatabaseNotExistException(dbName)
+        }
+    }
+  }
+
+  @throws[DatabaseNotExistException]
+  override def dropDatabase(
+      dbName: String,
+      ignoreIfNotExists: Boolean): Unit = synchronized {
+    if (databases.remove(dbName).isEmpty && !ignoreIfNotExists) {
+      throw new DatabaseNotExistException(dbName)
+    }
+  }
+
+  override def listDatabases(): JList[String] = synchronized {
+    databases.keys.toList.asJava
+  }
+
+  @throws[DatabaseNotExistException]
+  override def getDatabase(dbName: String): ExternalCatalogDatabase = synchronized {
+    databases.get(dbName) match {
+      case Some(database) => database.db
+      case None => null
+    }
+  }
+
+  private def getTables(db: String): HashMap[String, ExternalCatalogTable] =
+    databases.get(db) match {
+      case Some(database) => database.tables
+      case None => throw new DatabaseNotExistException(db)
+    }
+
+  private class Database(var db: ExternalCatalogDatabase) {
+    val tables = new HashMap[String, ExternalCatalogTable]
+  }
+
+}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ReadonlyExternalCatalog.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ReadonlyExternalCatalog.scala
new file mode 100644
index 0000000000000..ab1574996c699
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ReadonlyExternalCatalog.scala
@@ -0,0 +1,72 @@
+/*
+ * 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.flink.table.catalog
+
+import java.util.{List => JList}
+
+import org.apache.flink.table.api._
+
+/**
+  * This class is responsible for read table/database from external catalog.
+  * Its main responsibilities is provide tables for calcite catalog, it looks up databases or tables
+  * in the external catalog.
+  */
+trait ReadonlyExternalCatalog {
+
+  /**
+    * Gets table from external Catalog
+    *
+    * @param dbName    database name
+    * @param tableName table name
+    * @throws DatabaseNotExistException if database does not exist in the catalog yet
+    * @throws TableNotExistException    if table does not exist in the catalog yet
+    * @return found table
+    */
+  @throws[DatabaseNotExistException]
+  @throws[TableNotExistException]
+  def getTable(dbName: String, tableName: String): ExternalCatalogTable
+
+  /**
+    * Gets the table name lists from current external Catalog
+    *
+    * @param dbName database name
+    * @throws DatabaseNotExistException if database does not exist in the catalog yet
+    * @return lists of table name
+    */
+  @throws[DatabaseNotExistException]
+  def listTables(dbName: String): JList[String]
+
+  /**
+    * Gets database from external Catalog
+    *
+    * @param dbName database name
+    * @throws DatabaseNotExistException if database does not exist in the catalog yet
+    * @return found database
+    */
+  @throws[DatabaseNotExistException]
+  def getDatabase(dbName: String): ExternalCatalogDatabase
+
+  /**
+    * Gets the database name lists from current external Catalog
+    *
+    * @return list of database names
+    */
+  def listDatabases(): JList[String]
+
+}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/TableSourceConverter.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/TableSourceConverter.scala
new file mode 100644
index 0000000000000..13e54a63f1dce
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/TableSourceConverter.scala
@@ -0,0 +1,48 @@
+/*
+ * 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.flink.table.catalog
+
+import java.util.{Set => JSet}
+
+import org.apache.flink.table.sources.TableSource
+
+/** Defines a converter used to convert [[org.apache.flink.table.sources.TableSource]] to
+  * or from [[ExternalCatalogTable]].
+  *
+  * @tparam T The tableSource which to do convert operation on.
+  */
+trait TableSourceConverter[T <: TableSource[_]] {
+
+  /**
+    * Defines the required properties that must exists in the properties of an ExternalCatalogTable
+    * to ensure the input ExternalCatalogTable is compatible with the requirements of
+    * current converter.
+    * @return the required properties.
+    */
+  def requiredProperties: JSet[String]
+
+  /**
+    * Converts the input external catalog table instance to a tableSource instance.
+    *
+    * @param externalCatalogTable input external catalog table instance to convert
+    * @return converted tableSource instance from input external catalog table.
+    */
+  def fromExternalCatalogTable(externalCatalogTable: ExternalCatalogTable): T
+
+}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/CsvTableSourceConverter.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/CsvTableSourceConverter.scala
new file mode 100644
index 0000000000000..41a32673f1532
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/CsvTableSourceConverter.scala
@@ -0,0 +1,73 @@
+/*
+ * 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.flink.table.sources
+
+import org.apache.flink.table.annotation.TableType
+import org.apache.flink.table.catalog.{ExternalCatalogTable, TableSourceConverter}
+
+import scala.collection.JavaConverters._
+import java.util.{Set => JSet}
+
+import com.google.common.collect.ImmutableSet
+
+/**
+  * The class defines a converter used to convert [[CsvTableSource]] to
+  * or from [[ExternalCatalogTable]].
+  */
+@TableType(value = "csv")
+class CsvTableSourceConverter extends TableSourceConverter[CsvTableSource] {
+
+  private val required: JSet[String] = ImmutableSet.of("path")
+
+  override def requiredProperties: JSet[String] = required
+
+  override def fromExternalCatalogTable(
+      externalCatalogTable: ExternalCatalogTable): CsvTableSource = {
+    val params = externalCatalogTable.properties.asScala
+    val csvTableSourceBuilder = new CsvTableSource.Builder
+
+    params.get("path").foreach(csvTableSourceBuilder.path)
+    params.get("fieldDelim").foreach(csvTableSourceBuilder.fieldDelimiter)
+    params.get("rowDelim").foreach(csvTableSourceBuilder.lineDelimiter)
+    params.get("quoteCharacter").foreach(quoteStr =>
+      if (quoteStr.length != 1) {
+        throw new IllegalArgumentException("the value of param quoteCharacter is invalid")
+      } else {
+        csvTableSourceBuilder.quoteCharacter(quoteStr.charAt(0))
+      }
+    )
+    params.get("ignoreFirstLine").foreach(ignoreFirstLineStr =>
+        if(ignoreFirstLineStr.toBoolean) {
+          csvTableSourceBuilder.ignoreFirstLine()
+        }
+    )
+    params.get("ignoreComments").foreach(csvTableSourceBuilder.commentPrefix)
+    params.get("lenient").foreach(lenientStr =>
+        if(lenientStr.toBoolean) {
+          csvTableSourceBuilder.ignoreParseErrors
+        }
+    )
+    externalCatalogTable.schema.columnNames
+        .zip(externalCatalogTable.schema.columnTypes)
+        .foreach(field => csvTableSourceBuilder.field(field._1, field._2))
+
+    csvTableSourceBuilder.build()
+  }
+
+}
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/catalog/ExternalCatalogSchemaTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/catalog/ExternalCatalogSchemaTest.scala
new file mode 100644
index 0000000000000..6ffa8c6d33e1c
--- /dev/null
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/catalog/ExternalCatalogSchemaTest.scala
@@ -0,0 +1,88 @@
+/*
+ * 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.flink.table.catalog
+
+import java.util.Collections
+
+import com.google.common.collect.Lists
+import org.apache.calcite.jdbc.CalciteSchema
+import org.apache.calcite.prepare.CalciteCatalogReader
+import org.apache.calcite.schema.SchemaPlus
+import org.apache.calcite.sql.validate.SqlMonikerType
+import org.apache.commons.collections.CollectionUtils
+import org.apache.flink.table.calcite.{FlinkTypeFactory, FlinkTypeSystem}
+import org.apache.flink.table.plan.schema.TableSourceTable
+import org.apache.flink.table.sources.CsvTableSource
+import org.apache.flink.table.utils.CommonTestData
+import org.junit.{Before, Test}
+import org.junit.Assert._
+import scala.collection.JavaConverters._
+
+class ExternalCatalogSchemaTest {
+
+  private val schemaName: String = "test"
+  private var externalCatalogSchema: ExternalCatalogSchema = _
+  private var calciteCatalogReader: CalciteCatalogReader = _
+  private val db = "db1"
+  private val tb = "tb1"
+
+  @Before
+  def setUp(): Unit = {
+    val rootSchemaPlus: SchemaPlus = CalciteSchema.createRootSchema(true, false).plus()
+    val catalog = CommonTestData.getInMemoryTestCatalog
+    externalCatalogSchema = ExternalCatalogSchema.create(rootSchemaPlus, schemaName, catalog)
+    val typeFactory = new FlinkTypeFactory(new FlinkTypeSystem())
+    calciteCatalogReader = new CalciteCatalogReader(
+      CalciteSchema.from(rootSchemaPlus),
+      false,
+      Collections.emptyList(),
+      typeFactory)
+  }
+
+  @Test
+  def testGetSubSchema(): Unit = {
+    val allSchemaObjectNames = calciteCatalogReader
+        .getAllSchemaObjectNames(Lists.newArrayList(schemaName))
+    val subSchemas = allSchemaObjectNames.asScala
+        .filter(_.getType.equals(SqlMonikerType.SCHEMA))
+        .map(_.getFullyQualifiedNames.asScala.toList).toSet
+    assertTrue(Set(List(schemaName, "db1"), List(schemaName, "db2")) == subSchemas)
+  }
+
+  @Test
+  def testGetTable(): Unit = {
+    val relOptTable = calciteCatalogReader.getTable(Lists.newArrayList(schemaName, db, tb))
+    assertNotNull(relOptTable)
+    val tableSourceTable = relOptTable.unwrap(classOf[TableSourceTable[_]])
+    tableSourceTable match {
+      case tst: TableSourceTable[_] =>
+        assertTrue(tst.tableSource.isInstanceOf[CsvTableSource])
+      case _ =>
+        fail("unexpected table type!")
+    }
+  }
+
+  @Test
+  def testGetNotExistTable(): Unit = {
+    val relOptTable = calciteCatalogReader.getTable(
+      Lists.newArrayList(schemaName, db, "nonexist-tb"))
+    assertNull(relOptTable)
+  }
+
+}
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/catalog/InMemoryExternalCatalogTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/catalog/InMemoryExternalCatalogTest.scala
new file mode 100644
index 0000000000000..4199d536a31f6
--- /dev/null
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/catalog/InMemoryExternalCatalogTest.scala
@@ -0,0 +1,145 @@
+/*
+ * 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.flink.table.catalog
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo
+import org.apache.flink.table.api.{DatabaseAlreadyExistException, DatabaseNotExistException, TableAlreadyExistException, TableNotExistException}
+import org.junit.{Before, Test}
+import org.junit.Assert._
+
+class InMemoryExternalCatalogTest {
+
+  private val databaseName = "db1"
+
+  private var catalog: InMemoryExternalCatalog = _
+
+  @Before
+  def setUp(): Unit = {
+    catalog = new InMemoryExternalCatalog()
+    catalog.createDatabase(ExternalCatalogDatabase(databaseName), ignoreIfExists = false)
+  }
+
+  @Test
+  def testCreateTable(): Unit = {
+    assertTrue(catalog.listTables(databaseName).isEmpty)
+    catalog.createTable(createTableInstance(databaseName, "t1"), ignoreIfExists = false)
+    val tables = catalog.listTables(databaseName)
+    assertEquals(1, tables.size())
+    assertEquals("t1", tables.get(0))
+  }
+
+  @Test(expected = classOf[TableAlreadyExistException])
+  def testCreateExistedTable(): Unit = {
+    val tableName = "t1"
+    catalog.createTable(createTableInstance(databaseName, tableName), false)
+    catalog.createTable(createTableInstance(databaseName, tableName), false)
+  }
+
+  @Test
+  def testGetTable(): Unit = {
+    val originTable = createTableInstance(databaseName, "t1")
+    catalog.createTable(originTable, false)
+    assertEquals(catalog.getTable(databaseName, "t1"), originTable)
+  }
+
+  @Test(expected = classOf[DatabaseNotExistException])
+  def testGetTableUnderNotExistDatabaseName(): Unit = {
+    catalog.getTable("notexistedDb", "t1")
+  }
+
+  @Test(expected = classOf[TableNotExistException])
+  def testGetNotExistTable(): Unit = {
+    catalog.getTable(databaseName, "t1")
+  }
+
+  @Test
+  def testAlterTable(): Unit = {
+    val tableName = "t1"
+    val table = createTableInstance(databaseName, tableName)
+    catalog.createTable(table, false)
+    assertEquals(catalog.getTable(databaseName, tableName), table)
+    val newTable = createTableInstance(databaseName, tableName)
+    catalog.alterTable(newTable, false)
+    val currentTable = catalog.getTable(databaseName, tableName)
+    // validate the table is really replaced after alter table
+    assertNotEquals(table, currentTable)
+    assertEquals(newTable, currentTable)
+  }
+
+  @Test(expected = classOf[TableNotExistException])
+  def testAlterNotExistTable(): Unit = {
+    catalog.alterTable(createTableInstance(databaseName, "t1"), false)
+  }
+
+  @Test
+  def testDropTable(): Unit = {
+    val tableName = "t1"
+    catalog.createTable(createTableInstance(databaseName, tableName), false)
+    assertTrue(catalog.listTables(databaseName).contains(tableName))
+    catalog.dropTable(databaseName, tableName, false)
+    assertFalse(catalog.listTables(databaseName).contains(tableName))
+  }
+
+  @Test(expected = classOf[TableNotExistException])
+  def testDropNotExistTable(): Unit = {
+    catalog.dropTable(databaseName, "t1", false)
+  }
+
+  @Test
+  def testListDatabases(): Unit = {
+    val databases = catalog.listDatabases()
+    assertEquals(1, databases.size())
+    assertEquals(databaseName, databases.get(0))
+  }
+
+  @Test
+  def testGetDatabase(): Unit = {
+    assertNotNull(catalog.getDatabase(databaseName))
+  }
+
+  @Test
+  def testGetNotExistDatabase(): Unit = {
+    assertNull(catalog.getDatabase("notexistedDb"))
+  }
+
+  @Test
+  def testCreateDatabase(): Unit = {
+    val originDatabasesNum = catalog.listDatabases().size
+    catalog.createDatabase(ExternalCatalogDatabase("db2"), false)
+    assertEquals(catalog.listDatabases().size, originDatabasesNum + 1)
+  }
+
+  @Test(expected = classOf[DatabaseAlreadyExistException])
+  def testCreateExistedDatabase(): Unit = {
+    catalog.createDatabase(ExternalCatalogDatabase(databaseName), false)
+  }
+
+  private def createTableInstance(dbName: String, tableName: String): ExternalCatalogTable = {
+    val schema = new DataSchema(
+      Array(
+        BasicTypeInfo.STRING_TYPE_INFO,
+        BasicTypeInfo.INT_TYPE_INFO
+      ),
+      Array("first", "second"))
+    ExternalCatalogTable(
+      TableIdentifier(dbName, tableName),
+      "csv",
+      schema)
+  }
+}
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/utils/CommonTestData.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/utils/CommonTestData.scala
index 2364f23873976..c1739059aa7f6 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/utils/CommonTestData.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/utils/CommonTestData.scala
@@ -26,6 +26,7 @@ import org.apache.flink.api.java.typeutils.TypeExtractor
 import org.apache.flink.api.java.{DataSet, ExecutionEnvironment}
 import org.apache.flink.table.api.TableEnvironment
 import org.apache.flink.table.sources.{BatchTableSource, CsvTableSource}
+import org.apache.flink.table.catalog._
 
 object CommonTestData {
 
@@ -43,14 +44,9 @@ object CommonTestData {
       "Kelly#8#2.34#Williams"
     )
 
-    val tempFile = File.createTempFile("csv-test", "tmp")
-    tempFile.deleteOnExit()
-    val tmpWriter = new OutputStreamWriter(new FileOutputStream(tempFile), "UTF-8")
-    tmpWriter.write(csvRecords.mkString("$"))
-    tmpWriter.close()
-
+    val tempFilePath = writeToTempFile(csvRecords.mkString("$"), "csv-test", "tmp")
     new CsvTableSource(
-      tempFile.getAbsolutePath,
+      tempFilePath,
       Array("first", "id", "score", "last"),
       Array(
         BasicTypeInfo.STRING_TYPE_INFO,
@@ -65,6 +61,85 @@ object CommonTestData {
     )
   }
 
+  def getInMemoryTestCatalog: ReadonlyExternalCatalog = {
+    val csvRecord1 = Seq(
+      "1#1#Hi",
+      "2#2#Hello",
+      "3#2#Hello world"
+    )
+    val tempFilePath1 = writeToTempFile(csvRecord1.mkString("$"), "csv-test1", "tmp")
+    val properties1 = new util.HashMap[String, String]()
+    properties1.put("path", tempFilePath1)
+    properties1.put("fieldDelim", "#")
+    properties1.put("rowDelim", "$")
+    val externalCatalogTable1 = ExternalCatalogTable(
+      TableIdentifier("db1", "tb1"),
+      "csv",
+      DataSchema(
+        Array(
+          BasicTypeInfo.INT_TYPE_INFO,
+          BasicTypeInfo.LONG_TYPE_INFO,
+          BasicTypeInfo.STRING_TYPE_INFO),
+        Array("a", "b", "c")),
+      properties1
+    )
+
+    val csvRecord2 = Seq(
+      "1#1#0#Hallo#1",
+      "2#2#1#Hallo Welt#2",
+      "2#3#2#Hallo Welt wie#1",
+      "3#4#3#Hallo Welt wie gehts?#2",
+      "3#5#4#ABC#2",
+      "3#6#5#BCD#3",
+      "4#7#6#CDE#2",
+      "4#8#7#DEF#1",
+      "4#9#8#EFG#1",
+      "4#10#9#FGH#2",
+      "5#11#10#GHI#1",
+      "5#12#11#HIJ#3",
+      "5#13#12#IJK#3",
+      "5#14#13#JKL#2",
+      "5#15#14#KLM#2"
+    )
+    val tempFilePath2 = writeToTempFile(csvRecord2.mkString("$"), "csv-test2", "tmp")
+    val properties2 = new util.HashMap[String, String]()
+    properties2.put("path", tempFilePath1)
+    properties2.put("fieldDelim", "#")
+    properties2.put("rowDelim", "$")
+    val externalCatalogTable2 = ExternalCatalogTable(
+      TableIdentifier("db2", "tb2"),
+      "csv",
+      DataSchema(
+        Array(
+          BasicTypeInfo.INT_TYPE_INFO,
+          BasicTypeInfo.LONG_TYPE_INFO,
+          BasicTypeInfo.INT_TYPE_INFO,
+          BasicTypeInfo.STRING_TYPE_INFO,
+          BasicTypeInfo.LONG_TYPE_INFO),
+        Array("d", "e", "f", "g", "h")),
+      properties2
+    )
+    val catalog = new InMemoryExternalCatalog
+    catalog.createDatabase(ExternalCatalogDatabase("db1"), false)
+    catalog.createDatabase(ExternalCatalogDatabase("db2"), false)
+    catalog.createTable(externalCatalogTable1, false)
+    catalog.createTable(externalCatalogTable2, false)
+    catalog
+  }
+
+  private def writeToTempFile(
+      contents: String,
+      filePrefix: String,
+      fileSuffix: String,
+      charset: String = "UTF-8"): String = {
+    val tempFile = File.createTempFile(filePrefix, fileSuffix)
+    tempFile.deleteOnExit()
+    val tmpWriter = new OutputStreamWriter(new FileOutputStream(tempFile), charset)
+    tmpWriter.write(contents)
+    tmpWriter.close()
+    tempFile.getAbsolutePath
+  }
+
   def getNestedTableSource: BatchTableSource[Person] = {
     new BatchTableSource[Person] {
       override def getDataSet(execEnv: ExecutionEnvironment): DataSet[Person] = {

From 5c0cb62ce4e76ee35c033f8632ab1c41dce7a338 Mon Sep 17 00:00:00 2001
From: jingzhang <beyond1920@126.com>
Date: Fri, 17 Mar 2017 21:03:46 +0800
Subject: [PATCH 2/4] 1. Update code to adapt to specify multiple scan packages
 in one tableSourceConverter.properties 2. Update code based on fabian's
 comment

---
 .../resources/tableSourceConverter.properties |  2 +-
 ...atalog.scala => CrudExternalCatalog.scala} |  2 +-
 ...nalCatalog.scala => ExternalCatalog.scala} |  2 +-
 .../table/catalog/ExternalCatalogSchema.scala |  6 +-
 .../catalog/ExternalTableSourceUtil.scala     | 95 ++++++++++---------
 .../catalog/InMemoryExternalCatalog.scala     |  6 +-
 .../sources/CsvTableSourceConverter.scala     |  2 +-
 .../catalog/InMemoryExternalCatalogTest.scala |  4 +-
 .../flink/table/utils/CommonTestData.scala    |  8 +-
 9 files changed, 65 insertions(+), 62 deletions(-)
 rename flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/{CRUDExternalCatalog.scala => CrudExternalCatalog.scala} (98%)
 rename flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/{ReadonlyExternalCatalog.scala => ExternalCatalog.scala} (98%)

diff --git a/flink-libraries/flink-table/src/main/resources/tableSourceConverter.properties b/flink-libraries/flink-table/src/main/resources/tableSourceConverter.properties
index 7e2bb962a54a3..9f88eaabd269e 100644
--- a/flink-libraries/flink-table/src/main/resources/tableSourceConverter.properties
+++ b/flink-libraries/flink-table/src/main/resources/tableSourceConverter.properties
@@ -16,4 +16,4 @@
 # limitations under the License.
 ################################################################################
 
-scan.package=org.apache.flink.table.sources
\ No newline at end of file
+scan.packages=org.apache.flink.table.sources
\ No newline at end of file
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/CRUDExternalCatalog.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/CrudExternalCatalog.scala
similarity index 98%
rename from flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/CRUDExternalCatalog.scala
rename to flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/CrudExternalCatalog.scala
index 8f98256ed5bec..d93f140f65e11 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/CRUDExternalCatalog.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/CrudExternalCatalog.scala
@@ -28,7 +28,7 @@ import org.apache.flink.table.api._
   * <li> provide tables for calcite catalog, it looks up databases or tables in the external catalog
   * </ul>
   */
-trait CRUDExternalCatalog extends ReadonlyExternalCatalog {
+trait CrudExternalCatalog extends ExternalCatalog {
 
   /**
     * Adds table into external Catalog
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ReadonlyExternalCatalog.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalCatalog.scala
similarity index 98%
rename from flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ReadonlyExternalCatalog.scala
rename to flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalCatalog.scala
index ab1574996c699..58b62c7f36068 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ReadonlyExternalCatalog.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalCatalog.scala
@@ -27,7 +27,7 @@ import org.apache.flink.table.api._
   * Its main responsibilities is provide tables for calcite catalog, it looks up databases or tables
   * in the external catalog.
   */
-trait ReadonlyExternalCatalog {
+trait ExternalCatalog {
 
   /**
     * Gets table from external Catalog
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalCatalogSchema.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalCatalogSchema.scala
index 0357afbbecafb..e3ed96e79d7de 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalCatalogSchema.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalCatalogSchema.scala
@@ -40,7 +40,7 @@ import scala.collection.JavaConverters._
   */
 class ExternalCatalogSchema(
     catalogIdentifier: String,
-    catalog: ReadonlyExternalCatalog) extends Schema {
+    catalog: ExternalCatalog) extends Schema {
 
   private val LOG: Logger = LoggerFactory.getLogger(this.getClass)
 
@@ -96,7 +96,7 @@ class ExternalCatalogSchema(
 
   private class ExternalCatalogDatabaseSchema(
       schemaName: String,
-      flinkExternalCatalog: ReadonlyExternalCatalog) extends Schema {
+      flinkExternalCatalog: ExternalCatalog) extends Schema {
 
     override def getTable(name: String): Table = {
       try {
@@ -146,7 +146,7 @@ object ExternalCatalogSchema {
   def create(
       parentSchema: SchemaPlus,
       externalCatalogIdentifier: String,
-      externalCatalog: ReadonlyExternalCatalog): ExternalCatalogSchema = {
+      externalCatalog: ExternalCatalog): ExternalCatalogSchema = {
     val newSchema = new ExternalCatalogSchema(externalCatalogIdentifier, externalCatalog)
     val schemaPlusOfNewSchema = parentSchema.add(externalCatalogIdentifier, newSchema)
     newSchema.registerSubSchemas(schemaPlusOfNewSchema)
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalTableSourceUtil.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalTableSourceUtil.scala
index c16a8af78aa95..bb11a71b39865 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalTableSourceUtil.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalTableSourceUtil.scala
@@ -18,11 +18,10 @@
 
 package org.apache.flink.table.catalog
 
-import java.io.IOException
 import java.lang.reflect.Modifier
 import java.net.URL
-import java.util.Properties
 
+import org.apache.commons.configuration.{ConfigurationException, ConversionException, PropertiesConfiguration}
 import org.apache.flink.table.annotation.TableType
 import org.apache.flink.table.api.{AmbiguousTableSourceConverterException, NoMatchedTableSourceConverterException}
 import org.apache.flink.table.plan.schema.TableSourceTable
@@ -55,33 +54,31 @@ object ExternalTableSourceUtil {
     val resourceUrls = getClass.getClassLoader.getResources(tableSourceConverterConfigFileName)
     while (resourceUrls.hasMoreElements) {
       val url = resourceUrls.nextElement()
-      parseScanPackageFromConfigFile(url) match {
-        case Some(scanPackage) =>
-          val clazzWithAnnotations = new Reflections(scanPackage)
-              .getTypesAnnotatedWith(classOf[TableType])
-          clazzWithAnnotations.asScala.foreach(clazz =>
-            if (classOf[TableSourceConverter[_]].isAssignableFrom(clazz)) {
-              if (Modifier.isAbstract(clazz.getModifiers()) ||
-                  Modifier.isInterface(clazz.getModifiers)) {
-                LOG.warn(s"Class ${clazz.getName} is annotated with TableType " +
-                    s"but an abstract class or interface.")
-              } else {
-                val tableTypeAnnotation: TableType =
-                  clazz.getAnnotation(classOf[TableType])
-                val tableType = tableTypeAnnotation.value()
-                val converterClazz = clazz.asInstanceOf[Class[_ <: TableSourceConverter[_]]]
-                registeredConverters.addBinding(tableType, converterClazz)
-                LOG.info(s"Registers the converter ${clazz.getName} to table type [$tableType]. ")
-              }
+      val scanPackages = parseScanPackagesFromConfigFile(url)
+      scanPackages.foreach(scanPackage => {
+        val clazzWithAnnotations = new Reflections(scanPackage)
+            .getTypesAnnotatedWith(classOf[TableType])
+        clazzWithAnnotations.asScala.foreach(clazz =>
+          if (classOf[TableSourceConverter[_]].isAssignableFrom(clazz)) {
+            if (Modifier.isAbstract(clazz.getModifiers()) ||
+                Modifier.isInterface(clazz.getModifiers)) {
+              LOG.warn(s"Class ${clazz.getName} is annotated with TableType " +
+                  s"but an abstract class or interface.")
             } else {
-              LOG.warn(
-                s"Class ${clazz.getName} is annotated with TableType, " +
-                    s"but does not implement the TableSourceConverter interface.")
+              val tableTypeAnnotation: TableType =
+                clazz.getAnnotation(classOf[TableType])
+              val tableType = tableTypeAnnotation.value()
+              val converterClazz = clazz.asInstanceOf[Class[_ <: TableSourceConverter[_]]]
+              registeredConverters.addBinding(tableType, converterClazz)
+              LOG.info(s"Registers the converter ${clazz.getName} to table type [$tableType]. ")
             }
-          )
-        case None =>
-          LOG.warn(s"Fail to get scan package from config file [$url].")
-      }
+          } else {
+            LOG.warn(
+              s"Class ${clazz.getName} is annotated with TableType, " +
+                  s"but does not implement the TableSourceConverter interface.")
+          }
+        )
+      })
     }
     registeredConverters
   }
@@ -98,17 +95,23 @@ object ExternalTableSourceUtil {
     tableTypeToTableSourceConvertersClazz.get(tableType) match {
       case Some(converterClasses) =>
         val matchedConverters = converterClasses.map(InstantiationUtil.instantiate(_))
-            .filter(converter => propertyKeys.containsAll(converter.requiredProperties))
         if (matchedConverters.isEmpty) {
           LOG.error(s"Cannot find any TableSourceConverter binded to table type [$tableType]. " +
               s"Register TableSourceConverter via externalCatalogTable.properties file.")
           throw new NoMatchedTableSourceConverterException(tableType)
-        } else if (matchedConverters.size > 1) {
+        }
+        val filteredMatchedConverters = matchedConverters.filter(
+          converter => propertyKeys.containsAll(converter.requiredProperties))
+        if(filteredMatchedConverters.isEmpty) {
+          LOG.error(s"Cannot find any matched TableSourceConverter for type [$tableType], " +
+              s"because the required properties does not match.")
+          throw new NoMatchedTableSourceConverterException(tableType)
+        } else if (filteredMatchedConverters.size > 1) {
           LOG.error(s"Finds more than one matched TableSourceConverter for type [$tableType], " +
-              s"they are ${matchedConverters.map(_.getClass.getName)}")
+              s"they are ${filteredMatchedConverters.map(_.getClass.getName)}")
           throw new AmbiguousTableSourceConverterException(tableType)
         } else {
-          val convertedTableSource: TableSource[_] = matchedConverters.head
+          val convertedTableSource: TableSource[_] = filteredMatchedConverters.head
               .fromExternalCatalogTable(externalCatalogTable)
               .asInstanceOf[TableSource[_]]
           val flinkStatistic = if (externalCatalogTable.stats != null) {
@@ -125,21 +128,25 @@ object ExternalTableSourceUtil {
     }
   }
 
-  private def parseScanPackageFromConfigFile(url: URL): Option[String] = {
-    val properties = new Properties()
+  /**
+    * Parses scan package set from input config file
+    *
+    * @param url url of config file
+    * @return scan package set
+    */
+  private def parseScanPackagesFromConfigFile(url: URL): Set[String] = {
     try {
-      properties.load(url.openStream())
-      val scanPackage = properties.getProperty("scan.package")
-      if (scanPackage == null || scanPackage.isEmpty) {
-        LOG.warn(s"Config file $url does not contain scan package!")
-        None
-      } else {
-        Some(scanPackage)
-      }
+      val config = new PropertiesConfiguration(url)
+      config.setListDelimiter(',')
+      config.getStringArray("scan.packages").filterNot(_.isEmpty).toSet
     } catch {
-      case e: IOException =>
-        LOG.warn(s"Fail to open config file [$url]", e)
-        None
+      case e: ConfigurationException =>
+        LOG.warn(s"Error happened while loading the properties file [$url]", e)
+        Set.empty
+      case e1: ConversionException =>
+        LOG.warn(s"Error happened while parsing 'scan.packages' field of properties file [$url]. " +
+            s"The value is not a String or List of Strings.", e1)
+        Set.empty
     }
   }
 
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/InMemoryExternalCatalog.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/InMemoryExternalCatalog.scala
index a3d019dcc797e..6a6191616820d 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/InMemoryExternalCatalog.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/InMemoryExternalCatalog.scala
@@ -25,11 +25,11 @@ import scala.collection.mutable.HashMap
 import scala.collection.JavaConverters._
 
 /**
-  * This class is an in-memory implementation of [[ReadonlyExternalCatalog]].
+  * This class is an in-memory implementation of [[ExternalCatalog]].
   *
   * It could be used for testing or developing instead of used in production environment.
   */
-class InMemoryExternalCatalog extends CRUDExternalCatalog {
+class InMemoryExternalCatalog extends CrudExternalCatalog {
 
   private val databases = new HashMap[String, Database]
 
@@ -138,7 +138,7 @@ class InMemoryExternalCatalog extends CRUDExternalCatalog {
   override def getDatabase(dbName: String): ExternalCatalogDatabase = synchronized {
     databases.get(dbName) match {
       case Some(database) => database.db
-      case None => null
+      case None => throw new DatabaseNotExistException(dbName)
     }
   }
 
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/CsvTableSourceConverter.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/CsvTableSourceConverter.scala
index 41a32673f1532..af4d31df49398 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/CsvTableSourceConverter.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/CsvTableSourceConverter.scala
@@ -33,7 +33,7 @@ import com.google.common.collect.ImmutableSet
 @TableType(value = "csv")
 class CsvTableSourceConverter extends TableSourceConverter[CsvTableSource] {
 
-  private val required: JSet[String] = ImmutableSet.of("path")
+  private val required: JSet[String] = ImmutableSet.of("path", "fieldDelim", "rowDelim")
 
   override def requiredProperties: JSet[String] = required
 
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/catalog/InMemoryExternalCatalogTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/catalog/InMemoryExternalCatalogTest.scala
index 4199d536a31f6..65a9bf4fc987a 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/catalog/InMemoryExternalCatalogTest.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/catalog/InMemoryExternalCatalogTest.scala
@@ -113,9 +113,9 @@ class InMemoryExternalCatalogTest {
     assertNotNull(catalog.getDatabase(databaseName))
   }
 
-  @Test
+  @Test(expected = classOf[DatabaseNotExistException])
   def testGetNotExistDatabase(): Unit = {
-    assertNull(catalog.getDatabase("notexistedDb"))
+    catalog.getDatabase("notexistedDb")
   }
 
   @Test
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/utils/CommonTestData.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/utils/CommonTestData.scala
index c1739059aa7f6..ca77c8fa0d524 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/utils/CommonTestData.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/utils/CommonTestData.scala
@@ -24,7 +24,6 @@ import java.util
 import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
 import org.apache.flink.api.java.typeutils.TypeExtractor
 import org.apache.flink.api.java.{DataSet, ExecutionEnvironment}
-import org.apache.flink.table.api.TableEnvironment
 import org.apache.flink.table.sources.{BatchTableSource, CsvTableSource}
 import org.apache.flink.table.catalog._
 
@@ -61,7 +60,7 @@ object CommonTestData {
     )
   }
 
-  def getInMemoryTestCatalog: ReadonlyExternalCatalog = {
+  def getInMemoryTestCatalog: ExternalCatalog = {
     val csvRecord1 = Seq(
       "1#1#Hi",
       "2#2#Hello",
@@ -103,7 +102,7 @@ object CommonTestData {
     )
     val tempFilePath2 = writeToTempFile(csvRecord2.mkString("$"), "csv-test2", "tmp")
     val properties2 = new util.HashMap[String, String]()
-    properties2.put("path", tempFilePath1)
+    properties2.put("path", tempFilePath2)
     properties2.put("fieldDelim", "#")
     properties2.put("rowDelim", "$")
     val externalCatalogTable2 = ExternalCatalogTable(
@@ -170,7 +169,4 @@ object CommonTestData {
       this(null, null)
     }
   }
-
-  def getMockTableEnvironment: TableEnvironment = new MockTableEnvironment
-
 }

From 2e65848408a32f0b4253499f65fb73fdb08a524a Mon Sep 17 00:00:00 2001
From: jingzhang <beyond1920@126.com>
Date: Fri, 17 Mar 2017 21:08:06 +0800
Subject: [PATCH 3/4] 1. add better comments on tableSourceProperties file. 2.
 add check for instantiable of scanned TableSourceConverter subclazz in
 ExternalTableSourceUtil

---
 .../src/main/resources/tableSourceConverter.properties | 10 ++++++++++
 .../flink/table/catalog/ExternalTableSourceUtil.scala  |  8 ++++----
 .../flink/table/sources/CsvTableSourceConverter.scala  |  2 +-
 3 files changed, 15 insertions(+), 5 deletions(-)

diff --git a/flink-libraries/flink-table/src/main/resources/tableSourceConverter.properties b/flink-libraries/flink-table/src/main/resources/tableSourceConverter.properties
index 9f88eaabd269e..1632e1264d616 100644
--- a/flink-libraries/flink-table/src/main/resources/tableSourceConverter.properties
+++ b/flink-libraries/flink-table/src/main/resources/tableSourceConverter.properties
@@ -16,4 +16,14 @@
 # limitations under the License.
 ################################################################################
 
+################################################################################
+# The config file is used to specify the packages of current module where
+# to find TableSourceConverter implementation class annotationed with TableType.
+# If there are multiple packages to scan, put those packages together into a
+# string seperated with ',', for example, org.package1,org.package2.
+# Please notice:
+# It's better to have a tableSourceConverter.properties in each connector Module
+# which offers converters instead of put all information into the
+# tableSourceConverter.properties of flink-table module.
+################################################################################
 scan.packages=org.apache.flink.table.sources
\ No newline at end of file
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalTableSourceUtil.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalTableSourceUtil.scala
index bb11a71b39865..0964bbca0638f 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalTableSourceUtil.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalTableSourceUtil.scala
@@ -60,10 +60,10 @@ object ExternalTableSourceUtil {
             .getTypesAnnotatedWith(classOf[TableType])
         clazzWithAnnotations.asScala.foreach(clazz =>
           if (classOf[TableSourceConverter[_]].isAssignableFrom(clazz)) {
-            if (Modifier.isAbstract(clazz.getModifiers()) ||
-                Modifier.isInterface(clazz.getModifiers)) {
-              LOG.warn(s"Class ${clazz.getName} is annotated with TableType " +
-                  s"but an abstract class or interface.")
+            val errorInfo = InstantiationUtil.checkForInstantiationError(clazz)
+            if (errorInfo != null) {
+              LOG.warn(s"Class ${clazz.getName} is annotated with TableType, " +
+                  s"but is not instantiable because $errorInfo.")
             } else {
               val tableTypeAnnotation: TableType =
                 clazz.getAnnotation(classOf[TableType])
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/CsvTableSourceConverter.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/CsvTableSourceConverter.scala
index af4d31df49398..a093344e83dde 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/CsvTableSourceConverter.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/CsvTableSourceConverter.scala
@@ -47,7 +47,7 @@ class CsvTableSourceConverter extends TableSourceConverter[CsvTableSource] {
     params.get("rowDelim").foreach(csvTableSourceBuilder.lineDelimiter)
     params.get("quoteCharacter").foreach(quoteStr =>
       if (quoteStr.length != 1) {
-        throw new IllegalArgumentException("the value of param quoteCharacter is invalid")
+        throw new IllegalArgumentException("the value of param must only contain one character!")
       } else {
         csvTableSourceBuilder.quoteCharacter(quoteStr.charAt(0))
       }

From 00c09ce0ffebf875393f7370253bec17d1435ba7 Mon Sep 17 00:00:00 2001
From: jingzhang <beyond1920@126.com>
Date: Fri, 17 Mar 2017 21:24:55 +0800
Subject: [PATCH 4/4] delete DataSchema, reuse tableSchema

---
 .../table/catalog/ExternalCatalogTable.scala   | 18 ++----------------
 .../sources/CsvTableSourceConverter.scala      |  4 ++--
 .../catalog/InMemoryExternalCatalogTest.scala  |  9 +++++----
 .../flink/table/utils/CommonTestData.scala     | 15 +++++++++------
 4 files changed, 18 insertions(+), 28 deletions(-)

diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalCatalogTable.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalCatalogTable.scala
index 91c90ce2e657a..893cbb3671965 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalCatalogTable.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/ExternalCatalogTable.scala
@@ -22,6 +22,7 @@ import java.util.{HashMap => JHashMap, Map => JMap}
 import java.lang.{Long => JLong}
 
 import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.table.api.TableSchema
 import org.apache.flink.table.plan.stats.TableStats
 
 /**
@@ -39,7 +40,7 @@ import org.apache.flink.table.plan.stats.TableStats
 case class ExternalCatalogTable(
     identifier: TableIdentifier,
     tableType: String,
-    schema: DataSchema,
+    schema: TableSchema,
     properties: JMap[String, String] = new JHashMap(),
     stats: TableStats = null,
     comment: String = null,
@@ -59,18 +60,3 @@ case class TableIdentifier(
   override def toString: String = s"$database.$table"
 
 }
-
-/**
-  * Schema of External catalog table's columns
-  *
-  * @param columnTypes types of each column
-  * @param columnNames names of each column
-  */
-case class DataSchema(
-    columnTypes: Array[TypeInformation[_]],
-    columnNames: Array[String]) {
-
-  override def toString: String =
-    columnNames.zip(columnTypes).map(x => s"${x._1}: ${x._2}").mkString(", ")
-
-}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/CsvTableSourceConverter.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/CsvTableSourceConverter.scala
index a093344e83dde..68d2f55e83cd1 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/CsvTableSourceConverter.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/CsvTableSourceConverter.scala
@@ -63,8 +63,8 @@ class CsvTableSourceConverter extends TableSourceConverter[CsvTableSource] {
           csvTableSourceBuilder.ignoreParseErrors
         }
     )
-    externalCatalogTable.schema.columnNames
-        .zip(externalCatalogTable.schema.columnTypes)
+    externalCatalogTable.schema.getColumnNames
+        .zip(externalCatalogTable.schema.getTypes)
         .foreach(field => csvTableSourceBuilder.field(field._1, field._2))
 
     csvTableSourceBuilder.build()
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/catalog/InMemoryExternalCatalogTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/catalog/InMemoryExternalCatalogTest.scala
index 65a9bf4fc987a..5402780b3d3b4 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/catalog/InMemoryExternalCatalogTest.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/catalog/InMemoryExternalCatalogTest.scala
@@ -19,7 +19,7 @@
 package org.apache.flink.table.catalog
 
 import org.apache.flink.api.common.typeinfo.BasicTypeInfo
-import org.apache.flink.table.api.{DatabaseAlreadyExistException, DatabaseNotExistException, TableAlreadyExistException, TableNotExistException}
+import org.apache.flink.table.api._
 import org.junit.{Before, Test}
 import org.junit.Assert._
 
@@ -131,12 +131,13 @@ class InMemoryExternalCatalogTest {
   }
 
   private def createTableInstance(dbName: String, tableName: String): ExternalCatalogTable = {
-    val schema = new DataSchema(
+    val schema = new TableSchema(
+      Array("first", "second"),
       Array(
         BasicTypeInfo.STRING_TYPE_INFO,
         BasicTypeInfo.INT_TYPE_INFO
-      ),
-      Array("first", "second"))
+      )
+    )
     ExternalCatalogTable(
       TableIdentifier(dbName, tableName),
       "csv",
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/utils/CommonTestData.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/utils/CommonTestData.scala
index ca77c8fa0d524..a1bfd56d66805 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/utils/CommonTestData.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/utils/CommonTestData.scala
@@ -24,6 +24,7 @@ import java.util
 import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
 import org.apache.flink.api.java.typeutils.TypeExtractor
 import org.apache.flink.api.java.{DataSet, ExecutionEnvironment}
+import org.apache.flink.table.api.TableSchema
 import org.apache.flink.table.sources.{BatchTableSource, CsvTableSource}
 import org.apache.flink.table.catalog._
 
@@ -74,12 +75,12 @@ object CommonTestData {
     val externalCatalogTable1 = ExternalCatalogTable(
       TableIdentifier("db1", "tb1"),
       "csv",
-      DataSchema(
+      new TableSchema(
+        Array("a", "b", "c"),
         Array(
           BasicTypeInfo.INT_TYPE_INFO,
           BasicTypeInfo.LONG_TYPE_INFO,
-          BasicTypeInfo.STRING_TYPE_INFO),
-        Array("a", "b", "c")),
+          BasicTypeInfo.STRING_TYPE_INFO)),
       properties1
     )
 
@@ -108,14 +109,15 @@ object CommonTestData {
     val externalCatalogTable2 = ExternalCatalogTable(
       TableIdentifier("db2", "tb2"),
       "csv",
-      DataSchema(
+      new TableSchema(
+        Array("d", "e", "f", "g", "h"),
         Array(
           BasicTypeInfo.INT_TYPE_INFO,
           BasicTypeInfo.LONG_TYPE_INFO,
           BasicTypeInfo.INT_TYPE_INFO,
           BasicTypeInfo.STRING_TYPE_INFO,
-          BasicTypeInfo.LONG_TYPE_INFO),
-        Array("d", "e", "f", "g", "h")),
+          BasicTypeInfo.LONG_TYPE_INFO)
+      ),
       properties2
     )
     val catalog = new InMemoryExternalCatalog
@@ -169,4 +171,5 @@ object CommonTestData {
       this(null, null)
     }
   }
+
 }