added config changes

common/core/src/test/java/zingg/common/core/executor/ExecutorTester.java

@@ -1,18 +1,13 @@
 package zingg.common.core.executor;
 
 import java.io.IOException;
-import java.rmi.NoSuchObjectException;
-import java.util.Objects;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
-import zingg.common.client.arguments.ArgumentServiceImpl;
 import zingg.common.client.arguments.model.IArguments;
-import zingg.common.client.arguments.model.Arguments;
 import zingg.common.client.ClientOptions;
 import zingg.common.client.ZinggClientException;
-import zingg.common.client.pipe.Pipe;
 import zingg.common.client.util.DFObjectUtil;
 import zingg.common.core.executor.validate.ExecutorValidator;
 
@@ -23,39 +18,44 @@ public class ExecutorTester<S, D, R, C, T>{
 	public ZinggBase<S, D, R, C, T> executor;
 	public ExecutorValidator<S, D, R, C, T> validator;
 	protected IArguments args;
-	protected String configFile;
 	protected String modelId;
 	protected DFObjectUtil<S,D,R,C> dfObjectUtil;
 
-	public ExecutorTester(ZinggBase<S, D, R, C, T> executor,ExecutorValidator<S, D, R, C, T> validator, String configFile, String modelId, DFObjectUtil<S,D,R,C> dfObjectUtil) throws ZinggClientException, IOException {
+	/**
+	 * Args are built by the test (see TestArgumentsBuilder) and handed over ready to use;
+	 * no config file is read.
+	 */
+	public ExecutorTester(ZinggBase<S, D, R, C, T> executor,ExecutorValidator<S, D, R, C, T> validator, IArguments args, String modelId, DFObjectUtil<S,D,R,C> dfObjectUtil) throws ZinggClientException, IOException {
 		this.executor = executor;
 		this.validator = validator;
-		this.configFile = configFile;
+		this.args = args;
 		this.modelId = modelId;
 		this.dfObjectUtil = dfObjectUtil;
-		setupArgs();
 	}
 
-	public IArguments setupArgs(String configFile, String phase) throws ZinggClientException, NoSuchObjectException {
-		args = new ArgumentServiceImpl<Arguments>(Arguments.class).loadArguments(Objects.requireNonNull(getClass().getClassLoader().getResource(configFile)).getFile());
-		args = updateLocation(args);
-		args.setModelId(modelId);
-		return args;
-	}
-
-	public IArguments updateLocation(IArguments args){
-		for (Pipe p: args.getData()) {
-			if (p.getProps().containsKey("path")) {
-				String testOneFile = getClass().getClassLoader().getResource(p.get("path")).getFile();
-				// correct the location of test data
-				p.setProp("path", testOneFile);
-			}
+	/**
+	 * Gets args ready for the run: stamps the model id, then lets the tester adjust them
+	 * through updateArgs().
+	 *
+	 * The driver calls this before every execution (see TestExecutorsGeneric), so it is
+	 * deliberately not called from the constructor - a constructor call would run before
+	 * subclass fields are assigned.
+	 */
+	public void setupArgs() throws ZinggClientException, IOException{
+		// every run gets its own model id
+		this.args.setModelId(modelId);
+		try {
+			updateArgs(this.args);
+		} catch (Exception e) {
+			throw new ZinggClientException("Error while preparing args for the test run: ", e);
 		}
-		return args;
 	}
 
-	public void setupArgs() throws ZinggClientException, IOException{
-		this.args = setupArgs(configFile, "");
+	/**
+	 * Hook for testers that need to adjust args before the run, e.g. attach an in-memory
+	 * dataset to a pipe. Does nothing by default.
+	 */
+	protected void updateArgs(IArguments args) throws Exception {
 	}
 
 	public void initAndExecute(S session) throws ZinggClientException {

common/core/src/test/java/zingg/common/core/executor/FindAndLabellerExecutorTester.java

@@ -3,15 +3,16 @@
 import zingg.common.client.ClientOptions;
 import zingg.common.client.ZFrame;
 import zingg.common.client.ZinggClientException;
+import zingg.common.client.arguments.model.IArguments;
 import zingg.common.client.util.DFObjectUtil;
 import zingg.common.core.executor.validate.ExecutorValidator;
 
 import java.io.IOException;
 
 public class FindAndLabellerExecutorTester<S, D, R, C, T> extends MatchThresholdBasedExecutorTester<S, D, R, C, T> {
 
-    public FindAndLabellerExecutorTester(ZinggBase<S, D, R, C, T> ftdLabelerExecutor, ExecutorValidator<S, D, R, C, T> ftdLabelerValidator, String configFile, String modelId, DFObjectUtil<S, D, R, C> dfObjectUtil) throws ZinggClientException, IOException{
-        super(ftdLabelerExecutor, ftdLabelerValidator, configFile, modelId, dfObjectUtil);
+    public FindAndLabellerExecutorTester(ZinggBase<S, D, R, C, T> ftdLabelerExecutor, ExecutorValidator<S, D, R, C, T> ftdLabelerValidator, IArguments args, String modelId, DFObjectUtil<S, D, R, C> dfObjectUtil) throws ZinggClientException, IOException{
+        super(ftdLabelerExecutor, ftdLabelerValidator, args, modelId, dfObjectUtil);
     }
 
     //need to execute until we get

common/core/src/test/java/zingg/common/core/executor/FtdAndLabelCombinedExecutorTester.java

@@ -3,6 +3,7 @@
 import zingg.common.client.ClientOptions;
 import zingg.common.client.ZFrame;
 import zingg.common.client.ZinggClientException;
+import zingg.common.client.arguments.model.IArguments;
 import zingg.common.client.util.DFObjectUtil;
 import zingg.common.core.executor.validate.ExecutorValidator;
 
@@ -15,9 +16,9 @@ public class FtdAndLabelCombinedExecutorTester<S, D, R, C, T> extends MatchThres
 
     //setting labeller properties here
     //ftd properties are already set by super
-    public FtdAndLabelCombinedExecutorTester(ZinggBase<S, D, R, C, T> ftdExecutor, ExecutorValidator<S, D, R, C, T> ftdValidator, String configFile,
+    public FtdAndLabelCombinedExecutorTester(ZinggBase<S, D, R, C, T> ftdExecutor, ExecutorValidator<S, D, R, C, T> ftdValidator, IArguments args,
                                              ZinggBase<S, D, R, C, T> labelExecutor, ExecutorValidator<S, D, R, C, T> labelValidator, String modelId, DFObjectUtil<S,D,R,C> dfObjectUtil) throws ZinggClientException, IOException, NoSuchMethodException {
-        super(ftdExecutor, ftdValidator,configFile,modelId,dfObjectUtil);
+        super(ftdExecutor, ftdValidator,args,modelId,dfObjectUtil);
         this.labelExecutor = labelExecutor;
         this.labelValidator = labelValidator;
     }

common/core/src/test/java/zingg/common/core/executor/MatchThresholdBasedExecutorTester.java

@@ -4,6 +4,7 @@
 import org.apache.commons.logging.LogFactory;
 import zingg.common.client.ZFrame;
 import zingg.common.client.ZinggClientException;
+import zingg.common.client.arguments.model.IArguments;
 import zingg.common.client.util.DFObjectUtil;
 import zingg.common.core.ZinggException;
 import zingg.common.core.executor.validate.ExecutorValidator;
@@ -21,9 +22,9 @@ public abstract class MatchThresholdBasedExecutorTester<S, D, R, C, T> extends E
     protected long matchCount = 0;
     protected long notAMatchCount = 0;
 
-    public MatchThresholdBasedExecutorTester(ZinggBase<S, D, R, C, T> executor, ExecutorValidator<S, D, R, C, T> validator, String configFile, String modelId, DFObjectUtil<S, D, R, C> dfObjectUtil)
+    public MatchThresholdBasedExecutorTester(ZinggBase<S, D, R, C, T> executor, ExecutorValidator<S, D, R, C, T> validator, IArguments args, String modelId, DFObjectUtil<S, D, R, C> dfObjectUtil)
             throws ZinggClientException, IOException {
-        super(executor, validator, configFile, modelId, dfObjectUtil);
+        super(executor, validator, args, modelId, dfObjectUtil);
     }
 
     protected void runUntilThreshold() {

common/core/src/test/java/zingg/common/core/executor/TestExecutorsCompound.java

@@ -8,6 +8,7 @@
 import org.junit.jupiter.api.AfterEach;
 
 import zingg.common.client.ZinggClientException;
+import zingg.common.client.arguments.model.IArguments;
 import zingg.common.core.executor.validate.FindAndLabelValidator;
 import zingg.common.core.executor.validate.TrainMatchValidator;
 import zingg.common.core.util.IPerformCleanUpUtil;
@@ -24,14 +25,15 @@ public TestExecutorsCompound() {
 	public List<ExecutorTester<S, D, R, C, T>> getExecutors() throws ZinggClientException, IOException, NoSuchMethodException {
 		FindAndLabeller<S, D, R, C, T> findAndLabel = getFindAndLabeller();
 		FindAndLabelValidator<S, D, R, C, T> falValidator = new FindAndLabelValidator<S, D, R, C, T>(findAndLabel);
-		ExecutorTester<S, D, R, C, T> et = new FindAndLabellerExecutorTester<>(findAndLabel, falValidator,getConfigFile(),getModelId(),getDFObjectUtil());
+		ExecutorTester<S, D, R, C, T> et = new FindAndLabellerExecutorTester<>(findAndLabel, falValidator,getArgs(),getModelId(),getDFObjectUtil());
 		executorTesterList.add(et);
 		TrainMatcher<S, D, R, C, T> trainMatch = getTrainMatcher();
-		executorTesterList.add(new ExecutorTester<S, D, R, C, T>(trainMatch,getTrainMatchValidator(trainMatch), getConfigFile(),getModelId(),getDFObjectUtil()));
+		executorTesterList.add(new ExecutorTester<S, D, R, C, T>(trainMatch,getTrainMatchValidator(trainMatch), getArgs(),getModelId(),getDFObjectUtil()));
 		return executorTesterList;
 	}
 
-	public abstract String getConfigFile();
+	/** The args both compound phases run on; built in code, not read from a file. */
+	public abstract IArguments getArgs() throws ZinggClientException;
 
 	protected abstract FindAndLabeller<S, D, R, C, T> getFindAndLabeller() throws ZinggClientException;
 

common/core/src/test/java/zingg/common/core/executor/TestExecutorsSingle.java

@@ -8,6 +8,7 @@
 import org.junit.jupiter.api.AfterEach;
 
 import zingg.common.client.ZinggClientException;
+import zingg.common.client.arguments.model.IArguments;
 import zingg.common.core.executor.validate.LabellerValidator;
 import zingg.common.core.executor.validate.LinkerValidator;
 import zingg.common.core.executor.validate.MatcherValidator;
@@ -38,28 +39,30 @@ public void getBaseExecutors() throws ZinggClientException, IOException, NoSuchM
 
 		TrainingDataFinder<S, D, R, C, T> tdf = getTrainingDataFinder();
     	Labeller<S, D, R, C, T> labeler = getLabeller();
-		executorTesterList.add(new FtdAndLabelCombinedExecutorTester<S, D, R, C, T>(tdf, new TrainingDataFinderValidator<S, D, R, C, T>(tdf), getConfigFile(),
+		executorTesterList.add(new FtdAndLabelCombinedExecutorTester<S, D, R, C, T>(tdf, new TrainingDataFinderValidator<S, D, R, C, T>(tdf), getArgs(),
 				labeler, new LabellerValidator<S, D, R, C, T>(labeler), getModelId(), getDFObjectUtil()));
 
 
 		Trainer<S, D, R, C, T> trainer = getTrainer();
-		executorTesterList.add(new ExecutorTester<S, D, R, C, T>(trainer,getTrainerValidator(trainer),getConfigFile(),getModelId(),getDFObjectUtil()));
+		executorTesterList.add(new ExecutorTester<S, D, R, C, T>(trainer,getTrainerValidator(trainer),getArgs(),getModelId(),getDFObjectUtil()));
 
 	}
 
 	public void getAdditionalExecutors() throws ZinggClientException, IOException {
 
 		Matcher<S, D, R, C, T> matcher = getMatcher();
-		executorTesterList.add(new ExecutorTester<S, D, R, C, T>(matcher,new MatcherValidator<S, D, R, C, T>(matcher),getConfigFile(),getModelId(),getDFObjectUtil()));
+		executorTesterList.add(new ExecutorTester<S, D, R, C, T>(matcher,new MatcherValidator<S, D, R, C, T>(matcher),getArgs(),getModelId(),getDFObjectUtil()));
 
 		Linker<S, D, R, C, T> linker = getLinker();
-		executorTesterList.add(new ExecutorTester<S, D, R, C, T>(linker,new LinkerValidator<S, D, R, C, T>(linker),getLinkerConfigFile(),getModelId(),getDFObjectUtil()));
+		executorTesterList.add(new ExecutorTester<S, D, R, C, T>(linker,new LinkerValidator<S, D, R, C, T>(linker),getLinkerArgs(),getModelId(),getDFObjectUtil()));
 
 	}
 
-	public abstract String getConfigFile();
+	/** The args the single phase executors run on; built in code, not read from a file. */
+	public abstract IArguments getArgs() throws ZinggClientException;
 
-	public abstract String getLinkerConfigFile();
+	/** Linking needs its own args - two datasets and no training samples. */
+	public abstract IArguments getLinkerArgs() throws ZinggClientException;
 
     protected abstract TrainingDataFinder<S, D, R, C, T> getTrainingDataFinder() throws ZinggClientException;
 

common/core/src/test/java/zingg/common/core/executor/testData/TestArgumentsBuilder.java

@@ -0,0 +1,174 @@
+package zingg.common.core.executor.testData;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import zingg.common.client.FieldDefinition;
+import zingg.common.client.IMatchType;
+import zingg.common.client.MatchTypes;
+import zingg.common.client.ZinggClientException;
+import zingg.common.client.arguments.model.Arguments;
+import zingg.common.client.arguments.model.IArguments;
+import zingg.common.client.pipe.FilePipe;
+import zingg.common.client.pipe.Pipe;
+
+/**
+ * Builds the args the executor integration tests run on.
+ *
+ * These used to live in configSparkIntTest.json and configSparkLinkTest.json under
+ * src/test/resources. Keeping them in code means the field definitions and schemas
+ * are checked by the compiler, and a test can tweak one value without a second copy
+ * of the whole config.
+ *
+ * The paths are passed in by the caller because the test data sits on the classpath -
+ * only the driver knows where the class loader resolved it to.
+ */
+public class TestArgumentsBuilder {
+
+	public static final String DELIMITER = ",";
+	public static final String BAD_RECORDS_PATH = "/tmp/bad";
+	public static final float LABEL_DATA_SAMPLE_SIZE = 0.5f;
+	public static final int NUM_PARTITIONS = 4;
+
+	/**
+	 * Schema of test.csv - the record id followed by the ten compared fields.
+	 */
+	public static final String DATA_SCHEMA = "id string, fname string, lname string, stNo string, add1 string, add2 string, city string, state string, areacode string, dob string, ssn  string";
+
+	/**
+	 * Schema of training.csv - the two labelling columns followed by the record.
+	 * areacode comes before state here, the other way round from DATA_SCHEMA, because
+	 * that is the order the file itself is written in.
+	 */
+	public static final String TRAINING_SCHEMA = "z_cluster string, z_ismatch integer, id string, fname string, lname string, stNo string, add1 string, add2 string, city string, areacode string, state string, dob string, ssn  string";
+
+	/**
+	 * Schema of test1.csv and test2.csv - like DATA_SCHEMA but with areacode before state.
+	 */
+	public static final String LINK_DATA_SCHEMA = "id string, fname string, lname string, stNo string, add1 string, add2 string, city string, areacode string, state string, dob string, ssn  string";
+
+	protected TestArgumentsBuilder() {
+	}
+
+	/**
+	 * Args for the single and compound phases - one dataset, plus the training samples
+	 * the labeller seeds itself from.
+	 */
+	public static IArguments buildSingleArgs(String modelId, String zinggDir, String dataPath,
+			String trainingPath, String outputPath, String stopWordsPath) throws ZinggClientException {
+		IArguments args = buildCommonArgs(modelId, zinggDir, outputPath);
+		args.setData(new Pipe[] { csvPipe("test", dataPath, DATA_SCHEMA) });
+
+		Pipe trainingPipe = csvPipe("trainingPos", trainingPath, TRAINING_SCHEMA);
+		trainingPipe.setProp("badRecordsPath", BAD_RECORDS_PATH);
+		args.setTrainingSamples(new Pipe[] { trainingPipe });
+
+		args.setFieldDefinition(getMatchFieldDefinition(stopWordsPath));
+		return args;
+	}
+
+	/**
+	 * Args for the link phase - two datasets to link across, and no training samples,
+	 * since linking runs off the model the earlier phases trained.
+	 */
+	public static IArguments buildLinkArgs(String modelId, String zinggDir, String data1Path,
+			String data2Path, String outputPath) throws ZinggClientException {
+		IArguments args = buildCommonArgs(modelId, zinggDir, outputPath);
+		args.setData(new Pipe[] {
+				csvPipe("test1", data1Path, LINK_DATA_SCHEMA),
+				csvPipe("test2", data2Path, LINK_DATA_SCHEMA)
+		});
+		args.setFieldDefinition(getLinkFieldDefinition());
+		return args;
+	}
+
+	protected static IArguments buildCommonArgs(String modelId, String zinggDir, String outputPath)
+			throws ZinggClientException {
+		IArguments args = new Arguments();
+		args.setModelId(modelId);
+		args.setZinggDir(zinggDir);
+		args.setNumPartitions(NUM_PARTITIONS);
+		args.setLabelDataSampleSize(LABEL_DATA_SAMPLE_SIZE);
+
+		Pipe outputPipe = new Pipe();
+		outputPipe.setName("output");
+		outputPipe.setFormat(Pipe.FORMAT_CSV);
+		outputPipe.setProp(FilePipe.PATH, outputPath);
+		outputPipe.setProp(FilePipe.DELIMITER, DELIMITER);
+		outputPipe.setProp(FilePipe.HEADER, "true");
+		args.setOutput(new Pipe[] { outputPipe });
+
+		return args;
+	}
+
+	/**
+	 * id is carried through the output but not compared; everything else is fuzzy.
+	 * add1 gets the stop word list so the stop word removal path is exercised too.
+	 */
+	protected static List<FieldDefinition> getMatchFieldDefinition(String stopWordsPath) {
+		List<FieldDefinition> fieldDefinition = new ArrayList<FieldDefinition>();
+		fieldDefinition.add(dontUseField("id"));
+		fieldDefinition.add(fuzzyField("fname"));
+		fieldDefinition.add(fuzzyField("lname"));
+		fieldDefinition.add(fuzzyField("stNo"));
+
+		FieldDefinition add1 = fuzzyField("add1");
+		add1.setStopWords(stopWordsPath);
+		fieldDefinition.add(add1);
+
+		fieldDefinition.add(fuzzyField("add2"));
+		fieldDefinition.add(fuzzyField("city"));
+		fieldDefinition.add(fuzzyField("areacode"));
+		fieldDefinition.add(fuzzyField("state"));
+		fieldDefinition.add(fuzzyField("dob"));
+		fieldDefinition.add(fuzzyField("ssn"));
+		return fieldDefinition;
+	}
+
+	/**
+	 * Linking compares the ten fields and leaves id out of the definition altogether.
+	 */
+	protected static List<FieldDefinition> getLinkFieldDefinition() {
+		List<FieldDefinition> fieldDefinition = new ArrayList<FieldDefinition>();
+		fieldDefinition.add(fuzzyField("fname"));
+		fieldDefinition.add(fuzzyField("lname"));
+		fieldDefinition.add(fuzzyField("stNo"));
+		fieldDefinition.add(fuzzyField("add1"));
+		fieldDefinition.add(fuzzyField("add2"));
+		fieldDefinition.add(fuzzyField("city"));
+		fieldDefinition.add(fuzzyField("areacode"));
+		fieldDefinition.add(fuzzyField("state"));
+		fieldDefinition.add(fuzzyField("dob"));
+		fieldDefinition.add(fuzzyField("ssn"));
+		return fieldDefinition;
+	}
+
+	protected static Pipe csvPipe(String name, String path, String schema) {
+		Pipe pipe = new Pipe();
+		pipe.setName(name);
+		pipe.setFormat(Pipe.FORMAT_CSV);
+		pipe.setProp(FilePipe.PATH, path);
+		pipe.setProp(FilePipe.DELIMITER, DELIMITER);
+		pipe.setProp(FilePipe.HEADER, "false");
+		pipe.setSchema(schema);
+		return pipe;
+	}
+
+	protected static FieldDefinition fuzzyField(String name) {
+		return field(name, MatchTypes.FUZZY);
+	}
+
+	protected static FieldDefinition dontUseField(String name) {
+		return field(name, MatchTypes.DONT_USE);
+	}
+
+	protected static FieldDefinition field(String name, IMatchType matchType) {
+		FieldDefinition fieldDefinition = new FieldDefinition();
+		fieldDefinition.setFieldName(name);
+		fieldDefinition.setFields(name);
+		fieldDefinition.setDataType("string");
+		fieldDefinition.setMatchTypeInternal(matchType);
+		return fieldDefinition;
+	}
+
+}