diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java index 9a42ba9f01a5..f96d03a4889f 100644 --- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java +++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java @@ -420,12 +420,8 @@ public T unwrap(Class 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 openQueries = new LinkedList<>(); if (simpl instanceof RawStoreBundle rsb) { diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/iface/NotificationStore.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/iface/NotificationStore.java index a1d1a1db680d..7b8dbbf2af19 100644 --- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/iface/NotificationStore.java +++ b/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); /** diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/ColStatsStoreImpl.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/ColStatsStoreImpl.java index 874209c70fb6..2d47482a448f 100644 --- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/ColStatsStoreImpl.java +++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/ColStatsStoreImpl.java @@ -483,7 +483,6 @@ private List getMTableColumnStatistics(Table table, List validateTableCols(table, colNames); List result = Collections.emptyList(); - Query query = pm.newQuery(MTableColumnStatistics.class); result = Batchable.runBatched(batchSize, colNames, new Batchable() { @Override @@ -504,6 +503,7 @@ public List run(List input) params[i + 4] = input.get(i); } filter.append(")"); + Query query = pm.newQuery(MTableColumnStatistics.class); query.setFilter(filter.toString()); query.declareParameters(paramStr.toString()); List paritial = (List) query.executeWithArray(params); diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/NotificationStoreImpl.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/NotificationStoreImpl.java index b7159dfda158..4006212dd8a0 100644 --- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/NotificationStoreImpl.java +++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/NotificationStoreImpl.java @@ -247,16 +247,15 @@ private void cleanOlderEvents(int olderThan, Class table, String tableName) { final Optional 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(); @@ -264,6 +263,21 @@ private void cleanOlderEvents(int olderThan, Class table, String tableName) { TimeUnit.NANOSECONDS.toMillis(finish - start)); } + private int cleanNotificationEventsBatch(final int ageSec, final Optional batchSize, + Class 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 int doCleanNotificationEvents(final int ageSec, final Optional batchSize, Class tableClass, String tableName) { int eventsCount = 0; @@ -308,6 +322,7 @@ private int doCleanNotificationEvents(final int ageSec, final Optional 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(); } diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/TableStoreImpl.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/TableStoreImpl.java index 82900c08fa65..28ab0cc66190 100644 --- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/TableStoreImpl.java +++ b/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 part_vals, Par String dbname = normalizeIdentifier(tableName.getDb()); String name = normalizeIdentifier(tableName.getTable()); AtomicReference 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 part_vals, Partition newPart, String validWriteIds, AtomicReference 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 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(this, null) { @@ -2003,8 +1995,9 @@ private List alterPartitionsViaJdo(MTable table, List partNam mPartitionList = (List) 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, MPartition> mPartsMap = new HashMap(); @@ -2016,8 +2009,9 @@ private List alterPartitionsViaJdo(MTable table, List partNam AtomicReference 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()); diff --git a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/ExecutionContextTestUtils.java b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/ExecutionContextTestUtils.java new file mode 100644 index 000000000000..1423788e64bf --- /dev/null +++ b/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; + } +} diff --git a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/metastore/TestPersistenceContextEviction.java b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/metastore/TestPersistenceContextEviction.java new file mode 100644 index 000000000000..d798e238c758 --- /dev/null +++ b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/metastore/TestPersistenceContextEviction.java @@ -0,0 +1,124 @@ +/* + * 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.metastore; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hive.metastore.ExecutionContextTestUtils; +import org.apache.hadoop.hive.metastore.HMSHandler; +import org.apache.hadoop.hive.metastore.MetaStoreTestUtils; +import org.apache.hadoop.hive.metastore.ObjectStore; +import org.apache.hadoop.hive.metastore.Warehouse; +import org.apache.hadoop.hive.metastore.annotation.MetastoreUnitTest; +import org.apache.hadoop.hive.metastore.api.MetaException; +import org.apache.hadoop.hive.metastore.api.NotificationEvent; +import org.apache.hadoop.hive.metastore.api.NotificationEventRequest; +import org.apache.hadoop.hive.metastore.api.NotificationEventResponse; +import org.apache.hadoop.hive.metastore.conf.MetastoreConf; +import org.apache.hadoop.hive.metastore.conf.MetastoreConf.ConfVars; +import org.apache.hadoop.hive.metastore.messaging.EventMessage; +import org.apache.hadoop.hive.metastore.model.MNotificationLog; +import org.apache.hadoop.hive.metastore.utils.MetaStoreServerUtils; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.categories.Category; + +import javax.jdo.PersistenceManager; +import javax.jdo.Query; +import java.util.List; +import java.util.UUID; + +/** + * Verifies batched metastore operations evict loaded JDO entities from the persistence context. + * Without eviction, long-lived RawStore instances (for example the DB notification cleaner thread) + * accumulate deleted entities in the L1 cache and can OOM. + */ +@Category(MetastoreUnitTest.class) +public class TestPersistenceContextEviction { + private static final int BATCH_SIZE = 3; + private static final int NUM_EVENTS = 12; + + private ObjectStore objectStore; + private Configuration conf; + private PersistenceManager pm; + + @Before + public void setUp() throws Exception { + conf = MetastoreConf.newMetastoreConf(); + MetastoreConf.setBoolVar(conf, ConfVars.HIVE_IN_TEST, true); + MetastoreConf.setLongVar(conf, MetastoreConf.ConfVars.EVENT_CLEAN_MAX_EVENTS, BATCH_SIZE); + MetastoreConf.setLongVar(conf, ConfVars.RAWSTORE_PARTITION_BATCH_SIZE, BATCH_SIZE); + MetaStoreTestUtils.setConfForStandloneMode(conf); + + String currentUrl = MetastoreConf.getVar(conf, ConfVars.CONNECT_URL_KEY); + currentUrl = currentUrl.replace(MetaStoreServerUtils.JUNIT_DATABASE_PREFIX, + String.format("%s_%s", MetaStoreServerUtils.JUNIT_DATABASE_PREFIX, UUID.randomUUID())); + MetastoreConf.setVar(conf, ConfVars.CONNECT_URL_KEY, currentUrl); + + objectStore = new ObjectStore(); + objectStore.setConf(conf); + HMSHandler.createDefaultCatalog(objectStore, new Warehouse(conf)); + pm = objectStore.createRawStoreBundle().getPersistentManager(); + } + + @Test + public void testExecutionContextCountsLoadedNotificationEvents() throws MetaException { + insertNotificationEvents(5, "payload"); + + objectStore.openTransaction(); + try { + Query query = pm.newQuery(MNotificationLog.class); + List events = (List) query.execute(); + pm.retrieveAll(events); + Assert.assertTrue("expected loaded events to remain in the persistence context", + ExecutionContextTestUtils.countCachedInstances(pm, MNotificationLog.class) >= 5); + } finally { + objectStore.rollbackTransaction(); + } + + Assert.assertEquals(0, ExecutionContextTestUtils.countCachedInstances(pm, MNotificationLog.class)); + } + + @Test + public void testCleanNotificationEventsEvictsCachedEntities() throws MetaException { + insertNotificationEvents(NUM_EVENTS, "x".repeat(50)); + + objectStore.openTransaction(); + try { + objectStore.cleanNotificationEvents(0); + Assert.assertEquals("batched notification cleanup retains deleted events in the L1 cache", NUM_EVENTS, + ExecutionContextTestUtils.countCachedInstances(pm, MNotificationLog.class)); + } finally { + objectStore.commitTransaction(); + } + + Assert.assertEquals("batched notification cleanup must not retain deleted events in the L1 cache", 0, + ExecutionContextTestUtils.countCachedInstances(pm, MNotificationLog.class)); + NotificationEventResponse response = objectStore.getNextNotification(new NotificationEventRequest()); + Assert.assertEquals(0, response.getEventsSize()); + } + + private void insertNotificationEvents(int count, String message) throws MetaException { + for (int i = 0; i < count; i++) { + NotificationEvent event = new NotificationEvent(0, 0, + EventMessage.EventType.CREATE_DATABASE.toString(), message); + objectStore.addNotificationEvent(event); + } + } +}