Feature/extract execution synchronization #3519
Conversation
backend/src/main/java/com/bakdata/conquery/apiv1/QueryProcessor.java
Outdated
Show resolved
Hide resolved
backend/src/main/java/com/bakdata/conquery/models/query/DistributedExecutionManager.java
Show resolved
Hide resolved
backend/src/main/java/com/bakdata/conquery/models/query/ExecutionManager.java
Outdated
Show resolved
Hide resolved
…re/extract-execution-synchronization # Conflicts: # backend/src/test/java/com/bakdata/conquery/util/support/TestConquery.java
backend/src/main/java/com/bakdata/conquery/models/query/DistributedExecutionManager.java
Show resolved
Hide resolved
| query.finish(ExecutionState.DONE); | ||
| if (finishedWorkers.equals(getWorkerHandler(execution).getAllWorkerIds())) { | ||
| execution.finish(ExecutionState.DONE, this); | ||
| clearLockInternalExecution(execution.getId()); |
There was a problem hiding this comment.
ist das nicht eher teil von finish?
There was a problem hiding this comment.
habs zu finish gepackt
There was a problem hiding this comment.
habs in die finish methoden gepackt
backend/src/main/java/com/bakdata/conquery/models/query/ExecutionManager.java
Outdated
Show resolved
Hide resolved
| public void clearLockInternalExecution(ManagedExecutionId id) { | ||
| R result = Objects.requireNonNull(executionResults.getIfPresent(id), "Cannot clear lock on absent execution result"); | ||
|
|
||
| result.getExecutingLock().countDown(); | ||
| } | ||
|
|
||
| public void clearLockExternalExecution(ManagedExecutionId id) { | ||
| ExternalResult result = Objects.requireNonNull(externalExecutionResults.getIfPresent(id), "Cannot clear lock on absent execution result"); | ||
|
|
||
| result.getExecutingLock().countDown(); | ||
| } |
There was a problem hiding this comment.
warum sind die unterschiedlich?
There was a problem hiding this comment.
Ich schaue ob ich die alle unter ein Dach bringen kann
There was a problem hiding this comment.
Schwierig, executionResults ist parametrizert, damit jeweils der Sql/ DistributedExecutiomanager ohne casting auf das Result zugreifen kann.
Da die Externen Results reinzumischen bringt nichts.
| if (execution instanceof InternalExecution<?>) { | ||
| result = executionResults.getIfPresent(id); | ||
| } | ||
| else { | ||
| result = externalExecutionResults.getIfPresent(id); | ||
| } |
There was a problem hiding this comment.
die Result Klasse musst du noch komplett aufräumen glaube ich, das ist etwas strange
There was a problem hiding this comment.
ich hab sie jetzt umbenannt, aber an sich nichts geändert. Kannst du mir einen Tipp geben, was dich stört?
There was a problem hiding this comment.
annst du mir einen Tipp geben, was dich stört?
In der Klasse waren sehr viele instanceof calls. Der hier spezifisch irritiert mich, weil du 2 identitsch konfigurierte Caches gleich benutzt, aber auf die Execution Klasse unterscheidest. Wenn die Klassen sich keine Logik Teilen kannst du dann einfach hard-casten, sonst würde ich es eben in eine gemeinsame Klasse stecken.
backend/src/main/java/com/bakdata/conquery/models/query/ExternalResultImpl.java
Outdated
Show resolved
Hide resolved
backend/src/main/java/com/bakdata/conquery/models/query/ExecutionManager.java
Outdated
Show resolved
Hide resolved
backend/src/main/java/com/bakdata/conquery/models/query/ExternalStateImpl.java
Show resolved
Hide resolved
…onization' into feature/extract-execution-synchronization # Conflicts: # backend/src/main/java/com/bakdata/conquery/models/worker/Namespace.java
…re/extract-execution-synchronization # Conflicts: # backend/src/main/java/com/bakdata/conquery/mode/local/LocalNamespaceHandler.java # backend/src/main/java/com/bakdata/conquery/models/query/preview/EntityPreviewExecution.java # backend/src/main/java/com/bakdata/conquery/sql/conquery/SqlExecutionManager.java # backend/src/test/java/com/bakdata/conquery/integration/json/FormTest.java # backend/src/test/java/com/bakdata/conquery/io/result/ResultTestUtil.java # backend/src/test/java/com/bakdata/conquery/io/result/excel/ExcelResultRenderTest.java
|
|
||
| log.info("Executing Query[{}] in Dataset[{}]", execution.getQueryId(), execution.getDataset()); | ||
|
|
||
| addState(execution.getId(), new DistributedState(new ConcurrentHashMap<>(), new CountDownLatch(1))); |
There was a problem hiding this comment.
kannst du das nicht in einen default Constructor oder creator von DistributedState packen?
| public ExecutionState awaitDone(ManagedExecution execution, int time, TimeUnit unit) { | ||
| ManagedExecutionId id = execution.getId(); | ||
| ExecutionState state = execution.getState(); | ||
| if (state != ExecutionState.RUNNING) { | ||
| return state; | ||
| } | ||
|
|
||
| State result = executionStates.getIfPresent(id); | ||
|
|
||
| if (result == null) { | ||
| throw new IllegalStateException("Execution is running, but no result is registered"); | ||
| } | ||
| Uninterruptibles.awaitUninterruptibly(result.getExecutingLock(), time, unit); | ||
|
|
||
| return execution.getState(); | ||
| } |
Next step towards caching: remove synchronization object from persistable ManagedExecution to the ExecutionManager.