From b729b60bfb7884a7ce6507b84c96542ed694577f Mon Sep 17 00:00:00 2001 From: Ramin Gharib Date: Thu, 30 Jul 2026 12:31:23 +0200 Subject: [PATCH 1/2] [FSE-1855] Read Flink statement warnings from status.warnings The CLI showed statement warnings by printing status.detail as a warning banner. That field is a single flat string with no severity, reason, or timestamp, and it gets overwritten during the statement lifecycle, so warnings were short-lived and could not be rendered as a list. Statement warnings are now read from the structured status.warnings array and rendered most severe first, in the SQL shell, on dry runs, and in `confluent flink statement describe` and `create`. Serialized output carries the warnings verbatim under a new `warnings` field, so `-o json` and `-o yaml` consumers get severity, reason, message, and created_at. status.detail is still printed when a statement has no warnings, and always for a failed statement where it holds the failure reason. --- internal/flink/command_statement.go | 34 +++- internal/flink/command_statement_create.go | 12 +- internal/flink/command_statement_describe.go | 33 ++-- internal/flink/command_statement_list.go | 2 + ...TestExecuteStatementWithStructuredWarnings | 14 ++ .../controller/statement_controller_test.go | 37 +++++ pkg/flink/internal/store/store.go | 2 + pkg/flink/types/processed_statement.go | 31 +++- pkg/flink/types/processed_statement_test.go | 69 ++++++++ pkg/flink/types/statement_warning.go | 97 ++++++++++++ pkg/flink/types/statement_warning_test.go | 148 ++++++++++++++++++ .../statement/describe-warnings-yaml.golden | 22 +++ .../flink/statement/describe-warnings.golden | 22 +++ test/flink_test.go | 2 + test/test-server/flink_gateway_router.go | 17 ++ 15 files changed, 513 insertions(+), 29 deletions(-) create mode 100644 pkg/flink/internal/controller/.snapshots/TestStatementControllerTestSuite-TestExecuteStatementWithStructuredWarnings create mode 100644 pkg/flink/types/processed_statement_test.go create mode 100644 pkg/flink/types/statement_warning.go create mode 100644 pkg/flink/types/statement_warning_test.go create mode 100644 test/fixtures/output/flink/statement/describe-warnings-yaml.golden create mode 100644 test/fixtures/output/flink/statement/describe-warnings.golden diff --git a/internal/flink/command_statement.go b/internal/flink/command_statement.go index 377bac0fa3..2d928dfcb5 100644 --- a/internal/flink/command_statement.go +++ b/internal/flink/command_statement.go @@ -8,17 +8,35 @@ import ( cmfsdk "github.com/confluentinc/cmf-sdk-go/v1" "github.com/confluentinc/cli/v4/pkg/config" + "github.com/confluentinc/cli/v4/pkg/flink/types" + "github.com/confluentinc/cli/v4/pkg/output" ) +// printStatementWarnings renders warnings below the table, on stderr so that stdout stays the +// command's data. Serialized output already carries them in the warnings field. +func printStatementWarnings(cmd *cobra.Command, warnings []types.StatementWarning) { + if output.GetFormat(cmd) != output.Human { + return + } + + if block := types.FormatStatementWarnings(warnings); block != "" { + output.ErrPrintln(false, "") + output.ErrPrintln(false, block) + output.ErrPrintln(false, "") + } +} + type statementOut struct { - CreationDate time.Time `human:"Creation Date" serialized:"creation_date"` - Name string `human:"Name" serialized:"name"` - Statement string `human:"Statement" serialized:"statement"` - ComputePool string `human:"Compute Pool,omitempty" serialized:"compute_pool,omitempty"` - Status string `human:"Status" serialized:"status"` - StatusDetail string `human:"Status Detail,omitempty" serialized:"status_detail,omitempty"` - LatestOffsets map[string]string `human:"Latest Offsets" serialized:"latest_offsets"` - LatestOffsetsTimestamp *time.Time `human:"Latest Offsets Timestamp" serialized:"latest_offsets_timestamp"` + CreationDate time.Time `human:"Creation Date" serialized:"creation_date"` + Name string `human:"Name" serialized:"name"` + Statement string `human:"Statement" serialized:"statement"` + ComputePool string `human:"Compute Pool,omitempty" serialized:"compute_pool,omitempty"` + Status string `human:"Status" serialized:"status"` + StatusDetail string `human:"Status Detail,omitempty" serialized:"status_detail,omitempty"` + // Rendered below the table, since a warning message is too long for a cell. + Warnings []types.StatementWarning `human:"-" serialized:"warnings,omitempty"` + LatestOffsets map[string]string `human:"Latest Offsets" serialized:"latest_offsets"` + LatestOffsetsTimestamp *time.Time `human:"Latest Offsets Timestamp" serialized:"latest_offsets_timestamp"` } type statementOutOnPrem struct { diff --git a/internal/flink/command_statement_create.go b/internal/flink/command_statement_create.go index eda49f8606..9ee0462afd 100644 --- a/internal/flink/command_statement_create.go +++ b/internal/flink/command_statement_create.go @@ -173,6 +173,8 @@ func (c *command) statementCreate(cmd *cobra.Command, args []string) error { } } + warnings := types.NewStatementWarnings(statement.Status.GetWarnings()) + table := output.NewTable(cmd) table.Add(&statementOut{ CreationDate: statement.Metadata.GetCreatedAt(), @@ -181,7 +183,13 @@ func (c *command) statementCreate(cmd *cobra.Command, args []string) error { ComputePool: statement.Spec.GetComputePoolId(), Status: statement.Status.GetPhase(), StatusDetail: statement.Status.GetDetail(), + Warnings: warnings, }) - table.Filter([]string{"CreationDate", "Name", "Statement", "ComputePool", "Status", "StatusDetail"}) - return table.Print() + table.Filter([]string{"CreationDate", "Name", "Statement", "ComputePool", "Status", "StatusDetail", "Warnings"}) + if err := table.Print(); err != nil { + return err + } + + printStatementWarnings(cmd, warnings) + return nil } diff --git a/internal/flink/command_statement_describe.go b/internal/flink/command_statement_describe.go index d50928b681..2ca17c9432 100644 --- a/internal/flink/command_statement_describe.go +++ b/internal/flink/command_statement_describe.go @@ -8,20 +8,23 @@ import ( flinkgatewayv1 "github.com/confluentinc/ccloud-sdk-go-v2/flink-gateway/v1" pcmd "github.com/confluentinc/cli/v4/pkg/cmd" + "github.com/confluentinc/cli/v4/pkg/flink/types" "github.com/confluentinc/cli/v4/pkg/output" ) type describeStatementOut struct { - CreationDate time.Time `human:"Creation Date" serialized:"creation_date"` - Name string `human:"Name" serialized:"name"` - Statement string `human:"Statement" serialized:"statement"` - ComputePool string `human:"Compute Pool" serialized:"compute_pool"` - Status string `human:"Status" serialized:"status"` - StatusDetail string `human:"Status Detail,omitempty" serialized:"status_detail,omitempty"` - LatestOffsets map[string]string `human:"Latest Offsets" serialized:"latest_offsets"` - LatestOffsetsTimestamp *time.Time `human:"Latest Offsets Timestamp" serialized:"latest_offsets_timestamp"` - Properties map[string]string `human:"Properties" serialized:"properties"` - Principal string `human:"Principal" serialized:"principal"` + CreationDate time.Time `human:"Creation Date" serialized:"creation_date"` + Name string `human:"Name" serialized:"name"` + Statement string `human:"Statement" serialized:"statement"` + ComputePool string `human:"Compute Pool" serialized:"compute_pool"` + Status string `human:"Status" serialized:"status"` + StatusDetail string `human:"Status Detail,omitempty" serialized:"status_detail,omitempty"` + // Rendered below the table, since a warning message is too long for a cell. + Warnings []types.StatementWarning `human:"-" serialized:"warnings,omitempty"` + LatestOffsets map[string]string `human:"Latest Offsets" serialized:"latest_offsets"` + LatestOffsetsTimestamp *time.Time `human:"Latest Offsets Timestamp" serialized:"latest_offsets_timestamp"` + Properties map[string]string `human:"Properties" serialized:"properties"` + Principal string `human:"Principal" serialized:"principal"` } func (c *command) newStatementDescribeCommand() *cobra.Command { @@ -58,6 +61,8 @@ func (c *command) statementDescribe(cmd *cobra.Command, args []string) error { return err } + warnings := types.NewStatementWarnings(statement.Status.GetWarnings()) + table := output.NewTable(cmd) table.Add(&describeStatementOut{ CreationDate: statement.Metadata.GetCreatedAt(), @@ -66,10 +71,16 @@ func (c *command) statementDescribe(cmd *cobra.Command, args []string) error { ComputePool: statement.Spec.GetComputePoolId(), Status: statement.Status.GetPhase(), StatusDetail: statement.Status.GetDetail(), + Warnings: warnings, LatestOffsets: statement.Status.GetLatestOffsets(), LatestOffsetsTimestamp: flinkgatewayv1.PtrTime(statement.Status.GetLatestOffsetsTimestamp()), Properties: statement.Spec.GetProperties(), Principal: statement.Spec.GetPrincipal(), }) - return table.Print() + if err := table.Print(); err != nil { + return err + } + + printStatementWarnings(cmd, warnings) + return nil } diff --git a/internal/flink/command_statement_list.go b/internal/flink/command_statement_list.go index 42ba23f26a..e29e915bc2 100644 --- a/internal/flink/command_statement_list.go +++ b/internal/flink/command_statement_list.go @@ -13,6 +13,7 @@ import ( pcmd "github.com/confluentinc/cli/v4/pkg/cmd" "github.com/confluentinc/cli/v4/pkg/errors" "github.com/confluentinc/cli/v4/pkg/examples" + "github.com/confluentinc/cli/v4/pkg/flink/types" "github.com/confluentinc/cli/v4/pkg/log" "github.com/confluentinc/cli/v4/pkg/output" "github.com/confluentinc/cli/v4/pkg/utils" @@ -100,6 +101,7 @@ func (c *command) statementList(cmd *cobra.Command, _ []string) error { ComputePool: statement.Spec.GetComputePoolId(), Status: statement.Status.GetPhase(), StatusDetail: statement.Status.GetDetail(), + Warnings: types.NewStatementWarnings(statement.Status.GetWarnings()), LatestOffsets: statement.Status.GetLatestOffsets(), LatestOffsetsTimestamp: flinkgatewayv1.PtrTime(statement.Status.GetLatestOffsetsTimestamp()), }) diff --git a/pkg/flink/internal/controller/.snapshots/TestStatementControllerTestSuite-TestExecuteStatementWithStructuredWarnings b/pkg/flink/internal/controller/.snapshots/TestStatementControllerTestSuite-TestExecuteStatementWithStructuredWarnings new file mode 100644 index 0000000000..c95d3e60fd --- /dev/null +++ b/pkg/flink/internal/controller/.snapshots/TestStatementControllerTestSuite-TestExecuteStatementWithStructuredWarnings @@ -0,0 +1,14 @@ +Statement successfully submitted. +Waiting for statement to be ready. Statement phase: PENDING. +Warnings: + +CRITICAL [UPSERT_PRIMARY_KEY_MISMATCH] (Logged: 2026-07-30T09:15:00Z) +The primary key does not match the upsert key derived from the query. + +MODERATE [MISSING_WINDOW_START_END] (Logged: 2026-07-30T08:00:00Z) +The GROUP BY clause contains only `window_start` with no corresponding `window_end`. + +Statement phase is RUNNING. +Listening for execution errors. Press Enter to detach. +Finished statement execution. Statement phase: COMPLETED. + diff --git a/pkg/flink/internal/controller/statement_controller_test.go b/pkg/flink/internal/controller/statement_controller_test.go index cc9653c9e6..79c3e09284 100644 --- a/pkg/flink/internal/controller/statement_controller_test.go +++ b/pkg/flink/internal/controller/statement_controller_test.go @@ -333,6 +333,43 @@ func (s *StatementControllerTestSuite) TestExecuteStatementWithWarning() { cupaloy.SnapshotT(s.T(), stdout) } +func (s *StatementControllerTestSuite) TestExecuteStatementWithStructuredWarnings() { + statementToExecute := "insert into users values ('test');" + legacyDetail := "[Warning] The primary key does not match the upsert key derived from the query." + windowWarningTime := time.Date(2026, 7, 30, 8, 0, 0, 0, time.UTC) + upsertWarningTime := time.Date(2026, 7, 30, 9, 15, 0, 0, time.UTC) + warnings := []types.StatementWarning{ + { + Severity: "MODERATE", + Reason: "MISSING_WINDOW_START_END", + Message: "The GROUP BY clause contains only `window_start` with no corresponding `window_end`.", + CreatedAt: &windowWarningTime, + }, + { + Severity: "CRITICAL", + Reason: "UPSERT_PRIMARY_KEY_MISMATCH", + Message: "The primary key does not match the upsert key derived from the query.", + CreatedAt: &upsertWarningTime, + }, + } + processedStatement := types.ProcessedStatement{Status: types.PENDING, Principal: "sa-123", StatusDetail: legacyDetail, Warnings: warnings} + runningStatement := types.ProcessedStatement{Status: types.RUNNING, StatusDetail: legacyDetail, Warnings: warnings} + completedStatement := types.ProcessedStatement{Status: types.COMPLETED} + s.store.EXPECT().ProcessStatement(statementToExecute).Return(&processedStatement, nil) + s.consoleParser.EXPECT().Read().Return(nil, nil).AnyTimes() + s.store.EXPECT().WaitPendingStatement(gomock.Any(), processedStatement).Return(&runningStatement, nil) + s.store.EXPECT().FetchStatementResults(runningStatement).Return(&runningStatement, nil) + s.store.EXPECT().WaitForTerminalStatementState(gomock.Any(), runningStatement).Return(&completedStatement, nil) + + stdout := testUtils.RunAndCaptureSTDOUT(s.T(), func() { + returnedStatement, err := s.statementController.ExecuteStatement(statementToExecute) + require.Nil(s.T(), err) + require.Equal(s.T(), &completedStatement, returnedStatement) + }) + + cupaloy.SnapshotT(s.T(), stdout) +} + func (s *StatementControllerTestSuite) TestRenderMsgAndStatusLocalStatements() { tests := []struct { name string diff --git a/pkg/flink/internal/store/store.go b/pkg/flink/internal/store/store.go index 302414dfe1..442e7335ae 100644 --- a/pkg/flink/internal/store/store.go +++ b/pkg/flink/internal/store/store.go @@ -351,6 +351,8 @@ func (s *Store) WaitForTerminalStatementState(ctx context.Context, statement typ statement.Status = types.PHASE(statementObj.Status.GetPhase()) statement.StatusDetail = statusDetail + // Warnings can be added after submission, so refresh them on every poll. + statement.Warnings = types.NewStatementWarnings(statementObj.Status.GetWarnings()) if statement.IsTerminalState() { break } diff --git a/pkg/flink/types/processed_statement.go b/pkg/flink/types/processed_statement.go index f1e3861cd3..158693a7de 100644 --- a/pkg/flink/types/processed_statement.go +++ b/pkg/flink/types/processed_statement.go @@ -23,13 +23,14 @@ const ( // ProcessedStatement Custom Internal type that shall be used internally by the client type ProcessedStatement struct { - Statement string `json:"statement"` - StatementName string `json:"statement_name"` - Kind string `json:"kind"` - ComputePool string `json:"compute_pool"` - Principal string `json:"principal"` // Cloud only - Status PHASE `json:"status"` - StatusDetail string `json:"status_detail,omitempty"` // Shown at the top before the table + Statement string `json:"statement"` + StatementName string `json:"statement_name"` + Kind string `json:"kind"` + ComputePool string `json:"compute_pool"` + Principal string `json:"principal"` // Cloud only + Status PHASE `json:"status"` + StatusDetail string `json:"status_detail,omitempty"` // Shown at the top before the table + Warnings []StatementWarning `json:"warnings,omitempty"` IsLocalStatement bool IsSensitiveStatement bool PageToken string @@ -46,6 +47,7 @@ func NewProcessedStatement(statementObj flinkgatewayv1.SqlV1Statement) *Processe ComputePool: statementObj.Spec.GetComputePoolId(), Principal: statementObj.Spec.GetPrincipal(), StatusDetail: statementObj.Status.GetDetail(), + Warnings: NewStatementWarnings(statementObj.Status.GetWarnings()), Status: PHASE(statementObj.Status.GetPhase()), Properties: statementObj.Spec.GetProperties(), Traits: StatementTraits{FlinkGatewayV1StatementTraits: &traits}, @@ -92,10 +94,21 @@ func (s ProcessedStatement) printStatusMessageOfNonLocalStatement() { } } - if s.StatusDetail != "" { + // The status detail can repeat the warnings, so only print it when there are none. A failed + // statement is the exception: its detail holds the failure reason. + if s.StatusDetail != "" && (s.Status == FAILED || len(s.Warnings) == 0) { utils.OutputInfof("Details: ") utils.OutputWarn(s.StatusDetail) } + + s.printWarnings() +} + +func (s ProcessedStatement) printWarnings() { + if warnings := FormatStatementWarnings(s.Warnings); warnings != "" { + utils.OutputWarn(warnings) + utils.OutputInfo("") + } } func (s ProcessedStatement) PrintOutputDryRunStatement() { @@ -110,6 +123,8 @@ func (s ProcessedStatement) PrintOutputDryRunStatement() { utils.OutputInfof("Details: ") utils.OutputErr(s.StatusDetail) } + + s.printWarnings() } func (s ProcessedStatement) GetPageSize() int { diff --git a/pkg/flink/types/processed_statement_test.go b/pkg/flink/types/processed_statement_test.go new file mode 100644 index 0000000000..5aa9ec2987 --- /dev/null +++ b/pkg/flink/types/processed_statement_test.go @@ -0,0 +1,69 @@ +package types + +import ( + "testing" + + "github.com/stretchr/testify/require" + + testUtils "github.com/confluentinc/cli/v4/pkg/flink/test" +) + +func TestPrintStatusMessagePrintsStructuredWarningsInsteadOfStatusDetail(t *testing.T) { + statement := ProcessedStatement{ + Status: RUNNING, + StatusDetail: "[Warning] legacy inlined warning", + Warnings: []StatementWarning{{Severity: "CRITICAL", Reason: "SOME_REASON", Message: "Fix the query."}}, + } + + stdout := testUtils.RunAndCaptureSTDOUT(t, statement.PrintStatusMessage) + + require.Contains(t, stdout, "CRITICAL [SOME_REASON]") + require.Contains(t, stdout, "Fix the query.") + require.NotContains(t, stdout, "legacy inlined warning") + require.NotContains(t, stdout, "Details: ") +} + +func TestPrintStatusMessagePrintsStatusDetailOfFailedStatementAlongsideWarnings(t *testing.T) { + statement := ProcessedStatement{ + Status: FAILED, + StatusDetail: "the failure reason", + Warnings: []StatementWarning{{Severity: "LOW", Reason: "SOME_REASON", Message: "Fix the query."}}, + } + + stdout := testUtils.RunAndCaptureSTDOUT(t, statement.PrintStatusMessage) + + require.Contains(t, stdout, "the failure reason") + require.Contains(t, stdout, "LOW [SOME_REASON]") +} + +func TestPrintStatusMessagePrintsStatusDetailWhenThereAreNoWarnings(t *testing.T) { + statement := ProcessedStatement{Status: RUNNING, StatusDetail: "something worth knowing"} + + stdout := testUtils.RunAndCaptureSTDOUT(t, statement.PrintStatusMessage) + + require.Contains(t, stdout, "Details: ") + require.Contains(t, stdout, "something worth knowing") + require.NotContains(t, stdout, "Warnings:") +} + +func TestPrintStatusMessagePrintsNoWarningsBlockWhenThereAreNoWarnings(t *testing.T) { + statement := ProcessedStatement{Status: RUNNING} + + stdout := testUtils.RunAndCaptureSTDOUT(t, statement.PrintStatusMessage) + + require.Contains(t, stdout, "Statement successfully submitted.") + require.NotContains(t, stdout, "Warnings:") + require.NotContains(t, stdout, "Details: ") +} + +func TestPrintOutputDryRunStatementPrintsWarnings(t *testing.T) { + statement := ProcessedStatement{ + Status: COMPLETED, + Warnings: []StatementWarning{{Severity: "MODERATE", Reason: "SOME_REASON", Message: "Fix the query."}}, + } + + stdout := testUtils.RunAndCaptureSTDOUT(t, statement.PrintOutputDryRunStatement) + + require.Contains(t, stdout, "MODERATE [SOME_REASON]") + require.Contains(t, stdout, "Fix the query.") +} diff --git a/pkg/flink/types/statement_warning.go b/pkg/flink/types/statement_warning.go new file mode 100644 index 0000000000..eea5d45e55 --- /dev/null +++ b/pkg/flink/types/statement_warning.go @@ -0,0 +1,97 @@ +package types + +import ( + "fmt" + "slices" + "strings" + "time" + + flinkgatewayv1 "github.com/confluentinc/ccloud-sdk-go-v2/flink-gateway/v1" +) + +// severityOrder ranks severities for display, most severe first. Severity is an extensible enum, so +// an unrecognized value is still displayed, sorted last. +var severityOrder = []string{"CRITICAL", "MODERATE", "LOW"} + +// StatementWarning is a non-fatal issue reported for a statement. +type StatementWarning struct { + Severity string `json:"severity" yaml:"severity"` + Reason string `json:"reason" yaml:"reason"` + Message string `json:"message" yaml:"message"` + // A pointer so that a missing timestamp is omitted from serialized output. `omitempty` does not + // omit a zero `time.Time` value, only a nil pointer. + CreatedAt *time.Time `json:"created_at,omitempty" yaml:"created_at,omitempty"` +} + +// NewStatementWarnings converts a statement's warnings, ordered most severe first. +func NewStatementWarnings(warnings []flinkgatewayv1.SqlV1StatementWarning) []StatementWarning { + if len(warnings) == 0 { + return nil + } + + converted := make([]StatementWarning, len(warnings)) + for i, warning := range warnings { + converted[i] = StatementWarning{ + Severity: warning.GetSeverity(), + Reason: warning.GetReason(), + Message: warning.GetMessage(), + } + // The API models the timestamp as a value, so an absent one arrives as the zero time. + if createdAt := warning.GetCreatedAt(); !createdAt.IsZero() { + converted[i].CreatedAt = &createdAt + } + } + + sortBySeverity(converted) + + return converted +} + +// FormatStatementWarnings renders warnings for terminal output, most severe first. It returns an +// empty string when there are no warnings. +func FormatStatementWarnings(warnings []StatementWarning) string { + if len(warnings) == 0 { + return "" + } + + // Sort a copy so the order holds however the caller built the slice. + sorted := slices.Clone(warnings) + sortBySeverity(sorted) + + entries := make([]string, len(sorted)) + for i, warning := range sorted { + entries[i] = fmt.Sprintf("%s\n%s", warning.header(), warning.Message) + } + + return fmt.Sprintf("Warnings:\n\n%s", strings.Join(entries, "\n\n")) +} + +func sortBySeverity(warnings []StatementWarning) { + slices.SortStableFunc(warnings, func(a, b StatementWarning) int { + return severityRank(a.Severity) - severityRank(b.Severity) + }) +} + +func (w StatementWarning) header() string { + severity := w.Severity + if severity == "" { + severity = "WARNING" + } + + header := severity + if w.Reason != "" { + header += fmt.Sprintf(" [%s]", w.Reason) + } + if w.CreatedAt != nil { + header += fmt.Sprintf(" (Logged: %s)", w.CreatedAt.UTC().Format(time.RFC3339)) + } + + return header +} + +func severityRank(severity string) int { + if i := slices.Index(severityOrder, strings.ToUpper(severity)); i >= 0 { + return i + } + return len(severityOrder) +} diff --git a/pkg/flink/types/statement_warning_test.go b/pkg/flink/types/statement_warning_test.go new file mode 100644 index 0000000000..4cd0a42992 --- /dev/null +++ b/pkg/flink/types/statement_warning_test.go @@ -0,0 +1,148 @@ +package types + +import ( + "encoding/json" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/require" + + flinkgatewayv1 "github.com/confluentinc/ccloud-sdk-go-v2/flink-gateway/v1" +) + +func ptrTime(t time.Time) *time.Time { + return &t +} + +func TestNewStatementWarningsReturnsNilWhenThereAreNoWarnings(t *testing.T) { + require.Nil(t, NewStatementWarnings(nil)) + require.Nil(t, NewStatementWarnings([]flinkgatewayv1.SqlV1StatementWarning{})) +} + +func TestNewStatementWarningsSortsMostSevereFirst(t *testing.T) { + warnings := NewStatementWarnings([]flinkgatewayv1.SqlV1StatementWarning{ + {Severity: "LOW", Reason: "LOW_ONE"}, + {Severity: "CRITICAL", Reason: "CRITICAL_ONE"}, + {Severity: "MODERATE", Reason: "MODERATE_ONE"}, + {Severity: "CRITICAL", Reason: "CRITICAL_TWO"}, + }) + + reasons := make([]string, len(warnings)) + for i, warning := range warnings { + reasons[i] = warning.Reason + } + + require.Equal(t, []string{"CRITICAL_ONE", "CRITICAL_TWO", "MODERATE_ONE", "LOW_ONE"}, reasons) +} + +func TestNewStatementWarningsSortsUnrecognizedSeverityLast(t *testing.T) { + warnings := NewStatementWarnings([]flinkgatewayv1.SqlV1StatementWarning{ + {Severity: "SOMETHING_NEW", Reason: "NEW_ONE"}, + {Severity: "LOW", Reason: "LOW_ONE"}, + }) + + require.Equal(t, "LOW_ONE", warnings[0].Reason) + require.Equal(t, "NEW_ONE", warnings[1].Reason) +} + +func TestNewStatementWarningsCopiesEveryField(t *testing.T) { + createdAt := time.Date(2026, 7, 30, 9, 15, 0, 0, time.UTC) + + warnings := NewStatementWarnings([]flinkgatewayv1.SqlV1StatementWarning{{ + Severity: "CRITICAL", + Reason: "HIGH_STATE_OPERATOR_WITHOUT_TTL", + Message: "Your query includes one or more highly state-intensive operators.", + CreatedAt: createdAt, + }}) + + require.Equal(t, []StatementWarning{{ + Severity: "CRITICAL", + Reason: "HIGH_STATE_OPERATOR_WITHOUT_TTL", + Message: "Your query includes one or more highly state-intensive operators.", + CreatedAt: &createdAt, + }}, warnings) +} + +func TestNewStatementWarningsLeavesAnAbsentTimestampUnset(t *testing.T) { + warnings := NewStatementWarnings([]flinkgatewayv1.SqlV1StatementWarning{ + {Severity: "LOW", Reason: "SOME_REASON", Message: "Something to know."}, + }) + + require.Nil(t, warnings[0].CreatedAt) + + serialized, err := json.Marshal(warnings) + require.NoError(t, err) + require.NotContains(t, string(serialized), "created_at") + require.NotContains(t, string(serialized), "0001-01-01") +} + +func TestFormatStatementWarningsReturnsEmptyStringWhenThereAreNoWarnings(t *testing.T) { + require.Empty(t, FormatStatementWarnings(nil)) + require.Empty(t, FormatStatementWarnings([]StatementWarning{})) +} + +func TestFormatStatementWarnings(t *testing.T) { + warnings := []StatementWarning{ + { + Severity: "CRITICAL", + Reason: "HIGH_STATE_OPERATOR_WITHOUT_TTL", + Message: "Your query includes one or more highly state-intensive operators.", + CreatedAt: ptrTime(time.Date(2026, 7, 30, 9, 15, 0, 0, time.UTC)), + }, + { + Severity: "MODERATE", + Reason: "MISSING_WINDOW_START_END", + Message: "The GROUP BY clause contains only `window_start`.", + CreatedAt: ptrTime(time.Date(2026, 7, 30, 8, 0, 0, 0, time.UTC)), + }, + } + + expected := `Warnings: + +CRITICAL [HIGH_STATE_OPERATOR_WITHOUT_TTL] (Logged: 2026-07-30T09:15:00Z) +Your query includes one or more highly state-intensive operators. + +MODERATE [MISSING_WINDOW_START_END] (Logged: 2026-07-30T08:00:00Z) +The GROUP BY clause contains only ` + "`window_start`" + `.` + + require.Equal(t, expected, FormatStatementWarnings(warnings)) +} + +func TestFormatStatementWarningsSortsMostSevereFirstWithoutMutatingTheInput(t *testing.T) { + warnings := []StatementWarning{ + {Severity: "LOW", Reason: "LOW_ONE", Message: "Low."}, + {Severity: "CRITICAL", Reason: "CRITICAL_ONE", Message: "Critical."}, + } + + formatted := FormatStatementWarnings(warnings) + + require.Less(t, strings.Index(formatted, "CRITICAL_ONE"), strings.Index(formatted, "LOW_ONE")) + require.Equal(t, "LOW_ONE", warnings[0].Reason) +} + +func TestFormatStatementWarningsOmitsMissingHeaderParts(t *testing.T) { + warnings := []StatementWarning{{Severity: "LOW", Message: "Something to know."}} + + require.Equal(t, "Warnings:\n\nLOW\nSomething to know.", FormatStatementWarnings(warnings)) +} + +func TestFormatStatementWarningsFallsBackWhenSeverityIsMissing(t *testing.T) { + warnings := []StatementWarning{{Reason: "SOME_REASON", Message: "Something to know."}} + + require.Equal(t, "Warnings:\n\nWARNING [SOME_REASON]\nSomething to know.", FormatStatementWarnings(warnings)) +} + +func TestFormatStatementWarningsRendersTimestampInUtc(t *testing.T) { + berlin, err := time.LoadLocation("Europe/Berlin") + require.NoError(t, err) + + warnings := []StatementWarning{{ + Severity: "LOW", + Reason: "SOME_REASON", + Message: "Something to know.", + CreatedAt: ptrTime(time.Date(2026, 7, 30, 11, 15, 0, 0, berlin)), + }} + + require.Contains(t, FormatStatementWarnings(warnings), "(Logged: 2026-07-30T09:15:00Z)") +} diff --git a/test/fixtures/output/flink/statement/describe-warnings-yaml.golden b/test/fixtures/output/flink/statement/describe-warnings-yaml.golden new file mode 100644 index 0000000000..ace813bf22 --- /dev/null +++ b/test/fixtures/output/flink/statement/describe-warnings-yaml.golden @@ -0,0 +1,22 @@ +creation_date: 2022-01-01T00:00:00Z +name: my-statement-with-warnings +statement: CREATE TABLE test; +compute_pool: lfcp-123456 +status: COMPLETED +status_detail: SQL statement is completed +warnings: + - severity: CRITICAL + reason: HIGH_STATE_OPERATOR_WITHOUT_TTL + message: Your query includes one or more highly state-intensive operators but does not set a time-to-live (TTL) value. + created_at: 2022-01-01T00:00:00Z + - severity: MODERATE + reason: MISSING_WINDOW_START_END + message: The GROUP BY clause contains only `window_start` with no corresponding `window_end`. + created_at: 2022-01-01T00:00:00Z +latest_offsets: + customers_source: partition:0,offset:9223372036854775808 +latest_offsets_timestamp: 2022-01-01T00:00:00Z +properties: + sql.current-catalog: default + sql.current-database: my-cluster +principal: u-123456 diff --git a/test/fixtures/output/flink/statement/describe-warnings.golden b/test/fixtures/output/flink/statement/describe-warnings.golden new file mode 100644 index 0000000000..3a6483af10 --- /dev/null +++ b/test/fixtures/output/flink/statement/describe-warnings.golden @@ -0,0 +1,22 @@ ++--------------------------+---------------------------------------------------------+ +| Creation Date | 2022-01-01 00:00:00 +0000 UTC | +| Name | my-statement-with-warnings | +| Statement | CREATE TABLE test; | +| Compute Pool | lfcp-123456 | +| Status | COMPLETED | +| Status Detail | SQL statement is completed | +| Latest Offsets | customers_source=partition:0,offset:9223372036854775808 | +| Latest Offsets Timestamp | 2022-01-01 00:00:00 +0000 UTC | +| Properties | sql.current-catalog=default | +| | sql.current-database=my-cluster | +| Principal | u-123456 | ++--------------------------+---------------------------------------------------------+ + +Warnings: + +CRITICAL [HIGH_STATE_OPERATOR_WITHOUT_TTL] (Logged: 2022-01-01T00:00:00Z) +Your query includes one or more highly state-intensive operators but does not set a time-to-live (TTL) value. + +MODERATE [MISSING_WINDOW_START_END] (Logged: 2022-01-01T00:00:00Z) +The GROUP BY clause contains only `window_start` with no corresponding `window_end`. + diff --git a/test/flink_test.go b/test/flink_test.go index c07ef11c7f..c4cc25d3fa 100644 --- a/test/flink_test.go +++ b/test/flink_test.go @@ -442,6 +442,8 @@ func (s *CLITestSuite) TestFlinkStatement() { {args: "flink statement delete my-statement --force --cloud aws --region eu-west-1", fixture: "flink/statement/delete.golden"}, {args: "flink statement describe my-statement --cloud aws --region eu-west-1 -o yaml", fixture: "flink/statement/describe-yaml.golden"}, {args: "flink statement describe my-statement --cloud aws --region eu-west-1", fixture: "flink/statement/describe.golden"}, + {args: "flink statement describe my-statement-with-warnings --cloud aws --region eu-west-1", fixture: "flink/statement/describe-warnings.golden"}, + {args: "flink statement describe my-statement-with-warnings --cloud aws --region eu-west-1 -o yaml", fixture: "flink/statement/describe-warnings-yaml.golden"}, {args: "flink statement list --cloud aws --region eu-west-1", fixture: "flink/statement/list.golden"}, {args: "flink statement list --cloud aws --region eu-west-1 -o yaml", fixture: "flink/statement/list-yaml.golden"}, {args: "flink statement list --cloud aws --region eu-west-1 --status completed", fixture: "flink/statement/list-completed.golden"}, diff --git a/test/test-server/flink_gateway_router.go b/test/test-server/flink_gateway_router.go index affabe13a4..161ae44cbf 100644 --- a/test/test-server/flink_gateway_router.go +++ b/test/test-server/flink_gateway_router.go @@ -248,6 +248,23 @@ func handleStatementGet(t *testing.T) http.HandlerFunc { Metadata: &flinkgatewayv1.StatementObjectMeta{CreatedAt: flinkgatewayv1.PtrTime(time.Date(2022, 1, 1, 0, 0, 0, 0, time.UTC))}, } + if statement.GetName() == "my-statement-with-warnings" { + statement.Status.Warnings = &[]flinkgatewayv1.SqlV1StatementWarning{ + { + Severity: "MODERATE", + Reason: "MISSING_WINDOW_START_END", + Message: "The GROUP BY clause contains only `window_start` with no corresponding `window_end`.", + CreatedAt: time.Date(2022, 1, 1, 0, 0, 0, 0, time.UTC), + }, + { + Severity: "CRITICAL", + Reason: "HIGH_STATE_OPERATOR_WITHOUT_TTL", + Message: "Your query includes one or more highly state-intensive operators but does not set a time-to-live (TTL) value.", + CreatedAt: time.Date(2022, 1, 1, 0, 0, 0, 0, time.UTC), + }, + } + } + err := json.NewEncoder(w).Encode(statement) require.NoError(t, err) } From 433a79b8695632dfdefd313c65a538d891840f45 Mon Sep 17 00:00:00 2001 From: Ramin Gharib Date: Thu, 6 Aug 2026 08:48:57 +0200 Subject: [PATCH 2/2] [FSE-1855] Drop the warnings struct field comment The inline comment split gofmt's alignment group, so the fields below it aligned as a separate block. Removing it lets both statement output structs align as one block. The `human:"-"` tag already states that the field is excluded from the human table. --- internal/flink/command_statement.go | 13 ++++++------- internal/flink/command_statement_describe.go | 13 ++++++------- 2 files changed, 12 insertions(+), 14 deletions(-) diff --git a/internal/flink/command_statement.go b/internal/flink/command_statement.go index 2d928dfcb5..30ff2d18f0 100644 --- a/internal/flink/command_statement.go +++ b/internal/flink/command_statement.go @@ -27,13 +27,12 @@ func printStatementWarnings(cmd *cobra.Command, warnings []types.StatementWarnin } type statementOut struct { - CreationDate time.Time `human:"Creation Date" serialized:"creation_date"` - Name string `human:"Name" serialized:"name"` - Statement string `human:"Statement" serialized:"statement"` - ComputePool string `human:"Compute Pool,omitempty" serialized:"compute_pool,omitempty"` - Status string `human:"Status" serialized:"status"` - StatusDetail string `human:"Status Detail,omitempty" serialized:"status_detail,omitempty"` - // Rendered below the table, since a warning message is too long for a cell. + CreationDate time.Time `human:"Creation Date" serialized:"creation_date"` + Name string `human:"Name" serialized:"name"` + Statement string `human:"Statement" serialized:"statement"` + ComputePool string `human:"Compute Pool,omitempty" serialized:"compute_pool,omitempty"` + Status string `human:"Status" serialized:"status"` + StatusDetail string `human:"Status Detail,omitempty" serialized:"status_detail,omitempty"` Warnings []types.StatementWarning `human:"-" serialized:"warnings,omitempty"` LatestOffsets map[string]string `human:"Latest Offsets" serialized:"latest_offsets"` LatestOffsetsTimestamp *time.Time `human:"Latest Offsets Timestamp" serialized:"latest_offsets_timestamp"` diff --git a/internal/flink/command_statement_describe.go b/internal/flink/command_statement_describe.go index 2ca17c9432..9496a24cf4 100644 --- a/internal/flink/command_statement_describe.go +++ b/internal/flink/command_statement_describe.go @@ -13,13 +13,12 @@ import ( ) type describeStatementOut struct { - CreationDate time.Time `human:"Creation Date" serialized:"creation_date"` - Name string `human:"Name" serialized:"name"` - Statement string `human:"Statement" serialized:"statement"` - ComputePool string `human:"Compute Pool" serialized:"compute_pool"` - Status string `human:"Status" serialized:"status"` - StatusDetail string `human:"Status Detail,omitempty" serialized:"status_detail,omitempty"` - // Rendered below the table, since a warning message is too long for a cell. + CreationDate time.Time `human:"Creation Date" serialized:"creation_date"` + Name string `human:"Name" serialized:"name"` + Statement string `human:"Statement" serialized:"statement"` + ComputePool string `human:"Compute Pool" serialized:"compute_pool"` + Status string `human:"Status" serialized:"status"` + StatusDetail string `human:"Status Detail,omitempty" serialized:"status_detail,omitempty"` Warnings []types.StatementWarning `human:"-" serialized:"warnings,omitempty"` LatestOffsets map[string]string `human:"Latest Offsets" serialized:"latest_offsets"` LatestOffsetsTimestamp *time.Time `human:"Latest Offsets Timestamp" serialized:"latest_offsets_timestamp"`