Horton

Re: row filtering for logical replication

Andres Freund <andres@anarazel.de>

View thread
From: Andres Freund <andres@anarazel.de>
To: Euler Taveira <euler@eulerto.com>
Cc: Amit Kapila <amit.kapila16@gmail.com>, Rahila Syed <rahilasyed90@gmail.com>, Peter Eisentraut <peter.eisentraut@enterprisedb.com>, Önder Kalacı <onderkalaci@gmail.com>, japin <japinli@hotmail.com>, Michael Paquier <michael@paquier.xyz>, David Steele <david@pgmasters.net>, Craig Ringer <craig@2ndquadrant.com>, Tomas Vondra <tomas.vondra@2ndquadrant.com>, Amit Langote <amitlangote09@gmail.com>, PostgreSQL Hackers <pgsql-hackers@lists.postgresql.org>
Date: 2021-03-31T19:17:10Z
Lists: pgsql-hackers

Commits

Same data as JSON: GET /api/v1/messages/:b64id/commits the thread's linked commits as JSON, with link sources. API reference →

  1. Release cache tuple when no longer needed

  2. Add some additional tests for row filters in logical replication.

  3. Fix one of the tests introduced in commit 52e4f0cd47.

  4. Allow specifying row filters for logical replication of tables.

  5. Move scanint8() to numutils.c

  6. Replace Test::More plans with done_testing

  7. Reduce relcache access in WAL sender streaming logical changes

  8. Small cleanups related to PUBLICATION framework code

  9. Add a view to show the stats of subscription workers.

  10. Allow publishing the tables of schema.

  11. Doc: improve documentation of CREATE/ALTER SUBSCRIPTION.

  12. Add PublicationTable and PublicationRelInfo structs

  13. Remove unused argument "txn" in maybe_send_schema().

  14. Add prepare API support for streaming transactions in logical replication.

  15. Unify PostgresNode's new() and get_new_node() methods

  16. Use l*_node() family of functions where appropriate

  17. Add support for prepared transactions to built-in logical replication.

  18. Restore the portal-level snapshot after procedure COMMIT/ROLLBACK.

  19. Rename a parse node to be more general

  20. Remove unused column atttypmod from initial tablesync query

  21. SEARCH and CYCLE clauses

Hi,

As far as I can tell you have not *AT ALL* addressed that it is *NOT
SAFE* to evaluate arbitrary expressions from within an output
plugin. Despite that having been brought up multiple times.


> +static ExprState *
> +pgoutput_row_filter_prepare_expr(Node *rfnode, EState *estate)
> +{
> +	ExprState  *exprstate;
> +	Oid			exprtype;
> +	Expr	   *expr;
> +
> +	/* Prepare expression for execution */
> +	exprtype = exprType(rfnode);
> +	expr = (Expr *) coerce_to_target_type(NULL, rfnode, exprtype, BOOLOID, -1, COERCION_ASSIGNMENT, COERCE_IMPLICIT_CAST, -1);
> +
> +	if (expr == NULL)
> +		ereport(ERROR,
> +				(errcode(ERRCODE_CANNOT_COERCE),
> +				 errmsg("row filter returns type %s that cannot be coerced to the expected type %s",
> +						format_type_be(exprtype),
> +						format_type_be(BOOLOID)),
> +				 errhint("You will need to rewrite the row filter.")));
> +
> +	exprstate = ExecPrepareExpr(expr, estate);
> +
> +	return exprstate;
> +}
> +
> +/*
> + * Evaluates row filter.
> + *
> + * If the row filter evaluates to NULL, it is taken as false i.e. the change
> + * isn't replicated.
> + */
> +static inline bool
> +pgoutput_row_filter_exec_expr(ExprState *state, ExprContext *econtext)
> +{
> +	Datum		ret;
> +	bool		isnull;
> +
> +	Assert(state != NULL);
> +
> +	ret = ExecEvalExprSwitchContext(state, econtext, &isnull);
> +
> +	elog(DEBUG3, "row filter evaluates to %s (isnull: %s)",
> +		 DatumGetBool(ret) ? "true" : "false",
> +		 isnull ? "true" : "false");
> +
> +	if (isnull)
> +		return false;
> +
> +	return DatumGetBool(ret);
> +}

> +/*
> + * Change is checked against the row filter, if any.
> + *
> + * If it returns true, the change is replicated, otherwise, it is not.
> + */
> +static bool
> +pgoutput_row_filter(Relation relation, HeapTuple oldtuple, HeapTuple newtuple, List *rowfilter)
> +{
> +	TupleDesc	tupdesc;
> +	EState	   *estate;
> +	ExprContext *ecxt;
> +	MemoryContext oldcxt;
> +	ListCell   *lc;
> +	bool		result = true;
> +
> +	/* Bail out if there is no row filter */
> +	if (rowfilter == NIL)
> +		return true;
> +
> +	elog(DEBUG3, "table \"%s.%s\" has row filter",
> +		 get_namespace_name(get_rel_namespace(RelationGetRelid(relation))),
> +		 get_rel_name(relation->rd_id));
> +
> +	tupdesc = RelationGetDescr(relation);
> +
> +	estate = create_estate_for_relation(relation);
> +
> +	/* Prepare context per tuple */
> +	ecxt = GetPerTupleExprContext(estate);
> +	oldcxt = MemoryContextSwitchTo(estate->es_query_cxt);
> +	ecxt->ecxt_scantuple = ExecInitExtraTupleSlot(estate, tupdesc, &TTSOpsHeapTuple);
> +	MemoryContextSwitchTo(oldcxt);
> +
> +	ExecStoreHeapTuple(newtuple ? newtuple : oldtuple, ecxt->ecxt_scantuple, false);
> +	/*
> +	 * If the subscription has multiple publications and the same table has a
> +	 * different row filter in these publications, all row filters must be
> +	 * matched in order to replicate this change.
> +	 */
> +	foreach(lc, rowfilter)
> +	{
> +		Node	   *rfnode = (Node *) lfirst(lc);
> +		ExprState  *exprstate;
> +
> +		/* Prepare for expression execution */
> +		exprstate = pgoutput_row_filter_prepare_expr(rfnode, estate);
> +
> +		/* Evaluates row filter */
> +		result = pgoutput_row_filter_exec_expr(exprstate, ecxt);

Also, this still seems like an *extremely* expensive thing to do for
each tuple. It'll often be *vastly* faster to just send the data than to
the other side.

This just cannot be done once per tuple. It has to be cached.

I don't see how these issues can be addressed in the next 7 days,
therefore I think this unfortunately needs to be marked as returned with
feedback.

Greetings,

Andres Freund