RE: Support logical replication of DDLs

On Wednesday, April 12, 2023 7:24 PM Amit Kapila <amit(dot)kapila16(at)gmail(dot)com> wrote:
>
> On Fri, Apr 7, 2023 at 8:52 AM houzj(dot)fnst(at)fujitsu(dot)com
> <houzj(dot)fnst(at)fujitsu(dot)com> wrote:
> >
>
> Few comments on 0001

Thanks for the comments.

> ===================
> 1.
> + ConstrObjDomain,
> + ConstrObjForeignTable
> +} ConstraintObjType;
>
> These both object types don't seem to be supported by the first patch.
> So, I don't see why these should be part of it.
>

done, removed.

> 2.
> +append_string_object(ObjTree *tree, char *sub_fmt, char *
> +object_name,
>
> Extra space before object_name.

done

>
> 3. Is there a reason to keep format_type_detailed() in ddl_deparse.c
> instead of defining it in format_type.c where other format functions
> reside? Earlier, we were doing this deparsing as an extension, so it
> makes sense to define it locally but not sure if that is required now.
>

done, moved to format_type.c.

> 4.
> format_type_detailed()
> {
> ...
> + /*
> + * Check if it's a regular (variable length) array type. As above,
> + * fixed-length array types such as "name" shouldn't get deconstructed.
> + */
> + array_base_type = typeform->typelem;
>
> This comment gives incomplete information. I think it is better to
> say: "We switch our attention to the array element type for certain
> cases. See format_type_extended(). Then we can remove a similar
> comment later in the function.
>

Improved the comment here.

> 5.
> +
> + switch (type_oid)
> + {
> + case INTERVALOID:
> + *typename = pstrdup("INTERVAL");
> + break;
> + case TIMESTAMPTZOID:
> + if (typemod < 0)
> + *typename = pstrdup("TIMESTAMP WITH TIME ZONE"); else
> + /* otherwise, WITH TZ is added by typmod. */ *typename =
> + pstrdup("TIMESTAMP"); break; case TIMESTAMPOID:
> + *typename = pstrdup("TIMESTAMP");
> + break;
>
> In this switch case, use the type oid cases in the order of their value.
>

done

> 6.
> +static inline char *
> +get_type_storage(char storagetype)
>
> We already have a function with the name storage_name() which does
> exactly what this function is doing. Shall we expose that and use it?
>

done

> 7.
> +static ObjTree *
> +new_objtree(char *fmt)
> +{
> + ObjTree *params;
> +
> + params = palloc0(sizeof(ObjTree));
>
> Here, the variable name params appear a bit odd. Shall we change it to
> objtree or obj?
>

done

===============================

> Some more comments on 0001
> ==========================
>
> 1.
> +/*
> + * Subroutine for CREATE TABLE/CREATE DOMAIN deparsing.
> + *
> + * Given a table OID or domain OID, obtain its constraints and append
> +them to
> + * the given elements list. The updated list is returned.
> + *
> + * This works for typed tables, regular tables, and domains.
> + *
> + * Note that CONSTRAINT_FOREIGN constraints are always ignored.
> + */
> +static List *
> +obtainConstraints(List *elements, Oid relationId, Oid domainId,
> + ConstraintObjType objType)
>
> Why do we need to support DOMAIN in this patch? Isn't this only for tables?

Moved to later patch.

>
> 2.
> obtainConstraints()
> {
> ..
> + switch (constrForm->contype)
> + {
> + case CONSTRAINT_CHECK:
> + contype = "check";
> + break;
> + case CONSTRAINT_FOREIGN:
> + continue; /* not here */
> + case CONSTRAINT_PRIMARY:
> + contype = "primary key";
> + break;
> + case CONSTRAINT_UNIQUE:
> + contype = "unique";
> + break;
> + case CONSTRAINT_TRIGGER:
> + contype = "trigger";
> + break;
> + case CONSTRAINT_EXCLUSION:
> + contype = "exclusion";
> + break;
> + default:
> + elog(ERROR, "unrecognized constraint type");
>
> It looks a bit odd that except CONSTRAINT_NOTNULL all other
> constraints are handled here. I think the reason is callers themselves
> deal with not null constraints, if so, we can probably add a comment.
>

Since the CONSTRAINT_NOTNULL(9ce04b5) has been removed, I didn't add comments here.

> 3.
> obtainConstraints()
> {
> ...
> + if (constrForm->conindid &&
> + (constrForm->contype == CONSTRAINT_PRIMARY ||
> + constrForm->contype == CONSTRAINT_UNIQUE || contype ==
> + constrForm->CONSTRAINT_EXCLUSION))
> + {
> + Oid tblspc = get_rel_tablespace(constrForm->conindid);
> +
> + if (OidIsValid(tblspc))
> + append_string_object(constr,
> + "USING INDEX TABLESPACE %{tblspc}s", "tblspc",
> + get_tablespace_name(tblspc));
> ...
> }
>
> How is it guaranteed that we can get tablespace_name after getting id?
> If there is something that prevents tablespace from being removed
> between these two calls then it could be better to write a comment for
> the same.
>

Done, changed code to check if valid tablespace_name is received as
it may be concurrently dropped.

> 4. It seems RelationGetColumnDefault() assumed that the passed
> attribute always had a default because it didn't verify the return
> value of build_column_default(). Now, all but one of its callers in
> deparse_ColumnDef() check that attribute has a default value before
> calling this function. I think either we change that caller or have an
> error handling in RelationGetColumnDefault(). It might be better to
> add a comment in RelationGetColumnDefault() to reflect that callers
> ensure that the passed attribute has a default value and then have an
> assert for it as well.
>

Added a comments and assert.

> 5.
> +deparse_ColumnDef(Relation relation, List *dpcontext, bool composite,
> + ColumnDef *coldef, bool is_alter, List **exprs)
> {
> ...
> + attrTup = SearchSysCacheAttName(relid, coldef->colname); if
> + (!HeapTupleIsValid(attrTup)) elog(ERROR, "could not find cache entry
> + for column \"%s\" of relation %u",
> + coldef->colname, relid);
> + attrForm = (Form_pg_attribute) GETSTRUCT(attrTup);
> ...
> + /* IDENTITY COLUMN */
> + if (coldef->identity)
> + {
> + Oid attno = get_attnum(relid, coldef->colname);
> ...
>
> I think we don't need to perform additional syscache lookup to get
> attno as we already have that in this function and is used at other
> places.

done

>
> 6.
> +deparse_ColumnDef(Relation relation, List *dpcontext, bool composite,
> + ColumnDef *coldef, bool is_alter, List **exprs)
> {
> ...
>
> + seqrelid = getIdentitySequence(relid, attno, true); if
> + (OidIsValid(seqrelid) && coldef->identitySequence) seqrelid =
> + RangeVarGetRelid(coldef->identitySequence, NoLock, false);
> ...
>
> It may be better to add some comments to explain what exactly are we doing here.
>

Done

Best regards,
Hou zj