cartodb4/cartodb-postgresql
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Federated Server: Unify remote table listing

Browse Source 
- Removes DB_Federated_Server_List_Registered_Tables in favor
of CDB_Federated_Server_List_Remote_Tables that now returns
both registered and unregistered tables and, when they are
registered, the information about the associated view
This commit is contained in:
Raul Marin
2019-11-06 17:04:11 +01:00
parent 90b58fcdec
commit 01258530ff
6 changed files with 257 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files

110
scripts-available/CDB_FederatedServerListRemote.sql
View File

@@ -30,7 +30,8 @@ BEGIN
WHERE relnamespace = (SELECT oid FROM pg_namespace WHERE nspname = local_schema)
AND relname = remote_table
) THEN
EXECUTE format('IMPORT FOREIGN SCHEMA %I LIMIT TO (%I) FROM SERVER %I INTO %I', inf_schema, remote_table, server_internal, local_schema);
EXECUTE format('IMPORT FOREIGN SCHEMA %I LIMIT TO (%I) FROM SERVER %I INTO %I',
inf_schema, remote_table, server_internal, local_schema);
END IF;
-- Return the result we're interested in. Exclude toast and temp schemas
@@ -49,7 +50,7 @@ $$
LANGUAGE PLPGSQL VOLATILE PARALLEL UNSAFE;
--
-- List the tables from a remote PG schema
-- List the names of the tables in a remote PG schema
--
CREATE OR REPLACE FUNCTION @extschema@.__CDB_FS_List_Foreign_Tables_PG(server_internal name, remote_schema name)
RETURNS TABLE(remote_table name)
@@ -74,10 +75,10 @@ BEGIN
WHERE relnamespace = (SELECT oid FROM pg_namespace WHERE nspname = local_schema)
AND relname = remote_table
) THEN
EXECUTE format('IMPORT FOREIGN SCHEMA %I LIMIT TO (%I) FROM SERVER %I INTO %I', inf_schema, remote_table, server_internal, local_schema);
EXECUTE format('IMPORT FOREIGN SCHEMA %I LIMIT TO (%I) FROM SERVER %I INTO %I',
inf_schema, remote_table, server_internal, local_schema);
END IF;
-- Return the result we're interested in
-- Note: in this context, schema names are not to be quoted
RETURN QUERY EXECUTE format($q$
SELECT table_name::name AS remote_table FROM %I.%I WHERE table_schema = '%s' ORDER BY table_name
@@ -88,10 +89,10 @@ LANGUAGE PLPGSQL VOLATILE PARALLEL UNSAFE;
--
-- List the columns from a remote PG table
-- List the columns in a remote PG schema
--
CREATE OR REPLACE FUNCTION @extschema@.__CDB_FS_List_Foreign_Columns_PG(server_internal name, remote_schema name, remote_table name)
RETURNS TABLE(column_name name, column_type text)
CREATE OR REPLACE FUNCTION @extschema@.__CDB_FS_List_Foreign_Columns_PG(server_internal name, remote_schema name)
RETURNS TABLE(table_name name, column_name name, column_type text)
AS $func$
DECLARE
-- Import `columns` from the information schema
@@ -112,30 +113,37 @@ BEGIN
WHERE relnamespace = (SELECT oid FROM pg_namespace WHERE nspname = local_schema)
AND relname = remote_col_table
) THEN
EXECUTE format('IMPORT FOREIGN SCHEMA %I LIMIT TO (%I) FROM SERVER %I INTO %I', inf_schema, remote_col_table, server_internal, local_schema);
EXECUTE format('IMPORT FOREIGN SCHEMA %I LIMIT TO (%I) FROM SERVER %I INTO %I',
inf_schema, remote_col_table, server_internal, local_schema);
END IF;
-- Return the result we're interested in
-- Note: in this context, remote schema and remote table names are not to be quoted
-- Note: in this context, schema names are not to be quoted
-- We join with the geometry columns to change the type `USER-DEFINED`
-- by its appropiate geometry and srid
RETURN QUERY EXECUTE format($q$
SELECT
a.column_name::name, COALESCE(b.column_type, a.data_type)::TEXT as column_type
FROM %I.%I a
LEFT JOIN @extschema@.__CDB_FS_List_Foreign_Geometry_Columns_PG('%s', '%s', '%s') b ON a.column_name = b.column_name
WHERE table_schema = '%s' AND table_name = '%s'
ORDER BY column_name$q$,
a.table_name::name,
a.column_name::name,
COALESCE(b.column_type, a.data_type)::TEXT as column_type
FROM
%I.%I a
LEFT JOIN
@extschema@.__CDB_FS_List_Foreign_Geometry_Columns_PG('%s', '%s') b
ON a.table_name = b.table_name AND a.column_name = b.column_name
WHERE table_schema = '%s'
ORDER BY a.table_name, a.column_name $q$,
local_schema, remote_col_table,
server_internal, remote_schema, remote_table,
remote_schema, remote_table);
server_internal, remote_schema,
remote_schema);
END
$func$
LANGUAGE PLPGSQL VOLATILE PARALLEL UNSAFE;
--
-- List the columns from a remote PG table
-- List the geometry columns in a remote PG schema
--
CREATE OR REPLACE FUNCTION @extschema@.__CDB_FS_List_Foreign_Geometry_Columns_PG(server_internal name, remote_schema name, remote_table name, postgis_schema name DEFAULT 'public')
RETURNS TABLE(column_name name, column_type text)
CREATE OR REPLACE FUNCTION @extschema@.__CDB_FS_List_Foreign_Geometry_Columns_PG(server_internal name, remote_schema name, postgis_schema name DEFAULT 'public')
RETURNS TABLE(table_name name, column_name name, column_type text)
AS $func$
DECLARE
-- Import `geometry_columns` and `geography_columns` from the postgis schema
@@ -159,21 +167,22 @@ BEGIN
BEGIN
-- Note: We return both the type and srid as the type
RETURN QUERY EXECUTE format($q$
SELECT f_geometry_column::NAME as column_name,
type::TEXT || ',' || srid::TEXT as column_type
FROM
(
SELECT * FROM %I.%I UNION ALL SELECT * FROM %I.%I
) _geo_views
WHERE
f_table_schema = '%s' AND
f_table_name = '%s'
SELECT
f_table_name::NAME as table_name,
f_geometry_column::NAME as column_name,
type::TEXT || ',' || srid::TEXT as column_type
FROM
(
SELECT * FROM %I.%I UNION ALL SELECT * FROM %I.%I
) _geo_views
WHERE f_table_schema = '%s'
$q$,
local_schema, remote_geometry_view,
local_schema, remote_geography_view,
remote_schema, remote_table);
remote_schema);
EXCEPTION WHEN OTHERS THEN
RAISE NOTICE 'Could not find Postgis installation in the remote "%" schema', postgis_schema;
RAISE INFO 'Could not find Postgis installation in the remote "%" schema in server "%"',
postgis_schema, @extschema@.__CDB_FS_Extract_Server_Name(server_internal);
RETURN;
END;
END
@@ -207,9 +216,18 @@ LANGUAGE PLPGSQL VOLATILE PARALLEL UNSAFE;
--
-- List remote tables in a federated server that the current user has access to.
-- For registered tables it returns also the associated configuration
--
CREATE OR REPLACE FUNCTION @extschema@.CDB_Federated_Server_List_Remote_Tables(server TEXT, remote_schema TEXT)
RETURNS TABLE(remote_table name)
RETURNS TABLE(
registered boolean,
remote_table TEXT,
local_qualified_name TEXT,
id_column_name TEXT,
geom_column_name TEXT,
webmercator_column_name TEXT,
columns JSON
)
AS $$
DECLARE
server_internal name := @extschema@.__CDB_FS_Generate_Server_Name(input_name := server, check_existence := true);
@@ -217,7 +235,29 @@ DECLARE
BEGIN
CASE server_type
WHEN 'postgres_fdw' THEN
RETURN QUERY SELECT @extschema@.__CDB_FS_List_Foreign_Tables_PG(server_internal, remote_schema);
RETURN QUERY
SELECT
coalesce(registered_tables.registered, false)::boolean as registered,
foreign_tables.remote_table::text as remote_table,
registered_tables.local_qualified_name as local_qualified_name,
registered_tables.id_column_name as id_column_name,
registered_tables.geom_column_name as geom_column_name,
registered_tables.webmercator_column_name as webmercator_column_name,
remote_columns.columns as columns
FROM
@extschema@.__CDB_FS_List_Foreign_Tables_PG(server_internal, remote_schema) foreign_tables
LEFT JOIN
@extschema@.__CDB_FS_List_Registered_Tables(server_internal, remote_schema) registered_tables
ON foreign_tables.remote_table = registered_tables.remote_table
LEFT JOIN
( -- Extract and group columns with their remote table
SELECT table_name,
json_agg(json_build_object('Name', column_name, 'Type', column_type)) as columns
FROM @extschema@.__CDB_FS_List_Foreign_Columns_PG(server_internal, remote_schema)
GROUP BY table_name
) remote_columns
ON foreign_tables.remote_table = remote_columns.table_name
ORDER BY foreign_tables.remote_table;
ELSE
RAISE EXCEPTION 'Not implemented server type % for remote server %', server_type, remote_server;
END CASE;
@@ -232,7 +272,7 @@ CREATE OR REPLACE FUNCTION @extschema@.CDB_Federated_Server_List_Remote_Columns(
server TEXT,
remote_schema TEXT,
remote_table TEXT)
RETURNS TABLE(column_name name, column_type text)
RETURNS TABLE(column_n name, column_t text)
AS $$
DECLARE
server_internal name := @extschema@.__CDB_FS_Generate_Server_Name(input_name := server, check_existence := true);
@@ -244,7 +284,7 @@ BEGIN
CASE server_type
WHEN 'postgres_fdw' THEN
RETURN QUERY SELECT * FROM @extschema@.__CDB_FS_List_Foreign_Columns_PG(server_internal, remote_schema, remote_table);
RETURN QUERY SELECT column_name, column_type FROM @extschema@.__CDB_FS_List_Foreign_Columns_PG(server_internal, remote_schema) where table_name = remote_table;
ELSE
RAISE EXCEPTION 'Not implemented server type % for remote server %', server_type, remote_server;
END CASE;

169
scripts-available/CDB_FederatedServerTables.sql
View File

@@ -18,7 +18,7 @@ BEGIN
RETURN FOUND;
END
$$
LANGUAGE PLPGSQL VOLATILE PARALLEL UNSAFE;
LANGUAGE PLPGSQL STABLE PARALLEL UNSAFE;
--
-- Checks if a column is of geometry type
@@ -34,7 +34,7 @@ BEGIN
RETURN FOUND;
END
$$
LANGUAGE PLPGSQL VOLATILE PARALLEL UNSAFE;
LANGUAGE PLPGSQL STABLE PARALLEL UNSAFE;
--
-- Returns the name of all the columns from a table
@@ -55,7 +55,108 @@ AS $$
WHERE c.oid = input_table::oid
)
ORDER BY a.attnum;
$$ LANGUAGE SQL;
$$ LANGUAGE SQL STABLE PARALLEL UNSAFE;
--
-- Returns the id column from a view definition
--
CREATE OR REPLACE FUNCTION @extschema@.__CDB_FS_Get_View_id_column(view_def TEXT)
RETURNS TEXT
AS $$
WITH column_definitions AS
(
SELECT regexp_split_to_array(regexp_split_to_table(view_def, '\n'), ' ') AS col_def
)
SELECT split_part(col_def[array_length(col_def, 1) - 2], '.', 2)
FROM column_definitions where col_def[array_length(col_def, 1)] = 'cartodb_id,'
LIMIT 1;
$$ LANGUAGE SQL IMMUTABLE PARALLEL SAFE;
--
-- Returns the geom column from a view definition
--
CREATE OR REPLACE FUNCTION @extschema@.__CDB_FS_Get_View_geom_column(view_def TEXT)
RETURNS TEXT
AS $$
WITH column_definitions AS
(
SELECT regexp_split_to_array(regexp_split_to_table(view_def, '\n'), ' ') AS col_def
)
SELECT trim(trailing ',' FROM split_part(
CASE WHEN col_def[array_length(col_def, 1) - 2] = '4326)' THEN col_def[array_length(col_def, 1) - 3]
ELSE col_def[array_length(col_def, 1) - 2]
END, '.', 2))
FROM column_definitions
WHERE col_def[array_length(col_def, 1)] = 'the_geom,'
LIMIT 1;
$$ LANGUAGE SQL IMMUTABLE PARALLEL SAFE;
--
-- Returns the webmercatorcolumn from a view definition
--
CREATE OR REPLACE FUNCTION @extschema@.__CDB_FS_Get_View_webmercator_column(view_def TEXT)
RETURNS TEXT
AS $$
WITH column_definitions AS
(
SELECT regexp_split_to_array(regexp_split_to_table(view_def, '\n'), ' ') AS col_def
)
SELECT trim(trailing ',' FROM split_part(
CASE WHEN col_def[array_length(col_def, 1) - 2] = '3857)' THEN col_def[array_length(col_def, 1) - 3]
ELSE col_def[array_length(col_def, 1) - 2]
END, '.', 2))
FROM column_definitions
WHERE col_def[array_length(col_def, 1)] = 'the_geom_webmercator,'
LIMIT 1;
$$ LANGUAGE SQL IMMUTABLE PARALLEL SAFE;
--
-- List all registered tables in a server + schema
--
CREATE OR REPLACE FUNCTION @extschema@.__CDB_FS_List_Registered_Tables(
server_internal NAME,
remote_schema TEXT
)
RETURNS TABLE(
registered boolean,
remote_table TEXT,
local_qualified_name TEXT,
id_column_name TEXT,
geom_column_name TEXT,
webmercator_column_name TEXT
)
AS $$
DECLARE
local_schema name := @extschema@.__CDB_FS_Create_Schema(server_internal, remote_schema);
BEGIN
RETURN QUERY SELECT
true as registered,
source_table::text as remote_table,
format('%I.%I', dependent_schema, dependent_view)::text as local_qualified_name,
@extschema@.__CDB_FS_Get_View_id_column(view_definition) as id_column_name,
@extschema@.__CDB_FS_Get_View_geom_column(view_definition) as geom_column_name,
@extschema@.__CDB_FS_Get_View_webmercator_column(view_definition) as webmercator_column_name
FROM
(
SELECT DISTINCT
dependent_ns.nspname as dependent_schema,
dependent_view.relname as dependent_view,
source_table.relname as source_table,
pg_get_viewdef(dependent_view.oid) as view_definition
FROM pg_depend
JOIN pg_rewrite ON pg_depend.objid = pg_rewrite.oid
JOIN pg_class as dependent_view ON pg_rewrite.ev_class = dependent_view.oid
JOIN pg_class as source_table ON pg_depend.refobjid = source_table.oid
JOIN pg_namespace dependent_ns ON dependent_ns.oid = dependent_view.relnamespace
JOIN pg_namespace source_ns ON source_ns.oid = source_table.relnamespace
WHERE
source_ns.nspname = local_schema
ORDER BY 1,2
) _aux;
END
$$
LANGUAGE PLPGSQL VOLATILE PARALLEL UNSAFE;
--------------------------------------------------------------------------------
@@ -119,15 +220,22 @@ BEGIN
-- Import the foreign table
-- Drop the old view / table if there was one
IF EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = local_schema AND table_name = remote_table) THEN
IF EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = local_schema AND table_name = remote_table) THEN
EXECUTE @extschema@.CDB_Federated_Table_Unregister(server, remote_schema, remote_table);
END IF;
EXECUTE FORMAT ('IMPORT FOREIGN SCHEMA %I LIMIT TO (%I) FROM SERVER %I INTO %I;', remote_schema, remote_table, server_internal, local_schema);
src_table := format('%I.%I', local_schema, remote_table);
--- Grant SELECT to fdw role (TODO: Re-enable if needed)
--- EXECUTE FORMAT ('GRANT SELECT ON %I.%I TO %I;', fdw_objects_name, table_name, fdw_objects_name);
BEGIN
EXECUTE FORMAT('IMPORT FOREIGN SCHEMA %I LIMIT TO (%I) FROM SERVER %I INTO %I;',
remote_schema, remote_table, server_internal, local_schema);
EXCEPTION WHEN OTHERS THEN
RAISE EXCEPTION 'Could not import schema "%" of server "%"', remote_schema, server;
END;
BEGIN
src_table := format('%I.%I', local_schema, remote_table);
EXCEPTION WHEN OTHERS THEN
RAISE EXCEPTION 'Could not import table "%.%" of server "%"', remote_schema, remote_table, server;
END;
-- Check id_column is numeric
IF NOT @extschema@.__CDB_FS_Column_Is_Integer(src_table, id_column) THEN
RAISE EXCEPTION 'non integer id_column "%"', id_column;
@@ -155,7 +263,7 @@ BEGIN
geom_expression := format('t.%I AS the_geom', geom_column);
ELSE
-- It needs an ST_Transform to 4326
geom_expression := format('@postgisschema@.ST_Transform(t.%I, 4326) AS the_geom', geom_column);
geom_expression := format('@postgisschema@.ST_Transform(t.%I,4326) AS the_geom', geom_column);
END IF;
IF webmercator_column IS NULL
@@ -166,7 +274,7 @@ BEGIN
webmercator_expression := format('t.%I AS the_geom_webmercator', webmercator_column);
ELSE
-- It needs an ST_Transform to 3857
webmercator_expression := format('@postgisschema@.ST_Transform(t.%I, 3857) AS the_geom_webmercator', webmercator_column);
webmercator_expression := format('@postgisschema@.ST_Transform(t.%I,3857) AS the_geom_webmercator', webmercator_column);
END IF;
-- CARTO columns expressions
@@ -192,9 +300,6 @@ BEGIN
WHEN OTHERS THEN
RAISE EXCEPTION 'Could not import table "%" as "%": %', remote_table, local_name, SQLERRM;
END;
-- TODO: Handle this Grant perms to the view
-- EXECUTE format('GRANT SELECT ON %I TO %s', table_name, fdw_objects_name);
END
$$
LANGUAGE PLPGSQL VOLATILE PARALLEL UNSAFE;
@@ -217,39 +322,3 @@ BEGIN
END
$$
LANGUAGE PLPGSQL VOLATILE PARALLEL UNSAFE;
--
-- List all registered tables in a server + schema
--
CREATE OR REPLACE FUNCTION @extschema@.CDB_Federated_Server_List_Registered_Tables(
server TEXT,
remote_schema TEXT
)
RETURNS TABLE(remote_name TEXT, local_name TEXT)
AS $$
DECLARE
server_internal name := @extschema@.__CDB_FS_Generate_Server_Name(input_name := server, check_existence := false);
local_schema name := @extschema@.__CDB_FS_Create_Schema(server_internal, remote_schema);
BEGIN
RETURN QUERY SELECT
source_table::text as remote_table,
format('%I.%I', dependent_schema, dependent_view)::text as local_view
FROM
(
SELECT DISTINCT
dependent_ns.nspname as dependent_schema,
dependent_view.relname as dependent_view,
source_table.relname as source_table
FROM pg_depend
JOIN pg_rewrite ON pg_depend.objid = pg_rewrite.oid
JOIN pg_class as dependent_view ON pg_rewrite.ev_class = dependent_view.oid
JOIN pg_class as source_table ON pg_depend.refobjid = source_table.oid
JOIN pg_namespace dependent_ns ON dependent_ns.oid = dependent_view.relnamespace
JOIN pg_namespace source_ns ON source_ns.oid = source_table.relnamespace
WHERE
source_ns.nspname = local_schema
ORDER BY 1,2
) _aux;
END
$$
LANGUAGE PLPGSQL VOLATILE PARALLEL UNSAFE;