New Feature: Metastore Database Prefix

main/src/main/java/com/airbnb/reair/batch/hive/MetastoreReplicationJob.java

@@ -317,6 +317,7 @@ private void mergeConfiguration(Configuration inputConfig, Configuration merged)
  ConfigurationKeys.DEST_CLUSTER_METASTORE_URL,
  ConfigurationKeys.DEST_HDFS_ROOT,
  ConfigurationKeys.DEST_HDFS_TMP,
+ ConfigurationKeys.DEST_CLUSTER_METASTORE_DB_PREFIX,
  ConfigurationKeys.BATCH_JOB_METASTORE_BLACKLIST,
  ConfigurationKeys.BATCH_JOB_CLUSTER_FACTORY_CLASS,
  ConfigurationKeys.BATCH_JOB_OUTPUT_DIR,

main/src/main/java/com/airbnb/reair/incremental/configuration/ConfiguredClusterFactory.java

@@ -55,14 +55,19 @@ public Cluster getDestCluster() throws ConfigurationException {
  ConfigurationKeys.DEST_HDFS_ROOT);
  String destHdfsTmp = conf.get(
  ConfigurationKeys.DEST_HDFS_TMP);
+ // Note that the DEST_CLUSTER_METASTORE_DB_PREFIX is a special config that only exists on the
+ // dest cluster (but not on the src cluster).
+ String destDbPrefix = conf.get(
+ ConfigurationKeys.DEST_CLUSTER_METASTORE_DB_PREFIX);
  return new HardCodedCluster(
  destClusterName,
  destMetastoreUrl.getHost(),
  destMetastoreUrl.getPort(),
  null,
  null,
  new Path(destHdfsRoot),
- new Path(destHdfsTmp));
+ new Path(destHdfsTmp),
+ destDbPrefix);
  }
 
  @Override

main/src/main/java/com/airbnb/reair/incremental/configuration/HardCodedCluster.java

@@ -1,5 +1,7 @@
 package com.airbnb.reair.incremental.configuration;
 
+import com.airbnb.reair.common.DbPrefixHiveMetastoreClient;
+import com.airbnb.reair.common.HiveMetastoreClient;
 import com.airbnb.reair.common.HiveMetastoreException;
 import com.airbnb.reair.common.ThriftHiveMetastoreClient;
 
@@ -17,7 +19,8 @@ public class HardCodedCluster implements Cluster {
  private String jobtrackerPort;
  private Path hdfsRoot;
  private Path tmpDir;
- private ThreadLocal<ThriftHiveMetastoreClient> metastoreClient;
+ private String dbNamePrefix;
+ private ThreadLocal<HiveMetastoreClient> metastoreClient;
 
  /**
  * Constructor with specific values.
@@ -29,6 +32,7 @@ public class HardCodedCluster implements Cluster {
  * @param jobtrackerPort port of the job tracker
  * @param hdfsRoot the path for the root HDFS directory
  * @param tmpDir the path for the temporary HDFS directory (should be under root)
+ * @param dbNamePrefix optional prefix for all databases in the Hive metastore.
  */
  public HardCodedCluster(
  String name,
@@ -37,15 +41,38 @@ public HardCodedCluster(
  String jobtrackerHost,
  String jobtrackerPort,
  Path hdfsRoot,
- Path tmpDir) {
+ Path tmpDir,
+ String dbNamePrefix) {
  this.name = name;
  this.metastoreHost = metastoreHost;
  this.metastorePort = metastorePort;
  this.jobtrackerHost = jobtrackerHost;
  this.jobtrackerPort = jobtrackerPort;
  this.hdfsRoot = hdfsRoot;
  this.tmpDir = tmpDir;
- this.metastoreClient = new ThreadLocal<ThriftHiveMetastoreClient>();
+ this.dbNamePrefix = dbNamePrefix;
+ this.metastoreClient = new ThreadLocal<>();
+ }
+
+ /**
+ * Constructor with specific values, with a default dbNamePrefix of null.
+ */
+ public HardCodedCluster(
+ String name,
+ String metastoreHost,
+ int metastorePort,
+ String jobtrackerHost,
+ String jobtrackerPort,
+ Path hdfsRoot,
+ Path tmpDir) {
+ this(name,
+ metastoreHost,
+ metastorePort,
+ jobtrackerHost,
+ jobtrackerPort,
+ hdfsRoot,
+ tmpDir,
+ null);
  }
 
  public String getMetastoreHost() {
@@ -58,11 +85,17 @@ public int getMetastorePort() {
 
  /**
  * Get a cached ThreadLocal metastore client.
+ *
+ * <p>Also applies a DbPrefixHiveMetastoreClient wrapper if dbNamePrefix is not null or
+ * empty.
  */
- public ThriftHiveMetastoreClient getMetastoreClient() throws HiveMetastoreException {
- ThriftHiveMetastoreClient result = this.metastoreClient.get();
+ public HiveMetastoreClient getMetastoreClient() throws HiveMetastoreException {
+ HiveMetastoreClient result = this.metastoreClient.get();
  if (result == null) {
  result = new ThriftHiveMetastoreClient(getMetastoreHost(), getMetastorePort());
+ if (dbNamePrefix != null && !dbNamePrefix.isEmpty()) {
+ result = new DbPrefixHiveMetastoreClient(result, dbNamePrefix);
+ }
  this.metastoreClient.set(result);
  }
  return result;

main/src/main/java/com/airbnb/reair/incremental/deploy/ConfigurationKeys.java

@@ -52,6 +52,9 @@ public class ConfigurationKeys {
  public static final String DEST_HDFS_ROOT = "airbnb.reair.clusters.dest.hdfs.root";
  // The root of the temporary directory for storing temporary files on the destination cluster
  public static final String DEST_HDFS_TMP = "airbnb.reair.clusters.dest.hdfs.tmp";
+ // Prefix to add to any databases on the destination cluster. Mostly used for production testing.
+ public static final String DEST_CLUSTER_METASTORE_DB_PREFIX =
+ "airbnb.reair.clusters.dest.metastore.db.prefix";
 
  // Class to use for filtering out entries from the audit log
  public static final String OBJECT_FILTER_CLASS = "airbnb.reair.object.filter";

main/src/main/java/com/airbnb/reair/incremental/deploy/HiveCopy.java

@@ -2,8 +2,8 @@
 
 import com.airbnb.reair.common.ArgumentException;
 import com.airbnb.reair.common.CliUtils;
+import com.airbnb.reair.common.HiveMetastoreClient;
 import com.airbnb.reair.common.HiveObjectSpec;
-import com.airbnb.reair.common.ThriftHiveMetastoreClient;
 import com.airbnb.reair.incremental.DirectoryCopier;
 import com.airbnb.reair.incremental.ReplicationUtils;
 import com.airbnb.reair.incremental.RunInfo;
@@ -160,7 +160,7 @@ public static int main(String[] args) throws Exception {
 
  if ("copy-unpartitioned-table".equals(op)) {
  LOG.info("Copying an unpartitioned table");
- ThriftHiveMetastoreClient ms = srcCluster.getMetastoreClient();
+ HiveMetastoreClient ms = srcCluster.getMetastoreClient();
  Table srcTable = ms.getTable(spec.getDbName(), spec.getTableName());
  CopyUnpartitionedTableTask job = new CopyUnpartitionedTableTask(conf,
  destinationObjectFactory, conflictHandler, srcCluster, destCluster, spec,
@@ -172,7 +172,7 @@ public static int main(String[] args) throws Exception {
  }
  } else if ("copy-partitioned-table".equals(op)) {
  LOG.info("Copying a partitioned table");
- ThriftHiveMetastoreClient ms = srcCluster.getMetastoreClient();
+ HiveMetastoreClient ms = srcCluster.getMetastoreClient();
  Table srcTable = ms.getTable(spec.getDbName(), spec.getTableName());
  CopyPartitionedTableTask job = new CopyPartitionedTableTask(conf, destinationObjectFactory,
  conflictHandler, srcCluster, destCluster, spec, ReplicationUtils.getLocation(srcTable));
@@ -183,7 +183,7 @@ public static int main(String[] args) throws Exception {
  }
  } else if (op.equals("copy-partition")) {
  LOG.info("Copying a partition");
- ThriftHiveMetastoreClient ms = srcCluster.getMetastoreClient();
+ HiveMetastoreClient ms = srcCluster.getMetastoreClient();
  Partition srcPartition =
  ms.getPartition(spec.getDbName(), spec.getTableName(), spec.getPartitionName());
  CopyPartitionTask job = new CopyPartitionTask(conf, destinationObjectFactory, conflictHandler,

main/src/main/resources/batch_replication_configuration_template.xml

@@ -62,6 +62,12 @@
  </comment>
  </property>
 
+ <property>
+ <name>airbnb.reair.clusters.dest.metastore.db.prefix</name>
+ <value></value>
+ <comment>Optional prefix to add to destination Hive database names.</comment>
+ </property>
+
  <property>
  <name>airbnb.reair.clusters.batch.output.dir</name>
  <value>hdfs://airfs-test/user/test/test_output</value>