Implement PostgresComputationStatsService

[YuhongWang-Amazon]

This implements the PostgresComputationStatsService and partial PostgresComputationsService that is enough to run the PostgresComputationStatsServiceTest

[wfa-reviewable]

This change is 

[renjiezh]

Reviewed 9 of 13 files at r1, all commit messages.
Reviewable status: 9 of 13 files reviewed, 3 unresolved discussions (waiting on @YuhongWang-Amazon)

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/PostgresComputationsService.kt line 100 at r3 (raw file):

        ComputationReader(protocolStageEnumHelper)
          .readComputationToken(client.singleUse(), request.globalComputationId)
          ?: failGrpc(Status.UNKNOWN) { "Created computation not found." }

Status.NOT_FOUND might be more accurate here.

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/PostgresComputationsService.kt line 179 at r3 (raw file):

    attempt: Long = 0L
  ): CreateComputationLogEntryRequest {
    return CreateComputationLogEntryRequest.newBuilder()

Let's get rid of builders and use dsl.

---

src/test/kotlin/org/wfanet/measurement/duchy/deploy/postgres/PostgresComputationStatsServiceTest.kt line 74 at r3 (raw file):

      client,
      idGenerator,
      ALSACE,

Can we denote the parameter name like computationTypeEnumHelper = ComputationTypes,
Besides, do we need a fixed duchy name here?

[YuhongWang-Amazon]

Reviewable status: 8 of 13 files reviewed, 3 unresolved discussions (waiting on @renjiezh and @SanjayVas)

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/PostgresComputationsService.kt line 100 at r3 (raw file):

Previously, renjiezh wrote…

Status.NOT_FOUND might be more accurate here.

I was thinking about the NOT_FOUND as well. But exception will be raised from the createComputation request and after a computation that was supposed to be successfully created in a previous command. Hence, it is a bit weird to raise a NOT_FOUND exception here.
I decide to use UNKOWN here, because I don't know in which scenario this error could occur.

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/PostgresComputationsService.kt line 179 at r3 (raw file):

Previously, renjiezh wrote…

Let's get rid of builders and use dsl.

Done.

---

src/test/kotlin/org/wfanet/measurement/duchy/deploy/postgres/PostgresComputationStatsServiceTest.kt line 74 at r3 (raw file):

Previously, renjiezh wrote…

Can we denote the parameter name like computationTypeEnumHelper = ComputationTypes,
Besides, do we need a fixed duchy name here?

Added the parameter names.

For duchy name, it is used when sending update to kingdom. But it was never verified in any of the integration tests. So either a fixed string or a randomly generated name could work. Do you have any preference?

[SanjayVas]

Reviewed 4 of 13 files at r1, 1 of 4 files at r4, 1 of 4 files at r5.
Reviewable status: 9 of 16 files reviewed, 12 unresolved discussions (waiting on @renjiezh and @YuhongWang-Amazon)

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/PostgresComputationStatsService.kt line 36 at r4 (raw file):

    val localComputationId = request.localComputationId
    val metricName = request.metricName
    grpcRequire(localComputationId != 0L) { "Missing computation ID" }

nit: Reference proto field names in these error messages

Suggestion:

local_computation_id

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/PostgresComputationsService.kt line 100 at r3 (raw file):

Previously, YuhongWang-Amazon wrote…

I was thinking about the NOT_FOUND as well. But exception will be raised from the createComputation request and after a computation that was supposed to be successfully created in a previous command. Hence, it is a bit weird to raise a NOT_FOUND exception here.
I decide to use UNKOWN here, because I don't know in which scenario this error could occur.

INTERNAL might be a better fit here, as this is indicating a violation of a system invariant

Ideally this would all be in one transaction so it couldn't possibly happen.

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/PostgresComputationsService.kt line 83 at r4 (raw file):

  ): CreateComputationResponse {
    grpcRequire(request.globalComputationId.isNotEmpty()) {
      "globalComputationId is not specified."

nit: protobuf field name rather than language-dependent version

Suggestion:

global_computation_id 

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/PostgresComputationsService.kt line 86 at r4 (raw file):

    }

    return try {

nit: keep the try around the narrowest block that might throw that exception

e.g.

val result: ComputationReader.ReadComputationTokenResult = try {
  // ... create computation ...

  ComputationReader(protocolStageEnumHelper)
    .readComputationToken(client.singleUse(), request.globalComputationId)
    ?: failGrpc(Status.UNKNOWN) { "Created computation not found." }

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/PostgresComputationsService.kt line 140 at r4 (raw file):

          else -> throw ex
        }
      }

It's more idiomatic to use catch blocks for the specific exception types, even if there's a tiny amount of repetition.

That said, don't we expect the stage attempt details to exist in this case? Maybe that condition should be an INTERNAL, or not even have a corresponding DuchyInternalException that we catch (e.g. just IllegalStateException).

Suggestion:

      } catch (e: ComputationNotFoundException) {
        throw e.asStatusRuntimeException(Status.Code.NOT_FOUND)
      } catch (e: ComputationDetailsNotFoundException) {
        throw e.asStatusRuntimeException(Status.Code.INTERNAL)
      }

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/readers/ComputationReader.kt line 55 at r4 (raw file):

  /** A wrapper data class for ComputationToken query result */
  data class ReadComputationTokenResult(val computationToken: ComputationToken)

nit: we'd primarily been using Result wrappers when we've needed additional fields, e.g. returning an internal ID.

If you want to keep the wrapper, maybe rename it since it should always be referenced in the context of its enclosing type.

Suggestion:

Result

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/readers/ComputationReader.kt line 57 at r4 (raw file):

  data class ReadComputationTokenResult(val computationToken: ComputationToken)

  private fun buildComputationToken(row: ResultRow) =

Expression functions are only really cases for when it's either a one-liner or a block builder (e.g. runBlocking, coroutineScope, flow). Convert this to a block body instead. This will also allow you to pull the intermediate local variables out of the DSL builder scope.

Code quote:

 =

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/readers/ComputationReader.kt line 157 at r4 (raw file):

        cs.Details AS StageDetails,
        (
          SELECT ARRAY_AGG(

Note that we still don't know whether this is a good way to do this. For background, this is trying to emulate the Spanner SELECT AS STRUCT subquery which has additional support from the Spanner Java client library to easily parse arrays of structs (it uses the same data structure for a top-level row as for a struct subquery element). See the issue I filed with r2dbc-postgresql for additional context.

I believe the more "standard" way for other SQL dialects is to either do multiple queries in the same transaction or do JOINs (depending on whether the priority is to reduce redundant data in the response vs. reduce number of queries).

It's not strictly necessary to do it differently now, just something I wanted to point out as a potential issue (e.g. if using JSON arrays like this leads to some other problems such as parsing performance or potentially hitting a result row size limit).

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/readers/ComputationReader.kt line 256 at r4 (raw file):

  }

  /**

Drop this since you've now got the function from common-jvm.

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/writers/ClaimWork.kt line 63 at r4 (raw file):

  ComputationProtocolStageDetailsHelper<
    ProtocolT, StageT, StageDT, ComputationDT
  > by computationProtocolStageDetailsHelper {

I don't think you need to copy this from the Spanner impl. AFAICT this is a hack to get all the "helper" functions into the class scope, which is entirely unnecessary.

We had previously discussed how there are a lot of IMO bad abstractions in the DB layer. I personally would consider this a great opportunity to implement new abstractions and not use anything in theorg.wfanet.measurement.duchy.db.computation package, but I understand if you don't want to tackle that.

Code quote:

  ComputationTypeEnumHelper<ProtocolT> by computationTypeEnumHelper,
  ComputationProtocolStagesEnumHelper<ProtocolT, StageT> by computationProtocolStagesEnumHelper,
  ComputationProtocolStageDetailsHelper<
    ProtocolT, StageT, StageDT, ComputationDT
  > by computationProtocolStageDetailsHelper {

[YuhongWang-Amazon]

Reviewable status: 8 of 17 files reviewed, 9 unresolved discussions (waiting on @renjiezh and @SanjayVas)

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/PostgresComputationsService.kt line 100 at r3 (raw file):

Previously, SanjayVas (Sanjay Vasandani) wrote…

INTERNAL might be a better fit here, as this is indicating a violation of a system invariant

Ideally this would all be in one transaction so it couldn't possibly happen.

Done.

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/PostgresComputationsService.kt line 140 at r4 (raw file):

Previously, SanjayVas (Sanjay Vasandani) wrote…

It's more idiomatic to use catch blocks for the specific exception types, even if there's a tiny amount of repetition.

That said, don't we expect the stage attempt details to exist in this case? Maybe that condition should be an INTERNAL, or not even have a corresponding DuchyInternalException that we catch (e.g. just IllegalStateException).

IllegalStatException is probably better. Since this error is for the scenario when someone is holding the computation lock but there is NO computation stage details associated with it.

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/readers/ComputationReader.kt line 55 at r4 (raw file):

Previously, SanjayVas (Sanjay Vasandani) wrote…

nit: we'd primarily been using Result wrappers when we've needed additional fields, e.g. returning an internal ID.

If you want to keep the wrapper, maybe rename it since it should always be referenced in the context of its enclosing type.

Removed the wrapper.

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/readers/ComputationReader.kt line 57 at r4 (raw file):

Previously, SanjayVas (Sanjay Vasandani) wrote…

Expression functions are only really cases for when it's either a one-liner or a block builder (e.g. runBlocking, coroutineScope, flow). Convert this to a block body instead. This will also allow you to pull the intermediate local variables out of the DSL builder scope.

Done.

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/readers/ComputationReader.kt line 157 at r4 (raw file):

Previously, SanjayVas (Sanjay Vasandani) wrote…

Note that we still don't know whether this is a good way to do this. For background, this is trying to emulate the Spanner SELECT AS STRUCT subquery which has additional support from the Spanner Java client library to easily parse arrays of structs (it uses the same data structure for a top-level row as for a struct subquery element). See the issue I filed with r2dbc-postgresql for additional context.

I believe the more "standard" way for other SQL dialects is to either do multiple queries in the same transaction or do JOINs (depending on whether the priority is to reduce redundant data in the response vs. reduce number of queries).

It's not strictly necessary to do it differently now, just something I wanted to point out as a potential issue (e.g. if using JSON arrays like this leads to some other problems such as parsing performance or potentially hitting a result row size limit).

Thanks for the context.
I think this JSON approach is not very elegant; especially that extra step of encoding and decoding of the byte array data.
I updated this reader into multiple queries. Please help take another look.

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/readers/ComputationReader.kt line 256 at r4 (raw file):

Previously, SanjayVas (Sanjay Vasandani) wrote…

Drop this since you've now got the function from common-jvm.

Done.

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/writers/ClaimWork.kt line 63 at r4 (raw file):

Previously, SanjayVas (Sanjay Vasandani) wrote…

I don't think you need to copy this from the Spanner impl. AFAICT this is a hack to get all the "helper" functions into the class scope, which is entirely unnecessary.

We had previously discussed how there are a lot of IMO bad abstractions in the DB layer. I personally would consider this a great opportunity to implement new abstractions and not use anything in theorg.wfanet.measurement.duchy.db.computation package, but I understand if you don't want to tackle that.

I am interested into creating a better abstractions for the computation data. I looked into it a bit, but found it quite complicated to rebuild all those. Hence I think it would be better to create a design doc before the implementation.

Since I planned to finish the implementation of all Duchy Postgres services by the end of this month. I prefer to defer that work to after all duchy postgres service implementations.

[SanjayVas]

Reviewed 4 of 13 files at r1, 1 of 4 files at r4, 1 of 4 files at r5, 1 of 1 files at r6, 6 of 6 files at r7, all commit messages.
Reviewable status: all files reviewed, 6 unresolved discussions (waiting on @renjiezh and @YuhongWang-Amazon)

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/PostgresComputationsService.kt line 144 at r7 (raw file):

      )
      token.toClaimWorkResponse()
    } else ClaimWorkResponse.getDefaultInstance()

If you use braces on one part of an if block, use it on all parts. (Basically only omit braces when the whole if expression fits on one line.

Code quote:

 ClaimWorkResponse.getDefaultInstance()

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/readers/ComputationReader.kt line 106 at r7 (raw file):

  ): ComputationToken {
    return computationToken {
      this.globalComputationId = computation.globalComputationId

nit: I believe this is an unnecessary qualification since there's no other variable in scope with the same name

Code quote:

this.

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/readers/ComputationReader.kt line 249 at r7 (raw file):

    val readContext = client.readTransaction()
    try {
      val computation = readComputation(readContext, globalComputationId) ?: return null

Specify types explicitly where not obvious (e.g. not calling a constructor or clear factory/builder).

Code quote:

val computation

[YuhongWang-Amazon]

Reviewable status: 15 of 17 files reviewed, 5 unresolved discussions (waiting on @renjiezh and @SanjayVas)

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/PostgresComputationsService.kt line 144 at r7 (raw file):

Previously, SanjayVas (Sanjay Vasandani) wrote…

If you use braces on one part of an if block, use it on all parts. (Basically only omit braces when the whole if expression fits on one line.

Done.

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/readers/ComputationReader.kt line 249 at r7 (raw file):

Previously, SanjayVas (Sanjay Vasandani) wrote…

Specify types explicitly where not obvious (e.g. not calling a constructor or clear factory/builder).

Done.

[SanjayVas]

Reviewed 2 of 2 files at r8, all commit messages.
Reviewable status: all files reviewed, 3 unresolved discussions (waiting on @renjiezh)

[YuhongWang-Amazon]

Reviewed 7 of 13 files at r1, 1 of 4 files at r4, 1 of 4 files at r5, 1 of 1 files at r6, 4 of 6 files at r7, 2 of 2 files at r8, 1 of 1 files at r9, all commit messages.
Reviewable status: all files reviewed, 3 unresolved discussions (waiting on @renjiezh)

[YuhongWang-Amazon]

Reviewed 1 of 13 files at r1.
Reviewable status: all files reviewed, 3 unresolved discussions (waiting on @renjiezh)

---

src/test/kotlin/org/wfanet/measurement/duchy/deploy/postgres/PostgresComputationStatsServiceTest.kt line 74 at r3 (raw file):

Previously, YuhongWang-Amazon wrote…

Added the parameter names.

For duchy name, it is used when sending update to kingdom. But it was never verified in any of the integration tests. So either a fixed string or a randomly generated name could work. Do you have any preference?

@renjiezh do you have any other thoughts on this? If not, could you help resolve the feedbacks? Appreciated!

[renjiezh]

Reviewed 1 of 4 files at r4, 1 of 4 files at r5, 1 of 1 files at r6, 4 of 6 files at r7, 2 of 2 files at r8, 1 of 1 files at r9.
Reviewable status: all files reviewed, 1 unresolved discussion (waiting on @YuhongWang-Amazon)

---

src/main/kotlin/org/wfanet/measurement/duchy/deploy/postgres/PostgresComputationsService.kt line 134 at r9 (raw file):

      val token =
        ComputationReader(protocolStageEnumHelper).readComputationToken(client, claimed)
          ?: failGrpc(Status.UNKNOWN) { "Claimed computation $claimed not found." }

nit: As we discussed in another thread, "INTERNAL" or "NOT_FOUND" might be better.

---

src/test/kotlin/org/wfanet/measurement/duchy/deploy/postgres/PostgresComputationStatsServiceTest.kt line 74 at r3 (raw file):

Previously, YuhongWang-Amazon wrote…

@renjiezh do you have any other thoughts on this? If not, could you help resolve the feedbacks? Appreciated!

Sounds good.

[YuhongWang-Amazon]

Reviewed 1 of 1 files at r10, 1 of 1 files at r11, all commit messages.
Reviewable status: complete! all files reviewed, all discussions resolved (waiting on @YuhongWang-Amazon)

[YuhongWang-Amazon]

Reviewed 1 of 1 files at r12, all commit messages.
Reviewable status: complete! all files reviewed, all discussions resolved (waiting on @YuhongWang-Amazon)