SET execution.checkpointing.interval = 1h;

CREATE FUNCTION ARRAY_AGGR AS 'com.zerofinance.function.ArrayAggr';

CREATE TABLE products (
    id INT,
    name STRING,
    description STRING,
    PRIMARY KEY (id) NOT ENFORCED
  ) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = '${hostname}',
    'port' = '3306',
    'username' = '${username}',
    'password' = '${password}',
    'database-name' = 'demo_db',
    'table-name' = 'products'
  );
  
CREATE TABLE orders (
   order_id INT,
   order_date TIMESTAMP(0),
   customer_name STRING,
   price DECIMAL(10, 5),
   product_id INT,
   order_status BOOLEAN,
   PRIMARY KEY (order_id) NOT ENFORCED
 ) WITH (
   'connector' = 'mysql-cdc',
   'hostname' = '192.168.63.102',
   'port' = '3306',
   'username' = 'demo_db',
   'password' = 'Aa123456',
   'database-name' = 'demo_db',
   'table-name' = 'orders'
 );
 
CREATE TABLE enriched_orders (
   product_id INT,
   product_name STRING,
   product_description STRING,
   order_array ARRAY<ROW<order_id INT,order_date TIMESTAMP(0),customer_name STRING,price DECIMAL(10, 5),order_status BOOLEAN>>,
   PRIMARY KEY (product_id) NOT ENFORCED
 ) WITH (
     'connector' = 'elasticsearch-7',
     'hosts' = 'http://192.168.63.102:9200',
     'index' = 'enriched_orders_dave'
 );
 
INSERT INTO enriched_orders
  SELECT p.id AS product_id, p.name AS product_name, p.description AS product_description,
 (select ARRAY_AGGR(ROW(order_id,order_date,customer_name,price,order_status)) 
   from orders o 
   where o.product_id=p.id) as order_array
 FROM products AS p;