@Override public Table alterTable(String catName, String dbName, String tblName, Table newTable, String validWriteIds)
      throws InvalidObjectException, MetaException {
    newTable = rawStore.alterTable(catName, dbName, tblName, newTable, validWriteIds);
    // in case of event based cache update, cache will be updated during commit.
    if (canUseEvents) {
      return newTable;
    }
    catName = normalizeIdentifier(catName);
    dbName = normalizeIdentifier(dbName);
    tblName = normalizeIdentifier(tblName);
    String newTblName = normalizeIdentifier(newTable.getTableName());
    if (!shouldCacheTable(catName, dbName, tblName) && !shouldCacheTable(catName, dbName, newTblName)) {
      return newTable;
    }
    Table tbl = sharedCache.getTableFromCache(catName, dbName, tblName);
    if (tbl == null) {
      // The table is not yet loaded in cache
      return newTable;
    }
    if (shouldCacheTable(catName, dbName, tblName) && shouldCacheTable(catName, dbName, newTblName)) {
      // If old table is in the cache and the new table can also be cached
      sharedCache.alterTableInCache(catName, dbName, tblName, newTable);
    } else if (!shouldCacheTable(catName, dbName, tblName) && shouldCacheTable(catName, dbName, newTblName)) {
      // If old table is *not* in the cache but the new table can be cached
      sharedCache.addTableToCache(catName, dbName, newTblName, newTable);
    } else if (shouldCacheTable(catName, dbName, tblName) && !shouldCacheTable(catName, dbName, newTblName)) {
      // If old table is in the cache but the new table *cannot* be cached
      sharedCache.removeTableFromCache(catName, dbName, tblName);
    }
    return newTable;
  }
@Test public void testAlterTable() throws Exception {
    Configuration conf = MetastoreConf.newMetastoreConf();
    MetastoreConf.setBoolVar(conf, MetastoreConf.ConfVars.HIVE_IN_TEST, true);
    MetastoreConf.setVar(conf, MetastoreConf.ConfVars.CACHED_RAW_STORE_MAX_CACHE_MEMORY, "-1Kb");
    MetaStoreTestUtils.setConfForStandloneMode(conf);
    CachedStore cachedStore = new CachedStore();
    CachedStore.clearSharedCache();
    cachedStore.setConfForTest(conf);
    ObjectStore objectStore = (ObjectStore) cachedStore.getRawStore();
    // Prewarm CachedStore
    CachedStore.setCachePrewarmedState(false);
    CachedStore.prewarm(objectStore);
    List<String> db1Tables = cachedStore.getAllTables(DEFAULT_CATALOG_NAME, db1.getName());
    Assert.assertEquals(2, db1Tables.size());
    List<String> db2Tables = cachedStore.getAllTables(DEFAULT_CATALOG_NAME, db2.getName());
    Assert.assertEquals(2, db2Tables.size());
    // Alter table db1Utbl1 via CachedStore and read via ObjectStore
    Table db1Utbl1Read = cachedStore.getTable(DEFAULT_CATALOG_NAME, db1Utbl1.getDbName(), db1Utbl1.getTableName());
    String newOwner = "newOwner";
    Table db1Utbl1ReadAlt = new Table(db1Utbl1Read);
    db1Utbl1ReadAlt.setOwner(newOwner);
    cachedStore
        .alterTable(DEFAULT_CATALOG_NAME, db1Utbl1Read.getDbName(), db1Utbl1Read.getTableName(), db1Utbl1ReadAlt, "0");
    db1Utbl1Read =
        cachedStore.getTable(DEFAULT_CATALOG_NAME, db1Utbl1ReadAlt.getDbName(), db1Utbl1ReadAlt.getTableName());
    Table db1Utbl1ReadOS =
        objectStore.getTable(DEFAULT_CATALOG_NAME, db1Utbl1ReadAlt.getDbName(), db1Utbl1ReadAlt.getTableName());
    Assert.assertEquals(db1Utbl1Read, db1Utbl1ReadOS);
    // Alter table db2Utbl1 via ObjectStore and read via CachedStore
    Table db2Utbl1Read = objectStore.getTable(DEFAULT_CATALOG_NAME, db2Utbl1.getDbName(), db2Utbl1.getTableName());
    Table db2Utbl1ReadAlt = new Table(db2Utbl1Read);
    db2Utbl1ReadAlt.setOwner(newOwner);
    objectStore
        .alterTable(DEFAULT_CATALOG_NAME, db2Utbl1Read.getDbName(), db2Utbl1Read.getTableName(), db2Utbl1ReadAlt, "0");
    updateCache(cachedStore);
    db2Utbl1Read =
        objectStore.getTable(DEFAULT_CATALOG_NAME, db2Utbl1ReadAlt.getDbName(), db2Utbl1ReadAlt.getTableName());
    Table d21Utbl1ReadCS =
        cachedStore.getTable(DEFAULT_CATALOG_NAME, db2Utbl1ReadAlt.getDbName(), db2Utbl1ReadAlt.getTableName());
    Assert.assertEquals(db2Utbl1Read, d21Utbl1ReadCS);
    cachedStore.shutdown();
  }