Skip to content

Commit cab9aa8

Browse files
committed
changefeedccl: ensure that spans planned from cdc queries have end keys
When a cdc query results in a plan that scans a single key, the end key of the span is unset. This causes the feed to fail on startup. We now set span.EndKey = span.Key.Next(). Fixes: #143101 Release note (bug fix): Fixed a bug that caused feeds to fail on startup when scanning a single key.
1 parent cb94136 commit cab9aa8

File tree

2 files changed

+55
-1
lines changed

2 files changed

+55
-1
lines changed

pkg/ccl/changefeedccl/cdceval/plan.go

+9-1
Original file line numberDiff line numberDiff line change
@@ -132,7 +132,15 @@ func SpansForExpression(
132132
return nil, withErrorHint(err, d.FamilyName, d.HasOtherFamilies)
133133
}
134134

135-
return plan.Spans, nil
135+
spans := plan.Spans
136+
// Make sure any single-key spans are expanded to have end keys.
137+
for i := range spans {
138+
if len(spans[i].EndKey) == 0 {
139+
spans[i].EndKey = spans[i].Key.Clone().Next()
140+
}
141+
}
142+
143+
return spans, nil
136144
}
137145

138146
// withErrorHint wraps error with error hints.

pkg/ccl/changefeedccl/changefeed_test.go

+46
Original file line numberDiff line numberDiff line change
@@ -10794,6 +10794,52 @@ func TestChangefeedProtectedTimestampUpdate(t *testing.T) {
1079410794
cdcTest(t, testFn, feedTestForceSink("kafka"), withTxnRetries)
1079510795
}
1079610796

10797+
func TestCDCQuerySelectSingleRow(t *testing.T) {
10798+
defer leaktest.AfterTest(t)()
10799+
defer log.Scope(t).Close(t)
10800+
10801+
errCh := make(chan error, 1)
10802+
knobsFn := func(knobs *base.TestingKnobs) {
10803+
if knobs.DistSQL == nil {
10804+
knobs.DistSQL = &execinfra.TestingKnobs{}
10805+
}
10806+
if knobs.DistSQL.(*execinfra.TestingKnobs).Changefeed == nil {
10807+
knobs.DistSQL.(*execinfra.TestingKnobs).Changefeed = &TestingKnobs{}
10808+
}
10809+
cfKnobs := knobs.DistSQL.(*execinfra.TestingKnobs).Changefeed.(*TestingKnobs)
10810+
cfKnobs.HandleDistChangefeedError = func(err error) error {
10811+
errCh <- err
10812+
return err
10813+
}
10814+
}
10815+
10816+
testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
10817+
db := sqlutils.MakeSQLRunner(s.DB)
10818+
db.Exec(t, `CREATE TABLE foo (key INT PRIMARY KEY);`)
10819+
db.Exec(t, `INSERT INTO foo VALUES (1), (2), (3);`)
10820+
10821+
execed := make(chan struct{})
10822+
go func() {
10823+
defer close(execed)
10824+
// initial_scan='only' is not required, but it makes testing this easier.
10825+
foo := feed(t, f, `CREATE CHANGEFEED WITH initial_scan='only' AS SELECT * FROM foo WHERE key = 1`)
10826+
// db.Exec(t, "CREATE CHANGEFEED WITH initial_scan='only' AS SELECT * FROM foo WHERE key = 1")
10827+
defer closeFeed(t, foo)
10828+
assertPayloads(t, foo, []string{`foo: [1]->{"key": 1}`})
10829+
}()
10830+
10831+
select {
10832+
case err := <-errCh:
10833+
t.Fatalf("unexpected error: %v", err)
10834+
case <-time.After(10 * time.Second):
10835+
t.Fatal("timed out")
10836+
case <-execed:
10837+
}
10838+
}
10839+
// This fails with all sinks, but it's easier to detect failures with a sinkless feed.
10840+
cdcTest(t, testFn, feedTestForceSink("sinkless"), withKnobsFn(knobsFn))
10841+
}
10842+
1079710843
func assertReasonableMVCCTimestamp(t *testing.T, ts string) {
1079810844
epochNanos := parseTimeToHLC(t, ts).WallTime
1079910845
now := timeutil.Now()

0 commit comments

Comments
 (0)