diff --git ql/src/java/org/apache/hadoop/hive/ql/exec/repl/bootstrap/load/table/LoadPartitions.java ql/src/java/org/apache/hadoop/hive/ql/exec/repl/bootstrap/load/table/LoadPartitions.java index b36c4a531f..26a5ff9d6d 100644 --- ql/src/java/org/apache/hadoop/hive/ql/exec/repl/bootstrap/load/table/LoadPartitions.java +++ ql/src/java/org/apache/hadoop/hive/ql/exec/repl/bootstrap/load/table/LoadPartitions.java @@ -18,6 +18,7 @@ package org.apache.hadoop.hive.ql.exec.repl.bootstrap.load.table; import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.conf.HiveConf; import org.apache.hadoop.hive.metastore.TableType; import org.apache.hadoop.hive.metastore.Warehouse; import org.apache.hadoop.hive.metastore.api.Database; @@ -60,9 +61,11 @@ import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; +import java.util.LinkedList; import java.util.List; import java.util.Map; +import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.REPL_DUMP_METADATA_ONLY; import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.REPL_ENABLE_MOVE_OPTIMIZATION; import static org.apache.hadoop.hive.ql.exec.repl.bootstrap.load.ReplicationState.PartitionState; import static org.apache.hadoop.hive.ql.parse.ImportSemanticAnalyzer.isPartitioned; @@ -154,7 +157,44 @@ private ReplicationState initialReplicationState() throws SemanticException { ); } + private boolean isDDLOnlyOperation() { + return HiveConf.getBoolVar(context.hiveConf, REPL_DUMP_METADATA_ONLY); + } + + /** + * Get all partitions and consolidate them into single partition request. + * Also, copy relevant stats and other information from original request. + * + * @throws SemanticException + */ + private void addConsolidatedPartitionDesc() throws Exception { + List partitions = new LinkedList<>(); + for (AlterTableAddPartitionDesc alterTableAddPartitionDesc : event.partitionDescriptions(tableDesc)) { + + AlterTableAddPartitionDesc.PartitionDesc src = alterTableAddPartitionDesc.getPartitions().get(0); + + partitions.add(new AlterTableAddPartitionDesc.PartitionDesc( + src.getPartSpec(), src.getLocation(), src.getPartParams(), src.getInputFormat(), + src.getOutputFormat(), src.getNumBuckets(), src.getCols(), src.getSerializationLib(), + src.getSerdeParams(), src.getBucketCols(), src.getSortCols(), src.getColStats(), + src.getWriteId())); + } + AlterTableAddPartitionDesc consolidatedPartitionDesc = new AlterTableAddPartitionDesc(tableDesc.getDatabaseName(), + tableDesc.getTableName(), true, partitions); + + addPartition(false, consolidatedPartitionDesc, null); + if (partitions.size() > 0) { + LOG.info("Added {} partitions", partitions.size()); + } + } + private TaskTracker forNewTable() throws Exception { + if (isDDLOnlyOperation()) { + // Place all partitions in single task to reduce load on HMS. + addConsolidatedPartitionDesc(); + return tracker; + } + Iterator iterator = event.partitionDescriptions(tableDesc).iterator(); while (iterator.hasNext() && tracker.canAddMoreTasks()) { AlterTableAddPartitionDesc currentPartitionDesc = iterator.next();