add option to not write path_in_schema

Author: etseidl

parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java

@@ -66,6 +66,7 @@ public class ParquetProperties {
   public static final int DEFAULT_BLOOM_FILTER_CANDIDATES_NUMBER = 5;
   public static final boolean DEFAULT_STATISTICS_ENABLED = true;
   public static final boolean DEFAULT_SIZE_STATISTICS_ENABLED = true;
+  public static final boolean DEFAULT_WRITE_PATH_IN_SCHEMA_ENABLED = true;
 
   public static final boolean DEFAULT_PAGE_WRITE_CHECKSUM_ENABLED = true;
 
@@ -120,6 +121,7 @@ public static WriterVersion fromString(String name) {
   private final int statisticsTruncateLength;
   private final boolean statisticsEnabled;
   private final boolean sizeStatisticsEnabled;
+  private final boolean writePathInSchemaEnabled;
 
   // The expected NDV (number of distinct values) for each columns
   private final ColumnProperty<Long> bloomFilterNDVs;
@@ -154,6 +156,7 @@ private ParquetProperties(Builder builder) {
     this.statisticsTruncateLength = builder.statisticsTruncateLength;
     this.statisticsEnabled = builder.statisticsEnabled;
     this.sizeStatisticsEnabled = builder.sizeStatisticsEnabled;
+    this.writePathInSchemaEnabled = builder.writePathInSchemaEnabled;
     this.bloomFilterNDVs = builder.bloomFilterNDVs.build();
     this.bloomFilterFPPs = builder.bloomFilterFPPs.build();
     this.bloomFilterEnabled = builder.bloomFilterEnabled.build();
@@ -322,6 +325,10 @@ public boolean getPageWriteChecksumEnabled() {
     return pageWriteChecksumEnabled;
   }
 
+  public boolean getWritePathInSchemaEnabled() {
+    return writePathInSchemaEnabled;
+  }
+
   public OptionalLong getBloomFilterNDV(ColumnDescriptor column) {
     Long ndv = bloomFilterNDVs.getValue(column);
     return ndv == null ? OptionalLong.empty() : OptionalLong.of(ndv);
@@ -406,6 +413,7 @@ public static class Builder {
     private int statisticsTruncateLength = DEFAULT_STATISTICS_TRUNCATE_LENGTH;
     private boolean statisticsEnabled = DEFAULT_STATISTICS_ENABLED;
     private boolean sizeStatisticsEnabled = DEFAULT_SIZE_STATISTICS_ENABLED;
+    private boolean writePathInSchemaEnabled = DEFAULT_WRITE_PATH_IN_SCHEMA_ENABLED;
     private final ColumnProperty.Builder<Long> bloomFilterNDVs;
     private final ColumnProperty.Builder<Double> bloomFilterFPPs;
     private int maxBloomFilterBytes = DEFAULT_MAX_BLOOM_FILTER_BYTES;
@@ -756,6 +764,24 @@ public Builder withSizeStatisticsEnabled(String columnPath, boolean enabled) {
       return this;
     }
 
+    /**
+     * Sets whether to write the path_in_schema field in ColumnMetaData.
+     *
+     * The path_in_schema field in the Thrift metadata is redundant and wastes a great
+     * deal of space. Parquet file footers can be made much smaller by omitting this field.
+     * Because the field was originally a mandatory field, this property defaults to true
+     * to maintain compatibility with older readers that expect this field to be present.
+     * If one knows that all readers one plans to use are tolerant of the absense of this field,
+     * this may be safely set to false.
+     *
+     * @param enabled whether to write path_in_schema
+     * @return this builder for method chaining
+     */
+    public Builder withWritePathInSchemaEnabled(boolean enabled) {
+      this.writePathInSchemaEnabled = enabled;
+      return this;
+    }
+
     public ParquetProperties build() {
       ParquetProperties properties = new ParquetProperties(this);
       // we pass a constructed but uninitialized factory to ParquetProperties above as currently

parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java

@@ -163,14 +163,19 @@ public class ParquetMetadataConverter {
       new ConvertedTypeConverterVisitor();
   private final int statisticsTruncateLength;
   private final boolean useSignedStringMinMax;
+  private final boolean writePathInSchema;
   private final ParquetReadOptions options;
 
   public ParquetMetadataConverter() {
     this(false);
   }
 
   public ParquetMetadataConverter(int statisticsTruncateLength) {
-    this(false, statisticsTruncateLength);
+    this(false, statisticsTruncateLength, ParquetProperties.DEFAULT_WRITE_PATH_IN_SCHEMA_ENABLED);
+  }
+
+  public ParquetMetadataConverter(int statisticsTruncateLength, boolean writePathInSchema) {
+    this(false, statisticsTruncateLength, writePathInSchema);
   }
 
   /**
@@ -183,24 +188,36 @@ public ParquetMetadataConverter(Configuration conf) {
   }
 
   public ParquetMetadataConverter(ParquetReadOptions options) {
-    this(options.useSignedStringMinMax(), ParquetProperties.DEFAULT_STATISTICS_TRUNCATE_LENGTH, options);
+    this(
+        options.useSignedStringMinMax(),
+        ParquetProperties.DEFAULT_STATISTICS_TRUNCATE_LENGTH,
+        ParquetProperties.DEFAULT_WRITE_PATH_IN_SCHEMA_ENABLED,
+        options);
   }
 
   private ParquetMetadataConverter(boolean useSignedStringMinMax) {
-    this(useSignedStringMinMax, ParquetProperties.DEFAULT_STATISTICS_TRUNCATE_LENGTH);
+    this(
+        useSignedStringMinMax,
+        ParquetProperties.DEFAULT_STATISTICS_TRUNCATE_LENGTH,
+        ParquetProperties.DEFAULT_WRITE_PATH_IN_SCHEMA_ENABLED);
   }
 
-  private ParquetMetadataConverter(boolean useSignedStringMinMax, int statisticsTruncateLength) {
-    this(useSignedStringMinMax, statisticsTruncateLength, null);
+  private ParquetMetadataConverter(
+      boolean useSignedStringMinMax, int statisticsTruncateLength, boolean writePathInSchema) {
+    this(useSignedStringMinMax, statisticsTruncateLength, writePathInSchema, null);
   }
 
   private ParquetMetadataConverter(
-      boolean useSignedStringMinMax, int statisticsTruncateLength, ParquetReadOptions options) {
+      boolean useSignedStringMinMax,
+      int statisticsTruncateLength,
+      boolean writePathInSchema,
+      ParquetReadOptions options) {
     if (statisticsTruncateLength <= 0) {
       throw new IllegalArgumentException("Truncate length should be greater than 0");
     }
     this.useSignedStringMinMax = useSignedStringMinMax;
     this.statisticsTruncateLength = statisticsTruncateLength;
+    this.writePathInSchema = writePathInSchema;
     this.options = options;
   }
 
@@ -618,7 +635,7 @@ private void addRowGroup(
           || columnMetaData.hasDictionaryPage()) {
         metaData.setDictionary_page_offset(columnMetaData.getDictionaryPageOffset());
       }
-      if (path != null) {
+      if (path != null && this.writePathInSchema) {
         metaData.setPath_in_schema(path.toList());
       }
       long bloomFilterOffset = columnMetaData.getBloomFilterOffset();

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java

@@ -333,7 +333,8 @@ public ParquetFileWriter(OutputFile file, MessageType schema, Mode mode, long ro
         maxPaddingSize,
         ParquetProperties.DEFAULT_COLUMN_INDEX_TRUNCATE_LENGTH,
         ParquetProperties.DEFAULT_STATISTICS_TRUNCATE_LENGTH,
-        ParquetProperties.DEFAULT_PAGE_WRITE_CHECKSUM_ENABLED);
+        ParquetProperties.DEFAULT_PAGE_WRITE_CHECKSUM_ENABLED,
+        ParquetProperties.DEFAULT_WRITE_PATH_IN_SCHEMA_ENABLED);
   }
 
   @FunctionalInterface
@@ -373,8 +374,35 @@ private void withAbortOnFailure(IORunnable action) throws IOException {
    * @param columnIndexTruncateLength the length which the min/max values in column indexes tried to be truncated to
    * @param statisticsTruncateLength  the length which the min/max values in row groups tried to be truncated to
    * @param pageWriteChecksumEnabled  whether to write out page level checksums
+   * @param writePathInSchemaEnabled  whether to write path_in_schema to the column metadata
    * @throws IOException if the file can not be created
    */
+  public ParquetFileWriter(
+      OutputFile file,
+      MessageType schema,
+      Mode mode,
+      long rowGroupSize,
+      int maxPaddingSize,
+      int columnIndexTruncateLength,
+      int statisticsTruncateLength,
+      boolean pageWriteChecksumEnabled,
+      boolean writePathInSchemaEnabled)
+      throws IOException {
+    this(
+        file,
+        schema,
+        mode,
+        rowGroupSize,
+        maxPaddingSize,
+        columnIndexTruncateLength,
+        statisticsTruncateLength,
+        pageWriteChecksumEnabled,
+        writePathInSchemaEnabled,
+        null,
+        null,
+        null);
+  }
+
   public ParquetFileWriter(
       OutputFile file,
       MessageType schema,
@@ -394,7 +422,34 @@ public ParquetFileWriter(
         columnIndexTruncateLength,
         statisticsTruncateLength,
         pageWriteChecksumEnabled,
+        ParquetProperties.DEFAULT_WRITE_PATH_IN_SCHEMA_ENABLED,
+        null,
         null,
+        null);
+  }
+
+  public ParquetFileWriter(
+      OutputFile file,
+      MessageType schema,
+      Mode mode,
+      long rowGroupSize,
+      int maxPaddingSize,
+      int columnIndexTruncateLength,
+      int statisticsTruncateLength,
+      boolean pageWriteChecksumEnabled,
+      FileEncryptionProperties encryptionProperties)
+      throws IOException {
+    this(
+        file,
+        schema,
+        mode,
+        rowGroupSize,
+        maxPaddingSize,
+        columnIndexTruncateLength,
+        statisticsTruncateLength,
+        pageWriteChecksumEnabled,
+        ParquetProperties.DEFAULT_WRITE_PATH_IN_SCHEMA_ENABLED,
+        encryptionProperties,
         null,
         null);
   }
@@ -408,6 +463,7 @@ public ParquetFileWriter(
       int columnIndexTruncateLength,
       int statisticsTruncateLength,
       boolean pageWriteChecksumEnabled,
+      boolean writePathInSchemaEnabled,
       FileEncryptionProperties encryptionProperties)
       throws IOException {
     this(
@@ -419,6 +475,7 @@ public ParquetFileWriter(
         columnIndexTruncateLength,
         statisticsTruncateLength,
         pageWriteChecksumEnabled,
+        writePathInSchemaEnabled,
         encryptionProperties,
         null,
         null);
@@ -442,6 +499,7 @@ public ParquetFileWriter(
         props.getColumnIndexTruncateLength(),
         props.getStatisticsTruncateLength(),
         props.getPageWriteChecksumEnabled(),
+        props.getWritePathInSchemaEnabled(),
         encryptionProperties,
         null,
         props.getAllocator());
@@ -468,6 +526,7 @@ public ParquetFileWriter(
         columnIndexTruncateLength,
         statisticsTruncateLength,
         pageWriteChecksumEnabled,
+        ParquetProperties.DEFAULT_WRITE_PATH_IN_SCHEMA_ENABLED,
         null,
         encryptor,
         null);
@@ -482,6 +541,7 @@ private ParquetFileWriter(
       int columnIndexTruncateLength,
       int statisticsTruncateLength,
       boolean pageWriteChecksumEnabled,
+      boolean writePathInSchemaEnabled,
       FileEncryptionProperties encryptionProperties,
       InternalFileEncryptor encryptor,
       ByteBufferAllocator allocator)
@@ -512,7 +572,7 @@ private ParquetFileWriter(
         ? ReusingByteBufferAllocator.strict(allocator == null ? new HeapByteBufferAllocator() : allocator)
         : null;
 
-    this.metadataConverter = new ParquetMetadataConverter(statisticsTruncateLength);
+    this.metadataConverter = new ParquetMetadataConverter(statisticsTruncateLength, writePathInSchemaEnabled);
 
     if (null == encryptionProperties && null == encryptor) {
       this.fileEncryptor = null;
@@ -584,7 +644,9 @@ private ParquetFileWriter(
     this.crcAllocator = pageWriteChecksumEnabled
         ? ReusingByteBufferAllocator.strict(allocator == null ? new HeapByteBufferAllocator() : allocator)
         : null;
-    this.metadataConverter = new ParquetMetadataConverter(ParquetProperties.DEFAULT_STATISTICS_TRUNCATE_LENGTH);
+    this.metadataConverter = new ParquetMetadataConverter(
+        ParquetProperties.DEFAULT_STATISTICS_TRUNCATE_LENGTH,
+        ParquetProperties.DEFAULT_WRITE_PATH_IN_SCHEMA_ENABLED);
     this.fileEncryptor = null;
   }
 

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java

@@ -508,6 +508,7 @@ public RecordWriter<Void, T> getRecordWriter(Configuration conf, Path file, Comp
       throws IOException, InterruptedException {
     final WriteSupport<T> writeSupport = getWriteSupport(conf);
 
+    // TODO(ets): add write_path_in_schema to conf?
     ParquetProperties.Builder propsBuilder = ParquetProperties.builder()
         .withPageSize(getPageSize(conf))
         .withDictionaryPageSize(getDictionaryPageSize(conf))

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/ParquetRewriter.java

@@ -195,6 +195,7 @@ public ParquetRewriter(RewriteOptions options) throws IOException {
         DEFAULT_COLUMN_INDEX_TRUNCATE_LENGTH,
         DEFAULT_STATISTICS_TRUNCATE_LENGTH,
         ParquetProperties.DEFAULT_PAGE_WRITE_CHECKSUM_ENABLED,
+        ParquetProperties.DEFAULT_WRITE_PATH_IN_SCHEMA_ENABLED,
         options.getFileEncryptionProperties());
     writer.start();
     // column nullification requires a separate encryptor and forcing other columns encryption initialization

parquet-hadoop/src/test/java/org/apache/parquet/format/converter/TestParquetMetadataConverter.java

@@ -856,7 +856,8 @@ private void testBinaryStatsWithTruncation(int truncateLen, int minLen, int maxL
     byte[] max = generateRandomString("b", maxLen).getBytes();
     stats.updateStats(Binary.fromConstantByteArray(min));
     stats.updateStats(Binary.fromConstantByteArray(max));
-    ParquetMetadataConverter metadataConverter = new ParquetMetadataConverter(truncateLen);
+    ParquetMetadataConverter metadataConverter =
+        new ParquetMetadataConverter(truncateLen, ParquetProperties.DEFAULT_WRITE_PATH_IN_SCHEMA_ENABLED);
     org.apache.parquet.format.Statistics formatStats = metadataConverter.toParquetStatistics(stats);
 
     if (minLen + maxLen >= ParquetMetadataConverter.MAX_STATS_SIZE) {
@@ -1976,4 +1977,21 @@ public void testEdgeInterpolationAlgorithmConversion() {
     assertNull(ParquetMetadataConverter.fromParquetEdgeInterpolationAlgorithm(null));
     assertNull(ParquetMetadataConverter.toParquetEdgeInterpolationAlgorithm(null));
   }
+
+  @Test
+  public void testSkipPathInSchema() throws IOException {
+    ParquetMetadata origMetaData = createParquetMetaData(null, Encoding.PLAIN);
+    ParquetMetadataConverter converter =
+        new ParquetMetadataConverter(ParquetProperties.DEFAULT_STATISTICS_TRUNCATE_LENGTH, false);
+
+    // Without path_in_schema
+    FileMetaData footer = converter.toParquetMetadata(1, origMetaData);
+    assertFalse(
+        footer.getRow_groups().get(0).getColumns().get(0).getMeta_data().isSetPath_in_schema());
+
+    // With path_in_schema
+    converter = new ParquetMetadataConverter(ParquetProperties.DEFAULT_STATISTICS_TRUNCATE_LENGTH, true);
+    assertFalse(
+        footer.getRow_groups().get(0).getColumns().get(0).getMeta_data().isSetPath_in_schema());
+  }
 }