You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by de...@apache.org on 2025/07/02 06:13:58 UTC
(hive) branch master updated: HIVE-29050: Optimize the partition validation in add_partitions (#5910)
This is an automated email from the ASF dual-hosted git repository.
dengzh 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 895f3f3f5c5 HIVE-29050: Optimize the partition validation in add_partitions (#5910)
895f3f3f5c5 is described below
commit 895f3f3f5c5a1f5b4ecbba5af63e6a60db083dfc
Author: dengzh <de...@gmail.com>
AuthorDate: Wed Jul 2 14:13:53 2025 +0800
HIVE-29050: Optimize the partition validation in add_partitions (#5910)
---
.../apache/hadoop/hive/ql/TestTxnAddPartition.java | 4 +-
.../apache/hadoop/hive/metastore/HMSHandler.java | 164 ++++++---------------
2 files changed, 45 insertions(+), 123 deletions(-)
diff --git a/ql/src/test/org/apache/hadoop/hive/ql/TestTxnAddPartition.java b/ql/src/test/org/apache/hadoop/hive/ql/TestTxnAddPartition.java
index 59766990be6..4187a2fb9b1 100644
--- a/ql/src/test/org/apache/hadoop/hive/ql/TestTxnAddPartition.java
+++ b/ql/src/test/org/apache/hadoop/hive/ql/TestTxnAddPartition.java
@@ -114,7 +114,7 @@ private void addPartition() throws Exception {
CommandProcessorException e =
runStatementOnDriverNegative("ALTER TABLE T ADD PARTITION (p=0) location '" + getWarehouseDir() + "/3/data'");
Assert.assertTrue("add existing partition",
- e.getMessage() != null && e.getMessage().contains("Partition already exists"));
+ e.getMessage() != null && e.getMessage().contains("Partition(s) already exist"));
//should be no-op since p=3 exists
String stmt = "ALTER TABLE T ADD IF NOT EXISTS " +
@@ -188,7 +188,7 @@ private void addPartitionMM() throws Exception {
CommandProcessorException e =
runStatementOnDriverNegative("ALTER TABLE T ADD PARTITION (p=0) location '" + getWarehouseDir() + "/3/data'");
Assert.assertTrue("add existing partition",
- e.getMessage() != null && e.getMessage().contains("Partition already exists"));
+ e.getMessage() != null && e.getMessage().contains("Partition(s) already exist"));
//should be no-op since p=3 exists
runStatementOnDriver("ALTER TABLE T ADD IF NOT EXISTS " +
diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HMSHandler.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HMSHandler.java
index 9c3932b3853..57a897f6de6 100644
--- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HMSHandler.java
+++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HMSHandler.java
@@ -4264,13 +4264,15 @@ public boolean equals(Object obj) {
}
}
- private List<Partition> add_partitions_core(final RawStore ms, String catName,
- String dbName, String tblName, List<Partition> parts, final boolean ifNotExists,
- boolean isSkipColSchemaForPartition, EnvironmentContext envContext) throws TException {
+ private List<Partition> add_partitions_core(final RawStore ms,
+ AddPartitionsRequest request) throws TException {
+ String catName = request.getCatName();
+ String dbName = request.getDbName();
+ String tblName = request.getTblName();
if (dbName == null || tblName == null) {
throw new MetaException("The database and table name cannot be null.");
}
-
+ EnvironmentContext envContext = request.getEnvironmentContext();
boolean success = false;
// Ensures that the list doesn't have dups, and keeps track of directories we have created.
final Map<PartValEqWrapperLite, Boolean> addedPartitions = new ConcurrentHashMap<>();
@@ -4280,8 +4282,8 @@ private List<Partition> add_partitions_core(final RawStore ms, String catName,
Map<String, String> transactionalListenerResponses = Collections.emptyMap();
Database db = null;
- List<ColumnStatistics> partsColStats = new ArrayList<>(parts.size());
- List<Long> partsWriteIds = new ArrayList<>(parts.size());
+ List<ColumnStatistics> partsColStats = new ArrayList<>();
+ List<Long> partsWriteIds = new ArrayList<>();
throwUnsupportedExceptionIfRemoteDB(dbName, "add_partitions");
@@ -4292,17 +4294,16 @@ private List<Partition> add_partitions_core(final RawStore ms, String catName,
tbl = ms.getTable(catName, dbName, tblName, null);
if (tbl == null) {
throw new InvalidObjectException("Unable to add partitions because "
- + TableName.getQualified(catName, dbName, tblName) +
- " does not exist");
+ + TableName.getQualified(catName, dbName, tblName) + " does not exist");
}
MTable mTable = getMS().ensureGetMTable(catName, dbName, tblName);
db = ms.getDatabase(catName, dbName);
- Set<PartValEqWrapperLite> partsToAdd = new HashSet<>(parts.size());
- List<Partition> partitionsToAdd = new ArrayList<>(parts.size());
+ Set<PartValEqWrapperLite> partsToAdd = new HashSet<>();
List<FieldSchema> partitionKeys = tbl.getPartitionKeys();
- for (final Partition part : parts) {
- if(isSkipColSchemaForPartition) {
+ Map<String, Partition> nameToPart = new LinkedHashMap<>();
+ for (final Partition part : request.getParts()) {
+ if (request.isSkipColumnSchemaForPartition()) {
part.getSd().setCols(tbl.getSd().getCols());
}
// Collect partition column stats to be updated if present. Partition objects passed down
@@ -4319,19 +4320,30 @@ private List<Partition> add_partitions_core(final RawStore ms, String catName,
// incorrect, an exception will be thrown before the threads which create the partition
// folders are submitted. This way we can be sure that no partition and no partition
// folder will be created if the list contains an invalid partition.
- if (validatePartition(part, catName, tblName, dbName, partsToAdd, ms, ifNotExists,
- partitionKeys)) {
- partitionsToAdd.add(part);
- } else {
- existingParts.add(part);
- }
+ validatePartition(part, catName, tblName, dbName, partsToAdd);
+ nameToPart.put(Warehouse.makePartName(partitionKeys, part.getValues()), part);
}
- // Only authorize on newly created partitions
- if (!partitionsToAdd.isEmpty()) {
- firePreEvent(new PreAddPartitionEvent(tbl, partitionsToAdd, this));
+ List<Partition> existedParts =
+ ms.getPartitionsByNames(catName, dbName, tblName,
+ new GetPartitionsArgs.GetPartitionsArgsBuilder().partNames(new ArrayList<>(nameToPart.keySet())).build());
+ List<String> existedPartNames = new ArrayList<>();
+ for (Partition part : existedParts) {
+ String partName = Warehouse.makePartName(partitionKeys, part.getValues());
+ existedPartNames.add(partName);
+ existingParts.add(nameToPart.remove(partName));
+ }
+ if (!request.isIfNotExists() && !existedPartNames.isEmpty()) {
+ throw new AlreadyExistsException("Partition(s) already exist, partition name(s): " + existedPartNames);
}
+ if (nameToPart.isEmpty()) {
+ return Collections.emptyList();
+ }
+ List<Partition> partitionsToAdd = new ArrayList<>(nameToPart.values());
+ // Only authorize on newly created partitions
+ firePreEvent(new PreAddPartitionEvent(tbl, partitionsToAdd, this));
+
newParts.addAll(createPartitionFolders(partitionsToAdd, tbl, addedPartitions, envContext));
if (!newParts.isEmpty()) {
@@ -4439,16 +4451,14 @@ private void cleanupPartitionFolders(final Map<PartValEqWrapperLite, Boolean> ad
* @param tblName
* @param dbName
* @param partsToAdd
- * @param ms
- * @param ifNotExists
- * @return
* @throws MetaException
* @throws TException
*/
- private boolean validatePartition(final Partition part, final String catName,
- final String tblName, final String dbName, final Set<PartValEqWrapperLite> partsToAdd,
- final RawStore ms, final boolean ifNotExists, List<FieldSchema> partitionKeys) throws MetaException, TException {
-
+ private void validatePartition(final Partition part, final String catName,
+ final String tblName, final String dbName, final Set<PartValEqWrapperLite> partsToAdd)
+ throws MetaException, TException {
+ MetaStoreServerUtils.validatePartitionNameCharacters(part.getValues(),
+ partitionValidationPattern);
if (part.getDbName() == null || part.getTableName() == null) {
throw new MetaException("The database and table name must be set in the partition.");
}
@@ -4470,19 +4480,12 @@ private boolean validatePartition(final Partition part, final String catName,
throw new MetaException("Partition value cannot be null.");
}
- boolean shouldAdd = startAddPartition(ms, part, partitionKeys, ifNotExists);
- if (!shouldAdd) {
- LOG.info("Not adding partition {} as it already exists", part);
- return false;
- }
-
if (!partsToAdd.add(new PartValEqWrapperLite(part))) {
// Technically, for ifNotExists case, we could insert one and discard the other
// because the first one now "exists", but it seems better to report the problem
// upstream as such a command doesn't make sense.
throw new MetaException("Duplicate partitions in the list: " + part);
}
- return true;
}
/**
@@ -4592,9 +4595,8 @@ public AddPartitionsResult add_partitions_req(AddPartitionsRequest request)
boolean isColSkippedForPartitions = request.isSkipColumnSchemaForPartition();
// Make sure all the partitions have the catalog set as well
request.getParts().forEach(p -> p.setCatName(catName));
- List<Partition> parts = add_partitions_core(getMS(),catName, dbName, tblName,
- request.getParts(), request.isIfNotExists(),
- request.isSkipColumnSchemaForPartition(), request.getEnvironmentContext());
+ request.setCatName(catName);
+ List<Partition> parts = add_partitions_core(getMS(), request);
if (request.isNeedResult()) {
if (isColSkippedForPartitions) {
if (!parts.isEmpty()) {
@@ -4693,8 +4695,9 @@ public int add_partitions_pspec(final List<PartitionSpec> partSpecs)
partitionsToAdd.add(part);
partitionIterator.next();
}
- ret = add_partitions_core(getMS(), catName, dbName, tableName, partitionsToAdd,
- false, false, null).size();
+ AddPartitionsRequest request = new AddPartitionsRequest(dbName, tableName, partitionsToAdd, false);
+ request.setCatName(catName);
+ ret = add_partitions_core(getMS(), request).size();
} catch (Exception e) {
ex = e;
throw handleException(e)
@@ -4706,19 +4709,6 @@ public int add_partitions_pspec(final List<PartitionSpec> partSpecs)
return ret;
}
- private boolean startAddPartition(
- RawStore ms, Partition part, List<FieldSchema> partitionKeys, boolean ifNotExists)
- throws TException {
- MetaStoreServerUtils.validatePartitionNameCharacters(part.getValues(),
- partitionValidationPattern);
- boolean doesExist = ms.doesPartitionExist(part.getCatName(),
- part.getDbName(), part.getTableName(), partitionKeys, part.getValues());
- if (doesExist && !ifNotExists) {
- throw new AlreadyExistsException("Partition already exists: " + part);
- }
- return !doesExist;
- }
-
/**
* Handles the location for a partition being created.
* @param tbl Table.
@@ -4836,74 +4826,6 @@ private void initializePartitionParameters(final Table tbl,
}
}
- private Partition add_partition_core(final RawStore ms,
- final Partition part, final EnvironmentContext envContext)
- throws TException {
- boolean success = false;
- Table tbl = null;
- Map<String, String> transactionalListenerResponses = Collections.emptyMap();
- if (!part.isSetCatName()) {
- part.setCatName(getDefaultCatalog(conf));
- }
- try {
- ms.openTransaction();
- tbl = ms.getTable(part.getCatName(), part.getDbName(), part.getTableName(), null);
- if (tbl == null) {
- throw new InvalidObjectException(
- "Unable to add partition because table or database do not exist");
- }
-
- firePreEvent(new PreAddPartitionEvent(tbl, part, this));
-
- if (part.getValues() == null || part.getValues().isEmpty()) {
- throw new MetaException("The partition values cannot be null or empty.");
- }
- boolean shouldAdd = startAddPartition(ms, part, tbl.getPartitionKeys(), false);
- assert shouldAdd; // start would throw if it already existed here
- boolean madeDir = createLocationForAddedPartition(tbl, part);
- try {
- initializeAddedPartition(tbl, part, madeDir, envContext);
- initializePartitionParameters(tbl, part);
- success = ms.addPartition(part);
- } finally {
- if (!success && madeDir) {
- wh.deleteDir(new Path(part.getSd().getLocation()), true, false,
- ReplChangeManager.shouldEnableCm(ms.getDatabase(part.getCatName(), part.getDbName()), tbl));
- }
- }
-
- // Setting success to false to make sure that if the listener fails, rollback happens.
- success = false;
-
- if (!transactionalListeners.isEmpty()) {
- transactionalListenerResponses =
- MetaStoreListenerNotifier.notifyEvent(transactionalListeners,
- EventType.ADD_PARTITION,
- new AddPartitionEvent(tbl, Arrays.asList(part), true, this),
- envContext);
-
- }
-
- // we proceed only if we'd actually succeeded anyway, otherwise,
- // we'd have thrown an exception
- success = ms.commitTransaction();
- } finally {
- if (!success) {
- ms.rollbackTransaction();
- }
-
- if (!listeners.isEmpty()) {
- MetaStoreListenerNotifier.notifyEvent(listeners,
- EventType.ADD_PARTITION,
- new AddPartitionEvent(tbl, Arrays.asList(part), success, this),
- envContext,
- transactionalListenerResponses, ms);
-
- }
- }
- return part;
- }
-
@Deprecated
@Override
public Partition add_partition(final Partition part)