From 508d1a1a336beb9ed1218b4728d32dfdb9ef728b Mon Sep 17 00:00:00 2001 From: "Lei, HUANG" <6406592+v0y4g3r@users.noreply.github.com> Date: Tue, 20 Oct 2026 04:27:22 -0700 Subject: [PATCH] fix: backport clean must-have fixes to release/5e02efa9 (#9772) * fix(mito2): preserve effective sequences during compaction (#9147) * fix(mito2): preserve effective row sequences during compaction Signed-off-by: Lei, HUANG * fix(mito2): inherit compaction input sequence bounds Signed-off-by: Lei, HUANG --------- Signed-off-by: Lei, HUANG (cherry picked from commit e0552eee2dc1f2cfdf703cee9fccbb0ceca06699) Signed-off-by: Lei, HUANG * fix(meta-client): bound heartbeat handshake timeout (#9438) * fix(meta-client): bound heartbeat handshake timeout Signed-off-by: Aetherance * chore(meta-client): fix test license header Signed-off-by: Aetherance * test(meta-client): prefix heartbeat tests with test Signed-off-by: Aetherance --------- Signed-off-by: Aetherance (cherry picked from commit 9c69e381ae28dd86954530526239bbc6d4f17940) Signed-off-by: Lei, HUANG * fix(metasrv): stop holding the pusher registry lock across mailbox sends (#9445) fix(meta-srv): stop holding the pusher registry lock across mailbox sends Signed-off-by: lczllx <2181719471@qq.com> (cherry picked from commit 7605f27608716c5b03273906fe9f1ba912379b62) Signed-off-by: Lei, HUANG * fix(mito2): ignore missing inverted indexes in old SSTs (#9545) Signed-off-by: WenyXu (cherry picked from commit 287ced2de21b8e986f2becad59169132770f3ed0) Signed-off-by: Lei, HUANG * fix: prevent stale leader metadata after concurrent writes (#9549) fix: fence leader cache fills around store mutations Signed-off-by: jeremyhi (cherry picked from commit 394562aa9ed2ad3cc6fef1e15aa1c09f42f759e1) Signed-off-by: Lei, HUANG * fix(query): accept narrowed nullability in MergeScan remote schema check (#9592) On append-mode tables a region reports an exact row count. When it is zero, DataFusion's AggregateStatistics replaces the region-side __min_state/__max_state aggregate with a non-null struct literal, so the remote stream advertises the state column as non-nullable. The MergeScan schema check required equal nullability and failed the query. Accept a remote field that is non-nullable where the expected field is nullable, matching DataFusion's own physical-plan schema invariant. Batches are already rebuilt with the expected schema. Widening is still rejected, and nested field nullability is still compared exactly as part of the data type. Signed-off-by: Dennis Zhuang (cherry picked from commit ef1298ce026924bbd8f80cf279eedfed06718635) Signed-off-by: Lei, HUANG * test: regenerate missing inverted index sqlness result for release/5e02efa9 The release branch renders empty result sets as '++' instead of an empty table with headers. Only empty-result rendering changes; rows are identical. Signed-off-by: Lei, HUANG --------- Signed-off-by: Lei, HUANG Signed-off-by: Aetherance Signed-off-by: lczllx <2181719471@qq.com> Signed-off-by: WenyXu Signed-off-by: jeremyhi Signed-off-by: Dennis Zhuang Co-authored-by: Aether <98370028+Aetherance@users.noreply.github.com> Co-authored-by: Dragon Roar <2181719471@qq.com> Co-authored-by: Weny Xu Co-authored-by: jeremyhi Co-authored-by: dennis zhuang --- src/meta-client/src/client/heartbeat.rs | 29 +- src/meta-client/src/error.rs | 11 + .../tests/heartbeat_handshake_timeout.rs | 140 +++++++ src/meta-srv/src/handler.rs | 193 ++++++++-- src/meta-srv/src/service/store/cached_kv.rs | 361 +++++++++++++++++- src/mito2/src/access_layer.rs | 2 + src/mito2/src/compaction/compactor.rs | 337 ++++++++++++++-- src/mito2/src/engine/compaction_test.rs | 291 +++++++++++++- src/mito2/src/engine/merge_mode_test.rs | 189 +++++++++ src/mito2/src/engine/scan_test.rs | 153 +++++--- src/mito2/src/read/scan_region.rs | 7 +- src/mito2/src/region/options.rs | 3 +- src/mito2/src/sst/file.rs | 11 +- .../src/sst/index/inverted_index/applier.rs | 5 +- .../src/sst/index/inverted_index/creator.rs | 68 +++- src/query/src/dist_plan/merge_scan.rs | 87 ++++- .../aggregate/empty_region_stats.result | 126 ++++++ .../common/aggregate/empty_region_stats.sql | 42 ++ .../alter/change_col_inverted_index.result | 142 +++++++ .../alter/change_col_inverted_index.sql | 45 +++ 20 files changed, 2073 insertions(+), 169 deletions(-) create mode 100644 src/meta-client/tests/heartbeat_handshake_timeout.rs create mode 100644 tests/cases/standalone/common/aggregate/empty_region_stats.result create mode 100644 tests/cases/standalone/common/aggregate/empty_region_stats.sql diff --git a/src/meta-client/src/client/heartbeat.rs b/src/meta-client/src/client/heartbeat.rs index 96fc398556d..6dcc14c13f2 100644 --- a/src/meta-client/src/client/heartbeat.rs +++ b/src/meta-client/src/client/heartbeat.rs @@ -19,7 +19,7 @@ use std::time::Duration; use api::v1::meta::heartbeat_client::HeartbeatClient; use api::v1::meta::{HeartbeatRequest, HeartbeatResponse, RequestHeader, Role}; use common_grpc::channel_manager::ChannelManager; -use common_meta::distributed_time_constants::BASE_HEARTBEAT_INTERVAL; +use common_meta::distributed_time_constants::{BASE_HEARTBEAT_INTERVAL, HEARTBEAT_TIMEOUT}; use common_meta::util; use common_telemetry::tracing_context::TracingContext; use common_telemetry::{info, warn}; @@ -249,17 +249,26 @@ impl Inner { })?; let receiver = ReceiverStream::new(receiver); - let mut stream = leader - .heartbeat(receiver) - .await - .map_err(error::Error::from)? - .into_inner(); + let handshake = async { + let mut stream = leader + .heartbeat(receiver) + .await + .map_err(error::Error::from)? + .into_inner(); - let res = stream - .message() + let res = stream + .message() + .await + .map_err(error::Error::from)? + .context(error::CreateHeartbeatStreamSnafu)?; + Ok::<_, error::Error>((stream, res)) + }; + + let (stream, res) = tokio::time::timeout(HEARTBEAT_TIMEOUT, handshake) .await - .map_err(error::Error::from)? - .context(error::CreateHeartbeatStreamSnafu)?; + .with_context(|_| error::HeartbeatHandshakeTimeoutSnafu { + timeout: HEARTBEAT_TIMEOUT, + })??; let config = HeartbeatConfig::from_response(&res); diff --git a/src/meta-client/src/error.rs b/src/meta-client/src/error.rs index 4623cfee33b..e15c83be1d9 100644 --- a/src/meta-client/src/error.rs +++ b/src/meta-client/src/error.rs @@ -86,6 +86,15 @@ pub enum Error { location: Location, }, + #[snafu(display("Heartbeat handshake timeout after {timeout:?}"))] + HeartbeatHandshakeTimeout { + timeout: std::time::Duration, + #[snafu(implicit)] + location: Location, + #[snafu(source)] + error: tokio::time::error::Elapsed, + }, + #[snafu(display("Invalid response header"))] InvalidResponseHeader { #[snafu(implicit)] @@ -150,6 +159,7 @@ impl ErrorExt for Error { | Error::MissingQueryContext { .. } | Error::SendHeartbeat { .. } | Error::CreateHeartbeatStream { .. } + | Error::HeartbeatHandshakeTimeout { .. } | Error::CreateChannel { .. } | Error::RetryTimesExceeded { .. } | Error::ConvertMetaConfig { .. } => StatusCode::Internal, @@ -168,6 +178,7 @@ impl ErrorExt for Error { fn retry_hint(&self) -> RetryHint { match self { Error::MetaServer { retry_hint, .. } => *retry_hint, + Error::HeartbeatHandshakeTimeout { .. } => RetryHint::Retryable, Error::InvalidResponseHeader { source, .. } | Error::ConvertMetaRequest { source, .. } | Error::ConvertMetaResponse { source, .. } diff --git a/src/meta-client/tests/heartbeat_handshake_timeout.rs b/src/meta-client/tests/heartbeat_handshake_timeout.rs new file mode 100644 index 00000000000..2ee43fa74d3 --- /dev/null +++ b/src/meta-client/tests/heartbeat_handshake_timeout.rs @@ -0,0 +1,140 @@ +// 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::time::Duration; + +use api::v1::meta::heartbeat_server::{Heartbeat, HeartbeatServer}; +use api::v1::meta::{ + AskLeaderRequest, AskLeaderResponse, HeartbeatRequest, HeartbeatResponse, Peer, ResponseHeader, +}; +use common_error::ext::{ErrorExt, RetryHint}; +use common_meta::distributed_time_constants::HEARTBEAT_TIMEOUT; +use futures::Stream; +use meta_client::error::Error; +use meta_client::{MetaClientOptions, MetaClientRef, MetaClientType, create_meta_client}; +use tokio_stream::wrappers::TcpListenerStream; +use tonic::codec::CompressionEncoding; +use tonic::{Request, Response, Status, Streaming}; + +type HeartbeatStream = + Pin> + Send + 'static>>; + +struct MockMetasrv { + addr: String, + send_first_heartbeat_response: bool, +} + +#[async_trait::async_trait] +impl Heartbeat for MockMetasrv { + type HeartbeatStream = HeartbeatStream; + + async fn heartbeat( + &self, + request: Request>, + ) -> Result, Status> { + let send_first_heartbeat_response = self.send_first_heartbeat_response; + let stream = futures::stream::once(async move { + let mut request = request.into_inner(); + request + .message() + .await? + .ok_or_else(|| Status::invalid_argument("missing heartbeat request"))?; + + if send_first_heartbeat_response { + Ok(HeartbeatResponse::default()) + } else { + futures::future::pending::>().await + } + }); + Ok(Response::new(Box::pin(stream))) + } + + async fn ask_leader( + &self, + _request: Request, + ) -> Result, Status> { + Ok(Response::new(AskLeaderResponse { + header: Some(ResponseHeader::default()), + leader: Some(Peer { + id: 1, + addr: self.addr.clone(), + }), + })) + } +} + +async fn create_client(send_first_heartbeat_response: bool) -> MetaClientRef { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let mock = MockMetasrv { + addr: addr.to_string(), + send_first_heartbeat_response, + }; + + tokio::spawn(async move { + let service = HeartbeatServer::new(mock) + .accept_compressed(CompressionEncoding::Gzip) + .accept_compressed(CompressionEncoding::Zstd) + .send_compressed(CompressionEncoding::Gzip) + .send_compressed(CompressionEncoding::Zstd); + tonic::transport::Server::builder() + .add_service(service) + .serve_with_incoming(TcpListenerStream::new(listener)) + .await + .unwrap(); + }); + + let options = MetaClientOptions { + metasrv_addrs: vec![addr.to_string()], + connect_timeout: Duration::from_millis(500), + ..Default::default() + }; + create_meta_client( + MetaClientType::Datanode { member_id: 1 }, + &options, + None, + None, + ) + .await + .expect("the mock answers leader discovery") +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn test_heartbeat_handshake_succeeds_when_the_server_responds() { + let client = create_client(true).await; + + tokio::time::timeout(HEARTBEAT_TIMEOUT, client.heartbeat()) + .await + .expect("heartbeat handshake must complete before its deadline") + .expect("heartbeat handshake must succeed"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn test_heartbeat_handshake_times_out_when_the_stream_stays_silent() { + let client = create_client(false).await; + + let result = tokio::time::timeout(2 * HEARTBEAT_TIMEOUT, client.heartbeat()) + .await + .expect("heartbeat handshake must be bounded"); + let error = match result { + Ok(_) => panic!("a silent heartbeat stream must fail the handshake"), + Err(error) => error, + }; + assert!(matches!( + &error, + Error::HeartbeatHandshakeTimeout { timeout, .. } if *timeout == HEARTBEAT_TIMEOUT + )); + assert_eq!(error.retry_hint(), RetryHint::Retryable); +} diff --git a/src/meta-srv/src/handler.rs b/src/meta-srv/src/handler.rs index 50b53f4970a..b2fd83b9e96 100644 --- a/src/meta-srv/src/handler.rs +++ b/src/meta-srv/src/handler.rs @@ -223,18 +223,50 @@ impl Pusher { } #[inline] - pub async fn push(&self, res: HeartbeatResponse) -> Result<()> { - self.sender.send(Ok(res)).await.map_err(|e| { - error::PushMessageSnafu { - err_msg: e.to_string(), - } - .build() - }) + pub fn header(&self) -> ResponseHeader { + self.res_header.clone() + } +} + +/// An owned snapshot of the [`Pusher`] fields the push paths need, so a caller can drop +/// the [`Pushers`] read guard before awaiting the send. It must not hold a [`Pusher`]: +/// dropping one fires the deregister signal, which reads as a closed channel. +struct PusherHandle { + sender: Sender>, + res_header: ResponseHeader, + deregister_signal_receiver: DeregisterSignalReceiver, +} + +impl PusherHandle { + /// Snapshots `pusher`, which callers read under the [`Pushers`] guard. + fn from_pusher(pusher: &Pusher) -> Self { + Self { + sender: pusher.sender.clone(), + res_header: pusher.header(), + deregister_signal_receiver: pusher.deregister_signal_receiver.clone(), + } + } + + /// Sends one mailbox message, consuming the handle. + async fn push(self, mailbox_message: MailboxMessage) -> Result<()> { + self.sender + .send(Ok(HeartbeatResponse { + header: Some(self.res_header), + mailbox_message: Some(mailbox_message), + ..Default::default() + })) + .await + .map_err(|e| { + error::PushMessageSnafu { + err_msg: e.to_string(), + } + .build() + }) } #[inline] - pub fn header(&self) -> ResponseHeader { - self.res_header.clone() + fn deregister_signal_receiver(&self) -> DeregisterSignalReceiver { + self.deregister_signal_receiver.clone() } } @@ -248,41 +280,41 @@ impl Pushers { pusher_id: PusherId, mailbox_message: MailboxMessage, ) -> Result { - let pushers = self.0.read().await; - let pusher = pushers - .get(&pusher_id) - .with_context(|| error::PusherNotFoundSnafu { - pusher_id: pusher_id.to_string(), - })?; + // Releases the guard before the send below, which parks while the target's + // channel is full. `PusherNotFound` is still reported from inside the guard. + let handle = { + let pushers = self.0.read().await; + let pusher = pushers + .get(&pusher_id) + .with_context(|| error::PusherNotFoundSnafu { + pusher_id: pusher_id.to_string(), + })?; + PusherHandle::from_pusher(pusher) + }; - pusher - .push(HeartbeatResponse { - header: Some(pusher.header()), - mailbox_message: Some(mailbox_message), - ..Default::default() - }) - .await?; + let deregister_signal_receiver = handle.deregister_signal_receiver(); + handle.push(mailbox_message).await?; - Ok(pusher.deregister_signal_receiver.clone()) + Ok(deregister_signal_receiver) } async fn broadcast(&self, role: Role, mailbox_message: &MailboxMessage) -> Result<()> { - let pushers = self.0.read().await; - let pushers = pushers - .range(PusherId::role_range(role)) - .map(|(_, value)| value) - .collect::>(); - let mut results = Vec::with_capacity(pushers.len()); + // Collects owned handles, releasing the guard before the sends below. Collecting + // `&Pusher` would hold it for the whole `join_all`. + let targets = { + let pushers = self.0.read().await; + pushers + .range(PusherId::role_range(role)) + .map(|(_, pusher)| PusherHandle::from_pusher(pusher)) + .collect::>() + }; - for pusher in pushers { + let mut results = Vec::with_capacity(targets.len()); + for handle in targets { let mut mailbox_message = mailbox_message.clone(); mailbox_message.id = 0; // one-way message - results.push(pusher.push(HeartbeatResponse { - header: Some(pusher.header()), - mailbox_message: Some(mailbox_message), - ..Default::default() - })) + results.push(handle.push(mailbox_message)); } // Checks the error out of the loop. @@ -890,14 +922,15 @@ mod tests { use std::sync::Arc; use std::time::Duration; - use api::v1::meta::{HeartbeatRequest, MailboxMessage, RequestHeader, Role}; + use api::v1::meta::{HeartbeatRequest, HeartbeatResponse, MailboxMessage, RequestHeader, Role}; use common_meta::kv_backend::memory::MemoryKvBackend; use common_meta::sequence::SequenceBuilder; + use futures::future::poll_immediate; use tokio::sync::mpsc; use super::{ HandleControl, HeartbeatAccumulator, HeartbeatHandler, HeartbeatHandlerGroupBuilder, - PusherId, Pushers, + PusherHandle, PusherId, Pushers, }; use crate::error; use crate::handler::collect_stats_handler::CollectStatsHandler; @@ -1363,4 +1396,90 @@ mod tests { let extensions = res.extensions; assert_eq!(extensions.get("custom.key").unwrap(), b"custom-value"); } + + async fn full_pusher( + pushers: &Pushers, + id: PusherId, + ) -> mpsc::Receiver> { + let (tx, rx) = mpsc::channel(1); + tx.try_send(Ok(HeartbeatResponse::default())).unwrap(); + pushers.insert(id, Pusher::new(tx)).await; + // Callers keep `rx` alive: dropping it would make the send fail instead of park. + rx + } + + fn test_mailbox_message(subject: &str) -> MailboxMessage { + MailboxMessage { + id: 0, + subject: subject.to_string(), + timestamp_millis: 123, + ..Default::default() + } + } + + #[tokio::test] + async fn test_pushers_push_does_not_fence_the_registry() { + let pushers = Pushers::default(); + + let full_id = PusherId::new(Role::Datanode, 1); + let _full_rx = full_pusher(&pushers, full_id).await; + + let (healthy_tx, mut healthy_rx) = mpsc::channel(1); + let healthy_id = PusherId::new(Role::Datanode, 2); + pushers.insert(healthy_id, Pusher::new(healthy_tx)).await; + + let msg = test_mailbox_message("fence-test"); + + let mut parked = Box::pin(pushers.push(full_id, msg.clone())); + assert!(poll_immediate(parked.as_mut()).await.is_none()); + + let (writer_tx, _writer_rx) = mpsc::channel(1); + let mut writer = + Box::pin(pushers.insert(PusherId::new(Role::Datanode, 3), Pusher::new(writer_tx))); + assert!(poll_immediate(writer.as_mut()).await.is_some()); + + let mut healthy = Box::pin(pushers.push(healthy_id, msg)); + assert!(poll_immediate(healthy.as_mut()).await.is_some()); + assert!(healthy_rx.try_recv().is_ok()); + } + + #[tokio::test] + async fn test_pushers_broadcast_does_not_fence_the_registry() { + let pushers = Pushers::default(); + + let _full_rx = full_pusher(&pushers, PusherId::new(Role::Datanode, 1)).await; + let (healthy_tx, _healthy_rx) = mpsc::channel(1); + pushers + .insert(PusherId::new(Role::Datanode, 2), Pusher::new(healthy_tx)) + .await; + + let msg = test_mailbox_message("broadcast-fence-test"); + let mut parked = Box::pin(pushers.broadcast(Role::Datanode, &msg)); + assert!(poll_immediate(parked.as_mut()).await.is_none()); + + let (writer_tx, _writer_rx) = mpsc::channel(1); + let mut writer = + Box::pin(pushers.insert(PusherId::new(Role::Flownode, 1), Pusher::new(writer_tx))); + assert!(poll_immediate(writer.as_mut()).await.is_some()); + } + + #[tokio::test] + async fn test_pusher_handle_does_not_fire_the_deregister_signal() { + let (tx, _rx) = mpsc::channel(1); + let pusher = Pusher::new(tx); + let mut signal = pusher.deregister_signal_receiver.clone(); + + let handle = PusherHandle::from_pusher(&pusher); + drop(handle); + + assert!( + tokio::time::timeout(Duration::from_millis(100), signal.changed()) + .await + .is_err() + ); + + // The real `Pusher` still owns the signal. + drop(pusher); + assert!(signal.changed().await.is_ok()); + } } diff --git a/src/meta-srv/src/service/store/cached_kv.rs b/src/meta-srv/src/service/store/cached_kv.rs index 6526ebcd41e..1c2990f5446 100644 --- a/src/meta-srv/src/service/store/cached_kv.rs +++ b/src/meta-srv/src/service/store/cached_kv.rs @@ -69,6 +69,9 @@ pub struct LeaderCachedKvBackend { check_leader: CheckLeaderRef, store: KvBackendRef, cache: ResettableKvBackendRef, + /// Incremented before and after each store mutation. The first increment + /// detects overlapping writers; the second invalidates cache fills that + /// read the old value while the store mutation was in flight. version: AtomicUsize, name: String, } @@ -214,9 +217,11 @@ impl KvBackend for LeaderCachedKvBackend { let ver = self.create_new_version(); let res = self.store.put(req.clone()).await?; + self.create_new_version(); let _ = self.cache.put(req.clone()).await?; - if !self.validate_version(ver) { + // Only our own completion increment may follow the starting version. + if !self.validate_version(ver + 1) { self.invalid_key(req.key).await?; } @@ -231,9 +236,11 @@ impl KvBackend for LeaderCachedKvBackend { let ver = self.create_new_version(); let res = self.store.batch_put(req.clone()).await?; + self.create_new_version(); let _ = self.cache.batch_put(req.clone()).await?; - if !self.validate_version(ver) { + // Only our own completion increment may follow the starting version. + if !self.validate_version(ver + 1) { let keys = req.kvs.into_iter().map(|kv| kv.key).collect::>(); self.invalid_keys(keys).await?; } @@ -306,6 +313,7 @@ impl KvBackend for LeaderCachedKvBackend { let key = req.key.clone(); let res = self.store.compare_and_put(req).await?; + self.create_new_version(); // Delete key in the cache. // // Cache can not deal with the CAS operation, because it does @@ -322,6 +330,7 @@ impl KvBackend for LeaderCachedKvBackend { let _ = self.create_new_version(); let res = self.store.delete_range(req.clone()).await?; + self.create_new_version(); let _ = self.cache.delete_range(req).await?; Ok(res) } @@ -334,6 +343,7 @@ impl KvBackend for LeaderCachedKvBackend { let _ = self.create_new_version(); let res = self.store.batch_delete(req.clone()).await?; + self.create_new_version(); let _ = self.cache.batch_delete(req).await?; Ok(res) } @@ -351,6 +361,7 @@ impl TxnService for LeaderCachedKvBackend { let _ = self.create_new_version(); let res = self.store.txn(txn.clone()).await?; + self.create_new_version(); let TxnRequest { success, failure, .. } = txn.into(); @@ -384,6 +395,7 @@ impl TxnService for LeaderCachedKvBackend { impl ResettableKvBackend for LeaderCachedKvBackend { fn reset(&self) { + self.create_new_version(); self.cache.reset() } @@ -394,10 +406,355 @@ impl ResettableKvBackend for LeaderCachedKvBackend { #[cfg(test)] mod tests { + use std::sync::Mutex; + + use common_meta::kv_backend::txn::{Compare, CompareOp}; use common_meta::rpc::KeyValue; + use futures::poll; + use tokio::sync::oneshot; use super::*; + /// Pauses one backend call at a known point, without relying on task scheduling. + #[derive(Default)] + struct Pause { + receiver: Mutex>>, + } + + impl Pause { + fn arm(&self) -> oneshot::Sender<()> { + let (sender, receiver) = oneshot::channel(); + assert!(self.receiver.lock().unwrap().replace(receiver).is_none()); + sender + } + + async fn wait(&self) { + let receiver = self.receiver.lock().unwrap().take(); + if let Some(receiver) = receiver { + receiver.await.unwrap(); + } + } + } + + #[derive(Default)] + struct ControlledBackend { + inner: MemoryKvBackend, + before_write: Pause, + after_write: Pause, + after_read: Pause, + } + + #[async_trait::async_trait] + impl TxnService for ControlledBackend { + type Error = Error; + + async fn txn(&self, txn: Txn) -> Result { + self.before_write.wait().await; + let res = self.inner.txn(txn).await; + self.after_write.wait().await; + res + } + + fn max_txn_ops(&self) -> usize { + self.inner.max_txn_ops() + } + } + + #[async_trait::async_trait] + impl KvBackend for ControlledBackend { + fn name(&self) -> &str { + "ControlledBackend" + } + + fn as_any(&self) -> &dyn Any { + self + } + + async fn range(&self, req: RangeRequest) -> Result { + let res = self.inner.range(req).await; + self.after_read.wait().await; + res + } + + async fn batch_get(&self, req: BatchGetRequest) -> Result { + let res = self.inner.batch_get(req).await; + self.after_read.wait().await; + res + } + + async fn put(&self, req: PutRequest) -> Result { + self.before_write.wait().await; + let res = self.inner.put(req).await; + self.after_write.wait().await; + res + } + + async fn batch_put(&self, req: BatchPutRequest) -> Result { + self.before_write.wait().await; + let res = self.inner.batch_put(req).await; + self.after_write.wait().await; + res + } + + async fn delete_range(&self, req: DeleteRangeRequest) -> Result { + self.before_write.wait().await; + let res = self.inner.delete_range(req).await; + self.after_write.wait().await; + res + } + + async fn batch_delete(&self, req: BatchDeleteRequest) -> Result { + self.before_write.wait().await; + let res = self.inner.batch_delete(req).await; + self.after_write.wait().await; + res + } + } + + #[derive(Clone, Copy, Debug)] + enum Write { + Txn, + TxnFailure, + Put, + BatchPut, + CompareAndPut, + DeleteRange, + BatchDelete, + } + + const WRITES: [Write; 7] = [ + Write::Txn, + Write::TxnFailure, + Write::Put, + Write::BatchPut, + Write::CompareAndPut, + Write::DeleteRange, + Write::BatchDelete, + ]; + const KEY: &[u8] = b"route"; + + impl Write { + async fn execute(self, store: &LeaderCachedKvBackend, value: &[u8]) { + match self { + Self::Txn => { + store + .txn(Txn::new().and_then(vec![TxnOp::Put(KEY.to_vec(), value.to_vec())])) + .await + .unwrap(); + } + Self::TxnFailure => { + let res = store + .txn( + Txn::new() + .when(vec![Compare::with_value_not_exists( + KEY.to_vec(), + CompareOp::Equal, + )]) + .or_else(vec![TxnOp::Put(KEY.to_vec(), value.to_vec())]), + ) + .await + .unwrap(); + assert!(!res.succeeded); + } + Self::Put => { + store + .put(PutRequest::new().with_key(KEY).with_value(value)) + .await + .unwrap(); + } + Self::BatchPut => { + store + .batch_put(BatchPutRequest { + kvs: vec![KeyValue { + key: KEY.to_vec(), + value: value.to_vec(), + }], + ..Default::default() + }) + .await + .unwrap(); + } + Self::CompareAndPut => { + let res = store + .compare_and_put( + CompareAndPutRequest::new() + .with_key(KEY) + .with_expect(b"old") + .with_value(value), + ) + .await + .unwrap(); + assert!(res.success); + } + Self::DeleteRange => { + store + .delete_range(DeleteRangeRequest::new().with_key(KEY)) + .await + .unwrap(); + } + Self::BatchDelete => { + store + .batch_delete(BatchDeleteRequest { + keys: vec![KEY.to_vec()], + prev_kv: false, + }) + .await + .unwrap(); + } + } + } + } + + async fn read(store: &LeaderCachedKvBackend, batch: bool) -> Option { + if batch { + store + .batch_get(BatchGetRequest { + keys: vec![KEY.to_vec()], + }) + .await + .unwrap() + .kvs + .pop() + } else { + store.get(KEY).await.unwrap() + } + } + + /// A cache miss reads the old value after a writer starts, then returns after + /// the writer has committed and updated/invalidated the cache (#9548). + #[tokio::test] + async fn test_cache_miss_overlaps_write() { + for write in WRITES { + for batch in [true, false] { + let backing = Arc::new(ControlledBackend::default()); + backing + .inner + .put(PutRequest::new().with_key(KEY).with_value(b"old")) + .await + .unwrap(); + let store = LeaderCachedKvBackend::with_always_leader(backing.clone()); + + let commit = backing.before_write.arm(); + let writer = write.execute(&store, b"new"); + tokio::pin!(writer); + assert!(poll!(writer.as_mut()).is_pending()); + + let return_read = backing.after_read.arm(); + let reader = read(&store, batch); + tokio::pin!(reader); + assert!(poll!(reader.as_mut()).is_pending()); + + commit.send(()).unwrap(); + writer.await; + return_read.send(()).unwrap(); + assert_eq!(reader.await.unwrap().value(), b"old"); + + let durable = backing.inner.get(KEY).await.unwrap(); + assert_eq!( + read(&store, batch).await, + durable, + "{write:?}, batch={batch}" + ); + } + } + } + + /// Start order, commit order and response order can all differ. An older + /// put response must not overwrite the cache after a newer mutation. + #[tokio::test] + async fn test_overlapping_writes() { + for first in [Write::Put, Write::BatchPut] { + for second in WRITES { + for reverse_commit in [false, true] { + let backing = Arc::new(ControlledBackend::default()); + backing + .inner + .put(PutRequest::new().with_key(KEY).with_value(b"old")) + .await + .unwrap(); + let store = LeaderCachedKvBackend::with_always_leader(backing.clone()); + + let resume = if reverse_commit { + backing.before_write.arm() + } else { + backing.after_write.arm() + }; + // Keep the initial value so the second CAS succeeds. + let writer = first.execute(&store, b"old"); + tokio::pin!(writer); + assert!(poll!(writer.as_mut()).is_pending()); + second.execute(&store, b"new").await; + resume.send(()).unwrap(); + writer.await; + + let durable = backing.inner.get(KEY).await.unwrap(); + assert_eq!( + read(&store, true).await, + durable, + "{first:?}, {second:?}, reverse_commit={reverse_commit}" + ); + } + } + } + } + + /// The later-started put commits first but returns last, after its stored + /// value has already been replaced by the first writer. + #[tokio::test] + async fn test_write_response_reordering() { + for first in WRITES { + for second in [Write::Put, Write::BatchPut] { + let backing = Arc::new(ControlledBackend::default()); + backing + .inner + .put(PutRequest::new().with_key(KEY).with_value(b"old")) + .await + .unwrap(); + let store = LeaderCachedKvBackend::with_always_leader(backing.clone()); + + let commit_first = backing.before_write.arm(); + let first_writer = first.execute(&store, b"new"); + tokio::pin!(first_writer); + assert!(poll!(first_writer.as_mut()).is_pending()); + + let return_second = backing.after_write.arm(); + let second_writer = second.execute(&store, b"old"); + tokio::pin!(second_writer); + assert!(poll!(second_writer.as_mut()).is_pending()); + + commit_first.send(()).unwrap(); + first_writer.await; + return_second.send(()).unwrap(); + second_writer.await; + + let durable = backing.inner.get(KEY).await.unwrap(); + assert_eq!(read(&store, true).await, durable, "{first:?}, {second:?}"); + } + } + } + + #[tokio::test] + async fn test_cache_miss_overlaps_reset() { + for batch in [true, false] { + let backing = Arc::new(ControlledBackend::default()); + backing + .inner + .put(PutRequest::new().with_key(KEY).with_value(b"old")) + .await + .unwrap(); + let store = LeaderCachedKvBackend::with_always_leader(backing.clone()); + + let return_read = backing.after_read.arm(); + let reader = read(&store, batch); + tokio::pin!(reader); + assert!(poll!(reader.as_mut()).is_pending()); + store.reset(); + return_read.send(()).unwrap(); + reader.await; + assert!(store.cache.get(KEY).await.unwrap().is_none()); + } + } + fn create_leader_cached_kv_backend() -> LeaderCachedKvBackend { let store = Arc::new(MemoryKvBackend::new()); LeaderCachedKvBackend::with_always_leader(store) diff --git a/src/mito2/src/access_layer.rs b/src/mito2/src/access_layer.rs index 6d489ff07fb..e8146c11330 100644 --- a/src/mito2/src/access_layer.rs +++ b/src/mito2/src/access_layer.rs @@ -573,6 +573,8 @@ pub struct SstWriteRequest { pub cache_manager: CacheManagerRef, #[allow(dead_code)] pub storage: Option, + /// Optional uniform row sequence for writes that do not preserve sequences. + /// Compaction passes `None` to retain the reader's effective input sequences. pub max_sequence: Option, pub sst_write_format: FormatType, diff --git a/src/mito2/src/compaction/compactor.rs b/src/mito2/src/compaction/compactor.rs index 8ad1cdf0b22..598c0b7f207 100644 --- a/src/mito2/src/compaction/compactor.rs +++ b/src/mito2/src/compaction/compactor.rs @@ -404,10 +404,33 @@ pub trait SstMerger: Send + Sync + 'static { #[derive(Clone)] pub struct DefaultSstMerger; +/// File-level sequence metadata is independent of physical row encoding. +struct OutputSequenceMetadata { + file_sequence: Option, + exact_sequence_trusted: bool, +} + +/// Compaction does not admit new data, so it inherits the input sequence bounds +/// without allocating a new admission marker. Retaining physical row sequences +/// does not by itself restore exact-read capability for untrusted inputs. +fn output_sequence_metadata( + region: &CompactionRegion, + inputs: &[FileHandle], +) -> OutputSequenceMetadata { + let exact_sequence_trusted = region.region_options.preserve_row_sequence + && inputs + .iter() + .all(|f| f.is_effective_target_sequence_trusted(region.region_id)); + OutputSequenceMetadata { + file_sequence: known_max_input_sequence(inputs), + exact_sequence_trusted, + } +} + /// Computes the maximum target-domain sequence bound of the output of merging -/// `inputs`. Foreign files are described by their target-local barrier; local -/// trusted files use their physical sequence bound. Unknown bounds never become -/// trusted through a partial maximum. +/// `inputs`. A bound can be a row maximum or an admission marker; compaction +/// must preserve both for sequence-based pruning. Unknown bounds must remain +/// unknown rather than being replaced with a partial maximum. fn known_max_input_sequence(inputs: &[FileHandle]) -> Option { let mut max: Option = None; for input in inputs { @@ -452,40 +475,7 @@ impl SstMerger for DefaultSstMerger { .iter() .map(|f| f.file_id().to_string()) .join(","); - let input_max_sequence = known_max_input_sequence(&output.inputs); - // The output is trusted only when every input is trusted in the target - // domain. Foreign marker values describe the source and are ignored; - // their present FileMeta.sequence is the target-local barrier. - let output_preserves_sequence = compaction_region.region_options.preserve_row_sequence - && output - .inputs - .iter() - .all(|f| f.is_effective_target_sequence_trusted(region_id)); - let output_sequence = if output_preserves_sequence { - input_max_sequence - } else { - // The manifest records the region's latest accepted sequence for - // edit paths and its flushed frontier otherwise. Compaction only - // rewrites the immutable SST snapshot, so this is the same - // region-local admission barrier used when accepting files. - let manifest = compaction_region - .manifest_ctx - .manifest_manager - .read() - .await - .manifest(); - NonZeroU64::new( - manifest - .committed_sequence - .unwrap_or(manifest.flushed_sequence) - + 1, - ) - }; - // For untrusted output, keep the physical sequence override compatible - // with the main write path: use the known maximum sequence of the - // inputs. The FileMeta sequence remains the admission barrier, but it - // must not be written into the rows (or replaced with zero). - let write_max_sequence = input_max_sequence.map(NonZeroU64::get); + let sequence_metadata = output_sequence_metadata(&compaction_region, &output.inputs); let builder = CompactionSstReaderBuilder { metadata: compaction_region.region_metadata.clone(), sst_layer: compaction_region.access_layer.clone(), @@ -509,13 +499,16 @@ impl SstMerger for DefaultSstMerger { source, cache_manager: compaction_region.cache_manager.clone(), storage, - max_sequence: write_max_sequence, + // Readers resolve file overrides before merge/dedup. Replacing + // their effective sequences here could promote old rows above + // versions in SSTs that were not part of this merge. + max_sequence: None, sst_write_format: if flat_format { FormatType::Flat } else { FormatType::PrimaryKey }, - preserve_row_sequence: output_preserves_sequence, + preserve_row_sequence: sequence_metadata.exact_sequence_trusted, index_options, index_config, inverted_index_config, @@ -564,12 +557,12 @@ impl SstMerger for DefaultSstMerger { index_version: 0, num_rows: sst_info.num_rows as u64, num_row_groups: sst_info.num_row_groups, - sequence: output_sequence, + sequence: sequence_metadata.file_sequence, partition_expr: partition_expr.clone(), num_series: sst_info.num_series, primary_key_min, primary_key_max, - preserve_row_sequence: output_preserves_sequence, + preserve_row_sequence: sequence_metadata.exact_sequence_trusted, } }) .collect::>(); @@ -887,6 +880,268 @@ mod tests { assert_eq!(None, known_max_input_sequence(&[])); } + #[tokio::test] + async fn test_output_sequence_metadata_preserves_trust_and_unknown_bounds() { + let mut region = new_test_compaction_region().await; + let local = region.region_id; + let foreign = RegionId::new(1, 2); + let file = |region_id, sequence: Option, preserve_row_sequence| { + new_file_handle(FileMeta { + region_id, + sequence: sequence.and_then(NonZeroU64::new), + preserve_row_sequence, + ..dummy_file_meta() + }) + }; + // Bounds are independent of row trust: known, mixed trust, unknown in + // either sequence domain, empty inputs, and the largest possible bound. + let cases = [ + ( + vec![file(local, Some(3), true), file(foreign, Some(9), false)], + Some(9), + true, + ), + ( + vec![file(local, Some(3), true), file(local, Some(6), false)], + Some(6), + false, + ), + ( + vec![file(foreign, Some(9), false), file(foreign, Some(11), true)], + Some(11), + true, + ), + ( + vec![file(local, Some(3), true), file(local, None, true)], + None, + true, + ), + ( + vec![file(local, None, false), file(local, Some(6), true)], + None, + false, + ), + ( + vec![file(local, Some(3), true), file(foreign, None, true)], + None, + false, + ), + (vec![], None, true), + ( + vec![file(local, Some(u64::MAX), false)], + Some(u64::MAX), + false, + ), + ]; + for committed_sequence in [100, 101] { + region + .manifest_ctx + .manifest_manager + .write() + .await + .update( + RegionMetaActionList::with_action(RegionMetaAction::Edit(RegionEdit { + committed_sequence: Some(committed_sequence), + files_to_add: vec![], + files_to_remove: vec![], + timestamp_ms: None, + compaction_time_window: None, + flushed_entry_id: None, + flushed_sequence: None, + })), + false, + ) + .await + .unwrap(); + for preserve in [false, true] { + region.region_options.preserve_row_sequence = preserve; + for (inputs, known_bound, inputs_trusted) in &cases { + let metadata = output_sequence_metadata(®ion, inputs); + let trusted = preserve && *inputs_trusted; + assert_eq!(trusted, metadata.exact_sequence_trusted); + assert_eq!( + known_bound.and_then(NonZeroU64::new), + metadata.file_sequence + ); + } + } + } + } + + /// Independent outputs must be safe in either visibility order, including when + /// only one output succeeds. Merely committing both outputs together is not a + /// substitute for preserving version order across their separate merge streams. + #[rstest::rstest] + #[case(true)] + #[case(false)] + #[tokio::test] + async fn test_independent_outputs_preserve_delete_in_every_visibility_state( + #[case] flat_format: bool, + #[values(false, true)] unknown_put_bound: bool, + ) { + use datatypes::arrow::array::AsArray; + use datatypes::arrow::datatypes::TimestampMillisecondType; + use store_api::region_engine::RegionEngine; + use store_api::region_request::RegionRequest; + + use crate::compaction::CompactionOutput; + use crate::compaction::compactor::{CompactionRegion, Compactor, DefaultCompactor}; + use crate::compaction::picker::PickerOutput; + use crate::compaction::reader::CompactionSstReaderBuilder; + use crate::engine::compaction_test::{delete_and_flush, put_and_flush}; + use crate::region::options::MergeMode; + use crate::sst::file::FileHandle; + use crate::sst::file_purger::NoopFilePurger; + use crate::test_util::{CreateRequestBuilder, TestEnv, rows_schema}; + + let mut env = TestEnv::new().await; + let region_id = RegionId::new(1, 1); + let config = MitoConfig { + default_flat_format: flat_format, + min_compaction_interval: Duration::from_secs(3600), + ..Default::default() + }; + let engine = env.create_engine(config).await; + let request = CreateRequestBuilder::new().build(); + let columns = rows_schema(&request); + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + put_and_flush(&engine, region_id, &columns, 0..1).await; + delete_and_flush(&engine, region_id, &columns, 0..1).await; + put_and_flush(&engine, region_id, &columns, 10..11).await; + let region = engine.get_region(region_id).unwrap(); + let version = region.version(); + let mut files: Vec<_> = version + .ssts + .levels() + .iter() + .flat_map(|level| level.files()) + .cloned() + .collect(); + assert_eq!(3, files.len()); + files.sort_unstable_by_key(|file| file.meta_ref().sequence); + if unknown_put_bound { + // Model a legacy SST whose physical rows are readable but whose + // file-level boundary is unknown. Merging cannot invent that bound. + let mut meta = files[0].meta_ref().clone(); + meta.sequence = None; + files[0] = FileHandle::new(meta, Arc::new(NoopFilePurger)); + } + let compaction_region = CompactionRegion { + region_id, + region_options: version.options.clone(), + engine_config: Arc::new(MitoConfig { + default_flat_format: flat_format, + ..Default::default() + }), + region_metadata: version.metadata.clone(), + cache_manager: engine.cache_manager(), + access_layer: region.access_layer.clone(), + manifest_ctx: region.manifest_ctx.clone(), + current_version: version.into(), + file_purger: None, + ttl: None, + max_parallelism: 2, + plugins: common_base::Plugins::new(), + }; + let puts = vec![files[0].clone(), files[2].clone()]; + let deletes = vec![files[1].clone()]; + let picker_output = PickerOutput { + outputs: [puts.clone(), deletes.clone()] + .into_iter() + .map(|inputs| CompactionOutput { + output_level: 1, + inputs, + filter_deleted: false, + output_time_range: None, + }) + .collect(), + expired_ssts: vec![], + time_window_size: 3600, + max_file_size: None, + }; + let merged = DefaultCompactor::with_merger(DefaultSstMerger) + .merge_ssts(&compaction_region, picker_output) + .await + .unwrap(); + assert_eq!(2, merged.files_to_add.len()); + let outputs: Vec<_> = merged + .files_to_add + .into_iter() + .map(|meta| FileHandle::new(meta, Arc::new(NoopFilePurger))) + .collect(); + let put_output = outputs + .iter() + .find(|file| file.meta_ref().num_rows == 2) + .unwrap(); + let delete_output = outputs + .iter() + .find(|file| file.meta_ref().num_rows == 1) + .unwrap(); + assert_eq!( + if unknown_put_bound { + None + } else { + NonZeroU64::new(3) + }, + put_output.meta_ref().sequence, + ); + assert_eq!(NonZeroU64::new(2), delete_output.meta_ref().sequence); + assert!( + outputs + .iter() + .all(|file| !file.meta_ref().preserve_row_sequence) + ); + + // Cover the original snapshot, each partial success/cancellation state, + // and both completed outputs using real persisted SSTs and merge readers. + for (put_done, delete_done) in [(false, false), (true, false), (false, true), (true, true)] + { + let mut visible = if put_done { + vec![put_output.clone()] + } else { + puts.clone() + }; + visible.extend(if delete_done { + vec![delete_output.clone()] + } else { + deletes.clone() + }); + let mut reader = CompactionSstReaderBuilder { + metadata: compaction_region.region_metadata.clone(), + sst_layer: region.access_layer.clone(), + cache: engine.cache_manager(), + inputs: &visible, + append_mode: false, + filter_deleted: true, + time_range: None, + merge_mode: MergeMode::LastRow, + } + .build_flat_sst_reader() + .await + .unwrap(); + let mut timestamps = Vec::new(); + while let Some(batch) = reader.next_batch().await.unwrap() { + timestamps.extend( + batch + .column_by_name("ts") + .unwrap() + .as_primitive::() + .values() + .iter() + .copied(), + ); + } + assert_eq!( + vec![10_000], + timestamps, + "put_done={put_done}, delete_done={delete_done}" + ); + } + } + /// Build a minimal [`CompactionRegion`] suitable for tests where the /// [`SstMerger`] is mocked and never touches the access layer. async fn new_test_compaction_region() -> CompactionRegion { diff --git a/src/mito2/src/engine/compaction_test.rs b/src/mito2/src/engine/compaction_test.rs index c67afbf8c6f..c1704c2b1c9 100644 --- a/src/mito2/src/engine/compaction_test.rs +++ b/src/mito2/src/engine/compaction_test.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::ops::Range; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; @@ -136,6 +136,295 @@ async fn collect_stream_ts(stream: SendableRecordBatchStream) -> Vec { res } +/// Flush may collapse versions within one file, but compaction must not promote +/// them, even with no external overlaps or after rewriting a previous output. +#[rstest::rstest] +#[case(false, false)] +#[case(true, false)] +#[case(true, true)] +#[tokio::test] +async fn test_compaction_preserves_flush_sequences_across_generations( + #[values(false, true)] flat_format: bool, + #[case] append_mode: bool, + #[case] preserve_row_sequence: bool, +) { + let mut env = TestEnv::new().await; + let config = MitoConfig { + default_flat_format: flat_format, + min_compaction_interval: Duration::from_secs(3600), + ..Default::default() + }; + let engine = env.create_engine(config.clone()).await; + let region_id = RegionId::new(1, 1); + let request = CreateRequestBuilder::new() + .insert_option("append_mode", &append_mode.to_string()) + .insert_option("preserve_row_sequence", &preserve_row_sequence.to_string()) + .insert_option("compaction.type", "twcs") + .insert_option("compaction.twcs.time_window", "1h") + .build(); + let table_dir = request.table_dir.clone(); + let options = request.options.clone(); + let columns = crate::test_util::rows_schema(&request); + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + + async fn read_sequences(engine: &MitoEngine, region_id: RegionId) -> Vec { + let region = engine.get_region(region_id).unwrap(); + let mut sequences = Vec::new(); + for file in region + .version() + .ssts + .levels() + .iter() + .flat_map(|l| l.files()) + { + let mut reader = region + .access_layer + .read_sst(file.clone()) + .build() + .await + .unwrap() + .unwrap(); + while let Some(batch) = reader.next_record_batch().await.unwrap() { + sequences.extend_from_slice( + batch + .column(batch.num_columns() - 2) + .as_primitive::() + .values(), + ); + } + } + sequences.sort_unstable(); + sequences + } + + let mut expected = Vec::new(); + for generation in 0..=2 { + // Interleaving time ranges exercise a merge without duplicate rows. + let rows = [generation, generation + 10] + .into_iter() + .flat_map(|ts| build_rows_for_key("a", ts, ts + 1, 0)) + .collect(); + put_rows( + &engine, + region_id, + Rows { + schema: columns.clone(), + rows, + }, + ) + .await; + flush(&engine, region_id).await; + let max = (generation * 2 + 2) as u64; + expected.extend(if preserve_row_sequence { + [max - 1, max] + } else { + [max, max] + }); + assert_eq!(expected, read_sequences(&engine, region_id).await); + if generation == 0 { + continue; + } + engine + .handle_request( + region_id, + RegionRequest::Compact(RegionCompactRequest { + options: api::v1::region::compact_request::Options::StrictWindow( + api::v1::region::StrictWindow { + window_seconds: 3600, + }, + ), + ..Default::default() + }), + ) + .await + .unwrap(); + let version = engine.get_region(region_id).unwrap().version(); + let files: Vec<_> = version + .ssts + .levels() + .iter() + .flat_map(|l| l.files()) + .collect(); + assert_eq!( + 1, + files.len(), + "generation {generation} must merge all inputs" + ); + assert_eq!( + preserve_row_sequence, + files[0].meta_ref().preserve_row_sequence + ); + assert_eq!(Some(max), files[0].meta_ref().sequence.map(|s| s.get())); + assert_eq!( + expected, + read_sequences(&engine, region_id).await, + "generation {generation}" + ); + } + let engine = env.reopen_engine(engine, config).await; + crate::test_util::reopen_region(&engine, region_id, table_dir, false, options).await; + assert_eq!(expected, read_sequences(&engine, region_id).await); +} + +#[tokio::test] +async fn test_partial_compaction_preserves_delete_order_flat() { + assert_partial_compaction_preserves_delete_order(true).await; +} + +#[tokio::test] +async fn test_partial_compaction_preserves_delete_order_primary_key() { + assert_partial_compaction_preserves_delete_order(false).await; +} + +/// A newer unrelated input must not promote an old Put above an unselected Delete. +/// All files fit in the same one-hour window; no cross-window SST is required. +async fn assert_partial_compaction_preserves_delete_order(flat_format: bool) { + let mut env = TestEnv::new().await; + let region_id = RegionId::new(1, 1); + let (engine, columns) = env_for_manual_compaction(&mut env, region_id, flat_format).await; + + put_and_flush(&engine, region_id, &columns, 10..16).await; + let mut deletes = build_rows_for_key("a", 10, 16, 0); + // Distinct keys keep the Delete SST large even after compression, so the + // byte-balance rule excludes it without fabricating FileMeta sizes. + let mut rng = ::seed_from_u64(0); + for _ in 0..3000 { + deletes.extend(build_rows_for_key( + &format!("{:032x}", rand::Rng::random::(&mut rng)), + 1, + 2, + 0, + )); + } + engine + .handle_request( + region_id, + RegionRequest::Delete(RegionDeleteRequest { + rows: Rows { + schema: columns.clone(), + rows: deletes, + }, + hint: None, + partition_expr_version: None, + }), + ) + .await + .unwrap(); + flush(&engine, region_id).await; + // These newer writes are legitimate, but must not change the version of 10..16. + for i in 10..25 { + put_and_flush(&engine, region_id, &columns, i * 10..i * 10 + 6).await; + } + + let region = engine.get_region(region_id).unwrap(); + let table_dir = region.table_dir().to_string(); + let version = region.version(); + let files = version + .ssts + .levels() + .iter() + .flat_map(|level| level.files()) + .collect::>(); + assert_eq!(17, files.len()); + assert!(files.iter().all(|file| { + file.time_range().0 >= Timestamp::new_millisecond(0) + && file.time_range().1 < Timestamp::new_millisecond(3_600_000) + })); + let delete_file = files + .iter() + .find(|file| file.meta_ref().num_rows == 3006) + .unwrap(); + let delete_id = delete_file.file_id(); + let delete_sequence = delete_file.meta_ref().sequence; + let old_ids = files + .iter() + .map(|file| file.file_id()) + .collect::>(); + let before = collect_stream_ts( + engine + .scanner(region_id, ScanRequest::default()) + .await + .unwrap() + .scan() + .await + .unwrap(), + ) + .await; + let expected = (10..25) + .flat_map(|i| (i * 10..i * 10 + 6).map(|ts| ts * 1000)) + .collect::>(); + assert_eq!( + expected, before, + "the Delete must hide old rows before compaction" + ); + + compact(&engine, region_id).await; + + let scanner = engine + .scanner(region_id, ScanRequest::default()) + .await + .unwrap(); + let current_ids = scanner.file_ids(); + assert_eq!( + 2, + current_ids.len(), + "the 16 small inputs must merge, leaving the large Delete SST" + ); + assert!(current_ids.contains(&delete_id)); + assert_eq!( + 1, + current_ids.iter().filter(|id| old_ids.contains(id)).count() + ); + let after = collect_stream_ts(scanner.scan().await.unwrap()).await; + + let current = region.version(); + let output = current + .ssts + .levels() + .iter() + .flat_map(|level| level.files()) + .find(|file| !old_ids.contains(&file.file_id())) + .unwrap(); + let mut reader = region + .access_layer + .read_sst(output.clone()) + .build() + .await + .unwrap() + .unwrap(); + let batch = reader.next_record_batch().await.unwrap().unwrap(); + let output_sequences = batch + .column(batch.num_columns() - 2) + .as_primitive::(); + let first_output_sequence = output_sequences.value(0); + + crate::test_util::reopen_region(&engine, region_id, table_dir, true, HashMap::new()).await; + let reopened = collect_stream_ts( + engine + .scanner(region_id, ScanRequest::default()) + .await + .unwrap() + .scan() + .await + .unwrap(), + ) + .await; + let resurrected = after + .iter() + .filter(|ts| (10_000..16_000).contains(*ts)) + .collect::>(); + assert!( + after == expected && reopened == expected, + "partial compaction changed delete ordering (flat_format={flat_format}, unselected Delete sequence={delete_sequence:?}, first output row sequence={first_output_sequence}); expected rows={}, after rows={}, reopened rows={}, resurrected timestamps={resurrected:?}", + expected.len(), + after.len(), + reopened.len() + ); +} + struct CompactionListenerGuard(Option>); impl CompactionListenerGuard { diff --git a/src/mito2/src/engine/merge_mode_test.rs b/src/mito2/src/engine/merge_mode_test.rs index afb984e77bd..5430e6b246c 100644 --- a/src/mito2/src/engine/merge_mode_test.rs +++ b/src/mito2/src/engine/merge_mode_test.rs @@ -16,6 +16,9 @@ use api::v1::Rows; use common_recordbatch::RecordBatches; +use datafusion_expr::{col, lit}; +use datatypes::arrow::array::AsArray; +use datatypes::arrow::datatypes::UInt64Type; use store_api::region_engine::RegionEngine; use store_api::region_request::{RegionCompactRequest, RegionRequest}; use store_api::storage::{RegionId, ScanRequest}; @@ -27,6 +30,192 @@ use crate::test_util::{ delete_rows_schema, flush_region, put_rows, reopen_region, rows_schema, }; +#[rstest::rstest] +#[tokio::test] +async fn test_partial_compaction_preserves_last_row_put_versions( + #[values(false, true)] flat_format: bool, +) { + let mut env = TestEnv::new().await; + let config = MitoConfig { + default_flat_format: flat_format, + min_compaction_interval: std::time::Duration::from_secs(3600), + ..Default::default() + }; + let engine = env.create_engine(config.clone()).await; + let region_id = RegionId::new(1, 1); + env.get_schema_metadata_manager() + .register_region_table_info( + region_id.table_id(), + "test_table", + "test_catalog", + "test_schema", + None, + env.get_kv_backend(), + ) + .await; + let request = CreateRequestBuilder::new() + .field_num(2) + .insert_option("compaction.type", "twcs") + .insert_option("compaction.twcs.time_window", "1h") + .insert_option("merge_mode", "last_row") + .build(); + let table_dir = request.table_dir.clone(); + let region_opts = request.options.clone(); + let schema = rows_schema(&request); + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + + // A(x=1), B(x=2) share a key and timestamp; C is unrelated. + // Promoting A when merging A+C must not hide the unselected B. + let a = build_rows_with_fields("a", &[10], &[(Some(1), None)]); + put_rows( + &engine, + region_id, + Rows { + schema: schema.clone(), + rows: a, + }, + ) + .await; + flush_region(&engine, region_id, None).await; + let mut b = build_rows_with_fields("a", &[10], &[(Some(2), None)]); + let mut rng = ::seed_from_u64(0); + for _ in 0..3000 { + b.extend(build_rows_with_fields( + &format!("{:032x}", rand::Rng::random::(&mut rng)), + &[1], + &[(Some(2), None)], + )); + } + put_rows( + &engine, + region_id, + Rows { + schema: schema.clone(), + rows: b, + }, + ) + .await; + flush_region(&engine, region_id, None).await; + let b_id = engine + .get_region(region_id) + .unwrap() + .version() + .ssts + .levels() + .iter() + .flat_map(|level| level.files()) + .find(|file| file.meta_ref().num_rows == 3001) + .unwrap() + .file_id(); + let c = build_rows_with_fields("z", &[10], &[(None, Some(3))]); + put_rows( + &engine, + region_id, + Rows { + schema: schema.clone(), + rows: c, + }, + ) + .await; + flush_region(&engine, region_id, None).await; + // Sixteen small inputs plus the large middle-version file B. The normal + // byte-balanced candidate excludes B. + for i in 0..14 { + let rows = build_rows_with_fields("z", &[100 + i], &[(Some(4), None)]); + put_rows( + &engine, + region_id, + Rows { + schema: schema.clone(), + rows, + }, + ) + .await; + flush_region(&engine, region_id, None).await; + } + let scan = || ScanRequest { + filters: vec![col("tag_0").eq(lit("a"))], + ..Default::default() + }; + let before = + RecordBatches::try_collect(engine.scan_to_stream(region_id, scan()).await.unwrap()) + .await + .unwrap() + .pretty_print() + .unwrap(); + let expected = "\ ++-------+---------+---------+---------------------+ +| tag_0 | field_0 | field_1 | ts | ++-------+---------+---------+---------------------+ +| a | 2.0 | | 1970-01-01T00:00:10 | ++-------+---------+---------+---------------------+"; + assert_eq!(expected, before); + engine + .handle_request( + region_id, + RegionRequest::Compact(RegionCompactRequest::default()), + ) + .await + .unwrap(); + let after = RecordBatches::try_collect(engine.scan_to_stream(region_id, scan()).await.unwrap()) + .await + .unwrap() + .pretty_print() + .unwrap(); + assert_eq!(expected, after, "flat_format={flat_format}"); + let version = engine.get_region(region_id).unwrap().version(); + let files: Vec<_> = version + .ssts + .levels() + .iter() + .flat_map(|level| level.files()) + .collect(); + assert_eq!(2, files.len()); + assert!(files.iter().any(|file| file.file_id() == b_id)); + assert!( + files + .iter() + .all(|file| !file.meta_ref().preserve_row_sequence) + ); + let region = engine.get_region(region_id).unwrap(); + let mut output_sequences = std::collections::HashSet::new(); + for file in files.iter().filter(|file| file.file_id() != b_id) { + let mut reader = region + .access_layer + .read_sst((*file).clone()) + .build() + .await + .unwrap() + .unwrap(); + while let Some(batch) = reader.next_record_batch().await.unwrap() { + output_sequences.extend( + batch + .column(batch.num_columns() - 2) + .as_primitive::() + .values() + .iter() + .copied(), + ); + } + } + assert!( + output_sequences.len() > 1, + "compaction must retain effective input sequences" + ); + let engine = env.reopen_engine(engine, config).await; + reopen_region(&engine, region_id, table_dir, false, region_opts).await; + let reopened = + RecordBatches::try_collect(engine.scan_to_stream(region_id, scan()).await.unwrap()) + .await + .unwrap() + .pretty_print() + .unwrap(); + assert_eq!(expected, reopened); +} + #[tokio::test] async fn test_merge_mode_write_query() { test_merge_mode_write_query_with_format(false).await; diff --git a/src/mito2/src/engine/scan_test.rs b/src/mito2/src/engine/scan_test.rs index 6e54ab3a8ba..703d49fa75c 100644 --- a/src/mito2/src/engine/scan_test.rs +++ b/src/mito2/src/engine/scan_test.rs @@ -2788,9 +2788,8 @@ async fn test_bulk_write_sequence_not_committed_before_install() { } } -/// Non-preserving compaction must keep the physical input sequence below the -/// admission barrier. Otherwise a later row at the barrier can collide with -/// the compacted row when ordinary reads deduplicate overlapping SSTs. +/// Compaction must not allocate a new sequence that can collide with the next +/// write. Preserve the input's effective row sequence and file-level bound. #[tokio::test] async fn test_non_preserve_compaction_sequence_collision() { for flat_format in [false, true] { @@ -2860,8 +2859,7 @@ async fn test_non_preserve_compaction_sequence_collision_with_format(flat_format ); // Both current SSTs are compacted through the real picker, merger, writer, - // and manifest update. With committed/flushed sequence 2, the untrusted - // output's admission barrier is exactly 3. + // and manifest update. Both the row version and inherited file bound stay 2. engine .handle_request( region_id, @@ -2884,14 +2882,13 @@ async fn test_non_preserve_compaction_sequence_collision_with_format(flat_format assert!(!compacted.meta_ref().preserve_row_sequence); assert_eq!(1, compacted.num_rows()); assert_eq!( - Some(std::num::NonZeroU64::new(3).unwrap()), + Some(std::num::NonZeroU64::new(2).unwrap()), compacted.meta_ref().sequence, - "admission barrier is input max 2 plus one" + "compaction must inherit the input bound without allocating a sequence" ); // Hide FileMeta.sequence so this is a direct physical read, not a reader - // admission-barrier override. The output must contain physical sequence 2, - // not zero and not the output barrier 3. + // override. The output must contain physical sequence 2, not zero or 3. let mut physical_meta = compacted.meta_ref().clone(); physical_meta.sequence = None; let mut reader = region @@ -2918,10 +2915,8 @@ async fn test_non_preserve_compaction_sequence_collision_with_format(flat_format "after compaction: {after_compaction}" ); - // The next write receives physical sequence 3, equal to the compacted - // output's admission barrier. It must still win because the compacted row - // remains physically at sequence 2; an old zero/barrier encoding would - // collide here and incorrectly retain value 2. + // The next write receives sequence 3 and must win. The old zero/barrier + // encoding promoted the compacted row to 3 and could incorrectly retain 2. test_util::put_rows( &engine, region_id, @@ -2956,7 +2951,7 @@ async fn test_non_preserve_compaction_sequence_collision_with_format(flat_format .find(|file| file.meta_ref().level == 1) .expect("compacted L1 SST"); assert_eq!( - Some(std::num::NonZeroU64::new(3).unwrap()), + Some(std::num::NonZeroU64::new(2).unwrap()), compacted.meta_ref().sequence ); @@ -2967,24 +2962,27 @@ async fn test_non_preserve_compaction_sequence_collision_with_format(flat_format ); } -/// Compaction rewrites a legacy (unmarked) input as an untrusted output with -/// the region-local admission barrier. Its physical rows retain the known -/// input maximum, while exact scans skip it once C reaches that barrier and -/// fail closed while C is below it. +/// A legacy input must retain its admission boundary without gaining row trust. +/// A later flush can advance beyond manifest.committed_sequence, so compaction +/// must inherit the maximum input bound rather than sample that stale frontier. +#[rstest::rstest] #[tokio::test] -async fn test_compaction_output_non_preserve_not_laundered_from_legacy_input() { +async fn test_compaction_output_non_preserve_not_laundered_from_legacy_input( + #[values(false, true)] flat_format: bool, + #[values(false, true)] flush_after_edit: bool, +) { let mut env = TestEnv::with_prefix("test_compaction_output_not_laundered_from_legacy_input").await; // Suppress automatic edit-triggered compactions. The high TWCS trigger below // also prevents flush-triggered compaction from consuming the inputs, so the // explicit Compact below is the only compaction in flight (deterministic). - let engine = env - .create_engine(MitoConfig { - min_compaction_interval: std::time::Duration::from_secs(60 * 60), - schedule_compaction_after_edit: false, - ..Default::default() - }) - .await; + let config = MitoConfig { + default_flat_format: flat_format, + min_compaction_interval: std::time::Duration::from_secs(60 * 60), + schedule_compaction_after_edit: false, + ..Default::default() + }; + let engine = env.create_engine(config.clone()).await; let region_id = RegionId::new(1, 1); let request = CreateRequestBuilder::new() @@ -2994,6 +2992,8 @@ async fn test_compaction_output_non_preserve_not_laundered_from_legacy_input() { .insert_option("compaction.twcs.trigger_file_num", "100") .build(); let column_schemas = test_util::rows_schema(&request); + let table_dir = request.table_dir.clone(); + let region_options = request.options.clone(); engine .handle_request(region_id, RegionRequest::Create(request)) @@ -3053,6 +3053,30 @@ async fn test_compaction_output_non_preserve_not_laundered_from_legacy_input() { "seeded file should be unmarked" ); + let mut expected_sequences = vec![1, 2, 3, 4, 5, 6]; + let expected_bound = if flush_after_edit { + test_util::put_rows( + &engine, + region_id, + Rows { + schema: column_schemas, + rows: test_util::build_rows(6, 9), + }, + ) + .await; + test_util::flush_region(&engine, region_id, None).await; + expected_sequences.extend([8, 9, 10]); + 10 + } else { + 7 + }; + let manifest = region.manifest_ctx.manifest().await; + assert_eq!(Some(7), manifest.committed_sequence); + assert_eq!( + if flush_after_edit { 10 } else { 6 }, + manifest.flushed_sequence + ); + // Compact: the rewritten output must NOT be laundered back to marked. engine .handle_request( @@ -3079,25 +3103,24 @@ async fn test_compaction_output_non_preserve_not_laundered_from_legacy_input() { assert_eq!( 1, outputs.len(), - "two inputs should rewrite into one output file" + "all inputs should rewrite into one output file" ); assert!( !outputs[0].meta_ref().preserve_row_sequence, "legacy input must not be laundered into a marked output" ); - // The compacted output is sequence-less and carries the current region's - // admission barrier rather than any source-domain sequence. + // Inherit the admitted legacy bound and, if present, the newer flush bound. let barrier = outputs[0] .meta_ref() .sequence .expect("compaction output barrier") .get(); - assert_eq!(8, barrier); + assert_eq!(expected_bound, barrier); - // Reinstalling the legacy input assigns it sequence 7, so the physical - // parquet retains that known input maximum rather than encoding either - // zero or the output admission barrier 8. The manifest marker stays false. + // Reinstalling the legacy input assigns its metadata sequence 7, but the + // existing rows keep sequences 1..=6. Compaction must not rewrite them to + // the metadata bound, including when newer flushed rows join the merge. let mut sequence_meta = outputs[0].meta_ref().clone(); sequence_meta.sequence = None; let sequence_handle = FileHandle::new( @@ -3111,22 +3134,34 @@ async fn test_compaction_output_non_preserve_not_laundered_from_legacy_input() { .await .unwrap() .expect("compaction output reader"); - let batch = reader - .next_record_batch() - .await - .unwrap() - .expect("compaction output batch"); - let sequence = batch - .column(batch.num_columns() - 2) - .as_any() - .downcast_ref::() - .expect("sequence column"); - assert!( - sequence - .values() - .iter() - .all(|sequence| *sequence == barrier - 1) + let mut sequences = Vec::new(); + while let Some(batch) = reader.next_record_batch().await.unwrap() { + let sequence = batch + .column(batch.num_columns() - 2) + .as_any() + .downcast_ref::() + .expect("sequence column"); + sequences.extend_from_slice(sequence.values()); + } + sequences.sort_unstable(); + assert_eq!(expected_sequences, sequences); + + // The inherited bound and untrusted marker must survive manifest recovery. + let engine = env.reopen_engine(engine, config).await; + test_util::reopen_region(&engine, region_id, table_dir, false, region_options).await; + let version = engine.get_region(region_id).unwrap().version(); + let files: Vec<_> = version + .ssts + .levels() + .iter() + .flat_map(|level| level.files()) + .collect(); + assert_eq!(1, files.len()); + assert_eq!( + Some(expected_bound), + files[0].meta_ref().sequence.map(|s| s.get()) ); + assert!(!files[0].meta_ref().preserve_row_sequence); // A cursor before the barrier must fail closed. let err = engine @@ -3142,9 +3177,27 @@ async fn test_compaction_output_non_preserve_not_laundered_from_legacy_input() { .await .err() .expect("newer barrier must disable exact scanning"); - assert!(matches!(err, Error::SequenceRangeUnsupported { .. })); + if flush_after_edit { + // Without exact capability, the lower-bound fence is checked first. + assert!( + matches!( + err, + Error::IncrementalQueryStale { + given_seq: 9, + min_readable_seq: 10, + .. + } + ), + "unexpected error: {err:?}" + ); + } else { + assert!( + matches!(err, Error::SequenceRangeUnsupported { .. }), + "unexpected error: {err:?}" + ); + } - // Once C reaches the barrier the sequence-less file is skipped, so exact + // Once C reaches the barrier the untrusted file is skipped, so exact // capability is restored without attempting row-level filtering. let scanner = engine .scanner( diff --git a/src/mito2/src/read/scan_region.rs b/src/mito2/src/read/scan_region.rs index b7685314ef8..c38e5c9c596 100644 --- a/src/mito2/src/read/scan_region.rs +++ b/src/mito2/src/read/scan_region.rs @@ -1836,9 +1836,10 @@ fn pre_filter_mode(append_mode: bool, merge_mode: MergeMode) -> PreFilterMode { /// contribute a row to `(C, H]`. /// /// Unmarked local files use `FileMeta.sequence` as an admission barrier rather -/// than a row maximum. Compaction and edit assign it as `committed_sequence + 1`; -/// `C >= barrier` proves Flow has already consumed the entire file, so such a -/// file is excluded before the capability check. +/// than a row maximum. Region edits allocate this barrier; compaction inherits +/// the maximum input bound without assigning a new barrier. `C >= barrier` +/// proves Flow has already consumed the entire file, so such a file is excluded +/// before the capability check. An unknown bound cannot prove this exclusion. /// /// A foreign file is different: the parquet reader virtualizes every row to its /// target-local `FileMeta.sequence`. Consequently, a present sequence is the diff --git a/src/mito2/src/region/options.rs b/src/mito2/src/region/options.rs index 9a8d17c954c..7c5e44200ec 100644 --- a/src/mito2/src/region/options.rs +++ b/src/mito2/src/region/options.rs @@ -119,7 +119,8 @@ pub struct RegionOptions { /// the configured size; zero disables both limits. #[serde(skip_serializing_if = "Option::is_none")] pub write_buffer_size: Option, - /// Whether to preserve per-row sequence numbers through flush and compaction. + /// Whether to preserve original per-row sequence numbers when flushing. + /// Compaction always retains effective input sequences, regardless of this option. /// /// Only meaningful for append-only tables (`append_mode = true`): when enabled, /// every row keeps its exact sequence number in memtables, flushed SSTs and diff --git a/src/mito2/src/sst/file.rs b/src/mito2/src/sst/file.rs index b604275ce93..9702477b07a 100644 --- a/src/mito2/src/sst/file.rs +++ b/src/mito2/src/sst/file.rs @@ -243,10 +243,13 @@ pub struct FileMeta { /// the default value `0` doesn't means the file doesn't contains any rows, /// but instead means the number of rows is unknown. pub num_row_groups: u64, - /// Sequence in this file. + /// File-level sequence bound or admission marker in the target region. /// - /// This sequence is the only sequence in this file. And it's retrieved from the max - /// sequence of the rows on generating this file. + /// Flush records the maximum input row sequence. Compaction outputs inherit + /// the maximum input bound, or remain unknown if any input bound is unknown, + /// independently of per-row sequence trust. This does not imply that every + /// physical row has this sequence. + /// Readers also use it to normalize foreign and legacy all-zero files. pub sequence: Option, /// Partition expression from the region metadata when the file is created. /// @@ -284,6 +287,8 @@ pub struct FileMeta { pub primary_key_max: Option, /// Whether the file preserves per-row sequence numbers usable for exact /// row-level sequence filtering. + /// Merely retaining physical input sequences during compaction does not + /// restore this capability for untrusted inputs. #[serde(default, skip_serializing_if = "is_false")] pub preserve_row_sequence: bool, } diff --git a/src/mito2/src/sst/index/inverted_index/applier.rs b/src/mito2/src/sst/index/inverted_index/applier.rs index 0f665eec549..8f0c309a003 100644 --- a/src/mito2/src/sst/index/inverted_index/applier.rs +++ b/src/mito2/src/sst/index/inverted_index/applier.rs @@ -221,8 +221,9 @@ impl InvertedIndexApplier { let start = Instant::now(); let context = SearchContext { - // Encountering a non-existing column indicates that it doesn't match predicates. - index_not_found_strategy: IndexNotFoundStrategy::ReturnEmpty, + // Older SSTs may lack an index enabled by ALTER, or a column with a default value. + // Missing indexes cannot rule out matching rows. + index_not_found_strategy: IndexNotFoundStrategy::Ignore, }; let mut cache_miss = 0; diff --git a/src/mito2/src/sst/index/inverted_index/creator.rs b/src/mito2/src/sst/index/inverted_index/creator.rs index 9cad6fa84ee..6d58e71ef10 100644 --- a/src/mito2/src/sst/index/inverted_index/creator.rs +++ b/src/mito2/src/sst/index/inverted_index/creator.rs @@ -560,6 +560,7 @@ mod tests { async fn build_applier_factory( prefix: &str, rows: BTreeSet<(&'static str, i32, [u64; 2])>, + sst_indexed_column_ids: HashSet, ) -> impl Fn(DfExpr) -> BoxFuture<'static, Vec> { let (d, factory) = PuffinManagerFactory::new_for_test_async(prefix).await; let table_dir = "table0".to_string(); @@ -577,7 +578,7 @@ mod tests { intm_mgr, memory_threshold, NonZeroUsize::new(segment_row_count).unwrap(), - indexed_column_ids.clone(), + sst_indexed_column_ids, ); for (str_tag, i32_tag, u64_field) in &rows { @@ -628,6 +629,42 @@ mod tests { } } + #[tokio::test] + async fn test_create_and_query_missing_index() { + let rows = BTreeSet::from_iter([("aaa", 1, [1, 2]), ("bbb", 2, [3, 4])]); + // The SST predates enabling indexes on tag_i32 and field_u64. + let applier_factory = build_applier_factory( + "test_create_and_query_missing_index_", + rows, + HashSet::from([1]), + ) + .await; + + let cases = [ + (col("tag_i32").eq(lit(1)), vec![0, 1]), + ( + col("tag_i32") + .eq(lit(1)) + .and(col("field_u64").eq(lit(1u64))), + vec![0, 1], + ), + ( + col("tag_str").eq(lit("aaa")).and(col("tag_i32").eq(lit(1))), + vec![0], + ), + ( + col("tag_str") + .eq(lit("missing")) + .and(col("tag_i32").eq(lit(1))), + vec![], + ), + ]; + for (expr, expected) in cases { + let actual = applier_factory(expr.clone()).await; + assert_eq!(actual, expected, "{expr}"); + } + } + #[tokio::test] async fn test_create_and_query_get_key() { let rows = BTreeSet::from_iter([ @@ -642,7 +679,12 @@ mod tests { ("abc", 3, [9, 10]), ]); - let applier_factory = build_applier_factory("test_create_and_query_get_key_", rows).await; + let applier_factory = build_applier_factory( + "test_create_and_query_get_key_", + rows, + HashSet::from([1, 2, 4]), + ) + .await; let expr = col("tag_str").eq(lit("aaa")); let res = applier_factory(expr).await; @@ -685,7 +727,12 @@ mod tests { ("abc", 3, [9, 10]), ]); - let applier_factory = build_applier_factory("test_create_and_query_range_", rows).await; + let applier_factory = build_applier_factory( + "test_create_and_query_range_", + rows, + HashSet::from([1, 2, 4]), + ) + .await; let expr = col("tag_str").between(lit("aaa"), lit("aab")); let res = applier_factory(expr).await; @@ -722,8 +769,12 @@ mod tests { ("abc", 3, [9, 10]), ]); - let applier_factory = - build_applier_factory("test_create_and_query_comparison_", rows).await; + let applier_factory = build_applier_factory( + "test_create_and_query_comparison_", + rows, + HashSet::from([1, 2, 4]), + ) + .await; let expr = col("tag_str").lt(lit("aab")); let res = applier_factory(expr).await; @@ -804,7 +855,12 @@ mod tests { ("abc", 3, [9, 10]), ]); - let applier_factory = build_applier_factory("test_create_and_query_regex_", rows).await; + let applier_factory = build_applier_factory( + "test_create_and_query_regex_", + rows, + HashSet::from([1, 2, 4]), + ) + .await; let expr = binary_expr(col("tag_str"), Operator::RegexMatch, lit(".*")); let res = applier_factory(expr).await; diff --git a/src/query/src/dist_plan/merge_scan.rs b/src/query/src/dist_plan/merge_scan.rs index 668e811b95b..020a77bdf77 100644 --- a/src/query/src/dist_plan/merge_scan.rs +++ b/src/query/src/dist_plan/merge_scan.rs @@ -166,8 +166,20 @@ fn is_json_field(field: &Field) -> bool { == Some("Json") } +/// Returns true when the remote field's nullability fits the expected field. +/// +/// The remote physical optimizer may narrow a nullable column to non-nullable, +/// as DataFusion's own schema invariant permits: `AggregateStatistics` +/// replaces an aggregate over an empty region with a non-null literal. Batches +/// are rebuilt with the expected schema, so the narrowing does not leak. +/// Widening is rejected. Nested field nullability is part of the data type and +/// still has to match exactly. +fn nullability_compatible(expected_field: &Field, actual_field: &Field) -> bool { + expected_field.is_nullable() || !actual_field.is_nullable() +} + /// Returns true when two fields describe the same column semantics: same -/// name, same data type, and same nullability. +/// name, same data type, and compatible nullability. /// /// Arrow `Field` metadata is auxiliary information (indexing, encoding, /// compression hints, ...) and does not participate in the comparison. This @@ -178,7 +190,7 @@ fn is_json_field(field: &Field) -> bool { fn fields_semantically_equal(expected_field: &Field, actual_field: &Field) -> bool { expected_field.name() == actual_field.name() && expected_field.data_type() == actual_field.data_type() - && expected_field.is_nullable() == actual_field.is_nullable() + && nullability_compatible(expected_field, actual_field) } /// Returns true when two fields are semantically equivalent JSON columns. @@ -195,7 +207,7 @@ fn json_fields_compatible(expected_field: &Field, actual_field: &Field) -> bool is_json_field(expected_field) && is_json_field(actual_field) && expected_field.name() == actual_field.name() - && expected_field.is_nullable() == actual_field.is_nullable() + && nullability_compatible(expected_field, actual_field) // Both must carry the same JSON marker. && expected_field.metadata().get(datatypes::schema::TYPE_KEY) == actual_field.metadata().get(datatypes::schema::TYPE_KEY) @@ -211,14 +223,15 @@ fn json_fields_compatible(expected_field: &Field, actual_field: &Field) -> bool /// Validates the remote schema before positional column handling. /// /// Field metadata (indexing/encoding/compression hints) is non-semantic at -/// this boundary and is ignored: two fields are equal when their name, data -/// type, and nullability match (see [`fields_semantically_equal`]). A -/// timestamp timezone difference is the only intentional exception, accepted -/// when the fields share the same timestamp unit, distinct timezones, and -/// equal name and nullability. Top-level Arrow schema metadata is -/// non-semantic at this boundary as well. JSON columns are compared -/// semantically (see [`json_fields_compatible`]) because their wire and -/// decoded representations use different physical arrow types. +/// this boundary and is ignored: two fields are equal when their name and +/// data type match and their nullability is compatible (see +/// [`fields_semantically_equal`]). A timestamp timezone difference is the only +/// intentional exception, accepted when the fields share the same timestamp +/// unit, distinct timezones, equal name, and compatible nullability. +/// Top-level Arrow schema metadata is non-semantic at this boundary as well. +/// JSON columns are compared semantically (see [`json_fields_compatible`]) +/// because their wire and decoded representations use different physical +/// arrow types. fn validate_remote_schema( expected: &ArrowSchema, actual: &ArrowSchema, @@ -259,7 +272,7 @@ fn validate_remote_schema( ) if expected_unit == actual_unit && expected_timezone != actual_timezone && expected_field.name() == actual_field.name() - && expected_field.is_nullable() == actual_field.is_nullable() + && nullability_compatible(expected_field, actual_field) ); if !timezone_only_difference { return Err(remote_schema_mismatch(format!( @@ -2600,6 +2613,30 @@ mod tests { assert_eq!(batches[0].schema().as_ref(), &expected_schema); } + #[tokio::test] + async fn qbs_merge_scan_remote_schema_identity_emits_expected_schema_on_nullability_narrowing() + { + let batch = record_batch( + int64_schema(&["a", "b"]), + vec![ + Arc::new(Int64Vector::from_slice([11])) as _, + Arc::new(Int64Vector::from_slice([12])) as _, + ], + ); + let expected_schema = ArrowSchema::new(vec![ + Field::new("a", TestArrowDataType::Int64, true), + Field::new("b", TestArrowDataType::Int64, false), + ]); + let batches = collect_merge_scan(merge_scan_exec( + vec![(RegionId::new(1024, 1), batch)], + expected_schema.clone(), + 1, + )) + .await + .unwrap(); + assert_eq!(batches[0].schema().as_ref(), &expected_schema); + } + #[tokio::test] async fn qbs_merge_scan_remote_schema_identity_rejects_incompatible_type() { let batch = record_batch( @@ -2806,7 +2843,7 @@ mod tests { } #[test] - fn merge_scan_remote_schema_identity_ignores_field_metadata_but_rejects_nullability_mismatch() { + fn merge_scan_remote_schema_identity_ignores_field_metadata_but_rejects_nullability_widening() { let expected = expected_int64_schema(); let metadata_mismatch = ArrowSchema::new_with_metadata( vec![ @@ -2835,6 +2872,30 @@ mod tests { assert!(validate_remote_schema(&expected, &nullability_mismatch, "test").is_err()); } + #[test] + fn merge_scan_remote_schema_identity_accepts_top_level_nullability_narrowing_only() { + let state_type = |child_nullable| { + TestArrowDataType::Struct(arrow_schema::Fields::from(vec![Field::new( + "min[value]", + TestArrowDataType::Int64, + child_nullable, + )])) + }; + let schema = |nullable, child_nullable| { + ArrowSchema::new(vec![Field::new( + "__min_state(t.v)", + state_type(child_nullable), + nullable, + )]) + }; + + // `AggregateStatistics` turns the state aggregate over an empty region + // into a non-null struct literal. + assert!(validate_remote_schema(&schema(true, true), &schema(false, true), "test").is_ok()); + assert!(validate_remote_schema(&schema(false, true), &schema(true, true), "test").is_err()); + assert!(validate_remote_schema(&schema(true, true), &schema(true, false), "test").is_err()); + } + #[test] fn merge_scan_remote_schema_identity_ignores_skipping_index_field_metadata() { // Regression: a remote stream may advertise auxiliary field metadata diff --git a/tests/cases/standalone/common/aggregate/empty_region_stats.result b/tests/cases/standalone/common/aggregate/empty_region_stats.result new file mode 100644 index 00000000000..5f4a1bc59b0 --- /dev/null +++ b/tests/cases/standalone/common/aggregate/empty_region_stats.result @@ -0,0 +1,126 @@ +-- Append-mode regions report exact row counts, so the datanode can answer +-- count/min/max over a region from statistics. This only happens when every +-- aggregate in the query can be answered that way. +CREATE TABLE a1 (ts TIMESTAMP(9) TIME INDEX, k STRING, v BIGINT) +PARTITION ON COLUMNS (k) (k < 'b', k >= 'b' AND k < 'c', k >= 'c') +WITH (append_mode = 'true'); + +Affected Rows: 0 + +-- All regions are empty. +SELECT min(v), max(v) FROM a1; + ++-----------+-----------+ +| min(a1.v) | max(a1.v) | ++-----------+-----------+ +| | | ++-----------+-----------+ + +SELECT count(*), count(v) FROM a1; + ++----------+-------------+ +| count(*) | count(a1.v) | ++----------+-------------+ +| 0 | 0 | ++----------+-------------+ + +INSERT INTO a1 VALUES (1000, 'a', 1), (2000, 'a', 2); + +Affected Rows: 2 + +-- Two of three regions are empty. +SELECT min(v) FROM a1; + ++-----------+ +| min(a1.v) | ++-----------+ +| 1 | ++-----------+ + +SELECT max(v) FROM a1; + ++-----------+ +| max(a1.v) | ++-----------+ +| 2 | ++-----------+ + +SELECT min(ts), max(k) FROM a1; + ++----------------------------+-----------+ +| min(a1.ts) | max(a1.k) | ++----------------------------+-----------+ +| 1970-01-01T00:00:00.000001 | a | ++----------------------------+-----------+ + +SELECT count(*), count(v) FROM a1; + ++----------+-------------+ +| count(*) | count(a1.v) | ++----------+-------------+ +| 2 | 2 | ++----------+-------------+ + +SELECT min(v), count(*) FROM a1; + ++-----------+----------+ +| min(a1.v) | count(*) | ++-----------+----------+ +| 1 | 2 | ++-----------+----------+ + +SELECT sum(v), avg(v) FROM a1; + ++-----------+-----------+ +| sum(a1.v) | avg(a1.v) | ++-----------+-----------+ +| 3 | 1.5 | ++-----------+-----------+ + +SELECT k, min(v), max(v), count(*) FROM a1 GROUP BY k ORDER BY k; + ++---+-----------+-----------+----------+ +| k | min(a1.v) | max(a1.v) | count(*) | ++---+-----------+-----------+----------+ +| a | 1 | 2 | 2 | ++---+-----------+-----------+----------+ + +INSERT INTO a1 VALUES (3000, 'b', 3); + +Affected Rows: 1 + +-- One of three regions is empty. +SELECT min(v), max(v) FROM a1; + ++-----------+-----------+ +| min(a1.v) | max(a1.v) | ++-----------+-----------+ +| 1 | 3 | ++-----------+-----------+ + +SELECT k, min(v), max(v), count(*) FROM a1 GROUP BY k ORDER BY k; + ++---+-----------+-----------+----------+ +| k | min(a1.v) | max(a1.v) | count(*) | ++---+-----------+-----------+----------+ +| a | 1 | 2 | 2 | +| b | 3 | 3 | 1 | ++---+-----------+-----------+----------+ + +INSERT INTO a1 VALUES (4000, 'c', 4); + +Affected Rows: 1 + +-- No region is empty. +SELECT min(v), max(v), count(*) FROM a1; + ++-----------+-----------+----------+ +| min(a1.v) | max(a1.v) | count(*) | ++-----------+-----------+----------+ +| 1 | 4 | 4 | ++-----------+-----------+----------+ + +DROP TABLE a1; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/common/aggregate/empty_region_stats.sql b/tests/cases/standalone/common/aggregate/empty_region_stats.sql new file mode 100644 index 00000000000..a2d03e6b0e0 --- /dev/null +++ b/tests/cases/standalone/common/aggregate/empty_region_stats.sql @@ -0,0 +1,42 @@ +-- Append-mode regions report exact row counts, so the datanode can answer +-- count/min/max over a region from statistics. This only happens when every +-- aggregate in the query can be answered that way. +CREATE TABLE a1 (ts TIMESTAMP(9) TIME INDEX, k STRING, v BIGINT) +PARTITION ON COLUMNS (k) (k < 'b', k >= 'b' AND k < 'c', k >= 'c') +WITH (append_mode = 'true'); + +-- All regions are empty. +SELECT min(v), max(v) FROM a1; + +SELECT count(*), count(v) FROM a1; + +INSERT INTO a1 VALUES (1000, 'a', 1), (2000, 'a', 2); + +-- Two of three regions are empty. +SELECT min(v) FROM a1; + +SELECT max(v) FROM a1; + +SELECT min(ts), max(k) FROM a1; + +SELECT count(*), count(v) FROM a1; + +SELECT min(v), count(*) FROM a1; + +SELECT sum(v), avg(v) FROM a1; + +SELECT k, min(v), max(v), count(*) FROM a1 GROUP BY k ORDER BY k; + +INSERT INTO a1 VALUES (3000, 'b', 3); + +-- One of three regions is empty. +SELECT min(v), max(v) FROM a1; + +SELECT k, min(v), max(v), count(*) FROM a1 GROUP BY k ORDER BY k; + +INSERT INTO a1 VALUES (4000, 'c', 4); + +-- No region is empty. +SELECT min(v), max(v), count(*) FROM a1; + +DROP TABLE a1; diff --git a/tests/cases/standalone/common/alter/change_col_inverted_index.result b/tests/cases/standalone/common/alter/change_col_inverted_index.result index df684fec436..b1b4d456ea7 100644 --- a/tests/cases/standalone/common/alter/change_col_inverted_index.result +++ b/tests/cases/standalone/common/alter/change_col_inverted_index.result @@ -180,3 +180,145 @@ DROP TABLE test_pk; Affected Rows: 0 +-- The old SST has an inverted index, but not on host. +CREATE TABLE missing_inverted_index ( + ts TIMESTAMP TIME INDEX, + indexed_col STRING INVERTED INDEX, + host STRING +) WITH ('append_mode' = 'true'); + +Affected Rows: 0 + +INSERT INTO missing_inverted_index VALUES + (1, 'a', 'server1'), + (2, 'b', 'server2'); + +Affected Rows: 2 + +ADMIN FLUSH_TABLE('missing_inverted_index'); + ++---------------------------------------------+ +| ADMIN FLUSH_TABLE('missing_inverted_index') | ++---------------------------------------------+ +| 0 | ++---------------------------------------------+ + +SELECT * FROM missing_inverted_index WHERE host = 'server1' ORDER BY ts; + ++-------------------------+-------------+---------+ +| ts | indexed_col | host | ++-------------------------+-------------+---------+ +| 1970-01-01T00:00:00.001 | a | server1 | ++-------------------------+-------------+---------+ + +ALTER TABLE missing_inverted_index MODIFY COLUMN host SET INVERTED INDEX; + +Affected Rows: 0 + +SELECT * FROM missing_inverted_index WHERE host = 'server1' ORDER BY ts; + ++-------------------------+-------------+---------+ +| ts | indexed_col | host | ++-------------------------+-------------+---------+ +| 1970-01-01T00:00:00.001 | a | server1 | ++-------------------------+-------------+---------+ + +SELECT * FROM missing_inverted_index WHERE host = 'missing' ORDER BY ts; + +++ +++ + +SELECT * FROM missing_inverted_index WHERE indexed_col = 'a' AND host = 'server1' ORDER BY ts; + ++-------------------------+-------------+---------+ +| ts | indexed_col | host | ++-------------------------+-------------+---------+ +| 1970-01-01T00:00:00.001 | a | server1 | ++-------------------------+-------------+---------+ + +SELECT * FROM missing_inverted_index WHERE indexed_col = 'b' AND host = 'server1' ORDER BY ts; + +++ +++ + +-- New SSTs have both indexes; queries must also retain rows from the old SST. +INSERT INTO missing_inverted_index VALUES + (3, 'a', 'server1'), + (4, 'b', 'server2'); + +Affected Rows: 2 + +ADMIN FLUSH_TABLE('missing_inverted_index'); + ++---------------------------------------------+ +| ADMIN FLUSH_TABLE('missing_inverted_index') | ++---------------------------------------------+ +| 0 | ++---------------------------------------------+ + +SELECT * FROM missing_inverted_index WHERE host = 'server1' ORDER BY ts; + ++-------------------------+-------------+---------+ +| ts | indexed_col | host | ++-------------------------+-------------+---------+ +| 1970-01-01T00:00:00.001 | a | server1 | +| 1970-01-01T00:00:00.003 | a | server1 | ++-------------------------+-------------+---------+ + +SELECT * FROM missing_inverted_index WHERE indexed_col = 'a' AND host = 'server1' ORDER BY ts; + ++-------------------------+-------------+---------+ +| ts | indexed_col | host | ++-------------------------+-------------+---------+ +| 1970-01-01T00:00:00.001 | a | server1 | +| 1970-01-01T00:00:00.003 | a | server1 | ++-------------------------+-------------+---------+ + +-- SQLNESS ARG restart=true +SELECT * FROM missing_inverted_index WHERE host = 'server1' ORDER BY ts; + ++-------------------------+-------------+---------+ +| ts | indexed_col | host | ++-------------------------+-------------+---------+ +| 1970-01-01T00:00:00.001 | a | server1 | +| 1970-01-01T00:00:00.003 | a | server1 | ++-------------------------+-------------+---------+ + +-- A column absent from old SSTs can still match its default value. +ALTER TABLE missing_inverted_index ADD COLUMN source STRING DEFAULT 'legacy'; + +Affected Rows: 0 + +ALTER TABLE missing_inverted_index MODIFY COLUMN source SET INVERTED INDEX; + +Affected Rows: 0 + +SELECT * FROM missing_inverted_index WHERE source = 'legacy' ORDER BY ts; + ++-------------------------+-------------+---------+--------+ +| ts | indexed_col | host | source | ++-------------------------+-------------+---------+--------+ +| 1970-01-01T00:00:00.001 | a | server1 | legacy | +| 1970-01-01T00:00:00.002 | b | server2 | legacy | +| 1970-01-01T00:00:00.003 | a | server1 | legacy | +| 1970-01-01T00:00:00.004 | b | server2 | legacy | ++-------------------------+-------------+---------+--------+ + +SELECT * FROM missing_inverted_index WHERE source = 'missing' ORDER BY ts; + +++ +++ + +SELECT * FROM missing_inverted_index WHERE indexed_col = 'a' AND source = 'legacy' ORDER BY ts; + ++-------------------------+-------------+---------+--------+ +| ts | indexed_col | host | source | ++-------------------------+-------------+---------+--------+ +| 1970-01-01T00:00:00.001 | a | server1 | legacy | +| 1970-01-01T00:00:00.003 | a | server1 | legacy | ++-------------------------+-------------+---------+--------+ + +DROP TABLE missing_inverted_index; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/common/alter/change_col_inverted_index.sql b/tests/cases/standalone/common/alter/change_col_inverted_index.sql index aadc463d6fc..857bbb0aaf2 100644 --- a/tests/cases/standalone/common/alter/change_col_inverted_index.sql +++ b/tests/cases/standalone/common/alter/change_col_inverted_index.sql @@ -52,3 +52,48 @@ ALTER TABLE test_pk MODIFY COLUMN foo SET INVERTED INDEX; SHOW INDEX FROM test_pk; DROP TABLE test_pk; + +-- The old SST has an inverted index, but not on host. +CREATE TABLE missing_inverted_index ( + ts TIMESTAMP TIME INDEX, + indexed_col STRING INVERTED INDEX, + host STRING +) WITH ('append_mode' = 'true'); + +INSERT INTO missing_inverted_index VALUES + (1, 'a', 'server1'), + (2, 'b', 'server2'); + +ADMIN FLUSH_TABLE('missing_inverted_index'); + +SELECT * FROM missing_inverted_index WHERE host = 'server1' ORDER BY ts; + +ALTER TABLE missing_inverted_index MODIFY COLUMN host SET INVERTED INDEX; + +SELECT * FROM missing_inverted_index WHERE host = 'server1' ORDER BY ts; +SELECT * FROM missing_inverted_index WHERE host = 'missing' ORDER BY ts; +SELECT * FROM missing_inverted_index WHERE indexed_col = 'a' AND host = 'server1' ORDER BY ts; +SELECT * FROM missing_inverted_index WHERE indexed_col = 'b' AND host = 'server1' ORDER BY ts; + +-- New SSTs have both indexes; queries must also retain rows from the old SST. +INSERT INTO missing_inverted_index VALUES + (3, 'a', 'server1'), + (4, 'b', 'server2'); + +ADMIN FLUSH_TABLE('missing_inverted_index'); + +SELECT * FROM missing_inverted_index WHERE host = 'server1' ORDER BY ts; +SELECT * FROM missing_inverted_index WHERE indexed_col = 'a' AND host = 'server1' ORDER BY ts; + +-- SQLNESS ARG restart=true +SELECT * FROM missing_inverted_index WHERE host = 'server1' ORDER BY ts; + +-- A column absent from old SSTs can still match its default value. +ALTER TABLE missing_inverted_index ADD COLUMN source STRING DEFAULT 'legacy'; +ALTER TABLE missing_inverted_index MODIFY COLUMN source SET INVERTED INDEX; + +SELECT * FROM missing_inverted_index WHERE source = 'legacy' ORDER BY ts; +SELECT * FROM missing_inverted_index WHERE source = 'missing' ORDER BY ts; +SELECT * FROM missing_inverted_index WHERE indexed_col = 'a' AND source = 'legacy' ORDER BY ts; + +DROP TABLE missing_inverted_index;