insert into sink1 from select a,count(1) from source group by a;