value := ev.Val.Value
	if !value.IsPresent() {
		value = ev.Val.PrevValue
	}

	value := ev.Val.Value
	if !value.IsPresent() {
		value = ev.Val.PrevValue
	}

what is this case?

Why separate timestamp when it's already a field in the event? Function seems better.

Data structure question: we only care about the ordering when we go to deal with a frontier. It feels like it would be simpler to just sort.Sort at Flush point and append during execution? That also allows us to end up freeing up the buffer after a burst and to avoid allocating a new one on flush. The flip side is if there are many flushes before a lot of data, you'll sort for each flush, but I really don't care about that case. The bigger reason to do it is that I think it's simpler. Consider:

type rangefeedEvents []*roachpb.RangefeedEvent

func (events rangefeedEvents) Len() int { return len(events) }
func (events rangefeedEvents) Swap(i, j int) { events[i], events[j] = events[j], events[i] }
func (events rangefeedEvents) Less(i, j int) bool { return eventTS(events[i].Timestamp).Less(eventTS(events[j])) }

func eventTS(ev *roachpb.RangefeedEvent) hlc.Timestamp { /* ... */ }

// Add adds the given rangefeed entry to the buffer.
func (b *Buffer) Add(ctx context.Context, ev *roachpb.RangeFeedEvent) error {
        // ...

	b.mu.Lock()
	defer b.mu.Unlock()

	if value.Timestamp.LessEq(b.mu.frontier) {
		// If the rangefeed entry is at a timestamp less than or equal to our
		// last known checkpoint, we don't need to record it.
		return nil
	}

	if len(b.mu.events) > b.limit {
		return ErrBufferLimitExceeded
	}
        b.mu.events = append(b.mu.events, ev)
	return nil
}


func (b *Buffer) Flush(
	ctx context.Context, frontier hlc.Timestamp,
) (events []*roachpb.RangeFeedEvent) {
	b.mu.Lock()
	defer b.mu.Unlock()

	if frontier.Less(b.mu.frontier) {
		log.Fatalf(ctx, "frontier timestamp regressed: saw %s, previously %s", frontier, b.mu.frontier)
	}
        sort.Sort(rangefeedEvents(b.mu.events))
        idx := sort.Search(len(b.mu.events), func(i int) bool {
            return !eventTS(b.mu.events).Less(frontier)
        })
        events = b.mu.events[:idx]
        b.mu.events = b.mu.events[idx:]
	b.mu.frontier = frontier
	return events
}

Just a way to extract a timestamp for a rangefeed value deletion event.

Good idea, done for both. One think I'm not sure about is this bit:

 events = b.mu.events[:idx]
 b.mu.events = b.mu.events[idx:]

Are we creating an ever-growing slice here, discarding only the prefix when flushing out events lesseq than the frontier timestamp?

🤔 the delete should have a timestamp, no?

When the slice re-allocs, it'll be a new backing array.

Yea, isn't it ev.Val.PrevValue.Timestamp? That's what this code block was doing.

I don't think so. PrevValue will be empty unless you explicitly request it. It will also have the timestamp of the value that preceded the deletion.

Can you write a test that exercises the deletion path?

@arulajmani and I were talking yesterday about whether this structure could be used in the SQLWatcher as well in place of the current eventBuffer he introduced over there. That use would be a little different in that it is zipping together two rangefeeds, but it has a lot of the same requirements. We discussed whether this buffer should be made slightly more generic to accommodate other uses cases, essentially by replacing *roachpb.RangeFeedEvent with a type Event interface { Timestamp() hlc.Timestamp }.

We wanted to get your thoughts on that.

In TestRangefeedValueTimestamps I'm specifying rangefeed.WithDiff(), I'm seeing the previous values just fine -- the timestamps however are empty.

cockroach/pkg/kv/kvserver/rangefeed/processor.go

Lines 632 to 644 in 0417041

	var prevVal roachpb.Value
	if prevValue != nil {
	prevVal.RawBytes = prevValue
	}
	var event roachpb.RangeFeedEvent
	event.MustSetValue(&roachpb.RangeFeedValue{
	Key: key,
	Value: roachpb.Value{
	RawBytes: value,
	Timestamp: timestamp,
	},
	PrevValue: prevVal,
	})

Doesn't look like we set a prev-value timestamp above, unlike during the initial scan:

cockroach/pkg/kv/kvclient/rangefeed/rangefeed.go

Lines 258 to 278 in fc803ee

	func (f *RangeFeed) maybeRunInitialScan(
	ctx context.Context, n *log.EveryN, r *retry.Retry,
	) (canceled bool) {
	if !f.withInitialScan {
	return false // canceled
	}
	scan := func(kv roachpb.KeyValue) {
	v := roachpb.RangeFeedValue{
	Key: kv.Key,
	Value: kv.Value,
	}
	// Mark the data as occurring at the initial timestamp, which is the
	// timestamp at which it was read.
	v.Value.Timestamp = f.initialTimestamp
	// Supply the value from the scan as also the previous value to avoid
	// indicating that the value was previously deleted.
	if f.withDiff {
	v.PrevValue = v.Value
	}

Ah, guess we didn't expose the PrevValue timestamp. That was probably smart from an API contract perspective.

cockroach/pkg/kv/kvserver/rangefeed/processor.go

Lines 625 to 627 in efb1a95

	func (p *Processor) publishValue(
	ctx context.Context, key roachpb.Key, timestamp hlc.Timestamp, value, prevValue []byte,
	) {

More commentary on

cockroach/pkg/roachpb/api.proto

Lines 2384 to 2388 in efb1a95

	// prev_value is only populated if both:
	// 1. with_diff was passed in the corresponding RangeFeedRequest.
	// 2. the key-value was present and not a deletion tombstone before
	// this event.
	Value prev_value = 3 [(gogoproto.nullable) = false];

probably wouldn't hurt.

I don't mind the idea of hiding away the even through an interface. I imagine then in the SQLWatcher the events we'd store would be the decoded descriptor IDs (decoded as per the right table's schema), which feels cleaner than juggling two instances of this buffer; made that change.

There's little in this library that's now "rangefeed" specific, though I expect it to mostly be used in conjunction with it. Suggestions for where it should be placed or what it should be named?

Probably also worth reconciling those behaviors. That certainly was not intentional.

Another thing that has come up in offline discussions with both @irfansharif and @nvanbenschoten, and I'd like to get more thoughts on here, is how we feel about generalizing this even further so that this can track multiple frontier timestamps. Specifically for the SQLWatcher, this would allow us to push the "combined frontier timestamp" into the buffer (as opposed to tracking it in a layer above). We could extend this interface a bit to make this work.

In the EventBuffer from the SQLWatcher PR we were de-duplicating events from the buffer when flushing. Is that something we want to do here as well?

I'm 👎 on it, especially if you can simply add as buffer events as decoded IDs. There are no other instances (to my knowledge) where we're maintaining multiple (n=2) rangefeeds to keep track of multiple frontier timestamps, so that sounds overgeneralized. What do you imagine that API would look like? Would it give us much over something like the following:

type ... struct {
   rangefeedbuffer.Buffer // accumulates descpb.IDs
   zonesFrontierTS, descriptorFrontierTS hlc.Timestamp
}

+1 to what Irfan said.

I think that this is going to heap allocate because you have the interface implemented on a value. If you implement the interface on *events and then pass sort.Sort(&b.mu.events), you can avoid that.

TIL, you're right:

$ go build -gcflags '-m' github.com/cockroachdb/cockroach/pkg/kv/kvclient/rangefeed/rangefeedbuffer
...
pkg/kv/kvclient/rangefeed/rangefeedbuffer/buffer.go:86:16: b.mu.events escapes to heap

Done.