changefeeccl: Projections and Filters in CDC. #82562

miretskiy · 2022-06-07T23:13:10Z

Add a variant of CHANGEFEED statement that allows specification
of predicates and projections.

CREATE CHANGEFEED [INTO 'sink'] [WITH opt=val, ...]
AS SELECT .... FROM t WHERE ...

This changefeed variant can target at most 1 table (and 1 column
family) at a time. The expressions used as the projections and
filters can be almost any supported expression with some restrictions:

Volatile functions not allowed.
Sub-selects not allowed.
Aggregate and window functions (i.e. functions operating over many
rows) not allowed.
Some stable functions, notably functions which return MVCC
timestamp, are overridden to return MVCC timestamp of the event.

In addition, some CDC specific functions are provided:

cdc_is_delete: returns true if the event is a deletion event.
cdc_prev: returns JSON representation of the previous row state.
cdc_updated_timestamp: returns event update timestamp (usually MVCC
timestamp, but can be different if e.g. undergoing schema changes)
Additional CDC specific functions will be added in the follow on PRs.

Few examples:

Emit all but the deletion events:

CREATE CHANGEFEED INTO 'kafka://'
AS SELECT * FROM table
WHERE NOT cdc_is_delete()

Emit all events that modified important_col column:

CREATE CHANGEFEED INTO 'kafka://' WITH diff
AS SELECT *, cdc_prev() AS previous
FROM important_table
WHERE important_col != cdc_prev()->'important_col'

Emit few colums, as well as computed expresions:

CREATE CHANGEFEED INTO 'kafka://' WITH diff
AS SELECT warehouseID, (totalItems - orderedItems) as itemsAvailable
FROM warehouse
WHERE region='US/east';

When filter expression is specified, changefeed will now consult
optimizer so that the set of spans scanned by changefeed can be
restricted based on the predicate.

For example, given the following table and a changefeed:

CREATE TABLE warehouse (
  region STRING,
  warehouseID int,
  ....
  PRIMARY KEY (region, warehouseID)
);

CREATE CHANGEFEED INTO 'kafka://' WITH diff
AS SELECT *
FROM warehouse
WHERE region='US/east';

The create changefeed will only scan table spans that contain US/east
region (and ignore all other table spans).

For foundational work, see:

Addresses:

Release Notes (enterprise):
CHANGEFEED statement now supports general expressions -- predicates and projections.
Projections allow customers to emit only the data that they care about,
including computed columns, while predicates (i.e. filters) allow them
to restrict the data that's emitted only to those events that match the
filter.

CREATE CHANGEFEED INTO 'kafka://' AS SELECT * FROM t WHERE NOT cdc_is_delete()

cockroach-teamcity · 2022-06-07T23:13:19Z

This change is

miretskiy · 2022-06-07T23:14:50Z

Note to the reviewers: I plan on expanding changefeed_test with more end2end tests.
In the meantime, want to start the review and see what TC has to say.

rytaft

Cool! I left a few nits. I think I reviewed the optimizer-related code, but if you could point me to specific files you want me to look at more closely that would be helpful.

Reviewed 8 of 8 files at r1, 34 of 34 files at r2, all commit messages.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @HonoreDB, @miretskiy, and @stevendanna)

-- commits line 92 at r2:
now supports?

pkg/ccl/changefeedccl/cdceval/constraint.go line 25 at r2 (raw file):

// ConstrainPrimaryIndexSpanByFilter attempts to constrain table primary index span via specified
// filter.  Returns constrained span, and a possibly empty remaining filter that needs to be evaluated.

nit: comment should be <= 80 columns wide

pkg/ccl/changefeedccl/cdceval/expr_eval.go line 48 at r2 (raw file):

}

// NewEvaluatorForExpressions returns evaluator configured to process specified projection and filter expressions.

nit: wrap comment lines

pkg/ccl/changefeedccl/cdceval/validation.go line 85 at r2 (raw file):

		return nil, pgerror.Newf(pgcode.InvalidParameterValue,
			"projections and filter cannot be used when running against multifamily table (table has %d families)",
			desc.NumFamilies())

I think you can relax this restriction in the future if the filters and projections only use columns from the target family

pkg/ccl/changefeedccl/schemafeed/schema_feed.go line 534 at r2 (raw file):

func (tf *schemaFeed) validatePredicates(ctx context.Context, desc catalog.TableDescriptor) error {
	if len(tf.projection) == 0 {

did you mean predicate here?

pkg/clusterversion/cockroach_versions.go line 368 at r2 (raw file):

	// keys at the Pebble layer.
	EnablePebbleFormatVersionRangeKeys
	// EnablePredicateChangefeed indicates that changefeeds support predicates and projections.

nit: wrap comment

pkg/clusterversion/cockroach_versions.go line 369 at r2 (raw file):

	EnablePebbleFormatVersionRangeKeys
	// EnablePredicateChangefeed indicates that changefeeds support predicates and projections.
	EnablePredicateChangefeed

nit: maybe call this EnablePredicateProjectionChangefeed?

pkg/sql/execinfrapb/processors_changefeeds.proto line 61 at r2 (raw file):

  // Projection is the "select clause" for predicate changefeed.
  optional Expression projection = 6 [(gogoproto.nullable) = false];

should this be repeated instead of optional?

pkg/sql/opt/exec/execbuilder/builder.go line 132 at r1 (raw file):

	ContainsMutation bool

	// wrapFunction returns resolvable function referenced for function with

nit: referenced -> reference

pkg/ccl/changefeedccl/alter_changefeed_test.go line 1285 at r2 (raw file):

	// and removed 'primary_key_filter'.  Since predicates and projections are no longer
	// a "string" option, alter statement implementation (and grammar) needs to be updated, and
	// this test modified and re-enabled.

nit: wrap comment

miretskiy

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @HonoreDB, @rytaft, and @stevendanna)

-- commits line 92 at r2: