[
https://issues.apache.org/jira/browse/DRILL-6353?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16502508#comment-16502508
]
ASF GitHub Bot commented on DRILL-6353:
---------------------------------------
parthchandra closed pull request #1259: DRILL-6353: Upgrade Parquet MR
dependencies
URL: https://github.com/apache/drill/pull/1259
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/contrib/storage-hive/hive-exec-shade/pom.xml
b/contrib/storage-hive/hive-exec-shade/pom.xml
index 6f511adf71..98fd4b8150 100644
--- a/contrib/storage-hive/hive-exec-shade/pom.xml
+++ b/contrib/storage-hive/hive-exec-shade/pom.xml
@@ -31,6 +31,20 @@
<packaging>jar</packaging>
<name>contrib/hive-storage-plugin/hive-exec-shaded</name>
+ <properties>
+ <hive.parquet.version>1.8.3</hive.parquet.version>
+ </properties>
+
+ <dependencyManagement>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.parquet</groupId>
+ <artifactId>parquet-hadoop-bundle</artifactId>
+ <version>${hive.parquet.version}</version>
+ </dependency>
+ </dependencies>
+ </dependencyManagement>
+
<dependencies>
<dependency>
<groupId>org.apache.hive</groupId>
@@ -68,11 +82,6 @@
</exclusion>
</exclusions>
</dependency>
- <!--Once newer hive-exec version leverages parquet-column 1.9.0, this
dependency can be deleted -->
- <dependency>
- <groupId>org.apache.parquet</groupId>
- <artifactId>parquet-column</artifactId>
- </dependency>
</dependencies>
<build>
@@ -83,7 +92,7 @@
<artifactSet>
<includes>
<include>org.apache.hive:hive-exec</include>
- <include>org.apache.parquet:parquet-column</include>
+ <include>org.apache.parquet:parquet-hadoop-bundle</include>
<include>commons-codec:commons-codec</include>
<include>com.fasterxml.jackson.core:jackson-databind</include>
<include>com.fasterxml.jackson.core:jackson-annotations</include>
@@ -117,6 +126,10 @@
<pattern>org.apache.parquet.</pattern>
<shadedPattern>hive.org.apache.parquet.</shadedPattern>
</relocation>
+ <relocation>
+ <pattern>shaded.parquet.</pattern>
+ <shadedPattern>hive.shaded.parquet.</shadedPattern>
+ </relocation>
<relocation>
<pattern>org.apache.avro.</pattern>
<shadedPattern>hive.org.apache.avro.</shadedPattern>
diff --git a/exec/java-exec/pom.xml b/exec/java-exec/pom.xml
index 0d03cc8515..d0c6724b37 100644
--- a/exec/java-exec/pom.xml
+++ b/exec/java-exec/pom.xml
@@ -249,92 +249,17 @@
<dependency>
<groupId>org.apache.parquet</groupId>
<artifactId>parquet-hadoop</artifactId>
- <version>${parquet.version}</version>
<exclusions>
- <exclusion>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-client</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-common</artifactId>
- </exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.parquet</groupId>
<artifactId>parquet-format</artifactId>
- <version>2.3.0-incubating</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-client</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-common</artifactId>
- </exclusion>
- </exclusions>
</dependency>
<dependency>
<groupId>org.apache.parquet</groupId>
<artifactId>parquet-common</artifactId>
<version>${parquet.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-client</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-common</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
- <dependency>
- <groupId>org.apache.parquet</groupId>
- <artifactId>parquet-jackson</artifactId>
- <version>${parquet.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-client</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-common</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
- <dependency>
- <groupId>org.apache.parquet</groupId>
- <artifactId>parquet-encoding</artifactId>
- <version>${parquet.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-client</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-common</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
- <dependency>
- <groupId>org.apache.parquet</groupId>
- <artifactId>parquet-generator</artifactId>
- <version>${parquet.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-client</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-common</artifactId>
- </exclusion>
- </exclusions>
</dependency>
<dependency>
<groupId>javax.inject</groupId>
diff --git
a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/stat/ParquetComparisonPredicates.java
b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/stat/ParquetComparisonPredicates.java
index 5ba597c2a1..673d242d6d 100644
---
a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/stat/ParquetComparisonPredicates.java
+++
b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/stat/ParquetComparisonPredicates.java
@@ -20,274 +20,176 @@
import org.apache.drill.common.expression.LogicalExpression;
import org.apache.drill.common.expression.LogicalExpressionBase;
import org.apache.drill.common.expression.visitors.ExprVisitor;
+import org.apache.drill.exec.expr.fn.FunctionGenerationHelper;
import org.apache.parquet.column.statistics.Statistics;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
+import java.util.function.BiPredicate;
+
+import static
org.apache.drill.exec.expr.stat.ParquetPredicatesHelper.isNullOrEmpty;
+import static
org.apache.drill.exec.expr.stat.ParquetPredicatesHelper.isAllNulls;
/**
* Comparison predicates for parquet filter pushdown.
*/
-public class ParquetComparisonPredicates {
- public static abstract class ParquetCompPredicate extends
LogicalExpressionBase implements ParquetFilterPredicate {
- protected final LogicalExpression left;
- protected final LogicalExpression right;
-
- public ParquetCompPredicate(LogicalExpression left, LogicalExpression
right) {
- super(left.getPosition());
- this.left = left;
- this.right = right;
- }
-
- @Override
- public Iterator<LogicalExpression> iterator() {
- final List<LogicalExpression> args = new ArrayList<>();
- args.add(left);
- args.add(right);
- return args.iterator();
- }
+public class ParquetComparisonPredicates extends LogicalExpressionBase
implements ParquetFilterPredicate {
+ private final LogicalExpression left;
+ private final LogicalExpression right;
+ private final BiPredicate<Statistics, Statistics> predicate;
+
+ private ParquetComparisonPredicates(LogicalExpression left,
LogicalExpression right, BiPredicate<Statistics, Statistics> predicate) {
+ super(left.getPosition());
+ this.left = left;
+ this.right = right;
+ this.predicate = predicate;
+ }
- @Override
- public <T, V, E extends Exception> T accept(ExprVisitor<T, V, E> visitor,
V value) throws E {
- return visitor.visitUnknown(this, value);
- }
+ @Override
+ public Iterator<LogicalExpression> iterator() {
+ final List<LogicalExpression> args = new ArrayList<>();
+ args.add(left);
+ args.add(right);
+ return args.iterator();
+ }
+ @Override
+ public <T, V, E extends Exception> T accept(ExprVisitor<T, V, E> visitor, V
value) throws E {
+ return visitor.visitUnknown(this, value);
}
/**
- * EQ (=) predicate
+ * Semantics of canDrop() is very similar to what is implemented in Parquet
library's
+ * {@link org.apache.parquet.filter2.statisticslevel.StatisticsFilter} and
+ * {@link org.apache.parquet.filter2.predicate.FilterPredicate}
+ *
+ * Main difference :
+ * 1. A RangeExprEvaluator is used to compute the min/max of an expression,
such as CAST function
+ * of a column. CAST function could be explicitly added by Drill user (It's
recommended to use CAST
+ * function after DRILL-4372, if user wants to reduce planning time for
limit 0 query), or implicitly
+ * inserted by Drill, when the types of compare operands are not identical.
Therefore, it's important
+ * to allow CAST function to appear in the filter predicate.
+ * 2. We do not require list of ColumnChunkMetaData to do the evaluation,
while Parquet library's
+ * StatisticsFilter has such requirement. Drill's ParquetTableMetaData does
not maintain ColumnChunkMetaData,
+ * making it impossible to directly use Parquet library's StatisticFilter in
query planning time.
+ * 3. We allows both sides of comparison operator to be a min/max range. As
such, we support
+ * expression_of(Column1) < expression_of(Column2),
+ * where Column1 and Column2 are from same parquet table.
*/
- public static class EqualPredicate extends ParquetCompPredicate {
- public EqualPredicate(LogicalExpression left, LogicalExpression right) {
- super(left, right);
+ @Override
+ public boolean canDrop(RangeExprEvaluator evaluator) {
+ Statistics leftStat = left.accept(evaluator, null);
+ if (isNullOrEmpty(leftStat)) {
+ return false;
}
- /**
- Semantics of canDrop() is very similar to what is implemented in
Parquet library's
- {@link org.apache.parquet.filter2.statisticslevel.StatisticsFilter} and
- {@link org.apache.parquet.filter2.predicate.FilterPredicate}
-
- Main difference :
- 1. A RangeExprEvaluator is used to compute the min/max of an expression,
such as CAST function
- of a column. CAST function could be explicitly added by Drill user
(It's recommended to use CAST
- function after DRILL-4372, if user wants to reduce planning time for
limit 0 query), or implicitly
- inserted by Drill, when the types of compare operands are not
identical. Therefore, it's important
- to allow CAST function to appear in the filter predicate.
- 2. We do not require list of ColumnChunkMetaData to do the evaluation,
while Parquet library's
- StatisticsFilter has such requirement. Drill's ParquetTableMetaData
does not maintain ColumnChunkMetaData,
- making it impossible to directly use Parquet library's StatisticFilter
in query planning time.
- 3. We allows both sides of comparison operator to be a min/max range. As
such, we support
- expression_of(Column1) < expression_of(Column2),
- where Column1 and Column2 are from same parquet table.
- */
- @Override
- public boolean canDrop(RangeExprEvaluator evaluator) {
- Statistics leftStat = left.accept(evaluator, null);
- Statistics rightStat = right.accept(evaluator, null);
-
- if (leftStat == null ||
- rightStat == null ||
- leftStat.isEmpty() ||
- rightStat.isEmpty()) {
- return false;
- }
+ Statistics rightStat = right.accept(evaluator, null);
+ if (isNullOrEmpty(rightStat)) {
+ return false;
+ }
- // if either side is ALL null, = is evaluated to UNKNOW -> canDrop
- if (ParquetPredicatesHelper.isAllNulls(leftStat,
evaluator.getRowCount()) ||
- ParquetPredicatesHelper.isAllNulls(rightStat,
evaluator.getRowCount())) {
- return true;
- }
+ // if either side is ALL null, = is evaluated to UNKNOWN -> canDrop
+ if (isAllNulls(leftStat, evaluator.getRowCount()) || isAllNulls(rightStat,
evaluator.getRowCount())) {
+ return true;
+ }
- // can drop when left's max < right's min, or right's max < left's min
- if ( ( leftStat.genericGetMax().compareTo(rightStat.genericGetMin()) < 0
- || rightStat.genericGetMax().compareTo(leftStat.genericGetMin()) <
0)) {
- return true;
- } else {
- return false;
- }
+ if (leftStat.hasNonNullValue() && rightStat.hasNonNullValue()) {
+ return predicate.test(leftStat, rightStat);
+ } else {
+ return false;
}
+ }
- @Override
- public String toString() {
- return left.toString() + " = " + right.toString();
+ public static LogicalExpression createPredicate(String function,
LogicalExpression left, LogicalExpression right) {
+ switch (function) {
+ case FunctionGenerationHelper.EQ:
+ return createEqualPredicate(left, right);
+ case FunctionGenerationHelper.GT:
+ return createGTPredicate(left, right);
+ case FunctionGenerationHelper.GE:
+ return createGEPredicate(left, right);
+ case FunctionGenerationHelper.LT:
+ return createLTPredicate(left, right);
+ case FunctionGenerationHelper.LE:
+ return createLEPredicate(left, right);
+ case FunctionGenerationHelper.NE:
+ return createNEPredicate(left, right);
+ default:
+ return null;
}
}
/**
- * GT (>) predicate.
+ * EQ (=) predicate
*/
- public static class GTPredicate extends ParquetCompPredicate {
- public GTPredicate(LogicalExpression left, LogicalExpression right) {
- super(left, right);
- }
-
- @Override
- public boolean canDrop(RangeExprEvaluator evaluator) {
- Statistics leftStat = left.accept(evaluator, null);
- Statistics rightStat = right.accept(evaluator, null);
-
- if (leftStat == null ||
- rightStat == null ||
- leftStat.isEmpty() ||
- rightStat.isEmpty()) {
- return false;
- }
-
- // if either side is ALL null, = is evaluated to UNKNOW -> canDrop
- if (ParquetPredicatesHelper.isAllNulls(leftStat,
evaluator.getRowCount()) ||
- ParquetPredicatesHelper.isAllNulls(rightStat,
evaluator.getRowCount())) {
- return true;
- }
+ private static LogicalExpression createEqualPredicate(LogicalExpression
left, LogicalExpression right) {
+ return new ParquetComparisonPredicates(left, right, (leftStat, rightStat)
-> {
+ // can drop when left's max < right's min, or right's max < left's min
+ final Comparable leftMin = leftStat.genericGetMin();
+ final Comparable rightMin = rightStat.genericGetMin();
+ return leftStat.compareMaxToValue(rightMin) < 0 ||
rightStat.compareMaxToValue(leftMin) < 0;
+ }) {
+ @Override
+ public String toString() {
+ return left + " = " + right;
+ }
+ };
+ }
+ /**
+ * GT (>) predicate.
+ */
+ private static LogicalExpression createGTPredicate(LogicalExpression left,
LogicalExpression right) {
+ return new ParquetComparisonPredicates(left, right, (leftStat, rightStat)
-> {
// can drop when left's max <= right's min.
- if ( leftStat.genericGetMax().compareTo(rightStat.genericGetMin()) <= 0
) {
- return true;
- } else {
- return false;
- }
- }
+ final Comparable rightMin = rightStat.genericGetMin();
+ return leftStat.compareMaxToValue(rightMin) <= 0;
+ });
}
/**
* GE (>=) predicate.
*/
- public static class GEPredicate extends ParquetCompPredicate {
- public GEPredicate(LogicalExpression left, LogicalExpression right) {
- super(left, right);
- }
-
- @Override
- public boolean canDrop(RangeExprEvaluator evaluator) {
- Statistics leftStat = left.accept(evaluator, null);
- Statistics rightStat = right.accept(evaluator, null);
-
- if (leftStat == null ||
- rightStat == null ||
- leftStat.isEmpty() ||
- rightStat.isEmpty()) {
- return false;
- }
-
- // if either side is ALL null, = is evaluated to UNKNOW -> canDrop
- if (ParquetPredicatesHelper.isAllNulls(leftStat,
evaluator.getRowCount()) ||
- ParquetPredicatesHelper.isAllNulls(rightStat,
evaluator.getRowCount())) {
- return true;
- }
-
+ private static LogicalExpression createGEPredicate(LogicalExpression left,
LogicalExpression right) {
+ return new ParquetComparisonPredicates(left, right, (leftStat, rightStat)
-> {
// can drop when left's max < right's min.
- if ( leftStat.genericGetMax().compareTo(rightStat.genericGetMin()) < 0 )
{
- return true;
- } else {
- return false;
- }
- }
+ final Comparable rightMin = rightStat.genericGetMin();
+ return leftStat.compareMaxToValue(rightMin) < 0;
+ });
}
/**
* LT (<) predicate.
*/
- public static class LTPredicate extends ParquetCompPredicate {
- public LTPredicate(LogicalExpression left, LogicalExpression right) {
- super(left, right);
- }
-
- @Override
- public boolean canDrop(RangeExprEvaluator evaluator) {
- Statistics leftStat = left.accept(evaluator, null);
- Statistics rightStat = right.accept(evaluator, null);
-
- if (leftStat == null ||
- rightStat == null ||
- leftStat.isEmpty() ||
- rightStat.isEmpty()) {
- return false;
- }
-
- // if either side is ALL null, = is evaluated to UNKNOW -> canDrop
- if (ParquetPredicatesHelper.isAllNulls(leftStat,
evaluator.getRowCount()) ||
- ParquetPredicatesHelper.isAllNulls(rightStat,
evaluator.getRowCount())) {
- return true;
- }
-
+ private static LogicalExpression createLTPredicate(LogicalExpression left,
LogicalExpression right) {
+ return new ParquetComparisonPredicates(left, right, (leftStat, rightStat)
-> {
// can drop when right's max <= left's min.
- if ( rightStat.genericGetMax().compareTo(leftStat.genericGetMin()) <= 0
) {
- return true;
- } else {
- return false;
- }
- }
+ final Comparable leftMin = leftStat.genericGetMin();
+ return rightStat.compareMaxToValue(leftMin) <= 0;
+ });
}
/**
* LE (<=) predicate.
*/
- public static class LEPredicate extends ParquetCompPredicate {
- public LEPredicate(LogicalExpression left, LogicalExpression right) {
- super(left, right);
- }
-
- @Override
- public boolean canDrop(RangeExprEvaluator evaluator) {
- Statistics leftStat = left.accept(evaluator, null);
- Statistics rightStat = right.accept(evaluator, null);
-
- if (leftStat == null ||
- rightStat == null ||
- leftStat.isEmpty() ||
- rightStat.isEmpty()) {
- return false;
- }
-
- // if either side is ALL null, = is evaluated to UNKNOW -> canDrop
- if (ParquetPredicatesHelper.isAllNulls(leftStat,
evaluator.getRowCount()) ||
- ParquetPredicatesHelper.isAllNulls(rightStat,
evaluator.getRowCount())) {
- return true;
- }
-
+ private static LogicalExpression createLEPredicate(LogicalExpression left,
LogicalExpression right) {
+ return new ParquetComparisonPredicates(left, right, (leftStat, rightStat)
-> {
// can drop when right's max < left's min.
- if ( rightStat.genericGetMax().compareTo(leftStat.genericGetMin()) < 0 )
{
- return true;
- } else {
- return false;
- }
- }
+ final Comparable leftMin = leftStat.genericGetMin();
+ return rightStat.compareMaxToValue(leftMin) < 0;
+ });
}
/**
* NE (!=) predicate.
*/
- public static class NEPredicate extends ParquetCompPredicate {
- public NEPredicate(LogicalExpression left, LogicalExpression right) {
- super(left, right);
- }
-
- @Override
- public boolean canDrop(RangeExprEvaluator evaluator) {
- Statistics leftStat = left.accept(evaluator, null);
- Statistics rightStat = right.accept(evaluator, null);
-
- if (leftStat == null ||
- rightStat == null ||
- leftStat.isEmpty() ||
- rightStat.isEmpty()) {
- return false;
- }
-
- // if either side is ALL null, comparison is evaluated to UNKNOW ->
canDrop
- if (ParquetPredicatesHelper.isAllNulls(leftStat,
evaluator.getRowCount()) ||
- ParquetPredicatesHelper.isAllNulls(rightStat,
evaluator.getRowCount())) {
- return true;
- }
-
+ private static LogicalExpression createNEPredicate(LogicalExpression left,
LogicalExpression right) {
+ return new ParquetComparisonPredicates(left, right, (leftStat, rightStat)
-> {
// can drop when there is only one unique value.
- if ( leftStat.genericGetMin().compareTo(leftStat.genericGetMax()) == 0 &&
- rightStat.genericGetMin().compareTo(rightStat.genericGetMax()) ==0
&&
- leftStat.genericGetMax().compareTo(rightStat.genericGetMax()) == 0)
{
- return true;
- } else {
- return false;
- }
- }
+ final Comparable leftMax = leftStat.genericGetMax();
+ final Comparable rightMax = rightStat.genericGetMax();
+ return leftStat.compareMinToValue(leftMax) == 0 &&
rightStat.compareMinToValue(rightMax) == 0 &&
leftStat.compareMaxToValue(rightMax) == 0;
+ });
}
}
diff --git
a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/stat/ParquetIsPredicates.java
b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/stat/ParquetIsPredicates.java
index ef2b9406b2..e105e01838 100644
---
a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/stat/ParquetIsPredicates.java
+++
b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/stat/ParquetIsPredicates.java
@@ -28,6 +28,10 @@
import java.util.Iterator;
import java.util.List;
+import static
org.apache.drill.exec.expr.stat.ParquetPredicatesHelper.hasNoNulls;
+import static
org.apache.drill.exec.expr.stat.ParquetPredicatesHelper.isAllNulls;
+import static
org.apache.drill.exec.expr.stat.ParquetPredicatesHelper.isNullOrEmpty;
+
/**
* IS predicates for parquet filter pushdown.
*/
@@ -76,16 +80,12 @@ public boolean canDrop(RangeExprEvaluator evaluator) {
Statistics exprStat = expr.accept(evaluator, null);
- if (!ParquetPredicatesHelper.hasStats(exprStat)) {
+ if (isNullOrEmpty(exprStat)) {
return false;
}
//if there are no nulls -> canDrop
- if (!ParquetPredicatesHelper.hasNulls(exprStat)) {
- return true;
- } else {
- return false;
- }
+ return hasNoNulls(exprStat);
}
private boolean isArray(LogicalExpression expression) {
@@ -111,16 +111,12 @@ public IsNotNullPredicate(LogicalExpression expr) {
public boolean canDrop(RangeExprEvaluator evaluator) {
Statistics exprStat = expr.accept(evaluator, null);
- if (!ParquetPredicatesHelper.hasStats(exprStat)) {
+ if (isNullOrEmpty(exprStat)) {
return false;
}
//if there are all nulls -> canDrop
- if (ParquetPredicatesHelper.isAllNulls(exprStat,
evaluator.getRowCount())) {
- return true;
- } else {
- return false;
- }
+ return isAllNulls(exprStat, evaluator.getRowCount());
}
}
@@ -136,17 +132,12 @@ public IsTruePredicate(LogicalExpression expr) {
public boolean canDrop(RangeExprEvaluator evaluator) {
Statistics exprStat = expr.accept(evaluator, null);
- if (!ParquetPredicatesHelper.hasStats(exprStat)) {
+ if (isNullOrEmpty(exprStat)) {
return false;
}
//if max value is not true or if there are all nulls -> canDrop
- if (exprStat.genericGetMax().compareTo(true) != 0 ||
- ParquetPredicatesHelper.isAllNulls(exprStat,
evaluator.getRowCount())) {
- return true;
- } else {
- return false;
- }
+ return exprStat.compareMaxToValue(true) != 0 || isAllNulls(exprStat,
evaluator.getRowCount());
}
}
@@ -162,17 +153,12 @@ public IsFalsePredicate(LogicalExpression expr) {
public boolean canDrop(RangeExprEvaluator evaluator) {
Statistics exprStat = expr.accept(evaluator, null);
- if (!ParquetPredicatesHelper.hasStats(exprStat)) {
+ if (isNullOrEmpty(exprStat)) {
return false;
}
//if min value is not false or if there are all nulls -> canDrop
- if (exprStat.genericGetMin().compareTo(false) != 0 ||
- ParquetPredicatesHelper.isAllNulls(exprStat,
evaluator.getRowCount())) {
- return true;
- } else {
- return false;
- }
+ return exprStat.compareMinToValue(false) != 0 || isAllNulls(exprStat,
evaluator.getRowCount());
}
}
@@ -188,16 +174,12 @@ public IsNotTruePredicate(LogicalExpression expr) {
public boolean canDrop(RangeExprEvaluator evaluator) {
Statistics exprStat = expr.accept(evaluator, null);
- if (!ParquetPredicatesHelper.hasStats(exprStat)) {
+ if (isNullOrEmpty(exprStat)) {
return false;
}
//if min value is not false or if there are no nulls -> canDrop
- if (exprStat.genericGetMin().compareTo(false) != 0 &&
!ParquetPredicatesHelper.hasNulls(exprStat)) {
- return true;
- } else {
- return false;
- }
+ return exprStat.compareMinToValue(false) != 0 && hasNoNulls(exprStat);
}
}
@@ -213,16 +195,12 @@ public IsNotFalsePredicate(LogicalExpression expr) {
public boolean canDrop(RangeExprEvaluator evaluator) {
Statistics exprStat = expr.accept(evaluator, null);
- if (!ParquetPredicatesHelper.hasStats(exprStat)) {
+ if (isNullOrEmpty(exprStat)) {
return false;
}
//if max value is not true or if there are no nulls -> canDrop
- if (exprStat.genericGetMax().compareTo(true) != 0 &&
!ParquetPredicatesHelper.hasNulls(exprStat)) {
- return true;
- } else {
- return false;
- }
+ return exprStat.compareMaxToValue(true) != 0 && hasNoNulls(exprStat);
}
}
}
diff --git
a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/stat/ParquetPredicatesHelper.java
b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/stat/ParquetPredicatesHelper.java
index e83d393e2a..59038c8ac5 100644
---
a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/stat/ParquetPredicatesHelper.java
+++
b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/stat/ParquetPredicatesHelper.java
@@ -22,15 +22,14 @@
/**
* Parquet predicates class helper for filter pushdown.
*/
-@SuppressWarnings("rawtypes")
-public class ParquetPredicatesHelper {
+class ParquetPredicatesHelper {
/**
* @param stat statistics object
* @return true if the input stat object has valid statistics; false
otherwise
*/
- public static boolean hasStats(Statistics stat) {
- return stat != null && !stat.isEmpty();
+ static boolean isNullOrEmpty(Statistics stat) {
+ return stat == null || stat.isEmpty();
}
/**
@@ -41,8 +40,8 @@ public static boolean hasStats(Statistics stat) {
* @return True if all rows are null in the parquet file
* False if at least one row is not null.
*/
- public static boolean isAllNulls(Statistics stat, long rowCount) {
- return stat.getNumNulls() == rowCount;
+ static boolean isAllNulls(Statistics stat, long rowCount) {
+ return stat.isNumNullsSet() && stat.getNumNulls() == rowCount;
}
/**
@@ -52,8 +51,8 @@ public static boolean isAllNulls(Statistics stat, long
rowCount) {
* @return True if the parquet file has nulls
* False if the parquet file hasn't nulls.
*/
- public static boolean hasNulls(Statistics stat) {
- return stat.getNumNulls() > 0;
+ static boolean hasNoNulls(Statistics stat) {
+ return !stat.isNumNullsSet() || stat.getNumNulls() == 0;
}
}
diff --git
a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/AbstractParquetScanBatchCreator.java
b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/AbstractParquetScanBatchCreator.java
index 6a320b85b3..dc09ce1b69 100644
---
a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/AbstractParquetScanBatchCreator.java
+++
b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/AbstractParquetScanBatchCreator.java
@@ -33,11 +33,12 @@
import org.apache.drill.exec.store.parquet2.DrillParquetReader;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
+import org.apache.parquet.ParquetReadOptions;
import org.apache.parquet.column.ColumnDescriptor;
-import org.apache.parquet.format.converter.ParquetMetadataConverter;
import org.apache.parquet.hadoop.CodecFactory;
import org.apache.parquet.hadoop.ParquetFileReader;
import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.parquet.hadoop.util.HadoopInputFile;
import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.Type;
@@ -50,6 +51,9 @@
import java.util.Map;
import java.util.concurrent.TimeUnit;
+import static
org.apache.drill.exec.store.parquet.metadata.Metadata.PARQUET_STRINGS_SIGNED_MIN_MAX_ENABLED;
+import static
org.apache.parquet.format.converter.ParquetMetadataConverter.NO_FILTER;
+
public abstract class AbstractParquetScanBatchCreator {
private static final org.slf4j.Logger logger =
org.slf4j.LoggerFactory.getLogger(AbstractParquetScanBatchCreator.class);
@@ -146,11 +150,15 @@ protected ScanBatch getBatch(ExecutorFragmentContext
context, AbstractParquetRow
protected abstract AbstractDrillFileSystemManager
getDrillFileSystemCreator(OperatorContext operatorContext, OptionManager
optionManager);
private ParquetMetadata readFooter(Configuration conf, String path) throws
IOException {
- Configuration newConf = new Configuration(conf);
+ conf = new Configuration(conf);
conf.setBoolean(ENABLE_BYTES_READ_COUNTER, false);
conf.setBoolean(ENABLE_BYTES_TOTAL_COUNTER, false);
conf.setBoolean(ENABLE_TIME_READ_COUNTER, false);
- return ParquetFileReader.readFooter(newConf, new Path(path),
ParquetMetadataConverter.NO_FILTER);
+ conf.setBoolean(PARQUET_STRINGS_SIGNED_MIN_MAX_ENABLED, true);
+ ParquetReadOptions options =
ParquetReadOptions.builder().withMetadataFilter(NO_FILTER).build();
+ try (ParquetFileReader reader =
ParquetFileReader.open(HadoopInputFile.fromPath(new Path(path), conf),
options)) {
+ return reader.getFooter();
+ }
}
private boolean isComplex(ParquetMetadata footer) {
diff --git
a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ColumnDataReader.java
b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ColumnDataReader.java
index dcd40cf910..79294daea7 100644
---
a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ColumnDataReader.java
+++
b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ColumnDataReader.java
@@ -21,14 +21,13 @@
import java.io.IOException;
import java.io.OutputStream;
-import java.nio.ByteBuffer;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.parquet.bytes.BytesInput;
import org.apache.parquet.format.PageHeader;
import org.apache.parquet.format.Util;
-import org.apache.parquet.hadoop.util.CompatibilityUtil;
+import org.apache.parquet.hadoop.util.HadoopStreams;
public class ColumnDataReader {
static final org.slf4j.Logger logger =
org.slf4j.LoggerFactory.getLogger(ColumnDataReader.class);
@@ -58,11 +57,7 @@ public BytesInput getPageAsBytesInput(int pageLength) throws
IOException{
public void loadPage(DrillBuf target, int pageLength) throws IOException {
target.clear();
- ByteBuffer directBuffer = target.nioBuffer(0, pageLength);
- int lengthLeftToRead = pageLength;
- while (lengthLeftToRead > 0) {
- lengthLeftToRead -= CompatibilityUtil.getBuf(input, directBuffer,
lengthLeftToRead);
- }
+ HadoopStreams.wrap(input).read(target.nioBuffer(0, pageLength));
target.writerIndex(pageLength);
}
diff --git
a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetDirectByteBufferAllocator.java
b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetDirectByteBufferAllocator.java
index 09f1b26e24..ba6aac9431 100644
---
a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetDirectByteBufferAllocator.java
+++
b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetDirectByteBufferAllocator.java
@@ -17,10 +17,11 @@
*/
package org.apache.drill.exec.store.parquet;
-import io.netty.buffer.ByteBuf;
+import io.netty.buffer.DrillBuf;
import java.nio.ByteBuffer;
-import java.util.HashMap;
+import java.util.IdentityHashMap;
+import java.util.Map;
import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.ops.OperatorContext;
@@ -30,44 +31,42 @@
/**
* {@link ByteBufferAllocator} implementation that uses Drill's {@link
BufferAllocator} to allocate and release
* {@link ByteBuffer} objects.<br>
- * To properly release an allocated {@link ByteBuf}, this class keeps track of
it's corresponding {@link ByteBuffer}
+ * To properly release an allocated {@link DrillBuf}, this class keeps track
of it's corresponding {@link ByteBuffer}
* that was passed to the Parquet library.
*/
public class ParquetDirectByteBufferAllocator implements ByteBufferAllocator {
private static final org.slf4j.Logger logger =
org.slf4j.LoggerFactory.getLogger(ParquetDirectByteBufferAllocator.class);
private final BufferAllocator allocator;
- private final HashMap<Key, ByteBuf> allocatedBuffers = new HashMap<>();
+ private final Map<ByteBuffer, DrillBuf> allocatedBuffers = new
IdentityHashMap<>();
- public ParquetDirectByteBufferAllocator(OperatorContext o){
- allocator = o.getAllocator();
+ public ParquetDirectByteBufferAllocator(OperatorContext o) {
+ this(o.getAllocator());
}
public ParquetDirectByteBufferAllocator(BufferAllocator allocator) {
this.allocator = allocator;
}
-
@Override
public ByteBuffer allocate(int sz) {
- ByteBuf bb = allocator.buffer(sz);
- ByteBuffer b = bb.nioBuffer(0, sz);
- final Key key = new Key(b);
- allocatedBuffers.put(key, bb);
- logger.debug("ParquetDirectByteBufferAllocator: Allocated {} bytes.
Allocated ByteBuffer id: {}", sz, key.hash);
- return b;
+ DrillBuf drillBuf = allocator.buffer(sz);
+ ByteBuffer byteBuffer = drillBuf.nioBuffer(0, sz);
+ allocatedBuffers.put(byteBuffer, drillBuf);
+ logger.debug("{}: Allocated {} bytes. Allocated DrillBuf with id {} and
ByteBuffer {}", this, sz, drillBuf.getId(),
System.identityHashCode(byteBuffer));
+ return byteBuffer;
}
@Override
- public void release(ByteBuffer b) {
- final Key key = new Key(b);
- final ByteBuf bb = allocatedBuffers.get(key);
+ public void release(ByteBuffer byteBuffer) {
+ final DrillBuf drillBuf = allocatedBuffers.remove(byteBuffer);
// The ByteBuffer passed in may already have been freed or not allocated
by this allocator.
// If it is not found in the allocated buffers, do nothing
- if(bb != null) {
- logger.debug("ParquetDirectByteBufferAllocator: Freed byte buffer.
Allocated ByteBuffer id: {}", key.hash);
- bb.release();
- allocatedBuffers.remove(key);
+ if (drillBuf != null) {
+ logger.debug("{}: Freed DrillBuf with id {} and ByteBuffer {}", this,
drillBuf.getId(), System.identityHashCode(byteBuffer));
+ drillBuf.release();
+ } else {
+ logger.warn("{}: ByteBuffer {} is not present", this,
System.identityHashCode(byteBuffer));
}
}
@@ -75,41 +74,4 @@ public void release(ByteBuffer b) {
public boolean isDirect() {
return true;
}
-
- /**
- * ByteBuffer wrapper that computes a fixed hashcode.
- * <br><br>
- * Parquet only handles {@link ByteBuffer} objects, so we need to use them
as keys to keep track of their corresponding
- * {@link ByteBuf}, but {@link ByteBuffer} is mutable and it can't be used
as a {@link HashMap} key as it is.<br>
- * This class solves this by providing a fixed hashcode for {@link
ByteBuffer} and uses reference equality in case
- * of collisions (we don't need to compare the content of {@link ByteBuffer}
because the object passed to
- * {@link #release(ByteBuffer)} will be the same object returned from a
previous {@link #allocate(int)}.
- */
- private class Key {
- final int hash;
- final ByteBuffer buffer;
-
- Key(final ByteBuffer buffer) {
- this.buffer = buffer;
- // remember, we can't use buffer.hashCode()
- this.hash = System.identityHashCode(buffer);
- }
-
- @Override
- public int hashCode() {
- return hash;
- }
-
- @Override
- public boolean equals(Object obj) {
- if (this == obj) {
- return true;
- }
- if (!(obj instanceof Key)) {
- return false;
- }
- final Key key = (Key) obj;
- return hash == key.hash && buffer == key.buffer;
- }
- }
}
diff --git
a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetFilterBuilder.java
b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetFilterBuilder.java
index a8e101d6e0..5eb50153d0 100644
---
a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetFilterBuilder.java
+++
b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetFilterBuilder.java
@@ -272,22 +272,7 @@ private LogicalExpression
handleCompareFunction(FunctionHolderExpression functio
String funcName = ((DrillSimpleFuncHolder)
functionHolderExpression.getHolder()).getRegisteredNames()[0];
- switch (funcName) {
- case FunctionGenerationHelper.EQ :
- return new ParquetComparisonPredicates.EqualPredicate(newArgs.get(0),
newArgs.get(1));
- case FunctionGenerationHelper.GT :
- return new ParquetComparisonPredicates.GTPredicate(newArgs.get(0),
newArgs.get(1));
- case FunctionGenerationHelper.GE :
- return new ParquetComparisonPredicates.GEPredicate(newArgs.get(0),
newArgs.get(1));
- case FunctionGenerationHelper.LT :
- return new ParquetComparisonPredicates.LTPredicate(newArgs.get(0),
newArgs.get(1));
- case FunctionGenerationHelper.LE :
- return new ParquetComparisonPredicates.LEPredicate(newArgs.get(0),
newArgs.get(1));
- case FunctionGenerationHelper.NE :
- return new ParquetComparisonPredicates.NEPredicate(newArgs.get(0),
newArgs.get(1));
- default:
- return null;
- }
+ return ParquetComparisonPredicates.createPredicate(funcName,
newArgs.get(0), newArgs.get(1));
}
private LogicalExpression handleIsFunction(FunctionHolderExpression
functionHolderExpression, Set<LogicalExpression> value) {
diff --git
a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetRecordWriter.java
b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetRecordWriter.java
index 0e40c9e360..0917926608 100644
---
a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetRecordWriter.java
+++
b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetRecordWriter.java
@@ -54,8 +54,10 @@
import org.apache.hadoop.fs.Path;
import org.apache.parquet.bytes.CapacityByteArrayOutputStream;
import org.apache.parquet.column.ColumnWriteStore;
+import org.apache.parquet.column.ParquetProperties;
import org.apache.parquet.column.ParquetProperties.WriterVersion;
import org.apache.parquet.column.impl.ColumnWriteStoreV1;
+import org.apache.parquet.column.values.factory.DefaultV1ValuesWriterFactory;
import org.apache.parquet.hadoop.CodecFactory;
import org.apache.parquet.hadoop.ParquetColumnChunkPageWriteStore;
import org.apache.parquet.hadoop.ParquetFileWriter;
@@ -241,8 +243,15 @@ private void newSchema() throws IOException {
// once PARQUET-1006 will be resolved
pageStore = new
ParquetColumnChunkPageWriteStore(codecFactory.getCompressor(codec), schema,
initialSlabSize,
pageSize, new ParquetDirectByteBufferAllocator(oContext));
- store = new ColumnWriteStoreV1(pageStore, pageSize, initialPageBufferSize,
enableDictionary,
- writerVersion, new ParquetDirectByteBufferAllocator(oContext));
+ ParquetProperties parquetProperties = ParquetProperties.builder()
+ .withPageSize(pageSize)
+ .withDictionaryEncoding(enableDictionary)
+ .withDictionaryPageSize(initialPageBufferSize)
+ .withWriterVersion(writerVersion)
+ .withAllocator(new ParquetDirectByteBufferAllocator(oContext))
+ .withValuesWriterFactory(new DefaultV1ValuesWriterFactory())
+ .build();
+ store = new ColumnWriteStoreV1(pageStore, parquetProperties);
MessageColumnIO columnIO = new
ColumnIOFactory(false).getColumnIO(this.schema);
consumer = columnIO.getRecordWriter(store);
setUp(schema, consumer);
diff --git
a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/columnreaders/PageReader.java
b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/columnreaders/PageReader.java
index bf75695b6b..4aa2990508 100644
---
a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/columnreaders/PageReader.java
+++
b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/columnreaders/PageReader.java
@@ -17,6 +17,7 @@
*/
package org.apache.drill.exec.store.parquet.columnreaders;
+import com.google.common.base.Preconditions;
import com.google.common.base.Stopwatch;
import io.netty.buffer.ByteBufUtil;
import org.apache.drill.exec.util.filereader.BufferedDirectBufInputStream;
@@ -30,6 +31,7 @@
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.apache.parquet.bytes.ByteBufferInputStream;
import org.apache.parquet.bytes.BytesInput;
import org.apache.parquet.column.Dictionary;
import org.apache.parquet.column.Encoding;
@@ -250,7 +252,7 @@ private DrillBuf readPage(PageHeader pageHeader, int
compressedSize, int uncompr
}
public static BytesInput asBytesInput(DrillBuf buf, int offset, int length)
throws IOException {
- return BytesInput.from(buf.nioBuffer(offset, length), 0, length);
+ return BytesInput.from(buf.nioBuffer(offset, length));
}
@@ -319,41 +321,44 @@ public boolean next() throws IOException {
byteLength = pageHeader.uncompressed_page_size;
- final ByteBuffer pageDataBuffer = pageData.nioBuffer(0,
pageData.capacity());
+ final ByteBufferInputStream in =
ByteBufferInputStream.wrap(pageData.nioBuffer(0, pageData.capacity()));
readPosInBytes = 0;
if (parentColumnReader.getColumnDescriptor().getMaxRepetitionLevel() > 0) {
repetitionLevels =
rlEncoding.getValuesReader(parentColumnReader.columnDescriptor,
ValuesType.REPETITION_LEVEL);
- repetitionLevels.initFromPage(currentPageCount, pageDataBuffer, (int)
readPosInBytes);
+ repetitionLevels.initFromPage(currentPageCount, in);
// we know that the first value will be a 0, at the end of each list of
repeated values we will hit another 0 indicating
// a new record, although we don't know the length until we hit it (and
this is a one way stream of integers) so we
// read the first zero here to simplify the reading processes, and start
reading the first value the same as all
// of the rest. Effectively we are 'reading' the non-existent value in
front of the first allowing direct access to
// the first list of repetition levels
- readPosInBytes = repetitionLevels.getNextOffset();
+ readPosInBytes = in.position();
repetitionLevels.readInteger();
}
- if (parentColumnReader.columnDescriptor.getMaxDefinitionLevel() != 0){
+ if (parentColumnReader.columnDescriptor.getMaxDefinitionLevel() != 0) {
parentColumnReader.currDefLevel = -1;
definitionLevels =
dlEncoding.getValuesReader(parentColumnReader.columnDescriptor,
ValuesType.DEFINITION_LEVEL);
- definitionLevels.initFromPage(currentPageCount, pageDataBuffer, (int)
readPosInBytes);
- readPosInBytes = definitionLevels.getNextOffset();
+ definitionLevels.initFromPage(currentPageCount, in);
+ readPosInBytes = in.position();
if (!valueEncoding.usesDictionary()) {
valueReader =
valueEncoding.getValuesReader(parentColumnReader.columnDescriptor,
ValuesType.VALUES);
- valueReader.initFromPage(currentPageCount, pageDataBuffer, (int)
readPosInBytes);
+ valueReader.initFromPage(currentPageCount, in);
}
}
- if (parentColumnReader.columnDescriptor.getType() ==
PrimitiveType.PrimitiveTypeName.BOOLEAN) {
+ if (valueReader == null && parentColumnReader.columnDescriptor.getType()
== PrimitiveType.PrimitiveTypeName.BOOLEAN) {
valueReader =
valueEncoding.getValuesReader(parentColumnReader.columnDescriptor,
ValuesType.VALUES);
- valueReader.initFromPage(currentPageCount, pageDataBuffer, (int)
readPosInBytes);
+ valueReader.initFromPage(currentPageCount, in);
}
if (valueEncoding.usesDictionary()) {
// initialize two of the dictionary readers, one is for determining the
lengths of each value, the second is for
// actually copying the values out into the vectors
+ Preconditions.checkState(readPosInBytes < pageData.capacity());
+ int index = (int)readPosInBytes;
+ ByteBuffer byteBuffer = pageData.nioBuffer(index, pageData.capacity() -
index);
dictionaryLengthDeterminingReader = new
DictionaryValuesReader(dictionary);
- dictionaryLengthDeterminingReader.initFromPage(currentPageCount,
pageDataBuffer, (int) readPosInBytes);
+ dictionaryLengthDeterminingReader.initFromPage(currentPageCount,
ByteBufferInputStream.wrap(byteBuffer));
dictionaryValueReader = new DictionaryValuesReader(dictionary);
- dictionaryValueReader.initFromPage(currentPageCount, pageDataBuffer,
(int) readPosInBytes);
+ dictionaryValueReader.initFromPage(currentPageCount,
ByteBufferInputStream.wrap(byteBuffer));
parentColumnReader.usingDictionary = true;
} else {
parentColumnReader.usingDictionary = false;
@@ -445,25 +450,29 @@ public void clear(){
* @throws IOException An IO related condition
*/
void resetDefinitionLevelReader(int skipCount) throws IOException {
- if (parentColumnReader.columnDescriptor.getMaxDefinitionLevel() != 0) {
- throw new UnsupportedOperationException("Unsupoorted Operation");
- }
+
Preconditions.checkState(parentColumnReader.columnDescriptor.getMaxDefinitionLevel()
!= 0);
+ Preconditions.checkState(currentPageCount > 0);
+ final Encoding rlEncoding =
METADATA_CONVERTER.getEncoding(pageHeader.data_page_header.repetition_level_encoding);
final Encoding dlEncoding =
METADATA_CONVERTER.getEncoding(pageHeader.data_page_header.definition_level_encoding);
- final ByteBuffer pageDataBuffer = pageData.nioBuffer(0,
pageData.capacity());
- final int defStartPos = repetitionLevels != null ?
repetitionLevels.getNextOffset() : 0;
+
+ final ByteBufferInputStream in =
ByteBufferInputStream.wrap(pageData.nioBuffer(0, pageData.capacity()));
+
+ if (parentColumnReader.getColumnDescriptor().getMaxRepetitionLevel() > 0) {
+ repetitionLevels =
rlEncoding.getValuesReader(parentColumnReader.columnDescriptor,
ValuesType.REPETITION_LEVEL);
+ repetitionLevels.initFromPage(currentPageCount, in);
+ repetitionLevels.readInteger();
+ }
+
definitionLevels =
dlEncoding.getValuesReader(parentColumnReader.columnDescriptor,
ValuesType.DEFINITION_LEVEL);
parentColumnReader.currDefLevel = -1;
// Now reinitialize the underlying decoder
- assert currentPageCount > 0 : "Page count should be strictly upper than
zero";
- definitionLevels.initFromPage(currentPageCount, pageDataBuffer,
defStartPos);
+ definitionLevels.initFromPage(currentPageCount, in);
// Skip values if requested by caller
for (int idx = 0; idx < skipCount; ++idx) {
definitionLevels.skip();
}
}
-
-
}
diff --git
a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/Metadata.java
b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/Metadata.java
index ab655e9217..f39e9d1c0d 100644
---
a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/Metadata.java
+++
b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/Metadata.java
@@ -39,13 +39,13 @@
import org.apache.drill.exec.store.parquet.ParquetReaderUtility;
import org.apache.drill.exec.util.DrillFileSystemUtil;
import org.apache.drill.exec.util.ImpersonationUtil;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.parquet.column.statistics.Statistics;
-import org.apache.parquet.format.converter.ParquetMetadataConverter;
import org.apache.parquet.hadoop.ParquetFileReader;
import org.apache.parquet.hadoop.metadata.BlockMetaData;
import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
@@ -80,6 +80,7 @@
import static
org.apache.drill.exec.store.parquet.metadata.Metadata_V3.ParquetFileMetadata_v3;
import static
org.apache.drill.exec.store.parquet.metadata.Metadata_V3.ParquetTableMetadata_v3;
import static
org.apache.drill.exec.store.parquet.metadata.Metadata_V3.RowGroupMetadata_v3;
+import static
org.apache.parquet.format.converter.ParquetMetadataConverter.NO_FILTER;
public class Metadata {
private static final org.slf4j.Logger logger =
org.slf4j.LoggerFactory.getLogger(Metadata.class);
@@ -87,6 +88,7 @@
public static final String[] OLD_METADATA_FILENAMES =
{".drill.parquet_metadata.v2"};
public static final String METADATA_FILENAME = ".drill.parquet_metadata";
public static final String METADATA_DIRECTORIES_FILENAME =
".drill.parquet_metadata_directories";
+ public static final String PARQUET_STRINGS_SIGNED_MIN_MAX_ENABLED =
"parquet.strings.signed-min-max.enabled";
private final ParquetFormatConfig formatConfig;
@@ -409,9 +411,11 @@ private ParquetFileMetadata_v3
getParquetFileMetadata_v3(ParquetTableMetadata_v3
final FileStatus file, final FileSystem fs) throws IOException,
InterruptedException {
final ParquetMetadata metadata;
final UserGroupInformation processUserUgi =
ImpersonationUtil.getProcessUserUGI();
+ final Configuration conf = new Configuration(fs.getConf());
+ conf.setBoolean(PARQUET_STRINGS_SIGNED_MIN_MAX_ENABLED, true);
try {
metadata =
processUserUgi.doAs((PrivilegedExceptionAction<ParquetMetadata>)
- () -> ParquetFileReader.readFooter(fs.getConf(), file,
ParquetMetadataConverter.NO_FILTER));
+ () -> ParquetFileReader.readFooter(conf, file, NO_FILTER));
} catch(Exception e) {
logger.error("Exception while reading footer of parquet file [Details -
path: {}, owner: {}] as process user {}",
file.getPath(), file.getOwner(), processUserUgi.getShortUserName(), e);
diff --git
a/exec/java-exec/src/main/java/org/apache/drill/exec/util/filereader/BufferedDirectBufInputStream.java
b/exec/java-exec/src/main/java/org/apache/drill/exec/util/filereader/BufferedDirectBufInputStream.java
index f208d6e823..1d764b1b6d 100644
---
a/exec/java-exec/src/main/java/org/apache/drill/exec/util/filereader/BufferedDirectBufInputStream.java
+++
b/exec/java-exec/src/main/java/org/apache/drill/exec/util/filereader/BufferedDirectBufInputStream.java
@@ -21,7 +21,7 @@
import com.google.common.base.Stopwatch;
import io.netty.buffer.DrillBuf;
import org.apache.drill.exec.memory.BufferAllocator;
-import org.apache.parquet.hadoop.util.CompatibilityUtil;
+import org.apache.parquet.hadoop.util.HadoopStreams;
import java.io.Closeable;
import java.io.IOException;
@@ -179,7 +179,7 @@ private int getNextBlock() throws IOException {
int nBytes = 0;
if (bytesToRead > 0) {
try {
- nBytes = CompatibilityUtil.getBuf(getInputStream(), directBuffer,
bytesToRead);
+ nBytes = HadoopStreams.wrap(getInputStream()).read(directBuffer);
} catch (Exception e) {
logger.error("Error reading from stream {}. Error was : {}",
this.streamId, e.getMessage());
throw new IOException((e));
@@ -193,8 +193,8 @@ private int getNextBlock() throws IOException {
logger.trace(
"PERF: Disk read complete. {}, StartOffset: {}, TotalByteSize:
{}, BufferSize: {}, BytesRead: {}, Count: {}, "
+ "CurPosInStream: {}, CurPosInBuffer: {}, Time: {} ms",
this.streamId, this.startOffset,
- this.totalByteSize, this.bufSize, bytesRead, this.count,
this.curPosInStream, this.curPosInBuffer, ((double)
timer.elapsed(TimeUnit.MICROSECONDS))
- / 1000);
+ this.totalByteSize, this.bufSize, bytesRead, this.count,
this.curPosInStream, this.curPosInBuffer,
+ ((double) timer.elapsed(TimeUnit.MICROSECONDS)) / 1000);
}
}
}
diff --git
a/exec/java-exec/src/main/java/org/apache/drill/exec/util/filereader/DirectBufInputStream.java
b/exec/java-exec/src/main/java/org/apache/drill/exec/util/filereader/DirectBufInputStream.java
index ae09a3708c..ea2542eb80 100644
---
a/exec/java-exec/src/main/java/org/apache/drill/exec/util/filereader/DirectBufInputStream.java
+++
b/exec/java-exec/src/main/java/org/apache/drill/exec/util/filereader/DirectBufInputStream.java
@@ -23,7 +23,8 @@
import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.hadoop.fs.ByteBufferReadable;
import org.apache.hadoop.fs.FSDataInputStream;
-import org.apache.parquet.hadoop.util.CompatibilityUtil;
+import org.apache.parquet.hadoop.util.HadoopStreams;
+import org.apache.parquet.io.SeekableInputStream;
import java.io.FilterInputStream;
import java.io.IOException;
@@ -86,12 +87,16 @@ public synchronized int read(DrillBuf buf, int off, int
len) throws IOException
buf.clear();
ByteBuffer directBuffer = buf.nioBuffer(0, len);
int lengthLeftToRead = len;
+ SeekableInputStream seekableInputStream =
HadoopStreams.wrap(getInputStream());
while (lengthLeftToRead > 0) {
if(logger.isTraceEnabled()) {
logger.trace("PERF: Disk read start. {}, StartOffset: {},
TotalByteSize: {}", this.streamId, this.startOffset, this.totalByteSize);
}
Stopwatch timer = Stopwatch.createStarted();
- int bytesRead = CompatibilityUtil.getBuf(getInputStream(), directBuffer,
lengthLeftToRead);
+ int bytesRead = seekableInputStream.read(directBuffer);
+ if (bytesRead < 0) {
+ return bytesRead;
+ }
lengthLeftToRead -= bytesRead;
if(logger.isTraceEnabled()) {
logger.trace(
@@ -113,7 +118,7 @@ public synchronized DrillBuf getNext(int bytes) throws
IOException {
b.release();
throw e;
}
- if (bytesRead <= -1) {
+ if (bytesRead < 0) {
b.release();
return null;
}
diff --git
a/exec/java-exec/src/main/java/org/apache/parquet/hadoop/ColumnChunkIncReadStore.java
b/exec/java-exec/src/main/java/org/apache/parquet/hadoop/ColumnChunkIncReadStore.java
index 6e9db7e947..89731ff2a4 100644
---
a/exec/java-exec/src/main/java/org/apache/parquet/hadoop/ColumnChunkIncReadStore.java
+++
b/exec/java-exec/src/main/java/org/apache/parquet/hadoop/ColumnChunkIncReadStore.java
@@ -46,7 +46,7 @@
import org.apache.parquet.format.converter.ParquetMetadataConverter;
import org.apache.parquet.hadoop.CodecFactory.BytesDecompressor;
import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
-import org.apache.parquet.hadoop.util.CompatibilityUtil;
+import org.apache.parquet.hadoop.util.HadoopStreams;
import io.netty.buffer.ByteBuf;
@@ -163,12 +163,10 @@ public DataPage readPage() {
ByteBuf buf = allocator.buffer(pageHeader.compressed_page_size);
lastPage = buf;
ByteBuffer buffer = buf.nioBuffer(0,
pageHeader.compressed_page_size);
- int lengthLeftToRead = pageHeader.compressed_page_size;
- while (lengthLeftToRead > 0) {
- lengthLeftToRead -= CompatibilityUtil.getBuf(in, buffer,
lengthLeftToRead);
- }
+ HadoopStreams.wrap(in).readFully(buffer);
+ buffer.flip();
return new DataPageV1(
- decompressor.decompress(BytesInput.from(buffer, 0,
pageHeader.compressed_page_size), pageHeader.getUncompressed_page_size()),
+ decompressor.decompress(BytesInput.from(buffer),
pageHeader.getUncompressed_page_size()),
pageHeader.data_page_header.num_values,
pageHeader.uncompressed_page_size,
fromParquetStatistics(pageHeader.data_page_header.statistics,
columnDescriptor.getType()),
@@ -182,28 +180,33 @@ public DataPage readPage() {
buf = allocator.buffer(pageHeader.compressed_page_size);
lastPage = buf;
buffer = buf.nioBuffer(0, pageHeader.compressed_page_size);
- lengthLeftToRead = pageHeader.compressed_page_size;
- while (lengthLeftToRead > 0) {
- lengthLeftToRead -= CompatibilityUtil.getBuf(in, buffer,
lengthLeftToRead);
- }
+ HadoopStreams.wrap(in).readFully(buffer);
+ buffer.flip();
DataPageHeaderV2 dataHeaderV2 =
pageHeader.getData_page_header_v2();
int dataSize = compressedPageSize -
dataHeaderV2.getRepetition_levels_byte_length() -
dataHeaderV2.getDefinition_levels_byte_length();
BytesInput decompressedPageData =
decompressor.decompress(
- BytesInput.from(buffer, 0,
pageHeader.compressed_page_size),
+ BytesInput.from(buffer),
pageHeader.uncompressed_page_size);
+ ByteBuffer byteBuffer = decompressedPageData.toByteBuffer();
+ int limit = byteBuffer.limit();
+
byteBuffer.limit(dataHeaderV2.getRepetition_levels_byte_length());
+ BytesInput repetitionLevels =
BytesInput.from(byteBuffer.slice());
+
byteBuffer.position(dataHeaderV2.getRepetition_levels_byte_length());
+ byteBuffer.limit(dataHeaderV2.getRepetition_levels_byte_length()
+ dataHeaderV2.getDefinition_levels_byte_length());
+ BytesInput definitionLevels =
BytesInput.from(byteBuffer.slice());
+
byteBuffer.position(dataHeaderV2.getRepetition_levels_byte_length() +
dataHeaderV2.getDefinition_levels_byte_length());
+ byteBuffer.limit(limit);
+ BytesInput data = BytesInput.from(byteBuffer.slice());
+
return new DataPageV2(
dataHeaderV2.getNum_rows(),
dataHeaderV2.getNum_nulls(),
dataHeaderV2.getNum_values(),
- BytesInput.from(decompressedPageData.toByteBuffer(), 0,
dataHeaderV2.getRepetition_levels_byte_length()),
- BytesInput.from(decompressedPageData.toByteBuffer(),
- dataHeaderV2.getRepetition_levels_byte_length(),
- dataHeaderV2.getDefinition_levels_byte_length()),
+ repetitionLevels,
+ definitionLevels,
parquetMetadataConverter.getEncoding(dataHeaderV2.getEncoding()),
- BytesInput.from(decompressedPageData.toByteBuffer(),
- dataHeaderV2.getRepetition_levels_byte_length() +
dataHeaderV2.getDefinition_levels_byte_length(),
- dataSize),
+ data,
uncompressedPageSize,
fromParquetStatistics(dataHeaderV2.getStatistics(),
columnDescriptor.getType()),
dataHeaderV2.isIs_compressed()
diff --git
a/exec/java-exec/src/main/java/org/apache/parquet/hadoop/ParquetColumnChunkPageWriteStore.java
b/exec/java-exec/src/main/java/org/apache/parquet/hadoop/ParquetColumnChunkPageWriteStore.java
index 93f9920bd9..0ed2245250 100644
---
a/exec/java-exec/src/main/java/org/apache/parquet/hadoop/ParquetColumnChunkPageWriteStore.java
+++
b/exec/java-exec/src/main/java/org/apache/parquet/hadoop/ParquetColumnChunkPageWriteStore.java
@@ -17,8 +17,6 @@
*/
package org.apache.parquet.hadoop;
-import static
org.apache.parquet.column.statistics.Statistics.getStatsBasedOnType;
-
import java.io.Closeable;
import java.io.IOException;
import java.util.List;
@@ -119,7 +117,7 @@ private ColumnChunkPageWriter(ColumnDescriptor path,
this.path = path;
this.compressor = compressor;
this.buf = new CapacityByteArrayOutputStream(initialSlabSize,
maxCapacityHint, allocator);
- this.totalStatistics = getStatsBasedOnType(this.path.getType());
+ this.totalStatistics =
Statistics.createStats(this.path.getPrimitiveType());
}
@Override
@@ -226,11 +224,7 @@ public void writeToFileWriter(ParquetFileWriter writer)
throws IOException {
writer.writeDictionaryPage(dictionaryPage);
// tracking the dictionary encoding is handled in writeDictionaryPage
}
- List<Encoding> encodings = Lists.newArrayList();
- encodings.addAll(rlEncodings);
- encodings.addAll(dlEncodings);
- encodings.addAll(dataEncodings);
- writer.writeDataPages(BytesInput.from(buf), uncompressedLength,
compressedLength, totalStatistics, encodings);
+ writer.writeDataPages(BytesInput.from(buf), uncompressedLength,
compressedLength, totalStatistics, rlEncodings, dlEncodings, dataEncodings);
writer.endColumn();
logger.debug(
String.format(
diff --git
a/exec/java-exec/src/test/java/org/apache/drill/exec/store/parquet/TestParquetMetadataCache.java
b/exec/java-exec/src/test/java/org/apache/drill/exec/store/parquet/TestParquetMetadataCache.java
index 50e679ad64..c2d3608a57 100644
---
a/exec/java-exec/src/test/java/org/apache/drill/exec/store/parquet/TestParquetMetadataCache.java
+++
b/exec/java-exec/src/test/java/org/apache/drill/exec/store/parquet/TestParquetMetadataCache.java
@@ -30,6 +30,7 @@
import org.apache.drill.exec.planner.physical.PlannerSettings;
import org.junit.Assert;
import org.junit.BeforeClass;
+import org.junit.Ignore;
import org.junit.Test;
import org.junit.experimental.categories.Category;
@@ -737,6 +738,7 @@ public void testBooleanPartitionPruning() throws Exception {
}
}
+ @Ignore
@Test // DRILL-4139
public void testIntervalDayPartitionPruning() throws Exception {
final String intervalDayPartitionTable =
"dfs.tmp.`interval_day_partition`";
@@ -762,6 +764,7 @@ public void testIntervalDayPartitionPruning() throws
Exception {
}
}
+ @Ignore
@Test // DRILL-4139
public void testIntervalYearPartitionPruning() throws Exception {
final String intervalYearPartitionTable =
"dfs.tmp.`interval_yr_partition`";
@@ -812,6 +815,7 @@ public void testVarCharWithNullsPartitionPruning() throws
Exception {
}
}
+ @Ignore
@Test // DRILL-4139
public void testDecimalPartitionPruning() throws Exception {
List<String> ctasQueries = Lists.newArrayList();
diff --git a/pom.xml b/pom.xml
index 206af7e453..64f78ac225 100644
--- a/pom.xml
+++ b/pom.xml
@@ -44,7 +44,7 @@
<dep.slf4j.version>1.7.6</dep.slf4j.version>
<dep.guava.version>18.0</dep.guava.version>
<forkCount>2</forkCount>
- <parquet.version>1.8.1-drill-r0</parquet.version>
+ <parquet.version>1.10.0</parquet.version>
<calcite.version>1.16.0-drill-r3</calcite.version>
<avatica.version>1.11.0</avatica.version>
<janino.version>2.7.6</janino.version>
@@ -1394,6 +1394,36 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.parquet</groupId>
+ <artifactId>parquet-format</artifactId>
+ <version>2.5.0</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.parquet</groupId>
+ <artifactId>parquet-common</artifactId>
+ <version>${parquet.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
</dependencies>
</dependencyManagement>
@@ -1912,6 +1942,14 @@
<artifactId>parquet-hadoop</artifactId>
<version>${parquet.version}</version>
<exclusions>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ </exclusion>
<exclusion>
<groupId>org.xerial.snappy</groupId>
<artifactId>snappy-java</artifactId>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
> Upgrade Parquet MR dependencies
> -------------------------------
>
> Key: DRILL-6353
> URL: https://issues.apache.org/jira/browse/DRILL-6353
> Project: Apache Drill
> Issue Type: Task
> Reporter: Vlad Rozov
> Assignee: Vlad Rozov
> Priority: Major
> Fix For: 1.14.0
>
>
> Upgrade from a custom build {{1.8.1-drill-r0}} to Apache release {{1.10.0}}.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
