[zzylol]

Ingest SDK row-sampled points in the Rust collector

Why

The Go SDK patch (and the Rust SDK fork in progress) can sample each measurement per sketch row and send every admitted measurement as one Gauge point. The raw value is not pre-weighted, and three reserved attributes carry the sampling decision:

attribute	type	meaning
__asap_row_sampled_admitted_rows	int	bitmask of the rows that admit the point
__asap_row_sampled_rows	int	row count d the SDK sampled for
__asap_row_sampled_sample_p	double	per-row admission probability p

The Go collector (asapedgeprocessor/ingest.go, warm_sketch.go observeRowSampled) applies these. The Rust collector dropped them, because a non-string attribute that is not a reserved key was silently discarded. Each sampled point was then counted as a plain weight-1 sample, so CMS and Count Sketch counts and DDSketch counts came out about p times too small.

What

• The decoder reads the three keys into ObservationValue::row_sample and keeps them out of the labels (series identity).
• Count-Min and Count Sketch add 1/p only to the admitted rows.
• DDSketch inserts the value with weight 1/p. With the tail split, values above tau arrive at p = 1 and count once.
• A point that does not fit its sketch is dropped and counted in StatsSnapshot::dropped_row_sample. The checks are: row count equals the sketch's d, mask is nonzero and fits in d bits, 0 < p <= 1, and 1/p is an integer.
• KLL, HLL, and Count-Min with heap reject row-sampled points, as the Go collector does.
• A point without the keys behaves exactly as before.
• Fix: before this PR, an observation the observer rejected still opened an empty series (and under EvictOldest evicted another one). Now it changes no series.

How

sketch	row-sampled point	Go collector
Count-Min	1/p added to each admitted row's cell	same (cell gets 1/p, no envelope rescale)
Count Sketch	±count·1/p added to each admitted row's cell	same
DDSketch (d = 1, mask 1)	bucket count += 1/p	records once, stamps p on the envelope, requires one p per window
KLL, HLL, CMS+heap	rejected, counted	rejected, counted

• The DDSketch handling differs from the Go collector on purpose. The tail split sends two rates within one window, which the Go approach (one p stamped on the envelope) cannot express. Per-point weights can.
• Counters are integers (i64 cells, u64 DDSketch buckets), so 1/p must be an integer, for example p = 2^-k.
• The sketchlib envelope's admission_rate stays 1, because the counters already hold the weighted counts.
• sketchlib has no update for a subset of rows on the regular path. sketches::regular_path_cell therefore copies sketchlib's column and sign rule (hash64_seeded(row, key), low 32 bits mod cols, bit 63 = sign). The update_rows_match_sketchlib_update tests pin it to sketchlib's own update_weighted.

Key code interfaces added / modified

• observation::RowSample { admitted_rows: u64, rows: u32, sample_p: f64 }, with:
  • admitted_row_weight(sketch_rows) -> Result<i64, PrecomputeError>
  • admits(row)
• ObservationValue::row_sample: Option<RowSample>. It is skipped in serde when absent.
• observation::reject_row_sample(obs, sketch)
• PrecomputeError::InvalidRowSample(String) and StatsSnapshot::dropped_row_sample
• CMSWrapper::update_rows, CountSketchWrapper::update_rows, DDSketchWrapper::update_weighted
• otap::ATTR_ROW_SAMPLED_{ADMITTED_ROWS,ROWS,SAMPLE_P}

Usage:

// What the decoder builds from one SDK point (mask 0b1010, d = 4, p = 1/8):
let obs = Observation::new(ts_ms, "requests", vec![], labels, ObservationValue {
    row_sample: Some(RowSample { admitted_rows: 0b1010, rows: 4, sample_p: 0.125 }),
    ..ObservationValue::float(1.0)
});
// Rows 1 and 3 of the series' 4-row CMS each gain 8; rows 0 and 2 are untouched.
CMSObserver.observe(&mut cms, &obs)?;

Major algorithm changes

• Weighted per-row updates: an admitted row adds 1/p, so each row's cell is an unbiased count.
• Median readout: under per-row sampling, the unbiased readout is the median over rows. sketchlib's Count Sketch estimate already takes the median. Its regular-path Count-Min estimate takes the minimum, which is biased low on row-sampled data. This PR leaves the library and CMSWrapper::estimate_count unchanged, so a consumer of a row-sampled CMS has to read the median of rows.

Before this PR

An SDK sends 32 000 occurrences of key k, sampled per row at p = 1/8, d = 4. About 13 000 points arrive. The CMS for k holds about 13 000 in every row, against a true count of 32 000.

After this PR

Each row holds about 32 000 (each admitted row ≈ 4 000 × 8). The test row_sampled_stream_counts_are_unbiased checks that the median over rows is within 5% for every key and that the mean error over keys is under 1%.

Verification

• cargo fmt --check: clean
• cargo clippy --all-targets --all-features: no warnings
• cargo test --all-features: all pass (lib 110 passed / 1 ignored, integration suites pass)
• New tests:
  • observation::tests::admitted_row_weight_validates_against_the_sketch
  • otap::codec::tests::row_sampled_keys_are_parsed_and_kept_out_of_series_identity
  • sketches::{cms,countsketch}::tests::{update_rows_match_sketchlib_update, row_sampled_stream_*_are_unbiased, unsampled_tail_points_count_with_weight_one}
  • sketches::cms::tests::row_count_mismatch_is_rejected
  • sketches::ddsketch::tests::{row_sampled_body_with_unsampled_tail_matches_full_stream, unsampled_tail_points_count_with_weight_one}
  • tests/runtime.rs::invalid_row_sample_is_counted_and_opens_no_series. This one fails on the old window.rs (left: (1, 1) right: (1, 0)) and passes with the fix.