func()

in metastore/validator.go [80:177]
83 lines of code
32 McCabe index (conditional complexity)

func (v tableSchemaValidatorImpl) validateIndividualSchema(table *common.Table, creation bool) (err error) {
	var colIdDedup []bool

	nonDeletedColumnsCount := 0
	colNameDedup := make(map[string]bool)
	for columnID, column := range table.Columns {
		if !column.Deleted {
			nonDeletedColumnsCount++
		}
		if colNameDedup[column.Name] {
			return common.ErrDuplicatedColumnName
		}
		colNameDedup[column.Name] = true

		// validate data type
		if dataType := memCom.DataTypeFromString(column.Type); dataType == memCom.Unknown {
			return common.ErrInvalidDataType
		} else if table.IsFactTable && columnID == 0 && dataType != memCom.Uint32 {
			return common.ErrMissingTimeColumn
		}

		// validate hll config
		if err := validateColumnHLLConfig(column); err != nil {
			return err
		}

		// time column does not allow hll config
		if table.IsFactTable && columnID == 0 && column.HLLConfig.IsHLLColumn {
			return common.ErrTimeColumnDoesNotAllowHLLConfig
		}

		if column.DefaultValue != nil {
			if table.IsFactTable && columnID == 0 {
				return common.ErrTimeColumnDoesNotAllowDefault
			}

			if column.HLLConfig.IsHLLColumn {
				return common.ErrHLLColumnDoesNotAllowDefaultValue
			}

			err = ValidateDefaultValue(*column.DefaultValue, column.Type)
			if err != nil {
				return err
			}
		}
	}
	if nonDeletedColumnsCount == 0 {
		return common.ErrAllColumnsInvalid
	}

	if len(table.PrimaryKeyColumns) == 0 {
		return common.ErrMissingPrimaryKey
	}

	colIdDedup = make([]bool, len(table.Columns))
	for _, colId := range table.PrimaryKeyColumns {
		if colId >= len(table.Columns) {
			return common.ErrColumnNonExist
		}
		if table.Columns[colId].Deleted {
			return common.ErrColumnDeleted
		}
		if colIdDedup[colId] {
			return common.ErrDuplicatedColumn
		}
		colDataType := memCom.DataTypeFromString(table.Columns[colId].Type)
		if memCom.IsArrayType(colDataType) {
			return common.ErrInvalidPrimaryKeyDataType
		}
		colIdDedup[colId] = true
	}

	if err := validator.Validate(table.Config); err != nil {
		return utils.StackError(err, "invalid table config")
	}

	if table.IsFactTable {
		colIdDedup = make([]bool, len(table.Columns))
		for _, sortColumnId := range table.ArchivingSortColumns {
			if sortColumnId >= len(table.Columns) {
				return common.ErrColumnNonExist
			}
			if table.Columns[sortColumnId].Deleted {
				return common.ErrColumnDeleted
			}
			if colIdDedup[sortColumnId] {
				return common.ErrDuplicatedColumn
			}
			colDataType := memCom.DataTypeFromString(table.Columns[sortColumnId].Type)
			if colDataType >= memCom.ArrayBool && colDataType <= memCom.ArrayInt64 {
				return common.ErrInvalidSortColumnDataType
			}
			colIdDedup[sortColumnId] = true
		}
	}

	return
}