From 446621c51a22bca673d459727a80f682809ee641 Mon Sep 17 00:00:00 2001 From: "nitin.kashyap" Date: Wed, 20 Jan 2021 15:56:39 +0530 Subject: [PATCH] [I2E4X7] fixed nulls handling in predicate pushdown flow --- .../hive/TestHiveIntegrationSmokeTest.java | 76 +++++++++++++++++++ .../reader/BooleanSelectiveColumnReader.java | 6 +- .../reader/DoubleSelectiveColumnReader.java | 23 +++--- .../reader/FloatSelectiveColumnReader.java | 23 +++--- .../LongDecimalSelectiveColumnReader.java | 5 +- .../orc/reader/LongSelectiveColumnReader.java | 7 +- .../ShortDecimalSelectiveColumnReader.java | 5 +- .../SliceDictionarySelectiveColumnReader.java | 5 +- .../SliceDirectSelectiveColumnReader.java | 7 +- .../TimestampSelectiveColumnReader.java | 4 + 10 files changed, 125 insertions(+), 36 deletions(-) diff --git a/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveIntegrationSmokeTest.java b/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveIntegrationSmokeTest.java index 093b3112d..b03f88310 100644 --- a/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveIntegrationSmokeTest.java +++ b/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveIntegrationSmokeTest.java @@ -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 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() { diff --git a/presto-orc/src/main/java/io/prestosql/orc/reader/BooleanSelectiveColumnReader.java b/presto-orc/src/main/java/io/prestosql/orc/reader/BooleanSelectiveColumnReader.java index a3d240f43..5c421c4ce 100644 --- a/presto-orc/src/main/java/io/prestosql/orc/reader/BooleanSelectiveColumnReader.java +++ b/presto-orc/src/main/java/io/prestosql/orc/reader/BooleanSelectiveColumnReader.java @@ -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); } diff --git a/presto-orc/src/main/java/io/prestosql/orc/reader/DoubleSelectiveColumnReader.java b/presto-orc/src/main/java/io/prestosql/orc/reader/DoubleSelectiveColumnReader.java index 98d65ee38..64adf59a2 100644 --- a/presto-orc/src/main/java/io/prestosql/orc/reader/DoubleSelectiveColumnReader.java +++ b/presto-orc/src/main/java/io/prestosql/orc/reader/DoubleSelectiveColumnReader.java @@ -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 diff --git a/presto-orc/src/main/java/io/prestosql/orc/reader/FloatSelectiveColumnReader.java b/presto-orc/src/main/java/io/prestosql/orc/reader/FloatSelectiveColumnReader.java index 39a35da2b..25e102b42 100644 --- a/presto-orc/src/main/java/io/prestosql/orc/reader/FloatSelectiveColumnReader.java +++ b/presto-orc/src/main/java/io/prestosql/orc/reader/FloatSelectiveColumnReader.java @@ -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 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); } diff --git a/presto-orc/src/main/java/io/prestosql/orc/reader/LongDecimalSelectiveColumnReader.java b/presto-orc/src/main/java/io/prestosql/orc/reader/LongDecimalSelectiveColumnReader.java index f27915fe6..611b315f3 100644 --- a/presto-orc/src/main/java/io/prestosql/orc/reader/LongDecimalSelectiveColumnReader.java +++ b/presto-orc/src/main/java/io/prestosql/orc/reader/LongDecimalSelectiveColumnReader.java @@ -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); } diff --git a/presto-orc/src/main/java/io/prestosql/orc/reader/LongSelectiveColumnReader.java b/presto-orc/src/main/java/io/prestosql/orc/reader/LongSelectiveColumnReader.java index 64f366423..f8abdae9e 100644 --- a/presto-orc/src/main/java/io/prestosql/orc/reader/LongSelectiveColumnReader.java +++ b/presto-orc/src/main/java/io/prestosql/orc/reader/LongSelectiveColumnReader.java @@ -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); } diff --git a/presto-orc/src/main/java/io/prestosql/orc/reader/ShortDecimalSelectiveColumnReader.java b/presto-orc/src/main/java/io/prestosql/orc/reader/ShortDecimalSelectiveColumnReader.java index e943e16b4..4b206cf75 100644 --- a/presto-orc/src/main/java/io/prestosql/orc/reader/ShortDecimalSelectiveColumnReader.java +++ b/presto-orc/src/main/java/io/prestosql/orc/reader/ShortDecimalSelectiveColumnReader.java @@ -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); } diff --git a/presto-orc/src/main/java/io/prestosql/orc/reader/SliceDictionarySelectiveColumnReader.java b/presto-orc/src/main/java/io/prestosql/orc/reader/SliceDictionarySelectiveColumnReader.java index ec7cd0942..89bc25996 100644 --- a/presto-orc/src/main/java/io/prestosql/orc/reader/SliceDictionarySelectiveColumnReader.java +++ b/presto-orc/src/main/java/io/prestosql/orc/reader/SliceDictionarySelectiveColumnReader.java @@ -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); } diff --git a/presto-orc/src/main/java/io/prestosql/orc/reader/SliceDirectSelectiveColumnReader.java b/presto-orc/src/main/java/io/prestosql/orc/reader/SliceDirectSelectiveColumnReader.java index 591bfcf49..5a6d7eb13 100644 --- a/presto-orc/src/main/java/io/prestosql/orc/reader/SliceDirectSelectiveColumnReader.java +++ b/presto-orc/src/main/java/io/prestosql/orc/reader/SliceDirectSelectiveColumnReader.java @@ -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); } diff --git a/presto-orc/src/main/java/io/prestosql/orc/reader/TimestampSelectiveColumnReader.java b/presto-orc/src/main/java/io/prestosql/orc/reader/TimestampSelectiveColumnReader.java index b4289874b..604e5ebef 100644 --- a/presto-orc/src/main/java/io/prestosql/orc/reader/TimestampSelectiveColumnReader.java +++ b/presto-orc/src/main/java/io/prestosql/orc/reader/TimestampSelectiveColumnReader.java @@ -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++; }