[DWRF] Directly encoded ints optimization

Summary:
Analogous to my earlier diff "Directly encoded string optimization"

When the direct encoding for ints is carried over, we write the int to a long array first, and then to a stream to be compressed in the write stripe stage. The writing to the stream to be compressed can be done directly rather than going through the long array first. This also means that more data can be written into stripes since we store compressed data in memory, rather than the original uncompressed ints.

Test Plan: Ptests pass.
[DWRF] Avoid OOMs with many dynamic partitions/columns

Summary:
Sometimes queries will produce so many dynamic partitions with so many columns that the memory used just to initialize the writers will exceed the allocation given by the memory manager.  If this happens there's nothing the memory manager can do, and we tend to OOM.  Best case, we flush after every 10,000 rows.

To help avoid such cases, I added a check to the memory manager, so that if a Writer requests more than it's allocation just to initialize itself, dictionary encoding will be turned off on all writers (both new and old).  This greatly reduces the amount of memory required, to the point where we can generally avoid an OOM.  To compensate for the lack of dictionary compression, I also increased the Zlib level to 6 if we are using that compressor.

Unfortunately, when this happens ORC is slower than RC for writes.  To help compensate for this I added two optimizations.

If a writer flushes twice in a row on memory checks, and another writer does not flush for those two same memory checks, this implies that one writer is getting much more data than another.  To help with this I reallocate some memory from the unpopular writer to the popular writer, helping to reduce how often it needs to flush.

Also, when we flush the stripe, in the OutStream, if there is anything left in the buffer, we flush that to the OutputReceiver, and then immediately after flush the OutputReceivers data to disk.  We then have to reallocate buffers for all these streams when we start writing again.  To help with this, I allocated a buffer that can be shared across the streams for that list bit of data.  This greatly reduces the number of buffers that need to be allocated (e.g. present streams typically only need a single buffer) which helps with CPU.

It is still slower in the case I tried, so more work will be needed in the future.

Test Plan:
Ptests pass.  Added two new unit tests.

Tried writing to the 7 tables we saw fail in production, none of them experienced an OOM.
[DWRF] Don't read in all data at once

Summary:
Right now the way DWRF works is that it reads in all data for a stripe (minus excluded columns) from HDFS at once in a giant buffer and holds that in memory until the stripe is done being read.  The primary goals of this diff are to avoid this for 2 reasons:

1) We can null out buffers containing compressed index, dictionary data once they've been decompressed because we don't need the compressed data after this point.
2) We can only hold data for the current index stride (or a few strides) in memory (since this is the only data we'll be reading in the short term)

These should combine to make it DWRF much more memory efficient on read.

I will comment on this diff inline shortly (as there's a lot going on here).

Test Plan:
Ptests pass.

I did some performance testing too.

I tried preads instead of readfully, but readfully outperformed it, presumably because we read so much data in the same general area of the file.

I ran the perf tests, I haven't gotten conclusive results yet (I have to clean up some bad data first) but it looks like it's still at least 20% faster than RC (we were 25% faster originally).  For queries like count(*) we are significantly faster, I forget the exact number, but it's mostly due to the fact that we are reading a small fraction of the data from HDFS compared to before (I think just the stripe footers, dictionaries and indexes, this can be further optimized away in the future).  If we read e.g. every other column, we're about the same as before because in both cases we were reading a chunk of data for each column, rather than all the data as a giant chunk, which is similar to what we do now.  That 5% hit comes when we read all columns, this involves many more reads from HDFS than the single read from before, so this is expected, we make some of it back with fewer GCs, but not all of it.  I think in the future we can improve this, e.g. if we the nth stride for each column sequentially in the file we might get better performance (but there would be a cost since n strides for a single column would not be sequential).  Anyway, there's more work to do here.

In terms of memory, I was looking at a table with a lot of columns and would probably require a lot of memory (I don't know what table Presto was specifically having trouble with).  The amount of memory needed by the Reader dropped from ~43Mb to ~20Mb and could be as low as 18Mb adjusting configurations.  Tables with fewer columns should see a much more dramatic drop since they will have more index strides in a stripe (smaller rows = more rows per stripe).
[DWRF] Add config to start writing in low memory mode

Summary:
I got a task yesterday for a query that's failing with OOMs.

This query was writing a column with strings that could be > 3Mb.  DWRF cannot handle this well, mainly due to dictionaries and buffering of values.  If we turn on low memory mode immediately and turn off buffering the query can succeed.

In the future we can work on automatically detecting these, but this seems like a rare case and I didn't want to make queries less efficient by trying to detect these rare cases.

Test Plan: Running ptests.
[DWRF] bug in seek with chunks

Summary:
There is a bug (I think the comments in the test describe it well) where if you read from a stride in a chunk and then read from another stride at the same byte offset in another chunk the compressed offset will not be updated and you will get incorrect data.  This is because previousOffset will equal newCompressedOffset because both will be the byte offsets of the strides in the chunks.  We need to additionally check that we haven't read new data from disk (i.e. we aren't in a new chunk).

snapshottests

Test Plan:
Running ptests.

Verified it fixed the failing query in snapshot tests.

Added a very specific testcase.
[DWRF] Overflow error in large files

Summary: The InStream's offset is typed as an int which worked fine when we'd only pass in a byte array to it.  Now that we're passing the offset in the file, if the file is large enough, this can cause an error.  Changed the type to long.

Test Plan:
Added a testcase which simulates reading from a large file.

Ptests pass.
Support methods to fetch primitive from OrcLazyObjects

Summary:
Added methods like materializeInt(), materializeFloat(), ... to the OrcLazyObject class. Now,
the clients can use these methods to fetch primitives. To ensure backward compatibility,
the materlialize() method can be used in conjunction with materializeXXXX() methods.

The materializeXXXX() methods do not construct HadoopWritable objects now. It will be created
on  demand if the materialize() method is called immediately after materializeXXXX() method
without any intervening next() call.

Test Plan:
Added new tests to TestOrcFile.java and made sure the tests succeded. The new tests
invoke the materializeXXXX() operations in conjunction with materialize() method
to make sure the materializeXXXX() and materialize methods can play nicely with
each other.
API to read primitives broke deep copies

Summary:
The copy constructor for OrcLazyObjects should copy whether or not a writable object has already been created, otherwise the resulting copies can fail in materialize.

For reference, these deep copies are used by Hive in group by operators to copy grouping keys into the hash map.

Test Plan:
Added a test case for deep copies, which fails before the fix.

Running ptests.
[DWRF] Store unique values in stride level dictionary

Summary: Unique values don't need to be kept in the stripe level dictionary, storing them in a stride level dictionary (so they can still be sorted) is sufficient.  This helps with memory issues where there are some small strings that show up very often, and many large strings that show up once.  The stripe level dictionary just contains the small strings, and the large strings show up in the stride level dictionary, so they don't all show up in memory at once.

Test Plan:
Ptests pass.
[DWRF] Make reading from HDFS lazily optional

Summary:
Reading lazily from HDFS causes a significant decrease in CPU performance, which is unavoidable as no matter how we arrange the data in the file some queries will always effectively do random reads in the file.

Adding an option to read lazily and defaulting it to off for now.  Applications strained for memory can turn it on by default if they choose, but for now, we are running into few enough issues (one application on one table) that it's probably not worth turning it on by default everywhere yet.

Test Plan:
Added a testcase.

ptests pass.
[DWRF] bug casts all integers to short from the stride dictionaries

Summary: a left over cast, casts all numeric values from stride dictionaries to short causing wrong results

Test Plan: adjusted TestOrcFile so values in int and long columns can only be int or long, this fails before the fix
[DWRF] Dynamically turn on lazy HDFS reads if size is above threshold

Summary: Added a configurable threshold for the compressed size of the data to be read in by a DWRF RecordReader.  If the size of the data exceeds this threshold, it uses lazy reads to read that stripe, otherwise it performs the read eagerly.

Test Plan:
ptests pass

This fixed a few OOMs in snapshot tests (these must have been tests that were not converted to ORC the last time we rolled out, because they OOM with the currently deployed code as well).
[DWRF] Bug in StringTreeWriter across carry overs

Summary:
In the StringTreeWriter, if dictionary encoding is not used, and this is carried over, the dictionary will be nulled out to save memory.   If the number of strides between determining the encoding is exceeded, and then dictionary encoding is decided upon, the dictionary will be initialized without any parameters.

Before this just meant the data was always sorted, a bug, but it wouldn't lead to incorrect results.  Now however, the writer would expect the data to be sorted suitably for stride dictionaries (if the config was set) but the data would not be.  This led to incorrect results.

Test Plan:
Fixed the bug I saw in snapshot tests.

Added a few tests.  Only testStrideDictionariesWithoutStripeCarryover tests this bug.  The rest were tests I wrote in search of the bug, they seemed good, so I'm leaving them in.

ptests pass
[DWRF] Add close method to lazy tree readers

Summary:
One major source of OOMs in snapshot tests was the fact that when we start a new stripe or open a new file while reading DWRF data, there still maybe pointers to the previous row floating around (e.g. there's one in RecordReaderImpl itself).  This means that row can't be cleaned up from memory, and more importantly, the dictionaries, and other buffers it holds have to remain in memory.  So, as we read in a new stripe, since we unpack dictionaries eagerly, we have potentially two dictionaries per column in memory (plus some extra stuff).

When we start a new stripe, we must be in the next method, and the contract of this method means that it's safe to null out/clear all the structures the previous row pointed to.  Similarly, if we're opening a new file (within the CombineHiveRecordReader), close was called on the previous file, and the contract of close allows us to clear out all this memory.

Test Plan:
Ptests pass (note that many of the tests involve reading data from multiple stripes, so this code should be tested by the existing ones)

This fixes all but maybe one OOM in snapshot tests.
[DWRF] Add optional limit on the size of dictionaries

Summary:
Added a parameter that limits the size of the uncompressed dictionaries (by default to 100 Mb).  In OOMs we've been seeing, it's been pretty consistently because the size of the dictionaries are surprisingly large (e.g. 150-200 Mb).  These extreme sizes aren't strictly necessary for compression gains.  I converted all the inputs to snapshot tests using this and the total increase in size was about 0.002% (insignificant).

About 80 tables decreased in size, 481 remained unchanged, and 109 increased in size.  Of those 3 increased in size by more than 1% (1%, 3%, 4% respectively) on one decreased in size by almost 2%.

Given that this doesn't seem to have a significant effect on compression in most cases, and it could help with OOMs I think it's a reasonable trade off, and we can always turn it off on cold data or if we size a significant detriment.

Test Plan: ptests pass