queries.sql

CREATE STREAM IF NOT EXISTS ORDER_COMMANDS_AND_EVENTS (
  AGGREGATE_ID STRING KEY,
  IS_COMMAND BOOLEAN,
  TYPE STRING,
  JSON_DATA STRING
) WITH (
  KAFKA_TOPIC='order-commands-and-events',
  PARTITIONS=10,
  REPLICAS=1,
  VALUE_FORMAT='JSON'
);

CREATE TABLE IF NOT EXISTS ORDER_AGGREGATES WITH (
  KAFKA_TOPIC='order-aggregates',
  PARTITIONS=10,
  REPLICAS=1,
  VALUE_FORMAT='JSON'
) AS SELECT
  AGGREGATE_ID,
  COLLECT_LIST(IS_COMMAND) AS IS_COMMAND_LIST,
  COLLECT_LIST(TYPE) AS TYPE_LIST,
  COLLECT_LIST(JSON_DATA) AS JSON_DATA_LIST
FROM ORDER_COMMANDS_AND_EVENTS
GROUP BY AGGREGATE_ID
EMIT CHANGES;

CREATE STREAM IF NOT EXISTS ORDER_INTEGRATION_EVENTS (
  ORDER_ID STRING KEY,
  VERSION INT,
  EVENT_TYPE STRING,
  EVENT_TIMESTAMP BIGINT,
  STATUS STRING,
  RIDER_ID STRING,
  PRICE STRING,
  ROUTE ARRAY<STRUCT<ADDRESS STRING, LAT DOUBLE, LON DOUBLE>>,
  DRIVER_ID STRING
) WITH (
  KAFKA_TOPIC='order-integration-events',
  PARTITIONS=10,
  REPLICAS=1,
  VALUE_FORMAT='JSON'
);