Adds testoperator dispatch infrastructure for CH-BenCHmark (HTAP) workloads. (spiceai#10868)

sgrebnov · web-flow · commit 6964fb271b88 · 2026-05-16T11:57:44.000Z
diff --git a/.github/workflows/testoperator_dispatch.yml b/.github/workflows/testoperator_dispatch.yml
@@ -26,6 +26,7 @@ on:
           - 'text-to-sql'
           - 'streaming-bench'
           - 'streaming-correctness'
+          - 'htap'
       update_snapshots:
         description: 'Update snapshots?'
         required: false
@@ -115,6 +116,14 @@ jobs:
             "$SPICEPOD_VALIDATOR" "$file"
           done
 
+      - name: Validate spicepods - CH-BenCHmark - SF1
+        run: |
+          shopt -s globstar nullglob
+          for file in ./test/spicepods/chbench/sf1/**/*.yaml; do
+            echo "Validating $file"
+            "$SPICEPOD_VALIDATOR" "$file"
+          done
+
       - name: Validate spicepods - TPCH - SF100
         run: |
           shopt -s globstar nullglob
@@ -195,6 +204,14 @@ jobs:
           SPICED_COMMIT: ${{ steps.setup-spiced.outputs.SPICED_COMMIT }}
           WORKFLOW_COMMIT: ${{ matrix.branch }}
 
+      - name: Dispatch Testoperator - Scheduled CH-BenCHmark - SF1
+        run: |
+          testoperator dispatch ./tools/testoperator/dispatch/chbench/sf1 --workflow htap
+        env:
+          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          SPICED_COMMIT: ${{ steps.setup-spiced.outputs.SPICED_COMMIT }}
+          WORKFLOW_COMMIT: ${{ matrix.branch }}
+
   dispatch-input:
     name: Dispatch tests - Input
     runs-on: spiceai-dev-runners
@@ -266,6 +283,15 @@ jobs:
             "$SPICEPOD_VALIDATOR" "$file"
           done
 
+      - name: Validate spicepods - CH-BenCHmark - SF1
+        if: ${{ github.event.inputs.workflow_type == 'htap' }}
+        run: |
+          shopt -s globstar nullglob
+          for file in ./test/spicepods/chbench/sf1/**/*.yaml; do
+            echo "Validating $file"
+            "$SPICEPOD_VALIDATOR" "$file"
+          done
+
       - name: Validate spicepods - TPCH - SF100
         run: |
           shopt -s globstar nullglob
@@ -351,3 +377,13 @@ jobs:
           GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           SPICED_COMMIT: ${{ steps.setup-spiced.outputs.SPICED_COMMIT }}
           WORKFLOW_COMMIT: ${{ github.event.ref }}
+
+      - name: Dispatch Testoperator - ${{ github.event.inputs.workflow_type }} - CH-BenCHmark - SF1
+        run: |
+          testoperator dispatch ./tools/testoperator/dispatch/chbench/sf1 \
+            --workflow ${{ github.event.inputs.workflow_type }} \
+            --update-snapshots ${{ github.event.inputs.update_snapshots }}
+        env:
+          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          SPICED_COMMIT: ${{ steps.setup-spiced.outputs.SPICED_COMMIT }}
+          WORKFLOW_COMMIT: ${{ github.event.ref }}
diff --git a/crates/test-framework/src/lib.rs b/crates/test-framework/src/lib.rs
@@ -64,6 +64,7 @@ pub enum TestType {
     Streaming,
     StreamingCorrectness,
     Schema,
+    Htap,
 }
 
 impl TestType {
@@ -80,6 +81,7 @@ impl TestType {
             TestType::Streaming => "testoperator_run_streaming_bench.yml",
             TestType::StreamingCorrectness => "testoperator_run_streaming_correctness.yml",
             TestType::Schema => "testoperator_run_schema.yml",
+            TestType::Htap => "testoperator_run_htap.yml",
         }
     }
 }
@@ -97,6 +99,7 @@ impl Display for TestType {
             TestType::Streaming => write!(f, "streaming"),
             TestType::StreamingCorrectness => write!(f, "streaming_correctness"),
             TestType::Schema => write!(f, "schema"),
+            TestType::Htap => write!(f, "htap"),
         }
     }
 }
diff --git a/test/spicepods/chbench/sf1/accelerated/postgres-cayenne[file]-cdc-tuned.yaml b/test/spicepods/chbench/sf1/accelerated/postgres-cayenne[file]-cdc-tuned.yaml
@@ -0,0 +1,176 @@
+version: v1
+kind: Spicepod
+name: postgres-cayenne[file]-cdc-tuned
+
+
+runtime:
+  params:
+    cdc_prefetch_buffer: "1024"
+    cdc_max_coalesced_envelopes: "1024"
+    cdc_max_coalesced_bytes: "268435456"  # 256 MB
+
+datasets:
+  # ---- TPC-C core tables (mutated by OLTP workload) ----
+
+  - from: postgres:warehouse
+    name: warehouse
+    params: &postgres_params
+      pg_host: 127.0.0.1
+      pg_port: 5432
+      pg_db: chbench
+      pg_user: bench
+      pg_pass: bench
+      pg_sslmode: disable
+      pg_replication_slot: spice_warehouse
+      pg_replication_initial_snapshot: "true"
+    acceleration: &cayenne_accel
+      enabled: true
+      engine: cayenne
+      mode: file
+      refresh_mode: changes
+      primary_key: w_id
+      on_conflict:
+        w_id: upsert
+      params:
+        cayenne_write_concurrency: "4"
+        cayenne_upload_concurrency: "4"
+
+  - from: postgres:district
+    name: district
+    params:
+      <<: *postgres_params
+      pg_replication_slot: spice_district
+    acceleration:
+      <<: *cayenne_accel
+      primary_key: (d_w_id, d_id)
+      on_conflict:
+        (d_w_id, d_id): upsert
+      params:
+        cayenne_write_concurrency: "4"
+        cayenne_upload_concurrency: "4"
+
+  - from: postgres:customer
+    name: customer
+    params:
+      <<: *postgres_params
+      pg_replication_slot: spice_customer
+    acceleration:
+      <<: *cayenne_accel
+      primary_key: (c_w_id, c_d_id, c_id)
+      on_conflict:
+        (c_w_id, c_d_id, c_id): upsert
+      params:
+        cayenne_write_concurrency: "4"
+        cayenne_upload_concurrency: "4"
+
+  - from: postgres:new_order
+    name: new_order
+    params:
+      <<: *postgres_params
+      pg_replication_slot: spice_new_order
+    acceleration:
+      <<: *cayenne_accel
+      primary_key: (no_w_id, no_d_id, no_o_id)
+      on_conflict:
+        (no_w_id, no_d_id, no_o_id): upsert
+      params:
+        cayenne_write_concurrency: "4"
+        cayenne_upload_concurrency: "4"
+
+  - from: postgres:orders
+    name: orders
+    params:
+      <<: *postgres_params
+      pg_replication_slot: spice_orders
+    acceleration:
+      <<: *cayenne_accel
+      primary_key: (o_w_id, o_d_id, o_id)
+      on_conflict:
+        (o_w_id, o_d_id, o_id): upsert
+      params:
+        cayenne_write_concurrency: "4"
+        cayenne_upload_concurrency: "4"
+
+  - from: postgres:order_line
+    name: order_line
+    params:
+      <<: *postgres_params
+      pg_replication_slot: spice_order_line
+    acceleration:
+      <<: *cayenne_accel
+      primary_key: (ol_w_id, ol_d_id, ol_o_id, ol_number)
+      on_conflict:
+        (ol_w_id, ol_d_id, ol_o_id, ol_number): upsert
+      params:
+        cayenne_write_concurrency: "4"
+        cayenne_upload_concurrency: "4"
+
+  - from: postgres:item
+    name: item
+    params:
+      <<: *postgres_params
+      pg_replication_slot: spice_item
+    acceleration:
+      <<: *cayenne_accel
+      primary_key: i_id
+      on_conflict:
+        i_id: upsert
+      params:
+        cayenne_write_concurrency: "4"
+        cayenne_upload_concurrency: "4"
+
+  - from: postgres:stock
+    name: stock
+    params:
+      <<: *postgres_params
+      pg_replication_slot: spice_stock
+    acceleration:
+      <<: *cayenne_accel
+      primary_key: (s_w_id, s_i_id)
+      on_conflict:
+        (s_w_id, s_i_id): upsert
+      params:
+        cayenne_write_concurrency: "4"
+        cayenne_upload_concurrency: "4"
+
+  - from: postgres:region
+    name: region
+    params:
+      <<: *postgres_params
+      pg_replication_slot: spice_region
+    acceleration:
+      <<: *cayenne_accel
+      primary_key: r_regionkey
+      on_conflict:
+        r_regionkey: upsert
+      params:
+        cayenne_write_concurrency: "4"
+        cayenne_upload_concurrency: "4"
+
+  - from: postgres:nation
+    name: nation
+    params:
+      <<: *postgres_params
+      pg_replication_slot: spice_nation
+    acceleration:
+      <<: *cayenne_accel
+      primary_key: n_nationkey
+      on_conflict:
+        n_nationkey: upsert
+      params:
+        cayenne_write_concurrency: "4"
+        cayenne_upload_concurrency: "4"
+
+  - from: postgres:supplier
+    name: supplier
+    params:
+      <<: *postgres_params
+      pg_replication_slot: spice_supplier
+    acceleration:
+      <<: *cayenne_accel
+      primary_key: s_suppkey
+      on_conflict:
+        s_suppkey: upsert
+      params:
+        cayenne_write_concurrency: "4"
+        cayenne_upload_concurrency: "4"
diff --git a/tools/testoperator/dispatch/chbench/sf1/postgres-arrow.yaml b/tools/testoperator/dispatch/chbench/sf1/postgres-arrow.yaml
@@ -0,0 +1,7 @@
+tests:
+  htap:
+    spicepod_path: accelerated/postgres-arrow.yaml
+    runner_type: spiceai-dev-runners
+    scale_factor: 1
+    duration: 300
+    ready_wait: 60
diff --git a/tools/testoperator/dispatch/chbench/sf1/postgres-cayenne[file]-cdc-tuned.yaml b/tools/testoperator/dispatch/chbench/sf1/postgres-cayenne[file]-cdc-tuned.yaml
@@ -0,0 +1,7 @@
+tests:
+  htap:
+    spicepod_path: accelerated/postgres-cayenne[file]-cdc-tuned.yaml
+    runner_type: spiceai-dev-runners
+    scale_factor: 1
+    duration: 300
+    ready_wait: 60
diff --git a/tools/testoperator/dispatch/chbench/sf1/postgres-cayenne[file].yaml b/tools/testoperator/dispatch/chbench/sf1/postgres-cayenne[file].yaml
@@ -0,0 +1,7 @@
+tests:
+  htap:
+    spicepod_path: accelerated/postgres-cayenne[file].yaml
+    runner_type: spiceai-dev-runners
+    scale_factor: 1
+    duration: 300
+    ready_wait: 60
diff --git a/tools/testoperator/dispatch/chbench/sf1/postgres-duckdb[file].yaml b/tools/testoperator/dispatch/chbench/sf1/postgres-duckdb[file].yaml
@@ -0,0 +1,7 @@
+tests:
+  htap:
+    spicepod_path: accelerated/postgres-duckdb[file].yaml
+    runner_type: spiceai-dev-runners
+    scale_factor: 1
+    duration: 300
+    ready_wait: 60
diff --git a/tools/testoperator/src/args/dispatch.rs b/tools/testoperator/src/args/dispatch.rs
@@ -63,6 +63,7 @@ pub enum Workflow {
     StreamingBench,
     StreamingCorrectness,
     Schema,
+    Htap,
 }
 
 impl From<Workflow> for TestType {
@@ -77,6 +78,7 @@ impl From<Workflow> for TestType {
             Workflow::StreamingBench => TestType::Streaming,
             Workflow::StreamingCorrectness => TestType::StreamingCorrectness,
             Workflow::Schema => TestType::Schema,
+            Workflow::Htap => TestType::Htap,
         }
     }
 }
@@ -109,6 +111,8 @@ pub struct DispatchTests {
     pub streaming_correctness: Vec<StreamingCorrectnessDispatchArgs>,
     #[serde(deserialize_with = "deserialize_single_or_vec", default)]
     pub schema: Vec<SchemaArgs>,
+    #[serde(deserialize_with = "deserialize_single_or_vec", default)]
+    pub htap: Vec<HtapDispatchArgs>,
 }
 
 /// Benchmark and throughput workflow arguments, defined in the test files
@@ -413,6 +417,24 @@ pub struct StreamingCorrectnessDispatchArgs {
     pub mutation_seed: Option<u64>,
 }
 
+/// HTAP workflow arguments.
+///
+/// Mirrors the inputs of `testoperator_run_htap.yml`.
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct HtapDispatchArgs {
+    pub spicepod_path: PathBuf,
+    pub runner_type: RunnerType,
+    #[serde(
+        skip_serializing_if = "Option::is_none",
+        serialize_with = "serialize_scale_factor"
+    )]
+    pub scale_factor: Option<f64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub duration: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub ready_wait: Option<u64>,
+}
+
 fn default_queryset() -> String {
     "tpch".to_string()
 }
@@ -575,5 +597,39 @@ tests: {}
         assert_eq!(test_file.tests.bench.len(), 0);
         assert_eq!(test_file.tests.throughput.len(), 0);
         assert_eq!(test_file.tests.load.len(), 0);
+        assert_eq!(test_file.tests.htap.len(), 0);
+    }
+
+    #[test]
+    fn test_htap_section_deserialization() {
+        let yaml = "
+tests:
+  htap:
+    spicepod_path: accelerated/postgres-cayenne[file].yaml
+    runner_type: spiceai-dev-runners
+    scale_factor: 1
+    duration: 300
+    ready_wait: 60
+";
+
+        let test_file: DispatchTestFile = yaml::from_str(yaml).expect("Failed to deserialize");
+
+        assert_eq!(test_file.tests.htap.len(), 1);
+        assert_eq!(
+            test_file.tests.htap[0].spicepod_path.to_string_lossy(),
+            "accelerated/postgres-cayenne[file].yaml"
+        );
+        assert!(matches!(
+            test_file.tests.htap[0].runner_type,
+            RunnerType::Dev
+        ));
+        assert_eq!(test_file.tests.htap[0].scale_factor, Some(1.0));
+        assert_eq!(test_file.tests.htap[0].duration, Some(300));
+        assert_eq!(test_file.tests.htap[0].ready_wait, Some(60));
+
+        // Verify scale_factor serializes as integer 1 (not 1.0) for GitHub workflow inputs
+        let serialized =
+            serde_json::to_value(&test_file.tests.htap[0]).expect("Failed to serialize");
+        assert_eq!(serialized["scale_factor"], 1);
     }
 }
diff --git a/tools/testoperator/src/commands/dispatch/mod.rs b/tools/testoperator/src/commands/dispatch/mod.rs
diff --git a/tools/testoperator/src/commands/htap/mod.rs b/tools/testoperator/src/commands/htap/mod.rs

Original file line number	Diff line number	Diff line change
`@@ -64,6 +64,7 @@ pub enum TestType {`
`64`	`64`	`Streaming,`
`65`	`65`	`StreamingCorrectness,`
`66`	`66`	`Schema,`
	`67`	`+ Htap,`
`67`	`68`	`}`
`68`	`69`
`69`	`70`	`impl TestType {`
`@@ -80,6 +81,7 @@ impl TestType {`
`80`	`81`	`TestType::Streaming => "testoperator_run_streaming_bench.yml",`
`81`	`82`	`TestType::StreamingCorrectness => "testoperator_run_streaming_correctness.yml",`
`82`	`83`	`TestType::Schema => "testoperator_run_schema.yml",`
	`84`	`+ TestType::Htap => "testoperator_run_htap.yml",`
`83`	`85`	`}`
`84`	`86`	`}`
`85`	`87`	`}`
`@@ -97,6 +99,7 @@ impl Display for TestType {`
`97`	`99`	`TestType::Streaming => write!(f, "streaming"),`
`98`	`100`	`TestType::StreamingCorrectness => write!(f, "streaming_correctness"),`
`99`	`101`	`TestType::Schema => write!(f, "schema"),`
	`102`	`+ TestType::Htap => write!(f, "htap"),`
`100`	`103`	`}`
`101`	`104`	`}`
`102`	`105`	`}`