Fix handling of JSON columns

[dveeden]

What problem does this PR solve?

Closes #3624

What is changed and how it works?

When inserting valid JSON into a column of the json type this happens:

panic: interface conversion: interface {} is string, not json.BinaryJSON

goroutine 1228 [running]:
github.com/pingcap/ticdc/cdc/sink/codec.columnToAvroNativeData(0xc003f93bc0, 0xc000e14cb0, 0xc0027471c8, 0x2, 0xc00197a1a8, 0xd, 0x0, 0x0)
	github.com/pingcap/ticdc/cdc/sink/codec/avro.go:479 +0x19c5
github.com/pingcap/ticdc/cdc/sink/codec.rowToAvroNativeData(0xc00274e1f0, 0x2, 0x2, 0xc000e14cb0, 0x4, 0xc002747190, 0xc, 0x37)
	github.com/pingcap/ticdc/cdc/sink/codec/avro.go:274 +0xa8
github.com/pingcap/ticdc/cdc/sink/codec.avroEncode(0xc003f93bf0, 0xc0026817c0, 0x5f5736086ec0002, 0xc00274e1f0, 0x2, 0x2, 0xc000e14cb0, 0xc0044ef560, 0xc0044ef570, 0x3a071e8)
	github.com/pingcap/ticdc/cdc/sink/codec/avro.go:197 +0x191
github.com/pingcap/ticdc/cdc/sink/codec.(*AvroEventBatchEncoder).AppendRowChangedEvent(0xc002615200, 0xc001b52280, 0x0, 0x0, 0x3)
	github.com/pingcap/ticdc/cdc/sink/codec/avro.go:94 +0x830
github.com/pingcap/ticdc/cdc/sink.(*mqSink).runWorker(0xc0018fc280, 0x3a34708, 0xc0018aa1c0, 0x2, 0x0, 0x0)
	github.com/pingcap/ticdc/cdc/sink/mq.go:345 +0x3f1
github.com/pingcap/ticdc/cdc/sink.(*mqSink).run.func1(0x0, 0x0)
	github.com/pingcap/ticdc/cdc/sink/mq.go:275 +0x46
golang.org/x/sync/errgroup.(*Group).Go.func1(0xc0026151d0, 0xc002941400)
	golang.org/x/sync@v0.0.0-20210220032951-036812b2e83c/errgroup/errgroup.go:57 +0x59
created by golang.org/x/sync/errgroup.(*Group).Go
	golang.org/x/sync@v0.0.0-20210220032951-036812b2e83c/errgroup/errgroup.go:54 +0x66

The Value seems to be a string instead of json.BinaryJSON as the error indicates

(dlv) print col
*github.com/pingcap/ticdc/cdc/model.Column {
	Name: "json_payload",
	Type: 245,
	Flag: BinaryFlag|NullableFlag (65),
	Value: interface {}(string) "{\"foo\": \"bar\"}",}

Manual test

Setup a test environment:

tiup playground v5.2.2 --tiflash 0 --without-monitor=false
./bin/cdc server
curl --silent --output docker-compose.yml https://raw.githubusercontent.com/confluentinc/cp-all-in-one/6.2.0-post/cp-all-in-one-community/docker-compose.yml
sudo docker-compose up -d
./bin/cdc cli changefeed create --no-confirm --changefeed-id="simple-replication-task" --sort-engine="unified" --sink-uri="kafka://127.0.0.1:9092/cdc-test?protocol=avro&kafka-version=2.8.0" --log-level debug --opts registry="http://127.0.0.1:8081"

Create a table and insert some JSON:

CREATE TABLE `sample_table` (
  `id` bigint(20) unsigned primary key NOT NULL AUTO_INCREMENT,
  `json_payload` json DEFAULT NULL
);
INSERT INTO sample_table(json_payload) VALUES(NULL);
INSERT INTO sample_table(json_payload) VALUES(JSON_OBJECT('foo', 'bar'));
SELECT * FROM sample_table;

Start an Avro Console Consumer:

sudo docker-compose exec schema-registry /bin/sh
kafka-avro-console-consumer  --topic cdc-test --from-beginning --bootstrap-server broker:29092

Output:

{"id":"\t","json_payload":{"string":"{\"foo\": \"bar\"}"}}

Check List

Tests

• Manual test (add detailed scripts or steps below)

Related changes

• Need to cherry-pick to the release branch

Release note

The Avro sink was updated to handle JSON columns

[ti-chi-bot]

[REVIEW NOTIFICATION]

This pull request has been approved by:

• amyangfei
• ben1009

To complete the pull request process, please ask the reviewers in the list to review by filling /cc @reviewer in the comment.
After your PR has acquired the required number of LGTMs, you can assign this pull request to the committer in the list by filling /assign @committer in the comment to help you merge this pull request.

The full list of commands accepted by this bot can be found here.

Reviewer can indicate their review by submitting an approval review.
Reviewer can cancel approval by submitting a request changes review.

[codecov-commenter]

Codecov Report

Merging #3643 (7acb0b6) into master (24030f8) will increase coverage by 1.5616%.
The diff coverage is 68.7637%.

@@               Coverage Diff                @@
##             master      #3643        +/-   ##
================================================
+ Coverage   56.5356%   58.0973%   +1.5616%     
================================================
  Files           211        241        +30     
  Lines         22798      24502      +1704     
================================================
+ Hits          12889      14235      +1346     
- Misses         8598       8845       +247     
- Partials       1311       1422       +111     

[amyangfei]

/merge

[ti-chi-bot]

This pull request has been accepted and is ready to merge.

Commit hash: 7acb0b6

[ti-chi-bot]

In response to a cherrypick label: new pull request created: #3649.

[ti-chi-bot]

In response to a cherrypick label: new pull request created: #3650.

[ti-chi-bot]

In response to a cherrypick label: new pull request created: #3651.

[ti-chi-bot]

In response to a cherrypick label: new pull request created: #3652.

[ti-chi-bot]

In response to a cherrypick label: new pull request created: #3653.

[keweishang]

Thanks @dveeden for fixing this. Do you know which TiCDC release version would contain this fix?

[dveeden]

Thanks @dveeden for fixing this. Do you know which TiCDC release version would contain this fix?

That's not sure yet, but probably v5.3.1 and v5.2.4. You can follow the cherrypick requests that @ti-chi-bot posted to see when they are merged in the release-5.3 and release-5.2 branches.