Re: Support logical replication of DDLs

On Wed, 2 Nov 2022 at 05:13, vignesh C <vignesh21(at)gmail(dot)com> wrote:
>
> On Mon, 31 Oct 2022 at 16:17, vignesh C <vignesh21(at)gmail(dot)com> wrote:
> >
> > On Thu, 27 Oct 2022 at 16:02, vignesh C <vignesh21(at)gmail(dot)com> wrote:
> > >
> > > On Thu, 27 Oct 2022 at 02:09, Zheng Li <zhengli10(at)gmail(dot)com> wrote:
> > > >
> > > > > Adding support for deparsing of CREATE/ALTER/DROP LANGUAGE for ddl replication.
> > > >
> > > > Adding support for deparsing of:
> > > > COMMENT
> > > > ALTER DEFAULT PRIVILEGES
> > > > CREATE/DROP ACCESS METHOD
> > >
> > > Adding support for deparsing of:
> > > ALTER/DROP ROUTINE
> > >
> > > The patch also includes fixes for the following issues:
> >
>
Few comments:
1) If the user has specified a non-existing object, then we will throw
the wrong error.
+Datum
+publication_deparse_ddl_command_start(PG_FUNCTION_ARGS)
+{
+ EventTriggerData *trigdata;
+ char *command = psprintf("Drop table command start");
+ DropStmt *stmt;
+ ListCell *cell1;
+
+ if (!CALLED_AS_EVENT_TRIGGER(fcinfo))
+ elog(ERROR, "not fired by event trigger manager");
+
+ trigdata = (EventTriggerData *) fcinfo->context;
+ stmt = (DropStmt *) trigdata->parsetree;
+
+ /* extract the relid from the parse tree */
+ foreach(cell1, stmt->objects)
+ {
+ char relpersist;
+ Node *object = lfirst(cell1);
+ ObjectAddress address;
+ Relation relation = NULL;
+
+ address = get_object_address(stmt->removeType,
+ object,
+
&relation,
+
AccessExclusiveLock,
+ true);
+
+ relpersist = get_rel_persistence(address.objectId);

We could check relation is NULL after getting address and skip
processing that object

2) Materialized view handling is missing:
+ switch (rel->rd_rel->relkind)
+ {
+ case RELKIND_RELATION:
+ case RELKIND_PARTITIONED_TABLE:
+ reltype = "TABLE";
+ break;
+ case RELKIND_INDEX:
+ case RELKIND_PARTITIONED_INDEX:
+ reltype = "INDEX";
+ break;
+ case RELKIND_VIEW:
+ reltype = "VIEW";
+ break;
+ case RELKIND_COMPOSITE_TYPE:
+ reltype = "TYPE";
+ istype = true;
+ break;

We could use this scenario for debugging and verifying:
ALTER MATERIALIZED VIEW testschema.amv SET TABLESPACE regress_tblspace;

3) Readdition of alter table readd statistics is not handled:

+ case AT_DropIdentity:
+ tmpobj = new_objtree_VA("ALTER COLUMN
%{column}I DROP IDENTITY", 2,
+
"type", ObjTypeString, "drop identity",
+
"column", ObjTypeString, subcmd->name);
+
+ append_string_object(tmpobj,
"%{if_not_exists}s",
+
subcmd->missing_ok ? "IF EXISTS" : "");
+
+ subcmds = lappend(subcmds,
new_object_object(tmpobj));
+ break;
+ default:
+ elog(WARNING, "unsupported alter table
subtype %d",
+ subcmd->subtype);
+ break;
+ }

We could use this scenario for debugging and verifying:
CREATE TABLE functional_dependencies (
filler1 TEXT,
filler2 NUMERIC,
a INT,
b TEXT,
filler3 DATE,
c INT,
d TEXT
)
WITH (autovacuum_enabled = off);
CREATE STATISTICS func_deps_stat (dependencies) ON a, b, c FROM
functional_dependencies;
TRUNCATE functional_dependencies;
ALTER TABLE functional_dependencies ALTER COLUMN c TYPE numeric;

4) "Alter sequence as" option not hanlded

+ if (strcmp(elem->defname, "cache") == 0)
+ newelm = deparse_Seq_Cache(alterSeq, seqform, false);
+ else if (strcmp(elem->defname, "cycle") == 0)
+ newelm = deparse_Seq_Cycle(alterSeq, seqform, false);
+ else if (strcmp(elem->defname, "increment") == 0)
+ newelm = deparse_Seq_IncrementBy(alterSeq,
seqform, false);
+ else if (strcmp(elem->defname, "minvalue") == 0)
+ newelm = deparse_Seq_Minvalue(alterSeq, seqform, false);
+ else if (strcmp(elem->defname, "maxvalue") == 0)
+ newelm = deparse_Seq_Maxvalue(alterSeq, seqform, false);
+ else if (strcmp(elem->defname, "start") == 0)
+ newelm = deparse_Seq_Startwith(alterSeq,
seqform, false);
+ else if (strcmp(elem->defname, "restart") == 0)
+ newelm = deparse_Seq_Restart(alterSeq, seqdata);
+ else if (strcmp(elem->defname, "owned_by") == 0)
+ newelm = deparse_Seq_OwnedBy(alterSeq, objectId, false);
+ else
+ elog(ERROR, "invalid sequence option %s",
elem->defname);

We could use this scenario for debugging and verifying:
ALTER SEQUENCE seq1 AS smallint;

5) alter table row level security is not handled:

+ case AT_DropIdentity:
+ tmpobj = new_objtree_VA("ALTER COLUMN
%{column}I DROP IDENTITY", 2,
+
"type", ObjTypeString, "drop identity",
+
"column", ObjTypeString, subcmd->name);
+
+ append_string_object(tmpobj,
"%{if_not_exists}s",
+
subcmd->missing_ok ? "IF EXISTS" : "");
+
+ subcmds = lappend(subcmds,
new_object_object(tmpobj));
+ break;
+ default:
+ elog(WARNING, "unsupported alter table
subtype %d",
+ subcmd->subtype);
+ break;

We could use this scenario for debugging and verifying:
CREATE TABLE r1 (a int);
ALTER TABLE r1 FORCE ROW LEVEL SECURITY;

6) alter table add primary key is not handled:

+ case AT_DropIdentity:
+ tmpobj = new_objtree_VA("ALTER COLUMN
%{column}I DROP IDENTITY", 2,
+
"type", ObjTypeString, "drop identity",
+
"column", ObjTypeString, subcmd->name);
+
+ append_string_object(tmpobj,
"%{if_not_exists}s",
+
subcmd->missing_ok ? "IF EXISTS" : "");
+
+ subcmds = lappend(subcmds,
new_object_object(tmpobj));
+ break;
+ default:
+ elog(WARNING, "unsupported alter table
subtype %d",
+ subcmd->subtype);
+ break;

We could use this scenario for debugging and verifying:
create table idxpart (a int) partition by range (a);
create table idxpart0 (like idxpart);
alter table idxpart0 add primary key (a);
alter table idxpart attach partition idxpart0 for values from (0) to (1000);
alter table only idxpart add primary key (a);

7) Code not updated based on new change:

7.a) identity_column should be removed from new_objtree_VA
+ case AT_AddIdentity:
+ {
+ AttrNumber attnum;
+ Oid seq_relid;
+ ObjTree *seqdef;
+ ColumnDef *coldef =
(ColumnDef *) subcmd->def;
+
+ tmpobj = new_objtree_VA("ALTER
COLUMN %{column}I ADD %{identity_column}s", 2,
+
"type", ObjTypeString, "add identity",
+
"column", ObjTypeString, subcmd->name);
+
+ attnum =
get_attnum(RelationGetRelid(rel), subcmd->name);
+ seq_relid =
getIdentitySequence(RelationGetRelid(rel), attnum, true);
+ seqdef =
deparse_ColumnIdentity(seq_relid, coldef->identity, false);
+
+ append_object_object(tmpobj,
"identity_column", seqdef);

7.b) identity_column should be changed to "%{identity_column}s" in
append_object_object

We could use this scenario for debugging and verifying:
CREATE TABLE itest4 (a int NOT NULL, b text);
ALTER TABLE itest4 ALTER COLUMN a ADD GENERATED ALWAYS AS IDENTITY;

8) SearchSysCache1 copied twice, one of it should be removed
+ /*
+ * Lookup up object in the catalog, so we don't have to deal with
+ * current_user and such.
+ */
+
+ tp = SearchSysCache1(USERMAPPINGOID, ObjectIdGetDatum(objectId));
+ if (!HeapTupleIsValid(tp))
+ elog(ERROR, "cache lookup failed for user mapping %u",
objectId);
+
+ form = (Form_pg_user_mapping) GETSTRUCT(tp);
+
+ /*
+ * Lookup up object in the catalog, so we don't have to deal with
+ * current_user and such.
+ */
+
+ tp = SearchSysCache1(USERMAPPINGOID, ObjectIdGetDatum(objectId));
+ if (!HeapTupleIsValid(tp))
+ elog(ERROR, "cache lookup failed for user mapping %u",
objectId);

9) Create table with INCLUDING GENERATED not handled:
+ case AT_DropIdentity:
+ tmpobj = new_objtree_VA("ALTER COLUMN
%{column}I DROP IDENTITY", 2,
+
"type", ObjTypeString, "drop identity",
+
"column", ObjTypeString, subcmd->name);
+
+ append_string_object(tmpobj,
"%{if_not_exists}s",
+
subcmd->missing_ok ? "IF EXISTS" : "");
+
+ subcmds = lappend(subcmds,
new_object_object(tmpobj));
+ break;
+ default:
+ elog(WARNING, "unsupported alter table
subtype %d",
+ subcmd->subtype);
+ break;

We could use this scenario for debugging and verifying:
CREATE TABLE gtest28a (a int, b int, c int, x int GENERATED ALWAYS
AS (b * 2) STORED);
CREATE TABLE gtest28b (LIKE gtest28a INCLUDING GENERATED);

Regards,
Vignesh