exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/FunctionConverter.java - drill - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you 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.
  */
 package org.apache.drill.exec.expr.fn;

 import java.lang.annotation.Annotation;
 import java.lang.reflect.Field;
 import java.util.Arrays;
 import java.util.List;

 import javax.inject.Inject;

 import org.apache.drill.common.exceptions.DrillRuntimeException;
 import org.apache.drill.common.scanner.persistence.FieldDescriptor;
 import org.apache.drill.common.scanner.persistence.AnnotatedClassDescriptor;
 import org.apache.drill.common.types.TypeProtos.MajorType;
 import org.apache.drill.exec.expr.DrillFunc;
 import org.apache.drill.exec.expr.annotations.FunctionTemplate;
 import org.apache.drill.exec.expr.annotations.FunctionTemplate.FunctionScope;
 import org.apache.drill.exec.expr.annotations.Output;
 import org.apache.drill.exec.expr.annotations.Param;
 import org.apache.drill.exec.expr.annotations.Workspace;
 import org.apache.drill.exec.expr.holders.ValueHolder;
 import org.apache.drill.exec.ops.UdfUtilities;
 import org.apache.drill.exec.vector.complex.reader.FieldReader;
 import org.apache.drill.exec.vector.complex.writer.BaseWriter.ComplexWriter;

 import com.google.common.base.Joiner;
 import com.google.common.collect.Lists;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 /**
  * Converts FunctionCalls to Java Expressions.
  */
 public class FunctionConverter {
   private static final Logger logger = LoggerFactory.getLogger(FunctionConverter.class);

   public DrillFuncHolder getHolder(AnnotatedClassDescriptor func, ClassLoader classLoader) {
     FunctionTemplate template = func.getAnnotationProxy(FunctionTemplate.class);
     if (template == null) {
       return failure("Class does not declare FunctionTemplate annotation.", func);
     }

     String name = template.name();
     List<String> names = Arrays.asList(template.names());
     if (name.isEmpty() && names.isEmpty()) {
       // none set
       return failure("Must define 'name' or 'names'", func);
     }
     if (!name.isEmpty() && !names.isEmpty()) {
       // both are set
       return failure("Must use only one annotations 'name' or 'names', not both", func);
     }

     // start by getting field information.
     List<ValueReference> params = Lists.newArrayList();
     List<WorkspaceReference> workspaceFields = Lists.newArrayList();

     ValueReference outputField = null;
     int varArgsCount = 0;

     for (FieldDescriptor field : func.getFields()) {
       Param param = field.getAnnotationProxy(Param.class);
       Output output = field.getAnnotationProxy(Output.class);
       Workspace workspace = field.getAnnotationProxy(Workspace.class);
       Inject inject = field.getAnnotationProxy(Inject.class);

       Annotation[] annotations = {param, output, workspace, inject};
       int annotationCount = 0;
       for (Annotation annotationDescriptor : annotations) {
         if (annotationDescriptor != null) {
           annotationCount += 1;
         }
       }
       if (annotationCount == 0) {
         return failure("The field must be either a @Param, @Output, @Inject or @Workspace field.", func, field);
       } else if(annotationCount > 1) {
         return failure("The field must be only one of @Param, @Output, @Inject or @Workspace. It currently has more than one of these annotations.", func, field);
       }

       // TODO(Julien): verify there are a few of those and we can load them
       Class<?> fieldClass = field.getFieldClass();
       if (param != null || output != null) {
         if (Object[].class.isAssignableFrom(fieldClass)) {
           fieldClass = fieldClass.getComponentType();
           varArgsCount++;
         } else if (varArgsCount > 0 && param != null) {
           return failure("Vararg should be the last argument in the function.", func, field);
         }

         if (varArgsCount > 1) {
           return failure("Function should contain single vararg argument", func, field);
         }

         // Special processing for @Param FieldReader
         if (param != null && FieldReader.class.isAssignableFrom(fieldClass)) {
           ValueReference fieldReaderRef = ValueReference.createFieldReaderRef(field.getName());
           fieldReaderRef.setVarArg(varArgsCount > 0);
           params.add(fieldReaderRef);
           continue;
         }

         // Special processing for @Output ComplexWriter
         if (output != null && ComplexWriter.class.isAssignableFrom(fieldClass)) {
           if (outputField != null) {
             return failure("You've declared more than one @Output field.\n" +
                 "You must declare one and only @Output field per Function class.", func, field);
           } else {
             outputField = ValueReference.createComplexWriterRef(field.getName());
           }
           continue;
         }

         // check that param and output are value holders.
         if (!ValueHolder.class.isAssignableFrom(fieldClass)) {
           return failure(String.format("The field doesn't holds value of type %s which does not implement the ValueHolder or ComplexWriter interfaces.\n" +
               "All fields of type @Param or @Output must extend this interface.", fieldClass), func, field);
         }

         // get the type field from the value holder.
         MajorType type;
         try {
           type = getStaticFieldValue("TYPE", fieldClass, MajorType.class);
         } catch (Exception e) {
           return failure("Failure while trying to access the ValueHolder's TYPE static variable.  All ValueHolders must contain a static TYPE variable that defines their MajorType.", e, func, field);
         }

         ValueReference p = new ValueReference(type, field.getName());
         if (param != null) {
           p.setConstant(param.constant());
           p.setVarArg(varArgsCount > 0);
           params.add(p);
         } else {
           if (outputField != null) {
             return failure("You've declared more than one @Output field.  You must declare one and only @Output field per Function class.", func, field);
           } else {
             outputField = p;
           }
         }

       } else {
         // workspace work.
         boolean isInject = inject != null;
         if (isInject && UdfUtilities.INJECTABLE_GETTER_METHODS.get(fieldClass) == null) {
           return failure(
               String.format(
                   "A %s cannot be injected into a %s,"
                   + " available injectable classes are: %s.",
                   fieldClass, DrillFunc.class.getSimpleName(),
                   Joiner.on(",").join(UdfUtilities.INJECTABLE_GETTER_METHODS.keySet())
               ), func, field);
         }
         WorkspaceReference wsReference = new WorkspaceReference(fieldClass, field.getName(), isInject);

         if (!isInject && template.scope() == FunctionScope.POINT_AGGREGATE && !ValueHolder.class.isAssignableFrom(fieldClass) ) {
           return failure(String.format("Aggregate function '%s' workspace variable '%s' is of type '%s'. Please change it to Holder type.", func.getClassName(), field.getName(), fieldClass), func, field);
         }

         //If the workspace var is of Holder type, get its MajorType and assign to WorkspaceReference.
         if (ValueHolder.class.isAssignableFrom(fieldClass)) {
           MajorType majorType;
           try {
             majorType = getStaticFieldValue("TYPE", fieldClass, MajorType.class);
           } catch (Exception e) {
             return failure("Failure while trying to access the ValueHolder's TYPE static variable.  All ValueHolders must contain a static TYPE variable that defines their MajorType.", e, func, field);
           }
           wsReference.setMajorType(majorType);
         }
         workspaceFields.add(wsReference);
       }
     }

     if (outputField == null) {
       return failure("This function declares zero output fields.  A function must declare one output field.", func);
     }

     FunctionInitializer initializer = new FunctionInitializer(func.getClassName(), classLoader);
     try {
       // return holder
       ValueReference[] ps = params.toArray(new ValueReference[0]);
       WorkspaceReference[] works = workspaceFields.toArray(new WorkspaceReference[0]);

       FunctionAttributes functionAttributes = new FunctionAttributes(template, ps, outputField, works);

       switch (template.scope()) {
         case POINT_AGGREGATE:
           return outputField.isComplexWriter() ?
               new DrillComplexWriterAggFuncHolder(functionAttributes, initializer) :
               new DrillAggFuncHolder(functionAttributes, initializer);
         case SIMPLE:
           return outputField.isComplexWriter() ?
               new DrillComplexWriterFuncHolder(functionAttributes, initializer) :
               new DrillSimpleFuncHolder(functionAttributes, initializer);
         case HOLISTIC_AGGREGATE:
         case RANGE_AGGREGATE:
         default:
           return failure("Unsupported Function Type.", func);
       }
     } catch (Exception | NoSuchFieldError | AbstractMethodError ex) {
       return failure("Failure while creating function holder.", ex, func);
     }
   }

   @SuppressWarnings("unchecked")
   private <T> T getStaticFieldValue(String fieldName, Class<?> valueType, Class<T> c) throws NoSuchFieldException, SecurityException, IllegalArgumentException, IllegalAccessException{
       Field f = valueType.getDeclaredField(fieldName);
       Object val = f.get(null);
       return (T) val;
   }

   private static DrillFuncHolder failure(String message, Throwable t, AnnotatedClassDescriptor func, FieldDescriptor field) {
     return fieldFailure(message, t, func.getClassName(), field.getName());
   }

   private static DrillFuncHolder failure(String message, AnnotatedClassDescriptor func, FieldDescriptor field) {
     return fieldFailure(message, null, func.getClassName(), field.getName());
   }

   private DrillFuncHolder failure(String message, AnnotatedClassDescriptor func) {
     return classFailure(message, null, func.getClassName());
   }

   private DrillFuncHolder failure(String message, Throwable t, AnnotatedClassDescriptor func) {
     return classFailure(message, t, func.getClassName());
   }

   private static DrillFuncHolder classFailure(String message, Throwable t, String funcClassName) {
     return failure(String.format("Failure loading function class [%s]. Message: %s", funcClassName, message), t);
   }

   private static DrillFuncHolder fieldFailure(String message, Throwable t, String funcClassName, String fieldName) {
     return failure(String.format("Failure loading function class %s, field %s. Message: %s", funcClassName, fieldName, message), t);
   }

   private static DrillFuncHolder failure(String message, Throwable t) {
     if (t == null) {
       t = new DrillRuntimeException(message);
     }
     logger.warn(message, t);
     return null;
   }

 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you 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.
	*/
	package org.apache.drill.exec.expr.fn;

	import java.lang.annotation.Annotation;
	import java.lang.reflect.Field;
	import java.util.Arrays;
	import java.util.List;

	import javax.inject.Inject;

	import org.apache.drill.common.exceptions.DrillRuntimeException;
	import org.apache.drill.common.scanner.persistence.FieldDescriptor;
	import org.apache.drill.common.scanner.persistence.AnnotatedClassDescriptor;
	import org.apache.drill.common.types.TypeProtos.MajorType;
	import org.apache.drill.exec.expr.DrillFunc;
	import org.apache.drill.exec.expr.annotations.FunctionTemplate;
	import org.apache.drill.exec.expr.annotations.FunctionTemplate.FunctionScope;
	import org.apache.drill.exec.expr.annotations.Output;
	import org.apache.drill.exec.expr.annotations.Param;
	import org.apache.drill.exec.expr.annotations.Workspace;
	import org.apache.drill.exec.expr.holders.ValueHolder;
	import org.apache.drill.exec.ops.UdfUtilities;
	import org.apache.drill.exec.vector.complex.reader.FieldReader;
	import org.apache.drill.exec.vector.complex.writer.BaseWriter.ComplexWriter;

	import com.google.common.base.Joiner;
	import com.google.common.collect.Lists;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	/**
	* Converts FunctionCalls to Java Expressions.
	*/
	public class FunctionConverter {
	private static final Logger logger = LoggerFactory.getLogger(FunctionConverter.class);

	public DrillFuncHolder getHolder(AnnotatedClassDescriptor func, ClassLoader classLoader) {
	FunctionTemplate template = func.getAnnotationProxy(FunctionTemplate.class);
	if (template == null) {
	return failure("Class does not declare FunctionTemplate annotation.", func);
	}

	String name = template.name();
	List<String> names = Arrays.asList(template.names());
	if (name.isEmpty() && names.isEmpty()) {
	// none set
	return failure("Must define 'name' or 'names'", func);
	}
	if (!name.isEmpty() && !names.isEmpty()) {
	// both are set
	return failure("Must use only one annotations 'name' or 'names', not both", func);
	}

	// start by getting field information.
	List<ValueReference> params = Lists.newArrayList();
	List<WorkspaceReference> workspaceFields = Lists.newArrayList();

	ValueReference outputField = null;
	int varArgsCount = 0;

	for (FieldDescriptor field : func.getFields()) {
	Param param = field.getAnnotationProxy(Param.class);
	Output output = field.getAnnotationProxy(Output.class);
	Workspace workspace = field.getAnnotationProxy(Workspace.class);
	Inject inject = field.getAnnotationProxy(Inject.class);

	Annotation[] annotations = {param, output, workspace, inject};
	int annotationCount = 0;
	for (Annotation annotationDescriptor : annotations) {
	if (annotationDescriptor != null) {
	annotationCount += 1;
	}
	}
	if (annotationCount == 0) {
	return failure("The field must be either a @Param, @Output, @Inject or @Workspace field.", func, field);
	} else if(annotationCount > 1) {
	return failure("The field must be only one of @Param, @Output, @Inject or @Workspace. It currently has more than one of these annotations.", func, field);
	}

	// TODO(Julien): verify there are a few of those and we can load them
	Class<?> fieldClass = field.getFieldClass();
	if (param != null \|\| output != null) {
	if (Object[].class.isAssignableFrom(fieldClass)) {
	fieldClass = fieldClass.getComponentType();
	varArgsCount++;
	} else if (varArgsCount > 0 && param != null) {
	return failure("Vararg should be the last argument in the function.", func, field);
	}

	if (varArgsCount > 1) {
	return failure("Function should contain single vararg argument", func, field);
	}

	// Special processing for @Param FieldReader
	if (param != null && FieldReader.class.isAssignableFrom(fieldClass)) {
	ValueReference fieldReaderRef = ValueReference.createFieldReaderRef(field.getName());
	fieldReaderRef.setVarArg(varArgsCount > 0);
	params.add(fieldReaderRef);
	continue;
	}

	// Special processing for @Output ComplexWriter
	if (output != null && ComplexWriter.class.isAssignableFrom(fieldClass)) {
	if (outputField != null) {
	return failure("You've declared more than one @Output field.\n" +
	"You must declare one and only @Output field per Function class.", func, field);
	} else {
	outputField = ValueReference.createComplexWriterRef(field.getName());
	}
	continue;
	}

	// check that param and output are value holders.
	if (!ValueHolder.class.isAssignableFrom(fieldClass)) {
	return failure(String.format("The field doesn't holds value of type %s which does not implement the ValueHolder or ComplexWriter interfaces.\n" +
	"All fields of type @Param or @Output must extend this interface.", fieldClass), func, field);
	}

	// get the type field from the value holder.
	MajorType type;
	try {
	type = getStaticFieldValue("TYPE", fieldClass, MajorType.class);
	} catch (Exception e) {
	return failure("Failure while trying to access the ValueHolder's TYPE static variable. All ValueHolders must contain a static TYPE variable that defines their MajorType.", e, func, field);
	}

	ValueReference p = new ValueReference(type, field.getName());
	if (param != null) {
	p.setConstant(param.constant());
	p.setVarArg(varArgsCount > 0);
	params.add(p);
	} else {
	if (outputField != null) {
	return failure("You've declared more than one @Output field. You must declare one and only @Output field per Function class.", func, field);
	} else {
	outputField = p;
	}
	}

	} else {
	// workspace work.
	boolean isInject = inject != null;
	if (isInject && UdfUtilities.INJECTABLE_GETTER_METHODS.get(fieldClass) == null) {
	return failure(
	String.format(
	"A %s cannot be injected into a %s,"
	+ " available injectable classes are: %s.",
	fieldClass, DrillFunc.class.getSimpleName(),
	Joiner.on(",").join(UdfUtilities.INJECTABLE_GETTER_METHODS.keySet())
	), func, field);
	}
	WorkspaceReference wsReference = new WorkspaceReference(fieldClass, field.getName(), isInject);

	if (!isInject && template.scope() == FunctionScope.POINT_AGGREGATE && !ValueHolder.class.isAssignableFrom(fieldClass) ) {
	return failure(String.format("Aggregate function '%s' workspace variable '%s' is of type '%s'. Please change it to Holder type.", func.getClassName(), field.getName(), fieldClass), func, field);
	}

	//If the workspace var is of Holder type, get its MajorType and assign to WorkspaceReference.
	if (ValueHolder.class.isAssignableFrom(fieldClass)) {
	MajorType majorType;
	try {
	majorType = getStaticFieldValue("TYPE", fieldClass, MajorType.class);
	} catch (Exception e) {
	return failure("Failure while trying to access the ValueHolder's TYPE static variable. All ValueHolders must contain a static TYPE variable that defines their MajorType.", e, func, field);
	}
	wsReference.setMajorType(majorType);
	}
	workspaceFields.add(wsReference);
	}
	}

	if (outputField == null) {
	return failure("This function declares zero output fields. A function must declare one output field.", func);
	}

	FunctionInitializer initializer = new FunctionInitializer(func.getClassName(), classLoader);
	try {
	// return holder
	ValueReference[] ps = params.toArray(new ValueReference[0]);
	WorkspaceReference[] works = workspaceFields.toArray(new WorkspaceReference[0]);

	FunctionAttributes functionAttributes = new FunctionAttributes(template, ps, outputField, works);

	switch (template.scope()) {
	case POINT_AGGREGATE:
	return outputField.isComplexWriter() ?
	new DrillComplexWriterAggFuncHolder(functionAttributes, initializer) :
	new DrillAggFuncHolder(functionAttributes, initializer);
	case SIMPLE:
	return outputField.isComplexWriter() ?
	new DrillComplexWriterFuncHolder(functionAttributes, initializer) :
	new DrillSimpleFuncHolder(functionAttributes, initializer);
	case HOLISTIC_AGGREGATE:
	case RANGE_AGGREGATE:
	default:
	return failure("Unsupported Function Type.", func);
	}
	} catch (Exception \| NoSuchFieldError \| AbstractMethodError ex) {
	return failure("Failure while creating function holder.", ex, func);
	}
	}

	@SuppressWarnings("unchecked")
	private <T> T getStaticFieldValue(String fieldName, Class<?> valueType, Class<T> c) throws NoSuchFieldException, SecurityException, IllegalArgumentException, IllegalAccessException{
	Field f = valueType.getDeclaredField(fieldName);
	Object val = f.get(null);
	return (T) val;
	}

	private static DrillFuncHolder failure(String message, Throwable t, AnnotatedClassDescriptor func, FieldDescriptor field) {
	return fieldFailure(message, t, func.getClassName(), field.getName());
	}

	private static DrillFuncHolder failure(String message, AnnotatedClassDescriptor func, FieldDescriptor field) {
	return fieldFailure(message, null, func.getClassName(), field.getName());
	}

	private DrillFuncHolder failure(String message, AnnotatedClassDescriptor func) {
	return classFailure(message, null, func.getClassName());
	}

	private DrillFuncHolder failure(String message, Throwable t, AnnotatedClassDescriptor func) {
	return classFailure(message, t, func.getClassName());
	}

	private static DrillFuncHolder classFailure(String message, Throwable t, String funcClassName) {
	return failure(String.format("Failure loading function class [%s]. Message: %s", funcClassName, message), t);
	}

	private static DrillFuncHolder fieldFailure(String message, Throwable t, String funcClassName, String fieldName) {
	return failure(String.format("Failure loading function class %s, field %s. Message: %s", funcClassName, fieldName, message), t);
	}

	private static DrillFuncHolder failure(String message, Throwable t) {
	if (t == null) {
	t = new DrillRuntimeException(message);
	}
	logger.warn(message, t);
	return null;
	}

	}