blob: f76d0fe839de02299f7f6337a13a1d7cb669d451 [file] [log] [blame]
/**
* 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()));
}