Re: Support logical replication of DDLs - Mailing list pgsql-hackers

On Fri, 11 Nov 2022 at 20:09, vignesh C <vignesh21@gmail.com> wrote:
>
> On Fri, 4 Nov 2022 at 15:06, vignesh C <vignesh21@gmail.com> wrote:
> >
> > On Wed, 2 Nov 2022 at 05:13, vignesh C <vignesh21@gmail.com> wrote:
> > >
> > > On Mon, 31 Oct 2022 at 16:17, vignesh C <vignesh21@gmail.com> wrote:
> > > >
> > > > On Thu, 27 Oct 2022 at 16:02, vignesh C <vignesh21@gmail.com> wrote:
> > > > >
> > > > > On Thu, 27 Oct 2022 at 02:09, Zheng Li <zhengli10@gmail.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
>
> Modified
>
> > 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;
>
> Modified
>
> > 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;
>
> Modified
>
> > 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;
>
> Modified
>
> > 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;
>
> Modified
>
> > 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);
>
> Modified
>
> > 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;
>
> Modified
> > 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);
>
> Modified
>
> > 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);
>
> Modified
>
> The attached v36 patch has the changes for the same.

CFBot reported an issue with the patch, the updated patch has the
changes for the same.

Regards,
Vignesh