improved odps sql parser.

src/main/java/com/alibaba/druid/sql/dialect/odps/ast/OdpsCreateTableStatement.java

@@ -35,6 +35,11 @@ public class OdpsCreateTableStatement extends SQLCreateTableStatement {
 
     protected List<SQLColumnDefinition> partitionColumns = new ArrayList<SQLColumnDefinition>(2);
 
+    protected final List<SQLName> clusteredBy = new ArrayList<SQLName>();
+    protected final List<SQLName> sortedBy = new ArrayList<SQLName>();
+
+    protected int buckets;
+
     protected SQLExpr lifecycle;
 
     public OdpsCreateTableStatement(){
@@ -90,10 +95,27 @@ public class OdpsCreateTableStatement extends SQLCreateTableStatement {
             this.acceptChild(visitor, tableSource);
             this.acceptChild(visitor, tableElementList);
             this.acceptChild(visitor, partitionColumns);
+            this.acceptChild(visitor, clusteredBy);
+            this.acceptChild(visitor, sortedBy);
             this.acceptChild(visitor, lifecycle);
             this.acceptChild(visitor, select);
         }
         visitor.endVisit(this);
     }
 
+    public List<SQLName> getClusteredBy() {
+        return clusteredBy;
+    }
+
+    public List<SQLName> getSortedBy() {
+        return sortedBy;
+    }
+
+    public int getBuckets() {
+        return buckets;
+    }
+
+    public void setBuckets(int buckets) {
+        this.buckets = buckets;
+    }
 }

src/main/java/com/alibaba/druid/sql/dialect/odps/parser/OdpsCreateTableParser.java

@@ -136,6 +136,33 @@ public class OdpsCreateTableParser extends SQLCreateTableParser {
             
             accept(Token.RPAREN);
         }
+
+        if (identifierEquals("CLUSTERED")) {
+            lexer.nextToken();
+            accept(Token.BY);
+            accept(Token.LPAREN);
+            this.exprParser.names(stmt.getClusteredBy());
+            accept(Token.RPAREN);
+        }
+
+        if (identifierEquals("SORTED")) {
+            lexer.nextToken();
+            accept(Token.BY);
+            accept(Token.LPAREN);
+            this.exprParser.names(stmt.getSortedBy());
+            accept(Token.RPAREN);
+        }
+
+        if (stmt.getClusteredBy().size() > 0 || stmt.getSortedBy().size() > 0) {
+            accept(Token.INTO);
+            if (lexer.token() == Token.LITERAL_INT) {
+                stmt.setBuckets(lexer.integerValue().intValue());
+                lexer.nextToken();
+            } else {
+                throw new ParserException("into buckets must be integer.");
+            }
+            acceptIdentifier("BUCKETS");
+        }
         
         if (identifierEquals("LIFECYCLE")) {
             lexer.nextToken();

src/main/java/com/alibaba/druid/sql/dialect/odps/visitor/OdpsOutputVisitor.java

@@ -131,6 +131,30 @@ public class OdpsOutputVisitor extends SQLASTOutputVisitor implements OdpsASTVis
             print(')');
         }
 
+        List<SQLName> clusteredBy = x.getClusteredBy();
+        if (clusteredBy.size() > 0) {
+            println();
+            print0(ucase ? "CLUSTERED BY (" : "clustered by (");
+            printAndAccept(clusteredBy, ",");
+            print(')');
+        }
+
+        List<SQLName> sortedBy = x.getSortedBy();
+        if (sortedBy.size() > 0) {
+            println();
+            print0(ucase ? "SORTED BY (" : "sorted by (");
+            printAndAccept(sortedBy, ",");
+            print(')');
+        }
+
+        int buckets = x.getBuckets();
+        if (buckets > 0) {
+            println();
+            print0(ucase ? "INTO " : "into ");
+            print(buckets);
+            print0(ucase ? " BUCKETS" : " buckets");
+        }
+
         if (x.getLifecycle() != null) {
             println();
             print0(ucase ? "LIFECYCLE " : "lifecycle ");