Add operation run store methods by samdgupi · Pull Request #15320 · cdapio/cdap

samdgupi · 2023-09-13T18:00:47Z

Added

create run
update run status
update run metadata
fail run
get run
scan run

samdgupi · 2023-09-29T17:35:29Z

+   *     indexed column, or the type does not match the schema
+   * @throws IOException if there is an error scanning the table
+   */
+  default CloseableIterator<StructuredRow> scan(Range keyRange, int limit,


Currently only implemented in PostgresSql. Will handle nosql in a seperate PR

this isn't needed anymore right?

albertshau · 2023-10-03T22:47:06Z

+
+  @VisibleForTesting
+  // USE ONLY IN TESTS: WILL DELETE ALL OPERATION RUNS
+  public void deleteAllTables() throws IOException {


methods like this should be package private

a better name would be clear() or clearData(). This name makes it sound like the table will be dropped.

albertshau · 2023-10-03T22:51:41Z

+    );
+  }
+
+  private long getStartTime(OperationRunId runId)


Methods like this can lead to unnecessary DB reads over time, as callers may do a normal read and then call this method. We should remove this and force callers to get the start time from the row.

Done. Moved the logic directly to caller as there is only one usage

albertshau · 2023-10-03T22:53:31Z

+    return fields;
+  }
+
+  private OperationRunDetail<?> getCurrentRunDetail(OperationRunId runId)


this should be getRunDetail. The current makes it sound like there could be a previous run with the same id.

Also, I'm not sure how this would work with the <?>. What does it get deserialized to?

The generic type T refers to the type of operation request object. It is not needed for most of the cases when we retrieve the OperationRunDetail except when we plan to launch the operation. The current approach I am taking is based on the operation type the request object type will be determined and the OperationRunDetail will be casted accordingly.

I'm still confused how this will work, doesn't the type need to be passed into gson in order for it to be deserialized properly? If you just deserialize it here as ?, don't you get a Map or a JsonObject or something as the field? I don't think you'd be able to cast it directly.

albertshau · 2023-10-03T22:54:43Z

+
+  private Collection<Field<?>> getFilterIndexes(OperationRunFilter filter) {
+    Collection<Field<?>> filterIndexes = new ArrayList<>();
+    if (filter == null) {


don't need to handle his case, we should not allow the filter instance to be null

filter null represents no filter applied. I have added a new method in OperationRunFilter to get a empty filter to avoid a null check

albertshau · 2023-10-03T23:14:14Z

+   * @throws OperationRunNotFoundException run with id does not exist in namespace
+   */
+  public OperationRunDetail<?> getOperation(OperationRunId runId)
+      throws NotFoundException, IOException {


throws OperationRunNotFoundException

albertshau · 2023-10-03T23:17:00Z

+        Fields.stringField(StoreDefinition.OperationRunsStore.ID_FIELD, runId.getRun()));
+    fields.add(
+        Fields.longField(StoreDefinition.OperationRunsStore.UPDATE_TIME_FIELD,
+            System.currentTimeMillis()));


instead of System.currentTimeMillis(), it would be good to take a Clock in the constructor of this class and use that instead. It would make unit testing simpler, as you can expect a concrete time.

albertshau · 2023-10-03T23:19:24Z

+   * @param limit number of result
+   * @return a select query
+   */
+  private PreparedStatement prepareMultiScanQuery(Collection<Range> singletonRanges,


are there any logical changes being made here or is it just getting moved? (In the future, keep methods in the same location so it is easier to review the diff)

This was done to fix a checkstyle warning about keeping all the override methods together. There is no logical change. I will revert it back to reduce the diff

I see, if it is fixing checkstyle it's ok to move it.

albertshau · 2023-10-03T23:19:52Z

+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package io.cdap.cdap.proto.id;


fix checkstyle

albertshau · 2023-10-03T23:20:01Z

@@ -0,0 +1,90 @@
+/*
+ * Copyright © 2015-2019 Cask Data, Inc.


fix copyright year

samdgupi · 2023-10-04T17:41:13Z

+   * @param txBatchSize batch size of transaction
+   * @param consumer {@link Consumer} to process each scanned run
+   */
+  public boolean scanOperations(ScanOperationRunsRequest request, int txBatchSize,


Unit test will be added in a followup PR. Will add a TODO

albertshau · 2023-10-06T03:16:41Z

+
+    AtomicReference<ScanOperationRunsRequest> requestRef = new AtomicReference<>(request);
+    AtomicReference<String> lastKey = new AtomicReference<>(request.getScanAfter());
+    AtomicInteger currentLimit = new AtomicInteger(request.getLimit());


this looks like it can be a normal int

albertshau · 2023-10-06T03:17:18Z

+   *
+   * @param request scan request including filters and limit
+   * @param txBatchSize batch size of transaction
+   * @param consumer {@link Consumer} to process each scanned run


javadoc should have @return as well

though returning whether the limit was reached is kind of a weird thing to return, not sure how it would be used. I would either make it void or return the total number of operations scanned.

The boolean would be used in the handler to find if all the operations are scanned and send last operation id accordingly. We can use the total number of operation scanned also but wanted to keep this analogous to the application scan

see:

cdap/cdap-app-fabric/src/main/java/io/cdap/cdap/gateway/handlers/AppLifecycleHttpHandler.java

Line 307 in 550a8f3

return !pageLimitReached || record == null ? null :

hm I see, that's ok then

albertshau · 2023-10-06T03:24:52Z

+      requestRef.set(batchRequest);
+
+      TransactionRunners.run(transactionRunner, context -> {
+        lastKey.set(


you can use the run method with a return value instead of setting lastKey in the closure. Then it can be a normal String instead of an AtomicReference.

albertshau · 2023-10-06T03:26:30Z

+
+  private static final String SMALLEST_POSSIBLE_STRING = "";
+
+  @Inject


don't think we would use injection to create the store

albertshau · 2023-10-06T03:27:23Z

+  }
+
+  @VisibleForTesting
+    // USE ONLY IN TESTS: WILL DELETE ALL OPERATION RUNS


albertshau · 2023-10-06T03:28:41Z

+    Range range = Range.create(endFields, endBound, startFields, startBound);
+    Collection<Field<?>> filterIndexes = getFilterIndexes(request.getFilter());
+    StructuredTable table = getOperationRunsTable(context);
+    AtomicReference<String> lastKey = new AtomicReference<>();


this can be a normal String

albertshau · 2023-10-06T03:30:26Z

+  /**
+   * namespace to return applications for or null for all namespaces.
+   */
+  @Nullable


this should not be nullable

albertshau · 2023-10-06T03:30:36Z

+  }
+
+  /**
+   * namespace to return applications for or null for all namespaces.


update javadoc to not include null

albertshau · 2023-10-06T03:31:53Z

+  public void testScanOperation() throws Exception {
+    insertTestRuns();
+
+    // TODO(samik) verify the actual list


should do this now rather than later

albertshau · 2023-10-06T03:45:15Z

+          "is not an indexed column or primary key");
+    }
+    return scan(keyRange, limit, filterIndexes, Collections.singleton(orderByField), sortOrder,
+        true);


It is really confusing that this scan method is an 'and' filter while the other ones are an 'or' filter. Can we just make the operation type part of the primary key? That way it can go in the key range and we can use the existing methods without adding more to StructuredTable. Would like to avoid bloating the StructuredTable class, it's already pretty confusing as-is.

FYI, whenever we go to CDAP 7.0, we would like to remove Hadoop support, which would allow us to remove the NoSql implementations which would let us use normal SQL instead of this interface.

The reason operationtype is not part of the primary key because the operation id and namespace should uniquely identify a operation. This is in line with the API design /namespace/{namespace}/operation/{id}. If we add the operationtype in primary key we have to change the API also.
The API is designed to be compatible with GCP operations api which also does not include types.
I think we do need a option to select how the filters should be joined. I can create two different top level functions for AND and OR filter.

Removed and filter and added operation type in the range key

albertshau · 2023-10-10T03:41:00Z

+    }
+    return GSON.fromJson(
+        row.get().getString(StoreDefinition.OperationRunsStore.DETAILS_FIELD),
+        OperationRunDetail.class


I'm still not sure how this will work without a concrete class for the generic. Maybe it will be more clear when there is code around the caller using the result, but I get the feeling it will need to revisited.

albertshau · 2023-10-10T03:42:51Z

+   *     indexed column, or the type does not match the schema
+   * @throws IOException if there is an error scanning the table
+   */
+  default CloseableIterator<StructuredRow> scan(Range keyRange, int limit,


this isn't needed anymore right?

albertshau

lgtm assuming the new scan method is removed from StructuredTable

chtyim · 2023-10-10T18:54:28Z

+  private final TransactionRunner transactionRunner;
+
+  @Inject
+  public OperationLifecycleService(TransactionRunner transactionRunner) {


Remove public for injected constructor to avoid direct instantiation.

chtyim · 2023-10-10T18:55:36Z

+/**
+ * Service that manages lifecycle of Operation.
+ */
+public class OperationLifecycleService {


Usually we only name a class with Service suffix if it is a guava Service. This class seems more like a Manager?

We do have classes with suffix service like ProgramLifecycleService which is not a guava service but encapsulates all operations we can do upon an entity . I was following the apparent naming convention.

Updated to use Manager Suffix

chtyim · 2023-10-10T18:56:56Z

+   */
+  public boolean scanOperations(ScanOperationRunsRequest request, int txBatchSize,
+      Consumer<OperationRunDetail<?>> consumer) throws OperationRunNotFoundException, IOException {
+    AtomicReference<ScanOperationRunsRequest> requestRef = new AtomicReference<>(request);


Why we need an atomic reference? Seems like everything is happening in the local thread? A simple local variable would do the job.

chtyim · 2023-10-10T18:58:55Z

+      }
+      currentLimit -= txBatchSize;
+    }
+    return currentLimit == 0;


If the above loop having lastKey == null, the while loop would break and currentLimit may not be 0. Is this expected? If the caller is also having a while loop until this method return false, would it resulted in infinite loop?

It is expected to break the loop as lastKey == null will signify no more row left to scan.
The return value here will signify if the page limit was reached. The caller would also need to track the last record scanned as the following example in AppLifeCycleHttpHandler

boolean pageLimitReached = applicationLifecycleService.scanApplications(scanRequest, appDetail -> { ApplicationRecord record = new ApplicationRecord(appDetail); jsonListResponder.send(record); lastRecord.set(record); }); ApplicationRecord record = lastRecord.get(); return !pageLimitReached || record == null ? null : record.getName() + EntityId.IDSTRING_PART_SEPARATOR + record.getAppVersion();

The caller is requierd to update the ScanAfter field in the request for the next call if this method is called from a while loop

chtyim · 2023-10-10T18:59:05Z

+  private OperationRunsStore getOperationRunStore(StructuredTableContext context) {
+    return new OperationRunsStore(context);
+  }
+


Remove extra new lines.

chtyim · 2023-10-10T19:00:15Z

+/**
+ * This class defined various filters that can be applied during operation runs scanning.
+ */
+public class OperationRunFilter {


Usually a Filter is kind of a Function instead of just having getters. How is it supposed to be used?

It is just a container class to encapsulate what filter options are available for operations.

chtyim · 2023-10-10T19:00:36Z

+/**
+ * Store for operation runs.
+ */
+public class OperationRunsStore {


No plural in class name.

chtyim · 2023-10-10T19:02:12Z

+
+  private final Clock clock;
+
+  private static final String SMALLEST_POSSIBLE_STRING = "";


group static fields together

chtyim · 2023-10-10T19:05:40Z

+
+  private void writeOperationRun(OperationRunId runId, OperationRunDetail<?> detail)
+      throws IOException {
+    Collection<Field<?>> fields = new ArrayList<>();


Consider using Arrays.asList or ImmutableList.of to improve readability.

chtyim · 2023-10-10T19:07:18Z

+  private OperationRunDetail<?> getRunDetail(OperationRunId runId)
+      throws IOException, OperationRunNotFoundException {
+    Optional<StructuredRow> row = getOperationRunInternal(runId);
+    if (!row.isPresent()) {


Avoid using isPresent on Optional. In this case, use .map and orElseThrow.

return getOperationRunInternal(runId) .map(this::rowToRunDetail) .orElseThrow(() -> new OperationRunNotFoundException(...));

Please also fix other places that use isPresent and replace with more expressive way.

Updated. There is one case where we throw error if the row is present. As we can't throw checked error from within ifPresent I have not made any change

samdgupi · 2023-10-11T13:02:06Z

@chtyim Merging this PR to unblock followup PRs for LRO. Please add comments for any suggested change and I will raise a further PR to solve.

samdgupi changed the base branch from develop to operation-store September 13, 2023 18:10

samdgupi force-pushed the operation-store branch 5 times, most recently from 5c8c266 to 1941854 Compare September 21, 2023 15:35

Base automatically changed from operation-store to develop September 22, 2023 06:54

samdgupi force-pushed the operation-store-apis branch 4 times, most recently from 93d5151 to 52f7e9d Compare September 26, 2023 07:24

samdgupi added the build Triggers github actions build label Sep 26, 2023

samdgupi force-pushed the operation-store-apis branch from 52f7e9d to 03beee6 Compare September 26, 2023 12:45

samdgupi requested review from GnsP and albertshau September 26, 2023 12:46

samdgupi marked this pull request as ready for review September 26, 2023 12:46

samdgupi force-pushed the operation-store-apis branch 2 times, most recently from 23a35de to 3d30d35 Compare September 27, 2023 06:35

samdgupi requested a review from tivv September 28, 2023 08:52

samdgupi force-pushed the operation-store-apis branch from 3d30d35 to 18300f7 Compare September 29, 2023 17:34

samdgupi commented Sep 29, 2023

View reviewed changes

albertshau reviewed Oct 3, 2023

View reviewed changes

samdgupi force-pushed the operation-store-apis branch 2 times, most recently from 73ae585 to ce08626 Compare October 4, 2023 17:35

samdgupi commented Oct 4, 2023

View reviewed changes

samdgupi force-pushed the operation-store-apis branch from ce08626 to cb66ab8 Compare October 4, 2023 17:45

samdgupi requested a review from albertshau October 5, 2023 08:21

albertshau reviewed Oct 6, 2023

View reviewed changes

samdgupi force-pushed the operation-store-apis branch 2 times, most recently from af6aa13 to 57a1c6e Compare October 6, 2023 16:56

samdgupi force-pushed the operation-store-apis branch 3 times, most recently from df24aec to cf0f87b Compare October 9, 2023 17:39

samdgupi requested a review from albertshau October 9, 2023 17:40

albertshau reviewed Oct 10, 2023

View reviewed changes

albertshau approved these changes Oct 10, 2023

View reviewed changes

samdgupi force-pushed the operation-store-apis branch from cf0f87b to 26ad907 Compare October 10, 2023 18:38

chtyim reviewed Oct 10, 2023

View reviewed changes

Add operation store methods

6ef10d8

samdgupi force-pushed the operation-store-apis branch from 26ad907 to 6ef10d8 Compare October 11, 2023 11:01

samdgupi merged commit 30a1b67 into develop Oct 11, 2023

samdgupi deleted the operation-store-apis branch October 11, 2023 13:02

samdgupi added the LRO label for tracking Long Running Operation support label Nov 30, 2023


		private static final String SMALLEST_POSSIBLE_STRING = "";

		@Inject


		private final Clock clock;

		private static final String SMALLEST_POSSIBLE_STRING = "";

Conversation

samdgupi commented Sep 13, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

samdgupi Oct 10, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

albertshau Oct 5, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

samdgupi Oct 4, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

samdgupi commented Sep 13, 2023 •

edited

Loading

samdgupi Oct 10, 2023 •

edited

Loading

albertshau Oct 5, 2023 •

edited

Loading

samdgupi Oct 4, 2023 •

edited

Loading