The parsing logic for different flink version is different a lot, this ticket is to refactor FlinkSqlInterpreter to move the parsing logic to flink-shims