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

Fixes for IS NULL Operator OR pushdown issues

This commit is contained in:
nitin.kashyap 2020-09-14 09:45:40 +05:30
parent bcccf227ee
commit 996a4cdaec
20 changed files with 157 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
presto-hive/src/test/java/io/prestosql/plugin/hive/HiveQueryRunner.java
View File

@ -111,6 +111,7 @@ public final class HiveQueryRunner
Map<String, String> configProperties = new HashMap<>();
configProperties.put("auto-vacuum.enabled", "true");
configProperties.put("auto-vacuum.scan.interval", "15s");
configProperties.put("hetu.split-cache-map.enabled", "true");
queryRunner = DistributedQueryRunner
.builder(createSession(Optional.of(new SelectedRole(ROLE, Optional.of("admin")))))

3
presto-orc/src/main/java/io/prestosql/orc/reader/AbstractDecimalSelectiveColumnReader.java
View File

@ -176,6 +176,9 @@ public abstract class AbstractDecimalSelectiveColumnReader<T>
outputPositionCount = 0;
if (dataStream == null && scaleStream == null && presentStream != null) {
streamPosition = readAllNulls(positions, positionCount);
if (filters != null && filters.get(0).testNull() && accumulator != null) {
accumulator.set(positions[0], streamPosition);
}
}
else if (filters == null) {
streamPosition = readNoFilter(positions, positionCount);

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

@ -259,6 +259,9 @@ public class BooleanSelectiveColumnReader
int streamPosition = 0;
if (dataStream == null && presentStream != null) {
streamPosition = readAllNulls(positions, positionCount);
if (filters != null && filters.get(0).testNull() && accumulator != null) {
accumulator.set(positions[0], streamPosition);
}
}
else if (filters == null) {
streamPosition = readNoFilter(positions, positionCount);
@ -277,6 +280,10 @@ public class BooleanSelectiveColumnReader
if (outputRequired) {
nulls[outputPositionCount] = true;
}
if (accumulator != null) {
accumulator.set(position);
}
outputPositions[outputPositionCount] = position;
outputPositionCount++;
}

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

@ -176,6 +176,9 @@ public class DoubleSelectiveColumnReader
int streamPosition;
if (dataStream == null && presentStream != null) {
streamPosition = readAllNulls(positions, positionCount);
if (filters != null && filters.get(0).testNull() && accumulator != null) {
accumulator.set(positions[0], streamPosition);
}
}
else if (filters == null) {
streamPosition = readNoFilter(positions, positionCount);
@ -293,6 +296,9 @@ public class DoubleSelectiveColumnReader
if (outputRequired) {
nulls[outputPositionCount] = true;
}
if (accumulator != null) {
accumulator.set(position);
}
outputPositions[outputPositionCount] = position;
outputPositionCount++;
}

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

@ -152,6 +152,9 @@ public class FloatSelectiveColumnReader
int streamPosition = 0;
if (dataStream == null && presentStream != null) {
streamPosition = readAllNulls(positions, positionCount);
if (filters != null && filters.get(0).testNull() && accumulator != null) {
accumulator.set(positions[0], streamPosition);
}
}
else if (filters == null) {
streamPosition = readNoFilter(positions, positionCount);
@ -246,6 +249,9 @@ public class FloatSelectiveColumnReader
if (outputRequired) {
nulls[outputPositionCount] = true;
}
if (accumulator != null) {
accumulator.set(position);
}
outputPositions[outputPositionCount] = position;
outputPositionCount++;
}

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

@ -146,6 +146,11 @@ public class LongDecimalSelectiveColumnReader
if (outputRequired) {
nulls[outputPositionCount] = true;
}
if (accumulator != null) {
accumulator.set(position);
}
outputPositions[outputPositionCount] = position;
outputPositionCount++;
}

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

@ -203,6 +203,9 @@ public class LongSelectiveColumnReader
int streamPosition = 0;
if (dataStream == null && presentStream != null) {
streamPosition = readAllNulls(positions, positionCount);
if (filters != null && filters.get(0).testNull() && accumulator != null) {
accumulator.set(positions[0], streamPosition);
}
}
else {
for (int i = 0; i < positionCount; i++) {
@ -221,6 +224,9 @@ public class LongSelectiveColumnReader
if (filters != null) {
outputPositions[outputPositionCount] = position;
}
if (accumulator != null) {
accumulator.set(position);
}
outputPositionCount++;
}
}

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

@ -131,6 +131,9 @@ public class ShortDecimalSelectiveColumnReader
if (outputRequired) {
nulls[outputPositionCount] = true;
}
if (accumulator != null) {
accumulator.set(position);
}
outputPositions[outputPositionCount] = position;
outputPositionCount++;
}

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

@ -167,6 +167,9 @@ public class SliceDictionarySelectiveColumnReader
if (dataStream == null && presentStream != null) {
streamPosition = readAllNulls(positions, positionCount);
if (filters != null && filters.get(0).testNull() && accumulator != null) {
accumulator.set(positions[0], streamPosition);
}
}
else if (filters == null) {
streamPosition = readNoFilter(positions, positionCount);
@ -262,6 +265,9 @@ public class SliceDictionarySelectiveColumnReader
if (outputRequired) {
values[outputPositionCount] = dictionaryBlock.getPositionCount() - 1;
}
if (accumulator != null) {
accumulator.set(position);
}
outputPositions[outputPositionCount] = position;
outputPositionCount++;
}

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

@ -152,6 +152,9 @@ public class SliceDirectSelectiveColumnReader
if (lengthStream == null) {
streamPosition = readAllNulls(positions, positionCount);
if (filters != null && filters.get(0).testNull() && accumulator != null) {
accumulator.set(positions[0], streamPosition);
}
}
else if (filters == null) {
streamPosition = readNoFilter(positions, positionCount);
@ -229,6 +232,9 @@ public class SliceDirectSelectiveColumnReader
offsets[outputPositionCount + 1] = offset;
nulls[outputPositionCount] = true;
}
if (accumulator != null) {
accumulator.set(position);
}
outputPositions[outputPositionCount] = position;
outputPositionCount++;
}

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

@ -183,6 +183,9 @@ public class TimestampSelectiveColumnReader
int streamPosition = 0;
if (secondsStream == null && nanosStream == null && presentStream != null) {
streamPosition = readAllNulls(positions, positionCount);
if (filters != null && filters.get(0).testNull() && accumulator != null) {
accumulator.set(positions[0], streamPosition);
}
}
else if (filters == null) {
streamPosition = readNoFilter(positions, positionCount);

6
presto-spi/src/main/java/io/prestosql/spi/block/ByteArrayBlock.java
View File

@ -243,9 +243,11 @@ public class ByteArrayBlock
int matchCount = 0;
for (int i = 0; i < positionCount; i++) {
if (valueIsNull != null && valueIsNull[positions[i] + arrayOffset]) {
continue;
if (test.apply(null)) {
matchedPositions[matchCount++] = positions[i];
}
}
if (test.apply(values[positions[i] + arrayOffset])) {
else if (test.apply(values[positions[i] + arrayOffset])) {
matchedPositions[matchCount++] = positions[i];
}
}

16
presto-spi/src/main/java/io/prestosql/spi/block/DictionaryBlock.java
View File

@ -472,12 +472,15 @@ public class DictionaryBlock<T>
int matchCount = 0;
for (int i = 0; i < positionCount; i++) {
if (dictionary.isNull(getId(positions[i]))) {
continue;
if (test.apply(null)) {
matchedPositions[matchCount++] = positions[i];
}
}
T value = dictionary.get(getId(positions[i]));
if (test.apply(value)) {
matchedPositions[matchCount++] = positions[i];
else {
T value = dictionary.get(getId(positions[i]));
if (test.apply(value)) {
matchedPositions[matchCount++] = positions[i];
}
}
}
@ -487,6 +490,9 @@ public class DictionaryBlock<T>
@Override
public T get(int position)
{
if (dictionary.isNull(getId(position))) {
return null;
}
return dictionary.get(getId(position));
}
}

14
presto-spi/src/main/java/io/prestosql/spi/block/Int128ArrayBlock.java
View File

@ -253,12 +253,16 @@ public class Int128ArrayBlock
long[] val = new long[2];
for (int i = 0; i < positionCount; i++) {
if (valueIsNull != null && valueIsNull[positions[i] + positionOffset]) {
continue;
if (test.apply(null)) {
matchedPositions[matchCount++] = positions[i];
}
}
val[0] = values[(positions[i] + positionOffset) * 2];
val[1] = values[((positions[i] + positionOffset) * 2) + 1];
if (test.apply(val)) {
matchedPositions[matchCount++] = positions[i];
else {
val[0] = values[(positions[i] + positionOffset) * 2];
val[1] = values[((positions[i] + positionOffset) * 2) + 1];
if (test.apply(val)) {
matchedPositions[matchCount++] = positions[i];
}
}
}

6
presto-spi/src/main/java/io/prestosql/spi/block/IntArrayBlock.java
View File

@ -247,9 +247,11 @@ public class IntArrayBlock
int matchCount = 0;
for (int i = 0; i < positionCount; i++) {
if (valueIsNull != null && valueIsNull[positions[i] + arrayOffset]) {
continue;
if (test.apply(null)) {
matchedPositions[matchCount++] = positions[i];
}
}
if (test.apply(values[positions[i] + arrayOffset])) {
else if (test.apply(values[positions[i] + arrayOffset])) {
matchedPositions[matchCount++] = positions[i];
}
}

6
presto-spi/src/main/java/io/prestosql/spi/block/LongArrayBlock.java
View File

@ -299,9 +299,11 @@ public class LongArrayBlock
int matchCount = 0;
for (int i = 0; i < positionCount; i++) {
if (valueIsNull != null && valueIsNull[positions[i] + arrayOffset]) {
continue;
if (test.apply(null)) {
matchedPositions[matchCount++] = positions[i];
}
}
if (test.apply(values[positions[i] + arrayOffset])) {
else if (test.apply(values[positions[i] + arrayOffset])) {
matchedPositions[matchCount++] = positions[i];
}
}

30
presto-spi/src/main/java/io/prestosql/spi/block/RunLengthEncodedBlock.java
View File

@ -19,6 +19,7 @@ import io.prestosql.spi.type.Type;
import org.openjdk.jol.info.ClassLayout;
import java.util.function.BiConsumer;
import java.util.function.Function;
import static io.prestosql.spi.block.BlockUtil.checkArrayRange;
import static io.prestosql.spi.block.BlockUtil.checkValidPosition;
@ -304,4 +305,33 @@ public class RunLengthEncodedBlock<T>
throw new IllegalArgumentException("position is not valid");
}
}
@Override
public int filter(int[] positions, int positionCount, int[] matchedPositions, Function<Object, Boolean> test)
{
int matchCount = 0;
for (int i = 0; i < positionCount; i++) {
checkReadablePosition(positions[i]);
if (value.isNull(0)) {
if (test.apply(null)) {
matchedPositions[matchCount++] = positions[i];
}
}
else if (test.apply(value.get(0))) {
matchedPositions[matchCount++] = positions[i];
}
}
return matchCount;
}
@Override
public T get(int position)
{
checkReadablePosition(position);
if (value.isNull(0)) {
return null;
}
return value.get(0);
}
}

6
presto-spi/src/main/java/io/prestosql/spi/block/ShortArrayBlock.java
View File

@ -241,9 +241,11 @@ public class ShortArrayBlock
int matchCount = 0;
for (int i = 0; i < positionCount; i++) {
if (valueIsNull != null && valueIsNull[positions[i] + arrayOffset]) {
continue;
if (test.apply(null)) {
matchedPositions[matchCount++] = positions[i];
}
}
if (test.apply(values[positions[i] + arrayOffset])) {
else if (test.apply(values[positions[i] + arrayOffset])) {
matchedPositions[matchCount++] = positions[i];
}
}

16
presto-spi/src/main/java/io/prestosql/spi/block/VariableWidthBlock.java
View File

@ -242,12 +242,15 @@ public class VariableWidthBlock
int matchCount = 0;
for (int i = 0; i < positionCount; i++) {
if (valueIsNull != null && valueIsNull[positions[i] + arrayOffset]) {
continue;
if (test.apply(null)) {
matchedPositions[matchCount++] = positions[i];
}
}
byte[] value = slice.slice(offsets[i + arrayOffset], offsets[i + arrayOffset + 1] - offsets[i + arrayOffset]).getBytes();
if (test.apply(value)) {
matchedPositions[matchCount++] = positions[i];
else {
byte[] value = slice.slice(offsets[i + arrayOffset], offsets[i + arrayOffset + 1] - offsets[i + arrayOffset]).getBytes();
if (test.apply(value)) {
matchedPositions[matchCount++] = positions[i];
}
}
}
@ -257,6 +260,9 @@ public class VariableWidthBlock
@Override
public byte[] get(int position)
{
if (valueIsNull != null && valueIsNull[position + arrayOffset]) {
return null;
}
return slice.slice(offsets[position + arrayOffset], offsets[position + arrayOffset + 1] - offsets[position + arrayOffset]).getBytes();
}
}

34
presto-tests/src/main/java/io/prestosql/tests/AbstractTestDistributedQueries.java
View File

@ -1087,21 +1087,43 @@ public abstract class AbstractTestDistributedQueries
assertUpdate("INSERT INTO test_partition_predicate VALUES (4,'d',4), (5,'e',5), (6,'f',6)", 3);
assertUpdate("INSERT INTO test_partition_predicate VALUES (7,'g',7), (8,'h',8), (9,'i',9)", 3);
assertQuery(getSession(),
"SELECT id, p1, p2 FROM test_partition_predicate WHERE id > 0 and (p1='b' or p2>3) ORDER BY id",
String sql = "SELECT id, p1, p2 FROM test_partition_predicate WHERE id > 0 and (p1='b' or p2>3) ORDER BY id";
assertQuery(getSession(), sql,
"VALUES (2,'b',2), (4,'d',4), (5,'e',5), (6,'f',6), (7,'g',7), (8,'h',8), (9,'i',9)");
Session session1 = Session.builder(getSession())
.setCatalogSessionProperty(getSession().getCatalog().get(), "orc_predicate_pushdown_enabled", "true")
.build();
assertQuery(session1,
"SELECT id, p1, p2 FROM test_partition_predicate WHERE id > 0 and (p1='b' or p2>3) ORDER BY id",
assertQuery(session1, sql,
"VALUES (2,'b',2), (4,'d',4), (5,'e',5), (6,'f',6), (7,'g',7), (8,'h',8), (9,'i',9)");
assertUpdate("INSERT INTO test_partition_predicate VALUES (10,'j',10)", 1);
assertQuery(session1,
"SELECT id, p1, p2 FROM test_partition_predicate WHERE id > 0 and (p1='b' or p2>3) ORDER BY id",
assertQuery(session1, sql,
"VALUES (2,'b',2), (4,'d',4), (5,'e',5), (6,'f',6), (7,'g',7), (8,'h',8), (9,'i',9), (10,'j',10)");
assertUpdate("INSERT INTO test_partition_predicate VALUES (11,NULL,11), (12,NULL,NULL), (NULL,NULL,NULL)", 3);
/* NUlls Excluded */
MaterializedResult resultNormal = computeActual(sql);
MaterializedResult resultPushdown = computeActual(session1, sql);
assertEquals(resultNormal.getMaterializedRows(), resultPushdown.getMaterializedRows());
/* NUlls Included */
sql = "SELECT id, p1, p2 FROM test_partition_predicate WHERE id > 0 and (p1 IS NULL or p2<3) ORDER BY id, p1";
resultPushdown = computeActual(session1, sql);
resultNormal = computeActual(sql);
assertEquals(resultNormal.getMaterializedRows(), resultPushdown.getMaterializedRows());
/* Query Test with Cache */
Session session2 = Session.builder(getSession())
.setCatalogSessionProperty(getSession().getCatalog().get(), "orc_predicate_pushdown_enabled", "true")
.setCatalogSessionProperty(getSession().getCatalog().get(), "orc_row_data_cache_enabled", "true")
.build();
assertQuery(session2, "CACHE TABLE test_partition_predicate WHERE p2 > 0", "VALUES ('OK')");
MaterializedResult resultCachePushdown = computeActual(session2, sql);
assertEquals(resultNormal.getMaterializedRows(), resultCachePushdown.getMaterializedRows());
}
}
}