PARQUET-2417: Add statistics support to geometry logical type (#2971)

Author: zhangfengcdt

parquet-cli/README.md

@@ -121,6 +121,8 @@ Usage: parquet [options] [command] [command options]
         Rewrite one or more Parquet files to a new Parquet file
     size-stats
         Print size statistics for a Parquet file
+    geospatial-stats
+        Print geospatial statistics for a Parquet file
 
   Examples:
 

parquet-cli/src/main/java/org/apache/parquet/cli/Main.java

@@ -50,6 +50,7 @@
 import org.apache.parquet.cli.commands.ShowColumnIndexCommand;
 import org.apache.parquet.cli.commands.ShowDictionaryCommand;
 import org.apache.parquet.cli.commands.ShowFooterCommand;
+import org.apache.parquet.cli.commands.ShowGeospatialStatisticsCommand;
 import org.apache.parquet.cli.commands.ShowPagesCommand;
 import org.apache.parquet.cli.commands.ShowSizeStatisticsCommand;
 import org.apache.parquet.cli.commands.ToAvroCommand;
@@ -107,6 +108,7 @@ public class Main extends Configured implements Tool {
     jc.addCommand("scan", new ScanCommand(console));
     jc.addCommand("rewrite", new RewriteCommand(console));
     jc.addCommand("size-stats", new ShowSizeStatisticsCommand(console));
+    jc.addCommand("geospatial-stats", new ShowGeospatialStatisticsCommand(console));
   }
 
   @Override

parquet-cli/src/main/java/org/apache/parquet/cli/commands/ShowGeospatialStatisticsCommand.java

@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.parquet.cli.commands;
+
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.Parameters;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import java.io.IOException;
+import java.util.List;
+import org.apache.commons.text.TextStringBuilder;
+import org.apache.parquet.cli.BaseCommand;
+import org.apache.parquet.column.statistics.geospatial.GeospatialStatistics;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.metadata.BlockMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.parquet.schema.MessageType;
+import org.slf4j.Logger;
+
+@Parameters(commandDescription = "Print geospatial statistics for a Parquet file")
+public class ShowGeospatialStatisticsCommand extends BaseCommand {
+
+  public ShowGeospatialStatisticsCommand(Logger console) {
+    super(console);
+  }
+
+  @Parameter(description = "<parquet path>")
+  List<String> targets;
+
+  @Override
+  @SuppressWarnings("unchecked")
+  public int run() throws IOException {
+    Preconditions.checkArgument(targets != null && !targets.isEmpty(), "A Parquet file is required.");
+    Preconditions.checkArgument(targets.size() == 1, "Cannot process multiple Parquet files.");
+
+    String source = targets.get(0);
+    try (ParquetFileReader reader = ParquetFileReader.open(getConf(), qualifiedPath(source))) {
+      ParquetMetadata footer = reader.getFooter();
+      MessageType schema = footer.getFileMetaData().getSchema();
+
+      console.info("\nFile path: {}", source);
+
+      List<BlockMetaData> rowGroups = footer.getBlocks();
+      for (int index = 0, n = rowGroups.size(); index < n; index++) {
+        printRowGroupGeospatialStats(console, index, rowGroups.get(index), schema);
+        console.info("");
+      }
+    }
+
+    return 0;
+  }
+
+  private void printRowGroupGeospatialStats(Logger console, int index, BlockMetaData rowGroup, MessageType schema) {
+    int maxColumnWidth = Math.max(
+        "column".length(),
+        rowGroup.getColumns().stream()
+            .map(col -> col.getPath().toString().length())
+            .max(Integer::compare)
+            .orElse(0));
+
+    console.info(String.format("\nRow group %d\n%s", index, new TextStringBuilder(80).appendPadding(80, '-')));
+
+    String formatString = String.format("%%-%ds %%-15s %%-40s", maxColumnWidth);
+    console.info(String.format(formatString, "column", "bounding box", "geospatial types"));
+
+    for (ColumnChunkMetaData column : rowGroup.getColumns()) {
+      printColumnGeospatialStats(console, column, schema, maxColumnWidth);
+    }
+  }
+
+  private void printColumnGeospatialStats(
+      Logger console, ColumnChunkMetaData column, MessageType schema, int columnWidth) {
+    GeospatialStatistics stats = column.getGeospatialStatistics();
+
+    if (stats != null && stats.isValid()) {
+      String boundingBox =
+          stats.getBoundingBox() != null ? stats.getBoundingBox().toString() : "-";
+      String geospatialTypes = stats.getGeospatialTypes() != null
+          ? stats.getGeospatialTypes().toString()
+          : "-";
+      String formatString = String.format("%%-%ds %%-15s %%-40s", columnWidth);
+      console.info(String.format(formatString, column.getPath(), boundingBox, geospatialTypes));
+    } else {
+      String formatString = String.format("%%-%ds %%-15s %%-40s", columnWidth);
+      console.info(String.format(formatString, column.getPath(), "-", "-"));
+    }
+  }
+
+  @Override
+  public List<String> getExamples() {
+    return Lists.newArrayList("# Show geospatial statistics for a Parquet file", "sample.parquet");
+  }
+}

parquet-cli/src/test/java/org/apache/parquet/cli/commands/ShowGeospatialStatisticsCommandTest.java

@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.parquet.cli.commands;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+import org.apache.hadoop.conf.Configuration;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class ShowGeospatialStatisticsCommandTest extends ParquetFileTest {
+  @Test
+  public void testShowGeospatialStatisticsCommand() throws IOException {
+    File file = parquetFile();
+    ShowGeospatialStatisticsCommand command = new ShowGeospatialStatisticsCommand(createLogger());
+    command.targets = Arrays.asList(file.getAbsolutePath());
+    command.setConf(new Configuration());
+    Assert.assertEquals(0, command.run());
+  }
+}

parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnValueCollector.java

@@ -26,6 +26,7 @@
 import org.apache.parquet.column.ParquetProperties;
 import org.apache.parquet.column.statistics.SizeStatistics;
 import org.apache.parquet.column.statistics.Statistics;
+import org.apache.parquet.column.statistics.geospatial.GeospatialStatistics;
 import org.apache.parquet.column.values.bloomfilter.AdaptiveBlockSplitBloomFilter;
 import org.apache.parquet.column.values.bloomfilter.BlockSplitBloomFilter;
 import org.apache.parquet.column.values.bloomfilter.BloomFilter;
@@ -42,6 +43,7 @@ class ColumnValueCollector {
   private BloomFilter bloomFilter;
   private Statistics<?> statistics;
   private SizeStatistics.Builder sizeStatisticsBuilder;
+  private GeospatialStatistics.Builder geospatialStatisticsBuilder;
 
   ColumnValueCollector(ColumnDescriptor path, BloomFilterWriter bloomFilterWriter, ParquetProperties props) {
     this.path = path;
@@ -60,6 +62,9 @@ void resetPageStatistics() {
             path.getPrimitiveType(), path.getMaxRepetitionLevel(), path.getMaxDefinitionLevel())
         : SizeStatistics.noopBuilder(
             path.getPrimitiveType(), path.getMaxRepetitionLevel(), path.getMaxDefinitionLevel());
+    this.geospatialStatisticsBuilder = statisticsEnabled
+        ? GeospatialStatistics.newBuilder(path.getPrimitiveType())
+        : GeospatialStatistics.noopBuilder();
   }
 
   void writeNull(int repetitionLevel, int definitionLevel) {
@@ -99,6 +104,7 @@ void write(double value, int repetitionLevel, int definitionLevel) {
   void write(Binary value, int repetitionLevel, int definitionLevel) {
     statistics.updateStats(value);
     sizeStatisticsBuilder.add(repetitionLevel, definitionLevel, value);
+    geospatialStatisticsBuilder.update(value);
     bloomFilter.insertHash(bloomFilter.hash(value));
   }
 
@@ -199,4 +205,8 @@ Statistics<?> getStatistics() {
   SizeStatistics getSizeStatistics() {
     return sizeStatisticsBuilder.build();
   }
+
+  GeospatialStatistics getGeospatialStatistics() {
+    return geospatialStatisticsBuilder.build();
+  }
 }

parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriterBase.java

@@ -27,6 +27,7 @@
 import org.apache.parquet.column.page.PageWriter;
 import org.apache.parquet.column.statistics.SizeStatistics;
 import org.apache.parquet.column.statistics.Statistics;
+import org.apache.parquet.column.statistics.geospatial.GeospatialStatistics;
 import org.apache.parquet.column.values.ValuesWriter;
 import org.apache.parquet.column.values.bloomfilter.BloomFilterWriter;
 import org.apache.parquet.io.ParquetEncodingException;
@@ -380,6 +381,7 @@ void writePage() {
             valueCount,
             collector.getStatistics(),
             collector.getSizeStatistics(),
+            collector.getGeospatialStatistics(),
             repetitionLevelColumn,
             definitionLevelColumn,
             dataColumn);
@@ -403,6 +405,7 @@ abstract void writePage(
       int valueCount,
       Statistics<?> statistics,
       SizeStatistics sizeStatistics,
+      GeospatialStatistics geospatialStatistics,
       ValuesWriter repetitionLevels,
       ValuesWriter definitionLevels,
       ValuesWriter values)

parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriterV1.java

@@ -26,6 +26,7 @@
 import org.apache.parquet.column.page.PageWriter;
 import org.apache.parquet.column.statistics.SizeStatistics;
 import org.apache.parquet.column.statistics.Statistics;
+import org.apache.parquet.column.statistics.geospatial.GeospatialStatistics;
 import org.apache.parquet.column.values.ValuesWriter;
 import org.apache.parquet.column.values.bloomfilter.BloomFilterWriter;
 
@@ -62,6 +63,7 @@ void writePage(
       int valueCount,
       Statistics<?> statistics,
       SizeStatistics sizeStatistics,
+      GeospatialStatistics geospatialStatistics,
       ValuesWriter repetitionLevels,
       ValuesWriter definitionLevels,
       ValuesWriter values)
@@ -72,6 +74,7 @@ void writePage(
         rowCount,
         statistics,
         sizeStatistics,
+        geospatialStatistics,
         repetitionLevels.getEncoding(),
         definitionLevels.getEncoding(),
         values.getEncoding());

parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriterV2.java

@@ -26,6 +26,7 @@
 import org.apache.parquet.column.page.PageWriter;
 import org.apache.parquet.column.statistics.SizeStatistics;
 import org.apache.parquet.column.statistics.Statistics;
+import org.apache.parquet.column.statistics.geospatial.GeospatialStatistics;
 import org.apache.parquet.column.values.ValuesWriter;
 import org.apache.parquet.column.values.bitpacking.DevNullValuesWriter;
 import org.apache.parquet.column.values.bloomfilter.BloomFilterWriter;
@@ -88,6 +89,7 @@ void writePage(
       int valueCount,
       Statistics<?> statistics,
       SizeStatistics sizeStatistics,
+      GeospatialStatistics geospatialStatistics,
       ValuesWriter repetitionLevels,
       ValuesWriter definitionLevels,
       ValuesWriter values)
@@ -105,6 +107,7 @@ void writePage(
         encoding,
         bytes,
         statistics,
-        sizeStatistics);
+        sizeStatistics,
+        geospatialStatistics);
   }
 }

parquet-column/src/main/java/org/apache/parquet/column/page/PageWriter.java

@@ -23,6 +23,7 @@
 import org.apache.parquet.column.Encoding;
 import org.apache.parquet.column.statistics.SizeStatistics;
 import org.apache.parquet.column.statistics.Statistics;
+import org.apache.parquet.column.statistics.geospatial.GeospatialStatistics;
 
 /**
  * a writer for all the pages of a given column chunk
@@ -86,6 +87,7 @@ void writePage(
    * @param valuesEncoding values encoding
    * @throws IOException
    */
+  @Deprecated
   default void writePage(
       BytesInput bytesInput,
       int valueCount,
@@ -99,6 +101,33 @@ default void writePage(
     throw new UnsupportedOperationException("writePage with SizeStatistics is not implemented");
   }
 
+  /**
+   * writes a single page
+   * @param bytesInput the bytes for the page
+   * @param valueCount the number of values in that page
+   * @param rowCount the number of rows in that page
+   * @param statistics the statistics for that page
+   * @param sizeStatistics the size statistics for that page
+   * @param geospatialStatistics the geospatial statistics for that page
+   * @param rlEncoding repetition level encoding
+   * @param dlEncoding definition level encoding
+   * @param valuesEncoding values encoding
+   * @throws IOException
+   */
+  default void writePage(
+      BytesInput bytesInput,
+      int valueCount,
+      int rowCount,
+      Statistics<?> statistics,
+      SizeStatistics sizeStatistics,
+      GeospatialStatistics geospatialStatistics,
+      Encoding rlEncoding,
+      Encoding dlEncoding,
+      Encoding valuesEncoding)
+      throws IOException {
+    throw new UnsupportedOperationException("writePage with GeospatialStatistics is not implemented");
+  }
+
   /**
    * writes a single page in the new format
    *
@@ -136,6 +165,7 @@ void writePageV2(
    * @param sizeStatistics optional size stats for this page
    * @throws IOException if there is an exception while writing page data
    */
+  @Deprecated
   default void writePageV2(
       int rowCount,
       int nullCount,
@@ -150,6 +180,34 @@ default void writePageV2(
     throw new UnsupportedOperationException("writePageV2 with SizeStatistics is not implemented");
   }
 
+  /**
+   * writes a single page in the new format
+   * @param rowCount the number of rows in this page
+   * @param nullCount the number of null values (out of valueCount)
+   * @param valueCount the number of values in that page (there could be multiple values per row for repeated fields)
+   * @param repetitionLevels the repetition levels encoded in RLE without any size header
+   * @param definitionLevels the definition levels encoded in RLE without any size header
+   * @param dataEncoding the encoding for the data
+   * @param data the data encoded with dataEncoding
+   * @param statistics optional stats for this page
+   * @param sizeStatistics optional size stats for this page
+   * @throws IOException if there is an exception while writing page data
+   */
+  default void writePageV2(
+      int rowCount,
+      int nullCount,
+      int valueCount,
+      BytesInput repetitionLevels,
+      BytesInput definitionLevels,
+      Encoding dataEncoding,
+      BytesInput data,
+      Statistics<?> statistics,
+      SizeStatistics sizeStatistics,
+      GeospatialStatistics geospatialStatistics)
+      throws IOException {
+    throw new UnsupportedOperationException("writePageV2 with GeospatialStatistics is not implemented");
+  }
+
   /**
    * @return the current size used in the memory buffer for that column chunk
    */