DPP-368 clean up flags (digital-asset#10711)

* Make batching parallelism configurable changelog_begin changelog_end * Fail if incompatible cli flags are used
vasqcorp · Aug 31, 2021 · f058c2f · f058c2f
1 parent 90ad24f
commit f058c2f
Show file tree

Hide file tree

Showing 2 changed files with 48 additions and 7 deletions.
diff --git a/ledger/indexer-benchmark/src/main/scala/ledger/indexerbenchmark/Config.scala b/ledger/indexer-benchmark/src/main/scala/ledger/indexerbenchmark/Config.scala
@@ -54,37 +54,42 @@ object Config {
         .action((value, config) => config.copy(updateSource = value))
 
       opt[Int]("indexer-input-mapping-parallelism")
-        .text("[TODO] Sets the corresponding indexer parameter.")
+        .text("Sets the value of IndexerConfig.inputMappingParallelism.")
         .action((value, config) =>
           config.copy(indexerConfig = config.indexerConfig.copy(inputMappingParallelism = value))
         )
       opt[Int]("indexer-ingestion-parallelism")
-        .text("[TODO] Sets the corresponding indexer parameter.")
+        .text("Sets the value of IndexerConfig.ingestionParallelism.")
         .action((value, config) =>
           config.copy(indexerConfig = config.indexerConfig.copy(ingestionParallelism = value))
         )
+      opt[Int]("indexer-batching-parallelism")
+        .text("Sets the value of IndexerConfig.batchingParallelism.")
+        .action((value, config) =>
+          config.copy(indexerConfig = config.indexerConfig.copy(batchingParallelism = value))
+        )
       opt[Long]("indexer-submission-batch-size")
-        .text("[TODO] Sets the corresponding indexer parameter.")
+        .text("Sets the value of IndexerConfig.submissionBatchSize.")
         .action((value, config) =>
           config.copy(indexerConfig = config.indexerConfig.copy(submissionBatchSize = value))
         )
       opt[Int]("indexer-tailing-rate-limit-per-second")
-        .text("[TODO] Sets the corresponding indexer parameter.")
+        .text("Sets the value of IndexerConfig.tailingRateLimitPerSecond.")
         .action((value, config) =>
           config.copy(indexerConfig = config.indexerConfig.copy(tailingRateLimitPerSecond = value))
         )
       opt[Long]("indexer-batch-within-millis")
-        .text("[TODO] Sets the corresponding indexer parameter.")
+        .text("Sets the value of IndexerConfig.batchWithinMillis.")
         .action((value, config) =>
           config.copy(indexerConfig = config.indexerConfig.copy(batchWithinMillis = value))
         )
       opt[Boolean]("indexer-enable-compression")
-        .text("[TODO] Sets the corresponding indexer parameter.")
+        .text("Sets the value of IndexerConfig.enableCompression.")
         .action((value, config) =>
           config.copy(indexerConfig = config.indexerConfig.copy(enableCompression = value))
         )
       opt[Int]("indexer-max-input-buffer-size")
-        .text("[TODO] Sets the corresponding indexer parameter.")
+        .text("Sets the value of IndexerConfig.maxInputBufferSize.")
         .action((value, config) =>
           config.copy(indexerConfig = config.indexerConfig.copy(maxInputBufferSize = value))
         )

diff --git a/...ate/kvutils/app/src/main/scala/com/daml/ledger/participant/state/kvutils/app/Config.scala b/...ate/kvutils/app/src/main/scala/com/daml/ledger/participant/state/kvutils/app/Config.scala
@@ -174,6 +174,7 @@ object Config {
               "indexer-max-input-buffer-size, " +
               "indexer-input-mapping-parallelism, " +
               "indexer-ingestion-parallelism, " +
+              "indexer-batching-parallelism, " +
               "indexer-submission-batch-size, " +
               "indexer-tailing-rate-limit-per-second, " +
               "indexer-batch-within-millis, " +
@@ -570,6 +571,41 @@ object Config {
           )
           .action((_, config) => config.copy(enableAppendOnlySchema = true))
 
+        // TODO append-only: remove after removing support for the current (mutating) schema
+        checkConfig(c => {
+          if (
+            c.enableAppendOnlySchema && c.participants.exists(
+              _.indexerConfig.databaseConnectionPoolSize != ParticipantIndexerConfig.DefaultDatabaseConnectionPoolSize
+            )
+          ) {
+            failure(
+              "The following participant setting keys are not compatible with the --index-append-only-schema flag: " +
+                "indexer-connection-pool-size."
+            )
+          } else if (
+            !c.enableAppendOnlySchema && c.participants.exists(pc =>
+              pc.indexerConfig.maxInputBufferSize != ParticipantIndexerConfig.DefaultMaxInputBufferSize ||
+                pc.indexerConfig.inputMappingParallelism != ParticipantIndexerConfig.DefaultInputMappingParallelism ||
+                pc.indexerConfig.ingestionParallelism != ParticipantIndexerConfig.DefaultIngestionParallelism ||
+                pc.indexerConfig.submissionBatchSize != ParticipantIndexerConfig.DefaultSubmissionBatchSize ||
+                pc.indexerConfig.tailingRateLimitPerSecond != ParticipantIndexerConfig.DefaultTailingRateLimitPerSecond ||
+                pc.indexerConfig.batchWithinMillis != ParticipantIndexerConfig.DefaultBatchWithinMillis
+            )
+          ) {
+            failure(
+              "The following participant setting keys can only be used together with the --index-append-only-schema flag: " +
+                "indexer-max-input-buffer-size, " +
+                "indexer-input-mapping-parallelism, " +
+                "indexer-ingestion-parallelism, " +
+                "indexer-batching-parallelism, " +
+                "indexer-submission-batch-size, " +
+                "indexer-tailing-rate-limit-per-second, " +
+                "indexer-batch-within-millis. "
+            )
+          } else
+            success
+        })
+
         opt[Unit]("mutable-contract-state-cache")
           .optional()
           .hidden()