HIVE-21079: Address Sankar's comments

Do not change get_partitions_by_names metastore API. Instead add another one
get_partitions_by_names_req function accepting a GetPartitionsByNamesRequest argument, returning
GetPartitionsByNamesResult output.

Get or update partition statistics in the same transaction in wich the partiton was obtained or
added resp.

We do not dump partitions in a metadata-only dump and hence we shouldn't dump DROP PARTITION events
as well.

Some other cosmetic comments.

Ashutosh Bapat.

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

@@ -113,7 +113,7 @@ public void tearDown() throws Throwable {
 
 
   private Map<String, String> collectStatsParams(Map<String, String> allParams) {
-    Map<String, String> statsParams = new HashMap<String, String>();
+    Map<String, String> statsParams = new HashMap<>();
     List<String> params = new ArrayList<>(StatsSetupConst.SUPPORTED_STATS);
     params.add(StatsSetupConst.COLUMN_STATS_ACCURATE);
     for (String param : params) {
@@ -174,7 +174,6 @@ private void verifyNoStatsReplicationForMetadataOnly(String tableName) throws Th
     // or false. Either is fine with us so don't bother checking exact values.
     Map<String, String> rParams =
             collectStatsParams(replica.getTable(replicatedDbName, tableName).getParameters());
-    List<String> params = new ArrayList<>(StatsSetupConst.SUPPORTED_STATS);
     Map<String, String> expectedFalseParams = new HashMap<>();
     Map<String, String> expectedTrueParams = new HashMap<>();
     StatsSetupConst.setStatsStateForCreateTable(expectedTrueParams,
@@ -209,9 +208,13 @@ private void verifyNoPartitionStatsReplicationForMetadataOnly(String tableName)
   }
 
   private List<String> createBootStrapData() throws Throwable {
+    // Unpartitioned table with data
     String simpleTableName = "sTable";
+    // partitioned table with data
     String partTableName = "pTable";
+    // Unpartitioned table without data during bootstrap and hence no stats
     String ndTableName = "ndTable";
+    // Partitioned table without data during bootstrap and hence no stats.
     String ndPartTableName = "ndPTable";
 
     primary.run("use " + primaryDbName)
@@ -224,7 +227,7 @@ private List<String> createBootStrapData() throws Throwable {
             .run("create table " + ndTableName + " (str string)")
             .run("create table " + ndPartTableName + " (val string) partitioned by (pk int)");
 
-    List<String> tableNames = new ArrayList<String>(Arrays.asList(simpleTableName, partTableName,
+    List<String> tableNames = new ArrayList<>(Arrays.asList(simpleTableName, partTableName,
             ndTableName, ndPartTableName));
 
     // Run analyze on each of the tables, if they are not being gathered automatically.
@@ -267,7 +270,7 @@ private String dumpLoadVerify(List<String> tableNames, String lastReplicationId,
             .dump(primaryDbName, lastReplicationId, withClauseList);
 
     // Load, if necessary changing configuration.
-    if (parallelLoad && lastReplicationId == null) {
+    if (parallelLoad) {
       replica.hiveConf.setBoolVar(HiveConf.ConfVars.EXECPARALLEL, true);
     }
 
@@ -299,13 +302,16 @@ private String dumpLoadVerify(List<String> tableNames, String lastReplicationId,
   }
 
   private void createIncrementalData(List<String> tableNames) throws Throwable {
+    // Annotations for this table are same as createBootStrapData
     String simpleTableName = "sTable";
     String partTableName = "pTable";
     String ndTableName = "ndTable";
     String ndPartTableName = "ndPTable";
 
     Assert.assertTrue(tableNames.containsAll(Arrays.asList(simpleTableName, partTableName,
                                                          ndTableName, ndPartTableName)));
+    // New tables created during incremental phase and thus loaded with data and stats during
+    // incremental phase.
     String incTableName = "iTable"; // New table
     String incPartTableName = "ipTable"; // New partitioned table
 
@@ -341,7 +347,7 @@ private void createIncrementalData(List<String> tableNames) throws Throwable {
     }
   }
 
-  public void testStatsReplicationCommon(boolean parallelBootstrap, boolean metadataOnly) throws Throwable {
+  private void testStatsReplicationCommon(boolean parallelBootstrap, boolean metadataOnly) throws Throwable {
     List<String> tableNames = createBootStrapData();
     String lastReplicationId = dumpLoadVerify(tableNames, null, parallelBootstrap,
             metadataOnly);

ql/src/java/org/apache/hadoop/hive/ql/exec/repl/bootstrap/events/filesystem/FSTableEvent.java

@@ -195,7 +195,7 @@ private AddPartitionDesc partitionDesc(Path fromPath,
       if (partition.isSetColStats() && !replicationSpec().isMigratingToTxnTable()) {
         ColumnStatistics colStats = partition.getColStats();
         ColumnStatisticsDesc colStatsDesc = new ColumnStatisticsDesc(colStats.getStatsDesc());
-        colStatsDesc.setDbName(tblDesc.getTableName());
+        colStatsDesc.setTableName(tblDesc.getTableName());
         colStatsDesc.setDbName(tblDesc.getDatabaseName());
         partDesc.setColStats(new ColumnStatistics(colStatsDesc, colStats.getStatsObj()));
       }

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

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.hive.ql.parse.repl.dump.events;
 
+import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.metastore.api.NotificationEvent;
 
 import org.apache.hadoop.hive.metastore.messaging.DropPartitionMessage;
@@ -38,6 +39,14 @@ DropPartitionMessage eventMessage(String stringRepresentation) {
   @Override
   public void handle(Context withinContext) throws Exception {
     LOG.info("Processing#{} DROP_PARTITION message : {}", fromEventId(), eventMessageAsJSON);
+
+    // We do not dump partitions during metadata only bootstrap dump (See TableExport
+    // .getPartitions(), for bootstrap dump we pass tableSpec with TABLE_ONLY set.). So don't
+    // dump partition related events for metadata-only dump.
+    if (withinContext.hiveConf.getBoolVar(HiveConf.ConfVars.REPL_DUMP_METADATA_ONLY)) {
+      return;
+    }
+
     DumpMetaData dmd = withinContext.createDmd(this);
     dmd.setPayload(eventMessageAsJSON);
     dmd.write();

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

@@ -48,11 +48,6 @@ public void handle(Context withinContext) throws Exception {
       return;
     }
 
-    if (!Utils.shouldReplicate(withinContext.replicationSpec, new Table(tableObj),
-                              withinContext.hiveConf)) {
-      return;
-    }
-
     // Statistics without any data does not make sense.
     if (withinContext.replicationSpec.isMetadataOnly()) {
       return;
@@ -64,6 +59,11 @@ public void handle(Context withinContext) throws Exception {
       return;
     }
 
+    if (!Utils.shouldReplicate(withinContext.replicationSpec, new Table(tableObj),
+                              withinContext.hiveConf)) {
+      return;
+    }
+
     DumpMetaData dmd = withinContext.createDmd(this);
     dmd.setPayload(eventMessageAsJSON);
     dmd.write();

ql/src/java/org/apache/hadoop/hive/ql/parse/repl/load/message/UpdatePartColStatHandler.java

@@ -43,11 +43,11 @@ public List<Task<? extends Serializable>> handle(Context context)
     // Update tablename and database name in the statistics object
     ColumnStatistics colStats = upcsm.getColumnStatistics();
     ColumnStatisticsDesc colStatsDesc = colStats.getStatsDesc();
-    colStatsDesc.setDbName(context.dbName);
     if (!context.isTableNameEmpty()) {
       colStatsDesc.setTableName(context.tableName);
     }
     if (!context.isDbNameEmpty()) {
+      colStatsDesc.setDbName(context.dbName);
       updatedMetadata.set(context.dmd.getEventTo().toString(), context.dbName, context.tableName,
                   null);
     }