Method PrepareEnumCases

client/schema.go:199–255

PrepareEnumCases prepares enum cases

(tableName, columnName string, enumCases []string)

// PrepareEnumCases prepares enum cases
func (cf *CachedSchemaHandler) PrepareEnumCases(tableName, columnName string, enumCases []string) error {
	newEnumCases := make([]string, 0, len(enumCases))
	cf.RLock()
	schema, exist := cf.schemas[tableName]
	if !exist {
		cf.RUnlock()
		return nil
	}
	columnID, exist := schema.ColumnDict[columnName]
	if !exist {
		cf.RUnlock()
		return nil
	}

	caseInsensitive := schema.Table.Columns[columnID].CaseInsensitive
	disableAutoExpand := schema.Table.Columns[columnID].DisableAutoExpand
	for _, enumCase := range enumCases {
		if _, exist := cf.enumMappings[tableName][columnID][enumCase]; !exist {
			newEnumCases = append(newEnumCases, enumCase)
		}
	}
	cf.RUnlock()

	numNewEnumCases := len(newEnumCases)
	if disableAutoExpand && numNewEnumCases > 0 {
		// It's recommended to set up elk or sentry logging to catch this error.
		cf.logger.With(
			"TableName", tableName,
			"ColumnName", columnName,
			"ColumnID", columnID,
			"newEnumCasesSet", newEnumCases,
			"caseInsensitive", caseInsensitive,
		).Warn("Finding new enum cases during ingestion but enum auto expansion is disabled")
		cf.metricScope.Tagged(
			map[string]string{
				"TableName": tableName,
				"ColumnID":  strconv.Itoa(columnID),
			},
		).Counter("new_enum_cases_ignored").Inc(int64(numNewEnumCases))
		return nil
	}

	enumIDs, err := cf.schemaFetcher.ExtendEnumCases(tableName, columnName, newEnumCases)
	if err != nil {
		return err
	}

	cf.Lock()
	for index, enumCase := range newEnumCases {
		if caseInsensitive {
			enumCase = strings.ToLower(enumCase)
		}
		cf.enumMappings[tableName][columnID][enumCase] = enumIDs[index]
	}
	cf.Unlock()
	return nil
}