Class<?> procClass = null;
try {
procClass = Class.forName(className);
} catch (ClassNotFoundException e) {
String msg = "Cannot load class for procedure: " + className;
throw compiler.new VoltCompilerException(msg);
}
// get the short name of the class (no package)
String[] parts = className.split("\\.");
String shortName = parts[parts.length - 1];
// add an entry to the catalog
final Procedure procedure = db.getProcedures().add(shortName);
procedure.setId(compiler.getNextProcedureId());
for (String userName : procedureDescriptor.m_authUsers) {
final User user = db.getUsers().get(userName);
if (user == null) {
throw compiler.new VoltCompilerException("Procedure " + className + " has a user " + userName + " that does not exist");
}
final UserRef userRef = procedure.getAuthusers().add(userName);
userRef.setUser(user);
}
for (String groupName : procedureDescriptor.m_authGroups) {
final Group group = db.getGroups().get(groupName);
if (group == null) {
throw compiler.new VoltCompilerException("Procedure " + className + " has a group " + groupName + " that does not exist");
}
final GroupRef groupRef = procedure.getAuthgroups().add(groupName);
groupRef.setGroup(group);
}
procedure.setClassname(className);
// sysprocs don't use the procedure compiler
procedure.setSystemproc(false);
procedure.setHasjava(true);
// get the annotation
// first try to get one that has been passed from the compiler
ProcInfoData info = compiler.getProcInfoOverride(shortName);
// then check for the usual one in the class itself
// and create a ProcInfo.Data instance for it
if (info == null) {
info = new ProcInfoData();
ProcInfo annotationInfo = procClass.getAnnotation(ProcInfo.class);
if (annotationInfo != null) {
info.partitionInfo = annotationInfo.partitionInfo();
info.partitionParam = annotationInfo.partitionParam();
info.singlePartition = annotationInfo.singlePartition();
info.mapInputQuery = annotationInfo.mapInputQuery();
// info.mapEmitTable = annotationInfo.mapEmitTable();
info.reduceInputQuery = annotationInfo.reduceInputQuery();
// info.reduceEmitTable = annotationInfo.reduceEmitTable();
}
}
assert (info != null);
VoltProcedure procInstance = null;
try {
procInstance = (VoltProcedure) procClass.newInstance();
} catch (InstantiationException e1) {
e1.printStackTrace();
} catch (IllegalAccessException e1) {
e1.printStackTrace();
}
// MapReduce!
if (ClassUtil.getSuperClasses(procClass).contains(VoltMapReduceProcedure.class)) {
procedure.setMapreduce(true);
// The Map input query is required
// The Reduce input query is optional
if (info.mapInputQuery == null || info.mapInputQuery.isEmpty()) {
String msg = "Procedure: " + shortName + " must include a mapInputQuery";
throw compiler.new VoltCompilerException(msg);
}
Database catalog_db = CatalogUtil.getDatabase(procedure);
VoltMapReduceProcedure<?> mrInstance = (VoltMapReduceProcedure<?>) procInstance;
// Initialize the MapOutput table
// Create an invocation of the VoltMapProcedure so that we can grab
// the MapOutput's schema
VoltTable.ColumnInfo[] schema = mrInstance.getMapOutputSchema();
String tableMapOutput = "MAP_" + procedure.getName();
Table catalog_tbl = catalog_db.getTables().add(tableMapOutput);
assert (catalog_tbl != null);
for (int i = 0; i < schema.length; i++) {
Column catalog_col = catalog_tbl.getColumns().add(schema[i].getName());
catalog_col.setIndex(i);
catalog_col.setNullable(i > 0);
catalog_col.setType(schema[i].getType().getValue());
if (i == 0)
catalog_tbl.setPartitioncolumn(catalog_col);
} // FOR
catalog_tbl.setMapreduce(true);
catalog_tbl.setIsreplicated(false);
// Initialize the reduceOutput table
VoltTable.ColumnInfo[] schema_reduceOutput = mrInstance.getReduceOutputSchema();
String tableReduceOutput = "REDUCE_" + procedure.getName();
catalog_tbl = catalog_db.getTables().add(tableReduceOutput);
assert (catalog_tbl != null);
for (int i = 0; i < schema_reduceOutput.length; i++) {
Column catalog_col = catalog_tbl.getColumns().add(schema_reduceOutput[i].getName());
catalog_col.setIndex(i);
catalog_col.setNullable(i > 0);
catalog_col.setType(schema_reduceOutput[i].getType().getValue());
if (i == 0)
catalog_tbl.setPartitioncolumn(catalog_col);
} // FOR
catalog_tbl.setMapreduce(true);
catalog_tbl.setIsreplicated(false);
// Initialize the Procedure catalog object
procedure.setMapinputquery(info.mapInputQuery);
procedure.setMapemittable(tableMapOutput);
procedure.setReduceemittable(tableReduceOutput);
procedure.setReduceinputquery(info.reduceInputQuery);
}
// track if there are any writer stmts
boolean procHasWriteStmts = false;
// iterate through the fields and deal with
Field[] fields = procClass.getFields();
for (Field f : fields) {
if (f.getType() == SQLStmt.class) {
// String fieldName = f.getName();
SQLStmt stmt = null;
try {
stmt = (SQLStmt) f.get(procInstance);
} catch (IllegalArgumentException e) {
e.printStackTrace();
} catch (IllegalAccessException e) {
e.printStackTrace();
}
// add the statement to the catalog
Statement catalogStmt = procedure.getStatements().add(f.getName());
// compile the statement
try {
StatementCompiler.compile(compiler, hsql, catalog, db, estimates, catalogStmt, stmt.getText(), info.singlePartition);
} catch (VoltCompiler.VoltCompilerException e) {
e.printStackTrace();
String msg = shortName + "." + f.getName() + ": " + e.getMessage();
throw compiler.new VoltCompilerException(msg);
}
// If this Field has a Prefetchable annotation or the Statement was
// identified as prefetchable in the project XML, then we will want to
// set the "prefetchable" flag in the catalog for the Statement + Procedure
if (f.getAnnotation(Prefetchable.class) != null ||
procedureDescriptor.m_prefetchable.contains(catalogStmt.getName())) {
catalogStmt.setPrefetchable(true);
procedure.setPrefetchable(true);
}
// If this Field has a Deferrable annotation or the Statement was
// identified as deferrable in the project XML, then we will want to
// set the "deferrable" flag in the catalog for the Statement + Procedure
if (f.getAnnotation(Deferrable.class) != null) {
catalogStmt.setDeferrable(true);
procedure.setDeferrable(true);
}
// if a single stmt is not read only, then the proc is not read
// only
if (catalogStmt.getReadonly() == false)
procHasWriteStmts = true;
}
}
// set the read onlyness of a proc
procedure.setReadonly(procHasWriteStmts == false);
Class<?>[] paramTypes = populateProcedureParameters(compiler, procClass, procedure);
// parse the procinfo
procedure.setSinglepartition(info.singlePartition);
if (info.partitionInfo != null && info.partitionInfo.isEmpty() == false) {
parsePartitionInfo(compiler, db, procedure, info.partitionInfo);
if (procedure.getPartitionparameter() >= paramTypes.length) {
String msg = "PartitionInfo parameter not a valid parameter for procedure: " + procedure.getClassname();
throw compiler.new VoltCompilerException(msg);
}
// check the type of partition parameter meets our high standards
Class<?> partitionType = paramTypes[procedure.getPartitionparameter()];
Class<?>[] validPartitionClzzes = { Long.class, Integer.class, Short.class, Byte.class, long.class, int.class, short.class, byte.class, String.class };
boolean found = false;
for (Class<?> candidate : validPartitionClzzes) {
if (partitionType == candidate)
found = true;
}
// assume on of the two tests above passes and one fails
if (!found) {
String msg = "PartitionInfo parameter must be a String or Number for procedure: " + procedure.getClassname();
throw compiler.new VoltCompilerException(msg);
}
} else {
procedure.setPartitionparameter(NullProcParameter.PARAM_IDX);
}
// ProcInfo.partitionParam overrides everything else
if (info.partitionParam != -1) {
if (info.partitionParam >= paramTypes.length || info.partitionParam < 0) {
String msg = "PartitionInfo 'partitionParam' not a valid parameter for procedure: " + procedure.getClassname();
throw compiler.new VoltCompilerException(msg);
}
procedure.setPartitionparameter(info.partitionParam);
}
// put the compiled code for this procedure into the jarfile