ORC-22. Fixes #3 - Allow InputStreams to set the natural read size for their filesystem

c++/include/orc/OrcFile.hh

@@ -42,6 +42,12 @@ namespace orc {
      */
     virtual uint64_t getLength() const = 0;
 
+    /**
+     * Get the natural size for reads.
+     * @return the number of bytes that should be read at once
+     */
+    virtual uint64_t getNaturalReadSize() const = 0;
+
     /**
      * Read length bytes from the file starting at offset into
      * the buffer starting at buf.

c++/src/Compression.cc

@@ -69,20 +69,20 @@ namespace orc {
   SeekableArrayInputStream::SeekableArrayInputStream
                (const unsigned char* values,
                 uint64_t size,
-                int64_t blkSize
+                uint64_t blkSize
                 ): data(reinterpret_cast<const char*>(values)) {
     length = size;
     position = 0;
-    blockSize = blkSize == -1 ? length : static_cast<uint64_t>(blkSize);
+    blockSize = blkSize == 0 ? length : static_cast<uint64_t>(blkSize);
   }
 
   SeekableArrayInputStream::SeekableArrayInputStream(const char* values,
                                                      uint64_t size,
-                                                     int64_t blkSize
+                                                     uint64_t blkSize
                                                      ): data(values) {
     length = size;
     position = 0;
-    blockSize = blkSize == -1 ? length : static_cast<uint64_t>(blkSize);
+    blockSize = blkSize == 0 ? length : static_cast<uint64_t>(blkSize);
   }
 
   bool SeekableArrayInputStream::Next(const void** buffer, int*size) {
@@ -135,17 +135,15 @@ namespace orc {
     return result.str();
   }
 
-  static uint64_t computeBlock(int64_t request, uint64_t length) {
-    return std::min(length,
-                    static_cast<uint64_t>(request < 0 ?
-                                          256 * 1024 : request));
+  static uint64_t computeBlock(uint64_t request, uint64_t length) {
+    return std::min(length, request == 0 ? 256 * 1024 : request);
   }
 
   SeekableFileInputStream::SeekableFileInputStream(InputStream* stream,
                                                    uint64_t offset,
                                                    uint64_t byteCount,
                                                    MemoryPool& _pool,
-                                                   int64_t _blockSize
+                                                   uint64_t _blockSize
                                                    ):pool(_pool),
                                                      input(stream),
                                                      start(offset),

c++/src/Compression.hh

@@ -70,10 +70,10 @@ namespace orc {
   public:
     SeekableArrayInputStream(const unsigned char* list,
                              uint64_t length,
-                             int64_t block_size = -1);
+                             uint64_t block_size = 0);
     SeekableArrayInputStream(const char* list,
                              uint64_t length,
-                             int64_t block_size = -1);
+                             uint64_t block_size = 0);
     virtual ~SeekableArrayInputStream();
     virtual bool Next(const void** data, int*size) override;
     virtual void BackUp(int count) override;
@@ -102,7 +102,7 @@ namespace orc {
                             uint64_t offset,
                             uint64_t byteCount,
                             MemoryPool& pool,
-                            int64_t blockSize = -1);
+                            uint64_t blockSize = 0);
     virtual ~SeekableFileInputStream();
 
     virtual bool Next(const void** data, int*size) override;

c++/src/OrcFile.cc

@@ -57,6 +57,10 @@ namespace orc {
       return totalLength;
     }
 
+    uint64_t getNaturalReadSize() const override {
+      return 128 * 1024;
+    }
+
     void read(void* buf,
               uint64_t length,
               uint64_t offset) override {

c++/src/Reader.cc

@@ -1316,10 +1316,7 @@ namespace orc {
                          (new SeekableFileInputStream(stream.get(),
                                                       stripeFooterStart,
                                                       stripeFooterLength,
-                                                      memoryPool,
-                                                      static_cast<int64_t>
-                                                      (blockSize)
-                                                      )),
+                                                      memoryPool)),
                          blockSize,
                          memoryPool);
     proto::StripeFooter result;
@@ -1400,9 +1397,8 @@ namespace orc {
       if (stream.has_kind() &&
           stream.kind() == kind &&
           stream.column() == static_cast<uint64_t>(columnId)) {
-        int64_t myBlock = static_cast<int64_t>(shouldStream ?
-                                         1024 * 1024 :
-                                         stream.length());
+        uint64_t myBlock = shouldStream ? input.getNaturalReadSize():
+          stream.length();
         return createDecompressor(reader.getCompression(),
                                   std::unique_ptr<SeekableInputStream>
                                   (new SeekableFileInputStream

tools/test/TestReader.cc

@@ -2914,6 +2914,7 @@ class MockInputStream: public InputStream {
   MOCK_CONST_METHOD0(getLength, uint64_t());
   MOCK_CONST_METHOD0(getName, const std::string&());
   MOCK_METHOD3(read, void (void*, uint64_t, uint64_t));
+  MOCK_CONST_METHOD0(getNaturalReadSize, uint64_t());
 };
 
 MockInputStream::~MockInputStream() {