Skip to content

HIVE-29281: Make proactive cache eviction work with catalog #6379

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
Neer393 wants to merge 1 commit into
base: master
Choose a base branch
from
Open

HIVE-29281: Make proactive cache eviction work with catalog #6379

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
2 changes: 1 addition & 1 deletion iceberg/iceberg-handler/src/main/java/org/apache/iceberg/orc/VectorizedReadUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,7 @@ public static ByteBuffer getSerializedOrcTail(Path path, SyntheticFileId fileId,
// Note: Since Hive doesn't know about partition information of Iceberg tables, partitionDesc is only used to
// deduct the table (and DB) name here.
CacheTag cacheTag = HiveConf.getBoolVar(job, HiveConf.ConfVars.LLAP_TRACK_CACHE_USAGE) ?
LlapHiveUtils.getDbAndTableNameForMetrics(path, true, partitionDesc) : null;
LlapHiveUtils.getCacheTag(path, true, partitionDesc) : null;

try {
// Schema has to be serialized and deserialized as it is passed between different packages of TypeDescription:
Expand Down
3 changes: 2 additions & 1 deletion llap-common/src/protobuf/LlapDaemonProtocol.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -233,10 +233,11 @@ message SetCapacityRequestProto {
message SetCapacityResponseProto {
}

// Used for proactive eviction request. Must contain one DB name, and optionally table information.
// Used for proactive eviction request. Must contain a DB name, and optionally table information and catalog name.
message EvictEntityRequestProto {
required string db_name = 1;
repeated TableProto table = 2;
optional string catalog_name = 3 [default = "hive"];
}

// Used in EvictEntityRequestProto, can be used for non-partitioned and partitioned tables too.
Expand Down
16 changes: 12 additions & 4 deletions ...-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapCacheMetadataSerializer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
import org.apache.hadoop.hive.llap.io.encoded.LlapOrcCacheLoader;
import org.apache.hadoop.hive.ql.io.SyntheticFileId;
import org.apache.hadoop.hive.ql.io.orc.encoded.IoTrace;
import org.apache.hadoop.hive.ql.metadata.HiveUtils;
import org.apache.hive.common.util.FixedSizedObjectPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down Expand Up @@ -149,7 +150,7 @@ public void loadData(LlapDaemonProtocolProtos.CacheEntryList data) {
}

private void loadData(LlapDaemonProtocolProtos.CacheEntry ce) throws IOException {
CacheTag cacheTag = decodeCacheTag(ce.getCacheTag());
CacheTag cacheTag = decodeCacheTag(ce.getCacheTag(), conf);
DiskRangeList ranges = decodeRanges(ce.getRangesList());
Object fileKey = decodeFileKey(ce.getFileKey());
try (LlapOrcCacheLoader llr = new LlapOrcCacheLoader(new Path(ce.getFilePath()), fileKey, conf, cache,
Expand All @@ -167,9 +168,16 @@ private static DiskRangeList decodeRanges(List<LlapDaemonProtocolProtos.CacheEnt
return helper.get();
}

private static CacheTag decodeCacheTag(LlapDaemonProtocolProtos.CacheTag ct) {
return ct.getPartitionDescCount() == 0 ? CacheTag.build(ct.getTableName()) : CacheTag
.build(ct.getTableName(), ct.getPartitionDescList());
private static CacheTag decodeCacheTag(LlapDaemonProtocolProtos.CacheTag ct, Configuration conf) {
String tableName = ct.getTableName();
String[] parts = tableName.split("\\.");
if (parts.length == 2) {
// db.table without catalog, prepend current or default catalog
tableName = HiveUtils.getCurrentCatalogOrDefault(conf) + '.' + tableName;
}
return ct.getPartitionDescCount() == 0
? CacheTag.build(tableName)
: CacheTag.build(tableName, ct.getPartitionDescList());
}

@VisibleForTesting
Expand Down
9 changes: 6 additions & 3 deletions llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -324,9 +324,12 @@ public long evictEntity(LlapDaemonProtocolProtos.EvictEntityRequestProto protoRe
if (LOG.isDebugEnabled()) {
StringBuilder sb = new StringBuilder();
sb.append(markedBytes).append(" bytes marked for eviction from LLAP cache buffers that belong to table(s): ");
for (String table : request.getEntities().get(request.getSingleDbName()).keySet()) {
sb.append(table).append(" ");
}
request.getEntities().forEach((catalog, dbs) ->
dbs.forEach((db, tables) ->
tables.forEach((table, partitions) ->
sb.append(catalog + "." + db + "." + table).append(" "))
)
);
sb.append(" Duration: ").append(time).append(" ms");
LOG.debug(sb.toString());
}
Expand Down
2 changes: 1 addition & 1 deletion llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/OrcEncodedDataReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -235,7 +235,7 @@ public OrcEncodedDataReader(LowLevelCache lowLevelCache, BufferUsageManager buff
// LlapInputFormat needs to know the file schema to decide if schema evolution is supported.
PartitionDesc partitionDesc = LlapHiveUtils.partitionDescForPath(split.getPath(), parts);
cacheTag = HiveConf.getBoolVar(daemonConf, ConfVars.LLAP_TRACK_CACHE_USAGE)
? LlapHiveUtils.getDbAndTableNameForMetrics(split.getPath(), true, partitionDesc) : null;
? LlapHiveUtils.getCacheTag(split.getPath(), true, partitionDesc) : null;
// 1. Get file metadata from cache, or create the reader and read it.
// Don't cache the filesystem object for now; Tez closes it and FS cache will fix all that
fsSupplier = getFsSupplier(split.getPath(), jobConf);
Expand Down
2 changes: 1 addition & 1 deletion llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/SerDeEncodedDataReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -225,7 +225,7 @@ public MemoryBuffer create() {
PartitionDesc partitionDesc = LlapHiveUtils.partitionDescForPath(split.getPath(), parts);
fileKey = determineCacheKey(fs, split, partitionDesc, daemonConf);
cacheTag = HiveConf.getBoolVar(daemonConf, ConfVars.LLAP_TRACK_CACHE_USAGE)
? LlapHiveUtils.getDbAndTableNameForMetrics(split.getPath(), true, partitionDesc) : null;
? LlapHiveUtils.getCacheTag(split.getPath(), true, partitionDesc) : null;
this.sourceInputFormat = sourceInputFormat;
this.sourceSerDe = sourceSerDe;
this.reporter = reporter;
Expand Down
4 changes: 2 additions & 2 deletions llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcFileEstimateErrors.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import org.apache.hadoop.hive.common.io.DiskRangeList.MutateHelper;
import org.apache.hadoop.hive.common.io.CacheTag;
import org.apache.hadoop.hive.llap.cache.EvictionDispatcher;
import org.apache.hadoop.hive.metastore.Warehouse;
import org.apache.hadoop.hive.llap.cache.LlapCacheableBuffer;
import org.apache.hadoop.hive.ql.io.SyntheticFileId;
import org.apache.hadoop.hive.ql.io.orc.encoded.IncompleteCb;
Expand Down Expand Up @@ -140,7 +141,6 @@ public boolean isMarkedForEviction() {

@Override
public CacheTag getTag() {
// We don't care about these.
return CacheTag.build("OrcEstimates");
return CacheTag.build(Warehouse.DEFAULT_CATALOG_NAME, "OrcEstimates");
}
}
59 changes: 31 additions & 28 deletions llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestCacheContentsTracker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,11 +17,10 @@
*/
package org.apache.hadoop.hive.llap.cache;

import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;

import org.apache.hadoop.hive.common.io.CacheTag;
import org.apache.hadoop.hive.metastore.Warehouse;

import org.junit.BeforeClass;
import org.junit.Test;
Expand Down Expand Up @@ -127,7 +126,7 @@ public void testCacheTagComparison() {
public void testEncodingDecoding() throws Exception {
LinkedHashMap<String, String> partDescs = new LinkedHashMap<>();
partDescs.put("pytha=goras", "a2+b2=c2");
CacheTag tag = CacheTag.build("math.rules", partDescs);
CacheTag tag = CacheTag.build(Warehouse.DEFAULT_CATALOG_NAME + ".math.rules", partDescs);
CacheTag.SinglePartitionCacheTag stag = ((CacheTag.SinglePartitionCacheTag)tag);
assertEquals("pytha=goras=a2+b2=c2", stag.partitionDescToString());
assertEquals(1, stag.getPartitionDescMap().size());
Expand All @@ -136,7 +135,7 @@ public void testEncodingDecoding() throws Exception {
partDescs.clear();
partDescs.put("mutli=one", "one=/1");
partDescs.put("mutli=two/", "two=2");
tag = CacheTag.build("math.rules", partDescs);
tag = CacheTag.build(Warehouse.DEFAULT_CATALOG_NAME + ".math.rules", partDescs);
Copy link
Copy Markdown
Member

@deniskuzZ deniskuzZ Apr 7, 2026 •
edited
Loading

Choose a reason for hiding this comment

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

why not handle DEFAULT_CATALOG_NAME inside CacheTag.build when size ==2?
you wouldn't need CacheTag.build(Warehouse.DEFAULT_CATALOG_NAME, "math.rules") either

CacheTag.MultiPartitionCacheTag mtag = ((CacheTag.MultiPartitionCacheTag)tag);
assertEquals("mutli=one=one=/1/mutli=two/=two=2", mtag.partitionDescToString());
assertEquals(2, mtag.getPartitionDescMap().size());
Expand Down Expand Up @@ -168,6 +167,10 @@ private static LlapCacheableBuffer createMockBuffer(long size, CacheTag cacheTag
}

public static CacheTag cacheTagBuilder(String dbAndTable, String... partitions) {
String[] parts = dbAndTable.split("\\.");
if(parts.length < 3) {
Copy link
Copy Markdown
Member

@deniskuzZ deniskuzZ Apr 7, 2026

Choose a reason for hiding this comment

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

nit: space

dbAndTable = Warehouse.DEFAULT_CATALOG_NAME + "." + dbAndTable;
}
if (partitions != null && partitions.length > 0) {
LinkedHashMap<String, String> partDescs = new LinkedHashMap<>();
for (String partition : partitions) {
Expand Down Expand Up @@ -215,33 +218,33 @@ private static void evictSomeTestBuffers() {
private static final String EXPECTED_CACHE_STATE_WHEN_FULL =
"\n" +
"Cache state: \n" +
"default : 2/2, 2101248/2101248\n" +
"default.testtable : 2/2, 2101248/2101248\n" +
"otherdb : 7/7, 1611106304/1611106304\n" +
"otherdb.testtable : 4/4, 231424/231424\n" +
"otherdb.testtable/p=v1 : 3/3, 100352/100352\n" +
"otherdb.testtable/p=v1/pp=vv1 : 2/2, 34816/34816\n" +
"otherdb.testtable/p=v1/pp=vv2 : 1/1, 65536/65536\n" +
"otherdb.testtable/p=v2 : 1/1, 131072/131072\n" +
"otherdb.testtable/p=v2/pp=vv1 : 1/1, 131072/131072\n" +
"otherdb.testtable2 : 2/2, 537133056/537133056\n" +
"otherdb.testtable2/p=v3 : 2/2, 537133056/537133056\n" +
"otherdb.testtable3 : 1/1, 1073741824/1073741824";
"hive.default : 2/2, 2101248/2101248\n" +
"hive.default.testtable : 2/2, 2101248/2101248\n" +
"hive.otherdb : 7/7, 1611106304/1611106304\n" +
"hive.otherdb.testtable : 4/4, 231424/231424\n" +
"hive.otherdb.testtable/p=v1 : 3/3, 100352/100352\n" +
"hive.otherdb.testtable/p=v1/pp=vv1 : 2/2, 34816/34816\n" +
"hive.otherdb.testtable/p=v1/pp=vv2 : 1/1, 65536/65536\n" +
"hive.otherdb.testtable/p=v2 : 1/1, 131072/131072\n" +
"hive.otherdb.testtable/p=v2/pp=vv1 : 1/1, 131072/131072\n" +
"hive.otherdb.testtable2 : 2/2, 537133056/537133056\n" +
"hive.otherdb.testtable2/p=v3 : 2/2, 537133056/537133056\n" +
"hive.otherdb.testtable3 : 1/1, 1073741824/1073741824";

private static final String EXPECTED_CACHE_STATE_AFTER_EVICTION =
"\n" +
"Cache state: \n" +
"default : 0/2, 0/2101248\n" +
"default.testtable : 0/2, 0/2101248\n" +
"otherdb : 5/7, 1074202624/1611106304\n" +
"otherdb.testtable : 3/4, 198656/231424\n" +
"otherdb.testtable/p=v1 : 2/3, 67584/100352\n" +
"otherdb.testtable/p=v1/pp=vv1 : 1/2, 2048/34816\n" +
"otherdb.testtable/p=v1/pp=vv2 : 1/1, 65536/65536\n" +
"otherdb.testtable/p=v2 : 1/1, 131072/131072\n" +
"otherdb.testtable/p=v2/pp=vv1 : 1/1, 131072/131072\n" +
"otherdb.testtable2 : 1/2, 262144/537133056\n" +
"otherdb.testtable2/p=v3 : 1/2, 262144/537133056\n" +
"otherdb.testtable3 : 1/1, 1073741824/1073741824";
"hive.default : 0/2, 0/2101248\n" +
"hive.default.testtable : 0/2, 0/2101248\n" +
"hive.otherdb : 5/7, 1074202624/1611106304\n" +
"hive.otherdb.testtable : 3/4, 198656/231424\n" +
"hive.otherdb.testtable/p=v1 : 2/3, 67584/100352\n" +
"hive.otherdb.testtable/p=v1/pp=vv1 : 1/2, 2048/34816\n" +
"hive.otherdb.testtable/p=v1/pp=vv2 : 1/1, 65536/65536\n" +
"hive.otherdb.testtable/p=v2 : 1/1, 131072/131072\n" +
"hive.otherdb.testtable/p=v2/pp=vv1 : 1/1, 131072/131072\n" +
"hive.otherdb.testtable2 : 1/2, 262144/537133056\n" +
"hive.otherdb.testtable2/p=v3 : 1/2, 262144/537133056\n" +
"hive.otherdb.testtable3 : 1/1, 1073741824/1073741824";

}
3 changes: 2 additions & 1 deletion llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestFileCache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

import com.google.common.base.Function;
import org.apache.hadoop.hive.common.io.CacheTag;
import org.apache.hadoop.hive.metastore.Warehouse;
import org.junit.Test;

import java.util.concurrent.ConcurrentHashMap;
Expand All @@ -32,7 +33,7 @@ public void testFileCacheMetadata() {
ConcurrentHashMap<Object, FileCache<Object>> cache = new ConcurrentHashMap<>();
Object fileKey = 1234L;
Function<Void, Object> f = a -> new Object();
CacheTag tag = CacheTag.build("test_table");
CacheTag tag = CacheTag.build(Warehouse.DEFAULT_CATALOG_NAME, "test_db.test_table");

FileCache<Object> result = FileCache.getOrAddFileSubCache(cache, fileKey, f, tag);

Expand Down
8 changes: 5 additions & 3 deletions llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestLowLevelCacheImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@
import org.slf4j.LoggerFactory;

import org.apache.hadoop.hive.common.io.CacheTag;
import org.apache.hadoop.hive.metastore.Warehouse;
import org.apache.hadoop.hive.common.io.DiskRange;
import org.apache.hadoop.hive.common.io.DiskRangeList;
import org.apache.hadoop.hive.common.io.DataCache.DiskRangeListFactory;
Expand Down Expand Up @@ -309,13 +310,14 @@ private void _testProactiveEvictionMark(boolean isInstantDeallocation) {

LlapDataBuffer[] buffs1 = IntStream.range(0, 4).mapToObj(i -> fb()).toArray(LlapDataBuffer[]::new);
DiskRange[] drs1 = drs(IntStream.range(1, 5).toArray());
CacheTag tag1 = CacheTag.build("default.table1");
CacheTag tag1 = CacheTag.build(Warehouse.DEFAULT_CATALOG_NAME, "default.table1");

LlapDataBuffer[] buffs2 = IntStream.range(0, 41).mapToObj(i -> fb()).toArray(LlapDataBuffer[]::new);
DiskRange[] drs2 = drs(IntStream.range(1, 42).toArray());
CacheTag tag2 = CacheTag.build("default.table2");
CacheTag tag2 = CacheTag.build(Warehouse.DEFAULT_CATALOG_NAME, "default.table2");

Predicate<CacheTag> predicate = tag -> "default.table1".equals(tag.getTableName());
Predicate<CacheTag> predicate = tag ->
(Warehouse.DEFAULT_CATALOG_NAME + "." + "default.table1").equals(tag.getTableName());

cache.putFileData(fn1, drs1, buffs1, 0, Priority.NORMAL, null, tag1);
cache.putFileData(fn2, drs2, buffs2, 0, Priority.NORMAL, null, tag2);
Expand Down
19 changes: 12 additions & 7 deletions llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestOrcMetadataCache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@

import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.llap.IllegalCacheConfigurationException;
import org.apache.hadoop.hive.metastore.Warehouse;
import org.apache.hadoop.hive.llap.cache.LowLevelCache.Priority;
import org.apache.hadoop.hive.llap.io.encoded.OrcEncodedDataReader;
import org.apache.hadoop.hive.llap.io.metadata.MetadataCache;
Expand Down Expand Up @@ -249,7 +250,7 @@ public void testGetOrcTailForPath() throws Exception {
Path path = new Path("../data/files/alltypesorc");
Configuration jobConf = new Configuration();
Configuration daemonConf = new Configuration();
CacheTag tag = CacheTag.build("test-table");
CacheTag tag = CacheTag.build(Warehouse.DEFAULT_CATALOG_NAME, "test-db.test-table");
OrcTail uncached = OrcEncodedDataReader.getOrcTailForPath(path, jobConf, tag, daemonConf, cache, null);
jobConf.set(HiveConf.ConfVars.LLAP_IO_CACHE_ONLY.varname, "true");
OrcTail cached = OrcEncodedDataReader.getOrcTailForPath(path, jobConf, tag, daemonConf, cache, null);
Expand All @@ -270,7 +271,7 @@ public void testGetOrcTailForPathWithFileId() throws Exception {
Path path = new Path("../data/files/alltypesorc");
Configuration jobConf = new Configuration();
Configuration daemonConf = new Configuration();
CacheTag tag = CacheTag.build("test-table");
CacheTag tag = CacheTag.build(Warehouse.DEFAULT_CATALOG_NAME, "test-db.test-table");
FileSystem fs = FileSystem.get(daemonConf);
FileStatus fileStatus = fs.getFileStatus(path);
OrcTail uncached = OrcEncodedDataReader.getOrcTailForPath(fileStatus.getPath(), jobConf, tag, daemonConf, cache, new SyntheticFileId(fileStatus));
Expand All @@ -294,7 +295,7 @@ public void testGetOrcTailForPathWithFileIdChange() throws Exception {
Path path = new Path("../data/files/alltypesorc");
Configuration jobConf = new Configuration();
Configuration daemonConf = new Configuration();
CacheTag tag = CacheTag.build("test-table");
CacheTag tag = CacheTag.build(Warehouse.DEFAULT_CATALOG_NAME, "test-db.test-table");
OrcEncodedDataReader.getOrcTailForPath(path, jobConf, tag, daemonConf, cache, new SyntheticFileId(path, 100, 100));
jobConf.set(HiveConf.ConfVars.LLAP_IO_CACHE_ONLY.varname, "true");
Exception ex = null;
Expand Down Expand Up @@ -337,19 +338,23 @@ public void testProactiveEvictionMark() throws Exception {
// below is of length 65
ByteBuffer bb2 = ByteBuffer.wrap("-large-meta-data-content-large-meta-data-content-large-meta-data-".getBytes());

LlapBufferOrBuffers table1Buffers1 = cache.putFileMetadata(fn1, bb, CacheTag.build("default.table1"), isStopped);
LlapBufferOrBuffers table1Buffers1 = cache.putFileMetadata(fn1, bb,
CacheTag.build(Warehouse.DEFAULT_CATALOG_NAME, "default.table1"), isStopped);
assertNotNull(table1Buffers1.getSingleLlapBuffer());

LlapBufferOrBuffers table1Buffers2 = cache.putFileMetadata(fn2, bb2, CacheTag.build("default.table1"), isStopped);
LlapBufferOrBuffers table1Buffers2 = cache.putFileMetadata(fn2, bb2,
CacheTag.build(Warehouse.DEFAULT_CATALOG_NAME, "default.table1"), isStopped);
assertNotNull(table1Buffers2.getMultipleLlapBuffers());
assertEquals(2, table1Buffers2.getMultipleLlapBuffers().length);

// Case for when metadata consists of just 1 buffer (most of the realworld cases)
ByteBuffer bb3 = ByteBuffer.wrap("small-meta-data-content-for-otherFile".getBytes());
LlapBufferOrBuffers table2Buffers1 = cache.putFileMetadata(fn3, bb3, CacheTag.build("default.table2"), isStopped);
LlapBufferOrBuffers table2Buffers1 = cache.putFileMetadata(fn3, bb3,
CacheTag.build(Warehouse.DEFAULT_CATALOG_NAME, "default.table2"), isStopped);
assertNotNull(table2Buffers1.getSingleLlapBuffer());

Predicate<CacheTag> predicate = tag -> "default.table1".equals(tag.getTableName());
Predicate<CacheTag> predicate = tag ->
(Warehouse.DEFAULT_CATALOG_NAME + ".default.table1").equals(tag.getTableName());

// Simulating eviction on some buffers
table1Buffers2.getMultipleLlapBuffers()[1].decRef();
Expand Down
Loading
Loading