You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by jo...@apache.org on 2019/02/16 01:20:19 UTC
[impala] 04/04: IMPALA-7976 : Add a flag to disable sync using
events at a table level
This is an automated email from the ASF dual-hosted git repository.
joemcdonnell pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit f4dd98236a45fc0cc2b11a6db7c91a6f04c183d3
Author: Vihang Karajgaonkar <vi...@cloudera.com>
AuthorDate: Tue Jan 29 13:16:43 2019 -0800
IMPALA-7976 : Add a flag to disable sync using events at a table level
This change adds the ability to disable event processing at a table or
database level. If the user adds a property with the key
"impala.disableHmsSync" in the table or database properties, event
processing can be turned on/off at the respective level. The value of
the property is string value of Booleans ("true" or "false"). By default
event processing is enabled when the global config
hms_event_polling_interval_s is set to a non-zero value. The value of
property determines if the event processing needs to be disabled for a
particular table or database. So for example, if
"impala.disableHmsSync" is set to "true" then events for that
table/database are ignored. If the property is unset or a value of
"false" is used, the event processing is enabled (although it is not
necessary to explicitly enable event processing, given the global flag).
When both table and db level properties are set, the table level
property takes precedence. If the table level property is not set, then
database level property is used to evaluate if the event needs to be
processed or not.
Users can issue alter table command to change the value of this
property. When the flag value is changed, the alter table event
generated is always processed and subsequent events are either ignored
or processed based on the new value of table flag and database flag. In
theory, a database level property can also be changed. But Impala
doesn't currently support alter database operations. Users who may wish
to change the database level property should use alter database command
from Apache Hive (or any other metastore client)
When the flag is turned OFF and the previous value of ON or unset, it is
possible that the table does not exist in catalog anymore. For example,
when the table was created the event sync was disabled. Hence the create
event was skipped. Now when the user re-enables the sync again on such
table, event processor has no idea how many events were skipped during
this interval. Event processor errs on the side of caution and changes
the state to NEEDS_INVALIDATE in such case.
Testing done: Added new unit tests which execute all supported DDL
operations with all possible combination of db and table flags. Another
test executes alter table operations for changing flags for all combinations
of transitions at table level.
Change-Id: If116edba51bae139bc003be858b3214c3f0104e3
Reviewed-on: http://gerrit.cloudera.org:8080/12365
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../impala/catalog/CatalogServiceCatalog.java | 45 ++
...tastoreEventUtils.java => MetastoreEvents.java} | 482 +++++++++++++----
.../catalog/events/MetastoreEventsProcessor.java | 61 ++-
...astoreNotificationNeedsInvalidateException.java | 36 ++
.../events/MetastoreEventsProcessorTest.java | 597 ++++++++++++++++++++-
5 files changed, 1060 insertions(+), 161 deletions(-)
diff --git a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
index 0904714..55a854a 100644
--- a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
+++ b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
@@ -821,6 +821,51 @@ public class CatalogServiceCatalog extends Catalog {
}
/**
+ * Given a database name and a property key returns the value of the key from the
+ * parameters map of the HMS db object
+ * @param dbName name of the database
+ * @param propertyKey property key
+ * @return value of key from the db parameter. returns null if Db is not found or key
+ * does not exist in the parameters
+ */
+ public String getDbProperty(String dbName, String propertyKey) {
+ Preconditions.checkNotNull(dbName);
+ Preconditions.checkNotNull(propertyKey);
+ versionLock_.readLock().lock();
+ try {
+ Db db = getDb(dbName);
+ if (db == null) return null;
+ if (!db.getMetaStoreDb().isSetParameters()) return null;
+ return db.getMetaStoreDb().getParameters().get(propertyKey);
+ } finally {
+ versionLock_.readLock().unlock();
+ }
+ }
+
+ /**
+ * Given a dbname, table name and a key returns the value of the key from the cached
+ * Table object's parameters
+ * @return Value of the parameter which maps to property key, null if the table
+ * doesn't exist, if it is a incomplete table or if the parameter is not found
+ */
+ public String getTableProperty(String dbName, String tblName, String propertyKey) {
+ Preconditions.checkNotNull(dbName);
+ Preconditions.checkNotNull(tblName);
+ Preconditions.checkNotNull(propertyKey);
+ versionLock_.readLock().lock();
+ try {
+ Db db = getDb(dbName);
+ if (db == null) return null;
+ Table tbl = db.getTable(tblName);
+ if (tbl == null || tbl instanceof IncompleteTable) return null;
+ if (!tbl.getMetaStoreTable().isSetParameters()) return null;
+ return tbl.getMetaStoreTable().getParameters().get(propertyKey);
+ } finally {
+ versionLock_.readLock().unlock();
+ }
+ }
+
+ /**
* Adds a table in the topic update if its version is in the range
* ('ctx.fromVersion', 'ctx.toVersion']. If the table's version is larger than
* 'ctx.toVersion' and the table has skipped a topic update
diff --git a/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEventUtils.java b/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEvents.java
similarity index 58%
rename from fe/src/main/java/org/apache/impala/catalog/events/MetastoreEventUtils.java
rename to fe/src/main/java/org/apache/impala/catalog/events/MetastoreEvents.java
index aba698e..b69a74b 100644
--- a/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEventUtils.java
+++ b/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEvents.java
@@ -17,12 +17,18 @@
package org.apache.impala.catalog.events;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.Iterator;
import java.util.List;
+import java.util.Objects;
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.hadoop.hive.metastore.api.NotificationEvent;
+import org.apache.hadoop.hive.metastore.messaging.AddPartitionMessage;
+import org.apache.hadoop.hive.metastore.messaging.AlterPartitionMessage;
+import org.apache.hadoop.hive.metastore.messaging.CreateTableMessage;
import org.apache.hadoop.hive.metastore.messaging.json.JSONAlterTableMessage;
import org.apache.hadoop.hive.metastore.messaging.json.JSONCreateDatabaseMessage;
import org.apache.hadoop.hive.metastore.messaging.json.JSONDropTableMessage;
@@ -34,14 +40,18 @@ import org.apache.impala.catalog.Table;
import org.apache.impala.common.Pair;
import org.apache.impala.common.Reference;
import org.apache.impala.thrift.TTableName;
-import org.apache.log4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.Logger;
/**
- * Util class which provides Metastore event objects for various event types. Also
+ * Main class which provides Metastore event objects for various event types. Also
* provides a MetastoreEventFactory to get or create the event instances for a given event
* type
*/
-public class MetastoreEventUtils {
+public class MetastoreEvents {
+
+ // flag to be set in the table/database parameters to disable event based metadata sync
+ public static final String DISABLE_EVENT_HMS_SYNC_KEY = "impala.disableHmsSync";
public enum MetastoreEventType {
CREATE_TABLE("CREATE_TABLE"),
@@ -87,7 +97,9 @@ public class MetastoreEventUtils {
* Factory class to create various MetastoreEvents.
*/
public static class MetastoreEventFactory {
- private static final Logger LOG = Logger.getLogger(MetastoreEventFactory.class);
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(MetastoreEventFactory.class);
// catalog service instance to be used for creating eventHandlers
private final CatalogServiceCatalog catalog_;
@@ -121,13 +133,13 @@ public class MetastoreEventUtils {
return new IgnoredEvent(catalog_, event);
case ADD_PARTITION:
// add partition events triggers invalidate table currently
- return new TableInvalidatingEvent(catalog_, event);
+ return new AddPartitionEvent(catalog_, event);
case DROP_PARTITION:
// drop partition events triggers invalidate table currently
- return new TableInvalidatingEvent(catalog_, event);
+ return new DropPartitionEvent(catalog_, event);
case ALTER_PARTITION:
// alter partition events triggers invalidate table currently
- return new TableInvalidatingEvent(catalog_, event);
+ return new AlterPartitionEvent(catalog_, event);
default:
// ignore all the unknown events by creating a IgnoredEvent
return new IgnoredEvent(catalog_, event);
@@ -154,6 +166,8 @@ public class MetastoreEventUtils {
List<MetastoreEvent> getFilteredEvents(List<NotificationEvent> events)
throws MetastoreNotificationException {
Preconditions.checkNotNull(events);
+ if (events.isEmpty()) return Collections.emptyList();
+
List<MetastoreEvent> metastoreEvents = new ArrayList<>(events.size());
for (NotificationEvent event : events) {
metastoreEvents.add(get(event));
@@ -188,6 +202,12 @@ public class MetastoreEventUtils {
*/
public static abstract class MetastoreEvent {
+ // String.format compatible string to prepend event id and type
+ private static final String STR_FORMAT_EVENT_ID_TYPE = "EventId: %d EventType: %s ";
+
+ // logger format compatible string to prepend to a log formatted message
+ private static final String LOG_FORMAT_EVENT_ID_TYPE = "EventId: {} EventType: {} ";
+
// CatalogServiceCatalog instance on which the event needs to be acted upon
protected final CatalogServiceCatalog catalog_;
@@ -195,7 +215,7 @@ public class MetastoreEventUtils {
protected final NotificationEvent event_;
// Logger available for all the sub-classes
- protected final Logger LOG = Logger.getLogger(this.getClass());
+ protected final Logger LOG = LoggerFactory.getLogger(this.getClass());
// dbName from the event
protected final String dbName_;
@@ -213,25 +233,34 @@ public class MetastoreEventUtils {
this.event_ = event;
this.eventId_ = event_.getEventId();
this.eventType_ = MetastoreEventType.from(event.getEventType());
- LOG.debug(String
- .format("Creating event %d of type %s on table %s", event.getEventId(),
- event.getEventType(), event.getTableName()));
dbName_ = Preconditions.checkNotNull(event.getDbName());
metastoreNotificationEvent_ = event;
}
/**
+ * Process this event if it is enabled based on the flags on this object
+ * @throws CatalogException If Catalog operations fail
+ * @throws MetastoreNotificationException If NotificationEvent parsing fails
+ */
+ public void processIfEnabled()
+ throws CatalogException, MetastoreNotificationException {
+ if (isEventProcessingDisabled()) return;
+ process();
+ }
+
+ /**
* Process the information available in the NotificationEvent to take appropriate
* action on Catalog
*
* @throws MetastoreNotificationException in case of event parsing errors out
* @throws CatalogException in case catalog operations could not be performed
*/
- abstract void process() throws MetastoreNotificationException, CatalogException;
+ protected abstract void process() throws MetastoreNotificationException,
+ CatalogException;
/**
* Helper method to get debug string with helpful event information prepended to the
- * message
+ * message. This can be used to generate helpful exception messages
*
* @param msgFormatString String value to be used in String.format() for the given
* message
@@ -240,17 +269,55 @@ public class MetastoreEventUtils {
*/
protected String debugString(String msgFormatString, Object... args) {
String formatString =
- new StringBuilder("EventId: %d EventType: %s ").append(msgFormatString)
+ new StringBuilder(STR_FORMAT_EVENT_ID_TYPE).append(msgFormatString)
.toString();
+ Object[] formatArgs = getLogFormatArgs(args);
+ return String.format(formatString, formatArgs);
+ }
+
+ /**
+ * Helper method to generate the format args after prepending the event id and type
+ */
+ private Object[] getLogFormatArgs(Object[] args) {
Object[] formatArgs = new Object[args.length + 2];
formatArgs[0] = eventId_;
formatArgs[1] = eventType_;
- int i=2;
+ int i = 2;
for (Object arg : args) {
formatArgs[i] = arg;
i++;
}
- return String.format(formatString, formatArgs);
+ return formatArgs;
+ }
+
+ /**
+ * Logs at info level the given log formatted string and its args. The log
+ * formatted string should have {} pair at the appropriate location in the string
+ * for each arg value provided. This method prepends the event id and event type
+ * before logging the message. No-op if the log level is not at INFO
+
+ * @param logFormattedStr
+ * @param args
+ */
+ protected void infoLog(String logFormattedStr, Object... args) {
+ if (!LOG.isInfoEnabled()) return;
+ String formatString =
+ new StringBuilder(LOG_FORMAT_EVENT_ID_TYPE).append(logFormattedStr)
+ .toString();
+ Object[] formatArgs = getLogFormatArgs(args);
+ LOG.info(formatString, formatArgs);
+ }
+
+ /**
+ * Similar to infoLog excepts logs at debug level
+ */
+ protected void debugLog(String logFormattedStr, Object... args) {
+ if (!LOG.isDebugEnabled()) return;
+ String formatString =
+ new StringBuilder(LOG_FORMAT_EVENT_ID_TYPE).append(logFormattedStr)
+ .toString();
+ Object[] formatArgs = getLogFormatArgs(args);
+ LOG.debug(formatString, formatArgs);
}
/**
@@ -264,6 +331,12 @@ public class MetastoreEventUtils {
protected boolean isRemovedAfter(List<MetastoreEvent> events) {
return false;
}
+
+ /**
+ * Returns true if event based sync is disabled for this table/database associated
+ * with this event
+ */
+ protected abstract boolean isEventProcessingDisabled();
}
/**
@@ -274,10 +347,16 @@ public class MetastoreEventUtils {
// tblName from the event
protected final String tblName_;
+ // tbl object from the Notification event, corresponds to the before tableObj in
+ // case of alter events
+ protected org.apache.hadoop.hive.metastore.api.Table msTbl_;
+
private MetastoreTableEvent(CatalogServiceCatalog catalogServiceCatalog,
NotificationEvent event) {
super(catalogServiceCatalog, event);
tblName_ = Preconditions.checkNotNull(event.getTableName());
+ debugLog("Creating event {} of type {} on table {}", eventId_, eventType_,
+ getFullyQualifiedTblName());
}
@@ -297,6 +376,53 @@ public class MetastoreEventUtils {
protected boolean invalidateCatalogTable() {
return catalog_.invalidateTableIfExists(dbName_, tblName_) != null;
}
+
+ /**
+ * Checks if the table level property is set in the parameters of the table from
+ * the event. If it is available, it takes precedence over database level flag for
+ * this table. If the table level property is not set, returns the value from the
+ * database level property.f
+ *
+ * @return Boolean value of the table property with the key
+ * <code>DISABLE_EVENT_HMS_SYNC_KEY</code>. Else, returns the database property
+ * which is associated with this table. Returns false if neither of the properties
+ * are set.
+ */
+ @Override
+ protected boolean isEventProcessingDisabled() {
+ Preconditions.checkNotNull(msTbl_);
+ Boolean tblProperty = getHmsSyncProperty(msTbl_);
+ if (tblProperty != null) {
+ debugLog("Found table level flag {} is set to {}",
+ DISABLE_EVENT_HMS_SYNC_KEY, tblProperty.toString());
+ return tblProperty;
+ }
+ // if the tbl property is not set check at db level
+ String dbFlagVal = catalog_.getDbProperty(dbName_, DISABLE_EVENT_HMS_SYNC_KEY);
+ if (dbFlagVal != null) {
+ // no need to spew unnecessary logs. Most tables/databases are expected to not
+ // have this flag set when event based HMS polling is enabled
+ debugLog("Table level flag is not set. Db level flag {} at Db level is {}",
+ DISABLE_EVENT_HMS_SYNC_KEY, dbFlagVal);
+ }
+ // flag value of null also returns false
+ return Boolean.valueOf(dbFlagVal);
+ }
+
+ /**
+ * Gets the value of the parameter with the key
+ * <code>DISABLE_EVENT_HMS_SYNC_KEY</code> from the given table
+ *
+ * @return the Boolean value of the property with the key
+ * <code>DISABLE_EVENT_HMS_SYNC_KEY</code> if it is available else returns null
+ */
+ public static Boolean getHmsSyncProperty(
+ org.apache.hadoop.hive.metastore.api.Table tbl) {
+ if (!tbl.isSetParameters()) return null;
+ String val = tbl.getParameters().get(DISABLE_EVENT_HMS_SYNC_KEY);
+ if (val == null || val.isEmpty()) return null;
+ return Boolean.valueOf(val);
+ }
}
/**
@@ -307,20 +433,45 @@ public class MetastoreEventUtils {
MetastoreDatabaseEvent(CatalogServiceCatalog catalogServiceCatalog,
NotificationEvent event) {
super(catalogServiceCatalog, event);
+ debugLog("Creating event {} of type {} on database {}", eventId_,
+ eventType_, dbName_);
+ }
+
+ /**
+ * Even though there is a database level property
+ * <code>DISABLE_EVENT_HMS_SYNC_KEY</code> it is only used for tables within that
+ * database. As such this property does not control if the database level DDLs are
+ * skipped or not.
+ * @return false
+ */
+ @Override
+ protected boolean isEventProcessingDisabled() {
+ return false;
}
}
/**
* MetastoreEvent for CREATE_TABLE event type
*/
- private static class CreateTableEvent extends MetastoreTableEvent {
-
+ public static class CreateTableEvent extends MetastoreTableEvent {
/**
* Prevent instantiation from outside should use MetastoreEventFactory instead
*/
- private CreateTableEvent(CatalogServiceCatalog catalog, NotificationEvent event) {
+ private CreateTableEvent(CatalogServiceCatalog catalog, NotificationEvent event)
+ throws MetastoreNotificationException {
super(catalog, event);
Preconditions.checkArgument(MetastoreEventType.CREATE_TABLE.equals(eventType_));
+ Preconditions
+ .checkNotNull(event.getMessage(), debugString("Event message is null"));
+ CreateTableMessage createTableMessage =
+ MetastoreEventsProcessor.getMessageFactory().getDeserializer()
+ .getCreateTableMessage(event.getMessage());
+ try {
+ msTbl_ = createTableMessage.getTableObj();
+ } catch (Exception e) {
+ throw new MetastoreNotificationException(
+ debugString("Unable to deserialize the event message"), e);
+ }
}
/**
@@ -336,9 +487,12 @@ public class MetastoreEventUtils {
// table being dropped and recreated with the same name or in case this event is
// a self-event (see description of self-event in the class documentation of
// MetastoreEventsProcessor)
- boolean tableAdded;
try {
- tableAdded = catalog_.addTableIfNotExists(dbName_, tblName_);
+ if (!catalog_.addTableIfNotExists(dbName_, tblName_)) {
+ debugLog("Not adding the table {} since it already exists in catalog",
+ tblName_);
+ return;
+ }
} catch (CatalogException e) {
// if a exception is thrown, it could be due to the fact that the db did not
// exist in the catalog cache. This could only happen if the previous
@@ -349,13 +503,7 @@ public class MetastoreEventUtils {
+ "processing CREATE_DATABASE event for the database %s",
getFullyQualifiedTblName(), dbName_), e);
}
- if (!tableAdded) {
- LOG.debug(
- debugString("Not adding the table %s since it already exists in catalog",
- tblName_));
- return;
- }
- LOG.info(debugString("Added a table %s", getFullyQualifiedTblName()));
+ debugLog("Added a table {}", getFullyQualifiedTblName());
}
@Override
@@ -366,8 +514,8 @@ public class MetastoreEventUtils {
DropTableEvent dropTableEvent = (DropTableEvent) event;
if (dbName_.equalsIgnoreCase(dropTableEvent.dbName_) && tblName_
.equalsIgnoreCase(dropTableEvent.tblName_)) {
- LOG.info(debugString("Found table %s is removed later in event %d type %s",
- tblName_, dropTableEvent.eventId_, dropTableEvent.eventType_));
+ infoLog("Found table {} is removed later in event {} type {}",
+ tblName_, dropTableEvent.eventId_, dropTableEvent.eventType_);
return true;
}
} else if (event.eventType_.equals(MetastoreEventType.ALTER_TABLE)) {
@@ -380,11 +528,11 @@ public class MetastoreEventUtils {
// the create table event, the alter rename event just becomes a addIfNotExists
// event which is valid for both a self-event and external event cases
AlterTableEvent alterTableEvent = (AlterTableEvent) event;
- if (alterTableEvent.isRename_ && dbName_
- .equalsIgnoreCase(alterTableEvent.tableBefore_.getDbName()) && tblName_
- .equalsIgnoreCase(alterTableEvent.tableBefore_.getTableName())) {
- LOG.info(debugString("Found table %s is renamed later in event %d type %s",
- tblName_, alterTableEvent.eventId_, alterTableEvent.eventType_));
+ if (alterTableEvent.isRename_ &&
+ dbName_.equalsIgnoreCase(alterTableEvent.msTbl_.getDbName()) &&
+ tblName_.equalsIgnoreCase(alterTableEvent.msTbl_.getTableName())) {
+ infoLog("Found table {} is renamed later in event {} type {}",
+ tblName_, alterTableEvent.eventId_, alterTableEvent.eventType_);
return true;
}
}
@@ -396,19 +544,24 @@ public class MetastoreEventUtils {
/**
* MetastoreEvent for ALTER_TABLE event type
*/
- private static class AlterTableEvent extends MetastoreTableEvent {
+ public static class AlterTableEvent extends MetastoreTableEvent {
- // the table object before alter operation, as parsed from the NotificationEvent
- private final org.apache.hadoop.hive.metastore.api.Table tableBefore_;
// the table object after alter operation, as parsed from the NotificationEvent
- private final org.apache.hadoop.hive.metastore.api.Table tableAfter_;
+ protected org.apache.hadoop.hive.metastore.api.Table tableAfter_;
// true if this alter event was due to a rename operation
private final boolean isRename_;
+ // value of event sync flag for this table before the alter operation
+ private final Boolean eventSyncBeforeFlag_;
+ // value of the event sync flag if available at this table after the alter operation
+ private final Boolean eventSyncAfterFlag_;
+ // value of the db flag at the time of event creation
+ private final boolean dbFlagVal;
/**
* Prevent instantiation from outside should use MetastoreEventFactory instead
*/
- private AlterTableEvent(CatalogServiceCatalog catalog, NotificationEvent event)
+ @VisibleForTesting
+ AlterTableEvent(CatalogServiceCatalog catalog, NotificationEvent event)
throws MetastoreNotificationException {
super(catalog, event);
Preconditions.checkArgument(MetastoreEventType.ALTER_TABLE.equals(eventType_));
@@ -416,7 +569,7 @@ public class MetastoreEventUtils {
(JSONAlterTableMessage) MetastoreEventsProcessor.getMessageFactory()
.getDeserializer().getAlterTableMessage(event.getMessage());
try {
- tableBefore_ = Preconditions.checkNotNull(alterTableMessage.getTableObjBefore());
+ msTbl_ = Preconditions.checkNotNull(alterTableMessage.getTableObjBefore());
tableAfter_ = Preconditions.checkNotNull(alterTableMessage.getTableObjAfter());
} catch (Exception e) {
throw new MetastoreNotificationException(
@@ -424,8 +577,13 @@ public class MetastoreEventUtils {
}
// this is a rename event if either dbName or tblName of before and after object
// changed
- isRename_ = !tableBefore_.getDbName().equalsIgnoreCase(tableAfter_.getDbName())
- || !tableBefore_.getTableName().equalsIgnoreCase(tableAfter_.getTableName());
+ isRename_ = !msTbl_.getDbName().equalsIgnoreCase(tableAfter_.getDbName())
+ || !msTbl_.getTableName().equalsIgnoreCase(tableAfter_.getTableName());
+ eventSyncBeforeFlag_ =
+ getHmsSyncProperty(msTbl_);
+ eventSyncAfterFlag_ = getHmsSyncProperty(tableAfter_);
+ dbFlagVal =
+ Boolean.valueOf(catalog_.getDbProperty(dbName_, DISABLE_EVENT_HMS_SYNC_KEY));
}
/**
@@ -435,66 +593,108 @@ public class MetastoreEventUtils {
* in-place
*/
@Override
- public void process() throws MetastoreNotificationException {
+ public void process() throws MetastoreNotificationException, CatalogException {
// in case of table level alters from external systems it is better to do a full
// invalidate eg. this could be due to as simple as adding a new parameter or a
// full blown adding or changing column type
// detect the special where a table is renamed
- try {
- if (!isRename_) {
- // table is not renamed, need to invalidate
- if (!invalidateCatalogTable()) {
- LOG.debug(debugString("Table %s does not need to be "
- + "invalidated since it does not exist anymore",
- getFullyQualifiedTblName()));
- } else {
- LOG.info(debugString("Table %s is invalidated", getFullyQualifiedTblName()));
+ if (!isRename_) {
+ // table is not renamed, need to invalidate
+ if (!invalidateCatalogTable()) {
+ if (wasEventSyncTurnedOn()) {
+ // we received this alter table event on a non-existing table. We also
+ // detect that event sync was turned on in this event. This may mean that
+ // the table creation was skipped earlier because event sync was turned off
+ // we don't really know how many of events we have skipped till now because
+ // the sync was disabled all this while before we receive such a event. We
+ // error on the side of caution by stopping the event processing and
+ // letting the user to issue a invalidate metadata to reset the state
+ throw new MetastoreNotificationNeedsInvalidateException(debugString(
+ "Detected that event sync was turned on for the table %s "
+ + "and the table does not exist. Event processing cannot be "
+ + "continued further. Issue a invalidate metadata command to reset "
+ + "the event processing state", getFullyQualifiedTblName()));
}
- return;
- }
- // table was renamed, remove the old table
- LOG.info(debugString("Found that %s table was renamed. Renaming it by "
- + "remove and adding a new table", new TableName(tableBefore_.getDbName(),
- tableBefore_.getTableName())));
- TTableName oldTTableName =
- new TTableName(tableBefore_.getDbName(), tableBefore_.getTableName());
- TTableName newTTableName =
- new TTableName(tableAfter_.getDbName(), tableAfter_.getTableName());
-
- // atomically rename the old table to new table
- Pair<Boolean, Boolean> result =
- catalog_.renameOrAddTableIfNotExists(oldTTableName, newTTableName);
-
- // old table was not found. This could be because catalogD is stale and didn't
- // have any entry for the oldTable
- if (!result.first) {
- LOG.debug(debugString("Did not remove old table to rename table %s to %s since "
- + "it does not exist anymore", qualify(oldTTableName),
- qualify(newTTableName)));
+ debugLog("Table {} does not need to be "
+ + "invalidated since it does not exist anymore",
+ getFullyQualifiedTblName());
+ } else {
+ infoLog("Table {} is invalidated", getFullyQualifiedTblName());
}
- // the new table from the event was not added since it was already present
- if (!result.second) {
- LOG.debug(
- debugString("Did not add new table name while renaming table %s to %s",
- qualify(oldTTableName), qualify(newTTableName)));
- }
- } catch (Exception e) {
- throw new MetastoreNotificationException(e);
+ return;
+ }
+ // table was renamed, remove the old table
+ infoLog("Found that {} table was renamed. Renaming it by "
+ + "remove and adding a new table",
+ new TableName(msTbl_.getDbName(), msTbl_.getTableName()));
+ TTableName oldTTableName =
+ new TTableName(msTbl_.getDbName(), msTbl_.getTableName());
+ TTableName newTTableName =
+ new TTableName(tableAfter_.getDbName(), tableAfter_.getTableName());
+
+ // atomically rename the old table to new table
+ Pair<Boolean, Boolean> result = null;
+ result = catalog_.renameOrAddTableIfNotExists(oldTTableName, newTTableName);
+
+ // old table was not found. This could be because catalogD is stale and didn't
+ // have any entry for the oldTable
+ if (!result.first) {
+ debugLog("Did not remove old table to rename table {} to {} since "
+ + "it does not exist anymore", qualify(oldTTableName),
+ qualify(newTTableName));
}
+ // the new table from the event was not added since it was already present
+ if (!result.second) {
+ debugLog("Did not add new table name while renaming table {} to {}",
+ qualify(oldTTableName), qualify(newTTableName));
+ }
+ }
+
+ /**
+ * Detects a event sync flag was turned on in this event
+ */
+ private boolean wasEventSyncTurnedOn() {
+ // the eventsync flag was not changed
+ if (Objects.equals(eventSyncBeforeFlag_, eventSyncAfterFlag_)) return false;
+ // eventSync after flag is null or if it is explicitly set to false
+ if ((eventSyncAfterFlag_ == null && !dbFlagVal) || !eventSyncAfterFlag_) {
+ return true;
+ }
+ return false;
}
private String qualify(TTableName tTableName) {
return new TableName(tTableName.db_name, tTableName.table_name).toString();
}
+
+ /**
+ * In case of alter table events, it is possible that the alter event is generated
+ * because user changed the value of the parameter
+ * <code>DISABLE_EVENT_HMS_SYNC_KEY</code>. If the parameter is unchanged, it doesn't
+ * matter if you use the before or after table object here since the eventual
+ * action is going be invalidate or rename. If however, the parameter is changed,
+ * couple of things could happen. The flag changes from unset/false to true or it
+ * changes from true to false/unset. In the first case, we want to process the
+ * event (and ignore subsequent events on this table). In the second case, we
+ * should process the event (as well as all the subsequent events on the table). So
+ * we always process this event when the value of the flag is changed.
+ *
+ * @return true, if this event needs to be skipped. false if this event needs to be
+ * processed.
+ */
+ @Override
+ protected boolean isEventProcessingDisabled() {
+ // if the event sync flag was changed then we always process this event
+ if (!Objects.equals(eventSyncBeforeFlag_, eventSyncAfterFlag_)) return false;
+ // flag is unchanged, use the default impl from base class
+ return super.isEventProcessingDisabled();
+ }
}
/**
* MetastoreEvent for the DROP_TABLE event type
*/
- private static class DropTableEvent extends MetastoreTableEvent {
-
- // the metastore table object as parsed from the drop table event
- private final org.apache.hadoop.hive.metastore.api.Table droppedTable_;
+ public static class DropTableEvent extends MetastoreTableEvent {
/**
* Prevent instantiation from outside should use MetastoreEventFactory instead
@@ -507,7 +707,7 @@ public class MetastoreEventUtils {
(JSONDropTableMessage) MetastoreEventsProcessor.getMessageFactory()
.getDeserializer().getDropTableMessage(event.getMessage());
try {
- droppedTable_ = Preconditions.checkNotNull(dropTableMessage.getTableObj());
+ msTbl_ = Preconditions.checkNotNull(dropTableMessage.getTableObj());
} catch (Exception e) {
throw new MetastoreNotificationException(debugString(
"Could not parse event message. "
@@ -528,16 +728,15 @@ public class MetastoreEventUtils {
Reference<Boolean> tblWasFound = new Reference<>();
Reference<Boolean> tblMatched = new Reference<>();
Table removedTable =
- catalog_.removeTableIfExists(droppedTable_, tblWasFound, tblMatched);
+ catalog_.removeTableIfExists(msTbl_, tblWasFound, tblMatched);
if (removedTable != null) {
- LOG.info(debugString("Removed table %s ", getFullyQualifiedTblName()));
+ infoLog("Removed table {} ", getFullyQualifiedTblName());
} else if (!tblMatched.getRef()) {
LOG.warn(debugString("Table %s was not removed from "
+ "catalog since the creation time of the table did not match", tblName_));
} else if (!tblWasFound.getRef()) {
- LOG.debug(
- debugString("Table %s was not removed since it did not exist in catalog.",
- tblName_));
+ debugLog("Table {} was not removed since it did not exist in catalog.",
+ tblName_);
}
}
}
@@ -545,7 +744,7 @@ public class MetastoreEventUtils {
/**
* MetastoreEvent for CREATE_DATABASE event type
*/
- private static class CreateDatabaseEvent extends MetastoreDatabaseEvent {
+ public static class CreateDatabaseEvent extends MetastoreDatabaseEvent {
// metastore database object as parsed from NotificationEvent message
private final Database createdDatabase_;
@@ -587,9 +786,9 @@ public class MetastoreEventUtils {
// already existing database in catalog is a later version with the same name and
// this event can be ignored
if (catalog_.addDbIfNotExists(dbName_, createdDatabase_)) {
- LOG.info(debugString("Successfully added database %s", dbName_));
+ infoLog("Successfully added database {}", dbName_);
} else {
- LOG.info(debugString("Database %s already exists", dbName_));
+ infoLog("Database {} already exists", dbName_);
}
}
@@ -600,9 +799,9 @@ public class MetastoreEventUtils {
if (event.eventType_.equals(MetastoreEventType.DROP_DATABASE)) {
DropDatabaseEvent dropDatabaseEvent = (DropDatabaseEvent) event;
if (dbName_.equalsIgnoreCase(dropDatabaseEvent.dbName_)) {
- LOG.info(debugString(
- "Found database %s is removed later in event %d of " + "type %s ",
- dbName_, dropDatabaseEvent.eventId_, dropDatabaseEvent.eventType_));
+ infoLog(
+ "Found database {} is removed later in event {} of type {} ",
+ dbName_, dropDatabaseEvent.eventId_, dropDatabaseEvent.eventType_);
return true;
}
}
@@ -614,7 +813,7 @@ public class MetastoreEventUtils {
/**
* MetastoreEvent for the DROP_DATABASE event
*/
- private static class DropDatabaseEvent extends MetastoreDatabaseEvent {
+ public static class DropDatabaseEvent extends MetastoreDatabaseEvent {
/**
* Prevent instantiation from outside should use MetastoreEventFactory instead
@@ -641,7 +840,7 @@ public class MetastoreEventUtils {
Db removedDb = catalog_.removeDb(dbName_);
// if database did not exist in the cache there was nothing to do
if (removedDb != null) {
- LOG.info(debugString("Successfully removed database %s", dbName_));
+ infoLog("Successfully removed database {}", dbName_);
}
}
}
@@ -649,7 +848,7 @@ public class MetastoreEventUtils {
/**
* MetastoreEvent for which issues invalidate on a table from the event
*/
- private static class TableInvalidatingEvent extends MetastoreTableEvent {
+ public static abstract class TableInvalidatingEvent extends MetastoreTableEvent {
/**
* Prevent instantiation from outside should use MetastoreEventFactory instead
@@ -669,18 +868,84 @@ public class MetastoreEventUtils {
@Override
public void process() {
if (invalidateCatalogTable()) {
- LOG.info(debugString("Table %s is invalidated", getFullyQualifiedTblName()));
+ infoLog("Table {} is invalidated", getFullyQualifiedTblName());
} else {
- LOG.debug(debugString("Table %s does not need to be invalidated since "
- + "it does not exist anymore", getFullyQualifiedTblName()));
+ debugLog("Table {} does not need to be invalidated since "
+ + "it does not exist anymore", getFullyQualifiedTblName());
+ }
+ }
+ }
+
+ public static class AddPartitionEvent extends TableInvalidatingEvent {
+
+ /**
+ * Prevent instantiation from outside should use MetastoreEventFactory instead
+ */
+ private AddPartitionEvent(CatalogServiceCatalog catalog, NotificationEvent event)
+ throws MetastoreNotificationException {
+ super(catalog, event);
+ Preconditions.checkState(eventType_.equals(MetastoreEventType.ADD_PARTITION));
+ if (event.getMessage() == null) {
+ throw new IllegalStateException(debugString("Event message is null"));
+ }
+ AddPartitionMessage addPartitionMessage =
+ MetastoreEventsProcessor.getMessageFactory().getDeserializer()
+ .getAddPartitionMessage(event.getMessage());
+ try {
+ msTbl_ = addPartitionMessage.getTableObj();
+ } catch (Exception ex) {
+ throw new MetastoreNotificationException(ex);
+ }
+ }
+ }
+
+ public static class AlterPartitionEvent extends TableInvalidatingEvent {
+
+ /**
+ * Prevent instantiation from outside should use MetastoreEventFactory instead
+ */
+ private AlterPartitionEvent(CatalogServiceCatalog catalog, NotificationEvent event)
+ throws MetastoreNotificationException {
+ super(catalog, event);
+ Preconditions.checkState(eventType_.equals(MetastoreEventType.ALTER_PARTITION));
+ Preconditions.checkNotNull(event.getMessage());
+ AlterPartitionMessage alterPartitionMessage =
+ MetastoreEventsProcessor.getMessageFactory().getDeserializer()
+ .getAlterPartitionMessage(event.getMessage());
+ try {
+ msTbl_ = alterPartitionMessage.getTableObj();
+ } catch (Exception ex) {
+ throw new MetastoreNotificationException(ex);
}
}
}
+ public static class DropPartitionEvent extends TableInvalidatingEvent {
+
+ private final boolean isEventProcessingDisabled_;
+ /**
+ * Prevent instantiation from outside should use MetastoreEventFactory instead
+ */
+ private DropPartitionEvent(CatalogServiceCatalog catalog, NotificationEvent event) {
+ super(catalog, event);
+ Preconditions.checkState(eventType_.equals(MetastoreEventType.DROP_PARTITION));
+ Preconditions.checkNotNull(event.getMessage());
+ //TODO we should use DropPartitionMessage to get the table object here but
+ // current DropPartitionMessage does not provide the table object
+ isEventProcessingDisabled_ = Boolean.valueOf(
+ catalog_.getTableProperty(dbName_, tblName_, DISABLE_EVENT_HMS_SYNC_KEY));
+ }
+
+ @Override
+ public boolean isEventProcessingDisabled() {
+ return isEventProcessingDisabled_;
+ }
+ }
+
/**
* An event type which is ignored. Useful for unsupported metastore event types
*/
- private static class IgnoredEvent extends MetastoreEvent {
+ public static class IgnoredEvent extends MetastoreEvent {
/**
* Prevent instantiation from outside should use MetastoreEventFactory instead
@@ -691,7 +956,12 @@ public class MetastoreEventUtils {
@Override
public void process() {
- LOG.debug(debugString("Ignored"));
+ debugLog("Ignored");
+ }
+
+ @Override
+ protected boolean isEventProcessingDisabled() {
+ return false;
}
}
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEventsProcessor.java b/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEventsProcessor.java
index cca29bb..7bcae29 100644
--- a/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEventsProcessor.java
+++ b/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEventsProcessor.java
@@ -30,12 +30,11 @@ import org.apache.hadoop.hive.metastore.api.NotificationEvent;
import org.apache.hadoop.hive.metastore.api.NotificationEventResponse;
import org.apache.hadoop.hive.metastore.messaging.MessageFactory;
import org.apache.hadoop.hive.metastore.messaging.json.ExtendedJSONMessageFactory;
-import org.apache.hadoop.yarn.webapp.hamlet.HamletSpec.META;
import org.apache.impala.catalog.CatalogException;
import org.apache.impala.catalog.CatalogServiceCatalog;
import org.apache.impala.catalog.MetaStoreClientPool.MetaStoreClient;
-import org.apache.impala.catalog.events.MetastoreEventUtils.MetastoreEvent;
-import org.apache.impala.catalog.events.MetastoreEventUtils.MetastoreEventFactory;
+import org.apache.impala.catalog.events.MetastoreEvents.MetastoreEvent;
+import org.apache.impala.catalog.events.MetastoreEvents.MetastoreEventFactory;
import org.apache.log4j.Logger;
import org.apache.thrift.TException;
@@ -150,7 +149,9 @@ public class MetastoreEventsProcessor implements ExternalEventsProcessor {
public enum EventProcessorStatus {
STOPPED, // event processor is instantiated but not yet scheduled
ACTIVE, // event processor is scheduled at a given frequency
- ERROR // event processor is in error state and event processing has stopped
+ ERROR, // event processor is in error state and event processing has stopped
+ NEEDS_INVALIDATE // event processor could not resolve certain events and needs a
+ // manual invalidate command to reset the state (See AlterEvent for a example)
}
// current status of this event processor
@@ -197,7 +198,7 @@ public class MetastoreEventsProcessor implements ExternalEventsProcessor {
startScheduler();
eventProcessorStatus_ = EventProcessorStatus.ACTIVE;
LOG.info(String.format("Successfully started metastore event processing."
- + "Polling interval: %d seconds.", pollingFrequencyInSec_));
+ + " Polling interval: %d seconds.", pollingFrequencyInSec_));
}
/**
@@ -314,8 +315,7 @@ public class MetastoreEventsProcessor implements ExternalEventsProcessor {
NotificationEvent lastProcessedEvent = null;
try {
EventProcessorStatus currentStatus = eventProcessorStatus_;
- if (currentStatus == EventProcessorStatus.STOPPED
- || currentStatus == EventProcessorStatus.ERROR) {
+ if (currentStatus != EventProcessorStatus.ACTIVE) {
LOG.warn(String.format(
"Event processing is skipped since status is %s. Last synced event id is %d",
currentStatus, lastSyncedEventId_));
@@ -323,15 +323,16 @@ public class MetastoreEventsProcessor implements ExternalEventsProcessor {
}
List<NotificationEvent> events = getNextMetastoreEvents();
- lastProcessedEvent = processEvents(events);
- } catch (MetastoreNotificationFetchException ex) {
+ processEvents(events);
+ } catch (MetastoreNotificationFetchException fetchEx) {
updateStatus(EventProcessorStatus.ERROR);
- LOG.error("Unable to fetch the next batch of metastore events", ex);
- } catch (MetastoreNotificationException | CatalogException ex) {
+ LOG.error("Unable to fetch the next batch of metastore events", fetchEx);
+ } catch(MetastoreNotificationNeedsInvalidateException ex) {
+ updateStatus(EventProcessorStatus.NEEDS_INVALIDATE);
+ LOG.error("Event processing needs a invalidate command to resolve the state", ex);
+ } catch(MetastoreNotificationException ex) {
updateStatus(EventProcessorStatus.ERROR);
- LOG.error(String.format(
- "Unable to process notification event %d due to %s. Event processing will be "
- + "stopped", lastProcessedEvent.getEventId(), ex.getMessage()));
+ LOG.error("Unexpected exception received while processing event", ex);
dumpEventInfoToLog(lastProcessedEvent);
}
}
@@ -343,25 +344,31 @@ public class MetastoreEventsProcessor implements ExternalEventsProcessor {
* @return the last Notification event which was processed.
*/
@VisibleForTesting
- protected NotificationEvent processEvents(List<NotificationEvent> events)
- throws MetastoreNotificationException, CatalogException {
+ protected void processEvents(List<NotificationEvent> events)
+ throws MetastoreNotificationException {
List<MetastoreEvent> filteredEvents =
metastoreEventFactory_.getFilteredEvents(events);
NotificationEvent lastProcessedEvent = null;
- for (MetastoreEvent event : filteredEvents) {
- // synchronizing each event processing reduces the scope of the lock so the a
- // potential reset() during event processing is not blocked for longer than
- // necessary
- synchronized (this) {
- if (eventProcessorStatus_ != EventProcessorStatus.ACTIVE) {
- break;
+ try {
+ for (MetastoreEvent event : filteredEvents) {
+ // synchronizing each event processing reduces the scope of the lock so the a
+ // potential reset() during event processing is not blocked for longer than
+ // necessary
+ synchronized (this) {
+ if (eventProcessorStatus_ != EventProcessorStatus.ACTIVE) {
+ break;
+ }
+ lastProcessedEvent = event.metastoreNotificationEvent_;
+ event.processIfEnabled();
+ lastSyncedEventId_ = event.eventId_;
}
- lastProcessedEvent = event.metastoreNotificationEvent_;
- event.process();
- lastSyncedEventId_ = event.eventId_;
}
+ } catch (CatalogException e) {
+ throw new MetastoreNotificationException(String.format(
+ "Unable to process event %d of type %s. "
+ + "Event processing will be stopped.", lastProcessedEvent.getEventId(),
+ lastProcessedEvent.getEventType()), e);
}
- return lastProcessedEvent;
}
/**
diff --git a/fe/src/main/java/org/apache/impala/catalog/events/MetastoreNotificationNeedsInvalidateException.java b/fe/src/main/java/org/apache/impala/catalog/events/MetastoreNotificationNeedsInvalidateException.java
new file mode 100644
index 0000000..1887755
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/catalog/events/MetastoreNotificationNeedsInvalidateException.java
@@ -0,0 +1,36 @@
+// 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.impala.catalog.events;
+
+/**
+ * Exception class which is thrown by events when catalog state could not be resolved
+ * and it needs a invalidate metadata command to reset the state
+ */
+public class MetastoreNotificationNeedsInvalidateException extends
+ MetastoreNotificationException {
+
+ private static final long serialVersionUID = 4605519100581309214L;
+
+ public MetastoreNotificationNeedsInvalidateException(String msg) {
+ super(msg);
+ }
+
+ public MetastoreNotificationNeedsInvalidateException(String msg, Throwable cause) {
+ super(msg, cause);
+ }
+}
\ No newline at end of file
diff --git a/fe/src/test/java/org/apache/impala/catalog/events/MetastoreEventsProcessorTest.java b/fe/src/test/java/org/apache/impala/catalog/events/MetastoreEventsProcessorTest.java
index bba88c9..be891f5 100644
--- a/fe/src/test/java/org/apache/impala/catalog/events/MetastoreEventsProcessorTest.java
+++ b/fe/src/test/java/org/apache/impala/catalog/events/MetastoreEventsProcessorTest.java
@@ -24,6 +24,7 @@ import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
+import com.google.common.base.Preconditions;
import java.io.File;
import java.util.ArrayList;
import java.util.Arrays;
@@ -31,29 +32,36 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.hadoop.hive.metastore.api.CurrentNotificationEventId;
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.GetPartitionsRequest;
+import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.api.NotificationEvent;
import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.hadoop.hive.metastore.api.PrincipalType;
import org.apache.hadoop.hive.metastore.client.builder.DatabaseBuilder;
import org.apache.hadoop.hive.metastore.client.builder.PartitionBuilder;
import org.apache.hadoop.hive.metastore.client.builder.TableBuilder;
+import org.apache.impala.authorization.SentryConfig;
import org.apache.impala.catalog.CatalogException;
import org.apache.impala.catalog.CatalogServiceCatalog;
import org.apache.impala.catalog.DatabaseNotFoundException;
import org.apache.impala.catalog.HdfsFileFormat;
import org.apache.impala.catalog.HdfsTable;
import org.apache.impala.catalog.IncompleteTable;
+import org.apache.impala.catalog.MetaStoreClientPool;
import org.apache.impala.catalog.MetaStoreClientPool.MetaStoreClient;
import org.apache.impala.catalog.Table;
-import org.apache.impala.catalog.events.MetastoreEventUtils.MetastoreEvent;
-import org.apache.impala.catalog.events.MetastoreEventUtils.MetastoreEventType;
+import org.apache.impala.catalog.events.MetastoreEvents.AlterTableEvent;
+import org.apache.impala.catalog.events.MetastoreEvents.MetastoreEvent;
+import org.apache.impala.catalog.events.MetastoreEvents.MetastoreEventType;
import org.apache.impala.catalog.events.MetastoreEventsProcessor.EventProcessorStatus;
import org.apache.impala.common.ImpalaException;
+import org.apache.impala.common.Pair;
import org.apache.impala.service.CatalogOpExecutor;
+import org.apache.impala.service.FeSupport;
import org.apache.impala.testutil.CatalogServiceTestCatalog;
import org.apache.impala.thrift.TAlterTableOrViewRenameParams;
import org.apache.impala.thrift.TAlterTableParams;
@@ -72,14 +80,15 @@ import org.apache.impala.thrift.TScalarType;
import org.apache.impala.thrift.TTableName;
import org.apache.impala.thrift.TTypeNode;
import org.apache.impala.thrift.TTypeNodeType;
+import org.apache.impala.thrift.TUniqueId;
import org.apache.thrift.TException;
import org.junit.After;
import org.junit.AfterClass;
+import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Ignore;
import org.junit.Test;
-
import com.google.common.collect.Lists;
/**
@@ -180,7 +189,7 @@ public class MetastoreEventsProcessorTest {
public void testDropEmptyDatabaseEvent() throws TException, ImpalaException {
dropDatabaseCascade("database_to_be_dropped");
// create empty database
- createDatabase("database_to_be_dropped");
+ createDatabase("database_to_be_dropped", null);
eventsProcessor_.processEvents();
assertNotNull(catalog_.getDb("database_to_be_dropped"));
dropDatabaseCascade("database_to_be_dropped");
@@ -206,7 +215,7 @@ public class MetastoreEventsProcessorTest {
List<List<String>> partVals = new ArrayList<>(2);
partVals.add(Arrays.asList("1"));
partVals.add(Arrays.asList("2"));
- addPartitions(tblToBeDropped, partVals);
+ addPartitions(TEST_DB_NAME, tblToBeDropped, partVals);
eventsProcessor_.processEvents();
loadTable(tblToBeDropped);
// now drop the database with cascade option
@@ -326,7 +335,7 @@ public class MetastoreEventsProcessorTest {
partVals.add(Arrays.asList("2"));
partVals.add(Arrays.asList("3"));
partVals.add(Arrays.asList("4"));
- addPartitions(TEST_TABLE_NAME_PARTITIONED, partVals);
+ addPartitions(TEST_DB_NAME, TEST_TABLE_NAME_PARTITIONED, partVals);
eventsProcessor_.processEvents();
// after ADD_PARTITION event is received currently we just invalidate the table
@@ -454,7 +463,7 @@ public class MetastoreEventsProcessorTest {
List<List<String>> partVals = new ArrayList<>(2);
partVals.add(Arrays.asList("1"));
partVals.add(Arrays.asList("2"));
- addPartitions(TBL_TO_BE_DROPPED, partVals);
+ addPartitions(TEST_DB_NAME, TBL_TO_BE_DROPPED, partVals);
dropTable(TBL_TO_BE_DROPPED);
eventsProcessor_.processEvents();
assertTrue("Partitioned table should not be found after processing drop_table event",
@@ -685,18 +694,518 @@ public class MetastoreEventsProcessorTest {
catalog_.getDb(TEST_DB_NAME).getMetaStoreDb().getDescription());
}
- private void createDatabase() throws TException { createDatabase(TEST_DB_NAME); }
+ /**
+ * Test checks if the events are processed or ignored when the value of parameter
+ * <code>MetastoreEvents.DISABLE_EVENT_HMS_SYNC_KEY</code> is changed. Currently,
+ * this test only changes the flags for at the table level, since alter_database
+ * events are not supported currently. In order to confirm that the event processing
+ * happens as expected, this test generates a alter_table event using a mock
+ * notificationEvent and a mock catalog which returns the dbFlag flag as expected.
+ * Then it makes sure that the <code>isEventProcessingDisabled</code> method of the
+ * AlterTableEvent returns the expected result, given the flags. And then
+ * generates a additional alter table event to make sure that the subsequent event is
+ * processed/skipped based on the new flag values
+ */
+ @Test
+ public void testEventSyncFlagTransitions() throws Exception {
+ // each db level flag and tbl level flag can undergo following transition
+ // unset(defaults to false) -> true
+ // false -> true
+ // true -> false
+ Pair<String, String> unsetToTrue = new Pair<>(null, "true");
+ Pair<String, String> unsetToFalse = new Pair<>(null, "false");
+ Pair<String, String> falseToTrue = new Pair<>("false", "true");
+ Pair<String, String> falseToUnset = new Pair<>("false", null);
+ Pair<String, String> trueToFalse = new Pair<>("true", "false");
+ Pair<String, String> trueToUnset = new Pair<>("true", null);
+
+ List<Pair<String, String>> allTblTransitions = Arrays.asList(unsetToTrue,
+ unsetToFalse, falseToTrue, falseToUnset, trueToFalse, trueToUnset);
+ List<String> dbFlagVals = Arrays.asList(null, "true", "false");
+ // dbFlag transition is not tested here since ALTER_DATABASE events are ignored
+ // currently. dbFlags do not change in the following loop
+ for (String dbFlag : dbFlagVals) {
+ for (Pair<String, String> tblTransition : allTblTransitions) {
+ // subsequent event is skipped based on the new value of tblTransition flag or
+ // the dbFlag if the new value unsets it
+ boolean shouldSubsequentEventsBeSkipped = tblTransition.second == null ?
+ Boolean.valueOf(dbFlag) : Boolean.valueOf(tblTransition.second);
+ runDDLTestForFlagTransitionWithMock(TEST_DB_NAME, TEST_TABLE_NAME_NONPARTITIONED,
+ dbFlag, tblTransition, shouldSubsequentEventsBeSkipped);
+ }
+ }
+ }
+
+ /**
+ * Test exercises the error condition in event processing when a table creation is
+ * skipped because event processing is disabled for that table. But then user alters
+ * the flag to re-enable the event processing. Since the table doesn't exist in the
+ * catalog in the first place, event processing should stop and go into error state
+ */
+ @Test
+ public void testEventSyncFlagTurnedOnErrorCase()
+ throws TException, CatalogException {
+ // when the event sync flag is changed from true to false (or null), it is possible
+ // that the table is not existing in catalog anymore. Event processing should error
+ // out in such a case
+ Pair<String, String> trueToFalse = new Pair<>("true", "false");
+ Pair<String, String> trueToUnset = new Pair<>("true", null);
+ List<Pair<String, String>> tblFlagTransitions = Arrays.asList(trueToFalse,
+ trueToUnset);
+ List<String> dbFlagVals = Arrays.asList(null, "false");
+ for (String dbFlag : dbFlagVals) {
+ for (Pair<String, String> tblTransition : tblFlagTransitions) {
+ Map<String, String> dbParams = new HashMap<>(1);
+ if (dbFlag != null) {
+ dbParams.put(MetastoreEvents.DISABLE_EVENT_HMS_SYNC_KEY, dbFlag);
+ }
+ Map<String, String> tblParams = new HashMap<>(1);
+ if (tblTransition.first != null) {
+ tblParams.put(MetastoreEvents.DISABLE_EVENT_HMS_SYNC_KEY, tblTransition.first);
+ }
+ createDatabase(TEST_DB_NAME, dbParams);
+ createTable(TEST_DB_NAME, TEST_TABLE_NAME_NONPARTITIONED, tblParams, false);
+ eventsProcessor_.processEvents();
+ // table creation is skipped since the flag says so
+ assertNull(catalog_.getTable(TEST_DB_NAME, TEST_TABLE_NAME_NONPARTITIONED));
+ // now turn on the flag
+ alterTableAddParameter(TEST_TABLE_NAME_NONPARTITIONED,
+ MetastoreEvents.DISABLE_EVENT_HMS_SYNC_KEY, tblTransition.second);
+ eventsProcessor_.processEvents();
+ assertEquals(EventProcessorStatus.NEEDS_INVALIDATE, eventsProcessor_.getStatus());
+ // issue a catalog reset to make sure that table comes back again and event
+ // processing is active
+ catalog_.reset();
+ assertNotNull(catalog_.getTable(TEST_DB_NAME, TEST_TABLE_NAME_NONPARTITIONED));
+ assertEquals(EventProcessorStatus.ACTIVE, eventsProcessor_.getStatus());
+ dropDatabaseCascade(TEST_DB_NAME);
+ }
+ }
+ }
+
+ /**
+ * Test catalog service catalog which takes a value of db and tbl flags for a given
+ * table
+ */
+ private static class FakeCatalogServiceCatalogForFlagTests extends
+ CatalogServiceCatalog {
+
+ private String dbFlag_;
+ private String dbName_;
+ private String tblFlag_;
+ private String tblName_;
+
+ private FakeCatalogServiceCatalogForFlagTests(boolean loadInBackground,
+ int numLoadingThreads, SentryConfig sentryConfig, TUniqueId catalogServiceId,
+ String kerberosPrincipal, String localLibraryPath,
+ MetaStoreClientPool metaStoreClientPool) throws ImpalaException {
+ super(loadInBackground, numLoadingThreads, sentryConfig, catalogServiceId,
+ kerberosPrincipal, localLibraryPath, metaStoreClientPool);
+ }
+
+ public static CatalogServiceCatalog create() {
+ FeSupport.loadLibrary();
+ CatalogServiceCatalog cs;
+ try {
+ cs = new FakeCatalogServiceCatalogForFlagTests(false, 16, null, new TUniqueId(),
+ null, System.getProperty("java.io.tmpdir"), new MetaStoreClientPool(0, 0));
+ cs.reset();
+ } catch (ImpalaException e) {
+ throw new IllegalStateException(e.getMessage(), e);
+ }
+ return cs;
+ }
+
+ public void setFlags(String dbName, String tblName, String dbFlag,
+ String tblFlag) {
+ Preconditions.checkNotNull(dbName);
+ Preconditions.checkNotNull(tblName);
+ this.dbFlag_ = dbFlag;
+ this.dbName_ = dbName;
+ this.tblFlag_ = tblFlag;
+ this.tblName_ = tblName;
+ }
+
+ @Override
+ public String getDbProperty(String dbName, String propertyKey) {
+ if (dbName_.equals(dbName)) {
+ return dbFlag_;
+ }
+ return super.getDbProperty(dbName, propertyKey);
+ }
+
+ @Override
+ public String getTableProperty(String dbName, String tblName, String propertyKey) {
+ if (dbName_.equals(dbName) && tblName_.equals(tblName)) {
+ return tblFlag_;
+ }
+ return super.getTableProperty(dbName, tblName, propertyKey);
+ }
+ }
+
+ /**
+ * Method creates a test tableBefore and tableAfter with the given value of tblFlags.
+ * It then generates a mock notificationEvent which is used to confirm that
+ * AlterTableEvent is not skipped. A subsequent alter table event is generated to
+ * make sure that the new flag transition is working as expected
+ */
+ private void runDDLTestForFlagTransitionWithMock(String dbName, String tblName,
+ String dbFlag, Pair<String, String> tblFlagTransition,
+ boolean shouldNextEventBeSkipped) throws Exception {
+
+ Map<String, String> beforeParams = new HashMap<>();
+ if (tblFlagTransition.first != null) {
+ beforeParams
+ .put(MetastoreEvents.DISABLE_EVENT_HMS_SYNC_KEY, tblFlagTransition.first);
+ }
+
+ Map<String, String> afterParams = new HashMap<>(1);
+ if (tblFlagTransition.second != null) {
+ afterParams
+ .put(MetastoreEvents.DISABLE_EVENT_HMS_SYNC_KEY, tblFlagTransition.second);
+ }
+
+ org.apache.hadoop.hive.metastore.api.Table tableBefore =
+ getTestTable(dbName, tblName, beforeParams, false);
+ org.apache.hadoop.hive.metastore.api.Table tableAfter =
+ getTestTable(dbName, tblName, afterParams, false);
+
+ Map<String, String> dbParams = new HashMap<>(1);
+ if (dbFlag != null) {
+ dbParams.put(MetastoreEvents.DISABLE_EVENT_HMS_SYNC_KEY, dbFlag);
+ }
+
+ CatalogServiceCatalog fakeCatalog = FakeCatalogServiceCatalogForFlagTests.create();
+ ((FakeCatalogServiceCatalogForFlagTests) fakeCatalog)
+ .setFlags(dbName, tblName, dbFlag, tblFlagTransition.first);
+ NotificationEvent fakeAlterTableNotification =
+ createFakeAlterTableNotification(dbName, tblName, tableBefore, tableAfter);
+
+ AlterTableEvent alterTableEvent =
+ new AlterTableEvent(fakeCatalog, fakeAlterTableNotification);
+ Assert.assertFalse("Alter table which " + "changes the flags should not be skipped. "
+ + printFlagTransistions(dbFlag, tblFlagTransition),
+ alterTableEvent.isEventProcessingDisabled());
+
+ // issue a dummy alter table by adding a param
+ afterParams.put("dummy", "value");
+ org.apache.hadoop.hive.metastore.api.Table nextTable =
+ getTestTable(dbName, tblName, afterParams, false);
+ NotificationEvent nextNotification =
+ createFakeAlterTableNotification(dbName, tblName, tableAfter, nextTable);
+ alterTableEvent = new AlterTableEvent(fakeCatalog, nextNotification);
+ if (shouldNextEventBeSkipped) {
+ assertTrue("Alter table event should not skipped following this table flag "
+ + "transition. " + printFlagTransistions(dbFlag, tblFlagTransition),
+ alterTableEvent.isEventProcessingDisabled());
+ } else {
+ assertFalse("Alter table event should have been skipped following the table flag "
+ + "transistion. " + printFlagTransistions(dbFlag, tblFlagTransition),
+ alterTableEvent.isEventProcessingDisabled());
+ }
+ }
+
+ private AtomicLong eventIdGenerator = new AtomicLong(0);
+
+ private NotificationEvent createFakeAlterTableNotification(String dbName,
+ String tblName, org.apache.hadoop.hive.metastore.api.Table tableBefore,
+ org.apache.hadoop.hive.metastore.api.Table tableAfter) {
+ NotificationEvent fakeEvent = new NotificationEvent();
+ fakeEvent.setTableName(tblName);
+ fakeEvent.setDbName(dbName);
+ fakeEvent.setEventId(eventIdGenerator.incrementAndGet());
+ fakeEvent.setMessage(MetastoreEventsProcessor.getMessageFactory()
+ .buildAlterTableMessage(tableBefore, tableAfter).toString());
+ fakeEvent.setEventType("ALTER_TABLE");
+ return fakeEvent;
+ }
+
+ private String printFlagTransistions(String dbFlag,
+ Pair<String, String> tblFlagTransition) {
+ return new StringBuilder("Db flag value: ")
+ .append(dbFlag)
+ .append(" Tbl flag changed from ")
+ .append(tblFlagTransition.first)
+ .append(" -> ")
+ .append(tblFlagTransition.second)
+ .toString();
+ }
+
+ /**
+ * Test runs all the supported DDL operations with the given value of flag at
+ * database and table level
+ */
+ @Test
+ public void testDisableEventSyncFlag() throws Exception {
+ // base case, flags not present
+ runDDLTestsWithFlags(null, null, true);
+ // tbl level flag should get precedence which says event processing is not disabled
+ runDDLTestsWithFlags(false, false, true);
+ runDDLTestsWithFlags(true, false, true);
+ runDDLTestsWithFlags(null, false, true);
+
+ // tblFlag should get precedence which says event processing is disabled
+ runDDLTestsWithFlags(false, true, false);
+ runDDLTestsWithFlags(true, true, false);
+ runDDLTestsWithFlags(null, true, false);
+
+ // when tblFlag is not set, use dbFlag
+ runDDLTestsWithFlags(false, null, true);
+ runDDLTestsWithFlags(true, null, false);
+ }
+
+ /**
+ * Helper method to run all the supported DDL operations on table with the given
+ * values of flag at db and table levels. Takes in a boolean shouldEventGetProcessed
+ * which is used to determine the state of the table in catalog based on whether
+ * event is expected to be processed or skipped
+ */
+ void runDDLTestsWithFlags(Boolean dbFlag, Boolean tblFlag,
+ boolean shouldEventGetProcessed) throws Exception {
+ Map<String, String> dbParams = new HashMap<>(1);
+ Map<String, String> tblParams = new HashMap<>(1);
+ if (dbFlag == null) {
+ // if null, remove the flag
+ dbParams.remove(MetastoreEvents.DISABLE_EVENT_HMS_SYNC_KEY);
+ } else {
+ dbParams.put(MetastoreEvents.DISABLE_EVENT_HMS_SYNC_KEY, String.valueOf(dbFlag));
+ }
+ if (tblFlag == null) {
+ tblParams.remove(MetastoreEvents.DISABLE_EVENT_HMS_SYNC_KEY);
+ } else {
+ tblParams
+ .put(MetastoreEvents.DISABLE_EVENT_HMS_SYNC_KEY, String.valueOf(tblFlag));
+ }
+
+ testDDLOpUsingEvent(TEST_DB_NAME, TEST_TABLE_NAME_PARTITIONED, dbParams,
+ tblParams, MetastoreEventType.CREATE_DATABASE, shouldEventGetProcessed);
+ testDDLOpUsingEvent(TEST_DB_NAME, TEST_TABLE_NAME_PARTITIONED, dbParams,
+ tblParams, MetastoreEventType.ALTER_DATABASE, shouldEventGetProcessed);
+
+ testDDLOpUsingEvent(TEST_DB_NAME, TEST_TABLE_NAME_PARTITIONED, dbParams, tblParams,
+ MetastoreEventType.CREATE_TABLE, shouldEventGetProcessed);
+ testDDLOpUsingEvent(TEST_DB_NAME, TEST_TABLE_NAME_PARTITIONED, dbParams, tblParams,
+ MetastoreEventType.ALTER_TABLE, shouldEventGetProcessed);
+
+ testDDLOpUsingEvent(TEST_DB_NAME, TEST_TABLE_NAME_PARTITIONED, dbParams, tblParams,
+ MetastoreEventType.ADD_PARTITION, shouldEventGetProcessed);
+ testDDLOpUsingEvent(TEST_DB_NAME, TEST_TABLE_NAME_PARTITIONED, dbParams, tblParams,
+ MetastoreEventType.ALTER_PARTITION, shouldEventGetProcessed);
+ testDDLOpUsingEvent(TEST_DB_NAME, TEST_TABLE_NAME_PARTITIONED, dbParams, tblParams,
+ MetastoreEventType.DROP_PARTITION, shouldEventGetProcessed);
+
+ testDDLOpUsingEvent(TEST_DB_NAME, TEST_TABLE_NAME_PARTITIONED, dbParams, tblParams,
+ MetastoreEventType.DROP_TABLE, shouldEventGetProcessed);
+ testDDLOpUsingEvent(TEST_DB_NAME, TEST_TABLE_NAME_PARTITIONED, dbParams,
+ tblParams, MetastoreEventType.DROP_DATABASE, shouldEventGetProcessed);
+ }
+
+ private void cleanUpTblsForFlagTests(String dbName)
+ throws TException, MetastoreNotificationFetchException {
+ if (catalog_.getDb(dbName) == null) return;
+
+ dropDatabaseCascade(dbName);
+ assertFalse(eventsProcessor_.getNextMetastoreEvents().isEmpty());
+ eventsProcessor_.processEvents();
+ assertNull(catalog_.getDb(dbName));
+ }
+
+ private void initTblsForFlagTests(String dbName, String tblName,
+ Map<String, String> dbParams, Map<String, String> tblParams) throws Exception {
+ assertNull(catalog_.getDb(dbName));
+ createDatabase(dbName, dbParams);
+ createTable(dbName, tblName, tblParams, true);
+ List<List<String>> partVals = new ArrayList<>(3);
+ partVals.add(Arrays.asList("1"));
+ partVals.add(Arrays.asList("2"));
+ partVals.add(Arrays.asList("3"));
+ addPartitions(dbName, tblName, partVals);
+ assertEquals(3, eventsProcessor_.getNextMetastoreEvents().size());
+ }
+
+ private void testDDLOpUsingEvent(String dbName, String tblName,
+ Map<String, String> dbParams,
+ Map<String, String> tblParams, MetastoreEventType ddlOpCode,
+ boolean shouldEventBeProcessed)
+ throws Exception {
+ switch (ddlOpCode) {
+ case CREATE_TABLE: {
+ initTblsForFlagTests(dbName, tblName, dbParams, tblParams);
+ eventsProcessor_.processEvents();
+ if (shouldEventBeProcessed) {
+ assertNotNull(catalog_.getTable(dbName, tblName));
+ } else {
+ assertNull(catalog_.getTable(dbName, tblName));
+ }
+ cleanUpTblsForFlagTests(dbName);
+ return;
+ }
+ case ALTER_TABLE: {
+ initTblsForFlagTests(dbName, tblName, dbParams, tblParams);
+ eventsProcessor_.processEvents();
+ loadTable(tblName);
+ alterTableAddCol(tblName, "newCol", "string", "test new column");
+ alterTableAddParameter(tblName, "testParamKey", "somevalue");
+ alterTableRename(tblName, "newTblName");
+ altertableChangeCol("newTblName", "newCol", "int", "changed type to int");
+ alterTableRemoveCol("newTblName", "newCol");
+ alterTableRename("newTblName", tblName);
+ assertEquals(6, eventsProcessor_.getNextMetastoreEvents().size());
+ eventsProcessor_.processEvents();
+ if (shouldEventBeProcessed) {
+ assertNotNull(catalog_.getTable(dbName, tblName));
+ assertTrue(catalog_.getTable(dbName, tblName) instanceof IncompleteTable);
+ } else {
+ assertNull(catalog_.getTable(dbName, tblName));
+ }
+ cleanUpTblsForFlagTests(dbName);
+ return;
+ }
+ case DROP_TABLE: {
+ // in case of drop table use impala to directly create table since if you use
+ // hive to create the table the create table could be ignored given the flags
+ createDatabase(dbName, dbParams);
+ eventsProcessor_.processEvents();
+ createTableFromImpala(dbName, "impala_test_tbl", tblParams, true);
+ eventsProcessor_.processEvents();
+ assertNotNull(catalog_.getTable(dbName, "impala_test_tbl"));
+ dropTable("impala_test_tbl");
+ assertEquals(1, eventsProcessor_.getNextMetastoreEvents().size());
+ eventsProcessor_.processEvents();
+ if (shouldEventBeProcessed) {
+ assertNull(catalog_.getTable(dbName, "impala_test_tbl"));
+ } else {
+ assertNotNull(catalog_.getTable(dbName, "impala_test_tbl"));
+ }
+ cleanUpTblsForFlagTests(dbName);
+ return;
+ }
+ case ADD_PARTITION: {
+ initTblsForFlagTests(dbName, tblName, dbParams, tblParams);
+ eventsProcessor_.processEvents();
+ loadTable(tblName);
+ List<List<String>> partValues = new ArrayList<>(3);
+ partValues.add(Arrays.asList("4"));
+ partValues.add(Arrays.asList("5"));
+ partValues.add(Arrays.asList("6"));
+ addPartitions(dbName, tblName, partValues);
+ assertEquals(1, eventsProcessor_.getNextMetastoreEvents().size());
+ eventsProcessor_.processEvents();
+ if (shouldEventBeProcessed) {
+ assertTrue("Table should have been invalidated after add partition event",
+ catalog_.getTable(dbName, tblName) instanceof IncompleteTable);
+ } else {
+ assertFalse("Table should still have been in loaded state since sync is "
+ + "disabled",
+ catalog_.getTable(dbName, tblName) instanceof IncompleteTable);
+ }
+ cleanUpTblsForFlagTests(dbName);
+ return;
+ }
+ case DROP_PARTITION: {
+ initTblsForFlagTests(dbName, tblName, dbParams, tblParams);
+ eventsProcessor_.processEvents();
+ loadTable(tblName);
+ List<List<String>> partValues = new ArrayList<>(3);
+ partValues.add(Arrays.asList("3"));
+ dropPartitions(tblName, partValues);
+ assertEquals(1, eventsProcessor_.getNextMetastoreEvents().size());
+ eventsProcessor_.processEvents();
+ if (shouldEventBeProcessed) {
+ assertTrue("Table should have been invalidated after alter partition event",
+ catalog_.getTable(dbName, tblName) instanceof IncompleteTable);
+ } else {
+ assertFalse("Table should still have been in loaded state since sync is "
+ + "disabled",
+ catalog_.getTable(dbName, tblName) instanceof IncompleteTable);
+ }
+ cleanUpTblsForFlagTests(dbName);
+ return;
+ }
+ case ALTER_PARTITION: {
+ initTblsForFlagTests(dbName, tblName, dbParams, tblParams);
+ eventsProcessor_.processEvents();
+ loadTable(tblName);
+ List<List<String>> partValues = new ArrayList<>(1);
+ partValues.add(Arrays.asList("2"));
+ partValues.add(Arrays.asList("1"));
+ Map<String, String> newParams = new HashMap<>();
+ newParams.put("newParamk1", "newParamv1");
+ alterPartitions(tblName, partValues, newParams);
+ assertEquals(2, eventsProcessor_.getNextMetastoreEvents().size());
+ eventsProcessor_.processEvents();
+ if (shouldEventBeProcessed) {
+ assertTrue("Table should have been invalidated after alter partition event",
+ catalog_.getTable(dbName, tblName) instanceof IncompleteTable);
+ } else {
+ assertFalse("Table should still have been in loaded state since sync is "
+ + "disabled",
+ catalog_.getTable(dbName, tblName) instanceof IncompleteTable);
+ }
+ cleanUpTblsForFlagTests(dbName);
+ return;
+ }
+ case CREATE_DATABASE: {
+ initTblsForFlagTests(dbName, tblName, dbParams, tblParams);
+ eventsProcessor_.processEvents();
+ // database ops do not use disable flag, so they should always be processed
+ assertNotNull("Database should have been created after create database event",
+ catalog_.getDb(dbName));
+ cleanUpTblsForFlagTests(dbName);
+ return;
+ }
+ case DROP_DATABASE: {
+ initTblsForFlagTests(dbName, tblName, dbParams, tblParams);
+ eventsProcessor_.processEvents();
+ assertNotNull(catalog_.getDb(dbName));
+ dropDatabaseCascade(dbName);
+ eventsProcessor_.processEvents();
+ assertNull("Database should have been dropped after drop database event",
+ catalog_.getDb(dbName));
+ cleanUpTblsForFlagTests(dbName);
+ return;
+ }
+ case ALTER_DATABASE: {
+ // TODO alter database events are currently ignored
+ return;
+ }
+ }
+ }
+
+ @Ignore("Ignored we add alter database event support. See IMPALA-8149")
+ @Test
+ public void testAlterDisableFlagFromDb()
+ throws TException, CatalogException, MetastoreNotificationFetchException {
+ createDatabase();
+ eventsProcessor_.processEvents();
+ Database alteredDb = catalog_.getDb(TEST_DB_NAME).getMetaStoreDb().deepCopy();
+ alteredDb.putToParameters(MetastoreEvents.DISABLE_EVENT_HMS_SYNC_KEY, "true");
+ alterDatabase(alteredDb);
+
+ createTable(TEST_TABLE_NAME_NONPARTITIONED, false);
+ assertEquals(1, eventsProcessor_.getNextMetastoreEvents().size());
+ eventsProcessor_.processEvents();
+ assertNull("Table creation should be skipped when database level event sync flag is"
+ + " disabled", catalog_.getTable(TEST_DB_NAME, TEST_TABLE_NAME_NONPARTITIONED));
+ }
+
+ private void createDatabase() throws TException {
+ createDatabase(TEST_DB_NAME, null);
+ }
- private void createDatabase(String dbName) throws TException {
- Database database =
+ private void createDatabase(String dbName, Map<String, String> params)
+ throws TException {
+ DatabaseBuilder databaseBuilder =
new DatabaseBuilder()
.setName(dbName)
.setDescription("Notification test database")
- .addParam("dbparamkey", "dbparamValue")
.setOwnerName("NotificationTestOwner")
- .setOwnerType(PrincipalType.USER).build();
+ .setOwnerType(PrincipalType.USER);
+ if (params != null && !params.isEmpty()) {
+ databaseBuilder.setParams(params);
+ }
try (MetaStoreClient msClient = catalog_.getMetaStoreClient()) {
- msClient.getHiveClient().createDatabase(database);
+ msClient.getHiveClient().createDatabase(databaseBuilder.build());
}
}
@@ -710,30 +1219,48 @@ public class MetastoreEventsProcessorTest {
}
}
+
private void alterDatabase(Database newDatabase) throws TException {
try (MetaStoreClient msClient = catalog_.getMetaStoreClient()) {
msClient.getHiveClient().alterDatabase(newDatabase.getName(), newDatabase);
}
}
- private void createTable(String tblName, boolean isPartitioned) throws TException {
+ private void createTable(String dbName, String tblName, Map<String, String> params,
+ boolean isPartitioned) throws TException {
+ org.apache.hadoop.hive.metastore.api.Table
+ tbl = getTestTable(dbName, tblName, params, isPartitioned);
+
+ try (MetaStoreClient msClient = catalog_.getMetaStoreClient()) {
+ msClient.getHiveClient().createTable(tbl);
+ }
+ }
+
+ private org.apache.hadoop.hive.metastore.api.Table getTestTable(String dbName,
+ String tblName, Map<String, String> params, boolean isPartitioned)
+ throws MetaException {
TableBuilder tblBuilder =
new TableBuilder()
.setTableName(tblName)
- .setDbName(TEST_DB_NAME)
+ .setDbName(dbName)
.addTableParam("tblParamKey", "tblParamValue")
.addCol("c1", "string", "c1 description")
.addCol("c2", "string", "c2 description")
.setSerdeLib(HdfsFileFormat.PARQUET.serializationLib())
.setInputFormat(HdfsFileFormat.PARQUET.inputFormat())
.setOutputFormat(HdfsFileFormat.PARQUET.outputFormat());
+ // if params are provided use them
+ if (params != null && !params.isEmpty()) {
+ tblBuilder.setTableParams(params);
+ }
if (isPartitioned) {
tblBuilder.addPartCol("p1", "string", "partition p1 description");
}
+ return tblBuilder.build();
+ }
- try (MetaStoreClient msClient = catalog_.getMetaStoreClient()) {
- msClient.getHiveClient().createTable(tblBuilder.build());
- }
+ private void createTable(String tblName, boolean isPartitioned) throws TException {
+ createTable(TEST_DB_NAME, tblName, null, isPartitioned);
}
/**
@@ -775,22 +1302,27 @@ public class MetastoreEventsProcessorTest {
catalogOpExecutor_.execDdlRequest(req);
}
+ private void createTableFromImpala(String tblName, boolean isPartitioned)
+ throws ImpalaException {
+ createTableFromImpala(TEST_DB_NAME, tblName, null, isPartitioned);
+ }
/**
* Creates a table using CatalogOpExecutor to simulate a DDL operation from Impala
* client
*/
- private void createTableFromImpala(String tblName, boolean isPartitioned)
+ private void createTableFromImpala(String dbName, String tblName,
+ Map<String, String> tblParams, boolean isPartitioned)
throws ImpalaException {
TDdlExecRequest req = new TDdlExecRequest();
req.setDdl_type(TDdlType.CREATE_TABLE);
TCreateTableParams createTableParams = new TCreateTableParams();
- createTableParams.setTable_name(new TTableName(TEST_DB_NAME, tblName));
+ createTableParams.setTable_name(new TTableName(dbName, tblName));
createTableParams.setFile_format(THdfsFileFormat.PARQUET);
createTableParams.setIs_external(false);
createTableParams.setIf_not_exists(false);
- Map<String, String> properties = new HashMap<>();
- properties.put("tblParamKey", "tblParamValue");
- createTableParams.setTable_properties(properties);
+ if (tblParams != null) {
+ createTableParams.setTable_properties(tblParams);
+ }
List<TColumn> columns = new ArrayList<>(2);
columns.add(getScalarColumn("c1", TPrimitiveType.STRING));
columns.add(getScalarColumn("c2", TPrimitiveType.STRING));
@@ -851,7 +1383,11 @@ public class MetastoreEventsProcessorTest {
try (MetaStoreClient msClient = catalog_.getMetaStoreClient()) {
org.apache.hadoop.hive.metastore.api.Table msTable =
msClient.getHiveClient().getTable(TEST_DB_NAME, tblName);
- msTable.getParameters().put(key, val);
+ if (val == null) {
+ msTable.getParameters().remove(key);
+ } else {
+ msTable.getParameters().put(key, val);
+ }
msClient.getHiveClient().alter_table_with_environmentContext(
TEST_DB_NAME, tblName, msTable, null);
}
@@ -942,13 +1478,14 @@ public class MetastoreEventsProcessorTest {
}
}
- private void addPartitions(String tblName, List<List<String>> partitionValues)
+ private void addPartitions(String dbName, String tblName,
+ List<List<String>> partitionValues)
throws TException {
int i = 0;
List<Partition> partitions = new ArrayList<>(partitionValues.size());
try (MetaStoreClient msClient = catalog_.getMetaStoreClient()) {
org.apache.hadoop.hive.metastore.api.Table msTable =
- msClient.getHiveClient().getTable(TEST_DB_NAME, tblName);
+ msClient.getHiveClient().getTable(dbName, tblName);
for (List<String> partVals : partitionValues) {
partitions.add(
new PartitionBuilder()
@@ -965,10 +1502,14 @@ public class MetastoreEventsProcessorTest {
}
}
- private Table loadTable(String tblName) throws CatalogException {
- Table loadedTable = catalog_.getOrLoadTable(TEST_DB_NAME, tblName);
+ private Table loadTable(String dbName, String tblName) throws CatalogException {
+ Table loadedTable = catalog_.getOrLoadTable(dbName, tblName);
assertFalse("Table should have been loaded after getOrLoadTable call",
loadedTable instanceof IncompleteTable);
return loadedTable;
}
+
+ private Table loadTable(String tblName) throws CatalogException {
+ return loadTable(TEST_DB_NAME, tblName);
+ }
}