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/core/pom.xml

@@ -29,5 +29,39 @@
 			<artifactId>httpclient</artifactId>
 			<version>4.5.14</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/core/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/core/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/core/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/core/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")));
+	}
+
+}