APIE-744 - Flink Materialized Tables

cmd/lint/main.go

@@ -126,6 +126,9 @@ var flagRules = []linter.FlagRule{
 			"source-bootstrap-server",
 			"update-schema-registry",
 			"worker-configurations",
+			"watermark-column-name",
+			"distributed-by-column-names",
+			"distributed-by-buckets",
 		),
 	),
 	linter.FlagFilter(
@@ -139,6 +142,7 @@ var flagRules = []linter.FlagRule{
 			"schema-registry-api-key",
 			"schema-registry-api-secret",
 			"skip-message-on-error",
+			"distributed-by-column-names",
 		),
 	),
 }

go.mod

@@ -15,6 +15,7 @@ require (
 	github.com/charmbracelet/lipgloss v0.11.0
 	github.com/client9/gospell v0.0.0-20160306015952-90dfc71015df
 	github.com/confluentinc/ccloud-sdk-go-v1-public v0.0.0-20250521223017-0e8f6f971b52
+	github.com/confluentinc/ccloud-sdk-go-v2-internal/flink-gateway v0.27.0
 	github.com/confluentinc/ccloud-sdk-go-v2/ai v0.1.0
 	github.com/confluentinc/ccloud-sdk-go-v2/apikeys v0.4.0
 	github.com/confluentinc/ccloud-sdk-go-v2/billing v0.3.0

go.sum

@@ -188,6 +188,8 @@ github.com/compose-spec/compose-go/v2 v2.1.3 h1:bD67uqLuL/XgkAK6ir3xZvNLFPxPScEi
 github.com/compose-spec/compose-go/v2 v2.1.3/go.mod h1:lFN0DrMxIncJGYAXTfWuajfwj5haBJqrBkarHcnjJKc=
 github.com/confluentinc/ccloud-sdk-go-v1-public v0.0.0-20250521223017-0e8f6f971b52 h1:19qEGhkbZa5fopKCe0VPIV+Sasby4Pv10z9ZaktwWso=
 github.com/confluentinc/ccloud-sdk-go-v1-public v0.0.0-20250521223017-0e8f6f971b52/go.mod h1:62EMf+5uFEt1BJ2q8WMrUoI9VUSxAbDnmZCGRt/MbA0=
+github.com/confluentinc/ccloud-sdk-go-v2-internal/flink-gateway v0.27.0 h1:4nJe4mkLV/gHsYeVb0ZMXsgFbnastCiBv4JLkGTdN/4=
+github.com/confluentinc/ccloud-sdk-go-v2-internal/flink-gateway v0.27.0/go.mod h1:vN5BamRyiS+s+Jy3NNWJCDkgadC8xEKdK7wac970wOI=
 github.com/confluentinc/ccloud-sdk-go-v2/ai v0.1.0 h1:zSF4OQUJXWH2JeAo9rsq13ibk+JFdzITGR8S7cFMpzw=
 github.com/confluentinc/ccloud-sdk-go-v2/ai v0.1.0/go.mod h1:DoxqzzF3JzvJr3fWkvCiOHFlE0GoYpozWxFZ1Ud9ntA=
 github.com/confluentinc/ccloud-sdk-go-v2/apikeys v0.4.0 h1:8fWyLwMuy8ec0MVF5Avd54UvbIxhDFhZzanHBVwgxdw=

internal/flink/command.go

@@ -56,6 +56,7 @@ func New(cfg *config.Config, prerunner pcmd.PreRunner) *cobra.Command {
 	cmd.AddCommand(c.newConnectivityTypeCommand())
 	cmd.AddCommand(c.newEndpointCommand())
 	cmd.AddCommand(c.newRegionCommand())
+	cmd.AddCommand(c.newMaterializedTableCommand())
 
 	return cmd
 }

internal/flink/command_materialized_table.go

@@ -0,0 +1,87 @@
+package flink
+
+import (
+	"github.com/spf13/cobra"
+
+	pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
+)
+
+type materializedTableOut struct {
+	Name                     string   `human:"Name" serialized:"name"`
+	ClusterID                string   `human:"Kafka Cluster ID" serialized:"kafka_cluster_id"`
+	Environment              string   `human:"Environment" serialized:"environment"`
+	ComputePool              string   `human:"Compute Pool" serialized:"compute_pool"`
+	ServiceAccount           string   `human:"Service Account" serialized:"service_account"`
+	Stopped                  bool     `human:"Stopped" serialized:"stopped"`
+	Query                    string   `human:"Query,omitempty" serialized:"query,omitempty"`
+	Columns                  []string `human:"Columns,omitempty" serialized:"columns,omitempty"`
+	WaterMarkColumnName      string   `human:"Watermark Column Name,omitempty" serialized:"watermark_column_name,omitempty"`
+	WaterMarkExpression      string   `human:"Watermark Expression,omitempty" serialized:"watermark_expression,omitempty"`
+	Constraints              []string `human:"Constraints,omitempty" serialized:"constraints,omitempty"`
+	DistributedByColumnNames []string `human:"Distributed By Column Names,omitempty" serialized:"distributed_by_column_names,omitempty"`
+	DistributedByBuckets     int      `human:"Distributed By Buckets,omitempty" serialized:"distributed_by_buckets,omitempty"`
+}
+
+func (c *command) newMaterializedTableCommand() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:         "materialized-table",
+		Short:       "Manage Flink materialized tables.",
+		Annotations: map[string]string{pcmd.RunRequirement: pcmd.RequireNonAPIKeyCloudLogin},
+	}
+
+	cmd.AddCommand(c.newMaterializedTableCreateCommand())
+	cmd.AddCommand(c.newMaterializedTableDeleteCommand())
+	cmd.AddCommand(c.newMaterializedTableDescribeCommand())
+	cmd.AddCommand(c.newMaterializedTableListCommand())
+	cmd.AddCommand(c.newMaterializedTableUpdateCommand())
+	cmd.AddCommand(c.newMaterializedTableStopCommand())
+	cmd.AddCommand(c.newMaterializedTableResumeCommand())
+
+	return cmd
+}
+
+func (c *command) validMaterializedTableArgs(cmd *cobra.Command, args []string) []string {
+	if len(args) > 0 {
+		return nil
+	}
+
+	return c.validMaterializedTablesArgsMultiple(cmd, args)
+}
+
+func (c *command) validMaterializedTablesArgsMultiple(cmd *cobra.Command, args []string) []string {
+	if err := c.PersistentPreRunE(cmd, args); err != nil {
+		return nil
+	}
+
+	environmentId, err := c.Context.EnvironmentId()
+	if err != nil {
+		return nil
+	}
+
+	client, err := c.GetFlinkGatewayClientInternal(false)
+	if err != nil {
+		return nil
+	}
+
+	tables, err := client.ListMaterializedTables(environmentId, c.Context.GetCurrentOrganization())
+	if err != nil {
+		return nil
+	}
+
+	suggestions := make([]string, len(tables))
+	for i, table := range tables {
+		suggestions[i] = table.GetName()
+	}
+	return suggestions
+}
+
+func (c *command) addOptionalMaterializedTableFlags(cmd *cobra.Command) {
+	cmd.Flags().String("column-physical", "", "Path to the columns data for type physical.")
+	cmd.Flags().String("column-metadata", "", "Path to the columns data for type metadata.")
+	cmd.Flags().String("column-computed", "", "Path to the columns data for type computed.")
+	cmd.Flags().String("watermark-column-name", "", "The name of the watermark columns.")
+	cmd.Flags().String("watermark-expression", "", "The watermark expression.")
+	cmd.Flags().String("constraints", "", "Path to the constraints.")
+	cmd.Flags().String("distributed-by-column-names", "", "The names of the columns the table is distributed by.")
+	cmd.Flags().Int("distributed-by-buckets", 0, "The number of buckets.")
+}