Cache and reuse ValidTxnWriteIdList

Author: deniskuzZ

ql/src/java/org/apache/hadoop/hive/ql/DriverTxnHandler.java

@@ -246,7 +246,7 @@ private void acquireLocks() throws CommandProcessorException {
       String errorMessage;
       if (driverState.isDestroyed() || driverState.isAborted() || driverState.isClosed()) {
         errorMessage = String.format("Ignore lock acquisition related exception in terminal state (%s): %s",
-            driverState.toString(), e.getMessage());
+            driverState, e.getMessage());
         CONSOLE.printInfo(errorMessage);
       } else {
         errorMessage = String.format("FAILED: Error in acquiring locks: %s", e.getMessage());

ql/src/java/org/apache/hadoop/hive/ql/parse/BaseSemanticAnalyzer.java

@@ -2091,7 +2091,7 @@ public void startAnalysis() {
   public void endAnalysis(ASTNode tree) {
     if (ctx != null){
       queryProperties.setUsedTables(
-          CacheTableHelper.getUniqueNames(ctx.getParsedTables()));
+          TableHelper.getUniqueNames(ctx.getParsedTables()));
     }
     setQueryType(tree); // at this point we know the query type for sure
   }

ql/src/java/org/apache/hadoop/hive/ql/parse/CacheTableHelper.java

@@ -2458,8 +2458,7 @@ protected Set<TableName> getTablesUsed(RelNode plan) {
       new RelVisitor() {
         @Override
         public void visit(RelNode node, int ordinal, RelNode parent) {
-          if (node instanceof TableScan) {
-            TableScan ts = (TableScan) node;
+          if (node instanceof TableScan ts) {
             Table table = ((RelOptHiveTable) ts.getTable()).getHiveTableMD();
             if (AcidUtils.isTransactionalTable(table) ||
                   table.isNonNative() && table.getStorageHandler().areSnapshotsSupported()) {
@@ -5112,7 +5111,7 @@ private ImmutableMap<String, Integer> buildHiveColNameToInputPosMap(
       return hiveColNameToInputPosMapBuilder.build();
     }
 
-    private QBParseInfo getQBParseInfo(QB qb) throws CalciteSemanticException {
+    private QBParseInfo getQBParseInfo(QB qb) {
       return qb.getParseInfo();
     }
   }

ql/src/java/org/apache/hadoop/hive/ql/parse/CalcitePlanner.java

@@ -45,14 +45,12 @@ public static TableName of(Table table) throws SemanticException {
    * Set a @{@link Table} object's table and db names based on the provided string.
    * @param dbTable the dbtable string
    * @param table the table to update
-   * @return the table
    * @throws SemanticException
    */
-  public static Table setFrom(String dbTable, Table table) throws SemanticException{
+  public static void setFrom(String dbTable, Table table) throws SemanticException{
     TableName name = ofNullable(dbTable);
     table.setTableName(name.getTable());
     table.setDbName(name.getDb());
-    return table;
   }
 
   /**

ql/src/java/org/apache/hadoop/hive/ql/parse/HiveTableName.java

@@ -87,7 +87,6 @@
 import org.apache.hadoop.hive.common.StatsSetupConst.StatDB;
 import org.apache.hadoop.hive.common.StringInternUtils;
 import org.apache.hadoop.hive.common.TableName;
-import org.apache.hadoop.hive.common.ValidTxnList;
 import org.apache.hadoop.hive.common.ValidTxnWriteIdList;
 import org.apache.hadoop.hive.common.metrics.common.MetricsConstant;
 import org.apache.hadoop.hive.conf.Constants;
@@ -363,7 +362,6 @@ public class SemanticAnalyzer extends BaseSemanticAnalyzer {
   private final Map<TableScanOperator, Map<String, ExprNodeDesc>> opToPartToSkewedPruner;
   private Map<SelectOperator, Table> viewProjectToTableSchema;
   private Operator<? extends OperatorDesc> sinkOp;
-  private final CacheTableHelper cacheTableHelper = new CacheTableHelper();
 
   /**
    * a map for the split sampling, from alias to an instance of SplitSample
@@ -2883,8 +2881,7 @@ private void replaceViewReferenceWithDefinition(QB qb, Table tab,
       String viewText = tab.getViewExpandedText();
       TableMask viewMask = new TableMask(this, conf, false);
       viewTree = ParseUtils.parse(viewText, ctx, tab.getCompleteName());
-      cacheTableHelper.populateCacheForView(ctx.getParsedTables(), conf,
-          getTxnMgr(), tab.getDbName(), tab.getTableName());
+
       if (viewMask.isEnabled() && analyzeRewrite == null) {
         ParseResult parseResult = rewriteASTWithMaskAndFilter(viewMask, viewTree,
             ctx.getViewTokenRewriteStream(viewFullyQualifiedName),
@@ -13163,15 +13160,21 @@ void analyzeInternal(ASTNode ast, Supplier<PlannerContext> pcf) throws SemanticE
     perfLogger.perfLogBegin(this.getClass().getName(), PerfLogger.GENERATE_RESOLVED_PARSETREE);
     // 1. Generate Resolved Parse tree from syntax tree
     boolean needsTransform = needsTransform();
-    //change the location of position alias process here
+    // change the location of position alias process here
     processPositionAlias(ast);
-    cacheTableHelper.populateCache(ctx.getParsedTables(), conf, getTxnMgr());
     PlannerContext plannerCtx = pcf.get();
     if (!genResolvedParseTree(ast, plannerCtx)) {
       return;
     }
-    if (tablesFromReadEntities(inputs).stream().anyMatch(AcidUtils::isTransactionalTable)) {
-      queryState.getValidTxnList();
+
+    if (queryState.getHMSCache() != null) {
+      // this step primes the cache containing the validTxnWriteIdList. It will fetch
+      // all the tables into the MetaStore Client cache with one HMS call.
+      getQueryValidTxnWriteIdList();
+    } else {
+      if (tablesFromReadEntities(inputs).stream().anyMatch(AcidUtils::isTransactionalTable)) {
+        queryState.getValidTxnList();
+      }
     }
 
     if (HiveConf.getBoolVar(conf, ConfVars.HIVE_REMOVE_ORDERBY_IN_SUBQUERY)) {
@@ -15166,30 +15169,25 @@ private String getQueryStringForCache(ASTNode ast) {
 
   private ValidTxnWriteIdList getQueryValidTxnWriteIdList() throws SemanticException {
     // TODO: Once HIVE-18948 is in, should be able to retrieve writeIdList from the conf.
-    //cachedWriteIdList = AcidUtils.getValidTxnWriteIdList(conf);
+    // cachedWriteIdList = AcidUtils.getValidTxnWriteIdList(conf);
     //
-    List<String> transactionalTables = tablesFromReadEntities(inputs)
-      .stream()
-      .filter(AcidUtils::isTransactionalTable)
-      .map(Table::getFullyQualifiedName)
-      .collect(Collectors.toList());
-    
-    if (transactionalTables.size() > 0) {
-      String txnString = queryState.getValidTxnList();
-      if (txnString == null) {
-        return null;
-      }
-      try {
-        return getTxnMgr().getValidWriteIds(transactionalTables, txnString);
-      } catch (Exception err) {
-        String msg = "Error while getting the txnWriteIdList for tables " + transactionalTables
-                + " and validTxnList " + conf.get(ValidTxnList.VALID_TXNS_KEY);
-        throw new SemanticException(msg, err);
-      }
-    }
+    var transactionalTables = tablesFromReadEntities(inputs)
+        .stream()
+        .filter(AcidUtils::isTransactionalTable)
+        .map(Table::getFullyQualifiedName)
+        .toList();
 
-    // No transactional tables.
-    return null;
+    if (transactionalTables.isEmpty()) {
+      return null;
+    }
+    String txnString = queryState.getValidTxnList();
+    try {
+      return getTxnMgr().getValidWriteIds(transactionalTables, txnString);
+    } catch (Exception err) {
+      String msg = "Error while getting the txnWriteIdList for tables " + transactionalTables
+          + " and validTxnList " + txnString;
+      throw new SemanticException(msg, err);
+    }
   }
 
   private QueryResultsCache.LookupInfo createLookupInfoForQuery(ASTNode astNode) throws SemanticException {