package com.alibaba.druid.sql.dialect.spark.parser;

import com.alibaba.druid.sql.ast.SQLExpr;
import com.alibaba.druid.sql.ast.SQLName;
import com.alibaba.druid.sql.ast.expr.SQLListExpr;
import com.alibaba.druid.sql.ast.statement.SQLColumnDefinition;
import com.alibaba.druid.sql.ast.statement.SQLConstraint;
import com.alibaba.druid.sql.ast.statement.SQLCreateTableStatement;
import com.alibaba.druid.sql.ast.statement.SQLSelect;
import com.alibaba.druid.sql.ast.statement.SQLSelectOrderByItem;
import com.alibaba.druid.sql.ast.statement.SQLTableElement;
import com.alibaba.druid.sql.dialect.hive.ast.HiveInputOutputFormat;
import com.alibaba.druid.sql.dialect.spark.ast.expr.HiveRowFormat;
import com.alibaba.druid.sql.dialect.spark.ast.statement.SparkCreateTableStatement;
import com.alibaba.druid.sql.parser.ParserException;
import com.alibaba.druid.sql.parser.SQLCreateTableParser;
import com.alibaba.druid.sql.parser.SQLExprParser;
import com.alibaba.druid.sql.parser.Token;
import com.alibaba.druid.util.FnvHash;

public class SparkCreateTableParser extends SQLCreateTableParser {
   public SparkCreateTableParser(String sql) {
      super((SQLExprParser)(new SparkExprParser(sql)));
      ((SparkExprParser)this.exprParser).setSparkSqlParser(this);
   }

   public SparkCreateTableParser(SQLExprParser exprParser) {
      super(exprParser);
   }

   public SQLCreateTableStatement parseCreateTable(boolean acceptCreate) {
      SparkCreateTableStatement stmt = new SparkCreateTableStatement();
      if (acceptCreate) {
         if (this.lexer.hasComment() && this.lexer.isKeepComments()) {
            stmt.addBeforeComment(this.lexer.readAndResetComments());
         }

         this.accept(Token.CREATE);
      }

      if (this.lexer.identifierEquals(FnvHash.Constants.EXTERNAL)) {
         this.lexer.nextToken();
         stmt.setExternal(true);
      }

      if (this.lexer.identifierEquals(FnvHash.Constants.TEMPORARY)) {
         this.lexer.nextToken();
         stmt.setType(SQLCreateTableStatement.Type.TEMPORARY);
      }

      this.accept(Token.TABLE);
      if (this.lexer.token() == Token.IF || this.lexer.identifierEquals(FnvHash.Constants.IF)) {
         this.lexer.nextToken();
         this.accept(Token.NOT);
         this.accept(Token.EXISTS);
         stmt.setIfNotExiists(true);
      }

      stmt.setName(this.exprParser.name());
      if (this.lexer.token() == Token.LPAREN) {
         this.lexer.nextToken();

         do {
            Token token = this.lexer.token();
            if (token != Token.IDENTIFIER && token != Token.LITERAL_ALIAS) {
               if (token != Token.PRIMARY && token != Token.UNIQUE && token != Token.CHECK && token != Token.CONSTRAINT && token != Token.FOREIGN) {
                  if (token == Token.TABLESPACE) {
                     throw new ParserException("TODO " + this.lexer.info());
                  }

                  SQLColumnDefinition column = this.exprParser.parseColumn();
                  stmt.getTableElementList().add(column);
               } else {
                  SQLConstraint constraint = this.exprParser.parseConstaint();
                  constraint.setParent(stmt);
                  stmt.getTableElementList().add((SQLTableElement)constraint);
               }
            } else {
               SQLColumnDefinition column = this.exprParser.parseColumn();
               stmt.getTableElementList().add(column);
            }

            if (this.lexer.token() != Token.COMMA) {
               break;
            }

            this.lexer.nextToken();
         } while(this.lexer.token() != Token.RPAREN);

         this.accept(Token.RPAREN);
      }

      while(true) {
         label195:
         while(this.lexer.token() != Token.USING) {
            if (this.lexer.identifierEquals(FnvHash.Constants.OPTIONS)) {
               this.lexer.nextToken();
               this.accept(Token.LPAREN);
               this.parseAssignItems(stmt.getTableOptions(), stmt, false);
               this.accept(Token.RPAREN);
            } else if (this.lexer.token() == Token.COMMENT) {
               this.lexer.nextToken();
               SQLExpr comment = this.exprParser.expr();
               stmt.setComment(comment);
            } else if (this.lexer.identifierEquals(FnvHash.Constants.MAPPED)) {
               this.lexer.nextToken();
               this.accept(Token.BY);
               this.exprParser.parseAssignItem(stmt.getMappedBy(), stmt);
            } else if (this.lexer.token() == Token.PARTITIONED) {
               this.lexer.nextToken();
               this.accept(Token.BY);
               this.accept(Token.LPAREN);

               while(this.lexer.token() == Token.IDENTIFIER) {
                  SQLColumnDefinition column = this.exprParser.parseColumn();
                  stmt.addPartitionColumn(column);
                  if (this.lexer.isKeepComments() && this.lexer.hasComment()) {
                     column.addAfterComment(this.lexer.readAndResetComments());
                  }

                  if (this.lexer.token() != Token.COMMA) {
                     this.accept(Token.RPAREN);
                     continue label195;
                  }

                  this.lexer.nextToken();
                  if (this.lexer.isKeepComments() && this.lexer.hasComment()) {
                     column.addAfterComment(this.lexer.readAndResetComments());
                  }
               }

               throw new ParserException("expect identifier. " + this.lexer.info());
            } else if (this.lexer.identifierEquals(FnvHash.Constants.CLUSTERED)) {
               this.lexer.nextToken();
               this.accept(Token.BY);
               this.accept(Token.LPAREN);

               while(true) {
                  SQLSelectOrderByItem item = this.exprParser.parseSelectOrderByItem();
                  stmt.addClusteredByItem(item);
                  if (this.lexer.token() != Token.COMMA) {
                     this.accept(Token.RPAREN);
                     break;
                  }

                  this.lexer.nextToken();
               }
            } else if (this.lexer.identifierEquals(FnvHash.Constants.SKEWED)) {
               this.lexer.nextToken();
               this.accept(Token.BY);
               this.accept(Token.LPAREN);
               this.exprParser.exprList(stmt.getSkewedBy(), stmt);
               this.accept(Token.RPAREN);
               this.accept(Token.ON);
               this.accept(Token.LPAREN);

               while(true) {
                  if (this.lexer.token() == Token.LPAREN) {
                     SQLListExpr list = new SQLListExpr();
                     this.lexer.nextToken();
                     this.exprParser.exprList(list.getItems(), list);
                     this.accept(Token.RPAREN);
                     stmt.addSkewedByOn(list);
                  } else {
                     SQLExpr expr = this.exprParser.expr();
                     stmt.addSkewedByOn(expr);
                  }

                  if (this.lexer.token() != Token.COMMA) {
                     this.accept(Token.RPAREN);
                     break;
                  }

                  this.lexer.nextToken();
               }
            } else if (this.lexer.identifierEquals(FnvHash.Constants.SORTED)) {
               this.parseSortedBy(stmt);
            } else if (this.lexer.token() != Token.ROW && !this.lexer.identifierEquals(FnvHash.Constants.ROW)) {
               if (this.lexer.identifierEquals(FnvHash.Constants.SORTED)) {
                  this.parseSortedBy(stmt);
               } else if ((stmt.getClusteredBy().size() > 0 || stmt.getSortedBy().size() > 0) && this.lexer.token() == Token.INTO) {
                  this.accept(Token.INTO);
                  if (this.lexer.token() != Token.LITERAL_INT) {
                     throw new ParserException("into buckets must be integer. " + this.lexer.info());
                  }

                  stmt.setBuckets(this.lexer.integerValue().intValue());
                  this.lexer.nextToken();
                  this.acceptIdentifier("BUCKETS");
                  if (this.lexer.token() == Token.LPAREN) {
                     this.lexer.nextToken();
                     SQLSelect select = this.createSQLSelectParser().select();
                     stmt.setBucketSelect(select);
                     this.accept(Token.RPAREN);
                  }
               } else if (this.lexer.token() != Token.ROW && !this.lexer.identifierEquals(FnvHash.Constants.ROW)) {
                  if (this.lexer.identifierEquals(FnvHash.Constants.STORED)) {
                     this.lexer.nextToken();
                     this.accept(Token.AS);
                     if (this.lexer.identifierEquals(FnvHash.Constants.INPUTFORMAT)) {
                        HiveInputOutputFormat format = new HiveInputOutputFormat();
                        this.lexer.nextToken();
                        format.setInput(this.exprParser.primary());
                        if (this.lexer.identifierEquals(FnvHash.Constants.OUTPUTFORMAT)) {
                           this.lexer.nextToken();
                           format.setOutput(this.exprParser.primary());
                        }

                        stmt.setStoredAs(format);
                     } else {
                        SQLName name = this.exprParser.name();
                        stmt.setStoredAs(name);
                     }
                  } else if (this.lexer.identifierEquals(FnvHash.Constants.LOCATION)) {
                     this.lexer.nextToken();
                     SQLExpr location = this.exprParser.expr();
                     stmt.setLocation(location);
                  } else if (!this.lexer.identifierEquals(FnvHash.Constants.TBLPROPERTIES)) {
                     if (this.lexer.identifierEquals(FnvHash.Constants.META)) {
                        this.lexer.nextToken();
                        this.acceptIdentifier("LIFECYCLE");
                        stmt.setMetaLifeCycle(this.exprParser.primary());
                     } else if (this.lexer.token() == Token.LIKE) {
                        this.lexer.nextToken();
                        SQLName name = this.exprParser.name();
                        stmt.setLike(name);
                     } else {
                        if (this.lexer.token() != Token.COMMENT) {
                           if (this.lexer.token() == Token.AS) {
                              this.lexer.nextToken();
                              SQLSelect select = this.createSQLSelectParser().select();
                              stmt.setSelect(select);
                           }

                           return stmt;
                        }

                        this.lexer.nextToken();
                        SQLExpr comment = this.exprParser.expr();
                        stmt.setComment(comment);
                     }
                  } else {
                     this.lexer.nextToken();
                     this.accept(Token.LPAREN);

                     do {
                        String name = this.lexer.stringVal();
                        this.lexer.nextToken();
                        this.accept(Token.EQ);
                        SQLExpr value = this.exprParser.primary();
                        stmt.addTblProperty(name, value);
                        if (this.lexer.token() != Token.COMMA) {
                           break;
                        }

                        this.lexer.nextToken();
                     } while(this.lexer.token() != Token.RPAREN);

                     this.accept(Token.RPAREN);
                  }
               } else {
                  this.parseRowFormat(stmt);
               }
            } else {
               this.parseRowFormat(stmt);
            }
         }

         this.lexer.nextToken();
         SQLName expr = this.exprParser.name();
         stmt.setDatasource(expr);
      }
   }

   private void parseRowFormat(SparkCreateTableStatement stmt) {
      HiveRowFormat format = ((SparkExprParser)this.getExprParser()).parseHiveRowFormat();
      stmt.setHiveFormat(format);
      if (this.lexer.token() == Token.WITH) {
         this.lexer.nextToken();
         this.acceptIdentifier("SERDEPROPERTIES");
         this.accept(Token.LPAREN);

         while(true) {
            String name = this.lexer.stringVal();
            this.lexer.nextToken();
            this.accept(Token.EQ);
            SQLExpr value = this.exprParser.primary();
            stmt.getSerdeProperties().put(name, value);
            if (this.lexer.token() != Token.COMMA) {
               this.accept(Token.RPAREN);
               break;
            }

            this.lexer.nextToken();
         }
      }

   }

   private void parseSortedBy(SparkCreateTableStatement stmt) {
      this.lexer.nextToken();
      this.accept(Token.BY);
      this.accept(Token.LPAREN);

      while(true) {
         SQLSelectOrderByItem item = this.exprParser.parseSelectOrderByItem();
         stmt.addSortedByItem(item);
         if (this.lexer.token() != Token.COMMA) {
            this.accept(Token.RPAREN);
            return;
         }

         this.lexer.nextToken();
      }
   }
}
