wecharyu commented on code in PR #5851:
URL: https://github.com/apache/hive/pull/5851#discussion_r2662093568
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HMSHandler.java:
##########
@@ -5125,257 +4567,57 @@ public boolean drop_partition(final String db_name,
final String tbl_name,
null);
}
- /** Stores a path and its size. */
- private static class PathAndDepth implements Comparable<PathAndDepth> {
- final Path path;
- final int depth;
-
- public PathAndDepth(Path path, int depth) {
- this.path = path;
- this.depth = depth;
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(path.hashCode(), depth);
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- PathAndDepth that = (PathAndDepth) o;
- return depth == that.depth && Objects.equals(path, that.path);
- }
-
- /** The largest {@code depth} is processed first in a {@link
PriorityQueue}. */
- @Override
- public int compareTo(PathAndDepth o) {
- return o.depth - depth;
- }
- }
-
@Override
public DropPartitionsResult drop_partitions_req(
DropPartitionsRequest request) throws TException {
- RawStore ms = getMS();
- String dbName = request.getDbName(), tblName = request.getTblName();
- String catName = request.isSetCatName() ? request.getCatName() :
getDefaultCatalog(conf);
-
- boolean ifExists = request.isSetIfExists() && request.isIfExists();
- boolean deleteData = request.isSetDeleteData() && request.isDeleteData();
- boolean ignoreProtection = request.isSetIgnoreProtection() &&
request.isIgnoreProtection();
- boolean needResult = !request.isSetNeedResult() || request.isNeedResult();
-
- List<PathAndDepth> dirsToDelete = new ArrayList<>();
- List<Path> archToDelete = new ArrayList<>();
- EnvironmentContext envContext =
- request.isSetEnvironmentContext() ? request.getEnvironmentContext() :
null;
- boolean success = false;
-
- Table tbl = null;
- List<Partition> parts = null;
- boolean mustPurge = false;
- boolean tableDataShouldBeDeleted = false;
- long writeId = 0;
-
- Map<String, String> transactionalListenerResponses = null;
- boolean needsCm = false;
-
try {
- ms.openTransaction();
- // We need Partition-s for firing events and for result; DN needs
MPartition-s to drop.
- // Great... Maybe we could bypass fetching MPartitions by issuing direct
SQL deletes.
- GetTableRequest getTableRequest = new GetTableRequest(dbName, tblName);
- getTableRequest.setCatName(catName);
- tbl = get_table_core(getTableRequest);
- mustPurge = isMustPurge(envContext, tbl);
- tableDataShouldBeDeleted = checkTableDataShouldBeDeleted(tbl,
deleteData);
- writeId = getWriteId(envContext);
-
- boolean hasMissingParts = false;
- RequestPartsSpec spec = request.getParts();
- List<String> partNames = null;
-
- if (spec.isSetExprs()) {
- // Dropping by expressions.
- parts = new ArrayList<>(spec.getExprs().size());
- for (DropPartitionsExpr expr : spec.getExprs()) {
- List<Partition> result = new ArrayList<>();
- boolean hasUnknown = ms.getPartitionsByExpr(catName, dbName,
tblName, result,
- new GetPartitionsArgs.GetPartitionsArgsBuilder()
-
.expr(expr.getExpr()).skipColumnSchemaForPartition(request.isSkipColumnSchemaForPartition())
- .build());
- if (hasUnknown) {
- // Expr is built by DDLSA, it should only contain part cols and
simple ops
- throw new MetaException("Unexpected unknown partitions to drop");
- }
- // this is to prevent dropping archived partition which is archived
in a
- // different level the drop command specified.
- if (!ignoreProtection && expr.isSetPartArchiveLevel()) {
- for (Partition part : parts) {
- if (MetaStoreUtils.isArchived(part)
- && MetaStoreUtils.getArchivingLevel(part) <
expr.getPartArchiveLevel()) {
- throw new MetaException("Cannot drop a subset of partitions "
- + " in an archive, partition " + part);
- }
- }
- }
- if (result.isEmpty()) {
- hasMissingParts = true;
- if (!ifExists) {
- // fail-fast for missing partition expr
- break;
- }
+ DropPartitionsResult resp = new DropPartitionsResult();
+ AbstractOperationHandler<DropPartitionsRequest,
DropPartitionsHandler.DropPartitionsResult> dropPartsOp =
+ AbstractOperationHandler.offer(this, request);
+ AbstractOperationHandler.OperationStatus status =
dropPartsOp.getOperationStatus();
+ if (status.isFinished() && dropPartsOp.getResult() != null &&
dropPartsOp.getResult().success()) {
+ DropPartitionsHandler.DropPartitionsResult result =
dropPartsOp.getResult();
+ long writeId = getWriteId(request.getEnvironmentContext());
+ if (result.tableDataShouldBeDeleted()) {
+ LOG.info(result.mustPurge() ?
+ "dropPartition() will purge partition-directories directly,
skipping trash."
+ : "dropPartition() will move partition-directories to
trash-directory.");
+ // Archived partitions have har:/to_har_file as their location.
+ // The original directory was saved in params
+ for (Path path : result.getArchToDelete()) {
+ wh.deleteDir(path, result.mustPurge(), result.needCm());
}
- parts.addAll(result);
- }
- } else if (spec.isSetNames()) {
- partNames = spec.getNames();
- parts = ms.getPartitionsByNames(catName, dbName, tblName,
- new GetPartitionsArgs.GetPartitionsArgsBuilder()
-
.partNames(partNames).skipColumnSchemaForPartition(request.isSkipColumnSchemaForPartition())
- .build());
- hasMissingParts = (parts.size() != partNames.size());
- } else {
- throw new MetaException("Partition spec is not set");
- }
-
- if (hasMissingParts && !ifExists) {
- throw new NoSuchObjectException("Some partitions to drop are missing");
- }
-
- List<String> colNames = null;
- if (partNames == null) {
- partNames = new ArrayList<>(parts.size());
- colNames = new ArrayList<>(tbl.getPartitionKeys().size());
- for (FieldSchema col : tbl.getPartitionKeys()) {
- colNames.add(col.getName());
- }
- }
-
- for (Partition part : parts) {
- // TODO - we need to speed this up for the normal path where all
partitions are under
- // the table and we don't have to stat every partition
-
- firePreEvent(new PreDropPartitionEvent(tbl, part, deleteData, this));
- if (colNames != null) {
- partNames.add(FileUtils.makePartName(colNames, part.getValues()));
- }
- if (tableDataShouldBeDeleted) {
- if (MetaStoreUtils.isArchived(part)) {
- // Archived partition is only able to delete original location.
- Path archiveParentDir = MetaStoreUtils.getOriginalLocation(part);
- verifyIsWritablePath(archiveParentDir);
- archToDelete.add(archiveParentDir);
- } else if ((part.getSd() != null) && (part.getSd().getLocation() !=
null)) {
- Path partPath = new Path(part.getSd().getLocation());
- verifyIsWritablePath(partPath);
- dirsToDelete.add(new PathAndDepth(partPath,
part.getValues().size()));
- }
- }
- }
- ms.dropPartitions(catName, dbName, tblName, partNames);
- if (!parts.isEmpty() && !transactionalListeners.isEmpty()) {
- transactionalListenerResponses = MetaStoreListenerNotifier
- .notifyEvent(transactionalListeners, EventType.DROP_PARTITION,
- new DropPartitionEvent(tbl, parts, true, deleteData, this),
envContext);
- }
- success = ms.commitTransaction();
- needsCm = ReplChangeManager.shouldEnableCm(ms.getDatabase(catName,
dbName), tbl);
-
- DropPartitionsResult result = new DropPartitionsResult();
- if (needResult) {
- result.setPartitions(parts);
- }
- return result;
- } finally {
- if (!success) {
- ms.rollbackTransaction();
- } else if (tableDataShouldBeDeleted) {
- LOG.info(mustPurge ?
- "dropPartition() will purge partition-directories directly,
skipping trash."
- : "dropPartition() will move partition-directories to
trash-directory.");
- // Archived partitions have har:/to_har_file as their location.
- // The original directory was saved in params
- for (Path path : archToDelete) {
- wh.deleteDir(path, mustPurge, needsCm);
- }
-
- // Uses a priority queue to delete the parents of deleted directories
if empty.
- // Parents with the deepest path are always processed first. It
guarantees that the emptiness
- // of a parent won't be changed once it has been processed. So
duplicated processing can be
- // avoided.
- PriorityQueue<PathAndDepth> parentsToDelete = new PriorityQueue<>();
- for (PathAndDepth p : dirsToDelete) {
- wh.deleteDir(p.path, mustPurge, needsCm);
- addParentForDel(parentsToDelete, p);
- }
-
- HashSet<PathAndDepth> processed = new HashSet<>();
- while (!parentsToDelete.isEmpty()) {
- try {
- PathAndDepth p = parentsToDelete.poll();
- if (processed.contains(p)) {
- continue;
+ // Uses a priority queue to delete the parents of deleted
directories if empty.
+ // Parents with the deepest path are always processed first. It
guarantees that the emptiness
+ // of a parent won't be changed once it has been processed. So
duplicated processing can be
+ // avoided.
+ for (Iterator<DropPartitionsHandler.PathAndDepth> iterator =
result.getDirsToDelete();
+ iterator.hasNext();) {
+ DropPartitionsHandler.PathAndDepth p = iterator.next();
+ Path path = p.path();
+ if (p.isPartitionDir()) {
+ wh.deleteDir(path, result.mustPurge(), result.needCm());
+ } else if (wh.isWritable(path) && wh.isEmptyDir(path)) {
+ wh.deleteDir(path, result.mustPurge(), result.needCm());
}
- processed.add(p);
-
- Path path = p.path;
- if (wh.isWritable(path) && wh.isEmptyDir(path)) {
- wh.deleteDir(path, mustPurge, needsCm);
- addParentForDel(parentsToDelete, p);
- }
- } catch (IOException ex) {
- LOG.warn("Error from recursive parent deletion", ex);
- throw new MetaException("Failed to delete parent: " +
ex.getMessage());
}
- }
- } else if (TxnUtils.isTransactionalTable(tbl) && writeId > 0) {
- for (Partition part : parts) {
- if ((part.getSd() != null) && (part.getSd().getLocation() != null)) {
- Path partPath = new Path(part.getSd().getLocation());
- verifyIsWritablePath(partPath);
-
- addTruncateBaseFile(partPath, writeId, conf, DataFormat.DROPPED);
+ } else if (result.isTransactionalTable() && writeId > 0) {
+ for (Partition part : result.getPartitions()) {
+ if ((part.getSd() != null) && (part.getSd().getLocation() !=
null)) {
+ Path partPath = new Path(part.getSd().getLocation());
+ ((DropPartitionsHandler)
dropPartsOp).verifyIsWritablePath(partPath);
+ addTruncateBaseFile(partPath, writeId, conf, DataFormat.DROPPED);
+ }
}
}
- }
-
- if (parts != null) {
- if (!parts.isEmpty() && !listeners.isEmpty()) {
- MetaStoreListenerNotifier.notifyEvent(listeners,
- EventType.DROP_PARTITION,
- new DropPartitionEvent(tbl, parts, success, deleteData, this),
- envContext,
- transactionalListenerResponses, ms);
+ if (request.isNeedResult()) {
+ resp.setPartitions(result.getPartitions());
}
}
- }
- }
-
- private static void addParentForDel(PriorityQueue<PathAndDepth>
parentsToDelete, PathAndDepth p) {
- Path parent = p.path.getParent();
- if (parent != null && p.depth - 1 > 0) {
- parentsToDelete.add(new PathAndDepth(parent, p.depth - 1));
- }
- }
-
- private void verifyIsWritablePath(Path dir) throws MetaException {
- try {
- if (!wh.isWritable(dir.getParent())) {
- throw new MetaException("Table partition not deleted since " +
dir.getParent()
- + " is not writable by " + SecurityUtils.getUser());
- }
- } catch (IOException ex) {
- LOG.warn("Error from isWritable", ex);
- throw new MetaException("Table partition not deleted since " +
dir.getParent()
- + " access cannot be checked: " + ex.getMessage());
+ return resp;
Review Comment:
Got it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
