Merge pull request cockroachdb#18158 from dt/opts

sqlccl: let the reusable helper filter valueless opts

pkg/ccl/sqlccl/csv.go

@@ -55,9 +55,14 @@ const (
 	importOptionTemp          = "temp"
 )
 
-var valuelessImportOptions = map[string]bool{
-	importOptionTransformOnly: true,
-	importOptionDistributed:   true,
+var importOptionExpectValues = map[string]bool{
+	importOptionDelimiter:     true,
+	importOptionComment:       true,
+	importOptionDistributed:   false,
+	importOptionNullIf:        true,
+	importOptionTransformOnly: false,
+	importOptionSSTSize:       true,
+	importOptionTemp:          true,
 }
 
 // LoadCSV converts CSV files into enterprise backup format.
@@ -657,7 +662,7 @@ func importJobDescription(
 			hasTransformOnly = true
 		}
 		opt := parser.KVOption{Key: parser.Name(k)}
-		if !valuelessImportOptions[k] {
+		if importOptionExpectValues[k] {
 			opt.Value = parser.NewDString(v)
 		}
 		stmt.Options = append(stmt.Options, opt)
@@ -699,7 +704,7 @@ func importPlanHook(
 		return nil, nil, errors.Errorf("unsupported import format: %q", importStmt.FileFormat)
 	}
 
-	optsFn, err := p.TypeAsStringOpts(importStmt.Options)
+	optsFn, err := p.TypeAsStringOpts(importStmt.Options, importOptionExpectValues)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -724,13 +729,7 @@ func importPlanHook(
 			return err
 		}
 
-		transformOnly := false
-		if override, ok := opts[importOptionTransformOnly]; ok {
-			if override != "" {
-				return errors.Errorf("option '%s' does not take a value", importOptionTransformOnly)
-			}
-			transformOnly = true
-		}
+		_, transformOnly := opts[importOptionTransformOnly]
 
 		var targetDB string
 		if !transformOnly {
@@ -783,14 +782,6 @@ func importPlanHook(
 			sstSize = sz
 		}
 
-		distributed := false
-		if override, ok := opts[importOptionDistributed]; ok {
-			if override != "" {
-				return errors.New("option 'distributed' does not take a value")
-			}
-			distributed = true
-		}
-
 		var create *parser.CreateTable
 		if importStmt.CreateDefs != nil {
 			normName := parser.NormalizableTableName{TableNameReference: importStmt.Table}
@@ -842,7 +833,7 @@ func importPlanHook(
 		}
 
 		var importErr error
-		if distributed {
+		if _, distributed := opts[importOptionDistributed]; distributed {
 			_, importErr = doDistributedCSVTransform(
 				ctx, job, files, p, tableDesc, temp,
 				comma, comment, nullif, walltime,

pkg/ccl/sqlccl/csv_test.go

@@ -466,6 +466,30 @@ func TestImportStmt(t *testing.T) {
 			``,
 			"must provide a temporary storage location",
 		},
+		{
+			"bad-opt-name",
+			`IMPORT TABLE t (a INT PRIMARY KEY, b STRING, INDEX (b), INDEX (a, b)) CSV DATA (%s) WITH foo = 'bar'`,
+			nil,
+			files,
+			``,
+			"invalid option \"foo\"",
+		},
+		{
+			"bad-opt-arg",
+			`IMPORT TABLE t (a INT PRIMARY KEY, b STRING, INDEX (b), INDEX (a, b)) CSV DATA (%s) WITH distributed = 'bar'`,
+			nil,
+			files,
+			``,
+			"option \"distributed\" does not take a value",
+		},
+		{
+			"bad-opt-no-arg",
+			`IMPORT TABLE t (a INT PRIMARY KEY, b STRING, INDEX (b), INDEX (a, b)) CSV DATA (%s) WITH temp`,
+			nil,
+			files,
+			``,
+			"option \"temp\" requires a value",
+		},
 		{
 			"primary-key-dup",
 			`IMPORT TABLE t CREATE USING $1 CSV DATA (%s) WITH temp = $2, distributed`,

pkg/ccl/sqlccl/restore.go

@@ -47,6 +47,11 @@ const (
 	restoreOptSkipMissingFKs = "skip_missing_foreign_keys"
 )
 
+var restoreOptionExpectValues = map[string]bool{
+	restoreOptIntoDB:         true,
+	restoreOptSkipMissingFKs: false,
+}
+
 func loadBackupDescs(ctx context.Context, uris []string) ([]BackupDescriptor, error) {
 	backupDescs := make([]BackupDescriptor, len(uris))
 
@@ -125,10 +130,7 @@ func allocateTableRewrites(
 			if index.ForeignKey.IsSet() {
 				to := index.ForeignKey.Table
 				if _, ok := tablesByID[to]; !ok {
-					if empty, ok := opts[restoreOptSkipMissingFKs]; ok {
-						if empty != "" {
-							return errors.Errorf("option %q does not take a value", restoreOptSkipMissingFKs)
-						}
+					if _, ok := opts[restoreOptSkipMissingFKs]; ok {
 						index.ForeignKey = sqlbase.ForeignKeyReference{}
 					} else {
 						return errors.Errorf(
@@ -972,7 +974,7 @@ func restorePlanHook(
 		return nil, nil, err
 	}
 
-	optsFn, err := p.TypeAsStringOpts(restoreStmt.Options)
+	optsFn, err := p.TypeAsStringOpts(restoreStmt.Options, restoreOptionExpectValues)
 	if err != nil {
 		return nil, nil, err
 	}

pkg/sql/planhook.go

@@ -49,7 +49,9 @@ type PlanHookState interface {
 	DistLoader() *DistLoader
 	TypeAsString(e parser.Expr, op string) (func() (string, error), error)
 	TypeAsStringArray(e parser.Exprs, op string) (func() ([]string, error), error)
-	TypeAsStringOpts(opts parser.KVOptions) (func() (map[string]string, error), error)
+	TypeAsStringOpts(
+		opts parser.KVOptions, valuelessOpts map[string]bool,
+	) (func() (map[string]string, error), error)
 	User() string
 	AuthorizationAccessor
 }

pkg/sql/planner.go

@@ -355,15 +355,26 @@ func (p *planner) TypeAsString(e parser.Expr, op string) (func() (string, error)
 // typecheck as strings, and returns a function that can be called to
 // get the string value during (planNode).Start.
 func (p *planner) TypeAsStringOpts(
-	opts parser.KVOptions,
+	opts parser.KVOptions, expectValues map[string]bool,
 ) (func() (map[string]string, error), error) {
 	typed := make(map[string]parser.TypedExpr, len(opts))
 	for _, opt := range opts {
 		k := string(opt.Key)
+		takesValue, ok := expectValues[k]
+		if !ok {
+			return nil, errors.Errorf("invalid option %q", k)
+		}
+
 		if opt.Value == nil {
+			if takesValue {
+				return nil, errors.Errorf("option %q requires a value", k)
+			}
 			typed[k] = nil
 			continue
 		}
+		if !takesValue {
+			return nil, errors.Errorf("option %q does not take a value", k)
+		}
 		r, err := parser.TypeCheckAndRequire(opt.Value, &p.semaCtx, parser.TypeString, k)
 		if err != nil {
 			return nil, err