TestWriter.cc

/**
 * 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.
 */

#include "orc/ColumnPrinter.hh"
#include "orc/OrcFile.hh"

#include "MemoryInputStream.hh"
#include "MemoryOutputStream.hh"

#include "wrap/gmock.h"
#include "wrap/gtest-wrapper.h"

#include <cmath>
#include <ctime>
#include <sstream>

#ifdef __clang__
  DIAGNOSTIC_IGNORE("-Wmissing-variable-declarations")
#endif

namespace orc {

  using ::testing::TestWithParam;
  using ::testing::Values;

  const int DEFAULT_MEM_STREAM_SIZE = 100 * 1024 * 1024; // 100M

  std::unique_ptr<Writer> createWriter(
                                      uint64_t stripeSize,
                                      uint64_t compresionblockSize,
                                      CompressionKind compression,
                                      const Type& type,
                                      MemoryPool* memoryPool,
                                      OutputStream* stream,
                                      FileVersion version,
                                      uint64_t stride = 0){
    WriterOptions options;
    options.setStripeSize(stripeSize);
    options.setCompressionBlockSize(compresionblockSize);
    options.setCompression(compression);
    options.setMemoryPool(memoryPool);
    options.setRowIndexStride(stride);
    options.setFileVersion(version);
    return createWriter(type, stream, options);
  }

  std::unique_ptr<Reader> createReader(
                                      MemoryPool * memoryPool,
                                      std::unique_ptr<InputStream> stream) {
    ReaderOptions options;
    options.setMemoryPool(*memoryPool);
    return createReader(std::move(stream), options);
  }

  std::unique_ptr<RowReader> createRowReader(Reader* reader) {
    RowReaderOptions rowReaderOpts;
    return reader->createRowReader(rowReaderOpts);
  }

  class WriterTest : public TestWithParam<FileVersion> {
    // You can implement all the usual fixture class members here.
    // To access the test parameter, call GetParam() from class
    // TestWithParam<T>.
    virtual void SetUp();

    protected:
      FileVersion fileVersion;
    public:
      WriterTest(): fileVersion(FileVersion::v_0_11()) {}
  };

  void WriterTest::SetUp() {
    fileVersion = GetParam();
  }

  TEST_P(WriterTest, writeEmptyFile) {
    MemoryOutputStream memStream(DEFAULT_MEM_STREAM_SIZE);
    MemoryPool* pool = getDefaultPool();
    ORC_UNIQUE_PTR<Type> type(Type::buildTypeFromString("struct<col1:int>"));

    uint64_t stripeSize = 16 * 1024; // 16K
    uint64_t compressionBlockSize = 1024; // 1k

    std::unique_ptr<Writer> writer = createWriter(
                                      stripeSize,
                                      compressionBlockSize,
                                      CompressionKind_ZLIB,
                                      *type,
                                      pool,
                                      &memStream,
                                      fileVersion);
    writer->close();

    std::unique_ptr<InputStream> inStream(
            new MemoryInputStream (memStream.getData(), memStream.getLength()));
    std::unique_ptr<Reader> reader = createReader(
                                                pool,
                                                std::move(inStream));
    std::unique_ptr<RowReader> rowReader = createRowReader(reader.get());
    EXPECT_EQ(fileVersion, reader->getFormatVersion());
    EXPECT_EQ(WriterVersion_ORC_135, reader->getWriterVersion());
    EXPECT_EQ(0, reader->getNumberOfRows());

    WriterId writerId = WriterId::ORC_CPP_WRITER;
    EXPECT_EQ(writerId, reader->getWriterId());
    EXPECT_EQ(1, reader->getWriterIdValue());

    std::unique_ptr<ColumnVectorBatch> batch = rowReader->createRowBatch(1024);
    EXPECT_FALSE(rowReader->next(*batch));
  }

  TEST_P(WriterTest, writeIntFileOneStripe) {
    MemoryOutputStream memStream(DEFAULT_MEM_STREAM_SIZE);
    MemoryPool* pool = getDefaultPool();
    ORC_UNIQUE_PTR<Type> type(Type::buildTypeFromString("struct<col1:int>"));

    uint64_t stripeSize = 16 * 1024; // 16K
    uint64_t compressionBlockSize = 1024; // 1k

    std::unique_ptr<Writer> writer = createWriter(
                                      stripeSize,
                                      compressionBlockSize,
                                      CompressionKind_ZLIB,
                                      *type,
                                      pool,
                                      &memStream,
                                      fileVersion);
    std::unique_ptr<ColumnVectorBatch> batch = writer->createRowBatch(1024);
    StructVectorBatch* structBatch =
      dynamic_cast<StructVectorBatch *>(batch.get());
    LongVectorBatch* longBatch =
      dynamic_cast<LongVectorBatch *>(structBatch->fields[0]);

    for (uint64_t i = 0; i < 1024; ++i) {
      longBatch->data[i] = static_cast<int64_t>(i);
    }
    structBatch->numElements = 1024;
    longBatch->numElements = 1024;

    writer->add(*batch);

    for (uint64_t i = 1024; i < 2000; ++i) {
      longBatch->data[i - 1024] = static_cast<int64_t>(i);
    }
    structBatch->numElements = 2000 - 1024;
    longBatch->numElements = 2000 - 1024;

    writer->add(*batch);
    writer->addUserMetadata("name0","value0");
    writer->addUserMetadata("name1","value1");
    writer->close();

    std::unique_ptr<InputStream> inStream(
            new MemoryInputStream (memStream.getData(), memStream.getLength()));
    std::unique_ptr<Reader> reader = createReader(
                                                pool,
                                                std::move(inStream));
    std::unique_ptr<RowReader> rowReader = createRowReader(reader.get());
    EXPECT_EQ(2000, reader->getNumberOfRows());

    batch = rowReader->createRowBatch(2048);
    EXPECT_TRUE(rowReader->next(*batch));
    EXPECT_EQ(2000, batch->numElements);
    EXPECT_FALSE(rowReader->next(*batch));

    std::list<std::string> keys = reader->getMetadataKeys();
    EXPECT_EQ(keys.size(), 2);
    std::list<std::string>::const_iterator itr = keys.begin();
    EXPECT_EQ(*itr, "name0");
    EXPECT_EQ(reader->getMetadataValue(*itr), "value0");
    itr++;
    EXPECT_EQ(*itr, "name1");
    EXPECT_EQ(reader->getMetadataValue(*itr), "value1");

    for (uint64_t i = 0; i < 2000; ++i) {
      structBatch = dynamic_cast<StructVectorBatch *>(batch.get());
      longBatch = dynamic_cast<LongVectorBatch *>(structBatch->fields[0]);
      EXPECT_EQ(i, longBatch->data[i]);
    }
  }

  TEST_P(WriterTest, writeIntFileMultipleStripes) {
    MemoryOutputStream memStream(DEFAULT_MEM_STREAM_SIZE);
    MemoryPool* pool = getDefaultPool();
    ORC_UNIQUE_PTR<Type> type(Type::buildTypeFromString("struct<col1:int>"));

    uint64_t stripeSize = 1024; // 1K
    uint64_t compressionBlockSize = 1024; // 1k

    std::unique_ptr<Writer> writer = createWriter(
                                      stripeSize,
                                      compressionBlockSize,
                                      CompressionKind_ZLIB,
                                      *type,
                                      pool,
                                      &memStream,
                                      fileVersion);
    std::unique_ptr<ColumnVectorBatch> batch = writer->createRowBatch(65535);
    StructVectorBatch* structBatch =
      dynamic_cast<StructVectorBatch*>(batch.get());
    LongVectorBatch* longBatch =
      dynamic_cast<LongVectorBatch*>(structBatch->fields[0]);

    for (uint64_t j = 0; j < 10; ++j) {
      for (uint64_t i = 0; i < 65535; ++i) {
        longBatch->data[i] = static_cast<int64_t>(i);
      }
      structBatch->numElements = 65535;
      longBatch->numElements = 65535;

      writer->add(*batch);
    }

    writer->close();

    std::unique_ptr<InputStream> inStream(
            new MemoryInputStream (memStream.getData(), memStream.getLength()));
    std::unique_ptr<Reader> reader = createReader(
                                                pool,
                                                std::move(inStream));
    std::unique_ptr<RowReader> rowReader = createRowReader(reader.get());
    EXPECT_EQ(655350, reader->getNumberOfRows());

    batch = rowReader->createRowBatch(65535);
    for (uint64_t j = 0; j < 10; ++j) {
      EXPECT_TRUE(rowReader->next(*batch));
      EXPECT_EQ(65535, batch->numElements);

      for (uint64_t i = 0; i < 65535; ++i) {
        structBatch = dynamic_cast<StructVectorBatch*>(batch.get());
        longBatch = dynamic_cast<LongVectorBatch*>(structBatch->fields[0]);
        EXPECT_EQ(i, longBatch->data[i]);
      }
    }
    EXPECT_FALSE(rowReader->next(*batch));
  }

  TEST_P(WriterTest, writeStringAndBinaryColumn) {
    MemoryOutputStream memStream(DEFAULT_MEM_STREAM_SIZE);
    MemoryPool * pool = getDefaultPool();
    std::unique_ptr<Type> type(Type::buildTypeFromString(
      "struct<col1:string,col2:binary>"));

    uint64_t stripeSize = 1024;     // 1K
    uint64_t compressionBlockSize = 1024;      // 1k

    char dataBuffer[327675];
    uint64_t offset = 0;

    std::unique_ptr<Writer> writer = createWriter(stripeSize,
                                                  compressionBlockSize,
                                                  CompressionKind_ZLIB,
                                                  *type,
                                                  pool,
                                                  &memStream,
                                                  fileVersion);
    std::unique_ptr<ColumnVectorBatch> batch = writer->createRowBatch(65535);
    StructVectorBatch * structBatch =
      dynamic_cast<StructVectorBatch *>(batch.get());
    StringVectorBatch * strBatch =
      dynamic_cast<StringVectorBatch *>(structBatch->fields[0]);
    StringVectorBatch * binBatch =
      dynamic_cast<StringVectorBatch *>(structBatch->fields[1]);

    for (uint64_t i = 0; i < 65535; ++i) {
      std::ostringstream os;
      os << i;
      strBatch->data[i] = dataBuffer + offset;
      strBatch->length[i] = static_cast<int64_t>(os.str().size());
      binBatch->data[i] = dataBuffer + offset;
      binBatch->length[i] = static_cast<int64_t>(os.str().size());
      memcpy(dataBuffer + offset, os.str().c_str(), os.str().size());
      offset += os.str().size();
    }

    structBatch->numElements = 65535;
    strBatch->numElements = 65535;
    binBatch->numElements = 65535;

    writer->add(*batch);
    writer->close();

    std::unique_ptr<InputStream> inStream(
      new MemoryInputStream (memStream.getData(),
                             memStream.getLength()));
    std::unique_ptr<Reader> reader = createReader(pool, std::move(inStream));
    std::unique_ptr<RowReader> rowReader = createRowReader(reader.get());
    EXPECT_EQ(65535, reader->getNumberOfRows());

    batch = rowReader->createRowBatch(65535);
    EXPECT_EQ(true, rowReader->next(*batch));
    EXPECT_EQ(65535, batch->numElements);

    for (uint64_t i = 0; i < 65535; ++i) {
      structBatch = dynamic_cast<StructVectorBatch *>(batch.get());
      strBatch = dynamic_cast<StringVectorBatch *>(structBatch->fields[0]);
      binBatch = dynamic_cast<StringVectorBatch *>(structBatch->fields[1]);
      std::string str(
        strBatch->data[i],
        static_cast<size_t>(strBatch->length[i]));
      std::string bin(
        binBatch->data[i],
        static_cast<size_t>(binBatch->length[i]));
      EXPECT_EQ(i, static_cast<uint64_t>(atoi(str.c_str())));
      EXPECT_EQ(i, static_cast<uint64_t>(atoi(bin.c_str())));
    }

    EXPECT_FALSE(rowReader->next(*batch));
  }

  TEST_P(WriterTest, writeFloatAndDoubleColumn) {
    MemoryOutputStream memStream(DEFAULT_MEM_STREAM_SIZE);
    MemoryPool * pool = getDefaultPool();
    std::unique_ptr<Type> type(Type::buildTypeFromString(
      "struct<col1:double,col2:float>"));

    uint64_t stripeSize = 16 * 1024;
    uint64_t compressionBlockSize = 1024;
    uint64_t rowCount = 655350;

    std::vector<double> data(rowCount);
    for (uint64_t i = 0; i < rowCount; ++i) {
      data[i] = 100000 * (std::rand() * 1.0 / RAND_MAX);
    }

    std::unique_ptr<Writer> writer = createWriter(stripeSize,
                                                  compressionBlockSize,
                                                  CompressionKind_ZLIB,
                                                  *type,
                                                  pool,
                                                  &memStream,
                                                  fileVersion);
    std::unique_ptr<ColumnVectorBatch> batch = writer->createRowBatch(rowCount);
    StructVectorBatch * structBatch =
      dynamic_cast<StructVectorBatch *>(batch.get());
    DoubleVectorBatch * doubleBatch =
      dynamic_cast<DoubleVectorBatch *>(structBatch->fields[0]);
    DoubleVectorBatch * floatBatch =
      dynamic_cast<DoubleVectorBatch *>(structBatch->fields[1]);

    for (uint64_t i = 0; i < rowCount; ++i) {
      doubleBatch->data[i] = data[i];
      floatBatch->data[i] = data[i];
    }

    structBatch->numElements = rowCount;
    doubleBatch->numElements = rowCount;
    floatBatch->numElements = rowCount;

    writer->add(*batch);
    writer->close();

    std::unique_ptr<InputStream> inStream(
      new MemoryInputStream (memStream.getData(), memStream.getLength()));
    std::unique_ptr<Reader> reader = createReader(pool, std::move(inStream));
    std::unique_ptr<RowReader> rowReader = createRowReader(reader.get());
    EXPECT_EQ(rowCount, reader->getNumberOfRows());

    batch = rowReader->createRowBatch(rowCount);
    EXPECT_EQ(true, rowReader->next(*batch));
    EXPECT_EQ(rowCount, batch->numElements);

    structBatch = dynamic_cast<StructVectorBatch *>(batch.get());
    doubleBatch = dynamic_cast<DoubleVectorBatch *>(structBatch->fields[0]);
    floatBatch = dynamic_cast<DoubleVectorBatch *>(structBatch->fields[1]);
    for (uint64_t i = 0; i < rowCount; ++i) {
      EXPECT_TRUE(std::abs(data[i] - doubleBatch->data[i]) < 0.000001);
      EXPECT_TRUE(std::abs(static_cast<float>(data[i]) -
                           static_cast<float>(floatBatch->data[i])) < 0.000001f);
    }
    EXPECT_FALSE(rowReader->next(*batch));
  }

  TEST_P(WriterTest, writeShortIntLong) {
    MemoryOutputStream memStream(DEFAULT_MEM_STREAM_SIZE);
    MemoryPool * pool = getDefaultPool();
    std::unique_ptr<Type> type(Type::buildTypeFromString(
      "struct<col1:smallint,col2:int,col3:bigint>"));

    uint64_t stripeSize = 16 * 1024;
    uint64_t compressionBlockSize = 1024;
    uint64_t rowCount = 65535;

    std::unique_ptr<Writer> writer = createWriter(stripeSize,
                                                  compressionBlockSize,
                                                  CompressionKind_ZLIB,
                                                  *type,
                                                  pool,
                                                  &memStream,
                                                  fileVersion);
    std::unique_ptr<ColumnVectorBatch> batch = writer->createRowBatch(rowCount);
    StructVectorBatch * structBatch =
      dynamic_cast<StructVectorBatch *>(batch.get());
    LongVectorBatch * smallIntBatch =
      dynamic_cast<LongVectorBatch *>(structBatch->fields[0]);
    LongVectorBatch * intBatch =
      dynamic_cast<LongVectorBatch *>(structBatch->fields[1]);
    LongVectorBatch * bigIntBatch =
      dynamic_cast<LongVectorBatch *>(structBatch->fields[2]);

    for (uint64_t i = 0; i < rowCount; ++i) {
      smallIntBatch->data[i] = static_cast<int16_t>(i);
      intBatch->data[i] = static_cast<int32_t>(i);
      bigIntBatch->data[i] = static_cast<int64_t>(i);
    }
    structBatch->numElements = rowCount;
    smallIntBatch->numElements = rowCount;
    intBatch->numElements = rowCount;
    bigIntBatch->numElements = rowCount;

    writer->add(*batch);
    writer->close();

    std::unique_ptr<InputStream> inStream(
      new MemoryInputStream (memStream.getData(), memStream.getLength()));
    std::unique_ptr<Reader> reader = createReader(pool, std::move(inStream));
    std::unique_ptr<RowReader> rowReader = createRowReader(reader.get());
    EXPECT_EQ(rowCount, reader->getNumberOfRows());

    batch = rowReader->createRowBatch(rowCount);
    EXPECT_EQ(true, rowReader->next(*batch));

    structBatch = dynamic_cast<StructVectorBatch *>(batch.get());
    smallIntBatch = dynamic_cast<LongVectorBatch *>(structBatch->fields[0]);
    intBatch = dynamic_cast<LongVectorBatch *>(structBatch->fields[1]);
    bigIntBatch = dynamic_cast<LongVectorBatch *>(structBatch->fields[2]);
    for (uint64_t i = 0; i < rowCount; ++i) {
      EXPECT_EQ(static_cast<int16_t>(i), smallIntBatch->data[i]);
      EXPECT_EQ(static_cast<int32_t>(i), intBatch->data[i]);
      EXPECT_EQ(static_cast<int64_t>(i), bigIntBatch->data[i]);
    }
  }

  TEST_P(WriterTest, writeTinyint) {
    MemoryOutputStream memStream(DEFAULT_MEM_STREAM_SIZE);
    MemoryPool * pool = getDefaultPool();
    std::unique_ptr<Type> type(Type::buildTypeFromString(
      "struct<col1:tinyint>"));

    uint64_t stripeSize = 16 * 1024;
    uint64_t compressionBlockSize = 1024;
    uint64_t rowCount = 65535;

    std::unique_ptr<Writer> writer = createWriter(stripeSize,
                                                  compressionBlockSize,
                                                  CompressionKind_ZLIB,
                                                  *type,
                                                  pool,
                                                  &memStream,
                                                  fileVersion);
    std::unique_ptr<ColumnVectorBatch> batch = writer->createRowBatch(rowCount);
    StructVectorBatch * structBatch =
      dynamic_cast<StructVectorBatch *>(batch.get());
    LongVectorBatch * byteBatch =
      dynamic_cast<LongVectorBatch *>(structBatch->fields[0]);

    for (uint64_t i = 0; i < rowCount; ++i) {
      byteBatch->data[i] = static_cast<int8_t>(i);
    }
    structBatch->numElements = rowCount;
    byteBatch->numElements = rowCount;

    writer->add(*batch);
    writer->close();

    std::unique_ptr<InputStream> inStream(
      new MemoryInputStream (memStream.getData(), memStream.getLength()));
    std::unique_ptr<Reader> reader = createReader(pool, std::move(inStream));
    std::unique_ptr<RowReader> rowReader = createRowReader(reader.get());
    EXPECT_EQ(rowCount, reader->getNumberOfRows());

    batch = rowReader->createRowBatch(rowCount);
    EXPECT_EQ(true, rowReader->next(*batch));

    structBatch = dynamic_cast<StructVectorBatch *>(batch.get());
    byteBatch = dynamic_cast<LongVectorBatch *>(structBatch->fields[0]);
    for (uint64_t i = 0; i < rowCount; ++i) {
      EXPECT_EQ(static_cast<int8_t>(i), static_cast<int8_t>(byteBatch->data[i]));
    }
  }

  TEST_P(WriterTest, writeBooleanColumn) {
    MemoryOutputStream memStream(DEFAULT_MEM_STREAM_SIZE);
    MemoryPool* pool = getDefaultPool();
    std::unique_ptr<Type> type(Type::buildTypeFromString("struct<col1:boolean>"));

    uint64_t stripeSize = 16 * 1024;
    uint64_t compressionBlockSize = 1024;
    uint64_t rowCount = 65535;

    std::unique_ptr<Writer> writer = createWriter(stripeSize,
                                                  compressionBlockSize,
                                                  CompressionKind_ZLIB,
                                                  *type,
                                                  pool,
                                                  &memStream,
                                                  fileVersion);
    std::unique_ptr<ColumnVectorBatch> batch = writer->createRowBatch(rowCount);
    StructVectorBatch * structBatch =
      dynamic_cast<StructVectorBatch *>(batch.get());
    LongVectorBatch * byteBatch =
      dynamic_cast<LongVectorBatch *>(structBatch->fields[0]);

    for (uint64_t i = 0; i < rowCount; ++i) {
      byteBatch->data[i] = (i % 3) == 0 ? 1 : 0;
    }
    structBatch->numElements = rowCount;
    byteBatch->numElements = rowCount;

    writer->add(*batch);
    writer->close();

    std::unique_ptr<InputStream> inStream(
      new MemoryInputStream (memStream.getData(), memStream.getLength()));
    std::unique_ptr<Reader> reader = createReader(pool, std::move(inStream));
    std::unique_ptr<RowReader> rowReader = createRowReader(reader.get());
    EXPECT_EQ(rowCount, reader->getNumberOfRows());

    batch = rowReader->createRowBatch(rowCount);
    EXPECT_EQ(true, rowReader->next(*batch));

    structBatch = dynamic_cast<StructVectorBatch *>(batch.get());
    byteBatch = dynamic_cast<LongVectorBatch *>(structBatch->fields[0]);
    for (uint64_t i = 0; i < rowCount; ++i) {
      EXPECT_EQ((i % 3) == 0 ? 1 : 0, byteBatch->data[i]);
    }
  }

  TEST_P(WriterTest, writeDate) {
    MemoryOutputStream memStream(DEFAULT_MEM_STREAM_SIZE);
    MemoryPool* pool = getDefaultPool();
    std::unique_ptr<Type> type(Type::buildTypeFromString("struct<col1:date>"));

    uint64_t stripeSize = 16 * 1024;
    uint64_t compressionBlockSize = 1024;
    uint64_t rowCount = 1024;

    std::unique_ptr<Writer> writer = createWriter(stripeSize,
                                                  compressionBlockSize,
                                                  CompressionKind_ZLIB,
                                                  *type,
                                                  pool,
                                                  &memStream,
                                                  fileVersion);

    std::unique_ptr<ColumnVectorBatch> batch = writer->createRowBatch(rowCount);
    StructVectorBatch * structBatch =
      dynamic_cast<StructVectorBatch *>(batch.get());
    LongVectorBatch * longBatch =
      dynamic_cast<LongVectorBatch *>(structBatch->fields[0]);

    for (uint64_t i = 0; i < rowCount; ++i) {
      longBatch->data[i] = static_cast<int32_t>(i);
    }
    structBatch->numElements = rowCount;
    longBatch->numElements = rowCount;

    writer->add(*batch);
    writer->close();

    std::unique_ptr<InputStream> inStream(
      new MemoryInputStream (memStream.getData(), memStream.getLength()));
    std::unique_ptr<Reader> reader = createReader(pool, std::move(inStream));
    std::unique_ptr<RowReader> rowReader = createRowReader(reader.get());
    EXPECT_EQ(rowCount, reader->getNumberOfRows());

    batch = rowReader->createRowBatch(rowCount);
    EXPECT_EQ(true, rowReader->next(*batch));

    structBatch = dynamic_cast<StructVectorBatch *>(batch.get());
    longBatch = dynamic_cast<LongVectorBatch *>(structBatch->fields[0]);
    for (uint64_t i = 0; i < rowCount; ++i) {
      EXPECT_EQ(static_cast<int32_t>(i), longBatch->data[i]);
    }
  }

  TEST_P(WriterTest, writeTimestamp) {
    MemoryOutputStream memStream(DEFAULT_MEM_STREAM_SIZE);
    MemoryPool* pool = getDefaultPool();
    std::unique_ptr<Type> type(Type::buildTypeFromString("struct<col1:timestamp>"));

    uint64_t stripeSize = 16 * 1024;
    uint64_t compressionBlockSize = 1024;
    uint64_t rowCount = 102400;

    std::unique_ptr<Writer> writer = createWriter(stripeSize,
                                                  compressionBlockSize,
                                                  CompressionKind_ZLIB,
                                                  *type,
                                                  pool,
                                                  &memStream,
                                                  fileVersion);
    std::unique_ptr<ColumnVectorBatch> batch = writer->createRowBatch(rowCount);
    StructVectorBatch * structBatch =
      dynamic_cast<StructVectorBatch *>(batch.get());
    TimestampVectorBatch * tsBatch =
      dynamic_cast<TimestampVectorBatch *>(structBatch->fields[0]);

    std::vector<std::time_t> times(rowCount);
    for (uint64_t i = 0; i < rowCount; ++i) {
      time_t currTime = -14210715; // 1969-07-20 12:34:45
      times[i] = static_cast<int64_t>(currTime) + static_cast<int64_t >(i * 3660);
      tsBatch->data[i] = times[i];
      tsBatch->nanoseconds[i] = static_cast<int64_t>(i * 1000);
    }
    structBatch->numElements = rowCount;
    tsBatch->numElements = rowCount;

    writer->add(*batch);
    writer->close();

    std::unique_ptr<InputStream> inStream(
      new MemoryInputStream (memStream.getData(), memStream.getLength()));
    std::unique_ptr<Reader> reader = createReader(pool, std::move(inStream));
    std::unique_ptr<RowReader> rowReader = createRowReader(reader.get());
    EXPECT_EQ(rowCount, reader->getNumberOfRows());

    batch = rowReader->createRowBatch(rowCount);
    EXPECT_EQ(true, rowReader->next(*batch));

    structBatch = dynamic_cast<StructVectorBatch *>(batch.get());
    tsBatch = dynamic_cast<TimestampVectorBatch *>(structBatch->fields[0]);
    for (uint64_t i = 0; i < rowCount; ++i) {
      EXPECT_EQ(times[i], tsBatch->data[i]);
      EXPECT_EQ(i * 1000, tsBatch->nanoseconds[i]);
    }
  }

  TEST_P(WriterTest, writeCharAndVarcharColumn) {
    MemoryOutputStream memStream(DEFAULT_MEM_STREAM_SIZE);
    MemoryPool * pool = getDefaultPool();
    std::unique_ptr<Type> type(Type::buildTypeFromString(
      "struct<col1:char(3),col2:varchar(4)>"));

    uint64_t stripeSize = 1024;
    uint64_t compressionBlockSize = 1024;
    uint64_t rowCount = 65535;

    char dataBuffer[327675];
    uint64_t offset = 0;

    std::unique_ptr<Writer> writer = createWriter(stripeSize,
                                                  compressionBlockSize,
                                                  CompressionKind_ZLIB,
                                                  *type,
                                                  pool,
                                                  &memStream,
                                                  fileVersion);

    std::unique_ptr<ColumnVectorBatch> batch = writer->createRowBatch(rowCount);
    StructVectorBatch * structBatch =
      dynamic_cast<StructVectorBatch *>(batch.get());
    StringVectorBatch * charBatch =
      dynamic_cast<StringVectorBatch *>(structBatch->fields[0]);
    StringVectorBatch * varcharBatch =
      dynamic_cast<StringVectorBatch *>(structBatch->fields[1]);

    for (uint64_t i = 0; i < rowCount; ++i) {
      std::ostringstream os;
      os << i;
      charBatch->data[i] = dataBuffer + offset;
      charBatch->length[i] = static_cast<int64_t>(os.str().size());

      varcharBatch->data[i] = charBatch->data[i];
      varcharBatch->length[i] = charBatch->length[i];

      memcpy(dataBuffer + offset, os.str().c_str(), os.str().size());
      offset += os.str().size();
    }

    structBatch->numElements = rowCount;
    charBatch->numElements = rowCount;
    varcharBatch->numElements = rowCount;

    writer->add(*batch);
    writer->close();

    std::unique_ptr<InputStream> inStream(
      new MemoryInputStream (memStream.getData(), memStream.getLength()));
    std::unique_ptr<Reader> reader = createReader(pool, std::move(inStream));
    std::unique_ptr<RowReader> rowReader = createRowReader(reader.get());
    EXPECT_EQ(rowCount, reader->getNumberOfRows());

    batch = rowReader->createRowBatch(rowCount);
    EXPECT_EQ(true, rowReader->next(*batch));
    EXPECT_EQ(rowCount, batch->numElements);

    for (uint64_t i = 0; i < rowCount; ++i) {
      structBatch = dynamic_cast<StructVectorBatch *>(batch.get());
      charBatch = dynamic_cast<StringVectorBatch *>(structBatch->fields[0]);
      varcharBatch = dynamic_cast<StringVectorBatch *>(structBatch->fields[1]);

      EXPECT_EQ(3, charBatch->length[i]);
      EXPECT_FALSE(varcharBatch->length[i] > 4);

      // test char data
      std::string charsRead(
        charBatch->data[i],
        static_cast<size_t>(charBatch->length[i]));

      std::ostringstream os;
      os << i;
      std::string charsExpected = os.str().substr(0, 3);
      while (charsExpected.length() < 3) {
        charsExpected += ' ';
      }
      EXPECT_EQ(charsExpected, charsRead);

      // test varchar data
      std::string varcharRead(
        varcharBatch->data[i],
        static_cast<size_t>(varcharBatch->length[i]));
      std::string varcharExpected = os.str().substr(0, 4);
      EXPECT_EQ(varcharRead, varcharExpected);
    }

    EXPECT_FALSE(rowReader->next(*batch));
  }

  TEST_P(WriterTest, writeDecimal64Column) {
    const uint64_t maxPrecision = 18;
    MemoryOutputStream memStream(DEFAULT_MEM_STREAM_SIZE);
    MemoryPool* pool = getDefaultPool();
    std::unique_ptr<Type> type(Type::buildTypeFromString(
      "struct<col1:decimal(18,5)>"));

    uint64_t stripeSize = 16 * 1024; // 16K
    uint64_t compressionBlockSize = 1024; // 1k
    uint64_t rowCount = 1024;

    std::unique_ptr<Writer> writer = createWriter(stripeSize,
                                                  compressionBlockSize,
                                                  CompressionKind_ZLIB,
                                                  *type,
                                                  pool,
                                                  &memStream,
                                                  fileVersion);
    std::unique_ptr<ColumnVectorBatch> batch = writer->createRowBatch(rowCount);
    StructVectorBatch * structBatch =
      dynamic_cast<StructVectorBatch *>(batch.get());
    Decimal64VectorBatch * decBatch =
      dynamic_cast<Decimal64VectorBatch *>(structBatch->fields[0]);

    // write positive decimals
    for (uint64_t i = 0; i < rowCount; ++i) {
      decBatch->values[i] = static_cast<int64_t>(i + 10000);
    }
    structBatch->numElements = decBatch->numElements = rowCount;
    writer->add(*batch);

    // write negative decimals
    for (uint64_t i = 0; i < rowCount; ++i) {
      decBatch->values[i] = static_cast<int64_t>(i - 10000);
    }
    structBatch->numElements = decBatch->numElements = rowCount;
    writer->add(*batch);

    // write all precision decimals
    int64_t dec;
    for (uint64_t i = dec = 0; i < maxPrecision; ++i) {
      dec = dec * 10 + 9;
      decBatch->values[i] = dec;
      decBatch->values[i + maxPrecision] = -dec;
    }
    structBatch->numElements = decBatch->numElements = 2 * maxPrecision;
    writer->add(*batch);

    writer->close();

    std::unique_ptr<InputStream> inStream(
      new MemoryInputStream (memStream.getData(), memStream.getLength()));
    std::unique_ptr<Reader> reader = createReader(pool, std::move(inStream));
    std::unique_ptr<RowReader> rowReader = createRowReader(reader.get());
    EXPECT_EQ((rowCount + maxPrecision) * 2, reader->getNumberOfRows());

    // test reading positive decimals
    batch = rowReader->createRowBatch(rowCount);
    EXPECT_EQ(true, rowReader->next(*batch));
    structBatch = dynamic_cast<StructVectorBatch *>(batch.get());
    decBatch = dynamic_cast<Decimal64VectorBatch *>(structBatch->fields[0]);
    for (uint64_t i = 0; i < rowCount; ++i) {
      EXPECT_EQ(static_cast<int64_t>(i + 10000), decBatch->values[i]);
    }

    // test reading negative decimals
    EXPECT_EQ(true, rowReader->next(*batch));
    structBatch = dynamic_cast<StructVectorBatch *>(batch.get());
    decBatch = dynamic_cast<Decimal64VectorBatch *>(structBatch->fields[0]);
    for (uint64_t i = 0; i < rowCount; ++i) {
      EXPECT_EQ(static_cast<int64_t>(i - 10000), decBatch->values[i]);
    }

    // test reading all precision decimals
    EXPECT_EQ(true, rowReader->next(*batch));
    structBatch = dynamic_cast<StructVectorBatch *>(batch.get());
    decBatch = dynamic_cast<Decimal64VectorBatch *>(structBatch->fields[0]);
    for (uint64_t i = dec = 0; i < maxPrecision; ++i) {
      dec = dec * 10 + 9;
      EXPECT_EQ(dec, decBatch->values[i]);
      EXPECT_EQ(-dec, decBatch->values[i + maxPrecision]);
    }
  }

  TEST_P(WriterTest, writeDecimal128Column) {
    const uint64_t maxPrecision = 38;
    MemoryOutputStream memStream(DEFAULT_MEM_STREAM_SIZE);
    MemoryPool* pool = getDefaultPool();
    std::unique_ptr<Type> type(Type::buildTypeFromString(
      "struct<col1:decimal(38,10)>"));

    uint64_t stripeSize = 16 * 1024;
    uint64_t compressionBlockSize = 1024;
    uint64_t rowCount = 1024;

    std::unique_ptr<Writer> writer = createWriter(stripeSize,
                                                  compressionBlockSize,
                                                  CompressionKind_ZLIB,
                                                  *type,
                                                  pool,
                                                  &memStream,
                                                  fileVersion);
    std::unique_ptr<ColumnVectorBatch> batch = writer->createRowBatch(rowCount);
    StructVectorBatch * structBatch =
      dynamic_cast<StructVectorBatch *>(batch.get());
    Decimal128VectorBatch * decBatch =
      dynamic_cast<Decimal128VectorBatch *>(structBatch->fields[0]);

    // write positive decimals
    std::string base = "1" + std::string(1, '0');
    for (uint64_t i = 0; i < rowCount; ++i) {
      std::ostringstream os;
      os << i;
      decBatch->values[i] = Int128(base + os.str());
    }
    structBatch->numElements = decBatch->numElements = rowCount;
    writer->add(*batch);

    // write negative decimals
    std::string nbase = "-" + base;
    for (uint64_t i = 0; i < rowCount; ++i) {
      std::ostringstream os;
      os << i;
      decBatch->values[i] = Int128(nbase + os.str());
    }
    structBatch->numElements = rowCount;
    decBatch->numElements = rowCount;
    writer->add(*batch);

    // write all precision decimals
    for (uint64_t i = 0; i < maxPrecision; ++i) {
      std::string expected = std::string(i + 1, '9');
      decBatch->values[i] = Int128(expected);
      decBatch->values[i + maxPrecision] = Int128("-" + expected);
    }
    structBatch->numElements = decBatch->numElements = 2 * maxPrecision;
    writer->add(*batch);

    writer->close();

    std::unique_ptr<InputStream> inStream(
      new MemoryInputStream (memStream.getData(), memStream.getLength()));
    std::unique_ptr<Reader> reader = createReader(pool, std::move(inStream));
    std::unique_ptr<RowReader> rowReader = createRowReader(reader.get());
    EXPECT_EQ((rowCount + maxPrecision) * 2, reader->getNumberOfRows());

    // test reading positive decimals
    batch = rowReader->createRowBatch(rowCount);
    EXPECT_EQ(true, rowReader->next(*batch));
    structBatch = dynamic_cast<StructVectorBatch *>(batch.get());
    decBatch = dynamic_cast<Decimal128VectorBatch *>(structBatch->fields[0]);
    for (uint64_t i = 0; i < rowCount; ++i) {
      std::ostringstream os;
      os << i;
      EXPECT_EQ(base + os.str(), decBatch->values[i].toString());
    }

    // test reading negative decimals
    EXPECT_EQ(true, rowReader->next(*batch));
    structBatch = dynamic_cast<StructVectorBatch *>(batch.get());
    decBatch = dynamic_cast<Decimal128VectorBatch *>(structBatch->fields[0]);
    for (uint64_t i = 0; i < rowCount; ++i) {
      std::ostringstream os;
      os << i;
      EXPECT_EQ(nbase + os.str(), decBatch->values[i].toString());
    }

    // test reading all precision decimals
    EXPECT_EQ(true, rowReader->next(*batch));
    structBatch = dynamic_cast<StructVectorBatch *>(batch.get());
    decBatch = dynamic_cast<Decimal128VectorBatch *>(structBatch->fields[0]);
    for (uint64_t i = 0; i < maxPrecision; ++i) {
      std::string expected = std::string(i + 1, '9');
      EXPECT_EQ(expected, decBatch->values[i].toString());
      EXPECT_EQ("-" + expected, decBatch->values[i + maxPrecision].toString());
    }
  }

  TEST_P(WriterTest, writeListColumn) {
    MemoryOutputStream memStream(DEFAULT_MEM_STREAM_SIZE);
    MemoryPool * pool = getDefaultPool();

    std::unique_ptr<Type> type(Type::buildTypeFromString(
      "struct<col1:array<int>>"));

    uint64_t stripeSize = 1024 * 1024;
    uint64_t compressionBlockSize = 64 * 1024;
    uint64_t rowCount = 1024;
    uint64_t maxListLength = 10;
    uint64_t offset = 0;

    std::unique_ptr<Writer> writer = createWriter(stripeSize,
                                                  compressionBlockSize,
                                                  CompressionKind_ZLIB,
                                                  *type,
                                                  pool,
                                                  &memStream,
                                                  fileVersion);
    std::unique_ptr<ColumnVectorBatch> batch =
      writer->createRowBatch(rowCount * maxListLength);

    StructVectorBatch * structBatch =
      dynamic_cast<StructVectorBatch *>(batch.get());
    ListVectorBatch * listBatch =
      dynamic_cast<ListVectorBatch *>(structBatch->fields[0]);
    LongVectorBatch * intBatch =
      dynamic_cast<LongVectorBatch *>(listBatch->elements.get());
    int64_t * data = intBatch->data.data();
    int64_t * offsets = listBatch->offsets.data();

    for (uint64_t i = 0; i < rowCount; ++i) {
      offsets[i] = static_cast<int64_t>(offset);
      for (uint64_t length = i % maxListLength + 1; length != 0; --length) {
        data[offset++] = static_cast<int64_t >(i);
      }
    }
    offsets[rowCount] = static_cast<int64_t>(offset);

    structBatch->numElements = rowCount;
    listBatch->numElements = rowCount;

    writer->add(*batch);
    writer->close();

    std::unique_ptr<InputStream> inStream(
      new MemoryInputStream (memStream.getData(), memStream.getLength()));
    std::unique_ptr<Reader> reader = createReader(pool, std::move(inStream));
    std::unique_ptr<RowReader> rowReader = createRowReader(reader.get());
    EXPECT_EQ(rowCount, reader->getNumberOfRows());

    batch = rowReader->createRowBatch(rowCount * maxListLength);
    EXPECT_EQ(true, rowReader->next(*batch));

    structBatch = dynamic_cast<StructVectorBatch *>(batch.get());
    listBatch = dynamic_cast<ListVectorBatch *>(structBatch->fields[0]);
    intBatch = dynamic_cast<LongVectorBatch *>(listBatch->elements.get());
    data = intBatch->data.data();
    offsets = listBatch->offsets.data();

    EXPECT_EQ(rowCount, listBatch->numElements);
    EXPECT_EQ(offset, intBatch->numElements);

    for (uint64_t i = 0; i < rowCount; ++i) {
      uint64_t length = i % maxListLength + 1;
      for (int64_t j = 0; j != length; ++j) {
        EXPECT_EQ(static_cast<int64_t>(i), data[offsets[i] + j]);
      }
    }
  }

  TEST_P(WriterTest, writeMapColumn) {
    MemoryOutputStream memStream(DEFAULT_MEM_STREAM_SIZE);
    MemoryPool * pool = getDefaultPool();
    std::unique_ptr<Type> type(
      Type::buildTypeFromString("struct<col1:map<string,int>>"));

    uint64_t stripeSize = 16 * 1024;
    uint64_t compressionBlockSize = 1024;
    uint64_t rowCount = 1024, maxListLength = 10, offset = 0;

    std::unique_ptr<Writer> writer = createWriter(stripeSize,
                                                  compressionBlockSize,
                                                  CompressionKind_ZLIB,
                                                  *type,
                                                  pool,
                                                  &memStream,
                                                  fileVersion);
    std::unique_ptr<ColumnVectorBatch> batch =
      writer->createRowBatch(rowCount * maxListLength);
    StructVectorBatch * structBatch =
      dynamic_cast<StructVectorBatch *>(batch.get());
    MapVectorBatch * mapBatch =
      dynamic_cast<MapVectorBatch *>(structBatch->fields[0]);
    StringVectorBatch * keyBatch =
      dynamic_cast<StringVectorBatch *>(mapBatch->keys.get());
    LongVectorBatch * elemBatch =
      dynamic_cast<LongVectorBatch *>(mapBatch->elements.get());

    char dataBuffer[327675]; // 300k
    uint64_t strOffset = 0;

    int64_t * offsets = mapBatch->offsets.data();
    char ** keyData = keyBatch->data.data();
    int64_t * keyLength = keyBatch->length.data();
    int64_t * elemData = elemBatch->data.data();

    for (uint64_t i = 0; i < rowCount; ++i) {
      offsets[i] = static_cast<int64_t>(offset);
      for (uint64_t j = 0; j != i % maxListLength + 1; ++j) {
        std::ostringstream os;
        os << (i + j);
        memcpy(dataBuffer + strOffset, os.str().c_str(), os.str().size());
        keyData[offset] = dataBuffer + strOffset;

        keyLength[offset] = static_cast<int64_t>(os.str().size());
        elemData[offset++] = static_cast<int64_t>(i);

        strOffset += os.str().size();
      }
    }
    offsets[rowCount] = static_cast<int64_t>(offset);

    structBatch->numElements = rowCount;
    mapBatch->numElements = rowCount;

    writer->add(*batch);
    writer->close();

    std::unique_ptr<InputStream> inStream(
      new MemoryInputStream (memStream.getData(), memStream.getLength()));
    std::unique_ptr<Reader> reader = createReader(
      pool,
      std::move(inStream));
    std::unique_ptr<RowReader> rowReader = createRowReader(reader.get());
    EXPECT_EQ(rowCount, reader->getNumberOfRows());

    batch = rowReader->createRowBatch(rowCount * maxListLength);
    EXPECT_EQ(true, rowReader->next(*batch));

    structBatch = dynamic_cast<StructVectorBatch *>(batch.get());
    mapBatch = dynamic_cast<MapVectorBatch *>(structBatch->fields[0]);
    keyBatch = dynamic_cast<StringVectorBatch *>(mapBatch->keys.get());
    elemBatch = dynamic_cast<LongVectorBatch *>(mapBatch->elements.get());
    offsets = mapBatch->offsets.data();
    keyData = keyBatch->data.data();
    keyLength = keyBatch->length.data();
    elemData = elemBatch->data.data();

    EXPECT_EQ(rowCount, mapBatch->numElements);
    EXPECT_EQ(offset, keyBatch->numElements);
    EXPECT_EQ(offset, elemBatch->numElements);

    for (uint64_t i = 0; i != rowCount; ++i) {
      for (int64_t j = 0; j != i % maxListLength + 1; ++j) {
        std::ostringstream os;
        os << i + static_cast<uint64_t>(j);
        uint64_t lenRead = static_cast<uint64_t>(keyLength[offsets[i] + j]);
        EXPECT_EQ(os.str(), std::string(keyData[offsets[i] + j], lenRead));
        EXPECT_EQ(static_cast<int64_t>(i), elemData[offsets[i] + j]);
      }
    }
  }

  TEST_P(WriterTest, writeUnionColumn) {
    MemoryOutputStream memStream(DEFAULT_MEM_STREAM_SIZE);
    MemoryPool * pool = getDefaultPool();
    std::unique_ptr<Type> type(Type::buildTypeFromString(
      "struct<col1:uniontype<int,double,boolean>>"));

    uint64_t stripeSize = 16 * 1024;
    uint64_t compressionBlockSize = 1024;
    uint64_t rowCount = 3333;

    std::unique_ptr<Writer> writer = createWriter(stripeSize,
                                                  compressionBlockSize,
                                                  CompressionKind_ZLIB,
                                                  *type,
                                                  pool,
                                                  &memStream,
                                                  fileVersion);
    std::unique_ptr<ColumnVectorBatch> batch = writer->createRowBatch(rowCount);
    StructVectorBatch * structBatch =
      dynamic_cast<StructVectorBatch *>(batch.get());
    UnionVectorBatch * unionBatch =
      dynamic_cast<UnionVectorBatch *>(structBatch->fields[0]);
    unsigned char * tags = unionBatch->tags.data();
    uint64_t * offsets = unionBatch->offsets.data();

    LongVectorBatch * intBatch =
      dynamic_cast<LongVectorBatch *>(unionBatch->children[0]);
    DoubleVectorBatch * doubleBatch =
      dynamic_cast<DoubleVectorBatch *>(unionBatch->children[1]);
    LongVectorBatch * boolBatch =
      dynamic_cast<LongVectorBatch *>(unionBatch->children[2]);
    int64_t *intData = intBatch->data.data();
    double *doubleData = doubleBatch->data.data();
    int64_t *boolData = boolBatch->data.data();

    uint64_t intOffset = 0, doubleOffset = 0, boolOffset = 0, tag = 0;
    for (uint64_t i = 0; i < rowCount; ++i) {
      tags[i] = static_cast<unsigned char>(tag);
      switch(tag) {
        case 0:
          offsets[i] = intOffset;
          intData[intOffset++] = static_cast<int64_t>(i);
          break;
        case 1:
          offsets[i] = doubleOffset;
          doubleData[doubleOffset++] = static_cast<double>(i) + 0.5;
          break;
        case 2:
          offsets[i] = boolOffset;
          boolData[boolOffset++] = (i % 2 == 0) ? 1 : 0;
          break;
      }
      tag = (tag + 1) % 3;
    }

    structBatch->numElements = rowCount;
    unionBatch->numElements = rowCount;

    writer->add(*batch);
    writer->close();

    std::unique_ptr<InputStream> inStream(
      new MemoryInputStream (memStream.getData(), memStream.getLength()));
    std::unique_ptr<Reader> reader = createReader(
      pool,
      std::move(inStream));
    std::unique_ptr<RowReader> rowReader = createRowReader(reader.get());
    EXPECT_EQ(rowCount, reader->getNumberOfRows());

    batch = rowReader->createRowBatch(rowCount);
    EXPECT_EQ(true, rowReader->next(*batch));

    structBatch =dynamic_cast<StructVectorBatch *>(batch.get());
    unionBatch =dynamic_cast<UnionVectorBatch *>(structBatch->fields[0]);
    tags = unionBatch->tags.data();
    offsets = unionBatch->offsets.data();

    intBatch =  dynamic_cast<LongVectorBatch *>(unionBatch->children[0]);
    doubleBatch = dynamic_cast<DoubleVectorBatch *>(unionBatch->children[1]);
    boolBatch = dynamic_cast<LongVectorBatch *>(unionBatch->children[2]);
    intData = intBatch->data.data();
    doubleData = doubleBatch->data.data();
    boolData = boolBatch->data.data();

    EXPECT_EQ(rowCount, unionBatch->numElements);
    EXPECT_EQ(rowCount / 3, intBatch->numElements);
    EXPECT_EQ(rowCount / 3, doubleBatch->numElements);
    EXPECT_EQ(rowCount / 3, boolBatch->numElements);

    uint64_t offset;
    for (uint64_t i = 0; i < rowCount; ++i) {
      tag = tags[i];
      offset = offsets[i];

      switch(tag) {
        case 0:
          EXPECT_EQ(i, intData[offset]);
          break;
        case 1:
          EXPECT_TRUE(
            std::abs(static_cast<double>(i) + 0.5 - doubleData[offset]) < 0.000001);
          break;
        case 2:
          EXPECT_EQ(i % 2 == 0 ? 1 : 0, boolData[offset]);
          break;
      }
    }
  }

  TEST_P(WriterTest, writeUTF8CharAndVarcharColumn) {
    MemoryOutputStream memStream(DEFAULT_MEM_STREAM_SIZE);
    MemoryPool * pool = getDefaultPool();
    std::unique_ptr<Type> type(Type::buildTypeFromString(
      "struct<col1:char(2),col2:varchar(2)>"));

    uint64_t stripeSize = 1024;
    uint64_t compressionBlockSize = 1024;
    uint64_t rowCount = 3;
    std::unique_ptr<Writer> writer = createWriter(stripeSize,
                                                  compressionBlockSize,
                                                  CompressionKind_ZLIB,
                                                  *type,
                                                  pool,
                                                  &memStream,
                                                  fileVersion);
    std::unique_ptr<ColumnVectorBatch> batch = writer->createRowBatch(rowCount);
    StructVectorBatch * structBatch =
      dynamic_cast<StructVectorBatch *>(batch.get());
    StringVectorBatch * charBatch =
      dynamic_cast<StringVectorBatch *>(structBatch->fields[0]);
    StringVectorBatch * varcharBatch =
      dynamic_cast<StringVectorBatch *>(structBatch->fields[1]);
    std::vector<std::vector<char>> strs;

    // input character is 'à' (0xC3, 0xA0)
    // in total 3 rows, each has 1, 2, and 3 'à' respectively
    std::vector<char> vec;
    for (uint64_t i = 0; i != rowCount; ++i) {
      vec.push_back('\xC3');
      vec.push_back('\xA0');
      strs.push_back(vec);
      charBatch->data[i] = varcharBatch->data[i] = strs.back().data();
      charBatch->length[i] = varcharBatch->length[i] = static_cast<int64_t>(strs.back().size());
    }

    structBatch->numElements = rowCount;
    charBatch->numElements = rowCount;
    varcharBatch->numElements = rowCount;

    writer->add(*batch);
    writer->close();

    // read and verify data
    std::unique_ptr<InputStream> inStream(
      new MemoryInputStream (memStream.getData(), memStream.getLength()));
    std::unique_ptr<Reader> reader = createReader(pool, std::move(inStream));
    std::unique_ptr<RowReader> rowReader = createRowReader(reader.get());
    EXPECT_EQ(rowCount, reader->getNumberOfRows());

    batch = rowReader->createRowBatch(rowCount);
    structBatch = dynamic_cast<StructVectorBatch *>(batch.get());
    charBatch = dynamic_cast<StringVectorBatch *>(structBatch->fields[0]);
    varcharBatch = dynamic_cast<StringVectorBatch *>(structBatch->fields[1]);

    EXPECT_EQ(true, rowReader->next(*batch));
    EXPECT_EQ(rowCount, batch->numElements);

    char expectedPadded[3] = {'\xC3', '\xA0', ' '};
    char expectedOneChar[2] = {'\xC3', '\xA0'};
    char expectedTwoChars[4] = {'\xC3', '\xA0', '\xC3', '\xA0'};

    EXPECT_EQ(3, charBatch->length[0]);
    EXPECT_EQ(4, charBatch->length[1]);
    EXPECT_EQ(4, charBatch->length[2]);
    EXPECT_TRUE(memcmp(charBatch->data[0], expectedPadded, 3) == 0);
    EXPECT_TRUE(memcmp(charBatch->data[1], expectedTwoChars, 4) == 0);
    EXPECT_TRUE(memcmp(charBatch->data[2], expectedTwoChars, 4) == 0);

    EXPECT_EQ(2, varcharBatch->length[0]);
    EXPECT_EQ(4, varcharBatch->length[1]);
    EXPECT_EQ(4, varcharBatch->length[2]);
    EXPECT_TRUE(memcmp(varcharBatch->data[0], expectedOneChar, 2) == 0);
    EXPECT_TRUE(memcmp(varcharBatch->data[1], expectedTwoChars, 4) == 0);
    EXPECT_TRUE(memcmp(varcharBatch->data[2], expectedTwoChars, 4) == 0);

    EXPECT_FALSE(rowReader->next(*batch));
  }

  TEST_P(WriterTest, testWriteListColumnWithNull) {
    MemoryOutputStream memStream(DEFAULT_MEM_STREAM_SIZE);
    MemoryPool * pool = getDefaultPool();
    std::unique_ptr<Type> type(Type::buildTypeFromString(
      "struct<col1:array<tinyint>>"));

    uint64_t stripeSize = 1024;
    uint64_t compressionBlockSize = 1024;

    std::unique_ptr<Writer> writer = createWriter(stripeSize,
                                      compressionBlockSize,
                                      CompressionKind_ZLIB,
                                      *type,
                                      pool,
                                      &memStream,
                                      fileVersion);

    std::unique_ptr<ColumnVectorBatch> batch =
      writer->createRowBatch(4);
    StructVectorBatch * structBatch =
      dynamic_cast<StructVectorBatch *>(batch.get());
    ListVectorBatch * listBatch =
      dynamic_cast<ListVectorBatch *>(structBatch->fields[0]);
    LongVectorBatch * intBatch =
      dynamic_cast<LongVectorBatch *>(listBatch->elements.get());

    // test data looks like below -
    // {[1, 2]}
    // null
    // {[3, 4]}
    // {[5, 6]}
    int64_t * offsets = listBatch->offsets.data();
    offsets[0] = 0;
    offsets[1] = 2;
    offsets[2] = 2;
    offsets[3] = 4;
    offsets[4] = 6;

    intBatch->resize(6);
    for (uint64_t i = 0; i < 6; ++i) {
      intBatch->notNull[i] = 1;
    }

    int64_t * data = intBatch->data.data();
    for (int8_t i = 1; i < 7; ++i) {
      data[i - 1] = i;
    }

    structBatch->numElements = 4;

    listBatch->notNull[1] = 0;
    listBatch->hasNulls = true;
    listBatch->numElements = 4;

    writer->add(*batch);
    writer->close();

    std::unique_ptr<InputStream> inStream(
      new MemoryInputStream (memStream.getData(), memStream.getLength()));
    std::unique_ptr<Reader> reader = createReader(
      pool,
      std::move(inStream));
    std::unique_ptr<RowReader> rowReader = createRowReader(reader.get());
    EXPECT_EQ(4, reader->getNumberOfRows());

    batch = rowReader->createRowBatch(4 * 2);
    EXPECT_EQ(true, rowReader->next(*batch));

    structBatch = dynamic_cast<StructVectorBatch *>(batch.get());
    listBatch = dynamic_cast<ListVectorBatch *>(structBatch->fields[0]);
    intBatch = dynamic_cast<LongVectorBatch *>(listBatch->elements.get());

    data = intBatch->data.data();
    offsets = listBatch->offsets.data();

    EXPECT_EQ(4, structBatch->numElements);
    EXPECT_EQ(4, listBatch->numElements);
    EXPECT_EQ(1, listBatch->notNull[0]);
    EXPECT_EQ(0, listBatch->notNull[1]);
    EXPECT_EQ(1, listBatch->notNull[2]);
    EXPECT_EQ(1, listBatch->notNull[3]);

    EXPECT_EQ(0, offsets[0]);
    EXPECT_EQ(2, offsets[1]);
    EXPECT_EQ(2, offsets[2]);
    EXPECT_EQ(4, offsets[3]);
    EXPECT_EQ(6, offsets[4]);

    for (int8_t i = 1; i < 7; ++i) {
      EXPECT_EQ(i, data[i - 1]);
    }
  }

  TEST_P(WriterTest, testWriteNestedStructWithNull) {
    MemoryOutputStream memStream(DEFAULT_MEM_STREAM_SIZE);
    MemoryPool * pool = getDefaultPool();
    std::unique_ptr<Type> type(Type::buildTypeFromString(
      "struct<struct<bigint>>"));

    uint64_t stripeSize = 1024;
    uint64_t compressionBlockSize = 1024;

    std::unique_ptr<Writer> writer = createWriter(stripeSize,
                                      compressionBlockSize,
                                      CompressionKind_ZLIB,
                                      *type,
                                      pool,
                                      &memStream,
                                      fileVersion);

    // test data looks like below -
    // {0}
    // null
    // {1}
    // {2}
    // {3}
    std::unique_ptr<ColumnVectorBatch> batch =
      writer->createRowBatch(5);
    StructVectorBatch * structBatch =
      dynamic_cast<StructVectorBatch *>(batch.get());
    StructVectorBatch * structBatch2 =
      dynamic_cast<StructVectorBatch *>(structBatch->fields[0]);
    LongVectorBatch * intBatch =
      dynamic_cast<LongVectorBatch *>(structBatch2->fields[0]);

    structBatch->numElements = 5;
    structBatch2->numElements = 5;
    structBatch2->hasNulls = true;
    structBatch2->notNull[1] = 0;

    intBatch->resize(5);
    for (int64_t i = 0; i < 5; ++i) {
      intBatch->data.data()[i] = i;
    }
    intBatch->notNull[1] = 0;
    intBatch->hasNulls = true;

    writer->add(*batch);
    writer->close();

    std::unique_ptr<InputStream> inStream(
      new MemoryInputStream (memStream.getData(), memStream.getLength()));
    std::unique_ptr<Reader> reader = createReader(
      pool,
      std::move(inStream));
    std::unique_ptr<RowReader> rowReader = createRowReader(reader.get());
    EXPECT_EQ(5, reader->getNumberOfRows());

    batch = rowReader->createRowBatch(5);
    EXPECT_EQ(true, rowReader->next(*batch));

    structBatch = dynamic_cast<StructVectorBatch *>(batch.get());
    structBatch2 = dynamic_cast<StructVectorBatch *>(structBatch->fields[0]);
    intBatch = dynamic_cast<LongVectorBatch *>(structBatch2->fields[0]);

    for (uint64_t i = 0; i < 5; ++i) {
      EXPECT_EQ(1, structBatch->notNull[i]);
    }

    for (uint64_t i = 0; i < 5; ++i) {
      if (i == 1) {
        EXPECT_EQ(0, structBatch2->notNull[i]);
      } else {
        EXPECT_EQ(1, structBatch2->notNull[i]);
      }
    }

    for (uint64_t i = 0; i < 5; ++i) {
      if (i == 1) {
       EXPECT_EQ(0, intBatch->notNull[i]);
      } else {
       EXPECT_EQ(1, intBatch->notNull[i]);
      }
    }

    int64_t * data = intBatch->data.data();
    for (int8_t i = 0; i < 5; ++i) {
      if (i != 1) {
        EXPECT_EQ(i, data[i]);
      }
    }
  }

  TEST_P(WriterTest, testWriteNestedStructWithNullIndex) {
    MemoryOutputStream memStream(DEFAULT_MEM_STREAM_SIZE);
    MemoryPool * pool = getDefaultPool();
    std::unique_ptr<Type> type(Type::buildTypeFromString(
      "struct<struct<bigint>>"));

    uint64_t stripeSize = 1024;
    uint64_t compressionBlockSize = 1024;

    // 10000 rows with every 1000 row as an RG
    // Each RG has 100 null rows except that the 5th RG is all null
    std::unique_ptr<Writer> writer = createWriter(stripeSize,
                                      compressionBlockSize,
                                      CompressionKind_ZLIB,
                                      *type,
                                      pool,
                                      &memStream,
                                      fileVersion,
                                      1000);

    std::unique_ptr<ColumnVectorBatch> batch =
      writer->createRowBatch(10000);
    StructVectorBatch * structBatch =
      dynamic_cast<StructVectorBatch *>(batch.get());
    StructVectorBatch * structBatch2 =
      dynamic_cast<StructVectorBatch *>(structBatch->fields[0]);
    LongVectorBatch * intBatch =
      dynamic_cast<LongVectorBatch *>(structBatch2->fields[0]);

    structBatch->numElements = 10000;
    structBatch2->numElements = 10000;
    structBatch2->hasNulls = true;

    intBatch->resize(10000);
    intBatch->hasNulls = true;
    for (uint64_t i = 0; i < 10; ++i) {
      for (uint64_t j = i * 1000 + 100 * i; j < i * 1000 + 100 * i + 100; ++j) {
        structBatch2->notNull[j] = 0;
        intBatch->notNull[j] = 0;
      }
    }

    for (uint64_t i = 5000; i < 6000; ++i) {
      structBatch2->notNull[i] = 0;
      intBatch->notNull[i] = 0;
    }

    for (int64_t i = 0; i < 10000; ++i) {
      intBatch->data.data()[i] = i;
    }

    writer->add(*batch);
    writer->close();

    std::unique_ptr<InputStream> inStream(
      new MemoryInputStream (memStream.getData(), memStream.getLength()));
    std::unique_ptr<Reader> reader = createReader(
      pool,
      std::move(inStream));
    std::unique_ptr<RowReader> rowReader = createRowReader(reader.get());
    EXPECT_EQ(10000, reader->getNumberOfRows());

    batch = rowReader->createRowBatch(1000);
    structBatch = dynamic_cast<StructVectorBatch *>(batch.get());
    structBatch2 = dynamic_cast<StructVectorBatch *>(structBatch->fields[0]);
    intBatch = dynamic_cast<LongVectorBatch *>(structBatch2->fields[0]);

    // Read rows 0 - 1000
    EXPECT_EQ(true, rowReader->next(*batch));
    for (uint64_t i = 0; i < 1000; ++i) {
      EXPECT_EQ(1, structBatch->notNull[i]);
    }

    for (uint64_t i = 0; i < 1000; ++i) {
      if (i < 100) {
        EXPECT_EQ(0, structBatch2->notNull[i]);
      } else {
        EXPECT_EQ(1, structBatch2->notNull[i]);
      }
    }

    for (uint64_t i = 0; i < 1000; ++i) {
      if (i < 100) {
       EXPECT_EQ(0, intBatch->notNull[i]);
      } else {
       EXPECT_EQ(1, intBatch->notNull[i]);
       EXPECT_EQ(i, intBatch->data.data()[i]);
      }
    }

    // Read rows 1800 - 2800, in which 2200 - 2300 are nulls
    rowReader->seekToRow(1800);
    EXPECT_EQ(true, rowReader->next(*batch));
    for (uint64_t i = 0; i < 1000; ++i) {
      EXPECT_EQ(1, structBatch->notNull[i]);
    }

    for (uint64_t i = 0; i < 1000; ++i) {
      if (i >= 400 && i < 500) {
        EXPECT_EQ(0, structBatch2->notNull[i]);
      } else {
        EXPECT_EQ(1, structBatch2->notNull[i]);
      }
    }

    for (uint64_t i = 0; i < 1000; ++i) {
      if (i >= 400 && i < 500) {
       EXPECT_EQ(0, intBatch->notNull[i]);
      } else {
       EXPECT_EQ(1, intBatch->notNull[i]);
       EXPECT_EQ(i + 1800, intBatch->data.data()[i]);
      }
    }

    // Read rows 5000 - 6000, all nulls
    rowReader->seekToRow(5000);
    EXPECT_EQ(true, rowReader->next(*batch));
    for (uint64_t i = 0; i < 1000; ++i) {
      EXPECT_EQ(1, structBatch->notNull[i]);
    }

    for (uint64_t i = 0; i < 1000; ++i) {
      EXPECT_EQ(0, structBatch2->notNull[i]);
    }

    for (uint64_t i = 0; i < 1000; ++i) {
     EXPECT_EQ(0, intBatch->notNull[i]);
    }

    // Read rows 7200 - 8200, in which 7700 - 7800 are null
    rowReader->seekToRow(7200);
    EXPECT_EQ(true, rowReader->next(*batch));
    for (uint64_t i = 0; i < 1000; ++i) {
      EXPECT_EQ(1, structBatch->notNull[i]);
    }

    for (uint64_t i = 0; i < 1000; ++i) {
      if (i >= 500 && i < 600) {
        EXPECT_EQ(0, structBatch2->notNull[i]);
      } else {
        EXPECT_EQ(1, structBatch2->notNull[i]);
      }
    }

    for (uint64_t i = 0; i < 1000; ++i) {
      if (i >= 500 && i < 600) {
       EXPECT_EQ(0, intBatch->notNull[i]);
      } else {
       EXPECT_EQ(1, intBatch->notNull[i]);
       EXPECT_EQ(i + 7200, intBatch->data.data()[i]);
      }
    }
  }

  TEST_P(WriterTest, testBloomFilter) {
    WriterOptions options;
    options.setStripeSize(1024)
      .setCompressionBlockSize(64)
      .setCompression(CompressionKind_ZSTD)
      .setMemoryPool(getDefaultPool())
      .setRowIndexStride(10000)
      .setFileVersion(fileVersion)
      .setColumnsUseBloomFilter({1, 2});

    // write 65535 rows of data
    MemoryOutputStream memStream(DEFAULT_MEM_STREAM_SIZE);
    MemoryPool * pool = getDefaultPool();
    std::unique_ptr<Type> type(Type::buildTypeFromString(
      "struct<c1:bigint,c2:string>"));

    char dataBuffer[327675]; // 300k
    uint64_t offset = 0;
    uint64_t rowCount = 65535;

    std::unique_ptr<Writer> writer = createWriter(*type, &memStream, options);
    std::unique_ptr<ColumnVectorBatch> batch = writer->createRowBatch(rowCount);
    StructVectorBatch& structBatch = dynamic_cast<StructVectorBatch&>(*batch);
    LongVectorBatch& longBatch = dynamic_cast<LongVectorBatch&>(*structBatch.fields[0]);
    StringVectorBatch& strBatch = dynamic_cast<StringVectorBatch&>(*structBatch.fields[1]);

    for (uint64_t i = 0; i < rowCount; ++i) {
      // each row group has a unique value
      uint64_t data = (i / options.getRowIndexStride());

      // c1
      longBatch.data[i] = static_cast<int64_t>(data);

      // c2
      std::ostringstream os;
      os << data;
      strBatch.data[i] = dataBuffer + offset;
      strBatch.length[i] = static_cast<int64_t>(os.str().size());
      memcpy(dataBuffer + offset, os.str().c_str(), os.str().size());
      offset += os.str().size();
    }

    structBatch.numElements = rowCount;
    longBatch.numElements = rowCount;
    strBatch.numElements = rowCount;
    writer->add(*batch);
    writer->close();

    // verify bloomfilters
    std::unique_ptr<InputStream> inStream(new MemoryInputStream(
      memStream.getData(), memStream.getLength()));
    std::unique_ptr<Reader> reader = createReader(pool, std::move(inStream));
    EXPECT_EQ(rowCount, reader->getNumberOfRows());

    EXPECT_EQ(2, reader->getBloomFilters(0, {}).size());
    EXPECT_EQ(1, reader->getBloomFilters(0, {1}).size());
    EXPECT_EQ(1, reader->getBloomFilters(0, {2}).size());

    std::map<uint32_t, BloomFilterIndex> bfs = reader->getBloomFilters(0, {1, 2});
    EXPECT_EQ(2, bfs.size());
    EXPECT_EQ(7, bfs[1].entries.size());
    EXPECT_EQ(7, bfs[2].entries.size());

    // test bloomfilters
    for (uint64_t rg = 0; rg <= rowCount / options.getRowIndexStride(); ++rg) {
      for (uint64_t value = 0; value <= 100; ++value) {
        std::string str = to_string(static_cast<int64_t>(value));
        if (value == rg) {
          EXPECT_TRUE(bfs[1].entries[rg]->testLong(static_cast<int64_t>(value)));
          EXPECT_TRUE(bfs[2].entries[rg]->testBytes(str.c_str(), static_cast<int64_t>(str.size())));
        } else {
          EXPECT_FALSE(bfs[1].entries[rg]->testLong(static_cast<int64_t>(value)));
          EXPECT_FALSE(bfs[2].entries[rg]->testBytes(str.c_str(), static_cast<int64_t>(str.size())));
        }
      }
    }
  }

  INSTANTIATE_TEST_CASE_P(OrcTest, WriterTest, Values(FileVersion::v_0_11(), FileVersion::v_0_12()));
}