To make a sink plugged to a materialized view yield events only for the rows that changed in the view, you can use a combination of triggers and conditions to filter out unchanged rows. Here's a general approach:

1. Create a trigger function that checks for changes in the materialized view.
2. Use the trigger function to update a flag or timestamp column in the materialized view when a row changes.
3. Modify the sink query to filter rows based on the flag or timestamp column.

Example:

-- 1. Create a trigger function
CREATE OR REPLACE FUNCTION check_changes()
RETURNS TRIGGER AS $$
BEGIN
  NEW.changed := true; -- Set a flag to indicate a change
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

-- 2. Create a trigger on the materialized view
CREATE TRIGGER mat_view_trigger
AFTER INSERT OR UPDATE ON materialized_view
FOR EACH ROW
EXECUTE FUNCTION check_changes();

-- 3. Modify the sink query to filter out unchanged rows
SELECT *
FROM materialized_view
WHERE changed = true; -- Filter based on the flag