rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-21 11:22:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: backport clean must-have fixes to release/5e02efa9 (#9772)

Browse Source 
* fix(mito2): preserve effective sequences during compaction (#9147)

* fix(mito2): preserve effective row sequences during compaction

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(mito2): inherit compaction input sequence bounds

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

---------

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>
(cherry picked from commit e0552eee2d)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(meta-client): bound heartbeat handshake timeout (#9438)

* fix(meta-client): bound heartbeat handshake timeout

Signed-off-by: Aetherance <inkToAether@gmail.com>

* chore(meta-client): fix test license header

Signed-off-by: Aetherance <inkToAether@gmail.com>

* test(meta-client): prefix heartbeat tests with test

Signed-off-by: Aetherance <inkToAether@gmail.com>

---------

Signed-off-by: Aetherance <inkToAether@gmail.com>
(cherry picked from commit 9c69e381ae)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* 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 7605f27608)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(mito2): ignore missing inverted indexes in old SSTs (#9545)

Signed-off-by: WenyXu <wenymedia@gmail.com>
(cherry picked from commit 287ced2de2)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix: prevent stale leader metadata after concurrent writes (#9549)

fix: fence leader cache fills around store mutations

Signed-off-by: jeremyhi <fengjiachun@gmail.com>
(cherry picked from commit 394562aa9e)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* 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 <killme2008@gmail.com>
(cherry picked from commit ef1298ce02)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* 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 <ratuthomm@gmail.com>

---------

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>
Signed-off-by: Aetherance <inkToAether@gmail.com>
Signed-off-by: lczllx <2181719471@qq.com>
Signed-off-by: WenyXu <wenymedia@gmail.com>
Signed-off-by: jeremyhi <fengjiachun@gmail.com>
Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
Co-authored-by: Aether <98370028+Aetherance@users.noreply.github.com>
Co-authored-by: Dragon Roar <2181719471@qq.com>
Co-authored-by: Weny Xu <wenymedia@gmail.com>
Co-authored-by: jeremyhi <jiachun_feng@proton.me>
Co-authored-by: dennis zhuang <killme2008@gmail.com>
This commit is contained in:
Lei, HUANG
2026-10-20 19:27:22 +08:00
committed by GitHub
co-authored by Aether Dragon Roar Weny Xu jeremyhi dennis zhuang
parent ef3c9d7ccc
commit 508d1a1a33
20 changed files with 2073 additions and 169 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/meta-client/src/client/heartbeat.rs
+19 -10
View File
@@ -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);
src/meta-client/src/error.rs
+11
View File
@@ -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, .. }
src/meta-client/tests/heartbeat_handshake_timeout.rs
+140
View File
@@ -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<Box<dyn Stream<Item = Result<HeartbeatResponse, Status>> + 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<Streaming<HeartbeatRequest>>,
) -> Result<Response<Self::HeartbeatStream>, 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::<Result<HeartbeatResponse, Status>>().await
}
});
Ok(Response::new(Box::pin(stream)))
}
async fn ask_leader(
&self,
_request: Request<AskLeaderRequest>,
) -> Result<Response<AskLeaderResponse>, 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);
}
src/meta-srv/src/handler.rs
+156 -37
View File
@@ -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<std::result::Result<HeartbeatResponse, tonic::Status>>,
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<DeregisterSignalReceiver> {
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::<Vec<_>>();
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::<Vec<_>>()
};
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<std::result::Result<HeartbeatResponse, tonic::Status>> {
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());
}
}
src/meta-srv/src/service/store/cached_kv.rs
+359 -2
View File
@@ -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::<Vec<_>>();
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<Option<oneshot::Receiver<()>>>,
}
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<Error>,
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<TxnResponse> {
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<RangeResponse> {
let res = self.inner.range(req).await;
self.after_read.wait().await;
res
}
async fn batch_get(&self, req: BatchGetRequest) -> Result<BatchGetResponse> {
let res = self.inner.batch_get(req).await;
self.after_read.wait().await;
res
}
async fn put(&self, req: PutRequest) -> Result<PutResponse> {
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<BatchPutResponse> {
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<DeleteRangeResponse> {
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<BatchDeleteResponse> {
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<KeyValue> {
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)
src/mito2/src/access_layer.rs
+2
View File
@@ -573,6 +573,8 @@ pub struct SstWriteRequest {
pub cache_manager: CacheManagerRef,
#[allow(dead_code)]
pub storage: Option<String>,
/// 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<SequenceNumber>,
pub sst_write_format: FormatType,
src/mito2/src/compaction/compactor.rs
+296 -41
View File
@@ -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<NonZeroU64>,
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<NonZeroU64> {
let mut max: Option<NonZeroU64> = 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::<Vec<_>>();
@@ -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<u64>, 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(&region, 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::<TimestampMillisecondType>()
.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 {
src/mito2/src/engine/compaction_test.rs
+290 -1
View File
@@ -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<i64> {
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<u64> {
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::<datatypes::arrow::datatypes::UInt64Type>()
.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 = <rand::rngs::StdRng as rand::SeedableRng>::seed_from_u64(0);
for _ in 0..3000 {
deletes.extend(build_rows_for_key(
&format!("{:032x}", rand::Rng::random::<u128>(&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::<Vec<_>>();
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::<HashSet<_>>();
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::<Vec<i64>>();
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::<datatypes::arrow::datatypes::UInt64Type>();
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::<Vec<_>>();
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<Arc<CompactionListener>>);
impl CompactionListenerGuard {
src/mito2/src/engine/merge_mode_test.rs
+189
View File
@@ -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 = <rand::rngs::StdRng as rand::SeedableRng>::seed_from_u64(0);
for _ in 0..3000 {
b.extend(build_rows_with_fields(
&format!("{:032x}", rand::Rng::random::<u128>(&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::<UInt64Type>()
.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;
src/mito2/src/engine/scan_test.rs
+103 -50
View File
@@ -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::<datatypes::arrow::array::UInt64Array>()
.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::<datatypes::arrow::array::UInt64Array>()
.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(
src/mito2/src/read/scan_region.rs
+4 -3
View File
@@ -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
src/mito2/src/region/options.rs
+2 -1
View File
@@ -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<ReadableSize>,
/// 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
src/mito2/src/sst/file.rs
+8 -3
View File
@@ -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<NonZeroU64>,
/// Partition expression from the region metadata when the file is created.
///
@@ -284,6 +287,8 @@ pub struct FileMeta {
pub primary_key_max: Option<Bytes>,
/// 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,
}
src/mito2/src/sst/index/inverted_index/applier.rs
+3 -2
View File
@@ -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;
src/mito2/src/sst/index/inverted_index/creator.rs
+62 -6
View File
@@ -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<ColumnId>,
) -> impl Fn(DfExpr) -> BoxFuture<'static, Vec<usize>> {
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;
src/query/src/dist_plan/merge_scan.rs
+74 -13
View File
@@ -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
tests/cases/standalone/common/aggregate/empty_region_stats.result
+126
View File
@@ -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
tests/cases/standalone/common/aggregate/empty_region_stats.sql
+42
View File
@@ -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;
tests/cases/standalone/common/alter/change_col_inverted_index.result
+142
View File
@@ -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
tests/cases/standalone/common/alter/change_col_inverted_index.sql
+45
View File
@@ -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;