Parquet to 1.10.1

Author: wangyum

dev/deps/spark-deps-hadoop-2.7-hive-2.3

@@ -203,12 +203,12 @@ orc-shims/1.5.12//orc-shims-1.5.12.jar
 oro/2.0.8//oro-2.0.8.jar
 osgi-resource-locator/1.0.3//osgi-resource-locator-1.0.3.jar
 paranamer/2.8//paranamer-2.8.jar
-parquet-column/1.11.1//parquet-column-1.11.1.jar
-parquet-common/1.11.1//parquet-common-1.11.1.jar
-parquet-encoding/1.11.1//parquet-encoding-1.11.1.jar
-parquet-format-structures/1.11.1//parquet-format-structures-1.11.1.jar
-parquet-hadoop/1.11.1//parquet-hadoop-1.11.1.jar
-parquet-jackson/1.11.1//parquet-jackson-1.11.1.jar
+parquet-column/1.10.1//parquet-column-1.10.1.jar
+parquet-common/1.10.1//parquet-common-1.10.1.jar
+parquet-encoding/1.10.1//parquet-encoding-1.10.1.jar
+parquet-format/2.4.0//parquet-format-2.4.0.jar
+parquet-hadoop/1.10.1//parquet-hadoop-1.10.1.jar
+parquet-jackson/1.10.1//parquet-jackson-1.10.1.jar
 protobuf-java/2.5.0//protobuf-java-2.5.0.jar
 py4j/0.10.9//py4j-0.10.9.jar
 pyrolite/4.30//pyrolite-4.30.jar

dev/deps/spark-deps-hadoop-3.2-hive-2.3

@@ -173,12 +173,12 @@ orc-shims/1.5.12//orc-shims-1.5.12.jar
 oro/2.0.8//oro-2.0.8.jar
 osgi-resource-locator/1.0.3//osgi-resource-locator-1.0.3.jar
 paranamer/2.8//paranamer-2.8.jar
-parquet-column/1.11.1//parquet-column-1.11.1.jar
-parquet-common/1.11.1//parquet-common-1.11.1.jar
-parquet-encoding/1.11.1//parquet-encoding-1.11.1.jar
-parquet-format-structures/1.11.1//parquet-format-structures-1.11.1.jar
-parquet-hadoop/1.11.1//parquet-hadoop-1.11.1.jar
-parquet-jackson/1.11.1//parquet-jackson-1.11.1.jar
+parquet-column/1.10.1//parquet-column-1.10.1.jar
+parquet-common/1.10.1//parquet-common-1.10.1.jar
+parquet-encoding/1.10.1//parquet-encoding-1.10.1.jar
+parquet-format/2.4.0//parquet-format-2.4.0.jar
+parquet-hadoop/1.10.1//parquet-hadoop-1.10.1.jar
+parquet-jackson/1.10.1//parquet-jackson-1.10.1.jar
 protobuf-java/2.5.0//protobuf-java-2.5.0.jar
 py4j/0.10.9//py4j-0.10.9.jar
 pyrolite/4.30//pyrolite-4.30.jar

pom.xml

@@ -135,7 +135,7 @@
     <!-- note that this should be compatible with Kafka brokers version 0.10 and up -->
     <kafka.version>2.6.0</kafka.version>
     <derby.version>10.12.1.1</derby.version>
-    <parquet.version>1.11.1</parquet.version>
+    <parquet.version>1.10.1</parquet.version>
     <orc.version>1.5.12</orc.version>
     <jetty.version>9.4.28.v20200408</jetty.version>
     <javaxservlet.version>3.1.0</javaxservlet.version>

sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/SpecificParquetRecordReaderBase.java

@@ -146,7 +146,10 @@ public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptCont
     this.sparkSchema = StructType$.MODULE$.fromString(sparkRequestedSchemaString);
     this.reader = new ParquetFileReader(
         configuration, footer.getFileMetaData(), file, blocks, requestedSchema.getColumns());
-    this.totalRowCount += reader.getFilteredRecordCount();
+    // use the blocks from the reader in case some do not match filters and will not be read
+    for (BlockMetaData block : reader.getRowGroups()) {
+      this.totalRowCount += block.getRowCount();
+    }
 
     // For test purpose.
     // If the last external accumulator is `NumRowGroupsAccumulator`, the row group number to read
@@ -222,7 +225,10 @@ protected void initialize(String path, List<String> columns) throws IOException
     this.sparkSchema = new ParquetToSparkSchemaConverter(config).convert(requestedSchema);
     this.reader = new ParquetFileReader(
         config, footer.getFileMetaData(), file, blocks, requestedSchema.getColumns());
-    this.totalRowCount += reader.getFilteredRecordCount();
+    // use the blocks from the reader in case some do not match filters and will not be read
+    for (BlockMetaData block : reader.getRowGroups()) {
+      this.totalRowCount += block.getRowCount();
+    }
   }
 
   @Override

sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java

@@ -320,7 +320,7 @@ private void initializeInternal() throws IOException, UnsupportedOperationExcept
 
   private void checkEndOfRowGroup() throws IOException {
     if (rowsReturned != totalCountLoadedSoFar) return;
-    PageReadStore pages = reader.readNextFilteredRowGroup();
+    PageReadStore pages = reader.readNextRowGroup();
     if (pages == null) {
       throw new IOException("expecting more rows but reached last block. Read "
           + rowsReturned + " out of " + totalRowCount);

sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala

@@ -214,7 +214,7 @@ class StreamSuite extends StreamTest {
             .start(outputDir.getAbsolutePath)
           try {
             query.processAllAvailable()
-            val outputDf = spark.read.parquet(outputDir.getAbsolutePath).sort('a).as[Long]
+            val outputDf = spark.read.parquet(outputDir.getAbsolutePath).as[Long]
             checkDataset[Long](outputDf, (0L to 10L).toArray: _*)
           } finally {
             query.stop()