-
Notifications
You must be signed in to change notification settings - Fork 86
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
c22396f
commit c905762
Showing
99 changed files
with
21,264 additions
and
66 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1 +1 @@ | ||
create node table tableOfTypes (id INT64, int64Column INT64, doubleColumn DOUBLE, booleanColumn BOOLEAN, dateColumn DATE, timestampColumn TIMESTAMP, stringColumn STRING, listOfInt64 INT64[], listOfString STRING[], listOfListOfInt64 INT64[][], fixedSizeList INT64[3], listOfFixedSizeList INT64[3][], structColumn STRUCT(ID int64, name STRING), PRIMARY KEY (id)); | ||
create node table tableOfTypes (id INT64, int64Column INT64, doubleColumn DOUBLE, booleanColumn BOOLEAN, dateColumn DATE, stringColumn STRING, listOfInt64 INT64[], listOfString STRING[], listOfListOfInt64 INT64[][], structColumn STRUCT(ID int64, name STRING), PRIMARY KEY (id)); |
Binary file not shown.
Binary file not shown.
Binary file not shown.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
File renamed without changes.
45 changes: 45 additions & 0 deletions
45
src/include/processor/operator/persistent/reader/parquet/boolean_column_reader.h
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,45 @@ | ||
#pragma once | ||
|
||
#include "column_reader.h" | ||
#include "templated_column_reader.h" | ||
|
||
namespace kuzu { | ||
namespace processor { | ||
|
||
struct BooleanParquetValueConversion; | ||
|
||
class BooleanColumnReader : public TemplatedColumnReader<bool, BooleanParquetValueConversion> { | ||
public: | ||
static constexpr const common::PhysicalTypeID TYPE = common::PhysicalTypeID::BOOL; | ||
|
||
public: | ||
BooleanColumnReader(ParquetReader& reader, std::unique_ptr<common::LogicalType> type, | ||
const kuzu_parquet::format::SchemaElement& schema, uint64_t schemaIdx, uint64_t maxDefine, | ||
uint64_t maxRepeat) | ||
: TemplatedColumnReader<bool, BooleanParquetValueConversion>( | ||
reader, std::move(type), schema, schemaIdx, maxDefine, maxRepeat), | ||
bytePos(0){}; | ||
|
||
uint8_t bytePos; | ||
|
||
void initializeRead(uint64_t rowGroupIdx, | ||
const std::vector<kuzu_parquet::format::ColumnChunk>& columns, | ||
kuzu_apache::thrift::protocol::TProtocol& protocol) override; | ||
|
||
inline void resetPage() override { bytePos = 0; } | ||
}; | ||
|
||
struct BooleanParquetValueConversion { | ||
static bool dictRead(ByteBuffer& dict, uint32_t& offset, ColumnReader& reader) { | ||
throw common::CopyException{"Dicts for booleans make no sense"}; | ||
} | ||
|
||
static bool plainRead(ByteBuffer& plainData, ColumnReader& reader); | ||
|
||
static inline void plainSkip(ByteBuffer& plainData, ColumnReader& reader) { | ||
plainRead(plainData, reader); | ||
} | ||
}; | ||
|
||
} // namespace processor | ||
} // namespace kuzu |
40 changes: 40 additions & 0 deletions
40
src/include/processor/operator/persistent/reader/parquet/callback_column_reader.h
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,40 @@ | ||
#pragma once | ||
|
||
#include "column_reader.h" | ||
#include "parquet_reader.h" | ||
#include "templated_column_reader.h" | ||
|
||
namespace kuzu { | ||
namespace processor { | ||
|
||
template<class PARQUET_PHYSICAL_TYPE, class KU_PHYSICAL_TYPE, | ||
KU_PHYSICAL_TYPE (*FUNC)(const PARQUET_PHYSICAL_TYPE& input)> | ||
class CallbackColumnReader | ||
: public TemplatedColumnReader<KU_PHYSICAL_TYPE, | ||
CallbackParquetValueConversion<PARQUET_PHYSICAL_TYPE, KU_PHYSICAL_TYPE, FUNC>> { | ||
using BaseType = TemplatedColumnReader<KU_PHYSICAL_TYPE, | ||
CallbackParquetValueConversion<PARQUET_PHYSICAL_TYPE, KU_PHYSICAL_TYPE, FUNC>>; | ||
|
||
public: | ||
static constexpr const common::PhysicalTypeID TYPE = common::PhysicalTypeID::ANY; | ||
|
||
public: | ||
CallbackColumnReader(ParquetReader& reader, std::unique_ptr<common::LogicalType> type_p, | ||
const kuzu_parquet::format::SchemaElement& schema_p, uint64_t file_idx_p, | ||
uint64_t max_define_p, uint64_t max_repeat_p) | ||
: TemplatedColumnReader<KU_PHYSICAL_TYPE, | ||
CallbackParquetValueConversion<PARQUET_PHYSICAL_TYPE, KU_PHYSICAL_TYPE, FUNC>>( | ||
reader, std::move(type_p), schema_p, file_idx_p, max_define_p, max_repeat_p) {} | ||
|
||
protected: | ||
void dictionary(std::shared_ptr<ResizeableBuffer> dictionaryData, uint64_t numEntries) { | ||
BaseType::allocateDict(numEntries * sizeof(KU_PHYSICAL_TYPE)); | ||
auto dictPtr = (KU_PHYSICAL_TYPE*)this->dict->ptr; | ||
for (auto i = 0u; i < numEntries; i++) { | ||
dictPtr[i] = FUNC(dictionaryData->read<PARQUET_PHYSICAL_TYPE>()); | ||
} | ||
} | ||
}; | ||
|
||
} // namespace processor | ||
} // namespace kuzu |
119 changes: 119 additions & 0 deletions
119
src/include/processor/operator/persistent/reader/parquet/column_reader.h
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,119 @@ | ||
#pragma once | ||
|
||
#include <bitset> | ||
|
||
#include "common/constants.h" | ||
#include "common/types/types.h" | ||
#include "common/vector/value_vector.h" | ||
#include "parquet/parquet_types.h" | ||
#include "parquet_dbp_decoder.h" | ||
#include "parquet_rle_bp_decoder.h" | ||
#include "resizable_buffer.h" | ||
#include "thrift_tools.h" | ||
|
||
namespace kuzu { | ||
namespace processor { | ||
class ParquetReader; | ||
|
||
typedef std::bitset<common::DEFAULT_VECTOR_CAPACITY> parquet_filter_t; | ||
|
||
class ColumnReader { | ||
public: | ||
ColumnReader(ParquetReader& reader, std::unique_ptr<common::LogicalType> type, | ||
const kuzu_parquet::format::SchemaElement& schema, uint64_t fileIdx, uint64_t maxDefinition, | ||
uint64_t maxRepeat); | ||
virtual ~ColumnReader() = default; | ||
inline common::LogicalType* getDataType() const { return type.get(); } | ||
inline bool hasDefines() { return maxDefine > 0; } | ||
inline bool hasRepeats() { return maxRepeat > 0; } | ||
virtual inline void skip(uint64_t numValues) { pendingSkips += numValues; } | ||
virtual inline void dictionary(std::shared_ptr<ResizeableBuffer> data, uint64_t num_entries) { | ||
throw common::NotImplementedException{"Dictionary"}; | ||
} | ||
virtual inline void offsets(uint32_t* offsets, uint8_t* defines, uint64_t numValues, | ||
parquet_filter_t& filter, uint64_t resultOffset, common::ValueVector* result) { | ||
throw common::NotImplementedException{"ColumnReader::offsets"}; | ||
} | ||
virtual inline void plain(std::shared_ptr<ByteBuffer> plainData, uint8_t* defines, | ||
uint64_t numValues, parquet_filter_t& filter, uint64_t resultOffset, | ||
common::ValueVector* result) { | ||
throw common::NotImplementedException{"ColumnReader::plain"}; | ||
} | ||
virtual inline void resetPage() {} | ||
virtual inline uint64_t getGroupRowsAvailable() { return groupRowsAvailable; } | ||
virtual void initializeRead(uint64_t rowGroupIdx, | ||
const std::vector<kuzu_parquet::format::ColumnChunk>& columns, | ||
kuzu_apache::thrift::protocol::TProtocol& protocol); | ||
virtual uint64_t getTotalCompressedSize(); | ||
virtual void registerPrefetch(ThriftFileTransport& transport, bool allowMerge); | ||
virtual uint64_t fileOffset() const; | ||
virtual void applyPendingSkips(uint64_t numValues); | ||
virtual uint64_t read(uint64_t numValues, parquet_filter_t& filter, uint8_t* defineOut, | ||
uint8_t* repeatOut, common::ValueVector* resultOut); | ||
static std::unique_ptr<ColumnReader> createReader(ParquetReader& reader, | ||
std::unique_ptr<common::LogicalType> type, | ||
const kuzu_parquet::format::SchemaElement& schema, uint64_t fileIdx, uint64_t maxDefine, | ||
uint64_t maxRepeat); | ||
void prepareRead(parquet_filter_t& filter); | ||
void allocateBlock(uint64_t size); | ||
void allocateCompressed(uint64_t size); | ||
void decompressInternal(kuzu_parquet::format::CompressionCodec::type codec, const uint8_t* src, | ||
uint64_t srcSize, uint8_t* dst, uint64_t dstSize); | ||
void preparePageV2(kuzu_parquet::format::PageHeader& pageHdr); | ||
void preparePage(kuzu_parquet::format::PageHeader& pageHdr); | ||
void prepareDataPage(kuzu_parquet::format::PageHeader& pageHdr); | ||
template<class VALUE_TYPE, class CONVERSION> | ||
void plainTemplated(std::shared_ptr<ByteBuffer> plainData, uint8_t* defines, uint64_t numValues, | ||
parquet_filter_t& filter, uint64_t resultOffset, common::ValueVector* result) { | ||
for (auto i = 0u; i < numValues; i++) { | ||
if (hasDefines() && defines[i + resultOffset] != maxDefine) { | ||
result->setNull(i + resultOffset, true); | ||
continue; | ||
} | ||
if (filter[i + resultOffset]) { | ||
VALUE_TYPE val = CONVERSION::plainRead(*plainData, *this); | ||
result->setValue(i + resultOffset, val); | ||
} else { // there is still some data there that we have to skip over | ||
CONVERSION::plainSkip(*plainData, *this); | ||
} | ||
} | ||
} | ||
|
||
protected: | ||
const kuzu_parquet::format::SchemaElement& schema; | ||
|
||
uint64_t fileIdx; | ||
uint64_t maxDefine; | ||
uint64_t maxRepeat; | ||
|
||
ParquetReader& reader; | ||
std::unique_ptr<common::LogicalType> type; | ||
|
||
uint64_t pendingSkips = 0; | ||
|
||
const kuzu_parquet::format::ColumnChunk* chunk = nullptr; | ||
|
||
kuzu_apache::thrift::protocol::TProtocol* protocol; | ||
uint64_t pageRowsAvailable; | ||
uint64_t groupRowsAvailable; | ||
uint64_t chunkReadOffset; | ||
|
||
std::shared_ptr<ResizeableBuffer> block; | ||
|
||
ResizeableBuffer compressedBuffer; | ||
ResizeableBuffer offsetBuffer; | ||
|
||
std::unique_ptr<RleBpDecoder> dictDecoder; | ||
std::unique_ptr<RleBpDecoder> defineDecoder; | ||
std::unique_ptr<RleBpDecoder> repeatedDecoder; | ||
std::unique_ptr<DbpDecoder> dbpDecoder; | ||
std::unique_ptr<RleBpDecoder> rleDecoder; | ||
|
||
// dummies for Skip() | ||
parquet_filter_t noneFilter; | ||
ResizeableBuffer dummyDefine; | ||
ResizeableBuffer dummyRepeat; | ||
}; | ||
|
||
} // namespace processor | ||
} // namespace kuzu |
65 changes: 65 additions & 0 deletions
65
src/include/processor/operator/persistent/reader/parquet/decode_utils.h
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,65 @@ | ||
#pragma once | ||
|
||
#include "common/exception/copy.h" | ||
#include "common/string_utils.h" | ||
#include "resizable_buffer.h" | ||
|
||
namespace kuzu { | ||
namespace processor { | ||
class ParquetDecodeUtils { | ||
|
||
public: | ||
template<class T> | ||
static T ZigzagToInt(const T n) { | ||
return (n >> 1) ^ -(n & 1); | ||
} | ||
|
||
static const uint64_t BITPACK_MASKS[]; | ||
static const uint64_t BITPACK_MASKS_SIZE; | ||
static const uint8_t BITPACK_DLEN; | ||
|
||
template<typename T> | ||
static uint32_t BitUnpack( | ||
ByteBuffer& buffer, uint8_t& bitpack_pos, T* dest, uint32_t count, uint8_t width) { | ||
if (width >= ParquetDecodeUtils::BITPACK_MASKS_SIZE) { | ||
throw common::CopyException(common::StringUtils::string_format( | ||
"The width ({}) of the bitpacked data exceeds the supported max width ({}), " | ||
"the file might be corrupted.", | ||
width, ParquetDecodeUtils::BITPACK_MASKS_SIZE)); | ||
} | ||
auto mask = BITPACK_MASKS[width]; | ||
|
||
for (uint32_t i = 0; i < count; i++) { | ||
T val = (buffer.get<uint8_t>() >> bitpack_pos) & mask; | ||
bitpack_pos += width; | ||
while (bitpack_pos > BITPACK_DLEN) { | ||
buffer.inc(1); | ||
val |= (T(buffer.get<uint8_t>()) << T(BITPACK_DLEN - (bitpack_pos - width))) & mask; | ||
bitpack_pos -= BITPACK_DLEN; | ||
} | ||
dest[i] = val; | ||
} | ||
return count; | ||
} | ||
|
||
template<class T> | ||
static T VarintDecode(ByteBuffer& buf) { | ||
T result = 0; | ||
uint8_t shift = 0; | ||
while (true) { | ||
auto byte = buf.read<uint8_t>(); | ||
result |= T(byte & 127) << shift; | ||
if ((byte & 128) == 0) { | ||
break; | ||
} | ||
shift += 7; | ||
if (shift > sizeof(T) * 8) { | ||
throw std::runtime_error("Varint-decoding found too large number"); | ||
} | ||
} | ||
return result; | ||
} | ||
}; | ||
|
||
} // namespace processor | ||
} // namespace kuzu |
Oops, something went wrong.