aasha commented on a change in pull request #2365:
URL: https://github.com/apache/hive/pull/2365#discussion_r651473262
##########
File path:
hcatalog/server-extensions/src/main/java/org/apache/hive/hcatalog/listener/DbNotificationListener.java
##########
@@ -1194,108 +1205,82 @@ static String quoteString(String input) {
private void addNotificationLog(NotificationEvent event, ListenerEvent
listenerEvent, Connection dbConn,
SQLGenerator sqlGenerator) throws
MetaException, SQLException {
LOG.debug("DbNotificationListener: adding notification log for : {}",
event.getMessage());
+ addNotificationLogBatch(Collections.singletonList(event),
Collections.singletonList(listenerEvent),
+ dbConn, sqlGenerator);
+ }
+
+ private void addNotificationLogBatch(List<NotificationEvent> eventList,
List<ListenerEvent> listenerEventList,
+ Connection dbConn, SQLGenerator sqlGenerator)
throws MetaException, SQLException {
if ((dbConn == null) || (sqlGenerator == null)) {
LOG.info("connection or sql generator is not set so executing sql via
DN");
- process(event, listenerEvent);
+ for (int idx = 0; idx < eventList.size(); idx++) {
+ LOG.debug("DbNotificationListener: adding notification log for : {}",
eventList.get(idx).getMessage());
+ process(eventList.get(idx), listenerEventList.get(idx));
+ }
return;
}
- Statement stmt = null;
- PreparedStatement pst = null;
- ResultSet rs = null;
- try {
- stmt = dbConn.createStatement();
- event.setMessageFormat(msgEncoder.getMessageFormat());
+ try (Statement stmt = dbConn.createStatement()) {
if (sqlGenerator.getDbProduct().isMYSQL()) {
stmt.execute("SET @@session.sql_mode=ANSI_QUOTES");
}
+ }
- long nextEventId = getNextEventId(dbConn, sqlGenerator);
-
- long nextNLId = getNextNLId(dbConn, sqlGenerator,
- "org.apache.hadoop.hive.metastore.model.MNotificationLog");
-
- String insertVal;
- String columns;
- List<String> params = new ArrayList<String>();
-
- // Construct the values string, parameters and column string step by
step simultaneously so
- // that the positions of columns and of their corresponding values do
not go out of sync.
-
- // Notification log id
- columns = "\"NL_ID\"";
- insertVal = "" + nextNLId;
-
- // Event id
- columns = columns + ", \"EVENT_ID\"";
- insertVal = insertVal + "," + nextEventId;
-
- // Event time
- columns = columns + ", \"EVENT_TIME\"";
- insertVal = insertVal + "," + event.getEventTime();
+ long nextEventId = getNextEventId(dbConn, sqlGenerator, eventList.size());
+ long nextNLId = getNextNLId(dbConn, sqlGenerator,
+ "org.apache.hadoop.hive.metastore.model.MNotificationLog",
eventList.size());
- // Event type
- columns = columns + ", \"EVENT_TYPE\"";
- insertVal = insertVal + ", ?";
- params.add(event.getEventType());
+ String columns = "\"NL_ID\"" + ", \"EVENT_ID\"" + ", \"EVENT_TIME\"" + ",
\"EVENT_TYPE\"" + ", \"MESSAGE\""
+ + ", \"MESSAGE_FORMAT\"" + ", \"DB_NAME\"" + ", \"TBL_NAME\"" + ",
\"CAT_NAME\"";
+ String insertVal = "insert into \"NOTIFICATION_LOG\" (" + columns + ")
VALUES ("
+ + "?,?,?,?,?,?,?,?,?"
+ + ")";
- // Message
- columns = columns + ", \"MESSAGE\"";
- insertVal = insertVal + ", ?";
- params.add(event.getMessage());
+ try (PreparedStatement pst = dbConn.prepareStatement(insertVal)) {
+ int numRows = 0;
- // Message format
- columns = columns + ", \"MESSAGE_FORMAT\"";
- insertVal = insertVal + ", ?";
- params.add(event.getMessageFormat());
+ for (int idx = 0; idx < eventList.size(); idx++) {
+ NotificationEvent event = eventList.get(idx);
+ ListenerEvent listenerEvent = listenerEventList.get(idx);
- // Database name, optional
- String dbName = event.getDbName();
- if (dbName != null) {
- assert dbName.equals(dbName.toLowerCase());
- columns = columns + ", \"DB_NAME\"";
- insertVal = insertVal + ", ?";
- params.add(dbName);
- }
+ LOG.debug("DbNotificationListener: adding notification log for : {}",
event.getMessage());
+ event.setMessageFormat(msgEncoder.getMessageFormat());
- // Table name, optional
- String tableName = event.getTableName();
- if (tableName != null) {
- assert tableName.equals(tableName.toLowerCase());
- columns = columns + ", \"TBL_NAME\"";
- insertVal = insertVal + ", ?";
- params.add(tableName);
- }
+ pst.setLong(1, nextNLId);
+ pst.setLong(2, nextEventId);
+ pst.setLong(3, now());
+ pst.setString(4, event.getEventType());
+ pst.setString(5, event.getMessage());
+ pst.setString(6, event.getMessageFormat());
+ pst.setString(7, event.getDbName());
+ pst.setString(8, event.getTableName());
+ pst.setString(9, event.getCatName());
+
+ LOG.debug("Going to execute insert <" + insertVal + ">");
+ pst.addBatch();
+ numRows++;
+
+ if (numRows == maxBatchSize) {
+ pst.executeBatch();
+ numRows = 0;
+ }
- // Catalog name, optional
- String catName = event.getCatName();
- if (catName != null) {
- assert catName.equals(catName.toLowerCase());
- columns = columns + ", \"CAT_NAME\"";
- insertVal = insertVal + ", ?";
- params.add(catName);
+ event.setEventId(nextEventId);
+ // Set the DB_NOTIFICATION_EVENT_ID for future reference by other
listeners.
+ if (event.isSetEventId()) {
+ listenerEvent.putParameter(
+
MetaStoreEventListenerConstants.DB_NOTIFICATION_EVENT_ID_KEY_NAME,
+ Long.toString(event.getEventId()));
+ }
Review comment:
what happens in case of partial failures of the batch.
what will happen to the notification ids?
can there be a scenario when partially entries are inserted to notification
log?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
