Skip to content

Commit 642a058

Browse files
Neer393Neer393
committed
HIVE-29281: Make proactive cache eviction work with catalog
1 parent 24e835c commit 642a058

File tree

31 files changed

+481
-206
lines changed

31 files changed

+481
-206
lines changed

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -423,7 +423,7 @@ public static Pair<String, String> getDbAndTableName(String tableName) throws IO
423423
Properties props = inputJobInfo.getTableInfo().getStorerInfo().getProperties();
424424
props.put(serdeConstants.SERIALIZATION_LIB,storageHandler.getSerDeClass().getName());
425425
TableDesc tableDesc = new TableDesc(storageHandler.getInputFormatClass(),
426-
storageHandler.getOutputFormatClass(),props);
426+
storageHandler.getOutputFormatClass(), props, null);
427427
if (tableDesc.getJobProperties() == null) {
428428
tableDesc.setJobProperties(new HashMap<>());
429429
}
@@ -464,7 +464,7 @@ public static Pair<String, String> getDbAndTableName(String tableName) throws IO
464464
Properties props = outputJobInfo.getTableInfo().getStorerInfo().getProperties();
465465
props.put(serdeConstants.SERIALIZATION_LIB,storageHandler.getSerDeClass().getName());
466466
TableDesc tableDesc = new TableDesc(storageHandler.getInputFormatClass(),
467-
IgnoreKeyTextOutputFormat.class,props);
467+
IgnoreKeyTextOutputFormat.class, props, null);
468468
if (tableDesc.getJobProperties() == null)
469469
tableDesc.setJobProperties(new HashMap<>());
470470
for (Map.Entry<String, String> el : conf) {

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -123,7 +123,7 @@ public static MapWork mockMapWork(String tableName, int partitions,
123123
tblProps.put("serialization.lib", OrcSerde.class.getName());
124124
tblProps.put("columns", columnNames.toString());
125125
tblProps.put("columns.types", columnTypes.toString());
126-
TableDesc tbl = new TableDesc(OrcInputFormat.class, OrcOutputFormat.class, tblProps);
126+
TableDesc tbl = new TableDesc(OrcInputFormat.class, OrcOutputFormat.class, tblProps, null);
127127

128128
MapWork mapWork = new MapWork();
129129
mapWork.setVectorMode(true);

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -142,7 +142,7 @@ private static MapWork createFileSinkWork(TableDesc tableDesc) {
142142

143143
private static TableDesc createKafkaDesc(Properties props) {
144144
props.setProperty("name", "kafka_table_fake");
145-
return new TableDesc(KafkaInputFormat.class, KafkaOutputFormat.class, props);
145+
return new TableDesc(KafkaInputFormat.class, KafkaOutputFormat.class, props, null);
146146
}
147147

148148
/**

‎llap-common/src/protobuf/LlapDaemonProtocol.proto‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -233,10 +233,11 @@ message SetCapacityRequestProto {
233233
message SetCapacityResponseProto {
234234
}
235235

236-
// Used for proactive eviction request. Must contain one DB name, and optionally table information.
236+
// Used for proactive eviction request. Must contain a DB name, catalog name, and optionally table information.
237237
message EvictEntityRequestProto {
238238
required string db_name = 1;
239239
repeated TableProto table = 2;
240+
required string catalog_name = 3;
240241
}
241242

242243
// 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‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@
3131
import org.apache.hadoop.hive.llap.cache.PathCache;
3232
import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos;
3333
import org.apache.hadoop.hive.llap.io.encoded.LlapOrcCacheLoader;
34+
import org.apache.hadoop.hive.metastore.Warehouse;
3435
import org.apache.hadoop.hive.ql.io.SyntheticFileId;
3536
import org.apache.hadoop.hive.ql.io.orc.encoded.IoTrace;
3637
import org.apache.hive.common.util.FixedSizedObjectPool;
@@ -168,8 +169,14 @@ private static DiskRangeList decodeRanges(List<LlapDaemonProtocolProtos.CacheEnt
168169
}
169170

170171
private static CacheTag decodeCacheTag(LlapDaemonProtocolProtos.CacheTag ct) {
171-
return ct.getPartitionDescCount() == 0 ? CacheTag.build(ct.getTableName()) : CacheTag
172-
.build(ct.getTableName(), ct.getPartitionDescList());
172+
String tableName = ct.getTableName();
173+
if (tableName.indexOf('.') == tableName.lastIndexOf('.')) {
174+
// In case the CacheTag's table name does not contain catalog name
175+
tableName = Warehouse.DEFAULT_CATALOG_NAME + '.' + tableName;
176+
}
177+
return ct.getPartitionDescCount() == 0
178+
? CacheTag.build(tableName)
179+
: CacheTag.build(tableName, ct.getPartitionDescList());
173180
}
174181

175182
@VisibleForTesting

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

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -324,8 +324,12 @@ public long evictEntity(LlapDaemonProtocolProtos.EvictEntityRequestProto protoRe
324324
if (LOG.isDebugEnabled()) {
325325
StringBuilder sb = new StringBuilder();
326326
sb.append(markedBytes).append(" bytes marked for eviction from LLAP cache buffers that belong to table(s): ");
327-
for (String table : request.getEntities().get(request.getSingleDbName()).keySet()) {
328-
sb.append(table).append(" ");
327+
String catalog = request.getSingleCatalogName();
328+
String db = request.getSingleDbName();
329+
if (catalog != null && db != null) {
330+
for (String table : request.getEntities().get(catalog).get(db).keySet()) {
331+
sb.append(table).append(" ");
332+
}
329333
}
330334
sb.append(" Duration: ").append(time).append(" ms");
331335
LOG.debug(sb.toString());

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import org.apache.hadoop.hive.common.io.DiskRangeList.MutateHelper;
2727
import org.apache.hadoop.hive.common.io.CacheTag;
2828
import org.apache.hadoop.hive.llap.cache.EvictionDispatcher;
29+
import org.apache.hadoop.hive.metastore.Warehouse;
2930
import org.apache.hadoop.hive.llap.cache.LlapCacheableBuffer;
3031
import org.apache.hadoop.hive.ql.io.SyntheticFileId;
3132
import org.apache.hadoop.hive.ql.io.orc.encoded.IncompleteCb;
@@ -141,6 +142,6 @@ public boolean isMarkedForEviction() {
141142
@Override
142143
public CacheTag getTag() {
143144
// We don't care about these.
144-
return CacheTag.build("OrcEstimates");
145+
return CacheTag.build(Warehouse.DEFAULT_CATALOG_NAME + ".OrcEstimates");
145146
}
146147
}

‎llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestCacheContentsTracker.java‎

Lines changed: 54 additions & 49 deletions
Original file line numberDiff line numberDiff line change
@@ -17,11 +17,10 @@
1717
*/
1818
package org.apache.hadoop.hive.llap.cache;
1919

20-
import java.util.HashMap;
2120
import java.util.LinkedHashMap;
22-
import java.util.Map;
2321

2422
import org.apache.hadoop.hive.common.io.CacheTag;
23+
import org.apache.hadoop.hive.metastore.Warehouse;
2524

2625
import org.junit.BeforeClass;
2726
import org.junit.Test;
@@ -53,16 +52,20 @@ public static void setup() {
5352
*/
5453
@Test
5554
public void testParentCacheTagGeneration() {
56-
CacheTag db = cacheTagBuilder("dbname");
57-
CacheTag table = cacheTagBuilder("dbname.tablename");
58-
CacheTag p = cacheTagBuilder("dbname.tablename", "p=v1");
59-
CacheTag pp = cacheTagBuilder("dbname.tablename", "p=v1", "pp=vv1");
60-
CacheTag ppp = cacheTagBuilder("dbname.tablename", "p=v1", "pp=vv1", "ppp=vvv1");
55+
// DB-level tag: tableName = "hive.dbname" (catalog prepended, one dot).
56+
CacheTag db = cacheTagBuilder(Warehouse.DEFAULT_CATALOG_NAME + ".dbname");
57+
// Table-level tag: tableName = "hive.dbname.tablename" (two dots).
58+
CacheTag table = cacheTagBuilder(Warehouse.DEFAULT_CATALOG_NAME + ".dbname.tablename");
59+
CacheTag p = cacheTagBuilder(Warehouse.DEFAULT_CATALOG_NAME + ".dbname.tablename", "p=v1");
60+
CacheTag pp = cacheTagBuilder(Warehouse.DEFAULT_CATALOG_NAME + ".dbname.tablename", "p=v1", "pp=vv1");
61+
CacheTag ppp = cacheTagBuilder(Warehouse.DEFAULT_CATALOG_NAME + ".dbname.tablename", "p=v1", "pp=vv1", "ppp=vvv1");
6162

6263
assertTrue(pp.compareTo(CacheTag.createParentCacheTag(ppp)) == 0);
6364
assertTrue(p.compareTo(CacheTag.createParentCacheTag(pp)) == 0);
6465
assertTrue(table.compareTo(CacheTag.createParentCacheTag(p)) == 0);
66+
// createParentCacheTag(table) strips the table part → "hive.dbname" which equals db.
6567
assertTrue(db.compareTo(CacheTag.createParentCacheTag(table)) == 0);
68+
// DB-level tag has exactly one dot; the hierarchy stops here.
6669
assertNull(CacheTag.createParentCacheTag(db));
6770
}
6871

@@ -127,7 +130,7 @@ public void testCacheTagComparison() {
127130
public void testEncodingDecoding() throws Exception {
128131
LinkedHashMap<String, String> partDescs = new LinkedHashMap<>();
129132
partDescs.put("pytha=goras", "a2+b2=c2");
130-
CacheTag tag = CacheTag.build("math.rules", partDescs);
133+
CacheTag tag = CacheTag.build(Warehouse.DEFAULT_CATALOG_NAME + ".math.rules", partDescs);
131134
CacheTag.SinglePartitionCacheTag stag = ((CacheTag.SinglePartitionCacheTag)tag);
132135
assertEquals("pytha=goras=a2+b2=c2", stag.partitionDescToString());
133136
assertEquals(1, stag.getPartitionDescMap().size());
@@ -136,7 +139,7 @@ public void testEncodingDecoding() throws Exception {
136139
partDescs.clear();
137140
partDescs.put("mutli=one", "one=/1");
138141
partDescs.put("mutli=two/", "two=2");
139-
tag = CacheTag.build("math.rules", partDescs);
142+
tag = CacheTag.build(Warehouse.DEFAULT_CATALOG_NAME + ".math.rules", partDescs);
140143
CacheTag.MultiPartitionCacheTag mtag = ((CacheTag.MultiPartitionCacheTag)tag);
141144
assertEquals("mutli=one=one=/1/mutli=two/=two=2", mtag.partitionDescToString());
142145
assertEquals(2, mtag.getPartitionDescMap().size());
@@ -167,81 +170,83 @@ private static LlapCacheableBuffer createMockBuffer(long size, CacheTag cacheTag
167170
return llapCacheableBufferMock;
168171
}
169172

170-
public static CacheTag cacheTagBuilder(String dbAndTable, String... partitions) {
173+
public static CacheTag cacheTagBuilder(String fullTableName, String... partitions) {
171174
if (partitions != null && partitions.length > 0) {
172175
LinkedHashMap<String, String> partDescs = new LinkedHashMap<>();
173176
for (String partition : partitions) {
174177
String[] partDesc = partition.split("=");
175178
partDescs.put(partDesc[0], partDesc[1]);
176179
}
177-
return CacheTag.build(dbAndTable, partDescs);
180+
return CacheTag.build(fullTableName, partDescs);
178181
} else {
179-
return CacheTag.build(dbAndTable);
182+
return CacheTag.build(fullTableName);
180183
}
181184
}
182185

186+
private static final String DEF_CAT = Warehouse.DEFAULT_CATALOG_NAME + ".";
187+
183188
private static void cacheTestBuffers() {
184189
tracker.cache(createMockBuffer(4 * 1024L,
185-
cacheTagBuilder("default.testtable")), null);
190+
cacheTagBuilder(DEF_CAT + "default.testtable")), null);
186191
tracker.cache(createMockBuffer(2 * 1024L,
187-
cacheTagBuilder("otherdb.testtable", "p=v1", "pp=vv1")), null);
192+
cacheTagBuilder(DEF_CAT + "otherdb.testtable", "p=v1", "pp=vv1")), null);
188193
tracker.cache(createMockBuffer(32 * 1024L,
189-
cacheTagBuilder("otherdb.testtable", "p=v1", "pp=vv1")), null);
194+
cacheTagBuilder(DEF_CAT + "otherdb.testtable", "p=v1", "pp=vv1")), null);
190195
tracker.cache(createMockBuffer(64 * 1024L,
191-
cacheTagBuilder("otherdb.testtable", "p=v1", "pp=vv2")), null);
196+
cacheTagBuilder(DEF_CAT + "otherdb.testtable", "p=v1", "pp=vv2")), null);
192197
tracker.cache(createMockBuffer(128 * 1024L,
193-
cacheTagBuilder("otherdb.testtable", "p=v2", "pp=vv1")), null);
198+
cacheTagBuilder(DEF_CAT + "otherdb.testtable", "p=v2", "pp=vv1")), null);
194199
tracker.cache(createMockBuffer(256 * 1024L,
195-
cacheTagBuilder("otherdb.testtable2", "p=v3")), null);
200+
cacheTagBuilder(DEF_CAT + "otherdb.testtable2", "p=v3")), null);
196201
tracker.cache(createMockBuffer(512 * 1024 * 1024L,
197-
cacheTagBuilder("otherdb.testtable2", "p=v3")), null);
202+
cacheTagBuilder(DEF_CAT + "otherdb.testtable2", "p=v3")), null);
198203
tracker.cache(createMockBuffer(1024 * 1024 * 1024L,
199-
cacheTagBuilder("otherdb.testtable3")), null);
204+
cacheTagBuilder(DEF_CAT + "otherdb.testtable3")), null);
200205
tracker.cache(createMockBuffer(2 * 1024 * 1024L,
201-
cacheTagBuilder("default.testtable")), null);
206+
cacheTagBuilder(DEF_CAT + "default.testtable")), null);
202207
}
203208

204209
private static void evictSomeTestBuffers() {
205210
tracker.notifyEvicted(createMockBuffer(32 * 1024L,
206-
cacheTagBuilder("otherdb.testtable", "p=v1", "pp=vv1")));
211+
cacheTagBuilder(DEF_CAT + "otherdb.testtable", "p=v1", "pp=vv1")));
207212
tracker.notifyEvicted(createMockBuffer(512 * 1024 * 1024L,
208-
cacheTagBuilder("otherdb.testtable2", "p=v3")));
213+
cacheTagBuilder(DEF_CAT + "otherdb.testtable2", "p=v3")));
209214
tracker.notifyEvicted(createMockBuffer(2 * 1024 * 1024L,
210-
cacheTagBuilder("default.testtable")));
215+
cacheTagBuilder(DEF_CAT + "default.testtable")));
211216
tracker.notifyEvicted(createMockBuffer(4 * 1024L,
212-
cacheTagBuilder("default.testtable")));
217+
cacheTagBuilder(DEF_CAT + "default.testtable")));
213218
}
214219

215220
private static final String EXPECTED_CACHE_STATE_WHEN_FULL =
216221
"\n" +
217222
"Cache state: \n" +
218-
"default : 2/2, 2101248/2101248\n" +
219-
"default.testtable : 2/2, 2101248/2101248\n" +
220-
"otherdb : 7/7, 1611106304/1611106304\n" +
221-
"otherdb.testtable : 4/4, 231424/231424\n" +
222-
"otherdb.testtable/p=v1 : 3/3, 100352/100352\n" +
223-
"otherdb.testtable/p=v1/pp=vv1 : 2/2, 34816/34816\n" +
224-
"otherdb.testtable/p=v1/pp=vv2 : 1/1, 65536/65536\n" +
225-
"otherdb.testtable/p=v2 : 1/1, 131072/131072\n" +
226-
"otherdb.testtable/p=v2/pp=vv1 : 1/1, 131072/131072\n" +
227-
"otherdb.testtable2 : 2/2, 537133056/537133056\n" +
228-
"otherdb.testtable2/p=v3 : 2/2, 537133056/537133056\n" +
229-
"otherdb.testtable3 : 1/1, 1073741824/1073741824";
223+
"hive.default : 2/2, 2101248/2101248\n" +
224+
"hive.default.testtable : 2/2, 2101248/2101248\n" +
225+
"hive.otherdb : 7/7, 1611106304/1611106304\n" +
226+
"hive.otherdb.testtable : 4/4, 231424/231424\n" +
227+
"hive.otherdb.testtable/p=v1 : 3/3, 100352/100352\n" +
228+
"hive.otherdb.testtable/p=v1/pp=vv1 : 2/2, 34816/34816\n" +
229+
"hive.otherdb.testtable/p=v1/pp=vv2 : 1/1, 65536/65536\n" +
230+
"hive.otherdb.testtable/p=v2 : 1/1, 131072/131072\n" +
231+
"hive.otherdb.testtable/p=v2/pp=vv1 : 1/1, 131072/131072\n" +
232+
"hive.otherdb.testtable2 : 2/2, 537133056/537133056\n" +
233+
"hive.otherdb.testtable2/p=v3 : 2/2, 537133056/537133056\n" +
234+
"hive.otherdb.testtable3 : 1/1, 1073741824/1073741824";
230235

231236
private static final String EXPECTED_CACHE_STATE_AFTER_EVICTION =
232237
"\n" +
233238
"Cache state: \n" +
234-
"default : 0/2, 0/2101248\n" +
235-
"default.testtable : 0/2, 0/2101248\n" +
236-
"otherdb : 5/7, 1074202624/1611106304\n" +
237-
"otherdb.testtable : 3/4, 198656/231424\n" +
238-
"otherdb.testtable/p=v1 : 2/3, 67584/100352\n" +
239-
"otherdb.testtable/p=v1/pp=vv1 : 1/2, 2048/34816\n" +
240-
"otherdb.testtable/p=v1/pp=vv2 : 1/1, 65536/65536\n" +
241-
"otherdb.testtable/p=v2 : 1/1, 131072/131072\n" +
242-
"otherdb.testtable/p=v2/pp=vv1 : 1/1, 131072/131072\n" +
243-
"otherdb.testtable2 : 1/2, 262144/537133056\n" +
244-
"otherdb.testtable2/p=v3 : 1/2, 262144/537133056\n" +
245-
"otherdb.testtable3 : 1/1, 1073741824/1073741824";
239+
"hive.default : 0/2, 0/2101248\n" +
240+
"hive.default.testtable : 0/2, 0/2101248\n" +
241+
"hive.otherdb : 5/7, 1074202624/1611106304\n" +
242+
"hive.otherdb.testtable : 3/4, 198656/231424\n" +
243+
"hive.otherdb.testtable/p=v1 : 2/3, 67584/100352\n" +
244+
"hive.otherdb.testtable/p=v1/pp=vv1 : 1/2, 2048/34816\n" +
245+
"hive.otherdb.testtable/p=v1/pp=vv2 : 1/1, 65536/65536\n" +
246+
"hive.otherdb.testtable/p=v2 : 1/1, 131072/131072\n" +
247+
"hive.otherdb.testtable/p=v2/pp=vv1 : 1/1, 131072/131072\n" +
248+
"hive.otherdb.testtable2 : 1/2, 262144/537133056\n" +
249+
"hive.otherdb.testtable2/p=v3 : 1/2, 262144/537133056\n" +
250+
"hive.otherdb.testtable3 : 1/1, 1073741824/1073741824";
246251

247252
}

‎llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestFileCache.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919

2020
import com.google.common.base.Function;
2121
import org.apache.hadoop.hive.common.io.CacheTag;
22+
import org.apache.hadoop.hive.metastore.Warehouse;
2223
import org.junit.Test;
2324

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

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

‎llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestLowLevelCacheImpl.java‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@
3939
import org.slf4j.LoggerFactory;
4040

4141
import org.apache.hadoop.hive.common.io.CacheTag;
42+
import org.apache.hadoop.hive.metastore.Warehouse;
4243
import org.apache.hadoop.hive.common.io.DiskRange;
4344
import org.apache.hadoop.hive.common.io.DiskRangeList;
4445
import org.apache.hadoop.hive.common.io.DataCache.DiskRangeListFactory;
@@ -309,13 +310,13 @@ private void _testProactiveEvictionMark(boolean isInstantDeallocation) {
309310

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

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

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

320321
cache.putFileData(fn1, drs1, buffs1, 0, Priority.NORMAL, null, tag1);
321322
cache.putFileData(fn2, drs2, buffs2, 0, Priority.NORMAL, null, tag2);

0 commit comments

Comments
 (0)