diff --git a/ql/src/java/org/apache/hadoop/hive/ql/Driver.java b/ql/src/java/org/apache/hadoop/hive/ql/Driver.java index 9f4e6f2e53..52799b30c3 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/Driver.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/Driver.java @@ -1977,10 +1977,14 @@ private void preExecutionCacheActions() throws Exception { if (cacheUsage != null) { if (cacheUsage.getStatus() == CacheUsage.CacheStatus.CAN_CACHE_QUERY_RESULTS && plan.getFetchTask() != null) { + ValidTxnWriteIdList txnWriteIdList = null; + if (plan.hasAcidResourcesInQuery()) { + txnWriteIdList = AcidUtils.getValidTxnWriteIdList(conf); + } // The results of this query execution might be cacheable. // Add a placeholder entry in the cache so other queries know this result is pending. CacheEntry pendingCacheEntry = - QueryResultsCache.getInstance().addToCache(cacheUsage.getQueryInfo()); + QueryResultsCache.getInstance().addToCache(cacheUsage.getQueryInfo(), txnWriteIdList); if (pendingCacheEntry != null) { // Update cacheUsage to reference the pending entry. this.cacheUsage.setCacheEntry(pendingCacheEntry); @@ -2011,8 +2015,7 @@ private void postExecutionCacheActions() throws Exception { } boolean savedToCache = QueryResultsCache.getInstance().setEntryValid( cacheUsage.getCacheEntry(), - plan.getFetchTask().getWork(), - txnWriteIdList); + plan.getFetchTask().getWork()); LOG.info("savedToCache: {}", savedToCache); if (savedToCache) { useFetchFromCache(cacheUsage.getCacheEntry()); diff --git a/ql/src/java/org/apache/hadoop/hive/ql/cache/results/QueryResultsCache.java b/ql/src/java/org/apache/hadoop/hive/ql/cache/results/QueryResultsCache.java index 6734ac50f8..11b9c27126 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/cache/results/QueryResultsCache.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/cache/results/QueryResultsCache.java @@ -488,11 +488,12 @@ public CacheEntry lookup(LookupInfo request) { * @param queryInfo * @return */ - public CacheEntry addToCache(QueryInfo queryInfo) { + public CacheEntry addToCache(QueryInfo queryInfo, ValidTxnWriteIdList txnWriteIdList) { // Create placeholder entry with PENDING state. String queryText = queryInfo.getLookupInfo().getQueryText(); CacheEntry addedEntry = new CacheEntry(); addedEntry.queryInfo = queryInfo; + addedEntry.txnWriteIdList = txnWriteIdList; Lock writeLock = rwLock.writeLock(); try { @@ -522,7 +523,7 @@ public CacheEntry addToCache(QueryInfo queryInfo) { * @param fetchWork * @return */ - public boolean setEntryValid(CacheEntry cacheEntry, FetchWork fetchWork, ValidTxnWriteIdList txnWriteIdList) { + public boolean setEntryValid(CacheEntry cacheEntry, FetchWork fetchWork) { String queryText = cacheEntry.getQueryText(); boolean dataDirMoved = false; Path queryResultsPath = null; @@ -571,7 +572,6 @@ public boolean setEntryValid(CacheEntry cacheEntry, FetchWork fetchWork, ValidTx cacheEntry.cachedResultsPath = cachedResultsPath; cacheEntry.size = resultSize; this.cacheSize += resultSize; - cacheEntry.txnWriteIdList = txnWriteIdList; cacheEntry.setStatus(CacheEntryStatus.VALID); // Mark this entry as being in use. Caller will need to release later. @@ -700,6 +700,14 @@ private boolean entryMatches(LookupInfo lookupInfo, CacheEntry entry, Set