Skip to content

Fix loading from streams when there's worker that receives no chunks #98

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

Merged
merged 3 commits into from
Jan 22, 2021
Merged

Fix loading from streams when there's worker that receives no chunks #98

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5c28fcb
Fixes for loading from streams when some worker receive no chunks.
sighingnow Jan 20, 2021
510405c
Check if nullptr first.
sighingnow Jan 21, 2021
405ec85
Fixes...
sighingnow Jan 21, 2021
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
24 changes: 18 additions & 6 deletions analytical_engine/core/loader/arrow_fragment_loader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -437,7 +437,7 @@ class ArrowFragmentLoader {
vineyard::GatherVTables(
client_, {vineyard::VYObjectIDFromString(vertices[0]->values)},
comm_spec_.local_id(), comm_spec_.local_num()));
if (tables.size() == 1) {
if (tables.size() == 1 && tables[0] != nullptr) {
std::shared_ptr<arrow::KeyValueMetadata> meta;
if (tables[0]->schema()->metadata() == nullptr) {
meta = std::make_shared<arrow::KeyValueMetadata>();
Expand Down Expand Up @@ -485,11 +485,16 @@ class ArrowFragmentLoader {
total_parts, vertices[i]->properties));
table = tmp;
} else if (vertices[i]->protocol == "vineyard") {
VLOG(2) << "read vertex table from vineyard: " << vertices[i]->values;
VY_OK_OR_RAISE(vineyard::ReadTableFromVineyard(
client_, vineyard::VYObjectIDFromString(vertices[i]->values),
table, comm_spec_.local_id(), comm_spec_.local_num()));
LOG(INFO) << "read table from vineyard: " << vertices[i]->values
<< ": " << table->schema()->ToString();
if (table != nullptr) {
VLOG(2) << "schema of vertex table: "
<< table->schema()->ToString();
} else {
VLOG(2) << "vertex table is null";
}
} else {
LOG(ERROR) << "Unsupported protocol: " << vertices[i]->protocol;
}
Expand Down Expand Up @@ -612,7 +617,8 @@ class ArrowFragmentLoader {
{{vineyard::VYObjectIDFromString(
edges[0]->sub_labels[0].values)}},
comm_spec_.local_id(), comm_spec_.local_num()));
if (tables.size() == 1 && tables[0].size() == 1) {
if (tables.size() == 1 && tables[0].size() == 1 &&
tables[0][0] != nullptr) {
std::shared_ptr<arrow::KeyValueMetadata> meta;
if (tables[0][0]->schema()->metadata() == nullptr) {
meta = std::make_shared<arrow::KeyValueMetadata>();
Expand Down Expand Up @@ -675,11 +681,17 @@ class ArrowFragmentLoader {
sub_labels[j].column_num, index, total_parts,
sub_labels[j].properties));
} else if (sub_labels[j].protocol == "vineyard") {
LOG(INFO) << "read edge table from vineyard: "
<< sub_labels[j].values;
VY_OK_OR_RAISE(vineyard::ReadTableFromVineyard(
client_, vineyard::VYObjectIDFromString(sub_labels[j].values),
table, comm_spec_.local_id(), comm_spec_.local_num()));
LOG(INFO) << "read table from vineyard: " << sub_labels[j].values
<< ": " << table->schema()->ToString();
if (table == nullptr) {
VLOG(2) << "edge table is null";
} else {
VLOG(2) << "schema of edge table: "
<< table->schema()->ToString();
}
} else {
LOG(ERROR) << "Unrecognized protocol: " << sub_labels[j].protocol;
}
Expand Down