diff --git a/JDBCUtils.java b/JDBCUtils.java
index cea6d6f..dfc66a0 100644
--- a/JDBCUtils.java
+++ b/JDBCUtils.java
@@ -396,10 +396,14 @@ public String[] getColumnNamesByResultSetID(int resultSetID) throws SQLException
    *      Returns the column name
    */
   public String[] getTableNames() throws SQLException {
+    return getTableNames(null, null);
+  }
+
+  public String[] getTableNames(String catalog, String schema) throws SQLException {
     try {
       checkConnExist();
       DatabaseMetaData md = conn.getConnection().getMetaData();
-      ResultSet tmpResultSet = md.getTables(null, null, "%", null);
+      ResultSet tmpResultSet = md.getTables(catalog, schema, "%", null);
 
       List<String> tmpTableNamesList = new ArrayList<String>();
       while (tmpResultSet.next()) {
@@ -420,10 +424,14 @@ public String[] getTableNames() throws SQLException {
    *      Returns the column name
    */
   public String[] getColumnNames(String tableName) throws SQLException {
+    return getColumnNames(tableName, null, null);
+  }
+
+  public String[] getColumnNames(String tableName, String catalog, String schema) throws SQLException {
     try {
       checkConnExist();
       DatabaseMetaData md = conn.getConnection().getMetaData();
-      ResultSet tmpResultSet = md.getColumns(null, null, tableName, null);
+      ResultSet tmpResultSet = md.getColumns(catalog, schema, tableName, null);
       List<String> tmpColumnNamesList = new ArrayList<String>();
       while (tmpResultSet.next()) {
         tmpColumnNamesList.add(tmpResultSet.getString("COLUMN_NAME"));
@@ -443,10 +451,14 @@ public String[] getColumnNames(String tableName) throws SQLException {
    *      Returns the column name
    */
   public String[] getColumnTypes(String tableName) throws SQLException {
+    return getColumnTypes(tableName, null, null);
+  }
+
+  public String[] getColumnTypes(String tableName, String catalog, String schema) throws SQLException {
     try {
       checkConnExist();
       DatabaseMetaData md = conn.getConnection().getMetaData();
-      ResultSet tmpResultSet = md.getColumns(null, null, tableName, null);
+      ResultSet tmpResultSet = md.getColumns(catalog, schema, tableName, null);
       List<String> tmpColumnTypesList = new ArrayList<String>();
       while (tmpResultSet.next()) {
         tmpColumnTypesList.add(tmpResultSet.getString("TYPE_NAME"));
@@ -516,10 +528,14 @@ public String[] getColumnTypes(String tableName) throws SQLException {
    *      Returns the column name
    */
   public String[] getPrimaryKey(String tableName) throws SQLException {
+    return getPrimaryKey(tableName, null, null);
+  }
+
+  public String[] getPrimaryKey(String tableName, String catalog, String schema) throws SQLException {
     try {
       checkConnExist();
       DatabaseMetaData md = conn.getConnection().getMetaData();
-      ResultSet tmpResultSet = md.getPrimaryKeys(null, null, tableName);
+      ResultSet tmpResultSet = md.getPrimaryKeys(catalog, schema, tableName);
       List<String> tmpPrimaryKeyList = new ArrayList<String>();
       while (tmpResultSet.next()) {
         tmpPrimaryKeyList.add(tmpResultSet.getString("COLUMN_NAME"));
diff --git a/jdbc_fdw.c b/jdbc_fdw.c
index 2745744..6648f77 100644
--- a/jdbc_fdw.c
+++ b/jdbc_fdw.c
@@ -3123,9 +3123,53 @@ jdbcImportForeignSchema(ImportForeignSchemaStmt *stmt, Oid serverOid)
 	user = GetUserMapping(GetUserId(), server->serverid);
 	jdbcUtilsInfo = jdbc_get_jdbc_utils_obj(server, user, false);
 
-	schema_list = jq_get_schema_info(jdbcUtilsInfo);
+	/* Extract catalog from server options if set */
+	{
+		char	   *catalog = NULL;
+		ListCell   *opt_lc;
+
+		foreach(opt_lc, server->options)
+		{
+			DefElem    *def = (DefElem *) lfirst(opt_lc);
+
+			if (strcmp(def->defname, "catalog") == 0)
+				catalog = defGetString(def);
+		}
+
+		schema_list = jq_get_schema_info(jdbcUtilsInfo, catalog, stmt->remote_schema);
+	}
+
 	if (schema_list != NIL)
 	{
+		/* Filter tables based on LIMIT TO / EXCEPT */
+		if (stmt->list_type == FDW_IMPORT_SCHEMA_LIMIT_TO ||
+			stmt->list_type == FDW_IMPORT_SCHEMA_EXCEPT)
+		{
+			List	   *filtered_list = NIL;
+
+			foreach(table_lc, schema_list)
+			{
+				JtableInfo *tmpTableInfo = (JtableInfo *) lfirst(table_lc);
+				bool		found = false;
+				ListCell   *tbl_lc;
+
+				foreach(tbl_lc, stmt->table_list)
+				{
+					RangeVar   *rv = (RangeVar *) lfirst(tbl_lc);
+
+					if (strcmp(rv->relname, tmpTableInfo->table_name) == 0)
+					{
+						found = true;
+						break;
+					}
+				}
+				if ((stmt->list_type == FDW_IMPORT_SCHEMA_LIMIT_TO && found) ||
+					(stmt->list_type == FDW_IMPORT_SCHEMA_EXCEPT && !found))
+					filtered_list = lappend(filtered_list, tmpTableInfo);
+			}
+			schema_list = filtered_list;
+		}
+
 		initStringInfo(&buf);
 		/* schema_list includes tablename and tableinfos */
 		foreach(table_lc, schema_list)
@@ -3165,13 +3209,15 @@ jdbcImportForeignSchema(ImportForeignSchemaStmt *stmt, Oid serverOid)
 				}
 				/* Print column name and type */
 				appendStringInfo(&buf, "%s %s",
-								 columnInfo->column_name,
+								 quote_identifier(columnInfo->column_name),
 								 columnInfo->column_type);
 				/* Add option if the column is rowkey. */
 				if (columnInfo->primary_key)
 					appendStringInfoString(&buf, " OPTIONS (key 'true')");
 			}
-			appendStringInfo(&buf, ") SERVER %s;", quote_identifier(server->servername));
+			appendStringInfo(&buf, ") SERVER %s OPTIONS (table_name '%s');",
+							 quote_identifier(server->servername),
+							 tmpTableInfo->table_name);
 			commands = lappend(commands, pstrdup(buf.data));
 	NEXT_COLUMN:
 			resetStringInfo(&buf);
diff --git a/jq.c b/jq.c
index b8aa811..06f027c 100644
--- a/jq.c
+++ b/jq.c
@@ -125,8 +125,8 @@ void		jq_get_exception(void);
 /*
  * get table infomations for importForeignSchema
  */
-static List *jq_get_column_infos(JDBCUtilsInfo * jdbcUtilsInfo, char *tablename);
-static List *jq_get_table_names(JDBCUtilsInfo * jdbcUtilsInfo);
+static List *jq_get_column_infos(JDBCUtilsInfo * jdbcUtilsInfo, char *tablename, char *catalog, char *schema);
+static List *jq_get_table_names(JDBCUtilsInfo * jdbcUtilsInfo, char *catalog, char *schema);
 
 
 static void jq_get_JDBCUtils(JDBCUtilsInfo * jdbcUtilsInfo, jclass * JDBCUtilsClass, jobject * JDBCUtilsObject);
@@ -1487,11 +1487,13 @@ jq_get_exception()
 }
 
 static List *
-jq_get_column_infos(JDBCUtilsInfo * jdbcUtilsInfo, char *tablename)
+jq_get_column_infos(JDBCUtilsInfo * jdbcUtilsInfo, char *tablename, char *catalog, char *schema)
 {
 	jobject		JDBCUtilsObject;
 	jclass		JDBCUtilsClass;
 	jstring		jtablename = (*Jenv)->NewStringUTF(Jenv, tablename);
+	jstring		jcatalog = (catalog != NULL) ? (*Jenv)->NewStringUTF(Jenv, catalog) : NULL;
+	jstring		jschema = (schema != NULL) ? (*Jenv)->NewStringUTF(Jenv, schema) : NULL;
 	int			i;
 
 	/* getColumnNames */
@@ -1528,17 +1530,17 @@ jq_get_column_infos(JDBCUtilsInfo * jdbcUtilsInfo, char *tablename)
 	PG_END_TRY();
 
 	/* getColumnNames */
-	idGetColumnNames = (*Jenv)->GetMethodID(Jenv, JDBCUtilsClass, "getColumnNames", "(Ljava/lang/String;)[Ljava/lang/String;");
+	idGetColumnNames = (*Jenv)->GetMethodID(Jenv, JDBCUtilsClass, "getColumnNames", "(Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;)[Ljava/lang/String;");
 	if (idGetColumnNames == NULL)
 	{
 		(*Jenv)->DeleteLocalRef(Jenv, jtablename);
 		ereport(ERROR, (errmsg("Failed to find the JDBCUtils.getColumnNames method")));
 	}
 	jq_exception_clear();
-	columnNamesArray = (*Jenv)->CallObjectMethod(Jenv, JDBCUtilsObject, idGetColumnNames, jtablename);
+	columnNamesArray = (*Jenv)->CallObjectMethod(Jenv, JDBCUtilsObject, idGetColumnNames, jtablename, jcatalog, jschema);
 	jq_get_exception();
 	/* getColumnTypes */
-	idGetColumnTypes = (*Jenv)->GetMethodID(Jenv, JDBCUtilsClass, "getColumnTypes", "(Ljava/lang/String;)[Ljava/lang/String;");
+	idGetColumnTypes = (*Jenv)->GetMethodID(Jenv, JDBCUtilsClass, "getColumnTypes", "(Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;)[Ljava/lang/String;");
 	if (idGetColumnTypes == NULL)
 	{
 		(*Jenv)->DeleteLocalRef(Jenv, jtablename);
@@ -1549,10 +1551,10 @@ jq_get_column_infos(JDBCUtilsInfo * jdbcUtilsInfo, char *tablename)
 		ereport(ERROR, (errmsg("Failed to find the JDBCUtils.getColumnTypes method")));
 	}
 	jq_exception_clear();
-	columnTypesArray = (*Jenv)->CallObjectMethod(Jenv, JDBCUtilsObject, idGetColumnTypes, jtablename);
+	columnTypesArray = (*Jenv)->CallObjectMethod(Jenv, JDBCUtilsObject, idGetColumnTypes, jtablename, jcatalog, jschema);
 	jq_get_exception();
 	/* getPrimaryKey */
-	idGetPrimaryKey = (*Jenv)->GetMethodID(Jenv, JDBCUtilsClass, "getPrimaryKey", "(Ljava/lang/String;)[Ljava/lang/String;");
+	idGetPrimaryKey = (*Jenv)->GetMethodID(Jenv, JDBCUtilsClass, "getPrimaryKey", "(Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;)[Ljava/lang/String;");
 	if (idGetPrimaryKey == NULL)
 	{
 		(*Jenv)->DeleteLocalRef(Jenv, jtablename);
@@ -1567,7 +1569,7 @@ jq_get_column_infos(JDBCUtilsInfo * jdbcUtilsInfo, char *tablename)
 		ereport(ERROR, (errmsg("Failed to find the JDBCUtils.getColumnTypes method")));
 	}
 	jq_exception_clear();
-	primaryKeyArray = (*Jenv)->CallObjectMethod(Jenv, JDBCUtilsObject, idGetPrimaryKey, jtablename);
+	primaryKeyArray = (*Jenv)->CallObjectMethod(Jenv, JDBCUtilsObject, idGetPrimaryKey, jtablename, jcatalog, jschema);
 	jq_get_exception();
 	if (primaryKeyArray != NULL)
 	{
@@ -1628,6 +1630,10 @@ jq_get_column_infos(JDBCUtilsInfo * jdbcUtilsInfo, char *tablename)
 		(*Jenv)->DeleteLocalRef(Jenv, columnTypesArray);
 	}
 	(*Jenv)->DeleteLocalRef(Jenv, jtablename);
+	if (jcatalog != NULL)
+		(*Jenv)->DeleteLocalRef(Jenv, jcatalog);
+	if (jschema != NULL)
+		(*Jenv)->DeleteLocalRef(Jenv, jschema);
 
 	return columnInfoList;
 }
@@ -1755,7 +1761,7 @@ jq_get_column_infos_without_key(JDBCUtilsInfo * jdbcUtilsInfo, int *resultSetID,
  * jq_get_table_names
  */
 static List *
-jq_get_table_names(JDBCUtilsInfo * jdbcUtilsInfo)
+jq_get_table_names(JDBCUtilsInfo * jdbcUtilsInfo, char *catalog, char *schema)
 {
 	jobject		JDBCUtilsObject;
 	jclass		JDBCUtilsClass;
@@ -1764,16 +1770,18 @@ jq_get_table_names(JDBCUtilsInfo * jdbcUtilsInfo)
 	List	   *tableName = NIL;
 	jsize		numberOfTables;
 	int			i;
+	jstring		jcatalog = (catalog != NULL) ? (*Jenv)->NewStringUTF(Jenv, catalog) : NULL;
+	jstring		jschema = (schema != NULL) ? (*Jenv)->NewStringUTF(Jenv, schema) : NULL;
 
 	jq_get_JDBCUtils(jdbcUtilsInfo, &JDBCUtilsClass, &JDBCUtilsObject);
 
-	idGetTableNames = (*Jenv)->GetMethodID(Jenv, JDBCUtilsClass, "getTableNames", "()[Ljava/lang/String;");
+	idGetTableNames = (*Jenv)->GetMethodID(Jenv, JDBCUtilsClass, "getTableNames", "(Ljava/lang/String;Ljava/lang/String;)[Ljava/lang/String;");
 	if (idGetTableNames == NULL)
 	{
 		ereport(ERROR, (errmsg("Failed to find the JDBCUtils.getTableNames method")));
 	}
 	jq_exception_clear();
-	tableNameArray = (*Jenv)->CallObjectMethod(Jenv, JDBCUtilsObject, idGetTableNames);
+	tableNameArray = (*Jenv)->CallObjectMethod(Jenv, JDBCUtilsObject, idGetTableNames, jcatalog, jschema);
 	jq_get_exception();
 	if (tableNameArray != NULL)
 	{
@@ -1786,6 +1794,10 @@ jq_get_table_names(JDBCUtilsInfo * jdbcUtilsInfo)
 		}
 		(*Jenv)->DeleteLocalRef(Jenv, tableNameArray);
 	}
+	if (jcatalog != NULL)
+		(*Jenv)->DeleteLocalRef(Jenv, jcatalog);
+	if (jschema != NULL)
+		(*Jenv)->DeleteLocalRef(Jenv, jschema);
 	return tableName;
 }
 
@@ -1793,14 +1805,14 @@ jq_get_table_names(JDBCUtilsInfo * jdbcUtilsInfo)
  * jq_get_schema_info
  */
 List *
-jq_get_schema_info(JDBCUtilsInfo * jdbcUtilsInfo)
+jq_get_schema_info(JDBCUtilsInfo * jdbcUtilsInfo, char *catalog, char *schema)
 {
 	List	   *schema_list = NIL;
 	List	   *tableName = NIL;
 	JtableInfo *tableInfo;
 	ListCell   *lc;
 
-	tableName = jq_get_table_names(jdbcUtilsInfo);
+	tableName = jq_get_table_names(jdbcUtilsInfo, catalog, schema);
 
 	foreach(lc, tableName)
 	{
@@ -1811,7 +1823,7 @@ jq_get_schema_info(JDBCUtilsInfo * jdbcUtilsInfo)
 		if (tmpTableName != NULL)
 		{
 			tableInfo->table_name = tmpTableName;
-			tableInfo->column_info = jq_get_column_infos(jdbcUtilsInfo, tmpTableName);
+			tableInfo->column_info = jq_get_column_infos(jdbcUtilsInfo, tmpTableName, catalog, schema);
 			schema_list = lappend(schema_list, tableInfo);
 		}
 	}
diff --git a/jq.h b/jq.h
index e16be66..7a6cdb9 100644
--- a/jq.h
+++ b/jq.h
@@ -77,7 +77,7 @@ extern void jq_iterate_all_row(FunctionCallInfo fcinfo, JDBCUtilsInfo * jdbcUtil
 extern List *jq_get_column_infos_without_key(JDBCUtilsInfo * jdbcUtilsInfo, int *resultSetID, int *column_num);
 extern void *jq_bind_sql_var(JDBCUtilsInfo * jdbcUtilsInfo, Oid type, int attnum, Datum value, bool *isnull, int resultSetID);
 extern Datum jdbc_convert_to_pg(Oid pgtyp, int pgtypmod, char *value);
-extern List *jq_get_schema_info(JDBCUtilsInfo * jdbcUtilsInfo);
+extern List *jq_get_schema_info(JDBCUtilsInfo * jdbcUtilsInfo, char *catalog, char *schema);
 extern void jdbc_jvm_init(const ForeignServer *server, const UserMapping *user);
 extern void jq_cancel(JDBCUtilsInfo * jdbcUtilsInfo);
 void		jq_inval_callback(int cacheid, uint32 hashvalue);
diff --git a/option.c b/option.c
index 80f4249..4e9ef19 100644
--- a/option.c
+++ b/option.c
@@ -194,6 +194,7 @@ init_jdbcfdw_options(void)
 		{"querytimeout", ForeignServerRelationId, false},
 		{"jarfile", ForeignServerRelationId, false},
 		{"maxheapsize", ForeignServerRelationId, false},
+		{"catalog", ForeignServerRelationId, false},
 		{"username", UserMappingRelationId, false},
 		{"password", UserMappingRelationId, false},
 		/* use_remote_estimate is available on both server and table */