Merge branch 'main' into cdc

Author: shiyuhang0

internal/cli/serverless/export/create.go

@@ -73,6 +73,7 @@ func (c CreateOpts) NonInteractiveFlags() []string {
 		flag.OSSURI,
 		flag.OSSAccessKeyID,
 		flag.OSSAccessKeySecret,
+		flag.Partitions,
 	}
 }
 
@@ -161,6 +162,8 @@ func CreateCmd(h *internal.Helper) *cobra.Command {
 			// oss
 			var ossURI, ossAccessKeyID, ossAccessKeySecret string
 			var displayName string
+			// partitions
+			var partitions []string
 
 			if opts.interactive {
 				if !h.IOStreams.CanPrompt {
@@ -303,8 +306,8 @@ func CreateCmd(h *internal.Helper) *cobra.Command {
 						return errors.New("sql is empty")
 					}
 				case FilterTable:
-					fmt.Fprintln(h.IOStreams.Out, color.BlueString("Please input the following options, require at least one field"))
-					inputs := []string{flag.TableFilter, flag.TableWhere}
+					fmt.Fprintln(h.IOStreams.Out, color.BlueString("Please input the following options"))
+					inputs := []string{flag.TableFilter, flag.TableWhere, flag.Partitions}
 					textInput, err := ui.InitialInputModel(inputs, inputDescription)
 					if err != nil {
 						return err
@@ -314,9 +317,16 @@ func CreateCmd(h *internal.Helper) *cobra.Command {
 					if err != nil {
 						return err
 					}
+					if len(patterns) == 0 {
+						return errors.New("table filters are required, if want to export whole cluster, type *.*")
+					}
 					where = textInput.Inputs[1].Value()
-					if len(patterns) == 0 && where == "" {
-						return errors.New("both patterns and where are empty, require at least one field")
+					partitionString := textInput.Inputs[2].Value()
+					if partitionString != "" {
+						partitions, err = util.StringSliceConv(partitionString)
+						if err != nil {
+							return err
+						}
 					}
 				}
 
@@ -573,6 +583,10 @@ func CreateCmd(h *internal.Helper) *cobra.Command {
 				if err != nil {
 					return errors.Trace(err)
 				}
+				partitions, err = cmd.Flags().GetStringSlice(flag.Partitions)
+				if err != nil {
+					return errors.Trace(err)
+				}
 			}
 
 			if !opts.interactive && sql == "" && len(patterns) == 0 && !force {
@@ -669,11 +683,12 @@ func CreateCmd(h *internal.Helper) *cobra.Command {
 					Sql: &sql,
 				}
 			}
-			if len(patterns) > 0 || where != "" {
+			if len(patterns) > 0 || where != "" || len(partitions) > 0 {
 				params.ExportOptions.Filter = &export.ExportOptionsFilter{
 					Table: &export.ExportOptionsFilterTable{
-						Where:    &where,
-						Patterns: patterns,
+						Where:      &where,
+						Patterns:   patterns,
+						Partitions: partitions,
 					},
 				}
 			}
@@ -729,9 +744,11 @@ func CreateCmd(h *internal.Helper) *cobra.Command {
 	createCmd.Flags().String(flag.OSSAccessKeySecret, "", "The access key secret of the OSS.")
 	createCmd.Flags().String(flag.ParquetCompression, "ZSTD", fmt.Sprintf("The parquet compression algorithm. One of %q.", export.AllowedExportParquetCompressionTypeEnumEnumValues))
 	createCmd.Flags().String(flag.DisplayName, "", "The display name of the export. (default \"SNAPSHOT_<snapshot_time>\")")
+	createCmd.Flags().StringSlice(flag.Partitions, nil, "Filter the exported partition table(s) with specified partition(s).")
 
 	createCmd.MarkFlagsMutuallyExclusive(flag.TableFilter, flag.SQL)
 	createCmd.MarkFlagsMutuallyExclusive(flag.TableWhere, flag.SQL)
+	createCmd.MarkFlagsMutuallyExclusive(flag.Partitions, flag.SQL)
 	createCmd.MarkFlagsMutuallyExclusive(flag.S3RoleArn, flag.S3AccessKeyID)
 	createCmd.MarkFlagsMutuallyExclusive(flag.S3RoleArn, flag.S3SecretAccessKey)
 	return createCmd

internal/cli/serverless/export/create_test.go

@@ -348,8 +348,9 @@ func (suite *CreateExportSuite) TestCreateExportWithTableFilter() {
 	body := getDefaultCreateExportBody()
 	body.ExportOptions.Filter = &export.ExportOptionsFilter{
 		Table: &export.ExportOptionsFilterTable{
-			Patterns: []string{pattern1, pattern2},
-			Where:    &where,
+			Patterns:   []string{pattern1, pattern2},
+			Where:      &where,
+			Partitions: []string{},
 		},
 	}
 

internal/cli/serverless/export/ui.go

@@ -36,8 +36,8 @@ var inputDescription = map[string]string{
 	flag.GCSURI:               "Input your GCS URI in gs://<bucket>/<path> format",
 	flag.GCSServiceAccountKey: "Input your base64 encoded GCS service account key",
 	flag.SQL:                  "Input the SELECT SQL statement",
-	flag.TableFilter:          "Input the table filter patterns (comma separated). Example: database.table,database.*,`database-1`.`table-1`",
-	flag.TableWhere:           "Input the where clause which will apply to all filtered tables. Example: id > 10",
+	flag.TableFilter:          "Required, input the table filter patterns (comma separated). Example: database.table,database.*,`database-1`.`table-1`",
+	flag.TableWhere:           "Optional, input the where clause which will apply to all filtered tables. Example: id > 10",
 	flag.CSVSeparator:         "Input the CSV separator: separator of each value in CSV files, skip to use default value (,)",
 	flag.CSVDelimiter:         "Input the CSV delimiter: delimiter of string type variables in CSV files, skip to use default value (\"). If you want to set empty string, please use non-interactive mode",
 	flag.CSVNullValue:         "Input the CSV null value: representation of null values in CSV files, skip to use default value (\\N). If you want to set empty string, please use non-interactive mode",
@@ -46,6 +46,7 @@ var inputDescription = map[string]string{
 	flag.OSSURI:               "Input your OSS URI in oss://<bucket>/<path> format",
 	flag.OSSAccessKeyID:       "Input your OSS access key id",
 	flag.OSSAccessKeySecret:   "Input your OSS access key secret",
+	flag.Partitions:           "Optional, input the partitions (comma separated) which will apply to all filtered tables. Example: p1,p2. It is useful when you want to export specific partitions from a partition table.",
 }
 
 func GetSelectedParquetCompression() (export.ExportParquetCompressionTypeEnum, error) {

internal/flag/flag.go

@@ -98,6 +98,7 @@ const (
 	ParquetCompression     string = "parquet.compression"
 	StartDate              string = "start-date"
 	EndDate                string = "end-date"
+	Partitions             string = "partitions"
 
 	AuditLogUnRedacted      string = "unredacted"
 	AuditLogFilterRuleID    string = "filter-rule-id"

internal/service/cloud/api_client.go

@@ -405,17 +405,17 @@ func (d *ClientDelegate) CancelUpload(ctx context.Context, clusterId string, upl
 }
 
 func (d *ClientDelegate) GetExport(ctx context.Context, clusterId string, exportId string) (*export.Export, error) {
-	res, h, err := d.ec.ExportServiceAPI.ExportServiceGetExport(ctx, clusterId, exportId).Execute()
+	res, h, err := d.ec.ExportAPI.ExportServiceGetExport(ctx, clusterId, exportId).Execute()
 	return res, parseError(err, h)
 }
 
 func (d *ClientDelegate) CancelExport(ctx context.Context, clusterId string, exportId string) (*export.Export, error) {
-	res, h, err := d.ec.ExportServiceAPI.ExportServiceCancelExport(ctx, clusterId, exportId).Execute()
+	res, h, err := d.ec.ExportAPI.ExportServiceCancelExport(ctx, clusterId, exportId).Execute()
 	return res, parseError(err, h)
 }
 
 func (d *ClientDelegate) CreateExport(ctx context.Context, clusterId string, body *export.ExportServiceCreateExportBody) (*export.Export, error) {
-	r := d.ec.ExportServiceAPI.ExportServiceCreateExport(ctx, clusterId)
+	r := d.ec.ExportAPI.ExportServiceCreateExport(ctx, clusterId)
 	if body != nil {
 		r = r.Body(*body)
 	}
@@ -424,12 +424,12 @@ func (d *ClientDelegate) CreateExport(ctx context.Context, clusterId string, bod
 }
 
 func (d *ClientDelegate) DeleteExport(ctx context.Context, clusterId string, exportId string) (*export.Export, error) {
-	res, h, err := d.ec.ExportServiceAPI.ExportServiceDeleteExport(ctx, clusterId, exportId).Execute()
+	res, h, err := d.ec.ExportAPI.ExportServiceDeleteExport(ctx, clusterId, exportId).Execute()
 	return res, parseError(err, h)
 }
 
 func (d *ClientDelegate) ListExports(ctx context.Context, clusterId string, pageSize *int32, pageToken *string, orderBy *string) (*export.ListExportsResponse, error) {
-	r := d.ec.ExportServiceAPI.ExportServiceListExports(ctx, clusterId)
+	r := d.ec.ExportAPI.ExportServiceListExports(ctx, clusterId)
 	if pageSize != nil {
 		r = r.PageSize(*pageSize)
 	}