SET execution.checkpointing.interval = 10min;

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

-- source
CREATE TABLE novel (
  id int,
  name string,
  PRIMARY KEY (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' = 'novel_[0-1]?'
);

CREATE TABLE chapter (
  `id` int ,
  `chapter_name` varchar,
  `novel_id` int,
  PRIMARY KEY (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' = 'chapter_[0-1]?'
);

-- sink
CREATE TABLE novel_index (
    id INT,
  	name STRING,
    chapter_list ARRAY<ROW<id INT,chapter_name STRING>>,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (    
    'connector' = 'elasticsearch-7',
    'hosts' = 'http://192.168.63.102:9200',      
    'index' = 'novel'
);

INSERT INTO novel_index
  SELECT n.id, n.name,
  (select ARRAY_AGGR(ROW(id,chapter_name)) 
   from chapter c 
   where c.novel_id=n.id) as chapter_list
 FROM novel AS n;
