HIVE-28988: Handle materialized view commit txn during hive acid repl

* Details:
  * Commit txn of alter mv command was getting populated by write event infos.
  * This was causing a repl failure on load.
  * This is now fixed.

* Testing:
  * Added itest.

Author: shivamkmr02

itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenarios.java

@@ -3245,6 +3245,9 @@ public void testMaterializedViewsReplication() throws Exception {
     run("CREATE MATERIALIZED VIEW " + dbName + ".mat_view_boot2 disable rewrite  stored as textfile AS SELECT t1.a FROM " + dbName + ".unptned as t1 join " + dbName2 + ".unptned as t2 on t1.a = t2.a", driver);
     verifySetup("SELECT a from " + dbName + ".mat_view_boot2", unptn_data, driver);
 
+    run("CREATE MATERIALIZED VIEW " + dbName + ".mat_view_rebuild disable rewrite  stored as textfile AS SELECT a FROM " + dbName + ".unptned", driver);
+    verifySetup("SELECT a from " + dbName + ".mat_view_rebuild", unptn_data, driver);
+
     Tuple bootstrapDump = bootstrapLoadAndVerify(dbName, replDbName);
 
     verifyRun("SELECT * from " + replDbName + ".unptned", unptn_data, driverMirror);
@@ -3253,6 +3256,7 @@ public void testMaterializedViewsReplication() throws Exception {
     //verify source MVs are not on replica
     verifyIfTableNotExist(replDbName, "mat_view_boot", metaStoreClientMirror);
     verifyIfTableNotExist(replDbName, "mat_view_boot2", metaStoreClientMirror);
+    verifyIfTableNotExist(replDbName, "mat_view_rebuild", metaStoreClientMirror);
 
     //test alter materialized view with rename
     run("ALTER TABLE " + dbName + ".mat_view_boot RENAME TO " + dbName + ".mat_view_rename", driver);
@@ -3261,12 +3265,22 @@ public void testMaterializedViewsReplication() throws Exception {
     verifyIfTableNotExist(dbName, "mat_view_boot", metaStoreClient);
     verifyIfTableExist(dbName, "mat_view_rename", metaStoreClient);
 
+    String[] unptn_incr_data = new String[]{ "twenty", "thirty" };
+    String[] unptn_total_data = new String[]{ "twenty", "thirty", "eleven", "twelve" };
+    // Rebuild mv mat_view_rebuild after inserting some more data
+    run("INSERT INTO TABLE " + dbName + ".unptned values('" + unptn_incr_data[0] + "')", driver);
+    run("INSERT INTO TABLE " + dbName + ".unptned values('" + unptn_incr_data[1] + "')", driver);
+
+    run("ALTER MATERIALIZED VIEW " + dbName + ".mat_view_rebuild REBUILD" , driver);
+    verifyRun("SELECT a from " + dbName + ".mat_view_rebuild", unptn_total_data, driver);
+
     // Perform REPL-DUMP/LOAD
     Tuple incrementalDump = incrementalLoadAndVerify(dbName, replDbName);
 
     //verify source MVs are not on replica
     verifyIfTableNotExist(replDbName, "mat_view_rename", metaStoreClientMirror);
     verifyIfTableNotExist(replDbName, "mat_view_boot2", metaStoreClientMirror);
+    verifyIfTableNotExist(replDbName, "mat_view_rebuild", metaStoreClientMirror);
 
     //test alter materialized view rebuild
     run("ALTER MATERIALIZED VIEW " + dbName + ".mat_view_boot2 REBUILD" , driver);

ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/CommitTxnHandler.java

@@ -27,6 +27,7 @@
 import org.apache.hadoop.hive.metastore.api.NotificationEvent;
 import org.apache.hadoop.hive.metastore.api.WriteEventInfo;
 import org.apache.hadoop.hive.metastore.messaging.CommitTxnMessage;
+import org.apache.hadoop.hive.metastore.messaging.MessageBuilder;
 import org.apache.hadoop.hive.metastore.utils.StringUtils;
 import org.apache.hadoop.hive.ql.exec.repl.util.ReplUtils;
 import org.apache.hadoop.hive.ql.metadata.HiveFatalException;
@@ -43,6 +44,7 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.stream.Collectors;
 
 class CommitTxnHandler extends AbstractEventHandler<CommitTxnMessage> {
 
@@ -128,6 +130,20 @@ private List<WriteEventInfo> getAllWriteEventInfo(Context withinContext) throws
               })));
   }
 
+  private List<WriteEventInfo> getAllWriteEventInfoExceptMV(List<WriteEventInfo> writeEventInfoList) {
+    return writeEventInfoList.stream().filter(writeEventInfo -> {
+      try {
+        if (writeEventInfo.getTableObj() != null) {
+          Table table = new Table((org.apache.hadoop.hive.metastore.api.Table) MessageBuilder.getTObj(writeEventInfo.getTableObj(), org.apache.hadoop.hive.metastore.api.Table.class));
+          return !table.isMaterializedView();
+        }
+        return true;
+      } catch (Exception e) {
+        throw new RuntimeException(e);
+      }
+    }).collect(Collectors.toList());
+  }
+
   @Override
   public void handle(Context withinContext) throws Exception {
     if (!ReplUtils.includeAcidTableInDump(withinContext.hiveConf)) {
@@ -168,6 +184,10 @@ public void handle(Context withinContext) throws Exception {
         }
       }
 
+      // Filtering out all write event info related to materialized view
+      if (writeEventInfoList != null) {
+        writeEventInfoList = getAllWriteEventInfoExceptMV(writeEventInfoList);
+      }
       int numEntry = (writeEventInfoList != null ? writeEventInfoList.size() : 0);
       if (numEntry != 0) {
         eventMessage.addWriteEventInfo(writeEventInfoList);