You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by we...@apache.org on 2017/05/08 22:17:36 UTC
[19/50] [abbrv] hive git commit: HIVE-16268 : enable incremental repl
dump to handle functions metadata (Anishek Agarwal,
reviewed by Sushanth Sowmyan)
HIVE-16268 : enable incremental repl dump to handle functions metadata (Anishek Agarwal, reviewed by Sushanth Sowmyan)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/9d4f13af
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/9d4f13af
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/9d4f13af
Branch: refs/heads/hive-14535
Commit: 9d4f13afda34250b7cf722287a557426e90ff24d
Parents: 699d6ce
Author: Sushanth Sowmyan <kh...@gmail.com>
Authored: Thu May 4 02:48:27 2017 -0700
Committer: Sushanth Sowmyan <kh...@gmail.com>
Committed: Sun May 7 15:43:21 2017 -0700
----------------------------------------------------------------------
.../ql/parse/ReplicationSemanticAnalyzer.java | 306 ++-----------------
.../hadoop/hive/ql/parse/repl/DumpType.java | 1 +
.../parse/repl/dump/events/AbstractHandler.java | 46 +++
.../repl/dump/events/AddPartitionHandler.java | 114 +++++++
.../repl/dump/events/AlterPartitionHandler.java | 112 +++++++
.../repl/dump/events/AlterTableHandler.java | 102 +++++++
.../repl/dump/events/CreateFunctionHandler.java | 36 +++
.../repl/dump/events/CreateTableHandler.java | 86 ++++++
.../parse/repl/dump/events/DefaultHandler.java | 44 +++
.../repl/dump/events/DropPartitionHandler.java | 44 +++
.../repl/dump/events/DropTableHandler.java | 44 +++
.../ql/parse/repl/dump/events/EventHandler.java | 62 ++++
.../repl/dump/events/EventHandlerFactory.java | 76 +++++
.../parse/repl/dump/events/InsertHandler.java | 110 +++++++
.../ql/parse/repl/events/AbstractHandler.java | 46 ---
.../parse/repl/events/AddPartitionHandler.java | 114 -------
.../repl/events/AlterPartitionHandler.java | 112 -------
.../ql/parse/repl/events/AlterTableHandler.java | 102 -------
.../parse/repl/events/CreateTableHandler.java | 86 ------
.../ql/parse/repl/events/DefaultHandler.java | 44 ---
.../parse/repl/events/DropPartitionHandler.java | 44 ---
.../ql/parse/repl/events/DropTableHandler.java | 44 ---
.../hive/ql/parse/repl/events/EventHandler.java | 62 ----
.../parse/repl/events/EventHandlerFactory.java | 75 -----
.../ql/parse/repl/events/InsertHandler.java | 110 -------
.../load/message/AbstractMessageHandler.java | 67 ++++
.../parse/repl/load/message/DefaultHandler.java | 33 ++
.../repl/load/message/DropPartitionHandler.java | 108 +++++++
.../repl/load/message/DropTableHandler.java | 51 ++++
.../parse/repl/load/message/InsertHandler.java | 47 +++
.../parse/repl/load/message/MessageHandler.java | 91 ++++++
.../load/message/MessageHandlerFactory.java | 79 +++++
.../load/message/RenamePartitionHandler.java | 74 +++++
.../repl/load/message/RenameTableHandler.java | 81 +++++
.../parse/repl/load/message/TableHandler.java | 68 +++++
.../load/message/TruncatePartitionHandler.java | 69 +++++
.../repl/load/message/TruncateTableHandler.java | 50 +++
.../dump/events/TestEventHandlerFactory.java | 62 ++++
.../repl/events/TestEventHandlerFactory.java | 62 ----
39 files changed, 1781 insertions(+), 1183 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/9d4f13af/ql/src/java/org/apache/hadoop/hive/ql/parse/ReplicationSemanticAnalyzer.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/ReplicationSemanticAnalyzer.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/ReplicationSemanticAnalyzer.java
index 2daa123..5d1d2fd 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/ReplicationSemanticAnalyzer.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/ReplicationSemanticAnalyzer.java
@@ -27,22 +27,15 @@ 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.api.Database;
-import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.Function;
import org.apache.hadoop.hive.metastore.api.InvalidOperationException;
import org.apache.hadoop.hive.metastore.api.NotificationEvent;
+import org.apache.hadoop.hive.metastore.messaging.EventUtils;
+import org.apache.hadoop.hive.metastore.messaging.MessageFactory;
import org.apache.hadoop.hive.metastore.messaging.event.filters.AndFilter;
import org.apache.hadoop.hive.metastore.messaging.event.filters.DatabaseAndTableFilter;
import org.apache.hadoop.hive.metastore.messaging.event.filters.EventBoundaryFilter;
import org.apache.hadoop.hive.metastore.messaging.event.filters.MessageFormatFilter;
-import org.apache.hadoop.hive.metastore.messaging.AlterPartitionMessage;
-import org.apache.hadoop.hive.metastore.messaging.AlterTableMessage;
-import org.apache.hadoop.hive.metastore.messaging.DropPartitionMessage;
-import org.apache.hadoop.hive.metastore.messaging.DropTableMessage;
-import org.apache.hadoop.hive.metastore.messaging.EventUtils;
-import org.apache.hadoop.hive.metastore.messaging.InsertMessage;
-import org.apache.hadoop.hive.metastore.messaging.MessageDeserializer;
-import org.apache.hadoop.hive.metastore.messaging.MessageFactory;
import org.apache.hadoop.hive.ql.ErrorMsg;
import org.apache.hadoop.hive.ql.QueryState;
import org.apache.hadoop.hive.ql.exec.Task;
@@ -53,29 +46,23 @@ import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.metadata.Table;
import org.apache.hadoop.hive.ql.parse.repl.DumpType;
import org.apache.hadoop.hive.ql.parse.repl.dump.HiveWrapper;
+import org.apache.hadoop.hive.ql.parse.repl.dump.events.EventHandler;
+import org.apache.hadoop.hive.ql.parse.repl.dump.events.EventHandlerFactory;
import org.apache.hadoop.hive.ql.parse.repl.load.DumpMetaData;
import org.apache.hadoop.hive.ql.parse.repl.dump.io.FunctionSerializer;
import org.apache.hadoop.hive.ql.parse.repl.dump.io.JsonWriter;
import org.apache.hadoop.hive.ql.parse.repl.dump.Utils;
-import org.apache.hadoop.hive.ql.parse.repl.events.EventHandler;
-import org.apache.hadoop.hive.ql.parse.repl.events.EventHandlerFactory;
import org.apache.hadoop.hive.ql.parse.repl.load.MetaData;
+import org.apache.hadoop.hive.ql.parse.repl.load.message.MessageHandler;
+import org.apache.hadoop.hive.ql.parse.repl.load.message.MessageHandlerFactory;
import org.apache.hadoop.hive.ql.plan.AlterDatabaseDesc;
import org.apache.hadoop.hive.ql.plan.AlterTableDesc;
import org.apache.hadoop.hive.ql.plan.CreateDatabaseDesc;
import org.apache.hadoop.hive.ql.plan.CreateFunctionDesc;
import org.apache.hadoop.hive.ql.plan.DDLWork;
import org.apache.hadoop.hive.ql.plan.DependencyCollectionWork;
-import org.apache.hadoop.hive.ql.plan.DropTableDesc;
-import org.apache.hadoop.hive.ql.plan.ExprNodeColumnDesc;
-import org.apache.hadoop.hive.ql.plan.ExprNodeConstantDesc;
-import org.apache.hadoop.hive.ql.plan.ExprNodeGenericFuncDesc;
import org.apache.hadoop.hive.ql.plan.FunctionWork;
import org.apache.hadoop.hive.ql.plan.PlanUtils;
-import org.apache.hadoop.hive.ql.plan.RenamePartitionDesc;
-import org.apache.hadoop.hive.ql.plan.TruncateTableDesc;
-import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
-import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -87,7 +74,6 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
-import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -681,270 +667,26 @@ public class ReplicationSemanticAnalyzer extends BaseSemanticAnalyzer {
}
private List<Task<? extends Serializable>> analyzeEventLoad(
- String dbName, String tblName, String locn,
- Task<? extends Serializable> precursor,
- Map<String, Long> dbsUpdated, Map<String, Long> tablesUpdated,
- DumpMetaData dmd) throws SemanticException {
- MessageDeserializer md = MessageFactory.getInstance().getDeserializer();
- switch (dmd.getDumpType()) {
- case EVENT_CREATE_TABLE: {
- return analyzeTableLoad(dbName, tblName, locn, precursor, dbsUpdated, tablesUpdated);
- }
- case EVENT_ADD_PARTITION: {
- return analyzeTableLoad(dbName, tblName, locn, precursor, dbsUpdated, tablesUpdated);
- }
- case EVENT_DROP_TABLE: {
- DropTableMessage dropTableMessage = md.getDropTableMessage(dmd.getPayload());
- String actualDbName = ((dbName == null) || dbName.isEmpty() ? dropTableMessage.getDB() : dbName);
- String actualTblName = ((tblName == null) || tblName.isEmpty() ? dropTableMessage.getTable() : tblName);
- DropTableDesc dropTableDesc = new DropTableDesc(
- actualDbName + "." + actualTblName,
- null, true, true,
- getNewEventOnlyReplicationSpec(dmd.getEventFrom()));
- Task<DDLWork> dropTableTask = TaskFactory.get(new DDLWork(inputs, outputs, dropTableDesc), conf);
- if (precursor != null){
- precursor.addDependentTask(dropTableTask);
- }
- List<Task<? extends Serializable>> tasks = new ArrayList<Task<? extends Serializable>>();
- tasks.add(dropTableTask);
- LOG.debug("Added drop tbl task : {}:{}", dropTableTask.getId(), dropTableDesc.getTableName());
- dbsUpdated.put(actualDbName,dmd.getEventTo());
- return tasks;
- }
- case EVENT_DROP_PARTITION: {
- try {
- DropPartitionMessage dropPartitionMessage = md.getDropPartitionMessage(dmd.getPayload());
- String actualDbName = ((dbName == null) || dbName.isEmpty() ? dropPartitionMessage.getDB() : dbName);
- String actualTblName = ((tblName == null) || tblName.isEmpty() ? dropPartitionMessage.getTable() : tblName);
- Map<Integer, List<ExprNodeGenericFuncDesc>> partSpecs;
- partSpecs =
- genPartSpecs(new Table(dropPartitionMessage.getTableObj()),
- dropPartitionMessage.getPartitions());
- if (partSpecs.size() > 0) {
- DropTableDesc dropPtnDesc = new DropTableDesc(
- actualDbName + "." + actualTblName,
- partSpecs, null, true,
- getNewEventOnlyReplicationSpec(dmd.getEventFrom()));
- Task<DDLWork> dropPtnTask =
- TaskFactory.get(new DDLWork(inputs, outputs, dropPtnDesc), conf);
- if (precursor != null) {
- precursor.addDependentTask(dropPtnTask);
- }
- List<Task<? extends Serializable>> tasks = new ArrayList<Task<? extends Serializable>>();
- tasks.add(dropPtnTask);
- LOG.debug("Added drop ptn task : {}:{},{}", dropPtnTask.getId(),
- dropPtnDesc.getTableName(), dropPartitionMessage.getPartitions());
- dbsUpdated.put(actualDbName, dmd.getEventTo());
- tablesUpdated.put(actualDbName + "." + actualTblName, dmd.getEventTo());
- return tasks;
- } else {
- throw new SemanticException(
- "DROP PARTITION EVENT does not return any part descs for event message :"
- + dmd.getPayload());
- }
- } catch (Exception e) {
- if (!(e instanceof SemanticException)){
- throw new SemanticException("Error reading message members", e);
- } else {
- throw (SemanticException)e;
- }
- }
- }
- case EVENT_ALTER_TABLE: {
- return analyzeTableLoad(dbName, tblName, locn, precursor, dbsUpdated, tablesUpdated);
- }
- case EVENT_RENAME_TABLE: {
- AlterTableMessage renameTableMessage = md.getAlterTableMessage(dmd.getPayload());
- if ((tblName != null) && (!tblName.isEmpty())){
- throw new SemanticException("RENAMES of tables are not supported for table-level replication");
- }
- try {
- String oldDbName = renameTableMessage.getTableObjBefore().getDbName();
- String newDbName = renameTableMessage.getTableObjAfter().getDbName();
-
- if ((dbName != null) && (!dbName.isEmpty())){
- // If we're loading into a db, instead of into the warehouse, then the oldDbName and
- // newDbName must be the same
- if (!oldDbName.equalsIgnoreCase(newDbName)){
- throw new SemanticException("Cannot replicate an event renaming a table across"
- + " databases into a db level load " + oldDbName +"->" + newDbName);
- } else {
- // both were the same, and can be replaced by the new db we're loading into.
- oldDbName = dbName;
- newDbName = dbName;
- }
- }
-
- String oldName = oldDbName + "." + renameTableMessage.getTableObjBefore().getTableName();
- String newName = newDbName + "." + renameTableMessage.getTableObjAfter().getTableName();
- AlterTableDesc renameTableDesc = new AlterTableDesc(oldName, newName, false);
- Task<DDLWork> renameTableTask = TaskFactory.get(new DDLWork(inputs, outputs, renameTableDesc), conf);
- if (precursor != null){
- precursor.addDependentTask(renameTableTask);
- }
- List<Task<? extends Serializable>> tasks = new ArrayList<Task<? extends Serializable>>();
- tasks.add(renameTableTask);
- LOG.debug("Added rename table task : {}:{}->{}", renameTableTask.getId(), oldName, newName);
- dbsUpdated.put(newDbName, dmd.getEventTo()); // oldDbName and newDbName *will* be the same if we're here
- tablesUpdated.remove(oldName);
- tablesUpdated.put(newName, dmd.getEventTo());
- // Note : edge-case here in interaction with table-level REPL LOAD, where that nukes out tablesUpdated
- // However, we explicitly don't support repl of that sort, and error out above if so. If that should
- // ever change, this will need reworking.
- return tasks;
- } catch (Exception e) {
- if (!(e instanceof SemanticException)){
- throw new SemanticException("Error reading message members", e);
- } else {
- throw (SemanticException)e;
- }
- }
- }
- case EVENT_TRUNCATE_TABLE: {
- AlterTableMessage truncateTableMessage = md.getAlterTableMessage(dmd.getPayload());
- String actualDbName = ((dbName == null) || dbName.isEmpty() ? truncateTableMessage.getDB() : dbName);
- String actualTblName = ((tblName == null) || tblName.isEmpty() ? truncateTableMessage.getTable() : tblName);
-
- TruncateTableDesc truncateTableDesc = new TruncateTableDesc(
- actualDbName + "." + actualTblName, null);
- Task<DDLWork> truncateTableTask = TaskFactory.get(new DDLWork(inputs, outputs, truncateTableDesc), conf);
- if (precursor != null) {
- precursor.addDependentTask(truncateTableTask);
- }
-
- List<Task<? extends Serializable>> tasks = new ArrayList<Task<? extends Serializable>>();
- tasks.add(truncateTableTask);
- LOG.debug("Added truncate tbl task : {}:{}", truncateTableTask.getId(), truncateTableDesc.getTableName());
- dbsUpdated.put(actualDbName,dmd.getEventTo());
- return tasks;
- }
- case EVENT_ALTER_PARTITION: {
- return analyzeTableLoad(dbName, tblName, locn, precursor, dbsUpdated, tablesUpdated);
- }
- case EVENT_RENAME_PARTITION: {
- AlterPartitionMessage renamePtnMessage = md.getAlterPartitionMessage(dmd.getPayload());
- String actualDbName = ((dbName == null) || dbName.isEmpty() ? renamePtnMessage.getDB() : dbName);
- String actualTblName = ((tblName == null) || tblName.isEmpty() ? renamePtnMessage.getTable() : tblName);
-
- Map<String, String> newPartSpec = new LinkedHashMap<String,String>();
- Map<String, String> oldPartSpec = new LinkedHashMap<String,String>();
- String tableName = actualDbName + "." + actualTblName;
- try {
- org.apache.hadoop.hive.metastore.api.Table tblObj = renamePtnMessage.getTableObj();
- org.apache.hadoop.hive.metastore.api.Partition pobjBefore = renamePtnMessage.getPtnObjBefore();
- org.apache.hadoop.hive.metastore.api.Partition pobjAfter = renamePtnMessage.getPtnObjAfter();
- Iterator<String> beforeValIter = pobjBefore.getValuesIterator();
- Iterator<String> afterValIter = pobjAfter.getValuesIterator();
- for (FieldSchema fs : tblObj.getPartitionKeys()){
- oldPartSpec.put(fs.getName(), beforeValIter.next());
- newPartSpec.put(fs.getName(), afterValIter.next());
- }
- } catch (Exception e) {
- if (!(e instanceof SemanticException)){
- throw new SemanticException("Error reading message members", e);
- } else {
- throw (SemanticException)e;
- }
- }
-
- RenamePartitionDesc renamePtnDesc = new RenamePartitionDesc(tableName, oldPartSpec, newPartSpec);
- Task<DDLWork> renamePtnTask = TaskFactory.get(new DDLWork(inputs, outputs, renamePtnDesc), conf);
- if (precursor != null){
- precursor.addDependentTask(renamePtnTask);
- }
- List<Task<? extends Serializable>> tasks = new ArrayList<Task<? extends Serializable>>();
- tasks.add(renamePtnTask);
- LOG.debug("Added rename ptn task : {}:{}->{}", renamePtnTask.getId(), oldPartSpec, newPartSpec);
- dbsUpdated.put(actualDbName, dmd.getEventTo());
- tablesUpdated.put(tableName, dmd.getEventTo());
- return tasks;
- }
- case EVENT_TRUNCATE_PARTITION: {
- AlterPartitionMessage truncatePtnMessage = md.getAlterPartitionMessage(dmd.getPayload());
- String actualDbName = ((dbName == null) || dbName.isEmpty() ? truncatePtnMessage.getDB() : dbName);
- String actualTblName = ((tblName == null) || tblName.isEmpty() ? truncatePtnMessage.getTable() : tblName);
-
- Map<String, String> partSpec = new LinkedHashMap<String,String>();
- try {
- org.apache.hadoop.hive.metastore.api.Table tblObj = truncatePtnMessage.getTableObj();
- org.apache.hadoop.hive.metastore.api.Partition pobjAfter = truncatePtnMessage.getPtnObjAfter();
- Iterator<String> afterValIter = pobjAfter.getValuesIterator();
- for (FieldSchema fs : tblObj.getPartitionKeys()){
- partSpec.put(fs.getName(), afterValIter.next());
- }
- } catch (Exception e) {
- if (!(e instanceof SemanticException)){
- throw new SemanticException("Error reading message members", e);
- } else {
- throw (SemanticException)e;
- }
- }
-
- TruncateTableDesc truncateTableDesc = new TruncateTableDesc(
- actualDbName + "." + actualTblName, partSpec);
- Task<DDLWork> truncatePtnTask = TaskFactory.get(new DDLWork(inputs, outputs, truncateTableDesc), conf);
- if (precursor != null) {
- precursor.addDependentTask(truncatePtnTask);
- }
-
- List<Task<? extends Serializable>> tasks = new ArrayList<Task<? extends Serializable>>();
- tasks.add(truncatePtnTask);
- LOG.debug("Added truncate ptn task : {}:{}", truncatePtnTask.getId(), truncateTableDesc.getTableName());
- dbsUpdated.put(actualDbName,dmd.getEventTo());
- return tasks;
- }
- case EVENT_INSERT: {
- md = MessageFactory.getInstance().getDeserializer();
- InsertMessage insertMessage = md.getInsertMessage(dmd.getPayload());
- String actualDbName = ((dbName == null) || dbName.isEmpty() ? insertMessage.getDB() : dbName);
- String actualTblName = ((tblName == null) || tblName.isEmpty() ? insertMessage.getTable() : tblName);
-
- // Piggybacking in Import logic for now
- return analyzeTableLoad(actualDbName, actualTblName, locn, precursor, dbsUpdated, tablesUpdated);
- }
- case EVENT_UNKNOWN: {
- break;
- }
- default: {
- break;
- }
- }
- return null;
- }
-
- private Map<Integer, List<ExprNodeGenericFuncDesc>> genPartSpecs(Table table,
- List<Map<String, String>> partitions) throws SemanticException {
- Map<Integer, List<ExprNodeGenericFuncDesc>> partSpecs =
- new HashMap<Integer, List<ExprNodeGenericFuncDesc>>();
- int partPrefixLength = 0;
- if ((partitions != null) && (partitions.size() > 0)) {
- partPrefixLength = partitions.get(0).size();
- // pick the length of the first ptn, we expect all ptns listed to have the same number of
- // key-vals.
- }
- List<ExprNodeGenericFuncDesc> ptnDescs = new ArrayList<ExprNodeGenericFuncDesc>();
- for (Map<String, String> ptn : partitions) {
- // convert each key-value-map to appropriate expression.
- ExprNodeGenericFuncDesc expr = null;
- for (Map.Entry<String, String> kvp : ptn.entrySet()) {
- String key = kvp.getKey();
- Object val = kvp.getValue();
- String type = table.getPartColByName(key).getType();
- ;
- PrimitiveTypeInfo pti = TypeInfoFactory.getPrimitiveTypeInfo(type);
- ExprNodeColumnDesc column = new ExprNodeColumnDesc(pti, key, null, true);
- ExprNodeGenericFuncDesc op = DDLSemanticAnalyzer.makeBinaryPredicate(
- "=", column, new ExprNodeConstantDesc(pti, val));
- expr = (expr == null) ? op : DDLSemanticAnalyzer.makeBinaryPredicate("and", expr, op);
- }
- if (expr != null) {
- ptnDescs.add(expr);
+ String dbName, String tblName, String locn, Task<? extends Serializable> precursor,
+ Map<String, Long> dbsUpdated, Map<String, Long> tablesUpdated, DumpMetaData dmd)
+ throws SemanticException {
+ MessageHandler.Context context =
+ new MessageHandler.Context(dbName, tblName, locn, precursor, dmd, conf, db, ctx, LOG);
+ MessageHandler messageHandler = MessageHandlerFactory.handlerFor(dmd.getDumpType());
+ List<Task<? extends Serializable>> tasks = messageHandler.handle(context);
+
+ if (precursor != null) {
+ for (Task<? extends Serializable> t : tasks) {
+ precursor.addDependentTask(t);
+ LOG.debug("Added {}:{} as a precursor of {}:{}",
+ precursor.getClass(), precursor.getId(), t.getClass(), t.getId());
}
}
- if (ptnDescs.size() > 0) {
- partSpecs.put(partPrefixLength, ptnDescs);
- }
- return partSpecs;
+ dbsUpdated.putAll(messageHandler.databasesUpdated());
+ tablesUpdated.putAll(messageHandler.tablesUpdated());
+ inputs.addAll(messageHandler.readEntities());
+ outputs.addAll(messageHandler.writeEntities());
+ return tasks;
}
private boolean existEmptyDb(String dbName) throws InvalidOperationException, HiveException {
http://git-wip-us.apache.org/repos/asf/hive/blob/9d4f13af/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/DumpType.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/DumpType.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/DumpType.java
index b1df5a3..c2cffaa 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/DumpType.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/DumpType.java
@@ -31,6 +31,7 @@ public enum DumpType {
EVENT_RENAME_PARTITION("EVENT_RENAME_PARTITION"),
EVENT_TRUNCATE_PARTITION("EVENT_TRUNCATE_PARTITION"),
EVENT_INSERT("EVENT_INSERT"),
+ EVENT_CREATE_FUNCTION("EVENT_CREATE_FUNCTION"),
EVENT_UNKNOWN("EVENT_UNKNOWN");
String type = null;
http://git-wip-us.apache.org/repos/asf/hive/blob/9d4f13af/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/AbstractHandler.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/AbstractHandler.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/AbstractHandler.java
new file mode 100644
index 0000000..ba699e3
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/AbstractHandler.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hive.ql.parse.repl.dump.events;
+
+import org.apache.hadoop.hive.metastore.api.NotificationEvent;
+import org.apache.hadoop.hive.metastore.messaging.MessageDeserializer;
+import org.apache.hadoop.hive.metastore.messaging.MessageFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+abstract class AbstractHandler implements EventHandler {
+ static final Logger LOG = LoggerFactory.getLogger(AbstractHandler.class);
+
+ final NotificationEvent event;
+ final MessageDeserializer deserializer;
+
+ AbstractHandler(NotificationEvent event) {
+ this.event = event;
+ deserializer = MessageFactory.getInstance().getDeserializer();
+ }
+
+ @Override
+ public long fromEventId() {
+ return event.getEventId();
+ }
+
+ @Override
+ public long toEventId() {
+ return event.getEventId();
+ }
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/9d4f13af/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/AddPartitionHandler.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/AddPartitionHandler.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/AddPartitionHandler.java
new file mode 100644
index 0000000..f4239e5
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/AddPartitionHandler.java
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hive.ql.parse.repl.dump.events;
+
+import com.google.common.base.Function;
+import com.google.common.collect.Iterables;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.metastore.api.NotificationEvent;
+import org.apache.hadoop.hive.metastore.messaging.AddPartitionMessage;
+import org.apache.hadoop.hive.metastore.messaging.PartitionFiles;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.metadata.Partition;
+import org.apache.hadoop.hive.ql.metadata.Table;
+import org.apache.hadoop.hive.ql.parse.EximUtil;
+
+import javax.annotation.Nullable;
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.util.Iterator;
+
+import org.apache.hadoop.hive.ql.parse.repl.DumpType;
+
+class AddPartitionHandler extends AbstractHandler {
+ protected AddPartitionHandler(NotificationEvent notificationEvent) {
+ super(notificationEvent);
+ }
+
+ @Override
+ public void handle(Context withinContext) throws Exception {
+ AddPartitionMessage apm = deserializer.getAddPartitionMessage(event.getMessage());
+ LOG.info("Processing#{} ADD_PARTITION message : {}", fromEventId(), event.getMessage());
+ Iterable<org.apache.hadoop.hive.metastore.api.Partition> ptns = apm.getPartitionObjs();
+ if ((ptns == null) || (!ptns.iterator().hasNext())) {
+ LOG.debug("Event#{} was an ADD_PTN_EVENT with no partitions");
+ return;
+ }
+ org.apache.hadoop.hive.metastore.api.Table tobj = apm.getTableObj();
+ if (tobj == null) {
+ LOG.debug("Event#{} was a ADD_PTN_EVENT with no table listed");
+ return;
+ }
+
+ final Table qlMdTable = new Table(tobj);
+ Iterable<Partition> qlPtns = Iterables.transform(
+ ptns,
+ new Function<org.apache.hadoop.hive.metastore.api.Partition, Partition>() {
+ @Nullable
+ @Override
+ public Partition apply(@Nullable org.apache.hadoop.hive.metastore.api.Partition input) {
+ if (input == null) {
+ return null;
+ }
+ try {
+ return new Partition(qlMdTable, input);
+ } catch (HiveException e) {
+ throw new IllegalArgumentException(e);
+ }
+ }
+ }
+ );
+
+ Path metaDataPath = new Path(withinContext.eventRoot, EximUtil.METADATA_NAME);
+ EximUtil.createExportDump(
+ metaDataPath.getFileSystem(withinContext.hiveConf),
+ metaDataPath,
+ qlMdTable,
+ qlPtns,
+ withinContext.replicationSpec);
+
+ Iterator<PartitionFiles> partitionFilesIter = apm.getPartitionFilesIter().iterator();
+ for (Partition qlPtn : qlPtns) {
+ Iterable<String> files = partitionFilesIter.next().getFiles();
+ if (files != null) {
+ // encoded filename/checksum of files, write into _files
+ try (BufferedWriter fileListWriter = writer(withinContext, qlPtn)) {
+ for (String file : files) {
+ fileListWriter.write(file + "\n");
+ }
+ }
+ }
+ }
+ withinContext.createDmd(this).write();
+ }
+
+ private BufferedWriter writer(Context withinContext, Partition qlPtn)
+ throws IOException {
+ Path ptnDataPath = new Path(withinContext.eventRoot, qlPtn.getName());
+ FileSystem fs = ptnDataPath.getFileSystem(withinContext.hiveConf);
+ Path filesPath = new Path(ptnDataPath, EximUtil.FILES_NAME);
+ return new BufferedWriter(new OutputStreamWriter(fs.create(filesPath)));
+ }
+
+ @Override
+ public DumpType dumpType() {
+ return DumpType.EVENT_ADD_PARTITION;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/9d4f13af/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/AlterPartitionHandler.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/AlterPartitionHandler.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/AlterPartitionHandler.java
new file mode 100644
index 0000000..8a7e742
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/AlterPartitionHandler.java
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hive.ql.parse.repl.dump.events;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.metastore.api.NotificationEvent;
+import org.apache.hadoop.hive.metastore.messaging.AlterPartitionMessage;
+import org.apache.hadoop.hive.ql.metadata.Partition;
+import org.apache.hadoop.hive.ql.metadata.Table;
+import org.apache.hadoop.hive.ql.parse.EximUtil;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.ql.parse.repl.DumpType;
+
+import org.apache.hadoop.hive.ql.parse.repl.load.DumpMetaData;
+
+class AlterPartitionHandler extends AbstractHandler {
+ private final org.apache.hadoop.hive.metastore.api.Partition after;
+ private final org.apache.hadoop.hive.metastore.api.Table tableObject;
+ private final boolean isTruncateOp;
+ private final Scenario scenario;
+
+ AlterPartitionHandler(NotificationEvent event) throws Exception {
+ super(event);
+ AlterPartitionMessage apm = deserializer.getAlterPartitionMessage(event.getMessage());
+ tableObject = apm.getTableObj();
+ org.apache.hadoop.hive.metastore.api.Partition before = apm.getPtnObjBefore();
+ after = apm.getPtnObjAfter();
+ isTruncateOp = apm.getIsTruncateOp();
+ scenario = scenarioType(before, after);
+ }
+
+ private enum Scenario {
+ ALTER {
+ @Override
+ DumpType dumpType() {
+ return DumpType.EVENT_ALTER_PARTITION;
+ }
+ },
+ RENAME {
+ @Override
+ DumpType dumpType() {
+ return DumpType.EVENT_RENAME_PARTITION;
+ }
+ },
+ TRUNCATE {
+ @Override
+ DumpType dumpType() {
+ return DumpType.EVENT_TRUNCATE_PARTITION;
+ }
+ };
+
+ abstract DumpType dumpType();
+ }
+
+ private Scenario scenarioType(org.apache.hadoop.hive.metastore.api.Partition before,
+ org.apache.hadoop.hive.metastore.api.Partition after) {
+ Iterator<String> beforeValIter = before.getValuesIterator();
+ Iterator<String> afterValIter = after.getValuesIterator();
+ while(beforeValIter.hasNext()) {
+ if (!beforeValIter.next().equals(afterValIter.next())) {
+ return Scenario.RENAME;
+ }
+ }
+ return isTruncateOp ? Scenario.TRUNCATE : Scenario.ALTER;
+ }
+
+ @Override
+ public void handle(Context withinContext) throws Exception {
+ LOG.info("Processing#{} ALTER_PARTITION message : {}", fromEventId(), event.getMessage());
+
+ if (Scenario.ALTER == scenario) {
+ withinContext.replicationSpec.setIsMetadataOnly(true);
+ Table qlMdTable = new Table(tableObject);
+ List<Partition> partitions = new ArrayList<>();
+ partitions.add(new Partition(qlMdTable, after));
+ Path metaDataPath = new Path(withinContext.eventRoot, EximUtil.METADATA_NAME);
+ EximUtil.createExportDump(
+ metaDataPath.getFileSystem(withinContext.hiveConf),
+ metaDataPath,
+ qlMdTable,
+ partitions,
+ withinContext.replicationSpec);
+ }
+ DumpMetaData dmd = withinContext.createDmd(this);
+ dmd.setPayload(event.getMessage());
+ dmd.write();
+ }
+
+ @Override
+ public DumpType dumpType() {
+ return scenario.dumpType();
+ }
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/9d4f13af/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/AlterTableHandler.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/AlterTableHandler.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/AlterTableHandler.java
new file mode 100644
index 0000000..f457f23
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/AlterTableHandler.java
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hive.ql.parse.repl.dump.events;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.metastore.api.NotificationEvent;
+import org.apache.hadoop.hive.metastore.messaging.AlterTableMessage;
+import org.apache.hadoop.hive.ql.metadata.Table;
+import org.apache.hadoop.hive.ql.parse.EximUtil;
+
+import org.apache.hadoop.hive.ql.parse.repl.DumpType;
+
+import org.apache.hadoop.hive.ql.parse.repl.load.DumpMetaData;
+
+class AlterTableHandler extends AbstractHandler {
+ private final org.apache.hadoop.hive.metastore.api.Table after;
+ private final boolean isTruncateOp;
+ private final Scenario scenario;
+
+ private enum Scenario {
+ ALTER {
+ @Override
+ DumpType dumpType() {
+ return DumpType.EVENT_ALTER_TABLE;
+ }
+ },
+ RENAME {
+ @Override
+ DumpType dumpType() {
+ return DumpType.EVENT_RENAME_TABLE;
+ }
+ },
+ TRUNCATE {
+ @Override
+ DumpType dumpType() {
+ return DumpType.EVENT_TRUNCATE_TABLE;
+ }
+ };
+
+ abstract DumpType dumpType();
+ }
+
+ AlterTableHandler(NotificationEvent event) throws Exception {
+ super(event);
+ AlterTableMessage atm = deserializer.getAlterTableMessage(event.getMessage());
+ org.apache.hadoop.hive.metastore.api.Table before = atm.getTableObjBefore();
+ after = atm.getTableObjAfter();
+ isTruncateOp = atm.getIsTruncateOp();
+ scenario = scenarioType(before, after);
+ }
+
+ private Scenario scenarioType(org.apache.hadoop.hive.metastore.api.Table before,
+ org.apache.hadoop.hive.metastore.api.Table after) {
+ if (before.getDbName().equals(after.getDbName())
+ && before.getTableName().equals(after.getTableName())) {
+ return isTruncateOp ? Scenario.TRUNCATE : Scenario.ALTER;
+ } else {
+ return Scenario.RENAME;
+ }
+ }
+
+ @Override
+ public void handle(Context withinContext) throws Exception {
+ {
+ LOG.info("Processing#{} ALTER_TABLE message : {}", fromEventId(), event.getMessage());
+ if (Scenario.ALTER == scenario) {
+ withinContext.replicationSpec.setIsMetadataOnly(true);
+ Table qlMdTableAfter = new Table(after);
+ Path metaDataPath = new Path(withinContext.eventRoot, EximUtil.METADATA_NAME);
+ EximUtil.createExportDump(
+ metaDataPath.getFileSystem(withinContext.hiveConf),
+ metaDataPath,
+ qlMdTableAfter,
+ null,
+ withinContext.replicationSpec);
+ }
+ DumpMetaData dmd = withinContext.createDmd(this);
+ dmd.setPayload(event.getMessage());
+ dmd.write();
+ }
+ }
+
+ @Override
+ public DumpType dumpType() {
+ return scenario.dumpType();
+ }
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/9d4f13af/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/CreateFunctionHandler.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/CreateFunctionHandler.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/CreateFunctionHandler.java
new file mode 100644
index 0000000..bebf035
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/CreateFunctionHandler.java
@@ -0,0 +1,36 @@
+package org.apache.hadoop.hive.ql.parse.repl.dump.events;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.metastore.api.NotificationEvent;
+import org.apache.hadoop.hive.metastore.messaging.CreateFunctionMessage;
+import org.apache.hadoop.hive.ql.parse.EximUtil;
+import org.apache.hadoop.hive.ql.parse.repl.dump.io.FunctionSerializer;
+import org.apache.hadoop.hive.ql.parse.repl.dump.io.JsonWriter;
+
+import org.apache.hadoop.hive.ql.parse.repl.DumpType;
+
+class CreateFunctionHandler extends AbstractHandler {
+ CreateFunctionHandler(NotificationEvent event) {
+ super(event);
+ }
+
+ @Override
+ public void handle(Context withinContext) throws Exception {
+ CreateFunctionMessage createFunctionMessage =
+ deserializer.getCreateFunctionMessage(event.getMessage());
+ LOG.info("Processing#{} CREATE_MESSAGE message : {}", fromEventId(), event.getMessage());
+ Path metadataPath = new Path(withinContext.eventRoot, EximUtil.METADATA_NAME);
+ FileSystem fileSystem = metadataPath.getFileSystem(withinContext.hiveConf);
+
+ try (JsonWriter jsonWriter = new JsonWriter(fileSystem, metadataPath)) {
+ new FunctionSerializer(createFunctionMessage.getFunctionObj())
+ .writeTo(jsonWriter, withinContext.replicationSpec);
+ }
+ }
+
+ @Override
+ public DumpType dumpType() {
+ return DumpType.EVENT_CREATE_FUNCTION;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/9d4f13af/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/CreateTableHandler.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/CreateTableHandler.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/CreateTableHandler.java
new file mode 100644
index 0000000..ca3607f
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/CreateTableHandler.java
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hive.ql.parse.repl.dump.events;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.metastore.api.NotificationEvent;
+import org.apache.hadoop.hive.metastore.messaging.CreateTableMessage;
+import org.apache.hadoop.hive.ql.metadata.Table;
+import org.apache.hadoop.hive.ql.parse.EximUtil;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+
+import org.apache.hadoop.hive.ql.parse.repl.DumpType;
+
+class CreateTableHandler extends AbstractHandler {
+
+ CreateTableHandler(NotificationEvent event) {
+ super(event);
+ }
+
+ @Override
+ public void handle(Context withinContext) throws Exception {
+ CreateTableMessage ctm = deserializer.getCreateTableMessage(event.getMessage());
+ LOG.info("Processing#{} CREATE_TABLE message : {}", fromEventId(), event.getMessage());
+ org.apache.hadoop.hive.metastore.api.Table tobj = ctm.getTableObj();
+
+ if (tobj == null) {
+ LOG.debug("Event#{} was a CREATE_TABLE_EVENT with no table listed");
+ return;
+ }
+
+ Table qlMdTable = new Table(tobj);
+ if (qlMdTable.isView()) {
+ withinContext.replicationSpec.setIsMetadataOnly(true);
+ }
+
+ Path metaDataPath = new Path(withinContext.eventRoot, EximUtil.METADATA_NAME);
+ EximUtil.createExportDump(
+ metaDataPath.getFileSystem(withinContext.hiveConf),
+ metaDataPath,
+ qlMdTable,
+ null,
+ withinContext.replicationSpec);
+
+ Path dataPath = new Path(withinContext.eventRoot, "data");
+ Iterable<String> files = ctm.getFiles();
+ if (files != null) {
+ // encoded filename/checksum of files, write into _files
+ try (BufferedWriter fileListWriter = writer(withinContext, dataPath)) {
+ for (String file : files) {
+ fileListWriter.write(file + "\n");
+ }
+ }
+ }
+ withinContext.createDmd(this).write();
+ }
+
+ private BufferedWriter writer(Context withinContext, Path dataPath) throws IOException {
+ FileSystem fs = dataPath.getFileSystem(withinContext.hiveConf);
+ Path filesPath = new Path(dataPath, EximUtil.FILES_NAME);
+ return new BufferedWriter(new OutputStreamWriter(fs.create(filesPath)));
+ }
+
+ @Override
+ public DumpType dumpType() {
+ return DumpType.EVENT_CREATE_TABLE;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/9d4f13af/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/DefaultHandler.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/DefaultHandler.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/DefaultHandler.java
new file mode 100644
index 0000000..0d4665a
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/DefaultHandler.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hive.ql.parse.repl.dump.events;
+
+import org.apache.hadoop.hive.metastore.api.NotificationEvent;
+
+import org.apache.hadoop.hive.ql.parse.repl.DumpType;
+
+import org.apache.hadoop.hive.ql.parse.repl.load.DumpMetaData;
+
+class DefaultHandler extends AbstractHandler {
+
+ DefaultHandler(NotificationEvent event) {
+ super(event);
+ }
+
+ @Override
+ public void handle(Context withinContext) throws Exception {
+ LOG.info("Dummy processing#{} message : {}", fromEventId(), event.getMessage());
+ DumpMetaData dmd = withinContext.createDmd(this);
+ dmd.setPayload(event.getMessage());
+ dmd.write();
+ }
+
+ @Override
+ public DumpType dumpType() {
+ return DumpType.EVENT_UNKNOWN;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/9d4f13af/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/DropPartitionHandler.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/DropPartitionHandler.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/DropPartitionHandler.java
new file mode 100644
index 0000000..a4eacc4
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/DropPartitionHandler.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hive.ql.parse.repl.dump.events;
+
+import org.apache.hadoop.hive.metastore.api.NotificationEvent;
+
+import org.apache.hadoop.hive.ql.parse.repl.DumpType;
+
+import org.apache.hadoop.hive.ql.parse.repl.load.DumpMetaData;
+
+class DropPartitionHandler extends AbstractHandler {
+
+ DropPartitionHandler(NotificationEvent event) {
+ super(event);
+ }
+
+ @Override
+ public void handle(Context withinContext) throws Exception {
+ LOG.info("Processing#{} DROP_PARTITION message : {}", fromEventId(), event.getMessage());
+ DumpMetaData dmd = withinContext.createDmd(this);
+ dmd.setPayload(event.getMessage());
+ dmd.write();
+ }
+
+ @Override
+ public DumpType dumpType() {
+ return DumpType.EVENT_DROP_PARTITION;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/9d4f13af/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/DropTableHandler.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/DropTableHandler.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/DropTableHandler.java
new file mode 100644
index 0000000..40cd5cb
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/DropTableHandler.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hive.ql.parse.repl.dump.events;
+
+import org.apache.hadoop.hive.metastore.api.NotificationEvent;
+
+import org.apache.hadoop.hive.ql.parse.repl.DumpType;
+
+import org.apache.hadoop.hive.ql.parse.repl.load.DumpMetaData;
+
+class DropTableHandler extends AbstractHandler {
+
+ DropTableHandler(NotificationEvent event) {
+ super(event);
+ }
+
+ @Override
+ public void handle(Context withinContext) throws Exception {
+ LOG.info("Processing#{} DROP_TABLE message : {}", fromEventId(), event.getMessage());
+ DumpMetaData dmd = withinContext.createDmd(this);
+ dmd.setPayload(event.getMessage());
+ dmd.write();
+ }
+
+ @Override
+ public DumpType dumpType() {
+ return DumpType.EVENT_DROP_TABLE;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/9d4f13af/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/EventHandler.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/EventHandler.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/EventHandler.java
new file mode 100644
index 0000000..c0fa7b2
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/EventHandler.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hive.ql.parse.repl.dump.events;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.metadata.Hive;
+import org.apache.hadoop.hive.ql.parse.ReplicationSpec;
+
+import org.apache.hadoop.hive.ql.parse.repl.load.DumpMetaData;
+import org.apache.hadoop.hive.ql.parse.repl.DumpType;
+
+public interface EventHandler {
+ void handle(Context withinContext) throws Exception;
+
+ long fromEventId();
+
+ long toEventId();
+
+ DumpType dumpType();
+
+ class Context {
+ final Path eventRoot, cmRoot;
+ final Hive db;
+ final HiveConf hiveConf;
+ final ReplicationSpec replicationSpec;
+
+ public Context(Path eventRoot, Path cmRoot, Hive db, HiveConf hiveConf,
+ ReplicationSpec replicationSpec) {
+ this.eventRoot = eventRoot;
+ this.cmRoot = cmRoot;
+ this.db = db;
+ this.hiveConf = hiveConf;
+ this.replicationSpec = replicationSpec;
+ }
+
+ DumpMetaData createDmd(EventHandler eventHandler) {
+ return new DumpMetaData(
+ eventRoot,
+ eventHandler.dumpType(),
+ eventHandler.fromEventId(),
+ eventHandler.toEventId(),
+ cmRoot, hiveConf
+ );
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/9d4f13af/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/EventHandlerFactory.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/EventHandlerFactory.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/EventHandlerFactory.java
new file mode 100644
index 0000000..08dbd13
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/EventHandlerFactory.java
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hive.ql.parse.repl.dump.events;
+
+import org.apache.hadoop.hive.metastore.api.NotificationEvent;
+import org.apache.hadoop.hive.metastore.messaging.MessageFactory;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Modifier;
+import java.util.HashMap;
+import java.util.Map;
+
+public class EventHandlerFactory {
+ private EventHandlerFactory() {
+ }
+
+ private static Map<String, Class<? extends EventHandler>> registeredHandlers = new HashMap<>();
+
+ static {
+ register(MessageFactory.ADD_PARTITION_EVENT, AddPartitionHandler.class);
+ register(MessageFactory.ALTER_PARTITION_EVENT, AlterPartitionHandler.class);
+ register(MessageFactory.ALTER_TABLE_EVENT, AlterTableHandler.class);
+ register(MessageFactory.CREATE_FUNCTION_EVENT, CreateFunctionHandler.class);
+ register(MessageFactory.CREATE_TABLE_EVENT, CreateTableHandler.class);
+ register(MessageFactory.DROP_PARTITION_EVENT, DropPartitionHandler.class);
+ register(MessageFactory.DROP_TABLE_EVENT, DropTableHandler.class);
+ register(MessageFactory.INSERT_EVENT, InsertHandler.class);
+ }
+
+ static void register(String event, Class<? extends EventHandler> handlerClazz) {
+ try {
+ Constructor<? extends EventHandler> constructor =
+ handlerClazz.getDeclaredConstructor(NotificationEvent.class);
+ assert constructor != null;
+ assert !Modifier.isPrivate(constructor.getModifiers());
+ registeredHandlers.put(event, handlerClazz);
+ } catch (NoSuchMethodException e) {
+ throw new IllegalArgumentException("handler class: " + handlerClazz.getCanonicalName()
+ + " does not have the a constructor with only parameter of type:"
+ + NotificationEvent.class.getCanonicalName(), e);
+ }
+ }
+
+ public static EventHandler handlerFor(NotificationEvent event) {
+ if (registeredHandlers.containsKey(event.getEventType())) {
+ Class<? extends EventHandler> handlerClazz = registeredHandlers.get(event.getEventType());
+ try {
+ Constructor<? extends EventHandler> constructor =
+ handlerClazz.getDeclaredConstructor(NotificationEvent.class);
+ return constructor.newInstance(event);
+ } catch (NoSuchMethodException | IllegalAccessException | InstantiationException | InvocationTargetException e) {
+ // this should never happen. however we want to make sure we propagate the exception
+ throw new RuntimeException(
+ "failed when creating handler for " + event.getEventType()
+ + " with the responsible class being " + handlerClazz.getCanonicalName(), e);
+ }
+ }
+ return new DefaultHandler(event);
+ }
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/9d4f13af/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/InsertHandler.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/InsertHandler.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/InsertHandler.java
new file mode 100644
index 0000000..0393701
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/events/InsertHandler.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hive.ql.parse.repl.dump.events;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.metastore.api.NotificationEvent;
+import org.apache.hadoop.hive.metastore.messaging.InsertMessage;
+import org.apache.hadoop.hive.ql.metadata.Partition;
+import org.apache.hadoop.hive.ql.parse.EximUtil;
+import org.apache.thrift.TException;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hadoop.hive.ql.parse.repl.DumpType;
+
+import org.apache.hadoop.hive.ql.parse.repl.load.DumpMetaData;
+
+class InsertHandler extends AbstractHandler {
+
+ InsertHandler(NotificationEvent event) {
+ super(event);
+ }
+
+ @Override
+ public void handle(Context withinContext) throws Exception {
+ InsertMessage insertMsg = deserializer.getInsertMessage(event.getMessage());
+ org.apache.hadoop.hive.ql.metadata.Table qlMdTable = tableObject(withinContext, insertMsg);
+ Map<String, String> partSpec = insertMsg.getPartitionKeyValues();
+ List<Partition> qlPtns = null;
+ if (qlMdTable.isPartitioned() && !partSpec.isEmpty()) {
+ qlPtns = Collections.singletonList(withinContext.db.getPartition(qlMdTable, partSpec, false));
+ }
+ Path metaDataPath = new Path(withinContext.eventRoot, EximUtil.METADATA_NAME);
+
+ // Mark the replace type based on INSERT-INTO or INSERT_OVERWRITE operation
+ withinContext.replicationSpec.setIsReplace(insertMsg.isReplace());
+ EximUtil.createExportDump(metaDataPath.getFileSystem(withinContext.hiveConf), metaDataPath,
+ qlMdTable, qlPtns,
+ withinContext.replicationSpec);
+ Iterable<String> files = insertMsg.getFiles();
+
+ if (files != null) {
+ Path dataPath;
+ if ((null == qlPtns) || qlPtns.isEmpty()) {
+ dataPath = new Path(withinContext.eventRoot, EximUtil.DATA_PATH_NAME);
+ } else {
+ /*
+ * Insert into/overwrite operation shall operate on one or more partitions or even partitions from multiple
+ * tables. But, Insert event is generated for each partition to which the data is inserted. So, qlPtns list
+ * will have only one entry.
+ */
+ assert(1 == qlPtns.size());
+ dataPath = new Path(withinContext.eventRoot, qlPtns.get(0).getName());
+ }
+
+ // encoded filename/checksum of files, write into _files
+ try (BufferedWriter fileListWriter = writer(withinContext, dataPath)) {
+ for (String file : files) {
+ fileListWriter.write(file + "\n");
+ }
+ }
+ }
+
+ LOG.info("Processing#{} INSERT message : {}", fromEventId(), event.getMessage());
+ DumpMetaData dmd = withinContext.createDmd(this);
+ dmd.setPayload(event.getMessage());
+ dmd.write();
+ }
+
+ private org.apache.hadoop.hive.ql.metadata.Table tableObject(
+ Context withinContext, InsertMessage insertMsg) throws TException {
+ return new org.apache.hadoop.hive.ql.metadata.Table(
+ withinContext.db.getMSC().getTable(
+ insertMsg.getDB(), insertMsg.getTable()
+ )
+ );
+ }
+
+ private BufferedWriter writer(Context withinContext, Path dataPath) throws IOException {
+ Path filesPath = new Path(dataPath, EximUtil.FILES_NAME);
+ FileSystem fs = dataPath.getFileSystem(withinContext.hiveConf);
+ return new BufferedWriter(new OutputStreamWriter(fs.create(filesPath)));
+ }
+
+ @Override
+ public DumpType dumpType() {
+ return DumpType.EVENT_INSERT;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/9d4f13af/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/AbstractHandler.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/AbstractHandler.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/AbstractHandler.java
deleted file mode 100644
index ab059c2..0000000
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/AbstractHandler.java
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.hadoop.hive.ql.parse.repl.events;
-
-import org.apache.hadoop.hive.metastore.api.NotificationEvent;
-import org.apache.hadoop.hive.metastore.messaging.MessageDeserializer;
-import org.apache.hadoop.hive.metastore.messaging.MessageFactory;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public abstract class AbstractHandler implements EventHandler {
- static final Logger LOG = LoggerFactory.getLogger(AbstractHandler.class);
-
- final NotificationEvent event;
- final MessageDeserializer deserializer;
-
- AbstractHandler(NotificationEvent event) {
- this.event = event;
- deserializer = MessageFactory.getInstance().getDeserializer();
- }
-
- @Override
- public long fromEventId() {
- return event.getEventId();
- }
-
- @Override
- public long toEventId() {
- return event.getEventId();
- }
-}
http://git-wip-us.apache.org/repos/asf/hive/blob/9d4f13af/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/AddPartitionHandler.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/AddPartitionHandler.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/AddPartitionHandler.java
deleted file mode 100644
index 1616ab9..0000000
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/AddPartitionHandler.java
+++ /dev/null
@@ -1,114 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.hadoop.hive.ql.parse.repl.events;
-
-import com.google.common.base.Function;
-import com.google.common.collect.Iterables;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hive.metastore.api.NotificationEvent;
-import org.apache.hadoop.hive.metastore.messaging.AddPartitionMessage;
-import org.apache.hadoop.hive.metastore.messaging.PartitionFiles;
-import org.apache.hadoop.hive.ql.metadata.HiveException;
-import org.apache.hadoop.hive.ql.metadata.Partition;
-import org.apache.hadoop.hive.ql.metadata.Table;
-import org.apache.hadoop.hive.ql.parse.EximUtil;
-
-import javax.annotation.Nullable;
-import java.io.BufferedWriter;
-import java.io.IOException;
-import java.io.OutputStreamWriter;
-import java.util.Iterator;
-
-import org.apache.hadoop.hive.ql.parse.repl.DumpType;
-
-public class AddPartitionHandler extends AbstractHandler {
- protected AddPartitionHandler(NotificationEvent notificationEvent) {
- super(notificationEvent);
- }
-
- @Override
- public void handle(Context withinContext) throws Exception {
- AddPartitionMessage apm = deserializer.getAddPartitionMessage(event.getMessage());
- LOG.info("Processing#{} ADD_PARTITION message : {}", fromEventId(), event.getMessage());
- Iterable<org.apache.hadoop.hive.metastore.api.Partition> ptns = apm.getPartitionObjs();
- if ((ptns == null) || (!ptns.iterator().hasNext())) {
- LOG.debug("Event#{} was an ADD_PTN_EVENT with no partitions");
- return;
- }
- org.apache.hadoop.hive.metastore.api.Table tobj = apm.getTableObj();
- if (tobj == null) {
- LOG.debug("Event#{} was a ADD_PTN_EVENT with no table listed");
- return;
- }
-
- final Table qlMdTable = new Table(tobj);
- Iterable<Partition> qlPtns = Iterables.transform(
- ptns,
- new Function<org.apache.hadoop.hive.metastore.api.Partition, Partition>() {
- @Nullable
- @Override
- public Partition apply(@Nullable org.apache.hadoop.hive.metastore.api.Partition input) {
- if (input == null) {
- return null;
- }
- try {
- return new Partition(qlMdTable, input);
- } catch (HiveException e) {
- throw new IllegalArgumentException(e);
- }
- }
- }
- );
-
- Path metaDataPath = new Path(withinContext.eventRoot, EximUtil.METADATA_NAME);
- EximUtil.createExportDump(
- metaDataPath.getFileSystem(withinContext.hiveConf),
- metaDataPath,
- qlMdTable,
- qlPtns,
- withinContext.replicationSpec);
-
- Iterator<PartitionFiles> partitionFilesIter = apm.getPartitionFilesIter().iterator();
- for (Partition qlPtn : qlPtns) {
- Iterable<String> files = partitionFilesIter.next().getFiles();
- if (files != null) {
- // encoded filename/checksum of files, write into _files
- try (BufferedWriter fileListWriter = writer(withinContext, qlPtn)) {
- for (String file : files) {
- fileListWriter.write(file + "\n");
- }
- }
- }
- }
- withinContext.createDmd(this).write();
- }
-
- private BufferedWriter writer(Context withinContext, Partition qlPtn)
- throws IOException {
- Path ptnDataPath = new Path(withinContext.eventRoot, qlPtn.getName());
- FileSystem fs = ptnDataPath.getFileSystem(withinContext.hiveConf);
- Path filesPath = new Path(ptnDataPath, EximUtil.FILES_NAME);
- return new BufferedWriter(new OutputStreamWriter(fs.create(filesPath)));
- }
-
- @Override
- public DumpType dumpType() {
- return DumpType.EVENT_ADD_PARTITION;
- }
-}
http://git-wip-us.apache.org/repos/asf/hive/blob/9d4f13af/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/AlterPartitionHandler.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/AlterPartitionHandler.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/AlterPartitionHandler.java
deleted file mode 100644
index b6c3496..0000000
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/AlterPartitionHandler.java
+++ /dev/null
@@ -1,112 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.hadoop.hive.ql.parse.repl.events;
-
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hive.metastore.api.NotificationEvent;
-import org.apache.hadoop.hive.metastore.messaging.AlterPartitionMessage;
-import org.apache.hadoop.hive.ql.metadata.Partition;
-import org.apache.hadoop.hive.ql.metadata.Table;
-import org.apache.hadoop.hive.ql.parse.EximUtil;
-
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.List;
-
-import org.apache.hadoop.hive.ql.parse.repl.DumpType;
-
-import org.apache.hadoop.hive.ql.parse.repl.load.DumpMetaData;
-
-public class AlterPartitionHandler extends AbstractHandler {
- private final org.apache.hadoop.hive.metastore.api.Partition after;
- private final org.apache.hadoop.hive.metastore.api.Table tableObject;
- private final boolean isTruncateOp;
- private final Scenario scenario;
-
- AlterPartitionHandler(NotificationEvent event) throws Exception {
- super(event);
- AlterPartitionMessage apm = deserializer.getAlterPartitionMessage(event.getMessage());
- tableObject = apm.getTableObj();
- org.apache.hadoop.hive.metastore.api.Partition before = apm.getPtnObjBefore();
- after = apm.getPtnObjAfter();
- isTruncateOp = apm.getIsTruncateOp();
- scenario = scenarioType(before, after);
- }
-
- private enum Scenario {
- ALTER {
- @Override
- DumpType dumpType() {
- return DumpType.EVENT_ALTER_PARTITION;
- }
- },
- RENAME {
- @Override
- DumpType dumpType() {
- return DumpType.EVENT_RENAME_PARTITION;
- }
- },
- TRUNCATE {
- @Override
- DumpType dumpType() {
- return DumpType.EVENT_TRUNCATE_PARTITION;
- }
- };
-
- abstract DumpType dumpType();
- }
-
- private Scenario scenarioType(org.apache.hadoop.hive.metastore.api.Partition before,
- org.apache.hadoop.hive.metastore.api.Partition after) {
- Iterator<String> beforeValIter = before.getValuesIterator();
- Iterator<String> afterValIter = after.getValuesIterator();
- while(beforeValIter.hasNext()) {
- if (!beforeValIter.next().equals(afterValIter.next())) {
- return Scenario.RENAME;
- }
- }
- return isTruncateOp ? Scenario.TRUNCATE : Scenario.ALTER;
- }
-
- @Override
- public void handle(Context withinContext) throws Exception {
- LOG.info("Processing#{} ALTER_PARTITION message : {}", fromEventId(), event.getMessage());
-
- if (Scenario.ALTER == scenario) {
- withinContext.replicationSpec.setIsMetadataOnly(true);
- Table qlMdTable = new Table(tableObject);
- List<Partition> partitions = new ArrayList<>();
- partitions.add(new Partition(qlMdTable, after));
- Path metaDataPath = new Path(withinContext.eventRoot, EximUtil.METADATA_NAME);
- EximUtil.createExportDump(
- metaDataPath.getFileSystem(withinContext.hiveConf),
- metaDataPath,
- qlMdTable,
- partitions,
- withinContext.replicationSpec);
- }
- DumpMetaData dmd = withinContext.createDmd(this);
- dmd.setPayload(event.getMessage());
- dmd.write();
- }
-
- @Override
- public DumpType dumpType() {
- return scenario.dumpType();
- }
-}
http://git-wip-us.apache.org/repos/asf/hive/blob/9d4f13af/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/AlterTableHandler.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/AlterTableHandler.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/AlterTableHandler.java
deleted file mode 100644
index d553240..0000000
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/AlterTableHandler.java
+++ /dev/null
@@ -1,102 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.hadoop.hive.ql.parse.repl.events;
-
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hive.metastore.api.NotificationEvent;
-import org.apache.hadoop.hive.metastore.messaging.AlterTableMessage;
-import org.apache.hadoop.hive.ql.metadata.Table;
-import org.apache.hadoop.hive.ql.parse.EximUtil;
-
-import org.apache.hadoop.hive.ql.parse.repl.DumpType;
-
-import org.apache.hadoop.hive.ql.parse.repl.load.DumpMetaData;
-
-public class AlterTableHandler extends AbstractHandler {
- private final org.apache.hadoop.hive.metastore.api.Table after;
- private final boolean isTruncateOp;
- private final Scenario scenario;
-
- private enum Scenario {
- ALTER {
- @Override
- DumpType dumpType() {
- return DumpType.EVENT_ALTER_TABLE;
- }
- },
- RENAME {
- @Override
- DumpType dumpType() {
- return DumpType.EVENT_RENAME_TABLE;
- }
- },
- TRUNCATE {
- @Override
- DumpType dumpType() {
- return DumpType.EVENT_TRUNCATE_TABLE;
- }
- };
-
- abstract DumpType dumpType();
- }
-
- AlterTableHandler(NotificationEvent event) throws Exception {
- super(event);
- AlterTableMessage atm = deserializer.getAlterTableMessage(event.getMessage());
- org.apache.hadoop.hive.metastore.api.Table before = atm.getTableObjBefore();
- after = atm.getTableObjAfter();
- isTruncateOp = atm.getIsTruncateOp();
- scenario = scenarioType(before, after);
- }
-
- private Scenario scenarioType(org.apache.hadoop.hive.metastore.api.Table before,
- org.apache.hadoop.hive.metastore.api.Table after) {
- if (before.getDbName().equals(after.getDbName())
- && before.getTableName().equals(after.getTableName())) {
- return isTruncateOp ? Scenario.TRUNCATE : Scenario.ALTER;
- } else {
- return Scenario.RENAME;
- }
- }
-
- @Override
- public void handle(Context withinContext) throws Exception {
- {
- LOG.info("Processing#{} ALTER_TABLE message : {}", fromEventId(), event.getMessage());
- if (Scenario.ALTER == scenario) {
- withinContext.replicationSpec.setIsMetadataOnly(true);
- Table qlMdTableAfter = new Table(after);
- Path metaDataPath = new Path(withinContext.eventRoot, EximUtil.METADATA_NAME);
- EximUtil.createExportDump(
- metaDataPath.getFileSystem(withinContext.hiveConf),
- metaDataPath,
- qlMdTableAfter,
- null,
- withinContext.replicationSpec);
- }
- DumpMetaData dmd = withinContext.createDmd(this);
- dmd.setPayload(event.getMessage());
- dmd.write();
- }
- }
-
- @Override
- public DumpType dumpType() {
- return scenario.dumpType();
- }
-}
http://git-wip-us.apache.org/repos/asf/hive/blob/9d4f13af/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/CreateTableHandler.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/CreateTableHandler.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/CreateTableHandler.java
deleted file mode 100644
index 88600fd..0000000
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/CreateTableHandler.java
+++ /dev/null
@@ -1,86 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.hadoop.hive.ql.parse.repl.events;
-
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hive.metastore.api.NotificationEvent;
-import org.apache.hadoop.hive.metastore.messaging.CreateTableMessage;
-import org.apache.hadoop.hive.ql.metadata.Table;
-import org.apache.hadoop.hive.ql.parse.EximUtil;
-
-import java.io.BufferedWriter;
-import java.io.IOException;
-import java.io.OutputStreamWriter;
-
-import org.apache.hadoop.hive.ql.parse.repl.DumpType;
-
-public class CreateTableHandler extends AbstractHandler {
-
- CreateTableHandler(NotificationEvent event) {
- super(event);
- }
-
- @Override
- public void handle(Context withinContext) throws Exception {
- CreateTableMessage ctm = deserializer.getCreateTableMessage(event.getMessage());
- LOG.info("Processing#{} CREATE_TABLE message : {}", fromEventId(), event.getMessage());
- org.apache.hadoop.hive.metastore.api.Table tobj = ctm.getTableObj();
-
- if (tobj == null) {
- LOG.debug("Event#{} was a CREATE_TABLE_EVENT with no table listed");
- return;
- }
-
- Table qlMdTable = new Table(tobj);
- if (qlMdTable.isView()) {
- withinContext.replicationSpec.setIsMetadataOnly(true);
- }
-
- Path metaDataPath = new Path(withinContext.eventRoot, EximUtil.METADATA_NAME);
- EximUtil.createExportDump(
- metaDataPath.getFileSystem(withinContext.hiveConf),
- metaDataPath,
- qlMdTable,
- null,
- withinContext.replicationSpec);
-
- Path dataPath = new Path(withinContext.eventRoot, "data");
- Iterable<String> files = ctm.getFiles();
- if (files != null) {
- // encoded filename/checksum of files, write into _files
- try (BufferedWriter fileListWriter = writer(withinContext, dataPath)) {
- for (String file : files) {
- fileListWriter.write(file + "\n");
- }
- }
- }
- withinContext.createDmd(this).write();
- }
-
- private BufferedWriter writer(Context withinContext, Path dataPath) throws IOException {
- FileSystem fs = dataPath.getFileSystem(withinContext.hiveConf);
- Path filesPath = new Path(dataPath, EximUtil.FILES_NAME);
- return new BufferedWriter(new OutputStreamWriter(fs.create(filesPath)));
- }
-
- @Override
- public DumpType dumpType() {
- return DumpType.EVENT_CREATE_TABLE;
- }
-}
http://git-wip-us.apache.org/repos/asf/hive/blob/9d4f13af/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/DefaultHandler.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/DefaultHandler.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/DefaultHandler.java
deleted file mode 100644
index 78cd74f..0000000
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/DefaultHandler.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.hadoop.hive.ql.parse.repl.events;
-
-import org.apache.hadoop.hive.metastore.api.NotificationEvent;
-
-import org.apache.hadoop.hive.ql.parse.repl.DumpType;
-
-import org.apache.hadoop.hive.ql.parse.repl.load.DumpMetaData;
-
-public class DefaultHandler extends AbstractHandler {
-
- DefaultHandler(NotificationEvent event) {
- super(event);
- }
-
- @Override
- public void handle(Context withinContext) throws Exception {
- LOG.info("Dummy processing#{} message : {}", fromEventId(), event.getMessage());
- DumpMetaData dmd = withinContext.createDmd(this);
- dmd.setPayload(event.getMessage());
- dmd.write();
- }
-
- @Override
- public DumpType dumpType() {
- return DumpType.EVENT_UNKNOWN;
- }
-}
http://git-wip-us.apache.org/repos/asf/hive/blob/9d4f13af/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/DropPartitionHandler.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/DropPartitionHandler.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/DropPartitionHandler.java
deleted file mode 100644
index c4a0908..0000000
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/DropPartitionHandler.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.hadoop.hive.ql.parse.repl.events;
-
-import org.apache.hadoop.hive.metastore.api.NotificationEvent;
-
-import org.apache.hadoop.hive.ql.parse.repl.DumpType;
-
-import org.apache.hadoop.hive.ql.parse.repl.load.DumpMetaData;
-
-public class DropPartitionHandler extends AbstractHandler {
-
- DropPartitionHandler(NotificationEvent event) {
- super(event);
- }
-
- @Override
- public void handle(Context withinContext) throws Exception {
- LOG.info("Processing#{} DROP_PARTITION message : {}", fromEventId(), event.getMessage());
- DumpMetaData dmd = withinContext.createDmd(this);
- dmd.setPayload(event.getMessage());
- dmd.write();
- }
-
- @Override
- public DumpType dumpType() {
- return DumpType.EVENT_DROP_PARTITION;
- }
-}
http://git-wip-us.apache.org/repos/asf/hive/blob/9d4f13af/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/DropTableHandler.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/DropTableHandler.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/DropTableHandler.java
deleted file mode 100644
index e3addaf..0000000
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/events/DropTableHandler.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.hadoop.hive.ql.parse.repl.events;
-
-import org.apache.hadoop.hive.metastore.api.NotificationEvent;
-
-import org.apache.hadoop.hive.ql.parse.repl.DumpType;
-
-import org.apache.hadoop.hive.ql.parse.repl.load.DumpMetaData;
-
-public class DropTableHandler extends AbstractHandler {
-
- DropTableHandler(NotificationEvent event) {
- super(event);
- }
-
- @Override
- public void handle(Context withinContext) throws Exception {
- LOG.info("Processing#{} DROP_TABLE message : {}", fromEventId(), event.getMessage());
- DumpMetaData dmd = withinContext.createDmd(this);
- dmd.setPayload(event.getMessage());
- dmd.write();
- }
-
- @Override
- public DumpType dumpType() {
- return DumpType.EVENT_DROP_TABLE;
- }
-}