starred/openobserve
1
0
Fork 0
mirror of https://github.com/openobserve/openobserve.git synced 2026-11-05 00:09:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(metrics): TSID-major Parquet layout behind ZO_METRICS_TSID_MAJOR_ENABLED (#13882)

Browse Source 
The storage-layout half of #13839, without the `.sidx` series index and
its PromQL query path (those stay in #13839, which will rebase on this).

## What

Behind `ZO_METRICS_TSID_MAJOR_ENABLED` (default off), Parquet metrics
files of a stream whose `__hash__` column is `UInt64` are written
ordered by `(__hash__, _timestamp)` so every series is contiguous:

- **`config::meta::promql::tsid_layout`** —
`metrics_tsid_major_enabled(stream_type)` (flag + configured Parquet),
`metrics_tsid_major_stream(stream_type, schema)` (also `__hash__:
UInt64`; JSON-ingested metrics store `__hash__` as a string and keep the
classic layout), and `MetricsFileLayout { Legacy, HashSorted, TsidMajor
}` encoded in the file-name prefix (`tsid-sorted-*`, `tsid-major-v3-*`)
so readers and later merges know the physical order without opening the
file.
- **`MergeMode::HashSorted`** (open hour): the ingester movers and
incremental compactor merges write one hash-ordered file named
`tsid-sorted-*`. In pack mode the single-segment upload-as-is fast path
is skipped for such streams (segments are in arrival order).
- **`MergeMode::TsidMajor`** (closed hour): `merges_whole_batch()` —
every file of the hour, regardless of size — into size-split
`tsid-major-v3-*` files written at RecordBatch boundaries using
`ZO_COMPACT_MAX_FILE_SIZE` as logical original size
(`search/datafusion/merge/tsid_major.rs`). `MergeMode::for_compactor /
for_ingester` take the schema and pick the mode via
`metrics_tsid_major_stream`; downsampling keeps precedence. The
compactor skips a closed hour that is already all `tsid-major-v3-*`.
- **`MergedFile.layout`** decides the output file name
(`layout.file_name` / `mark_file_key`) in the compactor and the ingester
movers.
- **Merging pre-sorted inputs**: `MergeMode::input_sort_order(&files)`
declares `FileSortOrder::HashTimestampAsc` when every input is
hash-ordered (`tsid-sorted-*` / `tsid-major-v3-*`). Overlapping hash
ranges make DataFusion drop the order for its size-based groups; the
listing adapter regroups one file per partition (single-file groups are
trivially ordered) and keeps the declared `output_ordering`, so
`EnforceSorting` turns the `SortExec` into a `SortPreservingMergeExec`
over N pre-sorted streams (memory N × batch instead of the whole hour).
A batch that mixes layouts declares no order and falls back to the full
sort.

Two file states only: `tsid-sorted-*` (ordered, not finalized) and
`tsid-major-v3-*` (finalized). Nothing changes for streams without the
flag, non-Parquet metrics, or a non-UInt64 `__hash__`.

## Known limitation

The SQL query path still derives `sorted_by_time` from the SQL shape
only, so with the flag on a `SELECT … FROM <metrics> ORDER BY _timestamp
DESC LIMIT n` may assume a timestamp order that hash-sorted files do not
have. PromQL is unaffected (no ordering declared). To be addressed with
the query-side work.

## Validation

- `cargo fmt --check`; `cargo check` / `cargo clippy` on config, search,
compaction, jobs, promql-service, search_service (`--tests`) clean
- `cargo test -p config --lib -- tsid_layout` 2 passed; `cargo test -p
search --lib -- merge:: mode:: tsid_major listing_adapter` 9 passed
(mode selection, per-layout input order, hash `ORDER BY`, size-split
rotation, SPM plan shape); `cargo test -p compaction --lib -- merge` 31
passed
- End-to-end on #13839 (debug build, local disk, OTLP gauge 300 series):
ingester uploads `tsid-sorted-*` verified `(__hash__, _timestamp)`
ordered; hour-end `merge [tsid_major] input sort order:
hash_timestamp_asc` → plan `SortPreservingMergeExec` ← `DataSourceExec{N
single-file groups}` (no `SortExec`) → `tsid-major-v3-*`; incremental
`merge [hash_sorted]` 2 → 1 `tsid-sorted-*`; JSON-ingested stream (Utf8
`__hash__`) keeps the classic layout with no compactor errors.

## TODO (follow-up, not in this PR)

**Sort TSID-major streams by `(__hash__, _timestamp)` already at WAL
dump time.** `Partition::persist` full-sorts every dumped chunk anyway
(today by `_timestamp DESC` in `merge_record_batches`); switching the
sort key for TSID-major streams is near-free at dump time and removes
the biggest ingester-side cost of this layout — the full `SortExec` over
the merge input:

1. **persist**: lexsort the chunk by `(__hash__ ASC, _timestamp ASC)`
(same gate as the writers, `metrics_tsid_major_stream`) and stamp a
`sort_order: hash_timestamp_asc` key-value into the Parquet footer. A
file-name marker is not an option for WAL files —
`parse_time_range_from_filename` splits on `.` and a prefix would break
WAL time pruning.
2. **ingester mover**: it already reads every input's footer for the
schema; declare `HashTimestampAsc` on the table only when *all* inputs
carry the footer marker (else `FileSortOrder::None`, full sort — handles
mixed files during rolling upgrades / flag flips, same role
`MetricsFileLayout::of` plays for the compactor). The merge then becomes
a `SortPreservingMergeExec` over pre-sorted files instead of a full
sort.
3. **pack, single segment**: the as-is fast path (skipped for TSID-major
streams in this PR because segments were time-sorted) can come back —
the segment is already hash-sorted, upload it unchanged and mark it
`tsid-sorted-*`.
4. **pack, multiple segments**: one `MemTable` partition per segment
with a declared sort order, so the in-memory merge is also an SPM
instead of a full sort.

Query-side is already safe for this change: metrics SQL never declares
`sorted_by_time` and PromQL declares no order, so hash-sorted WAL files
cannot mislead a reader. The cost is slightly coarser time clustering
inside WAL files, negligible at WAL file sizes. Steps 1–2 carry most of
the win; 3–4 are incremental.
This commit is contained in:
Huaijin
2026-08-20 09:20:00 +00:00
committed by GitHub
parent 6930a9b724
commit 789eee463f
13 changed files with 930 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+67
View File
@@ -1122,6 +1122,7 @@ dependencies = [
"aws-credential-types",
"aws-sigv4",
"aws-smithy-async",
"aws-smithy-eventstream",
"aws-smithy-http",
"aws-smithy-runtime",
"aws-smithy-runtime-api",
@@ -1138,6 +1139,33 @@ dependencies = [
"uuid",
]
[[package]]
name = "aws-sdk-lambda"
version = "1.138.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b002aa94953a1aab8534280436ea3cf7805b821d51ebb49a89da41764650ce0b"
dependencies = [
"arc-swap",
"aws-credential-types",
"aws-runtime",
"aws-smithy-async",
"aws-smithy-eventstream",
"aws-smithy-http",
"aws-smithy-json",
"aws-smithy-observability",
"aws-smithy-runtime",
"aws-smithy-runtime-api",
"aws-smithy-schema",
"aws-smithy-types",
"aws-types",
"bytes",
"fastrand",
"http 0.2.12",
"http 1.5.0",
"regex-lite",
"tracing",
]
[[package]]
name = "aws-sdk-sns"
version = "1.107.0"
@@ -1251,6 +1279,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "723c2234ad7511ceef63eab016b7ba6ff7c55590fefb96fa8467af014a07309f"
dependencies = [
"aws-credential-types",
"aws-smithy-eventstream",
"aws-smithy-http",
"aws-smithy-runtime-api",
"aws-smithy-types",
@@ -1277,12 +1306,24 @@ dependencies = [
"tokio",
]
[[package]]
name = "aws-smithy-eventstream"
version = "0.61.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5a9381123ab62d20c13082b151f30f962a3b112b727345394536dfa39a482944"
dependencies = [
"aws-smithy-types",
"bytes",
"crc32fast",
]
[[package]]
name = "aws-smithy-http"
version = "0.64.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "37843d9add67c3aff5856f409c6dc315d3cdff60f9c0cb5b670dab1e9920306d"
dependencies = [
"aws-smithy-eventstream",
"aws-smithy-runtime-api",
"aws-smithy-types",
"bytes",
@@ -7882,6 +7923,7 @@ dependencies = [
"openobserve-node",
"openobserve-org-storage",
"openobserve-sourcemaps",
"openobserve-synthetics",
"opentelemetry",
"pprof",
"promql-service",
@@ -8121,6 +8163,7 @@ dependencies = [
"openobserve-core",
"openobserve-node",
"openobserve-org-storage",
"openobserve-synthetics",
"prometheus",
"promql-service",
"proto",
@@ -8217,6 +8260,30 @@ dependencies = [
"zip 8.6.0",
]
[[package]]
name = "openobserve-synthetics"
version = "0.1.0"
dependencies = [
"anyhow",
"aws-config",
"aws-sdk-lambda",
"chrono",
"config",
"dashmap",
"infra",
"log",
"o2_enterprise",
"o2_openfga",
"reqwest 0.13.4",
"sea-orm",
"serde",
"serde_json",
"svix-ksuid",
"tokio",
"tracing",
"urlencoding",
]
[[package]]
name = "openobserve-tls"
version = "0.1.0"
src/compaction/src/merge.rs → src/compaction/src/merge/mod.rs
+45 -6
View File
@@ -22,9 +22,12 @@ use config::{
FileFormat,
cluster::LOCAL_NODE,
get_config, ider, is_local_disk_storage,
meta::stream::{
FileKey, FileListDeleted, FileMeta, MergeStrategy, PartitionTimeLevel, StorageType,
StreamType,
meta::{
promql::tsid_layout::MetricsFileLayout,
stream::{
FileKey, FileListDeleted, FileMeta, MergeStrategy, PartitionTimeLevel, StorageType,
StreamType,
},
},
metrics,
utils::{
@@ -58,6 +61,9 @@ use tokio::{
};
use super::worker::{MergeBatch, MergeSender};
use crate::merge::tsid_major::{TsidMajorMergeScope, tsid_major_merge_scope};
mod tsid_major;
/// Last microsecond of the range a merge job at `offset` covers: the hour of
/// `offset`, or the whole day for daily-partitioned streams. No file of the
@@ -427,7 +433,13 @@ pub async fn merge_by_stream(
// the merge only read it — from the end of the range, so a whole-hour
// mode applies to the hour as a unit.
let range_end_ts = job_range_end(offset, partition_time_level);
let mode = MergeMode::for_compactor(stream_type, stream_name, range_end_ts, !is_incremental);
let mode = MergeMode::for_compactor(
stream_type,
stream_name,
&schema,
range_end_ts,
!is_incremental,
);
// a whole-hour merge needs every file of the hour, even those already
// above the size target that a normal merge would leave alone
let max_original_size = if mode.merges_whole_batch() {
@@ -496,6 +508,27 @@ pub async fn merge_by_stream(
if files_with_size.len() <= 1 && !mode.merges_whole_batch() {
return Ok(vec![]);
}
// what a closed TSID-major hour merges, see [`TsidMajorMergeScope`]
if mode.is_tsid_major() {
match tsid_major_merge_scope(&files_with_size, cfg.compact.max_file_size) {
TsidMajorMergeScope::Skip => return Ok(vec![]),
TsidMajorMergeScope::LateFilesOnly => {
files_with_size.retain(|f| {
MetricsFileLayout::of(&f.key) != MetricsFileLayout::TsidMajor
});
log::debug!(
"[COMPACTOR] merge_by_stream [{org_id}/{stream_type}/{stream_name}] tsid_major late merge of {} files, finalized files untouched",
files_with_size.len()
);
}
TsidMajorMergeScope::WholeHour => {
log::debug!(
"[COMPACTOR] merge_by_stream [{org_id}/{stream_type}/{stream_name}] tsid_major fragmentation cap hit, full rewrite of {} files",
files_with_size.len()
);
}
}
}
// group files need to merge
let mut batch_groups = Vec::new();
@@ -867,8 +900,13 @@ pub async fn merge_files(
target_partitions: 2,
};
let input_sort_order = mode.input_sort_order(&files);
log::debug!(
"[COMPACTOR:WORKER:{thread_id}] merge [{mode}] input sort order: {input_sort_order}, files: {}",
files.len()
);
let tables = match TableBuilder::new()
.sort_order(mode.input_sort_order())
.sort_order(input_sort_order)
.build(session, files.clone(), schema.clone())
.await
{
@@ -935,6 +973,7 @@ pub async fn merge_files(
for MergedFile {
buf,
meta: mut new_file_meta,
layout,
} in merged_files
{
new_file_meta.compressed_size = buf.len() as i64;
@@ -945,7 +984,7 @@ pub async fn merge_files(
}
let id = ider::generate_file_name();
let new_file_key = format!("{prefix}/{id}{}", file_format.extension());
let new_file_key = format!("{prefix}/{}", layout.file_name(&id, file_format));
// upload file to storage
let buf = Bytes::from(buf);
src/compaction/src/merge/tsid_major.rs
+131
View File
@@ -0,0 +1,131 @@
// Copyright 2026 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
//! What the compactor merges in a closed TSID-major hour: nothing when it is
//! finalized, the late files alone while fragmentation is low, else the
//! whole hour.
use config::meta::{promql::tsid_layout::MetricsFileLayout, stream::FileKey};
/// Finalized `tsid-major-v3-*` files a closed hour may hold beyond the ideal
/// `total_size / max_file_size` before the whole hour is rewritten.
const TSID_MAJOR_REWRITE_SLACK: usize = 4;
/// What the merge of a closed TSID-major hour covers.
#[derive(Debug, PartialEq, Eq)]
pub(super) enum TsidMajorMergeScope {
/// Every file is `tsid-major-v3-*` and the count fits the current
/// `max_file_size`: leave the hour alone.
Skip,
/// Merge only the late (non-v3) files into one new v3; finalized files
/// stay. The hour's v3 hash ranges then overlap, which readers handle
/// per file.
LateFilesOnly,
/// Too many finalized files: rewrite the whole hour. Reached by late
/// merges piling up fragments (amortizes their full-rewrite cost) or by
/// a raised `max_file_size` shrinking the ideal count (such hours are
/// re-enqueued by the old-data sweep).
WholeHour,
}
pub(super) fn tsid_major_merge_scope(
files: &[FileKey],
max_file_size: usize,
) -> TsidMajorMergeScope {
let finalized = files
.iter()
.filter(|f| MetricsFileLayout::of(&f.key) == MetricsFileLayout::TsidMajor)
.count();
let total_original_size: i64 = files.iter().map(|f| f.meta.original_size.max(0)).sum();
let ideal_file_count = (total_original_size as usize).div_ceil(max_file_size.max(1));
if finalized > ideal_file_count + TSID_MAJOR_REWRITE_SLACK {
TsidMajorMergeScope::WholeHour
} else if finalized == files.len() {
TsidMajorMergeScope::Skip
} else {
TsidMajorMergeScope::LateFilesOnly
}
}
#[cfg(test)]
mod tests {
use config::meta::stream::FileMeta;
use super::*;
fn file_key(name: &str, original_size: i64) -> FileKey {
let mut key = FileKey::from_file_name(&format!("files/o/metrics/m/2026/08/18/10/{name}"));
key.meta = FileMeta {
min_ts: 10,
max_ts: 20,
records: 100,
original_size,
..Default::default()
};
key
}
#[test]
fn test_tsid_major_merge_scope() {
let max_file_size = 100_usize;
let v3 = |id: usize| file_key(&format!("tsid-major-v3-{id}.parquet"), 10);
let late = |id: usize| file_key(&format!("tsid-sorted-{id}.parquet"), 10);
// every file finalized and acceptably sized: the hour is left alone
assert_eq!(
tsid_major_merge_scope(&[v3(1), v3(2)], max_file_size),
TsidMajorMergeScope::Skip
);
// max_file_size raised (files written at a smaller target): an all-v3
// hour above the cap is rewritten to the new size, no late data needed
let raised: Vec<FileKey> = (1..=10).map(v3).collect();
assert_eq!(
tsid_major_merge_scope(&raised, max_file_size),
TsidMajorMergeScope::WholeHour
);
// same files under the original target (ideal 10): left alone
assert_eq!(
tsid_major_merge_scope(&raised, 10),
TsidMajorMergeScope::Skip
);
// no finalized files yet (first hour-end pass): merge everything
assert_eq!(
tsid_major_merge_scope(&[late(1), late(2)], max_file_size),
TsidMajorMergeScope::LateFilesOnly
);
// few finalized files: only the late ones merge
assert_eq!(
tsid_major_merge_scope(&[v3(1), v3(2), v3(3), late(4)], max_file_size),
TsidMajorMergeScope::LateFilesOnly
);
// fragmentation cap: 8 finalized files, total size 90 -> ideal 1,
// 8 > 1 + TSID_MAJOR_REWRITE_SLACK -> rewrite the whole hour
let mut fragmented: Vec<FileKey> = (1..=8).map(v3).collect();
fragmented.push(late(9));
assert_eq!(
tsid_major_merge_scope(&fragmented, max_file_size),
TsidMajorMergeScope::WholeHour
);
// same file count but a large hour (ideal 9): still a late-only merge
let mut large: Vec<FileKey> = (1..=8)
.map(|id| file_key(&format!("tsid-major-v3-{id}.parquet"), 100))
.collect();
large.push(late(9));
assert_eq!(
tsid_major_merge_scope(&large, max_file_size),
TsidMajorMergeScope::LateFilesOnly
);
}
}
src/config/src/config.rs
+6
View File
@@ -2627,6 +2627,12 @@ pub struct Limit {
pub struct Compact {
#[env_config(name = "ZO_COMPACT_ENABLED", default = true)]
pub enabled: bool,
#[env_config(
name = "ZO_METRICS_TSID_MAJOR_ENABLED",
default = false,
help = "Experimental TSID-major metrics layout. The ingester writes Parquet metrics files ordered by (__hash__, _timestamp) instead of _timestamp DESC and marks them with a `tsid-sorted-` file name prefix; the compactor merges a closed hour into size-split `tsid-major-v3-` Parquet files in the same order. Only affects newly written Parquet metrics files of streams whose __hash__ column is UInt64; SQL queries on metrics streams must not assume a _timestamp order while it is on."
)]
pub metrics_tsid_major_enabled: bool,
#[env_config(name = "ZO_COMPACT_INTERVAL", default = 10)] // seconds
pub interval: u64,
#[env_config(
src/config/src/meta/promql/mod.rs
+1
View File
@@ -66,6 +66,7 @@ where
}
pub mod grpc;
pub mod tsid_layout;
pub mod value;
pub const NAME_LABEL: &str = "__name__";
src/config/src/meta/promql/tsid_layout.rs
+193
View File
@@ -0,0 +1,193 @@
// Copyright 2026 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
//! The TSID-major metrics layout (`ZO_METRICS_TSID_MAJOR_ENABLED`): when it
//! applies to a stream and how the physical layout of a metrics file is
//! encoded in its name.
use arrow_schema::{DataType, Schema};
use super::HASH_LABEL;
use crate::{FileFormat, meta::stream::StreamType};
/// True when `stream_type` uses the TSID-major layout
/// (`ZO_METRICS_TSID_MAJOR_ENABLED`): Parquet metrics files ordered by
/// `(__hash__, _timestamp)`, so readers must not assume a `_timestamp` order.
pub fn metrics_tsid_major_enabled(stream_type: StreamType) -> bool {
if stream_type != StreamType::Metrics {
return false;
}
let cfg = crate::get_config();
cfg.compact.metrics_tsid_major_enabled
&& cfg.common.file_format.for_stream(stream_type) == FileFormat::Parquet
}
/// [`metrics_tsid_major_enabled`] narrowed to one stream: the layout also
/// needs a `__hash__` column of type `UInt64` (remote-write / OTLP metrics).
pub fn metrics_tsid_major_stream(stream_type: StreamType, schema: &Schema) -> bool {
metrics_tsid_major_enabled(stream_type)
&& schema
.field_with_name(HASH_LABEL)
.is_ok_and(|field| field.data_type() == &DataType::UInt64)
}
/// Physical layout of a metrics data file, encoded in its file-name prefix so
/// readers and later merges know the row order without opening the file.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MetricsFileLayout {
/// Classic `_timestamp DESC` file (`{id}.parquet` / `{id}.vortex`).
Legacy,
/// Parquet ordered by `(__hash__ ASC, _timestamp ASC)` but not finalized:
/// written by the ingester and by incremental compactor merges of the
/// still-open hour (`tsid-sorted-{id}.parquet`).
HashSorted,
/// Size-bounded Parquet ordered by `(__hash__ ASC, _timestamp ASC)`,
/// written by the compactor's hour-end merge (`tsid-major-v3-{id}.parquet`).
TsidMajor,
}
impl MetricsFileLayout {
const HASH_SORTED_PREFIX: &'static str = "tsid-sorted-";
const TSID_MAJOR_PREFIX: &'static str = "tsid-major-v3-";
/// Layout of the file at `path` (a full object key or a bare file name).
pub fn of(path: &str) -> Self {
let file_name = path.rsplit('/').next().unwrap_or(path);
for (prefix, layout) in [
(Self::HASH_SORTED_PREFIX, Self::HashSorted),
(Self::TSID_MAJOR_PREFIX, Self::TsidMajor),
] {
if let Some(id) = file_name.strip_prefix(prefix)
&& let Some(id) = id.strip_suffix(".parquet")
&& !id.is_empty()
{
return layout;
}
}
Self::Legacy
}
fn prefix(self) -> &'static str {
match self {
Self::Legacy => "",
Self::HashSorted => Self::HASH_SORTED_PREFIX,
Self::TsidMajor => Self::TSID_MAJOR_PREFIX,
}
}
/// True when the rows are ordered by `(__hash__, _timestamp)`.
pub fn is_hash_ordered(self) -> bool {
!matches!(self, Self::Legacy)
}
/// File name for a new file of this layout. Marked layouts are always
/// Parquet; `Legacy` keeps the extension of `file_format`.
pub fn file_name(self, id: &str, file_format: FileFormat) -> String {
match self {
Self::Legacy => format!("{id}{}", file_format.extension()),
_ => format!("{}{id}.parquet", self.prefix()),
}
}
/// Add this layout's marker to the file name of an existing key
/// (`files/.../7099.parquet` -> `files/.../tsid-sorted-7099.parquet`).
pub fn mark_file_key(self, key: &str) -> String {
match key.rfind('/') {
Some(pos) => format!("{}/{}{}", &key[..pos], self.prefix(), &key[pos + 1..]),
None => format!("{}{key}", self.prefix()),
}
}
}
#[cfg(test)]
mod metrics_file_layout_tests {
use FileFormat;
use super::*;
#[test]
fn recognizes_layouts_from_file_names() {
assert_eq!(
MetricsFileLayout::of("files/default/metrics/cpu/2026/08/18/10/7099.parquet"),
MetricsFileLayout::Legacy
);
assert_eq!(
MetricsFileLayout::of("7099.vortex"),
MetricsFileLayout::Legacy
);
assert_eq!(
MetricsFileLayout::of(
"files/default/metrics/cpu/2026/08/18/10/tsid-sorted-7099.parquet"
),
MetricsFileLayout::HashSorted
);
assert_eq!(
MetricsFileLayout::of(
"files/default/metrics/test/2026/08/13/10/tsid-major-v3-456.parquet"
),
MetricsFileLayout::TsidMajor
);
// wrong format, empty id, other versions, marker in a directory name
for name in [
"tsid-major-v3-x.vortex",
"tsid-sorted-1.vortex",
"tsid-major-v3-.parquet",
"tsid-sorted-.parquet",
"tsid-major-v1-x.parquet",
"tsid-range-v3-b04-p000a-x.parquet",
"files/tsid-sorted-dir/1.parquet",
] {
assert_eq!(
MetricsFileLayout::of(name),
MetricsFileLayout::Legacy,
"{name}"
);
}
assert!(MetricsFileLayout::HashSorted.is_hash_ordered());
assert!(MetricsFileLayout::TsidMajor.is_hash_ordered());
assert!(!MetricsFileLayout::Legacy.is_hash_ordered());
}
#[test]
fn builds_and_marks_file_names() {
assert_eq!(
MetricsFileLayout::TsidMajor.file_name("456", FileFormat::Parquet),
"tsid-major-v3-456.parquet"
);
assert_eq!(
MetricsFileLayout::HashSorted.file_name("456", FileFormat::Parquet),
"tsid-sorted-456.parquet"
);
assert_eq!(
MetricsFileLayout::Legacy.file_name("456", FileFormat::Vortex),
"456.vortex"
);
let key = "files/default/metrics/cpu/2026/08/18/10/7099.parquet";
let marked = MetricsFileLayout::HashSorted.mark_file_key(key);
assert_eq!(
marked,
"files/default/metrics/cpu/2026/08/18/10/tsid-sorted-7099.parquet"
);
assert_eq!(
MetricsFileLayout::of(&marked),
MetricsFileLayout::HashSorted
);
assert_eq!(MetricsFileLayout::Legacy.mark_file_key(key), key);
assert_eq!(
MetricsFileLayout::HashSorted.mark_file_key("1.parquet"),
"tsid-sorted-1.parquet"
);
}
}
src/jobs/src/job/files/pack.rs
+24 -7
View File
@@ -26,7 +26,10 @@ use bytes::Bytes;
use chrono::Duration;
use config::{
FileFormat, cluster, get_config,
meta::stream::{FileMeta, StreamType},
meta::{
promql::tsid_layout::MetricsFileLayout,
stream::{FileMeta, StreamType},
},
metrics,
utils::{parquet::generate_filename_with_time_range, time::now_micros},
};
@@ -388,7 +391,11 @@ async fn upload_chunk(
bufs.push(data);
}
let (buf, mut new_file_meta, file_format) = if chunk.len() == 1 {
// Segments are persisted in arrival order. A hash-sorted layout must be
// produced by the merge, so it never takes the as-is fast path.
let mode = MergeMode::for_ingester(stream_type, stream_name, &latest_schema);
let single_segment_as_is = chunk.len() == 1 && mode.file_layout() == MetricsFileLayout::Legacy;
let (buf, mut new_file_meta, file_format, layout) = if single_segment_as_is {
// fast path: upload the parquet bytes as-is, no decode/re-encode
let buf = Bytes::from(bufs.pop().unwrap());
let file_meta = FileMeta {
@@ -399,7 +406,12 @@ async fn upload_chunk(
compressed_size: buf.len() as i64,
..Default::default()
};
(buf, file_meta, FileFormat::Parquet)
(
buf,
file_meta,
FileFormat::Parquet,
MetricsFileLayout::Legacy,
)
} else {
// merge multiple segments in memory
let mut segment_schemas = Vec::with_capacity(bufs.len());
@@ -437,13 +449,18 @@ async fn upload_chunk(
tables,
&bloom_filter_fields,
new_file_meta,
&MergeMode::for_ingester(stream_type, stream_name),
&mode,
MergeOutput::for_ingester(stream_type),
)
.await?;
// the ingester always merges into exactly one file
let (merged, file_format) = merge_result.into_single()?;
(Bytes::from(merged.buf), merged.meta, file_format)
(
Bytes::from(merged.buf),
merged.meta,
file_format,
merged.layout,
)
};
if new_file_meta.compressed_size == 0 {
@@ -458,13 +475,13 @@ async fn upload_chunk(
chunk[0].meta.partition_key,
generate_filename_with_time_range(min_ts, max_ts, 0)
);
let new_file_key = super::generate_ingester_storage_file_key(
let new_file_key = layout.mark_file_key(&super::generate_ingester_storage_file_key(
org_id,
stream_type,
stream_name,
&wal_like_name,
file_format,
);
));
log::info!(
"[INGESTER:PACK:JOB:{thread_id}] uploading {} segments into a new file: {new_file_key}, original_size: {}, compressed_size: {}, took: {} ms",
src/jobs/src/job/files/parquet.rs
+5 -4
View File
@@ -784,12 +784,13 @@ async fn merge_files(
.await?;
let start = std::time::Instant::now();
let mode = MergeMode::for_ingester(stream_type, &stream_name, &schema);
let merge_result = merge::merge_parquet_files(
schema,
tables,
&bloom_filter_fields,
new_file_meta,
&MergeMode::for_ingester(stream_type, &stream_name),
&mode,
MergeOutput::for_ingester(stream_type),
)
.await;
@@ -809,20 +810,20 @@ async fn merge_files(
// the ingester always merges into exactly one file
let (merged, file_format) = buf.into_single()?;
let (buf, mut new_file_meta) = (merged.buf, merged.meta);
let (buf, mut new_file_meta, layout) = (merged.buf, merged.meta, merged.layout);
if new_file_meta.compressed_size == 0 {
return Err(anyhow::anyhow!(
"merge_parquet_files error: compressed_size is 0"
));
}
let new_file_key = super::generate_ingester_storage_file_key(
let new_file_key = layout.mark_file_key(&super::generate_ingester_storage_file_key(
&org_id,
stream_type,
&stream_name,
&file_name,
file_format,
);
));
log::info!(
"[INGESTER:JOB:{thread_id}] merged {} files into a new file: {new_file_key}, original_size: {}, compressed_size: {}, took: {} ms",
retain_file_list.len(),
src/search/src/datafusion/merge/downsampling.rs
+8 -2
View File
@@ -19,7 +19,9 @@ use arrow::array::{Int64Array, RecordBatch};
use config::{
FileFormat, TIMESTAMP_COL_NAME, get_config,
meta::{
promql::{DownsamplingRule, Function, HASH_LABEL, VALUE_LABEL},
promql::{
DownsamplingRule, Function, HASH_LABEL, VALUE_LABEL, tsid_layout::MetricsFileLayout,
},
stream::FileMeta,
},
utils::parquet::new_parquet_writer,
@@ -72,7 +74,11 @@ pub(super) async fn write_files(
Ok(bufs
.into_iter()
.zip(file_metas)
.map(|(buf, meta)| MergedFile { buf, meta })
.map(|(buf, meta)| MergedFile {
buf,
meta,
layout: MetricsFileLayout::Legacy,
})
.collect())
}
src/search/src/datafusion/merge/mod.rs
+22 -5
View File
@@ -18,7 +18,7 @@ use std::sync::Arc;
use arrow::array::RecordBatch;
use config::{
FileFormat, get_config,
meta::stream::FileMeta,
meta::{promql::tsid_layout::MetricsFileLayout, stream::FileMeta},
utils::parquet::{VORTEX_FILE_META_KEY, encode_vortex_file_meta, new_parquet_writer},
};
use datafusion::{
@@ -52,6 +52,7 @@ use crate::datafusion::{
#[cfg(feature = "enterprise")]
pub mod downsampling;
pub mod mode;
mod tsid_major;
pub use mode::{MergeMode, MergeOutput};
@@ -59,6 +60,8 @@ pub use mode::{MergeMode, MergeOutput};
pub struct MergedFile {
pub buf: Vec<u8>,
pub meta: FileMeta,
/// Physical layout the file was written in; decides its file name.
pub layout: MetricsFileLayout,
}
pub struct MergeParquetResult {
@@ -98,7 +101,7 @@ pub async fn merge_parquet_files(
let sql = mode.sql(&schema);
log::debug!("merge_parquet_files [{mode}] sql: {sql}");
let (schema, mut rx, read_task) =
run_merge_query(&sql, mode.input_sort_order(), schema, tables).await?;
run_merge_query(&sql, mode.output_sort_order(), schema, tables).await?;
let files = match mode {
#[cfg(feature = "enterprise")]
@@ -113,6 +116,17 @@ pub async fn merge_parquet_files(
)
.await?
}
MergeMode::TsidMajor => {
tsid_major::write_files(
&schema,
bloom_filter_fields,
&metadata,
get_config().compact.max_file_size,
&mut rx,
read_task,
)
.await?
}
_ => {
let buf = match output.file_format {
FileFormat::Parquet => {
@@ -132,6 +146,7 @@ pub async fn merge_parquet_files(
vec![MergedFile {
buf,
meta: metadata,
layout: mode.file_layout(),
}]
}
};
@@ -149,9 +164,11 @@ pub async fn merge_parquet_files(
/// Plan and start `sql` over the union of `tables`; the record batches arrive
/// on the returned channel, the task reports the stream's completion / error.
/// `sort_order` only enables `split_file_groups_by_statistics` for that
/// order; the input tables declare what the files really carry.
async fn run_merge_query(
sql: &str,
input_sort_order: FileSortOrder,
sort_order: FileSortOrder,
schema: Arc<Schema>,
tables: Vec<Arc<dyn TableProvider>>,
) -> Result<(
@@ -162,7 +179,7 @@ async fn run_merge_query(
let cfg = get_config();
let ctx = DataFusionContextBuilder::new()
.trace_id("merge_parquet_files")
.sort_order(input_sort_order)
.sort_order(sort_order)
.build(cfg.limit.datafusion_min_partition_num)
.await?;
// register union table
@@ -378,7 +395,7 @@ mod tests {
.unwrap();
let table = Arc::new(MemTable::try_new(schema.clone(), vec![vec![batch]]).unwrap());
let mode = MergeMode::for_ingester(StreamType::Metadata, "trace_time_index_test");
let mode = MergeMode::for_ingester(StreamType::Metadata, "trace_time_index_test", &schema);
assert!(matches!(mode, MergeMode::TraceTimeIndex));
let merged = merge_parquet_files(
schema,
src/search/src/datafusion/merge/mode.rs
+131 -14
View File
@@ -28,7 +28,11 @@ use arrow_schema::Schema;
#[cfg(feature = "enterprise")]
use config::meta::promql::DownsamplingRule;
use config::{
FileFormat, FileFormatConfig, TIMESTAMP_COL_NAME, get_config, meta::stream::StreamType,
FileFormat, FileFormatConfig, TIMESTAMP_COL_NAME, get_config,
meta::{
promql::tsid_layout::{MetricsFileLayout, metrics_tsid_major_stream},
stream::{FileKey, StreamType},
},
utils::util::is_trace_time_index_stream,
};
#[cfg(feature = "enterprise")]
@@ -52,6 +56,13 @@ pub enum MergeMode {
/// as a whole.
#[cfg(feature = "enterprise")]
Downsampling(DownsamplingRule),
/// TSID-major metrics stream, hour still open: the ingester and the
/// incremental compactor merges write one `tsid-sorted-*` Parquet file
/// ordered by `(__hash__, _timestamp)`.
HashSorted,
/// TSID-major metrics stream, closed hour: the whole hour merges into
/// size-split `tsid-major-v3-*` files in the same order.
TsidMajor,
}
impl MergeMode {
@@ -63,23 +74,32 @@ impl MergeMode {
pub fn for_compactor(
stream_type: StreamType,
stream_name: &str,
max_ts: i64,
schema: &Schema,
_max_ts: i64,
finalize: bool,
) -> Self {
#[cfg(feature = "enterprise")]
if finalize
&& stream_type == StreamType::Metrics
&& let Some(rule) = get_largest_downsampling_rule(stream_name, max_ts)
&& let Some(rule) = get_largest_downsampling_rule(stream_name, _max_ts)
{
return Self::Downsampling(rule.clone());
}
#[cfg(not(feature = "enterprise"))]
let _ = (max_ts, finalize);
if metrics_tsid_major_stream(stream_type, schema) {
return if finalize {
Self::TsidMajor
} else {
Self::HashSorted
};
}
Self::for_stream(stream_type, stream_name)
}
/// Mode for the ingester movers: never downsamples.
pub fn for_ingester(stream_type: StreamType, stream_name: &str) -> Self {
/// Mode for the ingester movers: never downsamples, never finalizes.
pub fn for_ingester(stream_type: StreamType, stream_name: &str, schema: &Schema) -> Self {
if metrics_tsid_major_stream(stream_type, schema) {
return Self::HashSorted;
}
Self::for_stream(stream_type, stream_name)
}
@@ -91,6 +111,11 @@ impl MergeMode {
}
}
/// True for the TSID-major hour-end merge.
pub fn is_tsid_major(&self) -> bool {
matches!(self, Self::TsidMajor)
}
/// True when the whole hour must be merged as one batch — every file of
/// the hour, including ones already above the size target, and regardless
/// of `ZO_COMPACT_MAX_FILE_SIZE` (the writer splits the output itself).
@@ -98,13 +123,47 @@ impl MergeMode {
match self {
#[cfg(feature = "enterprise")]
Self::Downsampling(_) => true,
Self::TsidMajor => true,
_ => false,
}
}
/// Physical order of the input files the merge query may rely on.
pub fn input_sort_order(&self) -> FileSortOrder {
FileSortOrder::TimestampDesc
/// Row order the merge writes.
pub fn output_sort_order(&self) -> FileSortOrder {
match self {
Self::HashSorted | Self::TsidMajor => FileSortOrder::HashTimestampAsc,
_ => FileSortOrder::TimestampDesc,
}
}
/// Layout of the file(s) the merge writes.
pub fn file_layout(&self) -> MetricsFileLayout {
match self {
Self::HashSorted => MetricsFileLayout::HashSorted,
Self::TsidMajor => MetricsFileLayout::TsidMajor,
_ => MetricsFileLayout::Legacy,
}
}
/// Physical order of the input `files` the merge query may rely on.
///
/// A hash-ordered merge whose inputs are all hash-ordered declares that
/// order, so DataFusion merges the pre-sorted files instead of a full
/// sort. Any mix with other layouts declares no order; otherwise the
/// classic `_timestamp DESC`.
pub fn input_sort_order(&self, files: &[FileKey]) -> FileSortOrder {
let hash_ordered = files
.iter()
.filter(|f| MetricsFileLayout::of(&f.key).is_hash_ordered())
.count();
match self {
Self::HashSorted | Self::TsidMajor if hash_ordered == files.len() => {
FileSortOrder::HashTimestampAsc
}
_ if hash_ordered > 0 => FileSortOrder::None,
Self::HashSorted | Self::TsidMajor => FileSortOrder::None,
_ => FileSortOrder::TimestampDesc,
}
}
/// The merge query over the union table `tbl`.
@@ -129,6 +188,12 @@ impl MergeMode {
Self::Downsampling(rule) => {
super::downsampling::generate_downsampling_sql(schema, rule)
}
Self::HashSorted | Self::TsidMajor => format!(
"SELECT * FROM tbl ORDER BY {}",
FileSortOrder::HashTimestampAsc
.order_by_clause()
.expect("hash order has sort columns")
),
}
}
}
@@ -141,6 +206,8 @@ impl fmt::Display for MergeMode {
Self::FileList => write!(f, "file_list"),
#[cfg(feature = "enterprise")]
Self::Downsampling(rule) => write!(f, "downsampling(step={}s)", rule.step),
Self::HashSorted => write!(f, "hash_sorted"),
Self::TsidMajor => write!(f, "tsid_major"),
}
}
}
@@ -197,21 +264,67 @@ mod tests {
#[test]
fn mode_by_stream() {
let schema = Schema::empty();
assert!(matches!(
MergeMode::for_ingester(StreamType::Logs, "app"),
MergeMode::for_ingester(StreamType::Logs, "app", &schema),
MergeMode::Classic
));
assert!(matches!(
MergeMode::for_compactor(StreamType::Traces, "app", 0, true),
MergeMode::for_compactor(StreamType::Traces, "app", &schema, 0, true),
MergeMode::Classic
));
assert!(matches!(
MergeMode::for_compactor(StreamType::Filelist, "x", 0, false),
MergeMode::for_compactor(StreamType::Filelist, "x", &schema, 0, false),
MergeMode::FileList
));
assert!(!MergeMode::Classic.merges_whole_batch());
assert!(MergeMode::TsidMajor.merges_whole_batch());
assert!(!MergeMode::HashSorted.merges_whole_batch());
assert_eq!(
MergeMode::Classic.input_sort_order(),
MergeMode::Classic.input_sort_order(&[]),
FileSortOrder::TimestampDesc
);
assert_eq!(MergeMode::Classic.file_layout(), MetricsFileLayout::Legacy);
assert_eq!(
MergeMode::HashSorted.file_layout(),
MetricsFileLayout::HashSorted
);
assert_eq!(
MergeMode::TsidMajor.output_sort_order(),
FileSortOrder::HashTimestampAsc
);
}
#[test]
fn input_sort_order_by_file_layout() {
let legacy = FileKey::from_file_name("files/o/metrics/m/2026/08/18/10/1.parquet");
let sorted =
FileKey::from_file_name("files/o/metrics/m/2026/08/18/10/tsid-sorted-2.parquet");
let major =
FileKey::from_file_name("files/o/metrics/m/2026/08/18/10/tsid-major-v3-3.parquet");
// all inputs hash ordered: the hash modes merge them pre-sorted
for mode in [MergeMode::HashSorted, MergeMode::TsidMajor] {
assert_eq!(
mode.input_sort_order(std::slice::from_ref(&sorted)),
FileSortOrder::HashTimestampAsc
);
assert_eq!(
mode.input_sort_order(&[sorted.clone(), major.clone()]),
FileSortOrder::HashTimestampAsc
);
// a legacy file in the batch: nothing can be assumed
assert_eq!(
mode.input_sort_order(&[sorted.clone(), legacy.clone()]),
FileSortOrder::None
);
}
// classic merge over hash-ordered files (layout switched off): no order
assert_eq!(
MergeMode::Classic.input_sort_order(&[legacy.clone(), sorted.clone()]),
FileSortOrder::None
);
assert_eq!(
MergeMode::Classic.input_sort_order(std::slice::from_ref(&legacy)),
FileSortOrder::TimestampDesc
);
}
@@ -264,6 +377,10 @@ mod tests {
MergeMode::FileList.sql(&without_session),
"SELECT * FROM tbl ORDER BY min_ts DESC"
);
assert_eq!(
MergeMode::TsidMajor.sql(&without_session),
"SELECT * FROM tbl ORDER BY __hash__ ASC, _timestamp ASC"
);
assert!(
MergeMode::TraceTimeIndex
.sql(&with_session)
src/search/src/datafusion/merge/tsid_major.rs
+293
View File
@@ -0,0 +1,293 @@
// Copyright 2026 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use std::sync::Arc;
use arrow::{
array::{Array, Int64Array, RecordBatch},
compute::{max, min},
};
use config::{
TIMESTAMP_COL_NAME,
meta::{promql::tsid_layout::MetricsFileLayout, stream::FileMeta},
utils::parquet::new_parquet_writer,
};
use datafusion::{
arrow::datatypes::Schema,
error::{DataFusionError, Result},
};
use parquet::arrow::AsyncArrowWriter;
use super::{MergedFile, append_metadata};
/// Write a globally hash-sorted metrics stream into size-bounded files.
/// Rotation happens between input record batches. File and row-group
/// boundaries deliberately remain independent of TSID boundaries.
pub(super) async fn write_files(
schema: &Arc<Schema>,
bloom_filter_fields: &[String],
metadata: &FileMeta,
max_file_size: usize,
rx: &mut tokio::sync::mpsc::Receiver<RecordBatch>,
read_task: tokio::task::JoinHandle<Result<()>>,
) -> Result<Vec<MergedFile>> {
let timestamp_index = schema.index_of(TIMESTAMP_COL_NAME).map_err(|e| {
DataFusionError::Plan(format!(
"TSID-major layout requires {TIMESTAMP_COL_NAME}: {e}"
))
})?;
let max_file_size = i64::try_from(max_file_size).unwrap_or(i64::MAX).max(1);
let mut active: Option<ActiveSizeTsidWriter> = None;
let mut files: Vec<MergedFile> = Vec::new();
while let Some(batch) = rx.recv().await {
if batch.num_rows() == 0 {
continue;
}
// rotate between input batches once the logical size target is reached
if let Some(full) = active.take_if(|writer| {
proportional_original_size(metadata, writer.file_meta.records) >= max_file_size
}) {
files.push(full.finish(metadata, max_file_size).await?);
}
let writer = active.get_or_insert_with(|| {
ActiveSizeTsidWriter::new(schema, bloom_filter_fields, metadata, timestamp_index)
});
writer.write(&batch).await?;
}
read_task
.await
.map_err(|e| DataFusionError::External(Box::new(e)))??;
if let Some(active) = active.take() {
files.push(active.finish(metadata, max_file_size).await?);
}
if files.is_empty() {
return Err(DataFusionError::Execution(
"TSID-major merge produced no rows".to_string(),
));
}
Ok(files)
}
/// One output file in progress: the Parquet writer and the running file meta.
struct ActiveSizeTsidWriter {
writer: AsyncArrowWriter<Vec<u8>>,
file_meta: FileMeta,
timestamp_index: usize,
}
impl ActiveSizeTsidWriter {
fn new(
schema: &Arc<Schema>,
bloom_filter_fields: &[String],
metadata: &FileMeta,
timestamp_index: usize,
) -> Self {
let writer = new_parquet_writer(
Vec::new(),
schema,
bloom_filter_fields,
metadata,
false,
None,
);
Self {
writer,
file_meta: FileMeta::default(),
timestamp_index,
}
}
/// Append one (hash, ts)-ordered batch to the data file and its rows /
/// time range to the file meta.
async fn write(&mut self, batch: &RecordBatch) -> Result<()> {
self.writer.write(batch).await?;
let timestamps = batch
.column(self.timestamp_index)
.as_any()
.downcast_ref::<Int64Array>()
.ok_or_else(|| {
DataFusionError::Plan(format!(
"TSID-major layout requires Int64 {TIMESTAMP_COL_NAME}"
))
})?;
let meta = &mut self.file_meta;
if let (Some(batch_min), Some(batch_max)) = (min(timestamps), max(timestamps)) {
if meta.records == 0 {
(meta.min_ts, meta.max_ts) = (batch_min, batch_max);
} else {
meta.min_ts = meta.min_ts.min(batch_min);
meta.max_ts = meta.max_ts.max(batch_max);
}
}
meta.records += batch.num_rows() as i64;
Ok(())
}
async fn finish(self, source_meta: &FileMeta, max_file_size: i64) -> Result<MergedFile> {
let Self {
mut writer,
mut file_meta,
..
} = self;
// below the target so a finalized file never advertises >= max_file_size
file_meta.original_size =
proportional_original_size(source_meta, file_meta.records).min(max_file_size - 1);
append_metadata(&mut writer, &file_meta)?;
writer.finish().await?;
let buf = writer.into_inner();
file_meta.compressed_size = buf.len() as i64;
Ok(MergedFile {
buf,
meta: file_meta,
layout: MetricsFileLayout::TsidMajor,
})
}
}
/// The share of the source `original_size` that `records` rows carry.
fn proportional_original_size(source_meta: &FileMeta, records: i64) -> i64 {
let estimate = (i128::from(source_meta.original_size.max(0)) * i128::from(records.max(0)))
/ i128::from(source_meta.records.max(1));
i64::try_from(estimate).unwrap_or(i64::MAX)
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use arrow::array::{Float64Array, Int64Array, StringViewArray, UInt64Array};
use arrow_schema::{DataType, Field, Schema};
use config::meta::promql::{HASH_LABEL, VALUE_LABEL};
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
use super::*;
#[tokio::test]
async fn test_size_split_tsid_major_rotates_at_batch_boundary() {
let schema = Arc::new(Schema::new(vec![
Field::new(HASH_LABEL, DataType::UInt64, false),
Field::new(TIMESTAMP_COL_NAME, DataType::Int64, false),
Field::new(VALUE_LABEL, DataType::Float64, false),
Field::new("path", DataType::Utf8View, true),
]));
let batch1 = RecordBatch::try_new(
Arc::clone(&schema),
vec![
Arc::new(UInt64Array::from(vec![1, 1])),
Arc::new(Int64Array::from(vec![10, 20])),
Arc::new(Float64Array::from(vec![1.0, 2.0])),
Arc::new(StringViewArray::from(vec!["a", "a"])),
],
)
.unwrap();
let batch2 = RecordBatch::try_new(
Arc::clone(&schema),
vec![
Arc::new(UInt64Array::from(vec![1, 2, 2])),
Arc::new(Int64Array::from(vec![30, 10, 20])),
Arc::new(Float64Array::from(vec![3.0, 4.0, 5.0])),
Arc::new(StringViewArray::from(vec!["a", "b", "b"])),
],
)
.unwrap();
let batch3 = RecordBatch::try_new(
Arc::clone(&schema),
vec![
Arc::new(UInt64Array::from(vec![3])),
Arc::new(Int64Array::from(vec![10])),
Arc::new(Float64Array::from(vec![6.0])),
Arc::new(StringViewArray::from(vec!["c"])),
],
)
.unwrap();
let metadata = FileMeta {
min_ts: 10,
max_ts: 30,
records: 6,
original_size: 600,
compressed_size: 300,
..Default::default()
};
let (tx, mut rx) = tokio::sync::mpsc::channel(3);
tx.send(batch1).await.unwrap();
tx.send(batch2).await.unwrap();
tx.send(batch3).await.unwrap();
drop(tx);
let max_file_size = 151;
let files = write_files(
&schema,
&[],
&metadata,
max_file_size,
&mut rx,
tokio::spawn(async { Ok(()) }),
)
.await
.unwrap();
assert_eq!(files.len(), 3);
assert_eq!(files.iter().map(|f| f.meta.records).sum::<i64>(), 6);
assert_eq!(
files
.iter()
.map(|f| f.meta.original_size)
.collect::<Vec<_>>(),
vec![150, 150, 100]
);
assert!(files.iter().all(|f| {
f.layout == MetricsFileLayout::TsidMajor
&& f.meta.original_size < max_file_size as i64
&& f.meta.compressed_size == f.buf.len() as i64
}));
let mut file_hashes = Vec::new();
for file in files {
let bytes = bytes::Bytes::from(file.buf);
let footer = config::utils::parquet::read_metadata_from_bytes(&bytes)
.await
.unwrap();
assert_eq!(footer.min_ts, file.meta.min_ts);
assert_eq!(footer.max_ts, file.meta.max_ts);
assert_eq!(footer.records, file.meta.records);
assert_eq!(footer.original_size, file.meta.original_size);
let reader = ParquetRecordBatchReaderBuilder::try_new(bytes)
.unwrap()
.build()
.unwrap();
let mut hashes_in_file = Vec::new();
for batch in reader {
let batch = batch.unwrap();
let hashes = batch
.column_by_name(HASH_LABEL)
.unwrap()
.as_any()
.downcast_ref::<UInt64Array>()
.unwrap();
hashes_in_file.extend_from_slice(hashes.values());
}
file_hashes.push(hashes_in_file);
}
assert_eq!(file_hashes, vec![vec![1, 1], vec![1, 2, 2], vec![3]]);
assert!(file_hashes[0].contains(&1) && file_hashes[1].contains(&1));
}
}
src/search/src/sql/mod.rs
+4 -1
View File
@@ -182,7 +182,10 @@ impl Sql {
}
let need_sort_by_time = order_by.len() == 1
&& order_by[0].0 == TIMESTAMP_COL_NAME
&& order_by[0].1 == OrderBy::Desc;
&& order_by[0].1 == OrderBy::Desc
&& !total_schemas.iter().any(|(stream, _)| {
matches!(stream.get_stream_type(stream_type), StreamType::Metrics)
});
// check if need exact limit and offset
if (limit == -1 || limit == 0)