RKuttruff commented on code in PR #249:
URL:
https://github.com/apache/incubator-sdap-nexus/pull/249#discussion_r1279944890
##########
analysis/webservice/algorithms_spark/Matchup.py:
##########
@@ -283,37 +329,60 @@ def calc(self, request, **args):
if depth_max is not None:
args["depthMax"] = float(depth_max)
- total_keys = len(list(spark_result.keys()))
- total_values = sum(len(v) for v in spark_result.values())
- details = {
- "timeToComplete": int((end - start).total_seconds()),
- "numSecondaryMatched": total_values,
- "numPrimaryMatched": total_keys
- }
- matches = Matchup.convert_to_matches(spark_result)
+ with ResultsStorage(self.config) as resultsStorage:
+ execution_id = str(resultsStorage.insertInitialExecution(
+ params=args,
+ startTime=start,
+ status=ExecutionStatus.RUNNING.value
+ ))
- def do_result_insert():
- with ResultsStorage(self.config) as storage:
- storage.insertResults(results=matches, params=args,
stats=details,
- startTime=start, completeTime=end,
userEmail="",
- execution_id=execution_id)
+ self.log.debug("Querying for tiles in search domain")
+ # Get tile ids in box
+ tile_ids = [tile.tile_id for tile in
+
self._get_tile_service().find_tiles_in_polygon(bounding_polygon,
primary_ds_name,
+
start_seconds_from_epoch, end_seconds_from_epoch,
+ fetch_data=False,
fl='id',
+
sort=['tile_min_time_dt asc', 'tile_min_lon asc',
+
'tile_min_lat asc'], rows=5000)]
- threading.Thread(target=do_result_insert).start()
+ self.log.info('Found %s tile_ids', len(tile_ids))
- # Get only the first "result_size_limit" results
- # '0' means returns everything
- if result_size_limit > 0:
- return_matches = matches[0:result_size_limit]
- else:
- return_matches = matches
+ if not tile_ids:
+ # There are no matching tiles
+ end = datetime.utcnow()
+ with ResultsStorage(self.config) as storage:
+ storage.updateExecution(
+ uuid.UUID(execution_id),
+ completeTime=end,
+ status=ExecutionStatus.FAILED.value,
+ message='No tiles matched the provided domain'
+ )
- result = DomsQueryResults(results=return_matches, args=args,
- details=details, bounds=None,
- count=len(matches), computeOptions=None,
- executionId=execution_id)
+ # Start async processing with Spark. Do not wait for response
+ # before returning to user.
+ tornado_io_loop.run_in_executor(request.requestHandler.executor,
functools.partial(
+ self.async_calc,
+ execution_id=execution_id,
+ tile_ids=tile_ids,
+ bounding_polygon=bounding_polygon,
+ primary_ds_name=primary_ds_name,
+ secondary_ds_names=secondary_ds_names,
+ parameter_s=parameter_s,
+ start_time=start_time,
+ end_time=end_time,
+ depth_min=depth_min,
+ depth_max=depth_max,
+ time_tolerance=time_tolerance,
+ radius_tolerance=radius_tolerance,
+ platforms=platforms,
+ match_once=match_once,
+ result_size_limit=result_size_limit,
+ start=start
+ ))
Review Comment:
`prioritize_distance` should be passed here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscr...@sdap.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
