setop
/
sqlite-parquet-vtable
mirror of https://github.com/cldellow/sqlite-parquet-vtable.git
1
0
Fork 0
Code Releases Activity

Refactored the types to match the most recent Apache Arrow version

This commit is contained in:
Mikko Harju 2019-11-13 15:17:29 +02:00
parent e3a2837599
commit 96405b77dc
4 changed files with 60 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
parquet/parquet.cc
View File

@ -29,9 +29,9 @@ SQLITE_EXTENSION_INIT1
/* Forward references to the various virtual table methods implemented /* Forward references to the various virtual table methods implemented
* in this file. */ * in this file. */
static int parquetCreate(sqlite3*, void*, int, const char*const*, static int parquetCreate(sqlite3*, void*, int, const char*const*,
sqlite3_vtab**,char**); sqlite3_vtab**,char**);
static int parquetConnect(sqlite3*, void*, int, const char*const*, static int parquetConnect(sqlite3*, void*, int, const char*const*,
sqlite3_vtab**,char**); sqlite3_vtab**,char**);
static int parquetBestIndex(sqlite3_vtab*,sqlite3_index_info*); static int parquetBestIndex(sqlite3_vtab*,sqlite3_index_info*);
static int parquetDisconnect(sqlite3_vtab*); static int parquetDisconnect(sqlite3_vtab*);
@ -290,7 +290,7 @@ static int parquetColumn(
case parquet::Type::BYTE_ARRAY: case parquet::Type::BYTE_ARRAY:
{ {
parquet::ByteArray* rv = cursor->getByteArray(col); parquet::ByteArray* rv = cursor->getByteArray(col);
if(cursor->getLogicalType(col) == parquet::LogicalType::UTF8) { if(cursor->getLogicalType(col) == parquet::ConvertedType::UTF8) {
sqlite3_result_text(ctx, (const char*)rv->ptr, rv->len, SQLITE_TRANSIENT); sqlite3_result_text(ctx, (const char*)rv->ptr, rv->len, SQLITE_TRANSIENT);
} else { } else {
sqlite3_result_blob(ctx, (void*)rv->ptr, rv->len, SQLITE_TRANSIENT); sqlite3_result_blob(ctx, (void*)rv->ptr, rv->len, SQLITE_TRANSIENT);
@ -336,7 +336,7 @@ static int parquetColumn(
*/ */
static int parquetRowid(sqlite3_vtab_cursor *cur, sqlite_int64 *pRowid){ static int parquetRowid(sqlite3_vtab_cursor *cur, sqlite_int64 *pRowid){
ParquetCursor *cursor = ((sqlite3_vtab_cursor_parquet*)cur)->cursor; ParquetCursor *cursor = ((sqlite3_vtab_cursor_parquet*)cur)->cursor;
*pRowid = cursor->getRowId(); *pRowid = cursor->getRowId();
return SQLITE_OK; return SQLITE_OK;
} }
@ -741,15 +741,15 @@ static sqlite3_module ParquetModule = {
0, /* xRename */ 0, /* xRename */
}; };
/* /*
* This routine is called when the extension is loaded. The new * This routine is called when the extension is loaded. The new
* Parquet virtual table module is registered with the calling database * Parquet virtual table module is registered with the calling database
* connection. * connection.
*/ */
extern "C" { extern "C" {
int sqlite3_parquet_init( int sqlite3_parquet_init(
sqlite3 *db, sqlite3 *db,
char **pzErrMsg, char **pzErrMsg,
const sqlite3_api_routines *pApi const sqlite3_api_routines *pApi
){ ){
int rc; int rc;

56
parquet/parquet_cursor.cc
View File

@ -31,7 +31,7 @@ bool ParquetCursor::currentRowGroupSatisfiesRowIdFilter(Constraint& constraint)
} }
} }
bool ParquetCursor::currentRowGroupSatisfiesBlobFilter(Constraint& constraint, std::shared_ptr<parquet::RowGroupStatistics> _stats) { bool ParquetCursor::currentRowGroupSatisfiesBlobFilter(Constraint& constraint, std::shared_ptr<parquet::Statistics> _stats) {
if(!_stats->HasMinMax()) { if(!_stats->HasMinMax()) {
return true; return true;
} }
@ -48,8 +48,8 @@ bool ParquetCursor::currentRowGroupSatisfiesBlobFilter(Constraint& constraint, s
parquet::Type::type pqType = types[constraint.column]; parquet::Type::type pqType = types[constraint.column];
if(pqType == parquet::Type::BYTE_ARRAY) { if(pqType == parquet::Type::BYTE_ARRAY) {
parquet::TypedRowGroupStatistics<parquet::DataType<parquet::Type::BYTE_ARRAY>>* stats = parquet::TypedStatistics<parquet::ByteArrayType>* stats =
(parquet::TypedRowGroupStatistics<parquet::DataType<parquet::Type::BYTE_ARRAY>>*)_stats.get(); (parquet::TypedStatistics<parquet::ByteArrayType>*)_stats.get();
minPtr = stats->min().ptr; minPtr = stats->min().ptr;
minLen = stats->min().len; minLen = stats->min().len;
@ -137,9 +137,9 @@ bool ParquetCursor::currentRowGroupSatisfiesBlobFilter(Constraint& constraint, s
} }
} }
bool ParquetCursor::currentRowGroupSatisfiesTextFilter(Constraint& constraint, std::shared_ptr<parquet::RowGroupStatistics> _stats) { bool ParquetCursor::currentRowGroupSatisfiesTextFilter(Constraint& constraint, std::shared_ptr<parquet::Statistics> _stats) {
parquet::TypedRowGroupStatistics<parquet::DataType<parquet::Type::BYTE_ARRAY>>* stats = parquet::TypedStatistics<parquet::ByteArrayType>* stats =
(parquet::TypedRowGroupStatistics<parquet::DataType<parquet::Type::BYTE_ARRAY>>*)_stats.get(); (parquet::TypedStatistics<parquet::ByteArrayType>*)_stats.get();
if(!stats->HasMinMax()) { if(!stats->HasMinMax()) {
return true; return true;
@ -195,7 +195,7 @@ int64_t int96toMsSinceEpoch(const parquet::Int96& rv) {
return nsSinceEpoch; return nsSinceEpoch;
} }
bool ParquetCursor::currentRowGroupSatisfiesIntegerFilter(Constraint& constraint, std::shared_ptr<parquet::RowGroupStatistics> _stats) { bool ParquetCursor::currentRowGroupSatisfiesIntegerFilter(Constraint& constraint, std::shared_ptr<parquet::Statistics> _stats) {
if(!_stats->HasMinMax()) { if(!_stats->HasMinMax()) {
return true; return true;
} }
@ -211,27 +211,27 @@ bool ParquetCursor::currentRowGroupSatisfiesIntegerFilter(Constraint& constraint
parquet::Type::type pqType = types[column]; parquet::Type::type pqType = types[column];
if(pqType == parquet::Type::INT32) { if(pqType == parquet::Type::INT32) {
parquet::TypedRowGroupStatistics<parquet::DataType<parquet::Type::INT32>>* stats = parquet::TypedStatistics<parquet::Int32Type>* stats =
(parquet::TypedRowGroupStatistics<parquet::DataType<parquet::Type::INT32>>*)_stats.get(); (parquet::TypedStatistics<parquet::Int32Type>*)_stats.get();
min = stats->min(); min = stats->min();
max = stats->max(); max = stats->max();
} else if(pqType == parquet::Type::INT64) { } else if(pqType == parquet::Type::INT64) {
parquet::TypedRowGroupStatistics<parquet::DataType<parquet::Type::INT64>>* stats = parquet::TypedStatistics<parquet::Int64Type>* stats =
(parquet::TypedRowGroupStatistics<parquet::DataType<parquet::Type::INT64>>*)_stats.get(); (parquet::TypedStatistics<parquet::Int64Type>*)_stats.get();
min = stats->min(); min = stats->min();
max = stats->max(); max = stats->max();
} else if(pqType == parquet::Type::INT96) { } else if(pqType == parquet::Type::INT96) {
parquet::TypedRowGroupStatistics<parquet::DataType<parquet::Type::INT96>>* stats = parquet::TypedStatistics<parquet::Int96Type>* stats =
(parquet::TypedRowGroupStatistics<parquet::DataType<parquet::Type::INT96>>*)_stats.get(); (parquet::TypedStatistics<parquet::Int96Type>*)_stats.get();
min = int96toMsSinceEpoch(stats->min()); min = int96toMsSinceEpoch(stats->min());
max = int96toMsSinceEpoch(stats->max()); max = int96toMsSinceEpoch(stats->max());
} else if(pqType == parquet::Type::BOOLEAN) { } else if(pqType == parquet::Type::BOOLEAN) {
parquet::TypedRowGroupStatistics<parquet::DataType<parquet::Type::BOOLEAN>>* stats = parquet::TypedStatistics<parquet::BooleanType>* stats =
(parquet::TypedRowGroupStatistics<parquet::DataType<parquet::Type::BOOLEAN>>*)_stats.get(); (parquet::TypedStatistics<parquet::BooleanType>*)_stats.get();
min = stats->min(); min = stats->min();
max = stats->max(); max = stats->max();
@ -272,7 +272,7 @@ bool ParquetCursor::currentRowGroupSatisfiesIntegerFilter(Constraint& constraint
return true; return true;
} }
bool ParquetCursor::currentRowGroupSatisfiesDoubleFilter(Constraint& constraint, std::shared_ptr<parquet::RowGroupStatistics> _stats) { bool ParquetCursor::currentRowGroupSatisfiesDoubleFilter(Constraint& constraint, std::shared_ptr<parquet::Statistics> _stats) {
if(!_stats->HasMinMax()) { if(!_stats->HasMinMax()) {
return true; return true;
} }
@ -288,14 +288,14 @@ bool ParquetCursor::currentRowGroupSatisfiesDoubleFilter(Constraint& constraint,
parquet::Type::type pqType = types[column]; parquet::Type::type pqType = types[column];
if(pqType == parquet::Type::DOUBLE) { if(pqType == parquet::Type::DOUBLE) {
parquet::TypedRowGroupStatistics<parquet::DataType<parquet::Type::DOUBLE>>* stats = parquet::TypedStatistics<parquet::DoubleType>* stats =
(parquet::TypedRowGroupStatistics<parquet::DataType<parquet::Type::DOUBLE>>*)_stats.get(); (parquet::TypedStatistics<parquet::DoubleType>*)_stats.get();
min = stats->min(); min = stats->min();
max = stats->max(); max = stats->max();
} else if(pqType == parquet::Type::FLOAT) { } else if(pqType == parquet::Type::FLOAT) {
parquet::TypedRowGroupStatistics<parquet::DataType<parquet::Type::FLOAT>>* stats = parquet::TypedStatistics<parquet::FloatType>* stats =
(parquet::TypedRowGroupStatistics<parquet::DataType<parquet::Type::FLOAT>>*)_stats.get(); (parquet::TypedStatistics<parquet::FloatType>*)_stats.get();
min = stats->min(); min = stats->min();
max = stats->max(); max = stats->max();
@ -527,7 +527,7 @@ bool ParquetCursor::currentRowGroupSatisfiesFilter() {
} else { } else {
std::unique_ptr<parquet::ColumnChunkMetaData> md = rowGroupMetadata->ColumnChunk(column); std::unique_ptr<parquet::ColumnChunkMetaData> md = rowGroupMetadata->ColumnChunk(column);
if(md->is_stats_set()) { if(md->is_stats_set()) {
std::shared_ptr<parquet::RowGroupStatistics> stats = md->statistics(); std::shared_ptr<parquet::Statistics> stats = md->statistics();
// SQLite is much looser with types than you might expect if you // SQLite is much looser with types than you might expect if you
// come from a Postgres background. The constraint '30.0' (that is, // come from a Postgres background. The constraint '30.0' (that is,
@ -545,7 +545,7 @@ bool ParquetCursor::currentRowGroupSatisfiesFilter() {
} else { } else {
parquet::Type::type pqType = types[column]; parquet::Type::type pqType = types[column];
if(pqType == parquet::Type::BYTE_ARRAY && logicalTypes[column] == parquet::LogicalType::UTF8) { if(pqType == parquet::Type::BYTE_ARRAY && logicalTypes[column] == parquet::ConvertedType::UTF8) {
rv = currentRowGroupSatisfiesTextFilter(constraints[i], stats); rv = currentRowGroupSatisfiesTextFilter(constraints[i], stats);
} else if(pqType == parquet::Type::BYTE_ARRAY) { } else if(pqType == parquet::Type::BYTE_ARRAY) {
rv = currentRowGroupSatisfiesBlobFilter(constraints[i], stats); rv = currentRowGroupSatisfiesBlobFilter(constraints[i], stats);
@ -609,12 +609,12 @@ start:
} }
while(logicalTypes.size() < (unsigned int)rowGroupMetadata->num_columns()) { while(logicalTypes.size() < (unsigned int)rowGroupMetadata->num_columns()) {
logicalTypes.push_back(rowGroupMetadata->schema()->Column(0)->logical_type()); logicalTypes.push_back(rowGroupMetadata->schema()->Column(0)->converted_type());
} }
for(unsigned int i = 0; i < (unsigned int)rowGroupMetadata->num_columns(); i++) { for(unsigned int i = 0; i < (unsigned int)rowGroupMetadata->num_columns(); i++) {
types[i] = rowGroupMetadata->schema()->Column(i)->physical_type(); types[i] = rowGroupMetadata->schema()->Column(i)->physical_type();
logicalTypes[i] = rowGroupMetadata->schema()->Column(i)->logical_type(); logicalTypes[i] = rowGroupMetadata->schema()->Column(i)->converted_type();
} }
for(unsigned int i = 0; i < colRows.size(); i++) { for(unsigned int i = 0; i < colRows.size(); i++) {
@ -664,7 +664,7 @@ bool ParquetCursor::currentRowSatisfiesFilter() {
rv = !isNull(column); rv = !isNull(column);
} else { } else {
if(logicalTypes[column] == parquet::LogicalType::UTF8) { if(logicalTypes[column] == parquet::ConvertedType::UTF8) {
rv = currentRowSatisfiesTextFilter(constraints[i]); rv = currentRowSatisfiesTextFilter(constraints[i]);
} else { } else {
parquet::Type::type pqType = types[column]; parquet::Type::type pqType = types[column];
@ -775,7 +775,7 @@ void ParquetCursor::ensureColumn(int col) {
case parquet::Type::INT64: case parquet::Type::INT64:
{ {
parquet::Int64Scanner* s = (parquet::Int64Scanner*)scanners[col].get(); parquet::Int64Scanner* s = (parquet::Int64Scanner*)scanners[col].get();
long rv = 0; long long rv = 0;
s->NextValue(&rv, &wasNull); s->NextValue(&rv, &wasNull);
break; break;
} }
@ -859,7 +859,7 @@ void ParquetCursor::ensureColumn(int col) {
case parquet::Type::INT64: case parquet::Type::INT64:
{ {
parquet::Int64Scanner* s = (parquet::Int64Scanner*)scanners[col].get(); parquet::Int64Scanner* s = (parquet::Int64Scanner*)scanners[col].get();
long rv = 0; long long rv = 0;
hadValue = s->NextValue(&rv, &wasNull); hadValue = s->NextValue(&rv, &wasNull);
colIntValues[col] = rv; colIntValues[col] = rv;
break; break;
@ -928,7 +928,7 @@ parquet::Type::type ParquetCursor::getPhysicalType(int col) {
return types[col]; return types[col];
} }
parquet::LogicalType::type ParquetCursor::getLogicalType(int col) { parquet::ConvertedType::type ParquetCursor::getLogicalType(int col) {
return logicalTypes[col]; return logicalTypes[col];
} }

12
parquet/parquet_cursor.h
View File

@ -13,7 +13,7 @@ class ParquetCursor {
std::shared_ptr<parquet::RowGroupReader> rowGroup; std::shared_ptr<parquet::RowGroupReader> rowGroup;
std::vector<std::shared_ptr<parquet::Scanner>> scanners; std::vector<std::shared_ptr<parquet::Scanner>> scanners;
std::vector<parquet::Type::type> types; std::vector<parquet::Type::type> types;
std::vector<parquet::LogicalType::type> logicalTypes; std::vector<parquet::ConvertedType::type> logicalTypes;
std::vector<int> colRows; std::vector<int> colRows;
std::vector<bool> colNulls; std::vector<bool> colNulls;
@ -36,10 +36,10 @@ class ParquetCursor {
bool currentRowSatisfiesFilter(); bool currentRowSatisfiesFilter();
bool currentRowGroupSatisfiesFilter(); bool currentRowGroupSatisfiesFilter();
bool currentRowGroupSatisfiesRowIdFilter(Constraint& constraint); bool currentRowGroupSatisfiesRowIdFilter(Constraint& constraint);
bool currentRowGroupSatisfiesTextFilter(Constraint& constraint, std::shared_ptr<parquet::RowGroupStatistics> stats); bool currentRowGroupSatisfiesTextFilter(Constraint& constraint, std::shared_ptr<parquet::Statistics> stats);
bool currentRowGroupSatisfiesBlobFilter(Constraint& constraint, std::shared_ptr<parquet::RowGroupStatistics> stats); bool currentRowGroupSatisfiesBlobFilter(Constraint& constraint, std::shared_ptr<parquet::Statistics> stats);
bool currentRowGroupSatisfiesIntegerFilter(Constraint& constraint, std::shared_ptr<parquet::RowGroupStatistics> stats); bool currentRowGroupSatisfiesIntegerFilter(Constraint& constraint, std::shared_ptr<parquet::Statistics> stats);
bool currentRowGroupSatisfiesDoubleFilter(Constraint& constraint, std::shared_ptr<parquet::RowGroupStatistics> stats); bool currentRowGroupSatisfiesDoubleFilter(Constraint& constraint, std::shared_ptr<parquet::Statistics> stats);
bool currentRowSatisfiesTextFilter(Constraint& constraint); bool currentRowSatisfiesTextFilter(Constraint& constraint);
bool currentRowSatisfiesIntegerFilter(Constraint& constraint); bool currentRowSatisfiesIntegerFilter(Constraint& constraint);
@ -60,7 +60,7 @@ public:
unsigned int getNumConstraints() const; unsigned int getNumConstraints() const;
const Constraint& getConstraint(unsigned int i) const; const Constraint& getConstraint(unsigned int i) const;
parquet::Type::type getPhysicalType(int col); parquet::Type::type getPhysicalType(int col);
parquet::LogicalType::type getLogicalType(int col); parquet::ConvertedType::type getLogicalType(int col);
ParquetTable* getTable() const; ParquetTable* getTable() const;
int getInt32(int col); int getInt32(int col);

38
parquet/parquet_table.cc
View File

@ -66,33 +66,33 @@ std::string ParquetTable::CreateStatement() {
std::string type; std::string type;
parquet::Type::type physical = col->physical_type(); parquet::Type::type physical = col->physical_type();
parquet::LogicalType::type logical = col->logical_type(); parquet::ConvertedType::type logical = col->converted_type();
// Be explicit about which types we understand so we don't mislead someone // Be explicit about which types we understand so we don't mislead someone
// whose unsigned ints start getting interpreted as signed. (We could // whose unsigned ints start getting interpreted as signed. (We could
// support this for UINT_8/16/32 -- and for UINT_64 we could throw if // support this for UINT_8/16/32 -- and for UINT_64 we could throw if
// the high bit was set.) // the high bit was set.)
if(logical == parquet::LogicalType::NONE || if(logical == parquet::ConvertedType::NONE ||
logical == parquet::LogicalType::UTF8 || logical == parquet::ConvertedType::UTF8 ||
logical == parquet::LogicalType::DATE || logical == parquet::ConvertedType::DATE ||
logical == parquet::LogicalType::TIME_MILLIS || logical == parquet::ConvertedType::TIME_MILLIS ||
logical == parquet::LogicalType::TIMESTAMP_MILLIS || logical == parquet::ConvertedType::TIMESTAMP_MILLIS ||
logical == parquet::LogicalType::TIME_MICROS || logical == parquet::ConvertedType::TIME_MICROS ||
logical == parquet::LogicalType::TIMESTAMP_MICROS || logical == parquet::ConvertedType::TIMESTAMP_MICROS ||
logical == parquet::LogicalType::INT_8 || logical == parquet::ConvertedType::INT_8 ||
logical == parquet::LogicalType::INT_16 || logical == parquet::ConvertedType::INT_16 ||
logical == parquet::LogicalType::INT_32 || logical == parquet::ConvertedType::INT_32 ||
logical == parquet::LogicalType::INT_64) { logical == parquet::ConvertedType::INT_64) {
switch(physical) { switch(physical) {
case parquet::Type::BOOLEAN: case parquet::Type::BOOLEAN:
type = "TINYINT"; type = "TINYINT";
break; break;
case parquet::Type::INT32: case parquet::Type::INT32:
if(logical == parquet::LogicalType::NONE || if(logical == parquet::ConvertedType::NONE ||
logical == parquet::LogicalType::INT_32) { logical == parquet::ConvertedType::INT_32) {
type = "INT"; type = "INT";
} else if(logical == parquet::LogicalType::INT_8) { } else if(logical == parquet::ConvertedType::INT_8) {
type = "TINYINT"; type = "TINYINT";
} else if(logical == parquet::LogicalType::INT_16) { } else if(logical == parquet::ConvertedType::INT_16) {
type = "SMALLINT"; type = "SMALLINT";
} }
break; break;
@ -109,7 +109,7 @@ std::string ParquetTable::CreateStatement() {
type = "DOUBLE"; type = "DOUBLE";
break; break;
case parquet::Type::BYTE_ARRAY: case parquet::Type::BYTE_ARRAY:
if(logical == parquet::LogicalType::UTF8) { if(logical == parquet::ConvertedType::UTF8) {
type = "TEXT"; type = "TEXT";
} else { } else {
type = "BLOB"; type = "BLOB";
@ -126,7 +126,7 @@ std::string ParquetTable::CreateStatement() {
if(type.empty()) { if(type.empty()) {
std::ostringstream ss; std::ostringstream ss;
ss << __FILE__ << ":" << __LINE__ << ": column " << i << " has unsupported type: " << ss << __FILE__ << ":" << __LINE__ << ": column " << i << " has unsupported type: " <<
parquet::TypeToString(physical) << "/" << parquet::LogicalTypeToString(logical); parquet::TypeToString(physical) << "/" << parquet::ConvertedTypeToString(logical);
throw std::invalid_argument(ss.str()); throw std::invalid_argument(ss.str());
} }
@ -138,7 +138,7 @@ std::string ParquetTable::CreateStatement() {
col->physical_type(), col->physical_type(),
parquet::TypeToString(col->physical_type()).data(), parquet::TypeToString(col->physical_type()).data(),
col->logical_type(), col->logical_type(),
parquet::LogicalTypeToString(col->logical_type()).data(), parquet::ConvertedTypeToString(col->logical_type()).data(),
type.data()); type.data());
#endif #endif