Skip to content

HIVE-29820: Refactor and improve ObjectCache in some RecordProcessors #6709

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
abstractdog wants to merge 1 commit into
base: master
Choose a base branch
from
Open

HIVE-29820: Refactor and improve ObjectCache in some RecordProcessors #6709

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 6 additions & 10 deletions ql/src/java/org/apache/hadoop/hive/ql/exec/ObjectCacheFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,14 +19,14 @@

package org.apache.hadoop.hive.ql.exec;

import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.llap.io.api.LlapProxy;
import org.apache.hadoop.hive.llap.io.api.LlapProxy;
import org.apache.hadoop.hive.ql.exec.tez.LlapObjectCache;

/**
Expand Down Expand Up @@ -92,15 +92,11 @@ private static boolean isLlapCacheEnabled(Configuration conf, boolean isPlanCach
private static ObjectCache getLlapObjectCache(String queryId) {
// If order of events (i.e. dagstart and fragmentstart) was guaranteed, we could just
// create the cache when dag starts, and blindly return it to execution here.
if (queryId == null) throw new RuntimeException("Query ID cannot be null");
ObjectCache result = llapQueryCaches.get(queryId);
if (result != null) return result;
result = new LlapObjectCache();
ObjectCache old = llapQueryCaches.putIfAbsent(queryId, result);
if (old == null) {
LOG.info("Created object cache for " + queryId);
}
return (old != null) ? old : result;
Objects.requireNonNull(queryId, "Query ID cannot be null");
return llapQueryCaches.computeIfAbsent(queryId, k -> {
LOG.info("Created object cache for {}", k);
return new LlapObjectCache();
});
}

public static void removeLlapQueryCache(String queryId) {
Expand Down
32 changes: 5 additions & 27 deletions ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MapRecordProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,8 +46,6 @@
import org.apache.hadoop.hive.ql.exec.HashTableDummyOperator;
import org.apache.hadoop.hive.ql.exec.MapOperator;
import org.apache.hadoop.hive.ql.exec.MapredContext;
import org.apache.hadoop.hive.ql.exec.ObjectCache;
import org.apache.hadoop.hive.ql.exec.ObjectCacheFactory;
import org.apache.hadoop.hive.ql.exec.Operator;
import org.apache.hadoop.hive.ql.exec.OperatorUtils;
import org.apache.hadoop.hive.ql.exec.TezDummyStoreOperator;
Expand Down Expand Up @@ -93,20 +91,14 @@
private final ExecMapperContext execContext;
private MapWork mapWork;
private List<MapWork> mergeWorkList;
private final List<String> cacheKeys = new ArrayList<>();
private final List<String> dynamicValueCacheKeys = new ArrayList<>();
private final ObjectCache cache, dynamicValueCache;
// is this part of the query-based compaction process
private final boolean isInCompaction;

public MapRecordProcessor(final JobConf jconf, final ProcessorContext context) throws Exception {
super(jconf, context);
String queryId = HiveConf.getVar(jconf, HiveConf.ConfVars.HIVE_QUERY_ID);
if (LlapProxy.isDaemon()) {
setLlapOfFragmentId(context);
}
cache = ObjectCacheFactory.getCache(jconf, queryId, true);
dynamicValueCache = ObjectCacheFactory.getCache(jconf, queryId, false, true);
execContext = new ExecMapperContext(jconf);
execContext.setJc(jconf);
isInCompaction = CompactorUtil.COMPACTOR.equalsIgnoreCase(
Expand All @@ -129,12 +121,10 @@


String key = processorContext.getTaskVertexName() + MAP_PLAN_KEY;
cacheKeys.add(key);


// create map and fetch operators
if (!isInCompaction) {
mapWork = cache.retrieve(key, () -> Utilities.getMapWork(jconf));
mapWork = planCache.retrieve(key, () -> Utilities.getMapWork(jconf));
} else {
// During query-based compaction, we don't want to retrieve old MapWork from the cache, we want a new mapper
// and new UDF validate_acid_sort_order instance for each bucket, otherwise validate_acid_sort_order will fail.
Expand All @@ -160,11 +150,10 @@
}

key = processorContext.getTaskVertexName() + prefix;
cacheKeys.add(key);

checkAbortCondition();
mergeWorkList.add(
(MapWork) cache.retrieve(key, () -> Utilities.getMergeWork(jconf, prefix)));
(MapWork) planCache.retrieve(key, () -> Utilities.getMergeWork(jconf, prefix)));
}
}

Expand Down Expand Up @@ -307,9 +296,8 @@
checkAbortCondition();
String valueRegistryKey = DynamicValue.DYNAMIC_VALUE_REGISTRY_CACHE_KEY;
// On LLAP dynamic value registry might already be cached.
final DynamicValueRegistryTez registryTez = dynamicValueCache.retrieve(valueRegistryKey,
() -> new DynamicValueRegistryTez());
dynamicValueCacheKeys.add(valueRegistryKey);
final DynamicValueRegistryTez registryTez =
dynamicValueCache.retrieve(valueRegistryKey, () -> new DynamicValueRegistryTez());

Check warning on line 300 in ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MapRecordProcessor.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Replace this lambda with method reference 'DynamicValueRegistryTez::new'. 


See more on https://sonarcloud.io/project/issues?id=apache_hive&issues=AaAaeyclRUKBEW7t0nkI&open=AaAaeyclRUKBEW7t0nkI&pullRequest=6709
RegistryConfTez registryConf = new RegistryConfTez(jconf, mapWork, processorContext, inputs);
registryTez.init(registryConf);

Expand Down Expand Up @@ -442,17 +430,7 @@
setAborted(execContext.getIoCxt().getIOExceptions());
}

if (cache != null) {
for (String k: cacheKeys) {
cache.release(k);
}
}

if (dynamicValueCache != null) {
for (String k: dynamicValueCacheKeys) {
dynamicValueCache.release(k);
}
}
releaseCache();

// detecting failed executions by exceptions thrown by the operator tree
try {
Expand Down
20 changes: 2 additions & 18 deletions ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MergeFileRecordProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,9 +25,7 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.ql.exec.MapredContext;
import org.apache.hadoop.hive.ql.exec.ObjectCacheFactory;
import org.apache.hadoop.hive.ql.exec.Operator;
import org.apache.hadoop.hive.ql.exec.OperatorUtils;
import org.apache.hadoop.hive.ql.exec.Utilities;
Expand Down Expand Up @@ -59,11 +57,9 @@
protected Operator<? extends OperatorDesc> mergeOp;
private ExecMapperContext execContext = null;
protected static final String MAP_PLAN_KEY = "__MAP_PLAN__";
private String cacheKey;
private MergeFileWork mfWork;
MRInputLegacy mrInput = null;
private final Object[] row = new Object[2];
org.apache.hadoop.hive.ql.exec.ObjectCache cache;

public MergeFileRecordProcessor(final JobConf jconf, final ProcessorContext context) {
super(jconf, context);
Expand Down Expand Up @@ -94,20 +90,10 @@
.initialize();
}

String queryId = HiveConf.getVar(jconf, HiveConf.ConfVars.HIVE_QUERY_ID);
cache = ObjectCacheFactory.getCache(jconf, queryId, true);

try {
execContext.setJc(jconf);

cacheKey = MAP_PLAN_KEY;

MapWork mapWork = (MapWork) cache.retrieve(cacheKey, new Callable<Object>() {
@Override
public Object call() {
return Utilities.getMapWork(jconf);
}
});
MapWork mapWork = (MapWork) planCache.retrieve(MAP_PLAN_KEY, (Callable<Object>) () -> Utilities.getMapWork(jconf));

Check warning on line 96 in ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MergeFileRecordProcessor.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Line is longer than 120 characters (found 121). 


See more on https://sonarcloud.io/project/issues?id=apache_hive&issues=AaAaeyunRUKBEW7t0nkJ&open=AaAaeyunRUKBEW7t0nkJ&pullRequest=6709

@Aggarwal-Raghav Aggarwal-Raghav Aug 19, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: Extra casting ? in other hunks, the Callable / MapWork is not present in retrieve().

MapWork mapWork = planCache.retrieve(MAP_PLAN_KEY, () -> Utilities.getMapWork(jconf));

@Aggarwal-Raghav Aggarwal-Raghav Aug 19, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I just did java level review, The LLAP is not my forte.

Utilities.setMapWork(jconf, mapWork);

if (mapWork instanceof MergeFileWork) {
Expand Down Expand Up @@ -162,9 +148,7 @@
@Override
void close() {

if (cache != null && cacheKey != null) {
cache.release(cacheKey);
}
releaseCache();

// check if there are IOExceptions
if (!isAborted()) {
Expand Down
90 changes: 75 additions & 15 deletions ql/src/java/org/apache/hadoop/hive/ql/exec/tez/RecordProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,8 +22,11 @@
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.Callable;

import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.ql.exec.ObjectCache;
import org.apache.hadoop.hive.ql.exec.ObjectCacheFactory;
import org.apache.hadoop.hive.ql.exec.Utilities;
import org.apache.hadoop.hive.ql.exec.tez.TezProcessor.TezKVOutputCollector;
import org.apache.hadoop.hive.ql.log.PerfLogger;
Expand Down Expand Up @@ -59,9 +62,32 @@ public abstract class RecordProcessor extends InterruptibleProcessing {
protected PerfLogger perfLogger = SessionState.getPerfLogger();
protected String CLASS_NAME = RecordProcessor.class.getName();

protected final String queryId;

/**
* Per-processor plan cache — no daemon-wide sharing. Sharing would race on
* per-fragment operator state that {@code initializeOp()} resets (HIVE-14433:
* {@code FileSinkOperator.fsp}, {@code VectorGroupByOperator.aggregator},
* {@code Operator.childOperatorsArray}, {@code VectorTopNKeyOperator} filter
* state, ...) and yields NPE / {@code FileAlreadyExistsException}.
*/
protected final TrackedCache planCache;

/**
* Dynamic value cache. On LLAP this is the per-query daemon-wide
* {@link LlapObjectCache}, so dynamic values computed once (e.g. broadcast
* hash tables, DPP registries) are reused across fragments of the same query.
*/
protected final TrackedCache dynamicValueCache;

public RecordProcessor(JobConf jConf, ProcessorContext processorContext) {
this.jconf = jConf;
this.processorContext = processorContext;
this.queryId = HiveConf.getVar(jConf, HiveConf.ConfVars.HIVE_QUERY_ID);
this.planCache = new TrackedCache(
ObjectCacheFactory.getCache(jConf, queryId, true, false));
this.dynamicValueCache = new TrackedCache(
ObjectCacheFactory.getCache(jConf, queryId, false, true));
}

/**
Expand Down Expand Up @@ -98,26 +124,60 @@ protected void createOutputMap() {
}
}

public List<BaseWork> getMergeWorkList(final JobConf jconf, String key, String queryId,
ObjectCache cache, List<String> cacheKeys) throws HiveException {
/**
* Release every key retrieved through the plan and dynamic-value caches.
* A no-op for {@link LlapObjectCache} (which relies on soft references), but
* preserved for correctness against other {@link ObjectCache} implementations.
*/
protected void releaseCache() {
planCache.releaseAll();
dynamicValueCache.releaseAll();
}

public List<BaseWork> getMergeWorkList(final JobConf jconf) throws HiveException {
String prefixes = jconf.get(DagUtils.TEZ_MERGE_WORK_FILE_PREFIXES);
if (prefixes != null) {
List<BaseWork> mergeWorkList = new ArrayList<>();
if (prefixes == null) {
return null;
}
List<BaseWork> mergeWorkList = new ArrayList<>();
for (final String prefix : prefixes.split(",")) {
if (prefix.isEmpty()) {
continue;
}
mergeWorkList.add(planCache.retrieve(prefix, () -> Utilities.getMergeWork(jconf, prefix)));
}
return mergeWorkList;
}

/**
* An {@link ObjectCache} paired with the set of keys retrieved through it, so
* {@link #releaseAll()} releases exactly those keys at close time. All retrievals
* that need to be released should go through {@link #retrieve} — calling
* {@link ObjectCache#retrieve} directly on the underlying cache bypasses the
* tracking and leaks the key.
*/
protected static final class TrackedCache {
private final ObjectCache cache;
private final List<String> keys = new ArrayList<>();

for (final String prefix : prefixes.split(",")) {
if (prefix.isEmpty()) {
continue;
}
TrackedCache(ObjectCache cache) {
this.cache = cache;
}

key = prefix;
cacheKeys.add(key);
/** Retrieve (or compute) the value for {@code key}, tracking it for release. */
<T> T retrieve(String key, Callable<T> fn) throws HiveException {
keys.add(key);
return cache.retrieve(key, fn);
}

mergeWorkList.add(cache.retrieve(key, () -> Utilities.getMergeWork(jconf, prefix)));
/** Release every key retrieved through this wrapper. Null-safe on the underlying cache. */
void releaseAll() {
if (cache == null) {
return;
}
for (String k : keys) {
cache.release(k);
}

return mergeWorkList;
} else {
return null;
}
}
}
31 changes: 3 additions & 28 deletions ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,12 +30,9 @@
import org.apache.hadoop.hive.llap.LlapUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.ql.exec.DummyStoreOperator;
import org.apache.hadoop.hive.ql.exec.HashTableDummyOperator;
import org.apache.hadoop.hive.ql.exec.MapredContext;
import org.apache.hadoop.hive.ql.exec.ObjectCache;
import org.apache.hadoop.hive.ql.exec.ObjectCacheFactory;
import org.apache.hadoop.hive.ql.exec.Operator;
import org.apache.hadoop.hive.ql.exec.OperatorUtils;
import org.apache.hadoop.hive.ql.exec.Utilities;
Expand All @@ -56,8 +53,6 @@
import org.apache.tez.runtime.api.ProcessorContext;
import org.apache.tez.runtime.api.Reader;

import com.google.common.collect.Lists;

/**
* Process input from tez LogicalInput and write output - for a map plan
* Just pump the records through the query plan.
Expand All @@ -67,13 +62,9 @@ public class ReduceRecordProcessor extends RecordProcessor {

private static final String REDUCE_PLAN_KEY = "__REDUCE_PLAN__";

private final ObjectCache cache, dynamicValueCache;

private ReduceWork reduceWork;

private final List<BaseWork> mergeWorkList;
private final List<String> cacheKeys;
private final List<String> dynamicValueCacheKeys = new ArrayList<>();

private final Map<Integer, DummyStoreOperator> connectOps = new TreeMap<>();
private final Map<Integer, ReduceWork> tagToReducerMap = new HashMap<>();
Expand All @@ -87,16 +78,11 @@ public class ReduceRecordProcessor extends RecordProcessor {
public ReduceRecordProcessor(final JobConf jconf, final ProcessorContext context) throws Exception {
super(jconf, context);

String queryId = HiveConf.getVar(jconf, HiveConf.ConfVars.HIVE_QUERY_ID);
cache = ObjectCacheFactory.getCache(jconf, queryId, true);
dynamicValueCache = ObjectCacheFactory.getCache(jconf, queryId, false, true);

String cacheKey = processorContext.getTaskVertexName() + REDUCE_PLAN_KEY;
cacheKeys = Lists.newArrayList(cacheKey);
reduceWork = cache.retrieve(cacheKey, () -> Utilities.getReduceWork(jconf));
reduceWork = planCache.retrieve(cacheKey, () -> Utilities.getReduceWork(jconf));

Utilities.setReduceWork(jconf, reduceWork);
mergeWorkList = getMergeWorkList(jconf, cacheKey, queryId, cache, cacheKeys);
mergeWorkList = getMergeWorkList(jconf);
}

@Override
Expand Down Expand Up @@ -161,7 +147,6 @@ void init(MRTaskReporter mrReporter, Map<String, LogicalInput> inputs, Map<Strin
String valueRegistryKey = DynamicValue.DYNAMIC_VALUE_REGISTRY_CACHE_KEY;
DynamicValueRegistryTez registryTez =
dynamicValueCache.retrieve(valueRegistryKey, () -> new DynamicValueRegistryTez());
dynamicValueCacheKeys.add(valueRegistryKey);
RegistryConfTez registryConf = new RegistryConfTez(jconf, reduceWork, processorContext, inputs);
registryTez.init(registryConf);
checkAbortCondition();
Expand Down Expand Up @@ -339,17 +324,7 @@ private List<LogicalInput> getShuffleInputs(Map<String, LogicalInput> inputs) th

@Override
void close() {
if (cache != null) {
for (String key : cacheKeys) {
cache.release(key);
}
}

if (dynamicValueCache != null) {
for (String k : dynamicValueCacheKeys) {
dynamicValueCache.release(k);
}
}
releaseCache();

try {
boolean abort = isAborted();
Expand Down
Loading
Loading