Add Optional predicate to ParquetReader constructor used in Iceberg

Add requireNonNull on parquetPredicate and columnIndexStore in ParquetReader, pass Optional.empty() from Iceberg for parquetPredicate param
trinodb · Sep 7, 2022 · c72e3ba · c72e3ba
1 parent df4f4c8
commit c72e3ba
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 8 deletions.
diff --git a/lib/trino-parquet/src/main/java/io/trino/parquet/reader/ParquetReader.java b/lib/trino-parquet/src/main/java/io/trino/parquet/reader/ParquetReader.java
@@ -76,6 +76,7 @@
 import static java.lang.Math.max;
 import static java.lang.Math.min;
 import static java.lang.Math.toIntExact;
+import static java.util.Collections.nCopies;
 import static java.util.Objects.requireNonNull;
 
 public class ParquetReader
@@ -130,10 +131,11 @@ public ParquetReader(
             ParquetDataSource dataSource,
             DateTimeZone timeZone,
             AggregatedMemoryContext memoryContext,
-            ParquetReaderOptions options)
+            ParquetReaderOptions options,
+            Optional<Predicate> parquetPredicate)
             throws IOException
     {
-        this(fileCreatedBy, fields, blocks, firstRowsOfBlocks, dataSource, timeZone, memoryContext, options, null, null);
+        this(fileCreatedBy, fields, blocks, firstRowsOfBlocks, dataSource, timeZone, memoryContext, options, parquetPredicate, nCopies(blocks.size(), Optional.empty()));
     }
 
     public ParquetReader(
@@ -145,7 +147,7 @@ public ParquetReader(
             DateTimeZone timeZone,
             AggregatedMemoryContext memoryContext,
             ParquetReaderOptions options,
-            Predicate parquetPredicate,
+            Optional<Predicate> parquetPredicate,
             List<Optional<ColumnIndexStore>> columnIndexStore)
             throws IOException
     {
@@ -164,14 +166,16 @@ public ParquetReader(
 
         checkArgument(blocks.size() == firstRowsOfBlocks.size(), "elements of firstRowsOfBlocks must correspond to blocks");
 
-        this.columnIndexStore = columnIndexStore;
         this.blockRowRanges = listWithNulls(this.blocks.size());
         for (PrimitiveField field : primitiveFields) {
             ColumnDescriptor columnDescriptor = field.getDescriptor();
             this.paths.put(ColumnPath.get(columnDescriptor.getPath()), columnDescriptor);
         }
-        if (parquetPredicate != null && options.isUseColumnIndex()) {
-            this.filter = parquetPredicate.toParquetFilter(timeZone);
+
+        requireNonNull(parquetPredicate, "parquetPredicate is null");
+        this.columnIndexStore = requireNonNull(columnIndexStore, "columnIndexStore is null");
+        if (parquetPredicate.isPresent() && options.isUseColumnIndex()) {
+            this.filter = parquetPredicate.get().toParquetFilter(timeZone);
         }
         else {
             this.filter = Optional.empty();

diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/parquet/ParquetPageSourceFactory.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/parquet/ParquetPageSourceFactory.java
@@ -285,7 +285,7 @@ && predicateMatches(parquetPredicate, block, dataSource, descriptorsByPath, parq
                     timeZone,
                     newSimpleAggregatedMemoryContext(),
                     options,
-                    parquetPredicate,
+                    Optional.of(parquetPredicate),
                     columnIndexes.build());
 
             ConnectorPageSource parquetPageSource = new ParquetPageSource(

diff --git a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergPageSourceProvider.java b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergPageSourceProvider.java
@@ -1050,7 +1050,8 @@ else if (column.getId() == TRINO_MERGE_PARTITION_DATA) {
                     dataSource,
                     UTC,
                     memoryContext,
-                    options);
+                    options,
+                    Optional.empty());
 
             return new ReaderPageSourceWithRowPositions(
                     new ReaderPageSource(