Looking great so far. I'm pleasantly surprised how clean a lot of this is.

I'm a bit confused about why poller is handling this complexity; I thought we'd discussed offline that buffer would be doing it. This mostly matters once we do #28669, which is going to open up the possibility of shutting down RangeFeeds when the sink is slow. This seems similar to me. A backfill scan would also cause the buffer to back up and whatever mechanism is monitoring it would be able to shut down the RangeFeed.

On the other hand, it does seem like the timestamp boundaries are simpler this way since you can adjust the nextHighWater (though that advantage goes away when we switch to RangeFeed).

Update: After thinking about this overnight, I'm more convinced that poller is the right place to put this right now. Once the buffer actually buffers (which means we don't need to do this adjust the high-water dance, can avoid throwing away data, and don't need to restart rangefeeds for schemachanges) we should revisit.

Reviewable status: complete! 0 of 0 LGTMs obtained

pkg/ccl/changefeedccl/buffer.go, line 25 at r1 (raw file):

	// If the value's timestamp is greater than the minimum, the schema at the
	// value timestamp should be used instead.
	minSchemaTimestamp hlc.Timestamp

why is this a greater/lesser relationship? can we have it be an override when not the zero value or am i missing something?

pkg/ccl/changefeedccl/changefeed_test.go, line 433 at r1 (raw file):

// Test schema changes that require a backfill when the backfill option is
// allowed.
func TestChangefeedSchemaChangeAllowBackfill(t *testing.T) {

love the tests! 👏

pkg/ccl/changefeedccl/changefeed_test.go, line 499 at r1 (raw file):

			})
		})
	}

can we add a test that checks scanBoundaries queueing up many things (maybe by running many schema changes)

pkg/ccl/changefeedccl/poller.go, line 62 at r1 (raw file):

		// the initial scan of the table when starting a new Changefeed, and when
		// a backfilling schema change is marked as completed.
		scanBoundaries []hlc.Timestamp

does this have to stay sorted? seems like the code below is assuming that

pkg/ccl/changefeedccl/poller.go, line 578 at r1 (raw file):

		}
		if lastVersion.HasColumnBackfillMutation() && !desc.HasColumnBackfillMutation() {
			p.mu.scanBoundaries = append(p.mu.scanBoundaries, desc.GetModificationTime())

can you add a check that high-water is less than the boundary you're about to queue up? i think this means moving it inside your mu struct

pkg/ccl/changefeedccl/table_history_test.go, line 49 at r1 (raw file):

	// advance
	require.NoError(t, m.IngestDescriptors(context.TODO(), ts(0), ts(1), nil))

nit: these are all noisy. make a ctx variable above. also context.Background() is appropriate in tests

In terms of putting this in the poller: I think we're in agreement. In the poll/export world, I think this is obviously the best solution, as the tableHistory is polled on the same interval as the data itself, and we have an explicit maximum timestamp available when calling Export. I don't think it could get much cleaner.

In terms of rangefeeds, I think its going to be trickier than it appears at first glance to buffer their output and multiplexing it with a scan, but it is possible. I'm certain it can be done, but the new component stack at that point might not resemble the current buffer/poller combination, it might be a more considerable refactoring.

Reviewable status: complete! 0 of 0 LGTMs obtained

pkg/ccl/changefeedccl/buffer.go, line 25 at r1 (raw file):

I was thinking this might come up if we had a situation where a scan was returning keys that were created after the scan boundary, but that never happens anymore; i've changed it to a simple override, updated the field name and comment appropriately.

pkg/ccl/changefeedccl/changefeed_test.go, line 433 at r1 (raw file):

Thanks! But really you wrote these a while ago, i just reorganized them a bit.

pkg/ccl/changefeedccl/changefeed_test.go, line 499 at r1 (raw file):

That's a good idea, but i'm going to have to construct a test that's a bit different than the others (I think i'll have to use a hook). I'm going ahead and publishing the rest of the changes you requested, I'll get this one out soon.

pkg/ccl/changefeedccl/poller.go, line 62 at r1 (raw file):

Yes, it gets sorted in the only place it is updated, but I will add that information to the comment.

pkg/ccl/changefeedccl/poller.go, line 578 at r1 (raw file):

Done.

pkg/ccl/changefeedccl/table_history_test.go, line 49 at r1 (raw file):

Done.

Reviewable status: complete! 1 of 0 LGTMs obtained

pkg/ccl/changefeedccl/buffer.go, line 23 at r2 (raw file):

	resolved *jobspb.ResolvedSpan
	// Timestamp of the schema that should be used to read this KV.
	// If unset (default-valued), the value's timestamp will be used instead.

nit: I feel like I usually see "zero-valued" for this in our codebase

pkg/ccl/changefeedccl/changefeed.go, line 115 at r2 (raw file):

			r.row.deleted = rf.RowIsDeleted()
			r.row.timestamp = kv.Value.Timestamp

i think we also have to use schemaTimestamp here if it's set. otherwise, you'll get two copies of a row with the same timestamp but with different data, which is super confusing. especially in the presence of job restarts, how would the user know which one to keep? intuitively, it makes some sense to me that the backfill would be emitted with the descriptor timestamp. (oh, but we'll have to make sure we use the kv timestamps for the initial scan. that will have to be special cased)

also, this definitely deserves a test. maybe something based on TestChangefeedTimestamps

pkg/ccl/changefeedccl/changefeed_test.go, line 433 at r1 (raw file):

I liked the way you reorganized them!

pkg/ccl/changefeedccl/changefeed_test.go, line 469 at r2 (raw file):

			})
			sqlDB.Exec(t, `ALTER TABLE add_col_comp ADD COLUMN c INT AS (a + 10) STORED`)
			sqlDB.Exec(t, `INSERT INTO add_col_comp (a) VALUES (3)`)

should we look for this in the output or maybe remove it?

pkg/ccl/changefeedccl/poller.go, line 146 at r2 (raw file):

		// Determine if we are at a scanBoundary, and trigger a full scan if needed.
		isFullScan := false

is isFullScan redundant with lastHighwater == nextHighwater?

pkg/ccl/changefeedccl/poller.go, line 165 at r2 (raw file):

		p.mu.Unlock()

		if isFullScan {

is this backward?

pkg/ccl/changefeedccl/poller.go, line 227 at r2 (raw file):

		if performScan {
			if err := p.exportSpansParallel(
				ctx, spans, lastHighwater, lastHighwater, true, /* fullScan */

it's a little strange that this pops p.mu.scanBoundaries and then uses lastHighwater. i understand that they should match, but maybe there's some way we can make it more explicit

pkg/ccl/changefeedccl/poller.go, line 272 at r2 (raw file):

					switch t := e.GetValue().(type) {
					case *roachpb.RangeFeedValue:
						if err := p.tableHist.WaitForTS(ctx, t.Value.Timestamp); err != nil {

oof, this is not new in your PR, but this is making me notice that we backpressure raft on this. no way that cluster setting is even close to usable until we make buffer actually buffer and move all this stuff later in the changefeed flow

can you stick a TODO in here about it and reference #28669. also mention that we currently have to tear down and then set up all rangefeeds every time there is a backfill

pkg/ccl/changefeedccl/poller.go, line 328 at r2 (raw file):

}

func (p *poller) getSpansToProcess(ctx context.Context) ([]roachpb.Span, error) {

nit: can we make this a function that takes spans and db instead of a method on poller?

pkg/ccl/changefeedccl/poller.go, line 419 at r2 (raw file):

			// When outputting a full scan, we want to use the schema at the scan
			// timestamp, not the schema at the value timestamp.
			var minSchemaTimetamp hlc.Timestamp

s/minSchemaTimetamp/schemaTimestamp/ (remove "min" and correct typo in "timestamp")

pkg/ccl/changefeedccl/poller.go, line 580 at r2 (raw file):

}

func (p *poller) validateTables(ctx context.Context, desc *sqlbase.TableDescriptor) error {

nit: odd that the name is plural but it takes one table

pkg/ccl/changefeedccl/poller.go, line 606 at r2 (raw file):

			// To avoid race conditions with the lease manager, at this point we force
			// the manager to acquire the freshest descriptor of this table from the
			// store.

And the lease manager always returns the newest descriptor it has that matches the requested timestamp? Huh. TIL

Reviewable status: complete! 1 of 0 LGTMs obtained

pkg/ccl/changefeedccl/buffer.go, line 23 at r2 (raw file):

Done.

pkg/ccl/changefeedccl/changefeed.go, line 115 at r2 (raw file):

Why do we need to use the KV timestamps for the initial scan? It seems to me that the original timestamps don't matter in that case either; in fact, we want to use the current timestamp, as we need to pick up any backfilled schema changes that have occurred since the last modification of a row (in fact, I think this means our initial scans are currently wrong).

I've added a TODO here because this breaks a lot of the assumptions of our tests in subtle ways, it needs its own dedicated PR. Is that acceptable, or does it need to be swapped in the scope of this change?

pkg/ccl/changefeedccl/changefeed_test.go, line 499 at r1 (raw file):

I've added this test, it seems pretty good.

pkg/ccl/changefeedccl/changefeed_test.go, line 469 at r2 (raw file):

Just removed it.

pkg/ccl/changefeedccl/poller.go, line 146 at r2 (raw file):

Technically no, its based on detecting a scan boundary, after which nextHighwater is set to lastHightwater. However, it would be impossible to get into that situation otherwise.

This value gets passed to other functions that have different behavior for scans, I feel that its easier to read when its an explicit boolean like this.

pkg/ccl/changefeedccl/poller.go, line 165 at r2 (raw file):

Yes it is! Fixed, and good catch.

pkg/ccl/changefeedccl/poller.go, line 227 at r2 (raw file):

Instead of using the "perfomScan" boolean, I now use a "scanTime" hlc.Timestamp which gets set to the scan boundary if it exists.

pkg/ccl/changefeedccl/poller.go, line 272 at r2 (raw file):

TODOs added.

pkg/ccl/changefeedccl/poller.go, line 328 at r2 (raw file):

Done.

pkg/ccl/changefeedccl/poller.go, line 419 at r2 (raw file):

Done.

Reviewable status: complete! 1 of 0 LGTMs obtained

pkg/ccl/changefeedccl/poller.go, line 580 at r2 (raw file):

Done.

pkg/ccl/changefeedccl/poller.go, line 606 at r2 (raw file):

Yes indeed.

sorry to hold up the pr, but i think its worth hashing out the behavior of timestamps in scans first

Reviewable status: complete! 1 of 0 LGTMs obtained

pkg/ccl/changefeedccl/changefeed.go, line 115 at r2 (raw file):

Let's slack/hangout and figure this out today. I have reservations about landing this PR as-is in case the followup ends up not making 2.1.0 for some reason

pkg/ccl/changefeedccl/changefeed_test.go, line 499 at r1 (raw file):

yeah, this turned out great. nice use of the hook and waitgroup

pkg/ccl/changefeedccl/poller.go, line 146 at r2 (raw file):

👍

pkg/ccl/changefeedccl/poller.go, line 227 at r2 (raw file):

much better!

pkg/ccl/changefeedccl/poller.go, line 606 at r2 (raw file):

let's stick that in the comment here, too

@danhhz RFAL, I have made the emitted timestamp change and adjusted tests as per our offline discussion.

Reviewable status: complete! 1 of 0 LGTMs obtained

pkg/ccl/changefeedccl/changefeed.go, line 115 at r2 (raw file):

Okay, as discussed in our slack, we have made this change because we expect its behavior is correct (and that the previous behavior of initial scans might have been incorrect).

I have modified two tests accordingly: TestValidators has had the "fingerprintValidator" removed until it can be repaired, that is tracked by issue #31110. TestChangefeedTimestamps has been modified to account for the new initial scan timestamps.

pkg/ccl/changefeedccl/helpers_test.go, line 532 at r3 (raw file):

pkg/ccl/changefeedccl/poller.go, line 606 at r2 (raw file):

Done.

Reviewable status: complete! 1 of 0 LGTMs obtained

pkg/ccl/changefeedccl/validations_test.go, line 73 at r4 (raw file):

			var numResolved, rowsSinceResolved int

			// TODO(mrtracy): Re-enable Fingerprint validator.

normally, i'd ask that this have the pr that disabled it and some context so you don't have to git blame to dig up the context. but we're planning on fixing this RealSoonNow (tm), so don't eat a teamcity cycle for it

Reviewable status: complete! 1 of 0 LGTMs obtained

pkg/ccl/changefeedccl/validations_test.go, line 73 at r4 (raw file):

I ate the roundtrip anyway, done.