\brief Make the output schema of an as-of-join node \param[in] input_schema the schema of each input to the node \param[in] indices_of_on_key the on-key index of each input to the node \param[in] indices_of_by_key the by-key indices of each input to the node
| 1244 | /// \param[in] indices_of_on_key the on-key index of each input to the node |
| 1245 | /// \param[in] indices_of_by_key the by-key indices of each input to the node |
| 1246 | static arrow::Result<std::shared_ptr<Schema>> MakeOutputSchema( |
| 1247 | const std::vector<std::shared_ptr<Schema>> input_schema, |
| 1248 | const std::vector<col_index_t>& indices_of_on_key, |
| 1249 | const std::vector<std::vector<col_index_t>>& indices_of_by_key) { |
| 1250 | std::vector<std::shared_ptr<arrow::Field>> fields; |
| 1251 | |
| 1252 | size_t n_by = indices_of_by_key.size() == 0 ? 0 : indices_of_by_key[0].size(); |
| 1253 | const DataType* on_key_type = NULLPTR; |
| 1254 | std::vector<const DataType*> by_key_type(n_by, NULLPTR); |
| 1255 | // Take all non-key, non-time RHS fields |
| 1256 | for (size_t j = 0; j < input_schema.size(); ++j) { |
| 1257 | const auto& on_field_ix = indices_of_on_key[j]; |
| 1258 | const auto& by_field_ix = indices_of_by_key[j]; |
| 1259 | |
| 1260 | if ((on_field_ix == -1) || std_has(by_field_ix, -1)) { |
| 1261 | return Status::Invalid("Missing join key on table ", j); |
| 1262 | } |
| 1263 | |
| 1264 | const auto& on_field = input_schema[j]->fields()[on_field_ix]; |
| 1265 | std::vector<const Field*> by_field(n_by); |
| 1266 | for (size_t k = 0; k < n_by; k++) { |
| 1267 | by_field[k] = input_schema[j]->fields()[by_field_ix[k]].get(); |
| 1268 | } |
| 1269 | |
| 1270 | if (on_key_type == NULLPTR) { |
| 1271 | on_key_type = on_field->type().get(); |
| 1272 | } else if (*on_key_type != *on_field->type()) { |
| 1273 | return Status::Invalid("Expected on-key type ", *on_key_type, " but got ", |
| 1274 | *on_field->type(), " for field ", on_field->name(), |
| 1275 | " in input ", j); |
| 1276 | } |
| 1277 | for (size_t k = 0; k < n_by; k++) { |
| 1278 | if (by_key_type[k] == NULLPTR) { |
| 1279 | by_key_type[k] = by_field[k]->type().get(); |
| 1280 | } else if (*by_key_type[k] != *by_field[k]->type()) { |
| 1281 | return Status::Invalid("Expected by-key type ", *by_key_type[k], " but got ", |
| 1282 | *by_field[k]->type(), " for field ", by_field[k]->name(), |
| 1283 | " in input ", j); |
| 1284 | } |
| 1285 | } |
| 1286 | |
| 1287 | for (int i = 0; i < input_schema[j]->num_fields(); ++i) { |
| 1288 | const auto field = input_schema[j]->field(i); |
| 1289 | bool as_output; // true if the field appears as an output |
| 1290 | if (i == on_field_ix) { |
| 1291 | ARROW_RETURN_NOT_OK(is_valid_on_field(field)); |
| 1292 | // Only add on field from the left table |
| 1293 | as_output = (j == 0); |
| 1294 | } else if (std_has(by_field_ix, i)) { |
| 1295 | ARROW_RETURN_NOT_OK(is_valid_by_field(field)); |
| 1296 | // Only add by field from the left table |
| 1297 | as_output = (j == 0); |
| 1298 | } else { |
| 1299 | ARROW_RETURN_NOT_OK(is_valid_data_field(field)); |
| 1300 | as_output = true; |
| 1301 | } |
| 1302 | if (as_output) { |
| 1303 | fields.push_back(field); |