Add a test of realtime upsert

test/testdrive/compaction-kafka.td

@@ -7,9 +7,6 @@
 # the Business Source License, use of this software will be governed
 # by the Apache License, Version 2.0.
 
-# TODO: uncomment tests when timestamping problem is fixed.
-# TODO: add tests where the updated value is less than old value
-
 $ set keyschema={
     "type": "record",
     "name": "Key",
@@ -89,14 +86,17 @@ mammalmore: {"f1":"moose", "f2": 2}
 > CREATE MATERIALIZED SOURCE bytesavro
   FROM KAFKA BROKER '${testdrive.kafka-addr}' TOPIC 'testdrive-textavro-${testdrive.seed}'
     WITH (consistency = 'testdrive-data-consistency-${testdrive.seed}')
-  FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY '${testdrive.schema-registry-url}'
+  FORMAT AVRO USING SCHEMA '${schema}'
   ENVELOPE UPSERT FORMAT BYTES
 
+$ file-append path=data-schema.json
+\${schema}
+
 > CREATE MATERIALIZED SOURCE textavro
   FROM KAFKA BROKER '${testdrive.kafka-addr}' TOPIC
   'testdrive-textavro-${testdrive.seed}'
     WITH (consistency = 'testdrive-data-consistency-${testdrive.seed}')
-  FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY '${testdrive.schema-registry-url}'
+  FORMAT AVRO USING SCHEMA FILE '${testdrive.temp-dir}/data-schema.json'
   ENVELOPE UPSERT FORMAT TEXT
 
 > select * from bytesavro
@@ -218,3 +218,42 @@ key           data  mz_offset
 -----------------------------
 birdmore      geese 4
 mammalmore    moose 5
+
+$ kafka-create-topic topic=realtimeavroavro
+
+$ kafka-ingest format=avro topic=realtimeavroavro key-format=avro key-schema=${keyschema} schema=${schema} publish=true
+{"key": "lightning"} {"f1": "sheep", "f2": 53}
+{"key": "water"} {"f1":"plesiosaur", "f2": 224}
+{"key": "earth"} {"f1": "turtle", "f2": 34}
+{"key": "lightning"} {"f1": "sheep", "f2": 54}
+{"key": "earth"} {"f1": "snake", "f2": 68}
+{"key": "water"} {"f1": "crocodile", "f2": 7}
+{"key": "earth"} null
+
+> CREATE MATERIALIZED SOURCE realtimeavroavro
+  FROM KAFKA BROKER '${testdrive.kafka-addr}' TOPIC
+  'testdrive-realtimeavroavro-${testdrive.seed}'
+  FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY '${testdrive.schema-registry-url}'
+  ENVELOPE UPSERT
+
+> select * from realtimeavroavro
+key        f1             f2
+----------------------------
+lightning  sheep          54
+water      crocodile      7
+
+$ kafka-ingest format=avro topic=realtimeavroavro key-format=avro key-schema=${keyschema} schema=${schema} publish=true
+{"key": "air"} {"f1": "pigeon", "f2": 10}
+{"key": "air"} {"f1": "owl", "f2": 15}
+{"key": "earth"} {"f1": "rhinoceros", "f2": 211}
+{"key": "air"} {"f1": "chicken", "f2": 47}
+{"key": "lightning"} null
+{"key": "lightning"} {"f1":"dog", "f2": 243}
+{"key": "water"} null
+
+> select * from realtimeavroavro
+key        f1             f2
+------------------------------
+air        chicken        47
+lightning  dog            243
+earth      rhinoceros     211