integration Junit for OSS

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,82 @@
+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 abstract 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",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 = intersect(gold,result);
+		ZFrame<D, R, C> fp = result.except(gold);
+
+		long fnCount = fn.count();
+		long tpCount = tp.count();
+		long fpCount = fp.count();
+
+		LOG.info("False negative " + fnCount);
+		LOG.info("True positive " + tpCount);
+		LOG.info("False positive " + fpCount);
+		LOG.info("precision " + (tpCount*1.0d/(tpCount+fpCount)));
+		LOG.info("recall " + tpCount + " denom " + (tpCount+fnCount) + " overall " + (tpCount*1.0d/(tpCount+fnCount)));
+
+		assertTrue(0.8 < (tpCount*1.0d/(tpCount+fpCount)));
+		assertTrue(0.8 < (tpCount*1.0d/(tpCount+fnCount)));
+	}
+
+	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(gt(joined.col("fnameId"), joined.col("fnameId1")));
+	}
+
+
+	// returns df1.intersect(df2)
+	public abstract ZFrame<D, R, C> intersect(ZFrame<D, R, C> df1, ZFrame<D, R, C> df2);
+
+	// return col.substr(startPos,len)
+	public abstract C substr(C col, int startPos, int len);
+
+	// return c1.gt(c2)
+	public abstract C gt(C column1, C column2);
+
+}

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

@@ -0,0 +1,54 @@
+package zingg.common.core.executor;
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+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();
+
+	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();	
+
+}

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");
+	}
+
+}

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

@@ -0,0 +1,29 @@
+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;
+
+public class TrainingDataFinderTester<S, D, R, C, T> extends ExecutorTester<S, D, R, C, T> {
+
+	public static final Log LOG = LogFactory.getLog(TrainingDataFinderTester.class);
+
+	public TrainingDataFinderTester(TrainingDataFinder<S, D, R, C, T> executor) {
+		super(executor);
+	}
+
+	@Override
+	public void validateResults() throws ZinggClientException {
+		// check that unmarked data has at least 10 rows
+		ZFrame<D, R, C> df = executor.getContext().getPipeUtil().read(false, false, executor.getContext().getPipeUtil().getTrainingDataUnmarkedPipe(executor.getArgs()));
+
+		long trainingDataCount = df.count();
+		assertTrue(trainingDataCount > 10);
+		LOG.info("trainingDataCount : "+ trainingDataCount);				
+	}
+
+}

common/infra/src/main/java/zingg/common/infra/util/PojoToArrayConverter.java

@@ -1,4 +1,4 @@
-package zingg.common.infraForTest.util;
+package zingg.common.infra.util;
 
 import java.lang.reflect.*;
 import java.security.NoSuchAlgorithmException;

common/pom.xml

@@ -12,6 +12,7 @@
 		<module>infra</module>
 		<module>core</module>
 		<module>client</module>
+		<module>common-test</module>
 		<module>py</module>
 	</modules>
 </project>

pom.xml

@@ -247,7 +247,6 @@
 						</execution>
 					</executions>
 				</plugin>
-		</plugins>
-
+		</plugins>		
 	</build>
 </project>

spark/pom.xml

@@ -11,6 +11,7 @@
 	<modules>
 		<module>core</module>
 		<module>client</module>
+		<module>spark-test</module>
 	</modules>
 	<dependencies>
 		<dependency>