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

refactor(flow): trim eval offset compatibility scope

Browse Source 
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
discord9
2026-09-01 17:00:12 +08:00
parent 69cae1e069
commit 1f81fa7aa7
13 changed files with 9 additions and 516 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/catalog/src/system_schema/information_schema/flows.rs
+4 -71
View File
@@ -12,15 +12,11 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use std::collections::HashMap;
use std::sync::{Arc, Weak};
use common_catalog::consts::INFORMATION_SCHEMA_FLOW_TABLE_ID;
use common_error::ext::BoxedError;
use common_meta::ddl::create_flow::{
FlowType, INTERNAL_EVAL_OFFSET_KEY, INTERNAL_EVAL_SCHEDULE_KEY,
effective_eval_schedule_from_flow_info,
};
use common_meta::ddl::create_flow::{FlowType, effective_eval_schedule_from_flow_info};
use common_meta::key::FlowId;
use common_meta::key::flow::FlowMetadataManager;
use common_meta::key::flow::flow_info::FlowInfoValue;
@@ -180,16 +176,9 @@ impl InformationSchemaFlows {
.map(|schedule| schedule.anchor_secs)
.filter(|anchor_secs| *anchor_secs != 0),
comment,
// Defense-in-depth for `information_schema.flows.flow_definition`:
// never surface the internal transient transport keys, even if
// malformed/old metadata carries them in `options`.
flow_options: sql::statements::OptionMap::from_filtered_string_map(
flow_info.options(),
&[
FlowType::FLOW_TYPE_KEY,
INTERNAL_EVAL_OFFSET_KEY,
INTERNAL_EVAL_SCHEDULE_KEY,
],
&[FlowType::FLOW_TYPE_KEY],
),
query,
};
@@ -268,19 +257,6 @@ struct InformationSchemaFlowsBuilder {
flownode_addr_groups: StringVectorBuilder,
}
/// Filters the internal transient transport keys out of a flow's stored
/// options. `FlowInfoValue.options` should never contain them, but malformed
/// or old metadata might; `information_schema` must never surface them.
fn visible_flow_options(options: &HashMap<String, String>) -> HashMap<String, String> {
options
.iter()
.filter(|(key, _)| {
key.as_str() != INTERNAL_EVAL_OFFSET_KEY && key.as_str() != INTERNAL_EVAL_SCHEDULE_KEY
})
.map(|(key, value)| (key.clone(), value.clone()))
.collect()
}
impl InformationSchemaFlowsBuilder {
fn new(
schema: SchemaRef,
@@ -402,13 +378,10 @@ impl InformationSchemaFlowsBuilder {
}
})?,
));
// Never surface the internal transient transport keys in the OPTIONS
// column, even if malformed/old metadata carries them.
let visible_options = visible_flow_options(flow_info.options());
self.option_groups
.push(Some(&serde_json::to_string(&visible_options).context(
.push(Some(&serde_json::to_string(flow_info.options()).context(
JsonSnafu {
input: format!("{visible_options:?}"),
input: format!("{:?}", flow_info.options()),
},
)?));
self.created_time
@@ -592,44 +565,4 @@ mod tests {
};
assert_eq!(reparsed.eval_offset, None);
}
#[test]
fn test_generate_show_create_flow_filters_internal_keys_even_if_metadata_tainted() {
// Even if malformed/old metadata carries the internal transport keys
// in `options`, SHOW CREATE and flow_definition must not surface them.
let raw_sql = "SELECT max(c1) FROM public.src";
let options = HashMap::from([
("defer_on_missing_source".to_string(), "true".to_string()),
(INTERNAL_EVAL_OFFSET_KEY.to_string(), "120".to_string()),
(
INTERNAL_EVAL_SCHEDULE_KEY.to_string(),
"{\"x\":1}".to_string(),
),
(FlowType::FLOW_TYPE_KEY.to_string(), "batching".to_string()),
]);
let flow_info = flow_info_for_show_create(raw_sql, Some(3600), 120, options.clone());
let sql = InformationSchemaFlows::generate_show_create_flow(&flow_info).unwrap();
assert!(
!sql.contains("__greptime_internal"),
"internal keys must be absent, got:\n{sql}"
);
assert!(
!sql.contains("eval_schedule"),
"internal schedule must be absent, got:\n{sql}"
);
assert!(
!sql.contains(FlowType::FLOW_TYPE_KEY),
"flow_type must be filtered, got:\n{sql}"
);
assert!(
sql.contains("defer_on_missing_source = 'true'"),
"user option must be preserved, got:\n{sql}"
);
// The OPTIONS column uses the same filtering.
let visible = visible_flow_options(&options);
assert!(!visible.contains_key(INTERNAL_EVAL_OFFSET_KEY));
assert!(!visible.contains_key(INTERNAL_EVAL_SCHEDULE_KEY));
assert_eq!(visible.len(), 2); // defer_on_missing_source + flow_type
}
}
src/flow/src/adapter/flownode_impl.rs
-17
View File
@@ -687,23 +687,6 @@ impl FlowEngine for FlowDualEngine {
let flow_id = args.flow_id;
let src_table_ids = args.source_table_ids.clone();
// Historical persisted metadata may pair `flow_type=streaming` with an
// `EVAL INTERVAL` / schedule (created before EVAL OFFSET support, when
// an interval over an instant-TTL source silently fell back to
// streaming). The streaming engine has no scheduler, so recovering
// such metadata would silently discard the schedule. Reject it clearly
// at the dual-engine boundary; do not attempt to migrate it.
if flow_type == FlowType::Streaming
&& (args.eval_interval.is_some() || args.eval_schedule.is_some())
{
return UnexpectedSnafu {
reason: format!(
"flow {flow_id} is persisted with flow_type=streaming but carries an EVAL INTERVAL/schedule; the streaming engine cannot honor scheduled evaluation. Refusing to recover the flow instead of silently discarding its schedule. Recreate the flow without EVAL INTERVAL, or with a non-instant source TTL so it can use the batching scheduler"
),
}
.fail();
}
let res = match flow_type {
FlowType::Batching => self.batching_engine.create_flow(args).await,
FlowType::Streaming => self.streaming_engine.create_flow(args).await,
src/operator/src/expr_helper.rs
-104
View File
@@ -34,7 +34,6 @@ use api::v1::{
use common_datasource::object_store::LocalFileAccess;
use common_error::ext::BoxedError;
use common_grpc_expr::util::ColumnExpr;
use common_meta::ddl::create_flow::{INTERNAL_EVAL_OFFSET_KEY, INTERNAL_EVAL_SCHEDULE_KEY};
use common_time::Timezone;
use datafusion::sql::planner::object_name_to_table_reference;
use datatypes::prelude::ConcreteDataType;
@@ -1100,24 +1099,6 @@ pub fn to_create_flow_task_expr(
let eval_interval = create_flow.eval_interval;
let flow_options = stringify_flow_options(create_flow.flow_options)?;
// The internal offset/schedule transport keys are trusted only when
// inserted later by the operator procedure layer from the typed SQL AST
// (after user option validation). A user-supplied occurrence must be
// rejected instead of being honored.
if flow_options.contains_key(INTERNAL_EVAL_OFFSET_KEY)
|| flow_options.contains_key(INTERNAL_EVAL_SCHEDULE_KEY)
{
let key = if flow_options.contains_key(INTERNAL_EVAL_OFFSET_KEY) {
INTERNAL_EVAL_OFFSET_KEY
} else {
INTERNAL_EVAL_SCHEDULE_KEY
};
return InvalidSqlSnafu {
err_msg: format!("flow option '{key}' is reserved for internal use"),
}
.fail();
}
Ok(CreateFlowExpr {
catalog_name: query_ctx.current_catalog().to_string(),
flow_name: sanitize_flow_name(create_flow.flow_name)?,
@@ -2019,89 +2000,4 @@ WITH(
let new_sql = format!("{:#}", create_table);
assert_eq!(sql, new_sql);
}
#[test]
fn test_create_flow_expr_does_not_insert_internal_key_at_ast_layer() {
// The typed `EVAL OFFSET` is transported to meta by the operator
// procedure layer (after user option validation), NOT by the
// AST-to-expr helper. The expr produced here must therefore not carry
// the transient key at all.
let sql = r#"
CREATE FLOW calc_reqs
SINK TO cnt_reqs
EVAL INTERVAL '1 hour'
EVAL OFFSET '2 minutes'
AS
SELECT max(c1) FROM http_requests;"#;
let stmt =
ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
.unwrap()
.pop()
.unwrap();
let Statement::CreateFlow(create_flow) = stmt else {
unreachable!()
};
let expr = to_create_flow_task_expr(create_flow, &QueryContext::arc()).unwrap();
assert_eq!(expr.eval_interval.as_ref().map(|e| e.seconds), Some(3600));
assert!(
!expr.flow_options.contains_key(INTERNAL_EVAL_OFFSET_KEY),
"the transient key must be inserted by the procedure layer, not here"
);
// Zero offset (canonicalized to None by the parser) transports nothing.
let sql = r#"
CREATE FLOW calc_reqs_zero
SINK TO cnt_reqs
EVAL INTERVAL '1 hour'
EVAL OFFSET '0 seconds'
AS
SELECT max(c1) FROM http_requests;"#;
let stmt =
ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
.unwrap()
.pop()
.unwrap();
let Statement::CreateFlow(create_flow) = stmt else {
unreachable!()
};
let expr = to_create_flow_task_expr(create_flow, &QueryContext::arc()).unwrap();
assert!(!expr.flow_options.contains_key(INTERNAL_EVAL_OFFSET_KEY));
}
#[test]
fn test_create_flow_expr_rejects_user_supplied_internal_keys() {
for key in [
INTERNAL_EVAL_OFFSET_KEY,
INTERNAL_EVAL_SCHEDULE_KEY,
"__greptime_internal_eval_schedule",
"__greptime_internal_eval_offset_secs",
] {
let sql = format!(
r#"
CREATE FLOW calc_reqs
SINK TO cnt_reqs
EVAL INTERVAL '1 hour'
WITH ({key} = '120')
AS
SELECT max(c1) FROM http_requests;"#
);
let stmt = ParserContext::create_with_dialect(
&sql,
&GreptimeDbDialect {},
ParseOptions::default(),
)
.unwrap()
.pop()
.unwrap();
let Statement::CreateFlow(create_flow) = stmt else {
unreachable!()
};
let err = to_create_flow_task_expr(create_flow, &QueryContext::arc()).unwrap_err();
assert!(
err.to_string()
.contains(&format!("flow option '{key}' is reserved for internal use")),
"unexpected error for {key}: {err}"
);
}
}
}
src/query/src/sql.rs
+1 -8
View File
@@ -1236,16 +1236,9 @@ pub fn show_create_flow(
.map(|schedule| schedule.anchor_secs)
.filter(|anchor_secs| *anchor_secs != 0),
comment,
// Defense-in-depth for SHOW CREATE: never surface the internal
// transient transport keys, even if malformed/old metadata carries
// them in `options`.
flow_options: OptionMap::from_filtered_string_map(
flow_val.options(),
&[
FlowType::FLOW_TYPE_KEY,
common_meta::ddl::create_flow::INTERNAL_EVAL_OFFSET_KEY,
common_meta::ddl::create_flow::INTERNAL_EVAL_SCHEDULE_KEY,
],
&[FlowType::FLOW_TYPE_KEY],
),
query,
};
tests-integration/src/grpc.rs
+4 -92
View File
@@ -53,9 +53,9 @@ mod test {
use api::v1::region::QueryRequest as RegionQueryRequest;
use api::v1::{
AddColumn, AddColumns, AlterTableExpr, Column, ColumnDataType, ColumnDataTypeExtension,
ColumnDef, CreateDatabaseExpr, CreateFlowExpr, CreateTableExpr, DdlRequest, DeleteRequest,
DeleteRequests, DropTableExpr, EvalInterval, InsertIntoPlan, InsertRequest, InsertRequests,
QueryRequest, SemanticType, TableName, VectorTypeExtension, alter_table_expr,
ColumnDef, CreateDatabaseExpr, CreateTableExpr, DdlRequest, DeleteRequest, DeleteRequests,
DropTableExpr, InsertIntoPlan, InsertRequest, InsertRequests, QueryRequest, SemanticType,
VectorTypeExtension, alter_table_expr,
};
use auth::{
DefaultPermissionChecker, Identity, Password, PermissionCheckerRef, UserProvider,
@@ -64,7 +64,7 @@ mod test {
use client::OutputData;
use common_base::Plugins;
use common_catalog::consts::{
DEFAULT_CATALOG_NAME, DEFAULT_PRIVATE_SCHEMA_NAME, DEFAULT_SCHEMA_NAME, MITO_ENGINE,
DEFAULT_CATALOG_NAME, DEFAULT_PRIVATE_SCHEMA_NAME, MITO_ENGINE,
SEMANTIC_RELATIONSHIPS_DECLARED_TABLE_NAME,
};
use common_meta::rpc::router::region_distribution;
@@ -138,94 +138,6 @@ mod test {
.unwrap()
}
#[tokio::test(flavor = "multi_thread")]
async fn test_standalone_grpc_ddl_flow_rejects_internal_transport_keys() {
// Direct gRPC DDL requests bypass the SQL parser. A caller that crafts
// a `CreateFlowExpr` with an internal transport key in flow_options is
// trying to spoof a trusted `EVAL OFFSET` / schedule: the operator's
// `create_flow_inner` must reject it before any trusted extraction.
let standalone = GreptimeDbStandaloneBuilder::new(
"test_standalone_grpc_ddl_flow_rejects_internal_transport_keys",
)
.build()
.await;
let instance = standalone.fe_instance();
for key in [
"__greptime_internal_eval_offset_secs",
"__greptime_internal_eval_schedule",
] {
let expr = CreateFlowExpr {
catalog_name: DEFAULT_CATALOG_NAME.to_string(),
flow_name: format!("spoofed_flow_{}", key.len()),
source_table_names: vec![],
sink_table_name: Some(TableName {
catalog_name: DEFAULT_CATALOG_NAME.to_string(),
schema_name: DEFAULT_SCHEMA_NAME.to_string(),
table_name: "spoofed_sink".to_string(),
}),
or_replace: false,
create_if_not_exists: false,
expire_after: None,
eval_interval: Some(EvalInterval { seconds: 3600 }),
comment: String::new(),
sql: "SELECT 1".to_string(),
flow_options: HashMap::from([(key.to_string(), "120".to_string())]),
};
let request = Request::Ddl(DdlRequest {
expr: Some(DdlExpr::CreateFlow(expr)),
});
let err = GrpcQueryHandler::do_query(instance.as_ref(), request, QueryContext::arc())
.await
.unwrap_err();
let err_msg = format!("{err:?}");
assert!(
err_msg.contains(&format!("flow option '{key}' is reserved for internal use")),
"expected reserved-key rejection for {key}, got: {err_msg}"
);
}
// A benign flow_options map (no internal keys) passes the boundary
// guard; the request then proceeds to flow creation which fails on the
// missing source table (not on the transport guard).
let expr = CreateFlowExpr {
catalog_name: DEFAULT_CATALOG_NAME.to_string(),
flow_name: "benign_missing_source_flow".to_string(),
source_table_names: vec![TableName {
catalog_name: DEFAULT_CATALOG_NAME.to_string(),
schema_name: DEFAULT_SCHEMA_NAME.to_string(),
table_name: "definitely_missing_table".to_string(),
}],
sink_table_name: Some(TableName {
catalog_name: DEFAULT_CATALOG_NAME.to_string(),
schema_name: DEFAULT_SCHEMA_NAME.to_string(),
table_name: "benign_missing_source_sink".to_string(),
}),
or_replace: false,
create_if_not_exists: false,
expire_after: None,
eval_interval: Some(EvalInterval { seconds: 3600 }),
comment: String::new(),
sql: "SELECT * FROM definitely_missing_table".to_string(),
flow_options: HashMap::new(),
};
let request = Request::Ddl(DdlRequest {
expr: Some(DdlExpr::CreateFlow(expr)),
});
let err = GrpcQueryHandler::do_query(instance.as_ref(), request, QueryContext::arc())
.await
.unwrap_err();
let err_msg = format!("{err:?}");
assert!(
err_msg.contains("missing source tables"),
"benign request should fail on missing sources, got: {err_msg}"
);
assert!(
!err_msg.contains("reserved for internal use"),
"benign request must not trip the transport guard, got: {err_msg}"
);
}
#[tokio::test(flavor = "multi_thread")]
async fn test_grpc_insert_into_plan_rejects_readonly_user() {
let plugins = Plugins::new();
tests/compatibility/cases/flow_eval_interval_streaming_old_metadata/case.toml
-8
View File
@@ -1,8 +0,0 @@
name = "flow_eval_interval_streaming_old_metadata"
reason = "Old binary scheduled-streaming metadata (flow_type=streaming plus EVAL INTERVAL over an instant-TTL source, created before EVAL OFFSET support) must be explicitly rejected at the dual-engine boundary by a new binary instead of silently routed to the streaming engine and stripped of its schedule."
introduced_by = "EVAL OFFSET feature"
topologies = ["distributed", "standalone"]
from_range = ["<=v1.1.1"]
to_range = [">v1.1.1"]
features = ["flow", "table"]
owner = "flow"
tests/compatibility/cases/flow_eval_interval_streaming_old_metadata/setup.sql
-21
View File
@@ -1,21 +0,0 @@
-- Runs on the OLD binary (no EVAL OFFSET support): an `EVAL INTERVAL` flow
-- over an instant-TTL source is forced to the streaming engine, which persists
-- flow_type=streaming together with eval_interval in the metadata. A new binary
-- must reject this metadata on recovery instead of silently routing it to the
-- streaming engine (which would discard the schedule) or migrating it.
CREATE TABLE compat_eval_interval_streaming_old_input (
ts TIMESTAMP(3) TIME INDEX,
series STRING,
v DOUBLE,
PRIMARY KEY(series)
) WITH ('ttl' = 'instant');
CREATE FLOW compat_eval_interval_streaming_old_flow
SINK TO compat_eval_interval_streaming_old_sink
EVAL INTERVAL '1s'
AS
SELECT
date_trunc('second', now()) AS ts,
count(v) AS value_count
FROM compat_eval_interval_streaming_old_input
GROUP BY date_trunc('second', now());
tests/compatibility/cases/flow_eval_interval_streaming_old_metadata/verify.result
-44
View File
@@ -1,44 +0,0 @@
-- Runs on the NEW binary. Old scheduled-streaming metadata
-- (flow_type=streaming plus EVAL INTERVAL persisted by a binary without EVAL
-- OFFSET support) must be explicitly rejected at the dual-engine boundary on
-- recovery: the flow must not be silently routed to the streaming engine
-- (which would discard its schedule) and must not be migrated to batching.
-- The metadata is still readable; it is not deleted.
SELECT flow_name, source_table_names
FROM information_schema.flows
WHERE flow_name = 'compat_eval_interval_streaming_old_flow';
+-----------------------------------------+---------------------------------------------------------------------------------------------+
| flow_name | source_table_names |
+-----------------------------------------+---------------------------------------------------------------------------------------------+
| compat_eval_interval_streaming_old_flow | greptime.flow_eval_interval_streaming_old_metadata.compat_eval_interval_streaming_old_input |
+-----------------------------------------+---------------------------------------------------------------------------------------------+
SHOW CREATE FLOW compat_eval_interval_streaming_old_flow;
+-----------------------------------------+---------------------------------------------------------------------------------------------------------------------------------------------------------+
| Flow | Create Flow |
+-----------------------------------------+---------------------------------------------------------------------------------------------------------------------------------------------------------+
| compat_eval_interval_streaming_old_flow | CREATE FLOW IF NOT EXISTS compat_eval_interval_streaming_old_flow |
| | SINK TO flow_eval_interval_streaming_old_metadata.compat_eval_interval_streaming_old_sink |
| | EVAL INTERVAL '1 s' |
| | AS SELECT date_trunc('second', now()) AS ts, count(v) AS value_count FROM compat_eval_interval_streaming_old_input GROUP BY date_trunc('second', now()) |
+-----------------------------------------+---------------------------------------------------------------------------------------------------------------------------------------------------------+
-- Inserting into the instant-TTL source must NOT produce sink rows: the flow
-- was rejected on recovery instead of silently recovered as a streaming flow.
INSERT INTO compat_eval_interval_streaming_old_input VALUES
('2026-06-25 00:00:00', 'a', 1.0),
('2026-06-25 00:00:01', 'b', 2.0);
Affected Rows: 0
-- SQLNESS SLEEP 2s
SELECT count(*) AS sink_rows
FROM compat_eval_interval_streaming_old_sink;
+-----------+
| sink_rows |
+-----------+
| 0 |
+-----------+
tests/compatibility/cases/flow_eval_interval_streaming_old_metadata/verify.sql
-22
View File
@@ -1,22 +0,0 @@
-- Runs on the NEW binary. Old scheduled-streaming metadata
-- (flow_type=streaming plus EVAL INTERVAL persisted by a binary without EVAL
-- OFFSET support) must be explicitly rejected at the dual-engine boundary on
-- recovery: the flow must not be silently routed to the streaming engine
-- (which would discard its schedule) and must not be migrated to batching.
-- The metadata is still readable; it is not deleted.
SELECT flow_name, source_table_names
FROM information_schema.flows
WHERE flow_name = 'compat_eval_interval_streaming_old_flow';
SHOW CREATE FLOW compat_eval_interval_streaming_old_flow;
-- Inserting into the instant-TTL source must NOT produce sink rows: the flow
-- was rejected on recovery instead of silently recovered as a streaming flow.
INSERT INTO compat_eval_interval_streaming_old_input VALUES
('2026-06-25 00:00:00', 'a', 1.0),
('2026-06-25 00:00:01', 'b', 2.0);
-- SQLNESS SLEEP 2s
SELECT count(*) AS sink_rows
FROM compat_eval_interval_streaming_old_sink;
tests/compatibility/cases/flow_eval_offset_old_metadata/case.toml
-8
View File
@@ -1,8 +0,0 @@
name = "flow_eval_offset_old_metadata"
reason = "Old binary interval-flow metadata (created before EVAL OFFSET support) recovered by a new binary must default to an implicit zero offset / epoch anchor in SHOW CREATE and information_schema, and must never surface internal transport keys."
introduced_by = "EVAL OFFSET feature"
topologies = ["distributed", "standalone"]
from_range = ["<=v1.1.1"]
to_range = [">v1.1.1"]
features = ["flow", "table"]
owner = "flow"
tests/compatibility/cases/flow_eval_offset_old_metadata/setup.sql
-19
View File
@@ -1,19 +0,0 @@
-- Runs on the OLD binary (no EVAL OFFSET support): create an `EVAL INTERVAL`
-- flow without any offset. The old binary persists metadata without a typed
-- eval_schedule; the new binary must recover it with an implicit zero offset.
CREATE TABLE compat_eval_offset_old_input (
ts TIMESTAMP(3) TIME INDEX,
series STRING,
v DOUBLE,
PRIMARY KEY(series)
);
CREATE FLOW compat_eval_offset_old_flow
SINK TO compat_eval_offset_old_sink
EVAL INTERVAL '1s'
AS
SELECT
date_trunc('second', now()) AS ts,
count(v) AS value_count
FROM compat_eval_offset_old_input
GROUP BY date_trunc('second', now());
tests/compatibility/cases/flow_eval_offset_old_metadata/verify.result
-73
View File
@@ -1,73 +0,0 @@
-- Runs on the NEW binary: old interval-flow metadata must recover with an
-- implicit zero offset (no `EVAL OFFSET` line) and no internal transport keys.
SELECT flow_name, source_table_names
FROM information_schema.flows
WHERE flow_name = 'compat_eval_offset_old_flow';
+-----------------------------+---------------------------------------------------------------------+
| flow_name | source_table_names |
+-----------------------------+---------------------------------------------------------------------+
| compat_eval_offset_old_flow | greptime.flow_eval_offset_old_metadata.compat_eval_offset_old_input |
+-----------------------------+---------------------------------------------------------------------+
SHOW CREATE FLOW compat_eval_offset_old_flow;
+-----------------------------+---------------------------------------------------------------------------------------------------------------------------------------------+
| Flow | Create Flow |
+-----------------------------+---------------------------------------------------------------------------------------------------------------------------------------------+
| compat_eval_offset_old_flow | CREATE FLOW IF NOT EXISTS compat_eval_offset_old_flow |
| | SINK TO flow_eval_offset_old_metadata.compat_eval_offset_old_sink |
| | EVAL INTERVAL '1 s' |
| | AS SELECT date_trunc('second', now()) AS ts, count(v) AS value_count FROM compat_eval_offset_old_input GROUP BY date_trunc('second', now()) |
+-----------------------------+---------------------------------------------------------------------------------------------------------------------------------------------+
SELECT flow_definition
FROM information_schema.flows
WHERE flow_name = 'compat_eval_offset_old_flow';
+---------------------------------------------------------------------------------------------------------------------------------------------+
| flow_definition |
+---------------------------------------------------------------------------------------------------------------------------------------------+
| CREATE FLOW IF NOT EXISTS compat_eval_offset_old_flow |
| SINK TO flow_eval_offset_old_metadata.compat_eval_offset_old_sink |
| EVAL INTERVAL '1 s' |
| AS SELECT date_trunc('second', now()) AS ts, count(v) AS value_count FROM compat_eval_offset_old_input GROUP BY date_trunc('second', now()) |
+---------------------------------------------------------------------------------------------------------------------------------------------+
-- The recovered flow_definition must not surface internal transport keys.
SELECT count(*) AS leaked_internal_keys
FROM information_schema.flows
WHERE flow_name = 'compat_eval_offset_old_flow'
AND (flow_definition LIKE '%__greptime_internal%'
OR flow_definition LIKE '%eval_schedule%');
+----------------------+
| leaked_internal_keys |
+----------------------+
| 0 |
+----------------------+
-- The old flow still executes on the new binary.
INSERT INTO compat_eval_offset_old_input VALUES
('2026-06-25 00:00:00', 'a', 1.0),
('2026-06-25 00:00:01', 'b', 2.0);
Affected Rows: 2
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
ADMIN FLUSH_FLOW('compat_eval_offset_old_flow');
+-------------------------------------------------+
| ADMIN FLUSH_FLOW('compat_eval_offset_old_flow') |
+-------------------------------------------------+
| FLOW_FLUSHED |
+-------------------------------------------------+
SELECT count(*) > 0 AS old_flow_executed
FROM compat_eval_offset_old_sink;
+-------------------+
| old_flow_executed |
+-------------------+
| true |
+-------------------+
tests/compatibility/cases/flow_eval_offset_old_metadata/verify.sql
-29
View File
@@ -1,29 +0,0 @@
-- Runs on the NEW binary: old interval-flow metadata must recover with an
-- implicit zero offset (no `EVAL OFFSET` line) and no internal transport keys.
SELECT flow_name, source_table_names
FROM information_schema.flows
WHERE flow_name = 'compat_eval_offset_old_flow';
SHOW CREATE FLOW compat_eval_offset_old_flow;
SELECT flow_definition
FROM information_schema.flows
WHERE flow_name = 'compat_eval_offset_old_flow';
-- The recovered flow_definition must not surface internal transport keys.
SELECT count(*) AS leaked_internal_keys
FROM information_schema.flows
WHERE flow_name = 'compat_eval_offset_old_flow'
AND (flow_definition LIKE '%__greptime_internal%'
OR flow_definition LIKE '%eval_schedule%');
-- The old flow still executes on the new binary.
INSERT INTO compat_eval_offset_old_input VALUES
('2026-06-25 00:00:00', 'a', 1.0),
('2026-06-25 00:00:01', 'b', 2.0);
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
ADMIN FLUSH_FLOW('compat_eval_offset_old_flow');
SELECT count(*) > 0 AS old_flow_executed
FROM compat_eval_offset_old_sink;