csv: delimiter supports at most 2 characters

pkg/config/sink.go

@@ -179,7 +179,9 @@ func (s *SinkConfig) MaskSensitiveData() {
 
 // CSVConfig defines a series of configuration items for csv codec.
 type CSVConfig struct {
-	// delimiter between fields
+	// delimiter between fields, it can be 1 character or at most 2 characters
+	// It can not be CR or LF or contains CR or LF.
+	// It should have exclusive characters with quote.
 	Delimiter string `toml:"delimiter" json:"delimiter"`
 	// quoting character
 	Quote string `toml:"quote" json:"quote"`
@@ -214,19 +216,23 @@ func (c *CSVConfig) validateAndAdjust() error {
 	case 0:
 		return cerror.WrapError(cerror.ErrSinkInvalidConfig,
 			errors.New("csv config delimiter cannot be empty"))
-	case 1:
+	case 1, 2:
 		if strings.ContainsRune(c.Delimiter, CR) || strings.ContainsRune(c.Delimiter, LF) {
 			return cerror.WrapError(cerror.ErrSinkInvalidConfig,
 				errors.New("csv config delimiter contains line break characters"))
 		}
 	default:
 		return cerror.WrapError(cerror.ErrSinkInvalidConfig,
-			errors.New("csv config delimiter contains more than one character"))
+			errors.New("csv config delimiter contains more than two characters"))
 	}
 
-	if len(c.Quote) > 0 && strings.Contains(c.Delimiter, c.Quote) {
-		return cerror.WrapError(cerror.ErrSinkInvalidConfig,
-			errors.New("csv config quote and delimiter cannot be the same"))
+	if len(c.Quote) > 0 {
+		for _, r := range c.Delimiter {
+			if strings.ContainsRune(c.Quote, r) {
+				return cerror.WrapError(cerror.ErrSinkInvalidConfig,
+					errors.New("csv config quote and delimiter has common characters which is not allowed"))
+			}
+		}
 	}
 
 	// validate binary encoding method

pkg/config/sink_test.go

@@ -333,6 +333,15 @@ func TestValidateAndAdjustCSVConfig(t *testing.T) {
 			},
 			wantErr: "",
 		},
+		{
+			name: "valid delimiter with 2 characters",
+			config: &CSVConfig{
+				Quote:                "\"",
+				Delimiter:            "FE",
+				BinaryEncodingMethod: BinaryEncodingHex,
+			},
+			wantErr: "",
+		},
 		{
 			name: "delimiter is empty",
 			config: &CSVConfig{
@@ -350,20 +359,28 @@ func TestValidateAndAdjustCSVConfig(t *testing.T) {
 			wantErr: "csv config delimiter contains line break characters",
 		},
 		{
-			name: "delimiter contains more than one character",
+			name: "delimiter contains more than two characters",
 			config: &CSVConfig{
 				Quote:     "'",
-				Delimiter: "\r\t",
+				Delimiter: "FEF",
 			},
-			wantErr: "csv config delimiter contains more than one character",
+			wantErr: "csv config delimiter contains more than two characters",
 		},
 		{
 			name: "delimiter and quote are same",
 			config: &CSVConfig{
 				Quote:     "'",
 				Delimiter: "'",
 			},
-			wantErr: "csv config quote and delimiter cannot be the same",
+			wantErr: "csv config quote and delimiter has common characters which is not allowed",
+		},
+		{
+			name: "delimiter and quote contain common characters",
+			config: &CSVConfig{
+				Quote:     "EX",
+				Delimiter: "FE",
+			},
+			wantErr: "csv config quote and delimiter has common characters which is not allowed",
 		},
 		{
 			name: "invalid binary encoding method",