integration Junit for OSS

common/client/src/main/java/zingg/common/client/ZFrame.java

@@ -170,5 +170,10 @@ public interface ZFrame<D, R, C> {
 
     public ZFrame<D,R,C> groupByCount(String groupByCol1, String groupByCol2, String countColName);
 
-
+    public ZFrame<D,R,C> intersect(ZFrame<D,R,C> other);
+
+	public C substr(C col, int startPos, int len);
+
+	public C gt(C column1, C column2);
+
 }

common/common-test/pom.xml

@@ -0,0 +1,57 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<parent>
+		<groupId>zingg</groupId>
+		<artifactId>zingg-common</artifactId>
+		<version>${zingg.version}</version>
+	</parent>
+	<artifactId>zingg-common-test</artifactId>
+	<packaging>jar</packaging>
+	<dependencies>
+		<dependency>
+			<groupId>zingg</groupId>
+			<artifactId>zingg-common-core</artifactId>
+			<version>${zingg.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>zingg</groupId>
+			<artifactId>zingg-common-client</artifactId>
+			<version>${zingg.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>org.junit.jupiter</groupId>
+			<artifactId>junit-jupiter-engine</artifactId>
+			<version>5.8.1</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.junit.jupiter</groupId>
+			<artifactId>junit-jupiter-api</artifactId>
+			<version>5.8.1</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.junit.jupiter</groupId>
+			<artifactId>junit-jupiter-params</artifactId>
+			<version>5.8.1</version>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-jar-plugin</artifactId>
+				<version>2.3.2</version>
+				<executions>
+					<execution>
+						<goals>
+							<goal>test-jar</goal>
+						</goals>
+					</execution>
+				</executions>
+			</plugin>
+		</plugins>
+	</build>			
+</project>

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

@@ -0,0 +1,24 @@
+package zingg.common.core.executor;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import zingg.common.client.ZinggClientException;
+
+public abstract class ExecutorTester<S, D, R, C, T> {
+
+	public static final Log LOG = LogFactory.getLog(ExecutorTester.class);
+
+	public ZinggBase<S,D, R, C, T> executor;
+
+	public ExecutorTester(ZinggBase<S, D, R, C, T> executor) {
+		this.executor = executor;
+	}
+
+	public void execute() throws ZinggClientException {
+		executor.execute();
+	}
+
+	public abstract void validateResults() throws ZinggClientException;	
+
+}

common/common-test/src/test/java/zingg/common/core/executor/JunitLabeller.java

@@ -0,0 +1,60 @@
+package zingg.common.core.executor;
+
+import zingg.common.client.ZFrame;
+import zingg.common.client.ZinggClientException;
+import zingg.common.client.options.ZinggOptions;
+import zingg.common.client.util.ColName;
+import zingg.common.client.util.ColValues;
+import zingg.common.core.context.Context;
+
+public class JunitLabeller<S,D,R,C,T> extends Labeller<S,D,R,C,T> {
+
+	private static final long serialVersionUID = 1L;
+
+	public JunitLabeller(Context<S,D,R,C,T> context) {
+		setZinggOption(ZinggOptions.LABEL);
+		setContext(context);
+	}
+
+	@Override
+	public ZFrame<D, R, C> processRecordsCli(ZFrame<D, R, C> lines)
+			throws ZinggClientException {
+
+		// now get a list of all those rows which have same cluster and match due to fname => mark match
+		ZFrame<D, R, C> lines2 = getDSUtil().getPrefixedColumnsDS(lines);
+
+		// construct AND condition
+		C clusterCond = getJoinCondForCol(lines, lines2, ColName.CLUSTER_COLUMN,true);
+		C fnameCond = getJoinCondForCol(lines, lines2, "FNAME",true);
+		C idCond = getJoinCondForCol(lines, lines2, "ID",false);
+		C filterCond = lines2.and(lines2.and(clusterCond,idCond),fnameCond);
+
+		ZFrame<D, R, C> filtered = lines.joinOnCol(lines2, filterCond).cache();
+
+		ZFrame<D, R, C> matches = filtered.select(ColName.CLUSTER_COLUMN).distinct().withColumn(ColName.MATCH_FLAG_COL, ColValues.IS_MATCH_PREDICTION).cache();
+
+		ZFrame<D, R, C> nonMatches = lines.select(ColName.CLUSTER_COLUMN).except(matches.select(ColName.CLUSTER_COLUMN)).distinct().withColumn(ColName.MATCH_FLAG_COL, ColValues.IS_NOT_A_MATCH_PREDICTION).cache();
+
+		ZFrame<D, R, C> all = matches.unionAll(nonMatches);
+
+		ZFrame<D, R, C> linesMatched = lines;
+		linesMatched = linesMatched.drop(ColName.MATCH_FLAG_COL);
+		linesMatched = linesMatched.joinOnCol(all, ColName.CLUSTER_COLUMN);
+		linesMatched = linesMatched.select(lines.columns()); // make same order
+
+		return linesMatched;
+	}
+
+	private C getJoinCondForCol(ZFrame<D, R, C> df1, ZFrame<D, R, C> dfToJoin,String colName, boolean equal) {
+		C column = df1.col(colName);
+		C columnWithPrefix = dfToJoin.col(ColName.COL_PREFIX + colName);
+		C equalTo = df1.equalTo(column,columnWithPrefix);
+		if (equal) {
+			return equalTo;
+		} else {
+			return df1.not(equalTo);
+		}
+	}
+
+
+}

common/common-test/src/test/java/zingg/common/core/executor/LabellerTester.java

@@ -0,0 +1,36 @@
+package zingg.common.core.executor;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import zingg.common.client.ZFrame;
+import zingg.common.client.ZinggClientException;
+import zingg.common.client.util.ColName;
+
+public class LabellerTester<S, D, R, C, T> extends ExecutorTester<S, D, R, C, T> {
+
+	public static final Log LOG = LogFactory.getLog(LabellerTester.class);
+
+	public LabellerTester(Labeller<S, D, R, C, T> executor) {
+		super(executor);
+	}
+
+	@Override
+	public void validateResults() throws ZinggClientException {
+		// check that marked data has at least 1 match row and 1 unmatch row
+		ZFrame<D, R, C> dfMarked = executor.getContext().getPipeUtil().
+				read(false, false, executor.getContext().getPipeUtil().getTrainingDataMarkedPipe(executor.getArgs()));
+
+		C matchCond = dfMarked.equalTo(ColName.MATCH_FLAG_COL, 1);
+		C notMatchCond = dfMarked.equalTo(ColName.MATCH_FLAG_COL, 0);
+
+		long matchCount = dfMarked.filter(matchCond).count();
+		assertTrue(matchCount > 1);
+		long unmatchCount = dfMarked.filter(notMatchCond).count();
+		assertTrue(unmatchCount > 1);
+		LOG.info("matchCount : "+ matchCount + ", unmatchCount : " + unmatchCount);		
+	}
+
+}

common/common-test/src/test/java/zingg/common/core/executor/MatcherTester.java

@@ -0,0 +1,78 @@
+package zingg.common.core.executor;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import zingg.common.client.ZFrame;
+import zingg.common.client.ZinggClientException;
+import zingg.common.client.util.ColName;
+
+public class MatcherTester<S, D, R, C, T> extends ExecutorTester<S, D, R, C, T> {
+
+	public static final Log LOG = LogFactory.getLog(MatcherTester.class);
+
+	public MatcherTester(Matcher<S, D, R, C, T> executor) {
+		super(executor);
+	}
+
+	@Override
+	public void validateResults() throws ZinggClientException {
+		assessAccuracy();		
+	}
+
+	public String getClusterColName() {
+		return ColName.CLUSTER_COLUMN;	
+	}
+
+	protected void assessAccuracy() throws ZinggClientException {
+		ZFrame<D, R, C> df  = getOutputData();
+
+		df = df.withColumn("fnameId",df.concat(df.col("fname"), df.col("id")));
+		df = df.select("fnameId", getClusterColName());
+		df = df.withColumn("dupeFnameId",df.substr(df.col("fnameId"),0,8)).cache();
+		ZFrame<D, R, C> df1 = df.withColumnRenamed("fnameId", "fnameId1").withColumnRenamed("dupeFnameId", "dupeFnameId1")
+							.withColumnRenamed(getClusterColName(), getClusterColName() + "1").cache();
+
+
+		ZFrame<D, R, C> gold = joinAndFilter("dupeFnameId", df, df1).cache();
+		ZFrame<D, R, C> result = joinAndFilter(getClusterColName(), df, df1).cache();
+
+		ZFrame<D, R, C> fn = gold.except(result);
+		ZFrame<D, R, C> tp = gold.intersect(result);
+		ZFrame<D, R, C> fp = result.except(gold);
+
+		long fnCount = fn.count();
+		long tpCount = tp.count();
+		long fpCount = fp.count();
+		double score1 = tpCount*1.0d/(tpCount+fpCount);
+		double score2 = tpCount*1.0d/(tpCount+fnCount);
+
+		LOG.info("False negative " + fnCount);
+		LOG.info("True positive " + tpCount);
+		LOG.info("False positive " + fpCount);
+		LOG.info("precision " + score1);
+		LOG.info("recall " + tpCount + " denom " + (tpCount+fnCount) + " overall " + score2);
+
+		System.out.println("precision score1 " + score1);
+
+		System.out.println("recall score2 " + score2);
+
+		assertTrue(0.8 <= score1);
+		assertTrue(0.8 <= score2);
+	}
+
+	public ZFrame<D, R, C> getOutputData() throws ZinggClientException {
+		ZFrame<D, R, C> output = executor.getContext().getPipeUtil().read(false, false, executor.getArgs().getOutput()[0]);
+		return output;
+	}
+
+	protected ZFrame<D, R, C> joinAndFilter(String colName, ZFrame<D, R, C> df, ZFrame<D, R, C> df1){
+		C col1 = df.col(colName);
+		C col2 = df1.col(colName+"1");
+		ZFrame<D, R, C> joined = df.joinOnCol(df1, df.equalTo(col1, col2));
+		return joined.filter(joined.gt(joined.col("fnameId"), joined.col("fnameId1")));
+	}
+
+}

common/common-test/src/test/java/zingg/common/core/executor/TestExecutorsGeneric.java

@@ -0,0 +1,104 @@
+package zingg.common.core.executor;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.junit.jupiter.api.Test;
+
+import zingg.common.client.ArgumentsUtil;
+import zingg.common.client.IArguments;
+import zingg.common.client.ZinggClientException;
+
+public abstract class TestExecutorsGeneric<S, D, R, C, T> {
+
+	public static final Log LOG = LogFactory.getLog(TestExecutorsGeneric.class);
+
+	protected IArguments args;
+
+
+	protected S session;
+
+	public TestExecutorsGeneric() {
+
+	}
+
+	public TestExecutorsGeneric(S s) throws ZinggClientException, IOException {
+		init(s);					
+	}
+
+	public void init(S s) throws ZinggClientException, IOException {
+		this.session = s;
+		// set up args
+		setupArgs();					
+	}
+
+	public String setupArgs() throws ZinggClientException, IOException {
+		String configFile = getClass().getClassLoader().getResource(getConfigFile()).getFile();
+		args = new ArgumentsUtil().createArgumentsFromJSON(
+			configFile, 
+			"findTrainingData");
+		return configFile;
+	}
+
+	public abstract String getConfigFile();
+
+
+	@Test
+	public void testExecutors() throws ZinggClientException {	
+		List<ExecutorTester<S, D, R, C, T>> executorTesterList = new ArrayList<ExecutorTester<S, D, R, C, T>>();
+
+		TrainingDataFinder<S, D, R, C, T> trainingDataFinder = getTrainingDataFinder();
+		trainingDataFinder.init(args);
+		TrainingDataFinderTester<S, D, R, C, T> tdft = new TrainingDataFinderTester<S, D, R, C, T>(trainingDataFinder);
+		executorTesterList.add(tdft);
+
+		Labeller<S, D, R, C, T> labeller = getLabeller();
+		labeller.init(args);
+		LabellerTester<S, D, R, C, T> lt = new LabellerTester<S, D, R, C, T>(labeller);
+		executorTesterList.add(lt);
+
+		// training and labelling needed twice to get sufficient data
+		TrainingDataFinder<S, D, R, C, T> trainingDataFinder2 = getTrainingDataFinder();
+		trainingDataFinder2.init(args);
+		TrainingDataFinderTester<S, D, R, C, T> tdft2 = new TrainingDataFinderTester<S, D, R, C, T>(trainingDataFinder2);
+		executorTesterList.add(tdft2);
+
+		Labeller<S, D, R, C, T> labeller2 = getLabeller();
+		labeller2.init(args);
+		LabellerTester<S, D, R, C, T> lt2 = new LabellerTester<S, D, R, C, T>(labeller2);
+		executorTesterList.add(lt2);
+
+		Trainer<S, D, R, C, T> trainer = getTrainer();
+		trainer.init(args);
+		TrainerTester<S, D, R, C, T> tt = new TrainerTester<S, D, R, C, T>(trainer);
+		executorTesterList.add(tt);
+
+		Matcher<S, D, R, C, T> matcher = getMatcher();
+		matcher.init(args);
+		MatcherTester<S, D, R, C, T> mt = new MatcherTester(matcher);
+		executorTesterList.add(mt);
+
+		testExecutors(executorTesterList);
+	}
+
+
+	public void testExecutors(List<ExecutorTester<S, D, R, C, T>> executorTesterList) throws ZinggClientException {
+		for (ExecutorTester<S, D, R, C, T> executorTester : executorTesterList) {
+			executorTester.execute();
+			executorTester.validateResults();
+		}
+	}	
+
+	public abstract void tearDown();	
+
+	protected abstract TrainingDataFinder<S, D, R, C, T> getTrainingDataFinder() throws ZinggClientException;
+
+	protected abstract Labeller<S, D, R, C, T> getLabeller() throws ZinggClientException;
+
+	protected abstract Trainer<S, D, R, C, T> getTrainer() throws ZinggClientException;
+
+	protected abstract Matcher<S, D, R, C, T> getMatcher() throws ZinggClientException;	
+
+}

common/common-test/src/test/java/zingg/common/core/executor/TrainerTester.java

@@ -0,0 +1,19 @@
+package zingg.common.core.executor;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+public class TrainerTester<S, D, R, C, T> extends ExecutorTester<S, D, R, C, T> {
+
+	public static final Log LOG = LogFactory.getLog(TrainerTester.class);
+
+	public TrainerTester(Trainer<S, D, R, C, T> executor) {
+		super(executor);
+	}
+
+	@Override
+	public void validateResults() {
+		LOG.info("train successful");
+	}
+
+}