You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@orc.apache.org by om...@apache.org on 2015/07/06 23:52:42 UTC
[13/23] orc git commit: ORC-23. Simplify directory structure.
http://git-wip-us.apache.org/repos/asf/orc/blob/7f55b453/c++/test/TestColumnReader.cc
----------------------------------------------------------------------
diff --git a/c++/test/TestColumnReader.cc b/c++/test/TestColumnReader.cc
new file mode 100644
index 0000000..4c669d0
--- /dev/null
+++ b/c++/test/TestColumnReader.cc
@@ -0,0 +1,4308 @@
+/**
+ * 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 "Adaptor.hh"
+#include "ColumnReader.hh"
+#include "Exceptions.hh"
+#include "OrcTest.hh"
+
+#include "wrap/orc-proto-wrapper.hh"
+#include "wrap/gtest-wrapper.h"
+#include "wrap/gmock.h"
+
+#include <iostream>
+#include <vector>
+
+#ifdef __clang__
+ DIAGNOSTIC_IGNORE("-Winconsistent-missing-override")
+#endif
+
+namespace orc {
+
+class MockStripeStreams: public StripeStreams {
+public:
+ ~MockStripeStreams();
+ std::unique_ptr<SeekableInputStream> getStream(int64_t columnId,
+ proto::Stream_Kind kind,
+ bool stream) const override;
+ MOCK_CONST_METHOD0(getReaderOptions, const ReaderOptions&());
+ MOCK_CONST_METHOD0(getSelectedColumns, const std::vector<bool>());
+ MOCK_CONST_METHOD1(getEncoding, proto::ColumnEncoding (int64_t));
+ MOCK_CONST_METHOD3(getStreamProxy, SeekableInputStream*
+ (int64_t, proto::Stream_Kind, bool));
+ MemoryPool& getMemoryPool() const {
+ return *getDefaultPool();
+ }
+};
+
+MockStripeStreams::~MockStripeStreams() {
+ // PASS
+}
+
+std::unique_ptr<SeekableInputStream>
+MockStripeStreams::getStream(int64_t columnId,
+ proto::Stream_Kind kind,
+ bool shouldStream) const {
+ return std::unique_ptr < SeekableInputStream >
+ (getStreamProxy(columnId, kind, shouldStream));
+}
+
+TEST(TestColumnReader, testBooleanWithNulls) {
+ MockStripeStreams streams;
+
+ // set getSelectedColumns()
+ std::vector<bool> selectedColumns(2, true);
+ EXPECT_CALL(streams, getSelectedColumns())
+ .WillRepeatedly(testing::Return(selectedColumns));
+
+ // set getEncoding
+ proto::ColumnEncoding directEncoding;
+ directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+ EXPECT_CALL(streams, getEncoding(testing::_))
+ .WillRepeatedly(testing::Return(directEncoding));
+
+ // set getStream
+ EXPECT_CALL(streams, getStreamProxy(0, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+ // alternate 4 non-null and 4 null via [0xf0 for x in range(512 / 8)]
+ const unsigned char buffer1[] = { 0x3d, 0xf0 };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer1, ARRAY_SIZE(buffer1))));
+
+ // [0x0f for x in range(256 / 8)]
+ const unsigned char buffer2[] = {0x1d, 0x0f};
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer2, ARRAY_SIZE(buffer2))));
+
+ // create the row type
+ std::unique_ptr<Type> rowType = createStructType();
+ rowType->addStructField(createPrimitiveType(BOOLEAN), "col0");
+ rowType->assignIds(0);
+
+ std::unique_ptr<ColumnReader> reader = buildReader(*rowType, streams);
+ LongVectorBatch *longBatch = new LongVectorBatch(1024, *getDefaultPool());
+ StructVectorBatch batch(1024, *getDefaultPool());
+ batch.fields.push_back(longBatch);
+ reader->next(batch, 512, 0);
+ ASSERT_EQ(512, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(512, longBatch->numElements);
+ ASSERT_EQ(true, longBatch->hasNulls);
+ unsigned int next = 0;
+ for (size_t i = 0; i < batch.numElements; ++i) {
+ if (i & 4) {
+ EXPECT_EQ(0, longBatch->notNull[i]) << "Wrong value at " << i;
+ } else {
+ EXPECT_EQ(1, longBatch->notNull[i]) << "Wrong value at " << i;
+ EXPECT_EQ((next++ & 4) != 0, longBatch->data[i])
+ << "Wrong value at " << i;
+ }
+ }
+}
+
+TEST(TestColumnReader, testBooleanSkipsWithNulls) {
+ MockStripeStreams streams;
+
+ // set getSelectedColumns()
+ std::vector<bool> selectedColumns(2, true);
+ EXPECT_CALL(streams, getSelectedColumns())
+ .WillRepeatedly(testing::Return(selectedColumns));
+
+ // set getEncoding
+ proto::ColumnEncoding directEncoding;
+ directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+ EXPECT_CALL(streams, getEncoding(testing::_))
+ .WillRepeatedly(testing::Return(directEncoding));
+
+ // set getStream
+ EXPECT_CALL(streams, getStreamProxy(0, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+ // alternate 4 non-null and 4 null via [0xf0 for x in range(512 / 8)]
+ const unsigned char buffer1[] = {0x3d, 0xf0};
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer1, ARRAY_SIZE(buffer1))));
+ // [0x0f for x in range(128 / 8)]
+ const unsigned char buffer2[] = {0x1d, 0x0f};
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer2, ARRAY_SIZE(buffer2))));
+
+ // create the row type
+ std::unique_ptr<Type> rowType = createStructType();
+ rowType->addStructField(createPrimitiveType(BOOLEAN), "col0");
+ rowType->assignIds(0);
+
+ std::unique_ptr<ColumnReader> reader = buildReader(*rowType, streams);
+ LongVectorBatch *longBatch = new LongVectorBatch(1024, *getDefaultPool());
+ StructVectorBatch batch(1024, *getDefaultPool());
+ batch.fields.push_back(longBatch);
+ reader->next(batch, 1, 0);
+ ASSERT_EQ(1, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(1, longBatch->numElements);
+ ASSERT_EQ(true, !longBatch->hasNulls);
+ EXPECT_EQ(0, longBatch->data[0]);
+ reader->skip(506);
+ reader->next(batch, 5, 0);
+ ASSERT_EQ(5, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(5, longBatch->numElements);
+ ASSERT_EQ(true, longBatch->hasNulls);
+ EXPECT_EQ(1, longBatch->data[0]);
+ EXPECT_EQ(true, !longBatch->notNull[1]);
+ EXPECT_EQ(true, !longBatch->notNull[2]);
+ EXPECT_EQ(true, !longBatch->notNull[3]);
+ EXPECT_EQ(true, !longBatch->notNull[4]);
+}
+
+TEST(TestColumnReader, testByteWithNulls) {
+ MockStripeStreams streams;
+
+ // set getSelectedColumns()
+ std::vector<bool> selectedColumns(2, true);
+ EXPECT_CALL(streams, getSelectedColumns())
+ .WillRepeatedly(testing::Return(selectedColumns));
+
+ // set getEncoding
+ proto::ColumnEncoding directEncoding;
+ directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+ EXPECT_CALL(streams, getEncoding(testing::_))
+ .WillRepeatedly(testing::Return(directEncoding));
+
+ // set getStream
+ EXPECT_CALL(streams, getStreamProxy(0, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+ // alternate 4 non-null and 4 null via [0xf0 for x in range(512 / 8)]
+ const unsigned char buffer1[] = {0x3d, 0xf0};
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer1, ARRAY_SIZE(buffer1))));
+
+ // range(256)
+ char buffer[258];
+ buffer[0] = static_cast<char>(0x80);
+ for (unsigned int i = 0; i < 128; ++i) {
+ buffer[i + 1] = static_cast<char>(i);
+ }
+ buffer[129] = static_cast<char>(0x80);
+ for (unsigned int i = 128; i < 256; ++i) {
+ buffer[i + 2] = static_cast<char>(i);
+ }
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer, ARRAY_SIZE(buffer))));
+
+ // create the row type
+ std::unique_ptr<Type> rowType = createStructType();
+ rowType->addStructField(createPrimitiveType(BYTE), "col0");
+ rowType->assignIds(0);
+
+ std::unique_ptr<ColumnReader> reader =
+ buildReader(*rowType, streams);
+ LongVectorBatch *longBatch = new LongVectorBatch(1024, *getDefaultPool());
+ StructVectorBatch batch(1024, *getDefaultPool());
+ batch.fields.push_back(longBatch);
+ reader->next(batch, 512, 0);
+ ASSERT_EQ(512, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(512, longBatch->numElements);
+ ASSERT_EQ(true, longBatch->hasNulls);
+ unsigned int next = 0;
+ for (size_t i = 0; i < batch.numElements; ++i) {
+ if (i & 4) {
+ EXPECT_EQ(0, longBatch->notNull[i]) << "Wrong value at " << i;
+ } else {
+ EXPECT_EQ(1, longBatch->notNull[i]) << "Wrong value at " << i;
+ EXPECT_EQ(static_cast<char>(next++), longBatch->data[i])
+ << "Wrong value at " << i;
+ }
+ }
+}
+
+TEST(TestColumnReader, testByteSkipsWithNulls) {
+ MockStripeStreams streams;
+
+ // set getSelectedColumns()
+ std::vector<bool> selectedColumns(2, true);
+ EXPECT_CALL(streams, getSelectedColumns())
+ .WillRepeatedly(testing::Return(selectedColumns));
+
+ // set getEncoding
+ proto::ColumnEncoding directEncoding;
+ directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+ EXPECT_CALL(streams, getEncoding(testing::_))
+ .WillRepeatedly(testing::Return(directEncoding));
+
+ // set getStream
+ EXPECT_CALL(streams, getStreamProxy(0, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+ // alternate 4 non-null and 4 null via [0xf0 for x in range(512 / 8)]
+ const unsigned char buffer1[] = {0x3d, 0xf0};
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer1, ARRAY_SIZE(buffer1))));
+
+ // range(256)
+ char buffer[258];
+ buffer[0] = static_cast<char>(0x80);
+ for (unsigned int i = 0; i < 128; ++i) {
+ buffer[i + 1] = static_cast<char>(i);
+ }
+ buffer[129] = static_cast<char>(0x80);
+ for (unsigned int i = 128; i < 256; ++i) {
+ buffer[i + 2] = static_cast<char>(i);
+ }
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer, ARRAY_SIZE(buffer))));
+
+ // create the row type
+ std::unique_ptr<Type> rowType = createStructType();
+ rowType->addStructField(createPrimitiveType(BYTE), "col0");
+ rowType->assignIds(0);
+
+ std::unique_ptr<ColumnReader> reader =
+ buildReader(*rowType, streams);
+ LongVectorBatch *longBatch = new LongVectorBatch(1024, *getDefaultPool());
+ StructVectorBatch batch(1024, *getDefaultPool());
+ batch.fields.push_back(longBatch);
+ reader->next(batch, 1, 0);
+ ASSERT_EQ(1, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(1, longBatch->numElements);
+ ASSERT_EQ(true, !longBatch->hasNulls);
+ EXPECT_EQ(0, longBatch->data[0]);
+ reader->skip(506);
+ reader->next(batch, 5, 0);
+ ASSERT_EQ(5, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(5, longBatch->numElements);
+ ASSERT_EQ(true, longBatch->hasNulls);
+ EXPECT_EQ(static_cast<char>(-1), longBatch->data[0]);
+ EXPECT_EQ(true, !longBatch->notNull[1]);
+ EXPECT_EQ(true, !longBatch->notNull[2]);
+ EXPECT_EQ(true, !longBatch->notNull[3]);
+ EXPECT_EQ(true, !longBatch->notNull[4]);
+}
+
+TEST(TestColumnReader, testIntegerWithNulls) {
+ MockStripeStreams streams;
+
+ // set getSelectedColumns()
+ std::vector<bool> selectedColumns(2, true);
+
+ EXPECT_CALL(streams, getSelectedColumns())
+ .WillRepeatedly(testing::Return(selectedColumns));
+
+ // set getEncoding
+ proto::ColumnEncoding directEncoding;
+ directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+ EXPECT_CALL(streams, getEncoding(testing::_))
+ .WillRepeatedly(testing::Return(directEncoding));
+
+ // set getStream
+ EXPECT_CALL(streams, getStreamProxy(0, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+ const unsigned char buffer1[] = {0x16, 0xf0};
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer1, ARRAY_SIZE(buffer1))));
+ const unsigned char buffer2[] = {0x64, 0x01, 0x00};
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer2, ARRAY_SIZE(buffer2))));
+
+ // create the row type
+ std::unique_ptr<Type> rowType = createStructType();
+ rowType->addStructField(createPrimitiveType(INT), "myInt" );
+ rowType->assignIds(0);
+
+ std::unique_ptr<ColumnReader> reader =
+ buildReader(*rowType, streams);
+ LongVectorBatch *longBatch = new LongVectorBatch(1024, *getDefaultPool());
+ StructVectorBatch batch(1024, *getDefaultPool());
+ batch.fields.push_back(longBatch);
+ reader->next(batch, 200, 0);
+ ASSERT_EQ(200, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(200, longBatch->numElements);
+ ASSERT_EQ(true, longBatch->hasNulls);
+ long next = 0;
+ for (size_t i = 0; i < batch.numElements; ++i) {
+ if (i & 4) {
+ EXPECT_EQ(0, longBatch->notNull[i]);
+ } else {
+ EXPECT_EQ(1, longBatch->notNull[i]);
+ EXPECT_EQ(next++, longBatch->data[i]);
+ }
+ }
+}
+
+TEST(TestColumnReader, testDictionaryWithNulls) {
+ MockStripeStreams streams;
+
+ // set getSelectedColumns()
+ std::vector<bool> selectedColumns(2, true);
+ EXPECT_CALL(streams, getSelectedColumns())
+ .WillRepeatedly(testing::Return(selectedColumns));
+
+ // set getEncoding
+ proto::ColumnEncoding directEncoding;
+ directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+ EXPECT_CALL(streams, getEncoding(0))
+ .WillRepeatedly(testing::Return(directEncoding));
+ proto::ColumnEncoding dictionaryEncoding;
+ dictionaryEncoding.set_kind(proto::ColumnEncoding_Kind_DICTIONARY);
+ dictionaryEncoding.set_dictionarysize(2);
+ EXPECT_CALL(streams, getEncoding(1))
+ .WillRepeatedly(testing::Return(dictionaryEncoding));
+
+ // set getStream
+ EXPECT_CALL(streams, getStreamProxy(0, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+ const unsigned char buffer1[] = { 0x19, 0xf0 };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer1, ARRAY_SIZE(buffer1))));
+ const unsigned char buffer2[] = { 0x2f, 0x00, 0x00, 0x2f, 0x00, 0x01 };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer2, ARRAY_SIZE(buffer2))));
+ const unsigned char buffer3[] = { 0x4f, 0x52, 0x43, 0x4f, 0x77, 0x65, 0x6e };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_DICTIONARY_DATA,
+ false))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer3, ARRAY_SIZE(buffer3))));
+ const unsigned char buffer4[] = { 0x02, 0x01, 0x03 };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_LENGTH, false))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer4, ARRAY_SIZE(buffer4))));
+
+ // create the row type
+ std::unique_ptr<Type> rowType = createStructType();
+ rowType->addStructField(createPrimitiveType(STRING), "myString");
+ rowType->assignIds(0);
+
+ std::unique_ptr<ColumnReader> reader =
+ buildReader(*rowType, streams);
+ StringVectorBatch *stringBatch = new StringVectorBatch(1024,
+ *getDefaultPool());
+ StructVectorBatch batch(1024, *getDefaultPool());
+ batch.fields.push_back(stringBatch);
+ reader->next(batch, 200, 0);
+ ASSERT_EQ(200, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(200, stringBatch->numElements);
+ ASSERT_EQ(true, stringBatch->hasNulls);
+ for (size_t i = 0; i < batch.numElements; ++i) {
+ if (i & 4) {
+ EXPECT_EQ(0, stringBatch->notNull[i]);
+ } else {
+ EXPECT_EQ(1, stringBatch->notNull[i]);
+ const char* expected = i < 98 ? "ORC" : "Owen";
+ ASSERT_EQ(strlen(expected), stringBatch->length[i])
+ << "Wrong length at " << i;
+ for (size_t letter = 0; letter < strlen(expected); ++letter) {
+ EXPECT_EQ(expected[letter], stringBatch->data[i][letter])
+ << "Wrong contents at " << i << ", " << letter;
+ }
+ }
+ }
+}
+
+TEST(TestColumnReader, testVarcharDictionaryWithNulls) {
+ MockStripeStreams streams;
+
+ // set getSelectedColumns()
+ std::vector<bool> selectedColumns(3, true);
+ selectedColumns.push_back(false);
+
+ EXPECT_CALL(streams, getSelectedColumns())
+ .WillRepeatedly(testing::Return(selectedColumns));
+
+ // set getEncoding
+ proto::ColumnEncoding directEncoding;
+ directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+ EXPECT_CALL(streams, getEncoding(0))
+ .WillRepeatedly(testing::Return(directEncoding));
+
+ proto::ColumnEncoding dictionary2Encoding;
+ dictionary2Encoding.set_kind(proto::ColumnEncoding_Kind_DICTIONARY);
+ dictionary2Encoding.set_dictionarysize(2);
+ EXPECT_CALL(streams, getEncoding(1))
+ .WillRepeatedly(testing::Return(dictionary2Encoding));
+
+ proto::ColumnEncoding dictionary0Encoding;
+ dictionary0Encoding.set_kind(proto::ColumnEncoding_Kind_DICTIONARY);
+ dictionary0Encoding.set_dictionarysize(0);
+ EXPECT_CALL(streams, getEncoding(testing::Ge(2)))
+ .WillRepeatedly(testing::Return(dictionary0Encoding));
+
+ // set getStream
+ EXPECT_CALL(streams, getStreamProxy(0, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ const unsigned char buffer1[] = { 0x16, 0xff };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer1, ARRAY_SIZE(buffer1))));
+ const unsigned char buffer2[] = { 0x61, 0x00, 0x01, 0x61, 0x00, 0x00 };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer2, ARRAY_SIZE(buffer2))));
+ const unsigned char buffer3[] = { 0x4f, 0x52, 0x43, 0x4f, 0x77,
+ 0x65, 0x6e };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_DICTIONARY_DATA,
+ false))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer3, ARRAY_SIZE(buffer3))));
+ const unsigned char buffer4[] = { 0x02, 0x01, 0x03 };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_LENGTH, false))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer4, ARRAY_SIZE(buffer4))));
+
+ const unsigned char buffer5[] = {0x16, 0x00};
+ EXPECT_CALL(streams, getStreamProxy(2, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer5, ARRAY_SIZE(buffer5))));
+
+ // all three return an empty stream
+ EXPECT_CALL(streams, getStreamProxy(2, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer5, 0)));
+ EXPECT_CALL(streams, getStreamProxy(2, proto::Stream_Kind_DICTIONARY_DATA,
+ false))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer5, 0)));
+ EXPECT_CALL(streams, getStreamProxy(2, proto::Stream_Kind_LENGTH, false))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer5, 0)));
+
+ // create the row type
+ std::unique_ptr<Type> rowType = createStructType();
+ rowType->addStructField(createPrimitiveType(VARCHAR), "col0");
+ rowType->addStructField(createPrimitiveType(CHAR), "col1");
+ rowType->addStructField(createPrimitiveType(STRING), "col2");
+ rowType->assignIds(0);
+
+ std::unique_ptr<ColumnReader> reader =
+ buildReader(*rowType, streams);
+ StructVectorBatch batch(1024, *getDefaultPool());
+ StringVectorBatch *stringBatch = new StringVectorBatch(1024,
+ *getDefaultPool());
+ StringVectorBatch *nullBatch = new StringVectorBatch(1024,
+ *getDefaultPool());
+ batch.fields.push_back(stringBatch);
+ batch.fields.push_back(nullBatch);
+ reader->next(batch, 200, 0);
+ ASSERT_EQ(200, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(200, stringBatch->numElements);
+ ASSERT_EQ(true, !stringBatch->hasNulls);
+ ASSERT_EQ(200, nullBatch->numElements);
+ ASSERT_EQ(true, nullBatch->hasNulls);
+ for (size_t i = 0; i < batch.numElements; ++i) {
+ EXPECT_EQ(true, stringBatch->notNull[i]);
+ EXPECT_EQ(true, !nullBatch->notNull[i]);
+ const char* expected = i < 100 ? "Owen" : "ORC";
+ ASSERT_EQ(strlen(expected), stringBatch->length[i])
+ << "Wrong length at " << i;
+ for (size_t letter = 0; letter < strlen(expected); ++letter) {
+ EXPECT_EQ(expected[letter], stringBatch->data[i][letter])
+ << "Wrong contents at " << i << ", " << letter;
+ }
+ }
+}
+
+TEST(TestColumnReader, testSubstructsWithNulls) {
+ MockStripeStreams streams;
+
+ // set getSelectedColumns()
+ std::vector<bool> selectedColumns(4, true);
+ EXPECT_CALL(streams, getSelectedColumns())
+ .WillRepeatedly(testing::Return(selectedColumns));
+
+ // set getEncoding
+ proto::ColumnEncoding directEncoding;
+ directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+ EXPECT_CALL(streams, getEncoding(testing::_))
+ .WillRepeatedly(testing::Return(directEncoding));
+
+ // set getStream
+ EXPECT_CALL(streams, getStreamProxy(0, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ const unsigned char buffer1[] = {0x16, 0x0f};
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer1, ARRAY_SIZE(buffer1))));
+
+ const unsigned char buffer2[] = { 0x0a, 0x55 };
+ EXPECT_CALL(streams, getStreamProxy(2, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer2, ARRAY_SIZE(buffer2))));
+
+ const unsigned char buffer3[] = { 0x04, 0xf0 };
+ EXPECT_CALL(streams, getStreamProxy(3, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer3, ARRAY_SIZE(buffer3))));
+
+ const unsigned char buffer4[] = { 0x17, 0x01, 0x00 };
+ EXPECT_CALL(streams, getStreamProxy(3, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer4, ARRAY_SIZE(buffer4))));
+
+ // create the row type
+ std::unique_ptr<Type> rowType = createStructType();
+ rowType->addStructField(createStructType(), "col0")
+ .addStructField(createStructType(), "col1")
+ .addStructField(createPrimitiveType(LONG), "col2");
+ rowType->assignIds(0);
+
+ std::unique_ptr<ColumnReader> reader = buildReader(*rowType, streams);
+
+ StructVectorBatch batch(1024, *getDefaultPool());
+ StructVectorBatch *middle = new StructVectorBatch(1024, *getDefaultPool());
+ StructVectorBatch *inner = new StructVectorBatch(1024, *getDefaultPool());
+ LongVectorBatch *longs = new LongVectorBatch(1024, *getDefaultPool());
+ batch.fields.push_back(middle);
+ middle->fields.push_back(inner);
+ inner->fields.push_back(longs);
+ reader->next(batch, 200, 0);
+ ASSERT_EQ(200, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(200, middle->numElements);
+ ASSERT_EQ(true, middle->hasNulls);
+ ASSERT_EQ(200, inner->numElements);
+ ASSERT_EQ(true, inner->hasNulls);
+ ASSERT_EQ(200, longs->numElements);
+ ASSERT_EQ(true, longs->hasNulls);
+ long middleCount = 0;
+ long innerCount = 0;
+ long longCount = 0;
+ for (size_t i = 0; i < batch.numElements; ++i) {
+ if (i & 4) {
+ EXPECT_EQ(true, middle->notNull[i]) << "Wrong at " << i;
+ if (middleCount++ & 1) {
+ EXPECT_EQ(true, inner->notNull[i]) << "Wrong at " << i;
+ if (innerCount++ & 4) {
+ EXPECT_EQ(true, !longs->notNull[i]) << "Wrong at " << i;
+ } else {
+ EXPECT_EQ(true, longs->notNull[i]) << "Wrong at " << i;
+ EXPECT_EQ(longCount++, longs->data[i]) << "Wrong at " << i;
+ }
+ } else {
+ EXPECT_EQ(true, !inner->notNull[i]) << "Wrong at " << i;
+ EXPECT_EQ(true, !longs->notNull[i]) << "Wrong at " << i;
+ }
+ } else {
+ EXPECT_EQ(true, !middle->notNull[i]) << "Wrong at " << i;
+ EXPECT_EQ(true, !inner->notNull[i]) << "Wrong at " << i;
+ EXPECT_EQ(true, !longs->notNull[i]) << "Wrong at " << i;
+ }
+ }
+}
+
+TEST(TestColumnReader, testSkipWithNulls) {
+ MockStripeStreams streams;
+
+ // set getSelectedColumns()
+ std::vector<bool> selectedColumns(3, true);
+ EXPECT_CALL(streams, getSelectedColumns())
+ .WillRepeatedly(testing::Return(selectedColumns));
+
+ // set getEncoding
+ proto::ColumnEncoding directEncoding;
+ directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+ EXPECT_CALL(streams, getEncoding(testing::_))
+ .WillRepeatedly(testing::Return(directEncoding));
+ proto::ColumnEncoding dictionaryEncoding;
+ dictionaryEncoding.set_kind(proto::ColumnEncoding_Kind_DICTIONARY);
+ dictionaryEncoding.set_dictionarysize(100);
+ EXPECT_CALL(streams, getEncoding(2))
+ .WillRepeatedly(testing::Return(dictionaryEncoding));
+
+ // set getStream
+ EXPECT_CALL(streams, getStreamProxy(0, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+ const unsigned char buffer1[] = { 0x03, 0x00, 0xff, 0x3f, 0x08, 0xff,
+ 0xff, 0xfc, 0x03, 0x00 };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer1, ARRAY_SIZE(buffer1))));
+ EXPECT_CALL(streams, getStreamProxy(2, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer1, ARRAY_SIZE(buffer1))));
+
+ const unsigned char buffer2[] = { 0x61, 0x01, 0x00 };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer2, ARRAY_SIZE(buffer2))));
+ EXPECT_CALL(streams, getStreamProxy(2, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer2, ARRAY_SIZE(buffer2))));
+
+ // fill the dictionary with '00' to '99'
+ char digits[200];
+ for (int i = 0; i < 10; ++i) {
+ for (int j = 0; j < 10; ++j) {
+ digits[2 * (10 * i + j)] = static_cast<char>('0' + i);
+ digits[2 * (10 * i + j) + 1] = static_cast<char>('0' + j);
+ }
+ }
+ EXPECT_CALL(streams, getStreamProxy(2, proto::Stream_Kind_DICTIONARY_DATA,
+ false))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (digits, ARRAY_SIZE(digits))));
+ const unsigned char buffer3[] = {0x61, 0x00, 0x02};
+ EXPECT_CALL(streams, getStreamProxy(2, proto::Stream_Kind_LENGTH, false))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer3, ARRAY_SIZE(buffer3))));
+
+ // create the row type
+ std::unique_ptr<Type> rowType = createStructType();
+ rowType->addStructField(createPrimitiveType(INT), "myInt");
+ rowType->addStructField(createPrimitiveType(STRING), "myString");
+ rowType->assignIds(0);
+
+ std::unique_ptr<ColumnReader> reader =
+ buildReader(*rowType, streams);
+ StructVectorBatch batch(100, *getDefaultPool());
+ LongVectorBatch *longBatch = new LongVectorBatch(100, *getDefaultPool());
+ StringVectorBatch *stringBatch =
+ new StringVectorBatch(100, *getDefaultPool());
+ batch.fields.push_back(longBatch);
+ batch.fields.push_back(stringBatch);
+ reader->next(batch, 20, 0);
+ ASSERT_EQ(20, batch.numElements);
+ ASSERT_EQ(20, longBatch->numElements);
+ ASSERT_EQ(20, stringBatch->numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(true, longBatch->hasNulls);
+ ASSERT_EQ(true, stringBatch->hasNulls);
+ for (size_t i = 0; i < 20; ++i) {
+ EXPECT_EQ(true, !longBatch->notNull[i]) << "Wrong at " << i;
+ EXPECT_EQ(true, !stringBatch->notNull[i]) << "Wrong at " << i;
+ }
+ reader->skip(30);
+ reader->next(batch, 100, 0);
+ ASSERT_EQ(100, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(true, !longBatch->hasNulls);
+ ASSERT_EQ(true, !stringBatch->hasNulls);
+ for (size_t i = 0; i < 10; ++i) {
+ for (size_t j = 0; j < 10; ++j) {
+ size_t k = 10 * i + j;
+ EXPECT_EQ(1, longBatch->notNull[k]) << "Wrong at " << k;
+ ASSERT_EQ(2, stringBatch->length[k])<< "Wrong at " << k;
+ EXPECT_EQ('0' + static_cast<char>(i), stringBatch->data[k][0])
+ << "Wrong at " << k;
+ EXPECT_EQ('0' + static_cast<char>(j), stringBatch->data[k][1])
+ << "Wrong at " << k;
+ }
+ }
+ reader->skip(50);
+}
+
+TEST(TestColumnReader, testBinaryDirect) {
+ MockStripeStreams streams;
+
+ // set getSelectedColumns()
+ std::vector<bool> selectedColumns(2, true);
+ EXPECT_CALL(streams, getSelectedColumns())
+ .WillRepeatedly(testing::Return(selectedColumns));
+
+ // set getEncoding
+ proto::ColumnEncoding directEncoding;
+ directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+ EXPECT_CALL(streams, getEncoding(testing::_))
+ .WillRepeatedly(testing::Return(directEncoding));
+
+ // set getStream
+ EXPECT_CALL(streams, getStreamProxy(0, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ char blob[200];
+ for (size_t i = 0; i < 10; ++i) {
+ for (size_t j = 0; j < 10; ++j) {
+ blob[2 * (10 * i + j)] = static_cast<char>(i);
+ blob[2 * (10 * i + j) + 1] = static_cast<char>(j);
+ }
+ }
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (blob, ARRAY_SIZE(blob))));
+
+ const unsigned char buffer[] = { 0x61, 0x00, 0x02 };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_LENGTH, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer, ARRAY_SIZE(buffer))));
+
+ // create the row type
+ std::unique_ptr<Type> rowType = createStructType();
+ rowType->addStructField(createPrimitiveType(BINARY), "col0");
+ rowType->assignIds(0);
+
+ std::unique_ptr<ColumnReader> reader = buildReader(*rowType, streams);
+
+ StructVectorBatch batch(1024, *getDefaultPool());
+ StringVectorBatch *strings = new StringVectorBatch(1024, *getDefaultPool());
+ batch.fields.push_back(strings);
+ for (size_t i = 0; i < 2; ++i) {
+ reader->next(batch, 50, 0);
+ ASSERT_EQ(50, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(50, strings->numElements);
+ ASSERT_EQ(true, !strings->hasNulls);
+ for (size_t j = 0; j < batch.numElements; ++j) {
+ ASSERT_EQ(2, strings->length[j]);
+ EXPECT_EQ((50 * i + j) / 10, strings->data[j][0]);
+ EXPECT_EQ((50 * i + j) % 10, strings->data[j][1]);
+ }
+ }
+}
+
+TEST(TestColumnReader, testBinaryDirectWithNulls) {
+ MockStripeStreams streams;
+
+ // set getSelectedColumns()
+ std::vector<bool> selectedColumns(2, true);
+ EXPECT_CALL(streams, getSelectedColumns())
+ .WillRepeatedly(testing::Return(selectedColumns));
+
+ // set getEncoding
+ proto::ColumnEncoding directEncoding;
+ directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+ EXPECT_CALL(streams, getEncoding(testing::_))
+ .WillRepeatedly(testing::Return(directEncoding));
+
+ // set getStream
+ EXPECT_CALL(streams, getStreamProxy(0, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ const unsigned char buffer1[] = {0x1d, 0xf0};
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer1, ARRAY_SIZE(buffer1))));
+
+ char blob[256];
+ for (size_t i = 0; i < 8; ++i) {
+ for (size_t j = 0; j < 16; ++j) {
+ blob[2 * (16 * i + j)] = static_cast<char>('A' + i);
+ blob[2 * (16 * i + j) + 1] = static_cast<char>('A' + j);
+ }
+ }
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (blob, ARRAY_SIZE(blob))));
+
+ const unsigned char buffer2[] = {0x7d, 0x00, 0x02};
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_LENGTH, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer2, ARRAY_SIZE(buffer2))));
+
+ // create the row type
+ std::unique_ptr<Type> rowType = createStructType();
+ rowType->addStructField(createPrimitiveType(BINARY), "col0");
+ rowType->assignIds(0);
+
+ std::unique_ptr<ColumnReader> reader = buildReader(*rowType, streams);
+
+ StructVectorBatch batch(1024, *getDefaultPool());
+ StringVectorBatch *strings = new StringVectorBatch(1024, *getDefaultPool());
+ batch.fields.push_back(strings);
+ size_t next = 0;
+ for (size_t i = 0; i < 2; ++i) {
+ reader->next(batch, 128, 0);
+ ASSERT_EQ(128, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(128, strings->numElements);
+ ASSERT_EQ(true, strings->hasNulls);
+ for (size_t j = 0; j < batch.numElements; ++j) {
+ ASSERT_EQ(((128 * i + j) & 4) == 0, strings->notNull[j]);
+ if (strings->notNull[j]) {
+ ASSERT_EQ(2, strings->length[j]);
+ EXPECT_EQ('A' + static_cast<char>(next / 16), strings->data[j][0]);
+ EXPECT_EQ('A' + static_cast<char>(next % 16), strings->data[j][1]);
+ next += 1;
+ }
+ }
+ }
+}
+
+TEST(TestColumnReader, testShortBlobError) {
+ MockStripeStreams streams;
+
+ // set getSelectedColumns()
+ std::vector<bool> selectedColumns(2, true);
+ EXPECT_CALL(streams, getSelectedColumns())
+ .WillRepeatedly(testing::Return(selectedColumns));
+
+ // set getEncoding
+ proto::ColumnEncoding directEncoding;
+ directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+ EXPECT_CALL(streams, getEncoding(testing::_))
+ .WillRepeatedly(testing::Return(directEncoding));
+
+ // set getStream
+ EXPECT_CALL(streams, getStreamProxy(0, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ char blob[100];
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (blob, ARRAY_SIZE(blob))));
+
+ const unsigned char buffer1[] = {0x61, 0x00, 0x02};
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_LENGTH, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer1, ARRAY_SIZE(buffer1))));
+
+ // create the row type
+ std::unique_ptr<Type> rowType = createStructType();
+ rowType->addStructField(createPrimitiveType(STRING), "col0");
+ rowType->assignIds(0);
+
+ std::unique_ptr<ColumnReader> reader = buildReader(*rowType, streams);
+
+ StructVectorBatch batch(1024, *getDefaultPool());
+ StringVectorBatch *strings = new StringVectorBatch(1024, *getDefaultPool());
+ batch.fields.push_back(strings);
+ EXPECT_THROW(reader->next(batch, 100, 0), ParseError);
+}
+
+TEST(TestColumnReader, testStringDirectShortBuffer) {
+ MockStripeStreams streams;
+
+ // set getSelectedColumns()
+ std::vector<bool> selectedColumns(2, true);
+ EXPECT_CALL(streams, getSelectedColumns())
+ .WillRepeatedly(testing::Return(selectedColumns));
+
+ // set getEncoding
+ proto::ColumnEncoding directEncoding;
+ directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+ EXPECT_CALL(streams, getEncoding(testing::_))
+ .WillRepeatedly(testing::Return(directEncoding));
+
+ // set getStream
+ EXPECT_CALL(streams, getStreamProxy(0, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ char blob[200];
+ for (size_t i = 0; i < 10; ++i) {
+ for (size_t j = 0; j < 10; ++j) {
+ blob[2 * (10 * i + j)] = static_cast<char>(i);
+ blob[2 * (10 * i + j) + 1] = static_cast<char>(j);
+ }
+ }
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (blob, ARRAY_SIZE(blob), 3)));
+
+ const unsigned char buffer1[] = {0x61, 0x00, 0x02};
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_LENGTH, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer1, ARRAY_SIZE(buffer1))));
+
+ // create the row type
+ std::unique_ptr<Type> rowType = createStructType();
+ rowType->addStructField(createPrimitiveType(STRING), "col0");
+ rowType->assignIds(0);
+
+ std::unique_ptr<ColumnReader> reader = buildReader(*rowType, streams);
+
+ StructVectorBatch batch(25, *getDefaultPool());
+ StringVectorBatch *strings = new StringVectorBatch(25, *getDefaultPool());
+ batch.fields.push_back(strings);
+ for (size_t i = 0; i < 4; ++i) {
+ reader->next(batch, 25, 0);
+ ASSERT_EQ(25, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(25, strings->numElements);
+ ASSERT_EQ(true, !strings->hasNulls);
+ for (size_t j = 0; j < batch.numElements; ++j) {
+ ASSERT_EQ(2, strings->length[j]);
+ EXPECT_EQ((25 * i + j) / 10, strings->data[j][0]);
+ EXPECT_EQ((25 * i + j) % 10, strings->data[j][1]);
+ }
+ }
+}
+
+TEST(TestColumnReader, testStringDirectShortBufferWithNulls) {
+ MockStripeStreams streams;
+
+ // set getSelectedColumns()
+ std::vector<bool> selectedColumns(2, true);
+ EXPECT_CALL(streams, getSelectedColumns())
+ .WillRepeatedly(testing::Return(selectedColumns));
+
+ // set getEncoding
+ proto::ColumnEncoding directEncoding;
+ directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+ EXPECT_CALL(streams, getEncoding(testing::_))
+ .WillRepeatedly(testing::Return(directEncoding));
+
+ // set getStream
+ EXPECT_CALL(streams, getStreamProxy(0, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ const unsigned char buffer1[] = {0x3d, 0xf0};
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer1, ARRAY_SIZE(buffer1))));
+
+ char blob[512];
+ for (size_t i = 0; i < 16; ++i) {
+ for (size_t j = 0; j < 16; ++j) {
+ blob[2 * (16 * i + j)] = static_cast<char>('A' + i);
+ blob[2 * (16 * i + j) + 1] = static_cast<char>('A' + j);
+ }
+ }
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (blob, ARRAY_SIZE(blob), 30)));
+
+ const unsigned char buffer2[] = { 0x7d, 0x00, 0x02, 0x7d, 0x00, 0x02 };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_LENGTH, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer2, ARRAY_SIZE(buffer2))));
+
+ // create the row type
+ std::unique_ptr<Type> rowType = createStructType();
+ rowType->addStructField(createPrimitiveType(STRING), "col0");
+ rowType->assignIds(0);
+
+ std::unique_ptr<ColumnReader> reader = buildReader(*rowType, streams);
+
+ StructVectorBatch batch(64, *getDefaultPool());
+ StringVectorBatch *strings = new StringVectorBatch(64, *getDefaultPool());
+ batch.fields.push_back(strings);
+ size_t next = 0;
+ for (size_t i = 0; i < 8; ++i) {
+ reader->next(batch, 64, 0);
+ ASSERT_EQ(64, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(64, strings->numElements);
+ ASSERT_EQ(true, strings->hasNulls);
+ for (size_t j = 0; j < batch.numElements; ++j) {
+ ASSERT_EQ((j & 4) == 0, strings->notNull[j]);
+ if (strings->notNull[j]) {
+ ASSERT_EQ(2, strings->length[j]);
+ EXPECT_EQ('A' + next / 16, strings->data[j][0]);
+ EXPECT_EQ('A' + next % 16, strings->data[j][1]);
+ next += 1;
+ }
+ }
+ }
+}
+
+TEST(TestColumnReader, testStringDirectSkip) {
+ MockStripeStreams streams;
+
+ // set getSelectedColumns()
+ std::vector<bool> selectedColumns(2, true);
+ EXPECT_CALL(streams, getSelectedColumns())
+ .WillRepeatedly(testing::Return(selectedColumns));
+
+ // set getEncoding
+ proto::ColumnEncoding directEncoding;
+ directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+ EXPECT_CALL(streams, getEncoding(testing::_))
+ .WillRepeatedly(testing::Return(directEncoding));
+
+ // set getStream
+ EXPECT_CALL(streams, getStreamProxy(0, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ // sum(0 to 1199)
+ const size_t BLOB_SIZE = 719400;
+ char blob[BLOB_SIZE];
+ size_t posn = 0;
+ for (size_t item = 0; item < 1200; ++item) {
+ for (size_t ch = 0; ch < item; ++ch) {
+ blob[posn++] = static_cast<char>(ch);
+ }
+ }
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (blob, BLOB_SIZE, 200)));
+
+ // the stream of 0 to 1199
+ const unsigned char buffer1[] =
+ { 0x7f, 0x01, 0x00,
+ 0x7f, 0x01, 0x82, 0x01,
+ 0x7f, 0x01, 0x84, 0x02,
+ 0x7f, 0x01, 0x86, 0x03,
+ 0x7f, 0x01, 0x88, 0x04,
+ 0x7f, 0x01, 0x8a, 0x05,
+ 0x7f, 0x01, 0x8c, 0x06,
+ 0x7f, 0x01, 0x8e, 0x07,
+ 0x7f, 0x01, 0x90, 0x08,
+ 0x1b, 0x01, 0x92, 0x09 };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_LENGTH, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer1, ARRAY_SIZE(buffer1))));
+
+ // create the row type
+ std::unique_ptr<Type> rowType = createStructType();
+ rowType->addStructField(createPrimitiveType(STRING), "col0");
+ rowType->assignIds(0);
+
+ std::unique_ptr<ColumnReader> reader = buildReader(*rowType, streams);
+
+ StructVectorBatch batch(2, *getDefaultPool());
+ StringVectorBatch *strings = new StringVectorBatch(2, *getDefaultPool());
+ batch.fields.push_back(strings);
+ reader->next(batch, 2, 0);
+ ASSERT_EQ(2, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(2, strings->numElements);
+ ASSERT_EQ(true, !strings->hasNulls);
+ for (size_t i = 0; i < batch.numElements; ++i) {
+ ASSERT_EQ(i, strings->length[i]);
+ for (size_t j = 0; j < i; ++j) {
+ EXPECT_EQ(static_cast<char>(j), strings->data[i][j]);
+ }
+ }
+ reader->skip(14);
+ reader->next(batch, 2, 0);
+ ASSERT_EQ(2, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(2, strings->numElements);
+ ASSERT_EQ(true, !strings->hasNulls);
+ for (size_t i = 0; i < batch.numElements; ++i) {
+ ASSERT_EQ(16 + i, strings->length[i]);
+ for (size_t j = 0; j < 16 + i; ++j) {
+ EXPECT_EQ(static_cast<char>(j), strings->data[i][j]);
+ }
+ }
+ reader->skip(1180);
+ reader->next(batch, 2, 0);
+ ASSERT_EQ(2, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(2, strings->numElements);
+ ASSERT_EQ(true, !strings->hasNulls);
+ for (size_t i = 0; i < batch.numElements; ++i) {
+ ASSERT_EQ(1198 + i, strings->length[i]);
+ for (size_t j = 0; j < 1198 + i; ++j) {
+ EXPECT_EQ(static_cast<char>(j), strings->data[i][j]);
+ }
+ }
+}
+
+TEST(TestColumnReader, testStringDirectSkipWithNulls) {
+ MockStripeStreams streams;
+
+ // set getSelectedColumns()
+ std::vector<bool> selectedColumns(2, true);
+ EXPECT_CALL(streams, getSelectedColumns())
+ .WillRepeatedly(testing::Return(selectedColumns));
+
+ // set getEncoding
+ proto::ColumnEncoding directEncoding;
+ directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+ EXPECT_CALL(streams, getEncoding(testing::_))
+ .WillRepeatedly(testing::Return(directEncoding));
+
+ // set getStream
+ EXPECT_CALL(streams, getStreamProxy(0, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ // alternate 4 non-null and 4 null via [0xf0 for x in range(2400 / 8)]
+ const unsigned char buffer1[] = { 0x7f, 0xf0, 0x7f, 0xf0, 0x25, 0xf0 };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer1, ARRAY_SIZE(buffer1))));
+
+ // sum(range(1200))
+ const size_t BLOB_SIZE = 719400;
+
+ // each string is [x % 256 for x in range(r)]
+ char blob[BLOB_SIZE];
+ size_t posn = 0;
+ for (size_t item = 0; item < 1200; ++item) {
+ for (size_t ch = 0; ch < item; ++ch) {
+ blob[posn++] = static_cast<char>(ch);
+ }
+ }
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (blob, BLOB_SIZE, 200)));
+
+ // range(1200)
+ const unsigned char buffer2[] = { 0x7f, 0x01, 0x00,
+ 0x7f, 0x01, 0x82, 0x01,
+ 0x7f, 0x01, 0x84, 0x02,
+ 0x7f, 0x01, 0x86, 0x03,
+ 0x7f, 0x01, 0x88, 0x04,
+ 0x7f, 0x01, 0x8a, 0x05,
+ 0x7f, 0x01, 0x8c, 0x06,
+ 0x7f, 0x01, 0x8e, 0x07,
+ 0x7f, 0x01, 0x90, 0x08,
+ 0x1b, 0x01, 0x92, 0x09 };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_LENGTH, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer2, ARRAY_SIZE(buffer2))));
+
+ // create the row type
+ std::unique_ptr<Type> rowType = createStructType();
+ rowType->addStructField(createPrimitiveType(STRING), "col0");
+ rowType->assignIds(0);
+
+ std::unique_ptr<ColumnReader> reader = buildReader(*rowType, streams);
+
+ StructVectorBatch batch(2, *getDefaultPool());
+ StringVectorBatch *strings = new StringVectorBatch(2, *getDefaultPool());
+ batch.fields.push_back(strings);
+ reader->next(batch, 2, 0);
+ ASSERT_EQ(2, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(2, strings->numElements);
+ ASSERT_EQ(true, !strings->hasNulls);
+ for (size_t i = 0; i < batch.numElements; ++i) {
+ ASSERT_EQ(i, strings->length[i]);
+ for (size_t j = 0; j < i; ++j) {
+ EXPECT_EQ(static_cast<char>(j), strings->data[i][j]);
+ }
+ }
+ reader->skip(30);
+ reader->next(batch, 2, 0);
+ ASSERT_EQ(2, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(2, strings->numElements);
+ ASSERT_EQ(true, !strings->hasNulls);
+ for (size_t i = 0; i < batch.numElements; ++i) {
+ ASSERT_EQ(16 + i, strings->length[i]);
+ for (size_t j = 0; j < 16 + i; ++j) {
+ EXPECT_EQ(static_cast<char>(j), strings->data[i][j]);
+ }
+ }
+ reader->skip(2364);
+ reader->next(batch, 2, 0);
+ ASSERT_EQ(2, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(2, strings->numElements);
+ ASSERT_EQ(true, strings->hasNulls);
+ for (size_t i = 0; i < batch.numElements; ++i) {
+ EXPECT_EQ(true, !strings->notNull[i]);
+ }
+}
+
+TEST(TestColumnReader, testList) {
+ MockStripeStreams streams;
+
+ // set getSelectedColumns()
+ std::vector<bool> selectedColumns(3, true);
+ EXPECT_CALL(streams, getSelectedColumns())
+ .WillRepeatedly(testing::Return(selectedColumns));
+
+ // set getEncoding
+ proto::ColumnEncoding directEncoding;
+ directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+ EXPECT_CALL(streams, getEncoding(testing::_))
+ .WillRepeatedly(testing::Return(directEncoding));
+
+ // set getStream
+ EXPECT_CALL(streams, getStreamProxy(testing::_,
+ proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ // [2 for x in range(600)]
+ const unsigned char buffer1[] = { 0x7f, 0x00, 0x02,
+ 0x7f, 0x00, 0x02,
+ 0x7f, 0x00, 0x02,
+ 0x7f, 0x00, 0x02,
+ 0x4d, 0x00, 0x02 };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_LENGTH, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer1, ARRAY_SIZE(buffer1))));
+
+ // range(1200)
+ const unsigned char buffer2[] = { 0x7f, 0x01, 0x00,
+ 0x7f, 0x01, 0x84, 0x02,
+ 0x7f, 0x01, 0x88, 0x04,
+ 0x7f, 0x01, 0x8c, 0x06,
+ 0x7f, 0x01, 0x90, 0x08,
+ 0x7f, 0x01, 0x94, 0x0a,
+ 0x7f, 0x01, 0x98, 0x0c,
+ 0x7f, 0x01, 0x9c, 0x0e,
+ 0x7f, 0x01, 0xa0, 0x10,
+ 0x1b, 0x01, 0xa4, 0x12 };
+ EXPECT_CALL(streams, getStreamProxy(2, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer2, ARRAY_SIZE(buffer2))));
+
+ // create the row type
+ std::unique_ptr<Type> rowType = createStructType();
+ rowType->addStructField(createListType(createPrimitiveType(LONG)), "col0");
+ rowType->assignIds(0);
+
+ std::unique_ptr<ColumnReader> reader = buildReader(*rowType, streams);
+
+ StructVectorBatch batch(512, *getDefaultPool());
+ ListVectorBatch *lists = new ListVectorBatch(512, *getDefaultPool());
+ LongVectorBatch *longs = new LongVectorBatch(512, *getDefaultPool());
+ batch.fields.push_back(lists);
+ lists->elements = std::unique_ptr < ColumnVectorBatch > (longs);
+ reader->next(batch, 512, 0);
+ ASSERT_EQ(512, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(512, lists->numElements);
+ ASSERT_EQ(true, !lists->hasNulls);
+ ASSERT_EQ(1024, longs->numElements);
+ ASSERT_EQ(true, !longs->hasNulls);
+ for (size_t i = 0; i <= batch.numElements; ++i) {
+ EXPECT_EQ(2 * i, lists->offsets[i]);
+ }
+ for (size_t i = 0; i < longs->numElements; ++i) {
+ EXPECT_EQ(i, longs->data[i]);
+ }
+}
+
+TEST(TestColumnReader, testListPropagateNulls) {
+ MockStripeStreams streams;
+
+ // set getSelectedColumns()
+ std::vector<bool> selectedColumns(4, true);
+ EXPECT_CALL(streams, getSelectedColumns())
+ .WillRepeatedly(testing::Return(selectedColumns));
+
+ std::unique_ptr<Type> rowType = createStructType();
+ rowType->addStructField(createStructType(), "col0")
+ .addStructField(createListType(createPrimitiveType(LONG)), "col0_0");
+ rowType->assignIds(0);
+
+ // set getEncoding
+ proto::ColumnEncoding directEncoding;
+ directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+ EXPECT_CALL(streams, getEncoding(testing::_))
+ .WillRepeatedly(testing::Return(directEncoding));
+
+
+ // set getStream
+ EXPECT_CALL(streams, getStreamProxy(testing::_,
+ proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ // set getStream
+ const unsigned char buffer[] = {0xff, 0x00};
+ EXPECT_CALL(streams, getStreamProxy(1,proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return
+ (new SeekableArrayInputStream(buffer,
+ ARRAY_SIZE(buffer))));
+
+ EXPECT_CALL(streams, getStreamProxy(2, proto::Stream_Kind_LENGTH, true))
+ .WillRepeatedly(testing::Return
+ (new SeekableArrayInputStream(buffer, 0)));
+
+ EXPECT_CALL(streams, getStreamProxy(3, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return
+ (new SeekableArrayInputStream(buffer, 0)));
+
+ // create the row type
+ std::unique_ptr<ColumnReader> reader = buildReader(*rowType, streams);
+
+ StructVectorBatch batch(512, *getDefaultPool());
+ StructVectorBatch *structs = new StructVectorBatch(512, *getDefaultPool());
+ ListVectorBatch *lists = new ListVectorBatch(512, *getDefaultPool());
+ LongVectorBatch *longs = new LongVectorBatch(512, *getDefaultPool());
+ batch.fields.push_back(structs);
+ structs->fields.push_back(lists);
+ lists->elements = std::unique_ptr < ColumnVectorBatch > (longs);
+ reader->next(batch, 8, 0);
+ ASSERT_EQ(8, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(8, structs->numElements);
+ ASSERT_EQ(true, structs->hasNulls);
+ ASSERT_EQ(8, lists->numElements);
+ ASSERT_EQ(true, lists->hasNulls);
+ ASSERT_EQ(0, longs->numElements);
+ ASSERT_EQ(true, !longs->hasNulls);
+ for(size_t i=0; i < 8; ++i) {
+ EXPECT_EQ(true, !structs->notNull[i]);
+ }
+}
+
+TEST(TestColumnReader, testListWithNulls) {
+ MockStripeStreams streams;
+
+ // set getSelectedColumns()
+ std::vector<bool> selectedColumns(3, true);
+ EXPECT_CALL(streams, getSelectedColumns())
+ .WillRepeatedly(testing::Return(selectedColumns));
+
+ // set getEncoding
+ proto::ColumnEncoding directEncoding;
+ directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+ EXPECT_CALL(streams, getEncoding(testing::_))
+ .WillRepeatedly(testing::Return(directEncoding));
+
+ // set getStream
+ EXPECT_CALL(streams, getStreamProxy(0, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ // [0xaa for x in range(2048/8)]
+ const unsigned char buffer1[] = { 0x7f, 0xaa, 0x7b, 0xaa };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer1, ARRAY_SIZE(buffer1))));
+
+ EXPECT_CALL(streams, getStreamProxy(2, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ // [1 for x in range(260)] +
+ // [4 for x in range(260)] +
+ // [0 for x in range(260)] +
+ // [3 for x in range(243)] +
+ // [19]
+ const unsigned char buffer2[] = { 0x7f, 0x00, 0x01,
+ 0x7f, 0x00, 0x01,
+ 0x7f, 0x00, 0x04,
+ 0x7f, 0x00, 0x04,
+ 0x7f, 0x00, 0x00,
+ 0x7f, 0x00, 0x00,
+ 0x7f, 0x00, 0x03,
+ 0x6e, 0x00, 0x03,
+ 0xff, 0x13 };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_LENGTH, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer2, ARRAY_SIZE(buffer2))));
+
+ // range(2048)
+ const unsigned char buffer3[] = { 0x7f, 0x01, 0x00,
+ 0x7f, 0x01, 0x84, 0x02,
+ 0x7f, 0x01, 0x88, 0x04,
+ 0x7f, 0x01, 0x8c, 0x06,
+ 0x7f, 0x01, 0x90, 0x08,
+ 0x7f, 0x01, 0x94, 0x0a,
+ 0x7f, 0x01, 0x98, 0x0c,
+ 0x7f, 0x01, 0x9c, 0x0e,
+ 0x7f, 0x01, 0xa0, 0x10,
+ 0x7f, 0x01, 0xa4, 0x12,
+ 0x7f, 0x01, 0xa8, 0x14,
+ 0x7f, 0x01, 0xac, 0x16,
+ 0x7f, 0x01, 0xb0, 0x18,
+ 0x7f, 0x01, 0xb4, 0x1a,
+ 0x7f, 0x01, 0xb8, 0x1c,
+ 0x5f, 0x01, 0xbc, 0x1e };
+ EXPECT_CALL(streams, getStreamProxy(2, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer3, ARRAY_SIZE(buffer3))));
+
+ // create the row type
+ std::unique_ptr<Type> rowType = createStructType();
+ rowType->addStructField(createListType(createPrimitiveType(LONG)), "col0");
+ rowType->assignIds(0);
+
+ std::unique_ptr<ColumnReader> reader = buildReader(*rowType, streams);
+
+ StructVectorBatch batch(512, *getDefaultPool());
+ ListVectorBatch *lists = new ListVectorBatch(512, *getDefaultPool());
+ LongVectorBatch *longs = new LongVectorBatch(512, *getDefaultPool());
+ batch.fields.push_back(lists);
+ lists->elements = std::unique_ptr < ColumnVectorBatch > (longs);
+ reader->next(batch, 512, 0);
+ ASSERT_EQ(512, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(512, lists->numElements);
+ ASSERT_EQ(true, lists->hasNulls);
+ ASSERT_EQ(256, longs->numElements);
+ ASSERT_EQ(true, !longs->hasNulls);
+ for (size_t i = 0; i < batch.numElements; ++i) {
+ EXPECT_EQ(i % 2 == 0, lists->notNull[i]) << "Wrong value at " << i;
+ EXPECT_EQ((i + 1) / 2, lists->offsets[i]) << "Wrong value at " << i;
+ }
+ EXPECT_EQ(256, lists->offsets[512]);
+ for (size_t i = 0; i < longs->numElements; ++i) {
+ EXPECT_EQ(i, longs->data[i]);
+ }
+
+ reader->next(batch, 512, 0);
+ ASSERT_EQ(512, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(512, lists->numElements);
+ ASSERT_EQ(true, lists->hasNulls);
+ ASSERT_EQ(1012, longs->numElements);
+ ASSERT_EQ(true, !longs->hasNulls);
+ for (size_t i = 0; i < batch.numElements; ++i) {
+ EXPECT_EQ(i % 2 == 0, lists->notNull[i]) << "Wrong value at " << i;
+ if (i < 8) {
+ EXPECT_EQ((i + 1) / 2, lists->offsets[i])
+ << "Wrong value at " << i;
+ } else {
+ EXPECT_EQ(4 * ((i + 1) / 2) - 12, lists->offsets[i])
+ << "Wrong value at " << i;
+ }
+ }
+ EXPECT_EQ(1012, lists->offsets[512]);
+ for (size_t i = 0; i < longs->numElements; ++i) {
+ EXPECT_EQ(256 + i, longs->data[i]);
+ }
+
+ reader->next(batch, 512, 0);
+ ASSERT_EQ(512, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(512, lists->numElements);
+ ASSERT_EQ(true, lists->hasNulls);
+ ASSERT_EQ(32, longs->numElements);
+ ASSERT_EQ(true, !longs->hasNulls);
+ for (size_t i = 0; i < batch.numElements; ++i) {
+ EXPECT_EQ(i % 2 == 0, lists->notNull[i]) << "Wrong value at " << i;
+ if (i < 16) {
+ EXPECT_EQ(4 * ((i + 1) / 2), lists->offsets[i])
+ << "Wrong value at " << i;
+ } else {
+ EXPECT_EQ(32, lists->offsets[i]) << "Wrong value at " << i;
+ }
+ }
+ EXPECT_EQ(32, lists->offsets[512]);
+ for (size_t i = 0; i < longs->numElements; ++i) {
+ EXPECT_EQ(1268 + i, longs->data[i]);
+ }
+
+ reader->next(batch, 512, 0);
+ ASSERT_EQ(512, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(512, lists->numElements);
+ ASSERT_EQ(true, lists->hasNulls);
+ ASSERT_EQ(748, longs->numElements);
+ ASSERT_EQ(true, !longs->hasNulls);
+ for (size_t i = 0; i < batch.numElements; ++i) {
+ EXPECT_EQ(i % 2 == 0, lists->notNull[i]) << "Wrong value at " << i;
+ if (i < 24) {
+ EXPECT_EQ(0, lists->offsets[i]) << "Wrong value at " << i;
+ } else if (i < 510) {
+ EXPECT_EQ(3 * ((i - 23) / 2), lists->offsets[i])
+ << "Wrong value at " << i;
+ } else if (i < 511) {
+ EXPECT_EQ(729, lists->offsets[i]) << "Wrong value at " << i;
+ } else {
+ EXPECT_EQ(748, lists->offsets[i]) << "Wrong value at " << i;
+ }
+ }
+ EXPECT_EQ(748, lists->offsets[512]);
+ for (size_t i = 0; i < longs->numElements; ++i) {
+ EXPECT_EQ(1300 + i, longs->data[i]);
+ }
+}
+
+TEST(TestColumnReader, testListSkipWithNulls) {
+ MockStripeStreams streams;
+
+ // set getSelectedColumns()
+ std::vector<bool> selectedColumns(3, true);
+ EXPECT_CALL(streams, getSelectedColumns())
+ .WillRepeatedly(testing::Return(selectedColumns));
+
+ // set getEncoding
+ proto::ColumnEncoding directEncoding;
+ directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+ EXPECT_CALL(streams, getEncoding(testing::_))
+ .WillRepeatedly(testing::Return(directEncoding));
+
+ // set getStream
+ EXPECT_CALL(streams, getStreamProxy(0, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ // [0xaa for x in range(2048/8)]
+ const unsigned char buffer1[] = { 0x7f, 0xaa, 0x7b, 0xaa };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer1, ARRAY_SIZE(buffer1))));
+
+ EXPECT_CALL(streams, getStreamProxy(2, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ // [1 for x in range(260)] +
+ // [4 for x in range(260)] +
+ // [0 for x in range(260)] +
+ // [3 for x in range(243)] +
+ // [19]
+ const unsigned char buffer2[] = { 0x7f, 0x00, 0x01,
+ 0x7f, 0x00, 0x01,
+ 0x7f, 0x00, 0x04,
+ 0x7f, 0x00, 0x04,
+ 0x7f, 0x00, 0x00,
+ 0x7f, 0x00, 0x00,
+ 0x7f, 0x00, 0x03,
+ 0x6e, 0x00, 0x03,
+ 0xff, 0x13 };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_LENGTH, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer2, ARRAY_SIZE(buffer2))));
+
+ // range(2048)
+ const unsigned char buffer3[] = { 0x7f, 0x01, 0x00,
+ 0x7f, 0x01, 0x84, 0x02,
+ 0x7f, 0x01, 0x88, 0x04,
+ 0x7f, 0x01, 0x8c, 0x06,
+ 0x7f, 0x01, 0x90, 0x08,
+ 0x7f, 0x01, 0x94, 0x0a,
+ 0x7f, 0x01, 0x98, 0x0c,
+ 0x7f, 0x01, 0x9c, 0x0e,
+ 0x7f, 0x01, 0xa0, 0x10,
+ 0x7f, 0x01, 0xa4, 0x12,
+ 0x7f, 0x01, 0xa8, 0x14,
+ 0x7f, 0x01, 0xac, 0x16,
+ 0x7f, 0x01, 0xb0, 0x18,
+ 0x7f, 0x01, 0xb4, 0x1a,
+ 0x7f, 0x01, 0xb8, 0x1c,
+ 0x5f, 0x01, 0xbc, 0x1e };
+ EXPECT_CALL(streams, getStreamProxy(2, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer3, ARRAY_SIZE(buffer3))));
+
+ // create the row type
+ std::unique_ptr<Type> rowType = createStructType();
+ rowType->addStructField(createListType(createPrimitiveType(LONG)), "col0");
+ rowType->assignIds(0);
+
+ std::unique_ptr<ColumnReader> reader = buildReader(*rowType, streams);
+
+ StructVectorBatch batch(1, *getDefaultPool());
+ ListVectorBatch *lists = new ListVectorBatch(1, *getDefaultPool());
+ LongVectorBatch *longs = new LongVectorBatch(1, *getDefaultPool());
+ batch.fields.push_back(lists);
+ lists->elements = std::unique_ptr < ColumnVectorBatch > (longs);
+
+ reader->next(batch, 1, 0);
+ ASSERT_EQ(1, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(1, lists->numElements);
+ ASSERT_EQ(true, !lists->hasNulls);
+ ASSERT_EQ(1, longs->numElements);
+ ASSERT_EQ(true, !longs->hasNulls);
+ EXPECT_EQ(0, lists->offsets[0]);
+ EXPECT_EQ(1, lists->offsets[1]);
+ EXPECT_EQ(0, longs->data[0]);
+
+ reader->skip(13);
+ reader->next(batch, 1, 0);
+ ASSERT_EQ(1, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(1, lists->numElements);
+ ASSERT_EQ(true, !lists->hasNulls);
+ ASSERT_EQ(1, longs->numElements);
+ ASSERT_EQ(true, !longs->hasNulls);
+ EXPECT_EQ(0, lists->offsets[0]);
+ EXPECT_EQ(1, lists->offsets[1]);
+ EXPECT_EQ(7, longs->data[0]);
+
+ reader->skip(2031);
+ reader->next(batch, 2, 0);
+ ASSERT_EQ(2, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(2, lists->numElements);
+ ASSERT_EQ(true, lists->hasNulls);
+ ASSERT_EQ(19, longs->numElements);
+ ASSERT_EQ(true, !longs->hasNulls);
+ EXPECT_EQ(0, lists->offsets[0]);
+ EXPECT_EQ(19, lists->offsets[1]);
+ EXPECT_EQ(19, lists->offsets[2]);
+ for (size_t i = 0; i < longs->numElements; ++i) {
+ EXPECT_EQ(2029 + i, longs->data[i]);
+ }
+}
+
+TEST(TestColumnReader, testListSkipWithNullsNoData) {
+ MockStripeStreams streams;
+
+ // set getSelectedColumns()
+ std::vector<bool> selectedColumns(2, true);
+ selectedColumns.push_back(false);
+ EXPECT_CALL(streams, getSelectedColumns())
+ .WillRepeatedly(testing::Return(selectedColumns));
+
+ // set getEncoding
+ proto::ColumnEncoding directEncoding;
+ directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+ EXPECT_CALL(streams, getEncoding(testing::_))
+ .WillRepeatedly(testing::Return(directEncoding));
+
+ // set getStream
+ EXPECT_CALL(streams, getStreamProxy(0, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ // [0xaa for x in range(2048/8)]
+ const unsigned char buffer1[] = { 0x7f, 0xaa, 0x7b, 0xaa };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer1, ARRAY_SIZE(buffer1))));
+
+ EXPECT_CALL(streams, getStreamProxy(2, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ // [1 for x in range(260)] +
+ // [4 for x in range(260)] +
+ // [0 for x in range(260)] +
+ // [3 for x in range(243)] +
+ // [19]
+ const unsigned char buffer2[] = { 0x7f, 0x00, 0x01,
+ 0x7f, 0x00, 0x01,
+ 0x7f, 0x00, 0x04,
+ 0x7f, 0x00, 0x04,
+ 0x7f, 0x00, 0x00,
+ 0x7f, 0x00, 0x00,
+ 0x7f, 0x00, 0x03,
+ 0x6e, 0x00, 0x03,
+ 0xff, 0x13 };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_LENGTH, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer2, ARRAY_SIZE(buffer2))));
+
+ EXPECT_CALL(streams, getStreamProxy(2, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ // create the row type
+ std::unique_ptr<Type> rowType = createStructType();
+ rowType->addStructField(createListType(createPrimitiveType(LONG)), "col0");
+ rowType->assignIds(0);
+
+ std::unique_ptr<ColumnReader> reader = buildReader(*rowType, streams);
+
+ StructVectorBatch batch(1, *getDefaultPool());
+ ListVectorBatch *lists = new ListVectorBatch(1, *getDefaultPool());
+ batch.fields.push_back(lists);
+
+ reader->next(batch, 1, 0);
+ ASSERT_EQ(1, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(1, lists->numElements);
+ ASSERT_EQ(true, !lists->hasNulls);
+ EXPECT_EQ(0, lists->offsets[0]);
+ EXPECT_EQ(1, lists->offsets[1]);
+
+ reader->skip(13);
+ reader->next(batch, 1, 0);
+ ASSERT_EQ(1, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(1, lists->numElements);
+ ASSERT_EQ(true, !lists->hasNulls);
+ EXPECT_EQ(0, lists->offsets[0]);
+ EXPECT_EQ(1, lists->offsets[1]);
+
+ reader->skip(2031);
+ reader->next(batch, 2, 0);
+ ASSERT_EQ(2, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(2, lists->numElements);
+ ASSERT_EQ(true, lists->hasNulls);
+ EXPECT_EQ(0, lists->offsets[0]);
+ EXPECT_EQ(19, lists->offsets[1]);
+ EXPECT_EQ(19, lists->offsets[2]);
+}
+
+TEST(TestColumnReader, testMap) {
+ MockStripeStreams streams;
+
+ // set getSelectedColumns()
+ std::vector<bool> selectedColumns(4, true);
+ EXPECT_CALL(streams, getSelectedColumns())
+ .WillRepeatedly(testing::Return(selectedColumns));
+
+ // set getEncoding
+ proto::ColumnEncoding directEncoding;
+ directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+ EXPECT_CALL(streams, getEncoding(testing::_))
+ .WillRepeatedly(testing::Return(directEncoding));
+
+ // set getStream
+ EXPECT_CALL(streams, getStreamProxy(testing::_,
+ proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ // [2 for x in range(600)]
+ const unsigned char buffer1[] = { 0x7f, 0x00, 0x02,
+ 0x7f, 0x00, 0x02,
+ 0x7f, 0x00, 0x02,
+ 0x7f, 0x00, 0x02,
+ 0x4d, 0x00, 0x02 };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_LENGTH, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer1, ARRAY_SIZE(buffer1))));
+
+ // range(1200)
+ const unsigned char buffer2[] = { 0x7f, 0x01, 0x00,
+ 0x7f, 0x01, 0x84, 0x02,
+ 0x7f, 0x01, 0x88, 0x04,
+ 0x7f, 0x01, 0x8c, 0x06,
+ 0x7f, 0x01, 0x90, 0x08,
+ 0x7f, 0x01, 0x94, 0x0a,
+ 0x7f, 0x01, 0x98, 0x0c,
+ 0x7f, 0x01, 0x9c, 0x0e,
+ 0x7f, 0x01, 0xa0, 0x10,
+ 0x1b, 0x01, 0xa4, 0x12 };
+ EXPECT_CALL(streams, getStreamProxy(2, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer2, ARRAY_SIZE(buffer2))));
+
+ // range(8, 1208)
+ const unsigned char buffer3[] = { 0x7f, 0x01, 0x10,
+ 0x7f, 0x01, 0x94, 0x02,
+ 0x7f, 0x01, 0x98, 0x04,
+ 0x7f, 0x01, 0x9c, 0x06,
+ 0x7f, 0x01, 0xa0, 0x08,
+ 0x7f, 0x01, 0xa4, 0x0a,
+ 0x7f, 0x01, 0xa8, 0x0c,
+ 0x7f, 0x01, 0xac, 0x0e,
+ 0x7f, 0x01, 0xb0, 0x10,
+ 0x1b, 0x01, 0xb4, 0x12 };
+ EXPECT_CALL(streams, getStreamProxy(3, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer3, ARRAY_SIZE(buffer3))));
+
+ // create the row type
+ std::unique_ptr<Type> rowType = createStructType();
+ rowType->addStructField(createMapType(createPrimitiveType(LONG),
+ createPrimitiveType(LONG)),
+ "col0");
+ rowType->assignIds(0);
+
+ std::unique_ptr<ColumnReader> reader = buildReader(*rowType, streams);
+
+ StructVectorBatch batch(512, *getDefaultPool());
+ MapVectorBatch *maps = new MapVectorBatch(512, *getDefaultPool());
+ LongVectorBatch *keys = new LongVectorBatch(512, *getDefaultPool());
+ LongVectorBatch *elements = new LongVectorBatch(512, *getDefaultPool());
+ batch.fields.push_back(maps);
+ maps->keys = std::unique_ptr < ColumnVectorBatch > (keys);
+ maps->elements = std::unique_ptr < ColumnVectorBatch > (elements);
+ reader->next(batch, 512, 0);
+ ASSERT_EQ(512, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(512, maps->numElements);
+ ASSERT_EQ(true, !maps->hasNulls);
+ ASSERT_EQ(1024, keys->numElements);
+ ASSERT_EQ(true, !keys->hasNulls);
+ ASSERT_EQ(1024, elements->numElements);
+ ASSERT_EQ(true, !elements->hasNulls);
+ for (size_t i = 0; i <= batch.numElements; ++i) {
+ EXPECT_EQ(2 * i, maps->offsets[i]);
+ }
+ for (size_t i = 0; i < keys->numElements; ++i) {
+ EXPECT_EQ(i, keys->data[i]);
+ EXPECT_EQ(i + 8, elements->data[i]);
+ }
+}
+
+TEST(TestColumnReader, testMapWithNulls) {
+ MockStripeStreams streams;
+
+ // set getSelectedColumns()
+ std::vector<bool> selectedColumns(4, true);
+ EXPECT_CALL(streams, getSelectedColumns())
+ .WillRepeatedly(testing::Return(selectedColumns));
+
+ // set getEncoding
+ proto::ColumnEncoding directEncoding;
+ directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+ EXPECT_CALL(streams, getEncoding(testing::_))
+ .WillRepeatedly(testing::Return(directEncoding));
+
+ // set getStream
+ EXPECT_CALL(streams, getStreamProxy(0, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ // [0xaa for x in range(2048/8)]
+ const unsigned char buffer1[] = { 0x7f, 0xaa, 0x7b, 0xaa };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer1, ARRAY_SIZE(buffer1))));
+
+ EXPECT_CALL(streams, getStreamProxy(2, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ // [0x55 for x in range(2048/8)]
+ const unsigned char buffer2[] = { 0x7f, 0x55, 0x7b, 0x55 };
+ EXPECT_CALL(streams, getStreamProxy(3, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer2, ARRAY_SIZE(buffer2))));
+
+ // [1 for x in range(260)] +
+ // [4 for x in range(260)] +
+ // [0 for x in range(260)] +
+ // [3 for x in range(243)] +
+ // [19]
+ const unsigned char buffer3[] = { 0x7f, 0x00, 0x01,
+ 0x7f, 0x00, 0x01,
+ 0x7f, 0x00, 0x04,
+ 0x7f, 0x00, 0x04,
+ 0x7f, 0x00, 0x00,
+ 0x7f, 0x00, 0x00,
+ 0x7f, 0x00, 0x03,
+ 0x6e, 0x00, 0x03,
+ 0xff, 0x13 };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_LENGTH, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer3, ARRAY_SIZE(buffer3))));
+
+ // range(2048)
+ const unsigned char buffer4[] = { 0x7f, 0x01, 0x00,
+ 0x7f, 0x01, 0x84, 0x02,
+ 0x7f, 0x01, 0x88, 0x04,
+ 0x7f, 0x01, 0x8c, 0x06,
+ 0x7f, 0x01, 0x90, 0x08,
+ 0x7f, 0x01, 0x94, 0x0a,
+ 0x7f, 0x01, 0x98, 0x0c,
+ 0x7f, 0x01, 0x9c, 0x0e,
+ 0x7f, 0x01, 0xa0, 0x10,
+ 0x7f, 0x01, 0xa4, 0x12,
+ 0x7f, 0x01, 0xa8, 0x14,
+ 0x7f, 0x01, 0xac, 0x16,
+ 0x7f, 0x01, 0xb0, 0x18,
+ 0x7f, 0x01, 0xb4, 0x1a,
+ 0x7f, 0x01, 0xb8, 0x1c,
+ 0x5f, 0x01, 0xbc, 0x1e };
+ EXPECT_CALL(streams, getStreamProxy(2, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer4, ARRAY_SIZE(buffer4))));
+
+ // range(8, 1032)
+ const unsigned char buffer5[] = { 0x7f, 0x01, 0x10,
+ 0x7f, 0x01, 0x94, 0x02,
+ 0x7f, 0x01, 0x98, 0x04,
+ 0x7f, 0x01, 0x9c, 0x06,
+ 0x7f, 0x01, 0xa0, 0x08,
+ 0x7f, 0x01, 0xa4, 0x0a,
+ 0x7f, 0x01, 0xa8, 0x0c,
+ 0x6f, 0x01, 0xac, 0x0e };
+ EXPECT_CALL(streams, getStreamProxy(3, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer5, ARRAY_SIZE(buffer5))));
+
+ // create the row type
+ std::unique_ptr<Type> rowType = createStructType();
+ rowType->addStructField(createMapType(createPrimitiveType(LONG),
+ createPrimitiveType(LONG)),
+ "col0");
+ rowType->assignIds(0);
+
+ std::unique_ptr<ColumnReader> reader = buildReader(*rowType, streams);
+
+ StructVectorBatch batch(512, *getDefaultPool());
+ MapVectorBatch *maps = new MapVectorBatch(512, *getDefaultPool());
+ LongVectorBatch *keys = new LongVectorBatch(512, *getDefaultPool());
+ LongVectorBatch *elements = new LongVectorBatch(512, *getDefaultPool());
+ batch.fields.push_back(maps);
+ maps->keys = std::unique_ptr < ColumnVectorBatch > (keys);
+ maps->elements = std::unique_ptr < ColumnVectorBatch > (elements);
+ reader->next(batch, 512, 0);
+ ASSERT_EQ(512, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(512, maps->numElements);
+ ASSERT_EQ(true, maps->hasNulls);
+ ASSERT_EQ(256, keys->numElements);
+ ASSERT_EQ(true, !keys->hasNulls);
+ ASSERT_EQ(256, elements->numElements);
+ ASSERT_EQ(true, elements->hasNulls);
+ for (size_t i = 0; i < batch.numElements; ++i) {
+ EXPECT_EQ(i % 2 == 0, maps->notNull[i]) << "Wrong value at " << i;
+ EXPECT_EQ((i + 1) / 2, maps->offsets[i]) << "Wrong value at " << i;
+ }
+ EXPECT_EQ(256, maps->offsets[512]);
+ for (size_t i = 0; i < keys->numElements; ++i) {
+ EXPECT_EQ(i, keys->data[i]);
+ EXPECT_EQ(i & 1, elements->notNull[i]);
+ if (elements->notNull[i]) {
+ EXPECT_EQ(i / 2 + 8, elements->data[i]);
+ }
+ }
+
+ reader->next(batch, 512, 0);
+ ASSERT_EQ(512, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(512, maps->numElements);
+ ASSERT_EQ(true, maps->hasNulls);
+ ASSERT_EQ(1012, keys->numElements);
+ ASSERT_EQ(true, !keys->hasNulls);
+ ASSERT_EQ(1012, elements->numElements);
+ ASSERT_EQ(true, elements->hasNulls);
+ for (size_t i = 0; i < batch.numElements; ++i) {
+ EXPECT_EQ(i % 2 == 0, maps->notNull[i]) << "Wrong value at " << i;
+ if (i < 8) {
+ EXPECT_EQ((i + 1) / 2, maps->offsets[i])
+ << "Wrong value at " << i;
+ } else {
+ EXPECT_EQ(4 * ((i + 1) / 2) - 12, maps->offsets[i])
+ << "Wrong value at " << i;
+ }
+ }
+ EXPECT_EQ(1012, maps->offsets[512]);
+ for (size_t i = 0; i < keys->numElements; ++i) {
+ EXPECT_EQ(256 + i, keys->data[i]);
+ EXPECT_EQ(i & 1, elements->notNull[i]);
+ if (elements->notNull[i]) {
+ EXPECT_EQ(128 + 8 + i / 2, elements->data[i]);
+ }
+ }
+
+ reader->next(batch, 512, 0);
+ ASSERT_EQ(512, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(512, maps->numElements);
+ ASSERT_EQ(true, maps->hasNulls);
+ ASSERT_EQ(32, keys->numElements);
+ ASSERT_EQ(true, !keys->hasNulls);
+ ASSERT_EQ(32, elements->numElements);
+ ASSERT_EQ(true, elements->hasNulls);
+ for (size_t i = 0; i < batch.numElements; ++i) {
+ EXPECT_EQ(i % 2 == 0, maps->notNull[i]) << "Wrong value at " << i;
+ if (i < 16) {
+ EXPECT_EQ(4 * ((i + 1) / 2), maps->offsets[i])
+ << "Wrong value at " << i;
+ } else {
+ EXPECT_EQ(32, maps->offsets[i]) << "Wrong value at " << i;
+ }
+ }
+ EXPECT_EQ(32, maps->offsets[512]);
+ for (size_t i = 0; i < keys->numElements; ++i) {
+ EXPECT_EQ(1268 + i, keys->data[i]);
+ EXPECT_EQ(i & 1, elements->notNull[i]);
+ if (elements->notNull[i]) {
+ EXPECT_EQ(634 + 8 + i / 2, elements->data[i]);
+ }
+ }
+
+ reader->next(batch, 512, 0);
+ ASSERT_EQ(512, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(512, maps->numElements);
+ ASSERT_EQ(true, maps->hasNulls);
+ ASSERT_EQ(748, keys->numElements);
+ ASSERT_EQ(true, !keys->hasNulls);
+ ASSERT_EQ(748, elements->numElements);
+ ASSERT_EQ(true, elements->hasNulls);
+ for (size_t i = 0; i < batch.numElements; ++i) {
+ EXPECT_EQ(i % 2 == 0, maps->notNull[i]) << "Wrong value at " << i;
+ if (i < 24) {
+ EXPECT_EQ(0, maps->offsets[i]) << "Wrong value at " << i;
+ } else if (i < 510) {
+ EXPECT_EQ(3 * ((i - 23) / 2), maps->offsets[i])
+ << "Wrong value at " << i;
+ } else if (i < 511) {
+ EXPECT_EQ(729, maps->offsets[i]) << "Wrong value at " << i;
+ } else {
+ EXPECT_EQ(748, maps->offsets[i]) << "Wrong value at " << i;
+ }
+ }
+ EXPECT_EQ(748, maps->offsets[512]);
+ for (size_t i = 0; i < keys->numElements; ++i) {
+ EXPECT_EQ(1300 + i, keys->data[i]);
+ EXPECT_EQ(i & 1, elements->notNull[i]);
+ if (elements->notNull[i]) {
+ EXPECT_EQ(650 + 8 + i / 2, elements->data[i]);
+ }
+ }
+}
+
+TEST(TestColumnReader, testMapSkipWithNulls) {
+ MockStripeStreams streams;
+
+ // set getSelectedColumns()
+ std::vector<bool> selectedColumns(4, true);
+ EXPECT_CALL(streams, getSelectedColumns())
+ .WillRepeatedly(testing::Return(selectedColumns));
+
+ // set getEncoding
+ proto::ColumnEncoding directEncoding;
+ directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+ EXPECT_CALL(streams, getEncoding(testing::_))
+ .WillRepeatedly(testing::Return(directEncoding));
+
+ // set getStream
+ EXPECT_CALL(streams, getStreamProxy(testing::_,proto::Stream_Kind_PRESENT,
+ true))
+ .WillRepeatedly(testing::Return(nullptr));
+
+ // [0xaa for x in range(2048/8)]
+ const unsigned char buffer1[] = { 0x7f, 0xaa, 0x7b, 0xaa };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_PRESENT, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer1, ARRAY_SIZE(buffer1))));
+
+ // [1 for x in range(260)] +
+ // [4 for x in range(260)] +
+ // [0 for x in range(260)] +
+ // [3 for x in range(243)] +
+ // [19]
+ const unsigned char buffer2[] = { 0x7f, 0x00, 0x01,
+ 0x7f, 0x00, 0x01,
+ 0x7f, 0x00, 0x04,
+ 0x7f, 0x00, 0x04,
+ 0x7f, 0x00, 0x00,
+ 0x7f, 0x00, 0x00,
+ 0x7f, 0x00, 0x03,
+ 0x6e, 0x00, 0x03,
+ 0xff, 0x13 };
+ EXPECT_CALL(streams, getStreamProxy(1, proto::Stream_Kind_LENGTH, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer2, ARRAY_SIZE(buffer2))));
+
+ // range(2048)
+ const unsigned char buffer3[] = { 0x7f, 0x01, 0x00,
+ 0x7f, 0x01, 0x84, 0x02,
+ 0x7f, 0x01, 0x88, 0x04,
+ 0x7f, 0x01, 0x8c, 0x06,
+ 0x7f, 0x01, 0x90, 0x08,
+ 0x7f, 0x01, 0x94, 0x0a,
+ 0x7f, 0x01, 0x98, 0x0c,
+ 0x7f, 0x01, 0x9c, 0x0e,
+ 0x7f, 0x01, 0xa0, 0x10,
+ 0x7f, 0x01, 0xa4, 0x12,
+ 0x7f, 0x01, 0xa8, 0x14,
+ 0x7f, 0x01, 0xac, 0x16,
+ 0x7f, 0x01, 0xb0, 0x18,
+ 0x7f, 0x01, 0xb4, 0x1a,
+ 0x7f, 0x01, 0xb8, 0x1c,
+ 0x5f, 0x01, 0xbc, 0x1e };
+ EXPECT_CALL(streams, getStreamProxy(2, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer3, ARRAY_SIZE(buffer3))));
+
+ // range(8, 2056)
+ const unsigned char buffer4[] = { 0x7f, 0x01, 0x10,
+ 0x7f, 0x01, 0x94, 0x02,
+ 0x7f, 0x01, 0x98, 0x04,
+ 0x7f, 0x01, 0x9c, 0x06,
+ 0x7f, 0x01, 0xa0, 0x08,
+ 0x7f, 0x01, 0xa4, 0x0a,
+ 0x7f, 0x01, 0xa8, 0x0c,
+ 0x7f, 0x01, 0xac, 0x0e,
+ 0x7f, 0x01, 0xb0, 0x10,
+ 0x7f, 0x01, 0xb4, 0x12,
+ 0x7f, 0x01, 0xb8, 0x14,
+ 0x7f, 0x01, 0xbc, 0x16,
+ 0x7f, 0x01, 0xc0, 0x18,
+ 0x7f, 0x01, 0xc4, 0x1a,
+ 0x7f, 0x01, 0xc8, 0x1c,
+ 0x5f, 0x01, 0xcc, 0x1e };
+ EXPECT_CALL(streams, getStreamProxy(3, proto::Stream_Kind_DATA, true))
+ .WillRepeatedly(testing::Return(new SeekableArrayInputStream
+ (buffer4, ARRAY_SIZE(buffer4))));
+
+ // create the row type
+ std::unique_ptr<Type> rowType = createStructType();
+ rowType->addStructField(createMapType(createPrimitiveType(LONG),
+ createPrimitiveType(LONG)),
+ "col0");
+ rowType->assignIds(0);
+
+ std::unique_ptr<ColumnReader> reader = buildReader(*rowType, streams);
+
+ StructVectorBatch batch(1, *getDefaultPool());
+ MapVectorBatch *maps = new MapVectorBatch(1, *getDefaultPool());
+ LongVectorBatch *keys = new LongVectorBatch(1, *getDefaultPool());
+ LongVectorBatch *elements = new LongVectorBatch(1, *getDefaultPool());
+ batch.fields.push_back(maps);
+ maps->keys = std::unique_ptr < ColumnVectorBatch > (keys);
+ maps->elements = std::unique_ptr < ColumnVectorBatch > (elements);
+
+ reader->next(batch, 1, 0);
+ ASSERT_EQ(1, batch.numElements);
+ ASSERT_EQ(true, !batch.hasNulls);
+ ASSERT_EQ(1, maps->numElements);
+ ASSERT_EQ(true, !maps->hasNulls);
+ ASSERT_EQ(1, keys->numElements);
+ ASSERT_EQ(true, !keys->hasNulls);
+ ASSERT_EQ(1, elements->numElements);
+ ASSERT_EQ(true, !elements->hasNulls);
+ EXPECT_EQ(0, maps->offsets[0]);
+ EXPECT_EQ(1, maps->offsets[1]);
+ EXPECT_EQ(0, keys->data[0]);
+ EXPECT_EQ(8, elements->data[0]);
+
+ reader->skip(13);
+ rea
<TRUNCATED>