From cabc2f6cc667291d7bc3dcbb8f9cc82e27bd93a0 Mon Sep 17 00:00:00 2001 From: Palak Jha Date: Thu, 6 Aug 2026 12:54:59 +0530 Subject: [PATCH] feat(flow): add information_schema.flow_statistics and SHOW FLOW STATUS (#7987) (#8392) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(flow): add information_schema.flow_statistics and SHOW FLOW STATUS (fixes #7987) Signed-off-by: onepizzateam * feat(flow): add information_schema.flow_statistics and SHOW FLOW STATUS (fixes #7987) Signed-off-by: onepizzateam * test(flow): add sqlness golden result for flow_status Signed-off-by: Palak Jha * fix(catalog): remove unused OptionExt import in flow_statistics Signed-off-by: onepizzateam * docs(flow): fix stale 'recent errors' comment on QueryFlowExecStats Signed-off-by: onepizzateam * test: regenerate golden results for flow_statistics table Signed-off-by: onepizzateam * style: apply rustfmt to flow_statistics changes Signed-off-by: onepizzateam * refactor(catalog): hoist current_time_millis out of flow loop and clamp uptime Signed-off-by: onepizzateam * chore: remove accidentally committed fmt_check.log Signed-off-by: onepizzateam * Update flow_status.result del eof trailing blank line as per review Signed-off-by: onepizzateam * test(flow): restore runner-generated trailing blank line for sqlness Signed-off-by: onepizzateam * postgres: include SHOW FLOW STATUS in extended-query describe (return flow_statistics fields) Signed-off-by: onepizzateam * fix: address reviewer feedback on flow_statistics PR Signed-off-by: onepizzateam * fix(flow): resolve merge conflicts with main Signed-off-by: polar Signed-off-by: onepizzateam * sqlness check post gen (information_schema.result) Signed-off-by: onepizzateam * fix(flow): record start_time after req/snapshot_seqs built, before dispatch Signed-off-by: onepizzateam * fix(sql): handle ShowFlowStatus in match statement at util.rs Signed-off-by: onepizzateam * feat: review patch implementation Signed-off-by: onepizzateam * chore: remove accidentally committed local tool output files and fix fmt Signed-off-by: onepizzateam * fix worker.rs return type formatting Signed-off-by: onepizzateam * fix(flow): apply rustfmt to get_full_flow_stat return type Signed-off-by: onepizzateam * fix worker.rs return type formatting Signed-off-by: onepizzateam * fix(flow): re-apply rustfmt to get_full_flow_stat return type after merge Signed-off-by: onepizzateam * fix: merge conflicts Signed-off-by: onepizzateam * 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 * 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 --------- Signed-off-by: Dennis Zhuang Signed-off-by: onepizzateam * 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 Signed-off-by: onepizzateam * feat(query): plan native histogram functions (#8705) * feat(query): plan native histogram functions Signed-off-by: shuiyisong * fix: cr issue & add tests Signed-off-by: shuiyisong * fix: cr issue Signed-off-by: shuiyisong * fix: cr issue Signed-off-by: shuiyisong --------- Signed-off-by: shuiyisong Signed-off-by: onepizzateam * 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 * 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 * 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 * 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 * style(promql): apply pinned rustfmt Signed-off-by: discord9 * 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 --------- Signed-off-by: discord9 Signed-off-by: onepizzateam * ci: optimize fuzz and split workflows (#8710) * ci: batch fuzz targets in GitHub Actions Signed-off-by: WenyXu * ci: improve fuzz test observability Signed-off-by: WenyXu * fix(ci): preserve fuzz setup failure artifacts Signed-off-by: WenyXu * test(ci): keep fuzz mock output in logs Signed-off-by: WenyXu * ci: optimize fuzz worker cache Signed-off-by: WenyXu * ci: warm fuzz target binaries Signed-off-by: WenyXu * ci: isolate fuzz workflow Signed-off-by: WenyXu * ci: centralize fuzz target preparation Signed-off-by: WenyXu * ci: split general workflows Signed-off-by: WenyXu * ci: streamline docs required checks Signed-off-by: WenyXu * fix: transfer fuzz targets as artifacts Signed-off-by: WenyXu * fix: preserve fuzz binary permissions Signed-off-by: WenyXu * ci: streamline fuzz workers Signed-off-by: WenyXu * ci: cache PR build dependencies Signed-off-by: WenyXu * ci: retain main build cache policy Signed-off-by: WenyXu * ci: address fuzz review feedback Signed-off-by: WenyXu --------- Signed-off-by: WenyXu Signed-off-by: onepizzateam * fix: add public constructor for compactor (#8724) Signed-off-by: onepizzateam * feat(logging): add enable_file_logging option to disable file logging (#8721) Signed-off-by: xhwhis Signed-off-by: onepizzateam * avoid cloning final Prometheus remote write row (#8733) perf: avoid cloning final Prometheus remote write row Signed-off-by: lyang24 Signed-off-by: onepizzateam * feat(function): add json_object_keys scalar function (#8722) Expose JSON object key listing for outermost objects, with sqlness coverage. Signed-off-by: onepizzateam * 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 * fix(mito2): track automatic compaction follow-ups Signed-off-by: Lei, HUANG * docs(mito2): fix compaction transition rustdoc Signed-off-by: Lei, HUANG * fix(mito2): mark manual compaction conflict retryable Signed-off-by: Lei, HUANG * 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 * fix(mito2): gate test-only cancellation import Signed-off-by: Lei, HUANG * fix(mito2): prioritize DDL after compaction planning Signed-off-by: Lei, HUANG --------- Signed-off-by: Lei, HUANG Signed-off-by: Lei, HUANG Signed-off-by: onepizzateam * feat: update dashboard to v0.13.11 (#8737) Signed-off-by: onepizzateam * 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 * fix(query): preserve remote dynamic filter target (#8615) * fix(query): preserve remote dynamic filter target Signed-off-by: discord9 * fix(query): check RDF subscriber registration Signed-off-by: discord9 * 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 Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> Signed-off-by: onepizzateam * 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 * 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 * 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 * Update README.md Co-authored-by: Ning Sun --------- Signed-off-by: Dennis Zhuang Co-authored-by: Ning Sun Signed-off-by: onepizzateam * 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 * fix: tighten enterprise license checks Signed-off-by: Dennis Zhuang --------- Signed-off-by: Dennis Zhuang Signed-off-by: onepizzateam * fix(operator): invalidate local cache after dropping view (#8748) Signed-off-by: WenyXu Signed-off-by: onepizzateam * 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 * 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 * 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 * 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 * 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 * 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 * fix: gate fresh soft-drop procedures Signed-off-by: Lei, HUANG * test: gate soft-drop fallback coverage Signed-off-by: Lei, HUANG * fix: gate soft-drop procedure implementation Signed-off-by: Lei, HUANG * refactor: gate drop table soft-drop behavior Signed-off-by: Lei, HUANG * refactor: gate expired soft-drop gc behavior Signed-off-by: Lei, HUANG * ci: test enterprise table ddl lifecycle Signed-off-by: Lei, HUANG * 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 * 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 * 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 --------- Signed-off-by: Lei, HUANG Signed-off-by: onepizzateam * feat: make frontend heartbeat extensible and lifecycle-safe (#8726) * feat: make frontend heartbeat extensible and lifecycle-safe Signed-off-by: jeremyhi * fix: isolate heartbeat extension response handlers Signed-off-by: jeremyhi * fix: cancel in-flight heartbeat response handling Signed-off-by: jeremyhi * test: cover heartbeat wire compatibility Signed-off-by: jeremyhi * fix: clean up failed heartbeat startup Signed-off-by: jeremyhi * fix: address frontend heartbeat review feedback Signed-off-by: jeremyhi --------- Signed-off-by: jeremyhi Signed-off-by: onepizzateam * fix(meta): release region guards after drop rollback (#8751) Signed-off-by: WenyXu Signed-off-by: onepizzateam * 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 * chore: move prom_validation_mode as well Signed-off-by: shuiyisong --------- Signed-off-by: shuiyisong Signed-off-by: onepizzateam * feat: add health-aware gRPC client routing (#8684) * feat: add gRPC client health routing Signed-off-by: WenyXu * fix: harden gRPC client health routing Signed-off-by: WenyXu * fix: defer gRPC client health checks until first use Signed-off-by: WenyXu --------- Signed-off-by: WenyXu Signed-off-by: onepizzateam * feat(mito2): discard unflushed region data safely (#8600) * feat: support discarding unflushed region data Signed-off-by: evenyag * fix(mito2): wake stalled writers after discard Signed-off-by: evenyag * refactor(mito2): drop redundant manifest check for discarding unflushed data Signed-off-by: evenyag --------- Signed-off-by: evenyag Signed-off-by: onepizzateam * 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 Signed-off-by: onepizzateam * fix: support Utf8View labels in Prometheus response (#8754) Signed-off-by: evenyag Signed-off-by: onepizzateam * fix(query): validate merge scan remote schema (#8579) * fix(query): validate merge scan remote schema Signed-off-by: discord9 * 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 Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> Signed-off-by: onepizzateam * feat(grafana): add events dashboard (#8725) * feat(grafana): add events dashboard Signed-off-by: WenyXu * fix(grafana): tolerate evolving event schemas Signed-off-by: WenyXu * fix(grafana): address events dashboard review Signed-off-by: WenyXu * fix(grafana): restore events dashboard panels Signed-off-by: WenyXu * fix(grafana): bound events dashboard queries Signed-off-by: WenyXu * fix(grafana): include historical event catalogs Signed-off-by: WenyXu * fix(grafana): preserve events drill-down context Signed-off-by: WenyXu * fix(grafana): correct events lifecycle outcomes Signed-off-by: WenyXu * fix(grafana): handle empty event type ranges Signed-off-by: WenyXu * fix(grafana): scope event catalogs to submissions Signed-off-by: WenyXu * fix(grafana): handle empty events dashboard Signed-off-by: WenyXu * fix(grafana): refresh event schema variables Signed-off-by: WenyXu --------- Signed-off-by: WenyXu Signed-off-by: onepizzateam * refactor: separate a json2 extension type (#8745) Signed-off-by: luofucong Signed-off-by: onepizzateam * feat: add admin function registrar (#8762) * feat: add admin function registrar Signed-off-by: jeremyhi * fix: reject admin function name collisions Signed-off-by: jeremyhi * chore: fix typo in admin function test Signed-off-by: jeremyhi --------- Signed-off-by: jeremyhi Signed-off-by: onepizzateam * fix: information_schema.rs table initialization issue Signed-off-by: onepizzateam * rustfmt fix Signed-off-by: onepizzateam --------- Signed-off-by: onepizzateam Signed-off-by: Palak Jha Signed-off-by: polar Signed-off-by: Dennis Zhuang Signed-off-by: Lei, HUANG Signed-off-by: shuiyisong Signed-off-by: discord9 Signed-off-by: WenyXu Signed-off-by: xhwhis Signed-off-by: lyang24 Signed-off-by: Lei, HUANG Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> Signed-off-by: jeremyhi Signed-off-by: evenyag Signed-off-by: luofucong Co-authored-by: dennis zhuang 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 Co-authored-by: Weny Xu Co-authored-by: Ning Sun Co-authored-by: Whis Liao Co-authored-by: Lanqing Yang Co-authored-by: sun Co-authored-by: Ning Sun Co-authored-by: jeremyhi Co-authored-by: Yingwen Co-authored-by: LFC <990479+MichaelScofield@users.noreply.github.com> --- .../src/system_schema/information_schema.rs | 11 + .../information_schema/flow_statistics.rs | 285 ++++++++++++++++++ .../information_schema/table_names.rs | 1 + src/cli/src/metadata/control/put/key.rs | 4 +- src/common/catalog/src/consts.rs | 2 + src/common/meta/src/key/flow/flow_state.rs | 67 +++- src/flow/src/adapter/flownode_impl.rs | 4 + src/flow/src/adapter/stat.rs | 26 +- src/flow/src/adapter/worker.rs | 76 +++-- src/flow/src/batching_mode/engine.rs | 22 +- src/flow/src/batching_mode/state.rs | 22 +- src/flow/src/batching_mode/task.rs | 8 + src/flow/src/compute/state.rs | 12 + src/frontend/src/instance.rs | 5 + .../src/handler/flow_state_handler.rs | 6 +- src/operator/src/statement.rs | 1 + src/operator/src/statement/show.rs | 16 +- src/query/src/sql.rs | 48 ++- src/servers/src/postgres/handler.rs | 47 +++ src/sql/src/parsers/show_parser.rs | 32 +- src/sql/src/statements/show.rs | 15 + src/sql/src/statements/statement.rs | 8 +- src/sql/src/util.rs | 1 + .../standalone/common/flow/flow_status.result | 76 +++++ .../standalone/common/flow/flow_status.sql | 42 +++ .../common/show/show_databases_tables.result | 3 + .../common/system/information_schema.result | 7 + .../standalone/common/view/create.result | 1 + 28 files changed, 767 insertions(+), 81 deletions(-) create mode 100644 src/catalog/src/system_schema/information_schema/flow_statistics.rs create mode 100644 tests/cases/standalone/common/flow/flow_status.result create mode 100644 tests/cases/standalone/common/flow/flow_status.sql diff --git a/src/catalog/src/system_schema/information_schema.rs b/src/catalog/src/system_schema/information_schema.rs index a51810e048..d0539a062c 100644 --- a/src/catalog/src/system_schema/information_schema.rs +++ b/src/catalog/src/system_schema/information_schema.rs @@ -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(), diff --git a/src/catalog/src/system_schema/information_schema/flow_statistics.rs b/src/catalog/src/system_schema/information_schema/flow_statistics.rs new file mode 100644 index 0000000000..6163461eda --- /dev/null +++ b/src/catalog/src/system_schema/information_schema/flow_statistics.rs @@ -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, + flow_metadata_manager: Arc, +} + +impl InformationSchemaFlowStatistics { + pub(super) fn new( + catalog_name: String, + catalog_manager: Weak, + flow_metadata_manager: Arc, + ) -> 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 { + 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, + flow_metadata_manager: Arc, + + 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, + flow_metadata_manager: &Arc, + ) -> 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) -> Result { + 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, + 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 { + let columns: Vec = 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) -> DfSendableRecordBatchStream { + let schema: Arc = 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) + }), + )) + } +} diff --git a/src/catalog/src/system_schema/information_schema/table_names.rs b/src/catalog/src/system_schema/information_schema/table_names.rs index 5fa1aa517c..f161fecb54 100644 --- a/src/catalog/src/system_schema/information_schema/table_names.rs +++ b/src/catalog/src/system_schema/information_schema/table_names.rs @@ -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"; diff --git a/src/cli/src/metadata/control/put/key.rs b/src/cli/src/metadata/control/put/key.rs index 7becfd72dc..49d4b61e2f 100644 --- a/src/cli/src/metadata/control/put/key.rs +++ b/src/cli/src/metadata/control/put/key.rs @@ -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(); } diff --git a/src/common/catalog/src/consts.rs b/src/common/catalog/src/consts.rs index adbd3a8d47..3149f6ccd4 100644 --- a/src/common/catalog/src/consts.rs +++ b/src/common/catalog/src/consts.rs @@ -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 ----- diff --git a/src/common/meta/src/key/flow/flow_state.rs b/src/common/meta/src/key/flow/flow_state.rs index 1b161929fe..77ccd6206a 100644 --- a/src/common/meta/src/key/flow/flow_state.rs +++ b/src/common/meta/src/key/flow/flow_state.rs @@ -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, /// For each key, the last execution time of flow in unix timestamp milliseconds. pub last_exec_time_map: BTreeMap, + /// 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, } impl FlowStateValue { pub fn new( state_size: BTreeMap, last_exec_time_map: BTreeMap, + start_time_map: BTreeMap, ) -> 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, /// For each key, the last execution time of flow in unix timestamp milliseconds. pub last_exec_time_map: BTreeMap, + /// 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, } impl From for FlowStat { @@ -160,6 +169,7 @@ impl From 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 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); + } +} diff --git a/src/flow/src/adapter/flownode_impl.rs b/src/flow/src/adapter/flownode_impl.rs index 9e3399c182..7e83d5131d 100644 --- a/src/flow/src/adapter/flownode_impl.rs +++ b/src/flow/src/adapter/flownode_impl.rs @@ -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, } } diff --git a/src/flow/src/adapter/stat.rs b/src/flow/src/adapter/stat.rs index 68f8160c87..9521a24c5c 100644 --- a/src/flow/src/adapter/stat.rs +++ b/src/flow/src/adapter/stat.rs @@ -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, } } } diff --git a/src/flow/src/adapter/worker.rs b/src/flow/src/adapter/worker.rs index 32cc4eb7f5..71fbacb18f 100644 --- a/src/flow/src/adapter/worker.rs +++ b/src/flow/src/adapter/worker.rs @@ -202,30 +202,24 @@ impl WorkerHandle { } } - pub async fn get_state_size(&self) -> Result, Error> { + pub async fn get_full_flow_stat( + &self, + ) -> Result< + ( + BTreeMap, + BTreeMap, + BTreeMap, + ), + 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, 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, - }, - QueryLastExecTimeMap { - /// each flow tasks' last execution time - result: BTreeMap, + QueryFullFlowStat { + state_size: BTreeMap, + last_exec_time_map: BTreeMap, + start_time_map: BTreeMap, }, } @@ -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(); diff --git a/src/flow/src/batching_mode/engine.rs b/src/flow/src/batching_mode/engine.rs index 3f65761b55..d05eec9251 100644 --- a/src/flow/src/batching_mode/engine.rs +++ b/src/flow/src/batching_mode/engine.rs @@ -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, } } } diff --git a/src/flow/src/batching_mode/state.rs b/src/flow/src/batching_mode/state.rs index 225d76a864..59c4f0fe10 100644 --- a/src/flow/src/batching_mode/state.rs +++ b/src/flow/src/batching_mode/state.rs @@ -46,6 +46,8 @@ pub struct TaskState { last_query_duration: Duration, /// Last successful execution time in unix timestamp milliseconds. last_exec_time_millis: Option, + /// First execution time in unix timestamp milliseconds, set once. + start_time_millis: Option, /// 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 { + self.start_time_millis + } + pub fn checkpoint_mode(&self) -> CheckpointMode { self.checkpoint_mode } diff --git a/src/flow/src/batching_mode/task.rs b/src/flow/src/batching_mode/task.rs index 8428198de3..196c9ea42d 100644 --- a/src/flow/src/batching_mode/task.rs +++ b/src/flow/src/batching_mode/task.rs @@ -290,6 +290,10 @@ impl BatchingTask { self.state.read().unwrap().last_execution_time_millis() } + pub fn start_time_millis(&self) -> Option { + 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 { @@ -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, diff --git a/src/flow/src/compute/state.rs b/src/flow/src/compute/state.rs index 2ccd366194..b71633f517 100644 --- a/src/flow/src/compute/state.rs +++ b/src/flow/src/compute/state.rs @@ -47,6 +47,8 @@ pub struct DataflowState { expire_after: Option, /// the last time each subgraph executed last_exec_time: Option, + /// the time the flow first executed, in unix timestamp milliseconds + start_time: Option, } 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 { self.last_exec_time } + + /// Returns the time the flow first executed, in unix timestamp milliseconds. + pub fn start_time(&self) -> Option { + self.start_time + } } #[derive(Debug, Clone)] diff --git a/src/frontend/src/instance.rs b/src/frontend/src/instance.rs index 7a4b2f9a41..706e651462 100644 --- a/src/frontend/src/instance.rs +++ b/src/frontend/src/instance.rs @@ -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 diff --git a/src/meta-srv/src/handler/flow_state_handler.rs b/src/meta-srv/src/handler/flow_state_handler.rs index 9683fd760a..a13c18a9e0 100644 --- a/src/meta-srv/src/handler/flow_state_handler.rs +++ b/src/meta-srv/src/handler/flow_state_handler.rs @@ -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 diff --git a/src/operator/src/statement.rs b/src/operator/src/statement.rs index 92e0c04a92..320e7c2e64 100644 --- a/src/operator/src/statement.rs +++ b/src/operator/src/statement.rs @@ -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, diff --git a/src/operator/src/statement/show.rs b/src/operator/src/statement/show.rs index 33cdb0f367..4a3fa5ae9b 100644 --- a/src/operator/src/statement/show.rs +++ b/src/operator/src/statement/show.rs @@ -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 { + 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( diff --git a/src/query/src/sql.rs b/src/query/src/sql.rs index 90c3b4ccd4..fc755deb8a 100644 --- a/src/query/src/sql.rs +++ b/src/query/src/sql.rs @@ -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 { + 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, diff --git a/src/servers/src/postgres/handler.rs b/src/servers/src/postgres/handler.rs index e372a54237..484bb6a1f1 100644 --- a/src/servers/src/postgres/handler.rs +++ b/src/servers/src/postgres/handler.rs @@ -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(_), diff --git a/src/sql/src/parsers/show_parser.rs b/src/sql/src/parsers/show_parser.rs index d6fc35c675..075c17dabe 100644 --- a/src/sql/src/parsers/show_parser.rs +++ b/src/sql/src/parsers/show_parser.rs @@ -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 { + let kind = self.parse_show_kind()?; + + Ok(Statement::ShowFlowStatus(ShowFlowStatus { kind })) + } + fn parse_show_processlist(&mut self, full: bool) -> Result { 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"; diff --git a/src/sql/src/statements/show.rs b/src/sql/src/statements/show.rs index 77880e4a50..67624192b8 100644 --- a/src/sql/src/statements/show.rs +++ b/src/sql/src/statements/show.rs @@ -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 { diff --git a/src/sql/src/statements/statement.rs b/src/sql/src/statements/statement.rs index 0f652f0179..096acb654d 100644 --- a/src/sql/src/statements/statement.rs +++ b/src/sql/src/statements/statement.rs @@ -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), diff --git a/src/sql/src/util.rs b/src/sql/src/util.rs index 03071cfaa2..c0c237cb0c 100644 --- a/src/sql/src/util.rs +++ b/src/sql/src/util.rs @@ -335,6 +335,7 @@ fn extract_tables_from_statement(stmt: &Statement, names: &mut HashSet