From 9281965bacf6c7ef867a508199643d186f6005b5 Mon Sep 17 00:00:00 2001 From: discord9 Date: Wed, 23 Sep 2026 04:20:02 +0000 Subject: [PATCH 1/4] feat(flow): allow execution hook to rewrite completed plans (#9290) * refactor(flow): expose incremental aggregate plan analysis Signed-off-by: discord9 * feat(flow): allow execution hook to rewrite completed plans Signed-off-by: discord9 * test(flow): cover the execution hook receiving the completed plan The hook must see the plan that is dispatched after the incremental delta-sink merge, so the test records the plan a collaborator is handed and asserts the merge is already part of it. Signed-off-by: discord9 --------- Signed-off-by: discord9 --- .../src/batching_mode/batching_execution.rs | 6 ++ src/flow/src/batching_mode/task.rs | 4 + src/flow/src/batching_mode/task/test.rs | 93 +++++++++++++++++++ src/flow/src/lib.rs | 5 +- 4 files changed, 107 insertions(+), 1 deletion(-) diff --git a/src/flow/src/batching_mode/batching_execution.rs b/src/flow/src/batching_mode/batching_execution.rs index 0a699a227be..298522b8137 100644 --- a/src/flow/src/batching_mode/batching_execution.rs +++ b/src/flow/src/batching_mode/batching_execution.rs @@ -16,6 +16,7 @@ use std::sync::Arc; +use datafusion_expr::LogicalPlan; use query::QueryEngineRef; use table::TableRef; @@ -37,6 +38,11 @@ pub trait BatchingExecution: Send + Sync + 'static { max_window_cnt: Option, ) -> ExecuteOnceOutcome; + /// Rewrite the completed query plan after incremental merging and before execution. + fn rewrite_plan(&self, _task: &BatchingTask, plan: LogicalPlan) -> Result { + Ok(plan) + } + /// Retire this execution instance, rejecting new work and requesting that any /// retained local work stop. This is not an acknowledgement of remote quiescence. fn stop(&self) {} diff --git a/src/flow/src/batching_mode/task.rs b/src/flow/src/batching_mode/task.rs index c0ee6d1c1b5..346d39d1d46 100644 --- a/src/flow/src/batching_mode/task.rs +++ b/src/flow/src/batching_mode/task.rs @@ -828,6 +828,10 @@ impl BatchingTask { return Ok(None); } let plan = incremental_plan.unwrap_or_else(|| plan.clone()); + let plan = match &self.execution { + Some(execution) => execution.rewrite_plan(self, plan)?, + None => plan, + }; let extensions = self .build_flow_query_extensions(incremental_safe, coverage.is_incremental_delta()) diff --git a/src/flow/src/batching_mode/task/test.rs b/src/flow/src/batching_mode/task/test.rs index 7a56c70823d..7add59664a5 100644 --- a/src/flow/src/batching_mode/task/test.rs +++ b/src/flow/src/batching_mode/task/test.rs @@ -143,6 +143,99 @@ impl crate::BatchingExecution for RetainingExecution { } } +/// Records the completed plan handed to it by the task. +#[derive(Default)] +struct PlanRecordingExecution { + plans: std::sync::Mutex>, +} + +#[async_trait::async_trait] +impl crate::BatchingExecution for PlanRecordingExecution { + async fn execute_once( + self: Arc, + _guard: BatchingExecutionGuard, + _task: &BatchingTask, + _engine: &QueryEngineRef, + _frontend: &Arc, + _max_window_cnt: Option, + ) -> ExecuteOnceOutcome { + ExecuteOnceOutcome { + new_query: None, + result: Ok(None), + } + } + + fn rewrite_plan(&self, _task: &BatchingTask, plan: LogicalPlan) -> crate::Result { + self.plans.lock().unwrap().push(plan.clone()); + Ok(plan) + } +} + +#[tokio::test] +async fn test_execution_hook_receives_the_completed_incremental_plan() { + let sink_table = "hook_sink"; + let query = "SELECT max(number) AS number, ts FROM numbers_with_ts GROUP BY ts"; + let TestTaskParts { + task, query_engine, .. + } = new_test_task_engine_and_plan_with_query(query, sink_table).await; + register_auto_created_aggregate_sink(&query_engine, sink_table); + + let ctx = task.state.read().unwrap().query_ctx.clone(); + let plan = sql_to_df_plan(ctx, query_engine.clone(), query, true) + .await + .unwrap(); + let (sink, _) = get_table_info_df_schema( + query_engine.engine_state().catalog_manager().clone(), + [ + "greptime".to_string(), + "public".to_string(), + sink_table.to_string(), + ], + ) + .await + .unwrap(); + let table_provider = Arc::new(DfTableProviderAdapter::new(sink)); + let table_source = Arc::new(DefaultTableSource::new(table_provider)); + let dml_plan = LogicalPlan::Dml(DmlStatement::new( + datafusion_common::TableReference::bare("test"), + table_source, + WriteOp::Insert(datafusion_expr::dml::InsertOp::Append), + Arc::new(plan), + )); + + task.state + .write() + .unwrap() + .advance_checkpoints(HashMap::from([(1_u64, 10_u64)])); + + let execution = Arc::new(PlanRecordingExecution::default()); + let task = task.with_execution(Some(execution.clone())); + let (frontend_client, _handler) = + FrontendClient::from_empty_grpc_handler(QueryOptions::default()); + let frontend = Arc::new(frontend_client); + let dirty_restore = DirtyRestore::Unscoped(dirty_range(10, 15)); + + let _ = task + .execute_plan_unlocked( + &query_engine, + &frontend, + &dml_plan, + &dirty_restore, + &QueryCoverage::IncrementalDelta, + ) + .await; + + let recorded = execution.plans.lock().unwrap(); + let plan = recorded + .first() + .expect("the execution hook must see the plan that is dispatched"); + let plan_text = plan.to_string(); + assert!( + plan_text.contains("Left Join"), + "the hook must receive the completed delta-sink merge plan, got:\n{plan_text}" + ); +} + #[tokio::test] async fn test_execution_delegate_dispatch_is_serialized() { let TestTaskParts { diff --git a/src/flow/src/lib.rs b/src/flow/src/lib.rs index e9ef90e09d6..36dc7c21818 100644 --- a/src/flow/src/lib.rs +++ b/src/flow/src/lib.rs @@ -48,7 +48,10 @@ pub use batching_mode::task::{ QueryCoverage, TaskArgs, }; pub use batching_mode::time_window::{TimeWindowExpr, find_time_window_expr}; -pub use batching_mode::utils::sql_to_df_plan; +pub use batching_mode::utils::{ + IncrementalAggregateAnalysis, IncrementalAggregateMergeColumn, IncrementalAggregateMergeOp, + analyze_incremental_aggregate_plan, sql_to_df_plan, +}; pub(crate) use engine::{CreateFlowArgs, FlowId, TableName}; pub use error::{Error, Result}; pub use server::{FlownodeBuilder, FlownodeInstance, FlownodeServer, FlownodeServiceBuilder}; From 953d01ac546cf13937b8df8b6770fe9395dd11de Mon Sep 17 00:00:00 2001 From: Weny Xu Date: Wed, 23 Sep 2026 04:41:13 +0000 Subject: [PATCH 2/4] feat: support pending rows batching for MySQL and PostgreSQL (#9302) * feat: support pending rows batching for MySQL and PostgreSQL Signed-off-by: WenyXu * style: group batcher imports before item definitions Signed-off-by: WenyXu * fix: use 65536 as the default batcher worker channel capacity Signed-off-by: WenyXu * test: use a distinct custom worker channel capacity Signed-off-by: WenyXu * test: complete Prom config in worker capacity override case Signed-off-by: WenyXu --------- Signed-off-by: WenyXu --- config/config.md | 4 +- config/frontend.example.toml | 6 +- config/standalone.example.toml | 6 +- src/frontend/src/frontend.rs | 11 +- src/frontend/src/instance/builder.rs | 2 +- src/frontend/src/instance/logical_batcher.rs | 2 +- src/frontend/src/server.rs | 62 ++++-- .../service_config/pending_rows_batcher.rs | 44 ++++- src/frontend/src/service_config/prom_store.rs | 6 +- src/servers/src/batcher.rs | 39 ++++ src/servers/src/http.rs | 35 +--- src/servers/src/mysql/handler.rs | 31 ++- src/servers/src/mysql/server.rs | 11 +- src/servers/src/postgres.rs | 9 +- src/servers/src/postgres/handler.rs | 12 +- src/servers/src/postgres/server.rs | 12 +- src/servers/tests/http/prom_store_test.rs | 3 +- src/session/src/context.rs | 2 +- src/standalone/src/options.rs | 11 +- tests-integration/src/otlp.rs | 3 +- tests-integration/src/standalone.rs | 11 +- tests-integration/tests/sql.rs | 181 ++++++++++++++++++ 22 files changed, 406 insertions(+), 97 deletions(-) diff --git a/config/config.md b/config/config.md index d736c24d848..c33918a8573 100644 --- a/config/config.md +++ b/config/config.md @@ -75,7 +75,7 @@ | `influxdb` | -- | -- | InfluxDB protocol options. | | `influxdb.enable` | Bool | `true` | Whether to enable InfluxDB protocol in HTTP API. | | `influxdb.default_merge_mode` | String | `last_non_null` | Default merge mode for tables automatically created by InfluxDB protocol.
Available values: "last_non_null", "last_row". | -| `pending_rows_batcher` | -- | -- | Ordinary-table batching for opted-in HTTP ingestion protocols.
PENDING_ROWS_BATCH_SYNC defaults to true for both batchers. Set it to false to acknowledge
queue admission without waiting for storage; later failures cannot be returned to the client.
Omitted or empty protocols disables batching. Prom without metric engine uses this batcher.
OTLP logs, traces and ordinary metrics use this batcher. | +| `pending_rows_batcher` | -- | -- | Ordinary-table batching for opted-in ingestion protocols.
PENDING_ROWS_BATCH_SYNC defaults to true for both batchers. Set it to false to acknowledge
queue admission without waiting for storage; later failures cannot be returned to the client.
Omitted or empty protocols disables batching. Prom without metric engine uses this batcher.
OTLP logs, traces and ordinary metrics use this batcher.
MySQL and PostgreSQL use the same acknowledgement policy. Protocol timeouts are unchanged.
Single-connection writes and INSERT SELECT may incur additional flush waits. | | `pending_rows_batcher.pending_rows_flush_interval` | String | `0s` | Flush interval measured from the first pending submission. Zero disables batching. | | `pending_rows_batcher.max_batch_rows` | Integer | `100000` | Flush after a complete submission reaches this row threshold. | | `pending_rows_batcher.max_concurrent_flushes` | Integer | `256` | Maximum concurrent flushes shared by the frontend batcher. | @@ -343,7 +343,7 @@ | `influxdb` | -- | -- | InfluxDB protocol options. | | `influxdb.enable` | Bool | `true` | Whether to enable InfluxDB protocol in HTTP API. | | `influxdb.default_merge_mode` | String | `last_non_null` | Default merge mode for tables automatically created by InfluxDB protocol.
Available values: "last_non_null", "last_row". | -| `pending_rows_batcher` | -- | -- | Ordinary-table batching for opted-in HTTP ingestion protocols.
PENDING_ROWS_BATCH_SYNC defaults to true for both batchers. Set it to false to acknowledge
queue admission without waiting for storage; later failures cannot be returned to the client.
Omitted or empty protocols disables batching. Prom without metric engine uses this batcher.
OTLP logs, traces and ordinary metrics use this batcher. | +| `pending_rows_batcher` | -- | -- | Ordinary-table batching for opted-in ingestion protocols.
PENDING_ROWS_BATCH_SYNC defaults to true for both batchers. Set it to false to acknowledge
queue admission without waiting for storage; later failures cannot be returned to the client.
Omitted or empty protocols disables batching. Prom without metric engine uses this batcher.
OTLP logs, traces and ordinary metrics use this batcher.
MySQL and PostgreSQL use the same acknowledgement policy. Protocol timeouts are unchanged.
Single-connection writes and INSERT SELECT may incur additional flush waits. | | `pending_rows_batcher.pending_rows_flush_interval` | String | `0s` | Flush interval measured from the first pending submission. Zero disables batching. | | `pending_rows_batcher.max_batch_rows` | Integer | `100000` | Flush after a complete submission reaches this row threshold. | | `pending_rows_batcher.max_concurrent_flushes` | Integer | `256` | Maximum concurrent flushes shared by the frontend batcher. | diff --git a/config/frontend.example.toml b/config/frontend.example.toml index 275db3105c5..055acbb93bd 100644 --- a/config/frontend.example.toml +++ b/config/frontend.example.toml @@ -231,11 +231,13 @@ enable = true ## Available values: "last_non_null", "last_row". default_merge_mode = "last_non_null" -## Ordinary-table batching for opted-in HTTP ingestion protocols. +## Ordinary-table batching for opted-in ingestion protocols. ## PENDING_ROWS_BATCH_SYNC defaults to true for both batchers. Set it to false to acknowledge ## queue admission without waiting for storage; later failures cannot be returned to the client. ## Omitted or empty protocols disables batching. Prom without metric engine uses this batcher. ## OTLP logs, traces and ordinary metrics use this batcher. +## MySQL and PostgreSQL use the same acknowledgement policy. Protocol timeouts are unchanged. +## Single-connection writes and INSERT SELECT may incur additional flush waits. [pending_rows_batcher] # protocols = [ # "influxdb", @@ -246,6 +248,8 @@ default_merge_mode = "last_non_null" # "splunk", # "elasticsearch", # "http_sql", +# "mysql", +# "postgres", # "prom", # ] ## Flush interval measured from the first pending submission. Zero disables batching. diff --git a/config/standalone.example.toml b/config/standalone.example.toml index 257f48eda2c..357ec5904f7 100644 --- a/config/standalone.example.toml +++ b/config/standalone.example.toml @@ -210,11 +210,13 @@ enable = true ## Available values: "last_non_null", "last_row". default_merge_mode = "last_non_null" -## Ordinary-table batching for opted-in HTTP ingestion protocols. +## Ordinary-table batching for opted-in ingestion protocols. ## PENDING_ROWS_BATCH_SYNC defaults to true for both batchers. Set it to false to acknowledge ## queue admission without waiting for storage; later failures cannot be returned to the client. ## Omitted or empty protocols disables batching. Prom without metric engine uses this batcher. ## OTLP logs, traces and ordinary metrics use this batcher. +## MySQL and PostgreSQL use the same acknowledgement policy. Protocol timeouts are unchanged. +## Single-connection writes and INSERT SELECT may incur additional flush waits. [pending_rows_batcher] # protocols = [ # "influxdb", @@ -225,6 +227,8 @@ default_merge_mode = "last_non_null" # "splunk", # "elasticsearch", # "http_sql", +# "mysql", +# "postgres", # "prom", # ] ## Flush interval measured from the first pending submission. Zero disables batching. diff --git a/src/frontend/src/frontend.rs b/src/frontend/src/frontend.rs index 72bf322b1f1..51e291c7906 100644 --- a/src/frontend/src/frontend.rs +++ b/src/frontend/src/frontend.rs @@ -206,6 +206,7 @@ mod tests { use futures::Stream; use meta_client::MetaClientRef; use meta_client::client::MetaClientBuilder; + use servers::batcher::BatchingProtocol; use servers::grpc::{FlightCompression, GRPC_SERVER}; use servers::http::HTTP_SERVER; use servers::http::result::greptime_result_v1::GreptimedbV1Response; @@ -249,10 +250,7 @@ mod tests { .logical_table .unwrap() .protocols, - vec![ - servers::http::BatchingProtocol::Prom, - servers::http::BatchingProtocol::Otlp - ] + vec![BatchingProtocol::Prom, BatchingProtocol::Otlp] ); }, ); @@ -270,10 +268,7 @@ mod tests { FrontendOptions::load_layered_options(None, "FRONTEND_BATCHER_TEST").unwrap(); assert_eq!( options.pending_rows_batcher.table.protocols, - vec![ - servers::http::BatchingProtocol::Influxdb, - servers::http::BatchingProtocol::HttpSql - ] + vec![BatchingProtocol::Influxdb, BatchingProtocol::HttpSql] ); }, ); diff --git a/src/frontend/src/instance/builder.rs b/src/frontend/src/instance/builder.rs index d9b86ac5a05..d261daed48e 100644 --- a/src/frontend/src/instance/builder.rs +++ b/src/frontend/src/instance/builder.rs @@ -50,8 +50,8 @@ use partition::manager::PartitionRuleManager; use pipeline::pipeline_operator::PipelineOperator; use query::QueryEngineFactory; use query::region_query::RegionQueryHandlerFactoryRef; +use servers::batcher::BatchingProtocol; use servers::batcher::table::TablePendingRowsBatcher; -use servers::http::BatchingProtocol; use snafu::{OptionExt, ResultExt}; use crate::error::{self, DataFusionSnafu, ExternalSnafu, Result}; diff --git a/src/frontend/src/instance/logical_batcher.rs b/src/frontend/src/instance/logical_batcher.rs index 28ed19b3e14..e68dcf644d4 100644 --- a/src/frontend/src/instance/logical_batcher.rs +++ b/src/frontend/src/instance/logical_batcher.rs @@ -16,9 +16,9 @@ use std::sync::{Arc, Weak}; use api::v1::ColumnSchema; use async_trait::async_trait; +use servers::batcher::BatchingProtocol; use servers::batcher::logical_table::{LogicalTablePendingRowsBatcher, PendingRowsSchemaAlterer}; use servers::error::{BatcherChannelClosedSnafu, Result}; -use servers::http::BatchingProtocol; use session::context::QueryContextRef; use snafu::OptionExt; diff --git a/src/frontend/src/server.rs b/src/frontend/src/server.rs index 0a6612ba31c..fb1ab2f2603 100644 --- a/src/frontend/src/server.rs +++ b/src/frontend/src/server.rs @@ -24,7 +24,7 @@ use common_base::Plugins; use common_config::Configurable; use common_telemetry::{info, warn}; use meta_client::MetaClientOptions; -use servers::batcher::pending_rows_batch_sync_enabled; +use servers::batcher::{BatchingProtocol, pending_rows_batch_sync_enabled}; use servers::error::Error as ServerError; use servers::grpc::builder::GrpcServerBuilder; use servers::grpc::flight::FlightCraftRef; @@ -34,7 +34,7 @@ use servers::grpc::{GrpcOptions, GrpcServer}; use servers::http::event::LogValidatorRef; use servers::http::result::error_result::ErrorResponse; use servers::http::utils::router::RouterConfigurator; -use servers::http::{BatchingProtocol, HttpOptions, HttpServer, HttpServerBuilder}; +use servers::http::{HttpOptions, HttpServer, HttpServerBuilder}; use servers::interceptor::LogIngestInterceptorRef; use servers::metrics_handler::MetricsHandler; use servers::mysql::server::{MysqlServer, MysqlSpawnConfig, MysqlSpawnRef}; @@ -369,6 +369,15 @@ where } } + let table_batcher = opts.table_batcher_options(); + let batching_enabled = table_batcher.pending_rows_batching_enabled(); + let mysql_batching = + batching_enabled && table_batcher.protocols.contains(&BatchingProtocol::Mysql); + let postgres_batching = batching_enabled + && table_batcher + .protocols + .contains(&BatchingProtocol::Postgres); + if opts.mysql.enable { // Init MySQL server let opts = &opts.mysql; @@ -384,13 +393,16 @@ where let mysql_server = MysqlServer::create_server( common_runtime::global_runtime(), Arc::new(MysqlSpawnRef::new(instance.clone(), user_provider.clone())), - Arc::new(MysqlSpawnConfig::new( - opts.tls.should_force_tls(), - tls_server_config, - opts.keep_alive.as_secs(), - opts.reject_no_database.unwrap_or(false), - opts.prepared_stmt_cache_size, - )), + Arc::new( + MysqlSpawnConfig::new( + opts.tls.should_force_tls(), + tls_server_config, + opts.keep_alive.as_secs(), + opts.reject_no_database.unwrap_or(false), + opts.prepared_stmt_cache_size, + ) + .with_batching_enabled(mysql_batching), + ), Some(instance.process_manager().clone()), ); handlers.insert((mysql_server, mysql_addr)); @@ -407,15 +419,18 @@ where maybe_watch_server_tls_config(tls_server_config.clone()).context(StartServerSnafu)?; - let pg_server = Box::new(PostgresServer::new( - instance.clone(), - opts.tls.should_force_tls(), - tls_server_config, - opts.keep_alive.as_secs(), - common_runtime::global_runtime(), - user_provider.clone(), - Some(self.instance.process_manager().clone()), - )) as Box; + let pg_server = Box::new( + PostgresServer::new( + instance.clone(), + opts.tls.should_force_tls(), + tls_server_config, + opts.keep_alive.as_secs(), + common_runtime::global_runtime(), + user_provider.clone(), + Some(self.instance.process_manager().clone()), + ) + .with_batching_enabled(postgres_batching), + ) as Box; handlers.insert((pg_server, pg_addr)); } @@ -452,8 +467,11 @@ fn effective_http_options_with_sync(opts: &FrontendOptions, batch_sync: bool) -> let common_enabled = batch_sync && shared.pending_rows_batching_enabled() && shared.protocols.iter().any(|protocol| { - *protocol != BatchingProtocol::Prom - || (prom_store.enable && !prom_store.with_metric_engine) + !matches!( + protocol, + BatchingProtocol::Mysql | BatchingProtocol::Postgres + ) && (*protocol != BatchingProtocol::Prom + || (prom_store.enable && !prom_store.with_metric_engine)) }); let common_interval = common_enabled.then_some(shared.pending_rows_flush_interval); let prom_interval = (prom_store.pending_rows_batching_enabled() && batch_sync) @@ -655,6 +673,10 @@ mod tests { expected_secs, ) in [ (vec![BatchingProtocol::Prom], false, false, 5, 2, 1, 1), + (vec![BatchingProtocol::Mysql], false, false, 5, 0, 1, 1), + (vec![BatchingProtocol::Postgres], false, false, 5, 0, 1, 1), + (vec![BatchingProtocol::Mysql], false, true, 5, 0, 1, 1), + (vec![BatchingProtocol::Postgres], false, true, 5, 0, 1, 1), (vec![BatchingProtocol::Prom], true, false, 5, 2, 1, 1), (vec![BatchingProtocol::Prom], true, true, 5, 2, 1, 6), (vec![BatchingProtocol::Influxdb], true, false, 5, 2, 1, 1), diff --git a/src/frontend/src/service_config/pending_rows_batcher.rs b/src/frontend/src/service_config/pending_rows_batcher.rs index 61572015d7c..b671ee9a272 100644 --- a/src/frontend/src/service_config/pending_rows_batcher.rs +++ b/src/frontend/src/service_config/pending_rows_batcher.rs @@ -17,7 +17,7 @@ use std::time::Duration; use common_batcher::flush_policy::timing::TimingFlushPolicy; use serde::{Deserialize, Serialize}; -use servers::http::BatchingProtocol; +use servers::batcher::BatchingProtocol; use tokio::sync::Semaphore; use crate::frontend::FrontendOptions; @@ -38,11 +38,11 @@ pub struct PendingRowsBatcherOptions { pub logical_table: Option, } -/// Write batching controls shared by HTTP ingestion protocols. +/// Write batching controls shared by ingestion protocols. #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] #[serde(default)] pub struct BatcherOptions { - /// HTTP write protocols sharing this batcher; empty disables all entrances. + /// Write protocols sharing this batcher; empty disables all entrances. pub protocols: Vec, /// Time from the first pending submission to a timed flush. Zero disables batching. #[serde(with = "humantime_serde")] @@ -238,6 +238,8 @@ mod tests { "opentsdb", "elasticsearch", "splunk", + "mysql", + "postgres", ] { assert!( toml::from_str::(&format!( @@ -251,9 +253,11 @@ mod tests { #[test] fn test_protocols() { let options: BatcherOptions = toml::from_str( - "protocols = ['influxdb', 'opentsdb', 'otlp', 'logs', 'loki', 'splunk', 'elasticsearch', 'http_sql', 'prom']", + "protocols = ['influxdb', 'opentsdb', 'otlp', 'logs', 'loki', 'splunk', 'elasticsearch', 'http_sql', 'prom', 'mysql', 'postgres']", ).unwrap(); - assert_eq!(options.protocols.len(), 9); + assert_eq!(options.protocols.len(), 11); + assert!(options.protocols.contains(&BatchingProtocol::Mysql)); + assert!(options.protocols.contains(&BatchingProtocol::Postgres)); assert!(options.protocols.contains(&BatchingProtocol::HttpSql)); assert!(BatcherOptions::default().protocols.is_empty()); for invalid in ["sql", "jaeger", "unknown"] { @@ -289,6 +293,36 @@ mod tests { ); } + #[test] + fn test_worker_capacity_override() { + let options: FrontendOptions = toml::from_str( + r#" +[pending_rows_batcher] +worker_channel_capacity = 12345 +[pending_rows_batcher.logical_table] +worker_channel_capacity = 12345 +[prom_store] +enable = true +with_metric_engine = true +worker_channel_capacity = 12345 +"#, + ) + .unwrap(); + assert_eq!( + options.pending_rows_batcher.table.worker_channel_capacity, + 12_345 + ); + assert_eq!( + options + .pending_rows_batcher + .logical_table + .unwrap() + .worker_channel_capacity, + 12_345 + ); + assert_eq!(options.prom_store.worker_channel_capacity, 12_345); + } + #[test] fn test_partial_options_and_zero_controls() { let options: BatcherOptions = diff --git a/src/frontend/src/service_config/prom_store.rs b/src/frontend/src/service_config/prom_store.rs index 2fdd05043db..3d00f48cdda 100644 --- a/src/frontend/src/service_config/prom_store.rs +++ b/src/frontend/src/service_config/prom_store.rs @@ -108,7 +108,6 @@ mod tests { use crate::service_config::prom_store::{ default_flow_notification_queue_capacity, default_max_batch_rows, default_max_concurrent_flushes, default_max_inflight_requests, - default_worker_channel_capacity, }; #[test] @@ -160,10 +159,7 @@ mod tests { default.max_concurrent_flushes, default_max_concurrent_flushes() ); - assert_eq!( - default.worker_channel_capacity, - default_worker_channel_capacity() - ); + assert_eq!(default.worker_channel_capacity, 65_536); assert_eq!( default.max_inflight_requests, default_max_inflight_requests() diff --git a/src/servers/src/batcher.rs b/src/servers/src/batcher.rs index 6b6ca8f7ee7..3cd9a831aaa 100644 --- a/src/servers/src/batcher.rs +++ b/src/servers/src/batcher.rs @@ -22,6 +22,8 @@ pub mod table; #[cfg(test)] mod test_util; +use serde::{Deserialize, Serialize}; + /// Controls whether batching waits for storage before replying to the client. const PENDING_ROWS_BATCH_SYNC_ENV: &str = "PENDING_ROWS_BATCH_SYNC"; @@ -39,3 +41,40 @@ pub fn pending_rows_batch_sync_enabled() -> bool { .and_then(|v| v.parse::().ok()) .unwrap_or(true) } + +/// Ingestion protocols that can opt into pending-row batching. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum BatchingProtocol { + Prom, + Influxdb, + Opentsdb, + Otlp, + Logs, + Loki, + Splunk, + Elasticsearch, + HttpSql, + Mysql, + Postgres, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_protocol_names_reject_unknown_values() { + assert_eq!( + serde_json::from_str::("\"prom\"").unwrap(), + BatchingProtocol::Prom + ); + for name in ["sql", "unknown"] { + assert!(serde_json::from_str::(&format!("\"{name}\"")).is_err()); + } + assert_eq!( + serde_json::from_str::("\"http_sql\"").unwrap(), + BatchingProtocol::HttpSql + ); + } +} diff --git a/src/servers/src/http.rs b/src/servers/src/http.rs index 4d168111ade..f947ec1621b 100644 --- a/src/servers/src/http.rs +++ b/src/servers/src/http.rs @@ -53,6 +53,7 @@ use tower_http::trace::TraceLayer; use self::authorize::AuthState; use self::result::table_result::TableResponse; +use crate::batcher::BatchingProtocol; use crate::batcher::logical_table::LogicalTablePendingRowsBatcher; use crate::elasticsearch; use crate::error::{ @@ -189,22 +190,6 @@ pub(crate) enum HttpServerKind { Api, } -/// HTTP write protocols eligible for the shared pending-row batcher. -/// Prometheus uses this selector only when metric-engine storage is disabled. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum BatchingProtocol { - Prom, - Influxdb, - Opentsdb, - Otlp, - Logs, - Loki, - Splunk, - Elasticsearch, - HttpSql, -} - #[derive(Default)] pub struct HttpServer { router: StdMutex, @@ -2478,9 +2463,10 @@ mod batching_tests { use common_query::Output; use session::context::QueryContextRef; + use crate::batcher::BatchingProtocol; use crate::error::Result as ServerResult; use crate::http::test_helpers::TestClient; - use crate::http::{BatchingProtocol, HttpOptions, HttpServerBuilder}; + use crate::http::{HttpOptions, HttpServerBuilder}; use crate::influxdb::InfluxdbRequest; use crate::opentsdb::codec::DataPoint; use crate::query_handler::{InfluxdbLineProtocolHandler, OpentsdbProtocolHandler}; @@ -2523,21 +2509,6 @@ mod batching_tests { } } - #[test] - fn test_protocol_names_reject_unknown_values() { - assert_eq!( - serde_json::from_str::("\"prom\"").unwrap(), - BatchingProtocol::Prom - ); - for name in ["sql", "unknown"] { - assert!(serde_json::from_str::(&format!("\"{name}\"")).is_err()); - } - assert_eq!( - serde_json::from_str::("\"http_sql\"").unwrap(), - BatchingProtocol::HttpSql - ); - } - #[derive(Default)] struct RecordingWriteHandler { selections: Mutex>, diff --git a/src/servers/src/mysql/handler.rs b/src/servers/src/mysql/handler.rs index 3764591beaa..d0dd1f25279 100644 --- a/src/servers/src/mysql/handler.rs +++ b/src/servers/src/mysql/handler.rs @@ -85,6 +85,7 @@ pub struct MysqlInstanceShim { prepared_stmts_counter: AtomicU32, process_id: u32, prepared_stmt_cache_size: usize, + batching_enabled: bool, } impl MysqlInstanceShim { @@ -122,9 +123,22 @@ impl MysqlInstanceShim { prepared_stmts_counter: AtomicU32::new(1), process_id, prepared_stmt_cache_size, + batching_enabled: false, } } + /// Enables ordinary-table batching for this connection. + pub fn with_batching_enabled(mut self, enabled: bool) -> Self { + self.batching_enabled = enabled; + self + } + + fn new_query_context(&self) -> QueryContextRef { + let mut ctx = self.session.new_query_context(); + Arc::make_mut(&mut ctx).set_batching_enabled(self.batching_enabled); + ctx + } + #[tracing::instrument(skip_all, name = "mysql::do_query")] async fn do_query(&self, query: &str, query_ctx: QueryContextRef) -> Vec> { if let Some(output) = @@ -495,7 +509,7 @@ impl AsyncMysqlShim for MysqlInstanceShi raw_query: &'a str, w: StatementMetaWriter<'a, W>, ) -> Result<()> { - let query_ctx = self.session.new_query_context(); + let query_ctx = self.new_query_context(); let stmt_id = self.prepared_stmts_counter.fetch_add(1, Ordering::Relaxed); let stmt_key = uuid::Uuid::from_u128(stmt_id as u128).to_string(); let (params, columns) = match self @@ -525,7 +539,7 @@ impl AsyncMysqlShim for MysqlInstanceShi ) -> Result<()> { self.session.clear_warnings(); - let query_ctx = self.session.new_query_context(); + let query_ctx = self.new_query_context(); let db = query_ctx.get_db_string(); let _timer = crate::metrics::METRIC_MYSQL_QUERY_TIMER .with_label_values(&[crate::metrics::METRIC_MYSQL_BINQUERY, db.as_str()]) @@ -569,7 +583,7 @@ impl AsyncMysqlShim for MysqlInstanceShi query: &'a str, writer: QueryResultWriter<'a, W>, ) -> Result<()> { - let query_ctx = self.session.new_query_context(); + let query_ctx = self.new_query_context(); let db = query_ctx.get_db_string(); let _timer = crate::metrics::METRIC_MYSQL_QUERY_TIMER .with_label_values(&[crate::metrics::METRIC_MYSQL_TEXTQUERY, db.as_str()]) @@ -1039,6 +1053,17 @@ mod tests { ) } + #[test] + fn test_batching_context() { + for enabled in [false, true] { + let shim = create_shim().with_batching_enabled(enabled); + let ctx = shim.new_query_context(); + assert_eq!(ctx.batching_enabled(), enabled); + assert!(!ctx.logical_batching_enabled()); + assert_eq!(ctx.channel(), Channel::Mysql); + } + } + fn statement_with_transformed_placeholders(query: &str) -> Statement { let mut statements = ParserContext::create_with_dialect(query, &MySqlDialect {}, ParseOptions::default()) diff --git a/src/servers/src/mysql/server.rs b/src/servers/src/mysql/server.rs index 0f48cb0d598..811fc9827ec 100644 --- a/src/servers/src/mysql/server.rs +++ b/src/servers/src/mysql/server.rs @@ -86,6 +86,7 @@ pub struct MysqlSpawnConfig { reject_no_database: bool, // prepared statement cache capacity prepared_stmt_cache_size: usize, + batching_enabled: bool, } impl MysqlSpawnConfig { @@ -102,9 +103,16 @@ impl MysqlSpawnConfig { keep_alive_secs, reject_no_database, prepared_stmt_cache_size, + batching_enabled: false, } } + /// Enables ordinary-table batching for connections accepted by this server. + pub fn with_batching_enabled(mut self, enabled: bool) -> Self { + self.batching_enabled = enabled; + self + } + fn tls(&self) -> Option> { self.tls.get_config() } @@ -213,7 +221,8 @@ impl MysqlServer { stream.peer_addr()?, process_id, spawn_config.prepared_stmt_cache_size, - ); + ) + .with_batching_enabled(spawn_config.batching_enabled); let (mut r, w) = stream.into_split(); let mut w = BufWriter::with_capacity(DEFAULT_RESULT_SET_WRITE_BUFFER_SIZE, w); diff --git a/src/servers/src/postgres.rs b/src/servers/src/postgres.rs index 58ef4fdd7b7..1cd5907157b 100644 --- a/src/servers/src/postgres.rs +++ b/src/servers/src/postgres.rs @@ -80,6 +80,7 @@ pub struct PostgresServerHandlerInner { force_tls: bool, param_provider: Arc, + batching_enabled: bool, session: Arc, query_parser: Arc, } @@ -114,7 +115,12 @@ impl PgWireServerHandlers for PostgresServerHandler { } impl MakePostgresServerHandler { - fn make(&self, addr: Option, process_id: u32) -> PostgresServerHandler { + fn make( + &self, + addr: Option, + process_id: u32, + batching_enabled: bool, + ) -> PostgresServerHandler { let session = Arc::new(Session::new( addr, Channel::Postgres, @@ -127,6 +133,7 @@ impl MakePostgresServerHandler { force_tls: self.force_tls, param_provider: self.param_provider.clone(), + batching_enabled, session: session.clone(), query_parser: Arc::new(DefaultQueryParser::new(self.query_handler.clone(), session)), }; diff --git a/src/servers/src/postgres/handler.rs b/src/servers/src/postgres/handler.rs index e618c55614b..7bf1edca705 100644 --- a/src/servers/src/postgres/handler.rs +++ b/src/servers/src/postgres/handler.rs @@ -59,6 +59,14 @@ use crate::postgres::utils::convert_err; use crate::postgres::{PostgresServerHandlerInner, fixtures}; use crate::query_handler::sql::ServerSqlQueryHandlerRef; +impl PostgresServerHandlerInner { + fn new_query_context(&self) -> QueryContextRef { + let mut ctx = self.session.new_query_context(); + Arc::make_mut(&mut ctx).set_batching_enabled(self.batching_enabled); + ctx + } +} + #[async_trait] impl SimpleQueryHandler for PostgresServerHandlerInner { #[tracing::instrument(skip_all, fields(protocol = "postgres"))] @@ -68,7 +76,7 @@ impl SimpleQueryHandler for PostgresServerHandlerInner { C::Error: Debug, PgWireError: From<>::Error>, { - let query_ctx = self.session.new_query_context(); + let query_ctx = self.new_query_context(); let db = query_ctx.get_db_string(); let _timer = crate::metrics::METRIC_POSTGRES_QUERY_TIMER .with_label_values(&[crate::metrics::METRIC_POSTGRES_SIMPLE_QUERY, db.as_str()]) @@ -425,7 +433,7 @@ impl ExtendedQueryHandler for PostgresServerHandlerInner { C::Error: Debug, PgWireError: From<>::Error>, { - let query_ctx = self.session.new_query_context(); + let query_ctx = self.new_query_context(); let db = query_ctx.get_db_string(); let _timer = crate::metrics::METRIC_POSTGRES_QUERY_TIMER .with_label_values(&[crate::metrics::METRIC_POSTGRES_EXTENDED_QUERY, db.as_str()]) diff --git a/src/servers/src/postgres/server.rs b/src/servers/src/postgres/server.rs index 09bc5015a0b..6e4aa942f9d 100644 --- a/src/servers/src/postgres/server.rs +++ b/src/servers/src/postgres/server.rs @@ -38,6 +38,7 @@ pub struct PostgresServer { make_handler: Arc, tls_server_config: Arc, keep_alive_secs: u64, + batching_enabled: bool, bind_addr: Option, process_manager: Option, } @@ -66,17 +67,25 @@ impl PostgresServer { make_handler, tls_server_config, keep_alive_secs, + batching_enabled: false, bind_addr: None, process_manager, } } + /// Enables ordinary-table batching for connections accepted by this server. + pub fn with_batching_enabled(mut self, enabled: bool) -> Self { + self.batching_enabled = enabled; + self + } + fn accept( &self, io_runtime: Runtime, accepting_stream: AbortableStream, ) -> impl Future + use<> { let handler_maker = self.make_handler.clone(); + let batching_enabled = self.batching_enabled; let tls_server_config = self.tls_server_config.clone(); let process_manager = self.process_manager.clone(); accepting_stream.for_each(move |tcp_stream| { @@ -102,7 +111,8 @@ impl PostgresServer { let _handle = io_runtime.spawn(async move { crate::metrics::METRIC_POSTGRES_CONNECTIONS.inc(); - let pg_handler = Arc::new(handler_maker.make(addr, process_id)); + let pg_handler = + Arc::new(handler_maker.make(addr, process_id, batching_enabled)); let r = process_socket(io_stream, tls_acceptor.clone(), pg_handler).await; crate::metrics::METRIC_POSTGRES_CONNECTIONS.dec(); diff --git a/src/servers/tests/http/prom_store_test.rs b/src/servers/tests/http/prom_store_test.rs index 7c07e3b0344..ffd2c9dcf7f 100644 --- a/src/servers/tests/http/prom_store_test.rs +++ b/src/servers/tests/http/prom_store_test.rs @@ -36,11 +36,12 @@ use datafusion_expr::LogicalPlan; use prost::Message; use query::parser::PromQuery; use query::query_engine::DescribeResult; +use servers::batcher::BatchingProtocol; use servers::error::{self, Result}; use servers::http::header::{CONTENT_ENCODING_SNAPPY, CONTENT_TYPE_PROTOBUF}; use servers::http::prom_store::PHYSICAL_TABLE_PARAM; use servers::http::test_helpers::{TestClient, TestResponse}; -use servers::http::{BatchingProtocol, HttpOptions, HttpServerBuilder}; +use servers::http::{HttpOptions, HttpServerBuilder}; use servers::prom_remote_write::v2::test_util as remote_write_v2; use servers::prom_remote_write::validation::PromValidationMode; use servers::prom_store; diff --git a/src/session/src/context.rs b/src/session/src/context.rs index 5ebf8b36d21..ec10dc6cf3a 100644 --- a/src/session/src/context.rs +++ b/src/session/src/context.rs @@ -471,7 +471,7 @@ impl QueryContext { self.logical_batching_enabled = enabled; } - /// Whether the local HTTP entry point selected ordinary-table batching. + /// Whether the local protocol entry point selected ordinary-table batching. pub fn batching_enabled(&self) -> bool { self.batching_enabled } diff --git a/src/standalone/src/options.rs b/src/standalone/src/options.rs index d0de7408789..fbfcc2ccc1d 100644 --- a/src/standalone/src/options.rs +++ b/src/standalone/src/options.rs @@ -217,6 +217,7 @@ mod tests { use std::sync::Arc; use common_event_recorder::EventTypeFilter; + use servers::batcher::BatchingProtocol; use crate::options::*; @@ -266,10 +267,7 @@ mod tests { .logical_table .unwrap() .protocols, - vec![ - servers::http::BatchingProtocol::Prom, - servers::http::BatchingProtocol::Otlp - ] + vec![BatchingProtocol::Prom, BatchingProtocol::Otlp] ); }, ); @@ -288,10 +286,7 @@ mod tests { .unwrap(); assert_eq!( options.pending_rows_batcher.table.protocols, - vec![ - servers::http::BatchingProtocol::Influxdb, - servers::http::BatchingProtocol::HttpSql - ] + vec![BatchingProtocol::Influxdb, BatchingProtocol::HttpSql] ); }, ); diff --git a/tests-integration/src/otlp.rs b/tests-integration/src/otlp.rs index 338b03ff68f..9fbf80fbeea 100644 --- a/tests-integration/src/otlp.rs +++ b/tests-integration/src/otlp.rs @@ -514,8 +514,7 @@ WITH( ExponentialHistogram, ExponentialHistogramDataPoint, exponential_histogram_data_point, }; use prost::Message; - use servers::batcher::pending_rows_batch_sync_enabled; - use servers::http::BatchingProtocol; + use servers::batcher::{BatchingProtocol, pending_rows_batch_sync_enabled}; use servers::http::test_helpers::TestClient; use session::protocol_ctx::{OtlpMetricCtx, ProtocolCtx}; diff --git a/tests-integration/src/standalone.rs b/tests-integration/src/standalone.rs index 38f546e9ccb..6fc8ce231af 100644 --- a/tests-integration/src/standalone.rs +++ b/tests-integration/src/standalone.rs @@ -90,6 +90,7 @@ pub struct GreptimeDbStandaloneBuilder { auto_create_table: bool, experimental_metric_export: bool, logical_batcher: Option, + table_batcher: BatcherOptions, } impl GreptimeDbStandaloneBuilder { @@ -111,6 +112,7 @@ impl GreptimeDbStandaloneBuilder { auto_create_table: true, experimental_metric_export: false, logical_batcher: None, + table_batcher: BatcherOptions::default(), } } @@ -121,6 +123,13 @@ impl GreptimeDbStandaloneBuilder { self } + /// Configures ordinary-table batching for protocol integration tests. + #[must_use] + pub fn with_table_batcher(mut self, options: BatcherOptions) -> Self { + self.table_batcher = options; + self + } + /// Configures logical-table batching for integration tests. #[must_use] pub fn with_logical_batcher(mut self, options: BatcherOptions) -> Self { @@ -389,7 +398,7 @@ impl GreptimeDbStandaloneBuilder { experimental_metric_export: self.experimental_metric_export, pending_rows_batcher: PendingRowsBatcherOptions { logical_table: self.logical_batcher.clone(), - ..Default::default() + table: self.table_batcher.clone(), }, // Tests cover the descriptor, so they run with it enabled. otlp: frontend::service_config::OtlpOptions { diff --git a/tests-integration/tests/sql.rs b/tests-integration/tests/sql.rs index e3b66fc4447..cb3d151fa85 100644 --- a/tests-integration/tests/sql.rs +++ b/tests-integration/tests/sql.rs @@ -2672,3 +2672,184 @@ pub async fn test_declare_fetch_close_cursor(store_type: StorageType) { let _ = fe_pg_server.shutdown().await; guard.remove_all().await; } + +/// Keeps SQL and input rows fixed while changing only the selected batching protocols. +#[tokio::test(flavor = "multi_thread")] +async fn test_sql_batcher_alignment() { + use common_telemetry::{dump_metrics, info, init_default_ut_logging}; + use frontend::server::Services; + use frontend::service_config::BatcherOptions; + use servers::batcher::BatchingProtocol; + use servers::mysql::server::MYSQL_SERVER; + use tests_integration::standalone::GreptimeDbStandaloneBuilder; + + fn flushes() -> u64 { + dump_metrics() + .unwrap() + .lines() + .find_map(|line| { + line.strip_prefix("greptime_table_batcher_flush_total ") + .map(|value| value.parse().unwrap()) + }) + .unwrap_or(0) + } + + init_default_ut_logging(); + for protocols in [ + vec![], + vec![BatchingProtocol::Mysql], + vec![BatchingProtocol::Postgres], + vec![BatchingProtocol::Mysql, BatchingProtocol::Postgres], + ] { + let mysql_enabled = protocols.contains(&BatchingProtocol::Mysql); + let pg_enabled = protocols.contains(&BatchingProtocol::Postgres); + let mut instance = GreptimeDbStandaloneBuilder::new("sql_batcher_alignment") + .with_table_batcher(BatcherOptions { + protocols, + pending_rows_flush_interval: Duration::from_millis(10), + ..Default::default() + }) + .build() + .await; + let mut opts = instance.opts.clone(); + opts.http.addr = "127.0.0.1:0".into(); + opts.grpc.bind_addr = "127.0.0.1:0".into(); + opts.mysql.addr = "127.0.0.1:0".into(); + opts.postgres.addr = "127.0.0.1:0".into(); + let mut servers = Services::new(opts, instance.fe_instance().clone(), Default::default()) + .build() + .unwrap(); + servers.start_all().await.unwrap(); + let mysql_url = format!("mysql://{}/public", servers.addr(MYSQL_SERVER).unwrap()); + let mut mysql = MySqlConnection::connect(&mysql_url).await.unwrap(); + let (pg, connection) = tokio_postgres::connect( + &format!( + "postgres://{}/public", + servers.addr("POSTGRES_SERVER").unwrap() + ), + NoTls, + ) + .await + .unwrap(); + let pg_task = tokio::spawn(async move { connection.await.unwrap() }); + + for round in 0..2 { + mysql.execute("CREATE TABLE batch_sql (ts TIMESTAMP TIME INDEX, host STRING PRIMARY KEY, val INT DEFAULT 7)").await.unwrap(); + let started = std::time::Instant::now(); + let before = flushes(); + // COM_QUERY and COM_STMT_EXECUTE must both opt in. + assert_eq!( + mysql + .execute("INSERT INTO batch_sql (ts, host) VALUES (0, 'a')") + .await + .unwrap() + .rows_affected(), + 1 + ); + assert_eq!( + sqlx::query("INSERT INTO batch_sql VALUES (0, ?, ?)") + .persistent(false) + .bind("b") + .bind(8_i32) + .execute(&mut mysql) + .await + .unwrap() + .rows_affected(), + 1 + ); + assert_eq!(flushes() - before, if mysql_enabled { 2 } else { 0 }); + + let before = flushes(); + let result = pg + .simple_query("INSERT INTO batch_sql (ts, host) VALUES (0, 'c')") + .await + .unwrap(); + assert!(matches!( + result.as_slice(), + [SimpleQueryMessage::CommandComplete(1)] + )); + let statement = pg + .prepare("INSERT INTO batch_sql VALUES (0, $1, $2)") + .await + .unwrap(); + assert_eq!(pg.execute(&statement, &[&"d", &9_i32]).await.unwrap(), 1); + assert_eq!(flushes() - before, if pg_enabled { 2 } else { 0 }); + + // Independent connections can submit to the same worker concurrently. + let (mysql_result, pg_result) = tokio::join!( + mysql.execute("INSERT INTO batch_sql VALUES (0, 'e', 10)"), + pg.simple_query("INSERT INTO batch_sql VALUES (0, 'f', 11)") + ); + assert_eq!(mysql_result.unwrap().rows_affected(), 1); + assert!(matches!( + pg_result.unwrap().as_slice(), + [SimpleQueryMessage::CommandComplete(1)] + )); + + // Errors must reach the client without affecting later requests. + assert!( + mysql + .execute("INSERT INTO batch_sql (missing) VALUES (1)") + .await + .is_err() + ); + assert!( + pg.simple_query("INSERT INTO batch_sql (missing) VALUES (1)") + .await + .is_err() + ); + let expected = vec![ + ("a".to_string(), 7_i32), + ("b".to_string(), 8), + ("c".to_string(), 7), + ("d".to_string(), 9), + ("e".to_string(), 10), + ("f".to_string(), 11), + ]; + // No polling: successful protocol responses guarantee visibility. + let rows: Vec<(String, i32)> = + sqlx::query_as("SELECT host, val FROM batch_sql ORDER BY host") + .persistent(false) + .fetch_all(&mut mysql) + .await + .unwrap(); + assert_eq!(rows, expected); + let rows: Vec<(String, i32)> = pg + .query("SELECT host, val FROM batch_sql ORDER BY host", &[]) + .await + .unwrap() + .iter() + .map(|row| (row.get(0), row.get(1))) + .collect(); + assert_eq!(rows, expected); + + mysql.execute("CREATE TABLE batch_copy (ts TIMESTAMP TIME INDEX, host STRING PRIMARY KEY, val INT)").await.unwrap(); + assert_eq!( + mysql + .execute("INSERT INTO batch_copy SELECT * FROM batch_sql") + .await + .unwrap() + .rows_affected(), + 6 + ); + let rows: Vec<(String, i32)> = + sqlx::query_as("SELECT host, val FROM batch_copy ORDER BY host") + .persistent(false) + .fetch_all(&mut mysql) + .await + .unwrap(); + assert_eq!(rows, expected); + info!( + "sql_batcher_alignment mysql={mysql_enabled} postgres={pg_enabled} round={round} elapsed={:?}", + started.elapsed() + ); + mysql.execute("DROP TABLE batch_sql").await.unwrap(); + mysql.execute("DROP TABLE batch_copy").await.unwrap(); + } + mysql.close().await.unwrap(); + drop(pg); + pg_task.await.unwrap(); + servers.shutdown_all().await.unwrap(); + instance.guard.remove_all().await; + } +} From 1d8d95c12a61f33bf4116c1f3de940ad4eab2b87 Mon Sep 17 00:00:00 2001 From: Ning Sun Date: Wed, 23 Sep 2026 04:56:44 +0000 Subject: [PATCH 3/4] chore(deps): replace cargo-udeps with cargo-shear for unused dependency checks (#9294) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * chore(deps): replace cargo-udeps with cargo-shear for unused dependency checks cargo-udeps requires a nightly toolchain and its pinned version (0.1.61) no longer detects unused dependencies against current cargo internals — unused deps have landed on main undetected (e.g. humantime in common-frontend since #6689). cargo-shear is a standalone static analyzer that runs on any toolchain. - Swap 'make check-udeps' / 'make fix-udeps' recipes to 'cargo shear' / 'cargo shear --fix' and retire scripts/fix-udeps.py - CI: install cargo-shear in the check-udeps job; drop the build cache and protoc steps (cargo-shear never compiles) - Remove ~150 unused dependency declarations found by cargo-shear, move misplaced deps to the correct sections, drop orphaned [workspace.dependencies] entries (arrow-cast, rustc-hash) - Add [package.metadata.cargo-shear] ignored entries with explanations for dependencies that are structurally required despite no textual reference: sqlparser (required by sqlparser_derive expansions in datatypes, common-query), common-error (required by common-macro's stack_trace_debug expansions in session, tests-fuzz), k8s-openapi (feature-pinning for the transitive kube dependency in tests-fuzz), tikv-jemalloc-sys (link-only, enables jemalloc profiling features in common-mem-prof), protobuf (required by build.rs-generated bindings in log-store) - Drop the obsolete [package.metadata.cargo-udeps.ignore] sections Part of #9289 Signed-off-by: Ning Sun * fix(meta): populate physical metric table column ids (#9286) * fix(meta): populate physical metric table column ids Signed-off-by: dhruvxvaishnav * test(meta): verify physical metric column ids Signed-off-by: dhruvxvaishnav --------- Signed-off-by: dhruvxvaishnav * fix(postgres): return empty responses for comment-only SQL (#9295) fix(postgres): handle parsed empty queries in both protocols Signed-off-by: houyuwushang <180804215+houyuwushang@users.noreply.github.com> * ci: create docs follow-up issue on PR merge instead of on label (#9237) * ci: create docs follow-up issue on PR merge instead of on label The docbot workflow previously created a docs-repo issue as soon as the 'docs-required' condition was detected (PR opened/edited with the docs checkbox ticked), even if the PR was never merged. Now the workflow also triggers on PR 'closed': - opened/edited: only manage the docs-required/docs-not-required labels - closed: create the docs issue only when the PR was actually merged and carries the docs-required label This also lets maintainers control issue creation by manually adding or removing the docs-required label before merging. Signed-off-by: Ning Sun * fix: address review comments on docs issue creation timing - Only touch docs labels when the docs checkbox state actually changed in an edit. Previously, editing any other part of the PR body while the checkbox stayed checked removed the docs-required label, silently dropping the docs follow-up now that issue creation happens at merge. Unchanged checkbox now leaves labels untouched, which also preserves manual label overrides. - Do not trust the closed event's stale label snapshot at merge time: re-read the live PR via the API and create the docs issue if the docs-required label is present OR the checkbox is ticked in the current body. - Make the workflow concurrency group action-aware so a merge run does not cancel an in-flight label update from an edit run. Signed-off-by: Ning Sun * fix: make docs-required label the single source of truth at merge The label-OR-checkbox merge condition could not distinguish an intentional opt-out from an unfinished label update: removing docs-required while the checkbox stayed checked still produced an issue, and unchecking the box could still produce one if the merge read the stale label before the edit run removed it. At merge time, wait for any pending docbot runs on the PR head SHA to finish their label updates (bounded to 5 minutes), then decide solely by the live docs-required label. Adds actions: read permission for listing workflow runs. Signed-off-by: Ning Sun --------- Signed-off-by: Ning Sun * perf(promql): push label filters into grouped join inputs (#9280) * perf(promql): propagate matching filters through grouped joins Signed-off-by: Dennis Zhuang * perf(promql): check matcher safety on the receiving operand Signed-off-by: Dennis Zhuang * refactor(promql): spell out the shapes a filter may cross `preserves_filter` ended in `_ => true`, which was only sound because `selector_matchers` independently rejects label rewriting, `count_values`, subqueries and non-rollup calls on the same operand. Loosening the latter alone would have silently pushed a matcher below a label rewrite. List the shapes that carry a scan filter instead and default to `false`. Cite #9207 for the result labels the grouped cases record: the join projects the right operand's tag set, so `zone` is missing wherever the right side aggregates it away. No behavior change. Signed-off-by: Dennis Zhuang * test(promql): assert the new pushdowns reach the scan The grouped-join unit tests feed tag columns by hand and the SQLness case only checks results, which are identical whether or not the rewrite fires. Nothing would have failed if scalar arithmetic, ranking or grouped matching stopped propagating. Assert through the planner that the matcher reaches both scans, with a global topk one-side as the counter-example. Also state that the duplicate-one-side cases record a cross product Prometheus rejects (#9209), so the baseline is not read as intended semantics. Signed-off-by: Dennis Zhuang --------- Signed-off-by: Dennis Zhuang * fix(ci): build tests-integration lib with meta-srv/mock (#9299) * fix(ci): build tests-integration lib with meta-srv/mock tests-integration's lib code (src/cluster.rs) uses meta_srv::mocks, but the dependency carrying the mock feature sits in [dev-dependencies]. Builds that only touch the lib, such as the apidoc job's cargo doc --workspace, resolve meta-srv without mock and fail with E0432. --all-targets builds unify dev-dependency features, which is why check, clippy and nextest stayed green. Move the mock-enabled meta-srv entry back to [dependencies]. The other testing features moved out in #9072 are not needed by the lib and stay in [dev-dependencies]. Signed-off-by: Dennis Zhuang * test(repartition): split per-case repartition tests test_repartition_metric ran four format/primary-key-encoding cases in a single test function, and test_repartition_mito ran two format cases. Each case builds its own 3-datanode cluster and runs a full repartition plus GC cycle, so on S3 the metric test took 165-178s against the 180s nextest terminate-after. Merge queue runs failed on it at random. Split each case into its own test. Cases were already independent, so they now run in parallel and each stays far inside the timeout, and a failure points at one encoding instead of four. Signed-off-by: Dennis Zhuang --------- Signed-off-by: Dennis Zhuang * feat(json2): support altering JSON2 settings (#9029) * feat(sql): support alter syntax for JSON2 columns Signed-off-by: fys * fix(json2): preserve rows on type hint mismatch during compaction * refactor(json2): simplify alter settings handling * fix(json2): preserve coerced values during compaction * chore: remove unnecessary clone * chor: reduce memory allocations * fix: cargo clippy * chore: update greptime-proto to main branch * refactor(datatypes): unify string handling with other JSON type hints * fix: cr --------- Signed-off-by: fys * fix: keep compaction pruning, metadata, and index work on compact runtime (#9304) * fix: run compaction pruner tasks on compact runtime Signed-off-by: Lei, HUANG * fix: keep compaction metadata and index work on compact runtime Signed-off-by: Lei, HUANG --------- Signed-off-by: Lei, HUANG * feat: add AI matching, classification, and scoring functions (#9300) * feat: return matching scores from jev Replace the experimental three-argument Boolean function with jev(text, prompt) returning a Float64 probability in [0, 1]. Move threshold comparisons into SQL and update tests and migration examples. Signed-off-by: Lei, HUANG * feat: add Jev choice and score functions Share asynchronous execution across Noul, Choice, and Score. Validate JSON criteria before requests and return typed scalar answers. Add SQL and HTTP mock coverage with usage examples. Signed-off-by: Lei, HUANG * refactor: use generic AI SQL function names Expose ai_match, ai_choose, and ai_score and move their implementation, tests, and usage guide under generic AI names. Document the current unreleased interface without migration history. Signed-off-by: Lei, HUANG * fix: share constant AI criteria within each batch Borrow scalar string arguments and lazily parse constant criteria once per batch. Share the parsed allocation across requests while preserving NULL propagation and batch validation before HTTP calls. Signed-off-by: Lei, HUANG * feat: preserve AI score uncertainty in JSONB results Return score, confidence, and probabilities in criteria-level order from one evaluation. Validate the distribution and preserve provider precision. Add JSON extraction, uncertainty, and single-request regressions, and document confidence-aware ranking. Signed-off-by: Lei, HUANG * docs: explain reuse of volatile AI evaluations Document repeated SELECT and WHERE evaluation costs as N + M requests, and show subquery aliases for reusing scalar or structured AI results without additional model calls. Signed-off-by: Lei, HUANG --------- Signed-off-by: Lei, HUANG * feat: share logical table batching with OTLP metrics (#9288) * feat: share logical table batching with OTLP metrics Signed-off-by: WenyXu * fix: unify pending rows batch acknowledgement policy Signed-off-by: WenyXu * fix: align logical batcher example configuration expectations Signed-off-by: WenyXu * fix: align batcher worker channel defaults to 65536 Signed-off-by: WenyXu --------- Signed-off-by: WenyXu * perf(mito2): lazily decode dense primary key columns (#9226) * perf(mito2): lazily decode dense primary key columns Signed-off-by: Lei, HUANG * perf(mito2): bypass lazy decoding for full primary keys Signed-off-by: Lei, HUANG * fix(mito-codec): preserve prefix decoding errors Signed-off-by: Lei, HUANG * refactor(mito-codec): align encoded length helper naming Rename encoded_length to encoded_len and update all callers to match the other length helpers in the module. Signed-off-by: Lei, HUANG * refactor(mito2): clarify conditional dense key decoding Rename decode_dense_pk to ensure_dense_pk_decoded so callers can see that existing decoded values are preserved and only missing caches are populated. Signed-off-by: Lei, HUANG * refactor(mito-codec): share string framing in row converter Move encoded_string_len to the parent module so Dense and Sparse use the same framing helper without depending on each other. Preserve its implementation and visibility. Signed-off-by: Lei, HUANG --------- Signed-off-by: Lei, HUANG * chore: sync lock * fix: shear and check issues --------- Signed-off-by: Ning Sun Signed-off-by: dhruvxvaishnav Signed-off-by: houyuwushang <180804215+houyuwushang@users.noreply.github.com> Signed-off-by: Dennis Zhuang Signed-off-by: fys Signed-off-by: Lei, HUANG Signed-off-by: WenyXu Co-authored-by: Dhruv Vaishnav Co-authored-by: houyuwushang <180804215+houyuwushang@users.noreply.github.com> Co-authored-by: dennis zhuang Co-authored-by: fys <40801205+fengys1996@users.noreply.github.com> Co-authored-by: Lei, HUANG <6406592+v0y4g3r@users.noreply.github.com> Co-authored-by: Weny Xu --- .github/workflows/rust.yml | 15 +- .gitignore | 2 - Cargo.lock | 130 +------------ Cargo.toml | 2 - Makefile | 7 +- flake.nix | 2 +- scripts/fix-udeps.py | 265 --------------------------- src/catalog/Cargo.toml | 2 - src/cli/Cargo.toml | 8 - src/client/Cargo.toml | 4 - src/cmd/Cargo.toml | 10 +- src/common/config/Cargo.toml | 1 - src/common/frontend/Cargo.toml | 5 - src/common/function/Cargo.toml | 5 - src/common/grpc-expr/Cargo.toml | 4 - src/common/mem-prof/Cargo.toml | 6 + src/common/meta/Cargo.toml | 2 - src/common/procedure-test/Cargo.toml | 1 - src/common/query/Cargo.toml | 5 + src/common/runtime/Cargo.toml | 2 - src/common/sql/Cargo.toml | 1 - src/common/stat/Cargo.toml | 1 - src/common/substrait/Cargo.toml | 5 - src/datanode/Cargo.toml | 3 - src/datatypes/Cargo.toml | 5 + src/file-engine/Cargo.toml | 1 - src/flow/Cargo.toml | 17 -- src/frontend/Cargo.toml | 6 - src/log-store/Cargo.toml | 8 +- src/meta-client/Cargo.toml | 4 - src/meta-srv/Cargo.toml | 10 - src/metric-engine/Cargo.toml | 12 +- src/mito2/Cargo.toml | 20 +- src/object-store/Cargo.toml | 13 +- src/object-store/src/lib.rs | 1 + src/operator/Cargo.toml | 2 - src/partition/Cargo.toml | 4 - src/pipeline/Cargo.toml | 9 - src/query/Cargo.toml | 7 - src/servers/Cargo.toml | 4 +- src/session/Cargo.toml | 6 + src/sql/Cargo.toml | 7 - src/store-api/Cargo.toml | 5 - src/table/Cargo.toml | 3 - tests-fuzz/Cargo.toml | 9 + tests-integration/Cargo.toml | 39 ++-- tests/runner/Cargo.toml | 1 - 47 files changed, 79 insertions(+), 602 deletions(-) delete mode 100755 scripts/fix-udeps.py diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index e4483b4a42b..1326ab5fd3d 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -24,7 +24,7 @@ on: name: Rust CI env: - CARGO_UDEPS_VERSION: "0.1.61" + CARGO_SHEAR_VERSION: "1.13.4" concurrency: group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }} @@ -84,20 +84,11 @@ jobs: - uses: actions/checkout@v4 with: persist-credentials: false - - uses: arduino/setup-protoc@v3 - with: - repo-token: ${{ secrets.GITHUB_TOKEN }} - uses: actions-rust-lang/setup-rust-toolchain@v1 - - name: Rust Cache - uses: Swatinem/rust-cache@v2 - with: - shared-key: "check-udeps" - cache-all-crates: "true" - save-if: ${{ github.ref == 'refs/heads/main' }} - - name: Install cargo-udeps + - name: Install cargo-shear uses: taiki-e/install-action@v2 with: - tool: cargo-udeps@${{ env.CARGO_UDEPS_VERSION }} + tool: cargo-shear@${{ env.CARGO_SHEAR_VERSION }} fallback: none - name: Check unused dependencies run: make check-udeps diff --git a/.gitignore b/.gitignore index 90893f8aac1..dd60d96224e 100644 --- a/.gitignore +++ b/.gitignore @@ -52,8 +52,6 @@ venv/ tests-fuzz/artifacts/ tests-fuzz/corpus/ -# cargo-udeps reports -udeps-report.json # Nix .direnv diff --git a/Cargo.lock b/Cargo.lock index ea55f1b297d..6c7916d8194 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1673,7 +1673,6 @@ dependencies = [ "async-trait", "bytes", "cache", - "catalog", "chrono", "common-base", "common-catalog", @@ -1701,7 +1700,6 @@ dependencies = [ "lazy_static", "meta-client", "moka", - "object-store", "partition", "paste", "prometheus 0.14.0", @@ -2006,10 +2004,8 @@ version = "1.3.0-alpha.1" dependencies = [ "async-stream", "async-trait", - "auth", "base64 0.22.1", "cache", - "catalog", "chrono", "clap", "client", @@ -2022,13 +2018,10 @@ dependencies = [ "common-macro", "common-meta", "common-procedure", - "common-query", - "common-recordbatch", "common-runtime", "common-telemetry", "common-test-util", "common-time", - "common-version", "common-wal", "datatypes", "etcd-client", @@ -2048,7 +2041,6 @@ dependencies = [ "serde", "serde_json", "servers", - "session", "snafu 0.8.6", "sql", "standalone", @@ -2056,7 +2048,6 @@ dependencies = [ "table", "tempfile", "tokio", - "tracing-appender", "url", "uuid", ] @@ -2074,7 +2065,6 @@ dependencies = [ "common-catalog", "common-error", "common-grpc", - "common-grpc-expr", "common-macro", "common-meta", "common-query", @@ -2100,7 +2090,6 @@ dependencies = [ "tokio", "tokio-stream", "tonic 0.14.2", - "tracing", ] [[package]] @@ -2150,10 +2139,8 @@ dependencies = [ "api", "async-trait", "auth", - "base64 0.22.1", "cache", "catalog", - "chrono", "clap", "cli", "client", @@ -2170,7 +2157,6 @@ dependencies = [ "common-options", "common-procedure", "common-query", - "common-recordbatch", "common-runtime", "common-stat", "common-telemetry", @@ -2184,7 +2170,6 @@ dependencies = [ "datafusion-physical-plan", "datanode", "datatypes", - "either", "etcd-client", "file-engine", "flow", @@ -2192,7 +2177,6 @@ dependencies = [ "futures", "hex", "human-panic", - "humantime", "lazy_static", "log-store", "meta-client", @@ -2216,20 +2200,17 @@ dependencies = [ "serde", "serde_json", "servers", - "session", "similar-asserts", "smallvec", "snafu 0.8.6", "sqlparser", "standalone", "store-api", - "table", "temp-env", "tempfile", "tikv-jemallocator", "tokio", "toml 0.8.23", - "tonic 0.14.2", "tracing-appender", ] @@ -2355,7 +2336,6 @@ dependencies = [ "serde_json", "snafu 0.8.6", "temp-env", - "tempfile", "toml 0.8.23", ] @@ -2456,14 +2436,11 @@ dependencies = [ "common-grpc", "common-macro", "common-meta", - "common-time", "greptime-proto", - "humantime", "meta-client", "serde", "serde_json", "snafu 0.8.6", - "tokio", "tonic 0.14.2", ] @@ -2474,22 +2451,18 @@ dependencies = [ "ahash 0.8.12", "api", "approx 0.5.1", - "arc-swap", "arrow 59.2.0", - "arrow-cast 59.2.0", "arrow-schema 59.2.0", "async-trait", "axum 0.8.4", "bincode", "catalog", - "chrono", "common-base", "common-catalog", "common-error", "common-macro", "common-meta", "common-query", - "common-runtime", "common-telemetry", "common-time", "common-version", @@ -2514,7 +2487,6 @@ dependencies = [ "memchr", "mito-codec", "nalgebra", - "num", "num-traits", "paste", "pretty_assertions", @@ -2591,7 +2563,6 @@ name = "common-grpc-expr" version = "1.3.0-alpha.1" dependencies = [ "api", - "common-base", "common-catalog", "common-error", "common-macro", @@ -2599,7 +2570,6 @@ dependencies = [ "common-sql", "common-time", "datatypes", - "paste", "prost 0.14.1", "snafu 0.8.6", "store-api", @@ -2675,7 +2645,6 @@ dependencies = [ "common-wal", "common-workload", "datafusion-common", - "datafusion-expr", "datatypes", "deadpool", "deadpool-postgres", @@ -2684,7 +2653,6 @@ dependencies = [ "flexbuffers", "futures", "futures-util", - "greptime-proto", "hex", "humantime-serde", "itertools 0.14.0", @@ -2779,7 +2747,6 @@ version = "1.3.0-alpha.1" dependencies = [ "async-trait", "common-procedure", - "snafu 0.8.6", "tokio", ] @@ -2852,13 +2819,11 @@ dependencies = [ "lazy_static", "num_cpus", "once_cell", - "parking_lot 0.12.4", "paste", "pin-project", "prometheus 0.14.0", "ratelimit", "serde", - "serde_json", "snafu 0.8.6", "tempfile", "tokio", @@ -2885,7 +2850,6 @@ dependencies = [ "common-decimal", "common-error", "common-macro", - "common-telemetry", "common-time", "datafusion-sql", "datatypes", @@ -2903,7 +2867,6 @@ dependencies = [ "common-base", "common-runtime", "common-telemetry", - "lazy_static", "nix 0.30.1", "num_cpus", "prometheus 0.14.0", @@ -4533,7 +4496,6 @@ dependencies = [ "async-trait", "bytes", "cache", - "client", "common-base", "common-catalog", "common-config", @@ -4541,7 +4503,6 @@ dependencies = [ "common-error", "common-function", "common-greptimedb-telemetry", - "common-grpc", "common-macro", "common-meta", "common-options", @@ -5332,7 +5293,6 @@ dependencies = [ "common-datasource", "common-error", "common-macro", - "common-query", "common-recordbatch", "common-telemetry", "common-test-util", @@ -5472,19 +5432,14 @@ dependencies = [ "arrow 59.2.0", "arrow-flight", "arrow-schema 59.2.0", - "async-recursion", "async-trait", - "bytes", "catalog", "chrono", "client", "common-base", "common-catalog", "common-config", - "common-datasource", - "common-decimal", "common-error", - "common-frontend", "common-function", "common-grpc", "common-macro", @@ -5497,42 +5452,31 @@ dependencies = [ "common-telemetry", "common-time", "common-version", - "config", "datafusion", "datafusion-common", "datafusion-expr", "datafusion-physical-expr", - "datafusion-substrait", "datatypes", - "enum_dispatch", "futures", "get-size2", "greptime-proto", "hostname 0.4.1", - "http 1.5.0", "humantime-serde", "itertools 0.14.0", "lazy_static", "meta-client", - "nom 7.1.3", - "num-traits", - "num_cpus", "operator", - "partition", "pretty_assertions", "prometheus 0.14.0", - "prost 0.14.1", "query", "rand 0.9.4", "serde", "serde_json", "servers", "session", - "smallvec", "snafu 0.8.6", "sql", "store-api", - "strum 0.27.1", "substrait 1.3.0-alpha.1", "table", "tokio", @@ -5633,7 +5577,6 @@ name = "frontend" version = "1.3.0-alpha.1" dependencies = [ "api", - "arc-swap", "arrow-flight", "async-stream", "async-trait", @@ -5657,7 +5600,6 @@ dependencies = [ "common-memory-manager", "common-meta", "common-options", - "common-procedure", "common-query", "common-recordbatch", "common-runtime", @@ -5672,13 +5614,11 @@ dependencies = [ "datatypes", "futures", "hostname 0.4.1", - "humantime", "humantime-serde", "hyper-util", "lazy_static", "log-query", "meta-client", - "num_cpus", "opentelemetry-proto", "opentelemetry-semantic-conventions 0.31.0", "operator", @@ -5689,7 +5629,6 @@ dependencies = [ "promql-parser", "prost 0.14.1", "query", - "rand 0.9.4", "reqwest 0.13.4", "serde", "serde_json", @@ -5708,7 +5647,6 @@ dependencies = [ "tonic 0.14.2", "tower 0.5.2", "tracing", - "uuid", ] [[package]] @@ -8121,7 +8059,6 @@ dependencies = [ "derive_builder 0.20.2", "futures", "futures-util", - "itertools 0.14.0", "lazy_static", "object-store", "pin-project", @@ -8465,20 +8402,16 @@ dependencies = [ "common-options", "common-telemetry", "common-time", - "datatypes", "futures", - "futures-util", "humantime-serde", "meta-srv", "rand 0.9.4", "serde", "serde_json", - "session", "snafu 0.8.6", "tokio", "tokio-stream", "tonic 0.14.2", - "tower 0.5.2", "tracing", "tracing-subscriber", ] @@ -8490,9 +8423,7 @@ dependencies = [ "api", "async-trait", "axum 0.8.4", - "axum-extra", "axum-macros", - "bytes", "chrono", "clap", "client", @@ -8525,19 +8456,14 @@ dependencies = [ "futures", "h2 0.3.26", "hostname 0.4.1", - "http-body-util", - "humantime", "humantime-serde", "hyper-util", "itertools 0.14.0", "lazy_static", - "local-ip-address", "once_cell", "ordered-float 4.6.0", - "parking_lot 0.12.4", "partition", "prometheus 0.14.0", - "prost 0.14.1", "rand 0.9.4", "regex", "rskafka", @@ -8555,7 +8481,6 @@ dependencies = [ "toml 0.8.23", "tonic 0.14.2", "tower 0.5.2", - "tower-http 0.6.11", "tracing", "tracing-subscriber", "typetag", @@ -8603,7 +8528,6 @@ dependencies = [ "common-recordbatch", "common-runtime", "common-telemetry", - "common-test-util", "common-time", "common-wal", "criterion 0.7.0", @@ -8723,9 +8647,7 @@ dependencies = [ "bytes", "chrono", "common-base", - "common-config", "common-datasource", - "common-decimal", "common-error", "common-function", "common-grpc", @@ -8764,7 +8686,6 @@ dependencies = [ "object-store", "parquet", "partition", - "paste", "pin-project", "prometheus 0.14.0", "prost 0.14.1", @@ -8776,11 +8697,9 @@ dependencies = [ "rskafka", "rstest", "rstest_reuse", - "scopeguard", "serde", "serde_json", "serde_with", - "session", "smallvec", "snafu 0.8.6", "store-api", @@ -9562,10 +9481,8 @@ name = "object-store" version = "1.3.0-alpha.1" dependencies = [ "anyhow", - "async-trait", "bytes", "cap-std", - "chrono", "common-base", "common-error", "common-macro", @@ -9575,13 +9492,12 @@ dependencies = [ "derive_builder 0.20.2", "futures", "humantime-serde", - "lazy_static", + "object-store", "object_store", "object_store_opendal", "opendal", "opendal-http-transport-reqwest", "prometheus 0.14.0", - "rand 0.9.4", "reqwest 0.13.4", "serde", "snafu 0.8.6", @@ -10096,7 +10012,6 @@ dependencies = [ "ahash 0.8.12", "api", "arrow 59.2.0", - "arrow-ipc 59.2.0", "async-stream", "async-trait", "axum 0.8.4", @@ -10143,7 +10058,6 @@ dependencies = [ "path-slash", "percent-encoding", "prometheus 0.14.0", - "prost 0.14.1", "prost-types 0.14.1", "query", "regex", @@ -10492,19 +10406,15 @@ name = "partition" version = "1.3.0-alpha.1" dependencies = [ "api", - "async-trait", "common-base", "common-error", "common-macro", "common-meta", - "common-query", - "common-telemetry", "criterion 0.7.0", "datafusion-common", "datafusion-expr", "datafusion-physical-expr", "datatypes", - "itertools 0.14.0", "moka", "rand 0.8.5", "serde", @@ -10893,29 +10803,22 @@ dependencies = [ "api", "arrow 59.2.0", "arrow-schema 59.2.0", - "async-trait", "catalog", "chrono", "chrono-tz", "common-catalog", "common-error", - "common-function", "common-macro", "common-meta", "common-query", "common-recordbatch", - "common-runtime", "common-telemetry", "common-time", "criterion 0.7.0", - "crossbeam-utils", "csv", - "dashmap", "datafusion", "datafusion-common", "datafusion-expr", - "datafusion-functions", - "datafusion-physical-expr", "datatypes", "dyn-fmt", "enum_dispatch", @@ -10930,10 +10833,8 @@ dependencies = [ "once_cell", "operator", "ordered-float 4.6.0", - "paste", "prometheus 0.14.0", "query", - "rayon", "regex", "serde", "serde_json", @@ -11867,7 +11768,6 @@ version = "1.3.0-alpha.1" dependencies = [ "ahash 0.8.12", "api", - "arc-swap", "arrow 59.2.0", "arrow-schema 59.2.0", "async-recursion", @@ -11901,12 +11801,10 @@ dependencies = [ "datafusion-proto", "datafusion-sql", "datatypes", - "either", "fastrand", "futures", "futures-util", "greptime-proto", - "humantime", "itertools 0.14.0", "jsonb", "lazy_static", @@ -11915,11 +11813,8 @@ dependencies = [ "meter-macros", "moka", "nalgebra", - "num", - "num-traits", "object-store", "once_cell", - "parking_lot 0.12.4", "partition", "paste", "pretty_assertions", @@ -11939,7 +11834,6 @@ dependencies = [ "substrait 1.3.0-alpha.1", "table", "tokio", - "tokio-stream", "toml 0.8.23", "tracing", "uuid", @@ -13582,7 +13476,6 @@ dependencies = [ "common-catalog", "common-decimal", "common-error", - "common-frontend", "common-grpc", "common-macro", "common-mem-prof", @@ -13602,7 +13495,6 @@ dependencies = [ "common-version", "criterion 0.5.1", "csv", - "dashmap", "datafusion", "datafusion-common", "datafusion-expr", @@ -14070,25 +13962,19 @@ dependencies = [ "common-base", "common-catalog", "common-datasource", - "common-decimal", "common-error", "common-macro", "common-query", "common-sql", - "common-telemetry", "common-time", "datafusion", "datafusion-common", "datafusion-expr", - "datafusion-physical-expr", "datafusion-sql", "datatypes", "either", - "hex", - "humantime", "iso8601", "itertools 0.14.0", - "jsonb", "lazy_static", "pretty_assertions", "promql-parser", @@ -14098,7 +13984,6 @@ dependencies = [ "snafu 0.8.6", "sqlparser", "sqlparser_derive 0.1.1", - "store-api", "table", "uuid", ] @@ -14130,7 +14015,6 @@ dependencies = [ "client", "common-error", "common-query", - "common-recordbatch", "common-time", "datatypes", "flate2", @@ -14458,7 +14342,6 @@ dependencies = [ "api", "aquamarine", "arrow-schema 59.2.0", - "async-stream", "async-trait", "bytes", "common-base", @@ -14485,7 +14368,6 @@ dependencies = [ "serde", "serde_json", "snafu 0.8.6", - "sqlparser", "strum 0.27.1", "tokio", "uuid", @@ -14654,16 +14536,13 @@ dependencies = [ "common-error", "common-function", "common-macro", - "common-telemetry", "datafusion", "datafusion-common", "datafusion-expr", "datafusion-substrait", - "datatypes", "promql", "prost 0.14.1", "snafu 0.8.6", - "tokio", ] [[package]] @@ -14812,7 +14691,6 @@ dependencies = [ "futures", "greptime-proto", "humantime", - "humantime-serde", "lazy_static", "once_cell", "parquet", @@ -14820,10 +14698,8 @@ dependencies = [ "serde", "serde_json", "snafu 0.8.6", - "sqlparser", "store-api", "tokio", - "tokio-util", ] [[package]] @@ -15121,7 +14997,6 @@ version = "1.3.0-alpha.1" dependencies = [ "api", "arrow-flight", - "async-stream", "async-trait", "auth", "axum 0.8.4", @@ -15149,7 +15024,6 @@ dependencies = [ "common-query", "common-recordbatch", "common-runtime", - "common-stat", "common-telemetry", "common-test-util", "common-time", @@ -15180,7 +15054,6 @@ dependencies = [ "opentelemetry-proto", "operator", "otel-arrow-rust", - "partition", "paste", "pipeline", "plugins", @@ -15195,7 +15068,6 @@ dependencies = [ "servers", "session", "similar-asserts", - "snafu 0.8.6", "sql", "sqlx", "standalone", diff --git a/Cargo.toml b/Cargo.toml index 7c5c53d91c8..5cdc3cf63a2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -104,7 +104,6 @@ aquamarine = "0.6" arrow = { version = "=59.2.0", features = ["prettyprint"] } arrow-array = { version = "=59.2.0", default-features = false, features = ["chrono-tz"] } arrow-buffer = "=59.2.0" -arrow-cast = "=59.2.0" arrow-flight = "=59.2.0" arrow-ipc = { version = "=59.2.0", default-features = false, features = ["lz4", "zstd"] } arrow-schema = { version = "=59.2.0", features = ["serde"] } @@ -236,7 +235,6 @@ rskafka = { git = "https://github.com/influxdata/rskafka.git", rev = "80f42eb15e rstest = "0.25" rstest_reuse = "0.7" rust_decimal = "1.33" -rustc-hash = "2.0" rustls = { version = "0.23.25", default-features = false } ryu = "1.0" sea-query = "0.32" diff --git a/Makefile b/Makefile index 7176eafd0bb..1379d86683c 100644 --- a/Makefile +++ b/Makefile @@ -233,14 +233,11 @@ fix-clippy: ## Fix clippy violations. .PHONY: check-udeps check-udeps: ## Check unused dependencies. - cargo udeps --workspace --all-targets + cargo shear .PHONY: fix-udeps fix-udeps: ## Remove unused dependencies automatically. - @echo "Running cargo-udeps to find unused dependencies..." - @cargo udeps --workspace --all-targets --output json > udeps-report.json || true - @echo "Removing unused dependencies..." - @python3 scripts/fix-udeps.py udeps-report.json + cargo shear --fix .PHONY: fmt-check fmt-check: ## Check code format. diff --git a/flake.nix b/flake.nix index a5886b8add1..2d95d8dcfd6 100644 --- a/flake.nix +++ b/flake.nix @@ -46,7 +46,7 @@ ]) cargo-nextest cargo-llvm-cov - cargo-udeps + cargo-shear taplo curl gnuplot ## for cargo bench diff --git a/scripts/fix-udeps.py b/scripts/fix-udeps.py deleted file mode 100755 index 7547b897361..00000000000 --- a/scripts/fix-udeps.py +++ /dev/null @@ -1,265 +0,0 @@ -# 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. - -import json -import os -import re -import sys - - -def load_udeps_report(report_path): - try: - with open(report_path, "r") as f: - return json.load(f) - except FileNotFoundError: - print(f"Error: Report file '{report_path}' not found.") - return None - except json.JSONDecodeError as e: - print(f"Error: Invalid JSON in report file: {e}") - return None - - -def extract_unused_dependencies(report): - """ - Extract and organize unused dependencies from the cargo-udeps JSON report. - - The cargo-udeps report has this structure: - { - "unused_deps": { - "package_name v0.1.0 (/path/to/package)": { - "normal": ["dep1", "dep2"], - "development": ["dev_dep1"], - "build": ["build_dep1"], - "manifest_path": "/path/to/Cargo.toml" - } - } - } - - Args: - report (dict): The parsed JSON report from cargo-udeps - - Returns: - dict: Organized unused dependencies by package name: - { - "package_name": { - "dependencies": [("dep1", "normal"), ("dev_dep1", "dev")], - "manifest_path": "/path/to/Cargo.toml" - } - } - """ - if not report or "unused_deps" not in report: - return {} - - unused_deps = {} - for package_full_name, deps_info in report["unused_deps"].items(): - package_name = package_full_name.split(" ")[0] - - all_unused = [] - if deps_info.get("normal"): - all_unused.extend([(dep, "normal") for dep in deps_info["normal"]]) - if deps_info.get("development"): - all_unused.extend([(dep, "dev") for dep in deps_info["development"]]) - if deps_info.get("build"): - all_unused.extend([(dep, "build") for dep in deps_info["build"]]) - - if all_unused: - unused_deps[package_name] = { - "dependencies": all_unused, - "manifest_path": deps_info.get("manifest_path", "unknown"), - } - - return unused_deps - - -def get_section_pattern(dep_type): - """ - Get regex patterns to identify different dependency sections in Cargo.toml. - - Args: - dep_type (str): Type of dependency ("normal", "dev", or "build") - - Returns: - list: List of regex patterns to match the appropriate section headers - - """ - patterns = { - "normal": [r"\[dependencies\]", r"\[dependencies\..*?\]"], - "dev": [r"\[dev-dependencies\]", r"\[dev-dependencies\..*?\]"], - "build": [r"\[build-dependencies\]", r"\[build-dependencies\..*?\]"], - } - return patterns.get(dep_type, []) - - -def remove_dependency_line(content, dep_name, section_start, section_end): - """ - Remove a dependency line from a specific section of a Cargo.toml file. - - Args: - content (str): The entire content of the Cargo.toml file - dep_name (str): Name of the dependency to remove (e.g., "serde", "tokio") - section_start (int): Starting position of the section in the content - section_end (int): Ending position of the section in the content - - Returns: - tuple: (new_content, removed) where: - - new_content (str): The modified content with dependency removed - - removed (bool): True if dependency was found and removed, False otherwise - - Example input content format: - content = ''' - [package] - name = "my-crate" - version = "0.1.0" - - [dependencies] - serde = "1.0" - tokio = { version = "1.0", features = ["full"] } - serde_json.workspace = true - - [dev-dependencies] - tempfile = "3.0" - ''' - - # If dep_name = "serde", section_start = start of [dependencies], - # section_end = start of [dev-dependencies], this function will: - # 1. Extract the section: "serde = "1.0"\ntokio = { version = "1.0", features = ["full"] }\nserde_json.workspace = true\n" - # 2. Find and remove the line: "serde = "1.0"" - # 3. Return the modified content with that line removed - """ - section_content = content[section_start:section_end] - - dep_patterns = [ - rf"^{re.escape(dep_name)}\s*=.*$", # e.g., "serde = "1.0"" - rf"^{re.escape(dep_name)}\.workspace\s*=.*$", # e.g., "serde_json.workspace = true" - ] - - for pattern in dep_patterns: - match = re.search(pattern, section_content, re.MULTILINE) - if match: - line_start = section_start + match.start() # Start of the matched line - line_end = section_start + match.end() # End of the matched line - - if line_end < len(content) and content[line_end] == "\n": - line_end += 1 - - return content[:line_start] + content[line_end:], True - - return content, False - - -def remove_dependency_from_toml(file_path, dep_name, dep_type): - """ - Remove a specific dependency from a Cargo.toml file. - - Args: - file_path (str): Path to the Cargo.toml file - dep_name (str): Name of the dependency to remove - dep_type (str): Type of dependency ("normal", "dev", or "build") - - Returns: - bool: True if dependency was successfully removed, False otherwise - """ - try: - with open(file_path, "r") as f: - content = f.read() - - section_patterns = get_section_pattern(dep_type) - if not section_patterns: - return False - - for pattern in section_patterns: - section_match = re.search(pattern, content, re.IGNORECASE) - if not section_match: - continue - - section_start = section_match.end() - next_section = re.search(r"\n\s*\[", content[section_start:]) - section_end = ( - section_start + next_section.start() if next_section else len(content) - ) - - new_content, removed = remove_dependency_line( - content, dep_name, section_start, section_end - ) - if removed: - with open(file_path, "w") as f: - f.write(new_content) - return True - - return False - - except Exception as e: - print(f"Error processing {file_path}: {e}") - return False - - -def process_unused_dependencies(unused_deps): - """ - Process and remove all unused dependencies from their respective Cargo.toml files. - - Args: - unused_deps (dict): Dictionary of unused dependencies organized by package: - { - "package_name": { - "dependencies": [("dep1", "normal"), ("dev_dep1", "dev")], - "manifest_path": "/path/to/Cargo.toml" - } - } - - """ - if not unused_deps: - print("No unused dependencies found.") - return - - total_removed = 0 - total_failed = 0 - - for package, info in unused_deps.items(): - deps = info["dependencies"] - manifest_path = info["manifest_path"] - - if not os.path.exists(manifest_path): - print(f"Manifest file not found: {manifest_path}") - total_failed += len(deps) - continue - - for dep, dep_type in deps: - if remove_dependency_from_toml(manifest_path, dep, dep_type): - print(f"Removed {dep} from {package}") - total_removed += 1 - else: - print(f"Failed to remove {dep} from {package}") - total_failed += 1 - - print(f"Removed {total_removed} dependencies") - if total_failed > 0: - print(f"Failed to remove {total_failed} dependencies") - - -def main(): - if len(sys.argv) > 1: - report_path = sys.argv[1] - else: - report_path = "udeps-report.json" - - report = load_udeps_report(report_path) - if report is None: - sys.exit(1) - - unused_deps = extract_unused_dependencies(report) - process_unused_dependencies(unused_deps) - - -if __name__ == "__main__": - main() diff --git a/src/catalog/Cargo.toml b/src/catalog/Cargo.toml index f46887e148a..fe42ddc2d13 100644 --- a/src/catalog/Cargo.toml +++ b/src/catalog/Cargo.toml @@ -61,9 +61,7 @@ tokio-stream.workspace = true [dev-dependencies] cache.workspace = true -catalog = { workspace = true, features = ["testing"] } chrono.workspace = true common-meta = { workspace = true, features = ["testing"] } common-query = { workspace = true, features = ["testing"] } -object-store.workspace = true tokio.workspace = true diff --git a/src/cli/Cargo.toml b/src/cli/Cargo.toml index 587e5b8cffa..8178fa1b0b5 100644 --- a/src/cli/Cargo.toml +++ b/src/cli/Cargo.toml @@ -18,10 +18,8 @@ workspace = true [dependencies] async-stream.workspace = true async-trait.workspace = true -auth.workspace = true base64.workspace = true cache.workspace = true -catalog.workspace = true chrono.workspace = true clap.workspace = true client = { workspace = true, features = ["testing"] } @@ -34,14 +32,11 @@ common-grpc.workspace = true common-macro.workspace = true common-meta.workspace = true common-procedure.workspace = true -common-query.workspace = true -common-recordbatch.workspace = true common-runtime.workspace = true common-telemetry = { workspace = true, features = [ "deadlock_detection", ] } common-time.workspace = true -common-version.workspace = true common-wal.workspace = true datatypes.workspace = true etcd-client.workspace = true @@ -61,21 +56,18 @@ reqwest.workspace = true serde.workspace = true serde_json.workspace = true servers.workspace = true -session.workspace = true snafu.workspace = true sql.workspace = true standalone.workspace = true store-api.workspace = true table.workspace = true tokio.workspace = true -tracing-appender.workspace = true url.workspace = true uuid.workspace = true [dev-dependencies] common-meta = { workspace = true, features = ["testing"] } common-test-util.workspace = true -common-version.workspace = true object-store = { workspace = true, features = ["testing"] } serde.workspace = true tempfile.workspace = true diff --git a/src/client/Cargo.toml b/src/client/Cargo.toml index 5b479b1fbf2..36c3f60a9a4 100644 --- a/src/client/Cargo.toml +++ b/src/client/Cargo.toml @@ -45,7 +45,3 @@ substrait.workspace = true tokio.workspace = true tokio-stream = { workspace = true, features = ["net"] } tonic.workspace = true - -[dev-dependencies] -common-grpc-expr.workspace = true -tracing = "0.1" diff --git a/src/cmd/Cargo.toml b/src/cmd/Cargo.toml index fff268dea45..367b86809d8 100644 --- a/src/cmd/Cargo.toml +++ b/src/cmd/Cargo.toml @@ -43,10 +43,8 @@ workspace = true api.workspace = true async-trait.workspace = true auth.workspace = true -base64.workspace = true cache.workspace = true catalog.workspace = true -chrono.workspace = true clap.workspace = true cli.workspace = true client.workspace = true @@ -63,7 +61,6 @@ common-meta.workspace = true common-options.workspace = true common-procedure.workspace = true common-query.workspace = true -common-recordbatch.workspace = true common-runtime.workspace = true common-stat.workspace = true common-telemetry = { workspace = true, features = [ @@ -78,19 +75,16 @@ datafusion-physical-plan.workspace = true datafusion_object_store.workspace = true datanode.workspace = true datatypes.workspace = true -either = "1.15" etcd-client.workspace = true flow.workspace = true frontend = { workspace = true, default-features = false } futures.workspace = true hex.workspace = true human-panic = "2.0" -humantime.workspace = true lazy_static.workspace = true log-store.workspace = true meta-client.workspace = true meta-srv.workspace = true -metric-engine.workspace = true mito-codec.workspace = true mito2.workspace = true moka = { workspace = true, features = ["future"] } @@ -107,16 +101,13 @@ reqwest.workspace = true serde.workspace = true serde_json.workspace = true servers.workspace = true -session.workspace = true smallvec.workspace = true snafu.workspace = true sqlparser.workspace = true standalone.workspace = true store-api.workspace = true -table.workspace = true tokio.workspace = true toml.workspace = true -tonic.workspace = true tracing-appender.workspace = true [target.'cfg(unix)'.dependencies] @@ -134,6 +125,7 @@ client = { workspace = true, features = ["testing"] } common-test-util.workspace = true common-version.workspace = true file-engine.workspace = true +metric-engine.workspace = true mito2.workspace = true serde.workspace = true similar-asserts.workspace = true diff --git a/src/common/config/Cargo.toml b/src/common/config/Cargo.toml index 27b238add73..ad356716867 100644 --- a/src/common/config/Cargo.toml +++ b/src/common/config/Cargo.toml @@ -29,4 +29,3 @@ datanode.workspace = true meta-client.workspace = true serde.workspace = true temp-env = "0.3" -tempfile.workspace = true diff --git a/src/common/frontend/Cargo.toml b/src/common/frontend/Cargo.toml index 32a5146b566..26092ae13f7 100644 --- a/src/common/frontend/Cargo.toml +++ b/src/common/frontend/Cargo.toml @@ -12,14 +12,9 @@ common-event-recorder.workspace = true common-grpc.workspace = true common-macro.workspace = true common-meta.workspace = true -common-time.workspace = true greptime-proto.workspace = true -humantime.workspace = true meta-client.workspace = true serde.workspace = true serde_json.workspace = true snafu.workspace = true tonic.workspace = true - -[dev-dependencies] -tokio.workspace = true diff --git a/src/common/function/Cargo.toml b/src/common/function/Cargo.toml index ee6cfac9dfb..3a5da30c136 100644 --- a/src/common/function/Cargo.toml +++ b/src/common/function/Cargo.toml @@ -17,21 +17,17 @@ ai_functions = ["dep:futures", "dep:reqwest"] [dependencies] ahash.workspace = true api.workspace = true -arc-swap = "1.0" arrow.workspace = true -arrow-cast.workspace = true arrow-schema.workspace = true async-trait.workspace = true bincode = "=1.3.3" catalog.workspace = true -chrono.workspace = true common-base.workspace = true common-catalog.workspace = true common-error.workspace = true common-macro.workspace = true common-meta.workspace = true common-query.workspace = true -common-runtime.workspace = true common-telemetry.workspace = true common-time.workspace = true common-version.workspace = true @@ -55,7 +51,6 @@ jsonb.workspace = true memchr = "2.7" mito-codec.workspace = true nalgebra.workspace = true -num = "0.4" num-traits = "0.2" paste.workspace = true regex.workspace = true diff --git a/src/common/grpc-expr/Cargo.toml b/src/common/grpc-expr/Cargo.toml index a689de7c018..b735e8f9ba8 100644 --- a/src/common/grpc-expr/Cargo.toml +++ b/src/common/grpc-expr/Cargo.toml @@ -9,7 +9,6 @@ workspace = true [dependencies] api.workspace = true -common-base.workspace = true common-catalog.workspace = true common-error.workspace = true common-macro.workspace = true @@ -21,6 +20,3 @@ prost.workspace = true snafu.workspace = true store-api.workspace = true table.workspace = true - -[dev-dependencies] -paste.workspace = true diff --git a/src/common/mem-prof/Cargo.toml b/src/common/mem-prof/Cargo.toml index 10b67cf7fa7..46d959b02cc 100644 --- a/src/common/mem-prof/Cargo.toml +++ b/src/common/mem-prof/Cargo.toml @@ -26,3 +26,9 @@ jemalloc-pprof-mappings = { version = "0.7", package = "mappings" } # for get th [target.'cfg(not(windows))'.dependencies.tikv-jemalloc-sys] features = ["stats", "profiling", "unprefixed_malloc_on_supported_platforms"] version = "0.7" + +[package.metadata.cargo-shear] +# `tikv-jemalloc-sys` is a link-only dependency: it enables jemalloc +# profiling/stats features and pins the linked jemalloc instance, but is +# never referenced by name in source code. +ignored = ["tikv-jemalloc-sys"] diff --git a/src/common/meta/Cargo.toml b/src/common/meta/Cargo.toml index 8deb3d00814..bd974deb4cf 100644 --- a/src/common/meta/Cargo.toml +++ b/src/common/meta/Cargo.toml @@ -46,7 +46,6 @@ common-time.workspace = true common-wal.workspace = true common-workload.workspace = true datafusion-common.workspace = true -datafusion-expr.workspace = true datatypes.workspace = true deadpool = { workspace = true, optional = true } deadpool-postgres = { workspace = true, optional = true } @@ -55,7 +54,6 @@ etcd-client.workspace = true flexbuffers = "25.2" futures.workspace = true futures-util.workspace = true -greptime-proto.workspace = true hex.workspace = true humantime-serde.workspace = true itertools.workspace = true diff --git a/src/common/procedure-test/Cargo.toml b/src/common/procedure-test/Cargo.toml index e445a235bc9..907035dd052 100644 --- a/src/common/procedure-test/Cargo.toml +++ b/src/common/procedure-test/Cargo.toml @@ -10,5 +10,4 @@ workspace = true [dependencies] async-trait.workspace = true common-procedure = { workspace = true, features = ["testing"] } -snafu.workspace = true tokio.workspace = true diff --git a/src/common/query/Cargo.toml b/src/common/query/Cargo.toml index 9706098826f..4acdfafcb6c 100644 --- a/src/common/query/Cargo.toml +++ b/src/common/query/Cargo.toml @@ -39,3 +39,8 @@ common-base.workspace = true futures-util.workspace = true serde_json.workspace = true tokio.workspace = true + +[package.metadata.cargo-shear] +# `sqlparser` is required by `sqlparser_derive`: the generated code references +# `::sqlparser` paths, so it must stay a direct dependency. +ignored = ["sqlparser"] diff --git a/src/common/runtime/Cargo.toml b/src/common/runtime/Cargo.toml index 927aa17b9c2..410173f6445 100644 --- a/src/common/runtime/Cargo.toml +++ b/src/common/runtime/Cargo.toml @@ -25,13 +25,11 @@ futures.workspace = true lazy_static.workspace = true num_cpus.workspace = true once_cell.workspace = true -parking_lot.workspace = true paste.workspace = true pin-project.workspace = true prometheus.workspace = true ratelimit.workspace = true serde.workspace = true -serde_json.workspace = true snafu.workspace = true tempfile.workspace = true tokio.workspace = true diff --git a/src/common/sql/Cargo.toml b/src/common/sql/Cargo.toml index 3a66ec2d04e..318685996d5 100644 --- a/src/common/sql/Cargo.toml +++ b/src/common/sql/Cargo.toml @@ -10,7 +10,6 @@ common-base.workspace = true common-decimal.workspace = true common-error.workspace = true common-macro.workspace = true -common-telemetry.workspace = true common-time.workspace = true datafusion-sql.workspace = true datatypes.workspace = true diff --git a/src/common/stat/Cargo.toml b/src/common/stat/Cargo.toml index 2a99107c909..fcf927e9cca 100644 --- a/src/common/stat/Cargo.toml +++ b/src/common/stat/Cargo.toml @@ -8,7 +8,6 @@ license.workspace = true common-base.workspace = true common-runtime.workspace = true common-telemetry.workspace = true -lazy_static.workspace = true num_cpus.workspace = true prometheus.workspace = true sysinfo.workspace = true diff --git a/src/common/substrait/Cargo.toml b/src/common/substrait/Cargo.toml index a4e78475051..a56b56b8301 100644 --- a/src/common/substrait/Cargo.toml +++ b/src/common/substrait/Cargo.toml @@ -13,7 +13,6 @@ bytes.workspace = true common-error.workspace = true common-function.workspace = true common-macro.workspace = true -common-telemetry.workspace = true datafusion.workspace = true datafusion-common.workspace = true datafusion-expr.workspace = true @@ -21,7 +20,3 @@ datafusion-substrait.workspace = true promql.workspace = true prost.workspace = true snafu.workspace = true - -[dev-dependencies] -datatypes.workspace = true -tokio.workspace = true diff --git a/src/datanode/Cargo.toml b/src/datanode/Cargo.toml index 6b78858bd2f..7c126764723 100644 --- a/src/datanode/Cargo.toml +++ b/src/datanode/Cargo.toml @@ -16,7 +16,6 @@ api.workspace = true arrow-flight.workspace = true async-trait.workspace = true bytes.workspace = true -client.workspace = true common-base.workspace = true common-catalog.workspace = true common-config.workspace = true @@ -24,7 +23,6 @@ common-datasource.workspace = true common-error.workspace = true common-function.workspace = true common-greptimedb-telemetry.workspace = true -common-grpc.workspace = true common-macro.workspace = true common-meta.workspace = true common-options.workspace = true @@ -75,7 +73,6 @@ tracing.workspace = true [dev-dependencies] cache.workspace = true -client.workspace = true common-meta = { workspace = true, features = ["testing"] } common-query.workspace = true common-test-util.workspace = true diff --git a/src/datatypes/Cargo.toml b/src/datatypes/Cargo.toml index 124b1e0ea64..d1dd70acef1 100644 --- a/src/datatypes/Cargo.toml +++ b/src/datatypes/Cargo.toml @@ -37,3 +37,8 @@ serde_json.workspace = true snafu.workspace = true sqlparser.workspace = true sqlparser_derive = "0.1" + +[package.metadata.cargo-shear] +# `sqlparser` is required by `sqlparser_derive`: the generated code references +# `::sqlparser` paths, so it must stay a direct dependency. +ignored = ["sqlparser"] diff --git a/src/file-engine/Cargo.toml b/src/file-engine/Cargo.toml index ec904addf1e..dda5fb4221c 100644 --- a/src/file-engine/Cargo.toml +++ b/src/file-engine/Cargo.toml @@ -18,7 +18,6 @@ common-catalog.workspace = true common-datasource.workspace = true common-error.workspace = true common-macro.workspace = true -common-query.workspace = true common-recordbatch.workspace = true common-telemetry.workspace = true common-test-util = { workspace = true, optional = true } diff --git a/src/flow/Cargo.toml b/src/flow/Cargo.toml index 4586e612693..d0ac5eee9ac 100644 --- a/src/flow/Cargo.toml +++ b/src/flow/Cargo.toml @@ -11,18 +11,13 @@ workspace = true api.workspace = true arrow.workspace = true arrow-schema.workspace = true -async-recursion = "1.0" async-trait.workspace = true -bytes.workspace = true catalog.workspace = true chrono.workspace = true client.workspace = true common-base.workspace = true common-config.workspace = true -common-datasource.workspace = true -common-decimal.workspace = true common-error.workspace = true -common-frontend.workspace = true common-function.workspace = true common-grpc.workspace = true common-macro.workspace = true @@ -35,41 +30,30 @@ common-stat.workspace = true common-telemetry.workspace = true common-time.workspace = true common-version.workspace = true -config.workspace = true datafusion.workspace = true datafusion-common.workspace = true datafusion-expr.workspace = true datafusion-physical-expr.workspace = true -datafusion-substrait.workspace = true datatypes.workspace = true -enum_dispatch = "0.3" futures.workspace = true get-size2 = "0.1.2" greptime-proto.workspace = true hostname.workspace = true -http.workspace = true humantime-serde.workspace = true itertools.workspace = true lazy_static.workspace = true meta-client.workspace = true -nom = "7.1.3" -num-traits = "0.2" -num_cpus.workspace = true operator.workspace = true -partition.workspace = true prometheus.workspace = true -prost.workspace = true query.workspace = true rand.workspace = true serde.workspace = true serde_json.workspace = true servers.workspace = true session.workspace = true -smallvec.workspace = true snafu.workspace = true sql.workspace = true store-api.workspace = true -strum.workspace = true substrait.workspace = true table.workspace = true tokio.workspace = true @@ -81,7 +65,6 @@ catalog = { workspace = true, features = ["testing"] } common-catalog.workspace = true common-meta = { workspace = true, features = ["testing"] } pretty_assertions.workspace = true -prost.workspace = true query.workspace = true session.workspace = true table.workspace = true diff --git a/src/frontend/Cargo.toml b/src/frontend/Cargo.toml index b59e1e83a4a..1aa057d0fcd 100644 --- a/src/frontend/Cargo.toml +++ b/src/frontend/Cargo.toml @@ -19,7 +19,6 @@ workspace = true [dependencies] api.workspace = true -arc-swap = "1.0" async-stream.workspace = true async-trait.workspace = true auth.workspace = true @@ -42,7 +41,6 @@ common-macro.workspace = true common-memory-manager.workspace = true common-meta.workspace = true common-options.workspace = true -common-procedure.workspace = true common-query.workspace = true common-recordbatch.workspace = true common-runtime.workspace = true @@ -56,12 +54,10 @@ datafusion-expr.workspace = true datatypes.workspace = true futures.workspace = true hostname.workspace = true -humantime.workspace = true humantime-serde.workspace = true lazy_static.workspace = true log-query.workspace = true meta-client.workspace = true -num_cpus.workspace = true opentelemetry-proto.workspace = true opentelemetry-semantic-conventions = { version = "0.31", features = ["semconv_experimental"] } operator.workspace = true @@ -72,7 +68,6 @@ prometheus.workspace = true promql-parser.workspace = true prost.workspace = true query.workspace = true -rand.workspace = true serde.workspace = true serde_json.workspace = true servers.workspace = true @@ -100,4 +95,3 @@ serde_json.workspace = true strfmt = "0.2" temp-env.workspace = true tower.workspace = true -uuid.workspace = true diff --git a/src/log-store/Cargo.toml b/src/log-store/Cargo.toml index 15560b06b2c..5d068ca344d 100644 --- a/src/log-store/Cargo.toml +++ b/src/log-store/Cargo.toml @@ -31,7 +31,6 @@ delta-encoding = "0.4" derive_builder.workspace = true futures.workspace = true futures-util.workspace = true -itertools.workspace = true lazy_static.workspace = true object-store.workspace = true pin-project.workspace = true @@ -49,8 +48,13 @@ tokio.workspace = true common-meta = { workspace = true, features = ["testing"] } common-test-util.workspace = true common-wal = { workspace = true, features = ["testing"] } -itertools.workspace = true object-store = { workspace = true, features = ["testing"] } rand.workspace = true rskafka = { workspace = true, features = ["unstable-fuzzing"] } uuid.workspace = true + +[package.metadata.cargo-shear] +# `protobuf` is required by the code generated by `build.rs`: the generated +# protobuf bindings (included via `include!`) reference `::protobuf` paths, +# but no source file does. +ignored = ["protobuf"] diff --git a/src/meta-client/Cargo.toml b/src/meta-client/Cargo.toml index 984f434cc51..9863b32bbbf 100644 --- a/src/meta-client/Cargo.toml +++ b/src/meta-client/Cargo.toml @@ -19,7 +19,6 @@ common-options.workspace = true common-telemetry.workspace = true common-time.workspace = true futures.workspace = true -futures-util.workspace = true humantime-serde.workspace = true rand.workspace = true serde.workspace = true @@ -31,10 +30,7 @@ tonic.workspace = true [dev-dependencies] common-event-recorder.workspace = true -datatypes.workspace = true futures.workspace = true meta-srv = { workspace = true, features = ["mock"] } -session.workspace = true -tower.workspace = true tracing = "0.1" tracing-subscriber.workspace = true diff --git a/src/meta-srv/Cargo.toml b/src/meta-srv/Cargo.toml index e6585505db4..6baca0934e1 100644 --- a/src/meta-srv/Cargo.toml +++ b/src/meta-srv/Cargo.toml @@ -14,16 +14,11 @@ enterprise = ["common-meta/enterprise"] [lints] workspace = true -[target.'cfg(not(target_os = "android"))'.dependencies] -local-ip-address.workspace = true - [dependencies] api.workspace = true async-trait.workspace = true axum.workspace = true -axum-extra.workspace = true axum-macros.workspace = true -bytes.workspace = true chrono.workspace = true clap.workspace = true client.workspace = true @@ -55,18 +50,14 @@ etcd-client.workspace = true futures.workspace = true h2 = "0.3" hostname.workspace = true -http-body-util = "0.1" -humantime.workspace = true humantime-serde.workspace = true hyper-util = { workspace = true, features = ["tokio"] } itertools.workspace = true lazy_static.workspace = true once_cell.workspace = true ordered-float.workspace = true -parking_lot.workspace = true partition.workspace = true prometheus.workspace = true -prost.workspace = true rand.workspace = true regex.workspace = true rskafka.workspace = true @@ -87,7 +78,6 @@ tokio-stream = { workspace = true, features = ["net"] } toml.workspace = true tonic.workspace = true tower.workspace = true -tower-http.workspace = true tracing.workspace = true typetag.workspace = true url = "2.3" diff --git a/src/metric-engine/Cargo.toml b/src/metric-engine/Cargo.toml index 0005a4dc543..92000549832 100644 --- a/src/metric-engine/Cargo.toml +++ b/src/metric-engine/Cargo.toml @@ -14,7 +14,7 @@ async-stream.workspace = true async-trait.workspace = true base64.workspace = true bytes.workspace = true -fxhash = "0.2" +chrono.workspace = true common-base.workspace = true common-error.workspace = true common-grpc.workspace = true @@ -24,10 +24,10 @@ common-recordbatch.workspace = true common-runtime.workspace = true common-telemetry.workspace = true common-time.workspace = true -chrono.workspace = true datafusion.workspace = true datatypes.workspace = true futures-util.workspace = true +fxhash = "0.2" humantime-serde.workspace = true itertools.workspace = true lazy_static = "1.4" @@ -45,18 +45,14 @@ tokio.workspace = true tracing.workspace = true [dev-dependencies] +common-function = { workspace = true } common-meta = { workspace = true, features = ["testing"] } -common-test-util.workspace = true -mito2 = { workspace = true, features = ["test"] } common-wal = { workspace = true } criterion = { workspace = true, features = ["async", "async_tokio"] } +mito2 = { workspace = true, features = ["test"] } mur3 = "0.1" -common-function = { workspace = true } partition = { workspace = true } [[bench]] name = "bench_tsid_generator" harness = false - -[package.metadata.cargo-udeps.ignore] -normal = ["aquamarine"] diff --git a/src/mito2/Cargo.toml b/src/mito2/Cargo.toml index 2be72af57cf..82a2884549a 100644 --- a/src/mito2/Cargo.toml +++ b/src/mito2/Cargo.toml @@ -20,7 +20,6 @@ api.workspace = true aquamarine.workspace = true arrow-schema.workspace = true async-channel = "1.9" -common-stat.workspace = true async-stream.workspace = true async-trait.workspace = true base64.workspace = true @@ -28,58 +27,55 @@ bytemuck.workspace = true bytes.workspace = true chrono.workspace = true common-base.workspace = true -common-config.workspace = true common-datasource.workspace = true -common-decimal.workspace = true common-error.workspace = true -common-grpc.workspace = true common-function.workspace = true +common-grpc.workspace = true common-macro.workspace = true common-memory-manager.workspace = true common-meta.workspace = true common-query.workspace = true common-recordbatch.workspace = true common-runtime.workspace = true +common-stat.workspace = true common-telemetry.workspace = true common-test-util = { workspace = true, optional = true } common-time.workspace = true common-wal.workspace = true crc32fast = "1" crossbeam-utils.workspace = true +dashmap.workspace = true datafusion.workspace = true datafusion-common.workspace = true datafusion-expr.workspace = true datatypes.workspace = true -dashmap.workspace = true derive_more.workspace = true dotenv.workspace = true either.workspace = true futures.workspace = true +greptime-proto.workspace = true humantime.workspace = true humantime-serde.workspace = true index.workspace = true itertools.workspace = true -greptime-proto.workspace = true lazy_static = "1.4" log-store = { workspace = true } mito-codec.workspace = true moka = { workspace = true, features = ["sync", "future"] } object-store = { workspace = true, features = ["testing"] } parquet = { workspace = true, features = ["async", "variant_experimental"] } -paste.workspace = true +partition.workspace = true pin-project.workspace = true prometheus.workspace = true prost.workspace = true -partition.workspace = true puffin.workspace = true rand.workspace = true rayon = "1.10" -roaring = { version = "0.10", optional = true } regex.workspace = true +roaring = { version = "0.10", optional = true } rskafka = { workspace = true, optional = true } rstest = { workspace = true, optional = true } rstest_reuse = { workspace = true, optional = true } -scopeguard = "1.2" serde.workspace = true serde_json.workspace = true serde_with.workspace = true @@ -108,7 +104,6 @@ object-store = { workspace = true, features = ["services-memory"] } rskafka.workspace = true rstest.workspace = true rstest_reuse.workspace = true -session.workspace = true toml.workspace = true [[bench]] @@ -159,6 +154,3 @@ required-features = ["testing"] name = "bench_dense_index_update" harness = false required-features = ["testing"] - -[package.metadata.cargo-udeps.ignore] -normal = ["aquamarine"] diff --git a/src/object-store/Cargo.toml b/src/object-store/Cargo.toml index 412f3563298..4083271a967 100644 --- a/src/object-store/Cargo.toml +++ b/src/object-store/Cargo.toml @@ -10,23 +10,18 @@ workspace = true [features] mysql-object-store = ["opendal/services-mysql"] services-memory = ["opendal/services-memory"] -testing = ["derive_builder"] +testing = ["derive_builder", "uuid"] [dependencies] -async-trait.workspace = true bytes.workspace = true cap-std.workspace = true -chrono.workspace = true common-base.workspace = true common-error.workspace = true common-macro.workspace = true common-runtime.workspace = true common-telemetry.workspace = true -datafusion_object_store.workspace = true derive_builder = { workspace = true, optional = true } -futures.workspace = true humantime-serde.workspace = true -lazy_static.workspace = true opendal = { version = "0.59.2", features = [ "layers-tracing", "layers-prometheus", @@ -43,14 +38,16 @@ reqwest.workspace = true serde.workspace = true snafu.workspace = true tokio.workspace = true -uuid.workspace = true +uuid = { workspace = true, optional = true } [dev-dependencies] anyhow = "1.0" common-telemetry.workspace = true common-test-util.workspace = true +datafusion_object_store.workspace = true +futures.workspace = true +object-store = { workspace = true, features = ["testing"] } object_store_opendal.workspace = true -rand.workspace = true tempfile.workspace = true tokio.workspace = true toml.workspace = true diff --git a/src/object-store/src/lib.rs b/src/object-store/src/lib.rs index 05c344485c6..2de2b570082 100644 --- a/src/object-store/src/lib.rs +++ b/src/object-store/src/lib.rs @@ -26,6 +26,7 @@ pub mod layers; pub mod manager; mod metrics; pub mod secure_fs; +#[cfg(feature = "testing")] pub mod test_util; pub mod util; diff --git a/src/operator/Cargo.toml b/src/operator/Cargo.toml index ea1bc990f74..72f953e0f89 100644 --- a/src/operator/Cargo.toml +++ b/src/operator/Cargo.toml @@ -22,7 +22,6 @@ workspace = true ahash.workspace = true api.workspace = true arrow.workspace = true -arrow-ipc.workspace = true async-stream.workspace = true async-trait.workspace = true bytes.workspace = true @@ -66,7 +65,6 @@ object-store.workspace = true partition.workspace = true percent-encoding.workspace = true prometheus.workspace = true -prost.workspace = true prost-types = { workspace = true, optional = true } query.workspace = true regex.workspace = true diff --git a/src/partition/Cargo.toml b/src/partition/Cargo.toml index a8e3a8ae11a..6773cc5878b 100644 --- a/src/partition/Cargo.toml +++ b/src/partition/Cargo.toml @@ -9,18 +9,14 @@ workspace = true [dependencies] api.workspace = true -async-trait.workspace = true common-base.workspace = true common-error.workspace = true common-macro.workspace = true common-meta.workspace = true -common-query.workspace = true -common-telemetry.workspace = true datafusion-common.workspace = true datafusion-expr.workspace = true datafusion-physical-expr.workspace = true datatypes.workspace = true -itertools.workspace = true moka = { workspace = true, features = ["future"] } serde.workspace = true serde_json.workspace = true diff --git a/src/pipeline/Cargo.toml b/src/pipeline/Cargo.toml index 620ac36ac77..9a355f4c8dc 100644 --- a/src/pipeline/Cargo.toml +++ b/src/pipeline/Cargo.toml @@ -14,28 +14,21 @@ ahash.workspace = true api.workspace = true arrow.workspace = true arrow-schema.workspace = true -async-trait.workspace = true catalog.workspace = true chrono.workspace = true chrono-tz.workspace = true common-catalog.workspace = true common-error.workspace = true -common-function.workspace = true common-macro.workspace = true common-meta.workspace = true common-query.workspace = true common-recordbatch.workspace = true -common-runtime.workspace = true common-telemetry.workspace = true common-time.workspace = true -crossbeam-utils.workspace = true csv = "1.3" -dashmap.workspace = true datafusion.workspace = true datafusion-common.workspace = true datafusion-expr.workspace = true -datafusion-functions.workspace = true -datafusion-physical-expr.workspace = true datatypes.workspace = true dyn-fmt = "0.4" enum_dispatch = "0.3" @@ -50,7 +43,6 @@ moka = { workspace = true, features = ["future"] } once_cell.workspace = true operator.workspace = true ordered-float.workspace = true -paste.workspace = true prometheus.workspace = true query.workspace = true regex.workspace = true @@ -68,7 +60,6 @@ yaml-rust = "0.4" [dev-dependencies] catalog = { workspace = true, features = ["testing"] } criterion = { workspace = true, features = ["html_reports"] } -rayon = "1.0" session = { workspace = true, features = ["testing"] } [[bench]] diff --git a/src/query/Cargo.toml b/src/query/Cargo.toml index 480c95a5adb..e81d93b7781 100644 --- a/src/query/Cargo.toml +++ b/src/query/Cargo.toml @@ -16,7 +16,6 @@ vector_index = [] [dependencies] ahash.workspace = true api.workspace = true -arc-swap = "1.0" arrow.workspace = true arrow-schema.workspace = true async-recursion = "1.0" @@ -50,11 +49,9 @@ datafusion-physical-expr.workspace = true datafusion-proto.workspace = true datafusion-sql.workspace = true datatypes.workspace = true -either.workspace = true futures.workspace = true futures-util.workspace = true greptime-proto.workspace = true -humantime.workspace = true itertools.workspace = true jsonb.workspace = true lazy_static.workspace = true @@ -63,7 +60,6 @@ meter-core.workspace = true meter-macros.workspace = true object-store.workspace = true once_cell.workspace = true -parking_lot.workspace = true partition.workspace = true prometheus.workspace = true promql.workspace = true @@ -93,13 +89,10 @@ common-query = { workspace = true, features = ["testing"] } fastrand = "2.0" moka = { workspace = true, features = ["future"] } nalgebra.workspace = true -num = "0.4" -num-traits = "0.2" paste.workspace = true pretty_assertions.workspace = true rand.workspace = true session = { workspace = true, features = ["testing"] } store-api.workspace = true table = { workspace = true, features = ["testing"] } -tokio-stream.workspace = true toml.workspace = true diff --git a/src/servers/Cargo.toml b/src/servers/Cargo.toml index 0e2d896db8c..06648254a06 100644 --- a/src/servers/Cargo.toml +++ b/src/servers/Cargo.toml @@ -40,7 +40,6 @@ common-batcher.workspace = true common-catalog.workspace = true common-decimal.workspace = true common-error.workspace = true -common-frontend.workspace = true common-grpc.workspace = true common-macro.workspace = true common-mem-prof = { workspace = true, optional = true } @@ -58,7 +57,6 @@ common-telemetry.workspace = true common-time.workspace = true common-version = { workspace = true, features = ["codec"] } csv = "1.3" -dashmap.workspace = true datafusion.workspace = true datafusion-common.workspace = true datafusion-expr.workspace = true @@ -123,7 +121,6 @@ serde.workspace = true serde_json.workspace = true session.workspace = true simd-json.workspace = true -simdutf8 = "0.1" smallvec.workspace = true snafu.workspace = true snap = "1" @@ -169,6 +166,7 @@ rand.workspace = true serde_json.workspace = true servers = { workspace = true, features = ["testing"] } session = { workspace = true, features = ["testing"] } +simdutf8 = "0.1" table.workspace = true tempfile = "3.0.0" tokio = { workspace = true, features = ["test-util"] } diff --git a/src/session/Cargo.toml b/src/session/Cargo.toml index 889b911c3cc..01ef368f1f0 100644 --- a/src/session/Cargo.toml +++ b/src/session/Cargo.toml @@ -29,3 +29,9 @@ derive_more.workspace = true snafu.workspace = true sql.workspace = true uuid.workspace = true + +[package.metadata.cargo-shear] +# `common-error` is required by `common-macro`'s `stack_trace_debug`: the +# generated code references `::common_error` paths, so it must stay a direct +# dependency. +ignored = ["common-error"] diff --git a/src/sql/Cargo.toml b/src/sql/Cargo.toml index 1ab966fc7e3..685d96336b1 100644 --- a/src/sql/Cargo.toml +++ b/src/sql/Cargo.toml @@ -16,25 +16,19 @@ arrow-buffer.workspace = true chrono.workspace = true common-base.workspace = true common-catalog.workspace = true -common-decimal.workspace = true common-error.workspace = true common-macro.workspace = true common-query.workspace = true common-sql.workspace = true -common-telemetry.workspace = true common-time.workspace = true datafusion.workspace = true datafusion-common.workspace = true datafusion-expr.workspace = true -datafusion-physical-expr.workspace = true datafusion-sql.workspace = true datatypes.workspace = true either.workspace = true -hex = "0.4" -humantime.workspace = true iso8601 = "0.6.1" itertools.workspace = true -jsonb.workspace = true lazy_static.workspace = true promql-parser.workspace = true regex.workspace = true @@ -43,7 +37,6 @@ serde_json.workspace = true snafu.workspace = true sqlparser.workspace = true sqlparser_derive = "0.1" -store-api.workspace = true table.workspace = true uuid.workspace = true diff --git a/src/store-api/Cargo.toml b/src/store-api/Cargo.toml index 2e5a203912e..4a3616afb8a 100644 --- a/src/store-api/Cargo.toml +++ b/src/store-api/Cargo.toml @@ -36,15 +36,10 @@ prost.workspace = true serde.workspace = true serde_json.workspace = true snafu.workspace = true -sqlparser.workspace = true strum.workspace = true tokio.workspace = true uuid.workspace = true [dev-dependencies] -async-stream.workspace = true common-meta.workspace = true serde_json.workspace = true - -[package.metadata.cargo-udeps.ignore] -normal = ["aquamarine"] diff --git a/src/table/Cargo.toml b/src/table/Cargo.toml index 8ecc51a320c..357d005e612 100644 --- a/src/table/Cargo.toml +++ b/src/table/Cargo.toml @@ -34,13 +34,11 @@ derive_builder.workspace = true futures.workspace = true greptime-proto.workspace = true humantime.workspace = true -humantime-serde.workspace = true lazy_static.workspace = true once_cell.workspace = true paste.workspace = true serde.workspace = true snafu.workspace = true -sqlparser.workspace = true store-api.workspace = true tokio.workspace = true @@ -48,4 +46,3 @@ tokio.workspace = true common-test-util.workspace = true parquet = { workspace = true, features = ["async"] } serde_json.workspace = true -tokio-util = { version = "0.7", features = ["compat"] } diff --git a/tests-fuzz/Cargo.toml b/tests-fuzz/Cargo.toml index 2cd66b8afe5..1189f53c42f 100644 --- a/tests-fuzz/Cargo.toml +++ b/tests-fuzz/Cargo.toml @@ -188,3 +188,12 @@ path = "utils/kafka_wal_helper.rs" test = false bench = false doc = false + +[package.metadata.cargo-shear] +# `common-error` is required by `common-macro`'s `stack_trace_debug`: the +# generated code references `::common_error` paths, so it must stay a direct +# dependency. +# `k8s-openapi` is a feature-pinning dependency: it enables the `v1_30` +# version feature required by the transitive `kube` -> `k8s-openapi` chain, +# but is never referenced by name in source code. +ignored = ["common-error", "k8s-openapi"] diff --git a/tests-integration/Cargo.toml b/tests-integration/Cargo.toml index a5ac94924e4..a3d694f758b 100644 --- a/tests-integration/Cargo.toml +++ b/tests-integration/Cargo.toml @@ -20,16 +20,11 @@ workspace = true [dependencies] api.workspace = true arrow-flight.workspace = true -async-stream.workspace = true async-trait.workspace = true auth.workspace = true axum.workspace = true -base64.workspace = true cache.workspace = true catalog.workspace = true -chrono.workspace = true -clap.workspace = true -cli.workspace = true client.workspace = true cmd.workspace = true common-base.workspace = true @@ -38,16 +33,12 @@ common-config.workspace = true common-datasource.workspace = true common-error.workspace = true common-event-recorder.workspace = true -common-frontend.workspace = true common-grpc.workspace = true -common-memory-manager.workspace = true common-meta.workspace = true -common-options.workspace = true common-procedure.workspace = true common-query.workspace = true common-recordbatch.workspace = true common-runtime.workspace = true -common-stat.workspace = true common-telemetry.workspace = true common-test-util.workspace = true common-time.workspace = true @@ -55,62 +46,59 @@ common-wal.workspace = true datanode = { workspace = true } datatypes.workspace = true dotenv.workspace = true -flate2 = "1.0" flow.workspace = true frontend.workspace = true futures.workspace = true futures-util.workspace = true http.workspace = true hyper-util = { workspace = true, features = ["tokio"] } -log-query = { workspace = true } -loki-proto.workspace = true meta-client.workspace = true meta-srv = { workspace = true, features = ["mock"] } mito2.workspace = true object-store.workspace = true operator.workspace = true -plugins.workspace = true prost.workspace = true query.workspace = true rand.workspace = true serde_json.workspace = true servers.workspace = true session.workspace = true -snafu.workspace = true sql.workspace = true -sqlx = { workspace = true, features = [ - "mysql", - "postgres", - "chrono", -] } standalone.workspace = true store-api.workspace = true substrait.workspace = true table.workspace = true tempfile.workspace = true -time = "0.3" tokio.workspace = true tokio-stream = { workspace = true, features = ["net"] } tonic.workspace = true tower.workspace = true uuid.workspace = true -zstd.workspace = true [dev-dependencies] +base64.workspace = true catalog = { workspace = true, features = ["testing"] } +chrono.workspace = true +clap.workspace = true cli.workspace = true client = { workspace = true, features = ["testing"] } +common-frontend.workspace = true +common-memory-manager.workspace = true common-meta = { workspace = true, features = ["testing"] } +common-options.workspace = true common-procedure-test.workspace = true common-wal = { workspace = true, features = ["testing"] } datafusion.workspace = true datafusion-expr.workspace = true either.workspace = true +flate2 = "1.0" frontend = { workspace = true, features = ["testing"] } hex.workspace = true http.workspace = true itertools.workspace = true jsonb.workspace = true +log-query.workspace = true +loki-proto.workspace = true mito2 = { workspace = true, features = ["test", "test-shared-fs-region-migration"] } mysql_async = { version = "0.37", default-features = false, features = [ "time", @@ -119,9 +107,9 @@ mysql_async = { version = "0.37", default-features = false, features = [ opentelemetry-proto.workspace = true operator = { workspace = true, features = ["testing"] } otel-arrow-rust.workspace = true -partition.workspace = true paste.workspace = true pipeline.workspace = true +plugins.workspace = true prost.workspace = true rand.workspace = true rskafka.workspace = true @@ -131,9 +119,16 @@ sea-query.workspace = true servers = { workspace = true, features = ["testing"] } session = { workspace = true, features = ["testing"] } similar-asserts.workspace = true +sqlx = { workspace = true, features = [ + "mysql", + "postgres", + "chrono", +] } store-api.workspace = true +time = "0.3" tokio-postgres = { workspace = true } tokio-util.workspace = true url = "2.3" urlencoding = "2.1" yaml-rust = "0.4" +zstd.workspace = true diff --git a/tests/runner/Cargo.toml b/tests/runner/Cargo.toml index 2941e85fa5a..35de1443aa8 100644 --- a/tests/runner/Cargo.toml +++ b/tests/runner/Cargo.toml @@ -13,7 +13,6 @@ clap.workspace = true client.workspace = true common-error.workspace = true common-query.workspace = true -common-recordbatch.workspace = true common-time.workspace = true datatypes = { workspace = true } flate2 = "1.0" From a35f9d5c782ddccfe7c94fe9c7f291a6b7910518 Mon Sep 17 00:00:00 2001 From: Weny Xu Date: Wed, 23 Sep 2026 05:16:41 +0000 Subject: [PATCH 4/4] fix: address Windows test failures and run full Windows CI (#9305) * fix: use relative object keys for Windows filesystem access Signed-off-by: WenyXu * fix: cover Windows path and time limits in full test CI Signed-off-by: WenyXu * fix: use relative keys in metadata snapshot tests Signed-off-by: WenyXu --------- Signed-off-by: WenyXu --- .github/workflows/rust.yml | 42 ++++ src/cli/src/metadata/snapshot.rs | 218 ++++++++++++++---- src/common/datasource/src/file_format/csv.rs | 19 +- src/common/datasource/src/file_format/json.rs | 11 +- .../datasource/src/file_format/tests.rs | 28 +-- src/common/datasource/src/test_util.rs | 31 +-- src/common/datasource/src/tests.rs | 25 +- src/common/meta/src/snapshot.rs | 40 +--- src/mito2/src/cache/write_cache.rs | 3 +- src/query/src/promql/label_values.rs | 29 ++- 10 files changed, 303 insertions(+), 143 deletions(-) diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 1326ab5fd3d..5778d3856bb 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -148,6 +148,48 @@ jobs: env: RUSTFLAGS: "-D warnings" + test-windows: + if: ${{ github.repository == 'GreptimeTeam/greptimedb' && (github.event_name != 'pull_request' || github.event.pull_request.draft == false) }} + name: Test Windows + runs-on: windows-2022 + timeout-minutes: 90 + env: + CARGO_INCREMENTAL: 0 + CARGO_PROFILE_DEV_DEBUG: 0 + CARGO_PROFILE_TEST_DEBUG: 0 + RUSTFLAGS: "-D warnings -C linker=lld-link" + RUST_BACKTRACE: 1 + RUST_MIN_STACK: 8388608 # 8MB + steps: + - run: git config --global core.autocrlf false + - uses: actions/checkout@v4 + with: + persist-credentials: false + - uses: ./.github/actions/setup-cyborg + - uses: arduino/setup-protoc@v3 + with: + repo-token: ${{ secrets.GITHUB_TOKEN }} + - uses: KyleMayes/install-llvm-action@v1 + with: + version: "14.0" + - uses: actions-rust-lang/setup-rust-toolchain@v1 + with: + cache: false + - name: Rust Cache + uses: Swatinem/rust-cache@v2 + with: + shared-key: "test-windows" + cache-all-crates: "true" + save-if: ${{ github.ref == 'refs/heads/main' }} + - name: Install Cargo Nextest + uses: taiki-e/install-action@nextest + - name: Install WSL distribution + uses: Vampire/setup-wsl@v2 + with: + distribution: Ubuntu-22.04 + - name: Run Windows tests + run: cargo nextest run --locked --workspace -F dashboard --no-fail-fast + conflict-check: if: ${{ github.repository == 'GreptimeTeam/greptimedb' && (github.event_name != 'pull_request' || github.event.pull_request.draft == false) }} name: Check for conflict diff --git a/src/cli/src/metadata/snapshot.rs b/src/cli/src/metadata/snapshot.rs index 59b2599ad9c..9c010c2235b 100644 --- a/src/cli/src/metadata/snapshot.rs +++ b/src/cli/src/metadata/snapshot.rs @@ -12,11 +12,13 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::path::{Component, Path, PathBuf}; + use async_trait::async_trait; use clap::{Parser, Subcommand}; use common_error::ext::BoxedError; use common_meta::snapshot::MetadataSnapshotManager; -use object_store::{ObjectStore, services}; +use object_store::ObjectStore; use crate::Tool; use crate::common::{ObjectStoreConfig, StoreConfig, new_fs_object_store}; @@ -270,25 +272,44 @@ fn build_object_store_and_resolve_file_path( fs_root: &str, file_path: &str, ) -> Result<(ObjectStore, String), BoxedError> { - let object_store = object_store.build().map_err(BoxedError::new)?; - let object_store = match object_store { - Some(object_store) => object_store, - None => new_fs_object_store(fs_root)?, - }; + if let Some(object_store) = object_store.build().map_err(BoxedError::new)? { + return Ok((object_store, file_path.to_string())); + } - let file_path = if object_store.info().scheme() == services::FS_SCHEME { - resolve_relative_path_with_current_dir(file_path).map_err(BoxedError::new)? + let absolute_path = + resolve_relative_path_with_current_dir(file_path).map_err(BoxedError::new)?; + let (root, key) = split_fs_snapshot_path(fs_root, &absolute_path); + Ok((new_fs_object_store(&root)?, key)) +} + +/// Separates a local absolute path into an operator root and a relative key. +fn split_fs_snapshot_path(fs_root: &str, absolute_path: &str) -> (String, String) { + let path = Path::new(absolute_path); + // OpenDAL keys cannot contain a Windows drive/UNC prefix or a root directory. + // Keep the filesystem anchor in the operator root, not in the object key. + let anchor: PathBuf = path + .components() + .take_while(|c| matches!(c, Component::Prefix(_) | Component::RootDir)) + .collect(); + let key = path + .components() + .filter(|c| !matches!(c, Component::Prefix(_) | Component::RootDir)) + .map(|c| c.as_os_str().to_string_lossy()) + .collect::>() + .join("/"); + // Preserve the existing custom-root semantics: cwd-resolved keys are rooted + // under --dir. The default root follows the input path's drive on Windows. + let root = if fs_root == "/" { + anchor.to_string_lossy() } else { - file_path.to_string() + fs_root.into() }; - - Ok((object_store, file_path)) + (root.into_owned(), key) } #[cfg(test)] mod tests { use std::env; - use std::path::Path; use std::sync::Arc; use std::time::Duration; @@ -311,44 +332,159 @@ mod tests { #[tokio::test] async fn test_cmd_resolve_file_path() { - common_telemetry::init_default_ut_logging(); - let cmd = RestoreCommand::parse_from([ - "", - "--file_name", - "metadata_snapshot.metadata.fb", - "--backend", - "memory-store", - "--store-addrs", - "memory://", - ]); - let tool = cmd.build().await.unwrap(); let current_dir = env::current_dir().unwrap(); - let file_path = current_dir.join("metadata_snapshot.metadata.fb"); - assert_eq!(tool.file_path, file_path.to_string_lossy().to_string()); + for file_path in [ + "metadata_snapshot.metadata.fb".to_string(), + current_dir + .join("metadata_snapshot.metadata.fb") + .display() + .to_string(), + ] { + let cmd = RestoreCommand::parse_from([ + "", + "--file_name", + &file_path, + "--backend", + "memory-store", + "--store-addrs", + "memory://", + ]); + let tool = cmd.build().await.unwrap(); + assert!(!Path::new(&tool.file_path).is_absolute()); + assert!( + Path::new(&tool.file_path) + .components() + .all(|c| matches!(c, Component::Normal(_))) + ); + let expected = current_dir + .components() + .filter_map(|c| match c { + Component::Normal(part) => Some(part.to_str().unwrap()), + _ => None, + }) + .chain(["metadata_snapshot.metadata.fb"]) + .collect::>() + .join("/"); + assert_eq!(tool.file_path, expected); + } + } + #[cfg(windows)] + #[test] + fn test_split_windows_snapshot_path() { + for (path, root, key) in [ + ( + r"C:\backup\snapshot.metadata.fb", + r"C:\", + "backup/snapshot.metadata.fb", + ), + ( + r"D:\备份 dir\snapshot.metadata.fb", + r"D:\", + "备份 dir/snapshot.metadata.fb", + ), + ( + r"\\server\share\backup\snapshot.metadata.fb", + r"\\server\share\", + "backup/snapshot.metadata.fb", + ), + ( + r"\\?\C:\backup\snapshot.metadata.fb", + r"\\?\C:\", + "backup/snapshot.metadata.fb", + ), + ] { + assert_eq!( + split_fs_snapshot_path("/", path), + (root.to_string(), key.to_string()) + ); + assert_eq!( + split_fs_snapshot_path(r"D:\root", path), + (r"D:\root".to_string(), key.to_string()) + ); + } + } + + #[tokio::test] + async fn test_fs_snapshot_paths() { + let temp_dir = tempfile::tempdir().unwrap(); + for name in ["snapshot.metadata.fb", "nested dir/快照.metadata.fb"] { + let path = temp_dir.path().join(name); + let (store, key) = build_object_store_and_resolve_file_path( + ObjectStoreConfig::default(), + "/", + path.to_str().unwrap(), + ) + .unwrap(); + assert!(!path.exists()); + store.write(&key, "snapshot").await.unwrap(); + assert_eq!(std::fs::read(&path).unwrap(), b"snapshot"); + assert_eq!(store.read(&key).await.unwrap().to_bytes(), &b"snapshot"[..]); + } + } + + #[tokio::test] + async fn test_fs_snapshot_custom_root() { + let temp_dir = tempfile::tempdir().unwrap(); + let (store, key) = build_object_store_and_resolve_file_path( + ObjectStoreConfig::default(), + temp_dir.path().to_str().unwrap(), + "snapshot.metadata.fb", + ) + .unwrap(); + let current_dir = env::current_dir().unwrap(); + let expected = current_dir + .components() + .filter_map(|c| match c { + Component::Normal(part) => Some(part.to_str().unwrap()), + _ => None, + }) + .chain(["snapshot.metadata.fb"]) + .collect::>() + .join("/"); + assert_eq!(key, expected); + store.write(&key, "snapshot").await.unwrap(); + assert_eq!( + std::fs::read(temp_dir.path().join(expected)).unwrap(), + b"snapshot" + ); + } + + #[tokio::test] + async fn test_fs_snapshot_rejects_parent_traversal() { + let temp_dir = tempfile::tempdir().unwrap(); + let path = temp_dir.path().join("../snapshot.metadata.fb"); + let (store, key) = build_object_store_and_resolve_file_path( + ObjectStoreConfig::default(), + "/", + path.to_str().unwrap(), + ) + .unwrap(); + assert!(key.split('/').any(|part| part == "..")); + let err = store.write(&key, "snapshot").await.unwrap_err(); + assert_eq!(err.kind(), object_store::ErrorKind::NotFound); + assert!(err.to_string().contains("path escapes the configured root")); + } + + #[test] + fn test_remote_snapshot_key_unchanged() { let cmd = RestoreCommand::parse_from([ "", - "--file_name", - "metadata_snapshot.metadata.fb", "--backend", "memory-store", "--store-addrs", "memory://", + "--s3", + "--s3-bucket", + "snapshots", + "--s3-region", + "us-east-1", ]); - let tool = cmd.build().await.unwrap(); - assert_eq!(tool.file_path, file_path.to_string_lossy().to_string()); - - let cmd = RestoreCommand::parse_from([ - "", - "--file_name", - "metadata_snapshot.metadata.fb", - "--backend", - "memory-store", - "--store-addrs", - "memory://", - ]); - let tool = cmd.build().await.unwrap(); - assert_eq!(tool.file_path, file_path.to_string_lossy().to_string()); + let key = "nested/snapshot.metadata.fb"; + let (store, resolved) = + build_object_store_and_resolve_file_path(cmd.object_store, "/ignored", key).unwrap(); + assert_eq!(store.info().scheme(), object_store::services::S3_SCHEME); + assert_eq!(resolved, key); } async fn setup_backup_file(object_store: ObjectStore, file_path: &str) { diff --git a/src/common/datasource/src/file_format/csv.rs b/src/common/datasource/src/file_format/csv.rs index d016d4eabe9..d4cfb45088c 100644 --- a/src/common/datasource/src/file_format/csv.rs +++ b/src/common/datasource/src/file_format/csv.rs @@ -647,16 +647,15 @@ mod tests { let compressed_file_name = format!("test_compressed_csv.{}", compression_type.file_extension()); let compressed_file_path = temp_dir.path().join(&compressed_file_name); - let compressed_file_path_str = compressed_file_path.to_str().unwrap(); // Create a simple file store for testing - let store = test_store("/"); + let store = test_store(temp_dir.path().to_str().unwrap()); // Export CSV with compression let rows = stream_to_csv( Box::pin(DfRecordBatchStreamAdapter::new(recordbatches.as_stream())), store, - compressed_file_path_str, + &compressed_file_name, 1024, 1, &format, @@ -708,13 +707,13 @@ mod tests { } // Verify the compressed file can be decompressed and content matches original data - let store = test_store("/"); + let store = test_store(temp_dir.path().to_str().unwrap()); let schema = Arc::new( CsvFormat { compression_type, ..Default::default() } - .infer_schema(&store, compressed_file_path_str) + .infer_schema(&store, &compressed_file_name) .await .unwrap(), ); @@ -722,7 +721,7 @@ mod tests { let stream = file_to_stream( &store, - compressed_file_path_str, + &compressed_file_name, csv_source.clone(), None, compression_type, @@ -761,13 +760,13 @@ mod tests { ) .unwrap(); - let store = test_store("/"); + let store = test_store(temp_dir.path().to_str().unwrap()); let schema = Arc::new(arrow_schema::Schema::new(vec![ Field::new("id", DataType::UInt32, false), Field::new("name", DataType::Utf8, false), Field::new("value", DataType::Float64, false), ])); - let path = csv_file_path.to_str().unwrap(); + let path = "input.csv"; let stream = tolerant_csv_stream(&store, path, schema, vec![0, 1, 2], &CsvFormat::default()) @@ -793,13 +792,13 @@ mod tests { let csv_file_path = temp_dir.path().join("input.csv"); std::fs::write(&csv_file_path, "id,name,value\n1,Alice,10.5\n2,Bob\n").unwrap(); - let store = test_store("/"); + let store = test_store(temp_dir.path().to_str().unwrap()); let schema = Arc::new(arrow_schema::Schema::new(vec![ Field::new("id", DataType::UInt32, false), Field::new("name", DataType::Utf8, false), Field::new("value", DataType::Float64, false), ])); - let path = csv_file_path.to_str().unwrap(); + let path = "input.csv"; let stream = tolerant_csv_stream(&store, path, schema, vec![0, 1, 2], &CsvFormat::default()) diff --git a/src/common/datasource/src/file_format/json.rs b/src/common/datasource/src/file_format/json.rs index 3c24a4b7e27..23ce461fcb2 100644 --- a/src/common/datasource/src/file_format/json.rs +++ b/src/common/datasource/src/file_format/json.rs @@ -313,16 +313,15 @@ mod tests { let compressed_file_name = format!("test_compressed_json.{}", compression_type.file_extension()); let compressed_file_path = temp_dir.path().join(&compressed_file_name); - let compressed_file_path_str = compressed_file_path.to_str().unwrap(); // Create a simple file store for testing - let store = test_store("/"); + let store = test_store(temp_dir.path().to_str().unwrap()); // Export JSON with compression let rows = stream_to_json( Box::pin(DfRecordBatchStreamAdapter::new(recordbatches.as_stream())), store, - compressed_file_path_str, + &compressed_file_name, 1024, 1, &format, @@ -374,13 +373,13 @@ mod tests { } // Verify the compressed file can be decompressed and content matches original data - let store = test_store("/"); + let store = test_store(temp_dir.path().to_str().unwrap()); let schema = Arc::new( JsonFormat { compression_type, ..Default::default() } - .infer_schema(&store, compressed_file_path_str) + .infer_schema(&store, &compressed_file_name) .await .unwrap(), ); @@ -388,7 +387,7 @@ mod tests { let stream = file_to_stream( &store, - compressed_file_path_str, + &compressed_file_name, json_source.clone(), None, compression_type, diff --git a/src/common/datasource/src/file_format/tests.rs b/src/common/datasource/src/file_format/tests.rs index 27b86e1b114..14be2b1777f 100644 --- a/src/common/datasource/src/file_format/tests.rs +++ b/src/common/datasource/src/file_format/tests.rs @@ -66,13 +66,12 @@ impl Test<'_> { #[tokio::test] async fn test_json_opener() { - let store = test_store("/"); + let root = find_workspace_path("/src/common/datasource/tests"); + let store = test_store(root.to_str().unwrap()); let schema = basic_schema_with_time_format(); let file_source = Arc::new(JsonSource::new(schema)).with_batch_size(test_util::TEST_BATCH_SIZE); - let path = &find_workspace_path("/src/common/datasource/tests/json/basic.json") - .display() - .to_string(); + let path = "json/basic.json"; let tests = [ Test { @@ -108,11 +107,10 @@ async fn test_json_opener() { #[tokio::test] async fn test_csv_opener() { - let store = test_store("/"); + let root = find_workspace_path("/src/common/datasource/tests"); + let store = test_store(root.to_str().unwrap()); let schema = basic_schema_with_time_format(); - let path = &find_workspace_path("/src/common/datasource/tests/csv/basic.csv") - .display() - .to_string(); + let path = "csv/basic.csv"; let file_source = CsvSource::new(schema).with_batch_size(test_util::TEST_BATCH_SIZE); @@ -150,13 +148,12 @@ async fn test_csv_opener() { #[tokio::test(flavor = "multi_thread")] async fn test_parquet_exec() { - let store = test_store("/"); + let root = find_workspace_path("/src/common/datasource/tests"); + let store = test_store(root.to_str().unwrap()); let schema = test_basic_schema(); - let path = &find_workspace_path("/src/common/datasource/tests/parquet/basic.parquet") - .display() - .to_string(); + let path = "parquet/basic.parquet"; let parquet_source = ParquetSource::new(schema) .with_parquet_file_reader_factory(Arc::new(DefaultParquetFileReaderFactory::new(store))); @@ -191,11 +188,10 @@ async fn test_parquet_exec() { #[tokio::test] async fn test_orc_opener() { - let path = &find_workspace_path("/src/common/datasource/tests/orc/test.orc") - .display() - .to_string(); + let path = "orc/test.orc"; - let store = test_store("/"); + let root = find_workspace_path("/src/common/datasource/tests"); + let store = test_store(root.to_str().unwrap()); let schema = Arc::new(OrcFormat.infer_schema(&store, path).await.unwrap()); let file_source = Arc::new(OrcSource::new(schema.into())); diff --git a/src/common/datasource/src/test_util.rs b/src/common/datasource/src/test_util.rs index 4cefda88890..f95d24930b2 100644 --- a/src/common/datasource/src/test_util.rs +++ b/src/common/datasource/src/test_util.rs @@ -56,8 +56,8 @@ pub fn test_store(root: &str) -> ObjectStore { pub fn test_tmp_store(root: &str) -> (ObjectStore, TempDir) { let dir = create_temp_dir(root); - let builder = Fs::default(); - (ObjectStore::new(builder.root("/")).unwrap(), dir) + let store = test_store(dir.path().to_str().unwrap()); + (store, dir) } pub fn test_basic_schema() -> SchemaRef { @@ -94,9 +94,11 @@ pub(crate) fn scan_config( .build() } -pub async fn setup_stream_to_json_test(origin_path: &str, threshold: impl Fn(usize) -> usize) { - let store = test_store("/"); - +pub async fn setup_stream_to_json_test( + store: &ObjectStore, + origin_path: &str, + threshold: impl Fn(usize) -> usize, +) { let schema = basic_schema_with_time_format(); let json_opener = JsonOpener::new( @@ -117,9 +119,9 @@ pub async fn setup_stream_to_json_test(origin_path: &str, threshold: impl Fn(usi .build() .unwrap(); - let (tmp_store, dir) = test_tmp_store("test_stream_to_json"); + let (tmp_store, _dir) = test_tmp_store("test_stream_to_json"); - let output_path = format!("{}/{}", dir.path().display(), "output"); + let output_path = "output"; let json_format = JsonFormat::default(); @@ -127,7 +129,7 @@ pub async fn setup_stream_to_json_test(origin_path: &str, threshold: impl Fn(usi stream_to_json( Box::pin(stream), tmp_store.clone(), - &output_path, + output_path, threshold(size), 8, &json_format, @@ -136,18 +138,17 @@ pub async fn setup_stream_to_json_test(origin_path: &str, threshold: impl Fn(usi .is_ok() ); - let written = tmp_store.read(&output_path).await.unwrap(); + let written = tmp_store.read(output_path).await.unwrap(); let origin = store.read(origin_path).await.unwrap(); assert_eq_lines(written.to_vec(), origin.to_vec()); } pub async fn setup_stream_to_csv_test( + store: &ObjectStore, origin_path: &str, format_path: &str, threshold: impl Fn(usize) -> usize, ) { - let store = test_store("/"); - let schema = basic_schema_with_time_format(); let csv_source = CsvSource::new(schema).with_batch_size(TEST_BATCH_SIZE); @@ -168,9 +169,9 @@ pub async fn setup_stream_to_csv_test( .build() .unwrap(); - let (tmp_store, dir) = test_tmp_store("test_stream_to_csv"); + let (tmp_store, _dir) = test_tmp_store("test_stream_to_csv"); - let output_path = format!("{}/{}", dir.path().display(), "output"); + let output_path = "output"; let csv_format = CsvFormat { timestamp_format: Some("%m-%d-%Y".to_string()), @@ -183,7 +184,7 @@ pub async fn setup_stream_to_csv_test( stream_to_csv( Box::pin(stream), tmp_store.clone(), - &output_path, + output_path, threshold(size), 8, &csv_format, @@ -192,7 +193,7 @@ pub async fn setup_stream_to_csv_test( .is_ok() ); - let written = tmp_store.read(&output_path).await.unwrap(); + let written = tmp_store.read(output_path).await.unwrap(); let format_expect = store.read(format_path).await.unwrap(); assert_eq_lines(written.to_vec(), format_expect.to_vec()); } diff --git a/src/common/datasource/src/tests.rs b/src/common/datasource/src/tests.rs index 8be9add7c88..b983ef86627 100644 --- a/src/common/datasource/src/tests.rs +++ b/src/common/datasource/src/tests.rs @@ -18,34 +18,31 @@ use crate::test_util; #[tokio::test] async fn test_stream_to_json() { - let origin_path = &find_workspace_path("/src/common/datasource/tests/json/basic.json") - .display() - .to_string(); + let root = find_workspace_path("/src/common/datasource/tests/json"); + let store = test_util::test_store(root.to_str().unwrap()); + let origin_path = "basic.json"; // A small threshold // Triggers the flush each writes - test_util::setup_stream_to_json_test(origin_path, |size| size / 2).await; + test_util::setup_stream_to_json_test(&store, origin_path, |size| size / 2).await; // A large threshold // Only triggers the flush at last - test_util::setup_stream_to_json_test(origin_path, |size| size * 2).await; + test_util::setup_stream_to_json_test(&store, origin_path, |size| size * 2).await; } #[tokio::test] async fn test_stream_to_csv() { - let origin_path = &find_workspace_path("/src/common/datasource/tests/csv/basic.csv") - .display() - .to_string(); - - let format_path = &find_workspace_path("/src/common/datasource/tests/csv/basic_format.csv") - .display() - .to_string(); + let root = find_workspace_path("/src/common/datasource/tests/csv"); + let store = test_util::test_store(root.to_str().unwrap()); + let origin_path = "basic.csv"; + let format_path = "basic_format.csv"; // A small threshold // Triggers the flush each writes - test_util::setup_stream_to_csv_test(origin_path, format_path, |size| size / 2).await; + test_util::setup_stream_to_csv_test(&store, origin_path, format_path, |size| size / 2).await; // A large threshold // Only triggers the flush at last - test_util::setup_stream_to_csv_test(origin_path, format_path, |size| size * 2).await; + test_util::setup_stream_to_csv_test(&store, origin_path, format_path, |size| size * 2).await; } diff --git a/src/common/meta/src/snapshot.rs b/src/common/meta/src/snapshot.rs index f35c86e2e7b..3582b0acecb 100644 --- a/src/common/meta/src/snapshot.rs +++ b/src/common/meta/src/snapshot.rs @@ -408,7 +408,6 @@ mod tests { async fn test_dump_and_restore() { common_telemetry::init_default_ut_logging(); let (temp_dir, kv_backend, manager) = test_env("test_dump_and_restore"); - let temp_path = temp_dir.path(); for i in 0..10 { kv_backend @@ -420,32 +419,16 @@ mod tests { .await .unwrap(); } - let dump_path = temp_path.join("snapshot"); - manager - .dump(&format!( - "{}/metadata_snapshot", - &dump_path.as_path().display().to_string() - )) - .await - .unwrap(); + manager.dump("snapshot/metadata_snapshot").await.unwrap(); + let snapshot_key = "snapshot/metadata_snapshot.metadata.fb"; + assert!(temp_dir.path().join("data").join(snapshot_key).is_file()); // Clean up the kv backend kv_backend.clear(); - let err = manager - .dump(&format!( - "{}/metadata_snapshot.metadata.fb", - &dump_path.as_path().display().to_string() - )) - .await - .unwrap_err(); + let err = manager.dump(snapshot_key).await.unwrap_err(); assert_matches!(err, Error::Unexpected { .. }); assert!(err.to_string().contains("already exists")); - let restore_path = dump_path - .join("metadata_snapshot.metadata.fb") - .as_path() - .display() - .to_string(); - manager.restore(&restore_path).await.unwrap(); + manager.restore(snapshot_key).await.unwrap(); for i in 0..10 { let key = format!("test_{}", i); @@ -456,14 +439,11 @@ mod tests { #[tokio::test] async fn test_restore_from_nonexistent_file() { - let (temp_dir, _kv_backend, manager) = test_env("test_restore_from_nonexistent_file"); - let restore_path = temp_dir - .path() - .join("nonexistent.metadata.fb") - .as_path() - .display() - .to_string(); - let err = manager.restore(&restore_path).await.unwrap_err(); + let (_temp_dir, _kv_backend, manager) = test_env("test_restore_from_nonexistent_file"); + let err = manager + .restore("nonexistent.metadata.fb") + .await + .unwrap_err(); assert_matches!(err, Error::ReadObject { .. }) } } diff --git a/src/mito2/src/cache/write_cache.rs b/src/mito2/src/cache/write_cache.rs index 0e96c2f16a2..87ab4f0081a 100644 --- a/src/mito2/src/cache/write_cache.rs +++ b/src/mito2/src/cache/write_cache.rs @@ -790,7 +790,8 @@ mod tests { async fn test_read_metadata_from_write_cache() { common_telemetry::init_default_ut_logging(); let mut env = TestEnv::new().await; - let data_home = env.data_home().display().to_string(); + // Object keys are relative to the mock store root, not native paths. + let data_home = "write-cache-test".to_string(); let mock_store = env.init_object_store_manager(); let local_dir = create_temp_dir(""); diff --git a/src/query/src/promql/label_values.rs b/src/query/src/promql/label_values.rs index 44c173477bb..841666758e4 100644 --- a/src/query/src/promql/label_values.rs +++ b/src/query/src/promql/label_values.rs @@ -37,15 +37,15 @@ fn millis_since_epoch(time: SystemTime) -> Result { Ok(duration) => (duration.as_millis(), false), Err(earlier) => (earlier.duration().as_millis(), true), }; - let millis = i64::try_from(millis) - .ok() - .with_context(|| SystemTimeOutOfRangeSnafu { time })?; + let millis = + signed_millis(millis, before_epoch).with_context(|| SystemTimeOutOfRangeSnafu { time })?; + Ok(Timestamp::new_millisecond(millis)) +} - Ok(Timestamp::new_millisecond(if before_epoch { - -millis - } else { - millis - })) +/// Converts a millisecond magnitude and epoch direction without platform time limits. +fn signed_millis(millis: u128, before_epoch: bool) -> Option { + let millis = i64::try_from(millis).ok()?; + Some(if before_epoch { -millis } else { millis }) } fn build_time_filter(time_index_expr: Expr, start: Timestamp, end: Timestamp) -> Expr { @@ -139,7 +139,16 @@ mod tests { #[test] fn millis_beyond_i64_are_rejected() { - let time = UNIX_EPOCH + Duration::from_secs(1 << 60); - assert!(millis_since_epoch(time).is_err()); + let limit = i64::MAX as u128; + assert_eq!(signed_millis(limit, false), Some(i64::MAX)); + assert_eq!(signed_millis(limit, true), Some(-i64::MAX)); + assert_eq!(signed_millis(limit + 1, false), None); + assert_eq!(signed_millis(limit + 1, true), None); + + // Windows cannot represent this SystemTime. The conversion boundary is + // tested above on every platform; exercise the wrapper where possible. + if let Some(time) = UNIX_EPOCH.checked_add(Duration::from_millis(i64::MAX as u64 + 1)) { + assert!(millis_since_epoch(time).is_err()); + } } }