title: Exception in Flink TableAPI
body: <p>I am trying to run below simple Flink job to count words using TableAPI. Used DataStream API to read a stream of data and used StreamTableEnvironment API to create a Table environment. I am getting below exception. Can someone please help me what is wrong in code? I am using Flink 1.8 version.</p>
  <p><strong>Exception:</strong></p>
  <pre><code>**Exception in thread &quot;main&quot; org.apache.flink.table.api.TableException: Only the first field can reference an atomic type.**
      at org.apache.flink.table.api.TableEnvironment$$anonfun$5.apply(TableEnvironment.scala:1117)
      at org.apache.flink.table.api.TableEnvironment$$anonfun$5.apply(TableEnvironment.scala:1112)
      at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
      at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
      at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
      at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186)
      at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:241)
      at scala.collection.mutable.ArrayOps$ofRef.flatMap(ArrayOps.scala:186)
      at org.apache.flink.table.api.TableEnvironment.getFieldInfo(TableEnvironment.scala:1112)
      at org.apache.flink.table.api.StreamTableEnvironment.registerDataStreamInternal(StreamTableEnvironment.scala:546)
      at org.apache.flink.table.api.java.StreamTableEnvironment.fromDataStream(StreamTableEnvironment.scala:91)
      at Udemy_Course.TableAPIExample.CountWordExample.main(CountWordExample.java:30)
  </code></pre>
  <p><strong>Code</strong></p>
  <pre><code>public class CountWordExample {
  
      public static void main(String[] args) throws Exception {
          StreamExecutionEnvironment environment=StreamExecutionEnvironment.getExecutionEnvironment();
          StreamTableEnvironment streamTableEnvironment=StreamTableEnvironment.create(environment);
  
  
          DataStream&lt;WC&gt; streamOfWords =
                  environment.fromElements(
                          new WC(&quot;Hello&quot;,1L),
                          new WC(&quot;Howdy&quot;,1L),
                          new WC(&quot;Hello&quot;,1L),
                          new WC(&quot;Hello&quot;,1L));
  
          Table t1 = streamTableEnvironment.fromDataStream(streamOfWords, &quot;word, count&quot;);
  
          Table result = streamTableEnvironment.sqlQuery(&quot;select word, count(word) as wordcount 
          from &quot; + t1 + &quot; group by word&quot;);
  
  
          streamTableEnvironment.toRetractStream(result, CountWordExample.class ).print();
  
          environment.execute();
      }
  
      public static class WC {
  
          private String word;
          private Long count;
  
          public WC() {}
  
          public WC(String word,Long count) {
              this.word=word;
              this.count=count;
          }
      }
  }
  </code></pre>