diff --git a/pkg/picker/config.go b/pkg/picker/config.go
new file mode 100644
index 00000000..20134515
--- /dev/null
+++ b/pkg/picker/config.go
@@ -0,0 +1,113 @@
+package picker
+
+import (
+	"fmt"
+	"strings"
+)
+
+// Config is the configuration to build Picker
+// The priority is as follows:
+//		ConcatItems > Indices
+// 		Nullable
+// 		DefaultValue
+//    NullValue, if set to null, subsequent conversions will be skipped.
+// 		Type
+// 		Function
+// 		CheckOnPost
+type Config struct {
+	ConcatItems  ConcatItems        // Concat index column, constant, or mixed.
+	Indices      []int              // Set index columns, the first non-null.
+	Nullable     func(string) bool  // Determine whether it is null. Optional.
+	NullValue    string             // Set null value when it is null. Optional.
+	DefaultValue *string            // Set default value when it is null. Optional.
+	Type         string             // Set the type of value.
+	Function     *string            // Set the conversion function of value.
+	CheckOnPost  func(*Value) error // Set the value check function on post.
+}
+
+func (c *Config) Build() (Picker, error) {
+	var retPicker Picker
+	var nullHandled bool
+	switch {
+	case c.ConcatItems.Len() > 0:
+		retPicker = ConcatPicker{
+			items: c.ConcatItems,
+		}
+	case len(c.Indices) == 1:
+		retPicker = IndexPicker(c.Indices[0])
+	case len(c.Indices) > 1:
+		if c.Nullable == nil {
+			// the first must be picked
+			retPicker = IndexPicker(c.Indices[0])
+		} else {
+			pickers := make(NullablePickers, 0, len(c.Indices))
+			for _, index := range c.Indices {
+				pickers = append(pickers, ConverterPicker{
+					picker: IndexPicker(index),
+					converter: NullableConverters{
+						NullableConverter{
+							Nullable: c.Nullable,
+						},
+					},
+				})
+			}
+			retPicker = pickers
+		}
+		nullHandled = true
+	default:
+		return nil, fmt.Errorf("no indices or concat items")
+	}
+
+	var converters []Converter
+
+	if !nullHandled && c.Nullable != nil {
+		converters = append(converters, NullableConverter{
+			Nullable: c.Nullable,
+		})
+	}
+
+	if c.Nullable != nil {
+		if c.DefaultValue != nil {
+			converters = append(converters, DefaultConverter{
+				Value: *c.DefaultValue,
+			})
+		} else {
+			converters = append(converters, NullConverter{
+				Value: c.NullValue,
+			})
+		}
+	}
+
+	converters = append(converters, NewTypeConverter(c.Type))
+
+	if c.Function != nil && *c.Function != "" {
+		var functionConverter Converter = FunctionConverter{
+			Name: *c.Function,
+		}
+		if strings.EqualFold(*c.Function, "hash") && !strings.EqualFold(c.Type, "string") {
+			functionConverter = FunctionStringConverter{
+				Name: *c.Function,
+			}
+		}
+		converters = append(converters, functionConverter)
+	}
+
+	if c.CheckOnPost != nil {
+		converters = append(converters, ConverterFunc(func(v *Value) (*Value, error) {
+			if err := c.CheckOnPost(v); err != nil {
+				return nil, err
+			}
+			return v, nil
+		}))
+	}
+
+	var converter Converter = Converters(converters)
+	if c.Nullable != nil {
+		converter = NullableConverters(converters)
+	}
+
+	return ConverterPicker{
+		picker:    retPicker,
+		converter: converter,
+	}, nil
+}
diff --git a/pkg/picker/config_test.go b/pkg/picker/config_test.go
new file mode 100644
index 00000000..2a724862
--- /dev/null
+++ b/pkg/picker/config_test.go
@@ -0,0 +1,918 @@
+package picker
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestConfigBuildFailed(t *testing.T) {
+	var c Config
+	p, err := c.Build()
+	assert.Error(t, err)
+	assert.Nil(t, p)
+}
+
+func TestConverters(t *testing.T) {
+	var converter Converter = Converters(nil)
+	v, err := converter.Convert(&Value{
+		Val: "v",
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, &Value{
+		Val: "v",
+	}, v)
+
+	converter = Converters{NonConverter{}, ErrorConverter{Err: fmt.Errorf("test error")}}
+	v, err = converter.Convert(&Value{})
+	assert.Error(t, err)
+	assert.Equal(t, "test error", err.Error())
+	assert.Nil(t, v)
+
+	converter = NullableConverters(nil)
+	v, err = converter.Convert(&Value{
+		Val: "v",
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, &Value{
+		Val: "v",
+	}, v)
+
+	v, err = converter.Convert(&Value{
+		Val:       "v",
+		IsNull:    true,
+		isSetNull: true,
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, &Value{
+		Val:       "v",
+		IsNull:    true,
+		isSetNull: true,
+	}, v)
+}
+
+func TestConfig(t *testing.T) {
+	var (
+		strEmpty   = ""
+		strStr1    = "str1"
+		strInt1    = "1"
+		strFunHash = "hash"
+	)
+	type recordCase struct {
+		record        []string
+		wantValue     *Value
+		wantErrString string
+	}
+	testcases := []struct {
+		name  string
+		c     Config
+		fn    func(*Config)
+		cases []recordCase
+	}{
+		{
+			name: "index BOOL",
+			c: Config{
+				Indices: []int{1},
+				Type:    "BOOL",
+			},
+			cases: []recordCase{
+				{
+					record:        nil,
+					wantErrString: "prop index 1 out range 0 of record",
+				},
+				{
+					record:        []string{},
+					wantErrString: "prop index 1 out range 0 of record",
+				},
+				{
+					record:        []string{"0"},
+					wantErrString: "prop index 1 out range 1 of record",
+				},
+				{
+					record:    []string{"0", "1"},
+					wantValue: &Value{Val: "1", IsNull: false},
+				},
+				{
+					record:    []string{"0", "1", "2"},
+					wantValue: &Value{Val: "1", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "index iNt",
+			c: Config{
+				Indices: []int{1},
+				Type:    "iNt",
+			},
+			cases: []recordCase{
+				{
+					record:        nil,
+					wantErrString: "prop index 1 out range 0 of record",
+				},
+				{
+					record:        []string{},
+					wantErrString: "prop index 1 out range 0 of record",
+				},
+				{
+					record:        []string{"0"},
+					wantErrString: "prop index 1 out range 1 of record",
+				},
+				{
+					record:    []string{"0", "1"},
+					wantValue: &Value{Val: "1", IsNull: false},
+				},
+				{
+					record:    []string{"0", "1", "2"},
+					wantValue: &Value{Val: "1", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "index Float",
+			c: Config{
+				Indices: []int{2},
+				Type:    "Float",
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"0", "1.1", "2.2"},
+					wantValue: &Value{Val: "2.2", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "index double",
+			c: Config{
+				Indices: []int{3},
+				Type:    "double",
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"0", "1.1", "2.2", "3.3"},
+					wantValue: &Value{Val: "3.3", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "index string",
+			c: Config{
+				Indices: []int{1},
+				Type:    "string",
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"0", "str1", "str2"},
+					wantValue: &Value{Val: "\"str1\"", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "index date",
+			c: Config{
+				Indices: []int{0},
+				Type:    "date",
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"2020-01-02"},
+					wantValue: &Value{Val: "DATE(\"2020-01-02\")", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "index time",
+			c: Config{
+				Indices: []int{0},
+				Type:    "time",
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"18:38:23.284"},
+					wantValue: &Value{Val: "TIME(\"18:38:23.284\")", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "index datetime",
+			c: Config{
+				Indices: []int{0},
+				Type:    "datetime",
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"2020-01-11T19:28:23.284"},
+					wantValue: &Value{Val: "DATETIME(\"2020-01-11T19:28:23.284\")", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "index timestamp",
+			c: Config{
+				Indices: []int{0},
+				Type:    "timestamp",
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"2020-01-11T19:28:23"},
+					wantValue: &Value{Val: "TIMESTAMP(\"2020-01-11T19:28:23\")", IsNull: false},
+				},
+				{
+					record:    []string{"1578770903"},
+					wantValue: &Value{Val: "TIMESTAMP(1578770903)", IsNull: false},
+				},
+				{
+					record:    []string{""},
+					wantValue: &Value{Val: "TIMESTAMP(\"\")", IsNull: false},
+				},
+				{
+					record:    []string{"0"},
+					wantValue: &Value{Val: "TIMESTAMP(0)", IsNull: false},
+				},
+				{
+					record:    []string{"12"},
+					wantValue: &Value{Val: "TIMESTAMP(12)", IsNull: false},
+				},
+				{
+					record:    []string{"0x"},
+					wantValue: &Value{Val: "TIMESTAMP(\"0x\")", IsNull: false},
+				},
+				{
+					record:    []string{"0X"},
+					wantValue: &Value{Val: "TIMESTAMP(\"0X\")", IsNull: false},
+				},
+				{
+					record:    []string{"0123456789"},
+					wantValue: &Value{Val: "TIMESTAMP(0123456789)", IsNull: false},
+				},
+				{
+					record:    []string{"9876543210"},
+					wantValue: &Value{Val: "TIMESTAMP(9876543210)", IsNull: false},
+				},
+				{
+					record:    []string{"0x0123456789abcdef"},
+					wantValue: &Value{Val: "TIMESTAMP(0x0123456789abcdef)", IsNull: false},
+				},
+				{
+					record:    []string{"0X0123456789ABCDEF"},
+					wantValue: &Value{Val: "TIMESTAMP(0X0123456789ABCDEF)", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "index geography",
+			c: Config{
+				Indices: []int{0},
+				Type:    "geography",
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"Polygon((-85.1 34.8,-80.7 28.4,-76.9 34.9,-85.1 34.8))"},
+					wantValue: &Value{Val: "ST_GeogFromText(\"Polygon((-85.1 34.8,-80.7 28.4,-76.9 34.9,-85.1 34.8))\")", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "index geography(point)",
+			c: Config{
+				Indices: []int{0},
+				Type:    "geography(point)",
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"Point(0.0 0.0)"},
+					wantValue: &Value{Val: "ST_GeogFromText(\"Point(0.0 0.0)\")", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "index geography(linestring)",
+			c: Config{
+				Indices: []int{0},
+				Type:    "geography(linestring)",
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"linestring(0 1, 179.99 89.99)"},
+					wantValue: &Value{Val: "ST_GeogFromText(\"linestring(0 1, 179.99 89.99)\")", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "index geography(polygon)",
+			c: Config{
+				Indices: []int{0},
+				Type:    "geography(polygon)",
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"polygon((0 1, 2 4, 3 5, 4 9, 0 1))"},
+					wantValue: &Value{Val: "ST_GeogFromText(\"polygon((0 1, 2 4, 3 5, 4 9, 0 1))\")", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "index unsupported type",
+			c: Config{
+				Indices: []int{0},
+				Type:    "unsupported",
+			},
+			cases: []recordCase{
+				{
+					record:        []string{""},
+					wantErrString: "unsupported type",
+				},
+			},
+		},
+		{
+			name: "index Nullable",
+			c: Config{
+				Indices: []int{1},
+				Type:    "string",
+				Nullable: func(s string) bool {
+					return s == ""
+				},
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"str0", "", "str2", "str3"},
+					wantValue: &Value{Val: "", IsNull: true},
+				},
+			},
+		},
+		{
+			name: "index Nullable value",
+			c: Config{
+				Indices: []int{1},
+				Type:    "string",
+				Nullable: func(s string) bool {
+					return s == ""
+				},
+				NullValue: "",
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"str0", "", "str2", "str3"},
+					wantValue: &Value{Val: "", IsNull: true},
+				},
+			},
+		},
+		{
+			name: "index Nullable value changed",
+			c: Config{
+				Indices: []int{1},
+				Type:    "string",
+				Nullable: func(s string) bool {
+					return s == "__NULL__"
+				},
+				NullValue: "NULL",
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"str0", "__NULL__", "str2", "str3"},
+					wantValue: &Value{Val: "NULL", IsNull: true},
+				},
+			},
+		},
+		{
+			name: "index not Nullable",
+			c: Config{
+				Indices:   []int{1},
+				Type:      "string",
+				Nullable:  nil,
+				NullValue: "NULL",
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"str0", "", "str2", "str3"},
+					wantValue: &Value{Val: "\"\"", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "index not Nullable defaultValue",
+			c: Config{
+				Indices:      []int{1},
+				Type:         "string",
+				Nullable:     nil,
+				NullValue:    "NULL",
+				DefaultValue: &strStr1,
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"str0", "", "str2", "str3"},
+					wantValue: &Value{Val: "\"\"", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "index defaultValue string",
+			c: Config{
+				Indices: []int{1},
+				Type:    "string",
+				Nullable: func(s string) bool {
+					return s == ""
+				},
+				NullValue:    "NULL",
+				DefaultValue: &strStr1,
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"str0", "", "str2", "str3"},
+					wantValue: &Value{Val: "\"str1\"", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "index defaultValue string empty",
+			c: Config{
+				Indices: []int{1},
+				Type:    "string",
+				Nullable: func(s string) bool {
+					return s == "_NULL_"
+				},
+				NullValue:    "NULL",
+				DefaultValue: &strEmpty,
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"str0", "_NULL_", "str2", "str3"},
+					wantValue: &Value{Val: "\"\"", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "index defaultValue int",
+			c: Config{
+				Indices: []int{1},
+				Type:    "int",
+				Nullable: func(s string) bool {
+					return s == ""
+				},
+				NullValue:    "NULL",
+				DefaultValue: &strInt1,
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"0", "", "2", "3"},
+					wantValue: &Value{Val: "1", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "index Function string",
+			c: Config{
+				Indices:  []int{1},
+				Type:     "string",
+				Function: &strFunHash,
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"str0", "str1"},
+					wantValue: &Value{Val: "hash(\"str1\")", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "index Function int",
+			c: Config{
+				Indices:  []int{1, 2, 3},
+				Type:     "int",
+				Function: &strFunHash,
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"0", "1"},
+					wantValue: &Value{Val: "hash(\"1\")", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "index Function Nullable",
+			c: Config{
+				Indices: []int{1},
+				Type:    "string",
+				Nullable: func(s string) bool {
+					return s == ""
+				},
+				NullValue: "NULL",
+				Function:  &strFunHash,
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"str0", "", "str2", "str3"},
+					wantValue: &Value{Val: "NULL", IsNull: true},
+				},
+			},
+		},
+		{
+			name: "index Function defaultValue",
+			c: Config{
+				Indices: []int{1},
+				Type:    "string",
+				Nullable: func(s string) bool {
+					return s == ""
+				},
+				NullValue:    "NULL",
+				DefaultValue: &strStr1,
+				Function:     &strFunHash,
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"str0", "", "str2", "str3"},
+					wantValue: &Value{Val: "hash(\"str1\")", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "indices",
+			c: Config{
+				Indices: []int{1, 2, 3},
+				Type:    "string",
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"str0", "", "str2", "str3"},
+					wantValue: &Value{Val: "\"\"", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "indices unsupported type",
+			c: Config{
+				Indices: []int{1, 2, 3},
+				Type:    "unsupported",
+			},
+			cases: []recordCase{
+				{
+					record:        []string{"str0", "", "", ""},
+					wantErrString: "unsupported type",
+				},
+			},
+		},
+		{
+			name: "indices Nullable unsupported type",
+			c: Config{
+				Indices: []int{1, 2, 3},
+				Type:    "unsupported",
+				Nullable: func(s string) bool {
+					return s == ""
+				},
+				DefaultValue: &strEmpty,
+			},
+			cases: []recordCase{
+				{
+					record:        []string{"str0", "", "", ""},
+					wantErrString: "unsupported type",
+				},
+			},
+		},
+		{
+			name: "indices Nullable",
+			c: Config{
+				Indices: []int{1, 2, 3},
+				Type:    "string",
+				Nullable: func(s string) bool {
+					return s == ""
+				},
+			},
+			cases: []recordCase{
+				{
+					record:        []string{"str0", "", ""},
+					wantErrString: "prop index 3 out range 3 of record",
+				},
+				{
+					record:    []string{"str0", "", "", "str3"},
+					wantValue: &Value{Val: "\"str3\"", IsNull: false},
+				},
+				{
+					record:    []string{"str0", "", "", ""},
+					wantValue: &Value{Val: "", IsNull: true},
+				},
+			},
+		},
+		{
+			name: "indices Nullable value",
+			c: Config{
+				Indices: []int{1, 2, 3},
+				Type:    "string",
+				Nullable: func(s string) bool {
+					return s == ""
+				},
+				NullValue: "",
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"str0", "", "", ""},
+					wantValue: &Value{Val: "", IsNull: true},
+				},
+			},
+		},
+		{
+			name: "indices Nullable value changed",
+			c: Config{
+				Indices: []int{1, 2, 3},
+				Type:    "string",
+				Nullable: func(s string) bool {
+					return s == "__NULL__"
+				},
+				NullValue: "NULL",
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"str0", "__NULL__", "__NULL__", "__NULL__"},
+					wantValue: &Value{Val: "NULL", IsNull: true},
+				},
+			},
+		},
+		{
+			name: "indices not Nullable",
+			c: Config{
+				Indices:   []int{1, 2, 3},
+				Type:      "string",
+				Nullable:  nil,
+				NullValue: "NULL",
+			},
+			cases: []recordCase{
+				{
+					record:        []string{""},
+					wantErrString: "prop index 1 out range 1 of record",
+				},
+				{
+					record:    []string{"str0", "", "", ""},
+					wantValue: &Value{Val: "\"\"", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "indices not Nullable defaultValue",
+			c: Config{
+				Indices:      []int{1, 2, 3},
+				Type:         "string",
+				Nullable:     nil,
+				NullValue:    "NULL",
+				DefaultValue: &strStr1,
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"str0", "", "", ""},
+					wantValue: &Value{Val: "\"\"", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "indices defaultValue string",
+			c: Config{
+				Indices: []int{1, 2, 3},
+				Type:    "string",
+				Nullable: func(s string) bool {
+					return s == ""
+				},
+				NullValue:    "NULL",
+				DefaultValue: &strStr1,
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"str0", "", "", ""},
+					wantValue: &Value{Val: "\"str1\"", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "indices defaultValue string empty",
+			c: Config{
+				Indices: []int{1, 2, 3},
+				Type:    "string",
+				Nullable: func(s string) bool {
+					return s == "_NULL_"
+				},
+				NullValue:    "NULL",
+				DefaultValue: &strEmpty,
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"str0", "_NULL_", "_NULL_", "_NULL_"},
+					wantValue: &Value{Val: "\"\"", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "indices defaultValue int",
+			c: Config{
+				Indices: []int{1, 2, 3},
+				Type:    "int",
+				Nullable: func(s string) bool {
+					return s == ""
+				},
+				NullValue:    "NULL",
+				DefaultValue: &strInt1,
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"0", "", "", ""},
+					wantValue: &Value{Val: "1", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "indices Function string",
+			c: Config{
+				Indices:  []int{1, 2, 3},
+				Type:     "string",
+				Function: &strFunHash,
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"str0", "str1"},
+					wantValue: &Value{Val: "hash(\"str1\")", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "indices Function int",
+			c: Config{
+				Indices:  []int{1, 2, 3},
+				Type:     "int",
+				Function: &strFunHash,
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"0", "1"},
+					wantValue: &Value{Val: "hash(\"1\")", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "indices Function Nullable",
+			c: Config{
+				Indices: []int{1, 2, 3},
+				Type:    "string",
+				Nullable: func(s string) bool {
+					return s == ""
+				},
+				NullValue: "NULL",
+				Function:  &strFunHash,
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"str0", "", "", ""},
+					wantValue: &Value{Val: "NULL", IsNull: true},
+				},
+			},
+		},
+		{
+			name: "indices Function defaultValue",
+			c: Config{
+				Indices: []int{1, 2, 3},
+				Type:    "string",
+				Nullable: func(s string) bool {
+					return s == ""
+				},
+				NullValue:    "NULL",
+				DefaultValue: &strStr1,
+				Function:     &strFunHash,
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"str0", "", "", ""},
+					wantValue: &Value{Val: "hash(\"str1\")", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "concat items",
+			c: Config{
+				Indices: []int{1, 2, 3},
+				Type:    "string",
+				Nullable: func(s string) bool {
+					return s == ""
+				},
+				NullValue:    "NULL",
+				DefaultValue: &strStr1,
+			},
+			fn: func(c *Config) {
+				c.ConcatItems.
+					AddConstant("c1").
+					AddIndex(4).
+					AddIndex(5).
+					AddConstant("c2").
+					AddIndex(6).
+					AddConstant("c3")
+			},
+			cases: []recordCase{
+				{
+					record:        []string{"str0", "str1", "str2", "str3", "str4", "str5"},
+					wantErrString: "prop index 6 out range 6 of record",
+				},
+				{
+					record:    []string{"str0", "str1", "str2", "str3", "str4", "str5", "str6"},
+					wantValue: &Value{Val: "\"c1str4str5c2str6c3\"", IsNull: false},
+				},
+				{
+					record:    []string{"", "", "", "", "", "", ""},
+					wantValue: &Value{Val: "\"c1c2c3\"", IsNull: false},
+				},
+				{
+					record:    []string{"", "", "", "", "str4", "", ""},
+					wantValue: &Value{Val: "\"c1str4c2c3\"", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "concat items Function",
+			c: Config{
+				Indices: []int{1, 2, 3},
+				Type:    "string",
+				Nullable: func(s string) bool {
+					return s == ""
+				},
+				NullValue:    "NULL",
+				DefaultValue: &strStr1,
+				Function:     &strFunHash,
+			},
+			fn: func(c *Config) {
+				c.ConcatItems.
+					AddConstant("c1").
+					AddIndex(4).
+					AddIndex(5).
+					AddConstant("c2").
+					AddIndex(6).
+					AddConstant("c3")
+			},
+			cases: []recordCase{
+				{
+					record:        []string{"str0", "str1", "str2", "str3", "str4", "str5"},
+					wantErrString: "prop index 6 out range 6 of record",
+				},
+				{
+					record:    []string{"str0", "str1", "str2", "str3", "str4", "str5", "str6"},
+					wantValue: &Value{Val: "hash(\"c1str4str5c2str6c3\")", IsNull: false},
+				},
+				{
+					record:    []string{"", "", "", "", "", "", ""},
+					wantValue: &Value{Val: "hash(\"c1c2c3\")", IsNull: false},
+				},
+				{
+					record:    []string{"", "", "", "", "str4", "", ""},
+					wantValue: &Value{Val: "hash(\"c1str4c2c3\")", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "check",
+			c: Config{
+				Indices: []int{1},
+				Type:    "string",
+				CheckOnPost: func(value *Value) error {
+					return nil
+				},
+			},
+			cases: []recordCase{
+				{
+					record:    []string{"0", "str1", "str2"},
+					wantValue: &Value{Val: "\"str1\"", IsNull: false},
+				},
+			},
+		},
+		{
+			name: "check failed",
+			c: Config{
+				Indices: []int{1},
+				Type:    "string",
+				CheckOnPost: func(value *Value) error {
+					return fmt.Errorf("check failed")
+				},
+			},
+			cases: []recordCase{
+				{
+					record:        []string{"0", "str1", "str2"},
+					wantErrString: "check failed",
+				},
+			},
+		},
+	}
+
+	for _, tc := range testcases {
+		t.Run(tc.name, func(t *testing.T) {
+			ast := assert.New(t)
+			if tc.fn != nil {
+				tc.fn(&tc.c)
+			}
+			p, err := tc.c.Build()
+			ast.NoError(err)
+			for i, c := range tc.cases {
+				v, err := p.Pick(c.record)
+				if c.wantErrString == "" {
+					ast.NoError(err, "%d %v", i, c.record)
+					// isSetNull must equal to IsNull
+					c.wantValue.isSetNull = c.wantValue.IsNull
+					ast.Equal(c.wantValue, v, "%d %v", i, c.record)
+				} else {
+					ast.Error(err, "%d %v", i, c.record)
+					ast.Contains(err.Error(), c.wantErrString, "%d %v", i, c.record)
+					ast.Nil(v, "%d %v", i, c.record)
+				}
+			}
+		})
+	}
+}
diff --git a/pkg/picker/converter-default.go b/pkg/picker/converter-default.go
new file mode 100644
index 00000000..d3054e73
--- /dev/null
+++ b/pkg/picker/converter-default.go
@@ -0,0 +1,15 @@
+package picker
+
+var _ Converter = DefaultConverter{}
+
+type DefaultConverter struct {
+	Value string
+}
+
+func (dc DefaultConverter) Convert(v *Value) (*Value, error) {
+	if v.IsNull {
+		v.Val = dc.Value
+		v.IsNull = false
+	}
+	return v, nil
+}
diff --git a/pkg/picker/converter-error.go b/pkg/picker/converter-error.go
new file mode 100644
index 00000000..a5c0f034
--- /dev/null
+++ b/pkg/picker/converter-error.go
@@ -0,0 +1,11 @@
+package picker
+
+var _ Converter = ErrorConverter{}
+
+type ErrorConverter struct {
+	Err error
+}
+
+func (ec ErrorConverter) Convert(v *Value) (*Value, error) {
+	return nil, ec.Err
+}
diff --git a/pkg/picker/converter-function.go b/pkg/picker/converter-function.go
new file mode 100644
index 00000000..33c83d47
--- /dev/null
+++ b/pkg/picker/converter-function.go
@@ -0,0 +1,31 @@
+package picker
+
+import "fmt"
+
+var (
+	_ Converter = FunctionConverter{}
+	_ Converter = FunctionStringConverter{}
+)
+
+type (
+	FunctionConverter struct {
+		Name string
+	}
+	FunctionStringConverter struct {
+		Name string
+	}
+)
+
+func (fc FunctionConverter) Convert(v *Value) (*Value, error) {
+	v.Val = getFuncValue(fc.Name, v.Val)
+	return v, nil
+}
+
+func (fc FunctionStringConverter) Convert(v *Value) (*Value, error) {
+	v.Val = getFuncValue(fc.Name, fmt.Sprintf("%q", v.Val))
+	return v, nil
+}
+
+func getFuncValue(name, value string) string {
+	return name + "(" + value + ")"
+}
diff --git a/pkg/picker/converter-non.go b/pkg/picker/converter-non.go
new file mode 100644
index 00000000..34958a98
--- /dev/null
+++ b/pkg/picker/converter-non.go
@@ -0,0 +1,9 @@
+package picker
+
+var _ Converter = NonConverter{}
+
+type NonConverter struct{}
+
+func (NonConverter) Convert(v *Value) (*Value, error) {
+	return v, nil
+}
diff --git a/pkg/picker/converter-null.go b/pkg/picker/converter-null.go
new file mode 100644
index 00000000..b6c60a50
--- /dev/null
+++ b/pkg/picker/converter-null.go
@@ -0,0 +1,31 @@
+package picker
+
+var (
+	_ Converter = NullableConverter{}
+	_ Converter = NullConverter{}
+)
+
+type (
+	NullableConverter struct {
+		Nullable func(string) bool
+	}
+
+	NullConverter struct {
+		Value string
+	}
+)
+
+func (nc NullableConverter) Convert(v *Value) (*Value, error) {
+	if !v.IsNull && nc.Nullable(v.Val) {
+		v.IsNull = true
+	}
+	return v, nil
+}
+
+func (nc NullConverter) Convert(v *Value) (*Value, error) {
+	if v.IsNull {
+		v.Val = nc.Value
+		v.isSetNull = true
+	}
+	return v, nil
+}
diff --git a/pkg/picker/converter-type.go b/pkg/picker/converter-type.go
new file mode 100644
index 00000000..792d14f7
--- /dev/null
+++ b/pkg/picker/converter-type.go
@@ -0,0 +1,120 @@
+package picker
+
+import (
+	"fmt"
+	"strings"
+)
+
+var (
+	_ Converter = TypeBoolConverter{}
+	_ Converter = TypeIntConverter{}
+	_ Converter = TypeFloatConverter{}
+	_ Converter = TypeDoubleConverter{}
+	_ Converter = TypeStringConverter{}
+	_ Converter = TypeDateConverter{}
+	_ Converter = TypeTimeConverter{}
+	_ Converter = TypeDatetimeConverter{}
+	_ Converter = TypeTimestampConverter{}
+	_ Converter = TypeGeoConverter{}
+	_ Converter = TypeGeoPointConverter{}
+	_ Converter = TypeGeoLineStringConverter{}
+	_ Converter = TypeGeoPolygonConverter{}
+)
+
+type (
+	TypeBoolConverter = NonConverter
+
+	TypeIntConverter = NonConverter
+
+	TypeFloatConverter = NonConverter
+
+	TypeDoubleConverter = NonConverter
+
+	TypeStringConverter struct{}
+
+	TypeDateConverter = FunctionStringConverter
+
+	TypeTimeConverter = FunctionStringConverter
+
+	TypeDatetimeConverter = FunctionStringConverter
+
+	TypeTimestampConverter struct {
+		fc  FunctionConverter
+		fsc FunctionStringConverter
+	}
+
+	TypeGeoConverter = FunctionStringConverter
+
+	TypeGeoPointConverter = FunctionStringConverter
+
+	TypeGeoLineStringConverter = FunctionStringConverter
+
+	TypeGeoPolygonConverter = FunctionStringConverter
+)
+
+func NewTypeConverter(t string) Converter {
+	switch strings.ToUpper(t) {
+	case "BOOL":
+		return TypeBoolConverter{}
+	case "INT":
+		return TypeIntConverter{}
+	case "FLOAT":
+		return TypeFloatConverter{}
+	case "DOUBLE":
+		return TypeDoubleConverter{}
+	case "STRING":
+		return TypeStringConverter{}
+	case "DATE":
+		return TypeDateConverter{
+			Name: "DATE",
+		}
+	case "TIME":
+		return TypeTimeConverter{
+			Name: "TIME",
+		}
+	case "DATETIME":
+		return TypeDatetimeConverter{
+			Name: "DATETIME",
+		}
+	case "TIMESTAMP":
+		return TypeTimestampConverter{
+			fc: FunctionConverter{
+				Name: "TIMESTAMP",
+			},
+			fsc: FunctionStringConverter{
+				Name: "TIMESTAMP",
+			},
+		}
+	case "GEOGRAPHY":
+		return TypeGeoConverter{
+			Name: "ST_GeogFromText",
+		}
+	case "GEOGRAPHY(POINT)":
+		return TypeGeoPointConverter{
+			Name: "ST_GeogFromText",
+		}
+	case "GEOGRAPHY(LINESTRING)":
+		return TypeGeoLineStringConverter{
+			Name: "ST_GeogFromText",
+		}
+	case "GEOGRAPHY(POLYGON)":
+		return TypeGeoPolygonConverter{
+			Name: "ST_GeogFromText",
+		}
+	}
+	return ErrorConverter{
+		Err: fmt.Errorf("unsupported type %s", t),
+	}
+}
+
+func (tc TypeStringConverter) Convert(v *Value) (*Value, error) {
+	v.Val = fmt.Sprintf("%q", v.Val)
+	return v, nil
+}
+
+func (tc TypeTimestampConverter) Convert(v *Value) (*Value, error) {
+	if isUnsignedInteger(v.Val) {
+		return tc.fc.Convert(v)
+	}
+	return tc.fsc.Convert(v)
+}
diff --git a/pkg/picker/converter.go b/pkg/picker/converter.go
new file mode 100644
index 00000000..5eac6c26
--- /dev/null
+++ b/pkg/picker/converter.go
@@ -0,0 +1,66 @@
+package picker
+
+var _ Converter = Converters(nil)
+
+type (
+	Converter interface {
+		Convert(*Value) (*Value, error)
+	}
+
+	ConverterFunc func(v *Value) (*Value, error)
+
+	Converters         []Converter
+	NullableConverters []Converter
+)
+
+func (f ConverterFunc) Convert(v *Value) (*Value, error) {
+	return f(v)
+}
+
+func (cs Converters) Convert(v *Value) (*Value, error) {
+	switch len(cs) {
+	case 0:
+		return v, nil
+	case 1:
+		return cs[0].Convert(v)
+	}
+	return cs.convertSlow(v)
+}
+
+func (cs Converters) convertSlow(v *Value) (*Value, error) {
+	var err error
+	for _, c := range cs {
+		v, err = c.Convert(v)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return v, nil
+}
+
+func (ncs NullableConverters) Convert(v *Value) (*Value, error) {
+	if v.isSetNull {
+		return v, nil
+	}
+	switch len(ncs) {
+	case 0:
+		return v, nil
+	case 1:
+		return ncs[0].Convert(v)
+	}
+	return ncs.convertSlow(v)
+}
+
+func (ncs NullableConverters) convertSlow(v *Value) (*Value, error) {
+	var err error
+	for _, c := range ncs {
+		v, err = c.Convert(v)
+		if err != nil {
+			return nil, err
+		}
+		if v.isSetNull {
+			return v, nil
+		}
+	}
+	return v, nil
+}
diff --git a/pkg/picker/picker-concat.go b/pkg/picker/picker-concat.go
new file mode 100644
index 00000000..f84fd5dc
--- /dev/null
+++ b/pkg/picker/picker-concat.go
@@ -0,0 +1,47 @@
+package picker
+
+import (
+	"strings"
+)
+
+var _ Picker = ConcatPicker{}
+
+type (
+	ConcatItems struct {
+		pickers NullablePickers
+	}
+
+	ConcatPicker struct {
+		items ConcatItems
+	}
+)
+
+func (ci *ConcatItems) AddIndex(index int) *ConcatItems {
+	ci.pickers = append(ci.pickers, IndexPicker(index))
+	return ci
+}
+
+func (ci *ConcatItems) AddConstant(constant string) *ConcatItems {
+	ci.pickers = append(ci.pickers, ConstantPicker(constant))
+	return ci
+}
+
+func (ci ConcatItems) Len() int {
+	return len(ci.pickers)
+}
+
+func (cp ConcatPicker) Pick(record []string) (*Value, error) {
+	var sb strings.Builder
+	for _, p := range cp.items.pickers {
+		v, err := p.Pick(record)
+		if err != nil {
+			return nil, err
+		}
+		sb.WriteString(v.Val)
+	}
+
+	return &Value{
+		Val:    sb.String(),
+		IsNull: false,
+	}, nil
+}
diff --git a/pkg/picker/picker-constant.go b/pkg/picker/picker-constant.go
new file mode 100644
index 00000000..51abb60d
--- /dev/null
+++ b/pkg/picker/picker-constant.go
@@ -0,0 +1,14 @@
+package picker
+
+var (
+	_ Picker = ConstantPicker("")
+)
+
+type ConstantPicker string
+
+func (cp ConstantPicker) Pick(_ []string) (v *Value, err error) {
+	return &Value{
+		Val:    string(cp),
+		IsNull: false,
+	}, nil
+}
diff --git a/pkg/picker/picker-index.go b/pkg/picker/picker-index.go
new file mode 100644
index 00000000..fe5c23b9
--- /dev/null
+++ b/pkg/picker/picker-index.go
@@ -0,0 +1,22 @@
+package picker
+
+import "fmt"
+
+var (
+	_ Picker = IndexPicker(0)
+)
+
+type (
+	IndexPicker int
+)
+
+func (ip IndexPicker) Pick(record []string) (*Value, error) {
+	index := int(ip)
+	if index < 0 || index >= len(record) {
+		return nil, fmt.Errorf("prop index %d out range %d of record(%v)", index, len(record), record)
+	}
+	return &Value{
+		Val:    record[index],
+		IsNull: false,
+	}, nil
+}
diff --git a/pkg/picker/picker.go b/pkg/picker/picker.go
new file mode 100644
index 00000000..f1fd6431
--- /dev/null
+++ b/pkg/picker/picker.go
@@ -0,0 +1,37 @@
+package picker
+
+var _ Picker = ConverterPicker{}
+
+type (
+	Picker interface {
+		Pick([]string) (*Value, error)
+	}
+
+	ConverterPicker struct {
+		picker    Picker
+		converter Converter
+	}
+
+	NullablePickers []Picker
+)
+
+func (cp ConverterPicker) Pick(record []string) (*Value, error) {
+	v, err := cp.picker.Pick(record)
+	if err != nil {
+		return nil, err
+	}
+	return cp.converter.Convert(v)
+}
+
+func (nps NullablePickers) Pick(record []string) (v *Value, err error) {
+	for _, p := range nps {
+		v, err = p.Pick(record)
+		if err != nil {
+			return nil, err
+		}
+		if !v.IsNull {
+			return v, nil
+		}
+	}
+	return v, nil
+}
diff --git a/pkg/picker/utils.go b/pkg/picker/utils.go
new file mode 100644
index 00000000..518b24b2
--- /dev/null
+++ b/pkg/picker/utils.go
@@ -0,0 +1,36 @@
+package picker
+
+func isUnsignedInteger(s string) bool {
+	switch len(s) {
+	case 0:
+		return false
+	case 1:
+		return isDigit(s[0])
+	case 2:
+		return isDigit(s[0]) && isDigit(s[1])
+	}
+	return isIntegerSlow(s)
+}
+
+func isIntegerSlow(s string) bool {
+	f := isDigit
+	if len(s) > 2 && s[0] == '0' && (s[1] == 'x' || s[1] == 'X') {
+		s = s[2:]
+		f = isHexDigit
+	}
+
+	for _, b := range []byte(s) {
+		if !f(b) {
+			return false
+		}
+	}
+	return true
+}
+
+func isDigit(b byte) bool {
+	return '0' <= b && b <= '9'
+}
+
+func isHexDigit(b byte) bool {
+	return isDigit(b) || ('a' <= b && b <= 'f') || ('A' <= b && b <= 'F')
+}
diff --git a/pkg/picker/value.go b/pkg/picker/value.go
new file mode 100644
index 00000000..ac47d85a
--- /dev/null
+++ b/pkg/picker/value.go
@@ -0,0 +1,7 @@
+package picker
+
+type Value struct {
+	Val       string
+	IsNull    bool
+	isSetNull bool
+}