static final Logger LOG = LoggerFactory.getLogger(RecordReaderImpl.class);

private static final boolean isLogDebugEnabled = LOG.isDebugEnabled();

// as public for use with test cases

public static final OrcProto.ColumnStatistics EMPTY_COLUMN_STATISTICS =

OrcProto.ColumnStatistics.newBuilder().setNumberOfValues(0)

.setHasNull(false)

.setBytesOnDisk(0)

.build();

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;

// position of the follow reader within the stripe

private long followRowInStripe = 0;

private int currentStripe = -1;

private long rowBaseInStripe = 0;

private long rowCountInStripe = 0;

private final BatchReader reader;

private final OrcIndex indexes;

// identifies the columns requiring row indexes

private final boolean[] rowIndexColsToRead;

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;

// identifies the type of read, ALL(read everything), LEADERS(read only the filter columns)

private final TypeReader.ReadPhase startReadPhase;

// identifies that follow columns bytes must be read

private boolean needsFollowColumnsRead;

private final boolean noSelectedVector;

// identifies whether the file has bad bloom filters that we should not use.

private final boolean skipBloomFilters;

static final String[] BAD_CPP_BLOOM_FILTER_VERSIONS = {

"1.6.0", "1.6.1", "1.6.2", "1.6.3", "1.6.4", "1.6.5", "1.6.6", "1.6.7", "1.6.8",

"1.6.9", "1.6.10", "1.6.11", "1.7.0"};

/**

static int findColumns(SchemaEvolution evolution,

String columnName) {

TypeDescription fileColumn = findColumnType(evolution, columnName);

return fileColumn == null ? -1 : fileColumn.getId();

}

static TypeDescription findColumnType(SchemaEvolution evolution, String columnName) {

try {

TypeDescription readerColumn = evolution.getReaderBaseSchema().findSubtype(

columnName, evolution.isSchemaEvolutionCaseAware);

return evolution.getFileType(readerColumn);

} catch (IllegalArgumentException e) {

throw new IllegalArgumentException("Filter could not find column with name: " +

columnName + " on " + evolution.getReaderBaseSchema(),

e);

}

}

/**

static TypeDescription findMostCommonColumn(SchemaEvolution evolution, String columnName) {

try {

TypeDescription readerColumn = evolution.getReaderBaseSchema().findSubtype(

columnName, evolution.isSchemaEvolutionCaseAware);

TypeDescription fileColumn;

do {

fileColumn = evolution.getFileType(readerColumn);

if (fileColumn == null) {

readerColumn = readerColumn.getParent();

} else {

return fileColumn;

}

} while (readerColumn != null);

return null;

} catch (IllegalArgumentException e) {

throw new IllegalArgumentException("Filter could not find column with name: " +

columnName + " on " + evolution.getReaderBaseSchema(),

e);

}

}

/**

public static int[] mapSargColumnsToOrcInternalColIdx(

List<PredicateLeaf> sargLeaves,

SchemaEvolution evolution) {

int[] result = new int[sargLeaves.size()];

for (int i = 0; i < sargLeaves.size(); ++i) {

int colNum = -1;

try {

String colName = sargLeaves.get(i).getColumnName();

colNum = findColumns(evolution, colName);

} catch (IllegalArgumentException e) {

LOG.debug("{}", e.getMessage());

}

result[i] = colNum;

}

return result;

}

protected RecordReaderImpl(ReaderImpl fileReader,

Reader.Options options) throws IOException {

OrcFile.WriterVersion writerVersion = fileReader.getWriterVersion();

SchemaEvolution evolution;

if (options.getSchema() == null) {

LOG.debug("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.noSelectedVector = !options.useSelected();

LOG.debug("noSelectedVector={}", this.noSelectedVector);

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();

boolean[] rowIndexCols = new boolean[evolution.getFileIncluded().length];

if (sarg != null && rowIndexStride > 0) {

sargApp = new SargApplier(sarg,

rowIndexStride,

evolution,

writerVersion,

fileReader.useUTCTimestamp,

fileReader.writerUsedProlepticGregorian(),

fileReader.options.getConvertToProlepticGregorian());

sargApp.setRowIndexCols(rowIndexCols);

} 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)

.withMinSeekSize(options.minSeekSize())

.withMinSeekSizeTolerance(options.minSeekSizeTolerance());

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);

}

String[] filterCols = null;

Consumer<OrcFilterContext> filterCallBack = null;

String filePath = options.allowPluginFilters() ?

fileReader.getFileSystem().makeQualified(fileReader.path).toString() : null;

BatchFilter filter = FilterFactory.createBatchFilter(options,

evolution.getReaderBaseSchema(),

evolution.isSchemaEvolutionCaseAware(),

fileReader.getFileVersion(),

false,

filePath,

fileReader.conf);

if (filter != null) {

// If a filter is determined then use this

filterCallBack = filter;

filterCols = filter.getColumnNames();

}

// Map columnNames to ColumnIds

SortedSet<Integer> filterColIds = new TreeSet<>();

if (filterCols != null) {

for (String colName : filterCols) {

TypeDescription expandCol = findColumnType(evolution, colName);

// If the column is not present in the file then this can be ignored from read.

if (expandCol == null || expandCol.getId() == -1) {

// Add -1 to filter columns so that the NullTreeReader is invoked during the LEADERS phase

filterColIds.add(-1);

// Determine the common parent and include these

expandCol = findMostCommonColumn(evolution, colName);

}

while (expandCol != null && expandCol.getId() != -1) {

// classify the column and the parent branch as LEAD

filterColIds.add(expandCol.getId());

rowIndexCols[expandCol.getId()] = true;

expandCol = expandCol.getParent();

}

}

this.startReadPhase = TypeReader.ReadPhase.LEADERS;

LOG.debug("Using startReadPhase: {} with filter columns: {}", startReadPhase, filterColIds);

} else {

this.startReadPhase = TypeReader.ReadPhase.ALL;

}

var hasTrue = false;

for (boolean value: rowIndexCols) {

if (value) {

hasTrue = true;

break;

}

}

this.rowIndexColsToRead = hasTrue ? rowIndexCols : null;

TreeReaderFactory.ReaderContext readerContext =

new TreeReaderFactory.ReaderContext()

.setSchemaEvolution(evolution)

.setFilterCallback(filterColIds, filterCallBack)

.skipCorrupt(skipCorrupt)

.fileFormat(fileReader.getFileVersion())

.useUTCTimestamp(fileReader.useUTCTimestamp)

.setProlepticGregorian(fileReader.writerUsedProlepticGregorian(),

fileReader.options.getConvertToProlepticGregorian())

.setEncryption(encryption);

reader = TreeReaderFactory.createRootReader(evolution.getReaderSchema(), readerContext);

skipBloomFilters = hasBadBloomFilters(fileReader.getFileTail().getFooter());

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, filterColIds);

try {

advanceToNextRow(reader, 0L, true);

} catch (Exception e) {

// Try to close since this happens in constructor.

close();

long stripeId = stripes.size() == 0 ? 0 : stripes.get(0).getStripeId();

throw new IOException(String.format("Problem opening stripe %d footer in %s.",

stripeId, path), e);

}

}

/**

private boolean hasBadBloomFilters(OrcProto.Footer footer) {

// Only C++ writer in old releases could have bad bloom filters.

if (footer.getWriter() != 1) return false;

// 'softwareVersion' is added in 1.5.13, 1.6.11, and 1.7.0.

// 1.6.x releases before 1.6.11 won't have it. On the other side, the C++ writer

// supports writing bloom filters since 1.6.0. So files written by the C++ writer

// and with 'softwareVersion' unset would have bad bloom filters.

if (!footer.hasSoftwareVersion()) return true;

String fullVersion = footer.getSoftwareVersion();

String version = fullVersion;

// Deal with snapshot versions, e.g. 1.6.12-SNAPSHOT.

if (fullVersion.contains("-")) {

version = fullVersion.substring(0, fullVersion.indexOf('-'));

}

for (String v : BAD_CPP_BLOOM_FILTER_VERSIONS) {

if (v.equals(version)) {

return true;

}

}

return false;

}

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;

final boolean hasValue;

final boolean comparable;

ValueRange(PredicateLeaf predicate,

T lower, T upper,

boolean hasNulls,

boolean onlyLowerBound,

boolean onlyUpperBound,

boolean hasValue,

boolean comparable) {

PredicateLeaf.Type type = predicate.getType();

this.lower = getBaseObjectForComparison(type, lower);

this.upper = getBaseObjectForComparison(type, upper);

this.hasNulls = hasNulls;

this.onlyLowerBound = onlyLowerBound;

this.onlyUpperBound = onlyUpperBound;

this.hasValue = hasValue;

this.comparable = comparable;

}

ValueRange(PredicateLeaf predicate,

T lower, T upper,

boolean hasNulls,

boolean onlyLowerBound,

boolean onlyUpperBound) {

this(predicate, lower, upper, hasNulls, onlyLowerBound, onlyUpperBound,

lower != null, lower != null);

}

ValueRange(PredicateLeaf predicate, T lower, T upper,

boolean hasNulls) {

this(predicate, lower, upper, hasNulls, false, false);

}

/**

ValueRange(PredicateLeaf predicate, boolean hasNulls) {

this(predicate, null, null, hasNulls, false, false);

}

boolean hasValues() {

return hasValue;

}

/**

boolean isComparable() {

return hasValue && comparable;

}

/**

boolean isValid() {

return hasValue || hasNulls;

}

/**

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;

}

/**

boolean isSingleton() {

return lower != null && !onlyLowerBound && !onlyUpperBound &&

lower.equals(upper);

}

/**

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;

}

}

}

/**

static ValueRange getValueRange(ColumnStatistics index,

PredicateLeaf predicate,

boolean useUTCTimestamp) {

if (index.getNumberOfValues() == 0) {

return new ValueRange<>(predicate, index.hasNull());

} else if (index instanceof IntegerColumnStatistics stats) {

Long min = stats.getMinimum();

Long max = stats.getMaximum();

return new ValueRange<>(predicate, min, max, stats.hasNull());

} else if (index instanceof CollectionColumnStatistics stats) {

Long min = stats.getMinimumChildren();

Long max = stats.getMaximumChildren();

return new ValueRange<>(predicate, min, max, stats.hasNull());

}else if (index instanceof DoubleColumnStatistics stats) {

Double min = stats.getMinimum();

Double max = stats.getMaximum();

return new ValueRange<>(predicate, min, max, stats.hasNull());

} else if (index instanceof StringColumnStatistics stats) {

return new ValueRange<>(predicate, stats.getLowerBound(),

stats.getUpperBound(), stats.hasNull(), stats.getMinimum() == null,

stats.getMaximum() == null);

} else if (index instanceof DateColumnStatistics stats) {

ChronoLocalDate min = stats.getMinimumLocalDate();

ChronoLocalDate max = stats.getMaximumLocalDate();

return new ValueRange<>(predicate, min, max, stats.hasNull());

} else if (index instanceof DecimalColumnStatistics stats) {

HiveDecimal min = stats.getMinimum();

HiveDecimal max = stats.getMaximum();

return new ValueRange<>(predicate, min, max, stats.hasNull());

} else if (index instanceof TimestampColumnStatistics stats) {

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 stats) {

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, index.hasNull(), false, false, true, false);

}

}

/**

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, true, false);

}

/**

static TruthValue evaluatePredicateProto(OrcProto.ColumnStatistics statsProto,

PredicateLeaf predicate,

OrcProto.Stream.Kind kind,

OrcProto.ColumnEncoding encoding,

OrcProto.BloomFilter bloomFilter,

OrcFile.WriterVersion writerVersion,

TypeDescription type,

boolean writerUsedProlepticGregorian,

boolean useUTCTimestamp) {

// When statsProto is EMPTY_COLUMN_STATISTICS

// this column does not actually provide statistics

// we cannot make any assumptions, so we return YES_NO_NULL.

if (statsProto == EMPTY_COLUMN_STATISTICS) {

return TruthValue.YES_NO_NULL;

}

ColumnStatistics cs = ColumnStatisticsImpl.deserialize(

null, statsProto, writerUsedProlepticGregorian, true);

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;

} else if ((category == TypeDescription.Category.DOUBLE ||

category == TypeDescription.Category.FLOAT) && cs instanceof DoubleColumnStatistics dstas) {

if (Double.isNaN(dstas.getSum())) {

LOG.debug("Not using predication pushdown on {} because stats contain NaN values",

predicate.getColumnName());

return dstas.hasNull() ? TruthValue.YES_NO_NULL : TruthValue.YES_NO;

}

}

return evaluatePredicateRange(predicate, range,

BloomFilterIO.deserialize(kind, encoding, writerVersion, type.getCategory(),

bloomFilter), useUTCTimestamp);

}

/**

public static TruthValue evaluatePredicate(ColumnStatistics stats,

PredicateLeaf predicate,

BloomFilter bloomFilter) {

return evaluatePredicate(stats, predicate, bloomFilter, false);

}

/**

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 range is invalid, that means that no value (including null) is written to this column

// we should return TruthValue.NO for any predicate.

if (!range.isValid()) {

return TruthValue.NO;

}

// 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 if (predicate.getOperator() == PredicateLeaf.Operator.NULL_SAFE_EQUALS) {

Object literal = predicate.getLiteral();

if (literal == null) {

return TruthValue.YES;

} else {

return TruthValue.NO;

}

} else {

return TruthValue.NULL;

}

} else if (!range.isComparable()) {

return range.hasNulls ? TruthValue.YES_NO_NULL : TruthValue.YES_NO;

}

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

return 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));

}

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 ||

(loc == Location.MIN && range.isSingleton())) {

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()) {

predObj = getBaseObjectForComparison(predicate.getType(), (Comparable) arg);

if (range.compare(predObj) == 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)) {

result = TruthValue.YES_NO_NULL;

}

} else if (predObj instanceof Double) {

if (bf.testDouble((Double) predObj)) {

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 ChronoLocalDate) {

if (bf.testLong(((ChronoLocalDate) predObj).toEpochDay())) {

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;

}

LOG.debug("Bloom filter evaluation: {}", result);

return result;

}

/**

static class SargCastException extends IllegalArgumentException {

SargCastException(String string) {

super(string);

}

}

private static Comparable getBaseObjectForComparison(PredicateLeaf.Type type,

Comparable obj) {

if (obj == null) {

return null;

}

switch (type) {