Revert "PARQUET-1381: Add merge blocks command to parquet-tools (#512)" (#621)

This reverts commit 863a081.

The design of this feature has conceptional problems and also works incorrectly. See PARQUET-1381 for more details.

Author: gszadovszky

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

@@ -85,7 +85,7 @@ public ColumnReader getColumnReader(ColumnDescriptor path) {
     }
   }
 
-  public ColumnReaderImpl newMemColumnReader(ColumnDescriptor path, PageReader pageReader) {
+  private ColumnReaderImpl newMemColumnReader(ColumnDescriptor path, PageReader pageReader) {
     PrimitiveConverter converter = getPrimitiveConverter(path);
     return new ColumnReaderImpl(path, pageReader, converter, writerVersion);
   }

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

@@ -292,9 +292,4 @@ public void flushToFileWriter(ParquetFileWriter writer) throws IOException {
     }
   }
 
-  void flushToFileWriter(ColumnDescriptor path, ParquetFileWriter writer) throws IOException {
-    ColumnChunkPageWriter pageWriter = writers.get(path);
-    pageWriter.writeToFileWriter(writer);
-  }
-
 }

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

@@ -42,7 +42,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
@@ -67,7 +66,6 @@
 import org.apache.parquet.column.page.DataPageV2;
 import org.apache.parquet.column.page.DictionaryPage;
 import org.apache.parquet.column.page.DictionaryPageReadStore;
-import org.apache.parquet.column.page.PageReader;
 import org.apache.parquet.column.page.PageReadStore;
 import org.apache.parquet.compression.CompressionCodecFactory.BytesInputDecompressor;
 import org.apache.parquet.filter2.compat.FilterCompat;
@@ -1408,7 +1406,27 @@ public void addChunk(ChunkDescriptor descriptor) {
      * @throws IOException if there is an error while reading from the stream
      */
     public void readAll(SeekableInputStream f, ChunkListBuilder builder) throws IOException {
-      List<ByteBuffer> buffers = readBlocks(f, offset, length);
+      List<Chunk> result = new ArrayList<Chunk>(chunks.size());
+      f.seek(offset);
+
+      int fullAllocations = length / options.getMaxAllocationSize();
+      int lastAllocationSize = length % options.getMaxAllocationSize();
+
+      int numAllocations = fullAllocations + (lastAllocationSize > 0 ? 1 : 0);
+      List<ByteBuffer> buffers = new ArrayList<>(numAllocations);
+
+      for (int i = 0; i < fullAllocations; i += 1) {
+        buffers.add(options.getAllocator().allocate(options.getMaxAllocationSize()));
+      }
+
+      if (lastAllocationSize > 0) {
+        buffers.add(options.getAllocator().allocate(lastAllocationSize));
+      }
+
+      for (ByteBuffer buffer : buffers) {
+        f.readFully(buffer);
+        buffer.flip();
+      }
 
       // report in a counter the data we just scanned
       BenchmarkCounter.incrementBytesRead(length);
@@ -1428,72 +1446,4 @@ public long endPos() {
 
   }
 
-  /**
-   * @param f file to read the blocks from
-   * @return the ByteBuffer blocks
-   * @throws IOException if there is an error while reading from the stream
-   */
-  List<ByteBuffer> readBlocks(SeekableInputStream f, long offset, int length) throws IOException {
-    f.seek(offset);
-
-    int fullAllocations = length / options.getMaxAllocationSize();
-    int lastAllocationSize = length % options.getMaxAllocationSize();
-
-    int numAllocations = fullAllocations + (lastAllocationSize > 0 ? 1 : 0);
-    List<ByteBuffer> buffers = new ArrayList<>(numAllocations);
-
-    for (int i = 0; i < fullAllocations; i++) {
-      buffers.add(options.getAllocator().allocate(options.getMaxAllocationSize()));
-    }
-
-    if (lastAllocationSize > 0) {
-      buffers.add(options.getAllocator().allocate(lastAllocationSize));
-    }
-
-    for (ByteBuffer buffer : buffers) {
-      f.readFully(buffer);
-      buffer.flip();
-    }
-    return buffers;
-  }
-
-  Optional<PageReader> readColumnInBlock(int blockIndex, ColumnDescriptor columnDescriptor) {
-    BlockMetaData block = blocks.get(blockIndex);
-    if (block.getRowCount() == 0) {
-      throw new RuntimeException("Illegal row group of 0 rows");
-    }
-    Optional<ColumnChunkMetaData> mc = findColumnByPath(block, columnDescriptor.getPath());
-
-    return mc.map(column -> new ChunkDescriptor(columnDescriptor, column, column.getStartingPos(), (int) column.getTotalSize()))
-      .map(chunk -> readChunk(f, chunk));
-  }
-
-  private ColumnChunkPageReader readChunk(SeekableInputStream f, ChunkDescriptor descriptor) {
-    try {
-      List<ByteBuffer> buffers = readBlocks(f, descriptor.fileOffset, descriptor.size);
-      ByteBufferInputStream stream = ByteBufferInputStream.wrap(buffers);
-      Chunk chunk = new WorkaroundChunk(descriptor, stream.sliceBuffers(descriptor.size), f, null);
-      return chunk.readAllPages();
-    } catch (IOException e) {
-      throw new RuntimeException(e);
-    }
-  }
-
-  private Optional<ColumnChunkMetaData> findColumnByPath(BlockMetaData block, String[] path) {
-    for (ColumnChunkMetaData column : block.getColumns()) {
-      if (Arrays.equals(column.getPath().toArray(), path)) {
-        return Optional.of(column);
-      }
-    }
-    return Optional.empty();
-  }
-
-  public int blocksCount() {
-    return blocks.size();
-  }
-
-  public BlockMetaData getBlockMetaData(int blockIndex) {
-    return blocks.get(blockIndex);
-  }
-
 }

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

@@ -26,15 +26,12 @@
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.Optional;
 import java.util.Set;
 
 import org.apache.hadoop.conf.Configuration;
@@ -45,23 +42,14 @@
 import org.apache.parquet.Preconditions;
 import org.apache.parquet.Strings;
 import org.apache.parquet.Version;
-import org.apache.parquet.bytes.ByteBufferAllocator;
 import org.apache.parquet.bytes.BytesInput;
 import org.apache.parquet.bytes.BytesUtils;
-import org.apache.parquet.bytes.HeapByteBufferAllocator;
 import org.apache.parquet.column.ColumnDescriptor;
-import org.apache.parquet.column.ColumnReader;
-import org.apache.parquet.column.ColumnWriteStore;
-import org.apache.parquet.column.ColumnWriter;
 import org.apache.parquet.column.Encoding;
 import org.apache.parquet.column.EncodingStats;
 import org.apache.parquet.column.ParquetProperties;
-import org.apache.parquet.column.impl.ColumnReadStoreImpl;
-import org.apache.parquet.column.impl.ColumnWriteStoreV1;
 import org.apache.parquet.column.page.DictionaryPage;
-import org.apache.parquet.column.page.PageReader;
 import org.apache.parquet.column.statistics.Statistics;
-import org.apache.parquet.example.DummyRecordConverter;
 import org.apache.parquet.hadoop.ParquetOutputFormat.JobSummaryLevel;
 import org.apache.parquet.hadoop.metadata.ColumnPath;
 import org.apache.parquet.format.Util;
@@ -72,7 +60,6 @@
 import org.apache.parquet.hadoop.metadata.FileMetaData;
 import org.apache.parquet.hadoop.metadata.GlobalMetaData;
 import org.apache.parquet.hadoop.metadata.ParquetMetadata;
-import org.apache.parquet.hadoop.util.BlocksCombiner;
 import org.apache.parquet.hadoop.util.HadoopOutputFile;
 import org.apache.parquet.hadoop.util.HadoopStreams;
 import org.apache.parquet.internal.column.columnindex.ColumnIndex;
@@ -669,116 +656,6 @@ public void appendFile(InputFile file) throws IOException {
     }
   }
 
-  public int merge(List<InputFile> inputFiles, CodecFactory.BytesCompressor compressor, String createdBy, long maxBlockSize) throws IOException {
-    List<ParquetFileReader> readers = getReaders(inputFiles);
-    try {
-      ByteBufferAllocator allocator = new HeapByteBufferAllocator();
-      ColumnReadStoreImpl columnReadStore = new ColumnReadStoreImpl(null, new DummyRecordConverter(schema).getRootConverter(), schema, createdBy);
-      this.start();
-      List<BlocksCombiner.SmallBlocksUnion> largeBlocks = BlocksCombiner.combineLargeBlocks(readers, maxBlockSize);
-      for (BlocksCombiner.SmallBlocksUnion smallBlocks : largeBlocks) {
-        for (int columnIndex = 0; columnIndex < schema.getColumns().size(); columnIndex++) {
-          ColumnDescriptor path = schema.getColumns().get(columnIndex);
-          ColumnChunkPageWriteStore store = new ColumnChunkPageWriteStore(compressor, schema, allocator, ParquetProperties.DEFAULT_COLUMN_INDEX_TRUNCATE_LENGTH);
-          ColumnWriteStoreV1 columnWriteStoreV1 = new ColumnWriteStoreV1(schema, store, ParquetProperties.builder().build());
-          for (BlocksCombiner.SmallBlock smallBlock : smallBlocks.getBlocks()) {
-            ParquetFileReader parquetFileReader = smallBlock.getReader();
-            try {
-              Optional<PageReader> columnChunkPageReader = parquetFileReader.readColumnInBlock(smallBlock.getBlockIndex(), path);
-              ColumnWriter columnWriter = columnWriteStoreV1.getColumnWriter(path);
-              if (columnChunkPageReader.isPresent()) {
-                ColumnReader columnReader = columnReadStore.newMemColumnReader(path, columnChunkPageReader.get());
-                for (int i = 0; i < columnReader.getTotalValueCount(); i++) {
-                  consumeTriplet(columnWriteStoreV1, columnWriter, columnReader);
-                }
-              } else {
-                MessageType inputFileSchema = parquetFileReader.getFileMetaData().getSchema();
-                String[] parentPath = getExisingParentPath(path, inputFileSchema);
-                int def = parquetFileReader.getFileMetaData().getSchema().getMaxDefinitionLevel(parentPath);
-                int rep = parquetFileReader.getFileMetaData().getSchema().getMaxRepetitionLevel(parentPath);
-                for (int i = 0; i < parquetFileReader.getBlockMetaData(smallBlock.getBlockIndex()).getRowCount(); i++) {
-                  columnWriter.writeNull(rep, def);
-                  if (def == 0) {
-                    // V1 pages also respect record boundaries so we have to mark them
-                    columnWriteStoreV1.endRecord();
-                  }
-                }
-              }
-            } catch (Exception e) {
-              LOG.error("File {} is not readable", parquetFileReader.getFile(), e);
-            }
-          }
-          if (columnIndex == 0) {
-            this.startBlock(smallBlocks.getRowCount());
-          }
-          columnWriteStoreV1.flush();
-          store.flushToFileWriter(path, this);
-        }
-        this.endBlock();
-      }
-      this.end(Collections.emptyMap());
-    }finally {
-      BlocksCombiner.closeReaders(readers);
-    }
-    return 0;
-  }
-
-  private String[] getExisingParentPath(ColumnDescriptor path, MessageType inputFileSchema) {
-    List<String> parentPath = Arrays.asList(path.getPath());
-    while (parentPath.size() > 0 && !inputFileSchema.containsPath(parentPath.toArray(new String[parentPath.size()]))) {
-      parentPath = parentPath.subList(0, parentPath.size() - 1);
-    }
-    return parentPath.toArray(new String[parentPath.size()]);
-  }
-
-  private List<ParquetFileReader> getReaders(List<InputFile> inputFiles) throws IOException {
-    List<ParquetFileReader> readers = new ArrayList<>(inputFiles.size());
-    for (InputFile inputFile : inputFiles) {
-      readers.add(ParquetFileReader.open(inputFile));
-    }
-    return readers;
-  }
-
-  private void consumeTriplet(ColumnWriteStore columnWriteStore, ColumnWriter columnWriter, ColumnReader columnReader) {
-    int definitionLevel = columnReader.getCurrentDefinitionLevel();
-    int repetitionLevel = columnReader.getCurrentRepetitionLevel();
-    ColumnDescriptor column = columnReader.getDescriptor();
-    PrimitiveType type = column.getPrimitiveType();
-    if (definitionLevel < column.getMaxDefinitionLevel()) {
-      columnWriter.writeNull(repetitionLevel, definitionLevel);
-    } else {
-      switch (type.getPrimitiveTypeName()) {
-        case INT32:
-          columnWriter.write(columnReader.getInteger(), repetitionLevel, definitionLevel);
-          break;
-        case INT64:
-          columnWriter.write(columnReader.getLong(), repetitionLevel, definitionLevel);
-          break;
-        case BINARY:
-        case FIXED_LEN_BYTE_ARRAY:
-        case INT96:
-          columnWriter.write(columnReader.getBinary(), repetitionLevel, definitionLevel);
-          break;
-        case BOOLEAN:
-          columnWriter.write(columnReader.getBoolean(), repetitionLevel, definitionLevel);
-          break;
-        case FLOAT:
-          columnWriter.write(columnReader.getFloat(), repetitionLevel, definitionLevel);
-          break;
-        case DOUBLE:
-          columnWriter.write(columnReader.getDouble(), repetitionLevel, definitionLevel);
-          break;
-        default:
-          throw new IllegalArgumentException("Unknown primitive type " + type);
-      }
-    }
-    columnReader.consume();
-    if (repetitionLevel == 0) {
-      // V1 pages also respect record boundaries so we have to mark them
-      columnWriteStore.endRecord();
-    }
-  }
-
   /**
    * @param file a file stream to read from
    * @param rowGroups row groups to copy