Source Streams:
CREATE STREAM one ( ADC_ID BIGINT, ETL_DATATYPE VARCHAR(400)) WITH(KAFKA_TOPIC ='AMM_DATATYPE_CONVERSION',PARTITIONS=1,REPLICAS=1, VALUE_FORMAT='AVRO');
CREATE STREAM two ( ADT_ID BIGINT KEY,DATATYPE VARCHAR(400)) WITH(KAFKA_TOPIC ='AMM_DATATYPES',PARTITIONS=1,REPLICAS=1, VALUE_FORMAT='AVRO');
Target Stream:
CREATE STREAM TargetStream WITH (KAFKA_TOPIC ='TestTopic', VALUE_FORMAT='AVRO', PARTITIONS=3 )
AS SELECT CONCAT_WS('',one.ETL_DATATYPE,two.DATATYPE) AS ADDRESS,
one.ADC_ID as ADCID, one.ETL_DATATYPE as ETLDATATYPE, two.ADT_ID as ADTID
from one inner join two WITHIN 1 DAY on one.ADC_ID=two.ADT_ID EMIT CHANGES LIMIT 3;
here in this example ADCID column is missing in TestTopic.