title: Does Flink SQL support Java Map types?
body: <p>I'm trying to access a key from a map using Flink's SQL API. It fails with the error  Exception in thread "main" org.apache.flink.table.api.TableException: Type is not supported: ANY
  Please advise how i can fix it. 
  Here is my event class</p>
  
  <pre class="lang-java prettyprint-override"><code>     public class EventHolder {
  
          private Map&lt;String,String&gt; event;
  
          public Map&lt;String, String&gt; getEvent() {
              return event;
          }
  
          public void setEvent(Map&lt;String, String&gt; event) {
              this.event = event;
          }
      }
  </code></pre>
  
  <p>Here is the main class which submits the flink job</p>
  
  <pre class="lang-java prettyprint-override"><code>public class MapTableSource {
      public static void main(String[] args) throws Exception {
          final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
          DataStream&lt;EventHolder&gt; mapEventStream = env.fromCollection(getMaps());
  
          // register a table and use SQL
          StreamTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
          tableEnv.registerDataStream("mapEvent", mapEventStream); 
          //tableEnv.registerFunction("orderSizeType", new OrderSizeType());
  
          Table alerts = tableEnv.sql(
                  "select event['key'] from mapEvent ");
  
          DataStream&lt;String&gt; alertStream = tableEnv.toAppendStream(alerts, String.class);
  
          alertStream.filter(new FilterFunction&lt;String&gt;() {
              private static final long serialVersionUID = -2438621539037257735L;
  
              @Override
              public boolean filter(String value) throws Exception {
                  System.out.println("Key value is:"+value);
                  return value!=null;
              }
          });
  
          env.execute("map-tablsource-job");
      }
  
      private static List&lt;EventHolder&gt; getMaps(){
          List&lt;EventHolder&gt; list = new ArrayList&lt;&gt;();
          for(int i=0;i&lt;5;i++){
              EventHolder holder = new EventHolder();
              Map&lt;String,String&gt; map = new HashMap&lt;&gt;();
              map.put("key", "value");
              holder.setEvent(map);
              list.add(holder);
          }
          return list;
      }
  }
  </code></pre>
  
  <p>When I run it I'm getting the exception</p>
  
  <pre><code>Exception in thread "main" org.apache.flink.table.api.TableException: Type is not supported: ANY
  at org.apache.flink.table.api.TableException$.apply(exceptions.scala:53)
  at org.apache.flink.table.calcite.FlinkTypeFactory$.toTypeInfo(FlinkTypeFactory.scala:341)
  at org.apache.flink.table.plan.logical.LogicalRelNode$$anonfun$12.apply(operators.scala:530)
  at org.apache.flink.table.plan.logical.LogicalRelNode$$anonfun$12.apply(operators.scala:529)
  at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
  at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
  at scala.collection.Iterator$class.foreach(Iterator.scala:742)
  at scala.collection.AbstractIterator.foreach(Iterator.scala:1194)
  at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
  at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
  at scala.collection.TraversableLike$class.map(TraversableLike.scala:245)
  at scala.collection.AbstractTraversable.map(Traversable.scala:104)
  at org.apache.flink.table.plan.logical.LogicalRelNode.&lt;init&gt;(operators.scala:529)
  at org.apache.flink.table.api.TableEnvironment.sql(TableEnvironment.scala:503)
  at com.c.p.flink.MapTableSource.main(MapTableSource.java:25)
  </code></pre>
  
  <p>I'm using flink 1.3.1</p>