[GitHub] [carbondata] Pickupolddriver opened a new pull request #4032: WIP: Add MERGE INTO SQL Command support to carbondata

[GitBox]

CarbonDataQA2 commented on pull request #4032:
URL: https://github.com/apache/carbondata/pull/4032#issuecomment-740322409


   Build Failed  with Spark 2.4.5, Please check CI http://121.244.95.60:12444/job/ApacheCarbon_PR_Builder_2.4.5/3343/
   


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

CarbonDataQA2 commented on pull request #4032:
URL: https://github.com/apache/carbondata/pull/4032#issuecomment-740371852


   Build Success with Spark 2.3.4, Please check CI http://121.244.95.60:12444/job/ApacheCarbonPRBuilder2.3/5108/
   


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

CarbonDataQA2 commented on pull request #4032:
URL: https://github.com/apache/carbondata/pull/4032#issuecomment-740373756


   Build Success with Spark 2.4.5, Please check CI http://121.244.95.60:12444/job/ApacheCarbon_PR_Builder_2.4.5/3346/
   


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

ajantha-bhat commented on a change in pull request #4032:
URL: https://github.com/apache/carbondata/pull/4032#discussion_r542197802



##########
File path: integration/spark/src/main/antlr4/org/apache/spark/sql/parser/CarbonSqlBase.g4
##########
@@ -0,0 +1,1842 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");

Review comment:
       I see that 2 .g4 files created. This is required ? I think it is generated file




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

Zhangshunyu commented on a change in pull request #4032:
URL: https://github.com/apache/carbondata/pull/4032#discussion_r542212249



##########
File path: integration/spark/src/main/antlr4/org/apache/spark/sql/parser/CarbonSqlBase.g4
##########
@@ -0,0 +1,1842 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");

Review comment:
       @ajantha-bhat only 1 g4 file is there, this g4 file is used to gen the other java files, and the other java files will only generated during compile process.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

CarbonDataQA2 commented on pull request #4032:
URL: https://github.com/apache/carbondata/pull/4032#issuecomment-745043164


   Build Success with Spark 2.3.4, Please check CI http://121.244.95.60:12444/job/ApacheCarbonPRBuilder2.3/5161/
   


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

CarbonDataQA2 commented on pull request #4032:
URL: https://github.com/apache/carbondata/pull/4032#issuecomment-745044946


   Build Success with Spark 2.4.5, Please check CI http://121.244.95.60:12444/job/ApacheCarbon_PR_Builder_2.4.5/3399/
   


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

CarbonDataQA2 commented on pull request #4032:
URL: https://github.com/apache/carbondata/pull/4032#issuecomment-748003841


   Build Failed  with Spark 2.4.5, Please check CI http://121.244.95.60:12545/job/ApacheCarbon_PR_Builder_2.4.5/3301/
   


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

QiangCai commented on pull request #4032:
URL: https://github.com/apache/carbondata/pull/4032#issuecomment-750706348


   retest this please


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

QiangCai commented on a change in pull request #4032:
URL: https://github.com/apache/carbondata/pull/4032#discussion_r548352547



##########
File path: integration/spark/pom.xml
##########
@@ -264,6 +269,18 @@
       <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
+      <dependency>

Review comment:
       1. remove redundant dependency
   2. scope is test?
   

##########
File path: integration/spark/pom.xml
##########
@@ -528,6 +545,22 @@
           </execution>
         </executions>
       </plugin>
+      <plugin>
+        <groupId>org.antlr</groupId>
+        <artifactId>antlr4-maven-plugin</artifactId>
+        <executions>
+          <execution>
+            <goals>
+              <goal>antlr4</goal>
+            </goals>
+          </execution>
+        </executions>
+        <configuration>
+          <visitor>true</visitor>
+          <sourceDirectory>../spark/src/main/antlr4</sourceDirectory>

Review comment:
       how about src/main/antlr4?

##########
File path: integration/spark/src/main/antlr4/org/apache/spark/sql/parser/CarbonSqlBase.g4
##########
@@ -0,0 +1,1842 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * This file is an adaptation of Presto's presto-parser/src/main/antlr4/com/facebook/presto/sql/parser/CarbonSqlBase.g4 grammar.

Review comment:
       this comment is incorrect

##########
File path: integration/spark/src/main/antlr4/org/apache/spark/sql/parser/CarbonSqlBase.g4
##########
@@ -0,0 +1,1842 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * This file is an adaptation of Presto's presto-parser/src/main/antlr4/com/facebook/presto/sql/parser/CarbonSqlBase.g4 grammar.
+ */
+
+grammar CarbonSqlBase;
+
+@parser::members {

Review comment:
       better to simplify this file, now we only use mergeInto part

##########
File path: integration/spark/src/main/java/org/apache/spark/sql/CarbonAntlrSqlVisitor.java
##########
@@ -0,0 +1,353 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.spark.sql.catalyst.expressions.Expression;
+import org.apache.spark.sql.catalyst.parser.ParseException;
+import org.apache.spark.sql.catalyst.parser.ParserInterface;
+import org.apache.spark.sql.execution.command.mutation.merge.DeleteAction;
+import org.apache.spark.sql.execution.command.mutation.merge.InsertAction;
+import org.apache.spark.sql.execution.command.mutation.merge.MergeAction;
+import org.apache.spark.sql.execution.command.mutation.merge.UpdateAction;
+import org.apache.spark.sql.merge.model.CarbonJoinExpression;
+import org.apache.spark.sql.merge.model.CarbonMergeIntoModel;
+import org.apache.spark.sql.merge.model.ColumnModel;
+import org.apache.spark.sql.merge.model.TableModel;
+import org.apache.spark.sql.parser.CarbonSqlBaseBaseVisitor;
+import org.apache.spark.sql.parser.CarbonSqlBaseParser;
+import org.apache.spark.util.SparkUtil;
+
+public class CarbonAntlrSqlVisitor extends CarbonSqlBaseBaseVisitor {
+
+  private final ParserInterface sparkParser;
+
+  public CarbonAntlrSqlVisitor(ParserInterface sparkParser) {
+    this.sparkParser = sparkParser;
+  }
+
+  @Override
+  public String visitTableAlias(CarbonSqlBaseParser.TableAliasContext ctx) {
+    if (null == ctx.children) {
+      return null;
+    }
+    String res = ctx.getChild(1).getText();
+    System.out.println(res);
+    return res;
+  }
+
+  @Override
+  public MergeAction visitAssignmentList(CarbonSqlBaseParser.AssignmentListContext ctx) {
+    //  UPDATE SET assignmentList
+    Map<Column, Column> map = new HashMap<>();
+    for (int currIdx = 0; currIdx < ctx.getChildCount(); currIdx++) {
+      if (ctx.getChild(currIdx) instanceof CarbonSqlBaseParser.AssignmentContext) {
+        //Assume the actions are all use to pass value
+        String left = ctx.getChild(currIdx).getChild(0).getText();
+        if (left.split("\\.").length > 1) {
+          left = left.split("\\.")[1];
+        }
+        String right = ctx.getChild(currIdx).getChild(2).getText();
+        Column rightColumn = null;
+        try {
+          Expression expression = sparkParser.parseExpression(right);
+          rightColumn = new Column(expression);
+        } catch (Exception ex) {
+          // todo throw EX here
+        }
+        map.put(new Column(left), rightColumn);
+      }
+    }
+    return new UpdateAction(SparkUtil.convertMap(map), false);
+  }
+
+  @Override
+  public MergeAction visitMatchedAction(CarbonSqlBaseParser.MatchedActionContext ctx) {
+    int childCount = ctx.getChildCount();
+    if (childCount == 1) {
+      // when matched ** delete
+      return new DeleteAction();
+    } else {
+      if (ctx
+          .getChild(ctx.getChildCount() - 1) instanceof CarbonSqlBaseParser.AssignmentListContext) {
+        //UPDATE SET assignmentList
+        return visitAssignmentList(
+            (CarbonSqlBaseParser.AssignmentListContext) ctx.getChild(ctx.getChildCount() - 1));
+      } else {
+        //UPDATE SET *
+        return new UpdateAction(null, true);
+      }
+    }
+  }
+
+  @Override
+  public InsertAction visitNotMatchedAction(CarbonSqlBaseParser.NotMatchedActionContext ctx) {
+    if (ctx.getChildCount() <= 2) {
+      //INSERT *
+      return InsertAction.apply(null, true);
+    } else {
+      // INSERT '(' columns=multipartIdentifierList ')'VALUES '(' expression (',' expression)* ')'
+      // todo throw Exception here
+      return InsertAction.apply(null, false);
+    }
+  }
+
+  @Override
+  public MergeAction visitNotMatchedClause(CarbonSqlBaseParser.NotMatchedClauseContext ctx) {
+    int currIdx = 0;
+    for (; currIdx < ctx.getChildCount(); currIdx++) {
+      if (ctx.getChild(currIdx) instanceof CarbonSqlBaseParser.NotMatchedActionContext) {
+        break;
+      }
+    }
+    // Throw exception incase of not matched clause
+    return visitNotMatchedAction(
+        (CarbonSqlBaseParser.NotMatchedActionContext) ctx.getChild(currIdx));
+  }
+
+  @Override
+  public MergeAction visitMatchedClause(CarbonSqlBaseParser.MatchedClauseContext ctx) {
+    //There will be lots of childs at ctx,
+    // we need to find the predicate
+    int currIdx = 0;
+    for (; currIdx < ctx.getChildCount(); currIdx++) {
+      if (ctx.getChild(currIdx) instanceof CarbonSqlBaseParser.MatchedActionContext) {
+        break;
+      }
+    }
+    // Throw Exception in case of no Matched Action
+    return visitMatchedAction((CarbonSqlBaseParser.MatchedActionContext) ctx.getChild(currIdx));
+  }
+
+  public boolean containsWhenMatchedPredicateExpression(int childCount) {
+    return childCount > 4;
+  }
+
+  public boolean containsWhenNotMatchedPredicateExpression(int childCount) {
+    return childCount > 5;
+  }
+
+  @Override
+  public CarbonMergeIntoModel visitMergeInto(CarbonSqlBaseParser.MergeIntoContext ctx) {
+    TableModel targetTable = visitMultipartIdentifier(ctx.target);
+    TableModel sourceTable = visitMultipartIdentifier(ctx.source);
+
+    //Once get these two table,
+    //We can try to get CarbonTable
+
+    //Build a matched clause list to store the when matched and when not matched clause
+    int size = ctx.getChildCount();
+    int currIdx = 0;
+    Expression joinExpression = null;
+    List<Expression> mergeExpressions = new ArrayList<>();
+    List<MergeAction> mergeActions = new ArrayList<>();
+
+    // There should be two List to store the result retrieve from
+    // when matched / when not matched context
+    while (currIdx < size) {
+      if (ctx.getChild(currIdx) instanceof CarbonSqlBaseParser.PredicatedContext) {
+        //This branch will visit the Join Expression
+        ctx.getChild(currIdx).getChildCount();
+        joinExpression =
+            this.visitPredicated((CarbonSqlBaseParser.PredicatedContext) ctx.getChild(currIdx), "");
+      } else if (ctx.getChild(currIdx) instanceof CarbonSqlBaseParser.MatchedClauseContext) {
+        //This branch will deal with the Matched Clause
+        Expression whenMatchedExpression = null;
+        //Get the whenMatched expression
+        try {
+          if (this.containsWhenMatchedPredicateExpression(ctx.getChild(currIdx).getChildCount())) {
+            whenMatchedExpression = sparkParser.parseExpression(
+                ((CarbonSqlBaseParser.MatchedClauseContext) ctx.getChild(currIdx))
+                    .booleanExpression().getText());
+          }
+        } catch (ParseException e) {
+          e.printStackTrace();
+        }
+        mergeExpressions.add(whenMatchedExpression);
+        mergeActions.add(visitMatchedAction(
+            (CarbonSqlBaseParser.MatchedActionContext) ctx.getChild(currIdx)
+                .getChild(ctx.getChild(currIdx).getChildCount() - 1)));
+      } else if (ctx.getChild(currIdx) instanceof CarbonSqlBaseParser.NotMatchedClauseContext) {
+        //This branch will deal with the Matched Clause
+        Expression whenNotMatchedExpression = null;
+        //Get the whenMatched expression
+        try {
+          if (this
+              .containsWhenNotMatchedPredicateExpression(ctx.getChild(currIdx).getChildCount())) {
+            whenNotMatchedExpression = sparkParser.parseExpression(
+                ((CarbonSqlBaseParser.NotMatchedClauseContext) ctx.getChild(currIdx))
+                    .booleanExpression().getText());
+          }
+        } catch (ParseException e) {
+          e.printStackTrace();
+        }
+        mergeExpressions.add(whenNotMatchedExpression);
+        CarbonSqlBaseParser.NotMatchedActionContext notMatchedActionContext =
+                (CarbonSqlBaseParser.NotMatchedActionContext) ctx.getChild(currIdx)
+                        .getChild(ctx.getChild(currIdx).getChildCount() - 1);
+        if (notMatchedActionContext.getChildCount() <= 2) {
+          mergeActions.add(InsertAction.apply(null, true));
+        } else if (notMatchedActionContext.ASTERISK() == null) {
+          if (notMatchedActionContext.columns.multipartIdentifier().size() !=
+                  notMatchedActionContext.expression().size()) {
+            // todo throw EX here
+          }
+          Map<Column, Column> insertMap = new HashMap<>();
+          for (int i = 0; i < notMatchedActionContext.columns.multipartIdentifier().size(); i++) {
+            String left = visitMultipartIdentifier(
+                            notMatchedActionContext.columns.multipartIdentifier().get(i), "")
+                            .getColName();
+            String right = notMatchedActionContext.expression().get(i).getText();
+            // some times the right side is literal or expression, not table column
+            // so we need to check the left side is a column or expression
+            Column rightColumn = null;
+            try {
+              Expression expression = sparkParser.parseExpression(right);
+              rightColumn = new Column(expression);
+            } catch (Exception ex) {
+              // todo throw EX here
+            }
+            insertMap.put(new Column(left), rightColumn);
+          }
+          mergeActions.add(InsertAction.apply(SparkUtil.convertMap(insertMap), false));
+        } else {
+          mergeActions.add(InsertAction.apply(null, false));
+        }
+      }
+      currIdx++;
+    }
+    return new CarbonMergeIntoModel(targetTable, sourceTable, joinExpression,
+            mergeExpressions, mergeActions);
+  }
+
+  @Override
+  public CarbonJoinExpression visitComparison(CarbonSqlBaseParser.ComparisonContext ctx) {
+    // we need to get left Expression and Right Expression
+    // Even get the table name and col name
+    ctx.getText();
+    String t1Name = ctx.left.getChild(0).getChild(0).getText();
+    String c1Name = ctx.left.getChild(0).getChild(2).getText();
+    String t2Name = ctx.right.getChild(0).getChild(0).getText();
+    String c2Name = ctx.right.getChild(0).getChild(2).getText();
+    return new CarbonJoinExpression(t1Name, c1Name, t2Name, c2Name);
+  }
+
+  public Expression visitComparison(CarbonSqlBaseParser.ComparisonContext ctx, String x) {
+    Expression expression = null;
+    try {
+      expression = sparkParser.parseExpression(ctx.getText());
+    } catch (ParseException e) {
+      e.printStackTrace();
+    }
+    return expression;
+  }
+
+  @Override
+  public Object visitValueExpressionDefault(CarbonSqlBaseParser.ValueExpressionDefaultContext ctx) {
+    ctx.getText();
+    return super.visitValueExpressionDefault(ctx);
+  }
+
+  @Override
+  public CarbonJoinExpression visitPredicated(CarbonSqlBaseParser.PredicatedContext ctx) {
+    return visitComparison((CarbonSqlBaseParser.ComparisonContext) ctx.getChild(0));
+  }
+
+  public Expression visitPredicated(CarbonSqlBaseParser.PredicatedContext ctx, String type) {
+    return visitComparison((CarbonSqlBaseParser.ComparisonContext) ctx.getChild(0), "");
+  }
+
+  @Override
+  public ColumnModel visitDereference(CarbonSqlBaseParser.DereferenceContext ctx) {
+    // In this part, it will return two colunm name
+    int count = ctx.getChildCount();
+    ColumnModel col = new ColumnModel();
+    if (count == 3) {
+      String tableName = ctx.getChild(0).getText();
+      String colName = ctx.getChild(2).getText();
+      col = new ColumnModel(tableName, colName);
+    }
+    return col;
+  }
+
+  @Override
+  public TableModel visitMultipartIdentifier(CarbonSqlBaseParser.MultipartIdentifierContext ctx) {
+    TableModel table = new TableModel();
+    List<CarbonSqlBaseParser.ErrorCapturingIdentifierContext> parts = ctx.parts;
+    if (parts.size() == 2) {
+      table.setDatabase(parts.get(0).getText());
+      table.setTable(parts.get(1).getText());
+    }
+    if (parts.size() == 1) {
+      table.setTable(parts.get(0).getText());
+    }
+    return table;
+  }
+
+  public ColumnModel visitMultipartIdentifier(CarbonSqlBaseParser.MultipartIdentifierContext ctx,
+                                              String x) {
+    ColumnModel column = new ColumnModel();
+    List<CarbonSqlBaseParser.ErrorCapturingIdentifierContext> parts = ctx.parts;
+    if (parts.size() == 2) {
+      column.setTable(parts.get(0).getText());
+      column.setColName(parts.get(1).getText());
+    }
+    if (parts.size() == 1) {
+      column.setColName(parts.get(0).getText());
+    }
+    return column;
+  }
+
+  @Override
+  public String visitUnquotedIdentifier(CarbonSqlBaseParser.UnquotedIdentifierContext ctx) {
+    String res = ctx.getChild(0).getText();
+    System.out.println("ColName; " + res);
+    return res;
+  }
+
+  @Override
+  public String visitFromClause(CarbonSqlBaseParser.FromClauseContext ctx) {
+    String tableName = visitRelation(ctx.relation(0));
+    System.out.println("SQL table name: " + tableName);

Review comment:
       remove all "System.out.println" calls.
   change them to debug log if needed

##########
File path: integration/spark/src/main/java/org/apache/spark/sql/CarbonAntlrSqlVisitor.java
##########
@@ -0,0 +1,353 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.spark.sql.catalyst.expressions.Expression;
+import org.apache.spark.sql.catalyst.parser.ParseException;
+import org.apache.spark.sql.catalyst.parser.ParserInterface;
+import org.apache.spark.sql.execution.command.mutation.merge.DeleteAction;
+import org.apache.spark.sql.execution.command.mutation.merge.InsertAction;
+import org.apache.spark.sql.execution.command.mutation.merge.MergeAction;
+import org.apache.spark.sql.execution.command.mutation.merge.UpdateAction;
+import org.apache.spark.sql.merge.model.CarbonJoinExpression;
+import org.apache.spark.sql.merge.model.CarbonMergeIntoModel;
+import org.apache.spark.sql.merge.model.ColumnModel;
+import org.apache.spark.sql.merge.model.TableModel;
+import org.apache.spark.sql.parser.CarbonSqlBaseBaseVisitor;
+import org.apache.spark.sql.parser.CarbonSqlBaseParser;
+import org.apache.spark.util.SparkUtil;
+
+public class CarbonAntlrSqlVisitor extends CarbonSqlBaseBaseVisitor {
+
+  private final ParserInterface sparkParser;
+
+  public CarbonAntlrSqlVisitor(ParserInterface sparkParser) {
+    this.sparkParser = sparkParser;
+  }
+
+  @Override
+  public String visitTableAlias(CarbonSqlBaseParser.TableAliasContext ctx) {
+    if (null == ctx.children) {
+      return null;
+    }
+    String res = ctx.getChild(1).getText();
+    System.out.println(res);
+    return res;
+  }
+
+  @Override
+  public MergeAction visitAssignmentList(CarbonSqlBaseParser.AssignmentListContext ctx) {
+    //  UPDATE SET assignmentList
+    Map<Column, Column> map = new HashMap<>();
+    for (int currIdx = 0; currIdx < ctx.getChildCount(); currIdx++) {
+      if (ctx.getChild(currIdx) instanceof CarbonSqlBaseParser.AssignmentContext) {
+        //Assume the actions are all use to pass value
+        String left = ctx.getChild(currIdx).getChild(0).getText();
+        if (left.split("\\.").length > 1) {
+          left = left.split("\\.")[1];
+        }
+        String right = ctx.getChild(currIdx).getChild(2).getText();
+        Column rightColumn = null;
+        try {
+          Expression expression = sparkParser.parseExpression(right);
+          rightColumn = new Column(expression);
+        } catch (Exception ex) {
+          // todo throw EX here

Review comment:
       better to throw an exception to stop the parser flow

##########
File path: integration/spark/src/main/java/org/apache/spark/sql/CarbonMergeIntoSQLCommand.scala
##########
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql
+
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.execution.command.AtomicRunnableCommand
+import org.apache.spark.sql.execution.command.mutation.merge._
+import org.apache.spark.sql.functions.col
+import org.apache.spark.sql.merge.model.{CarbonMergeIntoModel, TableModel}
+import org.apache.spark.util.SparkUtil._
+import org.apache.spark.util.TableAPIUtil
+
+case class CarbonMergeIntoSQLCommand(mergeInto: CarbonMergeIntoModel)
+  extends AtomicRunnableCommand {
+
+  override def processMetadata(sparkSession: SparkSession): Seq[Row] = {
+    Seq.empty
+  }
+
+  override def processData(sparkSession: SparkSession): Seq[Row] = {
+    val sourceTable: TableModel = mergeInto.getSource
+    val targetTable: TableModel = mergeInto.getTarget
+    val mergeCondition: Expression = mergeInto.getMergeCondition
+    val mergeExpression: Seq[Expression] = convertExpressionList(mergeInto.getMergeExpressions)
+    val mergeActions: Seq[MergeAction] = convertMergeActionList(mergeInto.getMergeActions)
+
+    // validate the table
+    TableAPIUtil.validateTableExists(sparkSession,
+      if (sourceTable.getDatabase == null) {
+        "default"
+      } else {
+        sourceTable.getDatabase
+      },
+      sourceTable.getTable)
+    TableAPIUtil.validateTableExists(sparkSession,
+      if (targetTable.getDatabase == null) {
+        "default"
+      } else {
+        targetTable.getDatabase
+      },

Review comment:
       same as above

##########
File path: integration/spark/src/main/java/org/apache/spark/sql/CarbonMergeIntoSQLCommand.scala
##########
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql
+
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.execution.command.AtomicRunnableCommand
+import org.apache.spark.sql.execution.command.mutation.merge._
+import org.apache.spark.sql.functions.col
+import org.apache.spark.sql.merge.model.{CarbonMergeIntoModel, TableModel}
+import org.apache.spark.util.SparkUtil._
+import org.apache.spark.util.TableAPIUtil
+
+case class CarbonMergeIntoSQLCommand(mergeInto: CarbonMergeIntoModel)
+  extends AtomicRunnableCommand {
+
+  override def processMetadata(sparkSession: SparkSession): Seq[Row] = {
+    Seq.empty
+  }
+
+  override def processData(sparkSession: SparkSession): Seq[Row] = {
+    val sourceTable: TableModel = mergeInto.getSource
+    val targetTable: TableModel = mergeInto.getTarget
+    val mergeCondition: Expression = mergeInto.getMergeCondition
+    val mergeExpression: Seq[Expression] = convertExpressionList(mergeInto.getMergeExpressions)
+    val mergeActions: Seq[MergeAction] = convertMergeActionList(mergeInto.getMergeActions)
+
+    // validate the table
+    TableAPIUtil.validateTableExists(sparkSession,
+      if (sourceTable.getDatabase == null) {
+        "default"
+      } else {
+        sourceTable.getDatabase
+      },
+      sourceTable.getTable)
+    TableAPIUtil.validateTableExists(sparkSession,
+      if (targetTable.getDatabase == null) {
+        "default"
+      } else {
+        targetTable.getDatabase
+      },
+      targetTable.getTable)
+
+    val srcDf = sparkSession.sql(s"""SELECT * FROM ${ sourceTable.getTable }""")
+    val tgDf = sparkSession.sql(s"""SELECT * FROM ${ targetTable.getTable }""")
+
+    var matches = Seq.empty[MergeMatch]

Review comment:
       use ArrayBuffer

##########
File path: integration/spark/src/main/java/org/apache/spark/sql/CarbonMergeIntoSQLCommand.scala
##########
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql
+
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.execution.command.AtomicRunnableCommand
+import org.apache.spark.sql.execution.command.mutation.merge._
+import org.apache.spark.sql.functions.col
+import org.apache.spark.sql.merge.model.{CarbonMergeIntoModel, TableModel}
+import org.apache.spark.util.SparkUtil._
+import org.apache.spark.util.TableAPIUtil
+
+case class CarbonMergeIntoSQLCommand(mergeInto: CarbonMergeIntoModel)
+  extends AtomicRunnableCommand {
+
+  override def processMetadata(sparkSession: SparkSession): Seq[Row] = {
+    Seq.empty
+  }
+
+  override def processData(sparkSession: SparkSession): Seq[Row] = {
+    val sourceTable: TableModel = mergeInto.getSource
+    val targetTable: TableModel = mergeInto.getTarget
+    val mergeCondition: Expression = mergeInto.getMergeCondition
+    val mergeExpression: Seq[Expression] = convertExpressionList(mergeInto.getMergeExpressions)
+    val mergeActions: Seq[MergeAction] = convertMergeActionList(mergeInto.getMergeActions)
+
+    // validate the table
+    TableAPIUtil.validateTableExists(sparkSession,
+      if (sourceTable.getDatabase == null) {
+        "default"
+      } else {
+        sourceTable.getDatabase
+      },
+      sourceTable.getTable)
+    TableAPIUtil.validateTableExists(sparkSession,
+      if (targetTable.getDatabase == null) {
+        "default"
+      } else {
+        targetTable.getDatabase
+      },
+      targetTable.getTable)
+
+    val srcDf = sparkSession.sql(s"""SELECT * FROM ${ sourceTable.getTable }""")
+    val tgDf = sparkSession.sql(s"""SELECT * FROM ${ targetTable.getTable }""")

Review comment:
       need database qualifier

##########
File path: integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/iud/MergeIntoCarbonTableTestCase.scala
##########
@@ -0,0 +1,294 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.carbondata.spark.testsuite.iud
+
+import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.test.util.QueryTest
+import org.scalatest.BeforeAndAfterEach
+
+class MergeIntoCarbonTableTestCase extends QueryTest with BeforeAndAfterEach {

Review comment:
       need to add exception test case

##########
File path: integration/spark/src/main/java/org/apache/spark/sql/CarbonMergeIntoSQLCommand.scala
##########
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql
+
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.execution.command.AtomicRunnableCommand
+import org.apache.spark.sql.execution.command.mutation.merge._
+import org.apache.spark.sql.functions.col
+import org.apache.spark.sql.merge.model.{CarbonMergeIntoModel, TableModel}
+import org.apache.spark.util.SparkUtil._
+import org.apache.spark.util.TableAPIUtil
+
+case class CarbonMergeIntoSQLCommand(mergeInto: CarbonMergeIntoModel)
+  extends AtomicRunnableCommand {
+
+  override def processMetadata(sparkSession: SparkSession): Seq[Row] = {
+    Seq.empty
+  }
+
+  override def processData(sparkSession: SparkSession): Seq[Row] = {
+    val sourceTable: TableModel = mergeInto.getSource
+    val targetTable: TableModel = mergeInto.getTarget
+    val mergeCondition: Expression = mergeInto.getMergeCondition
+    val mergeExpression: Seq[Expression] = convertExpressionList(mergeInto.getMergeExpressions)
+    val mergeActions: Seq[MergeAction] = convertMergeActionList(mergeInto.getMergeActions)
+
+    // validate the table
+    TableAPIUtil.validateTableExists(sparkSession,
+      if (sourceTable.getDatabase == null) {
+        "default"
+      } else {
+        sourceTable.getDatabase
+      },

Review comment:
       use CarbonEnv.getDatabaseName to normalize the database name

##########
File path: integration/spark/src/main/scala/org/apache/spark/sql/parser/CarbonExtensionSqlParser.scala
##########
@@ -37,35 +37,42 @@ class CarbonExtensionSqlParser(
 ) extends SparkSqlParser(conf) {
 
   val parser = new CarbonExtensionSpark2SqlParser
+  val antlrParser = new CarbonAntlrParser(this)
 
   override def parsePlan(sqlText: String): LogicalPlan = {
     parser.synchronized {
       CarbonEnv.getInstance(sparkSession)
     }
     CarbonUtils.updateSessionInfoToCurrentThread(sparkSession)
     try {
-      val plan = parser.parse(sqlText)
-      plan
+      parser.parse(sqlText)
     } catch {
       case ce: MalformedCarbonCommandException =>
         throw ce
-      case ex: Throwable =>
+      case _: Throwable =>
         try {
-          val parsedPlan = initialParser.parsePlan(sqlText)
-          CarbonScalaUtil.cleanParserThreadLocals
-          parsedPlan
+          antlrParser.parse(sqlText)
         } catch {
-          case mce: MalformedCarbonCommandException =>
-            throw mce
-          case e: Throwable =>
-            e.printStackTrace(System.err)
-            CarbonScalaUtil.cleanParserThreadLocals
-            CarbonException.analysisException(
-              s"""== Parser1: ${parser.getClass.getName} ==
-                 |${ex.getMessage}
-                 |== Parser2: ${initialParser.getClass.getName} ==
-                 |${e.getMessage}
+          case ce: MalformedCarbonCommandException =>
+            throw ce
+          case ex: Throwable =>
+            try {
+              val parsedPlan = initialParser.parsePlan(sqlText)
+              CarbonScalaUtil.cleanParserThreadLocals
+              parsedPlan
+            } catch {
+              case mce: MalformedCarbonCommandException =>
+                throw mce
+              case e: Throwable =>
+                e.printStackTrace(System.err)
+                CarbonScalaUtil.cleanParserThreadLocals
+                CarbonException.analysisException(
+                  s"""== Parser1: ${ parser.getClass.getName } ==
+                     |${ ex.getMessage }
+                     |== Parser2: ${ initialParser.getClass.getName } ==
+                     |${ e.getMessage }

Review comment:
       1. adjust the order
   2. change the description:  Spark parser, Carbon parser instead of parser2, parser1
   3. output mergeInto parser exception under Carbon Parser

##########
File path: integration/spark/src/main/java/org/apache/spark/sql/CarbonAntlrSqlVisitor.java
##########
@@ -0,0 +1,353 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.spark.sql.catalyst.expressions.Expression;
+import org.apache.spark.sql.catalyst.parser.ParseException;
+import org.apache.spark.sql.catalyst.parser.ParserInterface;
+import org.apache.spark.sql.execution.command.mutation.merge.DeleteAction;
+import org.apache.spark.sql.execution.command.mutation.merge.InsertAction;
+import org.apache.spark.sql.execution.command.mutation.merge.MergeAction;
+import org.apache.spark.sql.execution.command.mutation.merge.UpdateAction;
+import org.apache.spark.sql.merge.model.CarbonJoinExpression;
+import org.apache.spark.sql.merge.model.CarbonMergeIntoModel;
+import org.apache.spark.sql.merge.model.ColumnModel;
+import org.apache.spark.sql.merge.model.TableModel;
+import org.apache.spark.sql.parser.CarbonSqlBaseBaseVisitor;
+import org.apache.spark.sql.parser.CarbonSqlBaseParser;
+import org.apache.spark.util.SparkUtil;
+
+public class CarbonAntlrSqlVisitor extends CarbonSqlBaseBaseVisitor {
+
+  private final ParserInterface sparkParser;
+
+  public CarbonAntlrSqlVisitor(ParserInterface sparkParser) {
+    this.sparkParser = sparkParser;
+  }
+
+  @Override
+  public String visitTableAlias(CarbonSqlBaseParser.TableAliasContext ctx) {
+    if (null == ctx.children) {
+      return null;
+    }
+    String res = ctx.getChild(1).getText();
+    System.out.println(res);
+    return res;
+  }
+
+  @Override
+  public MergeAction visitAssignmentList(CarbonSqlBaseParser.AssignmentListContext ctx) {
+    //  UPDATE SET assignmentList
+    Map<Column, Column> map = new HashMap<>();
+    for (int currIdx = 0; currIdx < ctx.getChildCount(); currIdx++) {
+      if (ctx.getChild(currIdx) instanceof CarbonSqlBaseParser.AssignmentContext) {
+        //Assume the actions are all use to pass value
+        String left = ctx.getChild(currIdx).getChild(0).getText();
+        if (left.split("\\.").length > 1) {
+          left = left.split("\\.")[1];
+        }
+        String right = ctx.getChild(currIdx).getChild(2).getText();
+        Column rightColumn = null;
+        try {
+          Expression expression = sparkParser.parseExpression(right);
+          rightColumn = new Column(expression);
+        } catch (Exception ex) {
+          // todo throw EX here

Review comment:
       please check all similar cases in this file

##########
File path: integration/spark/src/main/antlr4/org/apache/spark/sql/parser/CarbonSqlBase.g4
##########
@@ -0,0 +1,1842 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * This file is an adaptation of Presto's presto-parser/src/main/antlr4/com/facebook/presto/sql/parser/CarbonSqlBase.g4 grammar.
+ */
+
+grammar CarbonSqlBase;
+
+@parser::members {
+  /**
+   * When false, INTERSECT is given the greater precedence over the other set
+   * operations (UNION, EXCEPT and MINUS) as per the SQL standard.
+   */
+  public boolean legacy_setops_precedence_enbled = false;
+
+  /**
+   * When false, a literal with an exponent would be converted into
+   * double type rather than decimal type.
+   */
+  public boolean legacy_exponent_literal_as_decimal_enabled = false;
+
+  /**
+   * When true, the behavior of keywords follows ANSI SQL standard.
+   */
+  public boolean SQL_standard_keyword_behavior = false;
+}
+
+@lexer::members {
+  /**
+   * Verify whether current token is a valid decimal token (which contains dot).
+   * Returns true if the character that follows the token is not a digit or letter or underscore.
+   *
+   * For example:
+   * For char stream "2.3", "2." is not a valid decimal token, because it is followed by digit '3'.
+   * For char stream "2.3_", "2.3" is not a valid decimal token, because it is followed by '_'.
+   * For char stream "2.3W", "2.3" is not a valid decimal token, because it is followed by 'W'.
+   * For char stream "12.0D 34.E2+0.12 "  12.0D is a valid decimal token because it is followed
+   * by a space. 34.E2 is a valid decimal token because it is followed by symbol '+'
+   * which is not a digit or letter or underscore.
+   */
+  public boolean isValidDecimal() {
+    int nextChar = _input.LA(1);
+    if (nextChar >= 'A' && nextChar <= 'Z' || nextChar >= '0' && nextChar <= '9' ||
+      nextChar == '_') {
+      return false;
+    } else {
+      return true;
+    }
+  }
+
+  /**
+   * This method will be called when we see '/*' and try to match it as a bracketed comment.
+   * If the next character is '+', it should be parsed as hint later, and we cannot match
+   * it as a bracketed comment.
+   *
+   * Returns true if the next character is '+'.
+   */
+  public boolean isHint() {
+    int nextChar = _input.LA(1);
+    if (nextChar == '+') {
+      return true;
+    } else {
+      return false;
+    }
+  }
+}
+
+singleStatement
+    : statement ';'* EOF
+    ;
+
+singleExpression
+    : namedExpression EOF
+    ;
+
+singleTableIdentifier
+    : tableIdentifier EOF
+    ;
+
+singleMultipartIdentifier
+    : multipartIdentifier EOF
+    ;
+
+singleFunctionIdentifier
+    : functionIdentifier EOF
+    ;
+
+singleDataType
+    : dataType EOF
+    ;
+
+singleTableSchema
+    : colTypeList EOF
+    ;
+
+statement
+    : query                                                            #statementDefault
+    | ctes? dmlStatementNoWith                                         #dmlStatement
+    | USE NAMESPACE? multipartIdentifier                               #use
+    | CREATE namespace (IF NOT EXISTS)? multipartIdentifier
+        (commentSpec |
+         locationSpec |
+         (WITH (DBPROPERTIES | PROPERTIES) tablePropertyList))*        #createNamespace
+    | ALTER namespace multipartIdentifier
+        SET (DBPROPERTIES | PROPERTIES) tablePropertyList              #setNamespaceProperties
+    | ALTER namespace multipartIdentifier
+        SET locationSpec                                               #setNamespaceLocation
+    | DROP namespace (IF EXISTS)? multipartIdentifier
+        (RESTRICT | CASCADE)?                                          #dropNamespace
+    | SHOW (DATABASES | NAMESPACES) ((FROM | IN) multipartIdentifier)?
+        (LIKE? pattern=STRING)?                                        #showNamespaces
+    | createTableHeader ('(' colTypeList ')')? tableProvider
+        createTableClauses
+        (AS? query)?                                                   #createTable
+    | createTableHeader ('(' columns=colTypeList ')')?
+        (commentSpec |
+        (PARTITIONED BY '(' partitionColumns=colTypeList ')' |
+        PARTITIONED BY partitionColumnNames=identifierList) |
+        bucketSpec |
+        skewSpec |
+        rowFormat |
+        createFileFormat |
+        locationSpec |
+        (TBLPROPERTIES tableProps=tablePropertyList))*
+        (AS? query)?                                                   #createHiveTable
+    | CREATE TABLE (IF NOT EXISTS)? target=tableIdentifier
+        LIKE source=tableIdentifier
+        (tableProvider |
+        rowFormat |
+        createFileFormat |
+        locationSpec |
+        (TBLPROPERTIES tableProps=tablePropertyList))*                 #createTableLike
+    | replaceTableHeader ('(' colTypeList ')')? tableProvider
+        createTableClauses
+        (AS? query)?                                                   #replaceTable
+    | ANALYZE TABLE multipartIdentifier partitionSpec? COMPUTE STATISTICS
+        (identifier | FOR COLUMNS identifierSeq | FOR ALL COLUMNS)?    #analyze
+    | ALTER TABLE multipartIdentifier
+        ADD (COLUMN | COLUMNS)
+        columns=qualifiedColTypeWithPositionList                       #addTableColumns
+    | ALTER TABLE multipartIdentifier
+        ADD (COLUMN | COLUMNS)
+        '(' columns=qualifiedColTypeWithPositionList ')'               #addTableColumns
+    | ALTER TABLE table=multipartIdentifier
+        RENAME COLUMN
+        from=multipartIdentifier TO to=errorCapturingIdentifier        #renameTableColumn
+    | ALTER TABLE multipartIdentifier
+        DROP (COLUMN | COLUMNS)
+        '(' columns=multipartIdentifierList ')'                        #dropTableColumns
+    | ALTER TABLE multipartIdentifier
+        DROP (COLUMN | COLUMNS) columns=multipartIdentifierList        #dropTableColumns
+    | ALTER (TABLE | VIEW) from=multipartIdentifier
+        RENAME TO to=multipartIdentifier                               #renameTable
+    | ALTER (TABLE | VIEW) multipartIdentifier
+        SET TBLPROPERTIES tablePropertyList                            #setTableProperties
+    | ALTER (TABLE | VIEW) multipartIdentifier
+        UNSET TBLPROPERTIES (IF EXISTS)? tablePropertyList             #unsetTableProperties
+    | ALTER TABLE table=multipartIdentifier
+        (ALTER | CHANGE) COLUMN? column=multipartIdentifier
+        alterColumnAction?                                             #alterTableAlterColumn
+    | ALTER TABLE table=multipartIdentifier partitionSpec?
+        CHANGE COLUMN?
+        colName=multipartIdentifier colType colPosition?               #hiveChangeColumn
+    | ALTER TABLE table=multipartIdentifier partitionSpec?
+        REPLACE COLUMNS
+        '(' columns=qualifiedColTypeWithPositionList ')'               #hiveReplaceColumns
+    | ALTER TABLE multipartIdentifier (partitionSpec)?
+        SET SERDE STRING (WITH SERDEPROPERTIES tablePropertyList)?     #setTableSerDe
+    | ALTER TABLE multipartIdentifier (partitionSpec)?
+        SET SERDEPROPERTIES tablePropertyList                          #setTableSerDe
+    | ALTER (TABLE | VIEW) multipartIdentifier ADD (IF NOT EXISTS)?
+        partitionSpecLocation+                                         #addTablePartition
+    | ALTER TABLE multipartIdentifier
+        from=partitionSpec RENAME TO to=partitionSpec                  #renameTablePartition
+    | ALTER (TABLE | VIEW) multipartIdentifier
+        DROP (IF EXISTS)? partitionSpec (',' partitionSpec)* PURGE?    #dropTablePartitions
+    | ALTER TABLE multipartIdentifier
+        (partitionSpec)? SET locationSpec                              #setTableLocation
+    | ALTER TABLE multipartIdentifier RECOVER PARTITIONS               #recoverPartitions
+    | DROP TABLE (IF EXISTS)? multipartIdentifier PURGE?               #dropTable
+    | DROP VIEW (IF EXISTS)? multipartIdentifier                       #dropView
+    | CREATE (OR REPLACE)? (GLOBAL? TEMPORARY)?
+        VIEW (IF NOT EXISTS)? multipartIdentifier
+        identifierCommentList?
+        (commentSpec |
+         (PARTITIONED ON identifierList) |
+         (TBLPROPERTIES tablePropertyList))*
+        AS query                                                       #createView
+    | CREATE (OR REPLACE)? GLOBAL? TEMPORARY VIEW
+        tableIdentifier ('(' colTypeList ')')? tableProvider
+        (OPTIONS tablePropertyList)?                                   #createTempViewUsing
+    | ALTER VIEW multipartIdentifier AS? query                         #alterViewQuery
+    | CREATE (OR REPLACE)? TEMPORARY? FUNCTION (IF NOT EXISTS)?
+        multipartIdentifier AS className=STRING
+        (USING resource (',' resource)*)?                              #createFunction
+    | DROP TEMPORARY? FUNCTION (IF EXISTS)? multipartIdentifier        #dropFunction
+    | EXPLAIN (LOGICAL | FORMATTED | EXTENDED | CODEGEN | COST)?
+        statement                                                      #explain
+    | SHOW TABLES ((FROM | IN) multipartIdentifier)?
+        (LIKE? pattern=STRING)?                                        #showTables
+    | SHOW TABLE EXTENDED ((FROM | IN) ns=multipartIdentifier)?
+        LIKE pattern=STRING partitionSpec?                             #showTable
+    | SHOW TBLPROPERTIES table=multipartIdentifier
+        ('(' key=tablePropertyKey ')')?                                #showTblProperties
+    | SHOW COLUMNS (FROM | IN) table=multipartIdentifier
+        ((FROM | IN) ns=multipartIdentifier)?                          #showColumns
+    | SHOW VIEWS ((FROM | IN) multipartIdentifier)?
+        (LIKE? pattern=STRING)?                                        #showViews
+    | SHOW PARTITIONS multipartIdentifier partitionSpec?               #showPartitions
+    | SHOW identifier? FUNCTIONS
+        (LIKE? (multipartIdentifier | pattern=STRING))?                #showFunctions
+    | SHOW CREATE TABLE multipartIdentifier (AS SERDE)?                #showCreateTable
+    | SHOW CURRENT NAMESPACE                                           #showCurrentNamespace
+    | (DESC | DESCRIBE) FUNCTION EXTENDED? describeFuncName            #describeFunction
+    | (DESC | DESCRIBE) namespace EXTENDED?
+        multipartIdentifier                                            #describeNamespace
+    | (DESC | DESCRIBE) TABLE? option=(EXTENDED | FORMATTED)?
+        multipartIdentifier partitionSpec? describeColName?            #describeRelation
+    | (DESC | DESCRIBE) QUERY? query                                   #describeQuery
+    | COMMENT ON namespace multipartIdentifier IS
+        comment=(STRING | NULL)                                        #commentNamespace
+    | COMMENT ON TABLE multipartIdentifier IS comment=(STRING | NULL)  #commentTable
+    | REFRESH TABLE multipartIdentifier                                #refreshTable
+    | REFRESH FUNCTION multipartIdentifier                             #refreshFunction
+    | REFRESH (STRING | .*?)                                           #refreshResource
+    | CACHE LAZY? TABLE multipartIdentifier
+        (OPTIONS options=tablePropertyList)? (AS? query)?              #cacheTable
+    | UNCACHE TABLE (IF EXISTS)? multipartIdentifier                   #uncacheTable
+    | CLEAR CACHE                                                      #clearCache
+    | LOAD DATA LOCAL? INPATH path=STRING OVERWRITE? INTO TABLE
+        multipartIdentifier partitionSpec?                             #loadData
+    | TRUNCATE TABLE multipartIdentifier partitionSpec?                #truncateTable
+    | MSCK REPAIR TABLE multipartIdentifier                            #repairTable
+    | op=(ADD | LIST) identifier (STRING | .*?)                        #manageResource
+    | SET ROLE .*?                                                     #failNativeCommand
+    | SET TIME ZONE interval                                           #setTimeZone
+    | SET TIME ZONE timezone=(STRING | LOCAL)                          #setTimeZone
+    | SET TIME ZONE .*?                                                #setTimeZone
+    | SET configKey (EQ .*?)?                                          #setQuotedConfiguration
+    | SET .*?                                                          #setConfiguration
+    | RESET configKey                                                  #resetQuotedConfiguration
+    | RESET .*?                                                        #resetConfiguration
+    | unsupportedHiveNativeCommands .*?                                #failNativeCommand
+    ;
+
+configKey
+    : quotedIdentifier
+    ;
+
+unsupportedHiveNativeCommands
+    : kw1=CREATE kw2=ROLE
+    | kw1=DROP kw2=ROLE
+    | kw1=GRANT kw2=ROLE?
+    | kw1=REVOKE kw2=ROLE?
+    | kw1=SHOW kw2=GRANT
+    | kw1=SHOW kw2=ROLE kw3=GRANT?
+    | kw1=SHOW kw2=PRINCIPALS
+    | kw1=SHOW kw2=ROLES
+    | kw1=SHOW kw2=CURRENT kw3=ROLES
+    | kw1=EXPORT kw2=TABLE
+    | kw1=IMPORT kw2=TABLE
+    | kw1=SHOW kw2=COMPACTIONS
+    | kw1=SHOW kw2=CREATE kw3=TABLE
+    | kw1=SHOW kw2=TRANSACTIONS
+    | kw1=SHOW kw2=INDEXES
+    | kw1=SHOW kw2=LOCKS
+    | kw1=CREATE kw2=INDEX
+    | kw1=DROP kw2=INDEX
+    | kw1=ALTER kw2=INDEX
+    | kw1=LOCK kw2=TABLE
+    | kw1=LOCK kw2=DATABASE
+    | kw1=UNLOCK kw2=TABLE
+    | kw1=UNLOCK kw2=DATABASE
+    | kw1=CREATE kw2=TEMPORARY kw3=MACRO
+    | kw1=DROP kw2=TEMPORARY kw3=MACRO
+    | kw1=ALTER kw2=TABLE tableIdentifier kw3=NOT kw4=CLUSTERED
+    | kw1=ALTER kw2=TABLE tableIdentifier kw3=CLUSTERED kw4=BY
+    | kw1=ALTER kw2=TABLE tableIdentifier kw3=NOT kw4=SORTED
+    | kw1=ALTER kw2=TABLE tableIdentifier kw3=SKEWED kw4=BY
+    | kw1=ALTER kw2=TABLE tableIdentifier kw3=NOT kw4=SKEWED
+    | kw1=ALTER kw2=TABLE tableIdentifier kw3=NOT kw4=STORED kw5=AS kw6=DIRECTORIES
+    | kw1=ALTER kw2=TABLE tableIdentifier kw3=SET kw4=SKEWED kw5=LOCATION
+    | kw1=ALTER kw2=TABLE tableIdentifier kw3=EXCHANGE kw4=PARTITION
+    | kw1=ALTER kw2=TABLE tableIdentifier kw3=ARCHIVE kw4=PARTITION
+    | kw1=ALTER kw2=TABLE tableIdentifier kw3=UNARCHIVE kw4=PARTITION
+    | kw1=ALTER kw2=TABLE tableIdentifier kw3=TOUCH
+    | kw1=ALTER kw2=TABLE tableIdentifier partitionSpec? kw3=COMPACT
+    | kw1=ALTER kw2=TABLE tableIdentifier partitionSpec? kw3=CONCATENATE
+    | kw1=ALTER kw2=TABLE tableIdentifier partitionSpec? kw3=SET kw4=FILEFORMAT
+    | kw1=ALTER kw2=TABLE tableIdentifier partitionSpec? kw3=REPLACE kw4=COLUMNS
+    | kw1=START kw2=TRANSACTION
+    | kw1=COMMIT
+    | kw1=ROLLBACK
+    | kw1=DFS
+    ;
+
+createTableHeader
+    : CREATE TEMPORARY? EXTERNAL? TABLE (IF NOT EXISTS)? multipartIdentifier
+    ;
+
+replaceTableHeader
+    : (CREATE OR)? REPLACE TABLE multipartIdentifier
+    ;
+
+bucketSpec
+    : CLUSTERED BY identifierList
+      (SORTED BY orderedIdentifierList)?
+      INTO INTEGER_VALUE BUCKETS
+    ;
+
+skewSpec
+    : SKEWED BY identifierList
+      ON (constantList | nestedConstantList)
+      (STORED AS DIRECTORIES)?
+    ;
+
+locationSpec
+    : LOCATION STRING
+    ;
+
+commentSpec
+    : COMMENT STRING
+    ;
+
+query
+    : ctes? queryTerm queryOrganization
+    ;
+
+insertInto
+    : INSERT OVERWRITE TABLE? multipartIdentifier (partitionSpec (IF NOT EXISTS)?)?                         #insertOverwriteTable
+    | INSERT INTO TABLE? multipartIdentifier partitionSpec? (IF NOT EXISTS)?                                #insertIntoTable
+    | INSERT OVERWRITE LOCAL? DIRECTORY path=STRING rowFormat? createFileFormat?                            #insertOverwriteHiveDir
+    | INSERT OVERWRITE LOCAL? DIRECTORY (path=STRING)? tableProvider (OPTIONS options=tablePropertyList)?   #insertOverwriteDir
+    ;
+
+partitionSpecLocation
+    : partitionSpec locationSpec?
+    ;
+
+partitionSpec
+    : PARTITION '(' partitionVal (',' partitionVal)* ')'
+    ;
+
+partitionVal
+    : identifier (EQ constant)?
+    ;
+
+namespace
+    : NAMESPACE
+    | DATABASE
+    | SCHEMA
+    ;
+
+describeFuncName
+    : qualifiedName
+    | STRING
+    | comparisonOperator
+    | arithmeticOperator
+    | predicateOperator
+    ;
+
+describeColName
+    : nameParts+=identifier ('.' nameParts+=identifier)*
+    ;
+
+ctes
+    : WITH namedQuery (',' namedQuery)*
+    ;
+
+namedQuery
+    : name=errorCapturingIdentifier (columnAliases=identifierList)? AS? '(' query ')'
+    ;
+
+tableProvider
+    : USING multipartIdentifier
+    ;
+
+createTableClauses
+    :((OPTIONS options=tablePropertyList) |
+     (PARTITIONED BY partitioning=transformList) |
+     bucketSpec |
+     locationSpec |
+     commentSpec |
+     (TBLPROPERTIES tableProps=tablePropertyList))*
+    ;
+
+tablePropertyList
+    : '(' tableProperty (',' tableProperty)* ')'
+    ;
+
+tableProperty
+    : key=tablePropertyKey (EQ? value=tablePropertyValue)?
+    ;
+
+tablePropertyKey
+    : identifier ('.' identifier)*
+    | STRING
+    ;
+
+tablePropertyValue
+    : INTEGER_VALUE
+    | DECIMAL_VALUE
+    | booleanValue
+    | STRING
+    ;
+
+constantList
+    : '(' constant (',' constant)* ')'
+    ;
+
+nestedConstantList
+    : '(' constantList (',' constantList)* ')'
+    ;
+
+createFileFormat
+    : STORED AS fileFormat
+    | STORED BY storageHandler
+    ;
+
+fileFormat
+    : INPUTFORMAT inFmt=STRING OUTPUTFORMAT outFmt=STRING    #tableFileFormat
+    | identifier                                             #genericFileFormat
+    ;
+
+storageHandler
+    : STRING (WITH SERDEPROPERTIES tablePropertyList)?
+    ;
+
+resource
+    : identifier STRING
+    ;
+
+dmlStatementNoWith
+    : insertInto queryTerm queryOrganization                                       #singleInsertQuery
+    | fromClause multiInsertQueryBody+                                             #multiInsertQuery
+    | DELETE FROM multipartIdentifier tableAlias whereClause?                      #deleteFromTable
+    | UPDATE multipartIdentifier tableAlias setClause whereClause?                 #updateTable
+    | MERGE INTO target=multipartIdentifier targetAlias=tableAlias
+        USING (source=multipartIdentifier |
+          '(' sourceQuery=query')') sourceAlias=tableAlias
+        ON mergeCondition=booleanExpression
+        matchedClause*
+        notMatchedClause*                                                          #mergeIntoTable
+    ;
+
+mergeInto
+    : MERGE INTO target=multipartIdentifier targetAlias=tableAlias
+              USING (source=multipartIdentifier |
+                '(' sourceQuery=query')') sourceAlias=tableAlias
+              ON mergeCondition=booleanExpression
+              matchedClause*
+              notMatchedClause*

Review comment:
       how about  change it to following grammer:
   (matchedClause | notMatchedClause)
   (matchedClause | notMatchedClause) *




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

CarbonDataQA2 commented on pull request #4032:
URL: https://github.com/apache/carbondata/pull/4032#issuecomment-750726829


   Build Success with Spark 2.3.4, Please check CI http://121.244.95.60:12444/job/ApacheCarbonPRBuilder2.3/5243/
   


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

CarbonDataQA2 commented on pull request #4032:
URL: https://github.com/apache/carbondata/pull/4032#issuecomment-750728073


   Build Success with Spark 2.4.5, Please check CI http://121.244.95.60:12444/job/ApacheCarbon_PR_Builder_2.4.5/3482/
   


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

CarbonDataQA2 commented on pull request #4032:
URL: https://github.com/apache/carbondata/pull/4032#issuecomment-751671001


   Build Success with Spark 2.4.5, Please check CI http://121.244.95.60:12444/job/ApacheCarbon_PR_Builder_2.4.5/3488/
   


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

CarbonDataQA2 commented on pull request #4032:
URL: https://github.com/apache/carbondata/pull/4032#issuecomment-751672614


   Build Success with Spark 2.3.4, Please check CI http://121.244.95.60:12444/job/ApacheCarbonPRBuilder2.3/5249/
   


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

CarbonDataQA2 commented on pull request #4032:
URL: https://github.com/apache/carbondata/pull/4032#issuecomment-751944960


   Build Success with Spark 2.4.5, Please check CI http://121.244.95.60:12444/job/ApacheCarbon_PR_Builder_2.4.5/3492/
   


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

CarbonDataQA2 commented on pull request #4032:
URL: https://github.com/apache/carbondata/pull/4032#issuecomment-751945359


   Build Success with Spark 2.3.4, Please check CI http://121.244.95.60:12444/job/ApacheCarbonPRBuilder2.3/5253/
   


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

ajantha-bhat commented on a change in pull request #4032:
URL: https://github.com/apache/carbondata/pull/4032#discussion_r550194163



##########
File path: integration/spark/src/main/antlr4/org/apache/spark/sql/parser/CarbonSqlBase.g4
##########
@@ -0,0 +1,1842 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * This file is an adaptation of Presto's presto-parser/src/main/antlr4/com/facebook/presto/sql/parser/CarbonSqlBase.g4 grammar.
+ */
+
+grammar CarbonSqlBase;
+
+@parser::members {

Review comment:
       agree with @QiangCai




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

ajantha-bhat commented on a change in pull request #4032:
URL: https://github.com/apache/carbondata/pull/4032#discussion_r550380747



##########
File path: examples/spark/src/main/scala/org/apache/carbondata/examples/DataMergeIntoExample.scala
##########
@@ -0,0 +1,164 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.carbondata.examples
+
+import org.apache.spark.sql.SparkSession
+
+import org.apache.carbondata.examples.util.ExampleUtils
+
+object DataMergeIntoExample {

Review comment:
       Please also update `docs/scd-and-cdc-guide.md`
   
   it has below note, remove it and update the SQL syntax along with the link to this example file
   `**NOTE:** SQL syntax for merge is not yet supported.`




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

CarbonDataQA2 commented on pull request #4032:
URL: https://github.com/apache/carbondata/pull/4032#issuecomment-752839169


   Build Success with Spark 2.3.4, Please check CI http://121.244.95.60:12444/job/ApacheCarbonPRBuilder2.3/5267/
   


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

CarbonDataQA2 commented on pull request #4032:
URL: https://github.com/apache/carbondata/pull/4032#issuecomment-752839555


   Build Failed  with Spark 2.4.5, Please check CI http://121.244.95.60:12444/job/ApacheCarbon_PR_Builder_2.4.5/3506/
   


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]