jffwang
/
hetu-core
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

[I2E4X7] fixed nulls handling in predicate pushdown flow

This commit is contained in:
nitin.kashyap 2021-01-20 15:56:39 +05:30
parent 2ba9bdd21b
commit 446621c51a
10 changed files with 125 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

76
presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveIntegrationSmokeTest.java
View File

@ -66,6 +66,7 @@ import java.io.FilenameFilter;
import java.math.BigDecimal;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
@ -5109,6 +5110,81 @@ public class TestHiveIntegrationSmokeTest
}
}
@Test
public void testPushdownWithNullRows()
{
Session session = getSession();
Session session1 = Session.builder(session)
.setCatalogSessionProperty(session.getCatalog().get(), "orc_predicate_pushdown_enabled", "true")
.build();
Session session2 = Session.builder(session)
.setCatalogSessionProperty(session.getCatalog().get(), "orc_predicate_pushdown_enabled", "true")
.setCatalogSessionProperty(session.getCatalog().get(), "orc_disjunct_predicate_pushdown_enabled", "false")
.build();
String[] types = {"double", "decimal(7,2)", "decimal(38,7)", "integer", "bigint", "string", "boolean"};
for (String type : types) {
testPushdownNullForType(session1, session2, type);
}
}
private void testPushdownNullForType(Session sessionWithOr, Session sessionWithoutOR, String type)
{
try {
assertUpdate(sessionWithOr, "CREATE TABLE test_predicate_or_NULL (a " + type + ", b " + type + ", c int) with (transactional=true, format='orc')");
assertUpdate(sessionWithOr, "INSERT INTO test_predicate_or_NULL VALUES " +
"(cast(0 as " + type + "), cast(0 as " + type + "),0)," +
"(cast(1 as " + type + "), NULL, 1)," +
"(NULL,cast(2 as " + type + "), 2)," +
"(NULL,NULL,3)," +
"(cast(4 as " + type + "), cast(4 as " + type + "),4)", 5);
List<String> queries = new ArrayList<>();
queries.add("SELECT * FROM test_predicate_or_NULL WHERE " +
"c BETWEEN 0 AND 5 AND (a BETWEEN cast(0 as " + type + ") AND cast(5 as " + type + ") or b BETWEEN cast(0 as " + type + ") AND cast(5 as " + type + ")) " +
"ORDER BY a,b,c");
queries.add("SELECT * FROM test_predicate_or_NULL WHERE " +
"c BETWEEN 0 AND 5 " +
"AND (" +
"a BETWEEN cast(0 as " + type + ") and cast(5 as " + type + ") " +
"OR b BETWEEN cast(0 as " + type + ") and cast(5 as " + type + ") " +
"OR a IS NULL) " +
"ORDER BY a,b,c");
queries.add("SELECT * FROM test_predicate_or_NULL WHERE " +
"c BETWEEN 0 AND 5 " +
"AND (" +
"a BETWEEN cast(0 as " + type + ") and cast(5 as " + type + ") " +
"OR b BETWEEN cast(0 as " + type + ") and cast(5 as " + type + ") " +
"OR a IS NULL " +
"OR b IS NULL" +
") ORDER BY a,b,c");
queries.add("SELECT * FROM test_predicate_or_NULL WHERE " +
"c BETWEEN 0 AND 5 " +
"AND (" +
"a BETWEEN cast(0 as " + type + ") and cast(5 as " + type + ") " +
"OR b BETWEEN cast(0 as " + type + ") and cast(1 as " + type + ") " +
"OR a IS NOT NULL " +
"OR a BETWEEN cast(3 as " + type + ") and cast(5 as " + type + ") " +
") ORDER BY a,b,c");
MaterializedResult expected;
MaterializedResult resultPushdownOr;
MaterializedResult resultPushdown;
for (String query : queries) {
expected = computeActual(query);
resultPushdownOr = computeActual(sessionWithOr, query);
resultPushdown = computeActual(sessionWithoutOR, query);
assertEquals(expected.getMaterializedRows(), resultPushdown.getMaterializedRows());
assertEquals(expected.getMaterializedRows(), resultPushdownOr.getMaterializedRows());
System.out.println("Type(" + type + ")\n-------------\n" + resultPushdown.getMaterializedRows());
}
}
finally {
assertUpdate("DROP TABLE IF EXISTS test_predicate_or_NULL");
}
}
@Test
public void testUpdateAndDeleteForBooleanColumn()
{

6
presto-orc/src/main/java/io/prestosql/orc/reader/BooleanSelectiveColumnReader.java
View File

@ -256,9 +256,10 @@ public class BooleanSelectiveColumnReader
}
int streamPosition = 0;
boolean checkNulls = filters != null && filters.stream().anyMatch(f -> f.testNull());
if (dataStream == null && presentStream != null) {
streamPosition = readAllNulls(positions, positionCount);
if (filters != null && filters.get(0).testNull() && accumulator != null) {
if (checkNulls && accumulator != null) {
accumulator.set(positions[0], streamPosition);
}
}
@ -279,7 +280,8 @@ public class BooleanSelectiveColumnReader
if (outputRequired) {
nulls[outputPositionCount] = true;
}
if (accumulator != null) {
if (accumulator != null && checkNulls) {
accumulator.set(position);
}

23
presto-orc/src/main/java/io/prestosql/orc/reader/DoubleSelectiveColumnReader.java
View File

@ -165,6 +165,15 @@ public class DoubleSelectiveColumnReader
outputPositionCount = 0;
ensureValuesCapacity(outputRequired, positionCount, nullsAllowed && presentStream != null, positions);
if (filters != null) {
if (outputPositions == null || outputPositions.length < positionCount) {
outputPositions = new int[positionCount];
}
}
else {
outputPositions = positions;
}
// account memory used by values, nulls and outputPositions
systemMemoryContext.setBytes(getRetainedSizeInBytes());
@ -280,6 +289,7 @@ public class DoubleSelectiveColumnReader
{
allNulls = false;
int streamPosition = 0;
boolean checkNulls = filters != null && filters.stream().anyMatch(f -> f.testNull());
for (int i = 0; i < positionCount; i++) {
int position = positions[i];
@ -295,7 +305,7 @@ public class DoubleSelectiveColumnReader
if (outputRequired) {
nulls[outputPositionCount] = true;
}
if (accumulator != null) {
if (accumulator != null && checkNulls) {
accumulator.set(position);
}
outputPositions[outputPositionCount] = position;
@ -305,7 +315,7 @@ public class DoubleSelectiveColumnReader
else {
double value = dataStream.next();
if ((accumulator != null && accumulator.get(position))
|| filters == null || filters.get(0).testDouble(value)) {
|| filters == null || filters.stream().anyMatch(f -> f.testDouble(value))) {
if (accumulator != null) {
accumulator.set(position);
}
@ -355,15 +365,6 @@ public class DoubleSelectiveColumnReader
}
}
}
if (filter != null) {
if (outputPositions == null || outputPositions.length < capacity) {
outputPositions = new int[capacity];
}
}
else {
outputPositions = positions;
}
}
@Override

23
presto-orc/src/main/java/io/prestosql/orc/reader/FloatSelectiveColumnReader.java
View File

@ -142,6 +142,15 @@ public class FloatSelectiveColumnReader
ensureValuesCapacity(outputRequired, positionCount, nullsAllowed && presentStream != null, positions);
if (filters != null) {
if (outputPositions == null || outputPositions.length < positionCount) {
outputPositions = new int[positionCount];
}
}
else {
outputPositions = positions;
}
// account memory used by values, nulls and outputPositions
systemMemoryContext.setBytes(getRetainedSizeInBytes());
@ -183,15 +192,6 @@ public class FloatSelectiveColumnReader
}
}
}
if (filter != null) {
if (outputPositions == null || outputPositions.length < capacity) {
outputPositions = new int[capacity];
}
}
else {
outputPositions = positions;
}
}
private int readWithFilter(int[] positions, int positionCount, List<TupleDomainFilter> filters)
@ -237,6 +237,7 @@ public class FloatSelectiveColumnReader
throws IOException
{
int streamPosition = 0;
boolean checkNulls = filters != null && filters.stream().anyMatch(f -> f.testNull());
for (int i = 0; i < positionCount; i++) {
int position = positions[i];
if (position > streamPosition) {
@ -249,7 +250,7 @@ public class FloatSelectiveColumnReader
if (outputRequired) {
nulls[outputPositionCount] = true;
}
if (accumulator != null) {
if (accumulator != null && checkNulls) {
accumulator.set(position);
}
outputPositions[outputPositionCount] = position;
@ -259,7 +260,7 @@ public class FloatSelectiveColumnReader
else {
float value = dataStream.next();
if ((accumulator != null && accumulator.get(position))
|| filters == null || filters.get(0).testFloat(value)) {
|| filters == null || filters.stream().anyMatch(f -> f.testFloat(value))) {
if (accumulator != null) {
accumulator.set(position);
}

5
presto-orc/src/main/java/io/prestosql/orc/reader/LongDecimalSelectiveColumnReader.java
View File

@ -134,6 +134,7 @@ public class LongDecimalSelectiveColumnReader
outputPositionCount = 0;
long[] data = new long[2];
Slice rescaledDecimal = UnscaledDecimal128Arithmetic.unscaledDecimal();
boolean checkNulls = filters != null && filters.stream().anyMatch(f -> f.testNull());
for (int i = 0; i < positionCount; i++) {
int position = positions[i];
if (position > streamPosition) {
@ -147,7 +148,7 @@ public class LongDecimalSelectiveColumnReader
nulls[outputPositionCount] = true;
}
if (accumulator != null) {
if (accumulator != null && checkNulls) {
accumulator.set(position);
}
@ -163,7 +164,7 @@ public class LongDecimalSelectiveColumnReader
long low = UnsafeSlice.getLongUnchecked(rescaledDecimal, 0);
long high = UnsafeSlice.getLongUnchecked(rescaledDecimal, Long.BYTES);
if ((accumulator != null && accumulator.get(position))
|| filters == null || filters.get(0).testDecimal(low, high)) {
|| filters == null || filters.stream().anyMatch(f -> f.testDecimal(low, high))) {
if (accumulator != null) {
accumulator.set(position);
}

7
presto-orc/src/main/java/io/prestosql/orc/reader/LongSelectiveColumnReader.java
View File

@ -201,9 +201,10 @@ public class LongSelectiveColumnReader
outputPositionCount = 0;
int streamPosition = 0;
boolean checkNulls = filters != null && filters.stream().anyMatch(f -> f.testNull());
if (dataStream == null && presentStream != null) {
streamPosition = readAllNulls(positions, positionCount);
if (filters != null && filters.get(0).testNull() && accumulator != null) {
if (checkNulls && accumulator != null) {
accumulator.set(positions[0], streamPosition);
}
}
@ -224,7 +225,7 @@ public class LongSelectiveColumnReader
if (filters != null) {
outputPositions[outputPositionCount] = position;
}
if (accumulator != null) {
if (accumulator != null && checkNulls) {
accumulator.set(position);
}
outputPositionCount++;
@ -233,7 +234,7 @@ public class LongSelectiveColumnReader
else {
long value = dataStream.next();
if ((accumulator != null && accumulator.get(position))
|| filters == null || filters.size() <= 0 || filters.get(0).testLong(value)) {
|| filters == null || filters.size() <= 0 || filters.stream().anyMatch(f -> f.testLong(value))) {
if (accumulator != null) {
accumulator.set(position);
}

5
presto-orc/src/main/java/io/prestosql/orc/reader/ShortDecimalSelectiveColumnReader.java
View File

@ -119,6 +119,7 @@ public class ShortDecimalSelectiveColumnReader
int streamPosition = 0;
outputPositionCount = 0;
long[] data = new long[1];
boolean checkNulls = filters != null && filters.stream().anyMatch(f -> f.testNull());
for (int i = 0; i < positionCount; i++) {
int position = positions[i];
if (position > streamPosition) {
@ -131,7 +132,7 @@ public class ShortDecimalSelectiveColumnReader
if (outputRequired) {
nulls[outputPositionCount] = true;
}
if (accumulator != null) {
if (accumulator != null && checkNulls) {
accumulator.set(position);
}
outputPositions[outputPositionCount] = position;
@ -142,7 +143,7 @@ public class ShortDecimalSelectiveColumnReader
dataStream.nextShortDecimal(data, 1);
long rescale = Decimals.rescale(data[0], (int) scaleStream.next(), this.scale);
if ((accumulator != null && accumulator.get(position))
|| filters == null || filters.get(0).testLong(rescale)) {
|| filters == null || filters.stream().anyMatch(f -> f.testLong(rescale))) {
if (accumulator != null) {
accumulator.set(position);
}

5
presto-orc/src/main/java/io/prestosql/orc/reader/SliceDictionarySelectiveColumnReader.java
View File

@ -252,6 +252,7 @@ public class SliceDictionarySelectiveColumnReader
throws IOException
{
int streamPosition = 0;
boolean checkNulls = filters != null && filters.stream().anyMatch(f -> f.testNull());
for (int i = 0; i < positionCount; i++) {
int position = positions[i];
if (position > streamPosition) {
@ -264,7 +265,7 @@ public class SliceDictionarySelectiveColumnReader
if (outputRequired) {
values[outputPositionCount] = dictionaryBlock.getPositionCount() - 1;
}
if (accumulator != null) {
if (accumulator != null && checkNulls) {
accumulator.set(position);
}
outputPositions[outputPositionCount] = position;
@ -277,7 +278,7 @@ public class SliceDictionarySelectiveColumnReader
if (filters.get(0).testLength(currentPosLength)) {
Slice data = dictionaryBlock.getSlice(index, 0, currentPosLength);
if ((accumulator != null && accumulator.get(position))
|| filters == null || filters.get(0).testBytes(data.getBytes(), 0, currentPosLength)) {
|| filters == null || filters.stream().anyMatch(f -> f.testBytes(data.getBytes(), 0, currentPosLength))) {
if (accumulator != null) {
accumulator.set(position);
}

7
presto-orc/src/main/java/io/prestosql/orc/reader/SliceDirectSelectiveColumnReader.java
View File

@ -214,6 +214,7 @@ public class SliceDirectSelectiveColumnReader
allNulls = false;
int streamPosition = 0;
int dataToSkip = 0;
boolean checkNulls = filters != null && filters.stream().anyMatch(f -> f.testNull());
for (int i = 0; i < positionCount; i++) {
int position = positions[i];
@ -231,7 +232,7 @@ public class SliceDirectSelectiveColumnReader
offsets[outputPositionCount + 1] = offset;
nulls[outputPositionCount] = true;
}
if (accumulator != null) {
if (accumulator != null && checkNulls) {
accumulator.set(position);
}
outputPositions[outputPositionCount] = position;
@ -247,7 +248,7 @@ public class SliceDirectSelectiveColumnReader
dataToSkip = 0;
dataStream.next(data, dataOffset, dataOffset + length);
if ((accumulator != null && accumulator.get(position))
|| filters == null || filters.get(0).testBytes(data, dataOffset, length)) {
|| filters == null || filters.stream().anyMatch(f -> f.testBytes(data, dataOffset, length))) {
if (accumulator != null) {
accumulator.set(position);
}
@ -255,7 +256,7 @@ public class SliceDirectSelectiveColumnReader
}
else {
if ((accumulator != null && accumulator.get(position))
|| filters == null || filters.get(0).testBytes("".getBytes(), 0, 0)) {
|| filters == null || filters.stream().anyMatch(f -> f.testBytes("".getBytes(), 0, 0))) {
if (accumulator != null) {
accumulator.set(position);
}

4
presto-orc/src/main/java/io/prestosql/orc/reader/TimestampSelectiveColumnReader.java
View File

@ -206,6 +206,7 @@ public class TimestampSelectiveColumnReader
{
int streamPosition = 0;
outputPositionCount = 0;
boolean checkNulls = filters != null && filters.stream().anyMatch(f -> f.testNull());
for (int i = 0; i < positionCount; i++) {
int position = positions[i];
if (position > streamPosition) {
@ -218,6 +219,9 @@ public class TimestampSelectiveColumnReader
if (outputRequired) {
nulls[outputPositionCount] = true;
}
if (accumulator != null && checkNulls) {
accumulator.set(position);
}
outputPositions[outputPositionCount] = position;
outputPositionCount++;
}