chore: sync raw producer with merged series frame parent

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

src/common/query/Cargo.toml

@@ -25,6 +25,7 @@ datafusion-common.workspace = true
 datafusion-expr.workspace = true
 datafusion-proto.workspace = true
 datatypes.workspace = true
+futures-util.workspace = true
 once_cell.workspace = true
 prost.workspace = true
 serde.workspace = true
@@ -36,7 +37,6 @@ store-api.workspace = true
 
 [dev-dependencies]
 common-base.workspace = true
-futures-util.workspace = true
 serde_json.workspace = true
 tokio.workspace = true
 

src/common/query/src/lib.rs

@@ -20,6 +20,7 @@ pub mod prelude;
 pub mod prometheus;
 pub mod promql_annotations;
 pub mod request;
+pub mod series_frame;
 pub mod stream;
 #[cfg(any(test, feature = "testing"))]
 pub mod test_util;

src/common/query/src/series_frame.rs

@@ -0,0 +1,451 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Streaming boundary between finite raw-series chunks and DataFusion record batches.
+//!
+//! Schema fields identify the ordered logical payload fields and codec layout; additional
+//! application metadata is permitted when stable across the stream. The caller-provided
+//! [`RawLabelSource`] is the label binding, not authenticated by the frame schema or field
+//! provenance. This adapter preserves scan candidates only; it is not a residual filter,
+//! authorization check, remote transport, or compaction boundary. Only
+//! [`SeriesOrdering::Contiguous`] and [`SeriesOrdering::KeyThenTime`] are accepted;
+//! `Interleaved` is rejected before any poll.
+
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::{Context, Poll};
+
+use common_error::ext::BoxedError;
+use common_recordbatch::{DfRecordBatch, DfSendableRecordBatchStream};
+use datafusion_common::DataFusionError;
+use datatypes::error::InvalidSeriesSnafu;
+use datatypes::series::frame::{
+    RawLabelMode, RawSeriesChunkCursor, RawSeriesFrameBuilder, RawSeriesFrameDecoder,
+};
+use datatypes::series::raw::RawSeriesChunk;
+use datatypes::series::samples::SeriesSchema;
+use datatypes::series::{MetricName, SeriesOrdering};
+use futures_util::Stream;
+use store_api::region_engine::{RawLabelSource, RawSeriesStream, SendableRawSeriesStream};
+
+#[cfg(test)]
+mod tests;
+
+/// Packs a storage raw-series stream into DataFusion record batches incrementally.
+///
+/// Only [`SeriesOrdering::Contiguous`] and [`SeriesOrdering::KeyThenTime`] streams are
+/// accepted; an `Interleaved` declaration is rejected before the first poll, including for an
+/// empty stream. Rows accumulate across chunks into batches bounded by the builder's row and
+/// estimated payload limits; one legal oversized row may be emitted alone, and these limits are a
+/// work bound, not a resident-memory cap. A ready batch is yielded without polling the source
+/// again. Producers must keep each series' chunks contiguous. Validation checks adjacent chunk
+/// conditions but does not certify nonadjacent recurrence for `Contiguous` ordering.
+pub fn pack_raw_series(
+    source: SendableRawSeriesStream,
+) -> datafusion_common::Result<DfSendableRecordBatchStream> {
+    let ordering = source.ordering();
+    ensure_supported_ordering(ordering)?;
+    let schema = source.schema().clone();
+    let label_source = source.label_source().clone();
+    let mode = label_mode(&label_source);
+    let builder = RawSeriesFrameBuilder::try_new(
+        &schema,
+        mode,
+        RawSeriesFrameBuilder::DEFAULT_MAX_ROWS,
+        RawSeriesFrameBuilder::DEFAULT_MAX_BYTES,
+    )
+    .map_err(codec_error)?;
+    let frame_schema = builder.schema().clone();
+    Ok(Box::pin(PackStream {
+        source: Some(source),
+        schema,
+        mode,
+        frame_schema,
+        ordering,
+        validator: SequenceValidator::new(ordering),
+        builder: Some(builder),
+        cursor: None,
+    }))
+}
+
+/// Unpacks DataFusion batches using the caller's explicit scan binding and ordering.
+///
+/// Only [`SeriesOrdering::Contiguous`] and [`SeriesOrdering::KeyThenTime`] streams are
+/// accepted; an `Interleaved` declaration is rejected before the stream is returned, including
+/// for an empty stream. Producers must keep each series' chunks contiguous. Validation checks
+/// adjacent chunk conditions but does not certify nonadjacent recurrence for `Contiguous`
+/// ordering.
+///
+/// The supplied binding is retained unchanged and is not validated as field provenance.
+pub fn unpack_raw_series(
+    source: DfSendableRecordBatchStream,
+    expected: Arc<SeriesSchema>,
+    label_source: RawLabelSource,
+    ordering: SeriesOrdering,
+) -> datafusion_common::Result<SendableRawSeriesStream> {
+    ensure_supported_ordering(ordering)?;
+    let mode = label_mode(&label_source);
+    let frame_schema = source.schema();
+    let decoder = RawSeriesFrameDecoder::try_new(&expected, mode).map_err(codec_error)?;
+    // Validate even empty streams' declared schema and layout before returning the stream.
+    let empty = DfRecordBatch::new_empty(frame_schema.clone());
+    decoder.decode(&empty).map_err(codec_error)?;
+    Ok(Box::pin(UnpackStream {
+        source: Some(source),
+        schema: expected,
+        frame_schema,
+        label_source,
+        ordering,
+        validator: SequenceValidator::new(ordering),
+        pending: Vec::new().into_iter(),
+        decoder,
+    }))
+}
+
+fn label_mode(source: &RawLabelSource) -> RawLabelMode {
+    match source {
+        RawLabelSource::PrimaryKey { .. } => RawLabelMode::Key,
+        RawLabelSource::Explicit => RawLabelMode::Explicit,
+    }
+}
+
+/// Rejects orderings this adapter cannot validate; a bad declaration is an error, not a fallback.
+fn ensure_supported_ordering(ordering: SeriesOrdering) -> datafusion_common::Result<()> {
+    match ordering {
+        SeriesOrdering::Contiguous | SeriesOrdering::KeyThenTime => Ok(()),
+        SeriesOrdering::Interleaved => {
+            Err(invalid("interleaved raw series ordering is not supported"))
+        }
+    }
+}
+
+fn codec_error(error: datatypes::error::Error) -> DataFusionError {
+    DataFusionError::External(Box::new(BoxedError::new(error)))
+}
+
+fn upstream_error(error: BoxedError) -> DataFusionError {
+    DataFusionError::External(Box::new(error))
+}
+
+fn invalid(reason: impl Into<String>) -> DataFusionError {
+    DataFusionError::External(Box::new(BoxedError::new(
+        InvalidSeriesSnafu {
+            reason: reason.into(),
+        }
+        .build(),
+    )))
+}
+
+/// Cross-chunk validation state. The producer must keep each series' chunks contiguous; this
+/// validator checks adjacent chunks only and never sees `Interleaved` streams. It retains only
+/// the previous complete owned key (not an Arrow buffer), timestamp, metric, and finality.
+struct SequenceValidator {
+    ordering: SeriesOrdering,
+    previous_key: Option<Vec<u8>>,
+    previous_timestamp: Option<i64>,
+    previous_metric: Option<MetricName>,
+    previous_is_last: bool,
+}
+
+impl SequenceValidator {
+    fn new(ordering: SeriesOrdering) -> Self {
+        Self {
+            ordering,
+            previous_key: None,
+            previous_timestamp: None,
+            previous_metric: None,
+            previous_is_last: false,
+        }
+    }
+
+    fn finish(&self) -> datafusion_common::Result<()> {
+        if self.previous_key.is_some() && !self.previous_is_last {
+            return Err(invalid("raw series stream ended before key was final"));
+        }
+        Ok(())
+    }
+
+    fn validate(&mut self, chunk: &RawSeriesChunk) -> datafusion_common::Result<()> {
+        let key = chunk.primary_key.as_slice();
+        let timestamp = chunk
+            .time
+            .len()
+            .checked_sub(1)
+            .and_then(|last| chunk.time.timestamp(last))
+            .ok_or_else(|| invalid("empty raw series chunk"))?;
+        if let Some(previous) = &self.previous_key {
+            match previous.as_slice().cmp(key) {
+                std::cmp::Ordering::Equal => {
+                    if self.previous_is_last {
+                        return Err(invalid("raw series key recurred after final chunk"));
+                    }
+                    if self.previous_metric != chunk.metric {
+                        return Err(invalid("raw series metric changed within key"));
+                    }
+                    let first = chunk
+                        .time
+                        .timestamp(0)
+                        .ok_or_else(|| invalid("empty raw series chunk"))?;
+                    if self.previous_timestamp.is_some_and(|last| first <= last) {
+                        return Err(invalid(
+                            "raw series timestamps are not increasing across chunks",
+                        ));
+                    }
+                }
+                _ => {
+                    if !self.previous_is_last {
+                        return Err(invalid("raw series key changed before final chunk"));
+                    }
+                    if self.ordering == SeriesOrdering::KeyThenTime && previous.as_slice() > key {
+                        return Err(invalid("raw series keys are not ordered"));
+                    }
+                }
+            }
+        }
+        self.previous_key = Some(key.to_vec());
+        self.previous_timestamp = Some(timestamp);
+        self.previous_metric = chunk.metric.clone();
+        self.previous_is_last = chunk.is_last;
+        Ok(())
+    }
+}
+
+struct PackStream {
+    source: Option<SendableRawSeriesStream>,
+    schema: Arc<SeriesSchema>,
+    mode: RawLabelMode,
+    frame_schema: datatypes::arrow::datatypes::SchemaRef,
+    ordering: SeriesOrdering,
+    validator: SequenceValidator,
+    builder: Option<RawSeriesFrameBuilder>,
+    cursor: Option<RawSeriesChunkCursor>,
+}
+
+impl PackStream {
+    /// Releases the source and drops every buffered owner, leaving the stream terminal; batches
+    /// already yielded are unaffected.
+    fn finish(&mut self) {
+        self.source = None;
+        self.cursor = None;
+        self.builder = None;
+        self.validator = SequenceValidator::new(self.ordering);
+    }
+
+    /// Advances the installed cursor until it yields one ready batch or is exhausted, dropping an
+    /// exhausted cursor. `None` means no batch is ready, so the source may be polled again.
+    fn advance_cursor(&mut self) -> datatypes::error::Result<Option<DfRecordBatch>> {
+        let ready = match (self.cursor.as_mut(), self.builder.as_mut()) {
+            (Some(cursor), Some(builder)) => cursor.append_to(builder)?,
+            _ => return Ok(None),
+        };
+        match ready {
+            Some(batch) => Ok(Some(batch)),
+            None => {
+                self.cursor = None;
+                Ok(None)
+            }
+        }
+    }
+
+    /// Flushes the rows buffered so far, if any, so a pending source can still deliver data.
+    fn flush_buffered(&mut self) -> datatypes::error::Result<Option<DfRecordBatch>> {
+        match self.builder.as_mut() {
+            Some(builder) => builder.flush(),
+            None => Ok(None),
+        }
+    }
+}
+
+impl Stream for PackStream {
+    type Item = datafusion_common::Result<DfRecordBatch>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let this = &mut *self;
+        loop {
+            match this.advance_cursor() {
+                Ok(Some(batch)) => return Poll::Ready(Some(Ok(batch))),
+                Ok(None) => {}
+                Err(error) => {
+                    this.finish();
+                    return Poll::Ready(Some(Err(codec_error(error))));
+                }
+            }
+            let polled = match this.source.as_mut() {
+                Some(source) => source.as_mut().poll_next(cx),
+                None => return Poll::Ready(None),
+            };
+            match polled {
+                Poll::Pending => {
+                    let flushed = this.flush_buffered();
+                    return match flushed {
+                        Ok(Some(batch)) => Poll::Ready(Some(Ok(batch))),
+                        Ok(None) => Poll::Pending,
+                        Err(error) => {
+                            this.finish();
+                            Poll::Ready(Some(Err(codec_error(error))))
+                        }
+                    };
+                }
+                Poll::Ready(None) => {
+                    if let Err(error) = this.validator.finish() {
+                        this.finish();
+                        return Poll::Ready(Some(Err(error)));
+                    }
+                    let flushed = this.flush_buffered();
+                    this.finish();
+                    return match flushed {
+                        Ok(Some(batch)) => Poll::Ready(Some(Ok(batch))),
+                        Ok(None) => Poll::Ready(None),
+                        Err(error) => Poll::Ready(Some(Err(codec_error(error)))),
+                    };
+                }
+                Poll::Ready(Some(Err(error))) => {
+                    this.finish();
+                    return Poll::Ready(Some(Err(upstream_error(error))));
+                }
+                Poll::Ready(Some(Ok(chunk))) => {
+                    if let Err(error) = this.validator.validate(&chunk) {
+                        this.finish();
+                        return Poll::Ready(Some(Err(error)));
+                    }
+                    match RawSeriesChunkCursor::try_new(&this.schema, this.mode, chunk) {
+                        Ok(cursor) => this.cursor = Some(cursor),
+                        Err(error) => {
+                            this.finish();
+                            return Poll::Ready(Some(Err(codec_error(error))));
+                        }
+                    }
+                }
+            }
+        }
+    }
+}
+
+impl datafusion::physical_plan::RecordBatchStream for PackStream {
+    fn schema(&self) -> datatypes::arrow::datatypes::SchemaRef {
+        self.frame_schema.clone()
+    }
+}
+
+struct UnpackStream {
+    source: Option<DfSendableRecordBatchStream>,
+    schema: Arc<SeriesSchema>,
+    frame_schema: datatypes::arrow::datatypes::SchemaRef,
+    label_source: RawLabelSource,
+    ordering: SeriesOrdering,
+    validator: SequenceValidator,
+    pending: std::vec::IntoIter<RawSeriesChunk>,
+    decoder: RawSeriesFrameDecoder,
+}
+
+impl RawSeriesStream for UnpackStream {
+    fn schema(&self) -> &Arc<SeriesSchema> {
+        &self.schema
+    }
+
+    fn label_source(&self) -> &RawLabelSource {
+        &self.label_source
+    }
+
+    fn ordering(&self) -> SeriesOrdering {
+        self.ordering
+    }
+}
+
+impl UnpackStream {
+    fn finish(&mut self) {
+        self.source = None;
+        self.pending = Vec::new().into_iter();
+        self.validator = SequenceValidator::new(self.ordering);
+    }
+}
+
+impl Stream for UnpackStream {
+    type Item = Result<RawSeriesChunk, BoxedError>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        const MAX_EMPTY_BATCHES_PER_POLL: usize = 32;
+        let this = &mut *self;
+        let mut empty_batches = 0;
+        loop {
+            if let Some(chunk) = this.pending.next() {
+                return Poll::Ready(Some(Ok(chunk)));
+            }
+            let polled = match this.source.as_mut() {
+                Some(source) => source.as_mut().poll_next(cx),
+                None => return Poll::Ready(None),
+            };
+            match polled {
+                Poll::Pending => return Poll::Pending,
+                Poll::Ready(None) => {
+                    let result = this.validator.finish();
+                    this.finish();
+                    return Poll::Ready(result.err().map(|error| {
+                        Err(BoxedError::new(crate::error::Error::GeneralDataFusion {
+                            error,
+                            location: snafu::Location::default(),
+                        }))
+                    }));
+                }
+                Poll::Ready(Some(Err(error))) => {
+                    this.finish();
+                    return Poll::Ready(Some(Err(BoxedError::new(
+                        crate::error::Error::GeneralDataFusion {
+                            error,
+                            location: snafu::Location::default(),
+                        },
+                    ))));
+                }
+                Poll::Ready(Some(Ok(batch))) => {
+                    let result = (|| {
+                        if batch.schema() != this.frame_schema {
+                            return Err(invalid("raw frame schema mismatch"));
+                        }
+                        this.decoder
+                            .decode(&batch)
+                            .map_err(codec_error)
+                            .and_then(|chunks| {
+                                for chunk in &chunks {
+                                    this.validator.validate(chunk)?;
+                                }
+                                Ok(chunks)
+                            })
+                    })();
+                    match result {
+                        Ok(chunks) => {
+                            if chunks.is_empty() {
+                                empty_batches += 1;
+                                if empty_batches == MAX_EMPTY_BATCHES_PER_POLL {
+                                    cx.waker().wake_by_ref();
+                                    return Poll::Pending;
+                                }
+                            } else {
+                                this.pending = chunks.into_iter();
+                            }
+                        }
+                        Err(error) => {
+                            this.finish();
+                            return Poll::Ready(Some(Err(BoxedError::new(
+                                crate::error::Error::GeneralDataFusion {
+                                    error,
+                                    location: snafu::Location::default(),
+                                },
+                            ))));
+                        }
+                    }
+                }
+            }
+        }
+    }
+}

src/common/query/src/series_frame/tests/lifecycle.rs

@@ -0,0 +1,525 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::{Context, Poll};
+
+use common_error::ext::BoxedError;
+use datatypes::arrow::array::{Array, Float64Array, UInt32Array};
+use datatypes::arrow::buffer::{Buffer, ScalarBuffer};
+use datatypes::series::SeriesOrdering;
+use datatypes::series::histogram::{BucketSideView, CounterResetHint, HistogramBatchBuilder};
+use datatypes::series::raw::{RawSeriesChunk, RawTimestamps};
+use datatypes::series::samples::{
+    DenseSamplesRef, FieldSpec, HistogramSamples, SampleRef, Samples, SeriesFields, SeriesSchema,
+};
+use datatypes::series::{Label, MetricName};
+use futures_util::{Stream, StreamExt};
+use store_api::region_engine::{RawLabelSource, RawSeriesStream, SendableRawSeriesStream};
+
+use super::super::{pack_raw_series, unpack_raw_series};
+
+struct RawFixture {
+    schema: Arc<SeriesSchema>,
+    binding: RawLabelSource,
+    chunk: Option<RawSeriesChunk>,
+}
+
+impl Stream for RawFixture {
+    type Item = Result<RawSeriesChunk, BoxedError>;
+
+    fn poll_next(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        Poll::Ready(self.chunk.take().map(Ok))
+    }
+}
+
+impl RawSeriesStream for RawFixture {
+    fn schema(&self) -> &Arc<SeriesSchema> {
+        &self.schema
+    }
+
+    fn label_source(&self) -> &RawLabelSource {
+        &self.binding
+    }
+
+    fn ordering(&self) -> SeriesOrdering {
+        SeriesOrdering::Contiguous
+    }
+}
+
+fn histogram_batch() -> Arc<datatypes::series::histogram::HistogramBatch> {
+    let custom = [1.5, 4.5];
+    let positive_offsets = [1];
+    let positive_lengths = [2];
+    let positive_buckets = [3.25, 7.5];
+    let negative_offsets = [-2];
+    let negative_lengths = [1];
+    let negative_buckets = [0.75];
+    let mut builder = HistogramBatchBuilder::new();
+    for (sum, start_timestamp) in [(10.5, Some(-123)), (-9.5, Some(456)), (7.25, None)] {
+        builder
+            .push(datatypes::series::histogram::HistogramView {
+                schema: -7,
+                zero_threshold: 0.125,
+                sum,
+                count: 17.25,
+                zero_count: 2.5,
+                reset_hint: CounterResetHint::Unrecognized(37),
+                start_timestamp,
+                custom_values: &custom,
+                positive: BucketSideView {
+                    span_offsets: &positive_offsets,
+                    span_lengths: &positive_lengths,
+                    buckets: &positive_buckets,
+                },
+                negative: BucketSideView {
+                    span_offsets: &negative_offsets,
+                    span_lengths: &negative_lengths,
+                    buckets: &negative_buckets,
+                },
+            })
+            .unwrap();
+    }
+    builder.finish().unwrap()
+}
+
+fn assert_histogram(
+    view: datatypes::series::histogram::HistogramView<'_>,
+    sum: f64,
+    start: Option<i64>,
+) {
+    assert_eq!(view.schema, -7);
+    assert_eq!(view.zero_threshold, 0.125);
+    assert_eq!(view.sum.to_bits(), sum.to_bits());
+    assert_eq!(view.count, 17.25);
+    assert_eq!(view.zero_count, 2.5);
+    assert_eq!(view.reset_hint, CounterResetHint::Unrecognized(37));
+    assert_eq!(view.start_timestamp, start);
+    assert_eq!(view.custom_values, [1.5, 4.5]);
+    assert_eq!(view.positive.span_offsets, [1]);
+    assert_eq!(view.positive.span_lengths, [2]);
+    assert_eq!(view.positive.buckets, [3.25, 7.5]);
+    assert_eq!(view.negative.span_offsets, [-2]);
+    assert_eq!(view.negative.span_lengths, [1]);
+    assert_eq!(view.negative.buckets, [0.75]);
+}
+
+#[tokio::test]
+async fn streaming_round_trip_preserves_sliced_multifield_payload_and_owners() {
+    let schema = Arc::new(SeriesSchema {
+        fields: vec!["float", "histogram", "mixed"]
+            .into_iter()
+            .map(|name| FieldSpec {
+                name: Arc::from(name),
+            })
+            .collect::<Vec<_>>()
+            .into(),
+        eval_grid: None,
+    });
+    let time_owner = ScalarBuffer::from(vec![0, 11, 22, 33, 44]);
+    let times = RawTimestamps::try_new(time_owner.slice(1, 3)).unwrap();
+    let float_owner = Float64Array::from(vec![
+        Some(90.0),
+        None,
+        Some(f64::from_bits(0x7ff8_0000_0000_0042)),
+        Some(-0.0),
+        Some(80.0),
+    ]);
+    let float_a = float_owner.slice(1, 2);
+    let float_b = float_owner.slice(3, 1);
+    let floats = Samples::try_from_chunks(vec![
+        Samples::from_float(float_a.clone()),
+        Samples::from_float(float_b.clone()),
+    ])
+    .unwrap();
+
+    let table_owner = histogram_batch();
+    let table_weak = Arc::downgrade(&table_owner);
+    let source_table = table_owner.as_arrow().clone();
+    let sliced_table = Arc::new(
+        datatypes::series::histogram::HistogramBatch::try_from_arrow(source_table.slice(1, 2))
+            .unwrap(),
+    );
+    let sliced_weak = Arc::downgrade(&sliced_table);
+    let index_owner = UInt32Array::from(vec![Some(2), Some(0), Some(1), Some(0), Some(1)]);
+    let indices = index_owner.slice(1, 3);
+    let hist = Samples::try_from_histogram(indices.clone(), sliced_table.clone()).unwrap();
+    let mixed_float_owner = Float64Array::from(vec![Some(1.25), None, None]);
+    let mixed_index_owner = UInt32Array::from(vec![None, Some(0), None]);
+    let mixed = Samples::try_from_mixed(
+        mixed_float_owner.clone(),
+        HistogramSamples::try_new(mixed_index_owner.clone(), sliced_table.clone()).unwrap(),
+    )
+    .unwrap();
+    let chunk = RawSeriesChunk::try_new(
+        &schema,
+        Buffer::from(vec![0xaa_u8, 0x10, 0xbb]),
+        Some(
+            MetricName::try_new(
+                Some(Arc::from("cpu")),
+                vec![Label {
+                    name: Arc::from("host"),
+                    value: Arc::from("a"),
+                }],
+            )
+            .unwrap(),
+        ),
+        times,
+        SeriesFields::try_from_vec(vec![floats, hist, mixed]).unwrap(),
+        true,
+    )
+    .unwrap();
+    let input_times_ptr = chunk.time.as_buffer().as_ptr();
+    let input_float_ptr = float_a.values().as_ptr();
+    let input_hist_indices_ptr = indices.values().as_ptr();
+    let input_mixed_float_ptr = mixed_float_owner.values().as_ptr();
+    let input_mixed_indices_ptr = mixed_index_owner.values().as_ptr();
+
+    let input: SendableRawSeriesStream = Box::pin(RawFixture {
+        schema: schema.clone(),
+        binding: RawLabelSource::Explicit,
+        chunk: Some(chunk),
+    });
+    let frames = pack_raw_series(input).unwrap();
+    let mut decoded = unpack_raw_series(
+        frames,
+        schema,
+        RawLabelSource::Explicit,
+        SeriesOrdering::Contiguous,
+    )
+    .unwrap();
+    let mut output = Vec::new();
+    while let Some(item) = decoded.next().await {
+        output.push(item.unwrap());
+    }
+    assert_eq!(
+        output.len(),
+        2,
+        "one bounded batch still decodes one chunk per union-of-segment row"
+    );
+    assert_eq!(output[0].time.len(), 2);
+    assert!(
+        !output[0].is_last,
+        "a chunk that stops at the field segment boundary is not final"
+    );
+    assert_eq!(output[1].time.len(), 1);
+    assert!(output[1].is_last);
+    assert_eq!(
+        output[0].fields.get(0).unwrap().segments().len(),
+        1,
+        "the decoded chunk is dense again after the frame round trip"
+    );
+
+    // The batch now holds both rows, so its per-row Arrow constituents are concatenated: multi-row
+    // values are copied rather than aliasing the source slices. Multi-row copying is expected and
+    // does not change contents, offsets, or owner cleanup; only genuinely single-row frames keep
+    // the buffers by pointer (see the single-row zero-copy test below).
+    let DenseSamplesRef::Float(decoded_float) = output[0].fields.get(0).unwrap().dense().unwrap()
+    else {
+        panic!("expected float")
+    };
+    assert_ne!(
+        decoded_float.values().as_ptr(),
+        input_float_ptr,
+        "multi-row concatenation copies float values"
+    );
+    assert_eq!(decoded_float.len(), 2);
+    assert!(!decoded_float.is_valid(0));
+    assert_eq!(
+        decoded_float.value(1).to_bits(),
+        f64::from_bits(0x7ff8_0000_0000_0042).to_bits()
+    );
+    let DenseSamplesRef::Histogram(decoded_hist) =
+        output[0].fields.get(1).unwrap().dense().unwrap()
+    else {
+        panic!("expected histogram")
+    };
+    assert_ne!(
+        decoded_hist.indices().values().as_ptr(),
+        input_hist_indices_ptr,
+        "multi-row concatenation copies histogram indices"
+    );
+    assert!(
+        !decoded_hist
+            .batch()
+            .as_arrow()
+            .to_data()
+            .ptr_eq(&sliced_table.as_arrow().to_data()),
+        "multi-row concatenation copies the histogram table"
+    );
+    let DenseSamplesRef::Mixed(decoded_mixed) = output[0].fields.get(2).unwrap().dense().unwrap()
+    else {
+        panic!("expected mixed")
+    };
+    assert_ne!(
+        decoded_mixed.floats().values().as_ptr(),
+        input_mixed_float_ptr
+    );
+    assert_ne!(
+        decoded_mixed.histograms().indices().values().as_ptr(),
+        input_mixed_indices_ptr
+    );
+    assert!(
+        !decoded_mixed
+            .histograms()
+            .batch()
+            .as_arrow()
+            .to_data()
+            .ptr_eq(&sliced_table.as_arrow().to_data()),
+        "multi-row concatenation copies the mixed histogram table"
+    );
+    let first_times = output[0].time.as_buffer();
+    assert_ne!(
+        first_times.as_ptr(),
+        input_times_ptr,
+        "multi-row concatenation copies timestamps"
+    );
+    assert_eq!(&first_times[..], &[11_i64, 22]);
+
+    drop(decoded);
+    drop(time_owner);
+    drop(float_owner);
+    drop(float_a);
+    drop(float_b);
+    drop(index_owner);
+    drop(indices);
+    drop(mixed_float_owner);
+    drop(mixed_index_owner);
+    drop(source_table);
+    drop(table_owner);
+    drop(sliced_table);
+    assert!(table_weak.upgrade().is_none());
+    assert!(sliced_weak.upgrade().is_none());
+
+    let expected_histograms = [(-9.5, Some(456)), (7.25, None), (-9.5, Some(456))];
+    let mut times_seen = Vec::new();
+    let mut float_seen = Vec::new();
+    let mut histogram_seen = Vec::new();
+    let mut mixed_seen = Vec::new();
+    for chunk in &output {
+        assert_eq!(chunk.primary_key.as_slice(), [0xaa, 0x10, 0xbb]);
+        assert_eq!(chunk.metric.as_ref().unwrap().name(), Some("cpu"));
+        assert_eq!(chunk.metric.as_ref().unwrap().get("host"), Some("a"));
+        // Only the row that ends the chunk in the source stream is final.
+        assert_eq!(chunk.is_last, times_seen.len() + chunk.time.len() == 3);
+        for i in 0..chunk.time.len() {
+            let position = times_seen.len();
+            times_seen.push(chunk.time.timestamp(i).unwrap());
+            match chunk.fields.get(0).unwrap().sample(i) {
+                SampleRef::Float(value) => float_seen.push(Some(value.to_bits())),
+                SampleRef::Missing => float_seen.push(None),
+                SampleRef::Histogram(_) => panic!("expected float field"),
+            }
+            let SampleRef::Histogram(value) = chunk.fields.get(1).unwrap().sample(i) else {
+                panic!("expected histogram")
+            };
+            let (sum, start) = expected_histograms[position];
+            assert_histogram(value, sum, start);
+            histogram_seen.push(value.sum);
+            match (position, chunk.fields.get(2).unwrap().sample(i)) {
+                (0, SampleRef::Float(value)) => mixed_seen.push(Some(value.to_bits())),
+                (1, SampleRef::Histogram(value)) => {
+                    assert_histogram(value, -9.5, Some(456));
+                    mixed_seen.push(None);
+                }
+                (2, SampleRef::Missing) => mixed_seen.push(None),
+                _ => panic!("mixed payload variant did not match expected position {position}"),
+            }
+        }
+    }
+    assert_eq!(times_seen, [11, 22, 33]);
+    assert_eq!(
+        float_seen,
+        [
+            None,
+            Some(f64::from_bits(0x7ff8_0000_0000_0042).to_bits()),
+            Some((-0.0_f64).to_bits())
+        ]
+    );
+    assert_eq!(histogram_seen, [-9.5, 7.25, -9.5]);
+    assert_eq!(mixed_seen, [Some(1.25_f64.to_bits()), None, None]);
+}
+
+/// One row of one sample, with every array a nonzero-offset slice of a larger owner: the decoded
+/// chunk keeps the source buffers by pointer, and the values stay readable and bit-exact after
+/// every source owner is dropped.
+#[tokio::test]
+async fn single_row_sliced_payload_keeps_zero_copy_buffers_and_releases_owners() {
+    let schema = Arc::new(SeriesSchema {
+        fields: vec!["float", "histogram", "mixed"]
+            .into_iter()
+            .map(|name| FieldSpec {
+                name: Arc::from(name),
+            })
+            .collect::<Vec<_>>()
+            .into(),
+        eval_grid: None,
+    });
+    let metric = MetricName::try_new(
+        Some(Arc::from("cpu")),
+        vec![Label {
+            name: Arc::from("host"),
+            value: Arc::from("a"),
+        }],
+    )
+    .unwrap();
+
+    let time_owner = ScalarBuffer::from(vec![0_i64, 11, 22]);
+    let times = RawTimestamps::try_new(time_owner.slice(1, 1)).unwrap();
+    let float_owner = Float64Array::from(vec![
+        Some(90.0),
+        Some(f64::from_bits(0x7ff8_0000_0000_0042)),
+    ]);
+    let float = float_owner.slice(1, 1);
+    let table_owner = histogram_batch();
+    let table_weak = Arc::downgrade(&table_owner);
+    let source_table = table_owner.as_arrow().clone();
+    let sliced_table = Arc::new(
+        datatypes::series::histogram::HistogramBatch::try_from_arrow(source_table.slice(1, 1))
+            .unwrap(),
+    );
+    let sliced_weak = Arc::downgrade(&sliced_table);
+    let index_owner = UInt32Array::from(vec![Some(3), Some(0)]);
+    let indices = index_owner.slice(1, 1);
+    let hist = Samples::try_from_histogram(indices.clone(), sliced_table.clone()).unwrap();
+    let mixed_float_owner = Float64Array::from(vec![Some(0.0), Some(-0.0)]);
+    let mixed_floats = mixed_float_owner.slice(1, 1);
+    let mixed_index_owner = UInt32Array::from(vec![Some(0), None]);
+    let mixed_indices = mixed_index_owner.slice(1, 1);
+    let mixed = Samples::try_from_mixed(
+        mixed_floats.clone(),
+        HistogramSamples::try_new(mixed_indices.clone(), sliced_table.clone()).unwrap(),
+    )
+    .unwrap();
+    let chunk = RawSeriesChunk::try_new(
+        &schema,
+        Buffer::from(vec![0xaa_u8, 0x10, 0xbb]),
+        Some(metric),
+        times,
+        SeriesFields::try_from_vec(vec![Samples::from_float(float.clone()), hist, mixed]).unwrap(),
+        true,
+    )
+    .unwrap();
+
+    let input_times_ptr = chunk.time.as_buffer().as_ptr();
+    let input_float_ptr = float.values().as_ptr();
+    let input_hist_indices_ptr = indices.values().as_ptr();
+    let input_mixed_float_ptr = mixed_floats.values().as_ptr();
+    let input_mixed_indices_ptr = mixed_indices.values().as_ptr();
+
+    let input: SendableRawSeriesStream = Box::pin(RawFixture {
+        schema: schema.clone(),
+        binding: RawLabelSource::Explicit,
+        chunk: Some(chunk),
+    });
+    let frames = pack_raw_series(input).unwrap();
+    let mut decoded = unpack_raw_series(
+        frames,
+        schema,
+        RawLabelSource::Explicit,
+        SeriesOrdering::Contiguous,
+    )
+    .unwrap();
+    let mut output = Vec::new();
+    while let Some(item) = decoded.next().await {
+        output.push(item.unwrap());
+    }
+    assert_eq!(output.len(), 1, "one sample packs as one row");
+    assert_eq!(output[0].time.len(), 1);
+    assert!(output[0].is_last);
+
+    let DenseSamplesRef::Float(decoded_float) = output[0].fields.get(0).unwrap().dense().unwrap()
+    else {
+        panic!("expected float")
+    };
+    assert_eq!(decoded_float.values().as_ptr(), input_float_ptr);
+    let DenseSamplesRef::Histogram(decoded_hist) =
+        output[0].fields.get(1).unwrap().dense().unwrap()
+    else {
+        panic!("expected histogram")
+    };
+    assert_eq!(
+        decoded_hist.indices().values().as_ptr(),
+        input_hist_indices_ptr
+    );
+    assert!(
+        decoded_hist
+            .batch()
+            .as_arrow()
+            .to_data()
+            .ptr_eq(&sliced_table.as_arrow().to_data()),
+        "a single row keeps the sliced histogram table by pointer"
+    );
+    let DenseSamplesRef::Mixed(decoded_mixed) = output[0].fields.get(2).unwrap().dense().unwrap()
+    else {
+        panic!("expected mixed")
+    };
+    assert_eq!(
+        decoded_mixed.floats().values().as_ptr(),
+        input_mixed_float_ptr
+    );
+    assert_eq!(
+        decoded_mixed.histograms().indices().values().as_ptr(),
+        input_mixed_indices_ptr
+    );
+    assert!(
+        decoded_mixed
+            .histograms()
+            .batch()
+            .as_arrow()
+            .to_data()
+            .ptr_eq(&sliced_table.as_arrow().to_data())
+    );
+    assert_eq!(output[0].time.as_buffer().as_ptr(), input_times_ptr);
+
+    // The decoded chunk owns the slices, so the source owners can go while the values remain
+    // valid and bit-exact.
+    drop(decoded);
+    drop(time_owner);
+    drop(float_owner);
+    drop(float);
+    drop(index_owner);
+    drop(indices);
+    drop(mixed_float_owner);
+    drop(mixed_floats);
+    drop(mixed_index_owner);
+    drop(mixed_indices);
+    drop(source_table);
+    drop(table_owner);
+    drop(sliced_table);
+    assert!(table_weak.upgrade().is_none());
+    assert!(sliced_weak.upgrade().is_none());
+
+    let chunk = &output[0];
+    assert_eq!(chunk.primary_key.as_slice(), [0xaa, 0x10, 0xbb]);
+    assert_eq!(chunk.metric.as_ref().unwrap().name(), Some("cpu"));
+    assert_eq!(chunk.metric.as_ref().unwrap().get("host"), Some("a"));
+    assert_eq!(chunk.time.timestamp(0), Some(11));
+    match chunk.fields.get(0).unwrap().sample(0) {
+        SampleRef::Float(value) => assert_eq!(
+            value.to_bits(),
+            f64::from_bits(0x7ff8_0000_0000_0042).to_bits()
+        ),
+        SampleRef::Missing | SampleRef::Histogram(_) => panic!("expected float field"),
+    }
+    let SampleRef::Histogram(value) = chunk.fields.get(1).unwrap().sample(0) else {
+        panic!("expected histogram")
+    };
+    assert_histogram(value, -9.5, Some(456));
+    match chunk.fields.get(2).unwrap().sample(0) {
+        SampleRef::Float(value) => assert_eq!(value.to_bits(), (-0.0_f64).to_bits()),
+        SampleRef::Missing | SampleRef::Histogram(_) => panic!("expected mixed float"),
+    }
+}

src/datatypes/src/series.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+pub mod frame;
 pub mod histogram;
 pub mod raw;
 pub mod samples;

src/datatypes/src/series/samples.rs

@@ -199,6 +199,46 @@ impl Samples {
             storage: SampleStorage::Chunks(flat.into(), total),
         })
     }
+    /// Returns the dense segments in logical order without allocating, copying payloads, or
+    /// densifying the samples.
+    ///
+    /// Chunked samples expose their retained segment boundaries; different fields may therefore
+    /// have different boundaries. The logical start of each segment is the sum of the lengths of
+    /// all preceding segments, and the total length is [`Self::len`]. For non-chunked samples,
+    /// including empty samples, this returns a singleton slice containing `self`. Chunk
+    /// construction flattens nested chunks and drops empty segments; an all-empty input becomes
+    /// one empty float segment. Each segment is dense: [`Self::dense`] returns `Some` with its
+    /// typed Arrow view. Clone the samples cheaply if they need to outlive this borrow.
+    ///
+    /// ```
+    /// use arrow::array::Float64Array;
+    /// use datatypes::series::samples::{DenseSamplesRef, Samples};
+    ///
+    /// let samples = Samples::try_from_chunks(vec![
+    ///     Samples::from_float(Float64Array::from(vec![1.0, 2.0])),
+    ///     Samples::from_float(Float64Array::from(vec![3.0])),
+    /// ])
+    /// .unwrap();
+    ///
+    /// let mut logical_start = 0;
+    /// let mut starts = Vec::new();
+    /// for segment in samples.segments() {
+    ///     starts.push(logical_start);
+    ///     match segment.dense().unwrap() {
+    ///         DenseSamplesRef::Float(values) => assert_eq!(values.len(), segment.len()),
+    ///         _ => unreachable!("these segments contain floats"),
+    ///     }
+    ///     logical_start += segment.len();
+    /// }
+    /// assert_eq!(starts, vec![0, 2]);
+    /// assert_eq!(logical_start, samples.len());
+    /// ```
+    pub fn segments(&self) -> &[Samples] {
+        match &self.storage {
+            SampleStorage::Chunks(chunks, _) => chunks,
+            _ => std::slice::from_ref(self),
+        }
+    }
     /// Number of sample positions.
     pub fn len(&self) -> usize {
         match &self.storage {
@@ -315,8 +355,7 @@ impl Samples {
         }
     }
     /// Materialize chunks into one dense array. When all histogram-bearing chunks reference the
-    /// same table, reuse it; otherwise concatenate tables per chunk and remap indices, including
-    /// unused rows.
+    /// same Arrow table view, reuse it; otherwise concatenate tables per chunk and remap indices.
     pub fn into_dense(self) -> Result<Self> {
         let SampleStorage::Chunks(chunks, total_len) = self.storage else {
             return Ok(self);
@@ -350,9 +389,17 @@ impl Samples {
             SampleStorage::Mixed(mixed) => Some(&mixed.histograms.batch),
             SampleStorage::Float(_) | SampleStorage::Chunks(_, _) => None,
         });
-        let shared_batch = histogram_batches
-            .next()
-            .filter(|first| histogram_batches.all(|batch| Arc::ptr_eq(first, batch)));
+        let mut first_data = None;
+        let shared_batch = histogram_batches.next().filter(|first| {
+            histogram_batches.all(|batch| {
+                if Arc::ptr_eq(first, batch) {
+                    return true;
+                }
+                first_data
+                    .get_or_insert_with(|| first.as_arrow().to_data())
+                    .ptr_eq(&batch.as_arrow().to_data())
+            })
+        });
         let needs_float = chunks.iter().any(|chunk| {
             matches!(
                 chunk.storage,

src/datatypes/src/series/samples/tests.rs

@@ -15,8 +15,8 @@
 use super::*;
 use std::sync::atomic::{AtomicUsize, Ordering};
 
-use arrow::array::{Float64Array, UInt32Array};
-use arrow::buffer::{Buffer, ScalarBuffer};
+use arrow::array::{ArrayRef, Float64Array, Int64Array, StructArray, UInt32Array};
+use arrow::buffer::{BooleanBuffer, Buffer, NullBuffer, ScalarBuffer};
 
 use crate::series::histogram::{BucketSideView, CounterResetHint, HistogramBatchBuilder};
 
@@ -235,6 +235,308 @@ fn chunk_iterator_is_sequential_exact_and_handles_many_singletons() {
     assert!(iter.next().is_none());
 }
 
+#[test]
+fn densification_reuses_distinct_wrappers_of_the_same_arrow_table_view() {
+    let base_row = HistogramView {
+        schema: 2,
+        zero_threshold: 0.25,
+        sum: 7.0,
+        count: 8.0,
+        zero_count: 1.0,
+        reset_hint: CounterResetHint::Gauge,
+        start_timestamp: None,
+        custom_values: &[],
+        positive: BucketSideView {
+            span_offsets: &[0],
+            span_lengths: &[1],
+            buckets: &[9.0],
+        },
+        negative: BucketSideView {
+            span_offsets: &[],
+            span_lengths: &[],
+            buckets: &[],
+        },
+    };
+    let rows = [
+        HistogramView {
+            schema: 1,
+            zero_threshold: 0.0,
+            sum: 1.0,
+            count: 2.0,
+            zero_count: 0.0,
+            reset_hint: CounterResetHint::Unknown,
+            start_timestamp: Some(11),
+            custom_values: &[1.0],
+            positive: BucketSideView {
+                span_offsets: &[0],
+                span_lengths: &[1],
+                buckets: &[3.0],
+            },
+            negative: BucketSideView {
+                span_offsets: &[],
+                span_lengths: &[],
+                buckets: &[],
+            },
+        },
+        HistogramView {
+            sum: -0.0,
+            count: f64::NAN,
+            start_timestamp: Some(42),
+            custom_values: &[f64::from_bits(0x7ff8_0000_0000_0042)],
+            ..base_row
+        },
+    ];
+    let backing = histogram_batch(&[rows[0], rows[1], rows[1]]);
+    let original =
+        Arc::new(HistogramBatch::try_from_arrow(backing.as_arrow().slice(1, 1)).unwrap());
+    let wrapped = Arc::new(HistogramBatch::try_from_arrow(original.as_arrow().clone()).unwrap());
+    assert!(!Arc::ptr_eq(&original, &wrapped));
+    assert!(
+        original
+            .as_arrow()
+            .to_data()
+            .ptr_eq(&wrapped.as_arrow().to_data())
+    );
+
+    let mixed = Samples::try_from_mixed(
+        Float64Array::from(vec![
+            Some(-0.0),
+            Some(f64::from_bits(0x7ff8_0000_0000_0042)),
+            None,
+        ]),
+        HistogramSamples::try_new(UInt32Array::from(vec![None, None, Some(0)]), wrapped).unwrap(),
+    )
+    .unwrap();
+    let dense = Samples::try_from_chunks(vec![
+        Samples::try_from_histogram(UInt32Array::from(vec![Some(0), None]), original.clone())
+            .unwrap(),
+        mixed,
+    ])
+    .unwrap()
+    .into_dense()
+    .unwrap();
+    let DenseSamplesRef::Mixed(values) = dense.dense().unwrap() else {
+        panic!()
+    };
+    assert_eq!(values.histograms.batch().len(), 1);
+    assert!(Arc::ptr_eq(values.histograms.batch(), &original));
+    assert_eq!(values.histograms.indices().value(0), 0);
+    assert!(values.histograms.indices().is_null(1));
+    assert!(values.histograms.indices().is_null(2));
+    assert!(values.histograms.indices().is_null(3));
+    assert_eq!(values.histograms.indices().value(4), 0);
+    let expected = [
+        SampleRef::Histogram(original.view(0).unwrap()),
+        SampleRef::Missing,
+        SampleRef::Float(-0.0),
+        SampleRef::Float(f64::from_bits(0x7ff8_0000_0000_0042)),
+        SampleRef::Histogram(original.view(0).unwrap()),
+    ];
+    for (index, expected) in expected.into_iter().enumerate() {
+        assert!(
+            same_sample(dense.sample(index), expected),
+            "position {index}"
+        );
+    }
+}
+
+#[test]
+fn densification_does_not_merge_equal_content_distinct_buffers() {
+    let first = batch(2.0, 3.0, &[4.0]);
+    let equal = batch(2.0, 3.0, &[4.0]);
+    assert!(
+        !first
+            .as_arrow()
+            .to_data()
+            .ptr_eq(&equal.as_arrow().to_data())
+    );
+    let dense = Samples::try_from_chunks(vec![
+        Samples::try_from_histogram(UInt32Array::from(vec![Some(0)]), first.clone()).unwrap(),
+        Samples::try_from_histogram(UInt32Array::from(vec![Some(0)]), equal.clone()).unwrap(),
+    ])
+    .unwrap()
+    .into_dense()
+    .unwrap();
+    let DenseSamplesRef::Histogram(values) = dense.dense().unwrap() else {
+        panic!()
+    };
+    assert_eq!(values.batch().len(), 2);
+    assert_eq!(values.indices().value(0), 0);
+    assert_eq!(values.indices().value(1), 1);
+    assert!(same_sample(
+        dense.sample(0),
+        SampleRef::Histogram(first.view(0).unwrap())
+    ));
+    assert!(same_sample(
+        dense.sample(1),
+        SampleRef::Histogram(equal.view(0).unwrap())
+    ));
+}
+
+#[test]
+fn densification_does_not_merge_distinct_table_slices() {
+    let base_row = HistogramView {
+        schema: 1,
+        zero_threshold: 0.0,
+        sum: 2.0,
+        count: 3.0,
+        zero_count: 0.0,
+        reset_hint: CounterResetHint::Unknown,
+        start_timestamp: None,
+        custom_values: &[],
+        positive: BucketSideView {
+            span_offsets: &[0],
+            span_lengths: &[1],
+            buckets: &[4.0],
+        },
+        negative: BucketSideView {
+            span_offsets: &[],
+            span_lengths: &[],
+            buckets: &[],
+        },
+    };
+    let rows = [
+        base_row,
+        HistogramView {
+            start_timestamp: Some(42),
+            sum: 12.0,
+            ..base_row
+        },
+        HistogramView {
+            start_timestamp: Some(99),
+            sum: 19.0,
+            ..base_row
+        },
+    ];
+    let backing = histogram_batch(&rows);
+    let first = Arc::new(HistogramBatch::try_from_arrow(backing.as_arrow().slice(0, 1)).unwrap());
+    let equal_len_other_offset =
+        Arc::new(HistogramBatch::try_from_arrow(backing.as_arrow().slice(1, 1)).unwrap());
+    let longer_other_offset =
+        Arc::new(HistogramBatch::try_from_arrow(backing.as_arrow().slice(1, 2)).unwrap());
+
+    for (second, expected_table_len, second_expected) in [
+        (
+            equal_len_other_offset,
+            2,
+            SampleRef::Histogram(backing.view(1).unwrap()),
+        ),
+        (
+            longer_other_offset,
+            3,
+            SampleRef::Histogram(backing.view(1).unwrap()),
+        ),
+    ] {
+        let dense = Samples::try_from_chunks(vec![
+            Samples::try_from_histogram(UInt32Array::from(vec![Some(0)]), first.clone()).unwrap(),
+            Samples::try_from_histogram(UInt32Array::from(vec![Some(0)]), second).unwrap(),
+        ])
+        .unwrap()
+        .into_dense()
+        .unwrap();
+        let DenseSamplesRef::Histogram(values) = dense.dense().unwrap() else {
+            panic!()
+        };
+        assert_eq!(values.batch().len(), expected_table_len);
+        assert_eq!(values.indices().value(0), 0);
+        assert_eq!(values.indices().value(1), 1);
+        assert!(same_sample(
+            dense.sample(0),
+            SampleRef::Histogram(first.view(0).unwrap())
+        ));
+        assert!(same_sample(dense.sample(1), second_expected));
+    }
+}
+
+#[test]
+fn densification_does_not_merge_shared_value_buffer_with_different_null_bitmap_offsets() {
+    let source = batch(2.0, 3.0, &[4.0]);
+    let start_values = ScalarBuffer::from(vec![42_i64]);
+    let bitmap = Buffer::from(vec![0b10_u8]);
+    let missing_starts = Int64Array::new(
+        start_values.clone(),
+        Some(NullBuffer::new(BooleanBuffer::new(bitmap.clone(), 0, 1))),
+    );
+    let present_starts = Int64Array::new(
+        start_values.clone(),
+        Some(NullBuffer::new(BooleanBuffer::new(bitmap.clone(), 1, 1))),
+    );
+    assert_eq!(
+        missing_starts.values().inner().as_ptr(),
+        present_starts.values().inner().as_ptr()
+    );
+    let missing_nulls = missing_starts.nulls().unwrap();
+    let present_nulls = present_starts.nulls().unwrap();
+    assert_eq!(
+        missing_nulls.buffer().as_ptr(),
+        present_nulls.buffer().as_ptr()
+    );
+    assert_ne!(missing_nulls.offset(), present_nulls.offset());
+    assert!(missing_starts.is_null(0));
+    assert_eq!(present_starts.value(0), 42);
+
+    let with_start = |starts: Int64Array| {
+        let mut columns = source.as_arrow().columns().to_vec();
+        columns[6] = Arc::new(starts) as ArrayRef;
+        Arc::new(
+            HistogramBatch::try_from_arrow(
+                StructArray::try_new(source.as_arrow().fields().clone(), columns, None).unwrap(),
+            )
+            .unwrap(),
+        )
+    };
+    let missing = with_start(missing_starts);
+    let present = with_start(present_starts);
+    let missing_starts = missing
+        .as_arrow()
+        .column(6)
+        .as_any()
+        .downcast_ref::<Int64Array>()
+        .unwrap();
+    let present_starts = present
+        .as_arrow()
+        .column(6)
+        .as_any()
+        .downcast_ref::<Int64Array>()
+        .unwrap();
+    assert_eq!(
+        missing_starts.values().inner().as_ptr(),
+        present_starts.values().inner().as_ptr()
+    );
+    assert_eq!(
+        missing_starts.nulls().unwrap().buffer().as_ptr(),
+        present_starts.nulls().unwrap().buffer().as_ptr()
+    );
+    assert_ne!(
+        missing_starts.nulls().unwrap().offset(),
+        present_starts.nulls().unwrap().offset()
+    );
+    assert!(missing_starts.is_null(0));
+    assert_eq!(present_starts.value(0), 42);
+    assert!(
+        !missing
+            .as_arrow()
+            .to_data()
+            .ptr_eq(&present.as_arrow().to_data())
+    );
+    let dense = Samples::try_from_chunks(vec![
+        Samples::try_from_histogram(UInt32Array::from(vec![Some(0)]), missing.clone()).unwrap(),
+        Samples::try_from_histogram(UInt32Array::from(vec![Some(0)]), present.clone()).unwrap(),
+    ])
+    .unwrap()
+    .into_dense()
+    .unwrap();
+    assert_eq!(dense.len(), 2);
+    assert!(same_sample(
+        dense.sample(0),
+        SampleRef::Histogram(missing.view(0).unwrap())
+    ));
+    assert!(same_sample(
+        dense.sample(1),
+        SampleRef::Histogram(present.view(0).unwrap())
+    ));
+}
+
 #[test]
 fn densifying_slices_keeps_shared_histogram_table() {
     let table = batch(2.0, 3.0, &[4.0, 5.0]);