Skip to content

Execute all readColumnChunk concurrently for a given RowGroup #33

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ZJONSSON wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Execute all readColumnChunk concurrently for a given RowGroup #33

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6157f76
Execute all readColumnChunk concurrently for a given RowGroup
ZJONSSON Jan 22, 2018
6d1376a
Don't block on reading the header
ZJONSSON Feb 8, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 15 additions & 3 deletions lib/reader.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,8 +93,10 @@ class ParquetReader {
let envelopeReader = await ParquetEnvelopeReader.openFile(filePath);

try {
await envelopeReader.readHeader();
let metadata = await envelopeReader.readFooter();
let header = envelopeReader.readHeader();
let metadataPromise = envelopeReader.readFooter();
await header;
let metadata = await metadataPromise;
return new ParquetReader(metadata, envelopeReader);
} catch (err) {
await envelopeReader.close();
Expand Down Expand Up @@ -220,6 +222,8 @@ class ParquetEnvelopeReader {
columnData: {}
};

let requests = [];

for (let colChunk of rowGroup.columns) {
const colMetadata = colChunk.meta_data;
const colKey = colMetadata.path_in_schema;
Expand All @@ -228,9 +232,17 @@ class ParquetEnvelopeReader {
continue;
}

buffer.columnData[colKey] = await this.readColumnChunk(schema, colChunk);
requests.push(this.readColumnChunk(schema, colChunk).then(payload => ({
colKey, payload
})));
}

let data = await Promise.all(requests);

data.forEach(d => {
buffer.columnData[d.colKey] = d.payload;
});

return buffer;
}

Expand Down