import { isNil , isFunction , isString } from '@terascope/utils' ; import { validateFunctionArgs } from '../argument-validator' ; import { Column , validateAccepts , getFieldTypesFromFieldConfigAndChildConfig , ColumnOptions , mapVectorEach , mapVectorEachValue , dynamicMapVectorEach , dynamicMapVectorEachValue } from '../../column' ; import { DataFrame } from '../../data-frame' ; import { FieldTransformConfig , isFieldTransform , isFieldValidation , ProcessMode , FieldValidateConfig , DataTypeFieldAndChildren , FunctionDefinitionConfig , FunctionContext , DynamicFrameFunctionContext } from '../../function-configs/interfaces' ; import { Builder , copyVectorToBuilder } from '../../builder' ; import { WritableData } from '../../core' ; export interface DataFrameAdapterOptions < T extends Record < string , any > > { args ? : T | ( ( index : number ) => T ) ; field ? : string ; } export interface FrameAdapterFn { column ( input ) : Column < any > ; frame ( input ) : DataFrame < Record < string , unknown > > ; } function transformColumnData < T extends Record < string , any > > ( column , fnDef , options ) { const err = validateAccepts ( fnDef . accepts , getFieldTypesFromFieldConfigAndChildConfig ( column . vector . config , column . vector . childConfig ) ) ; if ( err ) throw err ; const columnOptions = { name : column . name , version : column . version , } ; const inputConfig = { field_config : column . config , child_config : column . vector . childConfig } ; let outputConfig ; const args = isFunction ( options . args ) ? options . args ( 0 ) : options . args as T ; if ( fnDef . output_type ) { outputConfig = fnDef . output_type ( inputConfig , args ) ; } else { outputConfig = inputConfig ; } const builder = Builder . make ( new WritableData ( column . vector . size ) , { childConfig : outputConfig . child_config , config : outputConfig . field_config , name : column . vector . name , } ) ; if ( isFunction ( options . args ) ) { const context = { args : options . args , parent : column , inputConfig , } ; if ( fnDef . process_mode === ProcessMode . NONE ) { return new Column ( copyVectorToBuilder ( column . vector , builder ) , columnOptions ) ; } if ( fnDef . process_mode === ProcessMode . FULL_VALUES ) { return new Column ( dynamicMapVectorEach ( column . vector , builder , dynamicTransformerFN ( fnDef , context ) ) , columnOptions ) ; } return new Column ( dynamicMapVectorEach ( column . vector , builder , dynamicTransformerFN ( fnDef , context ) ) , columnOptions ) ; } const context = { args , parent : column , inputConfig , outputConfig } ; const transformFn = fnDef . create ( context ) ; if ( fnDef . process_mode === ProcessMode . NONE ) { return new Column ( copyVectorToBuilder ( column . vector , builder ) , columnOptions ) ; } if ( fnDef . process_mode === ProcessMode . FULL_VALUES ) { return new Column ( mapVectorEach ( column . vector , builder , transformFn ) , columnOptions ) ; } return new Column ( mapVectorEachValue ( column . vector , builder , transformFn ) , columnOptions ) ; } function validatorTransformFN ( validatorFn ) { return function _validatorTransform ( value , index ) : unknown { if ( validatorFn ( value , index ) ) return value ; return null ; } ; } function dynamicValidatorFN < T > ( fnDef , context ) { return function _dynamicValidatorFN ( index ) { const newArgs = context . args ( index ) ; const args = validateFunctionArgs ( fnDef , newArgs ) ; const fullContext = { ... context , args } ; const validatorFn = fnDef . create ( fullContext ) ; return validatorTransformFN ( validatorFn ) ; } ; } function dynamicTransformerFN < T > ( fnDef , context ) { return function _dynamicTransformerFN ( index ) { const newArgs = context . args ( index ) ; const args = validateFunctionArgs ( fnDef , newArgs ) ; const fullContext = { ... context , args } ; return fnDef . create ( fullContext ) ; } ; } function validateColumnData < T extends Record < string , any > > ( column , fnDef , options ) { const err = validateAccepts ( fnDef . accepts , getFieldTypesFromFieldConfigAndChildConfig ( column . vector . config , column . vector . childConfig ) ) ; if ( err ) { return column . clearAll ( ) ; } const columnOptions = { name : column . name , version : column . version , } ; const inputConfig = { field_config : column . config , child_config : column . vector . childConfig , } ; const builder = Builder . make ( new WritableData ( column . vector . size ) , { childConfig : column . vector . childConfig , config : column . vector . config , name : column . vector . name , } ) ; if ( isFunction ( options . args ) ) { const context = { args : options . args , parent : column , inputConfig , } ; if ( fnDef . process_mode === ProcessMode . FULL_VALUES ) { return new Column ( dynamicMapVectorEach ( column . vector , builder , dynamicValidatorFN ( fnDef , context ) ) , columnOptions ) ; } return new Column ( dynamicMapVectorEachValue ( column . vector , builder , dynamicValidatorFN ( fnDef , context ) ) , columnOptions ) ; } const context = { args : options . args as T , parent : column , inputConfig } ; const validatorFn = fnDef . create ( context ) ; const validatorTransform = validatorTransformFN ( validatorFn ) ; if ( fnDef . process_mode === ProcessMode . FULL_VALUES ) { return new Column ( mapVectorEach ( column . vector , builder , validatorTransform ) , columnOptions ) ; } return new Column ( mapVectorEachValue ( column . vector , builder , validatorTransform ) , columnOptions ) ; } function validateColumn < T extends Record < string , any > > ( fnDef , options ) { return function _validateColumn ( column ) : Column < any > { return validateColumnData ( column , fnDef , options ) ; } ; } function transformColumn < T extends Record < string , any > > ( fnDef , options ) { return function _transformColumn ( column ) : Column { return transformColumnData ( column , fnDef , options ) ; } ; } function validateFrame < T extends Record < string , any > > ( fnDef , options ) { return function _validateFrame ( frame ) : DataFrame < Record < string , unknown > > { const { field } = options ; if ( isNil ( options . field ) || ! isString ( field ) ) throw new Error ( 'Must provide a field option when running a DataFrame' ) ; const col = frame . getColumnOrThrow ( field ) ; const validCol = validateColumnData ( col , fnDef , options ) ; return frame . assign ( [ validCol ] ) ; } ; } function transformFrame < T extends Record < string , any > > ( fnDef , options ) { return function _transformFrame ( frame ) : DataFrame < Record < string , unknown > > { const { field } = options ; if ( isNil ( field ) || ! isString ( field ) ) throw new Error ( 'Must provide a field option when running a DataFrame' ) ; const col = frame . getColumnOrThrow ( field ) ; const newCol = transformColumnData ( col , fnDef , options ) ; return frame . assign ( [ newCol ] ) ; } ; } export function dataFrameAdapter < T extends Record < string , any > = Record < string , unknown > > ( fnDef , options = { } ) { if ( isNil ( options . args ) ) options . args = { } as T ; if ( ! isFunction ( options . args ) ) { options . args = validateFunctionArgs ( fnDef , options . args ) ; } if ( isFieldValidation ( fnDef ) ) { return { column : validateColumn ( fnDef , options ) , frame : validateFrame ( fnDef , options ) } ; } if ( isFieldTransform ( fnDef ) ) { return { column : transformColumn ( fnDef , options ) , frame : transformFrame ( fnDef , options ) } ; } throw new Error ( `Function definition "${ fnDef . name }" (type: ${ fnDef . type }) is not supported` ) ; }