Changeset: a92ded53f773 for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB/rev/a92ded53f773
Modified Files:
sql/backends/monet5/vaults/parquet/parquet.c
sql/backends/monet5/vaults/parquet/pqc_dict_lookup.h
sql/backends/monet5/vaults/parquet/pqc_dict_lookup_var.h
sql/backends/monet5/vaults/parquet/pqc_filemetadata.c
sql/backends/monet5/vaults/parquet/pqc_reader.c
sql/backends/monet5/vaults/parquet/pqc_thrift.c
sql/backends/monet5/vaults/parquet/pqc_thrift.h
Branch: default
Log Message:
improved parquet reader for unalign reads
diffs (truncated from 526 to 300 lines):
diff --git a/sql/backends/monet5/vaults/parquet/parquet.c
b/sql/backends/monet5/vaults/parquet/parquet.c
--- a/sql/backends/monet5/vaults/parquet/parquet.c
+++ b/sql/backends/monet5/vaults/parquet/parquet.c
@@ -467,7 +467,7 @@ pqc_relation(mvc *sql, sql_subfunc *f, c
if (e->type == LT_UNKNOWN || e->type == listtype)
set_intern(ne);
list_append(res_exps, ne);
- if (e->precision && *est > ((lng) 1 << e->precision)) {
+ if (e->precision && e->precision < 64 && *est > ((lng)
1 << e->precision)) {
prop *p = ne->p = prop_create(sql->sa,
PROP_NUNIQUES, ne->p);
p->value.dval = 1L << e->precision;
}
@@ -678,11 +678,11 @@ PARQUETread_large(BAT **R, pqc_creader *
memset(h->base, 0, h->size);
#endif
rb->tascii = false; /* tobe fixed */
- size_t offset = 0;
+ size_t offset = 8;
if (dict >= 4 && is_string)
- offset = h->free;
+ offset = h->free+8;
int64_t rsz = 0, tsz = 0;
- if ((rsz = pqc_read_chunk(r->c[pse->ccnr], wnr,
rb->theap->base, ((char*)h->base)+h->free, sz, &offset, &dict)) < 0) {
+ if ((rsz = pqc_read_chunk(r->c[pse->ccnr], wnr,
rb->theap->base, ((char*)h->base)+h->free+8, sz, &offset, &dict)) < 0) {
BBPreclaim(rb);
const char *err = pqc_get_error(r->c[pse->ccnr]);
if (err)
diff --git a/sql/backends/monet5/vaults/parquet/pqc_dict_lookup.h
b/sql/backends/monet5/vaults/parquet/pqc_dict_lookup.h
--- a/sql/backends/monet5/vaults/parquet/pqc_dict_lookup.h
+++ b/sql/backends/monet5/vaults/parquet/pqc_dict_lookup.h
@@ -59,13 +59,13 @@ pqc_dict_lookup( pqc_creader_t *cr, void
pos += (sh)/8;
} else if (nr_bits < 16) {
for(; i < nrows && j < cr->remaining; j++, i++)
{
- uint16_t v =
pqc_sht(*(uint16_t*)(data+pos));
+ uint16_t v =
pqc_sht(get_uint16(data+pos));
uint32_t idx = (v >> sh)&mask;
sh += nr_bits;
if (sh >= 16) {
pos+=2;
sh -= 16;
- uint16_t v =
pqc_sht(*(uint16_t*)(data+pos));
+ uint16_t v =
pqc_sht(get_uint16(data+pos));
idx |= (v << (nr_bits-sh))&mask;
}
assert(idx < cr->dict_num_values);
@@ -75,13 +75,13 @@ pqc_dict_lookup( pqc_creader_t *cr, void
pos += (sh)/8;
} else if (nr_bits < 32) {
for(; i < nrows && j < cr->remaining; j++, i++)
{
- uint32_t v =
pqc_int(*(uint32_t*)(data+pos));
+ uint32_t v =
pqc_int(get_uint32(data+pos));
uint32_t idx = (v >> sh)&mask;
sh += nr_bits;
if (sh >= 32) {
pos+=4;
sh -= 32;
- uint32_t v =
pqc_int(*(uint32_t*)(data+pos));
+ uint32_t v =
pqc_int(get_uint32(data+pos));
idx |= (v << (nr_bits-sh))&mask;
}
assert(idx < cr->dict_num_values);
@@ -145,13 +145,13 @@ pqc_dict_lookup( pqc_creader_t *cr, void
}
} else if (nr_bits < 16) {
for (; i < nrows && j < m; j++, i++) {
- uint16_t v =
pqc_sht(*(uint16_t*)(data+pos));
+ uint16_t v =
pqc_sht(get_uint16(data+pos));
uint32_t idx = (v >> sh)&mask;
sh += nr_bits;
if (sh >= 16) {
pos+=2;
sh -= 16;
- uint16_t v =
pqc_sht(*(uint16_t*)(data+pos));
+ uint16_t v =
pqc_sht(get_uint16(data+pos));
idx |= (v << (nr_bits-sh))&mask;
}
assert(idx < cr->dict_num_values);
@@ -166,13 +166,13 @@ pqc_dict_lookup( pqc_creader_t *cr, void
}
} else if (nr_bits < 32) {
for (; i < nrows && j < m; j++, i++) {
- uint32_t v =
pqc_int(*(uint32_t*)(data+pos));
+ uint32_t v =
pqc_int(get_uint32(data+pos));
uint32_t idx = (v >> sh)&mask;
sh += nr_bits;
if (sh >= 32) {
pos+=4;
sh -= 32;
- uint32_t v =
pqc_int(*(uint32_t*)(data+pos));
+ uint32_t v =
pqc_int(get_uint32(data+pos));
idx |= (v << (nr_bits-sh))&mask;
}
assert(idx < cr->dict_num_values);
@@ -210,7 +210,7 @@ pqc_dict_lookup( pqc_creader_t *cr, void
}
} else if (nr_bits <= 16) { /* rle */
len>>=1;
- uint16_t idx = pqc_sht(*(uint16_t*)(data+pos));
+ uint16_t idx = pqc_sht(get_uint16(data+pos));
idx &= mask;
uint32_t j = 0;
pos += 2;
@@ -223,7 +223,7 @@ pqc_dict_lookup( pqc_creader_t *cr, void
}
} else if (nr_bits <= 24) { /* rle */
len>>=1;
- uint32_t idx = pqc_int(*(uint32_t*)(data+pos));
+ uint32_t idx = pqc_int(get_uint32(data+pos));
idx &= mask;
uint32_t j = 0;
pos += 3;
diff --git a/sql/backends/monet5/vaults/parquet/pqc_dict_lookup_var.h
b/sql/backends/monet5/vaults/parquet/pqc_dict_lookup_var.h
--- a/sql/backends/monet5/vaults/parquet/pqc_dict_lookup_var.h
+++ b/sql/backends/monet5/vaults/parquet/pqc_dict_lookup_var.h
@@ -60,13 +60,13 @@ pqc_dict_lookup( pqc_creader_t *cr, void
pos += (sh)/8;
} else if (nr_bits < 16) {
for(; i < nrows && j < cr->remaining; j++, i++)
{
- uint16_t v =
pqc_sht(*(uint16_t*)(data+pos));
+ uint16_t v =
pqc_sht(get_uint16(data+pos));
uint32_t idx = (v >> sh)&mask;
sh += nr_bits;
if (sh >= 16) {
pos+=2;
sh -= 16;
- uint16_t v =
pqc_sht(*(uint16_t*)(data+pos));
+ uint16_t v =
pqc_sht(get_uint16(data+pos));
idx |= (v << (nr_bits-sh))&mask;
}
assert(idx < cr->dict_num_values);
@@ -76,13 +76,13 @@ pqc_dict_lookup( pqc_creader_t *cr, void
pos += (sh)/8;
} else if (nr_bits < 32) {
for(; i < nrows && j < cr->remaining; j++, i++)
{
- uint32_t v =
pqc_int(*(uint32_t*)(data+pos));
+ uint32_t v =
pqc_int(get_uint32(data+pos));
uint32_t idx = (v >> sh)&mask;
sh += nr_bits;
if (sh >= 32) {
pos+=4;
sh -= 32;
- uint32_t v =
pqc_int(*(uint32_t*)(data+pos));
+ uint32_t v =
pqc_int(get_uint32(data+pos));
idx |= (v << (nr_bits-sh))&mask;
}
assert(idx < cr->dict_num_values);
@@ -156,13 +156,13 @@ pqc_dict_lookup( pqc_creader_t *cr, void
}
} else if (nr_bits < 16) {
for (; i < nrows && j < m; j++, i++) {
- uint16_t v =
pqc_sht(*(uint16_t*)(data+pos));
+ uint16_t v =
pqc_sht(get_uint16(data+pos));
uint32_t idx = (v >> sh)&mask;
sh += nr_bits;
if (sh >= 16) {
pos+=2;
sh -= 16;
- uint16_t v =
pqc_sht(*(uint16_t*)(data+pos));
+ uint16_t v =
pqc_sht(get_uint16(data+pos));
idx |= (v << (nr_bits-sh))&mask;
}
assert(idx < cr->dict_num_values);
@@ -177,13 +177,13 @@ pqc_dict_lookup( pqc_creader_t *cr, void
}
} else if (nr_bits < 32) {
for (; i < nrows && j < m; j++, i++) {
- uint32_t v =
pqc_int(*(uint32_t*)(data+pos));
+ uint32_t v =
pqc_int(get_uint32(data+pos));
uint32_t idx = (v >> sh)&mask;
sh += nr_bits;
if (sh >= 32) {
pos+=4;
sh -= 32;
- uint32_t v =
pqc_int(*(uint32_t*)(data+pos));
+ uint32_t v =
pqc_int(get_uint32(data+pos));
idx |= (v << (nr_bits-sh))&mask;
}
assert(idx < cr->dict_num_values);
@@ -219,7 +219,7 @@ pqc_dict_lookup( pqc_creader_t *cr, void
}
} else if (nr_bits <= 16) { /* rle */
len>>=1;
- uint16_t idx = pqc_sht(*(uint16_t*)(data+pos));
+ uint16_t idx = pqc_sht(get_uint16(data+pos));
pos += 2;
for(; i < nrows && j < len; j++, i++)
dst[i] = offset+offsets[idx];
@@ -230,7 +230,7 @@ pqc_dict_lookup( pqc_creader_t *cr, void
}
} else if (nr_bits <= 32) { /* rle */
len>>=1;
- uint32_t idx = pqc_int(*(uint32_t*)(data+pos));
+ uint32_t idx = pqc_int(get_uint32(data+pos));
pos += 4;
for(; i < nrows && j < len; j++, i++)
dst[i] = offset+offsets[idx];
diff --git a/sql/backends/monet5/vaults/parquet/pqc_filemetadata.c
b/sql/backends/monet5/vaults/parquet/pqc_filemetadata.c
--- a/sql/backends/monet5/vaults/parquet/pqc_filemetadata.c
+++ b/sql/backends/monet5/vaults/parquet/pqc_filemetadata.c
@@ -851,8 +851,8 @@ pqc_binary2string(pqc_schema_element *ps
if (pse->size == 64)
return snprintf(buf, sz, LLFMT,
*(int64_t*)bindata);
if (pse->size == 96) {
- uint64_t nanoseconds =
pqc_lng(*(uint64_t*)bindata);
- uint32_t julian_day =
pqc_int(*(uint32_t*)(bindata+8));
+ uint64_t nanoseconds =
pqc_lng(get_uint64((unsigned char*)bindata));
+ uint32_t julian_day =
pqc_int(get_uint32((unsigned char*)bindata+8));
nanoseconds /= LL_CONSTANT(1000);
julian_day -= 2440588;
diff --git a/sql/backends/monet5/vaults/parquet/pqc_reader.c
b/sql/backends/monet5/vaults/parquet/pqc_reader.c
--- a/sql/backends/monet5/vaults/parquet/pqc_reader.c
+++ b/sql/backends/monet5/vaults/parquet/pqc_reader.c
@@ -466,7 +466,7 @@ string_read_dict( pqc_creader_t *cr, uin
return -1;
if (num_values > 100000 || hsz > 2000000)
return -1;
- /* make one error to simplify memory management */
+ /* make one array to simplify memory management */
char *mem = NEW_ARRAY(char, (sizeof(char*) * num_values + sizeof(int) *
num_values) + hsz);
if (!mem)
return -1;
@@ -506,17 +506,19 @@ blob_read_dict( pqc_creader_t *cr, uint3
data += len + sizeof(int);
hsz += len + sizeof(size_t);
+ hsz = (hsz +15)&(~15);
}
+ hsz+=16;
if (num_values < 1)
return -1;
if (num_values > 100000 || hsz > 2000000)
return -1;
- /* make one error to simplify memory management */
+ /* make one array to simplify memory management */
char *mem = NEW_ARRAY(char, (sizeof(char*) * num_values + sizeof(int) *
num_values) + hsz);
if (!mem)
return -1;
char **rc = (char**)mem;
- char *buf = mem + sizeof(char*) * num_values + sizeof(int) *
num_values, *obuf = buf;
+ char *buf = mem + (((sizeof(char*) * num_values + sizeof(int) *
num_values)+15)&(~15)), *obuf = buf;
int *offsets = (int*)(mem + sizeof(char*) * num_values);
data = (uint8_t*)cr->dict;
for (i=0; i<num_values; i++) {
@@ -528,7 +530,7 @@ blob_read_dict( pqc_creader_t *cr, uint3
b->nitems = len;
offsets[i] = (int) (buf - obuf);
rc[i] = buf;
- buf += sizeof(size_t)+len;
+ buf += (sizeof(size_t)+len+15)&(~15);
data += len;
}
if (cr->dict_allocated)
@@ -754,6 +756,14 @@ pqc_page_header( pqc_reader_t *r, pqc_cr
} else {
pr->dict_allocated = false;
pr->dict = pr->buffer+pos;
+ if((((ulng)pr->dict) & 15) != 0) { /* copy for
alignment issues */
+ char *buf = pr->dict;
+ pr->dict = NEW_ARRAY(char, uncompressed_size);
+ if (!pr->dict)
+ return -1;
+ pr->dict_allocated = true;
+ memcpy(pr->dict, buf, uncompressed_size);
+ }
pr->dictsize = uncompressed_size;
pos += uncompressed_size;
if (!uncompressed_size)
@@ -773,21 +783,21 @@ pqc_page_header( pqc_reader_t *r, pqc_cr
char *s = pr->dict;
if (r->pse->size == 16)
for(uint32_t i = 0; i < num_values; i++,
s+=sizeof(uint16_t))
- *(uint16_t*)s = pqc_sht(*(uint16_t*)s);
+ *(uint16_t*)s = pqc_sht(get_uint16(s));
if (r->pse->size == 32)
for(uint32_t i = 0; i < num_values; i++,
s+=sizeof(uint32_t))
- *(uint32_t*)s = pqc_int(*(uint32_t*)s);
+ *(uint32_t*)s = pqc_int(get_uint32(s));
if (r->pse->size == 64)
for(uint32_t i = 0; i < num_values; i++,
s+=sizeof(uint64_t))
- *(uint64_t*)s = pqc_lng(*(uint64_t*)s);
+ *(uint64_t*)s = pqc_lng(get_uint64(s));
}
#endif
if (num_values && r->pse->type == inttype && r->pse->precision
== 96) { /* remove the 32 useless bits */
ulng *d = (ulng*)pr->dict;
- char *s = pr->dict;
+ unsigned char *s = (unsigned char*)pr->dict;
for(uint32_t i = 0; i < num_values; i++, s += 12) {
- uint64_t nanoseconds = pqc_lng(*(uint64_t*)s);
- uint32_t julian_day =
pqc_int(*(uint32_t*)(s+8));
+ uint64_t nanoseconds = pqc_lng(get_uint64(s));
+ uint32_t julian_day = pqc_int(get_uint32(s+8));
nanoseconds /= LL_CONSTANT(1000);
julian_day -= 2440588;
@@ -796,7 +806,8 @@ pqc_page_header( pqc_reader_t *r, pqc_cr
}
}
_______________________________________________
checkin-list mailing list -- [email protected]
To unsubscribe send an email to [email protected]
