-- start_matchignore
-- m/WARNING:  tablespace symlink path is too long for TAR/
-- m/DETAIL:  The symlinked path \".*\" will be truncated to 100 characters when sending a TAR to the utilities \(e.g. pg_basebackup\)/
-- end_matchignore
create or replace function has_init_file_for_oid(object oid) returns boolean as $$
	select count(pg_stat_file(p.pg_relation_filepath || '_init')) = 1 as success
	 from (select * from pg_relation_filepath(object)) p;
$$ language sql;
create or replace function has_init_file(aotablename regclass, tabletype text) returns boolean as $$
DECLARE
	relation_id regclass;
BEGIN
	case tabletype
	when 'table' then
		relation_id := aotablename;
	when 'index' then
		relation_id := aotablename;
	when 'aoseg' then
		select segrelid into relation_id from pg_appendonly where relid = aotablename;
	when 'blockdir' then
		select blkdirrelid into relation_id from pg_appendonly where relid = aotablename;
	when 'blockdirindex' then
		select blkdiridxid into relation_id from pg_appendonly where relid = aotablename;
	when 'visimap' then
		select visimaprelid into relation_id from pg_appendonly where relid = aotablename;
	when 'visimapindex' then
		select visimapidxid into relation_id from pg_appendonly where relid = aotablename;
	else
		raise notice 'Invalid tabletype for has_init_file %', tabletype;
		return false;
	end case;

	return has_init_file_for_oid(relation_id);
END;
$$ language plpgsql;
CREATE OR REPLACE FUNCTION get_tablespace_version_directory_name()
	RETURNS TEXT
	AS '@abs_builddir@/regress.so', 'get_tablespace_version_directory_name'
    LANGUAGE C;
-- create tablespaces we can use
CREATE TABLESPACE testspace LOCATION '@testtablespace@';
CREATE TABLESPACE ul_testspace LOCATION '@testtablespace@_unlogged';
SELECT gp_segment_id,
       CASE tblspc_loc
            WHEN '@testtablespace@' THEN 'testtablespace'
            ELSE 'testtablespace_unknown'
       END AS tblspc_loc
FROM gp_tablespace_location((SELECT oid FROM pg_tablespace WHERE spcname='testspace'));
 gp_segment_id |   tblspc_loc   
---------------+----------------
             1 | testtablespace
             2 | testtablespace
             0 | testtablespace
            -1 | testtablespace
(4 rows)

SELECT gp_segment_id,
       CASE tblspc_loc
            WHEN '@testtablespace@_unlogged' THEN 'testtablespace_unlogged'
            ELSE 'testtablespace_unknown'
       END AS tblspc_loc
FROM gp_tablespace_location((SELECT oid FROM pg_tablespace WHERE spcname='ul_testspace'));
 gp_segment_id |       tblspc_loc        
---------------+-------------------------
             2 | testtablespace_unlogged
             1 | testtablespace_unlogged
             0 | testtablespace_unlogged
            -1 | testtablespace_unlogged
(4 rows)

-- Test that test tablespaces have the catalog directory directly under their
-- symlink
SELECT pg_ls_dir('./pg_tblspc/' || oid) = get_tablespace_version_directory_name()
         AS has_version_dir
FROM pg_tablespace WHERE spcname = 'testspace';
 has_version_dir 
-----------------
 t
(1 row)

-- Ensure mirrors have applied filesystem changes
SELECT force_mirrors_to_catch_up();
 force_mirrors_to_catch_up 
---------------------------
 
(1 row)

\! ls @testtablespace@;
1
2
3
4
5
6
7
8
-- Test moving AO/AOCO tables from one tablespace to another.
CREATE TABLE ao_ts_table (id int4, t text) with (appendonly=true, orientation=row) distributed by (id);
CREATE TABLE aoco_ts_table (id int4, t text) with (appendonly=true, orientation=column) distributed by (id);
CREATE UNLOGGED TABLE ao_ul_ts_table (id int4, t text)
with (appendonly=true, orientation=row) distributed by (id);
CREATE UNLOGGED TABLE aoco_ul_ts_table (id int4, t text)
with (appendonly=true, orientation=column) distributed by (id);
insert into ao_ts_table select g, 'foo' || g from generate_series(1, 10000) g;
insert into aoco_ts_table select g, 'bar' || g from generate_series(1, 10000) g;
insert into ao_ul_ts_table select * from ao_ts_table;
insert into aoco_ul_ts_table select * from ao_ul_ts_table;
-- alter table to create scenario where .0 file also has data
ALTER TABLE ao_ts_table ALTER COLUMN id TYPE bigint;
-- to satisfy orca
ANALYZE ao_ts_table;
ALTER TABLE aoco_ts_table ALTER COLUMN id TYPE bigint;
-- to satisfy orca
ANALYZE aoco_ts_table;
insert into ao_ts_table select g, 'foofoo' || g from generate_series(10000, 10100) g;
insert into aoco_ts_table select g, 'barbar' || g from generate_series(10000, 10100) g;
CREATE INDEX ao_ts_index ON ao_ts_table(id);
CREATE INDEX ao_ul_ts_index ON ao_ul_ts_table(id);
SELECT COUNT(*) FROM ao_ts_table;
 count 
-------
 10101
(1 row)

SELECT COUNT(*) FROM aoco_ts_table;
 count 
-------
 10101
(1 row)

SELECT COUNT(*) FROM ao_ul_ts_table;
 count 
-------
 10000
(1 row)

SELECT COUNT(*) FROM aoco_ul_ts_table;
 count 
-------
 10000
(1 row)

ALTER TABLE ao_ts_table SET TABLESPACE testspace;
ALTER TABLE aoco_ts_table SET TABLESPACE testspace;
ALTER INDEX ao_ts_index SET TABLESPACE testspace;
ALTER TABLE ao_ul_ts_table SET TABLESPACE ul_testspace;
ALTER INDEX ao_ul_ts_index SET TABLESPACE ul_testspace;
ALTER TABLE aoco_ul_ts_table SET TABLESPACE ul_testspace;
-- Check that init fork exists on master
select has_init_file('ao_ul_ts_table', 'table');
 has_init_file 
---------------
 t
(1 row)

select has_init_file('ao_ul_ts_index', 'index');
 has_init_file 
---------------
 t
(1 row)

select has_init_file('ao_ul_ts_table', 'aoseg');
 has_init_file 
---------------
 t
(1 row)

select has_init_file('ao_ul_ts_table', 'blockdir');
 has_init_file 
---------------
 t
(1 row)

select has_init_file('ao_ul_ts_table', 'blockdirindex');
 has_init_file 
---------------
 t
(1 row)

select has_init_file('ao_ul_ts_table', 'visimap');
 has_init_file 
---------------
 t
(1 row)

select has_init_file('ao_ul_ts_table', 'visimapindex');
 has_init_file 
---------------
 t
(1 row)

INSERT INTO ao_ts_table VALUES(-1);
INSERT INTO aoco_ts_table VALUES(-1);
SELECT COUNT(*) FROM ao_ts_table;
 count 
-------
 10102
(1 row)

SELECT COUNT(*) FROM aoco_ts_table;
 count 
-------
 10102
(1 row)

INSERT INTO ao_ul_ts_table VALUES(-1);
INSERT INTO aoco_ul_ts_table VALUES(-1);
SELECT COUNT(*) FROM ao_ul_ts_table;
 count 
-------
 10001
(1 row)

SELECT COUNT(*) FROM aoco_ul_ts_table;
 count 
-------
 10001
(1 row)

-- Since count(*) for CO doesn't actually read all the columns, this query will
-- read all the columns and all the newly written files.
SELECT * FROM aoco_ts_table where id > 9995 and id < 10005;
  id   |      t      
-------+-------------
  9997 | bar9997
  9998 | bar9998
 10000 | bar10000
 10000 | barbar10000
 10001 | barbar10001
 10002 | barbar10002
 10003 | barbar10003
 10004 | barbar10004
  9996 | bar9996
  9999 | bar9999
(10 rows)

SELECT * FROM aoco_ul_ts_table where id > 9995 and id < 10005;
  id   |    t     
-------+----------
  9999 | foo9999
 10000 | foo10000
  9996 | foo9996
  9997 | foo9997
  9998 | foo9998
(5 rows)

-- Clean up. (It would be good to leave some extra tablespaces behind, so that
-- they would go through the gpcheckcat, pg_upgrade, etc. passes that run
-- after the main regression test suite. But all those tools are not up to
-- snuff yet to deal with tablespaces.)
drop table ao_ts_table;
drop table aoco_ts_table;
drop tablespace testspace;
drop table ao_ul_ts_table;
drop table aoco_ul_ts_table;
drop tablespace ul_testspace;
-- Greenplum tablespaces have the option to define tablespace location for specific segments
CREATE TABLESPACE testspace_otherloc LOCATION '@testtablespace@' WITH (content9999='@testtablespace@_otherloc'); -- should fail
ERROR:  segment content ID 9999 does not exist
HINT:  Segment content IDs can be found in gp_segment_configuration table.
CREATE TABLESPACE testspace_otherloc LOCATION '@testtablespace@' WITH (content1='@testtablespace@_otherloc');
SELECT gp_segment_id,
       CASE tblspc_loc
            WHEN '@testtablespace@' THEN 'testtablespace'
            WHEN '@testtablespace@_otherloc' THEN 'testtablespace_otherloc'
            ELSE 'testtablespace_unknown'
       END AS tblspc_loc
FROM gp_tablespace_location((SELECT oid FROM pg_tablespace WHERE spcname='testspace_otherloc'));
 gp_segment_id |       tblspc_loc        
---------------+-------------------------
             1 | testtablespace_otherloc
             2 | testtablespace
             0 | testtablespace
            -1 | testtablespace
(4 rows)

-- Create a tablespace with an existing GP_TABLESPACE_VERSION_DIRECTORY for
-- another version of GPDB.
CREATE TABLESPACE testspace_existing_version_dir LOCATION '@testtablespace@_existing_version_dir';
SELECT * FROM
  (SELECT pg_ls_dir('pg_tblspc/' || oid) AS versiondirs
    FROM pg_tablespace
    WHERE spcname = 'testspace_existing_version_dir'
  ) a
WHERE a.versiondirs != get_tablespace_version_directory_name();
    versiondirs    
-------------------
 GPDB_99_399999991
(1 row)

SELECT count(*) FROM
  (SELECT pg_ls_dir('pg_tblspc/' || oid) AS versiondirs
    FROM pg_tablespace
    WHERE spcname = 'testspace_existing_version_dir'
  ) a
WHERE a.versiondirs = get_tablespace_version_directory_name();
 count 
-------
     1
(1 row)

-- Do not drop the dbid directory, nor the existing version directory if you
-- drop this tablespace
DROP TABLESPACE testspace_existing_version_dir;
-- Ensure mirrors have applied filesystem changes
SELECT force_mirrors_to_catch_up();
 force_mirrors_to_catch_up 
---------------------------
 
(1 row)

\! ls @testtablespace@_existing_version_dir/*;
@testtablespace@_existing_version_dir/1:
GPDB_99_399999991

@testtablespace@_existing_version_dir/2:
GPDB_99_399999991

@testtablespace@_existing_version_dir/3:
GPDB_99_399999991

@testtablespace@_existing_version_dir/4:
GPDB_99_399999991

@testtablespace@_existing_version_dir/5:
GPDB_99_399999991

@testtablespace@_existing_version_dir/6:
GPDB_99_399999991

@testtablespace@_existing_version_dir/7:
GPDB_99_399999991

@testtablespace@_existing_version_dir/8:
GPDB_99_399999991
-- Test alter tablespace: PG does not seem to test these.
-- test SET & OWNER
ALTER TABLESPACE testspace_otherloc SET (random_page_cost=20.0);
SELECT spcoptions FROM pg_tablespace WHERE spcname = 'testspace_otherloc';
       spcoptions        
-------------------------
 {random_page_cost=20.0}
(1 row)

CREATE ROLE t_owner;
NOTICE:  resource queue required -- using default resource queue "pg_default"
ALTER TABLESPACE testspace_otherloc OWNER TO t_owner;
SELECT rolname from pg_roles WHERE oid in (SELECT spcowner FROM pg_tablespace WHERE spcname = 'testspace_otherloc');
 rolname 
---------
 t_owner
(1 row)

ALTER TABLESPACE testspace_otherloc OWNER TO @curusername@;
DROP ROLE t_owner;
-- test RENAME
ALTER TABLESPACE testspace_otherloc RENAME TO testspace_otherloc2;
SELECT spcname, spcoptions FROM pg_tablespace WHERE spcname = 'testspace_otherloc2';
       spcname       |       spcoptions        
---------------------+-------------------------
 testspace_otherloc2 | {random_page_cost=20.0}
(1 row)

ALTER TABLESPACE testspace_otherloc2 RENAME TO testspace_otherloc;
-- test RESET
ALTER TABLESPACE testspace_otherloc RESET (random_page_cost);
SELECT spcoptions FROM pg_tablespace WHERE spcname = 'testspace_otherloc';
 spcoptions 
------------
 
(1 row)

CREATE TABLE tblspc_otherloc_heap(a int, b text) distributed by (a);
INSERT INTO tblspc_otherloc_heap select i, 'foo' || i from generate_series(1,10000)i;
SELECT COUNT(*) FROM tblspc_otherloc_heap;
 count 
-------
 10000
(1 row)

DROP TABLE tblspc_otherloc_heap;
DROP TABLESPACE testspace_otherloc;
CREATE TABLESPACE testspace_dir_empty LOCATION '@testtablespace@';
CREATE TABLE t_dir_empty(a int);
\! rm -rf @testtablespace@/*;
DROP TABLE IF EXISTS t_dir_empty;
DROP TABLESPACE testspace_dir_empty;
WARNING:  could not open directory "@testtablespace@/2": No such file or directory  (seg0 127.0.0.1:7002 pid=3000)
WARNING:  could not open directory "@testtablespace@/3": No such file or directory  (seg1 127.0.0.1:7003 pid=3001)
WARNING:  could not open directory "@testtablespace@/4": No such file or directory  (seg2 127.0.0.1:7004 pid=3002)
WARNING:  could not open directory "@testtablespace@/1": No such file or directory