func HeadToUpper(str string) string {
	ln := len(str)
	if ln <= 0 {
		return str
	}
	c := str[0]
	if (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') {
		return strings.ToUpper(str[0:1]) + str[1:]
	}
	return "PARGO_PREFIX_" + str
}
Convert the first letter of a string to uppercase
package common
import (
	"bytes"
	"encoding/gob"
	"errors"
	"fmt"
	"reflect"
	"strconv"
	"strings"
	"github.com/apache/arrow/go/arrow"
	"github.com/apache/arrow/go/arrow/array"
	"github.com/xitongsys/parquet-go/parquet"
	"github.com/xitongsys/parquet-go/types"
)
type Tag struct {
	InName string
	ExName string
	Type      string
	KeyType   string
	ValueType string
	ConvertedType      string
	KeyConvertedType   string
	ValueConvertedType string
	Length      int32
	KeyLength   int32
	ValueLength int32
	Scale      int32
	KeyScale   int32
	ValueScale int32
	Precision      int32
	KeyPrecision   int32
	ValuePrecision int32
	IsAdjustedToUTC      bool
	KeyIsAdjustedToUTC   bool
	ValueIsAdjustedToUTC bool
	FieldID      int32
	KeyFieldID   int32
	ValueFieldID int32
	Encoding      parquet.Encoding
	KeyEncoding   parquet.Encoding
	ValueEncoding parquet.Encoding
	OmitStats      bool
	KeyOmitStats   bool
	ValueOmitStats bool
	RepetitionType      parquet.FieldRepetitionType
	KeyRepetitionType   parquet.FieldRepetitionType
	ValueRepetitionType parquet.FieldRepetitionType
	LogicalTypeFields      map[string]string
	KeyLogicalTypeFields   map[string]string
	ValueLogicalTypeFields map[string]string
}
func NewTag() *Tag {
	return &Tag{
		LogicalTypeFields:      make(map[string]string),
		KeyLogicalTypeFields:   make(map[string]string),
		ValueLogicalTypeFields: make(map[string]string),
	}
}
func StringToTag(tag string) (*Tag, error) {
	mp := NewTag()
	tagStr := strings.Replace(tag, "\t", "", -1)
	tags := strings.Split(tagStr, ",")
	for _, tag := range tags {
		tag = strings.TrimSpace(tag)
		kv := strings.SplitN(tag, "=", 2)
		if len(kv) != 2 {
			return nil, fmt.Errorf("expect 'key=value' but got '%s'", tag)
		}
		key := kv[0]
		key = strings.ToLower(key)
		key = strings.TrimSpace(key)
		val := kv[1]
		val = strings.TrimSpace(val)
		var err error
		switch key {
		case "type":
			mp.Type = val
		case "keytype":
			mp.KeyType = val
		case "valuetype":
			mp.ValueType = val
		case "convertedtype":
			mp.ConvertedType = val
		case "keyconvertedtype":
			mp.KeyConvertedType = val
		case "valueconvertedtype":
			mp.ValueConvertedType = val
		case "length":
			if mp.Length, err = Str2Int32(val); err != nil {
				return nil, fmt.Errorf("failed to parse length: %s", err.Error())
			}
		case "keylength":
			if mp.KeyLength, err = Str2Int32(val); err != nil {
				return nil, fmt.Errorf("failed to parse keylength: %s", err.Error())
			}
		case "valuelength":
			if mp.ValueLength, err = Str2Int32(val); err != nil {
				return nil, fmt.Errorf("failed to parse valuelength: %s", err.Error())
			}
		case "scale":
			if mp.Scale, err = Str2Int32(val); err != nil {
				return nil, fmt.Errorf("failed to parse scale: %s", err.Error())
			}
		case "keyscale":
			if mp.KeyScale, err = Str2Int32(val); err != nil {
				return nil, fmt.Errorf("failed to parse keyscale: %s", err.Error())
			}
		case "valuescale":
			if mp.ValueScale, err = Str2Int32(val); err != nil {
				return nil, fmt.Errorf("failed to parse valuescale: %s", err.Error())
			}
		case "precision":
			if mp.Precision, err = Str2Int32(val); err != nil {
				return nil, fmt.Errorf("failed to parse precision: %s", err.Error())
			}
		case "keyprecision":
			if mp.KeyPrecision, err = Str2Int32(val); err != nil {
				return nil, fmt.Errorf("failed to parse keyprecision: %s", err.Error())
			}
		case "valueprecision":
			if mp.ValuePrecision, err = Str2Int32(val); err != nil {
				return nil, fmt.Errorf("failed to parse valueprecision: %s", err.Error())
			}
		case "fieldid":
			if mp.FieldID, err = Str2Int32(val); err != nil {
				return nil, fmt.Errorf("failed to parse fieldid: %s", err.Error())
			}
		case "keyfieldid":
			if mp.KeyFieldID, err = Str2Int32(val); err != nil {
				return nil, fmt.Errorf("failed to parse keyfieldid: %s", err.Error())
			}
		case "valuefieldid":
			if mp.ValueFieldID, err = Str2Int32(val); err != nil {
				return nil, fmt.Errorf("failed to parse valuefieldid: %s", err.Error())
			}
		case "isadjustedtoutc":
			if mp.IsAdjustedToUTC, err = Str2Bool(val); err != nil {
				return nil, fmt.Errorf("failed to parse isadjustedtoutc: %s", err.Error())
			}
		case "keyisadjustedtoutc":
			if mp.KeyIsAdjustedToUTC, err = Str2Bool(val); err != nil {
				return nil, fmt.Errorf("failed to parse keyisadjustedtoutc: %s", err.Error())
			}
		case "valueisadjustedtoutc":
			if mp.ValueIsAdjustedToUTC, err = Str2Bool(val); err != nil {
				return nil, fmt.Errorf("failed to parse valueisadjustedtoutc: %s", err.Error())
			}
		case "name":
			if mp.InName == "" {
				mp.InName = StringToVariableName(val)
			}
			mp.ExName = val
		case "inname":
			mp.InName = val
		case "omitstats":
			if mp.OmitStats, err = Str2Bool(val); err != nil {
				return nil, fmt.Errorf("failed to parse omitstats: %s", err.Error())
			}
		case "keyomitstats":
			if mp.KeyOmitStats, err = Str2Bool(val); err != nil {
				return nil, fmt.Errorf("failed to parse keyomitstats: %s", err.Error())
			}
		case "valueomitstats":
			if mp.ValueOmitStats, err = Str2Bool(val); err != nil {
				return nil, fmt.Errorf("failed to parse valueomitstats: %s", err.Error())
			}
		case "repetitiontype":
			switch strings.ToLower(val) {
			case "repeated":
				mp.RepetitionType = parquet.FieldRepetitionType_REPEATED
			case "required":
				mp.RepetitionType = parquet.FieldRepetitionType_REQUIRED
			case "optional":
				mp.RepetitionType = parquet.FieldRepetitionType_OPTIONAL
			default:
				return nil, fmt.Errorf("unknown repetitiontype: '%v'", val)
			}
		case "keyrepetitiontype":
			switch strings.ToLower(val) {
			case "repeated":
				mp.KeyRepetitionType = parquet.FieldRepetitionType_REPEATED
			case "required":
				mp.KeyRepetitionType = parquet.FieldRepetitionType_REQUIRED
			case "optional":
				mp.KeyRepetitionType = parquet.FieldRepetitionType_OPTIONAL
			default:
				return nil, fmt.Errorf("unknown keyrepetitiontype: '%v'", val)
			}
		case "valuerepetitiontype":
			switch strings.ToLower(val) {
			case "repeated":
				mp.ValueRepetitionType = parquet.FieldRepetitionType_REPEATED
			case "required":
				mp.ValueRepetitionType = parquet.FieldRepetitionType_REQUIRED
			case "optional":
				mp.ValueRepetitionType = parquet.FieldRepetitionType_OPTIONAL
			default:
				return nil, fmt.Errorf("unknown valuerepetitiontype: '%v'", val)
			}
		case "encoding":
			switch strings.ToLower(val) {
			case "plain":
				mp.Encoding = parquet.Encoding_PLAIN
			case "rle":
				mp.Encoding = parquet.Encoding_RLE
			case "delta_binary_packed":
				mp.Encoding = parquet.Encoding_DELTA_BINARY_PACKED
			case "delta_length_byte_array":
				mp.Encoding = parquet.Encoding_DELTA_LENGTH_BYTE_ARRAY
			case "delta_byte_array":
				mp.Encoding = parquet.Encoding_DELTA_BYTE_ARRAY
			case "plain_dictionary":
				mp.Encoding = parquet.Encoding_PLAIN_DICTIONARY
			case "rle_dictionary":
				mp.Encoding = parquet.Encoding_RLE_DICTIONARY
			case "byte_stream_split":
				mp.Encoding = parquet.Encoding_BYTE_STREAM_SPLIT
			default:
				return nil, fmt.Errorf("unknown encoding type: '%v'", val)
			}
		case "keyencoding":
			switch strings.ToLower(val) {
			case "rle":
				mp.KeyEncoding = parquet.Encoding_RLE
			case "delta_binary_packed":
				mp.KeyEncoding = parquet.Encoding_DELTA_BINARY_PACKED
			case "delta_length_byte_array":
				mp.KeyEncoding = parquet.Encoding_DELTA_LENGTH_BYTE_ARRAY
			case "delta_byte_array":
				mp.KeyEncoding = parquet.Encoding_DELTA_BYTE_ARRAY
			case "plain_dictionary":
				mp.KeyEncoding = parquet.Encoding_PLAIN_DICTIONARY
			case "byte_stream_split":
				mp.KeyEncoding = parquet.Encoding_BYTE_STREAM_SPLIT
			default:
				return nil, fmt.Errorf("unknown keyencoding type: '%v'", val)
			}
		case "valueencoding":
			switch strings.ToLower(val) {
			case "rle":
				mp.ValueEncoding = parquet.Encoding_RLE
			case "delta_binary_packed":
				mp.ValueEncoding = parquet.Encoding_DELTA_BINARY_PACKED
			case "delta_length_byte_array":
				mp.ValueEncoding = parquet.Encoding_DELTA_LENGTH_BYTE_ARRAY
			case "delta_byte_array":
				mp.ValueEncoding = parquet.Encoding_DELTA_BYTE_ARRAY
			case "plain_dictionary":
				mp.ValueEncoding = parquet.Encoding_PLAIN_DICTIONARY
			case "byte_stream_split":
				mp.ValueEncoding = parquet.Encoding_BYTE_STREAM_SPLIT
			default:
				return nil, fmt.Errorf("unknown valueencoding type: '%v'", val)
			}
		default:
			if strings.HasPrefix(key, "logicaltype") {
				mp.LogicalTypeFields[key] = val
			} else if strings.HasPrefix(key, "keylogicaltype") {
				newKey := key[3:]
				mp.KeyLogicalTypeFields[newKey] = val
			} else if strings.HasPrefix(key, "valuelogicaltype") {
				newKey := key[5:]
				mp.ValueLogicalTypeFields[newKey] = val
			} else {
				return nil, fmt.Errorf("unrecognized tag '%v'", key)
			}
		}
	}
	return mp, nil
}
func NewSchemaElementFromTagMap(info *Tag) (*parquet.SchemaElement, error) {
	schema := parquet.NewSchemaElement()
	schema.Name = info.InName
	schema.TypeLength = &info.Length
	schema.Scale = &info.Scale
	schema.Precision = &info.Precision
	schema.FieldID = &info.FieldID
	schema.RepetitionType = &info.RepetitionType
	schema.NumChildren = nil
	if t, err := parquet.TypeFromString(info.Type); err == nil {
		schema.Type = &t
	} else {
		return nil, fmt.Errorf("type " + info.Type + ": " + err.Error())
	}
	if ct, err := parquet.ConvertedTypeFromString(info.ConvertedType); err == nil {
		schema.ConvertedType = &ct
	}
	var logicalType *parquet.LogicalType
	var err error
	if len(info.LogicalTypeFields) > 0 {
		logicalType, err = NewLogicalTypeFromFieldsMap(info.LogicalTypeFields)
		if err != nil {
			return nil, fmt.Errorf("failed to create logicaltype from field map: %s", err.Error())
		}
	} else {
		logicalType = NewLogicalTypeFromConvertedType(schema, info)
	}
	schema.LogicalType = logicalType
	return schema, nil
}
func NewLogicalTypeFromFieldsMap(mp map[string]string) (*parquet.LogicalType, error) {
	if val, ok := mp["logicaltype"]; !ok {
		return nil, errors.New("does not have logicaltype")
	} else {
		var err error
		logicalType := parquet.NewLogicalType()
		switch val {
		case "STRING":
			logicalType.STRING = parquet.NewStringType()
		case "MAP":
			logicalType.MAP = parquet.NewMapType()
		case "LIST":
			logicalType.LIST = parquet.NewListType()
		case "ENUM":
			logicalType.ENUM = parquet.NewEnumType()
		case "DECIMAL":
			logicalType.DECIMAL = parquet.NewDecimalType()
			logicalType.DECIMAL.Precision, err = Str2Int32(mp["logicaltype.precision"])
			if err != nil {
				return nil, fmt.Errorf("cannot parse logicaltype.precision as int32: %s", err.Error())
			}
			logicalType.DECIMAL.Scale, err = Str2Int32(mp["logicaltype.scale"])
			if err != nil {
				return nil, fmt.Errorf("cannot parse logicaltype.scale as int32: %s", err.Error())
			}
		case "DATE":
			logicalType.DATE = parquet.NewDateType()
		case "TIME":
			logicalType.TIME = parquet.NewTimeType()
			logicalType.TIME.IsAdjustedToUTC, err = Str2Bool(mp["logicaltype.isadjustedtoutc"])
			if err != nil {
				return nil, fmt.Errorf("cannot parse logicaltype.isadjustedtoutc as boolean: %s", err.Error())
			}
			switch mp["logicaltype.unit"] {
			case "MILLIS":
				logicalType.TIME.Unit = parquet.NewTimeUnit()
				logicalType.TIME.Unit.MILLIS = parquet.NewMilliSeconds()
			case "MICROS":
				logicalType.TIME.Unit = parquet.NewTimeUnit()
				logicalType.TIME.Unit.MICROS = parquet.NewMicroSeconds()
			case "NANOS":
				logicalType.TIME.Unit = parquet.NewTimeUnit()
				logicalType.TIME.Unit.NANOS = parquet.NewNanoSeconds()
			default:
				return nil, fmt.Errorf("logicaltype time error, unknown unit: %s", mp["logicaltype.unit"])
			}
		case "TIMESTAMP":
			logicalType.TIMESTAMP = parquet.NewTimestampType()
			logicalType.TIMESTAMP.IsAdjustedToUTC, err = Str2Bool(mp["logicaltype.isadjustedtoutc"])
			if err != nil {
				return nil, fmt.Errorf("cannot parse logicaltype.isadjustedtoutc as boolean: %s", err.Error())
			}
			switch mp["logicaltype.unit"] {
			case "MILLIS":
				logicalType.TIMESTAMP.Unit = parquet.NewTimeUnit()
				logicalType.TIMESTAMP.Unit.MILLIS = parquet.NewMilliSeconds()
			case "MICROS":
				logicalType.TIMESTAMP.Unit = parquet.NewTimeUnit()
				logicalType.TIMESTAMP.Unit.MICROS = parquet.NewMicroSeconds()
			case "NANOS":
				logicalType.TIMESTAMP.Unit = parquet.NewTimeUnit()
				logicalType.TIMESTAMP.Unit.NANOS = parquet.NewNanoSeconds()
			default:
				return nil, fmt.Errorf("logicaltype time error, unknown unit: %s", mp["logicaltype.unit"])
			}
		case "INTEGER":
			logicalType.INTEGER = parquet.NewIntType()
			bitWidth, err := Str2Int32(mp["logicaltype.bitwidth"])
			if err != nil {
				return nil, fmt.Errorf("cannot parse logicaltype.bitwidth as int32: %s", err.Error())
			}
			logicalType.INTEGER.BitWidth = int8(bitWidth)
			logicalType.INTEGER.IsSigned, err = Str2Bool(mp["logicaltype.issigned"])
			if err != nil {
				return nil, fmt.Errorf("cannot parse logicaltype.issigned as boolean: %s", err.Error())
			}
		case "JSON":
			logicalType.JSON = parquet.NewJsonType()
		case "BSON":
			logicalType.BSON = parquet.NewBsonType()
		case "UUID":
			logicalType.UUID = parquet.NewUUIDType()
		default:
			return nil, fmt.Errorf("unknow logicaltype: " + val)
		}
		return logicalType, nil
	}
}
func NewLogicalTypeFromConvertedType(schemaElement *parquet.SchemaElement, info *Tag) *parquet.LogicalType {
	_, ct := schemaElement.Type, schemaElement.ConvertedType
	if ct == nil {
		return nil
	}
	logicalType := parquet.NewLogicalType()
	switch *ct {
	case parquet.ConvertedType_INT_8:
		logicalType.INTEGER = parquet.NewIntType()
		logicalType.INTEGER.BitWidth = 8
		logicalType.INTEGER.IsSigned = true
	case parquet.ConvertedType_INT_16:
		logicalType.INTEGER = parquet.NewIntType()
		logicalType.INTEGER.BitWidth = 16
		logicalType.INTEGER.IsSigned = true
	case parquet.ConvertedType_INT_32:
		logicalType.INTEGER = parquet.NewIntType()
		logicalType.INTEGER.BitWidth = 32
		logicalType.INTEGER.IsSigned = true
	case parquet.ConvertedType_INT_64:
		logicalType.INTEGER = parquet.NewIntType()
		logicalType.INTEGER.BitWidth = 64
		logicalType.INTEGER.IsSigned = true
	case parquet.ConvertedType_UINT_8:
		logicalType.INTEGER = parquet.NewIntType()
		logicalType.INTEGER.BitWidth = 8
		logicalType.INTEGER.IsSigned = false
	case parquet.ConvertedType_UINT_16:
		logicalType.INTEGER = parquet.NewIntType()
		logicalType.INTEGER.BitWidth = 16
		logicalType.INTEGER.IsSigned = false
	case parquet.ConvertedType_UINT_32:
		logicalType.INTEGER = parquet.NewIntType()
		logicalType.INTEGER.BitWidth = 32
		logicalType.INTEGER.IsSigned = false
	case parquet.ConvertedType_UINT_64:
		logicalType.INTEGER = parquet.NewIntType()
		logicalType.INTEGER.BitWidth = 64
		logicalType.INTEGER.IsSigned = false
	case parquet.ConvertedType_DECIMAL:
		logicalType.DECIMAL = parquet.NewDecimalType()
		logicalType.DECIMAL.Precision = info.Precision
		logicalType.DECIMAL.Scale = info.Scale
	case parquet.ConvertedType_DATE:
		logicalType.DATE = parquet.NewDateType()
	case parquet.ConvertedType_TIME_MICROS:
		logicalType.TIME = parquet.NewTimeType()
		logicalType.TIME.IsAdjustedToUTC = info.IsAdjustedToUTC
		logicalType.TIME.Unit = parquet.NewTimeUnit()
		logicalType.TIME.Unit.MICROS = parquet.NewMicroSeconds()
	case parquet.ConvertedType_TIME_MILLIS:
		logicalType.TIME = parquet.NewTimeType()
		logicalType.TIME.IsAdjustedToUTC = info.IsAdjustedToUTC
		logicalType.TIME.Unit = parquet.NewTimeUnit()
		logicalType.TIME.Unit.MILLIS = parquet.NewMilliSeconds()
	case parquet.ConvertedType_TIMESTAMP_MICROS:
		logicalType.TIMESTAMP = parquet.NewTimestampType()
		logicalType.TIMESTAMP.IsAdjustedToUTC = info.IsAdjustedToUTC
		logicalType.TIMESTAMP.Unit = parquet.NewTimeUnit()
		logicalType.TIMESTAMP.Unit.MICROS = parquet.NewMicroSeconds()
	case parquet.ConvertedType_TIMESTAMP_MILLIS:
		logicalType.TIMESTAMP = parquet.NewTimestampType()
		logicalType.TIMESTAMP.IsAdjustedToUTC = info.IsAdjustedToUTC
		logicalType.TIMESTAMP.Unit = parquet.NewTimeUnit()
		logicalType.TIMESTAMP.Unit.MILLIS = parquet.NewMilliSeconds()
	case parquet.ConvertedType_BSON:
		logicalType.BSON = parquet.NewBsonType()
	case parquet.ConvertedType_ENUM:
		logicalType.ENUM = parquet.NewEnumType()
	case parquet.ConvertedType_JSON:
		logicalType.JSON = parquet.NewJsonType()
	case parquet.ConvertedType_LIST:
		logicalType.LIST = parquet.NewListType()
	case parquet.ConvertedType_MAP:
		logicalType.MAP = parquet.NewMapType()
	case parquet.ConvertedType_UTF8:
		logicalType.STRING = parquet.NewStringType()
	default:
		return nil
	}
	return logicalType
}
func DeepCopy(src, dst interface{}) {
	var buf bytes.Buffer
	gob.NewEncoder(&buf).Encode(src)
	gob.NewDecoder(bytes.NewBuffer(buf.Bytes())).Decode(dst)
	return
}
func GetKeyTagMap(src *Tag) *Tag {
	res := NewTag()
	res.InName = "Key"
	res.ExName = "key"
	res.Type = src.KeyType
	res.ConvertedType = src.KeyConvertedType
	res.IsAdjustedToUTC = src.KeyIsAdjustedToUTC
	res.Length = src.KeyLength
	res.Scale = src.KeyScale
	res.Precision = src.KeyPrecision
	res.FieldID = src.KeyFieldID
	res.Encoding = src.KeyEncoding
	res.OmitStats = src.KeyOmitStats
	res.RepetitionType = parquet.FieldRepetitionType_REQUIRED
	return res
}
func GetValueTagMap(src *Tag) *Tag {
	res := NewTag()
	res.InName = "Value"
	res.ExName = "value"
	res.Type = src.ValueType
	res.ConvertedType = src.ValueConvertedType
	res.IsAdjustedToUTC = src.ValueIsAdjustedToUTC
	res.Length = src.ValueLength
	res.Scale = src.ValueScale
	res.Precision = src.ValuePrecision
	res.FieldID = src.ValueFieldID
	res.Encoding = src.ValueEncoding
	res.OmitStats = src.ValueOmitStats
	res.RepetitionType = src.ValueRepetitionType
	return res
}
func StringToVariableName(str string) string {
	ln := len(str)
	if ln <= 0 {
		return str
	}
	name := ""
	for i := 0; i < ln; i++ {
		c := str[i]
		if (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') || c == '_' {
			name += string(c)
		} else {
			name += strconv.Itoa(int(c))
		}
	}
	name = HeadToUpper(name)
	return name
}