This is an automated email from the ASF dual-hosted git repository.
dengzhhu653 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new 4dc3c1eb171 HIVE-29701: Fix leaking queries in ObjectStore.unwrap()
(#6585)
4dc3c1eb171 is described below
commit 4dc3c1eb1718cdf71ca6823cbcea8d0c6e130167
Author: dengzh <[email protected]>
AuthorDate: Fri Jul 10 08:49:25 2026 +0800
HIVE-29701: Fix leaking queries in ObjectStore.unwrap() (#6585)
---
.../apache/hadoop/hive/metastore/ObjectStore.java | 28 +-
.../org/apache/hadoop/hive/metastore/RawStore.java | 5 +-
.../hive/metastore/metastore/RawStoreBundle.java | 6 +-
.../metastore/metastore/TransactionHandler.java | 19 ++
.../metastore/impl/ColStatsStoreImpl.java | 32 +-
.../metastore/impl/ConstraintStoreImpl.java | 19 +-
.../metastore/impl/PrivilegeStoreImpl.java | 22 +-
.../metastore/metastore/impl/TableStoreImpl.java | 51 +--
.../metastore/metastore/TestObjectStoreUnwrap.java | 365 +++++++++++++++++++++
9 files changed, 469 insertions(+), 78 deletions(-)
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 c26d2d2de7d..9a42ba9f01a 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
@@ -118,6 +118,7 @@
import org.apache.hadoop.hive.metastore.conf.MetastoreConf.ConfVars;
import org.apache.hadoop.hive.metastore.metastore.GetHelper;
import org.apache.hadoop.hive.metastore.metastore.GetListHelper;
+import org.apache.hadoop.hive.metastore.metastore.PersistenceManagerProxy;
import org.apache.hadoop.hive.metastore.metastore.iface.PrivilegeStore;
import org.apache.hadoop.hive.metastore.model.MCatalog;
import org.apache.hadoop.hive.metastore.model.MColumnDescriptor;
@@ -142,7 +143,6 @@
import org.apache.hadoop.hive.metastore.model.MReplicationMetrics;
import org.apache.hadoop.hive.metastore.properties.CachingPropertyStore;
import org.apache.hadoop.hive.metastore.properties.PropertyStore;
-import org.apache.hadoop.hive.metastore.metastore.PersistenceManagerProxy;
import org.apache.hadoop.hive.metastore.metastore.RawStoreBundle;
import org.apache.hadoop.hive.metastore.metastore.MetaDescriptor;
import org.apache.hadoop.hive.metastore.metastore.TransactionHandler;
@@ -207,7 +207,6 @@ private enum TXN_STATUS {
private Transaction currentTransaction = null;
private TXN_STATUS transactionStatus = TXN_STATUS.NO_STATE;
private PropertyStore propertyStore;
- private final Map<Class<?>, Object> cachedImpls = new HashMap<>();
public ObjectStore() {
}
@@ -382,7 +381,6 @@ public void shutdown() {
pm.close();
pm = null;
}
- cachedImpls.clear();
}
/**
@@ -412,27 +410,23 @@ public boolean openTransaction() {
return result;
}
- @SuppressWarnings("unchecked")
+ @SuppressWarnings("unchecked, rawtypes")
@Override
public <T> T unwrap(Class<T> iface) {
+ if (pm == null) {
+ throw new IllegalStateException("ObjectStore hasn't been initialized
yet");
+ }
MetaDescriptor descriptor = iface.getAnnotation(MetaDescriptor.class);
if (descriptor == null) {
throw new IllegalArgumentException("Unable to unwrap the store as " +
iface);
}
- String implClassName = conf.get("metastore." + descriptor.alias() +
".store.impl", "");
- T simpl;
- T impl = (T) cachedImpls.get(iface);
- if (impl != null &&
- (StringUtils.isEmpty(implClassName) ||
impl.getClass().getName().equals(implClassName))) {
- simpl = impl;
- } else {
- Class<?> ifaceImpl = descriptor.defaultImpl();
- if (StringUtils.isNotEmpty(implClassName)) {
- ifaceImpl = conf.getClass(implClassName, ifaceImpl);
- }
- simpl = (T) JavaUtils.newInstance(ifaceImpl);
- cachedImpls.put(iface, simpl);
+ String implClassName =
+ conf.get("metastore." + descriptor.alias() + ".store.impl", "");
+ Class<?> ifaceImpl = descriptor.defaultImpl();
+ if (StringUtils.isNotEmpty(implClassName)) {
+ ifaceImpl = conf.getClass(implClassName, ifaceImpl);
}
+ T simpl = (T) JavaUtils.newInstance(ifaceImpl);
List<Query> openQueries = new LinkedList<>();
if (simpl instanceof RawStoreBundle rsb) {
rsb.setBaseStore(this);
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/RawStore.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/RawStore.java
index ca9211f27d8..45f1a32bb4a 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/RawStore.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/RawStore.java
@@ -2475,10 +2475,6 @@ default MTable ensureGetMTable(String catName, String
dbName, String tblName) th
MDatabase ensureGetMDatabase(String catName, String dbName) throws
NoSuchObjectException;
- default MPartition ensureGetMPartition(TableName tableName, List<String>
partVals) throws MetaException {
- return unwrap(TableStore.class).ensureGetMPartition(tableName, partVals);
- }
-
/** Persistent Property Management. */
default PropertyStore getPropertyStore() {
return null;
@@ -2503,4 +2499,5 @@ default long updateParameterWithExpectedValue(Table
table, String key, String ex
* @throws RuntimeException If the context cannot be unwrapped to the
provided class
*/
<T> T unwrap(Class<T> iface);
+
}
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/RawStoreBundle.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/RawStoreBundle.java
index add2f07dfd2..66a5b4e6fb0 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/RawStoreBundle.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/RawStoreBundle.java
@@ -33,7 +33,7 @@ public void setBaseStore(RawStore store) {
}
public void setPersistentManager(PersistenceManager manager) {
- this.pm = Objects.requireNonNull(manager);
+ this.pm = manager;
}
public RawStore getBaseStore() {
@@ -43,4 +43,8 @@ public RawStore getBaseStore() {
public PersistenceManager getPersistentManager() {
return pm;
}
+
+ protected <T> T siblingStore(Class<T> iface) {
+ return baseStore.unwrap(iface);
+ }
}
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/TransactionHandler.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/TransactionHandler.java
index afde12e2231..512b815870b 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/TransactionHandler.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/TransactionHandler.java
@@ -28,6 +28,8 @@
import java.util.Arrays;
import java.util.List;
+import com.google.common.annotations.VisibleForTesting;
+
import org.apache.hadoop.hive.metastore.RawStore;
public record TransactionHandler<T> (RawStore rs, T simpl, List<Query>
openQueries)
@@ -44,6 +46,9 @@ public static <T> T getProxy(Class<T> iface,
TransactionHandler<T> handler) {
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws
Throwable {
+ if (method.getDeclaringClass() == Object.class) {
+ return handleObjectMethod(proxy, method, args);
+ }
boolean openTxn = method.getAnnotation(MetaDescriptor.NoTransaction.class)
== null;
boolean success = false;
if (openTxn) {
@@ -67,4 +72,18 @@ public Object invoke(Object proxy, Method method, Object[]
args) throws Throwabl
openQueries.clear();
}
}
+
+ @VisibleForTesting
+ public int getOpenQueryCount() {
+ return openQueries.size();
+ }
+
+ private static Object handleObjectMethod(Object proxy, Method method,
Object[] args) {
+ return switch (method.getName()) {
+ case "toString" -> proxy.getClass().getInterfaces()[0].getName() + "
proxy";
+ case "hashCode" -> System.identityHashCode(proxy);
+ case "equals" -> proxy == args[0];
+ default -> throw new UnsupportedOperationException(method.toString());
+ };
+ }
}
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 3a166afa1f5..874209c70fb 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
@@ -271,7 +271,9 @@ public Map<String, String>
updateTableColumnStatistics(ColumnStatistics colStats
String catName = statsDesc.isSetCatName() ? statsDesc.getCatName() :
getDefaultCatalog(conf);
// DataNucleus objects get detached all over the place for no (real)
reason.
// So let's not use them anywhere unless absolutely necessary.
- MTable mTable = baseStore.ensureGetMTable(catName, statsDesc.getDbName(),
statsDesc.getTableName());
+ TableStore tableStore = siblingStore(TableStore.class);
+ MTable mTable = tableStore
+ .ensureGetMTable(new TableName(catName, statsDesc.getDbName(),
statsDesc.getTableName()));
int maxRetries = MetastoreConf.getIntVar(conf,
MetastoreConf.ConfVars.METASTORE_S4U_NOWAIT_MAX_RETRIES);
long sleepInterval = MetastoreConf.getTimeVar(conf,
MetastoreConf.ConfVars.METASTORE_S4U_NOWAIT_RETRY_SLEEP_INTERVAL,
TimeUnit.MILLISECONDS);
@@ -292,7 +294,7 @@ public Map<String, String>
updateTableColumnStatistics(ColumnStatistics colStats
throw new RetryingExecutor.RetryException(exceptionRef.get());
}
pm.refresh(mTable);
- Table table = baseStore.unwrap(TableStore.class).getTable(new
TableName(catName, statsDesc.getDbName(),
+ Table table = tableStore.getTable(new TableName(catName,
statsDesc.getDbName(),
statsDesc.getTableName()), null, -1);
List<String> colNames = new ArrayList<>();
for (ColumnStatisticsObj statsObj : statsObjs) {
@@ -357,8 +359,9 @@ public Map<String, String>
updatePartitionColumnStatistics(Table table, MTable m
List<ColumnStatisticsObj> statsObjs = colStats.getStatsObj();
ColumnStatisticsDesc statsDesc = colStats.getStatsDesc();
String catName = statsDesc.isSetCatName() ? statsDesc.getCatName() :
getDefaultCatalog(conf);
- MPartition mPartition =
- baseStore.ensureGetMPartition(new TableName(catName,
statsDesc.getDbName(), statsDesc.getTableName()), partVals);
+ TableStore tableStore = siblingStore(TableStore.class);
+ MPartition mPartition = tableStore
+ .ensureGetMPartition(new TableName(catName, statsDesc.getDbName(),
statsDesc.getTableName()), partVals);
if (mPartition == null) {
throw new NoSuchObjectException("Partition for which stats is gathered
doesn't exist.");
}
@@ -387,7 +390,7 @@ public Map<String, String>
updatePartitionColumnStatistics(Table table, MTable m
throw new RetryingExecutor.RetryException(exceptionRef.get());
}
pm.refresh(mPartition);
- Partition partition = baseStore.unwrap(TableStore.class)
+ Partition partition = tableStore
.getPartition(new TableName(catName, statsDesc.getDbName(),
statsDesc.getTableName()), mPartition.getValues(), null);
Map<String, MPartitionColumnStatistics> oldStats = Maps.newHashMap();
List<MPartitionColumnStatistics> stats =
@@ -566,7 +569,7 @@ public ColumnStatistics getTableColumnStatistics(
String dbName = normalizeIdentifier(tableName.getDb());
String tblName = normalizeIdentifier(tableName.getTable());
if (writeIdList != null) {
- MTable table = baseStore.ensureGetMTable(catName, dbName, tblName);
+ MTable table = siblingStore(TableStore.class).ensureGetMTable(new
TableName(catName, dbName, tblName));
isCompliant = !TxnUtils.isTransactionalTable(table.getParameters())
|| (areTxnStatsSupported &&
isCurrentStatsValidForTheQuery(table.getParameters(), table.getWriteId(),
writeIdList, false));
}
@@ -678,9 +681,10 @@ public List<ColumnStatistics> getPartitionColumnStatistics(
cs.setIsStatsCompliant(false);
}
} else {
+ TableStore tableStore = siblingStore(TableStore.class);
// TODO: this could be improved to get partitions in bulk
for (ColumnStatistics cs : allStats) {
- MPartition mpart = baseStore.ensureGetMPartition(tableName,
+ MPartition mpart = tableStore.ensureGetMPartition(tableName,
Warehouse.getPartValuesFromPartName(cs.getStatsDesc().getPartName()));
if (mpart == null
|| !isCurrentStatsValidForTheQuery(mpart.getParameters(),
mpart.getWriteId(), writeIdList, false)) {
@@ -758,7 +762,8 @@ public AggrStats get_aggr_stats_for(TableName tableName,
return null;
}
- Table table = baseStore.unwrap(TableStore.class).getTable(tableName,
null, -1);
+ TableStore tableStore = siblingStore(TableStore.class);
+ Table table = tableStore.getTable(tableName, null, -1);
boolean isTxn = TxnUtils.isTransactionalTable(table.getParameters());
if (isTxn && !areTxnStatsSupported) {
return null;
@@ -769,7 +774,7 @@ public AggrStats get_aggr_stats_for(TableName tableName,
GetProjectionsSpec ps = new GetProjectionsSpec();
ps.setIncludeParamKeyPattern(StatsSetupConst.COLUMN_STATS_ACCURATE +
'%');
ps.setFieldList(Lists.newArrayList("writeId", "parameters", "values"));
- List<Partition> parts = baseStore.unwrap(TableStore.class)
+ List<Partition> parts = tableStore
.getPartitionSpecsByFilterAndProjection(table, ps, fs);
// Loop through the given "partNames" list
@@ -915,7 +920,6 @@ public void deleteAllPartitionColumnStatistics(TableName
tn, String writeIdList)
if (tableName == null) {
throw new RuntimeException("Table name is null.");
}
- MTable mTable = baseStore.ensureGetMTable(catName, dbName, tableName);
Query query = pm.newQuery(MPartitionColumnStatistics.class);
String filter = "partition.table.database.name == t2 &&
partition.table.tableName == t3 && partition.table.database.catalogName == t4";
String parameters = "java.lang.String t2, java.lang.String t3,
java.lang.String t4";
@@ -942,12 +946,12 @@ protected Integer getSqlResult() throws MetaException {
@Override
protected Integer getJdoResult() throws MetaException,
NoSuchObjectException {
try {
- List<Partition> parts = baseStore.unwrap(TableStore.class)
- .getPartitions(tn, GetPartitionsArgs.getAllPartitions());
+ TableStore tableStore = siblingStore(TableStore.class);
+ List<Partition> parts = tableStore.getPartitions(tn,
GetPartitionsArgs.getAllPartitions());
for (Partition part : parts) {
Partition newPart = new Partition(part);
StatsSetupConst.clearColumnStatsState(newPart.getParameters());
- baseStore.unwrap(TableStore.class).alterPartition(tn,
part.getValues(), newPart, writeIdList);
+ tableStore.alterPartition(tn, part.getValues(), newPart,
writeIdList);
}
return parts.size();
} catch (InvalidObjectException e) {
@@ -1133,7 +1137,7 @@ private boolean
deleteTableColumnStatisticsViaJdo(TableName tblName, List<String
pm.deletePersistentAll(mStatsObjColl);
}
- MTable mTable = baseStore.ensureGetMTable(catName, dbName, tableName);
+ MTable mTable = siblingStore(TableStore.class).ensureGetMTable(new
TableName(catName, dbName, tableName));
if (mTable != null) {
Map<String, String> tableParams = mTable.getParameters();
if (tableParams != null &&
tableParams.containsKey(StatsSetupConst.COLUMN_STATS_ACCURATE)) {
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/ConstraintStoreImpl.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/ConstraintStoreImpl.java
index d76cd8ff659..def4c89fe5c 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/ConstraintStoreImpl.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/ConstraintStoreImpl.java
@@ -77,7 +77,7 @@ public void setBaseStore(RawStore store) {
@Override
public SQLAllTableConstraints createTableWithConstraints(Table tbl,
SQLAllTableConstraints constraints)
throws InvalidObjectException, MetaException {
- baseStore.unwrap(TableStore.class).createTable(tbl);
+ siblingStore(TableStore.class).createTable(tbl);
// Add constraints.
// We need not do a deep retrieval of the Table Column Descriptor while
persisting the
// constraints since this transaction involving create table is not yet
committed.
@@ -184,6 +184,7 @@ public List<SQLForeignKey> addForeignKeys(
private List<SQLForeignKey> addForeignKeys(List<SQLForeignKey> foreignKeys,
boolean retrieveCD,
List<SQLPrimaryKey> primaryKeys, List<SQLUniqueConstraint>
uniqueConstraints)
throws InvalidObjectException, MetaException {
+ TableStore tableStore = siblingStore(TableStore.class);
if (CollectionUtils.isNotEmpty(foreignKeys)) {
List<MConstraint> mpkfks = new ArrayList<>();
String currentConstraintName = null;
@@ -208,7 +209,7 @@ private List<SQLForeignKey>
addForeignKeys(List<SQLForeignKey> foreignKeys, bool
// If retrieveCD is false, we do not need to do a deep retrieval of
the Table Column Descriptor.
// For instance, this is the case when we are creating the table.
final TableStore.AttachedMTableInfo nChildTable =
- baseStore.unwrap(TableStore.class).getMTable(new
TableName(catName, fkTableDB, fkTableName), retrieveCD);
+ tableStore.getMTable(new TableName(catName, fkTableDB,
fkTableName), retrieveCD);
final MTable childTable = nChildTable.mtbl;
if (childTable == null) {
throw new InvalidObjectException("Child table not found: " +
fkTableName);
@@ -242,7 +243,7 @@ private List<SQLForeignKey>
addForeignKeys(List<SQLForeignKey> foreignKeys, bool
existingTableUniqueConstraints = uniqueConstraints;
} else {
nParentTable =
- baseStore.unwrap(TableStore.class).getMTable(new
TableName(catName, pkTableDB, pkTableName), true);
+ tableStore.getMTable(new TableName(catName, pkTableDB,
pkTableName), true);
parentTable = nParentTable.mtbl;
if (parentTable == null) {
throw new InvalidObjectException("Parent table not found: " +
pkTableName);
@@ -441,7 +442,7 @@ private List<SQLPrimaryKey>
addPrimaryKeys(List<SQLPrimaryKey> pks, boolean retr
MetaException {
List<MConstraint> mpks = new ArrayList<>();
String constraintName = null;
-
+ TableStore tableStore = siblingStore(TableStore.class);
for (SQLPrimaryKey pk : pks) {
final String catName = normalizeIdentifier(pk.getCatName());
final String tableDB = normalizeIdentifier(pk.getTable_db());
@@ -450,7 +451,7 @@ private List<SQLPrimaryKey>
addPrimaryKeys(List<SQLPrimaryKey> pks, boolean retr
// If retrieveCD is false, we do not need to do a deep retrieval of the
Table Column Descriptor.
// For instance, this is the case when we are creating the table.
- TableStore.AttachedMTableInfo nParentTable =
baseStore.unwrap(TableStore.class)
+ TableStore.AttachedMTableInfo nParentTable = tableStore
.getMTable(new TableName(catName, tableDB, tableName), retrieveCD);
MTable parentTable = nParentTable.mtbl;
if (parentTable == null) {
@@ -527,6 +528,7 @@ private List<SQLUniqueConstraint>
addUniqueConstraints(List<SQLUniqueConstraint>
List<MConstraint> cstrs = new ArrayList<>();
String constraintName = null;
+ TableStore tableStore = siblingStore(TableStore.class);
for (SQLUniqueConstraint uk : uks) {
final String catName = normalizeIdentifier(uk.getCatName());
final String tableDB = normalizeIdentifier(uk.getTable_db());
@@ -535,7 +537,7 @@ private List<SQLUniqueConstraint>
addUniqueConstraints(List<SQLUniqueConstraint>
// If retrieveCD is false, we do not need to do a deep retrieval of the
Table Column Descriptor.
// For instance, this is the case when we are creating the table.
- TableStore.AttachedMTableInfo nParentTable =
baseStore.unwrap(TableStore.class)
+ TableStore.AttachedMTableInfo nParentTable = tableStore
.getMTable(new TableName(catName, tableDB, tableName), retrieveCD);
MTable parentTable = nParentTable.mtbl;
if (parentTable == null) {
@@ -652,7 +654,7 @@ private MConstraint addConstraint(String catName, String
tableDB, String tableNa
String constraintName = null;
// If retrieveCD is false, we do not need to do a deep retrieval of the
Table Column Descriptor.
// For instance, this is the case when we are creating the table.
- TableStore.AttachedMTableInfo nParentTable =
baseStore.unwrap(TableStore.class)
+ TableStore.AttachedMTableInfo nParentTable = siblingStore(TableStore.class)
.getMTable(new TableName(catName, tableDB, tableName), retrieveCD);
MTable parentTable = nParentTable.mtbl;
if (parentTable == null) {
@@ -735,6 +737,7 @@ private List<SQLNotNullConstraint>
addNotNullConstraints(List<SQLNotNullConstrai
List<MConstraint> cstrs = new ArrayList<>();
String constraintName;
+ TableStore tableStore = siblingStore(TableStore.class);
for (SQLNotNullConstraint nn : nns) {
final String catName = normalizeIdentifier(nn.getCatName());
final String tableDB = normalizeIdentifier(nn.getTable_db());
@@ -743,7 +746,7 @@ private List<SQLNotNullConstraint>
addNotNullConstraints(List<SQLNotNullConstrai
// If retrieveCD is false, we do not need to do a deep retrieval of the
Table Column Descriptor.
// For instance, this is the case when we are creating the table.
- TableStore.AttachedMTableInfo nParentTable =
baseStore.unwrap(TableStore.class)
+ TableStore.AttachedMTableInfo nParentTable = tableStore
.getMTable(new TableName(catName, tableDB, tableName), retrieveCD);
MTable parentTable = nParentTable.mtbl;
if (parentTable == null) {
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/PrivilegeStoreImpl.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/PrivilegeStoreImpl.java
index 17f58867e5b..8a42643ee9e 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/PrivilegeStoreImpl.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/PrivilegeStoreImpl.java
@@ -760,6 +760,7 @@ public boolean grantPrivileges(PrivilegeBag privileges)
throws InvalidObjectExce
if (CollectionUtils.isNotEmpty(privilegeList)) {
Iterator<HiveObjectPrivilege> privIter = privilegeList.iterator();
Set<String> privSet = new HashSet<>();
+ TableStore tableStore = siblingStore(TableStore.class);
while (privIter.hasNext()) {
HiveObjectPrivilege privDef = privIter.next();
HiveObjectRef hiveObject = privDef.getHiveObject();
@@ -836,8 +837,8 @@ public boolean grantPrivileges(PrivilegeBag privileges)
throws InvalidObjectExce
persistentObjs.add(mDc);
}
} else if (hiveObject.getObjectType() == HiveObjectType.TABLE) {
- MTable tblObj = baseStore.ensureGetMTable(catName,
hiveObject.getDbName(), hiveObject
- .getObjectName());
+ MTable tblObj = tableStore
+ .ensureGetMTable(new TableName(catName, hiveObject.getDbName(),
hiveObject.getObjectName()));
if (tblObj != null) {
List<MTablePrivilege> tablePrivs = this
.listAllMTableGrants(userName, principalType,
@@ -862,7 +863,7 @@ public boolean grantPrivileges(PrivilegeBag privileges)
throws InvalidObjectExce
}
}
} else if (hiveObject.getObjectType() == HiveObjectType.PARTITION) {
- MPartition partObj = baseStore.ensureGetMPartition(new
TableName(catName, hiveObject.getDbName(),
+ MPartition partObj = tableStore.ensureGetMPartition(new
TableName(catName, hiveObject.getDbName(),
hiveObject.getObjectName()), hiveObject.getPartValues());
String partName = null;
if (partObj != null) {
@@ -891,14 +892,15 @@ public boolean grantPrivileges(PrivilegeBag privileges)
throws InvalidObjectExce
}
}
} else if (hiveObject.getObjectType() == HiveObjectType.COLUMN) {
- MTable tblObj = baseStore.ensureGetMTable(catName,
hiveObject.getDbName(), hiveObject
- .getObjectName());
+ MTable tblObj = tableStore
+ .ensureGetMTable(new TableName(catName, hiveObject.getDbName(),
hiveObject.getObjectName()));
if (tblObj != null) {
if (hiveObject.getPartValues() != null) {
MPartition partObj = null;
List<MPartitionColumnPrivilege> colPrivs = null;
- partObj = baseStore.ensureGetMPartition(new TableName(catName,
hiveObject.getDbName(), hiveObject
- .getObjectName()), hiveObject.getPartValues());
+ partObj = tableStore
+ .ensureGetMPartition(new TableName(catName,
hiveObject.getDbName(), hiveObject.getObjectName()),
+ hiveObject.getPartValues());
if (partObj == null) {
continue;
}
@@ -973,7 +975,7 @@ public boolean revokePrivileges(PrivilegeBag privileges,
boolean grantOption)
if (CollectionUtils.isNotEmpty(privilegeList)) {
Iterator<HiveObjectPrivilege> privIter = privilegeList.iterator();
-
+ TableStore tableStore = siblingStore(TableStore.class);
while (privIter.hasNext()) {
HiveObjectPrivilege privDef = privIter.next();
HiveObjectRef hiveObject = privDef.getHiveObject();
@@ -1100,7 +1102,7 @@ public boolean revokePrivileges(PrivilegeBag privileges,
boolean grantOption)
}
} else if (hiveObject.getObjectType() == HiveObjectType.PARTITION) {
boolean found = false;
- Table tabObj = baseStore.unwrap(TableStore.class).getTable(
+ Table tabObj = tableStore.getTable(
new TableName(catName, hiveObject.getDbName(),
hiveObject.getObjectName()), null, -1);
String partName = null;
if (hiveObject.getPartValues() != null) {
@@ -1133,7 +1135,7 @@ public boolean revokePrivileges(PrivilegeBag privileges,
boolean grantOption)
}
}
} else if (hiveObject.getObjectType() == HiveObjectType.COLUMN) {
- Table tabObj = baseStore.unwrap(TableStore.class).getTable(
+ Table tabObj = tableStore.getTable(
new TableName(catName, hiveObject.getDbName(),
hiveObject.getObjectName()), null, -1);
String partName = null;
if (hiveObject.getPartValues() != null) {
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 28a34be97b5..e0e183f93f2 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
@@ -90,6 +90,8 @@
import org.apache.hadoop.hive.metastore.client.builder.GetPartitionsArgs;
import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
import org.apache.hadoop.hive.metastore.metastore.RawStoreBundle;
+import org.apache.hadoop.hive.metastore.metastore.iface.ColStatsStore;
+import org.apache.hadoop.hive.metastore.metastore.iface.PrivilegeStore;
import org.apache.hadoop.hive.metastore.model.FetchGroups;
import org.apache.hadoop.hive.metastore.model.MColumnDescriptor;
import org.apache.hadoop.hive.metastore.model.MConstraint;
@@ -194,7 +196,8 @@ public boolean dropTable(TableName table)
}
// delete column statistics if present
- baseStore.deleteTableColumnStatistics(catName, dbName, tableName, null,
null);
+ siblingStore(ColStatsStore.class)
+ .deleteTableColumnStatistics(new TableName(catName, dbName,
tableName), null, null);
List<MConstraint> tabConstraints =
listAllTableConstraintsWithOptionalConstraintName(
catName, dbName, tableName, null);
@@ -584,7 +587,7 @@ private MTable getMTable(String catName, String db, String
table) {
public AttachedMTableInfo getMTable(TableName tableName, boolean retrieveCD)
{
AttachedMTableInfo nmtbl = new AttachedMTableInfo();
String catName =
normalizeIdentifier(Optional.ofNullable(tableName.getCat())
- .orElse(getDefaultCatalog(baseStore.getConf())));
+ .orElse(getDefaultCatalog(conf)));
String db = normalizeIdentifier(tableName.getDb());
String table = normalizeIdentifier(tableName.getTable());
Query query = pm.newQuery(MTable.class,
@@ -616,7 +619,7 @@ public Table getTable(TableName table, String writeIdList,
long tableId)
String dbName = normalizeIdentifier(table.getDb());
String tableName = normalizeIdentifier(table.getTable());
MTable mtable = getMTable(catName, dbName, tableName);
- tbl = convertToTable(mtable, conf);
+ tbl = convertToTable(mtable);
// Retrieve creation metadata if needed
if (tbl != null &&
TableType.MATERIALIZED_VIEW.toString().equals(tbl.getTableType())) {
tbl.setCreationMetadata(
@@ -975,7 +978,7 @@ public Table alterTable(TableName tableName, Table
newTable, String queryValidWr
oldt.setWriteId(newTable.getWriteId());
}
}
- newTable = convertToTable(oldt, conf);
+ newTable = convertToTable(oldt);
return newTable;
}
@@ -1248,7 +1251,7 @@ public List<Table> getTableObjectsByName(String catName,
String db, List<String>
}
} else {
for (Iterator iter = mtables.iterator(); iter.hasNext(); ) {
- Table tbl = convertToTable((MTable) iter.next(), conf);
+ Table tbl = convertToTable((MTable) iter.next());
// Retrieve creation metadata if needed
if (TableType.MATERIALIZED_VIEW.toString().equals(tbl.getTableType()))
{
tbl.setCreationMetadata(
@@ -1450,7 +1453,7 @@ private String makeQueryFilterString(String catName,
String dbName, Table table,
params.put("catName", catName);
}
- tree.accept(new ExpressionTree.JDOFilterGenerator(baseStore.getConf(),
+ tree.accept(new ExpressionTree.JDOFilterGenerator(conf,
table != null ? table.getPartitionKeys() : null, queryBuilder,
params));
if (queryBuilder.hasError()) {
assert !isValidatedFilter;
@@ -1471,7 +1474,7 @@ private String makeQueryFilterString(String catName,
String dbName, String tblNa
params.put("t1", tblName);
params.put("t2", dbName);
params.put("t3", catName);
- tree.accept(new ExpressionTree.JDOFilterGenerator(baseStore.getConf(),
partitionKeys, queryBuilder, params));
+ tree.accept(new ExpressionTree.JDOFilterGenerator(conf, partitionKeys,
queryBuilder, params));
if (queryBuilder.hasError()) {
assert !isValidatedFilter;
LOG.debug("JDO filter pushdown cannot be used: {}",
queryBuilder.getErrorMessage());
@@ -1532,7 +1535,7 @@ private List<String> getPartitionNamesViaOrm(String
catName, String dbName, Stri
*/
private String getDefaultPartitionName(String inputDefaultPartName) {
return (((inputDefaultPartName == null) ||
(inputDefaultPartName.isEmpty()))
- ? MetastoreConf.getVar(baseStore.getConf(),
MetastoreConf.ConfVars.DEFAULTPARTITIONNAME)
+ ? MetastoreConf.getVar(conf,
MetastoreConf.ConfVars.DEFAULTPARTITIONNAME)
: inputDefaultPartName);
}
@@ -1563,7 +1566,7 @@ public List<String> listPartitionNames(TableName
tableName, String defaultPartNa
ExpressionTree tmp = null;
if (!isEmptyFilter) {
tmp = PartFilterExprUtil.makeExpressionTree(expressionProxy, exprBytes,
- getDefaultPartitionName(defaultPartName), baseStore.getConf());
+ getDefaultPartitionName(defaultPartName), conf);
}
String catName = normalizeIdentifier(tableName.getCat());
String dbName = normalizeIdentifier(tableName.getDb());
@@ -1647,7 +1650,7 @@ public boolean getPartitionsByExpr(TableName tableName,
List<Partition> result,
byte[] expr = args.getExpr();
final ExpressionTree exprTree = expr.length != 0 ?
PartFilterExprUtil.makeExpressionTree(
expressionProxy, expr,
- getDefaultPartitionName(args.getDefaultPartName()),
baseStore.getConf()) : ExpressionTree.EMPTY_TREE;
+ getDefaultPartitionName(args.getDefaultPartName()), conf) :
ExpressionTree.EMPTY_TREE;
final AtomicBoolean hasUnknownPartitions = new AtomicBoolean(false);
String catName = normalizeIdentifier(tableName.getCat());
@@ -1664,7 +1667,7 @@ protected List<Partition> getSqlResult() throws
MetaException {
MetaStoreDirectSql.SqlFilterForPushdown filter = new
MetaStoreDirectSql.SqlFilterForPushdown();
if (getDirectSql().generateSqlFilterForPushdown(catName, dbName,
tblName, partitionKeys,
exprTree, args.getDefaultPartName(), filter)) {
- String catalogName = (catName != null) ? catName :
getDefaultCatalog(baseStore.getConf());
+ String catalogName = (catName != null) ? catName :
getDefaultCatalog(conf);
return getDirectSql().getPartitionsViaSqlFilter(catalogName,
dbName, tblName, filter,
isAcidTable, args);
}
@@ -2210,11 +2213,12 @@ public List<Partition>
listPartitionsPsWithAuth(TableName tableName, GetPartitio
partitions = getPartitionsByPs(tableName, args);
}
if (getauth) {
+ PrivilegeStore privilegeStore = siblingStore(PrivilegeStore.class);
for (Partition part : partitions) {
String partName = Warehouse.makePartName(convertToFieldSchemas(mtbl
.getPartitionKeys()), part.getValues());
- PrincipalPrivilegeSet partAuth =
baseStore.getPartitionPrivilegeSet(catName, dbName,
- tblName, partName, userName, groupNames);
+ PrincipalPrivilegeSet partAuth = privilegeStore
+ .getPartitionPrivilegeSet(new TableName(catName, dbName, tblName),
partName, userName, groupNames);
part.setPrivileges(partAuth);
}
}
@@ -2699,8 +2703,8 @@ public Partition getPartitionWithAuth(TableName
tableName, List<String> partVals
if
("TRUE".equalsIgnoreCase(mtbl.getParameters().get("PARTITION_LEVEL_PRIVILEGE")))
{
String partName = Warehouse.makePartName(convertToFieldSchemas(mtbl
.getPartitionKeys()), partVals);
- PrincipalPrivilegeSet partAuth =
baseStore.getPartitionPrivilegeSet(catName, dbName,
- tblName, partName, user_name, group_names);
+ PrincipalPrivilegeSet partAuth = siblingStore(PrivilegeStore.class)
+ .getPartitionPrivilegeSet(new TableName(catName, dbName, tblName),
partName, user_name, group_names);
part.setPrivileges(partAuth);
}
return part;
@@ -2720,7 +2724,7 @@ private String makeQueryFilterString(String catName,
String dbName, MTable mtabl
Map<String, Object> params) throws MetaException {
ExpressionTree tree = (filter != null && !filter.isEmpty())
? PartFilterExprUtil.parseFilterTree(filter) :
ExpressionTree.EMPTY_TREE;
- return makeQueryFilterString(catName, dbName, convertToTable(mtable,
baseStore.getConf()), tree, params, true);
+ return makeQueryFilterString(catName, dbName, convertToTable(mtable),
tree, params, true);
}
@Override
@@ -2749,7 +2753,7 @@ public List<Table>
getAllMaterializedViewObjectsForRewriting(String catName) thr
Collection<MTable> mTbls = (Collection<MTable>) query.executeWithArray(
catName, TableType.MATERIALIZED_VIEW.toString(), true);
for (MTable mTbl : mTbls) {
- Table tbl = convertToTable(mTbl, conf);
+ Table tbl = convertToTable(mTbl);
tbl.setCreationMetadata(
convertToCreationMetadata(
getCreationMetadata(tbl.getCatName(), tbl.getDbName(),
tbl.getTableName())));
@@ -2959,7 +2963,7 @@ private void putPersistentPrivObjects(MTable mtbl,
List<Object> toPersistPrivObj
}
}
- private Table convertToTable(MTable mtbl, Configuration conf) throws
MetaException {
+ private Table convertToTable(MTable mtbl) throws MetaException {
if (mtbl == null) {
return null;
}
@@ -3013,7 +3017,7 @@ private MTable convertToMTable(Table tbl) throws
InvalidObjectException,
return null;
}
MDatabase mdb = null;
- String catName = tbl.isSetCatName() ? tbl.getCatName() :
getDefaultCatalog(baseStore.getConf());
+ String catName = tbl.isSetCatName() ? tbl.getCatName() :
getDefaultCatalog(conf);
try {
mdb = baseStore.ensureGetMDatabase(catName, tbl.getDbName());
} catch (NoSuchObjectException e) {
@@ -3241,7 +3245,7 @@ private MCreationMetadata
convertToMCreationMetadata(CreationMetadata m)
private MMVSource convertToSourceTable(String catalog, SourceTable
sourceTable)
throws NoSuchObjectException {
Table table = sourceTable.getTable();
- MTable mtbl = baseStore.ensureGetMTable(catalog, table.getDbName(),
table.getTableName());
+ MTable mtbl = ensureGetMTable(new TableName(catalog, table.getDbName(),
table.getTableName()));
MMVSource source = new MMVSource();
source.setTable(mtbl);
source.setInsertedCount(sourceTable.getInsertedCount());
@@ -3262,7 +3266,7 @@ private MMVSource convertToSourceTable(String catalog,
SourceTable sourceTable)
private MMVSource convertToSourceTable(String catalog, String
fullyQualifiedTableName)
throws NoSuchObjectException {
String[] names = fullyQualifiedTableName.split("\\.");
- MTable mtbl = baseStore.ensureGetMTable(catalog, names[0], names[1]);
+ MTable mtbl = ensureGetMTable(new TableName(catalog, names[0], names[1]));
MMVSource source = new MMVSource();
source.setTable(mtbl);
source.setInsertedCount(0L);
@@ -3300,9 +3304,8 @@ private SourceTable convertToSourceTable(MMVSource
mmvSource, String catalogName
throws MetaException, NoSuchObjectException {
SourceTable sourceTable = new SourceTable();
MTable mTable = mmvSource.getTable();
- Table table =
- convertToTable(baseStore.ensureGetMTable(catalogName,
mTable.getDatabase().getName(), mTable.getTableName()),
- baseStore.getConf());
+ Table table = convertToTable(
+ ensureGetMTable(new TableName(catalogName,
mTable.getDatabase().getName(), mTable.getTableName())));
sourceTable.setTable(table);
sourceTable.setInsertedCount(mmvSource.getInsertedCount());
sourceTable.setUpdatedCount(mmvSource.getUpdatedCount());
diff --git
a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/metastore/TestObjectStoreUnwrap.java
b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/metastore/TestObjectStoreUnwrap.java
new file mode 100644
index 00000000000..ac5c6ac07b6
--- /dev/null
+++
b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/metastore/TestObjectStoreUnwrap.java
@@ -0,0 +1,365 @@
+/*
+ * 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.common.TableName;
+import org.apache.hadoop.hive.metastore.Deadline;
+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.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.metastore.client.builder.DatabaseBuilder;
+import org.apache.hadoop.hive.metastore.client.builder.GetPartitionsArgs;
+import org.apache.hadoop.hive.metastore.client.builder.PartitionBuilder;
+import org.apache.hadoop.hive.metastore.client.builder.TableBuilder;
+import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
+import org.apache.hadoop.hive.metastore.conf.MetastoreConf.ConfVars;
+import org.apache.hadoop.hive.metastore.metastore.iface.TableStore;
+import org.apache.hadoop.hive.metastore.model.MTable;
+import org.apache.hadoop.hive.metastore.utils.DirectSqlConfigurator;
+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 java.lang.reflect.Proxy;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+
+import static org.apache.hadoop.hive.metastore.Warehouse.DEFAULT_CATALOG_NAME;
+
+@Category(MetastoreUnitTest.class)
+public class TestObjectStoreUnwrap {
+ private static final String DB = "unwrap_proxy_db";
+ private static final String TABLE = "unwrap_proxy_tbl";
+
+ private ObjectStore objectStore;
+ private Configuration conf;
+ private TableName tableName;
+
+ @Before
+ public void setUp() throws Exception {
+ conf = MetastoreConf.newMetastoreConf();
+ MetastoreConf.setBoolVar(conf, ConfVars.HIVE_IN_TEST, true);
+ 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));
+ tableName = new TableName(DEFAULT_CATALOG_NAME, DB, TABLE);
+ createPartitionedTable();
+ }
+
+ @Test
+ public void testQueryClosedAfterRead() throws Exception {
+ TableStore tableStore = objectStore.unwrap(TableStore.class);
+ TransactionHandler<?> handler = getHandler(tableStore);
+ GetPartitionsArgs args = new
GetPartitionsArgs.GetPartitionsArgsBuilder().max(10).build();
+
+ try (AutoCloseable ignored = deadline();
+ DirectSqlConfigurator directSql = new DirectSqlConfigurator(conf,
false)) {
+ List<Partition> partitions = tableStore.getPartitions(tableName, args);
+ Assert.assertEquals(3, partitions.size());
+ }
+
+ Assert.assertEquals(0, handler.getOpenQueryCount());
+ Assert.assertFalse(objectStore.isActiveTransaction());
+ }
+
+ @Test
+ public void testQueryClosedAfterDelete() throws Exception {
+ TableStore tableStore = objectStore.unwrap(TableStore.class);
+ TransactionHandler<?> handler = getHandler(tableStore);
+
+ try (AutoCloseable ignored = deadline();
+ DirectSqlConfigurator directSql = new DirectSqlConfigurator(conf,
false)) {
+ tableStore.dropPartitions(tableName, Arrays.asList("test_part_col=a0",
"test_part_col=a1"));
+ }
+
+ Assert.assertEquals(0, handler.getOpenQueryCount());
+ Assert.assertFalse(objectStore.isActiveTransaction());
+
+ try (AutoCloseable ignored = deadline();
+ DirectSqlConfigurator directSql = new DirectSqlConfigurator(conf,
false)) {
+ Assert.assertEquals(1, tableStore.getPartitions(tableName,
+ new
GetPartitionsArgs.GetPartitionsArgsBuilder().max(10).build()).size());
+ }
+ Assert.assertEquals(0, handler.getOpenQueryCount());
+ }
+
+ @Test
+ public void testRepeatedUnwrapNoLeaks() throws Exception {
+ GetPartitionsArgs args = new
GetPartitionsArgs.GetPartitionsArgsBuilder().max(10).build();
+ try (AutoCloseable ignored = deadline();
+ DirectSqlConfigurator directSql = new DirectSqlConfigurator(conf,
false)) {
+ for (int i = 0; i < 5; i++) {
+ TableStore tableStore = objectStore.unwrap(TableStore.class);
+ TransactionHandler<?> handler = getHandler(tableStore);
+ Assert.assertEquals(3, tableStore.getPartitions(tableName,
args).size());
+ Assert.assertEquals("queries leaked on iteration " + i, 0,
handler.getOpenQueryCount());
+ }
+ }
+ Assert.assertFalse(objectStore.isActiveTransaction());
+ }
+
+ @Test
+ public void testProxiesReadThenDelete() throws Exception {
+ TableStore readProxy = objectStore.unwrap(TableStore.class);
+ TableStore deleteProxy = objectStore.unwrap(TableStore.class);
+ Assert.assertNotSame(readProxy, deleteProxy);
+
+ TransactionHandler<?> readHandler = getHandler(readProxy);
+ TransactionHandler<?> deleteHandler = getHandler(deleteProxy);
+ Assert.assertNotSame(readHandler.openQueries(),
deleteHandler.openQueries());
+
+ GetPartitionsArgs args = new
GetPartitionsArgs.GetPartitionsArgsBuilder().max(10).build();
+
+ try (AutoCloseable ignored = deadline();
+ DirectSqlConfigurator directSql = new DirectSqlConfigurator(conf,
false)) {
+ Assert.assertEquals(3, readProxy.getPartitions(tableName, args).size());
+ Assert.assertEquals(0, readHandler.getOpenQueryCount());
+ Assert.assertEquals(0, deleteHandler.getOpenQueryCount());
+
+ Assert.assertEquals(3, deleteProxy.getPartitions(tableName,
args).size());
+ Assert.assertEquals(0, deleteHandler.getOpenQueryCount());
+ Assert.assertEquals(0, readHandler.getOpenQueryCount());
+
+ deleteProxy.dropPartitions(tableName, Arrays.asList("test_part_col=a0",
"test_part_col=a1"));
+ Assert.assertEquals(0, deleteHandler.getOpenQueryCount());
+
+ Assert.assertEquals(1, readProxy.getPartitions(tableName, args).size());
+ Assert.assertEquals(0, readHandler.getOpenQueryCount());
+ }
+ Assert.assertFalse(objectStore.isActiveTransaction());
+ }
+
+ @Test
+ public void testProxiesInterleavedOps() throws Exception {
+ TableStore proxyA = objectStore.unwrap(TableStore.class);
+ TableStore proxyB = objectStore.unwrap(TableStore.class);
+ TransactionHandler<?> handlerA = getHandler(proxyA);
+ TransactionHandler<?> handlerB = getHandler(proxyB);
+ Assert.assertNotSame(handlerA.openQueries(), handlerB.openQueries());
+ GetPartitionsArgs args = new
GetPartitionsArgs.GetPartitionsArgsBuilder().max(10).build();
+
+ try (AutoCloseable ignored = deadline();
+ DirectSqlConfigurator directSql = new DirectSqlConfigurator(conf,
false)) {
+ Assert.assertEquals(3, proxyA.getPartitions(tableName, args).size());
+ Assert.assertEquals(0, handlerA.getOpenQueryCount());
+
+ Assert.assertEquals(3, proxyB.getPartitions(tableName, args).size());
+ Assert.assertEquals(0, handlerB.getOpenQueryCount());
+
+ proxyA.dropPartitions(tableName,
Collections.singletonList("test_part_col=a0"));
+ Assert.assertEquals(0, handlerA.getOpenQueryCount());
+
+ Assert.assertEquals(2, proxyB.getPartitions(tableName, args).size());
+ Assert.assertEquals(0, handlerB.getOpenQueryCount());
+
+ proxyB.dropPartitions(tableName,
Collections.singletonList("test_part_col=a1"));
+ Assert.assertEquals(0, handlerB.getOpenQueryCount());
+
+ Assert.assertEquals(1, proxyA.getPartitions(tableName, args).size());
+ Assert.assertEquals(0, handlerA.getOpenQueryCount());
+ }
+ Assert.assertFalse(objectStore.isActiveTransaction());
+ }
+
+ @Test
+ public void testProxiesVisibleInTxn() throws Exception {
+ TableStore proxyA = objectStore.unwrap(TableStore.class);
+ TableStore proxyB = objectStore.unwrap(TableStore.class);
+ Assert.assertNotSame(proxyA, proxyB);
+
+ TransactionHandler<?> handlerA = getHandler(proxyA);
+ TransactionHandler<?> handlerB = getHandler(proxyB);
+ Assert.assertNotSame(handlerA.openQueries(), handlerB.openQueries());
+ GetPartitionsArgs args = new
GetPartitionsArgs.GetPartitionsArgsBuilder().max(10).build();
+
+ try (AutoCloseable ignored = deadline();
+ DirectSqlConfigurator directSql = new DirectSqlConfigurator(conf,
false)) {
+ objectStore.openTransaction();
+ try {
+ Assert.assertTrue(objectStore.isActiveTransaction());
+
+ Assert.assertEquals(3, proxyA.getPartitions(tableName, args).size());
+ Assert.assertEquals(0, handlerA.getOpenQueryCount());
+
+ proxyA.dropPartitions(tableName, Arrays.asList("test_part_col=a0",
"test_part_col=a1"));
+ Assert.assertEquals(0, handlerA.getOpenQueryCount());
+ Assert.assertTrue(objectStore.isActiveTransaction());
+
+ // Uncommitted deletes from proxyA are visible to proxyB within the
same transaction.
+ Assert.assertEquals(1, proxyB.getPartitions(tableName, args).size());
+ Assert.assertEquals(0, handlerB.getOpenQueryCount());
+
+ proxyB.dropPartitions(tableName,
Collections.singletonList("test_part_col=a2"));
+ Assert.assertEquals(0, handlerB.getOpenQueryCount());
+
+ // Uncommitted deletes from proxyB are visible to proxyA within the
same transaction.
+ Assert.assertEquals(0, proxyA.getPartitions(tableName, args).size());
+ Assert.assertEquals(0, handlerA.getOpenQueryCount());
+
+ objectStore.commitTransaction();
+ } catch (Exception e) {
+ objectStore.rollbackTransaction();
+ throw e;
+ }
+ }
+ Assert.assertFalse(objectStore.isActiveTransaction());
+
+ try (AutoCloseable ignored = deadline();
+ DirectSqlConfigurator directSql = new DirectSqlConfigurator(conf,
false)) {
+ Assert.assertEquals(0, proxyA.getPartitions(tableName, args).size());
+ Assert.assertEquals(0, proxyB.getPartitions(tableName, args).size());
+ }
+ }
+
+ @Test
+ public void testProxiesRollbackInTxn() throws Exception {
+ TableStore proxyA = objectStore.unwrap(TableStore.class);
+ TableStore proxyB = objectStore.unwrap(TableStore.class);
+ GetPartitionsArgs args = new
GetPartitionsArgs.GetPartitionsArgsBuilder().max(10).build();
+
+ try (AutoCloseable ignored = deadline();
+ DirectSqlConfigurator directSql = new DirectSqlConfigurator(conf,
false)) {
+ objectStore.openTransaction();
+ try {
+ proxyA.dropPartitions(tableName, Arrays.asList("test_part_col=a0",
"test_part_col=a1"));
+ Assert.assertEquals(1, proxyB.getPartitions(tableName, args).size());
+ objectStore.rollbackTransaction();
+ } catch (Exception e) {
+ objectStore.rollbackTransaction();
+ throw e;
+ }
+ }
+ Assert.assertFalse(objectStore.isActiveTransaction());
+
+ try (AutoCloseable ignored = deadline();
+ DirectSqlConfigurator directSql = new DirectSqlConfigurator(conf,
false)) {
+ Assert.assertEquals(3, proxyA.getPartitions(tableName, args).size());
+ Assert.assertEquals(3, proxyB.getPartitions(tableName, args).size());
+ }
+ }
+
+ @Test
+ public void testLeakingQueries() throws Exception {
+ TableStore proxyA = objectStore.unwrap(TableStore.class);
+ TransactionHandler<TableStore> handlerA = getHandler(proxyA);
+ RawStoreBundle rsbA = (RawStoreBundle) handlerA.simpl();
+ rsbA.getPersistentManager().newQuery(MTable.class);
+ rsbA.getPersistentManager().newQuery(MTable.class);
+ rsbA.getPersistentManager().newQuery(MTable.class);
+ // handlerA should have 3 opening queries
+ Assert.assertEquals(3, handlerA.getOpenQueryCount());
+
+ // Imaging proxyA creates a nested proxyB inside the same transaction
+ TableStore proxyB = objectStore.unwrap(TableStore.class);
+ TransactionHandler<TableStore> handlerB = getHandler(proxyB);
+ RawStoreBundle rsbB = (RawStoreBundle) handlerB.simpl();
+ rsbB.getPersistentManager().newQuery(MTable.class);
+ // Don't overwrite the queries opened by handlerA
+ Assert.assertEquals(3, handlerA.getOpenQueryCount());
+ Assert.assertEquals(1, handlerB.getOpenQueryCount());
+
+ proxyA.getTable(tableName, null, -1);
+ // handleA should close all opening queries at the end of transaction
+ Assert.assertEquals(0, handlerA.getOpenQueryCount());
+ // Shouldn't affect the opening queries in handlerB
+ Assert.assertEquals(1, handlerB.getOpenQueryCount());
+
+ rsbA.getPersistentManager().newQuery(MTable.class);
+ proxyB.getTable(tableName, null, -1);
+ Assert.assertEquals(0, handlerB.getOpenQueryCount());
+ // Shouldn't affect the opening queries in handlerA
+ Assert.assertEquals(1, handlerA.getOpenQueryCount());
+ }
+
+ @Test
+ public void testCreateTableSharesPm() throws Exception {
+ String otherDb = "unwrap_other_db";
+ String otherTable = "unwrap_other_tbl";
+ objectStore.createDatabase(new DatabaseBuilder()
+ .setName(otherDb)
+ .setDescription("description")
+ .setLocation("locationurl")
+ .build(conf));
+
+ Table table = new TableBuilder()
+ .setDbName(otherDb)
+ .setTableName(otherTable)
+ .addCol("c", "int")
+ .build(conf);
+
+ TableStore tableStore = objectStore.unwrap(TableStore.class);
+ tableStore.createTable(table);
+
+ Table fetched = tableStore.getTable(
+ new TableName(DEFAULT_CATALOG_NAME, otherDb, otherTable), null, -1);
+ Assert.assertNotNull(fetched);
+ Assert.assertEquals(0, getHandler(tableStore).getOpenQueryCount());
+ Assert.assertFalse(objectStore.isActiveTransaction());
+ }
+
+ private void createPartitionedTable() throws Exception {
+ objectStore.createDatabase(new DatabaseBuilder()
+ .setName(DB)
+ .setDescription("description")
+ .setLocation("locationurl")
+ .build(conf));
+
+ Table table = new TableBuilder()
+ .setDbName(DB)
+ .setTableName(TABLE)
+ .addCol("test_col1", "int")
+ .addPartCol("test_part_col", "int")
+ .build(conf);
+ objectStore.createTable(table);
+
+ for (int i = 0; i < 3; i++) {
+ Partition part = new PartitionBuilder()
+ .inTable(table)
+ .addValue("a" + i)
+ .build(conf);
+ objectStore.addPartition(part);
+ }
+ }
+
+ private static <T> TransactionHandler<T> getHandler(Object storeProxy) {
+ return (TransactionHandler<T>) Proxy.getInvocationHandler(storeProxy);
+ }
+
+ private AutoCloseable deadline() throws Exception {
+ Deadline.registerIfNot(100_000);
+ Deadline.startTimer("TestObjectStoreUnwrap");
+ return Deadline::stopTimer;
+ }
+}
