Feat/enable dryrun flow (#2879)

pkg/flink/app/application.go

@@ -149,8 +149,10 @@ func (a *Application) readEvalPrint() {
 		a.history.Append(userInput)
 	}
 
-	a.resultFetcher.Init(*executedStatement)
-	a.getOutputController(*executedStatement).VisualizeResults()
+	if !executedStatement.IsDryRunStatement() {
+		a.resultFetcher.Init(*executedStatement)
+		a.getOutputController(*executedStatement).VisualizeResults()
+	}
 }
 
 func (a *Application) panicRecovery() {

pkg/flink/config/local_statements.go

@@ -22,6 +22,7 @@ const (
 	KeyServiceAccount = "client.service-account"
 	KeyStatementName  = "client.statement-name"
 	KeyOutputFormat   = "client.output-format"
+	KeyDryRun         = "sql.dry-run"
 )
 
 type OutputFormat string

pkg/flink/internal/controller/.snapshots/TestStatementControllerTestSuite-TestOutputOfDryRunStatements-failed_dry_run

@@ -0,0 +1,5 @@
+Statement name: failed-test-statement
+Statement successfully submitted. Statement phase is FAILED.
+Dry run statement was verified and there were issues found.
+Error: Parse Error
+

pkg/flink/internal/controller/.snapshots/TestStatementControllerTestSuite-TestOutputOfDryRunStatements-successful_dry_run

@@ -0,0 +1,5 @@
+Statement name: successful-test-statement
+Statement successfully submitted. Statement phase is COMPLETED.
+Dry run statement was verified and there were no issues found.
+If you wish to submit your statement, disable dry run mode before submitting your statement with "set 'sql.dry-run' = 'false';"
+

pkg/flink/internal/controller/.snapshots/TestStatementControllerTestSuite-TestOutputOfDryRunStatements-unexpected_status_dry_run

@@ -0,0 +1,6 @@
+Statement name: unexpected-test-statement
+Statement successfully submitted. Statement phase is RUNNING.
+Dry run statement execution resulted in unexpected status.
+Status: RUNNING
+Details: unexpected status
+

pkg/flink/internal/controller/statement_controller.go

@@ -37,6 +37,12 @@ func (c *StatementController) ExecuteStatement(statementToExecute string) (*type
 		c.handleStatementError(*err)
 		return nil, err
 	}
+
+	if processedStatement.IsDryRunStatement() {
+		processedStatement.PrintOutputDryRunStatement()
+		return processedStatement, nil
+	}
+
 	c.createdStatementName = processedStatement.StatementName
 	processedStatement.PrintStatusMessage()
 

pkg/flink/internal/controller/statement_controller_test.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/confluentinc/go-prompt"
 
+	"github.com/confluentinc/cli/v3/pkg/flink/config"
 	testUtils "github.com/confluentinc/cli/v3/pkg/flink/test"
 	"github.com/confluentinc/cli/v3/pkg/flink/test/mock"
 	"github.com/confluentinc/cli/v3/pkg/flink/types"
@@ -386,6 +387,68 @@ func (s *StatementControllerTestSuite) TestRenderMsgAndStatusNonLocalNonFailedSt
 	}
 }
 
+func (s *StatementControllerTestSuite) TestOutputOfDryRunStatements() {
+	tests := []struct {
+		name      string
+		statement types.ProcessedStatement
+		want      string
+	}{
+		{
+			name:      "failed dry run",
+			statement: types.ProcessedStatement{StatementName: "failed-test-statement", Status: types.FAILED, StatusDetail: "Parse Error"},
+		},
+		{
+			name:      "successful dry run",
+			statement: types.ProcessedStatement{StatementName: "successful-test-statement", Status: types.COMPLETED, StatusDetail: "No errors"},
+		},
+		{
+			name:      "unexpected status dry run",
+			statement: types.ProcessedStatement{StatementName: "unexpected-test-statement", Status: types.RUNNING, StatusDetail: "unexpected status"},
+		},
+	}
+	for _, test := range tests {
+		s.T().Run(test.name, func(t *testing.T) {
+			actual := testUtils.RunAndCaptureSTDOUT(s.T(), test.statement.PrintOutputDryRunStatement)
+			cupaloy.SnapshotT(t, actual)
+		})
+	}
+}
+
+func (s *StatementControllerTestSuite) TestIsDryRunStatements() {
+	tests := []struct {
+		name      string
+		statement types.ProcessedStatement
+		expected  bool
+	}{
+		{
+			name: "is dry run statement",
+			statement: types.ProcessedStatement{Properties: map[string]string{
+				config.KeyDryRun: "true",
+			}},
+			expected: true,
+		},
+		{
+			name: "dry run is false",
+			statement: types.ProcessedStatement{Properties: map[string]string{
+				config.KeyDryRun: "false",
+			}},
+			expected: false,
+		},
+		{
+			name:      "dry run not set",
+			statement: types.ProcessedStatement{},
+			expected:  false,
+		},
+	}
+
+	for _, test := range tests {
+		s.T().Run(test.name, func(t *testing.T) {
+			actual := test.statement.IsDryRunStatement()
+			require.Equal(t, test.expected, actual)
+		})
+	}
+}
+
 func TestIsCancelEvent(t *testing.T) {
 	tests := []struct {
 		name string

pkg/flink/types/processed_statement.go

@@ -6,6 +6,7 @@ import (
 
 	flinkgatewayv1 "github.com/confluentinc/ccloud-sdk-go-v2/flink-gateway/v1"
 
+	"github.com/confluentinc/cli/v3/pkg/flink/config"
 	"github.com/confluentinc/cli/v3/pkg/flink/internal/utils"
 	"github.com/confluentinc/cli/v3/pkg/output"
 )
@@ -31,6 +32,7 @@ type ProcessedStatement struct {
 	IsLocalStatement     bool
 	IsSensitiveStatement bool
 	PageToken            string
+	Properties           map[string]string
 	StatementResults     *StatementResults
 	Traits               flinkgatewayv1.SqlV1StatementTraits
 }
@@ -43,6 +45,7 @@ func NewProcessedStatement(statementObj flinkgatewayv1.SqlV1Statement) *Processe
 		Principal:     statementObj.Spec.GetPrincipal(),
 		StatusDetail:  statementObj.Status.GetDetail(),
 		Status:        PHASE(statementObj.Status.GetPhase()),
+		Properties:    statementObj.Spec.GetProperties(),
 		Traits:        statementObj.Status.GetTraits(),
 	}
 }
@@ -75,6 +78,22 @@ func (s ProcessedStatement) printStatusMessageOfNonLocalStatement() {
 	}
 }
 
+func (s ProcessedStatement) PrintOutputDryRunStatement() {
+	if s.StatementName != "" {
+		utils.OutputInfof("Statement name: %s\n", s.StatementName)
+	}
+	utils.OutputInfo(fmt.Sprintf("Statement successfully submitted. Statement phase is %s.", s.Status))
+	if s.Status == "FAILED" {
+		utils.OutputErr(fmt.Sprintf("Dry run statement was verified and there were issues found.\nError: %s", s.StatusDetail))
+	} else if s.Status == "COMPLETED" {
+		utils.OutputInfo("Dry run statement was verified and there were no issues found.")
+		utils.OutputWarn("If you wish to submit your statement, disable dry run mode before submitting your statement with \"set 'sql.dry-run' = 'false';\"")
+	} else {
+		utils.OutputErr(fmt.Sprintf("Dry run statement execution resulted in unexpected status.\nStatus: %s", s.Status))
+		utils.OutputErr(fmt.Sprintf("Details: %s", s.StatusDetail))
+	}
+}
+
 func (s ProcessedStatement) GetPageSize() int {
 	return len(s.StatementResults.GetRows())
 }
@@ -98,3 +117,12 @@ func (s ProcessedStatement) IsSelectStatement() bool {
 	return strings.EqualFold(s.Traits.GetSqlKind(), "SELECT") ||
 		strings.HasPrefix(strings.ToUpper(s.Statement), "SELECT")
 }
+
+func (s ProcessedStatement) IsDryRunStatement() bool {
+	keyVal, ok := s.Properties[config.KeyDryRun]
+
+	if ok && strings.ToLower(keyVal) == "true" {
+		return true
+	}
+	return false
+}