diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out index 2e49ee3..0831c78 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -2834,3 +2834,51 @@ NOTICE: NEW: (13,"test triggered !") (0,27) (1 row) +-- Test IMPORT FOREIGN SCHEMA statement +CREATE schema import_destination; +CREATE schema import_source; +CREATE TABLE import_source.t1 (c1 int, c2 varchar NOT NULL); +CREATE TABLE import_source.t2 (c1 int, c2 varchar); +CREATE TABLE import_source.t3 (c1 int, c2 varchar); +IMPORT FOREIGN SCHEMA import_source FROM SERVER loopback INTO import_destination; +\det+ import_destination; + List of foreign tables + Schema | Table | Server | FDW Options | Description +--------------------+-------+----------+------------------------------------------------+------------- + import_destination | t1 | loopback | (schema_name 'import_source', table_name 't1') | + import_destination | t2 | loopback | (schema_name 'import_source', table_name 't2') | + import_destination | t3 | loopback | (schema_name 'import_source', table_name 't3') | +(3 rows) + +\d import_destination.t1; + Foreign table "import_destination.t1" + Column | Type | Modifiers | FDW Options +--------+-------------------+-----------+------------- + c1 | integer | | + c2 | character varying | not null | +Server: loopback +FDW Options: (schema_name 'import_source', table_name 't1') + +DROP SCHEMA import_destination cascade; +NOTICE: drop cascades to 3 other objects +DETAIL: drop cascades to foreign table import_destination.t1 +drop cascades to foreign table import_destination.t2 +drop cascades to foreign table import_destination.t3 +CREATE schema import_destination; +IMPORT FOREIGN SCHEMA import_source FROM SERVER loopback LIMIT TO (t1) INTO import_destination; +\det+ import_destination; + List of foreign tables + Schema | Table | Server | FDW Options | Description +--------------------+-------+----------+------------------------------------------------+------------- + import_destination | t1 | loopback | (schema_name 'import_source', table_name 't1') | +(1 row) + +IMPORT FOREIGN SCHEMA import_source FROM SERVER loopback EXCEPT (t1, t2) INTO import_destination; +\det+ import_destination; + List of foreign tables + Schema | Table | Server | FDW Options | Description +--------------------+-------+----------+------------------------------------------------+------------- + import_destination | t1 | loopback | (schema_name 'import_source', table_name 't1') | + import_destination | t3 | loopback | (schema_name 'import_source', table_name 't3') | +(2 rows) + diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c index 7dd43a9..79497f0 100644 --- a/contrib/postgres_fdw/postgres_fdw.c +++ b/contrib/postgres_fdw/postgres_fdw.c @@ -286,6 +286,7 @@ static void postgresExplainForeignModify(ModifyTableState *mtstate, static bool postgresAnalyzeForeignTable(Relation relation, AcquireSampleRowsFunc *func, BlockNumber *totalpages); +static List *postgresImportForeignSchema(ForeignServer *server, ImportForeignSchemaStmt * parsetree); /* * Helper functions @@ -326,6 +327,7 @@ static HeapTuple make_tuple_from_result_row(PGresult *res, List *retrieved_attrs, MemoryContext temp_context); static void conversion_error_callback(void *arg); +static PGresult *fetch_remote_tables(PGconn *conn, ImportForeignSchemaStmt * stmt); /* @@ -363,6 +365,9 @@ postgres_fdw_handler(PG_FUNCTION_ARGS) /* Support functions for ANALYZE */ routine->AnalyzeForeignTable = postgresAnalyzeForeignTable; + /* Support functions for IMPORT FOREIGN SCHEMA */ + routine->ImportForeignSchema = postgresImportForeignSchema; + PG_RETURN_POINTER(routine); } @@ -2347,6 +2352,176 @@ postgresAnalyzeForeignTable(Relation relation, return true; } +static PGresult * +fetch_remote_tables(PGconn *conn, ImportForeignSchemaStmt * stmt) +{ + StringInfoData buf; + int numparams; + const char **params = palloc0(sizeof(char *) * 2); + PGresult *res = NULL; + + initStringInfo(&buf); + params[0] = strdup((stmt->remote_schema)); + + // Check that the schema really exists + appendStringInfo(&buf, "SELECT 1 FROM pg_namespace WHERE nspname = $1"); + res = PQexecParams(conn, buf.data, 1, NULL, params, NULL, NULL, 0); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + pgfdw_report_error(ERROR, res, conn, true, buf.data); + if(PQntuples(res) != 1) + { + ereport(ERROR, + (errcode(ERRCODE_FDW_SCHEMA_NOT_FOUND), + errmsg("The schema %s does not exist on the server", params[0]))); + } + + + // Fetch all tables from this schema + resetStringInfo(&buf); + appendStringInfo(&buf, + "SELECT relname, " + "array_agg(attname::name) as colnames, " + "array_agg(atttypid::regtype) as coltypes, " + "array_agg(atttypmod::int4) as coltypmod, " + "array_agg(attnotnull::bool) as colnullable " + "FROM pg_class " + "INNER JOIN pg_namespace ON pg_class.relnamespace = pg_namespace.oid " + "LEFT JOIN pg_attribute ON pg_class.oid = pg_attribute.attrelid " + " AND pg_attribute.attnum >= 0 AND NOT pg_attribute.attisdropped " + "WHERE relkind IN ('r', 'v', 'f') " + "AND pg_namespace.nspname = $1 "); + /* If any LIMIT TO or EXCEPT clause is present, limit what is actually + * imported. */ + if (stmt->restriction->restriction_type != IMPORT_ALL) + { + Oid outfuncoid; + bool isvarlena; + Datum *elems = palloc0(list_length(stmt->restriction->table_list) * sizeof(Datum)); + ArrayType *array; + int i = 0; + ListCell *lc; + FmgrInfo *fmout = palloc0(sizeof(FmgrInfo)); + getTypeOutputInfo(CSTRINGARRAYOID, &outfuncoid, &isvarlena); + fmgr_info(outfuncoid, fmout); + foreach(lc, stmt->restriction->table_list) + { + elems[i] = CStringGetDatum(((RangeVar *) lfirst(lc))->relname); + i++; + } + array = construct_array(elems, i, CSTRINGOID, -2, false, 'c'); + params[1] = OutputFunctionCall(fmout, PointerGetDatum(array)); + appendStringInfo(&buf, "AND "); + if (stmt->restriction->restriction_type == IMPORT_EXCEPT) + { + appendStringInfo(&buf, "NOT "); + } + appendStringInfo(&buf, "pg_class.relname = ANY($2)"); + numparams = 2; + pfree(fmout); + pfree(elems); + } + else + { + numparams = 1; + } + appendStringInfo(&buf, "GROUP BY pg_class.oid, pg_class.relname"); + res = PQexecParams(conn, buf.data, numparams, NULL, params, NULL, NULL, 0); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + pgfdw_report_error(ERROR, res, conn, true, buf.data); + pfree(params); + return res; +} + +/* + * Map a remote schema to a local one. + */ +static List * +postgresImportForeignSchema(ForeignServer *server, + ImportForeignSchemaStmt * parsetree) +{ + List *tables = NULL; + Oid userid = GetUserId(); + UserMapping *mapping = GetUserMapping(userid, server->serverid); + PGconn *conn = GetConnection(server, mapping, false); + PGresult *res; + int numrows, + i, col; + #define numcols 4 + + /* Initialize FmgrInfo for parsing arrays */ + FmgrInfo *fmgr_infos = palloc0(sizeof(FmgrInfo) * numcols); + Oid typoid, + *typioparam = palloc0(sizeof(Oid) * numcols); + ArrayIterator *arrays = palloc0(sizeof(ArrayIterator) * numcols); + Oid column_types[numcols] = {NAMEARRAYOID, REGTYPEARRAYOID, INT4ARRAYOID, BOOLARRAYOID}; + Datum *array_item = palloc0(sizeof(Datum) * numcols); + bool *isnull = palloc0(sizeof(bool) * numcols); + + for (i = 0; i < numcols; i++) + { + getTypeInputInfo(column_types[i], &typoid, &typioparam[i]); + fmgr_info(typoid, &fmgr_infos[i]); + } + res = fetch_remote_tables(conn, parsetree); + numrows = PQntuples(res); + for (i = 0; i < numrows; i++) + { + CreateForeignTableStmt *stmt = makeNode(CreateForeignTableStmt); + char *tablename; + int colindex; + + tablename = PQgetvalue(res, i, 0); + /* setup the base relation information */ + stmt->base.relation = makeRangeVar(NULL, tablename, 0); + stmt->base.relation->schemaname = strdup(parsetree->local_schema); + stmt->servername = server->servername; + /* Parse arrays of columns from the result */ + for (colindex = 0; colindex < numcols; colindex++) + { + Datum array_datum = InputFunctionCall(&fmgr_infos[colindex], + PQgetvalue(res, i, colindex + 1), + typioparam[colindex], -1); + + arrays[colindex] = array_create_iterator(DatumGetArrayTypeP(array_datum), 0); + } + /* add the individual columns */ + while (array_iterate(arrays[0], &array_item[0], &isnull[0])) + { + ColumnDef *new_column = makeNode(ColumnDef); + for(col = 1; col < numcols; col++) + { + array_iterate(arrays[col], &array_item[col], &isnull[col]); + } + + new_column->colname = DatumGetCString(array_item[0]); + new_column->typeName = makeTypeNameFromOid(DatumGetObjectId(array_item[1]), + DatumGetInt32(array_item[2])); + new_column->is_not_null = DatumGetBool(array_item[3]); + stmt->base.tableElts = lappend(stmt->base.tableElts, new_column); + } + + /* + * Add schema_name and table_name options table_name is added to + * survive a foreign table rename. + */ + stmt->options = lappend(stmt->options, + makeDefElem("schema_name", (Node *) makeString(parsetree->remote_schema))); + stmt->options = lappend(stmt->options, + makeDefElem("table_name", (Node *) makeString(strdup(tablename)))); + tables = lappend(tables, stmt); + } + /* Cleanup */ + PQclear(res); + ReleaseConnection(conn); + pfree(array_item); + pfree(fmgr_infos); + pfree(arrays); + pfree(typioparam); + pfree(isnull); + + return tables; +} + /* * Acquire a random sample of rows from foreign table managed by postgres_fdw. * diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql index 6187839..e356fc4 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -609,3 +609,22 @@ UPDATE rem1 SET f2 = 'testo'; -- Test returning a system attribute INSERT INTO rem1(f2) VALUES ('test') RETURNING ctid; + + +-- Test IMPORT FOREIGN SCHEMA statement +CREATE schema import_destination; +CREATE schema import_source; +CREATE TABLE import_source.t1 (c1 int, c2 varchar NOT NULL); +CREATE TABLE import_source.t2 (c1 int, c2 varchar); +CREATE TABLE import_source.t3 (c1 int, c2 varchar); +IMPORT FOREIGN SCHEMA import_source FROM SERVER loopback INTO import_destination; +\det+ import_destination; +\d import_destination.t1; +DROP SCHEMA import_destination cascade; +CREATE schema import_destination; +IMPORT FOREIGN SCHEMA import_source FROM SERVER loopback LIMIT TO (t1) INTO import_destination; +\det+ import_destination; +IMPORT FOREIGN SCHEMA import_source FROM SERVER loopback EXCEPT (t1, t2) INTO import_destination; +\det+ import_destination; + +