PARQUET-2417: Add statistics support to geometry logical type

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
  */