feat(flow): add information_schema.flow_statistics and SHOW FLOW STATUS (#7987) (#8392)

* feat(flow): add information_schema.flow_statistics and SHOW FLOW STATUS (fixes #7987) Signed-off-by: onepizzateam <palakjha916@gmail.com> * feat(flow): add information_schema.flow_statistics and SHOW FLOW STATUS (fixes #7987) Signed-off-by: onepizzateam <palakjha916@gmail.com> * test(flow): add sqlness golden result for flow_status Signed-off-by: Palak Jha <palakjha916@gmail.com> * fix(catalog): remove unused OptionExt import in flow_statistics Signed-off-by: onepizzateam <palakjha916@gmail.com> * docs(flow): fix stale 'recent errors' comment on QueryFlowExecStats Signed-off-by: onepizzateam <palakjha916@gmail.com> * test: regenerate golden results for flow_statistics table Signed-off-by: onepizzateam <palakjha916@gmail.com> * style: apply rustfmt to flow_statistics changes Signed-off-by: onepizzateam <palakjha916@gmail.com> * refactor(catalog): hoist current_time_millis out of flow loop and clamp uptime Signed-off-by: onepizzateam <palakjha916@gmail.com> * chore: remove accidentally committed fmt_check.log Signed-off-by: onepizzateam <palakjha916@gmail.com> * Update flow_status.result del eof trailing blank line as per review Signed-off-by: onepizzateam <palakjha916@gmail.com> * test(flow): restore runner-generated trailing blank line for sqlness Signed-off-by: onepizzateam <palakjha916@gmail.com> * postgres: include SHOW FLOW STATUS in extended-query describe (return flow_statistics fields) Signed-off-by: onepizzateam <palakjha916@gmail.com> * fix: address reviewer feedback on flow_statistics PR Signed-off-by: onepizzateam <palakjha916@gmail.com> * fix(flow): resolve merge conflicts with main Signed-off-by: polar <palakjha916@gmail.com> Signed-off-by: onepizzateam <palakjha916@gmail.com> * sqlness check post gen (information_schema.result) Signed-off-by: onepizzateam <palakjha916@gmail.com> * fix(flow): record start_time after req/snapshot_seqs built, before dispatch Signed-off-by: onepizzateam <palakjha916@gmail.com> * fix(sql): handle ShowFlowStatus in match statement at util.rs Signed-off-by: onepizzateam <palakjha916@gmail.com> * feat: review patch implementation Signed-off-by: onepizzateam <palakjha916@gmail.com> * chore: remove accidentally committed local tool output files and fix fmt Signed-off-by: onepizzateam <palakjha916@gmail.com> * fix worker.rs return type formatting Signed-off-by: onepizzateam <palakjha916@gmail.com> * fix(flow): apply rustfmt to get_full_flow_stat return type Signed-off-by: onepizzateam <palakjha916@gmail.com> * fix worker.rs return type formatting Signed-off-by: onepizzateam <palakjha916@gmail.com> * fix(flow): re-apply rustfmt to get_full_flow_stat return type after merge Signed-off-by: onepizzateam <palakjha916@gmail.com> * fix: merge conflicts Signed-off-by: onepizzateam <palakjha916@gmail.com> * fix(auth): warn when credential load disables Postgres SCRAM or drops a line (#8652) * fix(auth): warn when credential load disables Postgres SCRAM or drops a line Static and watch user providers degraded silently in two ways: - A single non-SCRAM verifier (mysql_native_password, or a legacy pbkdf2_sha256 hash that predates SCRAM) disables Postgres SCRAM for every user and falls back to cleartext, with no signal to the operator. - A malformed credential line (commonly a plaintext password containing '=', which splits into more than two parts) was dropped without a trace. Emit a warning at each credential load for both cases so operators don't unknowingly serve cleartext passwords over Postgres or lose a user. This is logging only; authentication behavior is unchanged. The SCRAM check never logs secrets, and the malformed-line warning logs the line number and file, never the line content. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * fix(auth): warn on credential file read error before truncating A read error from lines() (I/O failure or invalid UTF-8) ends the iterator via map_while, silently dropping every remaining credential. Warn with the line number and file before truncating, matching the malformed-line handling, so the drop is observable. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> --------- Signed-off-by: Dennis Zhuang <killme2008@gmail.com> Signed-off-by: onepizzateam <palakjha916@gmail.com> * test(object-store): fix racy SecureFs abort test (#8720) test_writer_abort_is_unsupported_without_atomic_write asserted the file content immediately after abort() returned Unsupported. SecureFsWriter writes through tokio::fs::File, whose write_all() only enqueues a blocking write task (tokio's poll_write returns Ready before the write completes), so the data may not be visible yet when the test reads the file. Drop the race-prone content assertion and only verify the Unsupported contract. Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> Signed-off-by: onepizzateam <palakjha916@gmail.com> * feat(query): plan native histogram functions (#8705) * feat(query): plan native histogram functions Signed-off-by: shuiyisong <xixing.sys@gmail.com> * fix: cr issue & add tests Signed-off-by: shuiyisong <xixing.sys@gmail.com> * fix: cr issue Signed-off-by: shuiyisong <xixing.sys@gmail.com> * fix: cr issue Signed-off-by: shuiyisong <xixing.sys@gmail.com> --------- Signed-off-by: shuiyisong <xixing.sys@gmail.com> Signed-off-by: onepizzateam <palakjha916@gmail.com> * perf(promql): avoid repeated scans in sliding range evaluation (#8646) * perf(promql): use two pointers for sliding range boundaries Replace the stale cursor heuristic in RangeManipulateStream::calculate_range with monotonic left/right cursors. The old path rescanned each evaluation window (O(E x samples-per-window)) and could lose valid samples after sparse gaps or trailing empty windows. The two pointers keep strict monotonic progress, reducing boundary generation to O(N + E) while preserving (curr-range, curr] semantics, start/end shortening, and empty-window output. Controlled release benchmarks (fixed CPU, ABBA): - Public RangeManipulate wall time: ~28% faster at 1m/15s, ~66% at 5m/15s, ~96% at 1h/15s. - Warmed distributed TQL ANALYZE 1h queries: ~17-21% faster end to end; shorter windows stayed within run-order noise. Signed-off-by: discord9 <discord9@163.com> * perf(promql): specialize changes/resets with adaptive edge counting The generic range_fn macro slices, downcasts, and rescans every overlapping window for changes() and resets(). Replace the macro path for these two functions with hand-written UDF wrappers backed by a shared private edge-count kernel: direct raw-offset scans when requested edges are few, otherwise one global u64 edge prefix so each window is answered by a prefix difference. Behavior is preserved bit-for-bit, including raw null-buffer values, NaN semantics, signed zero, infinities, empty/singleton windows, independent timestamp/value offsets, arbitrary window layouts, and exact DataFusion error messages. The shared proc macro, planner, serializer, and other range functions are untouched. Controlled release benchmarks (fixed CPU, ABBA): - Dense sliding windows (k=4/20/240): 91.7-95.6% less public UDF wall time. - Low-coverage fallback (N=4096, 8 windows): 73.9-74.4% faster. - Warmed distributed TQL ANALYZE 5m/1h changes/resets: 12.1-19.7% client and 12.0-20.9% server latency improvement; controls stayed within drift. Signed-off-by: discord9 <discord9@163.com> * ci(query-regression): include PromQL range boundary case in defaults An audit of historical query-regression runs found zero range-query coverage: all 208 PromQL ANALYZE samples were bare selectors, so range evaluation could regress without CI noticing. Wire the promql_range_boundary case (introduced in #8646) into DEFAULT_CASES so label-triggered runs measure the range path. The case is cheap: a ~0.3s synthetic fixture and about a minute of query execution per base/candidate pass. Signed-off-by: discord9 <discord9@163.com> * chore(promql): address sliding range review nits Move test-only imports into their test modules and remove the unused pre-specialization changes and resets helpers. Signed-off-by: discord9 <discord9@163.com> * style(promql): apply pinned rustfmt Signed-off-by: discord9 <discord9@163.com> * test(promql): cover sparse range results Share the changes and resets test scaffolding while keeping their behavior oracles independent. Add an end-to-end sqlness regression for sparse samples, empty intermediate windows, and a valid trailing sample. Signed-off-by: discord9 <discord9@163.com> --------- Signed-off-by: discord9 <discord9@163.com> Signed-off-by: onepizzateam <palakjha916@gmail.com> * ci: optimize fuzz and split workflows (#8710) * ci: batch fuzz targets in GitHub Actions Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: improve fuzz test observability Signed-off-by: WenyXu <wenymedia@gmail.com> * fix(ci): preserve fuzz setup failure artifacts Signed-off-by: WenyXu <wenymedia@gmail.com> * test(ci): keep fuzz mock output in logs Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: optimize fuzz worker cache Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: warm fuzz target binaries Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: isolate fuzz workflow Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: centralize fuzz target preparation Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: split general workflows Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: streamline docs required checks Signed-off-by: WenyXu <wenymedia@gmail.com> * fix: transfer fuzz targets as artifacts Signed-off-by: WenyXu <wenymedia@gmail.com> * fix: preserve fuzz binary permissions Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: streamline fuzz workers Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: cache PR build dependencies Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: retain main build cache policy Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: address fuzz review feedback Signed-off-by: WenyXu <wenymedia@gmail.com> --------- Signed-off-by: WenyXu <wenymedia@gmail.com> Signed-off-by: onepizzateam <palakjha916@gmail.com> * fix: add public constructor for compactor (#8724) Signed-off-by: onepizzateam <palakjha916@gmail.com> * feat(logging): add enable_file_logging option to disable file logging (#8721) Signed-off-by: xhwhis <hi@whis.me> Signed-off-by: onepizzateam <palakjha916@gmail.com> * avoid cloning final Prometheus remote write row (#8733) perf: avoid cloning final Prometheus remote write row Signed-off-by: lyang24 <lanqingy93@gmail.com> Signed-off-by: onepizzateam <palakjha916@gmail.com> * feat(function): add json_object_keys scalar function (#8722) Expose JSON object key listing for outermost objects, with sqlness coverage. Signed-off-by: onepizzateam <palakjha916@gmail.com> * refactor(mito2): revise compaction trigger behavior (#8706) * refactor(mito2): revise compaction trigger behavior Distinguish automatic and manual triggers, coalesce explicit automatic follow-ups, and reject concurrent manual compactions. Remove implicit post-execution continuation and transient idle statuses so scheduler entries always represent an active lifecycle. Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> * fix(mito2): track automatic compaction follow-ups Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> * docs(mito2): fix compaction transition rustdoc Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> * fix(mito2): mark manual compaction conflict retryable Signed-off-by: Lei, HUANG <mrsatangel@gmail.com> * refactor(mito2): drop unused RequestCancelResult::NotRunning variant request_cancel is only called in tests where the region is guaranteed to be running, so the NotRunning case was dead code. Simplify to unwrap() and remove the variant. Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> * fix(mito2): gate test-only cancellation import Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> * fix(mito2): prioritize DDL after compaction planning Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> --------- Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> Signed-off-by: Lei, HUANG <mrsatangel@gmail.com> Signed-off-by: onepizzateam <palakjha916@gmail.com> * feat: update dashboard to v0.13.11 (#8737) Signed-off-by: onepizzateam <palakjha916@gmail.com> * fix(object-store): skip removed-entry lister test on Windows (#8735) DirEntry on Windows is a snapshot from FindFirstFileW: file_type() and metadata() keep returning cached data after the file is removed, so read_list_entry() cannot observe the deletion. The test asserts the Unix behavior (lstat returns ENOENT) and fails deterministically on Windows nightly CI (4/4 tries). Gate it with #[cfg(not(windows))]. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> Signed-off-by: onepizzateam <palakjha916@gmail.com> * fix(query): preserve remote dynamic filter target (#8615) * fix(query): preserve remote dynamic filter target Signed-off-by: discord9 <discord9@163.com> * fix(query): check RDF subscriber registration Signed-off-by: discord9 <discord9@163.com> * fix(query): refresh initial dyn filter snapshot before dispatch and handle RDF unregister The remote dynamic filter dispatch ordering regression: freeze the target, pre-register subscribers, refresh the initial snapshot, then dispatch. Also implement handle_remote_dyn_filter_unregister to keep unregister targets consistent with do_get/update. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(query): update test-only RegionQueryHandler impl to new trait signatures Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --------- Signed-off-by: discord9 <discord9@163.com> Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> Signed-off-by: onepizzateam <palakjha916@gmail.com> * docs: rework README release badges, drop star history, fix grpc flag (#8743) * docs: show stable, latest and nightly version badges in README The single release badge rendered whatever GitHub considered newest, so a pre-release such as v1.2.0-beta.1 looked like the recommended version. Split it into three self-updating badges using the shields.io `filter` parameter, keyed off the existing tag naming: - stable: `!*-*` matches tags without a hyphen (v1.1.4) - latest: `!*-*-*` excludes nightly and dev builds (v1.2.0-beta.1) - nightly: `*-nightly-*` matches the weekly build (v1.2.0-nightly-20260706) No workflow changes are needed; the badges track new releases on their own. A one-line caption below them says which channel to pick. The release-date badge is dropped as the three version badges already carry that signal. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * docs: remove star history chart from README The chart carried a sealed_token in three URLs and added a large third-party image to the Project Status section without saying anything the badges and case studies do not already cover. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * docs: use --grpc-bind-addr in README quickstart --rpc-bind-addr is now only a hidden alias of --grpc-bind-addr and no longer shows up in --help. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * Update README.md Co-authored-by: Ning Sun <classicning@gmail.com> --------- Signed-off-by: Dennis Zhuang <killme2008@gmail.com> Co-authored-by: Ning Sun <classicning@gmail.com> Signed-off-by: onepizzateam <palakjha916@gmail.com> * chore: check enterprise-gated files are listed in both license configs (#8750) * chore: check enterprise-gated files are listed in both license configs A file reachable only through `#[cfg(feature = "enterprise")] mod ...;` is governed by the GreptimeDB Enterprise License, so it must appear in the `includes` of licenserc-enterprise.toml and the `excludes` of licenserc.toml. hawkeye stays silent when it does not: the file keeps its Apache-2.0 header and passes the default check precisely because it was never excluded from it. scripts/check-enterprise-license.py walks enterprise-gated `mod` declarations, resolves them to files (submodules included) and diffs that set against both configs, also reporting stale entries. It runs in the license job in CI and as `make check-enterprise-license`. Documents the split it cannot decide for you — whole enterprise features get their own file, a gated match arm stays inline — in .agents/architecture-invariants.md. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * fix: tighten enterprise license checks Signed-off-by: Dennis Zhuang <killme2008@gmail.com> --------- Signed-off-by: Dennis Zhuang <killme2008@gmail.com> Signed-off-by: onepizzateam <palakjha916@gmail.com> * fix(operator): invalidate local cache after dropping view (#8748) Signed-off-by: WenyXu <wenymedia@gmail.com> Signed-off-by: onepizzateam <palakjha916@gmail.com> * chore!: gate soft-drop table behind the enterprise feature (#8747) * chore: gate soft-drop table behind the enterprise feature Soft-drop table becomes an enterprise-only feature: - metasrv rejects gc.experimental_soft_drop.enable=true at startup in non-enterprise builds, and ddl_soft_drop_enabled is hard-disabled without the enterprise feature as a second line of defense - the UNDROP TABLE parser/AST/statement variant, ADMIN purge_table() registration, and information_schema.recycle_bin registration are compiled out unless the enterprise feature is enabled - common-meta procedures, tombstone keys, and DdlTask serde stay unconditional for persisted-procedure recovery and wire compatibility - the [gc.experimental_soft_drop] section is removed from the OSS example config and generated docs (moving to the enterprise repo) - the soft-drop sqlness cases and their CI job are removed from OSS (moving to the enterprise repo); affected information_schema .result files are regenerated Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> * refactor: limit unused_variables allow to non-enterprise builds Addresses review comment: apply the allow via cfg_attr so enterprise builds still catch accidental unused variables in register_admin_only. Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> * refactor: include the config key in the soft-drop enterprise gate error Addresses review comment: name gc.experimental_soft_drop.enable in the startup validation error so users can locate the setting quickly when it is set via env vars or layered config. Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> * test: limit unused_mut allow to non-enterprise builds Addresses review comment: apply the allow via cfg_attr so enterprise builds still catch unused mut in the table_ddl_event test setup. Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> * feat: reject soft-drop DDL submissions in non-enterprise builds Addresses review comment: clients could bypass the SQL-level gates by submitting DdlTask::UndropTable or DdlTask::PurgeDroppedTable directly to the procedure service. Reject fresh submissions at the DdlManager boundary in non-enterprise builds while keeping the procedure loaders registered for crash recovery and wire compatibility. Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> * test: stop --enable-gc from enabling soft drop in the sqlness template Addresses review comment: the metasrv test template rendered [gc.experimental_soft_drop] enable = true under the generic --enable-gc flag, which non-enterprise metasrv now rejects at startup, making the documented --enable-gc mode unusable in OSS. Keep the flag scoped to plain GC; enterprise soft-drop coverage moves to the enterprise repo. Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> * fix: gate fresh soft-drop procedures Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> * test: gate soft-drop fallback coverage Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> * fix: gate soft-drop procedure implementation Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> * refactor: gate drop table soft-drop behavior Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> * refactor: gate expired soft-drop gc behavior Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> * ci: test enterprise table ddl lifecycle Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> * chore: mark purge_table as enterprise licensed The purge_table module is compiled only with the enterprise feature, so apply the Enterprise License header and register it with both license header configurations. Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> * chore: mark recycle_bin as enterprise licensed The recycle_bin module is compiled only with the enterprise feature, so apply the Enterprise License header and register it with both license header configurations. Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> * chore: mark soft-drop procedure sources as enterprise licensed The purge and undrop procedure implementations plus the recycle-bin test module compile only with the enterprise feature. Apply the Enterprise License header and register them with both license configurations. Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> --------- Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> Signed-off-by: onepizzateam <palakjha916@gmail.com> * feat: make frontend heartbeat extensible and lifecycle-safe (#8726) * feat: make frontend heartbeat extensible and lifecycle-safe Signed-off-by: jeremyhi <fengjiachun@gmail.com> * fix: isolate heartbeat extension response handlers Signed-off-by: jeremyhi <fengjiachun@gmail.com> * fix: cancel in-flight heartbeat response handling Signed-off-by: jeremyhi <fengjiachun@gmail.com> * test: cover heartbeat wire compatibility Signed-off-by: jeremyhi <fengjiachun@gmail.com> * fix: clean up failed heartbeat startup Signed-off-by: jeremyhi <fengjiachun@gmail.com> * fix: address frontend heartbeat review feedback Signed-off-by: jeremyhi <fengjiachun@gmail.com> --------- Signed-off-by: jeremyhi <fengjiachun@gmail.com> Signed-off-by: onepizzateam <palakjha916@gmail.com> * fix(meta): release region guards after drop rollback (#8751) Signed-off-by: WenyXu <wenymedia@gmail.com> Signed-off-by: onepizzateam <palakjha916@gmail.com> * refactor!: move native histogram config and `prom_validation_mode` to prom_store (#8744) * chore: adjust the position of experimental_enable_prometheus_native_histogram Signed-off-by: shuiyisong <xixing.sys@gmail.com> * chore: move prom_validation_mode as well Signed-off-by: shuiyisong <xixing.sys@gmail.com> --------- Signed-off-by: shuiyisong <xixing.sys@gmail.com> Signed-off-by: onepizzateam <palakjha916@gmail.com> * feat: add health-aware gRPC client routing (#8684) * feat: add gRPC client health routing Signed-off-by: WenyXu <wenymedia@gmail.com> * fix: harden gRPC client health routing Signed-off-by: WenyXu <wenymedia@gmail.com> * fix: defer gRPC client health checks until first use Signed-off-by: WenyXu <wenymedia@gmail.com> --------- Signed-off-by: WenyXu <wenymedia@gmail.com> Signed-off-by: onepizzateam <palakjha916@gmail.com> * feat(mito2): discard unflushed region data safely (#8600) * feat: support discarding unflushed region data Signed-off-by: evenyag <realevenyag@gmail.com> * fix(mito2): wake stalled writers after discard Signed-off-by: evenyag <realevenyag@gmail.com> * refactor(mito2): drop redundant manifest check for discarding unflushed data Signed-off-by: evenyag <realevenyag@gmail.com> --------- Signed-off-by: evenyag <realevenyag@gmail.com> Signed-off-by: onepizzateam <palakjha916@gmail.com> * docs: align wal.sync_period documented default with actual fallback (5s) (#8753) The example TOMLs and generated config.md documented the default of wal.sync_period as "10s", but since #5677 moved the WAL sync task to a background RepeatedTask, an unset sync_period falls back to 5s in RaftEngineLogStore. The two paths therefore had different fsync periods: deployments based on the example configs used 10s while bare configs used 5s. Align the documentation with the actual code behavior (5s) instead of changing the code fallback to 10s, so that no existing deployment silently gets a larger data-loss window on host power loss. - config/datanode.example.toml, config/standalone.example.toml: 10s -> 5s - config/config.md: regenerated via make config-docs - src/cmd/tests/load_config_test.rs: update assertions accordingly Signed-off-by: jeremyhi <fengjiachun@gmail.com> Signed-off-by: onepizzateam <palakjha916@gmail.com> * fix: support Utf8View labels in Prometheus response (#8754) Signed-off-by: evenyag <realevenyag@gmail.com> Signed-off-by: onepizzateam <palakjha916@gmail.com> * fix(query): validate merge scan remote schema (#8579) * fix(query): validate merge scan remote schema Signed-off-by: discord9 <discord9@163.com> * fix(query): treat JSON columns as schema-compatible across wire/decode forms CI (Sqlness json2_limit standalone + distributed) failed on the new remote-schema validation: a JSON column is Binary + extension metadata (ARROW:extension:name=greptime.json, greptime:type=Json) on the wire but decodes to Struct(...) with the extension metadata — validate_remote_schema compared raw arrow data_type and rejected it as a mismatch. Adds json_fields_compatible(): JSON fields are equal when name and nullability match, greptime:type matches, and the JSON2 settings (ARROW:extension:metadata type hints) match, ignoring the physical arrow type. Only JSON fields may bypass the raw-type comparison; non-JSON validation stays strict. Adds 4 regression tests mirroring the CI failure (wire-binary vs decoded-struct accepted both directions; different JSON2 settings rejected; JSON vs plain Binary rejected). Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --------- Signed-off-by: discord9 <discord9@163.com> Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> Signed-off-by: onepizzateam <palakjha916@gmail.com> * feat(grafana): add events dashboard (#8725) * feat(grafana): add events dashboard Signed-off-by: WenyXu <wenymedia@gmail.com> * fix(grafana): tolerate evolving event schemas Signed-off-by: WenyXu <wenymedia@gmail.com> * fix(grafana): address events dashboard review Signed-off-by: WenyXu <wenymedia@gmail.com> * fix(grafana): restore events dashboard panels Signed-off-by: WenyXu <wenymedia@gmail.com> * fix(grafana): bound events dashboard queries Signed-off-by: WenyXu <wenymedia@gmail.com> * fix(grafana): include historical event catalogs Signed-off-by: WenyXu <wenymedia@gmail.com> * fix(grafana): preserve events drill-down context Signed-off-by: WenyXu <wenymedia@gmail.com> * fix(grafana): correct events lifecycle outcomes Signed-off-by: WenyXu <wenymedia@gmail.com> * fix(grafana): handle empty event type ranges Signed-off-by: WenyXu <wenymedia@gmail.com> * fix(grafana): scope event catalogs to submissions Signed-off-by: WenyXu <wenymedia@gmail.com> * fix(grafana): handle empty events dashboard Signed-off-by: WenyXu <wenymedia@gmail.com> * fix(grafana): refresh event schema variables Signed-off-by: WenyXu <wenymedia@gmail.com> --------- Signed-off-by: WenyXu <wenymedia@gmail.com> Signed-off-by: onepizzateam <palakjha916@gmail.com> * refactor: separate a json2 extension type (#8745) Signed-off-by: luofucong <luofc@foxmail.com> Signed-off-by: onepizzateam <palakjha916@gmail.com> * feat: add admin function registrar (#8762) * feat: add admin function registrar Signed-off-by: jeremyhi <fengjiachun@gmail.com> * fix: reject admin function name collisions Signed-off-by: jeremyhi <fengjiachun@gmail.com> * chore: fix typo in admin function test Signed-off-by: jeremyhi <fengjiachun@gmail.com> --------- Signed-off-by: jeremyhi <fengjiachun@gmail.com> Signed-off-by: onepizzateam <palakjha916@gmail.com> * fix: information_schema.rs table initialization issue Signed-off-by: onepizzateam <palakjha916@gmail.com> * rustfmt fix Signed-off-by: onepizzateam <palakjha916@gmail.com> --------- Signed-off-by: onepizzateam <palakjha916@gmail.com> Signed-off-by: Palak Jha <palakjha916@gmail.com> Signed-off-by: polar <palakjha916@gmail.com> Signed-off-by: Dennis Zhuang <killme2008@gmail.com> Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> Signed-off-by: shuiyisong <xixing.sys@gmail.com> Signed-off-by: discord9 <discord9@163.com> Signed-off-by: WenyXu <wenymedia@gmail.com> Signed-off-by: xhwhis <hi@whis.me> Signed-off-by: lyang24 <lanqingy93@gmail.com> Signed-off-by: Lei, HUANG <mrsatangel@gmail.com> Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> Signed-off-by: jeremyhi <fengjiachun@gmail.com> Signed-off-by: evenyag <realevenyag@gmail.com> Signed-off-by: luofucong <luofc@foxmail.com> Co-authored-by: dennis zhuang <killme2008@gmail.com> Co-authored-by: Lei, HUANG <6406592+v0y4g3r@users.noreply.github.com> Co-authored-by: shuiyisong <113876041+shuiyisong@users.noreply.github.com> Co-authored-by: discord9 <discord9@163.com> Co-authored-by: Weny Xu <wenymedia@gmail.com> Co-authored-by: Ning Sun <sunng@protonmail.com> Co-authored-by: Whis Liao <xhwhis@gmail.com> Co-authored-by: Lanqing Yang <lanqingy93@gmail.com> Co-authored-by: sun <sunchang_long@163.com> Co-authored-by: Ning Sun <classicning@gmail.com> Co-authored-by: jeremyhi <jiachun_feng@proton.me> Co-authored-by: Yingwen <realevenyag@gmail.com> Co-authored-by: LFC <990479+MichaelScofield@users.noreply.github.com>
2026-08-18 12:08:22 +00:00 · 2026-08-06 12:54:59 +05:30
parent a28a0466b5
commit cabc2f6cc6
28 changed files with 767 additions and 81 deletions
@@ -14,6 +14,7 @@

 mod cluster_info;
 pub mod columns;
+pub mod flow_statistics;
 pub mod flows;
 mod information_memory_table;
 pub mod key_column_usage;
@@ -73,6 +74,7 @@ use crate::CatalogManager;
 use crate::error::{Error, Result};
 use crate::process_manager::ProcessManagerRef;
 use crate::system_schema::information_schema::cluster_info::InformationSchemaClusterInfo;
+use crate::system_schema::information_schema::flow_statistics::InformationSchemaFlowStatistics;
 use crate::system_schema::information_schema::flows::InformationSchemaFlows;
 use crate::system_schema::information_schema::information_memory_table::get_schema_columns;
 use crate::system_schema::information_schema::key_column_usage::InformationSchemaKeyColumnUsage;
@@ -271,6 +273,11 @@ impl SystemSchemaProviderInner for InformationSchemaProvider {
                self.catalog_manager.clone(),
                self.flow_metadata_manager.clone(),
            )) as _),
+            FLOW_STATISTICS => Some(Arc::new(InformationSchemaFlowStatistics::new(
+                self.catalog_name.clone(),
+                self.catalog_manager.clone(),
+                self.flow_metadata_manager.clone(),
+            )) as _),
            PROCEDURE_INFO => Some(
                Arc::new(procedure_info::InformationSchemaProcedureInfo::new(
                    self.catalog_manager.clone(),
@@ -406,6 +413,10 @@ impl InformationSchemaProvider {
            self.build_table(STATISTICS).unwrap(),
        );
        tables.insert(FLOWS.to_string(), self.build_table(FLOWS).unwrap());
+        tables.insert(
+            FLOW_STATISTICS.to_string(),
+            self.build_table(FLOW_STATISTICS).unwrap(),
+        );
        #[cfg(feature = "enterprise")]
        tables.insert(
            RECYCLE_BIN.to_string(),
@@ -0,0 +1,285 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::{Arc, Weak};
+
+use common_catalog::consts::INFORMATION_SCHEMA_FLOW_STATISTICS_TABLE_ID;
+use common_error::ext::BoxedError;
+use common_meta::key::FlowId;
+use common_meta::key::flow::FlowMetadataManager;
+use common_meta::key::flow::flow_state::FlowStat;
+use common_recordbatch::adapter::RecordBatchStreamAdapter;
+use common_recordbatch::{DfSendableRecordBatchStream, RecordBatch, SendableRecordBatchStream};
+use common_time::util::current_time_millis;
+use datafusion::execution::TaskContext;
+use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
+use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
+use datatypes::prelude::ConcreteDataType as CDT;
+use datatypes::scalars::ScalarVectorBuilder;
+use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
+use datatypes::timestamp::TimestampMillisecond;
+use datatypes::value::Value;
+use datatypes::vectors::{
+    Int64VectorBuilder, StringVectorBuilder, TimestampMillisecondVectorBuilder,
+    UInt32VectorBuilder, UInt64VectorBuilder, VectorRef,
+};
+use futures::TryStreamExt;
+use snafu::ResultExt;
+use store_api::storage::{ScanRequest, TableId};
+
+use crate::CatalogManager;
+use crate::error::{CreateRecordBatchSnafu, InternalSnafu, ListFlowsSnafu, Result};
+use crate::information_schema::{FLOW_STATISTICS, Predicates};
+use crate::system_schema::information_schema::InformationTable;
+use crate::system_schema::utils;
+
+const INIT_CAPACITY: usize = 42;
+
+// rows of information_schema.flow_statistics
+pub const FLOW_ID: &str = "flow_id";
+pub const FLOW_NAME: &str = "flow_name";
+pub const START_TIME: &str = "start_time";
+pub const LAST_EXECUTION_TIME: &str = "last_execution_time";
+pub const UPTIME_SECONDS: &str = "uptime_seconds";
+pub const STATE_SIZE: &str = "state_size";
+
+/// The `information_schema.flow_statistics` provides runtime statistics about flows.
+#[derive(Debug)]
+pub(super) struct InformationSchemaFlowStatistics {
+    schema: SchemaRef,
+    catalog_name: String,
+    catalog_manager: Weak<dyn CatalogManager>,
+    flow_metadata_manager: Arc<FlowMetadataManager>,
+}
+
+impl InformationSchemaFlowStatistics {
+    pub(super) fn new(
+        catalog_name: String,
+        catalog_manager: Weak<dyn CatalogManager>,
+        flow_metadata_manager: Arc<FlowMetadataManager>,
+    ) -> Self {
+        Self {
+            schema: Self::schema(),
+            catalog_name,
+            catalog_manager,
+            flow_metadata_manager,
+        }
+    }
+
+    pub(crate) fn schema() -> SchemaRef {
+        Arc::new(Schema::new(
+            vec![
+                (FLOW_ID, CDT::uint32_datatype(), false),
+                (FLOW_NAME, CDT::string_datatype(), false),
+                (START_TIME, CDT::timestamp_millisecond_datatype(), true),
+                (
+                    LAST_EXECUTION_TIME,
+                    CDT::timestamp_millisecond_datatype(),
+                    true,
+                ),
+                (UPTIME_SECONDS, CDT::int64_datatype(), true),
+                (STATE_SIZE, CDT::uint64_datatype(), true),
+            ]
+            .into_iter()
+            .map(|(name, ty, nullable)| ColumnSchema::new(name, ty, nullable))
+            .collect(),
+        ))
+    }
+
+    fn builder(&self) -> InformationSchemaFlowStatisticsBuilder {
+        InformationSchemaFlowStatisticsBuilder::new(
+            self.schema.clone(),
+            self.catalog_name.clone(),
+            self.catalog_manager.clone(),
+            &self.flow_metadata_manager,
+        )
+    }
+}
+
+impl InformationTable for InformationSchemaFlowStatistics {
+    fn table_id(&self) -> TableId {
+        INFORMATION_SCHEMA_FLOW_STATISTICS_TABLE_ID
+    }
+
+    fn table_name(&self) -> &'static str {
+        FLOW_STATISTICS
+    }
+
+    fn schema(&self) -> SchemaRef {
+        self.schema.clone()
+    }
+
+    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
+        let schema = self.schema.arrow_schema().clone();
+        let mut builder = self.builder();
+        let stream = Box::pin(DfRecordBatchStreamAdapter::new(
+            schema,
+            futures::stream::once(async move {
+                builder
+                    .make_flow_statistics(Some(request))
+                    .await
+                    .map(|x| x.into_df_record_batch())
+                    .map_err(|err| datafusion::error::DataFusionError::External(Box::new(err)))
+            }),
+        ));
+        Ok(Box::pin(
+            RecordBatchStreamAdapter::try_new(stream)
+                .map_err(BoxedError::new)
+                .context(InternalSnafu)?,
+        ))
+    }
+}
+
+/// Builds the `information_schema.flow_statistics` table row by row.
+struct InformationSchemaFlowStatisticsBuilder {
+    schema: SchemaRef,
+    catalog_name: String,
+    catalog_manager: Weak<dyn CatalogManager>,
+    flow_metadata_manager: Arc<FlowMetadataManager>,
+
+    flow_ids: UInt32VectorBuilder,
+    flow_names: StringVectorBuilder,
+    start_times: TimestampMillisecondVectorBuilder,
+    last_execution_times: TimestampMillisecondVectorBuilder,
+    uptime_seconds: Int64VectorBuilder,
+    state_sizes: UInt64VectorBuilder,
+}
+
+impl InformationSchemaFlowStatisticsBuilder {
+    fn new(
+        schema: SchemaRef,
+        catalog_name: String,
+        catalog_manager: Weak<dyn CatalogManager>,
+        flow_metadata_manager: &Arc<FlowMetadataManager>,
+    ) -> Self {
+        Self {
+            schema,
+            catalog_name,
+            catalog_manager,
+            flow_metadata_manager: flow_metadata_manager.clone(),
+
+            flow_ids: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
+            flow_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+            start_times: TimestampMillisecondVectorBuilder::with_capacity(INIT_CAPACITY),
+            last_execution_times: TimestampMillisecondVectorBuilder::with_capacity(INIT_CAPACITY),
+            uptime_seconds: Int64VectorBuilder::with_capacity(INIT_CAPACITY),
+            state_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
+        }
+    }
+
+    /// Construct the `information_schema.flow_statistics` virtual table.
+    async fn make_flow_statistics(&mut self, request: Option<ScanRequest>) -> Result<RecordBatch> {
+        let catalog_name = self.catalog_name.clone();
+        let predicates = Predicates::from_scan_request(&request);
+
+        let flow_info_manager = self.flow_metadata_manager.clone();
+
+        let mut stream = flow_info_manager
+            .flow_name_manager()
+            .flow_names(&catalog_name)
+            .await;
+
+        let flow_stat = {
+            let information_extension = utils::information_extension(&self.catalog_manager)?;
+            information_extension.flow_stats().await?
+        };
+
+        let now = current_time_millis();
+
+        while let Some((flow_name, flow_id)) = stream
+            .try_next()
+            .await
+            .map_err(BoxedError::new)
+            .context(ListFlowsSnafu {
+                catalog: &catalog_name,
+            })?
+        {
+            self.add_flow_statistic(&predicates, flow_id.flow_id(), &flow_name, &flow_stat, now);
+        }
+
+        self.finish()
+    }
+
+    fn add_flow_statistic(
+        &mut self,
+        predicates: &Predicates,
+        flow_id: FlowId,
+        flow_name: &str,
+        flow_stat: &Option<FlowStat>,
+        now: i64,
+    ) {
+        let row = [
+            (FLOW_ID, &Value::from(flow_id)),
+            (FLOW_NAME, &Value::from(flow_name.to_string())),
+        ];
+        if !predicates.eval(&row) {
+            return;
+        }
+
+        let start_time = flow_stat
+            .as_ref()
+            .and_then(|stat| stat.start_time_map.get(&flow_id).copied());
+
+        self.flow_ids.push(Some(flow_id));
+        self.flow_names.push(Some(flow_name));
+        self.start_times
+            .push(start_time.map(TimestampMillisecond::new));
+        self.last_execution_times
+            .push(flow_stat.as_ref().and_then(|stat| {
+                stat.last_exec_time_map
+                    .get(&flow_id)
+                    .map(|v| TimestampMillisecond::new(*v))
+            }));
+        self.uptime_seconds
+            .push(start_time.map(|start| ((now - start) / 1000).max(0)));
+        self.state_sizes.push(
+            flow_stat
+                .as_ref()
+                .and_then(|stat| stat.state_size.get(&flow_id).map(|v| *v as u64)),
+        );
+    }
+
+    fn finish(&mut self) -> Result<RecordBatch> {
+        let columns: Vec<VectorRef> = vec![
+            Arc::new(self.flow_ids.finish()),
+            Arc::new(self.flow_names.finish()),
+            Arc::new(self.start_times.finish()),
+            Arc::new(self.last_execution_times.finish()),
+            Arc::new(self.uptime_seconds.finish()),
+            Arc::new(self.state_sizes.finish()),
+        ];
+        RecordBatch::new(self.schema.clone(), columns).context(CreateRecordBatchSnafu)
+    }
+}
+
+impl DfPartitionStream for InformationSchemaFlowStatistics {
+    fn schema(&self) -> &arrow_schema::SchemaRef {
+        self.schema.arrow_schema()
+    }
+
+    fn execute(&self, _: Arc<TaskContext>) -> DfSendableRecordBatchStream {
+        let schema: Arc<arrow_schema::Schema> = self.schema.arrow_schema().clone();
+        let mut builder = self.builder();
+        Box::pin(DfRecordBatchStreamAdapter::new(
+            schema,
+            futures::stream::once(async move {
+                builder
+                    .make_flow_statistics(None)
+                    .await
+                    .map(|x| x.into_df_record_batch())
+                    .map_err(Into::into)
+            }),
+        ))
+    }
+}
@@ -44,6 +44,7 @@ pub const TABLE_CONSTRAINTS: &str = "table_constraints";
 pub const CLUSTER_INFO: &str = "cluster_info";
 pub const VIEWS: &str = "views";
 pub const FLOWS: &str = "flows";
+pub const FLOW_STATISTICS: &str = "flow_statistics";
 pub const PROCEDURE_INFO: &str = "procedure_info";
 pub const REGION_INFO: &str = "region_info";
 pub const REGION_STATISTICS: &str = "region_statistics";
@@ -345,9 +345,7 @@ mod tests {

    #[test]
    fn test_validate_exact_flow_state_key() {
-        let value = FlowStateValue::new(BTreeMap::new(), BTreeMap::new())
-            .try_as_raw_value()
-            .unwrap();
+        let value = FlowStateValue::default().try_as_raw_value().unwrap();

        validate_metadata_value(&flow_state_full_key(), &value).unwrap();
    }
@@ -120,6 +120,8 @@ pub const INFORMATION_SCHEMA_TABLE_SEMANTICS_TABLE_ID: u32 = 42;
 pub const INFORMATION_SCHEMA_STATISTICS_TABLE_ID: u32 = 43;
 /// id for information_schema.recycle_bin
 pub const INFORMATION_SCHEMA_RECYCLE_BIN_TABLE_ID: u32 = 44;
+/// id for information_schema.flow_statistics
+pub const INFORMATION_SCHEMA_FLOW_STATISTICS_TABLE_ID: u32 = 45;

 // ----- End of information_schema tables -----

@@ -93,22 +93,28 @@ impl<'a> MetadataKey<'a, FlowStateKey> for FlowStateKey {
 }

 /// The value of flow state size
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
 pub struct FlowStateValue {
    /// For each key, the bytes of the state in memory
    pub state_size: BTreeMap<FlowId, usize>,
    /// For each key, the last execution time of flow in unix timestamp milliseconds.
    pub last_exec_time_map: BTreeMap<FlowId, i64>,
+    /// For each flow, the time the flow first executed, in unix timestamp milliseconds.
+    /// TODO(#7987-followup): not yet propagated via the heartbeat wire format in distributed mode.
+    #[serde(default)]
+    pub start_time_map: BTreeMap<FlowId, i64>,
 }

 impl FlowStateValue {
    pub fn new(
        state_size: BTreeMap<FlowId, usize>,
        last_exec_time_map: BTreeMap<FlowId, i64>,
+        start_time_map: BTreeMap<FlowId, i64>,
    ) -> Self {
        Self {
            state_size,
            last_exec_time_map,
+            start_time_map,
        }
    }
 }
@@ -147,12 +153,15 @@ impl FlowStateManager {
 }

 /// Flow's state report, send regularly through heartbeat message
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Default)]
 pub struct FlowStat {
    /// For each key, the bytes of the state in memory
    pub state_size: BTreeMap<u32, usize>,
    /// For each key, the last execution time of flow in unix timestamp milliseconds.
    pub last_exec_time_map: BTreeMap<FlowId, i64>,
+    /// For each flow, the time the flow first executed, in unix timestamp milliseconds.
+    /// TODO(#7987-followup): not yet propagated via the heartbeat wire format in distributed mode.
+    pub start_time_map: BTreeMap<FlowId, i64>,
 }

 impl From<FlowStateValue> for FlowStat {
@@ -160,6 +169,7 @@ impl From<FlowStateValue> for FlowStat {
        Self {
            state_size: value.state_size,
            last_exec_time_map: value.last_exec_time_map,
+            start_time_map: value.start_time_map,
        }
    }
 }
@@ -169,6 +179,59 @@ impl From<FlowStat> for FlowStateValue {
        Self {
            state_size: value.state_size,
            last_exec_time_map: value.last_exec_time_map,
+            start_time_map: value.start_time_map,
        }
    }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::collections::BTreeMap;
+
+    use crate::key::FlowId;
+    use crate::key::flow::flow_state::FlowStateValue;
+
+    #[test]
+    fn test_deserialize_legacy_flow_state_value() {
+        // Legacy format: only state_size and last_exec_time_map are present,
+        // without the start_time_map field added in PR #8392.
+        let legacy_json =
+            r#"{"state_size":{"1":1024,"2":2048},"last_exec_time_map":{"1":1700000000000}}"#;
+        let value: FlowStateValue = serde_json::from_str(legacy_json).unwrap();
+
+        let mut expected_state_size = BTreeMap::new();
+        expected_state_size.insert(FlowId::from(1u32), 1024usize);
+        expected_state_size.insert(FlowId::from(2u32), 2048usize);
+        assert_eq!(value.state_size, expected_state_size);
+
+        let mut expected_last_exec_time_map = BTreeMap::new();
+        expected_last_exec_time_map.insert(FlowId::from(1u32), 1700000000000i64);
+        assert_eq!(value.last_exec_time_map, expected_last_exec_time_map);
+
+        // serde(default) kicks in: old persisted data must not break,
+        // and the new field defaults to empty.
+        assert!(value.start_time_map.is_empty());
+    }
+
+    #[test]
+    fn test_flow_state_value_roundtrip_includes_start_time_map() {
+        let mut state_size = BTreeMap::new();
+        state_size.insert(FlowId::from(1u32), 1024usize);
+        let mut last_exec_time_map = BTreeMap::new();
+        last_exec_time_map.insert(FlowId::from(1u32), 1700000000000i64);
+        let mut start_time_map = BTreeMap::new();
+        start_time_map.insert(FlowId::from(1u32), 1700000000000i64);
+
+        let value = FlowStateValue {
+            state_size,
+            last_exec_time_map,
+            start_time_map,
+        };
+
+        let json = serde_json::to_string(&value).unwrap();
+        assert!(json.contains("start_time_map"));
+
+        let decoded: FlowStateValue = serde_json::from_str(&json).unwrap();
+        assert_eq!(decoded, value);
+    }
+}
@@ -174,9 +174,13 @@ impl FlowDualEngine {
        let mut last_exec_time_map = streaming.last_exec_time_map;
        last_exec_time_map.extend(batching.last_exec_time_map);

+        let mut start_time_map = streaming.start_time_map;
+        start_time_map.extend(batching.start_time_map);
+
        FlowStat {
            state_size,
            last_exec_time_map,
+            start_time_map,
        }
    }

@@ -21,33 +21,27 @@ use crate::engine::FlowStatProvider;

 impl FlowStatProvider for StreamingEngine {
    async fn flow_stat(&self) -> FlowStat {
-        let mut full_report = BTreeMap::new();
+        let mut state_size_map = BTreeMap::new();
        let mut last_exec_time_map = BTreeMap::new();
+        let mut start_time_map = BTreeMap::new();

        for worker in self.worker_handles.iter() {
-            match worker.get_state_size().await {
-                Ok(state_size) => {
-                    full_report.extend(state_size.into_iter().map(|(k, v)| (k as u32, v)));
+            match worker.get_full_flow_stat().await {
+                Ok((sizes, exec_times, start_times)) => {
+                    state_size_map.extend(sizes.into_iter().map(|(k, v)| (k as u32, v)));
+                    last_exec_time_map.extend(exec_times.into_iter().map(|(k, v)| (k as u32, v)));
+                    start_time_map.extend(start_times.into_iter().map(|(k, v)| (k as u32, v)));
                }
                Err(err) => {
-                    common_telemetry::error!(err; "Get flow stat size error");
-                }
-            }
-
-            match worker.get_last_exec_time_map().await {
-                Ok(last_exec_time) => {
-                    last_exec_time_map
-                        .extend(last_exec_time.into_iter().map(|(k, v)| (k as u32, v)));
-                }
-                Err(err) => {
-                    common_telemetry::error!(err; "Get last exec time error");
+                    common_telemetry::error!(err; "Get full flow stat error");
                }
            }
        }

        FlowStat {
-            state_size: full_report,
+            state_size: state_size_map,
            last_exec_time_map,
+            start_time_map,
        }
    }
 }
@@ -202,30 +202,24 @@ impl WorkerHandle {
        }
    }

-    pub async fn get_state_size(&self) -> Result<BTreeMap<FlowId, usize>, Error> {
+    pub async fn get_full_flow_stat(
+        &self,
+    ) -> Result<
+        (
+            BTreeMap<FlowId, usize>,
+            BTreeMap<FlowId, i64>,
+            BTreeMap<FlowId, i64>,
+        ),
+        Error,
+    > {
        let ret = self
            .itc_client
-            .call_with_resp(Request::QueryStateSize)
+            .call_with_resp(Request::QueryFullFlowStat)
            .await?;
-        ret.into_query_state_size().map_err(|ret| {
+        ret.into_query_full_flow_stat().map_err(|ret| {
            InternalSnafu {
                reason: format!(
-                    "Flow Node/Worker itc failed, expect Response::QueryStateSize, found {ret:?}"
-                ),
-            }
-            .build()
-        })
-    }
-
-    pub async fn get_last_exec_time_map(&self) -> Result<BTreeMap<FlowId, i64>, Error> {
-        let ret = self
-            .itc_client
-            .call_with_resp(Request::QueryLastExecTimeMap)
-            .await?;
-        ret.into_query_last_exec_time_map().map_err(|ret| {
-            InternalSnafu {
-                reason: format!(
-                    "Flow Node/Worker get_last_exec_time_map failed, expect Response::QueryLastExecTimeMap, found {ret:?}"
+                    "Flow Node/Worker get_full_flow_stat failed, expected Response::QueryFullFlowStat, found {ret:?}"
                ),
            }
            .build()
@@ -408,21 +402,24 @@ impl<'s> Worker<'s> {
                Some(Response::ContainTask { result: ret })
            }
            Request::Shutdown => return Err(()),
-            Request::QueryStateSize => {
-                let mut ret = BTreeMap::new();
+            Request::QueryFullFlowStat => {
+                let mut state_size = BTreeMap::new();
+                let mut last_exec_time_map = BTreeMap::new();
+                let mut start_time_map = BTreeMap::new();
                for (flow_id, task_state) in self.task_states.iter() {
-                    ret.insert(*flow_id, task_state.state.get_state_size());
-                }
-                Some(Response::QueryStateSize { result: ret })
-            }
-            Request::QueryLastExecTimeMap => {
-                let mut ret = BTreeMap::new();
-                for (flow_id, task_state) in self.task_states.iter() {
-                    if let Some(last_exec_time) = task_state.state.last_exec_time() {
-                        ret.insert(*flow_id, last_exec_time);
+                    state_size.insert(*flow_id, task_state.state.get_state_size());
+                    if let Some(t) = task_state.state.last_exec_time() {
+                        last_exec_time_map.insert(*flow_id, t);
+                    }
+                    if let Some(t) = task_state.state.start_time() {
+                        start_time_map.insert(*flow_id, t);
                    }
                }
-                Some(Response::QueryLastExecTimeMap { result: ret })
+                Some(Response::QueryFullFlowStat {
+                    state_size,
+                    last_exec_time_map,
+                    start_time_map,
+                })
            }
        };
        Ok(ret)
@@ -455,8 +452,7 @@ pub enum Request {
        flow_id: FlowId,
    },
    Shutdown,
-    QueryStateSize,
-    QueryLastExecTimeMap,
+    QueryFullFlowStat,
 }

 #[derive(Debug, EnumAsInner)]
@@ -472,13 +468,10 @@ enum Response {
        result: bool,
    },
    RunAvail,
-    QueryStateSize {
-        /// each flow tasks' state size
-        result: BTreeMap<FlowId, usize>,
-    },
-    QueryLastExecTimeMap {
-        /// each flow tasks' last execution time
-        result: BTreeMap<FlowId, i64>,
+    QueryFullFlowStat {
+        state_size: BTreeMap<FlowId, usize>,
+        last_exec_time_map: BTreeMap<FlowId, i64>,
+        start_time_map: BTreeMap<FlowId, i64>,
    },
 }

@@ -604,7 +597,8 @@ mod test {
        );
        tx.send(Batch::empty()).unwrap();
        handle.run_available(0, true).await.unwrap();
-        assert_eq!(handle.get_state_size().await.unwrap().len(), 1);
+        let (state_size, _, _) = handle.get_full_flow_stat().await.unwrap();
+        assert_eq!(state_size.len(), 1);
        assert_eq!(sink_rx.recv().await.unwrap(), Batch::empty());
        drop(handle);
        worker_thread_handle.join().unwrap();
@@ -414,14 +414,24 @@ impl BatchingEngine {

 impl FlowStatProvider for BatchingEngine {
    async fn flow_stat(&self) -> FlowStat {
+        let runtime = self.runtime.read().await;
+        let mut last_exec_time_map = BTreeMap::new();
+        let mut start_time_map = BTreeMap::new();
+
+        for (flow_id, task) in runtime.tasks.iter() {
+            let id = *flow_id as u32;
+            if let Some(ts) = task.last_execution_time_millis() {
+                last_exec_time_map.insert(id, ts);
+            }
+            if let Some(ts) = task.start_time_millis() {
+                start_time_map.insert(id, ts);
+            }
+        }
+
        FlowStat {
            state_size: BTreeMap::new(),
-            last_exec_time_map: self
-                .get_last_exec_time_map()
-                .await
-                .into_iter()
-                .map(|(flow_id, timestamp)| (flow_id as u32, timestamp))
-                .collect(),
+            last_exec_time_map,
+            start_time_map,
        }
    }
 }
@@ -46,6 +46,8 @@ pub struct TaskState {
    last_query_duration: Duration,
    /// Last successful execution time in unix timestamp milliseconds.
    last_exec_time_millis: Option<i64>,
+    /// First execution time in unix timestamp milliseconds, set once.
+    start_time_millis: Option<i64>,
    /// Dirty Time windows need to be updated
    /// mapping of `start -> end` and non-overlapping
    pub(crate) dirty_time_windows: DirtyTimeWindows,
@@ -79,6 +81,7 @@ impl TaskState {
            last_update_time: Instant::now(),
            last_query_duration: Duration::from_secs(0),
            last_exec_time_millis: None,
+            start_time_millis: None,
            dirty_time_windows,
            checkpoint_mode: CheckpointMode::FullSnapshot,
            pending_fenced_repair: None,
@@ -90,8 +93,18 @@ impl TaskState {
        }
    }

-    /// called after last query is done
-    /// `is_succ` indicate whether the last query is successful
+    /// Record the first-execution start time. Call this once, just before
+    /// the first frontend query is dispatched, not after it completes.
+    pub fn record_start_time_if_first(&mut self) {
+        if self.start_time_millis.is_none() {
+            // start_time is recorded just before the first frontend query is dispatched
+            // (pre-execution), so it may be marginally earlier than the streaming engine's
+            // start_time which is set post-execution. Both are valid approximations of
+            // "when this flow first ran".
+            self.start_time_millis = Some(common_time::util::current_time_millis());
+        }
+    }
+
    pub fn after_query_exec(&mut self, elapsed: Duration, is_succ: bool) {
        self.exec_state = ExecState::Idle;
        self.last_query_duration = elapsed;
@@ -105,6 +118,11 @@ impl TaskState {
        self.last_exec_time_millis
    }

+    /// First execution time in unix timestamp milliseconds, set once.
+    pub fn start_time_millis(&self) -> Option<i64> {
+        self.start_time_millis
+    }
+
    pub fn checkpoint_mode(&self) -> CheckpointMode {
        self.checkpoint_mode
    }
@@ -290,6 +290,10 @@ impl BatchingTask {
        self.state.read().unwrap().last_execution_time_millis()
    }

+    pub fn start_time_millis(&self) -> Option<i64> {
+        self.state.read().unwrap().start_time_millis()
+    }
+
    /// Collect flow-related extensions from the task's query context that should be
    /// forwarded to the frontend (e.g. scheduled time).
    fn frontend_extensions(&self) -> HashMap<String, String> {
@@ -717,6 +721,10 @@ impl BatchingTask {
            };

            let snapshot_seqs = coverage.snapshot_seqs();
+            {
+                let mut state = self.state.write().unwrap();
+                state.record_start_time_if_first();
+            }
            frontend_client
                .query_with_terminal_metrics(
                    catalog,
@@ -47,6 +47,8 @@ pub struct DataflowState {
    expire_after: Option<Timestamp>,
    /// the last time each subgraph executed
    last_exec_time: Option<Timestamp>,
+    /// the time the flow first executed, in unix timestamp milliseconds
+    start_time: Option<Timestamp>,
 }

 impl DataflowState {
@@ -120,11 +122,21 @@ impl DataflowState {

    pub fn set_last_exec_time(&mut self, time: Timestamp) {
        self.last_exec_time = Some(time);
+        if self.start_time.is_none() {
+            // start_time is recorded at the completion of the first execution
+            // (post-execution), consistent with how last_exec_time is recorded.
+            self.start_time = Some(time);
+        }
    }

    pub fn last_exec_time(&self) -> Option<Timestamp> {
        self.last_exec_time
    }
+
+    /// Returns the time the flow first executed, in unix timestamp milliseconds.
+    pub fn start_time(&self) -> Option<Timestamp> {
+        self.start_time
+    }
 }

 #[derive(Debug, Clone)]
@@ -1559,6 +1559,11 @@ pub fn check_permission(
        Statement::ShowFlows(stmt) => {
            validate_db_permission!(stmt, query_ctx);
        }
+        Statement::ShowFlowStatus(_stmt) => {
+            // Flow statistics are organized based on the catalog dimension and
+            // filtered by the current catalog, so there is no need to check the
+            // permission of the database(schema).
+        }
        #[cfg(feature = "enterprise")]
        Statement::ShowTriggers(_stmt) => {
            // The trigger is organized based on the catalog dimension, so there
@@ -55,7 +55,11 @@ impl HeartbeatHandler for FlowStateHandler {
                .iter()
                .map(|(k, v)| (*k, *v))
                .collect();
-            let value: FlowStateValue = FlowStateValue::new(state_size, last_exec_time_map);
+            // TODO(#7987-followup): start_time_map is not yet propagated through the heartbeat
+            // wire format (`api::v1::meta::FlowStat`); it will always be empty in distributed
+            // mode until a follow-up PR adds heartbeat propagation.
+            let value: FlowStateValue =
+                FlowStateValue::new(state_size, last_exec_time_map, Default::default());
            self.flow_state_manager
                .put(value)
                .await
@@ -259,6 +259,7 @@ impl StatementExecutor {
            Statement::ShowViews(stmt) => self.show_views(stmt, query_ctx).await,

            Statement::ShowFlows(stmt) => self.show_flows(stmt, query_ctx).await,
+            Statement::ShowFlowStatus(stmt) => self.show_flow_status(stmt, query_ctx).await,

            #[cfg(feature = "enterprise")]
            Statement::ShowTriggers(stmt) => self.show_triggers(stmt, query_ctx).await,
@@ -26,8 +26,9 @@ use sql::ast::ObjectNamePartExt;
 use sql::statements::OptionMap;
 use sql::statements::create::Partitions;
 use sql::statements::show::{
-    ShowColumns, ShowCreateFlow, ShowCreateView, ShowDatabases, ShowFlows, ShowIndex, ShowKind,
-    ShowProcessList, ShowRegion, ShowTableStatus, ShowTables, ShowVariables, ShowViews,
+    ShowColumns, ShowCreateFlow, ShowCreateView, ShowDatabases, ShowFlowStatus, ShowFlows,
+    ShowIndex, ShowKind, ShowProcessList, ShowRegion, ShowTableStatus, ShowTables, ShowVariables,
+    ShowViews,
 };
 use table::TableRef;
 use table::metadata::{TableInfo, TableType};
@@ -250,6 +251,17 @@ impl StatementExecutor {
            .context(ExecuteStatementSnafu)
    }

+    #[tracing::instrument(skip_all)]
+    pub(super) async fn show_flow_status(
+        &self,
+        stmt: ShowFlowStatus,
+        query_ctx: QueryContextRef,
+    ) -> Result<Output> {
+        query::sql::show_flow_status(stmt, &self.query_engine, &self.catalog_manager, query_ctx)
+            .await
+            .context(ExecuteStatementSnafu)
+    }
+
    #[cfg(feature = "enterprise")]
    #[tracing::instrument(skip_all)]
    pub(super) async fn show_triggers(
@@ -20,8 +20,9 @@ use std::sync::Arc;

 use catalog::CatalogManagerRef;
 use catalog::information_schema::{
-    CHARACTER_SETS, COLLATIONS, COLUMNS, FLOWS, REGION_PEERS, SCHEMATA, STATISTICS, TABLES, VIEWS,
-    columns, flows, process_list, region_peers, schemata, statistics, tables,
+    CHARACTER_SETS, COLLATIONS, COLUMNS, FLOW_STATISTICS, FLOWS, REGION_PEERS, SCHEMATA,
+    STATISTICS, TABLES, VIEWS, columns, flow_statistics, flows, process_list, region_peers,
+    schemata, statistics, tables,
 };
 use common_catalog::consts::{
    INFORMATION_SCHEMA_NAME, SEMANTIC_TYPE_FIELD, SEMANTIC_TYPE_PRIMARY_KEY,
@@ -57,8 +58,8 @@ use sql::parser::ParserContext;
 use sql::statements::OptionMap;
 use sql::statements::create::{CreateDatabase, CreateFlow, CreateView, Partitions, SqlOrTql};
 use sql::statements::show::{
-    ShowColumns, ShowDatabases, ShowFlows, ShowIndex, ShowKind, ShowProcessList, ShowRegion,
-    ShowTableStatus, ShowTables, ShowVariables, ShowViews,
+    ShowColumns, ShowDatabases, ShowFlowStatus, ShowFlows, ShowIndex, ShowKind, ShowProcessList,
+    ShowRegion, ShowTableStatus, ShowTables, ShowVariables, ShowViews,
 };
 use sql::statements::statement::Statement;
 use sqlparser::ast::ObjectName;
@@ -971,6 +972,45 @@ pub async fn show_flows(
    .await
 }

+/// Execute [`ShowFlowStatus`] statement and return the [`Output`] if success.
+pub async fn show_flow_status(
+    stmt: ShowFlowStatus,
+    query_engine: &QueryEngineRef,
+    catalog_manager: &CatalogManagerRef,
+    query_ctx: QueryContextRef,
+) -> Result<Output> {
+    let projects = vec![
+        (flow_statistics::FLOW_ID, flow_statistics::FLOW_ID),
+        (flow_statistics::FLOW_NAME, flow_statistics::FLOW_NAME),
+        (flow_statistics::START_TIME, flow_statistics::START_TIME),
+        (
+            flow_statistics::LAST_EXECUTION_TIME,
+            flow_statistics::LAST_EXECUTION_TIME,
+        ),
+        (
+            flow_statistics::UPTIME_SECONDS,
+            flow_statistics::UPTIME_SECONDS,
+        ),
+        (flow_statistics::STATE_SIZE, flow_statistics::STATE_SIZE),
+    ];
+    let like_field = Some(flow_statistics::FLOW_NAME);
+    let sort = vec![col(flow_statistics::FLOW_NAME).sort(true, true)];
+
+    query_from_information_schema_table(
+        query_engine,
+        catalog_manager,
+        query_ctx,
+        FLOW_STATISTICS,
+        vec![],
+        projects,
+        vec![],
+        like_field,
+        sort,
+        stmt.kind,
+    )
+    .await
+}
+
 #[cfg(feature = "enterprise")]
 pub async fn show_triggers(
    stmt: sql::statements::show::trigger::ShowTriggers,
@@ -600,6 +600,53 @@ fn describe_fields(
                format.format_for(1),
            ),
        ]),
+        // SHOW FLOW STATUS returns six columns; return their descriptions so
+        // prepared/extended-protocol clients receive the correct row description.
+        SqlPlan::Statement(Statement::ShowFlowStatus(_), _) => Ok(vec![
+            FieldInfo::new(
+                "flow_id".to_string(),
+                None,
+                None,
+                Type::INT8, // matches type_gt_to_pg(UInt32) — do not use INT4
+                format.format_for(0),
+            ),
+            FieldInfo::new(
+                "flow_name".to_string(),
+                None,
+                None,
+                Type::TEXT,
+                format.format_for(1),
+            ),
+            FieldInfo::new(
+                "start_time".to_string(),
+                None,
+                None,
+                Type::TIMESTAMP,
+                format.format_for(2),
+            ),
+            FieldInfo::new(
+                "last_execution_time".to_string(),
+                None,
+                None,
+                Type::TIMESTAMP,
+                format.format_for(3),
+            ),
+            FieldInfo::new(
+                "uptime_seconds".to_string(),
+                None,
+                None,
+                Type::INT8,
+                format.format_for(4),
+            ),
+            FieldInfo::new(
+                "state_size".to_string(),
+                None,
+                None,
+                Type::NUMERIC,
+                format.format_for(5),
+            ),
+        ]),
+
        // single column show statements
        SqlPlan::Statement(
            Statement::ShowTables(_) | Statement::ShowFlows(_) | Statement::ShowViews(_),
@@ -26,8 +26,8 @@ use crate::error::{
 use crate::parser::ParserContext;
 use crate::statements::show::{
    ShowColumns, ShowCreateDatabase, ShowCreateFlow, ShowCreateTable, ShowCreateTableVariant,
-    ShowCreateView, ShowDatabases, ShowFlows, ShowIndex, ShowKind, ShowProcessList, ShowRegion,
-    ShowSearchPath, ShowStatus, ShowTableStatus, ShowTables, ShowVariables, ShowViews,
+    ShowCreateView, ShowDatabases, ShowFlowStatus, ShowFlows, ShowIndex, ShowKind, ShowProcessList,
+    ShowRegion, ShowSearchPath, ShowStatus, ShowTableStatus, ShowTables, ShowVariables, ShowViews,
 };
 use crate::statements::statement::Statement;

@@ -57,6 +57,12 @@ impl ParserContext<'_> {
            self.parse_show_views()
        } else if self.consume_token("FLOWS") {
            self.parse_show_flows()
+        } else if self.consume_token("FLOW") {
+            if self.consume_token("STATUS") {
+                self.parse_show_flow_status()
+            } else {
+                self.unsupported(self.peek_token_as_string())
+            }
        } else if self.matches_keyword(Keyword::CHARSET) {
            self.parser.next_token();
            Ok(Statement::ShowCharset(self.parse_show_kind()?))
@@ -587,6 +593,12 @@ impl ParserContext<'_> {
        Ok(Statement::ShowFlows(ShowFlows { kind, database }))
    }

+    fn parse_show_flow_status(&mut self) -> Result<Statement> {
+        let kind = self.parse_show_kind()?;
+
+        Ok(Statement::ShowFlowStatus(ShowFlowStatus { kind }))
+    }
+
    fn parse_show_processlist(&mut self, full: bool) -> Result<Statement> {
        match self.parser.next_token().token {
            Token::EOF | Token::SemiColon => {
@@ -1250,6 +1262,22 @@ mod tests {
        assert_eq!(sql, stmts[0].to_string());
    }

+    #[test]
+    pub fn test_show_flow_status() {
+        let sql = "SHOW FLOW STATUS";
+        let result =
+            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
+        let stmts = result.unwrap();
+        assert_eq!(1, stmts.len());
+        assert_eq!(
+            stmts[0],
+            Statement::ShowFlowStatus(ShowFlowStatus {
+                kind: ShowKind::All,
+            })
+        );
+        assert_eq!(sql, stmts[0].to_string());
+    }
+
    #[test]
    pub fn test_show_processlist() {
        let sql = "SHOW PROCESSLIST";
@@ -256,6 +256,21 @@ impl Display for ShowFlows {
    }
 }

+/// SQL structure for `SHOW FLOW STATUS`.
+#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut, Serialize)]
+pub struct ShowFlowStatus {
+    pub kind: ShowKind,
+}
+
+impl Display for ShowFlowStatus {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "SHOW FLOW STATUS")?;
+        format_kind!(self, f);
+
+        Ok(())
+    }
+}
+
 /// SQL structure for `SHOW CREATE VIEW`.
 #[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut, Serialize)]
 pub struct ShowCreateView {
@@ -40,8 +40,8 @@ use crate::statements::query::Query;
 use crate::statements::set_variables::SetVariables;
 use crate::statements::show::{
    ShowColumns, ShowCreateDatabase, ShowCreateFlow, ShowCreateTable, ShowCreateView,
-    ShowDatabases, ShowFlows, ShowIndex, ShowKind, ShowProcessList, ShowRegion, ShowSearchPath,
-    ShowStatus, ShowTableStatus, ShowTables, ShowVariables, ShowViews,
+    ShowDatabases, ShowFlowStatus, ShowFlows, ShowIndex, ShowKind, ShowProcessList, ShowRegion,
+    ShowSearchPath, ShowStatus, ShowTableStatus, ShowTables, ShowVariables, ShowViews,
 };
 use crate::statements::tql::Tql;
 use crate::statements::truncate::TruncateTable;
@@ -118,6 +118,8 @@ pub enum Statement {
    ShowCreateTrigger(crate::statements::show::trigger::ShowCreateTrigger),
    /// SHOW FLOWS
    ShowFlows(ShowFlows),
+    /// SHOW FLOW STATUS
+    ShowFlowStatus(ShowFlowStatus),
    // SHOW TRIGGERS
    #[cfg(feature = "enterprise")]
    ShowTriggers(crate::statements::show::trigger::ShowTriggers),
@@ -178,6 +180,7 @@ impl Statement {
            | Statement::ShowCreateTable(_)
            | Statement::ShowCreateFlow(_)
            | Statement::ShowFlows(_)
+            | Statement::ShowFlowStatus(_)
            | Statement::ShowCreateView(_)
            | Statement::ShowStatus(_)
            | Statement::ShowSearchPath(_)
@@ -268,6 +271,7 @@ impl Display for Statement {
            #[cfg(feature = "enterprise")]
            Statement::ShowCreateTrigger(s) => s.fmt(f),
            Statement::ShowFlows(s) => s.fmt(f),
+            Statement::ShowFlowStatus(s) => s.fmt(f),
            #[cfg(feature = "enterprise")]
            Statement::ShowTriggers(s) => s.fmt(f),
            Statement::ShowCreateDatabase(s) => s.fmt(f),
@@ -335,6 +335,7 @@ fn extract_tables_from_statement(stmt: &Statement, names: &mut HashSet<ObjectNam
        | Statement::ShowCreateDatabase(_)
        | Statement::ShowCreateFlow(_)
        | Statement::ShowFlows(_)
+        | Statement::ShowFlowStatus(_)
        | Statement::ShowStatus(_)
        | Statement::ShowSearchPath(_)
        | Statement::ShowViews(_)
@@ -0,0 +1,76 @@
+CREATE TABLE numbers_input_flow_status (
+    "number" INT,
+    ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    PRIMARY KEY(number),
+    TIME INDEX(ts)
+);
+
+Affected Rows: 0
+
+CREATE FLOW test_flow_status SINK TO out_flow_status AS
+SELECT
+    sum(number),
+    date_bin(INTERVAL '1 second', ts, '2021-07-01 00:00:00') as time_window
+FROM
+    numbers_input_flow_status
+GROUP BY
+    time_window;
+
+Affected Rows: 0
+
+INSERT INTO
+    numbers_input_flow_status
+VALUES
+    (20, "2021-07-01 00:00:00.200"),
+    (22, "2021-07-01 00:00:00.600");
+
+Affected Rows: 2
+
+-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED  |
+ADMIN FLUSH_FLOW('test_flow_status');
+
+--------------------------------------+
+| ADMIN FLUSH_FLOW('test_flow_status') |
+--------------------------------------+
+|  FLOW_FLUSHED  |
+--------------------------------------+
+
+-- flow_name is deterministic; the rest of the columns are runtime dependent.
+SELECT flow_name FROM information_schema.flow_statistics WHERE flow_name = 'test_flow_status';
+
+------------------+
+| flow_name        |
+------------------+
+| test_flow_status |
+------------------+
+
+-- SHOW FLOW STATUS is exercised below via the empty-result form (six-column header).
+-- Full row values differ between standalone (populated) and distributed (heartbeat
+-- has not yet propagated start_time), so assert only the stable columns here.
+-- SQLNESS REPLACE (\|\s+[0-9]+\s+\|) | ID |
+SELECT flow_id, flow_name FROM information_schema.flow_statistics WHERE flow_name = 'test_flow_status';
+
+---------+------------------+
+| flow_id | flow_name        |
+---------+------------------+
+| ID | test_flow_status |
+---------+------------------+
+
+-- the like filter matches against flow_name; no matching flow returns empty.
+SHOW FLOW STATUS LIKE 'no_such_flow';
+
++
++
+
+DROP FLOW test_flow_status;
+
+Affected Rows: 0
+
+DROP TABLE numbers_input_flow_status;
+
+Affected Rows: 0
+
+DROP TABLE out_flow_status;
+
+Affected Rows: 0
+
@@ -0,0 +1,42 @@
+CREATE TABLE numbers_input_flow_status (
+    "number" INT,
+    ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    PRIMARY KEY(number),
+    TIME INDEX(ts)
+);
+
+CREATE FLOW test_flow_status SINK TO out_flow_status AS
+SELECT
+    sum(number),
+    date_bin(INTERVAL '1 second', ts, '2021-07-01 00:00:00') as time_window
+FROM
+    numbers_input_flow_status
+GROUP BY
+    time_window;
+
+INSERT INTO
+    numbers_input_flow_status
+VALUES
+    (20, "2021-07-01 00:00:00.200"),
+    (22, "2021-07-01 00:00:00.600");
+
+-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED  |
+ADMIN FLUSH_FLOW('test_flow_status');
+
+-- flow_name is deterministic; the rest of the columns are runtime dependent.
+SELECT flow_name FROM information_schema.flow_statistics WHERE flow_name = 'test_flow_status';
+
+-- SHOW FLOW STATUS is exercised below via the empty-result form (six-column header).
+-- Full row values differ between standalone (populated) and distributed (heartbeat
+-- has not yet propagated start_time), so assert only the stable columns here.
+-- SQLNESS REPLACE (\|\s+[0-9]+\s+\|) | ID |
+SELECT flow_id, flow_name FROM information_schema.flow_statistics WHERE flow_name = 'test_flow_status';
+
+-- the like filter matches against flow_name; no matching flow returns empty.
+SHOW FLOW STATUS LIKE 'no_such_flow';
+
+DROP FLOW test_flow_status;
+
+DROP TABLE numbers_input_flow_status;
+
+DROP TABLE out_flow_status;
@@ -39,6 +39,7 @@ SHOW TABLES;
 | engines                               |
 | events                                |
 | files                                 |
+| flow_statistics                       |
 | flows                                 |
 | global_status                         |
 | key_column_usage                      |
@@ -92,6 +93,7 @@ SHOW FULL TABLES;
 | engines                               | LOCAL TEMPORARY |
 | events                                | LOCAL TEMPORARY |
 | files                                 | LOCAL TEMPORARY |
+| flow_statistics                       | LOCAL TEMPORARY |
 | flows                                 | LOCAL TEMPORARY |
 | global_status                         | LOCAL TEMPORARY |
 | key_column_usage                      | LOCAL TEMPORARY |
@@ -139,6 +141,7 @@ SHOW TABLE STATUS;
 |engines||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|||
 |events||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|||
 |files||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|||
+|flow_statistics||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|||
 |flows||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|||
 |global_status||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|||
 |key_column_usage||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|||
@@ -26,6 +26,7 @@ order by table_schema, table_name;
 |greptime|information_schema|engines|LOCALTEMPORARY|5|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|||Y|
 |greptime|information_schema|events|LOCALTEMPORARY|13|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|||Y|
 |greptime|information_schema|files|LOCALTEMPORARY|14|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|||Y|
+|greptime|information_schema|flow_statistics|LOCALTEMPORARY|45|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|||Y|
 |greptime|information_schema|flows|LOCALTEMPORARY|33|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|||Y|
 |greptime|information_schema|global_status|LOCALTEMPORARY|25|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|||Y|
 |greptime|information_schema|key_column_usage|LOCALTEMPORARY|16|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|||Y|
@@ -198,6 +199,12 @@ select * from information_schema.columns order by table_schema, table_name, colu
 | greptime      | information_schema | files                                 | update_count                      | 13               |                          |                        | 19                | 0             |                    |                    |                |            |       | select,insert |                       | Int64                | bigint              | FIELD         |                | NO          | bigint              |                |        |
 | greptime      | information_schema | files                                 | update_time                       | 34               |                          |                        |                   |               | 6                  |                    |                |            |       | select,insert |                       | TimestampMicrosecond | timestamp(6)        | FIELD         |                | NO          | timestamp(6)        |                |        |
 | greptime      | information_schema | files                                 | version                           | 25               | 2147483647               | 2147483647             |                   |               |                    | utf8               | utf8_bin       |            |       | select,insert |                       | String               | string              | FIELD         |                | NO          | string              |                |        |
+| greptime      | information_schema | flow_statistics                       | flow_id                           | 1                |                          |                        | 10                | 0             |                    |                    |                |            |       | select,insert |                       | UInt32               | int unsigned        | FIELD         |                | NO          | int unsigned        |                |        |
+| greptime      | information_schema | flow_statistics                       | flow_name                         | 2                | 2147483647               | 2147483647             |                   |               |                    | utf8               | utf8_bin       |            |       | select,insert |                       | String               | string              | FIELD         |                | NO          | string              |                |        |
+| greptime      | information_schema | flow_statistics                       | last_execution_time               | 4                |                          |                        |                   |               | 3                  |                    |                |            |       | select,insert |                       | TimestampMillisecond | timestamp(3)        | FIELD         |                | YES         | timestamp(3)        |                |        |
+| greptime      | information_schema | flow_statistics                       | start_time                        | 3                |                          |                        |                   |               | 3                  |                    |                |            |       | select,insert |                       | TimestampMillisecond | timestamp(3)        | FIELD         |                | YES         | timestamp(3)        |                |        |
+| greptime      | information_schema | flow_statistics                       | state_size                        | 6                |                          |                        | 20                | 0             |                    |                    |                |            |       | select,insert |                       | UInt64               | bigint unsigned     | FIELD         |                | YES         | bigint unsigned     |                |        |
+| greptime      | information_schema | flow_statistics                       | uptime_seconds                    | 5                |                          |                        | 19                | 0             |                    |                    |                |            |       | select,insert |                       | Int64                | bigint              | FIELD         |                | YES         | bigint              |                |        |
 | greptime      | information_schema | flows                                 | comment                           | 6                | 2147483647               | 2147483647             |                   |               |                    | utf8               | utf8_bin       |            |       | select,insert |                       | String               | string              | FIELD         |                | YES         | string              |                |        |
 | greptime      | information_schema | flows                                 | created_time                      | 12               |                          |                        |                   |               | 3                  |                    |                |            |       | select,insert |                       | TimestampMillisecond | timestamp(3)        | FIELD         |                | NO          | timestamp(3)        |                |        |
 | greptime      | information_schema | flows                                 | expire_after                      | 7                |                          |                        | 19                | 0             |                    |                    |                |            |       | select,insert |                       | Int64                | bigint              | FIELD         |                | YES         | bigint              |                |        |
@@ -105,6 +105,7 @@ SELECT * FROM INFORMATION_SCHEMA.TABLES ORDER BY TABLE_NAME, TABLE_TYPE;
 |greptime|information_schema|engines|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|||Y|
 |greptime|information_schema|events|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|||Y|
 |greptime|information_schema|files|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|||Y|
+|greptime|information_schema|flow_statistics|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|||Y|
 |greptime|information_schema|flows|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|||Y|
 |greptime|information_schema|global_status|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|||Y|
 |greptime|information_schema|key_column_usage|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|||Y|