| /* |
| * 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.orc.impl; |
| |
| import org.apache.hadoop.fs.FSDataInputStream; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.hive.common.type.HiveDecimal; |
| import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch; |
| import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf; |
| import org.apache.hadoop.hive.ql.io.sarg.SearchArgument; |
| import org.apache.hadoop.hive.ql.io.sarg.SearchArgument.TruthValue; |
| import org.apache.hadoop.hive.ql.util.TimestampUtils; |
| import org.apache.hadoop.hive.serde2.io.DateWritable; |
| import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable; |
| import org.apache.hadoop.io.Text; |
| import org.apache.orc.BooleanColumnStatistics; |
| import org.apache.orc.CollectionColumnStatistics; |
| import org.apache.orc.ColumnStatistics; |
| import org.apache.orc.CompressionCodec; |
| import org.apache.orc.DataReader; |
| import org.apache.orc.DateColumnStatistics; |
| import org.apache.orc.DecimalColumnStatistics; |
| import org.apache.orc.DoubleColumnStatistics; |
| import org.apache.orc.IntegerColumnStatistics; |
| import org.apache.orc.OrcConf; |
| import org.apache.orc.OrcFile; |
| import org.apache.orc.OrcProto; |
| import org.apache.orc.Reader; |
| import org.apache.orc.RecordReader; |
| import org.apache.orc.StringColumnStatistics; |
| import org.apache.orc.StripeInformation; |
| import org.apache.orc.TimestampColumnStatistics; |
| import org.apache.orc.TypeDescription; |
| import org.apache.orc.impl.reader.ReaderEncryption; |
| import org.apache.orc.impl.reader.StripePlanner; |
| import org.apache.orc.impl.reader.tree.BatchReader; |
| import org.apache.orc.util.BloomFilter; |
| import org.apache.orc.util.BloomFilterIO; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import java.io.IOException; |
| import java.math.BigDecimal; |
| import java.sql.Date; |
| import java.sql.Timestamp; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.List; |
| import java.util.TimeZone; |
| |
| public class RecordReaderImpl implements RecordReader { |
| static final Logger LOG = LoggerFactory.getLogger(RecordReaderImpl.class); |
| private static final boolean isLogDebugEnabled = LOG.isDebugEnabled(); |
| protected final Path path; |
| private final long firstRow; |
| private final List<StripeInformation> stripes = new ArrayList<>(); |
| private OrcProto.StripeFooter stripeFooter; |
| private final long totalRowCount; |
| protected final TypeDescription schema; |
| // the file included columns indexed by the file's column ids. |
| private final boolean[] fileIncluded; |
| private final long rowIndexStride; |
| private long rowInStripe = 0; |
| private int currentStripe = -1; |
| private long rowBaseInStripe = 0; |
| private long rowCountInStripe = 0; |
| private final BatchReader reader; |
| private final OrcIndex indexes; |
| private final SargApplier sargApp; |
| // an array about which row groups aren't skipped |
| private boolean[] includedRowGroups = null; |
| private final DataReader dataReader; |
| private final int maxDiskRangeChunkLimit; |
| private final StripePlanner planner; |
| |
| /** |
| * Given a list of column names, find the given column and return the index. |
| * |
| * @param evolution the mapping from reader to file schema |
| * @param columnName the fully qualified column name to look for |
| * @return the file column number or -1 if the column wasn't found |
| */ |
| static int findColumns(SchemaEvolution evolution, |
| String columnName) { |
| try { |
| TypeDescription readerColumn = |
| evolution.getReaderBaseSchema().findSubtype(columnName); |
| TypeDescription fileColumn = evolution.getFileType(readerColumn); |
| return fileColumn == null ? -1 : fileColumn.getId(); |
| } catch (IllegalArgumentException e) { |
| if (LOG.isDebugEnabled()){ |
| LOG.debug("{}", e.getMessage()); |
| } |
| return -1; |
| } |
| } |
| |
| /** |
| * Find the mapping from predicate leaves to columns. |
| * @param sargLeaves the search argument that we need to map |
| * @param evolution the mapping from reader to file schema |
| * @return an array mapping the sarg leaves to concrete column numbers in the |
| * file |
| */ |
| public static int[] mapSargColumnsToOrcInternalColIdx( |
| List<PredicateLeaf> sargLeaves, |
| SchemaEvolution evolution) { |
| int[] result = new int[sargLeaves.size()]; |
| Arrays.fill(result, -1); |
| for(int i=0; i < result.length; ++i) { |
| String colName = sargLeaves.get(i).getColumnName(); |
| result[i] = findColumns(evolution, colName); |
| } |
| return result; |
| } |
| |
| protected RecordReaderImpl(ReaderImpl fileReader, |
| Reader.Options options) throws IOException { |
| OrcFile.WriterVersion writerVersion = fileReader.getWriterVersion(); |
| SchemaEvolution evolution; |
| if (options.getSchema() == null) { |
| if (LOG.isInfoEnabled()) { |
| LOG.info("Reader schema not provided -- using file schema " + |
| fileReader.getSchema()); |
| } |
| evolution = new SchemaEvolution(fileReader.getSchema(), null, options); |
| } else { |
| |
| // Now that we are creating a record reader for a file, validate that |
| // the schema to read is compatible with the file schema. |
| // |
| evolution = new SchemaEvolution(fileReader.getSchema(), |
| options.getSchema(), |
| options); |
| if (LOG.isDebugEnabled() && evolution.hasConversion()) { |
| LOG.debug("ORC file " + fileReader.path.toString() + |
| " has data type conversion --\n" + |
| "reader schema: " + options.getSchema().toString() + "\n" + |
| "file schema: " + fileReader.getSchema()); |
| } |
| } |
| this.schema = evolution.getReaderSchema(); |
| this.path = fileReader.path; |
| this.rowIndexStride = fileReader.rowIndexStride; |
| boolean ignoreNonUtf8BloomFilter = OrcConf.IGNORE_NON_UTF8_BLOOM_FILTERS.getBoolean(fileReader.conf); |
| ReaderEncryption encryption = fileReader.getEncryption(); |
| this.fileIncluded = evolution.getFileIncluded(); |
| SearchArgument sarg = options.getSearchArgument(); |
| if (sarg != null && rowIndexStride != 0) { |
| sargApp = new SargApplier(sarg, |
| rowIndexStride, |
| evolution, |
| writerVersion, |
| fileReader.useUTCTimestamp, |
| fileReader.writerUsedProlepticGregorian(), |
| fileReader.options.getConvertToProlepticGregorian()); |
| } else { |
| sargApp = null; |
| } |
| |
| long rows = 0; |
| long skippedRows = 0; |
| long offset = options.getOffset(); |
| long maxOffset = options.getMaxOffset(); |
| for(StripeInformation stripe: fileReader.getStripes()) { |
| long stripeStart = stripe.getOffset(); |
| if (offset > stripeStart) { |
| skippedRows += stripe.getNumberOfRows(); |
| } else if (stripeStart < maxOffset) { |
| this.stripes.add(stripe); |
| rows += stripe.getNumberOfRows(); |
| } |
| } |
| this.maxDiskRangeChunkLimit = OrcConf.ORC_MAX_DISK_RANGE_CHUNK_LIMIT.getInt(fileReader.conf); |
| Boolean zeroCopy = options.getUseZeroCopy(); |
| if (zeroCopy == null) { |
| zeroCopy = OrcConf.USE_ZEROCOPY.getBoolean(fileReader.conf); |
| } |
| if (options.getDataReader() != null) { |
| this.dataReader = options.getDataReader().clone(); |
| } else { |
| InStream.StreamOptions unencryptedOptions = |
| InStream.options() |
| .withCodec(OrcCodecPool.getCodec(fileReader.getCompressionKind())) |
| .withBufferSize(fileReader.getCompressionSize()); |
| DataReaderProperties.Builder builder = |
| DataReaderProperties.builder() |
| .withCompression(unencryptedOptions) |
| .withFileSystemSupplier(fileReader.getFileSystemSupplier()) |
| .withPath(fileReader.path) |
| .withMaxDiskRangeChunkLimit(maxDiskRangeChunkLimit) |
| .withZeroCopy(zeroCopy); |
| FSDataInputStream file = fileReader.takeFile(); |
| if (file != null) { |
| builder.withFile(file); |
| } |
| this.dataReader = RecordReaderUtils.createDefaultDataReader( |
| builder.build()); |
| } |
| firstRow = skippedRows; |
| totalRowCount = rows; |
| Boolean skipCorrupt = options.getSkipCorruptRecords(); |
| if (skipCorrupt == null) { |
| skipCorrupt = OrcConf.SKIP_CORRUPT_DATA.getBoolean(fileReader.conf); |
| } |
| |
| TreeReaderFactory.ReaderContext readerContext = |
| new TreeReaderFactory.ReaderContext() |
| .setSchemaEvolution(evolution) |
| .skipCorrupt(skipCorrupt) |
| .fileFormat(fileReader.getFileVersion()) |
| .useUTCTimestamp(fileReader.useUTCTimestamp) |
| .setProlepticGregorian(fileReader.writerUsedProlepticGregorian(), |
| fileReader.options.getConvertToProlepticGregorian()) |
| .setEncryption(encryption); |
| reader = TreeReaderFactory.createRootReader(evolution.getReaderSchema(), readerContext); |
| |
| int columns = evolution.getFileSchema().getMaximumId() + 1; |
| indexes = new OrcIndex(new OrcProto.RowIndex[columns], |
| new OrcProto.Stream.Kind[columns], |
| new OrcProto.BloomFilterIndex[columns]); |
| |
| planner = new StripePlanner(evolution.getFileSchema(), encryption, |
| dataReader, writerVersion, ignoreNonUtf8BloomFilter, |
| maxDiskRangeChunkLimit); |
| |
| try { |
| advanceToNextRow(reader, 0L, true); |
| } catch (IOException e) { |
| // Try to close since this happens in constructor. |
| close(); |
| throw e; |
| } |
| } |
| |
| public static final class PositionProviderImpl implements PositionProvider { |
| private final OrcProto.RowIndexEntry entry; |
| private int index; |
| |
| public PositionProviderImpl(OrcProto.RowIndexEntry entry) { |
| this(entry, 0); |
| } |
| |
| public PositionProviderImpl(OrcProto.RowIndexEntry entry, int startPos) { |
| this.entry = entry; |
| this.index = startPos; |
| } |
| |
| @Override |
| public long getNext() { |
| return entry.getPositions(index++); |
| } |
| } |
| |
| public static final class ZeroPositionProvider implements PositionProvider { |
| @Override |
| public long getNext() { |
| return 0; |
| } |
| } |
| |
| public OrcProto.StripeFooter readStripeFooter(StripeInformation stripe |
| ) throws IOException { |
| return dataReader.readStripeFooter(stripe); |
| } |
| |
| enum Location { |
| BEFORE, MIN, MIDDLE, MAX, AFTER |
| } |
| |
| static class ValueRange<T extends Comparable> { |
| final Comparable lower; |
| final Comparable upper; |
| final boolean onlyLowerBound; |
| final boolean onlyUpperBound; |
| final boolean hasNulls; |
| |
| ValueRange(PredicateLeaf predicate, |
| T lower, T upper, |
| boolean hasNulls, |
| boolean onlyLowerBound, |
| boolean onlyUpperBound) { |
| PredicateLeaf.Type type = predicate.getType(); |
| this.lower = getBaseObjectForComparison(type, lower); |
| this.upper = getBaseObjectForComparison(type, upper); |
| this.hasNulls = hasNulls; |
| this.onlyLowerBound = onlyLowerBound; |
| this.onlyUpperBound = onlyUpperBound; |
| } |
| |
| ValueRange(PredicateLeaf predicate, T lower, T upper, |
| boolean hasNulls) { |
| this(predicate, lower, upper, hasNulls, false, false); |
| } |
| |
| boolean hasValues() { |
| return lower != null; |
| } |
| |
| /** |
| * Given a point and min and max, determine if the point is before, at the |
| * min, in the middle, at the max, or after the range. |
| * @param point the point to test |
| * @return the location of the point |
| */ |
| Location compare(Comparable point) { |
| int minCompare = point.compareTo(lower); |
| if (minCompare < 0) { |
| return Location.BEFORE; |
| } else if (minCompare == 0) { |
| return onlyLowerBound ? Location.BEFORE : Location.MIN; |
| } |
| int maxCompare = point.compareTo(upper); |
| if (maxCompare > 0) { |
| return Location.AFTER; |
| } else if (maxCompare == 0) { |
| return onlyUpperBound ? Location.AFTER : Location.MAX; |
| } |
| return Location.MIDDLE; |
| } |
| |
| /** |
| * Is this range a single point? |
| * @return true if min == max |
| */ |
| boolean isSingleton() { |
| return lower != null && !onlyLowerBound && !onlyUpperBound && |
| lower.equals(upper); |
| } |
| |
| /** |
| * Add the null option to the truth value, if the range includes nulls. |
| * @param value the original truth value |
| * @return the truth value extended with null if appropriate |
| */ |
| TruthValue addNull(TruthValue value) { |
| if (hasNulls) { |
| switch (value) { |
| case YES: |
| return TruthValue.YES_NULL; |
| case NO: |
| return TruthValue.NO_NULL; |
| case YES_NO: |
| return TruthValue.YES_NO_NULL; |
| default: |
| return value; |
| } |
| } else { |
| return value; |
| } |
| } |
| } |
| |
| /** |
| * Get the maximum value out of an index entry. |
| * Includes option to specify if timestamp column stats values |
| * should be in UTC. |
| * @param index the index entry |
| * @param predicate the kind of predicate |
| * @param useUTCTimestamp use UTC for timestamps |
| * @return the object for the maximum value or null if there isn't one |
| */ |
| static ValueRange getValueRange(ColumnStatistics index, |
| PredicateLeaf predicate, |
| boolean useUTCTimestamp) { |
| if (index instanceof IntegerColumnStatistics) { |
| IntegerColumnStatistics stats = (IntegerColumnStatistics) index; |
| Long min = stats.getMinimum(); |
| Long max = stats.getMaximum(); |
| return new ValueRange<>(predicate, min, max, stats.hasNull()); |
| } else if (index instanceof CollectionColumnStatistics) { |
| CollectionColumnStatistics stats = (CollectionColumnStatistics) index; |
| Long min = stats.getMinimumChildren(); |
| Long max = stats.getMaximumChildren(); |
| return new ValueRange<>(predicate, min, max, stats.hasNull()); |
| }else if (index instanceof DoubleColumnStatistics) { |
| DoubleColumnStatistics stats = (DoubleColumnStatistics) index; |
| Double min = stats.getMinimum(); |
| Double max = stats.getMaximum(); |
| return new ValueRange<>(predicate, min, max, stats.hasNull()); |
| } else if (index instanceof StringColumnStatistics) { |
| StringColumnStatistics stats = (StringColumnStatistics) index; |
| return new ValueRange<>(predicate, stats.getLowerBound(), |
| stats.getUpperBound(), stats.hasNull(), stats.getMinimum() == null, |
| stats.getMaximum() == null); |
| } else if (index instanceof DateColumnStatistics) { |
| DateColumnStatistics stats = (DateColumnStatistics) index; |
| java.util.Date min = stats.getMinimum(); |
| java.util.Date max = stats.getMaximum(); |
| return new ValueRange<>(predicate, min, max, stats.hasNull()); |
| } else if (index instanceof DecimalColumnStatistics) { |
| DecimalColumnStatistics stats = (DecimalColumnStatistics) index; |
| HiveDecimal min = stats.getMinimum(); |
| HiveDecimal max = stats.getMaximum(); |
| return new ValueRange<>(predicate, min, max, stats.hasNull()); |
| } else if (index instanceof TimestampColumnStatistics) { |
| TimestampColumnStatistics stats = (TimestampColumnStatistics) index; |
| Timestamp min = useUTCTimestamp ? stats.getMinimumUTC() : stats.getMinimum(); |
| Timestamp max = useUTCTimestamp ? stats.getMaximumUTC() : stats.getMaximum(); |
| return new ValueRange<>(predicate, min, max, stats.hasNull()); |
| } else if (index instanceof BooleanColumnStatistics) { |
| BooleanColumnStatistics stats = (BooleanColumnStatistics) index; |
| Boolean min = stats.getFalseCount() == 0; |
| Boolean max = stats.getTrueCount() != 0; |
| return new ValueRange<>(predicate, min, max, stats.hasNull()); |
| } else { |
| return new ValueRange(predicate, null, null, true); |
| } |
| } |
| |
| /** |
| * Evaluate a predicate with respect to the statistics from the column |
| * that is referenced in the predicate. |
| * @param statsProto the statistics for the column mentioned in the predicate |
| * @param predicate the leaf predicate we need to evaluation |
| * @param bloomFilter the bloom filter |
| * @param writerVersion the version of software that wrote the file |
| * @param type what is the kind of this column |
| * @return the set of truth values that may be returned for the given |
| * predicate. |
| */ |
| static TruthValue evaluatePredicateProto(OrcProto.ColumnStatistics statsProto, |
| PredicateLeaf predicate, |
| OrcProto.Stream.Kind kind, |
| OrcProto.ColumnEncoding encoding, |
| OrcProto.BloomFilter bloomFilter, |
| OrcFile.WriterVersion writerVersion, |
| TypeDescription type) { |
| return evaluatePredicateProto(statsProto, predicate, kind, encoding, bloomFilter, |
| writerVersion, type, false); |
| } |
| |
| /** |
| * Evaluate a predicate with respect to the statistics from the column |
| * that is referenced in the predicate. |
| * Includes option to specify if timestamp column stats values |
| * should be in UTC. |
| * @param statsProto the statistics for the column mentioned in the predicate |
| * @param predicate the leaf predicate we need to evaluation |
| * @param bloomFilter the bloom filter |
| * @param writerVersion the version of software that wrote the file |
| * @param type what is the kind of this column |
| * @param useUTCTimestamp |
| * @return the set of truth values that may be returned for the given |
| * predicate. |
| */ |
| |
| static TruthValue evaluatePredicateProto(OrcProto.ColumnStatistics statsProto, |
| PredicateLeaf predicate, |
| OrcProto.Stream.Kind kind, |
| OrcProto.ColumnEncoding encoding, |
| OrcProto.BloomFilter bloomFilter, |
| OrcFile.WriterVersion writerVersion, |
| TypeDescription type, |
| boolean useUTCTimestamp) { |
| ColumnStatistics cs = ColumnStatisticsImpl.deserialize(null, statsProto); |
| ValueRange range = getValueRange(cs, predicate, useUTCTimestamp); |
| |
| // files written before ORC-135 stores timestamp wrt to local timezone causing issues with PPD. |
| // disable PPD for timestamp for all old files |
| TypeDescription.Category category = type.getCategory(); |
| if (category == TypeDescription.Category.TIMESTAMP) { |
| if (!writerVersion.includes(OrcFile.WriterVersion.ORC_135)) { |
| LOG.debug("Not using predication pushdown on {} because it doesn't " + |
| "include ORC-135. Writer version: {}", |
| predicate.getColumnName(), writerVersion); |
| return range.addNull(TruthValue.YES_NO); |
| } |
| if (predicate.getType() != PredicateLeaf.Type.TIMESTAMP && |
| predicate.getType() != PredicateLeaf.Type.DATE && |
| predicate.getType() != PredicateLeaf.Type.STRING) { |
| return range.addNull(TruthValue.YES_NO); |
| } |
| } else if (writerVersion == OrcFile.WriterVersion.ORC_135 && |
| category == TypeDescription.Category.DECIMAL && |
| type.getPrecision() <= TypeDescription.MAX_DECIMAL64_PRECISION) { |
| // ORC 1.5.0 to 1.5.5, which use WriterVersion.ORC_135, have broken |
| // min and max values for decimal64. See ORC-517. |
| LOG.debug("Not using predicate push down on {}, because the file doesn't"+ |
| " include ORC-517. Writer version: {}", |
| predicate.getColumnName(), writerVersion); |
| return TruthValue.YES_NO_NULL; |
| } |
| return evaluatePredicateRange(predicate, range, |
| BloomFilterIO.deserialize(kind, encoding, writerVersion, type.getCategory(), |
| bloomFilter), useUTCTimestamp); |
| } |
| |
| /** |
| * Evaluate a predicate with respect to the statistics from the column |
| * that is referenced in the predicate. |
| * @param stats the statistics for the column mentioned in the predicate |
| * @param predicate the leaf predicate we need to evaluation |
| * @return the set of truth values that may be returned for the given |
| * predicate. |
| */ |
| public static TruthValue evaluatePredicate(ColumnStatistics stats, |
| PredicateLeaf predicate, |
| BloomFilter bloomFilter) { |
| return evaluatePredicate(stats, predicate, bloomFilter, false); |
| } |
| |
| /** |
| * Evaluate a predicate with respect to the statistics from the column |
| * that is referenced in the predicate. |
| * Includes option to specify if timestamp column stats values |
| * should be in UTC. |
| * @param stats the statistics for the column mentioned in the predicate |
| * @param predicate the leaf predicate we need to evaluation |
| * @param bloomFilter |
| * @param useUTCTimestamp |
| * @return the set of truth values that may be returned for the given |
| * predicate. |
| */ |
| public static TruthValue evaluatePredicate(ColumnStatistics stats, |
| PredicateLeaf predicate, |
| BloomFilter bloomFilter, |
| boolean useUTCTimestamp) { |
| ValueRange range = getValueRange(stats, predicate, useUTCTimestamp); |
| |
| return evaluatePredicateRange(predicate, range, bloomFilter, useUTCTimestamp); |
| } |
| |
| static TruthValue evaluatePredicateRange(PredicateLeaf predicate, |
| ValueRange range, |
| BloomFilter bloomFilter, |
| boolean useUTCTimestamp) { |
| // if we didn't have any values, everything must have been null |
| if (!range.hasValues()) { |
| if (predicate.getOperator() == PredicateLeaf.Operator.IS_NULL) { |
| return TruthValue.YES; |
| } else { |
| return TruthValue.NULL; |
| } |
| } |
| |
| TruthValue result; |
| Comparable baseObj = (Comparable) predicate.getLiteral(); |
| // Predicate object and stats objects are converted to the type of the predicate object. |
| Comparable predObj = getBaseObjectForComparison(predicate.getType(), baseObj); |
| |
| result = evaluatePredicateMinMax(predicate, predObj, range); |
| if (shouldEvaluateBloomFilter(predicate, result, bloomFilter)) { |
| return evaluatePredicateBloomFilter(predicate, predObj, bloomFilter, range.hasNulls, useUTCTimestamp); |
| } else { |
| return result; |
| } |
| } |
| |
| private static boolean shouldEvaluateBloomFilter(PredicateLeaf predicate, |
| TruthValue result, BloomFilter bloomFilter) { |
| // evaluate bloom filter only when |
| // 1) Bloom filter is available |
| // 2) Min/Max evaluation yield YES or MAYBE |
| // 3) Predicate is EQUALS or IN list |
| if (bloomFilter != null |
| && result != TruthValue.NO_NULL && result != TruthValue.NO |
| && (predicate.getOperator().equals(PredicateLeaf.Operator.EQUALS) |
| || predicate.getOperator().equals(PredicateLeaf.Operator.NULL_SAFE_EQUALS) |
| || predicate.getOperator().equals(PredicateLeaf.Operator.IN))) { |
| return true; |
| } |
| return false; |
| } |
| |
| private static TruthValue evaluatePredicateMinMax(PredicateLeaf predicate, |
| Comparable predObj, |
| ValueRange range) { |
| Location loc; |
| |
| switch (predicate.getOperator()) { |
| case NULL_SAFE_EQUALS: |
| loc = range.compare(predObj); |
| if (loc == Location.BEFORE || loc == Location.AFTER) { |
| return TruthValue.NO; |
| } else { |
| return TruthValue.YES_NO; |
| } |
| case EQUALS: |
| loc = range.compare(predObj); |
| if (range.isSingleton() && loc == Location.MIN) { |
| return range.addNull(TruthValue.YES); |
| } else if (loc == Location.BEFORE || loc == Location.AFTER) { |
| return range.addNull(TruthValue.NO); |
| } else { |
| return range.addNull(TruthValue.YES_NO); |
| } |
| case LESS_THAN: |
| loc = range.compare(predObj); |
| if (loc == Location.AFTER) { |
| return range.addNull(TruthValue.YES); |
| } else if (loc == Location.BEFORE || loc == Location.MIN) { |
| return range.addNull(TruthValue.NO); |
| } else { |
| return range.addNull(TruthValue.YES_NO); |
| } |
| case LESS_THAN_EQUALS: |
| loc = range.compare(predObj); |
| if (loc == Location.AFTER || loc == Location.MAX) { |
| return range.addNull(TruthValue.YES); |
| } else if (loc == Location.BEFORE) { |
| return range.addNull(TruthValue.NO); |
| } else { |
| return range.addNull(TruthValue.YES_NO); |
| } |
| case IN: |
| if (range.isSingleton()) { |
| // for a single value, look through to see if that value is in the |
| // set |
| for (Object arg : predicate.getLiteralList()) { |
| if (range.compare((Comparable) arg) == Location.MIN) { |
| return range.addNull(TruthValue.YES); |
| } |
| } |
| return range.addNull(TruthValue.NO); |
| } else { |
| // are all of the values outside of the range? |
| for (Object arg : predicate.getLiteralList()) { |
| predObj = getBaseObjectForComparison(predicate.getType(), (Comparable) arg); |
| loc = range.compare(predObj); |
| if (loc == Location.MIN || loc == Location.MIDDLE || |
| loc == Location.MAX) { |
| return range.addNull(TruthValue.YES_NO); |
| } |
| } |
| return range.addNull(TruthValue.NO); |
| } |
| case BETWEEN: |
| List<Object> args = predicate.getLiteralList(); |
| if (args == null || args.isEmpty()) { |
| return range.addNull(TruthValue.YES_NO); |
| } |
| Comparable predObj1 = getBaseObjectForComparison(predicate.getType(), |
| (Comparable) args.get(0)); |
| |
| loc = range.compare(predObj1); |
| if (loc == Location.BEFORE || loc == Location.MIN) { |
| Comparable predObj2 = getBaseObjectForComparison(predicate.getType(), |
| (Comparable) args.get(1)); |
| Location loc2 = range.compare(predObj2); |
| if (loc2 == Location.AFTER || loc2 == Location.MAX) { |
| return range.addNull(TruthValue.YES); |
| } else if (loc2 == Location.BEFORE) { |
| return range.addNull(TruthValue.NO); |
| } else { |
| return range.addNull(TruthValue.YES_NO); |
| } |
| } else if (loc == Location.AFTER) { |
| return range.addNull(TruthValue.NO); |
| } else { |
| return range.addNull(TruthValue.YES_NO); |
| } |
| case IS_NULL: |
| // min = null condition above handles the all-nulls YES case |
| return range.hasNulls ? TruthValue.YES_NO : TruthValue.NO; |
| default: |
| return range.addNull(TruthValue.YES_NO); |
| } |
| } |
| |
| private static TruthValue evaluatePredicateBloomFilter(PredicateLeaf predicate, |
| final Object predObj, BloomFilter bloomFilter, boolean hasNull, boolean useUTCTimestamp) { |
| switch (predicate.getOperator()) { |
| case NULL_SAFE_EQUALS: |
| // null safe equals does not return *_NULL variant. So set hasNull to false |
| return checkInBloomFilter(bloomFilter, predObj, false, useUTCTimestamp); |
| case EQUALS: |
| return checkInBloomFilter(bloomFilter, predObj, hasNull, useUTCTimestamp); |
| case IN: |
| for (Object arg : predicate.getLiteralList()) { |
| // if atleast one value in IN list exist in bloom filter, qualify the row group/stripe |
| Object predObjItem = getBaseObjectForComparison(predicate.getType(), (Comparable) arg); |
| TruthValue result = checkInBloomFilter(bloomFilter, predObjItem, hasNull, useUTCTimestamp); |
| if (result == TruthValue.YES_NO_NULL || result == TruthValue.YES_NO) { |
| return result; |
| } |
| } |
| return hasNull ? TruthValue.NO_NULL : TruthValue.NO; |
| default: |
| return hasNull ? TruthValue.YES_NO_NULL : TruthValue.YES_NO; |
| } |
| } |
| |
| private static TruthValue checkInBloomFilter(BloomFilter bf, |
| Object predObj, |
| boolean hasNull, |
| boolean useUTCTimestamp) { |
| TruthValue result = hasNull ? TruthValue.NO_NULL : TruthValue.NO; |
| |
| if (predObj instanceof Long) { |
| if (bf.testLong(((Long) predObj).longValue())) { |
| result = TruthValue.YES_NO_NULL; |
| } |
| } else if (predObj instanceof Double) { |
| if (bf.testDouble(((Double) predObj).doubleValue())) { |
| result = TruthValue.YES_NO_NULL; |
| } |
| } else if (predObj instanceof String || predObj instanceof Text || |
| predObj instanceof HiveDecimalWritable || |
| predObj instanceof BigDecimal) { |
| if (bf.testString(predObj.toString())) { |
| result = TruthValue.YES_NO_NULL; |
| } |
| } else if (predObj instanceof Timestamp) { |
| if (useUTCTimestamp) { |
| if (bf.testLong(((Timestamp) predObj).getTime())) { |
| result = TruthValue.YES_NO_NULL; |
| } |
| } else { |
| if (bf.testLong(SerializationUtils.convertToUtc(TimeZone.getDefault(), ((Timestamp) predObj).getTime()))) { |
| result = TruthValue.YES_NO_NULL; |
| } |
| } |
| } else if (predObj instanceof Date) { |
| if (bf.testLong(DateWritable.dateToDays((Date) predObj))) { |
| result = TruthValue.YES_NO_NULL; |
| } |
| } else { |
| // if the predicate object is null and if hasNull says there are no nulls then return NO |
| if (predObj == null && !hasNull) { |
| result = TruthValue.NO; |
| } else { |
| result = TruthValue.YES_NO_NULL; |
| } |
| } |
| |
| if (result == TruthValue.YES_NO_NULL && !hasNull) { |
| result = TruthValue.YES_NO; |
| } |
| |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("Bloom filter evaluation: " + result.toString()); |
| } |
| |
| return result; |
| } |
| |
| /** |
| * An exception for when we can't cast things appropriately |
| */ |
| static class SargCastException extends IllegalArgumentException { |
| |
| public SargCastException(String string) { |
| super(string); |
| } |
| } |
| |
| private static Comparable getBaseObjectForComparison(PredicateLeaf.Type type, |
| Comparable obj) { |
| if (obj == null) { |
| return null; |
| } |
| switch (type) { |
| case BOOLEAN: |
| if (obj instanceof Boolean) { |
| return obj; |
| } else { |
| // will only be true if the string conversion yields "true", all other values are |
| // considered false |
| return Boolean.valueOf(obj.toString()); |
| } |
| case DATE: |
| if (obj instanceof Date) { |
| return obj; |
| } else if (obj instanceof String) { |
| return Date.valueOf((String) obj); |
| } else if (obj instanceof Timestamp) { |
| return DateWritable.timeToDate(((Timestamp) obj).getTime() / 1000L); |
| } |
| // always string, but prevent the comparison to numbers (are they days/seconds/milliseconds?) |
| break; |
| case DECIMAL: |
| if (obj instanceof Boolean) { |
| return new HiveDecimalWritable(((Boolean) obj).booleanValue() ? |
| HiveDecimal.ONE : HiveDecimal.ZERO); |
| } else if (obj instanceof Integer) { |
| return new HiveDecimalWritable(((Integer) obj).intValue()); |
| } else if (obj instanceof Long) { |
| return new HiveDecimalWritable(((Long) obj)); |
| } else if (obj instanceof Float || obj instanceof Double || |
| obj instanceof String) { |
| return new HiveDecimalWritable(obj.toString()); |
| } else if (obj instanceof BigDecimal) { |
| return new HiveDecimalWritable(HiveDecimal.create((BigDecimal) obj)); |
| } else if (obj instanceof HiveDecimal) { |
| return new HiveDecimalWritable((HiveDecimal) obj); |
| } else if (obj instanceof HiveDecimalWritable) { |
| return obj; |
| } else if (obj instanceof Timestamp) { |
| return new HiveDecimalWritable(Double.toString( |
| TimestampUtils.getDouble((Timestamp) obj))); |
| } |
| break; |
| case FLOAT: |
| if (obj instanceof Number) { |
| // widening conversion |
| return ((Number) obj).doubleValue(); |
| } else if (obj instanceof HiveDecimal) { |
| return ((HiveDecimal) obj).doubleValue(); |
| } else if (obj instanceof String) { |
| return Double.valueOf(obj.toString()); |
| } else if (obj instanceof Timestamp) { |
| return TimestampUtils.getDouble((Timestamp) obj); |
| } else if (obj instanceof HiveDecimal) { |
| return ((HiveDecimal) obj).doubleValue(); |
| } else if (obj instanceof BigDecimal) { |
| return ((BigDecimal) obj).doubleValue(); |
| } |
| break; |
| case LONG: |
| if (obj instanceof Number) { |
| // widening conversion |
| return ((Number) obj).longValue(); |
| } else if (obj instanceof HiveDecimal) { |
| return ((HiveDecimal) obj).longValue(); |
| } else if (obj instanceof String) { |
| return Long.valueOf(obj.toString()); |
| } |
| break; |
| case STRING: |
| return (obj.toString()); |
| case TIMESTAMP: |
| if (obj instanceof Timestamp) { |
| return obj; |
| } else if (obj instanceof Integer) { |
| return new Timestamp(((Number) obj).longValue()); |
| } else if (obj instanceof Float) { |
| return TimestampUtils.doubleToTimestamp(((Float) obj).doubleValue()); |
| } else if (obj instanceof Double) { |
| return TimestampUtils.doubleToTimestamp(((Double) obj).doubleValue()); |
| } else if (obj instanceof HiveDecimal) { |
| return TimestampUtils.decimalToTimestamp((HiveDecimal) obj); |
| } else if (obj instanceof HiveDecimalWritable) { |
| return TimestampUtils.decimalToTimestamp(((HiveDecimalWritable) obj).getHiveDecimal()); |
| } else if (obj instanceof Date) { |
| return new Timestamp(((Date) obj).getTime()); |
| } |
| // float/double conversion to timestamp is interpreted as seconds whereas integer conversion |
| // to timestamp is interpreted as milliseconds by default. The integer to timestamp casting |
| // is also config driven. The filter operator changes its promotion based on config: |
| // "int.timestamp.conversion.in.seconds". Disable PPD for integer cases. |
| break; |
| default: |
| break; |
| } |
| |
| throw new SargCastException(String.format( |
| "ORC SARGS could not convert from %s to %s", obj.getClass() |
| .getSimpleName(), type)); |
| } |
| |
| public static class SargApplier { |
| public final static boolean[] READ_ALL_RGS = null; |
| public final static boolean[] READ_NO_RGS = new boolean[0]; |
| |
| private final OrcFile.WriterVersion writerVersion; |
| private final SearchArgument sarg; |
| private final List<PredicateLeaf> sargLeaves; |
| private final int[] filterColumns; |
| private final long rowIndexStride; |
| // same as the above array, but indices are set to true |
| private final boolean[] sargColumns; |
| private SchemaEvolution evolution; |
| private final long[] exceptionCount; |
| private final boolean useUTCTimestamp; |
| private final boolean writerUsedProlepticGregorian; |
| private final boolean convertToProlepticGregorian; |
| |
| public SargApplier(SearchArgument sarg, |
| long rowIndexStride, |
| SchemaEvolution evolution, |
| OrcFile.WriterVersion writerVersion, |
| boolean useUTCTimestamp, |
| boolean writerUsedProlepticGregorian, |
| boolean convertToProlepticGregorian) { |
| this.writerVersion = writerVersion; |
| this.sarg = sarg; |
| sargLeaves = sarg.getLeaves(); |
| this.writerUsedProlepticGregorian = writerUsedProlepticGregorian; |
| this.convertToProlepticGregorian = convertToProlepticGregorian; |
| filterColumns = mapSargColumnsToOrcInternalColIdx(sargLeaves, |
| evolution); |
| this.rowIndexStride = rowIndexStride; |
| // included will not be null, row options will fill the array with |
| // trues if null |
| sargColumns = new boolean[evolution.getFileIncluded().length]; |
| for (int i : filterColumns) { |
| // filter columns may have -1 as index which could be partition |
| // column in SARG. |
| if (i > 0) { |
| sargColumns[i] = true; |
| } |
| } |
| this.evolution = evolution; |
| exceptionCount = new long[sargLeaves.size()]; |
| this.useUTCTimestamp = useUTCTimestamp; |
| } |
| |
| /** |
| * Pick the row groups that we need to load from the current stripe. |
| * |
| * @return an array with a boolean for each row group or null if all of the |
| * row groups must be read. |
| * @throws IOException |
| */ |
| public boolean[] pickRowGroups(StripeInformation stripe, |
| OrcProto.RowIndex[] indexes, |
| OrcProto.Stream.Kind[] bloomFilterKinds, |
| List<OrcProto.ColumnEncoding> encodings, |
| OrcProto.BloomFilterIndex[] bloomFilterIndices, |
| boolean returnNone) throws IOException { |
| long rowsInStripe = stripe.getNumberOfRows(); |
| int groupsInStripe = (int) ((rowsInStripe + rowIndexStride - 1) / rowIndexStride); |
| boolean[] result = new boolean[groupsInStripe]; // TODO: avoid alloc? |
| TruthValue[] leafValues = new TruthValue[sargLeaves.size()]; |
| boolean hasSelected = false; |
| boolean hasSkipped = false; |
| TruthValue[] exceptionAnswer = new TruthValue[leafValues.length]; |
| for (int rowGroup = 0; rowGroup < result.length; ++rowGroup) { |
| for (int pred = 0; pred < leafValues.length; ++pred) { |
| int columnIx = filterColumns[pred]; |
| if (columnIx == -1) { |
| // the column is a virtual column |
| leafValues[pred] = TruthValue.YES_NO_NULL; |
| } else if (exceptionAnswer[pred] != null) { |
| leafValues[pred] = exceptionAnswer[pred]; |
| } else { |
| if (indexes[columnIx] == null) { |
| throw new AssertionError("Index is not populated for " + columnIx); |
| } |
| OrcProto.RowIndexEntry entry = indexes[columnIx].getEntry(rowGroup); |
| if (entry == null) { |
| throw new AssertionError("RG is not populated for " + columnIx + " rg " + rowGroup); |
| } |
| OrcProto.ColumnStatistics stats = entry.getStatistics(); |
| OrcProto.BloomFilter bf = null; |
| OrcProto.Stream.Kind bfk = null; |
| if (bloomFilterIndices != null && bloomFilterIndices[columnIx] != null) { |
| bfk = bloomFilterKinds[columnIx]; |
| bf = bloomFilterIndices[columnIx].getBloomFilter(rowGroup); |
| } |
| if (evolution != null && evolution.isPPDSafeConversion(columnIx)) { |
| PredicateLeaf predicate = sargLeaves.get(pred); |
| try { |
| leafValues[pred] = evaluatePredicateProto(stats, |
| predicate, bfk, encodings.get(columnIx), bf, |
| writerVersion, evolution.getFileSchema(). |
| findSubtype(columnIx), |
| useUTCTimestamp); |
| } catch (Exception e) { |
| exceptionCount[pred] += 1; |
| if (e instanceof SargCastException) { |
| LOG.info("Skipping ORC PPD - " + e.getMessage() + " on " |
| + predicate); |
| } else { |
| if (LOG.isWarnEnabled()) { |
| final String reason = e.getClass().getSimpleName() + " when evaluating predicate." + |
| " Skipping ORC PPD." + |
| " Stats: " + stats + |
| " Predicate: " + predicate; |
| LOG.warn(reason, e); |
| } |
| } |
| boolean hasNoNull = stats.hasHasNull() && !stats.getHasNull(); |
| if (predicate.getOperator().equals(PredicateLeaf.Operator.NULL_SAFE_EQUALS) |
| || hasNoNull) { |
| exceptionAnswer[pred] = TruthValue.YES_NO; |
| } else { |
| exceptionAnswer[pred] = TruthValue.YES_NO_NULL; |
| } |
| leafValues[pred] = exceptionAnswer[pred]; |
| } |
| } else { |
| leafValues[pred] = TruthValue.YES_NO_NULL; |
| } |
| if (LOG.isTraceEnabled()) { |
| LOG.trace("Stats = " + stats); |
| LOG.trace("Setting " + sargLeaves.get(pred) + " to " + leafValues[pred]); |
| } |
| } |
| } |
| result[rowGroup] = sarg.evaluate(leafValues).isNeeded(); |
| hasSelected = hasSelected || result[rowGroup]; |
| hasSkipped = hasSkipped || (!result[rowGroup]); |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("Row group " + (rowIndexStride * rowGroup) + " to " + |
| (rowIndexStride * (rowGroup + 1) - 1) + " is " + |
| (result[rowGroup] ? "" : "not ") + "included."); |
| } |
| } |
| |
| return hasSkipped ? ((hasSelected || !returnNone) ? result : READ_NO_RGS) : READ_ALL_RGS; |
| } |
| |
| /** |
| * Get the count of exceptions for testing. |
| * @return |
| */ |
| long[] getExceptionCount() { |
| return exceptionCount; |
| } |
| } |
| |
| /** |
| * Pick the row groups that we need to load from the current stripe. |
| * |
| * @return an array with a boolean for each row group or null if all of the |
| * row groups must be read. |
| * @throws IOException |
| */ |
| protected boolean[] pickRowGroups() throws IOException { |
| // if we don't have a sarg or indexes, we read everything |
| if (sargApp == null) { |
| return null; |
| } |
| readRowIndex(currentStripe, fileIncluded, sargApp.sargColumns); |
| return sargApp.pickRowGroups(stripes.get(currentStripe), |
| indexes.getRowGroupIndex(), |
| indexes.getBloomFilterKinds(), stripeFooter.getColumnsList(), |
| indexes.getBloomFilterIndex(), false); |
| } |
| |
| private void clearStreams() { |
| planner.clearStreams(); |
| } |
| |
| /** |
| * Read the current stripe into memory. |
| * |
| * @throws IOException |
| */ |
| private void readStripe() throws IOException { |
| StripeInformation stripe = beginReadStripe(); |
| planner.parseStripe(stripe, fileIncluded); |
| includedRowGroups = pickRowGroups(); |
| |
| // move forward to the first unskipped row |
| if (includedRowGroups != null) { |
| while (rowInStripe < rowCountInStripe && |
| !includedRowGroups[(int) (rowInStripe / rowIndexStride)]) { |
| rowInStripe = Math.min(rowCountInStripe, rowInStripe + rowIndexStride); |
| } |
| } |
| |
| // if we haven't skipped the whole stripe, read the data |
| if (rowInStripe < rowCountInStripe) { |
| planner.readData(indexes, includedRowGroups, false); |
| reader.startStripe(planner); |
| // if we skipped the first row group, move the pointers forward |
| if (rowInStripe != 0) { |
| seekToRowEntry(reader, (int) (rowInStripe / rowIndexStride)); |
| } |
| } |
| } |
| |
| private StripeInformation beginReadStripe() throws IOException { |
| StripeInformation stripe = stripes.get(currentStripe); |
| stripeFooter = readStripeFooter(stripe); |
| clearStreams(); |
| // setup the position in the stripe |
| rowCountInStripe = stripe.getNumberOfRows(); |
| rowInStripe = 0; |
| rowBaseInStripe = 0; |
| for (int i = 0; i < currentStripe; ++i) { |
| rowBaseInStripe += stripes.get(i).getNumberOfRows(); |
| } |
| // reset all of the indexes |
| OrcProto.RowIndex[] rowIndex = indexes.getRowGroupIndex(); |
| for (int i = 0; i < rowIndex.length; ++i) { |
| rowIndex[i] = null; |
| } |
| return stripe; |
| } |
| |
| /** |
| * Read the next stripe until we find a row that we don't skip. |
| * |
| * @throws IOException |
| */ |
| private void advanceStripe() throws IOException { |
| rowInStripe = rowCountInStripe; |
| while (rowInStripe >= rowCountInStripe && |
| currentStripe < stripes.size() - 1) { |
| currentStripe += 1; |
| readStripe(); |
| } |
| } |
| |
| /** |
| * Skip over rows that we aren't selecting, so that the next row is |
| * one that we will read. |
| * |
| * @param nextRow the row we want to go to |
| * @throws IOException |
| */ |
| private boolean advanceToNextRow( |
| BatchReader reader, long nextRow, boolean canAdvanceStripe) |
| throws IOException { |
| long nextRowInStripe = nextRow - rowBaseInStripe; |
| // check for row skipping |
| if (rowIndexStride != 0 && |
| includedRowGroups != null && |
| nextRowInStripe < rowCountInStripe) { |
| int rowGroup = (int) (nextRowInStripe / rowIndexStride); |
| if (!includedRowGroups[rowGroup]) { |
| while (rowGroup < includedRowGroups.length && !includedRowGroups[rowGroup]) { |
| rowGroup += 1; |
| } |
| if (rowGroup >= includedRowGroups.length) { |
| if (canAdvanceStripe) { |
| advanceStripe(); |
| } |
| return canAdvanceStripe; |
| } |
| nextRowInStripe = Math.min(rowCountInStripe, rowGroup * rowIndexStride); |
| } |
| } |
| if (nextRowInStripe >= rowCountInStripe) { |
| if (canAdvanceStripe) { |
| advanceStripe(); |
| } |
| return canAdvanceStripe; |
| } |
| if (nextRowInStripe != rowInStripe) { |
| if (rowIndexStride != 0) { |
| int rowGroup = (int) (nextRowInStripe / rowIndexStride); |
| seekToRowEntry(reader, rowGroup); |
| reader.skipRows(nextRowInStripe - rowGroup * rowIndexStride); |
| } else { |
| reader.skipRows(nextRowInStripe - rowInStripe); |
| } |
| rowInStripe = nextRowInStripe; |
| } |
| return true; |
| } |
| |
| @Override |
| public boolean nextBatch(VectorizedRowBatch batch) throws IOException { |
| try { |
| if (rowInStripe >= rowCountInStripe) { |
| currentStripe += 1; |
| if (currentStripe >= stripes.size()) { |
| batch.size = 0; |
| return false; |
| } |
| readStripe(); |
| } |
| |
| int batchSize = computeBatchSize(batch.getMaxSize()); |
| |
| rowInStripe += batchSize; |
| reader.setVectorColumnCount(batch.getDataColumnCount()); |
| reader.nextBatch(batch, batchSize); |
| advanceToNextRow(reader, rowInStripe + rowBaseInStripe, true); |
| return batch.size != 0; |
| } catch (IOException e) { |
| // Rethrow exception with file name in log message |
| throw new IOException("Error reading file: " + path, e); |
| } |
| } |
| |
| private int computeBatchSize(long targetBatchSize) { |
| final int batchSize; |
| // In case of PPD, batch size should be aware of row group boundaries. If only a subset of row |
| // groups are selected then marker position is set to the end of range (subset of row groups |
| // within strip). Batch size computed out of marker position makes sure that batch size is |
| // aware of row group boundary and will not cause overflow when reading rows |
| // illustration of this case is here https://issues.apache.org/jira/browse/HIVE-6287 |
| if (rowIndexStride != 0 && includedRowGroups != null && rowInStripe < rowCountInStripe) { |
| int startRowGroup = (int) (rowInStripe / rowIndexStride); |
| if (!includedRowGroups[startRowGroup]) { |
| while (startRowGroup < includedRowGroups.length && !includedRowGroups[startRowGroup]) { |
| startRowGroup += 1; |
| } |
| } |
| |
| int endRowGroup = startRowGroup; |
| while (endRowGroup < includedRowGroups.length && includedRowGroups[endRowGroup]) { |
| endRowGroup += 1; |
| } |
| |
| final long markerPosition = |
| (endRowGroup * rowIndexStride) < rowCountInStripe ? (endRowGroup * rowIndexStride) |
| : rowCountInStripe; |
| batchSize = (int) Math.min(targetBatchSize, (markerPosition - rowInStripe)); |
| |
| if (isLogDebugEnabled && batchSize < targetBatchSize) { |
| LOG.debug("markerPosition: " + markerPosition + " batchSize: " + batchSize); |
| } |
| } else { |
| batchSize = (int) Math.min(targetBatchSize, (rowCountInStripe - rowInStripe)); |
| } |
| return batchSize; |
| } |
| |
| @Override |
| public void close() throws IOException { |
| clearStreams(); |
| dataReader.close(); |
| } |
| |
| @Override |
| public long getRowNumber() { |
| return rowInStripe + rowBaseInStripe + firstRow; |
| } |
| |
| /** |
| * Return the fraction of rows that have been read from the selected. |
| * section of the file |
| * |
| * @return fraction between 0.0 and 1.0 of rows consumed |
| */ |
| @Override |
| public float getProgress() { |
| return ((float) rowBaseInStripe + rowInStripe) / totalRowCount; |
| } |
| |
| private int findStripe(long rowNumber) { |
| for (int i = 0; i < stripes.size(); i++) { |
| StripeInformation stripe = stripes.get(i); |
| if (stripe.getNumberOfRows() > rowNumber) { |
| return i; |
| } |
| rowNumber -= stripe.getNumberOfRows(); |
| } |
| throw new IllegalArgumentException("Seek after the end of reader range"); |
| } |
| |
| public OrcIndex readRowIndex(int stripeIndex, boolean[] included, |
| boolean[] sargColumns) throws IOException { |
| // if this is the current stripe, use the cached objects. |
| if (stripeIndex == currentStripe) { |
| return planner.readRowIndex( |
| sargColumns != null || sargApp == null |
| ? sargColumns : sargApp.sargColumns, indexes); |
| } else { |
| StripePlanner copy = new StripePlanner(planner); |
| if (included == null) { |
| included = new boolean[schema.getMaximumId() + 1]; |
| Arrays.fill(included, true); |
| } |
| copy.parseStripe(stripes.get(stripeIndex), included); |
| return copy.readRowIndex(sargColumns, null); |
| } |
| } |
| |
| private void seekToRowEntry(BatchReader reader, int rowEntry) |
| throws IOException { |
| OrcProto.RowIndex[] rowIndices = indexes.getRowGroupIndex(); |
| PositionProvider[] index = new PositionProvider[rowIndices.length]; |
| for (int i = 0; i < index.length; ++i) { |
| if (rowIndices[i] != null) { |
| OrcProto.RowIndexEntry entry = rowIndices[i].getEntry(rowEntry); |
| // This is effectively a test for pre-ORC-569 files. |
| if (rowEntry == 0 && entry.getPositionsCount() == 0) { |
| index[i] = new ZeroPositionProvider(); |
| } else { |
| index[i] = new PositionProviderImpl(entry); |
| } |
| } |
| } |
| reader.seek(index); |
| } |
| |
| @Override |
| public void seekToRow(long rowNumber) throws IOException { |
| if (rowNumber < 0) { |
| throw new IllegalArgumentException("Seek to a negative row number " + |
| rowNumber); |
| } else if (rowNumber < firstRow) { |
| throw new IllegalArgumentException("Seek before reader range " + |
| rowNumber); |
| } |
| // convert to our internal form (rows from the beginning of slice) |
| rowNumber -= firstRow; |
| |
| // move to the right stripe |
| int rightStripe = findStripe(rowNumber); |
| if (rightStripe != currentStripe) { |
| currentStripe = rightStripe; |
| readStripe(); |
| } |
| readRowIndex(currentStripe, fileIncluded, sargApp == null ? null : sargApp.sargColumns); |
| |
| // if we aren't to the right row yet, advance in the stripe. |
| advanceToNextRow(reader, rowNumber, true); |
| } |
| |
| private static final String TRANSLATED_SARG_SEPARATOR = "_"; |
| public static String encodeTranslatedSargColumn(int rootColumn, Integer indexInSourceTable) { |
| return rootColumn + TRANSLATED_SARG_SEPARATOR |
| + ((indexInSourceTable == null) ? -1 : indexInSourceTable); |
| } |
| |
| public static int[] mapTranslatedSargColumns( |
| List<OrcProto.Type> types, List<PredicateLeaf> sargLeaves) { |
| int[] result = new int[sargLeaves.size()]; |
| OrcProto.Type lastRoot = null; // Root will be the same for everyone as of now. |
| String lastRootStr = null; |
| for (int i = 0; i < result.length; ++i) { |
| String[] rootAndIndex = sargLeaves.get(i).getColumnName().split(TRANSLATED_SARG_SEPARATOR); |
| assert rootAndIndex.length == 2; |
| String rootStr = rootAndIndex[0], indexStr = rootAndIndex[1]; |
| int index = Integer.parseInt(indexStr); |
| // First, check if the column even maps to anything. |
| if (index == -1) { |
| result[i] = -1; |
| continue; |
| } |
| assert index >= 0; |
| // Then, find the root type if needed. |
| if (!rootStr.equals(lastRootStr)) { |
| lastRoot = types.get(Integer.parseInt(rootStr)); |
| lastRootStr = rootStr; |
| } |
| // Subtypes of the root types correspond, in order, to the columns in the table schema |
| // (disregarding schema evolution that doesn't presently work). Get the index for the |
| // corresponding subtype. |
| result[i] = lastRoot.getSubtypes(index); |
| } |
| return result; |
| } |
| |
| public CompressionCodec getCompressionCodec() { |
| return dataReader.getCompressionOptions().getCodec(); |
| } |
| |
| public int getMaxDiskRangeChunkLimit() { |
| return maxDiskRangeChunkLimit; |
| } |
| } |
