diff --git a/be/src/service/client-request-state.cc b/be/src/service/client-request-state.cc
index 60acc3c..84c250b 100644
--- a/be/src/service/client-request-state.cc
+++ b/be/src/service/client-request-state.cc
@@ -1241,6 +1241,7 @@
       catalog_update.is_overwrite = finalize_params.is_overwrite;
       if (InTransaction()) {
         catalog_update.__set_transaction_id(finalize_params.transaction_id);
+        catalog_update.__set_write_id(finalize_params.write_id);
       }
 
       Status cnxn_status;
diff --git a/common/thrift/CatalogService.thrift b/common/thrift/CatalogService.thrift
index de3cd4f..5acdb7f 100644
--- a/common/thrift/CatalogService.thrift
+++ b/common/thrift/CatalogService.thrift
@@ -215,7 +215,10 @@
   7: required bool is_overwrite;
 
   // ACID transaction ID for transactional inserts.
-  8: optional i64 transaction_id;
+  8: optional i64 transaction_id
+
+  // ACID write ID for transactional inserts.
+  9: optional i64 write_id
 }
 
 // Response from a TUpdateCatalogRequest
diff --git a/fe/src/compat-hive-3/java/org/apache/impala/compat/MetastoreShim.java b/fe/src/compat-hive-3/java/org/apache/impala/compat/MetastoreShim.java
index dd41715..de7c586 100644
--- a/fe/src/compat-hive-3/java/org/apache/impala/compat/MetastoreShim.java
+++ b/fe/src/compat-hive-3/java/org/apache/impala/compat/MetastoreShim.java
@@ -25,8 +25,10 @@
 import static org.apache.impala.service.MetadataOp.TABLE_TYPE_VIEW;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Stopwatch;
 import com.google.common.collect.ImmutableMap;
 
+import com.google.common.collect.Iterables;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
 import java.util.ArrayList;
@@ -35,6 +37,7 @@
 import java.util.Collections;
 import java.util.EnumSet;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 import org.apache.hadoop.hive.common.StatsSetupConst;
@@ -74,6 +77,7 @@
 import org.apache.hadoop.hive.metastore.api.Table;
 import org.apache.hadoop.hive.metastore.api.TableValidWriteIds;
 import org.apache.hadoop.hive.metastore.api.TxnAbortedException;
+import org.apache.hadoop.hive.metastore.api.WriteNotificationLogRequest;
 import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
 import org.apache.hadoop.hive.metastore.messaging.AlterTableMessage;
 import org.apache.hadoop.hive.metastore.messaging.EventMessage;
@@ -102,7 +106,7 @@
 import org.apache.impala.thrift.TValidWriteIdList;
 import org.apache.impala.util.AcidUtils;
 import org.apache.impala.util.AcidUtils.TblTransaction;
-import org.apache.impala.util.MetaStoreUtil.InsertEventInfo;
+import org.apache.impala.util.MetaStoreUtil.TableInsertEventInfo;
 import org.apache.thrift.TException;
 
 import com.google.common.base.Preconditions;
@@ -1061,60 +1065,102 @@
    * In case of any exception, we just log the failure of firing insert events.
    */
   public static List<Long> fireInsertEvents(MetaStoreClient msClient,
-      List<InsertEventInfo> insertEventInfos, String dbName, String tableName) {
+      TableInsertEventInfo insertEventInfo, String dbName, String tableName) {
+    Stopwatch sw = Stopwatch.createStarted();
     try {
-      return fireInsertEventHelper(msClient.getHiveClient(), insertEventInfos, dbName, tableName);
+      if (insertEventInfo.isTransactional()) {
+        // if the table is transactional we use a different API to fire these
+        // events. Note that we don't really need the event ids here for self-event
+        // detection since these events are not fetched by the EventsProcessor later
+        // These events are mostly required for incremental replication in Hive
+        fireInsertTransactionalEventHelper(msClient.getHiveClient(),
+            insertEventInfo, dbName, tableName);
+      } else {
+        return fireInsertEventHelper(msClient.getHiveClient(),
+            insertEventInfo.getInsertEventReqData(), dbName,
+            tableName);
+      }
     } catch (Exception e) {
       LOG.error("Failed to fire insert event. Some tables might not be"
               + " refreshed on other impala clusters.", e);
     } finally {
+      LOG.info("Time taken to fire insert events on table {}.{}: {} msec", dbName,
+          tableName, sw.stop().elapsed(TimeUnit.MILLISECONDS));
       msClient.close();
     }
     return Collections.emptyList();
   }
 
   /**
+   * Fires a listener event of the type ACID_WRITE on a transactional table in metastore.
+   * This event is polled by other external systems to detect insert operations into
+   * ACID tables.
+   * @throws TException in case of errors during HMS API call.
+   */
+  private static void fireInsertTransactionalEventHelper(
+      IMetaStoreClient hiveClient, TableInsertEventInfo insertEventInfo, String dbName,
+      String tableName) throws TException {
+    for (InsertEventRequestData insertData : insertEventInfo.getInsertEventReqData()) {
+      // TODO(Vihang) unfortunately there is no bulk insert event API for transactional
+      // tables. It is possible that this may take long time here if there are lots of
+      // partitions which were inserted.
+      if (LOG.isDebugEnabled()) {
+        String msg =
+            "Firing write notification log request for table " + dbName + "." + tableName
+                + (insertData.isSetPartitionVal() ? " on partition " + insertData
+                .getPartitionVal() : "");
+        LOG.debug(msg);
+      }
+      WriteNotificationLogRequest rqst = new WriteNotificationLogRequest(
+          insertEventInfo.getTxnId(), insertEventInfo.getWriteId(), dbName, tableName,
+          insertData);
+      if (insertData.isSetPartitionVal()) {
+        rqst.setPartitionVals(insertData.getPartitionVal());
+      }
+      // TODO(Vihang) metastore should return the event id here so that we can get rid
+      // of firing INSERT event types for transactional tables.
+      hiveClient.addWriteNotificationLog(rqst);
+    }
+  }
+
+  /**
    *  Fires an insert event to HMS notification log. In Hive-3 for partitioned table,
    *  all partition insert events will be fired by a bulk API.
    *
    * @param msClient Metastore client,
-   * @param insertEventInfos A list of insert event encapsulating the information needed
-   * to fire insert
+   * @param insertEventDataList A list of insert event info encapsulating the information
+   *                            needed to fire insert events.
    * @param dbName
    * @param tableName
    * @return a list of eventIds for the insert events
    */
   @VisibleForTesting
   public static List<Long> fireInsertEventHelper(IMetaStoreClient msClient,
-      List<InsertEventInfo> insertEventInfos, String dbName, String tableName)
+      List<InsertEventRequestData> insertEventDataList, String dbName, String tableName)
       throws TException {
     Preconditions.checkNotNull(msClient);
     Preconditions.checkNotNull(dbName);
     Preconditions.checkNotNull(tableName);
-    Preconditions.checkState(!insertEventInfos.isEmpty(), "Atleast one insert event "
+    Preconditions.checkState(!insertEventDataList.isEmpty(), "Atleast one insert event "
         + "info must be provided.");
     LOG.debug(String.format(
-        "Firing %s insert event for %s", insertEventInfos.size(), tableName));
+        "Firing %s insert event(s) for %s.%s", insertEventDataList.size(), dbName,
+        tableName));
     FireEventRequestData data = new FireEventRequestData();
     FireEventRequest rqst = new FireEventRequest(true, data);
     rqst.setDbName(dbName);
     rqst.setTableName(tableName);
-    List<InsertEventRequestData> insertDatas = new ArrayList<>();
-    for (InsertEventInfo info : insertEventInfos) {
-      InsertEventRequestData insertData = new InsertEventRequestData();
-      Preconditions.checkNotNull(info.getNewFiles());
-      insertData.setFilesAdded(new ArrayList<>(info.getNewFiles()));
-      insertData.setReplace(info.isOverwrite());
-      if (info.getPartVals() != null) insertData.setPartitionVal(info.getPartVals());
-      insertDatas.add(insertData);
-    }
-    if (insertDatas.size() == 1) {
-      if (insertEventInfos.get(0).getPartVals() != null) {
-        rqst.setPartitionVals(insertEventInfos.get(0).getPartVals());
+    if (insertEventDataList.size() == 1) {
+      InsertEventRequestData insertEventData = Iterables
+          .getOnlyElement(insertEventDataList);
+      if (insertEventData.getPartitionVal() != null) {
+        rqst.setPartitionVals(insertEventData.getPartitionVal());
       }
-      data.setInsertData(insertDatas.get(0));
+      // single insert event API
+      data.setInsertData(insertEventData);
     } else {
-      data.setInsertDatas(insertDatas);
+      // use bulk insert API
+      data.setInsertDatas(insertEventDataList);
     }
     FireEventResponse response = msClient.fireListenerEvent(rqst);
     if (!response.isSetEventIds()) {
diff --git a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
index f4e562a..0400d3a 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
@@ -735,11 +735,13 @@
   }
 
   public FileSystem getFileSystem() throws CatalogException {
+    FileSystem tableFs;
     try {
-      return (new Path(getLocation())).getFileSystem(CONF);
+      tableFs = (new Path(getLocation())).getFileSystem(CONF);
     } catch (IOException e) {
       throw new CatalogException("Invalid table path for table: " + getFullName(), e);
     }
+    return tableFs;
   }
 
   /**
diff --git a/fe/src/main/java/org/apache/impala/catalog/HiveStorageDescriptorFactory.java b/fe/src/main/java/org/apache/impala/catalog/HiveStorageDescriptorFactory.java
index c649808..c645cfc 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HiveStorageDescriptorFactory.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HiveStorageDescriptorFactory.java
@@ -17,6 +17,7 @@
 
 package org.apache.impala.catalog;
 
+import java.util.ArrayList;
 import java.util.HashMap;
 
 import org.apache.hadoop.hive.metastore.api.SerDeInfo;
@@ -47,6 +48,8 @@
     sd.setInputFormat(hdfsFileFormat.inputFormat());
     sd.setOutputFormat(hdfsFileFormat.outputFormat());
     sd.getSerdeInfo().setSerializationLib(hdfsFileFormat.serializationLib());
+    sd.setBucketCols(new ArrayList<>(0));
+    sd.setSortCols(new ArrayList<>(0));
     setSerdeInfo(rowFormat, sd.getSerdeInfo());
     return sd;
   }
diff --git a/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEventsProcessor.java b/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEventsProcessor.java
index 8b9bba3..2e38d4c 100644
--- a/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEventsProcessor.java
+++ b/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEventsProcessor.java
@@ -233,7 +233,7 @@
   private final long pollingFrequencyInSec_;
 
   // catalog service instance to be used while processing events
-  private final CatalogServiceCatalog catalog_;
+  protected final CatalogServiceCatalog catalog_;
 
   // scheduler daemon thread executor for processing events at given frequency
   private final ScheduledExecutorService scheduler_ = Executors
diff --git a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
index e8270bb..88801a5 100644
--- a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
+++ b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
@@ -19,9 +19,15 @@
 
 import static org.apache.impala.analysis.Analyzer.ACCESSTYPE_READWRITE;
 
+import com.codahale.metrics.Timer;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Joiner;
+import com.google.common.base.Preconditions;
 import com.google.common.base.Stopwatch;
 import com.google.common.collect.Iterables;
-
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.util.ArrayList;
@@ -36,8 +42,8 @@
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
-
 import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileChecksum;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hive.common.StatsSetupConst;
@@ -54,6 +60,7 @@
 import org.apache.hadoop.hive.metastore.api.DataOperationType;
 import org.apache.hadoop.hive.metastore.api.Database;
 import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.InsertEventRequestData;
 import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
 import org.apache.hadoop.hive.metastore.api.Partition;
 import org.apache.hadoop.hive.metastore.api.PrincipalType;
@@ -61,7 +68,9 @@
 import org.apache.hadoop.hive.metastore.api.SQLPrimaryKey;
 import org.apache.hadoop.hive.metastore.api.SerDeInfo;
 import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.TruncateTableRequest;
 import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.util.StringUtils;
 import org.apache.impala.analysis.AlterTableSortByStmt;
 import org.apache.impala.analysis.FunctionName;
 import org.apache.impala.analysis.TableName;
@@ -100,8 +109,8 @@
 import org.apache.impala.catalog.Type;
 import org.apache.impala.catalog.View;
 import org.apache.impala.catalog.events.MetastoreEvents.MetastoreEventPropertyKey;
-import org.apache.impala.catalog.monitor.CatalogOperationMetrics;
 import org.apache.impala.catalog.monitor.CatalogMonitor;
+import org.apache.impala.catalog.monitor.CatalogOperationMetrics;
 import org.apache.impala.common.FileSystemUtil;
 import org.apache.impala.common.ImpalaException;
 import org.apache.impala.common.ImpalaRuntimeException;
@@ -141,6 +150,7 @@
 import org.apache.impala.thrift.TColumnType;
 import org.apache.impala.thrift.TColumnValue;
 import org.apache.impala.thrift.TCommentOnParams;
+import org.apache.impala.thrift.TCopyTestCaseReq;
 import org.apache.impala.thrift.TCreateDataSourceParams;
 import org.apache.impala.thrift.TCreateDbParams;
 import org.apache.impala.thrift.TCreateDropRoleParams;
@@ -163,7 +173,6 @@
 import org.apache.impala.thrift.TGrantRevokeRoleParams;
 import org.apache.impala.thrift.THdfsCachingOp;
 import org.apache.impala.thrift.THdfsFileFormat;
-import org.apache.impala.thrift.TCopyTestCaseReq;
 import org.apache.impala.thrift.TPartitionDef;
 import org.apache.impala.thrift.TPartitionKeyValue;
 import org.apache.impala.thrift.TPartitionStats;
@@ -183,25 +192,16 @@
 import org.apache.impala.thrift.TTruncateParams;
 import org.apache.impala.thrift.TUpdateCatalogRequest;
 import org.apache.impala.thrift.TUpdateCatalogResponse;
-import org.apache.impala.util.CompressionUtil;
 import org.apache.impala.util.AcidUtils;
 import org.apache.impala.util.AcidUtils.TblTransaction;
+import org.apache.impala.util.CompressionUtil;
 import org.apache.impala.util.FunctionUtils;
 import org.apache.impala.util.HdfsCachingUtil;
 import org.apache.impala.util.KuduUtil;
 import org.apache.impala.util.MetaStoreUtil;
-import org.apache.impala.util.MetaStoreUtil.InsertEventInfo;
-import org.slf4j.Logger;
+import org.apache.impala.util.MetaStoreUtil.TableInsertEventInfo;
 import org.apache.thrift.TException;
-
-import com.codahale.metrics.Timer;
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Joiner;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
-
+import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
@@ -2080,33 +2080,45 @@
         catalog_.getLock().writeLock().unlock();
         TblTransaction tblTxn = MetastoreShim.createTblTransaction(hmsClient,
             table.getMetaStoreTable(), txn.getId());
-        HdfsTable hdfsTable = (HdfsTable)table;
-        Collection<? extends FeFsPartition> parts =
-            FeCatalogUtils.loadAllPartitions(hdfsTable);
-        createEmptyBaseDirectories(parts, tblTxn.writeId);
-        // Currently Impala cannot update the statistics properly. So instead of
-        // writing correct stats, let's just remove COLUMN_STATS_ACCURATE parameter from
-        // each partition.
-        // TODO(IMPALA-8883): properly update statistics
-        List<org.apache.hadoop.hive.metastore.api.Partition> hmsPartitions =
-            Lists.newArrayListWithCapacity(parts.size());
-        if (table.getNumClusteringCols() > 0) {
-          for (FeFsPartition part: parts) {
-            org.apache.hadoop.hive.metastore.api.Partition hmsPart =
-                ((HdfsPartition)part).toHmsPartition();
-            Preconditions.checkNotNull(hmsPart);
-            if (hmsPart.getParameters() != null) {
-              hmsPart.getParameters().remove(StatsSetupConst.COLUMN_STATS_ACCURATE);
-              hmsPartitions.add(hmsPart);
+        HdfsTable hdfsTable = (HdfsTable) table;
+        // if the table is replicated we should use the HMS API to truncate it so that
+        // if moves the files into in replication change manager location which is later
+        // used for replication.
+        if (isTableBeingReplicated(hmsClient, hdfsTable)) {
+          Stopwatch sw = Stopwatch.createStarted();
+          String dbName = Preconditions.checkNotNull(hdfsTable.getDb()).getName();
+          hmsClient.truncateTable(dbName, hdfsTable.getName(), null, tblTxn.validWriteIds,
+              tblTxn.writeId);
+          LOG.debug("Time taken to truncate table {} using HMS API: {} msec",
+              hdfsTable.getFullName(), sw.stop().elapsed(TimeUnit.MILLISECONDS));
+        } else {
+          Collection<? extends FeFsPartition> parts =
+              FeCatalogUtils.loadAllPartitions(hdfsTable);
+          createEmptyBaseDirectories(parts, tblTxn.writeId);
+          // Currently Impala cannot update the statistics properly. So instead of
+          // writing correct stats, let's just remove COLUMN_STATS_ACCURATE parameter from
+          // each partition.
+          // TODO(IMPALA-8883): properly update statistics
+          List<org.apache.hadoop.hive.metastore.api.Partition> hmsPartitions =
+              Lists.newArrayListWithCapacity(parts.size());
+          if (table.getNumClusteringCols() > 0) {
+            for (FeFsPartition part : parts) {
+              org.apache.hadoop.hive.metastore.api.Partition hmsPart =
+                  ((HdfsPartition) part).toHmsPartition();
+              Preconditions.checkNotNull(hmsPart);
+              if (hmsPart.getParameters() != null) {
+                hmsPart.getParameters().remove(StatsSetupConst.COLUMN_STATS_ACCURATE);
+                hmsPartitions.add(hmsPart);
+              }
             }
           }
+          // For partitioned tables we need to alter all the partitions in HMS.
+          if (!hmsPartitions.isEmpty()) {
+            unsetPartitionsColStats(table.getMetaStoreTable(), hmsPartitions, tblTxn);
+          }
+          // Remove COLUMN_STATS_ACCURATE property from the table.
+          unsetTableColStats(table.getMetaStoreTable(), tblTxn);
         }
-        // For partitioned tables we need to alter all the partitions in HMS.
-        if (!hmsPartitions.isEmpty()) {
-          unsetPartitionsColStats(table.getMetaStoreTable(), hmsPartitions, tblTxn);
-        }
-        // Remove COLUMN_STATS_ACCURATE property from the table.
-        unsetTableColStats(table.getMetaStoreTable(), tblTxn);
         txn.commit();
       }
       return newCatalogVersion;
@@ -2116,6 +2128,27 @@
     }
   }
 
+  /**
+   * Helper method to check if the database which table belongs to is a source of
+   * Hive replication. We cannot rely on the Db object here due to eventual nature of
+   * cache updates.
+   */
+  private boolean isTableBeingReplicated(IMetaStoreClient metastoreClient,
+      HdfsTable tbl) throws CatalogException {
+    Preconditions.checkState(tbl.getLock().isHeldByCurrentThread());
+    String dbName = tbl.getDb().getName();
+    try {
+      Database db = metastoreClient.getDatabase(dbName);
+      if (!db.isSetParameters()) return false;
+      return org.apache.commons.lang.StringUtils
+          .isNotEmpty(db.getParameters().get("repl.source.for"));
+    } catch (TException tException) {
+      throw new CatalogException(
+          String.format("Could not determine if the table %s is a replication source",
+          tbl.getFullName()), tException);
+    }
+  }
+
   private long truncateNonTransactionalTable(TTruncateParams params, Table table)
       throws Exception {
     Preconditions.checkState(table.getLock().isHeldByCurrentThread());
@@ -2123,6 +2156,18 @@
     long newCatalogVersion = catalog_.incrementAndGetCatalogVersion();
     catalog_.getLock().writeLock().unlock();
     HdfsTable hdfsTable = (HdfsTable) table;
+    // if the table is being replicated we issue the HMS API to truncate the table
+    // since it generates additional events which are used by Hive Replication.
+    try (MetaStoreClient metaStoreClient = catalog_.getMetaStoreClient()) {
+      if (isTableBeingReplicated(metaStoreClient.getHiveClient(), hdfsTable)) {
+        String dbName = Preconditions.checkNotNull(hdfsTable.getDb()).getName();
+        Stopwatch sw = Stopwatch.createStarted();
+        metaStoreClient.getHiveClient().truncateTable(dbName, hdfsTable.getName(), null);
+        LOG.debug("Time taken to truncate table {} using HMS API: {} msec",
+            hdfsTable.getFullName(), sw.stop().elapsed(TimeUnit.MILLISECONDS));
+        return newCatalogVersion;
+      }
+    }
     Collection<? extends FeFsPartition> parts = FeCatalogUtils
         .loadAllPartitions(hdfsTable);
     for (FeFsPartition part : parts) {
@@ -4516,16 +4561,37 @@
       // Commit transactional inserts on success. We don't abort the transaction
       // here in case of failures, because the client, i.e. query coordinator, is
       // always responsible for aborting transactions when queries hit errors.
+      boolean skipTransactionalInsertEvent = false;
       if (update.isSetTransaction_id()) {
         if (response.getResult().getStatus().getStatus_code() == TErrorCode.OK) {
           commitTransaction(update.getTransaction_id());
+        } else {
+          // do not generate the insert event if the transaction was aborted.
+          skipTransactionalInsertEvent = true;
         }
       }
+      // the load operation below changes the partitions in-place (this was later
+      // changed in upstream). If the partitions are loaded in-place, we cannot keep track
+      // of files before the load and hence the firing of insert events later below
+      // cannot find the new files.
+      Map<String, Set<String>> filesBeforeInsert = Maps.newHashMap();
+      if (shouldGenerateInsertEvents()) {
+        filesBeforeInsert = getPartitionNameFilesMap(affectedExistingPartitions);
+      }
       loadTableMetadata(table, newCatalogVersion, true, false, partsToLoadMetadata,
           "INSERT");
       // After loading metadata, fire insert events if external event processing is
       // enabled.
-      createInsertEvents(table, affectedExistingPartitions, update.is_overwrite);
+      if (!skipTransactionalInsertEvent) {
+        // new parts are only created in case the table is partitioned.
+        Set<String> newPartsCreated = table.getNumClusteringCols() > 0 ? Sets
+            .difference(Preconditions.checkNotNull(partsToLoadMetadata),
+                filesBeforeInsert.keySet()) : Sets.newHashSetWithExpectedSize(0);
+        long txnId = tblTxn == null ? -1 : tblTxn.txnId;
+        long writeId = tblTxn == null ? -1: tblTxn.writeId;
+        createInsertEvents(table, filesBeforeInsert, newPartsCreated,
+            update.is_overwrite, txnId, writeId);
+      }
       addTableToCatalogUpdate(table, response.result);
     } finally {
       context.stop();
@@ -4544,89 +4610,91 @@
    * Populates insert event data and calls fireInsertEvents() if external event
    * processing is enabled. This is no-op if event processing is disabled or there are
    * no existing partitions affected by this insert.
-   *
-   * @param affectedExistingPartitions List of existing partitions touched by the insert.
-   * @param isInsertOverwrite indicates if the operation was an insert overwrite. If it
-   *     is not, all the new files added by this insert is calculated.
+   *  @param partitionFilesMapBeforeInsert List of files for each partition in which data
+   *                                     was inserted.
+   * @param newPartsCreated represents all the partitions names which got created by
+   *                       the insert operation.
+   * @param isInsertOverwrite indicates if the operation was an insert overwrite.
    */
   private void createInsertEvents(Table table,
-      List<FeFsPartition> affectedExistingPartitions, boolean isInsertOverwrite) {
-    boolean isInsertEventsEnabled = BackendConfig.INSTANCE.isInsertEventsEnabled();
-    if (!catalog_.isEventProcessingActive() || !isInsertEventsEnabled
-        || affectedExistingPartitions.size() == 0) {
+      Map<String, Set<String>> partitionFilesMapBeforeInsert,
+      Set<String> newPartsCreated, boolean isInsertOverwrite,
+      long txnId, long writeId) throws CatalogException {
+    if (!shouldGenerateInsertEvents()) {
       return;
     }
 
     // List of all insert events that we call HMS fireInsertEvent() on.
-    List<InsertEventInfo> insertEventInfos = new ArrayList<>();
+    List<InsertEventRequestData> insertEventReqDatas = new ArrayList<>();
     // List of all partitions that we insert into
     List<HdfsPartition> partitions = new ArrayList<>();
-
-    // Map of partition names to file names of all existing partitions touched by the
-    // insert.
-    Map<String, Set<String>> partitionFilesMapBeforeInsert = new HashMap<>();
-    if (!isInsertOverwrite) {
-      partitionFilesMapBeforeInsert =
-          getPartitionNameFilesMap(affectedExistingPartitions);
-    }
-    // If table is partitioned, we add all existing partitions touched by this insert
-    // to the insert event.
     Collection<? extends FeFsPartition> partsPostInsert;
     partsPostInsert =
-        ((HdfsTable) table).getPartitionsForNames(
-            partitionFilesMapBeforeInsert.keySet());
+        ((HdfsTable) table).getPartitionsForNames(partitionFilesMapBeforeInsert.keySet());
 
-    // If it is not an insert overwrite operation, we find new files added by this insert.
-    Map<String, Set<String>> partitionFilesMapPostInsert = new HashMap<>();
-    if (!isInsertOverwrite) {
-      partitionFilesMapPostInsert =
-          getPartitionNameFilesMap(partsPostInsert);
-    }
+    Map<String, Set<String>> partitionFilesMapPostInsert = getPartitionNameFilesMap(
+        partsPostInsert);
 
     for (FeFsPartition part : partsPostInsert) {
       // Find the delta of the files added by the insert if it is not an overwrite
       // operation. HMS fireListenerEvent() expects an empty list if no new files are
       // added or if the operation is an insert overwrite.
       HdfsPartition hdfsPartition = (HdfsPartition) part;
-      Set<String> deltaFiles = new HashSet<>();
+      // newFiles keeps track of newly added files by this insert operation.
+      Set<String> newFiles;
       List<String> partVals = null;
-      if (!isInsertOverwrite) {
-        String partitionName = hdfsPartition.getPartitionName() + "/";
-        Set<String> filesPostInsert =
-            partitionFilesMapPostInsert.get(partitionName);
-        if (table.getNumClusteringCols() > 0) {
-          Set<String> filesBeforeInsert =
-              partitionFilesMapBeforeInsert.get(partitionName);
-          deltaFiles = Sets.difference(filesBeforeInsert, filesPostInsert);
-          partVals = hdfsPartition.getPartitionValuesAsStrings(true);
-        } else {
-          Map.Entry<String, Set<String>> entry =
-              partitionFilesMapBeforeInsert.entrySet().iterator().next();
-          deltaFiles = Sets.difference(entry.getValue(), filesPostInsert);
-        }
-        LOG.info("{} new files detected for table {} partition {}.",
-            filesPostInsert.size(), table.getTableName(),
-            hdfsPartition.getPartitionName());
-      }
-      if (deltaFiles != null || isInsertOverwrite) {
+      String partitionName = hdfsPartition.getPartitionName() + "/";
+      Set<String> filesPostInsert =
+          partitionFilesMapPostInsert.get(partitionName);
+      Set<String> filesBeforeInsert =
+          partitionFilesMapBeforeInsert.get(partitionName);
+      newFiles = isInsertOverwrite ? filesPostInsert
+          : Sets.difference(filesPostInsert, filesBeforeInsert);
+      // if the table is unpartitioned partVals will be empty
+      partVals = hdfsPartition.getPartitionValuesAsStrings(true);
+      LOG.info(String.format("%s new files detected for table %s%s", newFiles.size(),
+          table.getFullName(),
+          ((HdfsTable) table).isPartitioned() ? " partition " + hdfsPartition
+              .getPartitionName() : ""));
+      if (!newFiles.isEmpty()) {
         // Collect all the insert events.
-        insertEventInfos.add(
-            new InsertEventInfo(partVals, deltaFiles, isInsertOverwrite));
-        if (partVals != null) {
+        insertEventReqDatas.add(
+            makeInsertEventData((HdfsTable) table, partVals, newFiles,
+                isInsertOverwrite));
+        if (!partVals.isEmpty()) {
           // insert into partition
           partitions.add(hdfsPartition);
         }
-      } else {
-        LOG.info("No new files were created, and is not a replace. Skipping "
-            + "generating INSERT event.");
       }
     }
 
-    if (insertEventInfos.isEmpty()) return;
+    // if the table is transaction table we should generate a transactional
+    // insert event type.
+    boolean isTransactional = AcidUtils.isTransactionalTable(table.getMetaStoreTable()
+        .getParameters());
+    Preconditions.checkState(!isTransactional || txnId > 0, String
+        .format("Invalid transaction id %s for generating insert events on table %s",
+            txnId, table.getFullName()));
+    Preconditions.checkState(!isTransactional || writeId > 0,
+        String.format("Invalid write id %s for generating insert events on table %s",
+            writeId, table.getFullName()));
+    // if the table is transactional we fire the event for new partitions as well.
+    if (isTransactional) {
+      for (HdfsPartition newPart : ((HdfsTable) table)
+          .getPartitionsForNames(newPartsCreated)) {
+        insertEventReqDatas.add(makeInsertEventData((HdfsTable) table,
+            newPart.getPartitionValuesAsStrings(true), newPart.getFileNames(),
+            isInsertOverwrite));
+      }
+    }
+
+    if (insertEventReqDatas.isEmpty()) return;
 
     MetaStoreClient metaStoreClient = catalog_.getMetaStoreClient();
+    TableInsertEventInfo insertEventInfo = new TableInsertEventInfo(
+        insertEventReqDatas, isTransactional, txnId, writeId);
     List<Long> eventIds = MetastoreShim.fireInsertEvents(metaStoreClient,
-        insertEventInfos, table.getDb().getName(), table.getName());
+        insertEventInfo, table.getDb().getName(), table.getName());
     if (!eventIds.isEmpty()) {
       if (partitions.size() == 0) { // insert into table
         catalog_.addVersionsForInflightEvents(true, table, eventIds.get(0));
@@ -4639,14 +4707,58 @@
     }
   }
 
+  private boolean shouldGenerateInsertEvents() {
+    return catalog_.isEventProcessingActive() && BackendConfig.INSTANCE
+        .isInsertEventsEnabled();
+  }
+
+  private InsertEventRequestData makeInsertEventData(HdfsTable tbl, List<String> partVals,
+      Set<String> newFiles, boolean isInsertOverwrite) throws CatalogException {
+    Preconditions.checkNotNull(newFiles);
+    Preconditions.checkNotNull(partVals);
+    InsertEventRequestData insertEventRequestData = new InsertEventRequestData(
+        Lists.newArrayListWithCapacity(
+            newFiles.size()));
+    boolean isTransactional = AcidUtils
+        .isTransactionalTable(tbl.getMetaStoreTable().getParameters());
+    // in case of unpartitioned table, partVals will be empty
+    if (!partVals.isEmpty()) insertEventRequestData.setPartitionVal(partVals);
+    FileSystem fs = tbl.getFileSystem();
+    for (String file : newFiles) {
+      try {
+        Path filePath = new Path(file);
+        FileChecksum checkSum = fs.getFileChecksum(filePath);
+        String checksumStr = checkSum == null ? ""
+            : StringUtils.byteToHexString(checkSum.getBytes(), 0, checkSum.getLength());
+        insertEventRequestData.addToFilesAdded(file);
+        insertEventRequestData.addToFilesAddedChecksum(checksumStr);
+        if (isTransactional) {
+          String acidDirPath = AcidUtils.getFirstLevelAcidDirPath(filePath, fs);
+          if (acidDirPath != null) {
+            insertEventRequestData.addToSubDirectoryList(acidDirPath);
+          }
+        }
+        insertEventRequestData.setReplace(isInsertOverwrite);
+      } catch (IOException e) {
+        throw new CatalogException("Could not get the file checksum for file " + file, e);
+      }
+    }
+    return insertEventRequestData;
+  }
+
   /**
    * Util method to return a map of partition names to list of files for that partition.
    */
   private static Map<String, Set<String>> getPartitionNameFilesMap(Collection<?
       extends FeFsPartition> partitions) {
-        return partitions.stream().collect(
-            Collectors.toMap(p -> p.getPartitionName() + "/",
-                p -> ((HdfsPartition) p).getFileNames()));
+    Map<String, Set<String>> partitionFilePaths = new HashMap<>();
+    for (FeFsPartition partition : partitions) {
+      String key = partition.getPartitionName() + "/";
+      Set<String> filenames = ((HdfsPartition) partition).getFileNames();
+      partitionFilePaths.putIfAbsent(key, new HashSet<>(filenames.size()));
+      partitionFilePaths.get(key).addAll(filenames);
+    }
+    return partitionFilePaths;
   }
 
   /**
diff --git a/fe/src/main/java/org/apache/impala/util/AcidUtils.java b/fe/src/main/java/org/apache/impala/util/AcidUtils.java
index 7671318..b9fef23 100644
--- a/fe/src/main/java/org/apache/impala/util/AcidUtils.java
+++ b/fe/src/main/java/org/apache/impala/util/AcidUtils.java
@@ -22,7 +22,9 @@
 import com.google.common.collect.Range;
 import com.google.errorprone.annotations.Immutable;
 
+import java.io.IOException;
 import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hive.common.ValidTxnList;
 import org.apache.hadoop.hive.common.ValidWriteIdList;
@@ -173,6 +175,45 @@
   }
 
   /**
+   * This method is copied from Hive's org.apache.hadoop.hive.ql.io.AcidUtils.java
+   * (commit hash 17ac1d9f230b8d663c09c22016753012a9b91edf). It is used to generate
+   * the ACID directory names to be added to the insert events on the transactional
+   * tables.
+   */
+  //Get the first level acid directory (if any) from a given path
+  public static String getFirstLevelAcidDirPath(Path dataPath, FileSystem fileSystem)
+      throws IOException {
+    if (dataPath == null) {
+      return null;
+    }
+    String firstLevelAcidDir = getAcidSubDir(dataPath);
+    if (firstLevelAcidDir != null) {
+      return firstLevelAcidDir;
+    }
+
+    String acidDirPath = getFirstLevelAcidDirPath(dataPath.getParent(), fileSystem);
+    if (acidDirPath == null) {
+      return null;
+    }
+
+    // We need the path for directory so no need to append file name
+    if (fileSystem.isDirectory(dataPath)) {
+      return acidDirPath + Path.SEPARATOR + dataPath.getName();
+    }
+    return acidDirPath;
+  }
+
+  private static String getAcidSubDir(Path dataPath) {
+    String dataDir = dataPath.getName();
+    if (dataDir.startsWith("base_")
+        || dataDir.startsWith("delta_")
+        || dataDir.startsWith("delete_delta_")) {
+      return dataDir;
+    }
+    return null;
+  }
+
+  /**
    * Predicate that checks if the file or directory is relevant for a given WriteId list.
    * The class does not implement a Predicate interface since we want to support a strict
    * mode which throw exception in certain cases.
diff --git a/fe/src/main/java/org/apache/impala/util/MetaStoreUtil.java b/fe/src/main/java/org/apache/impala/util/MetaStoreUtil.java
index 81b62f6..db5655b 100644
--- a/fe/src/main/java/org/apache/impala/util/MetaStoreUtil.java
+++ b/fe/src/main/java/org/apache/impala/util/MetaStoreUtil.java
@@ -25,11 +25,14 @@
 import java.util.Map;
 
 import java.util.Collection;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.metastore.IMetaStoreClient;
 import org.apache.hadoop.hive.metastore.Warehouse;
 import org.apache.hadoop.hive.metastore.api.ConfigValSecurityException;
 import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.InsertEventRequestData;
 import org.apache.hadoop.hive.metastore.api.MetaException;
 import org.apache.hadoop.hive.metastore.api.Partition;
 import org.apache.hadoop.hive.metastore.api.SQLPrimaryKey;
@@ -321,33 +324,39 @@
     return msTbl.getSd().getNumBuckets() > 0;
   }
 
-  /**
-   * A helper class that encapsulates all the information needed to fire and insert event
-   * with HMS.
-   */
-  public static class InsertEventInfo {
-    // List of partition values corresponding to the partition keys in
-    // a partitioned table. This is null for non-partitioned table.
-    private List<String> partVals;
+  public static class TableInsertEventInfo {
+    // list of partition level insert event info
+    private final List<InsertEventRequestData> insertEventRequestData_;
+    // transaction id in case this represents a transactional table.
+    private final long txnId_;
+    // writeId in case this is for a transactional table.
+    private final long writeId_;
+    // true in case this is for transactional table.
+    private final boolean isTransactional_;
 
-    // Set of all the 'new' files added by this insert. This is empty in
-    // case of insert overwrite.
-    private Collection<String> newFiles;
-
-    // If true, sets the 'replace' flag to true indicating that the
-    // operation was an insert overwrite in the notification log. Will set the same to
-    // false otherwise.
-    private boolean isOverwrite;
-
-    public InsertEventInfo(
-        List<String> partVals, Collection<String> newFiles, boolean isOverwrite) {
-      this.partVals = partVals;
-      this.newFiles = newFiles;
-      this.isOverwrite = isOverwrite;
+    public TableInsertEventInfo(
+        List<InsertEventRequestData> insertEventInfos_, boolean isTransactional,
+        long txnId, long writeId) {
+      this.insertEventRequestData_ = insertEventInfos_;
+      this.txnId_ = txnId;
+      this.writeId_ = writeId;
+      this.isTransactional_ = isTransactional;
     }
 
-    public List<String> getPartVals() { return this.partVals; }
-    public Collection<String> getNewFiles() { return this.newFiles; }
-    public boolean isOverwrite() { return this.isOverwrite; }
+    public boolean isTransactional() {
+      return isTransactional_;
+    }
+
+    public List<InsertEventRequestData> getInsertEventReqData() {
+      return insertEventRequestData_;
+    }
+
+    public long getTxnId() {
+      return txnId_;
+    }
+
+    public long getWriteId() {
+      return writeId_;
+    }
   }
 }
diff --git a/fe/src/test/java/org/apache/impala/catalog/events/MetastoreEventsProcessorTest.java b/fe/src/test/java/org/apache/impala/catalog/events/MetastoreEventsProcessorTest.java
index 338d6ff..27e1ab1 100644
--- a/fe/src/test/java/org/apache/impala/catalog/events/MetastoreEventsProcessorTest.java
+++ b/fe/src/test/java/org/apache/impala/catalog/events/MetastoreEventsProcessorTest.java
@@ -33,6 +33,7 @@
 import com.google.common.collect.Iterables;
 import java.io.File;
 import java.io.IOException;
+import java.io.InputStream;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -46,12 +47,16 @@
 import java.util.concurrent.atomic.AtomicLong;
 
 import org.apache.commons.lang.RandomStringUtils;
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RemoteIterator;
 import org.apache.hadoop.hive.metastore.api.CurrentNotificationEventId;
 import org.apache.hadoop.hive.metastore.api.Database;
 import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.InsertEventRequestData;
 import org.apache.hadoop.hive.metastore.api.MetaException;
 import org.apache.hadoop.hive.metastore.api.NotificationEvent;
 import org.apache.hadoop.hive.metastore.api.Partition;
@@ -59,6 +64,9 @@
 import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
 import org.apache.hadoop.hive.metastore.api.PrincipalType;
 import org.apache.hadoop.hive.metastore.messaging.AlterTableMessage;
+import org.apache.hadoop.hive.metastore.messaging.CommitTxnMessage;
+import org.apache.hadoop.hive.metastore.messaging.InsertMessage;
+import org.apache.hadoop.io.IOUtils;
 import org.apache.impala.analysis.FunctionName;
 import org.apache.impala.analysis.HdfsUri;
 import org.apache.impala.authorization.NoopAuthorizationFactory;
@@ -80,6 +88,7 @@
 import org.apache.impala.catalog.Type;
 import org.apache.impala.catalog.events.ConfigValidator.ValidationResult;
 import org.apache.impala.catalog.events.MetastoreEvents.AlterTableEvent;
+import org.apache.impala.catalog.events.MetastoreEvents.InsertEvent;
 import org.apache.impala.catalog.events.MetastoreEvents.MetastoreEvent;
 import org.apache.impala.catalog.events.MetastoreEvents.MetastoreEventPropertyKey;
 import org.apache.impala.catalog.events.MetastoreEvents.MetastoreEventType;
@@ -114,6 +123,7 @@
 import org.apache.impala.thrift.TColumnType;
 import org.apache.impala.thrift.TCreateDbParams;
 import org.apache.impala.thrift.TCreateFunctionParams;
+import org.apache.impala.thrift.TCreateTableLikeParams;
 import org.apache.impala.thrift.TCreateTableParams;
 import org.apache.impala.thrift.TDdlExecRequest;
 import org.apache.impala.thrift.TDdlType;
@@ -138,7 +148,6 @@
 import org.apache.impala.thrift.TTypeNodeType;
 import org.apache.impala.thrift.TUniqueId;
 import org.apache.impala.thrift.TUpdateCatalogRequest;
-import org.apache.impala.util.MetaStoreUtil.InsertEventInfo;
 import org.apache.thrift.TException;
 import org.junit.After;
 import org.junit.AfterClass;
@@ -720,50 +729,148 @@
    * should be treated as self-event.
    */
   @Test
-  public void testInsertFromImpala() throws ImpalaException {
+  public void testInsertFromImpala() throws Exception {
     Assume.assumeTrue("Skipping this test because it only works with Hive-3 or greater",
         TestUtils.getHiveMajorVersion() >= 3);
     // Test insert into multiple partitions
     createDatabaseFromImpala(TEST_DB_NAME, null);
     String tableToInsertPart = "tbl_with_mul_part";
-    createTableFromImpala(TEST_DB_NAME, tableToInsertPart, true);
     String tableToInsertMulPart = "tbl_to_insert_mul_part";
-    createTableFromImpala(TEST_DB_NAME, tableToInsertMulPart, true);
-    // add first partition
-    TPartitionDef partitionDef = new TPartitionDef();
-    partitionDef.addToPartition_spec(new TPartitionKeyValue("p1", "1"));
-    partitionDef.addToPartition_spec(new TPartitionKeyValue("p2", "100"));
-    alterTableAddPartition(TEST_DB_NAME, tableToInsertPart, partitionDef);
-    alterTableAddPartition(TEST_DB_NAME, tableToInsertMulPart, partitionDef);
-    // add second partition
-    partitionDef = new TPartitionDef();
-    partitionDef.addToPartition_spec(new TPartitionKeyValue("p1", "1"));
-    partitionDef.addToPartition_spec(new TPartitionKeyValue("p2", "200"));
-    alterTableAddPartition(TEST_DB_NAME, tableToInsertPart, partitionDef);
-    alterTableAddPartition(TEST_DB_NAME, tableToInsertMulPart, partitionDef);
-    eventsProcessor_.processEvents();
+    createInsertTestTbls(tableToInsertPart, tableToInsertMulPart);
     // count self event from here, numberOfSelfEventsBefore=4 as we have 4 ADD PARTITION
     // events
     long numberOfSelfEventsBefore =
         eventsProcessor_.getMetrics()
             .getCounter(MetastoreEventsProcessor.NUMBER_OF_SELF_EVENTS)
             .getCount();
+    runInsertTest(tableToInsertPart, tableToInsertMulPart, numberOfSelfEventsBefore,
+        false);
+  }
+
+  @Test
+  public void testInsertOverwriteFromImpala() throws Exception {
+    Assume.assumeTrue("Skipping this test because it only works with Hive-3 or greater",
+        TestUtils.getHiveMajorVersion() >= 3);
+    // Test insert into multiple partitions
+    createDatabaseFromImpala(TEST_DB_NAME, null);
+    String tableToInsertPart = "tbl_with_mul_part";
+    String tableToInsertMulPart = "tbl_to_insert_mul_part";
+    createInsertTestTbls(tableToInsertPart, tableToInsertMulPart);
+    // count self event from here, numberOfSelfEventsBefore=4 as we have 4 ADD PARTITION
+    // events
+    long numberOfSelfEventsBefore =
+        eventsProcessor_.getMetrics()
+            .getCounter(MetastoreEventsProcessor.NUMBER_OF_SELF_EVENTS)
+            .getCount();
+    runInsertTest(tableToInsertPart, tableToInsertMulPart, numberOfSelfEventsBefore,
+        true);
+  }
+
+  private void createInsertTestTbls(String tableToInsertPart, String tableToInsertMulPart)
+      throws Exception {
+    createTableLike("functional", "alltypes", TEST_DB_NAME, tableToInsertPart);
+    createTableLike("functional", "alltypes", TEST_DB_NAME, tableToInsertMulPart);
+    // add first partition
+    TPartitionDef partitionDef = new TPartitionDef();
+    partitionDef.addToPartition_spec(new TPartitionKeyValue("year", "2009"));
+    partitionDef.addToPartition_spec(new TPartitionKeyValue("month", "1"));
+    alterTableAddPartition(TEST_DB_NAME, tableToInsertPart, partitionDef);
+    alterTableAddPartition(TEST_DB_NAME, tableToInsertMulPart, partitionDef);
+    // add second partition
+    partitionDef = new TPartitionDef();
+    partitionDef.addToPartition_spec(new TPartitionKeyValue("year", "2009"));
+    partitionDef.addToPartition_spec(new TPartitionKeyValue("month", "2"));
+    alterTableAddPartition(TEST_DB_NAME, tableToInsertPart, partitionDef);
+    alterTableAddPartition(TEST_DB_NAME, tableToInsertMulPart, partitionDef);
+
+    HdfsTable allTypes = (HdfsTable) catalog_
+        .getOrLoadTable("functional", "alltypes", "test", null);
+    HdfsTable insertTbl = (HdfsTable) catalog_
+        .getOrLoadTable(TEST_DB_NAME, tableToInsertPart, "test", null);
+    HdfsTable multiInsertTbl = (HdfsTable) catalog_
+        .getOrLoadTable(TEST_DB_NAME, tableToInsertMulPart, "test", null);
+    // copy files from the source tables so that we have some data
+    copyFiles(allTypes.getFileSystem(),
+        new Path(allTypes.getHdfsBaseDir() + "/year=2009/month=1"),
+        insertTbl.getFileSystem(),
+        new Path(insertTbl.getHdfsBaseDir() + "/year=2009/month=1"), true, null);
+    copyFiles(allTypes.getFileSystem(),
+        new Path(allTypes.getHdfsBaseDir() + "/year=2009/month=2"),
+        insertTbl.getFileSystem(),
+        new Path(insertTbl.getHdfsBaseDir() + "/year=2009/month=2"), true, null);
+    copyFiles(allTypes.getFileSystem(),
+        new Path(allTypes.getHdfsBaseDir() + "/year=2009/month=1"),
+        multiInsertTbl.getFileSystem(),
+        new Path(multiInsertTbl.getHdfsBaseDir() + "/year=2009/month=1"), true, null);
+    copyFiles(allTypes.getFileSystem(),
+        new Path(allTypes.getHdfsBaseDir() + "/year=2009/month=2"),
+        multiInsertTbl.getFileSystem(),
+        new Path(multiInsertTbl.getHdfsBaseDir() + "/year=2009/month=2"), true, null);
+    // load the created tables
+    catalog_.reloadTable(multiInsertTbl, "test");
+    catalog_.reloadTable(insertTbl, "test");
+    eventsProcessor_.processEvents();
+  }
+
+  private void runInsertTest(String tableToInsertPart, String tableToInsertMulPart,
+      long numberOfSelfEventsBefore, boolean overwrite) throws Exception {
     // insert into partition
-    insertFromImpala(tableToInsertPart, true, "1", "100");
-    insertFromImpala(tableToInsertPart, true, "1", "200");
+    HdfsTable allTypes = (HdfsTable) catalog_
+        .getOrLoadTable("functional", "alltypes", "test", null);
+    HdfsTable insertTbl = (HdfsTable) catalog_
+        .getOrLoadTable(TEST_DB_NAME, tableToInsertPart, "test", null);
+    HdfsTable multiInsertTbl = (HdfsTable) catalog_
+        .getOrLoadTable(TEST_DB_NAME, tableToInsertMulPart, "test", null);
+    // we copy files from the src tbl and then issue a insert catalogOp to simulate a
+    // insert operation
+    List<String> tbl1Part1Files = copyFiles(allTypes.getFileSystem(),
+        new Path(allTypes.getHdfsBaseDir() + "/year=2009/month=1"),
+        insertTbl.getFileSystem(),
+        new Path(insertTbl.getHdfsBaseDir() + "/year=2009/month=1"), overwrite, "copy_");
+    List<String> tbl1Part2Files = copyFiles(allTypes.getFileSystem(),
+        new Path(allTypes.getHdfsBaseDir() + "/year=2009/month=2"),
+        insertTbl.getFileSystem(),
+        new Path(insertTbl.getHdfsBaseDir() + "/year=2009/month=2"), overwrite, "copy_");
+    List<String> tbl2Part1Files = copyFiles(allTypes.getFileSystem(),
+        new Path(allTypes.getHdfsBaseDir() + "/year=2009/month=1"),
+        multiInsertTbl.getFileSystem(),
+        new Path(multiInsertTbl.getHdfsBaseDir() + "/year=2009/month=1"), overwrite,
+        "copy_");
+    List<String> tbl2Part2Files = copyFiles(allTypes.getFileSystem(),
+        new Path(allTypes.getHdfsBaseDir() + "/year=2009/month=2"),
+        multiInsertTbl.getFileSystem(),
+        new Path(multiInsertTbl.getHdfsBaseDir() + "/year=2009/month=2"), overwrite,
+        "copy_");
+    insertFromImpala(tableToInsertPart, true, "year=2009", "month=1", overwrite);
+    insertFromImpala(tableToInsertPart, true, "year=2009", "month=2", overwrite);
     // insert into multiple partition
     Set<String> created_partitions = new HashSet<String>();
-    String partition1 = "p1=1/p2=100/";
-    String partition2 = "p1=1/p2=200/";
+    String partition1 = "year=2009/month=1/";
+    String partition2 = "year=2009/month=2/";
     created_partitions.add(partition1);
     created_partitions.add(partition2);
-    insertMulPartFromImpala(tableToInsertMulPart, tableToInsertPart, created_partitions);
+    insertMulPartFromImpala(tableToInsertMulPart, tableToInsertPart, created_partitions,
+        overwrite);
+    // we expect 3 INSERT events (2 for the insertTbl and 1 for multiInsertTbl)
+    List<NotificationEvent> events = eventsProcessor_.getNextMetastoreEvents();
+    assertEquals(4, events.size());
+    assertEquals(tbl1Part1Files, getFilesFromEvent(events.get(0)));
+    assertEquals(tbl1Part2Files, getFilesFromEvent(events.get(1)));
+    assertEquals(tbl2Part1Files, getFilesFromEvent(events.get(2)));
+    assertEquals(tbl2Part2Files, getFilesFromEvent(events.get(3)));
     eventsProcessor_.processEvents();
 
     // Test insert into table
-    String tableToInsert = "tbl_to_insert";
-    createTableFromImpala(TEST_DB_NAME, tableToInsert, false);
-    insertFromImpala(tableToInsert, false, "", "");
+    String unpartitionedTbl = "tbl_to_insert";
+    createTableLike("functional", "tinytable", TEST_DB_NAME, unpartitionedTbl);
+    HdfsTable tinyTable = (HdfsTable) catalog_
+        .getOrLoadTable("functional", "tinytable", "test", null);
+    HdfsTable unpartTable =
+        (HdfsTable) catalog_.getOrLoadTable(TEST_DB_NAME, unpartitionedTbl, "test", null);
+    copyFiles(tinyTable.getFileSystem(), new Path(tinyTable.getHdfsBaseDir()),
+        unpartTable.getFileSystem(), new Path(unpartTable.getHdfsBaseDir()), overwrite,
+        "copy_");
+    insertFromImpala(unpartitionedTbl, false, "", "", overwrite);
     eventsProcessor_.processEvents();
 
     long selfEventsCountAfter =
@@ -776,6 +883,50 @@
         numberOfSelfEventsBefore + 5, selfEventsCountAfter);
   }
 
+  private List<String> getFilesFromEvent(NotificationEvent event) {
+    assertEquals("INSERT", event.getEventType());
+    List<String> files= new ArrayList<>();
+    for (String f : MetastoreEventsProcessor.getMessageDeserializer()
+        .getInsertMessage(event.getMessage()).getFiles()) {
+      // Metastore InsertMessage appends "###" for some reason. Ignoring that bit in the
+      // comparison below.
+      files.add(f.replaceAll("###", ""));
+    }
+    return files;
+  }
+
+  private static final Configuration CONF = new Configuration();
+
+  private static List<String> copyFiles(FileSystem srcFs, Path src, FileSystem destFs,
+      Path dest, boolean overwrite, String prefix) throws Exception {
+    FSDataOutputStream out = null;
+    try {
+      if (srcFs.isDirectory(src)) {
+        if (!destFs.exists(dest)) {
+          destFs.mkdirs(dest);
+        } else if(overwrite) {
+          destFs.delete(dest, true);
+          destFs.mkdirs(dest);
+        }
+      }
+      List<String> filesCopied = new ArrayList<>();
+      RemoteIterator<? extends FileStatus> it = FileSystemUtil
+          .listStatus(srcFs, src, true);
+      while (it.hasNext()) {
+        FileStatus status = it.next();
+        if (status.isDirectory()) continue;
+        InputStream in = srcFs.open(status.getPath());
+        String copyFileName = (prefix == null ? "" : prefix) + status.getPath().getName();
+        out = destFs.create(new Path(dest, copyFileName), false);
+        IOUtils.copyBytes(in, out, CONF, true);
+        filesCopied.add(new Path(dest, copyFileName).toString());
+      }
+      return filesCopied;
+    } catch (IOException ex) {
+      IOUtils.closeStream(out);
+      throw ex;
+    }
+  }
   /**
    * Test generates a sequence of create_table, insert and drop_table in the event stream
    * to make sure when the insert event is processed on a removed table, it doesn't cause
@@ -900,10 +1051,14 @@
     List <String> newFiles = addFilesToDirectory(parentPath, "testFile.",
         totalNumberOfFilesToAdd, isOverwrite);
     try (MetaStoreClient metaStoreClient = catalog_.getMetaStoreClient()) {
-      List<InsertEventInfo> insertEventInfos = new ArrayList<>();
-      insertEventInfos.add(new InsertEventInfo(null, newFiles, isOverwrite));
+      List<InsertEventRequestData> partitionInsertEventInfos = new ArrayList<>();
+      InsertEventRequestData insertEventRequestData = new InsertEventRequestData();
+      insertEventRequestData.setFilesAdded(newFiles);
+      insertEventRequestData.setReplace(isOverwrite);
+      partitionInsertEventInfos
+          .add(insertEventRequestData);
       MetastoreShim.fireInsertEventHelper(metaStoreClient.getHiveClient(),
-          insertEventInfos, msTbl.getDbName(), msTbl.getTableName());
+          partitionInsertEventInfos, msTbl.getDbName(), msTbl.getTableName());
     }
   }
 
@@ -2436,6 +2591,19 @@
     catalogOpExecutor_.execDdlRequest(req);
   }
 
+  private void createTableLike(String srcDb, String srcTbl, String destDb, String destTbl)
+      throws Exception {
+    HdfsTable table = (HdfsTable) catalog_.getOrLoadTable(srcDb, srcTbl, "Test", null);
+    TCreateTableLikeParams createTableLikeParams = new TCreateTableLikeParams();
+    createTableLikeParams.setSrc_table_name(new TTableName(srcDb, srcTbl));
+    createTableLikeParams.setTable_name(new TTableName(destDb, destTbl));
+    createTableLikeParams.setIs_external(false);
+    TDdlExecRequest req = new TDdlExecRequest();
+    req.setDdl_type(TDdlType.CREATE_TABLE_LIKE);
+    req.create_table_like_params = createTableLikeParams;
+    catalogOpExecutor_.execDdlRequest(req);
+  }
+
   private void createTableFromImpala(String dbName, String tblName, boolean isPartitioned)
       throws ImpalaException {
     createTableFromImpala(dbName, tblName, null, isPartitioned);
@@ -2781,11 +2949,11 @@
    * Insert multiple partitions into table from Impala
    */
   private void insertMulPartFromImpala(String tblName1, String tblName2,
-      Set<String> created_partitions) throws ImpalaException {
+      Set<String> created_partitions, boolean overwrite) throws ImpalaException {
     String insert_mul_part = String.format(
         "insert into table %s partition(p1, p2) select * from %s", tblName1, tblName2);
     TUpdateCatalogRequest testInsertRequest = createTestTUpdateCatalogRequest(
-        TEST_DB_NAME, tblName1, insert_mul_part, created_partitions);
+        TEST_DB_NAME, tblName1, insert_mul_part, created_partitions, overwrite);
     catalogOpExecutor_.updateCatalog(testInsertRequest);
   }
 
@@ -2796,16 +2964,16 @@
    * @return
    */
   private void insertFromImpala(String tblName, boolean isPartitioned, String p1val,
-      String p2val) throws ImpalaException {
-    String partition = String.format("partition (p1=%s, p2='%s')", p1val, p2val);
+      String p2val, boolean isOverwrite) throws ImpalaException {
+    String partition = String.format("partition (%s, %s)", p1val, p2val);
     String test_insert_tbl = String.format("insert into table %s %s values ('a','aa') ",
         tblName, isPartitioned ? partition : "");
-    Set<String> created_partitions = new HashSet<String>();
+    Set<String> created_partitions = new HashSet<>();
     String created_part_str =
-        isPartitioned ? String.format("p1=%s/p2=%s/", p1val, p2val) : "";
+        isPartitioned ? String.format("%s/%s/", p1val, p2val) : "";
     created_partitions.add(created_part_str);
     TUpdateCatalogRequest testInsertRequest = createTestTUpdateCatalogRequest(
-        TEST_DB_NAME, tblName, test_insert_tbl, created_partitions);
+        TEST_DB_NAME, tblName, test_insert_tbl, created_partitions, isOverwrite);
     catalogOpExecutor_.updateCatalog(testInsertRequest);
   }
 
@@ -2818,13 +2986,15 @@
    * @return
    */
   private TUpdateCatalogRequest createTestTUpdateCatalogRequest(String dBName,
-      String tableName, String redacted_sql_stmt, Set<String> created_partitions) {
+      String tableName, String redacted_sql_stmt, Set<String> created_partitions,
+      boolean isOverwrite) {
     TUpdateCatalogRequest tUpdateCatalogRequest = new TUpdateCatalogRequest();
     tUpdateCatalogRequest.setDb_name(dBName);
     tUpdateCatalogRequest.setTarget_table(tableName);
     tUpdateCatalogRequest.setCreated_partitions((created_partitions));
     tUpdateCatalogRequest.setHeader(new TCatalogServiceRequestHeader());
     tUpdateCatalogRequest.getHeader().setRedacted_sql_stmt(redacted_sql_stmt);
+    if (isOverwrite) tUpdateCatalogRequest.setIs_overwrite(true);
     return tUpdateCatalogRequest;
   }
 
diff --git a/fe/src/test/java/org/apache/impala/catalog/events/SynchronousHMSEventProcessorForTests.java b/fe/src/test/java/org/apache/impala/catalog/events/SynchronousHMSEventProcessorForTests.java
index f71dfc6..b8c4474 100644
--- a/fe/src/test/java/org/apache/impala/catalog/events/SynchronousHMSEventProcessorForTests.java
+++ b/fe/src/test/java/org/apache/impala/catalog/events/SynchronousHMSEventProcessorForTests.java
@@ -17,8 +17,15 @@
 
 package org.apache.impala.catalog.events;
 
+import java.util.Collections;
+import java.util.List;
+import org.apache.hadoop.hive.metastore.api.CurrentNotificationEventId;
+import org.apache.hadoop.hive.metastore.api.NotificationEvent;
+import org.apache.hadoop.hive.metastore.api.NotificationEventResponse;
 import org.apache.impala.catalog.CatalogException;
 import org.apache.impala.catalog.CatalogServiceCatalog;
+import org.apache.impala.catalog.MetaStoreClientPool.MetaStoreClient;
+import org.apache.thrift.TException;
 
 /**
  * A test MetastoreEventProcessor which executes in the same thread. Useful for testing
@@ -35,4 +42,29 @@
   public void startScheduler() {
     // nothing to do here; there is no background thread for this processor
   }
+
+  public List<NotificationEvent> getNextNotificationEvents(long eventId)
+      throws MetastoreNotificationFetchException {
+    try (MetaStoreClient msClient = catalog_.getMetaStoreClient()) {
+      // fetch the current notification event id. We assume that the polling interval
+      // is small enough that most of these polling operations result in zero new
+      // events. In such a case, fetching current notification event id is much faster
+      // (and cheaper on HMS side) instead of polling for events directly
+      CurrentNotificationEventId currentNotificationEventId =
+          msClient.getHiveClient().getCurrentNotificationEventId();
+      long currentEventId = currentNotificationEventId.getEventId();
+
+      // no new events since we last polled
+      if (currentEventId <= eventId) {
+        return Collections.emptyList();
+      }
+
+      NotificationEventResponse response = msClient.getHiveClient()
+          .getNextNotification(eventId, 1000, null);
+      return response.getEvents();
+    } catch (TException e) {
+      throw new MetastoreNotificationFetchException(
+          "Unable to fetch notifications from metastore", e);
+    }
+  }
 }
diff --git a/fe/src/test/resources/hive-site.xml.py b/fe/src/test/resources/hive-site.xml.py
index 3fc6a80..c14f982 100644
--- a/fe/src/test/resources/hive-site.xml.py
+++ b/fe/src/test/resources/hive-site.xml.py
@@ -56,6 +56,12 @@
   # TODO(todd): should we be enabling stats autogather?
   'hive.stats.autogather': 'false',
   'hive.support.concurrency': 'true',
+  # There are tests which issue Hive's replication command. The repl
+  # dump command will wait in case there are any open transactions.
+  # The default value of this configuration is 1h which will block
+  # the test for long time. Overriding this configuration helps with
+  # the runtime of the test.
+  'hive.repl.bootstrap.dump.open.txn.timeout': '120s'
 })
 
 if variant == 'changed_external_dir':
diff --git a/tests/custom_cluster/test_event_processing.py b/tests/custom_cluster/test_event_processing.py
index d88e388..47510ae 100644
--- a/tests/custom_cluster/test_event_processing.py
+++ b/tests/custom_cluster/test_event_processing.py
@@ -187,6 +187,179 @@
     self.__run_self_events_test(unique_database, True)
     self.__run_self_events_test(unique_database, False)
 
+  @CustomClusterTestSuite.with_args(catalogd_args="--hms_event_polling_interval_s=1")
+  def test_event_based_replication(self):
+    self.__run_event_based_replication_tests()
+
+  def __run_event_based_replication_tests(self, transactional=True):
+    """Hive Replication relies on the insert events generated on the tables.
+    This test issues some basic replication commands from Hive and makes sure
+    that the replicated table has correct data."""
+    TBLPROPERTIES = self.__get_transactional_tblproperties(transactional)
+    source_db = self.__get_random_name("repl_source_")
+    target_db = self.__get_random_name("repl_target_")
+    unpartitioned_tbl = "unpart_tbl"
+    partitioned_tbl = "part_tbl"
+    try:
+      self.run_stmt_in_hive("create database {0}".format(source_db))
+      self.run_stmt_in_hive(
+        "alter database {0} set dbproperties ('repl.source.for'='xyz')".format(source_db))
+      # explicit create table command since create table like doesn't allow tblproperties
+      self.client.execute("create table {0}.{1} (a string, b string) stored as parquet"
+        " {2}".format(source_db, unpartitioned_tbl, TBLPROPERTIES))
+      EventProcessorUtils.wait_for_event_processing(self)
+      self.client.execute(
+        "create table {0}.{1} (id int, bool_col boolean, tinyint_col tinyint, "
+        "smallint_col smallint, int_col int, bigint_col bigint, float_col float, "
+        "double_col double, date_string string, string_col string, "
+        "timestamp_col timestamp) partitioned by (year int, month int) stored as parquet"
+        " {2}".format(source_db, partitioned_tbl, TBLPROPERTIES))
+
+      # case I: insert
+      # load the table with some data from impala, this also creates new partitions.
+      self.client.execute("insert into {0}.{1}"
+        " select * from functional.tinytable".format(source_db,
+          unpartitioned_tbl))
+      self.client.execute("insert into {0}.{1} partition(year,month)"
+        " select * from functional_parquet.alltypessmall".format(
+          source_db, partitioned_tbl))
+      rows_in_unpart_tbl = int(self.execute_scalar(
+        "select count(*) from {0}.{1}".format(source_db, unpartitioned_tbl)).split('\t')[
+        0])
+      rows_in_part_tbl = int(self.execute_scalar(
+        "select count(*) from {0}.{1}".format(source_db, partitioned_tbl)).split('\t')[0])
+      assert rows_in_unpart_tbl > 0
+      assert rows_in_part_tbl > 0
+      # bootstrap the replication
+      self.run_stmt_in_hive("repl dump {0}".format(source_db))
+      # create a target database where tables will be replicated
+      self.client.execute("create database {0}".format(target_db))
+      # replicate the table from source to target
+      self.run_stmt_in_hive("repl load {0} into {1}".format(source_db, target_db))
+      EventProcessorUtils.wait_for_event_processing(self)
+      assert unpartitioned_tbl in self.client.execute(
+        "show tables in {0}".format(target_db)).get_data()
+      assert partitioned_tbl in self.client.execute(
+        "show tables in {0}".format(target_db)).get_data()
+      # confirm the number of rows in target match with the source table.
+      rows_in_unpart_tbl_target = int(self.execute_scalar(
+        "select count(*) from {0}.{1}".format(target_db, unpartitioned_tbl))
+          .split('\t')[0])
+      rows_in_part_tbl_target = int(self.execute_scalar(
+        "select count(*) from {0}.{1}".format(target_db, partitioned_tbl))
+          .split('\t')[0])
+      assert rows_in_unpart_tbl == rows_in_unpart_tbl_target
+      assert rows_in_part_tbl == rows_in_part_tbl_target
+
+      # case II: insert into existing partitions.
+      self.client.execute("insert into {0}.{1}"
+        " select * from functional.tinytable".format(
+          source_db, unpartitioned_tbl))
+      self.client.execute("insert into {0}.{1} partition(year,month)"
+        " select * from functional_parquet.alltypessmall".format(
+          source_db, partitioned_tbl))
+      self.run_stmt_in_hive("repl dump {0}".format(source_db))
+      # replicate the table from source to target
+      self.run_stmt_in_hive("repl load {0} into {1}".format(source_db, target_db))
+      # we wait until the events catch up in case repl command above did some HMS
+      # operations.
+      EventProcessorUtils.wait_for_event_processing(self)
+      # confirm the number of rows in target match with the source table.
+      rows_in_unpart_tbl_target = int(self.execute_scalar(
+        "select count(*) from {0}.{1}".format(target_db, unpartitioned_tbl))
+          .split('\t')[0])
+      rows_in_part_tbl_target = int(self.execute_scalar(
+        "select count(*) from {0}.{1}".format(target_db, partitioned_tbl)).split('\t')[0])
+      assert 2 * rows_in_unpart_tbl == rows_in_unpart_tbl_target
+      assert 2 * rows_in_part_tbl == rows_in_part_tbl_target
+
+      # Case III: insert overwrite
+      # impala does a insert overwrite of the tables.
+      self.client.execute("insert overwrite table {0}.{1}"
+        " select * from functional.tinytable".format(
+          source_db, unpartitioned_tbl))
+      self.client.execute("insert overwrite table {0}.{1} partition(year,month)"
+        " select * from functional_parquet.alltypessmall".format(
+          source_db, partitioned_tbl))
+      self.run_stmt_in_hive("repl dump {0}".format(source_db))
+      # replicate the table from source to target
+      self.run_stmt_in_hive("repl load {0} into {1}".format(source_db, target_db))
+      # we wait until the events catch up in case repl command above did some HMS
+      # operations.
+      EventProcessorUtils.wait_for_event_processing(self)
+      # confirm the number of rows in target match with the source table.
+      rows_in_unpart_tbl_target = int(self.execute_scalar(
+        "select count(*) from {0}.{1}".format(target_db, unpartitioned_tbl))
+          .split('\t')[0])
+      rows_in_part_tbl_target = int(self.execute_scalar(
+        "select count(*) from {0}.{1}".format(target_db, partitioned_tbl)).split('\t')[0])
+      assert rows_in_unpart_tbl == rows_in_unpart_tbl_target
+      assert rows_in_part_tbl == rows_in_part_tbl_target
+
+      # Case IV: CTAS which creates a transactional table.
+      self.client.execute(
+        "create table {0}.insertonly_nopart_ctas {1} as "
+        "select * from {0}.{2}".format(source_db, TBLPROPERTIES, unpartitioned_tbl))
+      self.client.execute(
+        "create table {0}.insertonly_part_ctas partitioned by (year, month) {1}"
+        " as select * from {0}.{2}".format(source_db, TBLPROPERTIES, partitioned_tbl))
+      self.run_stmt_in_hive("repl dump {0}".format(source_db))
+      # replicate the table from source to target
+      self.run_stmt_in_hive("repl load {0} into {1}".format(source_db, target_db))
+      # we wait until the events catch up in case repl command above did some HMS
+      # operations.
+      EventProcessorUtils.wait_for_event_processing(self)
+      # confirm the number of rows in target match with the source table.
+      rows_in_unpart_tbl_source = int(self.execute_scalar("select count(*) from "
+        "{0}.insertonly_nopart_ctas".format(source_db)).split('\t')[0])
+      rows_in_unpart_tbl_target = int(self.execute_scalar("select count(*) from "
+          "{0}.insertonly_nopart_ctas".format(target_db)).split('\t')[0])
+      assert rows_in_unpart_tbl_source == rows_in_unpart_tbl_target
+      rows_in_unpart_tbl_source = int(self.execute_scalar("select count(*) from "
+        "{0}.insertonly_part_ctas".format(source_db)).split('\t')[0])
+      rows_in_unpart_tbl_target = int(self.execute_scalar("select count(*) from "
+        "{0}.insertonly_part_ctas".format(target_db)).split('\t')[0])
+      assert rows_in_unpart_tbl_source == rows_in_unpart_tbl_target
+
+      # Case V: truncate table
+      # impala truncates both the tables. Make sure replication sees that.
+      self.client.execute("truncate table {0}.{1}".format(source_db, unpartitioned_tbl))
+      self.client.execute("truncate table {0}.{1}".format(source_db, partitioned_tbl))
+      self.run_stmt_in_hive("repl dump {0}".format(source_db))
+      # replicate the table from source to target
+      self.run_stmt_in_hive("repl load {0} into {1}".format(source_db, target_db))
+      # we wait until the events catch up in case repl command above did some HMS
+      # operations.
+      EventProcessorUtils.wait_for_event_processing(self)
+      # confirm the number of rows in target match with the source table.
+      rows_in_unpart_tbl_target = int(self.execute_scalar(
+        "select count(*) from {0}.{1}".format(target_db, unpartitioned_tbl))
+          .split('\t')[0])
+      rows_in_part_tbl_target = int(self.execute_scalar(
+        "select count(*) from {0}.{1}".format(target_db, partitioned_tbl)).split('\t')[0])
+      assert rows_in_unpart_tbl_target == 0
+      assert rows_in_part_tbl_target == 0
+    finally:
+      src_db = self.__get_db_nothrow(source_db)
+      target_db_obj = self.__get_db_nothrow(target_db)
+      if src_db is not None:
+        self.run_stmt_in_hive(
+          "alter database {0} set dbproperties ('repl.source.for'='')".format(source_db))
+        self.run_stmt_in_hive("drop database if exists {0} cascade".format(source_db))
+      if target_db_obj is not None:
+        self.run_stmt_in_hive("drop database if exists {0} cascade".format(target_db))
+      # workaround for HIVE-24135. the managed db location doesn't get cleaned up
+      if src_db is not None and src_db.managedLocationUri is not None:
+        self.filesystem_client.delete_file_dir(src_db.managedLocationUri, True)
+      if target_db_obj is not None and target_db_obj.managedLocationUri is not None:
+        self.filesystem_client.delete_file_dir(target_db.src_db.managedLocationUri, True)
+
+  def __get_db_nothrow(self, name):
+    try:
+      return self.hive_client.get_database(name)
+    except:
+      return None
+
   def _run_test_empty_partition_events(self, unique_database, is_transactional):
     test_tbl = unique_database + ".test_events"
     TBLPROPERTIES = self.__get_transactional_tblproperties(is_transactional)
@@ -321,6 +494,7 @@
           # drop non-existing partition; essentially this is a no-op
           "alter table {0}.{1} drop if exists partition (year=2100, month=1)".format(
             db_name, tbl_name),
+          # empty table case where no insert events are generated
           "insert overwrite {0}.{1} select * from {0}.{1}".format(
             db_name, empty_unpartitioned_tbl),
           "insert overwrite {0}.{1} partition(part) select * from {0}.{1}".format(
@@ -328,10 +502,14 @@
       ]
     }
     if HIVE_MAJOR_VERSION >= 3:
-      # insert into a existing partition; generates INSERT events
+      # insert into a existing partition; generates INSERT self-event
       self_event_test_queries[True].append("insert into table {0}.{1} partition "
           "(year, month) select * from functional.alltypessmall where year=2009 "
           "and month=1".format(db_name, tbl2))
+      # insert overwrite query from Impala also generates a INSERT self-event
+      self_event_test_queries[True].append("insert overwrite table {0}.{1} partition "
+         "(year, month) select * from functional.alltypessmall where year=2009 "
+         "and month=1".format(db_name, tbl2))
     return self_event_test_queries
 
   def __get_hive_test_queries(self, db_name, recover_tbl_name):