Merge branch 'master' into vinooganesh/thrift0.22.0

Author: vinooganesh

parquet-avro/src/test/java/org/apache/parquet/avro/TestReadWrite.java

@@ -962,9 +962,9 @@ private ParquetWriter<GenericRecord> writer(String file, Schema schema) throws I
           writerBuilder.withCodecFactory(HadoopCodecs.newFactory(ParquetProperties.DEFAULT_PAGE_SIZE));
     }
     if (conf == ConfigurationType.PLAIN_PARQUET_INTERFACE) {
-      return writerBuilder.withConf(hadoopConfWithInterface).build();
-    } else if (conf == ConfigurationType.HADOOP_PARQUET_INTERFACE) {
       return writerBuilder.withConf(plainParquetConf).build();
+    } else if (conf == ConfigurationType.HADOOP_PARQUET_INTERFACE) {
+      return writerBuilder.withConf(hadoopConfWithInterface).build();
     } else {
       return writerBuilder.withConf(testConf).build();
     }
@@ -983,9 +983,9 @@ private ParquetReader<GenericRecord> reader(String file) throws IOException {
           readerBuilder.withCodecFactory(HadoopCodecs.newFactory(ParquetProperties.DEFAULT_PAGE_SIZE));
     }
     if (conf == ConfigurationType.PLAIN_PARQUET_INTERFACE) {
-      return readerBuilder.withConf(hadoopConfWithInterface).build();
-    } else if (conf == ConfigurationType.HADOOP_PARQUET_INTERFACE) {
       return readerBuilder.withConf(plainParquetConf).build();
+    } else if (conf == ConfigurationType.HADOOP_PARQUET_INTERFACE) {
+      return readerBuilder.withConf(hadoopConfWithInterface).build();
     } else {
       return readerBuilder.withConf(testConf).build();
     }

parquet-cli/README.md

@@ -31,13 +31,13 @@ You can build this project using maven:
 The build produces a shaded Jar that can be run using the `hadoop` command:
 
 ```
-hadoop jar parquet-cli-1.12.3-runtime.jar org.apache.parquet.cli.Main
+hadoop jar parquet-cli-1.16.0-runtime.jar org.apache.parquet.cli.Main
 ```
 
 For a shorter command-line invocation, add an alias to your shell like this:
 
 ```
-alias parquet="hadoop jar /path/to/parquet-cli-1.12.3-runtime.jar org.apache.parquet.cli.Main --dollar-zero parquet"
+alias parquet="hadoop jar /path/to/parquet-cli-1.16.0-runtime.jar org.apache.parquet.cli.Main --dollar-zero parquet"
 ```
 
 ### Running without Hadoop
@@ -51,7 +51,7 @@ To run from the target directory instead of using the `hadoop` command, first co
 Then, run the command-line and add `target/dependencies/*` to the classpath:
 
 ```
-java -cp 'target/parquet-cli-1.12.3.jar:target/dependency/*' org.apache.parquet.cli.Main
+java -cp 'target/parquet-cli-1.16.0.jar:target/dependency/*' org.apache.parquet.cli.Main
 ```
 
 Note that you shouldn't include the runtime jar used above into the classpath in this case.
@@ -79,6 +79,8 @@ Usage: parquet [options] [command] [command options]
 
     help
         Retrieves details on the functions of other commands
+    version
+        Print version of the Parquet CLI tool
     meta
         Print a Parquet file's metadata
     pages
@@ -126,7 +128,7 @@ Usage: parquet [options] [command] [command options]
 
   Examples:
 
-    # print information for create
+    # print information for meta
     parquet help meta
 
   See 'parquet help <command>' for more information on a specific command.

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

@@ -25,6 +25,8 @@
 import com.beust.jcommander.ParameterDescription;
 import com.beust.jcommander.Parameters;
 import com.google.common.collect.Lists;
+import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import org.slf4j.Logger;
 
@@ -116,7 +118,7 @@ public void printGenericHelp() {
     }
 
     jc.getCommands().keySet().stream()
-        .filter(s -> !s.equals("help"))
+        .filter(s -> !Arrays.asList("version", "help").contains(s))
         .findFirst()
         .ifPresent(command -> {
           console.info("\n  Examples:");
@@ -148,6 +150,6 @@ private String formatDefault(ParameterDescription param) {
 
   @Override
   public List<String> getExamples() {
-    return null;
+    return Collections.emptyList();
   }
 }

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

@@ -53,6 +53,7 @@
 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.ShowVersionCommand;
 import org.apache.parquet.cli.commands.ToAvroCommand;
 import org.apache.parquet.cli.commands.TransCompressionCommand;
 import org.slf4j.Logger;
@@ -87,6 +88,7 @@ public class Main extends Configured implements Tool {
     this.help = new Help(jc, console);
     jc.setProgramName(DEFAULT_PROGRAM_NAME);
     jc.addCommand("help", help, "-h", "-help", "--help");
+    jc.addCommand("version", new ShowVersionCommand(console), "-version", "--version");
     jc.addCommand("meta", new ParquetMetadataCommand(console));
     jc.addCommand("pages", new ShowPagesCommand(console));
     jc.addCommand("dictionary", new ShowDictionaryCommand(console));

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

@@ -0,0 +1,45 @@
+/*
+ * 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.Parameters;
+import java.util.Collections;
+import java.util.List;
+import org.apache.parquet.Version;
+import org.apache.parquet.cli.BaseCommand;
+import org.slf4j.Logger;
+
+@Parameters(commandDescription = "Print version of the Parquet CLI tool")
+public class ShowVersionCommand extends BaseCommand {
+
+  public ShowVersionCommand(Logger console) {
+    super(console);
+  }
+
+  @Override
+  public int run() {
+    console.info(Version.FULL_VERSION);
+    return 0;
+  }
+
+  @Override
+  public List<String> getExamples() {
+    return Collections.emptyList();
+  }
+}

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

@@ -19,12 +19,17 @@
 package org.apache.parquet.cli.commands;
 
 import java.io.File;
+import java.util.Queue;
+import java.util.concurrent.LinkedBlockingQueue;
 import org.apache.commons.logging.LogFactory;
 import org.apache.log4j.PropertyConfigurator;
 import org.junit.Rule;
 import org.junit.rules.TemporaryFolder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.slf4j.event.LoggingEvent;
+import org.slf4j.event.SubstituteLoggingEvent;
+import org.slf4j.helpers.SubstituteLoggerFactory;
 
 public abstract class FileTest {
 
@@ -52,4 +57,24 @@ protected static Logger createLogger() {
         .setAttribute("org.apache.commons.logging.Log", "org.apache.commons.logging.impl.Log4JLogger");
     return console;
   }
+
+  @FunctionalInterface
+  public interface ThrowableBiConsumer<T, U> {
+    void accept(T t, U u) throws Exception;
+  }
+
+  protected static void withLogger(ThrowableBiConsumer<Logger, Queue<? extends LoggingEvent>> body) {
+    SubstituteLoggerFactory loggerFactory = new SubstituteLoggerFactory();
+    LinkedBlockingQueue<SubstituteLoggingEvent> loggingEvents = loggerFactory.getEventQueue();
+    Logger console = loggerFactory.getLogger(ParquetFileTest.class.getName());
+    try {
+      body.accept(console, loggingEvents);
+    } catch (RuntimeException rethrow) {
+      throw rethrow;
+    } catch (Exception checkedException) {
+      throw new RuntimeException(checkedException);
+    } finally {
+      loggerFactory.clear();
+    }
+  }
 }

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

@@ -0,0 +1,43 @@
+/*
+ * 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.util.Queue;
+import org.apache.parquet.Version;
+import org.junit.Assert;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.event.LoggingEvent;
+
+public class ShowVersionCommandTest extends FileTest {
+
+  @Test
+  public void testVersionCommand() {
+    withLogger(this::testVersionCommand0);
+  }
+
+  private void testVersionCommand0(Logger console, Queue<? extends LoggingEvent> loggingEvents) {
+    ShowVersionCommand command = new ShowVersionCommand(console);
+    Assert.assertEquals(0, command.run());
+    Assert.assertEquals(1, loggingEvents.size());
+    LoggingEvent loggingEvent = loggingEvents.remove();
+    Assert.assertEquals(Version.FULL_VERSION, loggingEvent.getMessage());
+    loggingEvents.clear();
+  }
+}

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

@@ -57,6 +57,7 @@ public class ParquetProperties {
   public static final int DEFAULT_PAGE_VALUE_COUNT_THRESHOLD = Integer.MAX_VALUE / 2;
   public static final int DEFAULT_COLUMN_INDEX_TRUNCATE_LENGTH = 64;
   public static final int DEFAULT_STATISTICS_TRUNCATE_LENGTH = Integer.MAX_VALUE;
+  public static final int DEFAULT_ROW_GROUP_ROW_COUNT_LIMIT = Integer.MAX_VALUE;
   public static final int DEFAULT_PAGE_ROW_COUNT_LIMIT = 20_000;
   public static final int DEFAULT_MAX_BLOOM_FILTER_BYTES = 1024 * 1024;
   public static final boolean DEFAULT_BLOOM_FILTER_ENABLED = false;
@@ -122,6 +123,7 @@ public static WriterVersion fromString(String name) {
   private final ColumnProperty<Boolean> bloomFilterEnabled;
   private final ColumnProperty<Boolean> adaptiveBloomFilterEnabled;
   private final ColumnProperty<Integer> numBloomFilterCandidates;
+  private final int rowGroupRowCountLimit;
   private final int pageRowCountLimit;
   private final boolean pageWriteChecksumEnabled;
   private final ColumnProperty<ByteStreamSplitMode> byteStreamSplitEnabled;
@@ -153,6 +155,7 @@ private ParquetProperties(Builder builder) {
     this.maxBloomFilterBytes = builder.maxBloomFilterBytes;
     this.adaptiveBloomFilterEnabled = builder.adaptiveBloomFilterEnabled.build();
     this.numBloomFilterCandidates = builder.numBloomFilterCandidates.build();
+    this.rowGroupRowCountLimit = builder.rowGroupRowCountLimit;
     this.pageRowCountLimit = builder.pageRowCountLimit;
     this.pageWriteChecksumEnabled = builder.pageWriteChecksumEnabled;
     this.byteStreamSplitEnabled = builder.byteStreamSplitEnabled.build();
@@ -302,6 +305,10 @@ public boolean estimateNextSizeCheck() {
     return estimateNextSizeCheck;
   }
 
+  public int getRowGroupRowCountLimit() {
+    return rowGroupRowCountLimit;
+  }
+
   public int getPageRowCountLimit() {
     return pageRowCountLimit;
   }
@@ -400,6 +407,7 @@ public static class Builder {
     private final ColumnProperty.Builder<Boolean> adaptiveBloomFilterEnabled;
     private final ColumnProperty.Builder<Integer> numBloomFilterCandidates;
     private final ColumnProperty.Builder<Boolean> bloomFilterEnabled;
+    private int rowGroupRowCountLimit = DEFAULT_ROW_GROUP_ROW_COUNT_LIMIT;
     private int pageRowCountLimit = DEFAULT_PAGE_ROW_COUNT_LIMIT;
     private boolean pageWriteChecksumEnabled = DEFAULT_PAGE_WRITE_CHECKSUM_ENABLED;
     private final ColumnProperty.Builder<ByteStreamSplitMode> byteStreamSplitEnabled;
@@ -679,6 +687,12 @@ public Builder withBloomFilterEnabled(String columnPath, boolean enabled) {
       return this;
     }
 
+    public Builder withRowGroupRowCountLimit(int rowCount) {
+      Preconditions.checkArgument(rowCount > 0, "Invalid row count limit for row groups: %s", rowCount);
+      rowGroupRowCountLimit = rowCount;
+      return this;
+    }
+
     public Builder withPageRowCountLimit(int rowCount) {
       Preconditions.checkArgument(rowCount > 0, "Invalid row count limit for pages: %s", rowCount);
       pageRowCountLimit = rowCount;

parquet-hadoop/README.md

@@ -266,12 +266,16 @@ conf.set("parquet.bloom.filter.fpp#column.path", 0.02)
 
 ---
 
-
 **Property:** `parquet.decrypt.off-heap.buffer.enabled`  
 **Description:** Whether to use direct buffers to decrypt encrypted files. This should be set to 
 true if the reader is using a `DirectByteBufferAllocator`
 **Default value:** `false`
 
+---
+
+**Property:** `parquet.block.row.count.limit`
+**Description:** The maximum number of rows per row group.
+**Default value:** `2147483647` (Integer.MAX_VALUE)
 
 ---
 **Property:** `parquet.page.row.count.limit`  

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

@@ -48,8 +48,8 @@ class InternalParquetRecordWriter<T> {
   private final WriteSupport<T> writeSupport;
   private final MessageType schema;
   private final Map<String, String> extraMetaData;
-  private final long rowGroupSize;
   private long rowGroupSizeThreshold;
+  private final int rowGroupRecordCountThreshold;
   private long nextRowGroupSize;
   private final BytesInputCompressor compressor;
   private final boolean validating;
@@ -91,8 +91,8 @@ public InternalParquetRecordWriter(
     this.writeSupport = Objects.requireNonNull(writeSupport, "writeSupport cannot be null");
     this.schema = schema;
     this.extraMetaData = extraMetaData;
-    this.rowGroupSize = rowGroupSize;
     this.rowGroupSizeThreshold = rowGroupSize;
+    this.rowGroupRecordCountThreshold = props.getRowGroupRowCountLimit();
     this.nextRowGroupSize = rowGroupSizeThreshold;
     this.compressor = compressor;
     this.validating = validating;
@@ -166,9 +166,16 @@ public long getDataSize() {
   }
 
   private void checkBlockSizeReached() throws IOException {
-    if (recordCount
-        >= recordCountForNextMemCheck) { // checking the memory size is relatively expensive, so let's not do it
-      // for every record.
+    if (recordCount >= rowGroupRecordCountThreshold) {
+      LOG.debug("record count reaches threshold: flushing {} records to disk.", recordCount);
+      flushRowGroupToStore();
+      initStore();
+      recordCountForNextMemCheck = min(
+          max(props.getMinRowCountForPageSizeCheck(), recordCount / 2),
+          props.getMaxRowCountForPageSizeCheck());
+      this.lastRowGroupEndPos = parquetFileWriter.getPos();
+    } else if (recordCount >= recordCountForNextMemCheck) {
+      // checking the memory size is relatively expensive, so let's not do it for every record.
       long memSize = columnStore.getBufferedSize();
       long recordSize = memSize / recordCount;
       // flush the row group if it is within ~2 records of the limit