ORC-1986: Trigger flush stripe for large input rows

For large input rows, the stripe may excessively large , requiring more memory for both reading and writing one strip.
We can check the tree write size in bytes and flush the strip even when the input rows count is less than 5000.

Author: WanKun

java/core/src/java/org/apache/orc/OrcConf.java

@@ -118,6 +118,11 @@ public enum OrcConf {
       "If the number of distinct keys in a dictionary is greater than this\n" +
           "fraction of the total number of non-null rows, turn off \n" +
           "dictionary encoding.  Use 1 to always use dictionary encoding."),
+  DICTIONARY_MAX_SIZE_IN_BYTES("orc.dictionary.maxSizeInBytes",
+      "hive.exec.orc.dictionary.maxSizeInBytes",
+      16 * 1024 * 1024,
+      "If the total size of the dictionary is greater than this\n" +
+          ", turn off dictionary encoding. Use 0 to disable this check."),
   ROW_INDEX_STRIDE_DICTIONARY_CHECK("orc.dictionary.early.check",
       "hive.orc.row.index.stride.dictionary.check",
       true,
@@ -182,6 +187,10 @@ public enum OrcConf {
       "added to all of the writers.  Valid range is [1,10000] and is primarily meant for" +
       "testing.  Setting this too low may negatively affect performance."
         + " Use orc.stripe.row.count instead if the value larger than orc.stripe.row.count."),
+  STRIPE_SIZE_CHECK("orc.stripe.size.check", "hive.exec.orc.default.stripe.size.check",
+      128L * 1024 * 1024,
+      "Flush stripe if the tree writer size in bytes is larger than this, " +
+          "use 0 to disable this check."),
   OVERWRITE_OUTPUT_FILE("orc.overwrite.output.file", "orc.overwrite.output.file", false,
     "A boolean flag to enable overwriting of the output file if it already exists.\n"),
   IS_SCHEMA_EVOLUTION_CASE_SENSITIVE("orc.schema.evolution.case.sensitive",

java/core/src/java/org/apache/orc/impl/WriterImpl.java

@@ -112,6 +112,7 @@ public class WriterImpl implements WriterInternal, MemoryManager.Callback {
   private long previousAllocation = -1;
   private long memoryLimit;
   private final long ROWS_PER_CHECK;
+  private final long STRIPE_SIZE_CHECK;
   private long rowsSinceCheck = 0;
   private final OrcFile.Version version;
   private final Configuration conf;
@@ -221,6 +222,7 @@ public WriterImpl(FileSystem fs,
     // ensure that we are able to handle callbacks before we register ourselves
     ROWS_PER_CHECK = Math.min(opts.getStripeRowCountValue(),
         OrcConf.ROWS_BETWEEN_CHECKS.getLong(conf));
+    STRIPE_SIZE_CHECK = OrcConf.STRIPE_SIZE_CHECK.getLong(conf);
     this.stripeRowCount= opts.getStripeRowCountValue();
     this.stripeSize = opts.getStripeSize();
     memoryLimit = stripeSize;
@@ -325,9 +327,9 @@ public boolean checkMemory(double newScale) throws IOException {
   }
 
   private boolean checkMemory() throws IOException {
-    if (rowsSinceCheck >= ROWS_PER_CHECK) {
+    long size = treeWriter.estimateMemory();
+    if (rowsSinceCheck >= ROWS_PER_CHECK || (STRIPE_SIZE_CHECK > 0 && size > STRIPE_SIZE_CHECK)) {
       rowsSinceCheck = 0;
-      long size = treeWriter.estimateMemory();
       if (LOG.isDebugEnabled()) {
         LOG.debug("ORC writer " + physicalWriter + " size = " + size +
             " memoryLimit = " + memoryLimit + " rowsInStripe = " + rowsInStripe +

java/core/src/java/org/apache/orc/impl/writer/StringBaseTreeWriter.java

@@ -59,6 +59,7 @@ public abstract class StringBaseTreeWriter extends TreeWriterBase {
   // If the number of keys in a dictionary is greater than this fraction of
   //the total number of non-null rows, turn off dictionary encoding
   private final double dictionaryKeySizeThreshold;
+  private final long dictionaryMaxSizeInBytes;
   protected Dictionary dictionary;
   protected boolean useDictionaryEncoding = true;
   private boolean isDirectV2 = true;
@@ -101,6 +102,7 @@ private static Dictionary createDict(Configuration conf) {
     rowIndexValueCount.add(0L);
     buildIndex = context.buildIndex();
     dictionaryKeySizeThreshold = context.getDictionaryKeySizeThreshold(id);
+    dictionaryMaxSizeInBytes = OrcConf.DICTIONARY_MAX_SIZE_IN_BYTES.getLong(conf);
     strideDictionaryCheck =
         OrcConf.ROW_INDEX_STRIDE_DICTIONARY_CHECK.getBoolean(conf);
     if (dictionaryKeySizeThreshold <= 0.0) {
@@ -118,7 +120,9 @@ private void checkDictionaryEncoding() {
       // based on whether or not the fraction of distinct keys over number of
       // non-null rows is less than the configured threshold
       float ratio = rows.size() > 0 ? (float) (dictionary.size()) / rows.size() : 0.0f;
-      useDictionaryEncoding = !isDirectV2 || ratio <= dictionaryKeySizeThreshold;
+      useDictionaryEncoding = !isDirectV2 || (ratio <= dictionaryKeySizeThreshold &&
+          (dictionaryMaxSizeInBytes <= 0 ||
+              dictionary.getSizeInBytes() <= dictionaryMaxSizeInBytes));
       doneDictionaryCheck = true;
     }
   }