This is an automated email from the ASF dual-hosted git repository.
szita pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new a35a6ea HIVE-25628: Avoid unnecessary file ops if Iceberg table is
LLAP cached (Adam Szita, reviewed by Marton Bod)
a35a6ea is described below
commit a35a6ea7c16888664a631470e1773319cab59ff6
Author: Adam Szita <[email protected]>
AuthorDate: Thu Oct 28 15:12:08 2021 +0200
HIVE-25628: Avoid unnecessary file ops if Iceberg table is LLAP cached
(Adam Szita, reviewed by Marton Bod)
---
.../mr/hive/vector/HiveVectorizedReader.java | 12 ++-
.../apache/iceberg/orc/VectorizedReadUtils.java | 113 +++++++++++++++------
2 files changed, 90 insertions(+), 35 deletions(-)
diff --git
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vector/HiveVectorizedReader.java
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vector/HiveVectorizedReader.java
index 4f5acea..1375ee8 100644
---
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vector/HiveVectorizedReader.java
+++
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vector/HiveVectorizedReader.java
@@ -27,6 +27,7 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.llap.io.api.LlapProxy;
import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.hadoop.hive.ql.io.SyntheticFileId;
import org.apache.hadoop.hive.ql.io.orc.OrcSplit;
import org.apache.hadoop.hive.ql.io.orc.VectorizedOrcInputFormat;
import org.apache.hadoop.hive.serde2.ColumnProjectionUtils;
@@ -114,19 +115,24 @@ public class HiveVectorizedReader {
// Need to turn positional schema evolution off since we use column
name based schema evolution for projection
// and Iceberg will make a mapping between the file schema and the
current reading schema.
job.setBoolean(OrcConf.FORCE_POSITIONAL_EVOLUTION.getHiveConfName(),
false);
- VectorizedReadUtils.handleIcebergProjection(inputFile, task, job);
+
+ // TODO: Iceberg currently does not track the last modification time
of a file. Until that's added,
+ // we need to set Long.MIN_VALUE as last modification time in the
fileId triplet.
+ SyntheticFileId fileId = new SyntheticFileId(path,
task.file().fileSizeInBytes(), Long.MIN_VALUE);
+
+ VectorizedReadUtils.handleIcebergProjection(inputFile, task, job,
fileId);
RecordReader<NullWritable, VectorizedRowBatch> recordReader = null;
// If LLAP enabled, try to retrieve an LLAP record reader - this
might yield to null in some special cases
if (HiveConf.getBoolVar(job, HiveConf.ConfVars.LLAP_IO_ENABLED,
LlapProxy.isDaemon()) &&
LlapProxy.getIo() != null) {
- recordReader =
LlapProxy.getIo().llapVectorizedOrcReaderForPath(null, path, null,
readColumnIds,
+ recordReader =
LlapProxy.getIo().llapVectorizedOrcReaderForPath(fileId, path, null,
readColumnIds,
job, task.start(), task.length(), reporter);
}
if (recordReader == null) {
- InputSplit split = new OrcSplit(path, null, task.start(),
task.length(), (String[]) null, null, false,
+ InputSplit split = new OrcSplit(path, fileId, task.start(),
task.length(), (String[]) null, null, false,
false, com.google.common.collect.Lists.newArrayList(), 0,
task.length(), path.getParent(), null);
recordReader = new
VectorizedOrcInputFormat().getRecordReader(split, job, reporter);
}
diff --git
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/orc/VectorizedReadUtils.java
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/orc/VectorizedReadUtils.java
index addf202..30f66b4 100644
---
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/orc/VectorizedReadUtils.java
+++
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/orc/VectorizedReadUtils.java
@@ -20,73 +20,122 @@
package org.apache.iceberg.orc;
import java.io.IOException;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.common.io.CacheTag;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.llap.LlapHiveUtils;
+import org.apache.hadoop.hive.llap.io.api.LlapProxy;
+import org.apache.hadoop.hive.ql.io.SyntheticFileId;
import org.apache.hadoop.hive.ql.io.sarg.ConvertAstToSearchArg;
+import org.apache.hadoop.hive.ql.plan.MapWork;
+import org.apache.hadoop.hive.ql.plan.PartitionDesc;
import org.apache.hadoop.hive.ql.plan.TableScanDesc;
import org.apache.hadoop.hive.serde2.ColumnProjectionUtils;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hive.iceberg.org.apache.orc.Reader;
import org.apache.hive.iceberg.org.apache.orc.TypeDescription;
+import org.apache.hive.iceberg.org.apache.orc.impl.ReaderImpl;
import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.Schema;
import org.apache.iceberg.expressions.Binder;
import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.io.InputFile;
import org.apache.iceberg.mapping.MappingUtil;
+import org.apache.orc.impl.BufferChunk;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Utilities that rely on Iceberg code from org.apache.iceberg.orc package.
*/
public class VectorizedReadUtils {
+ private static final Logger LOG =
LoggerFactory.getLogger(VectorizedReadUtils.class);
+
private VectorizedReadUtils() {
}
+ private static TypeDescription getSchemaForFile(InputFile inputFile,
SyntheticFileId fileId, JobConf job)
+ throws IOException {
+ TypeDescription schema = null;
+
+ if (HiveConf.getBoolVar(job, HiveConf.ConfVars.LLAP_IO_ENABLED,
LlapProxy.isDaemon()) &&
+ LlapProxy.getIo() != null) {
+ MapWork mapWork = LlapHiveUtils.findMapWork(job);
+ Path path = new Path(inputFile.location());
+ PartitionDesc partitionDesc = LlapHiveUtils.partitionDescForPath(path,
mapWork.getPathToPartitionInfo());
+
+ // 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;
+
+ try {
+ // Schema has to be serialized and deserialized as it is passed
between different packages of TypeDescription:
+ // Iceberg expects
org.apache.hive.iceberg.org.apache.orc.TypeDescription as it shades ORC, while
LLAP provides
+ // the unshaded org.apache.orc.TypeDescription type.
+ BufferChunk tailBuffer = LlapProxy.getIo().getOrcTailFromCache(path,
job, cacheTag, fileId).getTailBuffer();
+ schema = ReaderImpl.extractFileTail(tailBuffer.getData()).getSchema();
+ } catch (IOException ioe) {
+ LOG.warn("LLAP is turned on but was unable to get file metadata
information through its cache for {}",
+ path, ioe);
+ }
+
+ }
+
+ // Fallback to simple ORC reader file opening method in lack of or failure
of LLAP.
+ if (schema == null) {
+ try (Reader orcFileReader = ORC.newFileReader(inputFile, job)) {
+ schema = orcFileReader.getSchema();
+ }
+ }
+
+ return schema;
+
+ }
+
/**
* Adjusts the jobConf so that column reorders and renames that might have
happened since this ORC file was written
* are properly mapped to the schema of the original file.
* @param inputFile - the original ORC file - this needs to be accessed to
retrieve the original schema for mapping
* @param task - Iceberg task - required for
* @param job - JobConf instance to adjust
+ * @param fileId - FileID for the input file, serves as cache key in an LLAP
setup
* @throws IOException - errors relating to accessing the ORC file
*/
- public static void handleIcebergProjection(InputFile inputFile, FileScanTask
task, JobConf job)
- throws IOException {
- Reader orcFileReader = ORC.newFileReader(inputFile, job);
-
- try {
- // We need to map with the current (i.e. current Hive table columns)
full schema (without projections),
- // as OrcInputFormat will take care of the projections by the use of an
include boolean array
- Schema currentSchema = task.spec().schema();
- TypeDescription fileSchema = orcFileReader.getSchema();
-
- TypeDescription readOrcSchema;
- if (ORCSchemaUtil.hasIds(fileSchema)) {
- readOrcSchema = ORCSchemaUtil.buildOrcProjection(currentSchema,
fileSchema);
- } else {
- TypeDescription typeWithIds =
- ORCSchemaUtil.applyNameMapping(fileSchema,
MappingUtil.create(currentSchema));
- readOrcSchema = ORCSchemaUtil.buildOrcProjection(currentSchema,
typeWithIds);
- }
+ public static void handleIcebergProjection(InputFile inputFile, FileScanTask
task, JobConf job,
+ SyntheticFileId fileId) throws IOException {
+
+ // We need to map with the current (i.e. current Hive table columns) full
schema (without projections),
+ // as OrcInputFormat will take care of the projections by the use of an
include boolean array
+ Schema currentSchema = task.spec().schema();
+ TypeDescription fileSchema = getSchemaForFile(inputFile, fileId, job);
+
+ TypeDescription readOrcSchema;
+ if (ORCSchemaUtil.hasIds(fileSchema)) {
+ readOrcSchema = ORCSchemaUtil.buildOrcProjection(currentSchema,
fileSchema);
+ } else {
+ TypeDescription typeWithIds =
+ ORCSchemaUtil.applyNameMapping(fileSchema,
MappingUtil.create(currentSchema));
+ readOrcSchema = ORCSchemaUtil.buildOrcProjection(currentSchema,
typeWithIds);
+ }
- job.set(ColumnProjectionUtils.ORC_SCHEMA_STRING,
readOrcSchema.toString());
+ job.set(ColumnProjectionUtils.ORC_SCHEMA_STRING, readOrcSchema.toString());
- // Predicate pushdowns needs to be adjusted too in case of column
renames, we let Iceberg generate this into job
- if (task.residual() != null) {
- Expression boundFilter = Binder.bind(currentSchema.asStruct(),
task.residual(), false);
+ // Predicate pushdowns needs to be adjusted too in case of column renames,
we let Iceberg generate this into job
+ if (task.residual() != null) {
+ Expression boundFilter = Binder.bind(currentSchema.asStruct(),
task.residual(), false);
- // Note the use of the unshaded version of this class here (required
for SARG deseralization later)
- org.apache.hadoop.hive.ql.io.sarg.SearchArgument sarg =
- ExpressionToOrcSearchArgument.convert(boundFilter, readOrcSchema);
- if (sarg != null) {
- job.unset(TableScanDesc.FILTER_EXPR_CONF_STR);
- job.unset(ConvertAstToSearchArg.SARG_PUSHDOWN);
+ // Note the use of the unshaded version of this class here (required for
SARG deseralization later)
+ org.apache.hadoop.hive.ql.io.sarg.SearchArgument sarg =
+ ExpressionToOrcSearchArgument.convert(boundFilter, readOrcSchema);
+ if (sarg != null) {
+ job.unset(TableScanDesc.FILTER_EXPR_CONF_STR);
+ job.unset(ConvertAstToSearchArg.SARG_PUSHDOWN);
- job.set(ConvertAstToSearchArg.SARG_PUSHDOWN,
ConvertAstToSearchArg.sargToKryo(sarg));
- }
+ job.set(ConvertAstToSearchArg.SARG_PUSHDOWN,
ConvertAstToSearchArg.sargToKryo(sarg));
}
- } finally {
- orcFileReader.close();
}
}
}
