HIVE-29281: Make proactive cache eviction work with catalog

Author: Neer393

hcatalog/core/src/main/java/org/apache/hive/hcatalog/common/HCatUtil.java

@@ -423,7 +423,7 @@ public static Pair<String, String> getDbAndTableName(String tableName) throws IO
     Properties props = inputJobInfo.getTableInfo().getStorerInfo().getProperties();
     props.put(serdeConstants.SERIALIZATION_LIB,storageHandler.getSerDeClass().getName());
     TableDesc tableDesc = new TableDesc(storageHandler.getInputFormatClass(),
-      storageHandler.getOutputFormatClass(),props);
+      storageHandler.getOutputFormatClass(), props, null);
     if (tableDesc.getJobProperties() == null) {
       tableDesc.setJobProperties(new HashMap<>());
     }
@@ -464,7 +464,7 @@ public static Pair<String, String> getDbAndTableName(String tableName) throws IO
     Properties props = outputJobInfo.getTableInfo().getStorerInfo().getProperties();
     props.put(serdeConstants.SERIALIZATION_LIB,storageHandler.getSerDeClass().getName());
     TableDesc tableDesc = new TableDesc(storageHandler.getInputFormatClass(),
-        IgnoreKeyTextOutputFormat.class,props);
+        IgnoreKeyTextOutputFormat.class, props, null);
     if (tableDesc.getJobProperties() == null)
       tableDesc.setJobProperties(new HashMap<>());
     for (Map.Entry<String, String> el : conf) {

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/orc/VectorizedReadUtils.java

@@ -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.getDbAndTableNameForMetrics(partitionDesc.getCatalogName(), path, true, partitionDesc) : null;
 
       try {
         // Schema has to be serialized and deserialized as it is passed between different packages of TypeDescription:

itests/hive-jmh/src/main/java/org/apache/hive/benchmark/ql/exec/KryoBench.java

@@ -123,7 +123,7 @@ public static MapWork mockMapWork(String tableName, int partitions,
     tblProps.put("serialization.lib", OrcSerde.class.getName());
     tblProps.put("columns", columnNames.toString());
     tblProps.put("columns.types", columnTypes.toString());
-    TableDesc tbl = new TableDesc(OrcInputFormat.class, OrcOutputFormat.class, tblProps);
+    TableDesc tbl = new TableDesc(OrcInputFormat.class, OrcOutputFormat.class, tblProps, null);
 
     MapWork mapWork = new MapWork();
     mapWork.setVectorMode(true);

kafka-handler/src/test/org/apache/hadoop/hive/kafka/KafkaDagCredentialSupplierTest.java

@@ -142,7 +142,7 @@ private static MapWork createFileSinkWork(TableDesc tableDesc) {
 
   private static TableDesc createKafkaDesc(Properties props) {
     props.setProperty("name", "kafka_table_fake");
-    return new TableDesc(KafkaInputFormat.class, KafkaOutputFormat.class, props);
+    return new TableDesc(KafkaInputFormat.class, KafkaOutputFormat.class, props, null);
   }
 
   /**

llap-common/src/protobuf/LlapDaemonProtocol.proto

@@ -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, catalog name, and optionally table information.
 message EvictEntityRequestProto {
   required string db_name = 1;
   repeated TableProto table = 2;
+  required string catalog_name = 3;
 }
 
 // Used in EvictEntityRequestProto, can be used for non-partitioned and partitioned tables too.

llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapCacheMetadataSerializer.java

@@ -31,6 +31,7 @@
 import org.apache.hadoop.hive.llap.cache.PathCache;
 import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos;
 import org.apache.hadoop.hive.llap.io.encoded.LlapOrcCacheLoader;
+import org.apache.hadoop.hive.metastore.Warehouse;
 import org.apache.hadoop.hive.ql.io.SyntheticFileId;
 import org.apache.hadoop.hive.ql.io.orc.encoded.IoTrace;
 import org.apache.hive.common.util.FixedSizedObjectPool;
@@ -168,8 +169,14 @@ private static DiskRangeList decodeRanges(List<LlapDaemonProtocolProtos.CacheEnt
   }
 
   private static CacheTag decodeCacheTag(LlapDaemonProtocolProtos.CacheTag ct) {
-    return ct.getPartitionDescCount() == 0 ? CacheTag.build(ct.getTableName()) : CacheTag
-        .build(ct.getTableName(), ct.getPartitionDescList());
+    String tableName = ct.getTableName();
+    if (tableName.indexOf('.') == tableName.lastIndexOf('.')) {
+      // In case the CacheTag's table name does not contain catalog name
+      tableName = Warehouse.DEFAULT_CATALOG_NAME + '.' + tableName;
+    }
+    return ct.getPartitionDescCount() == 0
+        ? CacheTag.build(tableName)
+        : CacheTag.build(tableName, ct.getPartitionDescList());
   }
 
   @VisibleForTesting

llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java

@@ -324,8 +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(" ");
+      String catalog = request.getSingleCatalogName();
+      String db = request.getSingleDbName();
+      if (catalog != null && db != null) {
+        for (String table : request.getEntities().get(catalog).get(db).keySet()) {
+          sb.append(table).append(" ");
+        }
       }
       sb.append(" Duration: ").append(time).append(" ms");
       LOG.debug(sb.toString());

llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/OrcEncodedDataReader.java

@@ -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.getDbAndTableNameForMetrics(partitionDesc.getCatalogName(), 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);

llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/SerDeEncodedDataReader.java

@@ -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.getDbAndTableNameForMetrics(partitionDesc.getCatalogName(), split.getPath(), true, partitionDesc) : null;
     this.sourceInputFormat = sourceInputFormat;
     this.sourceSerDe = sourceSerDe;
     this.reporter = reporter;

llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcFileEstimateErrors.java

@@ -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;
@@ -141,6 +142,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");
   }
 }