// Generated by tmpl
// https://github.com/benbjohnson/tmpl
//
// DO NOT EDIT!
// Source: fill.gen.go.tmpl

package universe

import (
	"fmt"

	"github.com/influxdata/flux"
	"github.com/influxdata/flux/array"
	"github.com/influxdata/flux/values"
)

func (t *fillTransformation) fillColumn(typ flux.ColType, arr array.Interface, fillValue *interface{}) array.Interface {
	switch typ {
	case flux.TInt:
		return t.fillIntColumn(arr.(*array.Int), fillValue)
	case flux.TUInt:
		return t.fillUintColumn(arr.(*array.Uint), fillValue)
	case flux.TFloat:
		return t.fillFloatColumn(arr.(*array.Float), fillValue)
	case flux.TBool:
		return t.fillBooleanColumn(arr.(*array.Boolean), fillValue)
	case flux.TString:
		return t.fillStringColumn(arr.(*array.String), fillValue)
	case flux.TTime:
		return t.fillTimeColumn(arr.(*array.Int), fillValue)

	default:
		panic(fmt.Errorf("unsupported array data type: %s", arr.DataType()))
	}
}

func (t *fillTransformation) fillIntColumn(arr *array.Int, fillValue *interface{}) array.Interface {
	fillValueNull := *fillValue == nil
	var fillValueInt int64
	if !fillValueNull {
		fillValueInt = (*fillValue).(int64)
	}
	b := array.NewIntBuilder(t.alloc)
	b.Resize(arr.Len())
	for i := 0; i < arr.Len(); i++ {
		if arr.IsNull(i) {
			if fillValueNull {
				b.AppendNull()
			} else {
				b.Append(fillValueInt)
			}
		} else {
			v := arr.Value(i)
			b.Append(v)
			if t.spec.UsePrevious {
				fillValueInt = v
				fillValueNull = false
			}
		}
	}
	if t.spec.UsePrevious && !fillValueNull {
		*fillValue = fillValueInt
	}
	return b.NewArray()
}

func (t *fillTransformation) fillUintColumn(arr *array.Uint, fillValue *interface{}) array.Interface {
	fillValueNull := *fillValue == nil
	var fillValueUint uint64
	if !fillValueNull {
		fillValueUint = (*fillValue).(uint64)
	}
	b := array.NewUintBuilder(t.alloc)
	b.Resize(arr.Len())
	for i := 0; i < arr.Len(); i++ {
		if arr.IsNull(i) {
			if fillValueNull {
				b.AppendNull()
			} else {
				b.Append(fillValueUint)
			}
		} else {
			v := arr.Value(i)
			b.Append(v)
			if t.spec.UsePrevious {
				fillValueUint = v
				fillValueNull = false
			}
		}
	}
	if t.spec.UsePrevious && !fillValueNull {
		*fillValue = fillValueUint
	}
	return b.NewArray()
}

func (t *fillTransformation) fillFloatColumn(arr *array.Float, fillValue *interface{}) array.Interface {
	fillValueNull := *fillValue == nil
	var fillValueFloat float64
	if !fillValueNull {
		fillValueFloat = (*fillValue).(float64)
	}
	b := array.NewFloatBuilder(t.alloc)
	b.Resize(arr.Len())
	for i := 0; i < arr.Len(); i++ {
		if arr.IsNull(i) {
			if fillValueNull {
				b.AppendNull()
			} else {
				b.Append(fillValueFloat)
			}
		} else {
			v := arr.Value(i)
			b.Append(v)
			if t.spec.UsePrevious {
				fillValueFloat = v
				fillValueNull = false
			}
		}
	}
	if t.spec.UsePrevious && !fillValueNull {
		*fillValue = fillValueFloat
	}
	return b.NewArray()
}

func (t *fillTransformation) fillBooleanColumn(arr *array.Boolean, fillValue *interface{}) array.Interface {
	fillValueNull := *fillValue == nil
	var fillValueBoolean bool
	if !fillValueNull {
		fillValueBoolean = (*fillValue).(bool)
	}
	b := array.NewBooleanBuilder(t.alloc)
	b.Resize(arr.Len())
	for i := 0; i < arr.Len(); i++ {
		if arr.IsNull(i) {
			if fillValueNull {
				b.AppendNull()
			} else {
				b.Append(fillValueBoolean)
			}
		} else {
			v := arr.Value(i)
			b.Append(v)
			if t.spec.UsePrevious {
				fillValueBoolean = v
				fillValueNull = false
			}
		}
	}
	if t.spec.UsePrevious && !fillValueNull {
		*fillValue = fillValueBoolean
	}
	return b.NewArray()
}

func (t *fillTransformation) fillStringColumn(arr *array.String, fillValue *interface{}) array.Interface {
	fillValueNull := *fillValue == nil
	var fillValueString string
	if !fillValueNull {
		fillValueString = (*fillValue).(string)
	}
	b := array.NewStringBuilder(t.alloc)
	b.Resize(arr.Len())
	for i := 0; i < arr.Len(); i++ {
		if arr.IsNull(i) {
			if fillValueNull {
				b.AppendNull()
			} else {
				b.Append(fillValueString)
			}
		} else {
			v := arr.Value(i)
			b.Append(v)
			if t.spec.UsePrevious {
				fillValueString = v
				fillValueNull = false
			}
		}
	}
	if t.spec.UsePrevious && !fillValueNull {
		*fillValue = fillValueString
	}
	return b.NewArray()
}

func (t *fillTransformation) fillTimeColumn(arr *array.Int, fillValue *interface{}) array.Interface {
	fillValueNull := *fillValue == nil
	var fillValueTime int64
	if !fillValueNull {
		fillValueTime = int64((*fillValue).(values.Time))
	}
	b := array.NewIntBuilder(t.alloc)
	b.Resize(arr.Len())
	for i := 0; i < arr.Len(); i++ {
		if arr.IsNull(i) {
			if fillValueNull {
				b.AppendNull()
			} else {
				b.Append(fillValueTime)
			}
		} else {
			v := arr.Value(i)
			b.Append(v)
			if t.spec.UsePrevious {
				fillValueTime = v
				fillValueNull = false
			}
		}
	}
	if t.spec.UsePrevious && !fillValueNull {
		*fillValue = fillValueTime
	}
	return b.NewArray()
}
