HIVE-29784: Cleaning lots of expired notifications could lead to OOM

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java

@@ -420,12 +420,8 @@ public <T> T unwrap(Class<T> iface) {
     if (descriptor == null) {
       throw new IllegalArgumentException("Unable to unwrap the store as " + iface);
     }
-    String implClassName =
-        conf.get("metastore." + descriptor.alias() + ".store.impl", "");
-    Class<?> ifaceImpl = descriptor.defaultImpl();
-    if (StringUtils.isNotEmpty(implClassName)) {
-      ifaceImpl = conf.getClass(implClassName, ifaceImpl);
-    }
+    Class<?> ifaceImpl =
+        conf.getClass("metastore." + descriptor.alias() + ".store.impl", descriptor.defaultImpl());
     T simpl = (T) JavaUtils.newInstance(ifaceImpl);
     List<Query> openQueries = new LinkedList<>();
     if (simpl instanceof RawStoreBundle rsb) {

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/iface/NotificationStore.java

@@ -49,10 +49,11 @@ public interface NotificationStore {
   void addNotificationEvent(NotificationEvent event) throws MetaException;
 
   /**
-   * Remove older notification events.
+   * Remove older notification events, transaction is explicitly handled inside.
    *
    * @param olderThan Remove any events older or equal to a given number of seconds
    */
+  @MetaDescriptor.NoTransaction
   void cleanNotificationEvents(int olderThan);
 
   /**
@@ -71,9 +72,10 @@ public interface NotificationStore {
   NotificationEventsCountResponse getNotificationEventsCount(NotificationEventsCountRequest rqst);
 
   /**
-   * Remove older notification events.
+   * Remove older notification events, transaction is explicitly handled inside.
    * @param olderThan Remove any events older or equal to a given number of seconds
    */
+  @MetaDescriptor.NoTransaction
   void cleanWriteNotificationEvents(int olderThan);
 
   /**

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/ColStatsStoreImpl.java

@@ -483,7 +483,6 @@ private List<MTableColumnStatistics> getMTableColumnStatistics(Table table, List
     validateTableCols(table, colNames);
 
     List<MTableColumnStatistics> result = Collections.emptyList();
-    Query query = pm.newQuery(MTableColumnStatistics.class);
     result =
         Batchable.runBatched(batchSize, colNames, new Batchable<String, MTableColumnStatistics>() {
           @Override
@@ -504,6 +503,7 @@ public List<MTableColumnStatistics> run(List<String> input)
               params[i + 4] = input.get(i);
             }
             filter.append(")");
+            Query query = pm.newQuery(MTableColumnStatistics.class);
             query.setFilter(filter.toString());
             query.declareParameters(paramStr.toString());
             List<MTableColumnStatistics> paritial = (List<MTableColumnStatistics>) query.executeWithArray(params);

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/NotificationStoreImpl.java

@@ -247,23 +247,37 @@ private void cleanOlderEvents(int olderThan, Class table, String tableName) {
     final Optional<Integer> batchSize = (eventBatchSize > 0) ? Optional.of(eventBatchSize) : Optional.empty();
 
     final long start = System.nanoTime();
-    int deleteCount = doCleanNotificationEvents(tooOld, batchSize, table, tableName);
+    int deleteCount = 0;
+    int batchCount;
+    do {
+      batchCount = cleanNotificationEventsBatch(tooOld, batchSize, table, tableName);
+      deleteCount += batchCount;
+    } while (batchCount > 0);
 
     if (deleteCount == 0) {
       LOG.info("No {} events found to be cleaned with eventTime < {}", tableName, tooOld);
-    } else {
-      int batchCount = 0;
-      do {
-        batchCount = doCleanNotificationEvents(tooOld, batchSize, table, tableName);
-        deleteCount += batchCount;
-      } while (batchCount > 0);
     }
 
     final long finish = System.nanoTime();
     LOG.info("Deleted {} {} events older than epoch:{} in {}ms", deleteCount, tableName, tooOld,
         TimeUnit.NANOSECONDS.toMillis(finish - start));
   }
 
+  private <T> int cleanNotificationEventsBatch(final int ageSec, final Optional<Integer> batchSize,
+      Class<T> tableClass, String tableName) {
+    boolean committed = false;
+    baseStore.openTransaction();
+    try {
+      int deleted = doCleanNotificationEvents(ageSec, batchSize, tableClass, tableName);
+      committed = baseStore.commitTransaction();
+      return deleted;
+    } finally {
+      if (!committed && baseStore.isActiveTransaction()) {
+        baseStore.rollbackTransaction();
+      }
+    }
+  }
+
   private <T> int doCleanNotificationEvents(final int ageSec, final Optional<Integer> batchSize,
       Class<T> tableClass, String tableName) {
     int eventsCount = 0;
@@ -308,6 +322,7 @@ private <T> int doCleanNotificationEvents(final int ageSec, final Optional<Integ
       }
       pm.deletePersistentAll(events);
     }
+    query.closeAll();
     return eventsCount;
   }
 
@@ -327,15 +342,11 @@ public NotificationEventsCountResponse getNotificationEventsCount(NotificationEv
     Long result = 0L;
     long fromEventId = rqst.getFromEventId();
     String inputDbName = rqst.getDbName();
-    String catName = rqst.isSetCatName() ? rqst.getCatName() : getDefaultCatalog(conf);
+    String catName = rqst.isSetCatName() ? normalizeIdentifier(rqst.getCatName()) : getDefaultCatalog(conf);
     long toEventId;
     String paramSpecs;
     List<Object> paramVals = new ArrayList<>();
 
-    // We store a catalog name in lower case in metastore and also use the same way everywhere in
-    // hive.
-    assert catName.equals(catName.toLowerCase());
-
     // Build the query to count events, part by part
     String queryStr = "select count(eventId) from " + MNotificationLog.class.getName();
     // count fromEventId onwards events
@@ -352,8 +363,7 @@ public NotificationEventsCountResponse getNotificationEventsCount(NotificationEv
       // counted.
       queryStr = queryStr + " && (dbName == inputDbName || dbName == null)";
       paramSpecs = paramSpecs + ", java.lang.String inputDbName";
-      // We store a database name in lower case in metastore.
-      paramVals.add(inputDbName.toLowerCase());
+      paramVals.add(normalizeIdentifier(inputDbName));
     }
 
     // catName could be NULL in case of transaction related events, which also need to be
@@ -373,7 +383,7 @@ public NotificationEventsCountResponse getNotificationEventsCount(NotificationEv
     if (rqst.isSetTableNames() && !rqst.getTableNames().isEmpty()) {
       queryStr = queryStr + " && (";
       for (String tableName : rqst.getTableNames()) {
-        paramVals.add(tableName.toLowerCase());
+        paramVals.add(normalizeIdentifier(tableName));
         queryStr = queryStr + "tableName == tableName" + paramVals.size() + " || ";
         paramSpecs = paramSpecs + ", java.lang.String tableName" + paramVals.size();
       }
@@ -385,7 +395,7 @@ public NotificationEventsCountResponse getNotificationEventsCount(NotificationEv
     query.declareParameters(paramSpecs);
     result = (Long) query.executeWithArray(paramVals.toArray());
     // Cap the event count by limit if specified.
-    long  eventCount = result.longValue();
+    long eventCount = result.longValue();
     if (rqst.isSetLimit() && eventCount > rqst.getLimit()) {
       eventCount = rqst.getLimit();
     }

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/TableStoreImpl.java

@@ -937,7 +937,7 @@ public Table alterTable(TableName tableName, Table newTable, String queryValidWr
     boolean isToTxn = isTxn && !TxnUtils.isTransactionalTable(oldt.getParameters());
     if (!isToTxn && isTxn && areTxnStatsSupported) {
       // Transactional table is altered without a txn. Make sure there are no changes to the flag.
-      String errorMsg = verifyStatsChangeCtx(TableName.getDbTable(name, dbname), oldt.getParameters(),
+      String errorMsg = verifyStatsChangeCtx(TableName.getDbTable(dbname, name), oldt.getParameters(),
           newTable.getParameters(), newTable.getWriteId(), queryValidWriteIds, false);
       if (errorMsg != null) {
         throw new MetaException(errorMsg);
@@ -1845,25 +1845,17 @@ public Partition alterPartition(TableName tableName, List<String> part_vals, Par
     String dbname = normalizeIdentifier(tableName.getDb());
     String name = normalizeIdentifier(tableName.getTable());
     AtomicReference<MColumnDescriptor> oldCd = new AtomicReference<>();
-    Partition result = alterPartitionNoTxn(catName, dbname, name, part_vals, new_part, queryValidWriteIds, oldCd);
+    MTable table = this.getMTable(new_part.getCatName(), new_part.getDbName(), new_part.getTableName());
+    MPartition oldp = getMPartition(catName, dbname, name, part_vals, table);
+    Partition result = alterPartitionNoTxn(catName, dbname, name, oldp, new_part, queryValidWriteIds, oldCd, table);
     removeUnusedColumnDescriptor(oldCd.get());
     return result;
   }
 
   /**
    * Alters an existing partition. Initiates copy of SD. Returns the old CD.
-   * @param part_vals Partition values (of the original partition instance)
    * @param newPart Partition object containing new information
    */
-  private Partition alterPartitionNoTxn(String catName, String dbname, String name,
-      List<String> part_vals, Partition newPart, String validWriteIds, AtomicReference<MColumnDescriptor> oldCd)
-      throws InvalidObjectException, MetaException {
-    MTable table = this.getMTable(newPart.getCatName(), newPart.getDbName(), newPart.getTableName());
-    MPartition oldp = getMPartition(catName, dbname, name, part_vals, table);
-    return alterPartitionNoTxn(catName, dbname, name, oldp, newPart,
-        validWriteIds, oldCd, table);
-  }
-
   private Partition alterPartitionNoTxn(String catName, String dbname,
       String name, MPartition oldp, Partition newPart,
       String validWriteIds,
@@ -1872,15 +1864,15 @@ private Partition alterPartitionNoTxn(String catName, String dbname,
     catName = normalizeIdentifier(catName);
     name = normalizeIdentifier(name);
     dbname = normalizeIdentifier(dbname);
+    if (oldp == null) {
+      throw new InvalidObjectException("partition does not exist.");
+    }
     MPartition newp = convertToMPart(newPart, table);
     MColumnDescriptor oldCD = null;
     MStorageDescriptor oldSD = oldp.getSd();
     if (oldSD != null) {
       oldCD = oldSD.getCD();
     }
-    if (newp == null) {
-      throw new InvalidObjectException("partition does not exist.");
-    }
     oldp.setValues(newp.getValues());
     oldp.setPartitionName(newp.getPartitionName());
     boolean isTxn = TxnUtils.isTransactionalTable(table.getParameters());
@@ -1968,7 +1960,7 @@ protected List<Partition> alterPartitionsInternal(MTable table,
         throw new MetaException("Invalid DB name : " + tmpPart.getDbName());
       }
       if (!tmpPart.getTableName().equalsIgnoreCase(tblName)) {
-        throw new MetaException("Invalid table name : " + tmpPart.getDbName());
+        throw new MetaException("Invalid table name : " + tmpPart.getTableName());
       }
     }
     return new GetListHelper<TableName, Partition>(this, null) {
@@ -2003,8 +1995,9 @@ private List<Partition> alterPartitionsViaJdo(MTable table, List<String> partNam
       mPartitionList = (List<MPartition>) query.executeWithArray(tblName, dbName, partNames, catName);
       pm.retrieveAll(mPartitionList);
 
-      if (mPartitionList.size() > newParts.size()) {
-        throw new MetaException("Expecting only one partition but more than one partitions are found.");
+      if (mPartitionList.size() != newParts.size()) {
+        throw new MetaException("Expected " + newParts.size() + " partitions but found "
+            + mPartitionList.size());
       }
 
       Map<List<String>, MPartition> mPartsMap = new HashMap();
@@ -2016,8 +2009,9 @@ private List<Partition> alterPartitionsViaJdo(MTable table, List<String> partNam
       AtomicReference<MColumnDescriptor> oldCdRef = new AtomicReference<>();
       for (Partition tmpPart : newParts) {
         oldCdRef.set(null);
+        MPartition mPart = mPartsMap.get(tmpPart.getValues());
         Partition result = alterPartitionNoTxn(catName, dbName, tblName,
-            mPartsMap.get(tmpPart.getValues()), tmpPart, queryWriteIdList, oldCdRef, table);
+            mPart, tmpPart, queryWriteIdList, oldCdRef, table);
         results.add(result);
         if (oldCdRef.get() != null) {
           oldCds.add(oldCdRef.get());

standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/ExecutionContextTestUtils.java

@@ -0,0 +1,57 @@
+/*
+ * 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.metastore;
+
+import org.apache.hadoop.hive.metastore.metastore.PersistenceManagerProxy;
+import org.datanucleus.ExecutionContext;
+import org.datanucleus.api.jdo.JDOPersistenceManager;
+import org.datanucleus.cache.Level1Cache;
+import org.datanucleus.state.DNStateManager;
+
+import javax.jdo.PersistenceManager;
+
+/**
+ * Helpers for inspecting DataNucleus L1 (persistence context) cache in unit tests.
+ */
+public final class ExecutionContextTestUtils {
+
+  private ExecutionContextTestUtils() {
+  }
+
+  public static ExecutionContext getExecutionContext(PersistenceManager pm) {
+    if (pm instanceof JDOPersistenceManager) {
+      return ((JDOPersistenceManager) pm).getExecutionContext();
+    }
+    if (pm instanceof PersistenceManagerProxy.ExecutionContextReference) {
+      return ((PersistenceManagerProxy.ExecutionContextReference) pm).getExecutionContext();
+    }
+    throw new IllegalArgumentException("Unsupported PersistenceManager: " + pm.getClass());
+  }
+
+  public static int countCachedInstances(PersistenceManager pm, Class<?> clazz) {
+    Level1Cache l1Cache = getExecutionContext(pm).getLevel1Cache();
+    int count = 0;
+    for (DNStateManager stateManager : l1Cache.values()) {
+      if (clazz.isInstance(stateManager.getObject())) {
+        count++;
+      }
+    }
+    return count;
+  }
+}