You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by kr...@apache.org on 2025/05/15 07:50:57 UTC
(hive) branch master updated: HIVE-28947: Add ability to specify Compaction options in stats updater of compactor (#5807)
This is an automated email from the ASF dual-hosted git repository.
krisztiankasa 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 eb472324faa HIVE-28947: Add ability to specify Compaction options in stats updater of compactor (#5807)
eb472324faa is described below
commit eb472324faa81f1ba81beee555785089a75ea741
Author: Krisztian Kasa <ka...@gmail.com>
AuthorDate: Thu May 15 09:50:51 2025 +0200
HIVE-28947: Add ability to specify Compaction options in stats updater of compactor (#5807)
---
.../hive/ql/txn/compactor/TestCompactor.java | 8 +-
.../ql/txn/compactor/TestCrudCompactorOnTez.java | 5 +-
.../hive/ql/txn/compactor/CompactorUtil.java | 20 +++++
.../hive/ql/txn/compactor/QueryCompactor.java | 16 +---
.../hadoop/hive/ql/txn/compactor/StatsUpdater.java | 41 ++++++---
.../hadoop/hive/ql/txn/compactor/Worker.java | 9 +-
.../hive/ql/txn/compactor/TestStatsUpdater.java | 97 ++++++++++++++++++++++
.../hadoop/hive/ql/txn/compactor/TestWorker.java | 2 +-
.../metastore/txn/entities/CompactionInfo.java | 4 +-
9 files changed, 169 insertions(+), 33 deletions(-)
diff --git a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestCompactor.java b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestCompactor.java
index 8b93bc42a53..c8e21876134 100644
--- a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestCompactor.java
+++ b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestCompactor.java
@@ -66,6 +66,7 @@
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -74,6 +75,7 @@
import java.util.TreeSet;
import java.util.concurrent.atomic.AtomicBoolean;
+import static java.util.Collections.emptyMap;
import static org.apache.hadoop.hive.common.AcidConstants.VISIBILITY_PATTERN;
import static org.apache.hadoop.hive.ql.TestTxnCommands2.runCleaner;
import static org.apache.hadoop.hive.ql.TestTxnCommands2.runInitiator;
@@ -378,10 +380,10 @@ public void testStatsAfterCompactionPartTbl() throws Exception {
//compute stats before compaction
CompactionInfo ci = new CompactionInfo(dbName, tblName, "bkt=0", CompactionType.MAJOR);
- statsUpdater.gatherStats(ci, conf, System.getProperty("user.name"),
+ statsUpdater.gatherStats(conf, ci, emptyMap(), System.getProperty("user.name"),
CompactorUtil.getCompactorJobQueueName(conf, ci, table), msClient);
ci = new CompactionInfo(dbName, tblName, "bkt=1", CompactionType.MAJOR);
- statsUpdater.gatherStats(ci, conf, System.getProperty("user.name"),
+ statsUpdater.gatherStats(conf, ci, emptyMap(), System.getProperty("user.name"),
CompactorUtil.getCompactorJobQueueName(conf, ci, table), msClient);
//Check basic stats are collected
@@ -472,7 +474,7 @@ public void testStatsAfterCompactionTbl() throws Exception {
//compute stats before compaction
CompactionInfo ci = new CompactionInfo(dbName, tblName, null, CompactionType.MAJOR);
- statsUpdater.gatherStats(ci, conf, System.getProperty("user.name"),
+ statsUpdater.gatherStats(conf, ci, emptyMap(), System.getProperty("user.name"),
CompactorUtil.getCompactorJobQueueName(conf, ci, table), msClient);
//Check basic stats are collected
diff --git a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestCrudCompactorOnTez.java b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestCrudCompactorOnTez.java
index b8d1827de0a..6f662f2cfee 100644
--- a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestCrudCompactorOnTez.java
+++ b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestCrudCompactorOnTez.java
@@ -82,6 +82,7 @@
import org.mockito.Mockito;
import org.mockito.internal.util.reflection.FieldSetter;
+import static java.util.Collections.emptyMap;
import static org.apache.hadoop.hive.ql.TxnCommandsBaseForTests.runWorker;
import static org.apache.hadoop.hive.ql.txn.compactor.TestCompactor.execSelectAndDumpData;
import static org.apache.hadoop.hive.ql.txn.compactor.TestCompactor.executeStatementOnDriver;
@@ -858,7 +859,7 @@ public void testStatsAfterQueryCompactionOnTez() throws Exception {
//compute stats before compaction
CompactionInfo ci = new CompactionInfo(dbName, tblName, null, CompactionType.MAJOR);
- new StatsUpdater().gatherStats(ci, conf, System.getProperty("user.name"),
+ new StatsUpdater().gatherStats(conf, ci, emptyMap(), System.getProperty("user.name"),
CompactorUtil.getCompactorJobQueueName(conf, ci, table), msClient);
//Check basic stats are collected
@@ -3165,7 +3166,7 @@ public void testStatsAfterCompactionPartTbl(boolean isQueryBased, boolean isAuto
//compute stats before compaction
CompactionInfo ci = new CompactionInfo(dbName, tblName, "bkt=1", compactionType);
- new StatsUpdater().gatherStats(ci, conf, System.getProperty("user.name"),
+ new StatsUpdater().gatherStats(conf, ci, emptyMap(), System.getProperty("user.name"),
CompactorUtil.getCompactorJobQueueName(conf, ci, table), msClient);
//Check basic stats are collected
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorUtil.java b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorUtil.java
index 05e932d0f91..2b5b9e97e0b 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorUtil.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorUtil.java
@@ -75,6 +75,7 @@
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
import java.util.HashMap;
import java.util.concurrent.ExecutorService;
@@ -84,6 +85,7 @@
import java.util.function.Function;
import java.util.regex.Matcher;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static java.lang.String.format;
import static org.apache.hadoop.hive.metastore.HMSHandler.getMSForConf;
@@ -92,8 +94,10 @@
public class CompactorUtil {
private static final Logger LOG = LoggerFactory.getLogger(CompactorUtil.class);
public static final String COMPACTOR = "compactor";
+ public static final String COMPACTOR_PREFIX = COMPACTOR + ".";
private static final String COMPACTOR_THRESHOLD_PREFIX = "compactorthreshold.";
+
/**
* List of accepted properties for defining the compactor's job queue.
*
@@ -579,4 +583,20 @@ public static String getPoolName(HiveConf conf, Table t, MetadataCache metadataC
}
return poolName;
}
+
+ public static void overrideConfProps(HiveConf conf, CompactionInfo ci, Map<String, String> properties) {
+ overrideConfProps(conf, new StringableMap(ci.properties), properties);
+ }
+
+ public static void overrideConfProps(
+ HiveConf conf, Map<String, String> ciProperties, Map<String, String> properties) {
+ Stream.of(properties, ciProperties)
+ .filter(Objects::nonNull)
+ .flatMap(map -> map.entrySet().stream())
+ .filter(entry -> entry.getKey().startsWith(COMPACTOR_PREFIX))
+ .forEach(entry -> {
+ String property = entry.getKey().substring(COMPACTOR_PREFIX.length());
+ conf.set(property, entry.getValue());
+ });
+ }
}
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/QueryCompactor.java b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/QueryCompactor.java
index 4638451f3b5..afb01e7fd4c 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/QueryCompactor.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/QueryCompactor.java
@@ -46,13 +46,14 @@
import java.util.Objects;
import java.util.stream.Stream;
+import static org.apache.hadoop.hive.ql.txn.compactor.CompactorUtil.overrideConfProps;
+
/**
* Common interface for query based compactions.
*/
public abstract class QueryCompactor implements Compactor {
private static final Logger LOG = LoggerFactory.getLogger(QueryCompactor.class.getName());
- private static final String COMPACTOR_PREFIX = "compactor.";
/**
* This is the final step of the compaction, which can vary based on compaction type. Usually this involves some file
@@ -74,7 +75,7 @@ protected SessionState setupQueryCompactionSession(HiveConf conf, CompactionInfo
conf.set(HiveConf.ConfVars.HIVE_QUOTEDID_SUPPORT.varname, "column");
conf.setBoolVar(HiveConf.ConfVars.HIVE_SERVER2_ENABLE_DOAS, true);
conf.setBoolVar(HiveConf.ConfVars.HIVE_HDFS_ENCRYPTION_SHIM_CACHE_ON, false);
- Util.overrideConfProps(conf, compactionInfo, tblProperties);
+ overrideConfProps(conf, compactionInfo, tblProperties);
String user = compactionInfo.runAs;
SessionState sessionState = DriverUtils.setUpAndStartSessionState(conf, user);
@@ -290,16 +291,5 @@ static void removeFilesForMmTable(HiveConf conf, AcidDirectory dir) throws IOExc
fs.delete(dead, true);
}
}
-
- static void overrideConfProps(HiveConf conf, CompactionInfo ci, Map<String, String> properties) {
- Stream.of(properties, new StringableMap(ci.properties))
- .filter(Objects::nonNull)
- .flatMap(map -> map.entrySet().stream())
- .filter(entry -> entry.getKey().startsWith(COMPACTOR_PREFIX))
- .forEach(entry -> {
- String property = entry.getKey().substring(COMPACTOR_PREFIX.length());
- conf.set(property, entry.getValue());
- });
- }
}
}
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/StatsUpdater.java b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/StatsUpdater.java
index a3b940e74d5..c22d4c2c51d 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/StatsUpdater.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/StatsUpdater.java
@@ -22,6 +22,7 @@
import org.apache.hadoop.hive.metastore.IMetaStoreClient;
import org.apache.hadoop.hive.metastore.Warehouse;
import org.apache.hadoop.hive.metastore.txn.entities.CompactionInfo;
+import org.apache.hadoop.hive.metastore.utils.StringableMap;
import org.apache.hadoop.hive.ql.DriverUtils;
import org.apache.hadoop.hive.ql.session.SessionState;
import org.apache.hadoop.hive.ql.stats.StatsUtils;
@@ -32,6 +33,9 @@
import java.util.List;
import java.util.Map;
+import static org.apache.commons.lang3.StringUtils.isNotBlank;
+import static org.apache.hadoop.hive.ql.txn.compactor.CompactorUtil.overrideConfProps;
+
/**
* Updates table/partition statistics.
* Intended to run after a successful compaction.
@@ -52,7 +56,7 @@ public final class StatsUpdater {
* @param userName The user to run the statistic collection with
* @param compactionQueueName The name of the compaction queue
*/
- public void gatherStats(CompactionInfo ci, HiveConf hiveConf,
+ public void gatherStats(HiveConf hiveConf, CompactionInfo ci, Map<String, String> tableProperties,
String userName, String compactionQueueName,
IMetaStoreClient msc) {
try {
@@ -60,9 +64,8 @@ public void gatherStats(CompactionInfo ci, HiveConf hiveConf,
throw new IllegalArgumentException("Metastore client is missing");
}
- HiveConf conf = new HiveConf(hiveConf);
- //so that Driver doesn't think it's already in a transaction
- conf.unset(ValidTxnList.VALID_TXNS_KEY);
+ HiveConf conf = setUpDriverSession(
+ hiveConf, compactionQueueName, tableProperties, new StringableMap(ci.properties));
//e.g. analyze table page_view partition(dt='10/15/2014',country=’US’)
// compute statistics for columns viewtime
@@ -86,15 +89,33 @@ public void gatherStats(CompactionInfo ci, HiveConf hiveConf,
} else {
sb.append(" noscan");
}
- LOG.info(ci + ": running '" + sb + "'");
- if (compactionQueueName != null && compactionQueueName.length() > 0) {
- conf.set(TezConfiguration.TEZ_QUEUE_NAME, compactionQueueName);
- }
+
+ LOG.info("{}: running '{}'", ci, sb);
SessionState sessionState = DriverUtils.setUpAndStartSessionState(conf, userName);
DriverUtils.runOnDriver(conf, sessionState, sb.toString());
} catch (Throwable t) {
- LOG.error(ci + ": gatherStats(" + ci.dbname + "," + ci.tableName + "," + ci.partName +
- ") failed due to: " + t.getMessage(), t);
+ LOG.error("{}: gatherStats({},{},{}) failed due to: {}",
+ ci, ci.dbname, ci.tableName, ci.partName, t.getMessage(), t);
}
}
+
+ HiveConf setUpDriverSession(
+ HiveConf sourceConf,
+ String compactionQueueName,
+ Map<String, String> tableProperties,
+ Map<String, String> ciProperties) {
+
+ HiveConf conf = new HiveConf(sourceConf);
+
+ //so that Driver doesn't think it's already in a transaction
+ conf.unset(ValidTxnList.VALID_TXNS_KEY);
+
+ overrideConfProps(conf, ciProperties, tableProperties);
+
+ if (isNotBlank(compactionQueueName)) {
+ conf.set(TezConfiguration.TEZ_QUEUE_NAME, compactionQueueName);
+ }
+
+ return conf;
+ }
}
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java
index c5f478f7ba1..c08da2b86e6 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java
@@ -286,8 +286,13 @@ protected Boolean findNextCompactionAndExecute(boolean collectGenericStats, bool
}
if (success && compactionService.computeStats()) {
- statsUpdater.gatherStats(ci, conf, CompactorUtil.runJobAsSelf(ci.runAs) ? ci.runAs : table.getOwner(),
- CompactorUtil.getCompactorJobQueueName(conf, ci, table), msc);
+ statsUpdater.gatherStats(
+ conf,
+ ci,
+ table.getParameters(),
+ CompactorUtil.runJobAsSelf(ci.runAs) ? ci.runAs : table.getOwner(),
+ CompactorUtil.getCompactorJobQueueName(conf, ci, table),
+ msc);
}
return success;
}
diff --git a/ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestStatsUpdater.java b/ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestStatsUpdater.java
new file mode 100644
index 00000000000..0a51f30ae83
--- /dev/null
+++ b/ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestStatsUpdater.java
@@ -0,0 +1,97 @@
+/*
+ * 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.ql.txn.compactor;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static java.util.Collections.emptyMap;
+import static org.apache.hadoop.hive.ql.txn.compactor.CompactorUtil.COMPACTOR_PREFIX;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+public class TestStatsUpdater {
+ @Test
+ public void testSetUpDriverSessionAddsCompactorPropertiesFromTableProperties() {
+ StatsUpdater statsUpdater = new StatsUpdater();
+
+ HiveConf conf = new HiveConf();
+ Map<String, String> tableProperties = new HashMap<String, String>() {{
+ put(COMPACTOR_PREFIX + "test.property", "test-value");
+ }};
+
+ Map<String, String> ciProperties = new HashMap<>();
+ HiveConf statsUpdaterConf = statsUpdater.setUpDriverSession(conf, "testQueue", tableProperties, ciProperties);
+
+ assertThat(statsUpdaterConf.get("test.property"), is("test-value"));
+ }
+
+ @Test
+ public void testSetUpDriverSessionAddsCompactorPropertiesFromCompactionInfo() {
+ StatsUpdater statsUpdater = new StatsUpdater();
+
+ HiveConf conf = new HiveConf();
+ Map<String, String> tableProperties = new HashMap<>();
+ Map<String, String> ciProperties = new HashMap<String, String>() {{
+ put(COMPACTOR_PREFIX + "test.property", "test-value");
+ }};
+
+ HiveConf statsUpdaterConf = statsUpdater.setUpDriverSession(conf, "testQueue", tableProperties, ciProperties);
+
+ assertThat(statsUpdaterConf.get("test.property"), is("test-value"));
+ }
+
+ @Test
+ public void testSetUpDriverSessionOverridesExistingCompactorPropertiesWithTableProperties() {
+ StatsUpdater statsUpdater = new StatsUpdater();
+
+ HiveConf conf = new HiveConf();
+ conf.set(COMPACTOR_PREFIX + "test.property", "test-value");
+
+ Map<String, String> tableProperties = new HashMap<String, String>() {{
+ put(COMPACTOR_PREFIX + "test.property", "changed-in-table-properties");
+ }};
+
+ HiveConf statsUpdaterConf = statsUpdater.setUpDriverSession(conf, "testQueue", tableProperties, emptyMap());
+
+ assertThat(statsUpdaterConf.get("test.property"), is("changed-in-table-properties"));
+ }
+
+ @Test
+ public void testSetUpDriverSessionOverridesExistingCompactorPropertiesWithPropertiesInCompactionInfo() {
+ StatsUpdater statsUpdater = new StatsUpdater();
+
+ HiveConf conf = new HiveConf();
+ conf.set(COMPACTOR_PREFIX + "test.property", "test-value");
+
+ Map<String, String> tableProperties = new HashMap<String, String>() {{
+ put(COMPACTOR_PREFIX + "test.property", "changed-in-table-properties");
+ }};
+
+ Map<String, String> ciProperties = new HashMap<String, String>() {{
+ put(COMPACTOR_PREFIX + "test.property", "changed-in-compaction-info");
+ }};
+
+ HiveConf statsUpdaterConf = statsUpdater.setUpDriverSession(conf, "testQueue", tableProperties, ciProperties);
+
+ assertThat(statsUpdaterConf.get("test.property"), is("changed-in-compaction-info"));
+ }
+}
diff --git a/ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestWorker.java b/ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestWorker.java
index 9a833d504f1..bf01034711e 100644
--- a/ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestWorker.java
+++ b/ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestWorker.java
@@ -1051,7 +1051,7 @@ public void testDoesNotGatherStatsIfCompactionFails() throws Exception {
worker.findNextCompactionAndExecute(true, true);
- Mockito.verify(statsUpdater, Mockito.never()).gatherStats(any(), any(), any(), any(), any());
+ Mockito.verify(statsUpdater, Mockito.never()).gatherStats(any(), any(), any(), any(), any(), any());
}
@Test
diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/entities/CompactionInfo.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/entities/CompactionInfo.java
index d0039378ff5..4d24f1d23ed 100644
--- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/entities/CompactionInfo.java
+++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/entities/CompactionInfo.java
@@ -207,7 +207,6 @@ public boolean equals(Object obj) {
CompactionInfo info = (CompactionInfo) obj;
return this.compareTo(info) == 0;
}
-
/**
* loads object from a row in Select * from COMPACTION_QUEUE
* @param rs ResultSet after call to rs.next()
@@ -242,6 +241,7 @@ public static CompactionInfo loadFullFromCompactionQueue(ResultSet rs) throws SQ
fullCi.orderByClause = rs.getString(25);
return fullCi;
}
+
static void insertIntoCompletedCompactions(PreparedStatement pStmt, CompactionInfo ci, long endTime) throws SQLException, MetaException {
pStmt.setLong(1, ci.id);
pStmt.setString(2, ci.dbname);
@@ -339,7 +339,6 @@ public static CompactionInfoStruct compactionInfoToStruct(CompactionInfo ci) {
cr.setOrderByClause(ci.orderByClause);
return cr;
}
-
public static OptionalCompactionInfoStruct compactionInfoToOptionalStruct(CompactionInfo ci) {
CompactionInfoStruct cis = compactionInfoToStruct(ci);
OptionalCompactionInfoStruct ocis = new OptionalCompactionInfoStruct();
@@ -348,6 +347,7 @@ public static OptionalCompactionInfoStruct compactionInfoToOptionalStruct(Compac
}
return ocis;
}
+
public static CompactionInfo optionalCompactionInfoStructToInfo(OptionalCompactionInfoStruct ocis) {
if (ocis.isSetCi()) {
return compactionStructToInfo(ocis.getCi());