mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-11-14 23:06:59 +00:00
feat(datatypes): add finite raw series frame codec (#9607)
I hereby agree to the terms of the [GreptimeDB CLA](https://github.com/GreptimeTeam/.github/blob/main/CLA.md). ## Refer to a related PR or issue link (optional) - Depends on: #9603 (`feat/narthex-raw-contract`). - Base of: #9611 (`feat/narthex-raw-frame-stream`). - The finite raw SeriesFrame codec remains independent of the Mito producer in #9608. ## What's changed and what's your intention? Adds an Arrow `RecordBatch` representation for the raw series chunks defined in the stack, so subsequent execution and transport code can exchange typed samples without rebuilding every sample, plus the bounded multirow encoder that packs raw series chunks into multi-row batches directly. **Codec.** The new `series::frame` module provides raw-frame schema construction, chunk packing and whole-batch unpacking, in encoded-key and explicit-label modes, preserving explicit timestamps, float/histogram/mixed payloads, nulls, field order and `is_last`. Different fields may have different chunk boundaries: packing splits at their union and retains Arrow views rather than densifying the inputs. `Samples::segments()` exposes the zero-copy dense segment view the codec walks, with a public-call doctest, so the codec can consume existing dense segments without a new segment type or allocation. Decode validates schema metadata, payload lengths, indices, histogram structure and timestamps, and any malformed row fails the entire unpacking call rather than returning a successful prefix. **Bounded multirow encoding.** `RawSeriesFrameBuilder` buffers rows and emits one batch when either bound is reached (`DEFAULT_MAX_ROWS` 256, `DEFAULT_MAX_BYTES` 1 MiB); the effective byte cap is stored as `min(requested, i32::MAX)`, which keeps every cumulative Arrow offset domain representable for any admitted group. `RawSeriesChunkCursor` walks one already validated chunk at the union of the field segment boundaries, holds at most one pending measured row, and measures each row exactly once; the pending row is retried after a flush, so no row is dropped or split, and per-field offsets advance against each field's original segment length, which keeps nonzero-offset (sliced) sample arrays correct. The byte measure dominates the encoder's offset domains: key and timestamp bytes, list element counts and bitmap bytes, and the visible extent of each histogram table list column read from the rebased `value_offsets` view rather than from retained capacity. It is an internal batching work estimate — not an exact IPC size and not a resident-memory or RSS cap — so one individually legal row that exceeds the requested cap is still accepted and emitted alone, because that slice cannot be split into smaller legal rows. **Histogram layout and sharing.** Row-local histogram indices are never remapped; physical tables are concatenated under cumulative row list offsets, so each row still addresses its own table through a nonzero slice of the child. A row group with a single Arrow constituent reuses it by clone, which shares buffers and keeps the single-row zero-copy and owner-drop behavior; a group with several constituents is concatenated, which copies. Both cases are asserted. The `Samples::into_dense` sharing check compares the Arrow table view (`ArrayData` identity) in addition to `Arc` identity, and the first histogram batch's `ArrayData` is computed lazily and cached: that is pointer-identity caching, not content deduplication, and distinct buffers with equal content are never merged. Local buffer sharing does not prevent a full table from being encoded in every batch; any future trimming or remapping belongs at the actual IPC/network/spill boundary, reusing Arrow slicing and selection and the existing writer, and preserving duplicate indices without content hashing. **Preserved interfaces.** `pack_raw_series_chunk` keeps its `Vec<RecordBatch>` return type and `unpack_raw_series_frame` keeps its signature; `RawSeriesFrameDecoder` prepares the expected frame schema once for a stream of batches, while the wrapper still builds a decoder per call. The raw contract's `RawTimestamps` validates strictly increasing source timestamps once at construction behind a private buffer. The frozen `v1` contract is untouched: the four reserved metadata entries, the column names (`read_key`, `is_last`, `time`, `fields`, plus `name`/`labels` in explicit mode) and the wire column literal `read_key` stay as they were (`primary_key` is only the Rust-side name of that buffer). Histogram-only rows decode through `Samples::try_from_histogram` directly, without a duplicate bounds check. **Tests.** The frame tests cover schema and metadata acceptance and rejection, bit-exact payloads (including NaN payloads and `-0.0`), sliced-payload owner lifetimes, row and byte bounds (bounds taken from real measured rows, flush-before-exceed with pending retry, legal oversized row), row-local histogram indices and complete tables with a real Arrow IPC round trip, decoder schema caching and label-mode separation. The `samples` tests pin both sides of the sharing rule: distinct wrappers of one Arrow table view are reused, while distinct buffers, distinct table slices, and shared value buffers with different null-bitmap offsets are never merged. **Limitations and boundaries.** Each histogram-bearing row carries its complete histogram table, including entries no index references, and a large table referenced by several segments can be repeated per row; there is no deduplication and no global reindex, so an encoded batch can exceed the source sample bytes. These entry points are exercised by unit tests, and the builder, cursor and decoder are also used by the in-process stream adapter of #9611, which is not yet wired into a production query path; there is no transport, persistence or query-performance change. This PR handles finite raw frames only: it does not add stream-wide key/order/EOF validation (that is #9611), evaluated frames, a Flight/IPC execution path, logical-name authentication, a native query consumer, a production query-path switch, distributed or cross-region merging, or a prepared producer. The frame preserves ordered logical field names and rejects mismatched supplied schemas; that is not authentication of label binding or data provenance. Existing persisted storage formats and query entrypoints are unchanged, and this is not a cross-version wire-compatibility claim. Open and unmerged; current-head CI has completed successfully. ### Bounded local Frame-boundary measurement (B baseline single-row, C selected multirow) Provenance: measured at the frozen local candidate `eddd595b82888452b3c066678fabc6c1200c74f7`, whose bounded-multirow finite codec content is `3406a5334574a518924c67c319e2d92b6cac943b`, against the old public baseline `36843e8cebeded1328dd7005ed668300a606f182`. The later normal parent synchronization and the strict-lint syntax-only fix are not separately rebenchmarked, so these numbers describe that frozen local content, not the pushed head; no remote CI or performance result is implied by this local measurement. This PR now implements the selected C direction and native A is unchanged: the native direct path remains the control and is not replaced. A bounded local-only experiment measured this codec's role as the pack/unpack kernel of that direction. Fixture: six synthetic raw-series Frame-boundary cases in two sequential campaigns (100000 series in the first, 1000 in the second, one chunk per series in both), CPU affinity 2, seven repetitions, default warmup, plain-timing and allocator-instrumented runs kept separate, no retries and no excluded rounds. Stage wall time from the plain runner, milliseconds, median only, for the six cases in table order: | Case | Series / rows per series | A0 native, baseline build | A1 native, candidate build | B baseline single-row | C selected multirow | |---|---:|---:|---:|---:|---:| | float2_key16 | 100000 / 2 | 42.313 | 43.520 | 2549.316 | 206.888 | | float3_key16 | 100000 / 3 | 59.434 | 60.111 | 3261.311 | 260.933 | | float3_key128 | 100000 / 3 | 64.968 | 62.999 | 3240.701 | 291.846 | | long_float1024_key16 | 1000 / 1024 | 7.556 | 7.113 | 26.311 | 10.468 | | histogram3_key16 | 1000 / 3 | 0.452 | 0.448 | 17.250 | 5.632 | | multifield_float_hist3_key16 | 1000 / 3 | 0.593 | 0.588 | 25.194 | 6.328 | The same run's full min–max, MAD, all seven rounds per role, allocation counters, RSS, frame counts and acceptance checks are in #9611, which carries the canonical full results — including the frame totals counted from the runs (B emitted 100000 / 100000 / 100000 / 1000 / 1000 / 1000 batches and C 391 / 391 / 391 / 17 / 4 / 4, with frame rows equal to series × chunks in every case and no frames on the native path). Reading: the A0 and A1 native control ranges overlap in every case, so the native controls establish no candidate-build difference, and the C/B range disjointness in this bounded seven-round sample is descriptive for these synthetic fixtures only. Relative to B, C lowers stage wall time and allocation request counts while raising phase peak-live bytes and whole-process RSS; that batching-versus-retained-memory tradeoff of the selected direction is accepted once, as part of that direction, and is not a veto or an open design question (largest observed increase: the long-float case, peak live Δ 2156848 B versus 50438 B and lifetime RSS 9176 kB versus 6080 kB). Scope limits, as in #9611: every SOURCE_KEY group uses one identical explicit metric label set, so this is not real PromQL label diversity or cardinality evidence; the histogram table template is built outside the measured phase; no `Samples::Mixed` inside one field. The measured scope is local Frame generation, the codec boundary, digests and phase-owned cleanup — not a Flight/IPC network path, not a whole-query or query-regression benchmark, not a storage or service measurement, and not a CI result. No query speedup is claimed from this codec or from the selected direction. ### Verification Pushed head: `ba88f817cdcffeae6594943c55aaced2ffa769e5`, synchronized normally with #9603 at `794389688084ee97a7760d1ae993bcf1c109489e`. Checks on the exact frozen source recorded by this head: - `cargo +1.98.1 check --locked -p datatypes --all-targets` — passed. - `cargo +1.98.1 clippy --locked -p datatypes --all-targets --all-features -- -D warnings` — passed with the official pinned 1.98.1 Clippy component. - `cargo +1.98.1 nextest run --locked -p datatypes --retries 0 --no-fail-fast` — **363 passed, 0 skipped**. - `cargo +1.98.1 test --locked -p datatypes --doc series::samples::Samples::segments` — **1 passed**. - Rust 1.99 formatting, TOML/license/typo checks, `git diff --check`, and normal signed-off commit/push hooks passed. Source hashes were verified unchanged after the hooks. The final closeout only collapses the full-builder flush guard into a short-circuit-preserving let-chain for strict Clippy; it changes no batching behavior or assertion. The local measurements above remain attributed to their frozen candidate. Current-head remote CI completed successfully in all three workflows below. The PR remains open and unmerged until the normal merge gates and parent-first integration are checked. ## PR Checklist Please convert it to a draft if some of the following conditions are not met. - [x] I have written the necessary rustdoc comments. - [x] I have added the necessary unit tests and integration tests. - [ ] This PR requires documentation updates. - [x] API changes are backward compatible. - [x] Schema or data changes are backward compatible. - [ ] This PR needs to be backported to release branches, and I have added the `backport-<target>` labels (e.g. `backport-v1.3` targets `release/v1.3`). ### Verified current-head remote CI Exact pushed head: `ba88f817cdcffeae6594943c55aaced2ffa769e5`. - [Rust CI](https://github.com/GreptimeTeam/greptimedb/actions/runs/38882922544) — **success**. - [Checks](https://github.com/GreptimeTeam/greptimedb/actions/runs/38882922504) — **success**. - [Integration CI](https://github.com/GreptimeTeam/greptimedb/actions/runs/38882922665) — **success**. Linux and Windows test jobs, strict Clippy, builds, SQLness and the executed integration/fuzz jobs succeeded. Coverage and recent-release compatibility jobs were skipped by their workflow conditions; they are not claimed as executed. These CI results are independent of the frozen local experiment provenance above.
This commit is contained in:
discord9
@@ -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;
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -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,
|
||||
|
||||
@@ -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]);
|
||||
|
||||
Reference in New Issue
Block a user