You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by bs...@apache.org on 2018/12/03 19:57:09 UTC
[2/4] hive git commit: HIVE-20987: Split the MiniDruid Cli tests into 3 different ones eg Druid_only, Druid_kakfa and kafka_only (Slim B via Jesus Camacho Rodriguez) for each cluster we are using different ports kafka_only Kafka broker 9093 for druid_kaf

http://git-wip-us.apache.org/repos/asf/hive/blob/8e30b5e0/ql/src/test/results/clientpositive/kafka/kafka_storage_handler.q.out
----------------------------------------------------------------------
diff --git a/ql/src/test/results/clientpositive/kafka/kafka_storage_handler.q.out b/ql/src/test/results/clientpositive/kafka/kafka_storage_handler.q.out
new file mode 100644
index 0000000..82984fd
--- /dev/null
+++ b/ql/src/test/results/clientpositive/kafka/kafka_storage_handler.q.out
@@ -0,0 +1,1717 @@
+PREHOOK: query: CREATE EXTERNAL TABLE kafka_table
+(`__time` timestamp , `page` string, `user` string, `language` string,
+`country` string,`continent` string, `namespace` string, `newPage` boolean, `unpatrolled` boolean,
+`anonymous` boolean, `robot` boolean, added int, deleted int, delta bigint)
+STORED BY 'org.apache.hadoop.hive.kafka.KafkaStorageHandler'
+WITH SERDEPROPERTIES ("timestamp.formats"="yyyy-MM-dd\'T\'HH:mm:ss\'Z\'")
+TBLPROPERTIES
+("kafka.topic" = "test-topic",
+"kafka.bootstrap.servers"="localhost:9093",
+"kafka.serde.class"="org.apache.hadoop.hive.serde2.JsonSerDe")
+PREHOOK: type: CREATETABLE
+PREHOOK: Output: database:default
+PREHOOK: Output: default@kafka_table
+POSTHOOK: query: CREATE EXTERNAL TABLE kafka_table
+(`__time` timestamp , `page` string, `user` string, `language` string,
+`country` string,`continent` string, `namespace` string, `newPage` boolean, `unpatrolled` boolean,
+`anonymous` boolean, `robot` boolean, added int, deleted int, delta bigint)
+STORED BY 'org.apache.hadoop.hive.kafka.KafkaStorageHandler'
+WITH SERDEPROPERTIES ("timestamp.formats"="yyyy-MM-dd\'T\'HH:mm:ss\'Z\'")
+TBLPROPERTIES
+("kafka.topic" = "test-topic",
+"kafka.bootstrap.servers"="localhost:9093",
+"kafka.serde.class"="org.apache.hadoop.hive.serde2.JsonSerDe")
+POSTHOOK: type: CREATETABLE
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@kafka_table
+PREHOOK: query: DESCRIBE EXTENDED kafka_table
+PREHOOK: type: DESCTABLE
+PREHOOK: Input: default@kafka_table
+POSTHOOK: query: DESCRIBE EXTENDED kafka_table
+POSTHOOK: type: DESCTABLE
+POSTHOOK: Input: default@kafka_table
+__time              	timestamp           	from deserializer   
+page                	string              	from deserializer   
+user                	string              	from deserializer   
+language            	string              	from deserializer   
+country             	string              	from deserializer   
+continent           	string              	from deserializer   
+namespace           	string              	from deserializer   
+newpage             	boolean             	from deserializer   
+unpatrolled         	boolean             	from deserializer   
+anonymous           	boolean             	from deserializer   
+robot               	boolean             	from deserializer   
+added               	int                 	from deserializer   
+deleted             	int                 	from deserializer   
+delta               	bigint              	from deserializer   
+__key               	binary              	from deserializer   
+__partition         	int                 	from deserializer   
+__offset            	bigint              	from deserializer   
+__timestamp         	bigint              	from deserializer   
+	 	 
+#### A masked pattern was here ####
+StorageHandlerInfo	 	 
+Partition(topic = test-topic, partition = 0, leader = 1, replicas = [1], isr = [1], offlineReplicas = []) [start offset = [0], end offset = [10]]	 	 
+PREHOOK: query: Select `__partition` , `__offset`,`__key`, `__time`, `page`, `user`, `language`, `country`,`continent`, `namespace`, `newPage` ,
+`unpatrolled` , `anonymous` , `robot` , added , deleted , delta FROM kafka_table
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: Select `__partition` , `__offset`,`__key`, `__time`, `page`, `user`, `language`, `country`,`continent`, `namespace`, `newPage` ,
+`unpatrolled` , `anonymous` , `robot` , added , deleted , delta FROM kafka_table
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+0	0	key	NULL	Gypsy Danger	nuclear	en	United States	North America	article	true	true	false	false	57	200	-143
+0	1	key	NULL	Striker Eureka	speed	en	Australia	Australia	wikipedia	true	false	false	true	459	129	330
+0	2	key	NULL	Cherno Alpha	masterYi	ru	Russia	Asia	article	true	false	false	true	123	12	111
+0	3	key	NULL	Crimson Typhoon	triplets	zh	China	Asia	wikipedia	false	true	false	true	905	5	900
+0	4	key	NULL	Coyote Tango	stringer	ja	Japan	Asia	wikipedia	false	true	false	true	1	10	-9
+0	5	key	NULL	Gypsy Danger	nuclear	en	United States	North America	article	true	true	false	false	57	200	-143
+0	6	key	NULL	Striker Eureka	speed	en	Australia	Australia	wikipedia	true	false	false	true	459	129	330
+0	7	key	NULL	Cherno Alpha	masterYi	ru	Russia	Asia	article	true	false	false	true	123	12	111
+0	8	key	NULL	Crimson Typhoon	triplets	zh	China	Asia	wikipedia	false	true	false	true	905	5	900
+0	9	key	NULL	Coyote Tango	stringer	ja	Japan	Asia	wikipedia	false	true	false	true	1	10	-9
+PREHOOK: query: Select count(*) FROM kafka_table
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: Select count(*) FROM kafka_table
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+10
+PREHOOK: query: Select `__partition`, `__offset`, `__time`, `page`, `user`, `language`, `country`,`continent`, `namespace`, `newPage` ,
+`unpatrolled` , `anonymous` , `robot` , added , deleted , delta
+from kafka_table where `__timestamp` > 1533960760123
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: Select `__partition`, `__offset`, `__time`, `page`, `user`, `language`, `country`,`continent`, `namespace`, `newPage` ,
+`unpatrolled` , `anonymous` , `robot` , added , deleted , delta
+from kafka_table where `__timestamp` > 1533960760123
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+0	0	NULL	Gypsy Danger	nuclear	en	United States	North America	article	true	true	false	false	57	200	-143
+0	1	NULL	Striker Eureka	speed	en	Australia	Australia	wikipedia	true	false	false	true	459	129	330
+0	2	NULL	Cherno Alpha	masterYi	ru	Russia	Asia	article	true	false	false	true	123	12	111
+0	3	NULL	Crimson Typhoon	triplets	zh	China	Asia	wikipedia	false	true	false	true	905	5	900
+0	4	NULL	Coyote Tango	stringer	ja	Japan	Asia	wikipedia	false	true	false	true	1	10	-9
+0	5	NULL	Gypsy Danger	nuclear	en	United States	North America	article	true	true	false	false	57	200	-143
+0	6	NULL	Striker Eureka	speed	en	Australia	Australia	wikipedia	true	false	false	true	459	129	330
+0	7	NULL	Cherno Alpha	masterYi	ru	Russia	Asia	article	true	false	false	true	123	12	111
+0	8	NULL	Crimson Typhoon	triplets	zh	China	Asia	wikipedia	false	true	false	true	905	5	900
+0	9	NULL	Coyote Tango	stringer	ja	Japan	Asia	wikipedia	false	true	false	true	1	10	-9
+PREHOOK: query: Select `__partition`, `__offset` ,`__time`, `page`, `user`, `language`, `country`,`continent`, `namespace`, `newPage` ,
+`unpatrolled` , `anonymous` , `robot` , added , deleted , delta
+from kafka_table where `__timestamp` > 533960760123
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: Select `__partition`, `__offset` ,`__time`, `page`, `user`, `language`, `country`,`continent`, `namespace`, `newPage` ,
+`unpatrolled` , `anonymous` , `robot` , added , deleted , delta
+from kafka_table where `__timestamp` > 533960760123
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+0	0	NULL	Gypsy Danger	nuclear	en	United States	North America	article	true	true	false	false	57	200	-143
+0	1	NULL	Striker Eureka	speed	en	Australia	Australia	wikipedia	true	false	false	true	459	129	330
+0	2	NULL	Cherno Alpha	masterYi	ru	Russia	Asia	article	true	false	false	true	123	12	111
+0	3	NULL	Crimson Typhoon	triplets	zh	China	Asia	wikipedia	false	true	false	true	905	5	900
+0	4	NULL	Coyote Tango	stringer	ja	Japan	Asia	wikipedia	false	true	false	true	1	10	-9
+0	5	NULL	Gypsy Danger	nuclear	en	United States	North America	article	true	true	false	false	57	200	-143
+0	6	NULL	Striker Eureka	speed	en	Australia	Australia	wikipedia	true	false	false	true	459	129	330
+0	7	NULL	Cherno Alpha	masterYi	ru	Russia	Asia	article	true	false	false	true	123	12	111
+0	8	NULL	Crimson Typhoon	triplets	zh	China	Asia	wikipedia	false	true	false	true	905	5	900
+0	9	NULL	Coyote Tango	stringer	ja	Japan	Asia	wikipedia	false	true	false	true	1	10	-9
+PREHOOK: query: Select `__partition`, `__offset`,`__time`, `page`, `user`, `language`, `country`,`continent`, `namespace`, `newPage` ,
+`unpatrolled` , `anonymous` , `robot` , added , deleted , delta
+from kafka_table where (`__offset` > 7 and `__partition` = 0 and `__offset` <9 ) OR
+`__offset` = 4 and `__partition` = 0 OR (`__offset` <= 1 and `__partition` = 0 and `__offset` > 0)
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: Select `__partition`, `__offset`,`__time`, `page`, `user`, `language`, `country`,`continent`, `namespace`, `newPage` ,
+`unpatrolled` , `anonymous` , `robot` , added , deleted , delta
+from kafka_table where (`__offset` > 7 and `__partition` = 0 and `__offset` <9 ) OR
+`__offset` = 4 and `__partition` = 0 OR (`__offset` <= 1 and `__partition` = 0 and `__offset` > 0)
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+0	1	NULL	Striker Eureka	speed	en	Australia	Australia	wikipedia	true	false	false	true	459	129	330
+0	4	NULL	Coyote Tango	stringer	ja	Japan	Asia	wikipedia	false	true	false	true	1	10	-9
+0	8	NULL	Crimson Typhoon	triplets	zh	China	Asia	wikipedia	false	true	false	true	905	5	900
+PREHOOK: query: Select `__key`,`__partition`, `__offset`,`__time`, `page`, `user` from kafka_table where `__offset` = 5
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: Select `__key`,`__partition`, `__offset`,`__time`, `page`, `user` from kafka_table where `__offset` = 5
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+key	0	5	NULL	Gypsy Danger	nuclear
+PREHOOK: query: Select `__key`,`__partition`, `__offset`,`__time`, `page`, `user` from kafka_table where `__offset` < 5
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: Select `__key`,`__partition`, `__offset`,`__time`, `page`, `user` from kafka_table where `__offset` < 5
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+key	0	0	NULL	Gypsy Danger	nuclear
+key	0	1	NULL	Striker Eureka	speed
+key	0	2	NULL	Cherno Alpha	masterYi
+key	0	3	NULL	Crimson Typhoon	triplets
+key	0	4	NULL	Coyote Tango	stringer
+PREHOOK: query: Select `__key`,`__partition`, `__offset`,`__time`, `page`, `user` from kafka_table where `__offset` > 5
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: Select `__key`,`__partition`, `__offset`,`__time`, `page`, `user` from kafka_table where `__offset` > 5
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+key	0	6	NULL	Striker Eureka	speed
+key	0	7	NULL	Cherno Alpha	masterYi
+key	0	8	NULL	Crimson Typhoon	triplets
+key	0	9	NULL	Coyote Tango	stringer
+PREHOOK: query: Select `__partition`, `__offset`, `user` from kafka_table where
+`__timestamp` >  to_epoch_milli(CURRENT_TIMESTAMP - interval '1' HOURS)
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: Select `__partition`, `__offset`, `user` from kafka_table where
+`__timestamp` >  to_epoch_milli(CURRENT_TIMESTAMP - interval '1' HOURS)
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+0	0	nuclear
+0	1	speed
+0	2	masterYi
+0	3	triplets
+0	4	stringer
+0	5	nuclear
+0	6	speed
+0	7	masterYi
+0	8	triplets
+0	9	stringer
+PREHOOK: query: Select  count(*) from kafka_table where `__partition` = 1
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: Select  count(*) from kafka_table where `__partition` = 1
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+0
+PREHOOK: query: Select count(*) from kafka_table where `__offset` = 100
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: Select count(*) from kafka_table where `__offset` = 100
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+0
+PREHOOK: query: Select count(*) from kafka_table where `__offset` <= 100 and `__partition` <= 100
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: Select count(*) from kafka_table where `__offset` <= 100 and `__partition` <= 100
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+10
+PREHOOK: query: Drop table kafka_table_offsets
+PREHOOK: type: DROPTABLE
+POSTHOOK: query: Drop table kafka_table_offsets
+POSTHOOK: type: DROPTABLE
+PREHOOK: query: create table kafka_table_offsets(partition_id int, max_offset bigint, insert_time timestamp)
+PREHOOK: type: CREATETABLE
+PREHOOK: Output: database:default
+PREHOOK: Output: default@kafka_table_offsets
+POSTHOOK: query: create table kafka_table_offsets(partition_id int, max_offset bigint, insert_time timestamp)
+POSTHOOK: type: CREATETABLE
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@kafka_table_offsets
+PREHOOK: query: insert overwrite table kafka_table_offsets select `__partition`, min(`__offset`) - 1, CURRENT_TIMESTAMP from kafka_table group by `__partition`, CURRENT_TIMESTAMP
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table
+PREHOOK: Output: default@kafka_table_offsets
+POSTHOOK: query: insert overwrite table kafka_table_offsets select `__partition`, min(`__offset`) - 1, CURRENT_TIMESTAMP from kafka_table group by `__partition`, CURRENT_TIMESTAMP
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table
+POSTHOOK: Output: default@kafka_table_offsets
+POSTHOOK: Lineage: kafka_table_offsets.insert_time SIMPLE []
+POSTHOOK: Lineage: kafka_table_offsets.max_offset EXPRESSION [(kafka_table)kafka_table.FieldSchema(name:__offset, type:bigint, comment:from deserializer), ]
+POSTHOOK: Lineage: kafka_table_offsets.partition_id SIMPLE [(kafka_table)kafka_table.FieldSchema(name:__partition, type:int, comment:from deserializer), ]
+PREHOOK: query: select partition_id, max_offset from kafka_table_offsets
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table_offsets
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: select partition_id, max_offset from kafka_table_offsets
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table_offsets
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+0	-1
+PREHOOK: query: Drop table orc_kafka_table
+PREHOOK: type: DROPTABLE
+POSTHOOK: query: Drop table orc_kafka_table
+POSTHOOK: type: DROPTABLE
+PREHOOK: query: Create table orc_kafka_table (partition_id int, row_offset bigint, kafka_ts bigint,
+ `__time` timestamp , `page` string, `user` string, `language` string,
+`country` string,`continent` string, `namespace` string, `newPage` boolean, `unpatrolled` boolean,
+`anonymous` boolean, `robot` boolean, added int, deleted int, delta bigint
+) stored as ORC
+PREHOOK: type: CREATETABLE
+PREHOOK: Output: database:default
+PREHOOK: Output: default@orc_kafka_table
+POSTHOOK: query: Create table orc_kafka_table (partition_id int, row_offset bigint, kafka_ts bigint,
+ `__time` timestamp , `page` string, `user` string, `language` string,
+`country` string,`continent` string, `namespace` string, `newPage` boolean, `unpatrolled` boolean,
+`anonymous` boolean, `robot` boolean, added int, deleted int, delta bigint
+) stored as ORC
+POSTHOOK: type: CREATETABLE
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@orc_kafka_table
+PREHOOK: query: From kafka_table ktable JOIN kafka_table_offsets offset_table
+on (ktable.`__partition` = offset_table.partition_id and ktable.`__offset` > offset_table.max_offset and  ktable.`__offset` < 3 )
+insert into table orc_kafka_table select `__partition`, `__offset`, `__timestamp`,
+`__time`, `page`, `user`, `language`, `country`,`continent`, `namespace`, `newPage` ,
+`unpatrolled` , `anonymous` , `robot` , added , deleted , delta
+Insert overwrite table kafka_table_offsets select
+`__partition`, max(`__offset`), CURRENT_TIMESTAMP group by `__partition`, CURRENT_TIMESTAMP
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table
+PREHOOK: Input: default@kafka_table_offsets
+PREHOOK: Output: default@kafka_table_offsets
+PREHOOK: Output: default@orc_kafka_table
+POSTHOOK: query: From kafka_table ktable JOIN kafka_table_offsets offset_table
+on (ktable.`__partition` = offset_table.partition_id and ktable.`__offset` > offset_table.max_offset and  ktable.`__offset` < 3 )
+insert into table orc_kafka_table select `__partition`, `__offset`, `__timestamp`,
+`__time`, `page`, `user`, `language`, `country`,`continent`, `namespace`, `newPage` ,
+`unpatrolled` , `anonymous` , `robot` , added , deleted , delta
+Insert overwrite table kafka_table_offsets select
+`__partition`, max(`__offset`), CURRENT_TIMESTAMP group by `__partition`, CURRENT_TIMESTAMP
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table
+POSTHOOK: Input: default@kafka_table_offsets
+POSTHOOK: Output: default@kafka_table_offsets
+POSTHOOK: Output: default@orc_kafka_table
+POSTHOOK: Lineage: kafka_table_offsets.insert_time EXPRESSION []
+POSTHOOK: Lineage: kafka_table_offsets.max_offset EXPRESSION [(kafka_table)ktable.FieldSchema(name:__offset, type:bigint, comment:from deserializer), ]
+POSTHOOK: Lineage: kafka_table_offsets.partition_id SIMPLE [(kafka_table)ktable.FieldSchema(name:__partition, type:int, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.__time SIMPLE [(kafka_table)ktable.FieldSchema(name:__time, type:timestamp, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.added SIMPLE [(kafka_table)ktable.FieldSchema(name:added, type:int, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.anonymous SIMPLE [(kafka_table)ktable.FieldSchema(name:anonymous, type:boolean, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.continent SIMPLE [(kafka_table)ktable.FieldSchema(name:continent, type:string, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.country SIMPLE [(kafka_table)ktable.FieldSchema(name:country, type:string, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.deleted SIMPLE [(kafka_table)ktable.FieldSchema(name:deleted, type:int, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.delta SIMPLE [(kafka_table)ktable.FieldSchema(name:delta, type:bigint, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.kafka_ts SIMPLE [(kafka_table)ktable.FieldSchema(name:__timestamp, type:bigint, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.language SIMPLE [(kafka_table)ktable.FieldSchema(name:language, type:string, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.namespace SIMPLE [(kafka_table)ktable.FieldSchema(name:namespace, type:string, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.newpage SIMPLE [(kafka_table)ktable.FieldSchema(name:newpage, type:boolean, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.page SIMPLE [(kafka_table)ktable.FieldSchema(name:page, type:string, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.partition_id SIMPLE [(kafka_table)ktable.FieldSchema(name:__partition, type:int, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.robot SIMPLE [(kafka_table)ktable.FieldSchema(name:robot, type:boolean, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.row_offset SIMPLE [(kafka_table)ktable.FieldSchema(name:__offset, type:bigint, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.unpatrolled SIMPLE [(kafka_table)ktable.FieldSchema(name:unpatrolled, type:boolean, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.user SIMPLE [(kafka_table)ktable.FieldSchema(name:user, type:string, comment:from deserializer), ]
+PREHOOK: query: select count(*) from  orc_kafka_table
+PREHOOK: type: QUERY
+PREHOOK: Input: default@orc_kafka_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: select count(*) from  orc_kafka_table
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@orc_kafka_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+3
+PREHOOK: query: select partition_id, max_offset from kafka_table_offsets
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table_offsets
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: select partition_id, max_offset from kafka_table_offsets
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table_offsets
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+0	2
+PREHOOK: query: select `partition_id`, `row_offset`,`__time`, `page`, `user`, `language`, `country`,`continent`, `namespace`, `newPage` ,
+`unpatrolled` , `anonymous` , `robot` , added , deleted , delta from  orc_kafka_table
+PREHOOK: type: QUERY
+PREHOOK: Input: default@orc_kafka_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: select `partition_id`, `row_offset`,`__time`, `page`, `user`, `language`, `country`,`continent`, `namespace`, `newPage` ,
+`unpatrolled` , `anonymous` , `robot` , added , deleted , delta from  orc_kafka_table
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@orc_kafka_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+0	0	NULL	Gypsy Danger	nuclear	en	United States	North America	article	true	true	false	false	57	200	-143
+0	1	NULL	Striker Eureka	speed	en	Australia	Australia	wikipedia	true	false	false	true	459	129	330
+0	2	NULL	Cherno Alpha	masterYi	ru	Russia	Asia	article	true	false	false	true	123	12	111
+PREHOOK: query: From kafka_table ktable JOIN kafka_table_offsets offset_table
+on (ktable.`__partition` = offset_table.partition_id and ktable.`__offset` > offset_table.max_offset)
+insert into table orc_kafka_table select `__partition`, `__offset`, `__timestamp`,
+`__time`, `page`, `user`, `language`, `country`,`continent`, `namespace`, `newPage` ,
+`unpatrolled` , `anonymous` , `robot` , added , deleted , delta
+Insert overwrite table kafka_table_offsets select
+`__partition`, max(`__offset`), CURRENT_TIMESTAMP group by `__partition`, CURRENT_TIMESTAMP
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table
+PREHOOK: Input: default@kafka_table_offsets
+PREHOOK: Output: default@kafka_table_offsets
+PREHOOK: Output: default@orc_kafka_table
+POSTHOOK: query: From kafka_table ktable JOIN kafka_table_offsets offset_table
+on (ktable.`__partition` = offset_table.partition_id and ktable.`__offset` > offset_table.max_offset)
+insert into table orc_kafka_table select `__partition`, `__offset`, `__timestamp`,
+`__time`, `page`, `user`, `language`, `country`,`continent`, `namespace`, `newPage` ,
+`unpatrolled` , `anonymous` , `robot` , added , deleted , delta
+Insert overwrite table kafka_table_offsets select
+`__partition`, max(`__offset`), CURRENT_TIMESTAMP group by `__partition`, CURRENT_TIMESTAMP
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table
+POSTHOOK: Input: default@kafka_table_offsets
+POSTHOOK: Output: default@kafka_table_offsets
+POSTHOOK: Output: default@orc_kafka_table
+POSTHOOK: Lineage: kafka_table_offsets.insert_time EXPRESSION []
+POSTHOOK: Lineage: kafka_table_offsets.max_offset EXPRESSION [(kafka_table)ktable.FieldSchema(name:__offset, type:bigint, comment:from deserializer), ]
+POSTHOOK: Lineage: kafka_table_offsets.partition_id SIMPLE [(kafka_table)ktable.FieldSchema(name:__partition, type:int, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.__time SIMPLE [(kafka_table)ktable.FieldSchema(name:__time, type:timestamp, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.added SIMPLE [(kafka_table)ktable.FieldSchema(name:added, type:int, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.anonymous SIMPLE [(kafka_table)ktable.FieldSchema(name:anonymous, type:boolean, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.continent SIMPLE [(kafka_table)ktable.FieldSchema(name:continent, type:string, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.country SIMPLE [(kafka_table)ktable.FieldSchema(name:country, type:string, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.deleted SIMPLE [(kafka_table)ktable.FieldSchema(name:deleted, type:int, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.delta SIMPLE [(kafka_table)ktable.FieldSchema(name:delta, type:bigint, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.kafka_ts SIMPLE [(kafka_table)ktable.FieldSchema(name:__timestamp, type:bigint, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.language SIMPLE [(kafka_table)ktable.FieldSchema(name:language, type:string, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.namespace SIMPLE [(kafka_table)ktable.FieldSchema(name:namespace, type:string, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.newpage SIMPLE [(kafka_table)ktable.FieldSchema(name:newpage, type:boolean, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.page SIMPLE [(kafka_table)ktable.FieldSchema(name:page, type:string, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.partition_id SIMPLE [(kafka_table)ktable.FieldSchema(name:__partition, type:int, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.robot SIMPLE [(kafka_table)ktable.FieldSchema(name:robot, type:boolean, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.row_offset SIMPLE [(kafka_table)ktable.FieldSchema(name:__offset, type:bigint, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.unpatrolled SIMPLE [(kafka_table)ktable.FieldSchema(name:unpatrolled, type:boolean, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.user SIMPLE [(kafka_table)ktable.FieldSchema(name:user, type:string, comment:from deserializer), ]
+PREHOOK: query: select partition_id, max_offset from kafka_table_offsets
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table_offsets
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: select partition_id, max_offset from kafka_table_offsets
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table_offsets
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+0	9
+PREHOOK: query: select count(*) from  orc_kafka_table
+PREHOOK: type: QUERY
+PREHOOK: Input: default@orc_kafka_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: select count(*) from  orc_kafka_table
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@orc_kafka_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+10
+PREHOOK: query: select `partition_id`, `row_offset`,`__time`, `page`, `user`, `language`, `country`,`continent`, `namespace`, `newPage` ,
+`unpatrolled` , `anonymous` , `robot` , added , deleted , delta from  orc_kafka_table
+PREHOOK: type: QUERY
+PREHOOK: Input: default@orc_kafka_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: select `partition_id`, `row_offset`,`__time`, `page`, `user`, `language`, `country`,`continent`, `namespace`, `newPage` ,
+`unpatrolled` , `anonymous` , `robot` , added , deleted , delta from  orc_kafka_table
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@orc_kafka_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+0	0	NULL	Gypsy Danger	nuclear	en	United States	North America	article	true	true	false	false	57	200	-143
+0	1	NULL	Striker Eureka	speed	en	Australia	Australia	wikipedia	true	false	false	true	459	129	330
+0	2	NULL	Cherno Alpha	masterYi	ru	Russia	Asia	article	true	false	false	true	123	12	111
+0	3	NULL	Crimson Typhoon	triplets	zh	China	Asia	wikipedia	false	true	false	true	905	5	900
+0	4	NULL	Coyote Tango	stringer	ja	Japan	Asia	wikipedia	false	true	false	true	1	10	-9
+0	5	NULL	Gypsy Danger	nuclear	en	United States	North America	article	true	true	false	false	57	200	-143
+0	6	NULL	Striker Eureka	speed	en	Australia	Australia	wikipedia	true	false	false	true	459	129	330
+0	7	NULL	Cherno Alpha	masterYi	ru	Russia	Asia	article	true	false	false	true	123	12	111
+0	8	NULL	Crimson Typhoon	triplets	zh	China	Asia	wikipedia	false	true	false	true	905	5	900
+0	9	NULL	Coyote Tango	stringer	ja	Japan	Asia	wikipedia	false	true	false	true	1	10	-9
+PREHOOK: query: Drop table kafka_table_offsets
+PREHOOK: type: DROPTABLE
+PREHOOK: Input: default@kafka_table_offsets
+PREHOOK: Output: default@kafka_table_offsets
+POSTHOOK: query: Drop table kafka_table_offsets
+POSTHOOK: type: DROPTABLE
+POSTHOOK: Input: default@kafka_table_offsets
+POSTHOOK: Output: default@kafka_table_offsets
+PREHOOK: query: create table kafka_table_offsets(partition_id int, max_offset bigint, insert_time timestamp)
+PREHOOK: type: CREATETABLE
+PREHOOK: Output: database:default
+PREHOOK: Output: default@kafka_table_offsets
+POSTHOOK: query: create table kafka_table_offsets(partition_id int, max_offset bigint, insert_time timestamp)
+POSTHOOK: type: CREATETABLE
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@kafka_table_offsets
+PREHOOK: query: Drop table orc_kafka_table
+PREHOOK: type: DROPTABLE
+PREHOOK: Input: default@orc_kafka_table
+PREHOOK: Output: default@orc_kafka_table
+POSTHOOK: query: Drop table orc_kafka_table
+POSTHOOK: type: DROPTABLE
+POSTHOOK: Input: default@orc_kafka_table
+POSTHOOK: Output: default@orc_kafka_table
+PREHOOK: query: Create table orc_kafka_table (partition_id int, row_offset bigint, kafka_ts bigint,
+ `__time` timestamp , `page` string, `user` string, `language` string,
+`country` string,`continent` string, `namespace` string, `newPage` boolean, `unpatrolled` boolean,
+`anonymous` boolean, `robot` boolean, added int, deleted int, delta bigint
+) stored as ORC
+PREHOOK: type: CREATETABLE
+PREHOOK: Output: database:default
+PREHOOK: Output: default@orc_kafka_table
+POSTHOOK: query: Create table orc_kafka_table (partition_id int, row_offset bigint, kafka_ts bigint,
+ `__time` timestamp , `page` string, `user` string, `language` string,
+`country` string,`continent` string, `namespace` string, `newPage` boolean, `unpatrolled` boolean,
+`anonymous` boolean, `robot` boolean, added int, deleted int, delta bigint
+) stored as ORC
+POSTHOOK: type: CREATETABLE
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@orc_kafka_table
+PREHOOK: query: From kafka_table ktable LEFT OUTER JOIN kafka_table_offsets offset_table
+on (ktable.`__partition` = offset_table.partition_id and ktable.`__offset` > offset_table.max_offset )
+insert into table orc_kafka_table select `__partition`, `__offset`, `__timestamp`,
+`__time`, `page`, `user`, `language`, `country`,`continent`, `namespace`, `newPage` ,
+`unpatrolled` , `anonymous` , `robot` , added , deleted , delta
+Insert overwrite table kafka_table_offsets select
+`__partition`, max(`__offset`), CURRENT_TIMESTAMP group by `__partition`, CURRENT_TIMESTAMP
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table
+PREHOOK: Input: default@kafka_table_offsets
+PREHOOK: Output: default@kafka_table_offsets
+PREHOOK: Output: default@orc_kafka_table
+POSTHOOK: query: From kafka_table ktable LEFT OUTER JOIN kafka_table_offsets offset_table
+on (ktable.`__partition` = offset_table.partition_id and ktable.`__offset` > offset_table.max_offset )
+insert into table orc_kafka_table select `__partition`, `__offset`, `__timestamp`,
+`__time`, `page`, `user`, `language`, `country`,`continent`, `namespace`, `newPage` ,
+`unpatrolled` , `anonymous` , `robot` , added , deleted , delta
+Insert overwrite table kafka_table_offsets select
+`__partition`, max(`__offset`), CURRENT_TIMESTAMP group by `__partition`, CURRENT_TIMESTAMP
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table
+POSTHOOK: Input: default@kafka_table_offsets
+POSTHOOK: Output: default@kafka_table_offsets
+POSTHOOK: Output: default@orc_kafka_table
+POSTHOOK: Lineage: kafka_table_offsets.insert_time EXPRESSION []
+POSTHOOK: Lineage: kafka_table_offsets.max_offset EXPRESSION [(kafka_table)ktable.FieldSchema(name:__offset, type:bigint, comment:from deserializer), ]
+POSTHOOK: Lineage: kafka_table_offsets.partition_id SIMPLE [(kafka_table)ktable.FieldSchema(name:__partition, type:int, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.__time SIMPLE [(kafka_table)ktable.FieldSchema(name:__time, type:timestamp, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.added SIMPLE [(kafka_table)ktable.FieldSchema(name:added, type:int, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.anonymous SIMPLE [(kafka_table)ktable.FieldSchema(name:anonymous, type:boolean, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.continent SIMPLE [(kafka_table)ktable.FieldSchema(name:continent, type:string, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.country SIMPLE [(kafka_table)ktable.FieldSchema(name:country, type:string, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.deleted SIMPLE [(kafka_table)ktable.FieldSchema(name:deleted, type:int, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.delta SIMPLE [(kafka_table)ktable.FieldSchema(name:delta, type:bigint, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.kafka_ts SIMPLE [(kafka_table)ktable.FieldSchema(name:__timestamp, type:bigint, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.language SIMPLE [(kafka_table)ktable.FieldSchema(name:language, type:string, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.namespace SIMPLE [(kafka_table)ktable.FieldSchema(name:namespace, type:string, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.newpage SIMPLE [(kafka_table)ktable.FieldSchema(name:newpage, type:boolean, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.page SIMPLE [(kafka_table)ktable.FieldSchema(name:page, type:string, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.partition_id SIMPLE [(kafka_table)ktable.FieldSchema(name:__partition, type:int, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.robot SIMPLE [(kafka_table)ktable.FieldSchema(name:robot, type:boolean, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.row_offset SIMPLE [(kafka_table)ktable.FieldSchema(name:__offset, type:bigint, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.unpatrolled SIMPLE [(kafka_table)ktable.FieldSchema(name:unpatrolled, type:boolean, comment:from deserializer), ]
+POSTHOOK: Lineage: orc_kafka_table.user SIMPLE [(kafka_table)ktable.FieldSchema(name:user, type:string, comment:from deserializer), ]
+PREHOOK: query: select count(*) from  orc_kafka_table
+PREHOOK: type: QUERY
+PREHOOK: Input: default@orc_kafka_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: select count(*) from  orc_kafka_table
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@orc_kafka_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+10
+PREHOOK: query: select partition_id, max_offset from kafka_table_offsets
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table_offsets
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: select partition_id, max_offset from kafka_table_offsets
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table_offsets
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+0	9
+PREHOOK: query: select `partition_id`, `row_offset`,`__time`, `page`, `user`, `language`, `country`,`continent`, `namespace`, `newPage` ,
+`unpatrolled` , `anonymous` , `robot` , added , deleted , delta from  orc_kafka_table
+PREHOOK: type: QUERY
+PREHOOK: Input: default@orc_kafka_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: select `partition_id`, `row_offset`,`__time`, `page`, `user`, `language`, `country`,`continent`, `namespace`, `newPage` ,
+`unpatrolled` , `anonymous` , `robot` , added , deleted , delta from  orc_kafka_table
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@orc_kafka_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+0	0	NULL	Gypsy Danger	nuclear	en	United States	North America	article	true	true	false	false	57	200	-143
+0	1	NULL	Striker Eureka	speed	en	Australia	Australia	wikipedia	true	false	false	true	459	129	330
+0	2	NULL	Cherno Alpha	masterYi	ru	Russia	Asia	article	true	false	false	true	123	12	111
+0	3	NULL	Crimson Typhoon	triplets	zh	China	Asia	wikipedia	false	true	false	true	905	5	900
+0	4	NULL	Coyote Tango	stringer	ja	Japan	Asia	wikipedia	false	true	false	true	1	10	-9
+0	5	NULL	Gypsy Danger	nuclear	en	United States	North America	article	true	true	false	false	57	200	-143
+0	6	NULL	Striker Eureka	speed	en	Australia	Australia	wikipedia	true	false	false	true	459	129	330
+0	7	NULL	Cherno Alpha	masterYi	ru	Russia	Asia	article	true	false	false	true	123	12	111
+0	8	NULL	Crimson Typhoon	triplets	zh	China	Asia	wikipedia	false	true	false	true	905	5	900
+0	9	NULL	Coyote Tango	stringer	ja	Japan	Asia	wikipedia	false	true	false	true	1	10	-9
+PREHOOK: query: CREATE EXTERNAL TABLE kafka_table_2
+(`__time` timestamp with local time zone , `page` string, `user` string, `language` string,
+`country` string,`continent` string, `namespace` string, `newPage` boolean, `unpatrolled` boolean,
+`anonymous` boolean, `robot` boolean, added int, deleted int, delta bigint)
+STORED BY 'org.apache.hadoop.hive.kafka.KafkaStorageHandler'
+TBLPROPERTIES
+("kafka.topic" = "test-topic",
+"kafka.bootstrap.servers"="localhost:9093")
+PREHOOK: type: CREATETABLE
+PREHOOK: Output: database:default
+PREHOOK: Output: default@kafka_table_2
+POSTHOOK: query: CREATE EXTERNAL TABLE kafka_table_2
+(`__time` timestamp with local time zone , `page` string, `user` string, `language` string,
+`country` string,`continent` string, `namespace` string, `newPage` boolean, `unpatrolled` boolean,
+`anonymous` boolean, `robot` boolean, added int, deleted int, delta bigint)
+STORED BY 'org.apache.hadoop.hive.kafka.KafkaStorageHandler'
+TBLPROPERTIES
+("kafka.topic" = "test-topic",
+"kafka.bootstrap.servers"="localhost:9093")
+POSTHOOK: type: CREATETABLE
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@kafka_table_2
+PREHOOK: query: Select `__partition`, `__offset`, `__time`, `page`, `user`, `language`, `country`,`continent`, `namespace`, `newPage` ,
+`unpatrolled` , `anonymous` , `robot` , added , deleted , delta
+FROM kafka_table_2
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table_2
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: Select `__partition`, `__offset`, `__time`, `page`, `user`, `language`, `country`,`continent`, `namespace`, `newPage` ,
+`unpatrolled` , `anonymous` , `robot` , added , deleted , delta
+FROM kafka_table_2
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table_2
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+0	0	NULL	Gypsy Danger	nuclear	en	United States	North America	article	true	true	false	false	57	200	-143
+0	1	NULL	Striker Eureka	speed	en	Australia	Australia	wikipedia	true	false	false	true	459	129	330
+0	2	NULL	Cherno Alpha	masterYi	ru	Russia	Asia	article	true	false	false	true	123	12	111
+0	3	NULL	Crimson Typhoon	triplets	zh	China	Asia	wikipedia	false	true	false	true	905	5	900
+0	4	NULL	Coyote Tango	stringer	ja	Japan	Asia	wikipedia	false	true	false	true	1	10	-9
+0	5	NULL	Gypsy Danger	nuclear	en	United States	North America	article	true	true	false	false	57	200	-143
+0	6	NULL	Striker Eureka	speed	en	Australia	Australia	wikipedia	true	false	false	true	459	129	330
+0	7	NULL	Cherno Alpha	masterYi	ru	Russia	Asia	article	true	false	false	true	123	12	111
+0	8	NULL	Crimson Typhoon	triplets	zh	China	Asia	wikipedia	false	true	false	true	905	5	900
+0	9	NULL	Coyote Tango	stringer	ja	Japan	Asia	wikipedia	false	true	false	true	1	10	-9
+PREHOOK: query: Select count(*) FROM kafka_table_2
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table_2
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: Select count(*) FROM kafka_table_2
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table_2
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+10
+PREHOOK: query: CREATE EXTERNAL TABLE wiki_kafka_avro_table
+STORED BY 'org.apache.hadoop.hive.kafka.KafkaStorageHandler'
+TBLPROPERTIES
+("kafka.topic" = "wiki_kafka_avro_table",
+"kafka.bootstrap.servers"="localhost:9093",
+"kafka.serde.class"="org.apache.hadoop.hive.serde2.avro.AvroSerDe",
+'avro.schema.literal'='{
+  "type" : "record",
+  "name" : "Wikipedia",
+  "namespace" : "org.apache.hive.kafka",
+  "version": "1",
+  "fields" : [ {
+    "name" : "isrobot",
+    "type" : "boolean"
+  }, {
+    "name" : "channel",
+    "type" : "string"
+  }, {
+    "name" : "timestamp",
+    "type" : "string"
+  }, {
+    "name" : "flags",
+    "type" : "string"
+  }, {
+    "name" : "isunpatrolled",
+    "type" : "boolean"
+  }, {
+    "name" : "page",
+    "type" : "string"
+  }, {
+    "name" : "diffurl",
+    "type" : "string"
+  }, {
+    "name" : "added",
+    "type" : "long"
+  }, {
+    "name" : "comment",
+    "type" : "string"
+  }, {
+    "name" : "commentlength",
+    "type" : "long"
+  }, {
+    "name" : "isnew",
+    "type" : "boolean"
+  }, {
+    "name" : "isminor",
+    "type" : "boolean"
+  }, {
+    "name" : "delta",
+    "type" : "long"
+  }, {
+    "name" : "isanonymous",
+    "type" : "boolean"
+  }, {
+    "name" : "user",
+    "type" : "string"
+  }, {
+    "name" : "deltabucket",
+    "type" : "double"
+  }, {
+    "name" : "deleted",
+    "type" : "long"
+  }, {
+    "name" : "namespace",
+    "type" : "string"
+  } ]
+}'
+)
+PREHOOK: type: CREATETABLE
+PREHOOK: Output: database:default
+PREHOOK: Output: default@wiki_kafka_avro_table
+POSTHOOK: query: CREATE EXTERNAL TABLE wiki_kafka_avro_table
+STORED BY 'org.apache.hadoop.hive.kafka.KafkaStorageHandler'
+TBLPROPERTIES
+("kafka.topic" = "wiki_kafka_avro_table",
+"kafka.bootstrap.servers"="localhost:9093",
+"kafka.serde.class"="org.apache.hadoop.hive.serde2.avro.AvroSerDe",
+'avro.schema.literal'='{
+  "type" : "record",
+  "name" : "Wikipedia",
+  "namespace" : "org.apache.hive.kafka",
+  "version": "1",
+  "fields" : [ {
+    "name" : "isrobot",
+    "type" : "boolean"
+  }, {
+    "name" : "channel",
+    "type" : "string"
+  }, {
+    "name" : "timestamp",
+    "type" : "string"
+  }, {
+    "name" : "flags",
+    "type" : "string"
+  }, {
+    "name" : "isunpatrolled",
+    "type" : "boolean"
+  }, {
+    "name" : "page",
+    "type" : "string"
+  }, {
+    "name" : "diffurl",
+    "type" : "string"
+  }, {
+    "name" : "added",
+    "type" : "long"
+  }, {
+    "name" : "comment",
+    "type" : "string"
+  }, {
+    "name" : "commentlength",
+    "type" : "long"
+  }, {
+    "name" : "isnew",
+    "type" : "boolean"
+  }, {
+    "name" : "isminor",
+    "type" : "boolean"
+  }, {
+    "name" : "delta",
+    "type" : "long"
+  }, {
+    "name" : "isanonymous",
+    "type" : "boolean"
+  }, {
+    "name" : "user",
+    "type" : "string"
+  }, {
+    "name" : "deltabucket",
+    "type" : "double"
+  }, {
+    "name" : "deleted",
+    "type" : "long"
+  }, {
+    "name" : "namespace",
+    "type" : "string"
+  } ]
+}'
+)
+POSTHOOK: type: CREATETABLE
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@wiki_kafka_avro_table
+PREHOOK: query: describe extended wiki_kafka_avro_table
+PREHOOK: type: DESCTABLE
+PREHOOK: Input: default@wiki_kafka_avro_table
+POSTHOOK: query: describe extended wiki_kafka_avro_table
+POSTHOOK: type: DESCTABLE
+POSTHOOK: Input: default@wiki_kafka_avro_table
+isrobot             	boolean             	from deserializer   
+channel             	string              	from deserializer   
+timestamp           	string              	from deserializer   
+flags               	string              	from deserializer   
+isunpatrolled       	boolean             	from deserializer   
+page                	string              	from deserializer   
+diffurl             	string              	from deserializer   
+added               	bigint              	from deserializer   
+comment             	string              	from deserializer   
+commentlength       	bigint              	from deserializer   
+isnew               	boolean             	from deserializer   
+isminor             	boolean             	from deserializer   
+delta               	bigint              	from deserializer   
+isanonymous         	boolean             	from deserializer   
+user                	string              	from deserializer   
+deltabucket         	double              	from deserializer   
+deleted             	bigint              	from deserializer   
+namespace           	string              	from deserializer   
+__key               	binary              	from deserializer   
+__partition         	int                 	from deserializer   
+__offset            	bigint              	from deserializer   
+__timestamp         	bigint              	from deserializer   
+	 	 
+#### A masked pattern was here ####
+StorageHandlerInfo	 	 
+Partition(topic = wiki_kafka_avro_table, partition = 0, leader = 1, replicas = [1], isr = [1], offlineReplicas = []) [start offset = [0], end offset = [11]]	 	 
+PREHOOK: query: select cast (`__timestamp` as timestamp) as kafka_record_ts, `__partition`, `__offset`, `timestamp`, `user`,
+ `page`, `deleted`, `deltabucket`, `isanonymous`, `commentlength` from wiki_kafka_avro_table
+PREHOOK: type: QUERY
+PREHOOK: Input: default@wiki_kafka_avro_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: select cast (`__timestamp` as timestamp) as kafka_record_ts, `__partition`, `__offset`, `timestamp`, `user`,
+ `page`, `deleted`, `deltabucket`, `isanonymous`, `commentlength` from wiki_kafka_avro_table
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@wiki_kafka_avro_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+2018-08-20 03:37:05.09	0	0	08/19/2018 20:37:05	test-user-0	page is 0	0	0.0	false	0
+2018-08-20 04:37:05.09	0	1	08/19/2018 21:37:05	test-user-1	page is 100	-1	100.4	true	1
+2018-08-20 05:37:05.09	0	2	08/19/2018 22:37:05	test-user-2	page is 200	-2	200.8	true	2
+2018-08-20 06:37:05.09	0	3	08/19/2018 23:37:05	test-user-3	page is 300	-3	301.20000000000005	false	3
+2018-08-20 07:37:05.09	0	4	08/20/2018 00:37:05	test-user-4	page is 400	-4	401.6	true	4
+2018-08-20 08:37:05.09	0	5	08/20/2018 01:37:05	test-user-5	page is 500	-5	502.0	true	5
+2018-08-20 09:37:05.09	0	6	08/20/2018 02:37:05	test-user-6	page is 600	-6	602.4000000000001	false	6
+2018-08-20 10:37:05.09	0	7	08/20/2018 03:37:05	test-user-7	page is 700	-7	702.8000000000001	true	7
+2018-08-20 11:37:05.09	0	8	08/20/2018 04:37:05	test-user-8	page is 800	-8	803.2	true	8
+2018-08-20 12:37:05.09	0	9	08/20/2018 05:37:05	test-user-9	page is 900	-9	903.6	false	9
+2018-08-20 13:37:05.09	0	10	08/20/2018 06:37:05	test-user-10	page is 1000	-10	1004.0	true	10
+PREHOOK: query: select count(*) from wiki_kafka_avro_table
+PREHOOK: type: QUERY
+PREHOOK: Input: default@wiki_kafka_avro_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: select count(*) from wiki_kafka_avro_table
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@wiki_kafka_avro_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+11
+PREHOOK: query: select count(distinct `user`) from  wiki_kafka_avro_table
+PREHOOK: type: QUERY
+PREHOOK: Input: default@wiki_kafka_avro_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: select count(distinct `user`) from  wiki_kafka_avro_table
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@wiki_kafka_avro_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+11
+PREHOOK: query: select sum(deltabucket), min(commentlength) from wiki_kafka_avro_table
+PREHOOK: type: QUERY
+PREHOOK: Input: default@wiki_kafka_avro_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: select sum(deltabucket), min(commentlength) from wiki_kafka_avro_table
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@wiki_kafka_avro_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+5522.000000000001	0
+PREHOOK: query: select cast (`__timestamp` as timestamp) as kafka_record_ts, `__timestamp` as kafka_record_ts_long,
+`__partition`, `__key`, `__offset`, `timestamp`, `user`, `page`, `deleted`, `deltabucket`,
+`isanonymous`, `commentlength` from wiki_kafka_avro_table where `__timestamp` > 1534750625090
+PREHOOK: type: QUERY
+PREHOOK: Input: default@wiki_kafka_avro_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: select cast (`__timestamp` as timestamp) as kafka_record_ts, `__timestamp` as kafka_record_ts_long,
+`__partition`, `__key`, `__offset`, `timestamp`, `user`, `page`, `deleted`, `deltabucket`,
+`isanonymous`, `commentlength` from wiki_kafka_avro_table where `__timestamp` > 1534750625090
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@wiki_kafka_avro_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+2018-08-20 08:37:05.09	1534754225090	0	key-5	5	08/20/2018 01:37:05	test-user-5	page is 500	-5	502.0	true	5
+2018-08-20 09:37:05.09	1534757825090	0	key-6	6	08/20/2018 02:37:05	test-user-6	page is 600	-6	602.4000000000001	false	6
+2018-08-20 10:37:05.09	1534761425090	0	key-7	7	08/20/2018 03:37:05	test-user-7	page is 700	-7	702.8000000000001	true	7
+2018-08-20 11:37:05.09	1534765025090	0	key-8	8	08/20/2018 04:37:05	test-user-8	page is 800	-8	803.2	true	8
+2018-08-20 12:37:05.09	1534768625090	0	key-9	9	08/20/2018 05:37:05	test-user-9	page is 900	-9	903.6	false	9
+2018-08-20 13:37:05.09	1534772225090	0	key-10	10	08/20/2018 06:37:05	test-user-10	page is 1000	-10	1004.0	true	10
+PREHOOK: query: CREATE EXTERNAL TABLE kafka_table_insert
+(c_name string, c_int int, c_float float)
+STORED BY 'org.apache.hadoop.hive.kafka.KafkaStorageHandler'
+WITH SERDEPROPERTIES ("timestamp.formats"="yyyy-MM-dd\'T\'HH:mm:ss\'Z\'")
+TBLPROPERTIES
+("kafka.topic" = "test-topic-write-json",
+"kafka.bootstrap.servers"="localhost:9093",
+"kafka.serde.class"="org.apache.hadoop.hive.serde2.JsonSerDe")
+PREHOOK: type: CREATETABLE
+PREHOOK: Output: database:default
+PREHOOK: Output: default@kafka_table_insert
+POSTHOOK: query: CREATE EXTERNAL TABLE kafka_table_insert
+(c_name string, c_int int, c_float float)
+STORED BY 'org.apache.hadoop.hive.kafka.KafkaStorageHandler'
+WITH SERDEPROPERTIES ("timestamp.formats"="yyyy-MM-dd\'T\'HH:mm:ss\'Z\'")
+TBLPROPERTIES
+("kafka.topic" = "test-topic-write-json",
+"kafka.bootstrap.servers"="localhost:9093",
+"kafka.serde.class"="org.apache.hadoop.hive.serde2.JsonSerDe")
+POSTHOOK: type: CREATETABLE
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@kafka_table_insert
+PREHOOK: query: insert into table kafka_table_insert (c_name,c_int, c_float,`__key`, `__partition`, `__offset`, `__timestamp`)
+values ('test1',5, 4.999,'key',null ,-1,1536449552290)
+PREHOOK: type: QUERY
+PREHOOK: Input: _dummy_database@_dummy_table
+PREHOOK: Output: default@kafka_table_insert
+POSTHOOK: query: insert into table kafka_table_insert (c_name,c_int, c_float,`__key`, `__partition`, `__offset`, `__timestamp`)
+values ('test1',5, 4.999,'key',null ,-1,1536449552290)
+POSTHOOK: type: QUERY
+POSTHOOK: Input: _dummy_database@_dummy_table
+POSTHOOK: Output: default@kafka_table_insert
+PREHOOK: query: insert into table kafka_table_insert (c_name,c_int, c_float, `__key`, `__partition`, `__offset`, `__timestamp`)
+values ('test2',15, 14.9996666, null ,null ,-1,1536449552285)
+PREHOOK: type: QUERY
+PREHOOK: Input: _dummy_database@_dummy_table
+PREHOOK: Output: default@kafka_table_insert
+POSTHOOK: query: insert into table kafka_table_insert (c_name,c_int, c_float, `__key`, `__partition`, `__offset`, `__timestamp`)
+values ('test2',15, 14.9996666, null ,null ,-1,1536449552285)
+POSTHOOK: type: QUERY
+POSTHOOK: Input: _dummy_database@_dummy_table
+POSTHOOK: Output: default@kafka_table_insert
+PREHOOK: query: select * from kafka_table_insert
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table_insert
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: select * from kafka_table_insert
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table_insert
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+test1	5	4.999	key	0	0	1536449552290
+test2	15	14.999666	NULL	0	1	1536449552285
+PREHOOK: query: insert into table wiki_kafka_avro_table select
+isrobot as isrobot, channel as channel,`timestamp` as `timestamp`,  flags as flags,  isunpatrolled as isunpatrolled, page as page,
+diffurl as diffurl, added as added, comment as comment, commentlength as commentlength, isnew as isnew, isminor as isminor,
+delta as delta, isanonymous as isanonymous, `user` as `user`,  deltabucket as detlabucket, deleted as deleted, namespace as namespace,
+`__key`, `__partition`, -1 as `__offset`,`__timestamp`
+from wiki_kafka_avro_table
+PREHOOK: type: QUERY
+PREHOOK: Input: default@wiki_kafka_avro_table
+PREHOOK: Output: default@wiki_kafka_avro_table
+POSTHOOK: query: insert into table wiki_kafka_avro_table select
+isrobot as isrobot, channel as channel,`timestamp` as `timestamp`,  flags as flags,  isunpatrolled as isunpatrolled, page as page,
+diffurl as diffurl, added as added, comment as comment, commentlength as commentlength, isnew as isnew, isminor as isminor,
+delta as delta, isanonymous as isanonymous, `user` as `user`,  deltabucket as detlabucket, deleted as deleted, namespace as namespace,
+`__key`, `__partition`, -1 as `__offset`,`__timestamp`
+from wiki_kafka_avro_table
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@wiki_kafka_avro_table
+POSTHOOK: Output: default@wiki_kafka_avro_table
+PREHOOK: query: select cast ((`__timestamp`/1000) as timestamp) as kafka_record_ts, `__partition`, `__offset`, `timestamp`, `user`,
+`page`, `deleted`, `deltabucket`, `isanonymous`, `commentlength` from wiki_kafka_avro_table
+PREHOOK: type: QUERY
+PREHOOK: Input: default@wiki_kafka_avro_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: select cast ((`__timestamp`/1000) as timestamp) as kafka_record_ts, `__partition`, `__offset`, `timestamp`, `user`,
+`page`, `deleted`, `deltabucket`, `isanonymous`, `commentlength` from wiki_kafka_avro_table
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@wiki_kafka_avro_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+2018-08-20 03:37:05.09	0	0	08/19/2018 20:37:05	test-user-0	page is 0	0	0.0	false	0
+2018-08-20 04:37:05.09	0	1	08/19/2018 21:37:05	test-user-1	page is 100	-1	100.4	true	1
+2018-08-20 05:37:05.09	0	2	08/19/2018 22:37:05	test-user-2	page is 200	-2	200.8	true	2
+2018-08-20 06:37:05.09	0	3	08/19/2018 23:37:05	test-user-3	page is 300	-3	301.20000000000005	false	3
+2018-08-20 07:37:05.09	0	4	08/20/2018 00:37:05	test-user-4	page is 400	-4	401.6	true	4
+2018-08-20 08:37:05.09	0	5	08/20/2018 01:37:05	test-user-5	page is 500	-5	502.0	true	5
+2018-08-20 09:37:05.09	0	6	08/20/2018 02:37:05	test-user-6	page is 600	-6	602.4000000000001	false	6
+2018-08-20 10:37:05.09	0	7	08/20/2018 03:37:05	test-user-7	page is 700	-7	702.8000000000001	true	7
+2018-08-20 11:37:05.09	0	8	08/20/2018 04:37:05	test-user-8	page is 800	-8	803.2	true	8
+2018-08-20 12:37:05.09	0	9	08/20/2018 05:37:05	test-user-9	page is 900	-9	903.6	false	9
+2018-08-20 13:37:05.09	0	10	08/20/2018 06:37:05	test-user-10	page is 1000	-10	1004.0	true	10
+2018-08-20 03:37:05.09	0	11	08/19/2018 20:37:05	test-user-0	page is 0	0	0.0	false	0
+2018-08-20 04:37:05.09	0	12	08/19/2018 21:37:05	test-user-1	page is 100	-1	100.4	true	1
+2018-08-20 05:37:05.09	0	13	08/19/2018 22:37:05	test-user-2	page is 200	-2	200.8	true	2
+2018-08-20 06:37:05.09	0	14	08/19/2018 23:37:05	test-user-3	page is 300	-3	301.20000000000005	false	3
+2018-08-20 07:37:05.09	0	15	08/20/2018 00:37:05	test-user-4	page is 400	-4	401.6	true	4
+2018-08-20 08:37:05.09	0	16	08/20/2018 01:37:05	test-user-5	page is 500	-5	502.0	true	5
+2018-08-20 09:37:05.09	0	17	08/20/2018 02:37:05	test-user-6	page is 600	-6	602.4000000000001	false	6
+2018-08-20 10:37:05.09	0	18	08/20/2018 03:37:05	test-user-7	page is 700	-7	702.8000000000001	true	7
+2018-08-20 11:37:05.09	0	19	08/20/2018 04:37:05	test-user-8	page is 800	-8	803.2	true	8
+2018-08-20 12:37:05.09	0	20	08/20/2018 05:37:05	test-user-9	page is 900	-9	903.6	false	9
+2018-08-20 13:37:05.09	0	21	08/20/2018 06:37:05	test-user-10	page is 1000	-10	1004.0	true	10
+PREHOOK: query: select `__key`, count(1)  FROM wiki_kafka_avro_table group by `__key` order by `__key`
+PREHOOK: type: QUERY
+PREHOOK: Input: default@wiki_kafka_avro_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: select `__key`, count(1)  FROM wiki_kafka_avro_table group by `__key` order by `__key`
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@wiki_kafka_avro_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+key-0	2
+key-1	2
+key-10	2
+key-2	2
+key-3	2
+key-4	2
+key-5	2
+key-6	2
+key-7	2
+key-8	2
+key-9	2
+PREHOOK: query: select `__timestamp`, count(1) from wiki_kafka_avro_table group by `__timestamp` order by `__timestamp`
+PREHOOK: type: QUERY
+PREHOOK: Input: default@wiki_kafka_avro_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: select `__timestamp`, count(1) from wiki_kafka_avro_table group by `__timestamp` order by `__timestamp`
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@wiki_kafka_avro_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+1534736225090	2
+1534739825090	2
+1534743425090	2
+1534747025090	2
+1534750625090	2
+1534754225090	2
+1534757825090	2
+1534761425090	2
+1534765025090	2
+1534768625090	2
+1534772225090	2
+PREHOOK: query: CREATE EXTERNAL TABLE kafka_table_csv
+(c_name string, c_int int, c_float float)
+STORED BY 'org.apache.hadoop.hive.kafka.KafkaStorageHandler'
+TBLPROPERTIES
+("kafka.topic" = "test-topic-write-csv",
+"kafka.bootstrap.servers"="localhost:9093",
+"kafka.serde.class"="org.apache.hadoop.hive.serde2.OpenCSVSerde")
+PREHOOK: type: CREATETABLE
+PREHOOK: Output: database:default
+PREHOOK: Output: default@kafka_table_csv
+POSTHOOK: query: CREATE EXTERNAL TABLE kafka_table_csv
+(c_name string, c_int int, c_float float)
+STORED BY 'org.apache.hadoop.hive.kafka.KafkaStorageHandler'
+TBLPROPERTIES
+("kafka.topic" = "test-topic-write-csv",
+"kafka.bootstrap.servers"="localhost:9093",
+"kafka.serde.class"="org.apache.hadoop.hive.serde2.OpenCSVSerde")
+POSTHOOK: type: CREATETABLE
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@kafka_table_csv
+PREHOOK: query: ALTER TABLE kafka_table_csv SET TBLPROPERTIES ("hive.kafka.optimistic.commit"="false", "kafka.write.semantic"="EXACTLY_ONCE")
+PREHOOK: type: ALTERTABLE_PROPERTIES
+PREHOOK: Input: default@kafka_table_csv
+PREHOOK: Output: default@kafka_table_csv
+POSTHOOK: query: ALTER TABLE kafka_table_csv SET TBLPROPERTIES ("hive.kafka.optimistic.commit"="false", "kafka.write.semantic"="EXACTLY_ONCE")
+POSTHOOK: type: ALTERTABLE_PROPERTIES
+POSTHOOK: Input: default@kafka_table_csv
+POSTHOOK: Output: default@kafka_table_csv
+PREHOOK: query: insert into table kafka_table_csv select c_name, c_int, c_float, `__key`, `__partition`, -1 as `__offset`, `__timestamp` from kafka_table_insert
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table_insert
+PREHOOK: Output: default@kafka_table_csv
+POSTHOOK: query: insert into table kafka_table_csv select c_name, c_int, c_float, `__key`, `__partition`, -1 as `__offset`, `__timestamp` from kafka_table_insert
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table_insert
+POSTHOOK: Output: default@kafka_table_csv
+PREHOOK: query: insert into table kafka_table_csv (c_name,c_int, c_float,`__key`, `__partition`, `__offset`, `__timestamp`)
+values ('test4',-5, -4.999,'key-2',null ,-1,1536449552291)
+PREHOOK: type: QUERY
+PREHOOK: Input: _dummy_database@_dummy_table
+PREHOOK: Output: default@kafka_table_csv
+POSTHOOK: query: insert into table kafka_table_csv (c_name,c_int, c_float,`__key`, `__partition`, `__offset`, `__timestamp`)
+values ('test4',-5, -4.999,'key-2',null ,-1,1536449552291)
+POSTHOOK: type: QUERY
+POSTHOOK: Input: _dummy_database@_dummy_table
+POSTHOOK: Output: default@kafka_table_csv
+PREHOOK: query: insert into table kafka_table_csv (c_name,c_int, c_float, `__key`, `__partition`, `__offset`, `__timestamp`)
+values ('test5',-15, -14.9996666, 'key-3' ,null ,-1,1536449552284)
+PREHOOK: type: QUERY
+PREHOOK: Input: _dummy_database@_dummy_table
+PREHOOK: Output: default@kafka_table_csv
+POSTHOOK: query: insert into table kafka_table_csv (c_name,c_int, c_float, `__key`, `__partition`, `__offset`, `__timestamp`)
+values ('test5',-15, -14.9996666, 'key-3' ,null ,-1,1536449552284)
+POSTHOOK: type: QUERY
+POSTHOOK: Input: _dummy_database@_dummy_table
+POSTHOOK: Output: default@kafka_table_csv
+PREHOOK: query: select * from kafka_table_csv
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table_csv
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: select * from kafka_table_csv
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table_csv
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+test1	5	4.999	key	0	0	1536449552290
+test2	15	14.999666	NULL	0	1	1536449552285
+test4	-5	-4.999	key-2	0	3	1536449552291
+test5	-15	-14.9996666	key-3	0	5	1536449552284
+PREHOOK: query: select distinct `__key`, c_name from kafka_table_csv
+PREHOOK: type: QUERY
+PREHOOK: Input: default@kafka_table_csv
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: select distinct `__key`, c_name from kafka_table_csv
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@kafka_table_csv
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+key	test1
+key-2	test4
+key-3	test5
+NULL	test2
+PREHOOK: query: explain extended select distinct `__offset`, cast(`__timestamp` as timestamp ) , `__key` from wiki_kafka_avro_table
+PREHOOK: type: QUERY
+PREHOOK: Input: default@wiki_kafka_avro_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: explain extended select distinct `__offset`, cast(`__timestamp` as timestamp ) , `__key` from wiki_kafka_avro_table
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@wiki_kafka_avro_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+OPTIMIZED SQL: SELECT `__offset` AS `$f0`, CAST(`__timestamp` AS TIMESTAMP(9)) AS `$f1`, `__key` AS `$f2`
+FROM `default`.`wiki_kafka_avro_table`
+GROUP BY `__offset`, CAST(`__timestamp` AS TIMESTAMP(9)), `__key`
+STAGE DEPENDENCIES:
+  Stage-1 is a root stage
+  Stage-0 depends on stages: Stage-1
+
+STAGE PLANS:
+  Stage: Stage-1
+    Tez
+#### A masked pattern was here ####
+      Edges:
+        Reducer 2 <- Map 1 (SIMPLE_EDGE)
+#### A masked pattern was here ####
+      Vertices:
+        Map 1 
+            Map Operator Tree:
+                TableScan
+                  alias: wiki_kafka_avro_table
+                  Statistics: Num rows: 1 Data size: 160 Basic stats: COMPLETE Column stats: NONE
+                  GatherStats: false
+                  Select Operator
+                    expressions: __offset (type: bigint), CAST( __timestamp AS TIMESTAMP) (type: timestamp), __key (type: binary)
+                    outputColumnNames: _col0, _col1, _col2
+                    Statistics: Num rows: 1 Data size: 160 Basic stats: COMPLETE Column stats: NONE
+                    Group By Operator
+                      keys: _col0 (type: bigint), _col1 (type: timestamp), _col2 (type: binary)
+                      mode: hash
+                      outputColumnNames: _col0, _col1, _col2
+                      Statistics: Num rows: 1 Data size: 160 Basic stats: COMPLETE Column stats: NONE
+                      Reduce Output Operator
+                        key expressions: _col0 (type: bigint), _col1 (type: timestamp), _col2 (type: binary)
+                        null sort order: aaa
+                        sort order: +++
+                        Map-reduce partition columns: _col0 (type: bigint), _col1 (type: timestamp), _col2 (type: binary)
+                        Statistics: Num rows: 1 Data size: 160 Basic stats: COMPLETE Column stats: NONE
+                        tag: -1
+                        auto parallelism: true
+            Execution mode: llap
+            LLAP IO: no inputs
+            Path -> Alias:
+              hdfs://### HDFS PATH ### [wiki_kafka_avro_table]
+            Path -> Partition:
+              hdfs://### HDFS PATH ### 
+                Partition
+                  base file name: wiki_kafka_avro_table
+                  input format: org.apache.hadoop.hive.kafka.KafkaInputFormat
+                  output format: org.apache.hadoop.hive.kafka.KafkaOutputFormat
+                  properties:
+                    EXTERNAL TRUE
+                    avro.schema.literal {
+  "type" : "record",
+  "name" : "Wikipedia",
+  "namespace" : "org.apache.hive.kafka",
+  "version": "1",
+  "fields" : [ {
+    "name" : "isrobot",
+    "type" : "boolean"
+  }, {
+    "name" : "channel",
+    "type" : "string"
+  }, {
+    "name" : "timestamp",
+    "type" : "string"
+  }, {
+    "name" : "flags",
+    "type" : "string"
+  }, {
+    "name" : "isunpatrolled",
+    "type" : "boolean"
+  }, {
+    "name" : "page",
+    "type" : "string"
+  }, {
+    "name" : "diffurl",
+    "type" : "string"
+  }, {
+    "name" : "added",
+    "type" : "long"
+  }, {
+    "name" : "comment",
+    "type" : "string"
+  }, {
+    "name" : "commentlength",
+    "type" : "long"
+  }, {
+    "name" : "isnew",
+    "type" : "boolean"
+  }, {
+    "name" : "isminor",
+    "type" : "boolean"
+  }, {
+    "name" : "delta",
+    "type" : "long"
+  }, {
+    "name" : "isanonymous",
+    "type" : "boolean"
+  }, {
+    "name" : "user",
+    "type" : "string"
+  }, {
+    "name" : "deltabucket",
+    "type" : "double"
+  }, {
+    "name" : "deleted",
+    "type" : "long"
+  }, {
+    "name" : "namespace",
+    "type" : "string"
+  } ]
+}
+                    bucket_count -1
+                    bucketing_version 2
+                    column.name.delimiter ,
+                    columns isrobot,channel,timestamp,flags,isunpatrolled,page,diffurl,added,comment,commentlength,isnew,isminor,delta,isanonymous,user,deltabucket,deleted,namespace,__key,__partition,__offset,__timestamp
+                    columns.comments 'from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer'
+                    columns.types boolean:string:string:string:boolean:string:string:bigint:string:bigint:boolean:boolean:bigint:boolean:string:double:bigint:string:binary:int:bigint:bigint
+                    discover.partitions true
+#### A masked pattern was here ####
+                    hive.kafka.max.retries 6
+                    hive.kafka.metadata.poll.timeout.ms 30000
+                    hive.kafka.optimistic.commit false
+                    hive.kafka.poll.timeout.ms 5000
+                    kafka.bootstrap.servers localhost:9093
+                    kafka.serde.class org.apache.hadoop.hive.serde2.avro.AvroSerDe
+                    kafka.topic wiki_kafka_avro_table
+                    kafka.write.semantic AT_LEAST_ONCE
+#### A masked pattern was here ####
+                    location hdfs://### HDFS PATH ###
+                    name default.wiki_kafka_avro_table
+                    numFiles 0
+                    numRows 0
+                    rawDataSize 0
+                    serialization.ddl struct wiki_kafka_avro_table { bool isrobot, string channel, string timestamp, string flags, bool isunpatrolled, string page, string diffurl, i64 added, string comment, i64 commentlength, bool isnew, bool isminor, i64 delta, bool isanonymous, string user, double deltabucket, i64 deleted, string namespace, binary __key, i32 __partition, i64 __offset, i64 __timestamp}
+                    serialization.format 1
+                    serialization.lib org.apache.hadoop.hive.kafka.KafkaSerDe
+                    storage_handler org.apache.hadoop.hive.kafka.KafkaStorageHandler
+                    totalSize 0
+#### A masked pattern was here ####
+                  serde: org.apache.hadoop.hive.kafka.KafkaSerDe
+                
+                    input format: org.apache.hadoop.hive.kafka.KafkaInputFormat
+                    jobProperties:
+                      hive.kafka.max.retries 6
+                      hive.kafka.metadata.poll.timeout.ms 30000
+                      hive.kafka.optimistic.commit false
+                      hive.kafka.poll.timeout.ms 5000
+                      kafka.bootstrap.servers localhost:9093
+                      kafka.serde.class org.apache.hadoop.hive.serde2.avro.AvroSerDe
+                      kafka.topic wiki_kafka_avro_table
+                      kafka.write.semantic AT_LEAST_ONCE
+                    output format: org.apache.hadoop.hive.kafka.KafkaOutputFormat
+                    properties:
+                      EXTERNAL TRUE
+                      avro.schema.literal {
+  "type" : "record",
+  "name" : "Wikipedia",
+  "namespace" : "org.apache.hive.kafka",
+  "version": "1",
+  "fields" : [ {
+    "name" : "isrobot",
+    "type" : "boolean"
+  }, {
+    "name" : "channel",
+    "type" : "string"
+  }, {
+    "name" : "timestamp",
+    "type" : "string"
+  }, {
+    "name" : "flags",
+    "type" : "string"
+  }, {
+    "name" : "isunpatrolled",
+    "type" : "boolean"
+  }, {
+    "name" : "page",
+    "type" : "string"
+  }, {
+    "name" : "diffurl",
+    "type" : "string"
+  }, {
+    "name" : "added",
+    "type" : "long"
+  }, {
+    "name" : "comment",
+    "type" : "string"
+  }, {
+    "name" : "commentlength",
+    "type" : "long"
+  }, {
+    "name" : "isnew",
+    "type" : "boolean"
+  }, {
+    "name" : "isminor",
+    "type" : "boolean"
+  }, {
+    "name" : "delta",
+    "type" : "long"
+  }, {
+    "name" : "isanonymous",
+    "type" : "boolean"
+  }, {
+    "name" : "user",
+    "type" : "string"
+  }, {
+    "name" : "deltabucket",
+    "type" : "double"
+  }, {
+    "name" : "deleted",
+    "type" : "long"
+  }, {
+    "name" : "namespace",
+    "type" : "string"
+  } ]
+}
+                      bucket_count -1
+                      bucketing_version 2
+                      column.name.delimiter ,
+                      columns isrobot,channel,timestamp,flags,isunpatrolled,page,diffurl,added,comment,commentlength,isnew,isminor,delta,isanonymous,user,deltabucket,deleted,namespace,__key,__partition,__offset,__timestamp
+                      columns.comments 'from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer'
+                      columns.types boolean:string:string:string:boolean:string:string:bigint:string:bigint:boolean:boolean:bigint:boolean:string:double:bigint:string:binary:int:bigint:bigint
+                      discover.partitions true
+#### A masked pattern was here ####
+                      hive.kafka.max.retries 6
+                      hive.kafka.metadata.poll.timeout.ms 30000
+                      hive.kafka.optimistic.commit false
+                      hive.kafka.poll.timeout.ms 5000
+                      kafka.bootstrap.servers localhost:9093
+                      kafka.serde.class org.apache.hadoop.hive.serde2.avro.AvroSerDe
+                      kafka.topic wiki_kafka_avro_table
+                      kafka.write.semantic AT_LEAST_ONCE
+#### A masked pattern was here ####
+                      location hdfs://### HDFS PATH ###
+                      name default.wiki_kafka_avro_table
+                      numFiles 0
+                      numRows 0
+                      rawDataSize 0
+                      serialization.ddl struct wiki_kafka_avro_table { bool isrobot, string channel, string timestamp, string flags, bool isunpatrolled, string page, string diffurl, i64 added, string comment, i64 commentlength, bool isnew, bool isminor, i64 delta, bool isanonymous, string user, double deltabucket, i64 deleted, string namespace, binary __key, i32 __partition, i64 __offset, i64 __timestamp}
+                      serialization.format 1
+                      serialization.lib org.apache.hadoop.hive.kafka.KafkaSerDe
+                      storage_handler org.apache.hadoop.hive.kafka.KafkaStorageHandler
+                      totalSize 0
+#### A masked pattern was here ####
+                    serde: org.apache.hadoop.hive.kafka.KafkaSerDe
+                    name: default.wiki_kafka_avro_table
+                  name: default.wiki_kafka_avro_table
+            Truncated Path -> Alias:
+              /wiki_kafka_avro_table [wiki_kafka_avro_table]
+        Reducer 2 
+            Execution mode: llap
+            Needs Tagging: false
+            Reduce Operator Tree:
+              Group By Operator
+                keys: KEY._col0 (type: bigint), KEY._col1 (type: timestamp), KEY._col2 (type: binary)
+                mode: mergepartial
+                outputColumnNames: _col0, _col1, _col2
+                Statistics: Num rows: 1 Data size: 160 Basic stats: COMPLETE Column stats: NONE
+                File Output Operator
+                  compressed: false
+                  GlobalTableId: 0
+                  directory: hdfs://### HDFS PATH ###
+                  NumFilesPerFileSink: 1
+                  Statistics: Num rows: 1 Data size: 160 Basic stats: COMPLETE Column stats: NONE
+                  Stats Publishing Key Prefix: hdfs://### HDFS PATH ###
+                  table:
+                      input format: org.apache.hadoop.mapred.SequenceFileInputFormat
+                      output format: org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat
+                      properties:
+                        columns _col0,_col1,_col2
+                        columns.types bigint:timestamp:binary
+                        escape.delim \
+                        hive.serialization.extend.additional.nesting.levels true
+                        serialization.escape.crlf true
+                        serialization.format 1
+                        serialization.lib org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
+                      serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
+                  TotalFiles: 1
+                  GatherStats: false
+                  MultiFileSpray: false
+
+  Stage: Stage-0
+    Fetch Operator
+      limit: -1
+      Processor Tree:
+        ListSink
+
+PREHOOK: query: select distinct `__offset`, cast(`__timestamp` as timestamp ) , `__key` from wiki_kafka_avro_table
+PREHOOK: type: QUERY
+PREHOOK: Input: default@wiki_kafka_avro_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: select distinct `__offset`, cast(`__timestamp` as timestamp ) , `__key` from wiki_kafka_avro_table
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@wiki_kafka_avro_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+0	2018-08-20 03:37:05.09	key-0
+1	2018-08-20 04:37:05.09	key-1
+3	2018-08-20 06:37:05.09	key-3
+4	2018-08-20 07:37:05.09	key-4
+5	2018-08-20 08:37:05.09	key-5
+7	2018-08-20 10:37:05.09	key-7
+8	2018-08-20 11:37:05.09	key-8
+9	2018-08-20 12:37:05.09	key-9
+10	2018-08-20 13:37:05.09	key-10
+11	2018-08-20 03:37:05.09	key-0
+13	2018-08-20 05:37:05.09	key-2
+15	2018-08-20 07:37:05.09	key-4
+17	2018-08-20 09:37:05.09	key-6
+19	2018-08-20 11:37:05.09	key-8
+21	2018-08-20 13:37:05.09	key-10
+2	2018-08-20 05:37:05.09	key-2
+6	2018-08-20 09:37:05.09	key-6
+12	2018-08-20 04:37:05.09	key-1
+14	2018-08-20 06:37:05.09	key-3
+16	2018-08-20 08:37:05.09	key-5
+18	2018-08-20 10:37:05.09	key-7
+20	2018-08-20 12:37:05.09	key-9
+PREHOOK: query: explain extended select distinct `__offset`, cast(`__timestamp` as timestamp ) , `__key` from wiki_kafka_avro_table
+PREHOOK: type: QUERY
+PREHOOK: Input: default@wiki_kafka_avro_table
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: explain extended select distinct `__offset`, cast(`__timestamp` as timestamp ) , `__key` from wiki_kafka_avro_table
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@wiki_kafka_avro_table
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+OPTIMIZED SQL: SELECT `__offset` AS `$f0`, CAST(`__timestamp` AS TIMESTAMP(9)) AS `$f1`, `__key` AS `$f2`
+FROM `default`.`wiki_kafka_avro_table`
+GROUP BY `__offset`, CAST(`__timestamp` AS TIMESTAMP(9)), `__key`
+STAGE DEPENDENCIES:
+  Stage-1 is a root stage
+  Stage-0 depends on stages: Stage-1
+
+STAGE PLANS:
+  Stage: Stage-1
+    Tez
+#### A masked pattern was here ####
+      Edges:
+        Reducer 2 <- Map 1 (SIMPLE_EDGE)
+#### A masked pattern was here ####
+      Vertices:
+        Map 1 
+            Map Operator Tree:
+                TableScan
+                  alias: wiki_kafka_avro_table
+                  Statistics: Num rows: 1 Data size: 160 Basic stats: COMPLETE Column stats: NONE
+                  GatherStats: false
+                  Select Operator
+                    expressions: __offset (type: bigint), CAST( __timestamp AS TIMESTAMP) (type: timestamp), __key (type: binary)
+                    outputColumnNames: _col0, _col1, _col2
+                    Statistics: Num rows: 1 Data size: 160 Basic stats: COMPLETE Column stats: NONE
+                    Group By Operator
+                      keys: _col0 (type: bigint), _col1 (type: timestamp), _col2 (type: binary)
+                      mode: hash
+                      outputColumnNames: _col0, _col1, _col2
+                      Statistics: Num rows: 1 Data size: 160 Basic stats: COMPLETE Column stats: NONE
+                      Reduce Output Operator
+                        key expressions: _col0 (type: bigint), _col1 (type: timestamp), _col2 (type: binary)
+                        null sort order: aaa
+                        sort order: +++
+                        Map-reduce partition columns: _col0 (type: bigint), _col1 (type: timestamp), _col2 (type: binary)
+                        Statistics: Num rows: 1 Data size: 160 Basic stats: COMPLETE Column stats: NONE
+                        tag: -1
+                        auto parallelism: true
+            Execution mode: vectorized, llap
+            LLAP IO: no inputs
+            Path -> Alias:
+              hdfs://### HDFS PATH ### [wiki_kafka_avro_table]
+            Path -> Partition:
+              hdfs://### HDFS PATH ### 
+                Partition
+                  base file name: wiki_kafka_avro_table
+                  input format: org.apache.hadoop.hive.kafka.KafkaInputFormat
+                  output format: org.apache.hadoop.hive.kafka.KafkaOutputFormat
+                  properties:
+                    EXTERNAL TRUE
+                    avro.schema.literal {
+  "type" : "record",
+  "name" : "Wikipedia",
+  "namespace" : "org.apache.hive.kafka",
+  "version": "1",
+  "fields" : [ {
+    "name" : "isrobot",
+    "type" : "boolean"
+  }, {
+    "name" : "channel",
+    "type" : "string"
+  }, {
+    "name" : "timestamp",
+    "type" : "string"
+  }, {
+    "name" : "flags",
+    "type" : "string"
+  }, {
+    "name" : "isunpatrolled",
+    "type" : "boolean"
+  }, {
+    "name" : "page",
+    "type" : "string"
+  }, {
+    "name" : "diffurl",
+    "type" : "string"
+  }, {
+    "name" : "added",
+    "type" : "long"
+  }, {
+    "name" : "comment",
+    "type" : "string"
+  }, {
+    "name" : "commentlength",
+    "type" : "long"
+  }, {
+    "name" : "isnew",
+    "type" : "boolean"
+  }, {
+    "name" : "isminor",
+    "type" : "boolean"
+  }, {
+    "name" : "delta",
+    "type" : "long"
+  }, {
+    "name" : "isanonymous",
+    "type" : "boolean"
+  }, {
+    "name" : "user",
+    "type" : "string"
+  }, {
+    "name" : "deltabucket",
+    "type" : "double"
+  }, {
+    "name" : "deleted",
+    "type" : "long"
+  }, {
+    "name" : "namespace",
+    "type" : "string"
+  } ]
+}
+                    bucket_count -1
+                    bucketing_version 2
+                    column.name.delimiter ,
+                    columns isrobot,channel,timestamp,flags,isunpatrolled,page,diffurl,added,comment,commentlength,isnew,isminor,delta,isanonymous,user,deltabucket,deleted,namespace,__key,__partition,__offset,__timestamp
+                    columns.comments 'from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer'
+                    columns.types boolean:string:string:string:boolean:string:string:bigint:string:bigint:boolean:boolean:bigint:boolean:string:double:bigint:string:binary:int:bigint:bigint
+                    discover.partitions true
+#### A masked pattern was here ####
+                    hive.kafka.max.retries 6
+                    hive.kafka.metadata.poll.timeout.ms 30000
+                    hive.kafka.optimistic.commit false
+                    hive.kafka.poll.timeout.ms 5000
+                    kafka.bootstrap.servers localhost:9093
+                    kafka.serde.class org.apache.hadoop.hive.serde2.avro.AvroSerDe
+                    kafka.topic wiki_kafka_avro_table
+                    kafka.write.semantic AT_LEAST_ONCE
+#### A masked pattern was here ####
+                    location hdfs://### HDFS PATH ###
+                    name default.wiki_kafka_avro_table
+                    numFiles 0
+                    numRows 0
+                    rawDataSize 0
+                    serialization.ddl struct wiki_kafka_avro_table { bool isrobot, string channel, string timestamp, string flags, bool isunpatrolled, string page, string diffurl, i64 added, string comment, i64 commentlength, bool isnew, bool isminor, i64 delta, bool isanonymous, string user, double deltabucket, i64 deleted, string namespace, binary __key, i32 __partition, i64 __offset, i64 __timestamp}
+                    serialization.format 1
+                    serialization.lib org.apache.hadoop.hive.kafka.KafkaSerDe
+                    storage_handler org.apache.hadoop.hive.kafka.KafkaStorageHandler
+                    totalSize 0
+#### A masked pattern was here ####
+                  serde: org.apache.hadoop.hive.kafka.KafkaSerDe
+                
+                    input format: org.apache.hadoop.hive.kafka.KafkaInputFormat
+                    jobProperties:
+                      hive.kafka.max.retries 6
+                      hive.kafka.metadata.poll.timeout.ms 30000
+                      hive.kafka.optimistic.commit false
+                      hive.kafka.poll.timeout.ms 5000
+                      kafka.bootstrap.servers localhost:9093
+                      kafka.serde.class org.apache.hadoop.hive.serde2.avro.AvroSerDe
+                      kafka.topic wiki_kafka_avro_table
+                      kafka.write.semantic AT_LEAST_ONCE
+                    output format: org.apache.hadoop.hive.kafka.KafkaOutputFormat
+                    properties:
+                      EXTERNAL TRUE
+                      avro.schema.literal {
+  "type" : "record",
+  "name" : "Wikipedia",
+  "namespace" : "org.apache.hive.kafka",
+  "version": "1",
+  "fields" : [ {
+    "name" : "isrobot",
+    "type" : "boolean"
+  }, {
+    "name" : "channel",
+    "type" : "string"
+  }, {
+    "name" : "timestamp",
+    "type" : "string"
+  }, {
+    "name" : "flags",
+    "type" : "string"
+  }, {
+    "name" : "isunpatrolled",
+    "type" : "boolean"
+  }, {
+    "name" : "page",
+    "type" : "string"
+  }, {
+    "name" : "diffurl",
+    "type" : "string"
+  }, {
+    "name" : "added",
+    "type" : "long"
+  }, {
+    "name" : "comment",
+    "type" : "string"
+  }, {
+    "name" : "commentlength",
+    "type" : "long"
+  }, {
+    "name" : "isnew",
+    "type" : "boolean"
+  }, {
+    "name" : "isminor",
+    "type" : "boolean"
+  }, {
+    "name" : "delta",
+    "type" : "long"
+  }, {
+    "name" : "isanonymous",
+    "type" : "boolean"
+  }, {
+    "name" : "user",
+    "type" : "string"
+  }, {
+    "name" : "deltabucket",
+    "type" : "double"
+  }, {
+    "name" : "deleted",
+    "type" : "long"
+  }, {
+    "name" : "namespace",
+    "type" : "string"
+  } ]
+}
+                      bucket_count -1
+                      bucketing_version 2
+                      column.name.delimiter ,
+                      columns isrobot,channel,timestamp,flags,isunpatrolled,page,diffurl,added,comment,commentlength,isnew,isminor,delta,isanonymous,user,deltabucket,deleted,namespace,__key,__partition,__offset,__timestamp
+                      columns.comments 'from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer','from deserializer'
+                      columns.types boolean:string:string:string:boolean:string:string:bigint:string:bigint:boolean:boolean:bigint:boolean:string:double:bigint:string:binary:int:bigint:bigint
+                      discover.partitions true
+#### A masked pattern was here ####
+                      hive.kafka.max.retries 6
+                      hive.kafka.metadata.poll.timeout.ms 30000
+                      hive.kafka.optimistic.commit false
+                      hive.kafka.poll.timeout.ms 5000
+                      kafka.bootstrap.servers localhost:9093
+                      kafka.serde.class org.apache.hadoop.hive.serde2.avro.AvroSerDe
+                      kafka.topic wiki_kafka_avro_table
+                      kafka.write.semantic AT_LEAST_ONCE
+#### A masked pattern was here ####
+                      location hdfs://### HDFS PATH ###
+                      name default.wiki_kafka_avro_table
+                      numFiles 0
+                      numRows 0
+                      rawDataSize 0
+                      serialization.ddl struct wiki_kafka_avro_table { bool isrobot, string channel, string timestamp, string flags, bool isunpatrolled, string page, string diffurl, i64 added, string comment, i64 commentlength, bool isnew, bool isminor, i64 delta, bool isanonymous, string user, double deltabucket, i64 deleted, string namespace, binary __key, i32 __partition, i64 __offset, i64 __timestamp}
+                      serialization.format 1
+                      serialization.lib org.apache.hadoop.hive.kafka.KafkaSerDe
+                      storage_handler org.apache.hadoop.hive.kafka.KafkaStorageHandler
+                      totalSize 0
+#### A masked pattern was here ####
+                    serde: org.apache.hadoop.hive.kafka.KafkaSerDe
+                    name: default.wiki_kafka_avro_table
+                  name: default.wiki_kafka_avro_table
+            Truncated Path -> Alias:
+              /wiki_kafka_avro_table [wiki_kafka_avro_table]
+        Reducer 2 
+            Execution mode: vectorized, llap
+            Needs Tagging: false
+            Reduce Operator Tree:
+              Group By Operator
+                keys: KEY._col0 (type: bigint), KEY._col1 (type: timestamp), KEY._col2 (type: binary)
+                mode: mergepartial
+                outputColumnNames: _col0, _col1, _col2
+                Statistics: Num rows: 1 Data size: 160 Basic stats: COMPLETE Column stats: NONE
+                File Output Operator
+                  compressed: false
+                  GlobalTableId: 0
+                  directory: hdfs://### HDFS PATH ###
+                  NumFilesPerFileSink: 1
+                  Statistics: Num rows: 1 Data size: 160 Basic stats: COMPLETE Column stats: NONE
+                  Stats Publishing Key Prefix: hdfs://### HDFS PATH ###
+                  table:
+                      input format: org.apache.hadoop.mapred.SequenceFileInputFormat
+                      output format: org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat
+                      properties:
+                        columns _col0,_col1,_col2
+                        columns.types bigint:timestamp:binary
+                        escape.delim

<TRUNCATED>