HIVE-30059: LLAP IO Cache: native parquet data cache

common/src/java/org/apache/hadoop/hive/conf/HiveConf.java

@@ -261,6 +261,13 @@ private static URL checkConfigFile(File f) {
   public static final String HIVE_SERVER2_AUTHENTICATION_LDAP_USERMEMBERSHIPKEY_NAME =
       "hive.server2.authentication.ldap.userMembershipKey";
 
+  /**
+   * Token that {@link ConfVars#HIVE_VECTORIZED_INPUT_FORMAT_SUPPORTS_ENABLED} must contain for
+   * decimal_64 vectorization to be enabled. Extracted here so call sites can reference the same
+   * constant instead of repeating the literal.
+   */
+  public static final String HIVE_VECTORIZED_INPUT_FORMAT_SUPPORTS_DECIMAL_64 = "decimal_64";
+
   /**
    * dbVars are the parameters can be set per database. If these
    * parameters are set as a database property, when switching to that
@@ -4641,7 +4648,7 @@ public static enum ConfVars {
         "evaluate call and turn them into NULLs. Assume, by default, this is not needed"),
     HIVE_VECTORIZED_INPUT_FORMAT_SUPPORTS_ENABLED(
         "hive.vectorized.input.format.supports.enabled",
-        "decimal_64",
+        HIVE_VECTORIZED_INPUT_FORMAT_SUPPORTS_DECIMAL_64,
         "Which vectorized input format support features are enabled for vectorization.\n" +
         "That is, if a VectorizedInputFormat input format does support \"decimal_64\" for example\n" +
         "this variable must enable that to be used in vectorization"),
@@ -4976,6 +4983,9 @@ public static enum ConfVars {
     LLAP_IO_ENCODE_ENABLED("hive.llap.io.encode.enabled", true,
         "Whether LLAP should try to re-encode and cache data for non-ORC formats. This is used\n" +
         "on LLAP Server side to determine if the infrastructure for that is initialized."),
+    LLAP_IO_PARQUET_NATIVE_ENABLED("hive.llap.io.parquet.native.enabled", false,
+        "Whether LLAP IO caches Parquet column chunks and decodes them from the cache, like ORC.\n" +
+        "A projection that reaches into a nested type falls back to the vectorized Parquet reader."),
     LLAP_IO_ENCODE_FORMATS("hive.llap.io.encode.formats",
         "org.apache.hadoop.mapred.TextInputFormat,",
         "The table input formats for which LLAP IO should re-encode and cache data.\n" +

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java

@@ -84,7 +84,7 @@
 import org.apache.hadoop.hive.ql.exec.Utilities;
 import org.apache.hadoop.hive.ql.hooks.WriteEntity;
 import org.apache.hadoop.hive.ql.io.StorageFormatDescriptor;
-import org.apache.hadoop.hive.ql.io.parquet.vector.VectorizedParquetRecordReader;
+import org.apache.hadoop.hive.ql.io.parquet.vector.ParquetRowGroupDecoder;
 import org.apache.hadoop.hive.ql.io.sarg.ConvertAstToSearchArg;
 import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
 import org.apache.hadoop.hive.ql.metadata.DefaultStorageHandler;
@@ -1890,7 +1890,8 @@ private static boolean hasOrcTimeInSchema(Properties tableProps, Schema tableSch
   /**
    * Vectorized reads of parquet files from columns with list or map type is only supported if the nested types are of
    * primitive type category
-   * check {@link VectorizedParquetRecordReader#checkListColumnSupport} for details on nested types under lists
+   * check {@link ParquetRowGroupDecoder#checkListColumnSupport} for
+   * details on nested types under lists
    * @param tableProps iceberg table properties
    * @param tableSchema iceberg table schema
    * @return true if having nested types

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vector/HiveVectorizedReader.java

@@ -261,7 +261,9 @@ private static RecordReader<NullWritable, VectorizedRowBatch> parquetRecordReade
     if (HiveConf.getBoolVar(job, HiveConf.ConfVars.LLAP_IO_ENABLED, LlapProxy.isDaemon()) &&
         LlapProxy.getIo() != null && LlapProxy.getIo().usingLowLevelCache()) {
       LlapProxy.getIo().initCacheOnlyInputFormat(inputFormat);
-      footerData = LlapProxy.getIo().getParquetFooterBuffersFromCache(path, job, fileId);
+      // No per-fragment counters on the Iceberg vectorized path yet; footer lookups are
+      // recorded as cache traffic but not as META hits/misses in the LLAP IO summary.
+      footerData = LlapProxy.getIo().getParquetFooterBuffersFromCache(path, job, fileId, null);
     }
 
     ParquetMetadata parquetMetadata = HiveParquetUtil.readFooter(task.file(), io, job, footerData);

llap-client/src/java/org/apache/hadoop/hive/llap/io/api/LlapIo.java

@@ -23,9 +23,9 @@
 import java.util.List;
 
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hive.common.io.CacheTag;
+import org.apache.hadoop.hive.common.io.DataCache.BooleanRef;
 import org.apache.hadoop.hive.common.io.encoded.MemoryBufferOrBuffers;
 import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos;
 import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
@@ -74,11 +74,15 @@ InputFormat<NullWritable, T> getInputFormat(
    * @param fileKey fileId of the Parquet file (either the Long fileId of HDFS or the SyntheticFileId).
    *                Optional, if it is not provided, it will be generated, see:
    *                org.apache.hadoop.hive.ql.io.HdfsUtils#getFileId()
+   * @param cacheHit optional out-parameter; when supplied, its {@code value} field is written
+   *                 with {@code true} on a cache hit and {@code false} on a miss, so the caller
+   *                 (which lives in llap-server and holds {@code QueryFragmentCounters}) can bump
+   *                 {@code METADATA_CACHE_HIT} / {@code METADATA_CACHE_MISS} for the LLAP IO summary.
    * @return
    * @throws IOException
    */
-  MemoryBufferOrBuffers getParquetFooterBuffersFromCache(Path path, JobConf conf, @Nullable Object fileKey)
-      throws IOException;
+  MemoryBufferOrBuffers getParquetFooterBuffersFromCache(Path path, JobConf conf, @Nullable Object fileKey,
+      @Nullable BooleanRef cacheHit) throws IOException;
 
   /**
    * Handles request to evict entities specified in the request object.
@@ -107,6 +111,14 @@ MemoryBufferOrBuffers getParquetFooterBuffersFromCache(Path path, JobConf conf,
   RecordReader<NullWritable, VectorizedRowBatch> llapVectorizedOrcReaderForPath(Object fileKey, Path path, CacheTag tag,
       List<Integer> tableIncludedCols, JobConf conf, long offset, long length, Reporter reporter) throws IOException;
 
+  /**
+   * Parquet counterpart of {@link #llapVectorizedOrcReaderForPath}: reads the column chunks of the split through the
+   * LLAP data cache. Returns null when the file cannot be served this way (native Parquet IO disabled, no MapWork,
+   * unsupported schema). See {@link LlapParquetReadRequest} for the identity / split / projection fields.
+   */
+  RecordReader<NullWritable, VectorizedRowBatch> llapVectorizedParquetReaderForPath(
+      LlapParquetReadRequest request, JobConf conf, Reporter reporter) throws IOException;
+
   /**
    * Extract and return the cache content metadata.
    */

llap-client/src/java/org/apache/hadoop/hive/llap/io/api/LlapParquetReadRequest.java

@@ -0,0 +1,44 @@
+/*
+ * 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.llap.io.api;
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.common.io.CacheTag;
+
+/**
+ * Everything that identifies a single Parquet split served through the LLAP data cache: the file
+ * (its optional {@code fileKey}, {@code path} and {@code tag}), the split ({@code offset} and
+ * {@code length}), the projection ({@code tableIncludedCols}), and the defaults for columns
+ * absent from the file ({@code initialDefaults}). Kept together so
+ * {@link LlapIo#llapVectorizedParquetReaderForPath} stays a three-parameter call with the job
+ * conf and reporter alongside.
+ */
+public record LlapParquetReadRequest(
+    Object fileKey,
+    Path path,
+    CacheTag tag,
+    List<Integer> tableIncludedCols,
+    long offset,
+    long length,
+    Map<String, Object> initialDefaults) {
+}

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

@@ -48,7 +48,9 @@
 import org.apache.hadoop.hive.ql.exec.vector.VectorizedInputFormatInterface;
 import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
 import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatchCtx;
+import org.apache.hadoop.hive.llap.io.api.LlapProxy;
 import org.apache.hadoop.hive.ql.io.CombineHiveInputFormat.AvoidSplitCombination;
+import org.apache.hadoop.hive.ql.io.LlapCacheOnlyInputFormatInterface;
 import org.apache.hadoop.hive.ql.io.LlapAwareSplit;
 import org.apache.hadoop.hive.ql.io.NullRowsInputFormat.NullRowsRecordReader;
 import org.apache.hadoop.hive.ql.io.SelfDescribingInputFormatInterface;
@@ -123,19 +125,15 @@ public RecordReader<NullWritable, VectorizedRowBatch> getRecordReader(
           cvp, executor, sourceInputFormat, sourceSerDe, reporter, daemonConf);
       if (rr == null) {
         // Reader-specific incompatibility like SMB or schema evolution.
+        if (sourceInputFormat instanceof LlapCacheOnlyInputFormatInterface) {
+          LlapProxy.getIo().initCacheOnlyInputFormat(sourceInputFormat);
+        }
         return sourceInputFormat.getRecordReader(split, job, reporter);
       }
-      // For non-vectorized operator case, wrap the reader if possible.
-      RecordReader<NullWritable, VectorizedRowBatch> result = rr;
-      if (!Utilities.getIsVectorized(job)) {
-        result = null;
-        if (HiveConf.getBoolVar(job, ConfVars.LLAP_IO_ROW_WRAPPER_ENABLED)) {
-          result = wrapLlapReader(tableIncludedCols, rr, split);
-        }
-        if (result == null) {
-          // Cannot wrap a reader for non-vectorized pipeline.
-          return sourceInputFormat.getRecordReader(split, job, reporter);
-        }
+      RecordReader<NullWritable, VectorizedRowBatch> result =
+          wrapOrFallback(rr, tableIncludedCols, split, job);
+      if (result == null) {
+        return sourceInputFormat.getRecordReader(split, job, reporter);
       }
       // This starts the reader in the background.
       rr.start();
@@ -154,6 +152,24 @@ public RecordReader<NullWritable, VectorizedRowBatch> getRecordReader(
     }
   }
 
+  /**
+   * Adapts the LLAP reader (which always produces vectorized batches) to what the surrounding
+   * operator pipeline expects. In a vectorized pipeline it is handed back as-is; in a
+   * non-vectorized one a row wrapper is attempted, and if that isn't possible (wrapper disabled
+   * or unavailable for this reader) null is returned so the caller falls back to
+   * {@link #sourceInputFormat}.
+   */
+  private RecordReader<NullWritable, VectorizedRowBatch> wrapOrFallback(LlapRecordReader rr,
+      List<Integer> tableIncludedCols, InputSplit split, JobConf job) throws IOException {
+    if (Utilities.getIsVectorized(job)) {
+      return rr;
+    }
+    if (!HiveConf.getBoolVar(job, ConfVars.LLAP_IO_ROW_WRAPPER_ENABLED)) {
+      return null;
+    }
+    return wrapLlapReader(tableIncludedCols, rr, split);
+  }
+
   private boolean checkLimitReached(JobConf job) {
     /*
      * 2 assumptions here when using "tez.mapreduce.vertex.name"

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

@@ -23,6 +23,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
@@ -49,6 +50,7 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hive.common.io.Allocator;
 import org.apache.hadoop.hive.common.io.DataCache;
+import org.apache.hadoop.hive.common.io.DataCache.BooleanRef;
 import org.apache.hadoop.hive.common.io.DiskRange;
 import org.apache.hadoop.hive.common.io.DiskRangeList;
 import org.apache.hadoop.hive.common.io.FileMetadataCache;
@@ -74,9 +76,12 @@
 import org.apache.hadoop.hive.llap.cache.LowLevelCache.Priority;
 import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos;
 import org.apache.hadoop.hive.llap.io.api.LlapIo;
+import org.apache.hadoop.hive.llap.io.api.LlapParquetReadRequest;
 import org.apache.hadoop.hive.llap.io.decode.ColumnVectorProducer;
 import org.apache.hadoop.hive.llap.io.decode.GenericColumnVectorProducer;
 import org.apache.hadoop.hive.llap.io.decode.OrcColumnVectorProducer;
+import org.apache.hadoop.hive.llap.io.decode.ParquetColumnVectorProducer;
+import org.apache.hadoop.hive.llap.io.decode.ParquetEncodedDataConsumer;
 import org.apache.hadoop.hive.llap.io.encoded.OrcEncodedDataReader;
 import org.apache.hadoop.hive.llap.io.metadata.MetadataCache;
 import org.apache.hadoop.hive.llap.metrics.LlapDaemonCacheMetrics;
@@ -87,11 +92,13 @@
 import org.apache.hadoop.hive.ql.io.orc.OrcSplit;
 import org.apache.hadoop.hive.ql.io.orc.encoded.IoTrace;
 import org.apache.hadoop.hive.ql.io.orc.OrcInputFormat;
+import org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat;
 import org.apache.hadoop.hive.ql.io.parquet.vector.ParquetFooterInputFromCache;
 import org.apache.hadoop.hive.ql.io.parquet.vector.VectorizedParquetRecordReader;
 import org.apache.hadoop.hive.ql.metadata.HiveException;
 import org.apache.hadoop.hive.serde2.Deserializer;
 import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.mapred.FileSplit;
 import org.apache.hadoop.mapred.InputFormat;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.RecordReader;
@@ -122,6 +129,7 @@
 
   // TODO: later, we may have a map
   private final ColumnVectorProducer orcCvp, genericCvp;
+  private final ColumnVectorProducer parquetCvp;
   private final ExecutorService executor;
   private final ExecutorService encodeExecutor;
   private final LlapDaemonCacheMetrics cacheMetrics;
@@ -171,7 +179,7 @@
 
     MetadataCache metadataCache = null;
     SerDeLowLevelCacheImpl serdeCache = null; // TODO: extract interface when needed
-    BufferUsageManager bufferManagerOrc = null, bufferManagerGeneric = null;
+    BufferUsageManager bufferManagerData = null, bufferManagerGeneric = null;
     boolean isEncodeEnabled = useLowLevelCache
         && HiveConf.getBoolVar(conf, ConfVars.LLAP_IO_ENCODE_ENABLED);
     if (useLowLevelCache) {
@@ -219,7 +227,7 @@
       cachePolicyWrapper.setEvictionListener(e);
 
       cacheImpl.startThreads(); // Start the cache threads.
-      bufferManager = bufferManagerOrc = cacheImpl; // Cache also serves as buffer manager.
+      bufferManager = bufferManagerData = cacheImpl; // Cache also serves as buffer manager.
       bufferManagerGeneric = serdeCache;
       if (trackUsage) {
         debugDumpComponents.add(cachePolicyWrapper); // Cache contents tracker.
@@ -238,7 +246,7 @@
       this.allocator = new SimpleAllocator(conf);
       fileMetadataCache = null;
       SimpleBufferManager sbm = new SimpleBufferManager(allocator, cacheMetrics);
-      bufferManager = bufferManagerOrc = bufferManagerGeneric = sbm;
+      bufferManager = bufferManagerData = bufferManagerGeneric = sbm;
       dataCache = sbm;
       this.memoryManager = null;
       debugDumpComponents.add(new LlapIoDebugDump() {
@@ -269,9 +277,13 @@
 
     // TODO: this should depends on input format and be in a map, or something.
     this.orcCvp = new OrcColumnVectorProducer(
-        metadataCache, dataCache, pathCache, bufferManagerOrc, conf, cacheMetrics, ioMetrics, tracePool);
+        metadataCache, dataCache, pathCache, bufferManagerData, conf, cacheMetrics, ioMetrics, tracePool);
     this.genericCvp = isEncodeEnabled ? new GenericColumnVectorProducer(
         serdeCache, bufferManagerGeneric, conf, cacheMetrics, ioMetrics, tracePool, encodeExecutor) : null;
+    // Native Parquet IO is gated per query by the job conf at the dispatch sites.
+    this.parquetCvp = dataCache != null
+        ? new ParquetColumnVectorProducer(dataCache, bufferManagerData, conf, cacheMetrics, ioMetrics)
+        : null;
     LOG.info("LLAP IO initialized");
 
     registerMXBeans();
@@ -342,6 +354,13 @@
     ColumnVectorProducer cvp = genericCvp;
     if (sourceInputFormat instanceof OrcInputFormat) {
       cvp = orcCvp; // Special-case for ORC.
+    } else if (sourceInputFormat instanceof MapredParquetInputFormat && sourceSerDe == null) {
+      // Parquet arrives without a SerDe only from HiveInputFormat's native-cache dispatch; with one
+      // it is the encode.formats configuration, which keeps the re-encoding producer.
+      if (parquetCvp == null) {
+        return null;
+      }
+      cvp = parquetCvp;
     } else if (cvp == null) {
       LOG.warn("LLAP encode is disabled; cannot use for " + sourceInputFormat.getClass());
       return null;
@@ -474,8 +493,30 @@
   }
 
   @Override
-  public MemoryBufferOrBuffers getParquetFooterBuffersFromCache(Path path, JobConf conf, @Nullable Object fileKey)
-      throws IOException {
+  public RecordReader<NullWritable, VectorizedRowBatch> llapVectorizedParquetReaderForPath(
+      LlapParquetReadRequest request, JobConf conf, Reporter reporter) throws IOException {
+    if (parquetCvp == null) {
+      return null;
+    }
+    FileSplit split = new FileSplit(request.path(), request.offset(), request.length(), (String[]) null);
+    try {
+      LlapRecordReader rr = LlapRecordReader.create(conf, split, request.tableIncludedCols(),
+          HiveStringUtils.getHostname(), parquetCvp, executor, null, null, reporter, daemonConf);
+      if (rr == null) {  // NOSONAR - S2583: create() has null-return paths Sonar's data-flow does not model.
+        return null;
+      }
+      ((ParquetEncodedDataConsumer) rr.getReadPipeline()).setInitialDefaults(request.initialDefaults());
+      rr.setPartitionValues(null);
+      rr.start();
+      return rr;
+    } catch (HiveException e) {
+      throw new IOException(e);
+    }
+  }
+
+  @Override
+  public MemoryBufferOrBuffers getParquetFooterBuffersFromCache(Path path, JobConf conf, @Nullable Object fileKey,
+      @Nullable BooleanRef cacheHit) throws IOException {
 
     Preconditions.checkNotNull(fileMetadataCache, "Metadata cache must not be null");
 
@@ -484,6 +525,9 @@
 
     MemoryBufferOrBuffers footerData = (fileKey == null ) ? null
         : fileMetadataCache.getFileMetadata(fileKey);
+    if (cacheHit != null) {
+      cacheHit.value = footerData != null;
+    }
     if (footerData != null) {
       LOG.info("Found the footer in cache for " + fileKey);
       try {