fill_processor.gen.go

/*
Copyright 2022 Huawei Cloud Computing Technologies Co., Ltd.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

 http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
// Code generated by tmpl; DO NOT EDIT.
// https://github.com/benbjohnson/tmpl
//
// Source: fill_processor.gen.go.tmpl

package executor

import "github.com/openGemini/openGemini/engine/hybridqp"

type FloatLinearFillProcessor struct {
	inOrdinal  int
	outOrdinal int
}

func NewFloatLinearFillProcessor(inOrdinal, outOrdinal int) *FloatLinearFillProcessor {
	return &FloatLinearFillProcessor{
		inOrdinal:  inOrdinal,
		outOrdinal: outOrdinal,
	}
}

func (f *FloatLinearFillProcessor) fillHelperFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	if (prev != nil && fillItem.inputReadAt >= fillItem.prevReadAt) &&
		(fillItem.prevReadAt < prev.Len() && !prev.Column(f.inOrdinal).IsNilV2(fillItem.prevReadAt)) &&
		(fillItem.inputReadAt < input.Len() && !input.Column(f.inOrdinal).IsNilV2(fillItem.inputReadAt)) {
		inputValueIndex := input.Column(f.inOrdinal).GetValueIndexV2(fillItem.inputReadAt)
		prevValueIndex := prev.Column(f.inOrdinal).GetValueIndexV2(fillItem.prevReadAt)
		output.Column(f.outOrdinal).AppendFloatValue(
			hybridqp.LinearInterpolateFloat(
				fillItem.start,
				prev.TimeByIndex(fillItem.prevReadAt)/fillItem.interval,
				input.TimeByIndex(fillItem.inputReadAt)/fillItem.interval,
				prev.Column(f.inOrdinal).FloatValue(prevValueIndex),
				input.Column(f.inOrdinal).FloatValue(inputValueIndex),
			),
		)
		output.Column(f.outOrdinal).AppendNotNil()
	} else if (prev == nil || fillItem.inputReadAt < fillItem.prevReadAt) ||
		(fillItem.prevReadAt < prev.Len() && prev.Column(f.inOrdinal).IsNilV2(fillItem.prevReadAt)) ||
		(fillItem.inputReadAt == input.Len()-1 && input.Column(f.inOrdinal).IsNilV2(fillItem.inputReadAt)) ||
		fillItem.inputReadAt == input.Len() {
		output.Column(f.outOrdinal).AppendNil()
	}
}

func (f *FloatLinearFillProcessor) fillAppendFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	if !input.Column(f.inOrdinal).IsNilV2(fillItem.currIndex) {
		valueIndex := input.Column(f.inOrdinal).GetValueIndexV2(fillItem.currIndex)
		output.Column(f.outOrdinal).AppendFloatValue(input.Column(f.inOrdinal).FloatValue(valueIndex))
		if input.Column(f.inOrdinal).ColumnTimes() != nil {
			output.Column(f.outOrdinal).AppendColumnTime(input.Column(f.inOrdinal).ColumnTime(valueIndex))
		}
		output.Column(f.outOrdinal).AppendNotNil()
	} else {
		f.fillHelperFunc(input, output, prev, fillItem, prevWindow)
	}
}

type IntegerLinearFillProcessor struct {
	inOrdinal  int
	outOrdinal int
}

func NewIntegerLinearFillProcessor(inOrdinal, outOrdinal int) *IntegerLinearFillProcessor {
	return &IntegerLinearFillProcessor{
		inOrdinal:  inOrdinal,
		outOrdinal: outOrdinal,
	}
}

func (f *IntegerLinearFillProcessor) fillHelperFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	if (prev != nil && fillItem.inputReadAt >= fillItem.prevReadAt) &&
		(fillItem.prevReadAt < prev.Len() && !prev.Column(f.inOrdinal).IsNilV2(fillItem.prevReadAt)) &&
		(fillItem.inputReadAt < input.Len() && !input.Column(f.inOrdinal).IsNilV2(fillItem.inputReadAt)) {
		inputValueIndex := input.Column(f.inOrdinal).GetValueIndexV2(fillItem.inputReadAt)
		prevValueIndex := prev.Column(f.inOrdinal).GetValueIndexV2(fillItem.prevReadAt)
		output.Column(f.outOrdinal).AppendIntegerValue(
			hybridqp.LinearInterpolateInteger(
				fillItem.start,
				prev.TimeByIndex(fillItem.prevReadAt)/fillItem.interval,
				input.TimeByIndex(fillItem.inputReadAt)/fillItem.interval,
				prev.Column(f.inOrdinal).IntegerValue(prevValueIndex),
				input.Column(f.inOrdinal).IntegerValue(inputValueIndex),
			),
		)
		output.Column(f.outOrdinal).AppendNotNil()
	} else if (prev == nil || fillItem.inputReadAt < fillItem.prevReadAt) ||
		(fillItem.prevReadAt < prev.Len() && prev.Column(f.inOrdinal).IsNilV2(fillItem.prevReadAt)) ||
		(fillItem.inputReadAt == input.Len()-1 && input.Column(f.inOrdinal).IsNilV2(fillItem.inputReadAt)) ||
		fillItem.inputReadAt == input.Len() {
		output.Column(f.outOrdinal).AppendNil()
	}
}

func (f *IntegerLinearFillProcessor) fillAppendFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	if !input.Column(f.inOrdinal).IsNilV2(fillItem.currIndex) {
		valueIndex := input.Column(f.inOrdinal).GetValueIndexV2(fillItem.currIndex)
		output.Column(f.outOrdinal).AppendIntegerValue(input.Column(f.inOrdinal).IntegerValue(valueIndex))
		if input.Column(f.inOrdinal).ColumnTimes() != nil {
			output.Column(f.outOrdinal).AppendColumnTime(input.Column(f.inOrdinal).ColumnTime(valueIndex))
		}
		output.Column(f.outOrdinal).AppendNotNil()
	} else {
		f.fillHelperFunc(input, output, prev, fillItem, prevWindow)
	}
}

type StringLinearFillProcessor struct {
	inOrdinal  int
	outOrdinal int
}

func NewStringLinearFillProcessor(inOrdinal, outOrdinal int) *StringLinearFillProcessor {
	return &StringLinearFillProcessor{
		inOrdinal:  inOrdinal,
		outOrdinal: outOrdinal,
	}
}

func (f *StringLinearFillProcessor) fillHelperFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	output.Column(f.outOrdinal).AppendNil()
}

func (f *StringLinearFillProcessor) fillAppendFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	if !input.Column(f.inOrdinal).IsNilV2(fillItem.currIndex) {
		valueIndex := input.Column(f.inOrdinal).GetValueIndexV2(fillItem.currIndex)
		output.Column(f.outOrdinal).AppendStringValue(input.Column(f.inOrdinal).StringValue(valueIndex))
		if input.Column(f.inOrdinal).ColumnTimes() != nil {
			output.Column(f.outOrdinal).AppendColumnTime(input.Column(f.inOrdinal).ColumnTime(valueIndex))
		}
		output.Column(f.outOrdinal).AppendNotNil()
	} else {
		f.fillHelperFunc(input, output, prev, fillItem, prevWindow)
	}
}

type BooleanLinearFillProcessor struct {
	inOrdinal  int
	outOrdinal int
}

func NewBooleanLinearFillProcessor(inOrdinal, outOrdinal int) *BooleanLinearFillProcessor {
	return &BooleanLinearFillProcessor{
		inOrdinal:  inOrdinal,
		outOrdinal: outOrdinal,
	}
}

func (f *BooleanLinearFillProcessor) fillHelperFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	output.Column(f.outOrdinal).AppendNil()
}

func (f *BooleanLinearFillProcessor) fillAppendFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	if !input.Column(f.inOrdinal).IsNilV2(fillItem.currIndex) {
		valueIndex := input.Column(f.inOrdinal).GetValueIndexV2(fillItem.currIndex)
		output.Column(f.outOrdinal).AppendBooleanValue(input.Column(f.inOrdinal).BooleanValue(valueIndex))
		if input.Column(f.inOrdinal).ColumnTimes() != nil {
			output.Column(f.outOrdinal).AppendColumnTime(input.Column(f.inOrdinal).ColumnTime(valueIndex))
		}
		output.Column(f.outOrdinal).AppendNotNil()
	} else {
		f.fillHelperFunc(input, output, prev, fillItem, prevWindow)
	}
}

type FloatNullFillProcessor struct {
	inOrdinal  int
	outOrdinal int
}

func NewFloatNullFillProcessor(inOrdinal, outOrdinal int) *FloatNullFillProcessor {
	return &FloatNullFillProcessor{
		inOrdinal:  inOrdinal,
		outOrdinal: outOrdinal,
	}
}

func (f *FloatNullFillProcessor) fillHelperFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	// append nil value
	output.Column(f.outOrdinal).AppendNil()
}

func (f *FloatNullFillProcessor) fillAppendFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	if !input.Column(f.inOrdinal).IsNilV2(fillItem.currIndex) {
		valueIndex := input.Column(f.inOrdinal).GetValueIndexV2(fillItem.currIndex)
		output.Column(f.outOrdinal).AppendFloatValue(input.Column(f.inOrdinal).FloatValue(valueIndex))
		if input.Column(f.inOrdinal).ColumnTimes() != nil {
			output.Column(f.outOrdinal).AppendColumnTime(input.Column(f.inOrdinal).ColumnTime(valueIndex))
		}
		output.Column(f.outOrdinal).AppendNotNil()
	} else {
		f.fillHelperFunc(input, output, prev, fillItem, prevWindow)
	}
}

type IntegerNullFillProcessor struct {
	inOrdinal  int
	outOrdinal int
}

func NewIntegerNullFillProcessor(inOrdinal, outOrdinal int) *IntegerNullFillProcessor {
	return &IntegerNullFillProcessor{
		inOrdinal:  inOrdinal,
		outOrdinal: outOrdinal,
	}
}

func (f *IntegerNullFillProcessor) fillHelperFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	// append nil value
	output.Column(f.outOrdinal).AppendNil()
}

func (f *IntegerNullFillProcessor) fillAppendFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	if !input.Column(f.inOrdinal).IsNilV2(fillItem.currIndex) {
		valueIndex := input.Column(f.inOrdinal).GetValueIndexV2(fillItem.currIndex)
		output.Column(f.outOrdinal).AppendIntegerValue(input.Column(f.inOrdinal).IntegerValue(valueIndex))
		if input.Column(f.inOrdinal).ColumnTimes() != nil {
			output.Column(f.outOrdinal).AppendColumnTime(input.Column(f.inOrdinal).ColumnTime(valueIndex))
		}
		output.Column(f.outOrdinal).AppendNotNil()
	} else {
		f.fillHelperFunc(input, output, prev, fillItem, prevWindow)
	}
}

type StringNullFillProcessor struct {
	inOrdinal  int
	outOrdinal int
}

func NewStringNullFillProcessor(inOrdinal, outOrdinal int) *StringNullFillProcessor {
	return &StringNullFillProcessor{
		inOrdinal:  inOrdinal,
		outOrdinal: outOrdinal,
	}
}

func (f *StringNullFillProcessor) fillHelperFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	// append nil value
	output.Column(f.outOrdinal).AppendNil()
}

func (f *StringNullFillProcessor) fillAppendFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	if !input.Column(f.inOrdinal).IsNilV2(fillItem.currIndex) {
		valueIndex := input.Column(f.inOrdinal).GetValueIndexV2(fillItem.currIndex)
		output.Column(f.outOrdinal).AppendStringValue(input.Column(f.inOrdinal).StringValue(valueIndex))
		if input.Column(f.inOrdinal).ColumnTimes() != nil {
			output.Column(f.outOrdinal).AppendColumnTime(input.Column(f.inOrdinal).ColumnTime(valueIndex))
		}
		output.Column(f.outOrdinal).AppendNotNil()
	} else {
		f.fillHelperFunc(input, output, prev, fillItem, prevWindow)
	}
}

type BooleanNullFillProcessor struct {
	inOrdinal  int
	outOrdinal int
}

func NewBooleanNullFillProcessor(inOrdinal, outOrdinal int) *BooleanNullFillProcessor {
	return &BooleanNullFillProcessor{
		inOrdinal:  inOrdinal,
		outOrdinal: outOrdinal,
	}
}

func (f *BooleanNullFillProcessor) fillHelperFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	// append nil value
	output.Column(f.outOrdinal).AppendNil()
}

func (f *BooleanNullFillProcessor) fillAppendFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	if !input.Column(f.inOrdinal).IsNilV2(fillItem.currIndex) {
		valueIndex := input.Column(f.inOrdinal).GetValueIndexV2(fillItem.currIndex)
		output.Column(f.outOrdinal).AppendBooleanValue(input.Column(f.inOrdinal).BooleanValue(valueIndex))
		if input.Column(f.inOrdinal).ColumnTimes() != nil {
			output.Column(f.outOrdinal).AppendColumnTime(input.Column(f.inOrdinal).ColumnTime(valueIndex))
		}
		output.Column(f.outOrdinal).AppendNotNil()
	} else {
		f.fillHelperFunc(input, output, prev, fillItem, prevWindow)
	}
}

type FloatNumberFillProcessor struct {
	inOrdinal  int
	outOrdinal int
}

func NewFloatNumberFillProcessor(inOrdinal, outOrdinal int) *FloatNumberFillProcessor {
	return &FloatNumberFillProcessor{
		inOrdinal:  inOrdinal,
		outOrdinal: outOrdinal,
	}
}

func (f *FloatNumberFillProcessor) fillHelperFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	value, _ := hybridqp.TransToFloat(fillItem.fillValue)
	output.Column(f.outOrdinal).AppendFloatValue(value)
	output.Column(f.outOrdinal).AppendNotNil()
}

func (f *FloatNumberFillProcessor) fillAppendFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	if !input.Column(f.inOrdinal).IsNilV2(fillItem.currIndex) {
		valueIndex := input.Column(f.inOrdinal).GetValueIndexV2(fillItem.currIndex)
		output.Column(f.outOrdinal).AppendFloatValue(input.Column(f.inOrdinal).FloatValue(valueIndex))
		if input.Column(f.inOrdinal).ColumnTimes() != nil {
			output.Column(f.outOrdinal).AppendColumnTime(input.Column(f.inOrdinal).ColumnTime(valueIndex))
		}
		output.Column(f.outOrdinal).AppendNotNil()
	} else {
		f.fillHelperFunc(input, output, prev, fillItem, prevWindow)
	}
}

type IntegerNumberFillProcessor struct {
	inOrdinal  int
	outOrdinal int
}

func NewIntegerNumberFillProcessor(inOrdinal, outOrdinal int) *IntegerNumberFillProcessor {
	return &IntegerNumberFillProcessor{
		inOrdinal:  inOrdinal,
		outOrdinal: outOrdinal,
	}
}

func (f *IntegerNumberFillProcessor) fillHelperFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	value, _ := hybridqp.TransToInteger(fillItem.fillValue)
	output.Column(f.outOrdinal).AppendIntegerValue(value)
	output.Column(f.outOrdinal).AppendNotNil()
}

func (f *IntegerNumberFillProcessor) fillAppendFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	if !input.Column(f.inOrdinal).IsNilV2(fillItem.currIndex) {
		valueIndex := input.Column(f.inOrdinal).GetValueIndexV2(fillItem.currIndex)
		output.Column(f.outOrdinal).AppendIntegerValue(input.Column(f.inOrdinal).IntegerValue(valueIndex))
		if input.Column(f.inOrdinal).ColumnTimes() != nil {
			output.Column(f.outOrdinal).AppendColumnTime(input.Column(f.inOrdinal).ColumnTime(valueIndex))
		}
		output.Column(f.outOrdinal).AppendNotNil()
	} else {
		f.fillHelperFunc(input, output, prev, fillItem, prevWindow)
	}
}

type StringNumberFillProcessor struct {
	inOrdinal  int
	outOrdinal int
}

func NewStringNumberFillProcessor(inOrdinal, outOrdinal int) *StringNumberFillProcessor {
	return &StringNumberFillProcessor{
		inOrdinal:  inOrdinal,
		outOrdinal: outOrdinal,
	}
}

func (f *StringNumberFillProcessor) fillHelperFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	value, _ := hybridqp.TransToString(fillItem.fillValue)
	output.Column(f.outOrdinal).AppendStringValue(value)
	output.Column(f.outOrdinal).AppendNotNil()
}

func (f *StringNumberFillProcessor) fillAppendFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	if !input.Column(f.inOrdinal).IsNilV2(fillItem.currIndex) {
		valueIndex := input.Column(f.inOrdinal).GetValueIndexV2(fillItem.currIndex)
		output.Column(f.outOrdinal).AppendStringValue(input.Column(f.inOrdinal).StringValue(valueIndex))
		if input.Column(f.inOrdinal).ColumnTimes() != nil {
			output.Column(f.outOrdinal).AppendColumnTime(input.Column(f.inOrdinal).ColumnTime(valueIndex))
		}
		output.Column(f.outOrdinal).AppendNotNil()
	} else {
		f.fillHelperFunc(input, output, prev, fillItem, prevWindow)
	}
}

type BooleanNumberFillProcessor struct {
	inOrdinal  int
	outOrdinal int
}

func NewBooleanNumberFillProcessor(inOrdinal, outOrdinal int) *BooleanNumberFillProcessor {
	return &BooleanNumberFillProcessor{
		inOrdinal:  inOrdinal,
		outOrdinal: outOrdinal,
	}
}

func (f *BooleanNumberFillProcessor) fillHelperFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	value, _ := hybridqp.TransToBoolean(fillItem.fillValue)
	output.Column(f.outOrdinal).AppendBooleanValue(value)
	output.Column(f.outOrdinal).AppendNotNil()
}

func (f *BooleanNumberFillProcessor) fillAppendFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	if !input.Column(f.inOrdinal).IsNilV2(fillItem.currIndex) {
		valueIndex := input.Column(f.inOrdinal).GetValueIndexV2(fillItem.currIndex)
		output.Column(f.outOrdinal).AppendBooleanValue(input.Column(f.inOrdinal).BooleanValue(valueIndex))
		if input.Column(f.inOrdinal).ColumnTimes() != nil {
			output.Column(f.outOrdinal).AppendColumnTime(input.Column(f.inOrdinal).ColumnTime(valueIndex))
		}
		output.Column(f.outOrdinal).AppendNotNil()
	} else {
		f.fillHelperFunc(input, output, prev, fillItem, prevWindow)
	}
}

type FloatPreviousFillProcessor struct {
	inOrdinal  int
	outOrdinal int
}

func NewFloatPreviousFillProcessor(inOrdinal, outOrdinal int) *FloatPreviousFillProcessor {
	return &FloatPreviousFillProcessor{
		inOrdinal:  inOrdinal,
		outOrdinal: outOrdinal,
	}
}

func (f *FloatPreviousFillProcessor) fillHelperFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	if prev != nil && !prev.Column(f.inOrdinal).IsNilV2(fillItem.prevReadAt) {
		valueIndex := prev.Column(f.inOrdinal).GetValueIndexV2(fillItem.prevReadAt)
		output.Column(f.outOrdinal).AppendFloatValue(prev.Column(f.inOrdinal).FloatValue(valueIndex))
		output.Column(f.outOrdinal).AppendNotNil()
	} else if prevWindow.value != nil && !prevWindow.nil[f.inOrdinal] {
		output.Column(f.outOrdinal).AppendFloatValue(prevWindow.value[f.inOrdinal].(float64))
		output.Column(f.outOrdinal).AppendNotNil()
	} else {
		// append nil value
		output.Column(f.outOrdinal).AppendNil()
	}
}

func (f *FloatPreviousFillProcessor) fillAppendFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	if !input.Column(f.inOrdinal).IsNilV2(fillItem.currIndex) {
		valueIndex := input.Column(f.inOrdinal).GetValueIndexV2(fillItem.currIndex)
		output.Column(f.outOrdinal).AppendFloatValue(input.Column(f.inOrdinal).FloatValue(valueIndex))
		if input.Column(f.inOrdinal).ColumnTimes() != nil {
			output.Column(f.outOrdinal).AppendColumnTime(input.Column(f.inOrdinal).ColumnTime(valueIndex))
		}
		output.Column(f.outOrdinal).AppendNotNil()
	} else {
		f.fillHelperFunc(input, output, prev, fillItem, prevWindow)
	}
}

type IntegerPreviousFillProcessor struct {
	inOrdinal  int
	outOrdinal int
}

func NewIntegerPreviousFillProcessor(inOrdinal, outOrdinal int) *IntegerPreviousFillProcessor {
	return &IntegerPreviousFillProcessor{
		inOrdinal:  inOrdinal,
		outOrdinal: outOrdinal,
	}
}

func (f *IntegerPreviousFillProcessor) fillHelperFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	if prev != nil && !prev.Column(f.inOrdinal).IsNilV2(fillItem.prevReadAt) {
		valueIndex := prev.Column(f.inOrdinal).GetValueIndexV2(fillItem.prevReadAt)
		output.Column(f.outOrdinal).AppendIntegerValue(prev.Column(f.inOrdinal).IntegerValue(valueIndex))
		output.Column(f.outOrdinal).AppendNotNil()
	} else if prevWindow.value != nil && !prevWindow.nil[f.inOrdinal] {
		output.Column(f.outOrdinal).AppendIntegerValue(prevWindow.value[f.inOrdinal].(int64))
		output.Column(f.outOrdinal).AppendNotNil()
	} else {
		// append nil value
		output.Column(f.outOrdinal).AppendNil()
	}
}

func (f *IntegerPreviousFillProcessor) fillAppendFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	if !input.Column(f.inOrdinal).IsNilV2(fillItem.currIndex) {
		valueIndex := input.Column(f.inOrdinal).GetValueIndexV2(fillItem.currIndex)
		output.Column(f.outOrdinal).AppendIntegerValue(input.Column(f.inOrdinal).IntegerValue(valueIndex))
		if input.Column(f.inOrdinal).ColumnTimes() != nil {
			output.Column(f.outOrdinal).AppendColumnTime(input.Column(f.inOrdinal).ColumnTime(valueIndex))
		}
		output.Column(f.outOrdinal).AppendNotNil()
	} else {
		f.fillHelperFunc(input, output, prev, fillItem, prevWindow)
	}
}

type StringPreviousFillProcessor struct {
	inOrdinal  int
	outOrdinal int
}

func NewStringPreviousFillProcessor(inOrdinal, outOrdinal int) *StringPreviousFillProcessor {
	return &StringPreviousFillProcessor{
		inOrdinal:  inOrdinal,
		outOrdinal: outOrdinal,
	}
}

func (f *StringPreviousFillProcessor) fillHelperFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	if prev != nil && !prev.Column(f.inOrdinal).IsNilV2(fillItem.prevReadAt) {
		valueIndex := prev.Column(f.inOrdinal).GetValueIndexV2(fillItem.prevReadAt)
		output.Column(f.outOrdinal).AppendStringValue(prev.Column(f.inOrdinal).StringValue(valueIndex))
		output.Column(f.outOrdinal).AppendNotNil()
	} else if prevWindow.value != nil && !prevWindow.nil[f.inOrdinal] {
		output.Column(f.outOrdinal).AppendStringValue(prevWindow.value[f.inOrdinal].(string))
		output.Column(f.outOrdinal).AppendNotNil()
	} else {
		// append nil value
		output.Column(f.outOrdinal).AppendNil()
	}
}

func (f *StringPreviousFillProcessor) fillAppendFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	if !input.Column(f.inOrdinal).IsNilV2(fillItem.currIndex) {
		valueIndex := input.Column(f.inOrdinal).GetValueIndexV2(fillItem.currIndex)
		output.Column(f.outOrdinal).AppendStringValue(input.Column(f.inOrdinal).StringValue(valueIndex))
		if input.Column(f.inOrdinal).ColumnTimes() != nil {
			output.Column(f.outOrdinal).AppendColumnTime(input.Column(f.inOrdinal).ColumnTime(valueIndex))
		}
		output.Column(f.outOrdinal).AppendNotNil()
	} else {
		f.fillHelperFunc(input, output, prev, fillItem, prevWindow)
	}
}

type BooleanPreviousFillProcessor struct {
	inOrdinal  int
	outOrdinal int
}

func NewBooleanPreviousFillProcessor(inOrdinal, outOrdinal int) *BooleanPreviousFillProcessor {
	return &BooleanPreviousFillProcessor{
		inOrdinal:  inOrdinal,
		outOrdinal: outOrdinal,
	}
}

func (f *BooleanPreviousFillProcessor) fillHelperFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	if prev != nil && !prev.Column(f.inOrdinal).IsNilV2(fillItem.prevReadAt) {
		valueIndex := prev.Column(f.inOrdinal).GetValueIndexV2(fillItem.prevReadAt)
		output.Column(f.outOrdinal).AppendBooleanValue(prev.Column(f.inOrdinal).BooleanValue(valueIndex))
		output.Column(f.outOrdinal).AppendNotNil()
	} else if prevWindow.value != nil && !prevWindow.nil[f.inOrdinal] {
		output.Column(f.outOrdinal).AppendBooleanValue(prevWindow.value[f.inOrdinal].(bool))
		output.Column(f.outOrdinal).AppendNotNil()
	} else {
		// append nil value
		output.Column(f.outOrdinal).AppendNil()
	}
}

func (f *BooleanPreviousFillProcessor) fillAppendFunc(input, output, prev Chunk, fillItem *FillItem, prevWindow *prevWindow) {
	if !input.Column(f.inOrdinal).IsNilV2(fillItem.currIndex) {
		valueIndex := input.Column(f.inOrdinal).GetValueIndexV2(fillItem.currIndex)
		output.Column(f.outOrdinal).AppendBooleanValue(input.Column(f.inOrdinal).BooleanValue(valueIndex))
		if input.Column(f.inOrdinal).ColumnTimes() != nil {
			output.Column(f.outOrdinal).AppendColumnTime(input.Column(f.inOrdinal).ColumnTime(valueIndex))
		}
		output.Column(f.outOrdinal).AppendNotNil()
	} else {
		f.fillHelperFunc(input, output, prev, fillItem, prevWindow)
	}
}