dbsp_adapters: Specify max step size as per-thread value for Nexmark input.

blp · blp · commit 057758fab66a · 2024-10-21T20:32:04.000Z
This is a little easier to dimension than the maximum total step size, at
least if the number of generator threads matches the number of worker
threads as `run-nexmark.sh` does.

Signed-off-by: Ben Pfaff &lt;blp@feldera.com&gt;
diff --git a/benchmark/feldera-sql/benchmarks/nexmark/table.sql b/benchmark/feldera-sql/benchmarks/nexmark/table.sql
@@ -20,8 +20,8 @@ CREATE TABLE person (
                 "options": {{
                     "events": {events},
                     "threads": {cores},
-                    "batch_size_per_thread": {batchsize},
-                    "max_step_size": {batchsize}
+                    "batch_size_per_thread": 1000,
+                    "max_step_size_per_thread": 10000
                 }}
             }}
         }}
diff --git a/crates/adapters/src/transport/nexmark.rs b/crates/adapters/src/transport/nexmark.rs
@@ -119,8 +119,8 @@ impl InputReader for InputGenerator {
         let consumable_batches = self.inner.consumable_batches.load(Ordering::Acquire);
         let options = self.inner.options.get().unwrap();
         let max_batches = options
-            .max_step_size
-            .div_ceil(options.batch_size_per_thread * options.threads as u64);
+            .max_step_size_per_thread
+            .div_ceil(options.batch_size_per_thread);
 
         let mut total = 0;
         for _ in 0..max_batches {
diff --git a/crates/feldera-types/src/transport/nexmark.rs b/crates/feldera-types/src/transport/nexmark.rs
@@ -55,12 +55,14 @@ pub struct NexmarkInputOptions {
     /// batches of size `threads × batch_size_per_thread`.
     pub batch_size_per_thread: u64,
 
-    /// Maximum number of events to submit in a single step.  This should be a
-    /// multiple of `batch_size`.
+    /// Maximum number of events to submit in a single step, per thread.
+    ///
+    /// This should really be per worker thread, not per generator thread, but
+    /// the connector does not know how many worker threads there are.
     ///
     /// This stands in for `max_batch_size` from the connector configuration
     /// because it must be a constant across all three of the nexmark tables.
-    pub max_step_size: u64,
+    pub max_step_size_per_thread: u64,
 }
 
 impl Default for NexmarkInputOptions {
@@ -69,7 +71,7 @@ impl Default for NexmarkInputOptions {
             events: 100_000_000,
             threads: 4,
             batch_size_per_thread: 1_000,
-            max_step_size: 4_000_000,
+            max_step_size_per_thread: 10_000,
         }
     }
 }
diff --git a/openapi.json b/openapi.json
@@ -3058,11 +3058,11 @@
             "default": 100000000,
             "minimum": 0
           },
-          "max_step_size": {
+          "max_step_size_per_thread": {
             "type": "integer",
             "format": "int64",
-            "description": "Maximum number of events to submit in a single step.  This should be a\nmultiple of `batch_size`.\n\nThis stands in for `max_batch_size` from the connector configuration\nbecause it must be a constant across all three of the nexmark tables.",
-            "default": 4000000,
+            "description": "Maximum number of events to submit in a single step, per thread.\n\nThis should really be per worker thread, not per generator thread, but\nthe connector does not know how many worker threads there are.\n\nThis stands in for `max_batch_size` from the connector configuration\nbecause it must be a constant across all three of the nexmark tables.",
+            "default": 10000,
             "minimum": 0
           },
           "threads": {
diff --git a/web-console/src/lib/services/manager/schemas.gen.ts b/web-console/src/lib/services/manager/schemas.gen.ts
@@ -1361,15 +1361,17 @@ batches of size \`threads × batch_size_per_thread\`.`,
       default: 100000000,
       minimum: 0
     },
-    max_step_size: {
+    max_step_size_per_thread: {
       type: 'integer',
       format: 'int64',
-      description: `Maximum number of events to submit in a single step.  This should be a
-multiple of \`batch_size\`.
+      description: `Maximum number of events to submit in a single step, per thread.
+
+This should really be per worker thread, not per generator thread, but
+the connector does not know how many worker threads there are.
 
 This stands in for \`max_batch_size\` from the connector configuration
 because it must be a constant across all three of the nexmark tables.`,
-      default: 4000000,
+      default: 10000,
       minimum: 0
     },
     threads: {
diff --git a/web-console/src/lib/services/manager/types.gen.ts b/web-console/src/lib/services/manager/types.gen.ts
@@ -950,13 +950,15 @@ export type NexmarkInputOptions = {
    */
   events?: number
   /**
-   * Maximum number of events to submit in a single step.  This should be a
-   * multiple of `batch_size`.
+   * Maximum number of events to submit in a single step, per thread.
+   *
+   * This should really be per worker thread, not per generator thread, but
+   * the connector does not know how many worker threads there are.
    *
    * This stands in for `max_batch_size` from the connector configuration
    * because it must be a constant across all three of the nexmark tables.
    */
-  max_step_size?: number
+  max_step_size_per_thread?: number
   /**
    * Number of event generator threads.
    *

Original file line number	Diff line number	Diff line change
`@@ -55,12 +55,14 @@ pub struct NexmarkInputOptions {`
`55`	`55`	/// batches of size `threads × batch_size_per_thread`.
`56`	`56`	`pub batch_size_per_thread: u64,`
`57`	`57`
`58`		`- /// Maximum number of events to submit in a single step. This should be a`
`59`		- /// multiple of `batch_size`.
	`58`	`+ /// Maximum number of events to submit in a single step, per thread.`
	`59`	`+ ///`
	`60`	`+ /// This should really be per worker thread, not per generator thread, but`
	`61`	`+ /// the connector does not know how many worker threads there are.`
`60`	`62`	`///`
`61`	`63`	/// This stands in for `max_batch_size` from the connector configuration
`62`	`64`	`/// because it must be a constant across all three of the nexmark tables.`
`63`		`- pub max_step_size: u64,`
	`65`	`+ pub max_step_size_per_thread: u64,`
`64`	`66`	`}`
`65`	`67`
`66`	`68`	`impl Default for NexmarkInputOptions {`
`@@ -69,7 +71,7 @@ impl Default for NexmarkInputOptions {`
`69`	`71`	`events: 100_000_000,`
`70`	`72`	`threads: 4,`
`71`	`73`	`batch_size_per_thread: 1_000,`
`72`		`- max_step_size: 4_000_000,`
	`74`	`+ max_step_size_per_thread: 10_000,`
`73`	`75`	`}`
`74`	`76`	`}`
`75`	`77`	`}`