upsert-unordered-key.td 2.5 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788
  1. # Copyright Materialize, Inc. and contributors. All rights reserved.
  2. #
  3. # Use of this software is governed by the Business Source License
  4. # included in the LICENSE file at the root of this repository.
  5. #
  6. # As of the Change Date specified in that file, in accordance with
  7. # the Business Source License, use of this software will be governed
  8. # by the Apache License, Version 2.0.
  9. $ set-arg-default single-replica-cluster=quickstart
  10. # Must be a subset of the keys in the rows AND
  11. # in a different order than the value.
  12. $ set keyschema={
  13. "type": "record",
  14. "name": "Key",
  15. "fields": [
  16. {"name": "b", "type": "string"},
  17. {"name": "a", "type": "long"}
  18. ]
  19. }
  20. $ set schema={
  21. "type" : "record",
  22. "name" : "envelope",
  23. "fields" : [
  24. {
  25. "name": "before",
  26. "type": [
  27. {
  28. "name": "row",
  29. "type": "record",
  30. "fields": [
  31. {
  32. "name": "a",
  33. "type": "long"
  34. },
  35. {
  36. "name": "data",
  37. "type": "string"
  38. },
  39. {
  40. "name": "b",
  41. "type": "string"
  42. }]
  43. },
  44. "null"
  45. ]
  46. },
  47. {
  48. "name": "after",
  49. "type": ["row", "null"]
  50. }
  51. ]
  52. }
  53. $ kafka-create-topic topic=dbzupsert partitions=1
  54. $ kafka-ingest format=avro topic=dbzupsert key-format=avro key-schema=${keyschema} schema=${schema}
  55. {"b": "bdata", "a": 1} {"before": {"row": {"a": 1, "data": "fish", "b": "bdata"}}, "after": {"row": {"a": 1, "data": "fish2", "b": "bdata"}}}
  56. > CREATE CONNECTION IF NOT EXISTS csr_conn TO CONFLUENT SCHEMA REGISTRY (
  57. URL '${testdrive.schema-registry-url}'
  58. );
  59. > CREATE CONNECTION kafka_conn
  60. TO KAFKA (BROKER '${testdrive.kafka-addr}', SECURITY PROTOCOL PLAINTEXT);
  61. > CREATE SOURCE doin_upsert
  62. IN CLUSTER ${arg.single-replica-cluster}
  63. FROM KAFKA CONNECTION kafka_conn (TOPIC 'testdrive-dbzupsert-${testdrive.seed}')
  64. > CREATE TABLE doin_upsert_tbl FROM SOURCE doin_upsert (REFERENCE "testdrive-dbzupsert-${testdrive.seed}")
  65. FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr_conn
  66. ENVELOPE DEBEZIUM
  67. > SELECT * FROM doin_upsert_tbl
  68. a data b
  69. -----------
  70. 1 fish2 bdata
  71. $ kafka-ingest format=avro topic=dbzupsert key-format=avro key-schema=${keyschema} schema=${schema}
  72. {"b": "bdata", "a": 1} {"before": {"row": {"a": 1, "data": "fish2", "b": "bdata"}}, "after": {"row": {"a": 1, "data": "fish3", "b": "bdata"}}}
  73. > SELECT * FROM doin_upsert_tbl
  74. a data b
  75. -----------
  76. 1 fish3 bdata