study: --phase hydra on the eval's datasets; Hydra cost rows

aqpbm-cli/src/cli.rs

@@ -236,6 +236,17 @@ pub struct SketchbenchArgs {
     /// merge exactly, so a gap is a defect; for KLL it is the result.
     #[arg(long, default_value_t = 2, help_heading = "Measurement content")]
     pub merge_shards: usize,
+    /// Which label columns the `hydra-*` rows group by, comma-separated
+    /// indices into the columns before the value column: any non-empty subset
+    /// (`0`, `1`, `0,1`). The grid stores every subset, so this picks the one
+    /// asked and scored. Every other target ignores it.
+    #[arg(
+        long,
+        value_delimiter = ',',
+        default_value = "0",
+        help_heading = "Measurement content"
+    )]
+    pub group_columns: Vec<usize>,
     /// How the merge operation splits the stream into its shards: `contiguous`
     /// cuts it into stretches, `interleaved` deals it round-robin by record, so
     /// each shard is a sample of the whole stream. The shard sizes are the same
@@ -248,17 +259,6 @@ pub struct SketchbenchArgs {
         help_heading = "Measurement content"
     )]
     pub merge_split: MergeSplit,
-    /// Which label columns the `hydra-*` rows group by, comma-separated
-    /// indices into the columns before the value column: any non-empty subset
-    /// (`0`, `1`, `0,1`). The grid stores every subset, so this picks the one
-    /// asked and scored. Every other target ignores it.
-    #[arg(
-        long,
-        value_delimiter = ',',
-        default_value = "0",
-        help_heading = "Measurement content"
-    )]
-    pub group_columns: Vec<usize>,
 
     /// Path to append JSONL records to. `-` or omitted sends them to stdout.
     #[arg(long, help_heading = "Output")]

aqpbm-cli/src/external.rs

@@ -271,6 +271,8 @@ fn shape_column(data_type: &str) -> ColumnSpec {
         string: None,
         child_of: None,
         fan_out: None,
+        scale_by: None,
+        scale_range: None,
     }
 }
 

aqpbm-cli/src/flatten_record.rs

@@ -259,6 +259,8 @@ mod tests {
                 string: None,
                 child_of: None,
                 fan_out: None,
+                scale_by: None,
+                scale_range: None,
             },
             1000,
         );

aqpbm-cli/src/main.rs

@@ -260,6 +260,8 @@ fn dataset_spec(args: &SketchbenchArgs) -> Result<InputDataSetSpec> {
             },
             child_of: None,
             fan_out: None,
+            scale_by: None,
+            scale_range: None,
         },
         args.size.expect("clap requires --size") as u64,
     )))
@@ -374,9 +376,9 @@ fn run_sketchbench(args: SketchbenchArgs) -> Result<()> {
         width,
         workers: args.workers.max(1),
         merge_shards: args.merge_shards,
-        merge_split: args.merge_split,
         group_columns,
         per_group_out: args.per_group_out.as_ref().map(std::path::PathBuf::from),
+        merge_split: args.merge_split,
         comparator: args.comparator.clone(),
         runs: args.runs,
         warmup_runs: args.warmup_runs,

aqpbm-cli/src/repeat.rs

@@ -193,6 +193,8 @@ mod tests {
                 string: None,
                 child_of: None,
                 fan_out: None,
+                scale_by: None,
+                scale_range: None,
             },
             1000,
         );

aqpbm-core/schema/merged_record.schema.json

@@ -431,6 +431,28 @@
           "format": "uint64",
           "minimum": 0
         },
+        "scale_by": {
+          "description": "Index of an earlier `string` column whose values scale this one. Each\ndistinct label value gets a fixed factor in `scale_range`, and every\nrow's value is multiplied by its label's factor. `f64` columns only.",
+          "type": [
+            "integer",
+            "null"
+          ],
+          "format": "uint",
+          "minimum": 0
+        },
+        "scale_range": {
+          "description": "`[lo, hi]` with `0 < lo <= hi`: the factors are log-uniform in it.\nRequired with `scale_by`.",
+          "type": [
+            "array",
+            "null"
+          ],
+          "items": {
+            "type": "number",
+            "format": "double"
+          },
+          "maxItems": 2,
+          "minItems": 2
+        },
         "shift": {
           "description": "Added to every value, for a column that should not start where its\ndistribution naturally does. Under the monotonic rule it is the value the\nseries starts from.",
           "type": [

aqpbm-core/src/atomic_costs.rs

@@ -74,6 +74,15 @@ pub struct MeasuredAt {
     pub merge_operand_items: Option<u64>,
     /// The value column's distribution as generated, e.g. Zipf with its skew.
     pub distribution: Option<DataDistribution>,
+    /// The datagen spec (yaml stem) a `hydra-*` row was measured on, named by
+    /// the driver: the record carries the description, not its name.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub dataset: Option<String>,
+    /// A `hydra-*` row's label columns `d`: each record is inserted into
+    /// `2^d - 1` subsets, so its insert cost holds at this width only. Absent
+    /// for every other row.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub schema_width: Option<u32>,
 }
 
 impl MeasuredAt {
@@ -275,7 +284,10 @@ pub(crate) fn reduce_scores(record: &MergedRecord) -> Result<AtomicCostEntry, Sk
         insert_cpu_secs,
         merge_cpu_secs,
         query_cpu_secs,
-        measured_at: measured_at(record, groups_measured),
+        measured_at: MeasuredAt {
+            schema_width: query_accuracy.get("schema_width").map(|&d| d as u32),
+            ..measured_at(record, groups_measured)
+        },
         query_accuracy,
         accuracy_metric: String::new(),
     })
@@ -320,6 +332,8 @@ fn measured_at(record: &MergedRecord, groups_measured: Option<f64>) -> MeasuredA
             .filter(|&m| m > 0)
             .map(|m| items / m as u64),
         distribution: value.map(|c| c.distribution.clone()),
+        dataset: None,
+        schema_width: None,
     }
 }
 
@@ -420,6 +434,8 @@ mod tests {
                 string: None,
                 child_of: None,
                 fan_out: None,
+                scale_by: None,
+                scale_range: None,
             },
             1_000_000,
         )
@@ -483,6 +499,25 @@ mod tests {
 
     /// A one-column table: its column gives the keys and the values, and the
     /// merge operand is the stream split `merge_shards` ways.
+    /// A Hydra row's insert is per record at its schema width, which the
+    /// row names; a row without the score names none.
+    #[test]
+    fn measured_at_carries_a_hydra_rows_schema_width() {
+        let mut record = full_record();
+        assert_eq!(
+            reduce_one(&record, METRIC)
+                .unwrap()
+                .measured_at
+                .schema_width,
+            None
+        );
+        record.query.accuracy =
+            Some(serde_json::json!({"relative_error_mean": 0.1, "schema_width": 4.0}));
+        let entry = reduce_one(&record, METRIC).expect("fully populated record");
+        assert_eq!(entry.measured_at.schema_width, Some(4));
+        assert!((entry.insert_cpu_secs - 0.5e-6).abs() < 1e-12);
+    }
+
     #[test]
     fn measured_at_reads_a_single_column_workload() {
         let mut record = full_record();
@@ -721,6 +756,8 @@ mod tests {
                 value_range: None,
                 merge_operand_items: None,
                 distribution: None,
+                dataset: None,
+                schema_width: None,
             },
         };
         let json = serde_json::to_string(&entry).unwrap();

aqpbm-core/src/benchmark_result/schema.rs

@@ -525,6 +525,8 @@ mod tests {
                 string: None,
                 child_of: None,
                 fan_out: None,
+                scale_by: None,
+                scale_range: None,
             },
             1_000_000,
         )

aqpbm-core/src/erp.rs

@@ -183,6 +183,8 @@ mod tests {
                     string: None,
                     child_of: None,
                     fan_out: None,
+                    scale_by: None,
+                    scale_range: None,
                 },
                 100_000,
             ),

aqpbm-datagen/src/column.rs

@@ -45,6 +45,15 @@ pub struct ColumnSpec {
     /// checked against the distribution's domain like `cardinality` is.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub fan_out: Option<u64>,
+    /// Index of an earlier `string` column whose values scale this one. Each
+    /// distinct label value gets a fixed factor in `scale_range`, and every
+    /// row's value is multiplied by its label's factor. `f64` columns only.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub scale_by: Option<usize>,
+    /// `[lo, hi]` with `0 < lo <= hi`: the factors are log-uniform in it.
+    /// Required with `scale_by`.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub scale_range: Option<[f64; 2]>,
 }
 
 impl ColumnSpec {
@@ -77,6 +86,35 @@ impl ColumnSpec {
             (Some(_), Some(fan_out)) => self.validate_child(fan_out)?,
         }
 
+        match (self.scale_by, self.scale_range) {
+            (None, None) => {}
+            (Some(_), None) => {
+                return Err(DataGenError::BadParam(
+                    "scale_by: needs scale_range, the [lo, hi] the factors are drawn from".into(),
+                ))
+            }
+            (None, Some(_)) => {
+                return Err(DataGenError::BadParam(
+                    "scale_range: applies only to a scale_by column".into(),
+                ))
+            }
+            (Some(_), Some([lo, hi])) => {
+                // An integer column would round the scaled value back down.
+                if self.data_type != "f64" {
+                    return Err(DataGenError::BadParam(format!(
+                        "scale_by: a scaled value is fractional, so data_type has to be \
+                         `f64`, not '{}'",
+                        self.data_type
+                    )));
+                }
+                if !(lo > 0.0 && hi >= lo && hi.is_finite()) {
+                    return Err(DataGenError::BadParam(format!(
+                        "scale_range [{lo}, {hi}] needs 0 < lo <= hi"
+                    )));
+                }
+            }
+        }
+
         // `cardinality` restates the domain. Silently preferring one over the
         // other would let a report cite a key space the column never had.
         if let Some(claimed) = self.cardinality {
@@ -246,6 +284,19 @@ impl ColumnSpec {
         }
     }
 
+    /// The fixed factor of one `scale_by` label value: a SplitMix64 hash of
+    /// the value's FNV-1a and this column's seed, as `u` in `[0, 1)`, mapped to
+    /// `lo·(hi/lo)^u`.
+    pub(crate) fn scale_factor(&self, label: &str) -> f64 {
+        let [lo, hi] = self.scale_range.expect("scale_range was validated above");
+        let fnv = label.bytes().fold(0xCBF2_9CE4_8422_2325u64, |h, b| {
+            (h ^ b as u64).wrapping_mul(0x0000_0100_0000_01B3)
+        });
+        let x = crate::table::mix_connected_bits(fnv ^ self.distribution.seed());
+        let u = (x >> 11) as f64 / (1u64 << 53) as f64;
+        lo * (hi / lo).powf(u)
+    }
+
     /// Render a `child_of` column: each row's parent value, a `.`, and the
     /// draw's 0-based rank in the domain as the child index.
     pub(crate) fn render_child(