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

@@ -30,7 +30,7 @@ require (
 	github.com/confluentinc/ccloud-sdk-go-v2/connect-custom-plugin v0.0.9
 	github.com/confluentinc/ccloud-sdk-go-v2/flink v0.9.0
 	github.com/confluentinc/ccloud-sdk-go-v2/flink-artifact v0.3.0
-	github.com/confluentinc/ccloud-sdk-go-v2/flink-gateway v0.17.0
+	github.com/confluentinc/ccloud-sdk-go-v2/flink-gateway v0.18.0
 	github.com/confluentinc/ccloud-sdk-go-v2/iam v0.15.0
 	github.com/confluentinc/ccloud-sdk-go-v2/iam-ip-filtering v0.5.0
 	github.com/confluentinc/ccloud-sdk-go-v2/identity-provider v0.3.0
@@ -164,6 +164,7 @@ require (
 	github.com/charmbracelet/x/term v0.1.1 // indirect
 	github.com/charmbracelet/x/windows v0.1.0 // indirect
 	github.com/cloudflare/circl v1.6.1 // indirect
+	github.com/confluentinc/ccloud-sdk-go-v2-internal/flink-gateway v0.27.0 // indirect
 	github.com/confluentinc/proto-go-setter v0.3.0 // indirect
 	github.com/cyphar/filepath-securejoin v0.2.5 // indirect
 	github.com/distribution/reference v0.6.0 // indirect

go.sum

@@ -188,6 +188,10 @@ 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.26.0 h1:SpINYIWdqK1tRGjHsN5Fwq/k/+YIgbJdksnHzB796s0=
+github.com/confluentinc/ccloud-sdk-go-v2-internal/flink-gateway v0.26.0/go.mod h1:zNoX10GjoVTzo4dYii6eXaYhfIi7J61qCYiSNEGGbrI=
+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=
@@ -218,8 +222,8 @@ github.com/confluentinc/ccloud-sdk-go-v2/flink v0.9.0 h1:QqtIFEB5E3CIyGMJd7NQBEt
 github.com/confluentinc/ccloud-sdk-go-v2/flink v0.9.0/go.mod h1:GPj4sfR85OyiFQUMNEq1DtPOjYVAuE222Z6Mcapwa48=
 github.com/confluentinc/ccloud-sdk-go-v2/flink-artifact v0.3.0 h1:DVWL3Y4b5azgCADubtyp3EhGZuyJkleINaTy2V3iius=
 github.com/confluentinc/ccloud-sdk-go-v2/flink-artifact v0.3.0/go.mod h1:P4fdIkI1ynjSvhDEGX283KhtzG51eGHQc5Cqtp7bu1Q=
-github.com/confluentinc/ccloud-sdk-go-v2/flink-gateway v0.17.0 h1:8Y1uXjolI2d5mawcfLn4OfJ81WRMQpjMFWdBm3dLdrk=
-github.com/confluentinc/ccloud-sdk-go-v2/flink-gateway v0.17.0/go.mod h1:cJ6erfVlWSyz6L+2dR46cF2+s5I2r+pTNrPm2fNbcqU=
+github.com/confluentinc/ccloud-sdk-go-v2/flink-gateway v0.18.0 h1:KzlhRDrUsXbs4ZPZy6T9OWmFIVkZWHxNsDorHHSnwFs=
+github.com/confluentinc/ccloud-sdk-go-v2/flink-gateway v0.18.0/go.mod h1:CuvhIQpYj/LbQeYzp7Sw2LJkTKzLh8xlFdQoKq9ZQlY=
 github.com/confluentinc/ccloud-sdk-go-v2/iam v0.15.0 h1:37Gjdo+0Ev3g2NPEXyiVm7yTT85AlWbjXYRLvq6Aj9E=
 github.com/confluentinc/ccloud-sdk-go-v2/iam v0.15.0/go.mod h1:jnWqax4kM22sutPGMtGmHqe2usgfqYig4UtmHsLENz0=
 github.com/confluentinc/ccloud-sdk-go-v2/iam-ip-filtering v0.5.0 h1:xD7CXcyAqezFnSVFB4U27oWUY4FlbyciVP0ftDIiI18=

internal/flink/command.go

@@ -55,6 +55,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,86 @@
+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"`
+	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.ListMaterializedTable(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.")
+}