CREATE TABLE kafka_upsert_sink_table (
  id   INT,
  name STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  -- Define Upsert Kafka parameters.
  'connector' = 'upsert-kafka',  -- Specify the connector.
  'topic' = 'topic',  -- Replace this with the topic you want to write to.
  'properties.bootstrap.servers' = '...',  -- Replace this with your Kafka connection address.
  'key.format' = 'json',  -- Define the data format of keys.
  'value.format' = 'json'  -- Define the data format of values.
);

CREATE TABLE `kafka_json_source_table` (
  `id`   INT,
  `name` STRING
) WITH (
  -- Define Kafka parameters.
  'connector' = 'kafka',
  'topic' = 'Data-Input',  -- Replace this with the topic you want to consume data from.
  'scan.startup.mode' = 'latest-offset', -- Valid values include latest-offset, earliest-offset, specific-offsets, group-offsets, and timestamp.
  'properties.bootstrap.servers' = '172.28.28.13:9092',  -- Replace this with your Kafka connection address.
  'properties.group.id' = 'testGroup', -- (Required) The group ID.

  -- Define the data format (JSON).
  'format' = 'json',
  'json.fail-on-missing-field' = 'false',  -- If this is 'false', no errors will occur even when parameters are missing.
  'json.ignore-parse-errors' = 'true'    -- If this is 'true', all parse errors will be ignored.
);

CREATE TABLE kafka_upsert_sink_table (
  id   INT,
  name STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  -- Define Upsert Kafka parameters.
  'connector' = 'upsert-kafka',  -- Specify the connector.
  'topic' = 'topic',  -- Replace this with the topic you want to consume data from.
  'properties.bootstrap.servers' = '...',  -- Replace this with your Kafka connection address.
  'key.format' = 'json',  -- Define the data format of keys.
  'value.format' = 'json'  -- Define the data format of values.
);

-- Calculate pv and uv and insert them to upsert-kafka sink.
INSERT INTO kafka_upsert_sink_table
SELECT * FROM kafka_json_source_table;