HIVE-29807

ql/src/java/org/apache/hadoop/hive/ql/exec/ExplainTask.java

@@ -665,7 +665,7 @@ JSONObject collectAuthRelatedEntities(PrintStream out, ExplainWork work)
     HiveOperation operation = queryState.getHiveOperation();
 
     JSONObject object = new JSONObject(new LinkedHashMap<>());
-    Object jsonInput = toJson("INPUTS", toString(analyzer.getInputs()), out, work);
+    Object jsonInput = toJson("INPUTS", toString(analyzer.getAllInputs()), out, work);
     if (work.isFormatted()) {
       object.put("INPUTS", jsonInput);
     }

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

@@ -1649,7 +1649,7 @@ public boolean hasTransactionalInQuery() {
 
   public boolean isRequiresOpenTransaction() {
     return hasTransactionalInQuery() || getAcidDdlDesc() != null ||
-      Stream.of(getInputs(), getOutputs()).flatMap(Collection::stream)
+      Stream.of(getAllInputs(), getOutputs()).flatMap(Collection::stream)
         .filter(entity -> entity.getType() == Entity.Type.TABLE || entity.getType() == Entity.Type.PARTITION)
         .flatMap(entity -> {
           Table tbl = entity.getTable();

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

@@ -34,7 +34,8 @@ public ColumnAccessAnalyzer(ParseContext pactx) {
     pGraphContext = pactx;
   }
 
-  public ColumnAccessInfo analyzeColumnAccess(ColumnAccessInfo columnAccessInfo) throws SemanticException {
+  public ColumnAccessInfo analyzeColumnAccess(SemanticAnalyzer analyzer) throws SemanticException {
+    ColumnAccessInfo columnAccessInfo = analyzer.getColumnAccessInfo();
     if (columnAccessInfo == null) {
       columnAccessInfo = new ColumnAccessInfo();
     }
@@ -58,6 +59,22 @@ public ColumnAccessInfo analyzeColumnAccess(ColumnAccessInfo columnAccessInfo) t
         }
       }
     }
+    // Every Analyzer holds its private rootClause
+    columnAccessInfo.merge(getMaterializedCteColumnAccessInfo(analyzer.rootClause.asExecutionOrder()));
+    return columnAccessInfo;
+  }
+
+  /**
+   * Merge column access recorded by materialized CTE sub-analyzers into this analyzer's
+   * column access info so authorization sees base-table columns, not only the temp CTE table.
+   */
+  private ColumnAccessInfo getMaterializedCteColumnAccessInfo(List<SemanticAnalyzer.CTEClause> cteClauses) {
+    ColumnAccessInfo columnAccessInfo = new ColumnAccessInfo();
+    for (SemanticAnalyzer.CTEClause cte : cteClauses) {
+      if (cte.source != null && cte.source.getColumnAccessInfo() != null) {
+        columnAccessInfo.merge(cte.source.getColumnAccessInfo());
+      }
+    }
     return columnAccessInfo;
   }
 }

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

@@ -92,6 +92,20 @@ public Map<String, List<String>> getTableToColumnAllAccessMap() {
     return mapping;
   }
 
+  /**
+   * Merge direct column accesses from another ColumnAccessInfo into this one.
+   */
+  public void merge(ColumnAccessInfo other) {
+    if (other == null) {
+      return;
+    }
+    for (Map.Entry<String, List<String>> entry : other.getTableToColumnAccessMap().entrySet()) {
+      for (String col : entry.getValue()) {
+        add(entry.getKey(), col);
+      }
+    }
+  }
+
   /**
    * Strip a virtual column out of the set of columns.  This is useful in cases where we do not
    * want to be checking against the user reading virtual columns, namely update and delete.

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

@@ -133,7 +133,7 @@ private boolean tryMetadataUpdate(Table table, ASTNode tabNameNode, ASTNode wher
 
     DDLWork ddlWork = createDDLWorkOfMetadataUpdate(tableName, sarg);
     rootTasks = Collections.singletonList(TaskFactory.get(ddlWork));
-    inputs = sem.getInputs();
+    inputs = sem.getAllInputs();
     outputs = sem.getOutputs();
     updateOutputs(table);
     return true;

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

@@ -186,7 +186,7 @@ public void analyzeInternal(ASTNode ast) throws SemanticException {
     BaseSemanticAnalyzer sem = SemanticAnalyzerFactory.get(queryState, input);
     sem.analyze(input, ctx);
     sem.validate();
-    inputs = sem.getInputs();
+    inputs = sem.getAllInputs();
     outputs = sem.getOutputs();
 
     ctx.setResFile(ctx.getLocalTmpPath());

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

@@ -56,7 +56,7 @@ public void setConf(Configuration conf) {
 
   @Override
   public void update(BaseSemanticAnalyzer sem) {
-    this.inputs = sem.getInputs();
+    this.inputs = sem.getAllInputs();
     this.outputs = sem.getOutputs();
     this.commandType = sem.getQueryState().getHiveOperation();
   }

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

@@ -1406,7 +1406,7 @@ private void addCTEAsSubQuery(QB qb, String cteName, String cteAlias)
     qb.rewriteCTEToSubq(cteAlias, cteName, cteQBExpr);
   }
 
-  private final CTEClause rootClause = new CTEClause(null, null, null);
+  final CTEClause rootClause = new CTEClause(null, null, null);
 
   @Override
   public List<Task<?>> getAllRootTasks() {
@@ -1421,10 +1421,10 @@ public List<Task<?>> getAllRootTasks() {
 
   @Override
   public Set<ReadEntity> getAllInputs() {
-    Set<ReadEntity> readEntities = new HashSet<ReadEntity>(getInputs());
+    Set<ReadEntity> readEntities = new LinkedHashSet<>(getInputs());
     for (CTEClause cte : rootClause.asExecutionOrder()) {
       if (cte.source != null) {
-        readEntities.addAll(cte.source.getInputs());
+        readEntities.addAll(cte.source.getAllInputs());
       }
     }
     return readEntities;
@@ -1435,7 +1435,7 @@ public Set<WriteEntity> getAllOutputs() {
     Set<WriteEntity> writeEntities = new HashSet<WriteEntity>(getOutputs());
     for (CTEClause cte : rootClause.asExecutionOrder()) {
       if (cte.source != null) {
-        writeEntities.addAll(cte.source.getOutputs());
+        writeEntities.addAll(cte.source.getAllOutputs());
       }
     }
     return writeEntities;
@@ -1595,9 +1595,8 @@ Table materializeCTE(String cteName, CTEClause cte) throws HiveException {
 
     LOG.info("{} will be materialized into {}", cteName, location);
     cte.source = analyzer;
-
+    
     ctx.addMaterializedTable(cteName, table, getMaterializedTableStats(analyzer.getSinkOp()));
-
     return table;
   }
 
@@ -13387,7 +13386,7 @@ void analyzeInternal(ASTNode ast, Supplier<PlannerContext> pcf) throws SemanticE
               || HiveConf.getBoolVar(this.conf, HiveConf.ConfVars.HIVE_STATS_COLLECT_SCANCOLS)) {
         ColumnAccessAnalyzer columnAccessAnalyzer = new ColumnAccessAnalyzer(pCtx);
         // view column access info is carried by this.getColumnAccessInfo().
-        setColumnAccessInfo(columnAccessAnalyzer.analyzeColumnAccess(this.getColumnAccessInfo()));
+        setColumnAccessInfo(columnAccessAnalyzer.analyzeColumnAccess(this));
       }
     }
     perfLogger.perfLogEnd(this.getClass().getName(), PerfLogger.LOGICAL_OPTIMIZATION);
@@ -13426,7 +13425,7 @@ void analyzeInternal(ASTNode ast, Supplier<PlannerContext> pcf) throws SemanticE
 
     // 11. put accessed columns to readEntity
     if (HiveConf.getBoolVar(this.conf, HiveConf.ConfVars.HIVE_STATS_COLLECT_SCANCOLS)) {
-      putAccessedColumnsToReadEntity(inputs, columnAccessInfo);
+      putAccessedColumnsToReadEntity(getAllInputs(), columnAccessInfo);
     }
 
     if (isCacheEnabled && lookupInfo != null) {
@@ -15288,7 +15287,7 @@ private void useCachedResult(QueryResultsCache.CacheEntry cacheEntry, boolean ne
   private QueryResultsCache.QueryInfo createCacheQueryInfoForQuery(QueryResultsCache.LookupInfo lookupInfo) {
     long queryTime = SessionState.get().getQueryCurrentTimestamp().toEpochMilli();
     return new QueryResultsCache.QueryInfo(queryTime, lookupInfo, queryState.getHiveOperation(),
-        resultSchema, getTableAccessInfo(), getColumnAccessInfo(), inputs);
+        resultSchema, getTableAccessInfo(), getColumnAccessInfo(), getAllInputs());
   }
 
   /**

ql/src/java/org/apache/hadoop/hive/ql/plan/ExplainWork.java

@@ -81,7 +81,7 @@ public ExplainWork(Path resFile,
     }
     this.analyzer = analyzer;
     if (analyzer != null) {
-      this.inputs = analyzer.getInputs();
+      this.inputs = analyzer.getAllInputs();
     }
     if (analyzer != null) {
       this.outputs = analyzer.getAllOutputs();

ql/src/java/org/apache/hadoop/hive/ql/security/authorization/command/CommandAuthorizer.java

@@ -75,14 +75,15 @@ private static boolean skip(HiveOperation op, BaseSemanticAnalyzer sem) throws H
 
   private static Set<ReadEntity> getInputs(BaseSemanticAnalyzer sem) {
     Set<ReadEntity> additionalInputs = new HashSet<ReadEntity>();
-    for (Entity e : sem.getInputs()) {
+    for (Entity e : sem.getAllInputs()) {
       if (e.getType() == Entity.Type.PARTITION) {
         additionalInputs.add(new ReadEntity(e.getTable()));
       }
     }
 
+    // getAllInputs() includes tables read by materialized CTE sub-analyzers.
     // Sets.union keeps the values from the first set if they are present in both
-    return Sets.union(sem.getInputs(), additionalInputs);
+    return Sets.union(sem.getAllInputs(), additionalInputs);
   }
 
   private static Set<WriteEntity> getOutputs(BaseSemanticAnalyzer sem) {

ql/src/test/org/apache/hadoop/hive/ql/parse/TestMaterializedCTEInputs.java

@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hive.ql.parse;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Stream;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.conf.HiveConfForTest;
+import org.apache.hadoop.hive.metastore.client.builder.DatabaseBuilder;
+import org.apache.hadoop.hive.metastore.client.builder.TableBuilder;
+import org.apache.hadoop.hive.ql.Context;
+import org.apache.hadoop.hive.ql.QueryState;
+import org.apache.hadoop.hive.ql.ddl.database.drop.DropDatabaseDesc;
+import org.apache.hadoop.hive.ql.hooks.ReadEntity;
+import org.apache.hadoop.hive.ql.metadata.Hive;
+import org.apache.hadoop.hive.ql.session.SessionState;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class TestMaterializedCTEInputs {
+  private static final String DB_NAME = "test_materialized_cte";
+  private static final String TABLE_FQ_NAME = DB_NAME + "@src";
+
+  private static Hive db;
+  private static HiveConf conf;
+
+  @BeforeAll
+  public static void beforeClass() throws Exception {
+    conf = new HiveConfForTest(TestMaterializedCTEInputs.class);
+    conf.set("hive.security.authorization.enabled", "false");
+    conf.set("hive.security.authorization.manager",
+        "org.apache.hadoop.hive.ql.security.authorization.plugin.sqlstd.SQLStdConfOnlyAuthorizerFactory");
+    conf.setIntVar(HiveConf.ConfVars.HIVE_CTE_MATERIALIZE_THRESHOLD, 1);
+    conf.setBoolVar(HiveConf.ConfVars.HIVE_CTE_MATERIALIZE_FULL_AGGREGATE_ONLY, false);
+    conf.setBoolVar(HiveConf.ConfVars.HIVE_STATS_COLLECT_SCANCOLS, true);
+    db = Hive.get(conf);
+    SessionState.start(conf);
+    new DatabaseBuilder().setName(DB_NAME).create(db.getMSC(), conf);
+    SessionState.get().setCurrentDatabase(DB_NAME);
+    new TableBuilder().setDbName(DB_NAME).setTableName("src")
+        .addCol("key", "string")
+        .addCol("value", "string")
+        .addCol("col1", "int")
+        .create(db.getMSC(), conf);
+  }
+
+  public static Stream<Arguments> casesForMaterializedCteInputs() {
+    return Stream.of(
+        Arguments.of("chain cte", "with q1 as ( select key from q2 where key = '5'),"
+            + "q2 as ( select key from test_materialized_cte.src where key = '5') "
+            + "select * from (select key from q1) a", Set.of("key")),
+        Arguments.of("nested cte", "WITH q1 AS ("
+            + "WITH q2 AS (SELECT key, value FROM test_materialized_cte.src WHERE key = '4') "
+            + "SELECT * FROM q2 UNION ALL SELECT * FROM q2) "
+            + "SELECT * FROM q1 t1 JOIN q1 t2 ON t1.key = t2.key", Set.of("key", "value"))
+    );
+  }
+
+  @ParameterizedTest(name = "{0}")
+  @MethodSource("casesForMaterializedCteInputs")
+  void testMaterializedCteInputs(String type, String query, Set<String> expectedCols) throws Exception {
+    HiveConf testConf = new HiveConf(conf);
+    Context ctx = new Context(testConf);
+    ASTNode astNode = ParseUtils.parse(query, ctx);
+    QueryState queryState = new QueryState.Builder().withHiveConf(testConf).build();
+    SemanticAnalyzer analyzer = (SemanticAnalyzer) SemanticAnalyzerFactory.get(queryState, astNode);
+    analyzer.initCtx(ctx);
+    analyzer.analyze(astNode, ctx);
+
+    Set<ReadEntity> directInputs = analyzer.getInputs();
+    Set<ReadEntity> allInputs = analyzer.getAllInputs();
+
+    assertTrue(directInputs.stream().noneMatch(e -> isTableNamed(e, "src")),
+        "Materialized CTE should not expose base table in direct inputs");
+    assertTrue(allInputs.stream().anyMatch(e -> isTableNamed(e, "src")),
+        "Materialized CTE base table must appear in getAllInputs");
+
+    ColumnAccessInfo columnAccessInfo = analyzer.getColumnAccessInfo();
+    assertNotNull(columnAccessInfo);
+    List<String> srcCols = columnAccessInfo.getTableToColumnAccessMap().get(TABLE_FQ_NAME);
+    assertNotNull(srcCols, "Column must include materialized CTE base table");
+    assertTrue(new HashSet<>(srcCols).containsAll(expectedCols),
+        () -> "Expected columns " + expectedCols + " but got " + srcCols);
+    ctx.clear();
+  }
+
+  private static boolean isTableNamed(ReadEntity entity, String tableName) {
+    return entity.getTable() != null && tableName.equals(entity.getTable().getTableName());
+  }
+
+  @AfterAll
+  public static void afterClass() throws Exception {
+    try {
+      db.dropDatabase(new DropDatabaseDesc(DB_NAME, DB_NAME, true, true, true));
+    } catch (Exception ignored) {
+    }
+    db.close(true);
+  }
+}

ql/src/test/results/clientpositive/llap/cte_3.q.out

@@ -4,13 +4,15 @@ select *
 from q1
 PREHOOK: type: QUERY
 PREHOOK: Input: default@q1
+PREHOOK: Input: default@src
 #### A masked pattern was here ####
 POSTHOOK: query: explain
 with q1 as ( select key from src where key = '5')
 select *
 from q1
 POSTHOOK: type: QUERY
 POSTHOOK: Input: default@q1
+POSTHOOK: Input: default@src
 #### A masked pattern was here ####
 Plan optimized by CBO.
 
@@ -63,12 +65,14 @@ with q1 as ( select key from src where key = '5')
 select * from (select key from q1) a
 PREHOOK: type: QUERY
 PREHOOK: Input: default@q1
+PREHOOK: Input: default@src
 #### A masked pattern was here ####
 POSTHOOK: query: explain
 with q1 as ( select key from src where key = '5')
 select * from (select key from q1) a
 POSTHOOK: type: QUERY
 POSTHOOK: Input: default@q1
+POSTHOOK: Input: default@src
 #### A masked pattern was here ####
 Plan optimized by CBO.
 
@@ -120,13 +124,17 @@ q2 as ( select key from src where key = '5')
 select * from (select key from q1) a
 PREHOOK: type: QUERY
 PREHOOK: Input: default@q1
+PREHOOK: Input: default@q2
+PREHOOK: Input: default@src
 #### A masked pattern was here ####
 POSTHOOK: query: explain
 with q1 as ( select key from q2 where key = '5'),
 q2 as ( select key from src where key = '5')
 select * from (select key from q1) a
 POSTHOOK: type: QUERY
 POSTHOOK: Input: default@q1
+POSTHOOK: Input: default@q2
+POSTHOOK: Input: default@src
 #### A masked pattern was here ####
 Plan optimized by CBO.