ForeignKeys { static double calculateEarthMoverDistance(double[] sourceNodes, double[] targetNodes, double[][] distance) { logEmdState("Initial problem:", sourceNodes, targetNodes, distance); double sourceSum = Arrays.stream(sourceNodes).sum(); double targetSum = Arrays.stream(targetNodes).sum(); assert approximatelyEquals(sourceSum, targetSum); double[][] distanceCopy = new double[distance.length][]; for (int i = 0; i < distance.length; i++) { distanceCopy[i] = distance[i].clone(); } for (int source = 0; source < sourceNodes.length; source++) { double min = Double.POSITIVE_INFINITY; for (int target = 0; target < targetNodes.length; target++) { min = Math.min(min, distanceCopy[source][target]); } for (int target = 0; target < targetNodes.length; target++) { distanceCopy[source][target] -= min; } } logEmdState("Distance matrix after per-source reduction:", sourceNodes, targetNodes, distanceCopy); for (int target = 0; target < targetNodes.length; target++) { double min = Double.POSITIVE_INFINITY; for (int source = 0; source < sourceNodes.length; source++) { min = Math.min(min, distanceCopy[source][target]); } for (int source = 0; source < sourceNodes.length; source++) { distanceCopy[source][target] -= min; } } logEmdState("Distance matrix after per-target reduction:", sourceNodes, targetNodes, distanceCopy); double[] sourceCapacity = sourceNodes.clone(); double[] targetCapacity = targetNodes.clone(); Map<FlowGraphNode, List<FlowGraphNode>> flowGraphEdges = new HashMap<>(); for (int source = 0; source < sourceNodes.length; source++) { FlowGraphNode sourceNode = new FlowGraphNode(source, true); for (int target = 0; target < targetNodes.length; target++) { FlowGraphNode targetNode = new FlowGraphNode(target, false); if (isZeroOrNegative(distanceCopy[source][target])) { flowGraphEdges.computeIfAbsent(sourceNode, k -> new ArrayList<>()).add(targetNode); flowGraphEdges.computeIfAbsent(targetNode, k -> new ArrayList<>()).add(sourceNode); logger.debug("Adding {} -> {}.", sourceNode, targetNode); } } } double[][] flow = new double[sourceCapacity.length][targetCapacity.length]; while (true) { logEmdState("Current distances:", sourceCapacity, targetCapacity, distanceCopy); while (true) { PriorityQueue<FlowGraphPath> queue = new PriorityQueue<>(FlowGraphPath.distanceComparator); for (int i = 0; i < sourceCapacity.length; i++) { if (!isZero(sourceCapacity[i])) { queue.add(new FlowGraphPath(new FlowGraphNode(i, true), null, 0)); } } Set<FlowGraphNode> visitedNodes = new HashSet<>(); Deque<FlowGraphNode> shortestPath = null; while (!queue.isEmpty()) { FlowGraphPath path = queue.poll(); if (!visitedNodes.add(path.node)) continue; if (!path.node.isSourceNode && isGreaterThanZero(targetCapacity[path.node.index])) { shortestPath = new LinkedList<>(); do { shortestPath.addFirst(path.node); path = path.predecessor; } while (path != null); break; } for (FlowGraphNode nextNode : flowGraphEdges.getOrDefault(path.node, Collections.emptyList())) { if (!visitedNodes.contains(nextNode) && (!nextNode.isSourceNode || isGreaterThanZero(flow[nextNode.index][path.node.index]))) { queue.offer(new FlowGraphPath(nextNode, path, path.distance + 1)); } } } if (shortestPath != null) { double pathFlow = sourceCapacity[shortestPath.getFirst().index]; FlowGraphNode lastNode = null; for (FlowGraphNode currentNode : shortestPath) { if (lastNode != null) { if (lastNode.isSourceNode) { if (!isZero(targetCapacity[currentNode.index])) pathFlow = Math.min(pathFlow, targetCapacity[currentNode.index]); } else { pathFlow = Math.min(pathFlow, flow[currentNode.index][lastNode.index]); } } lastNode = currentNode; } lastNode = null; for (FlowGraphNode currentNode : shortestPath) { if (lastNode != null) { if (lastNode.isSourceNode) { flow[lastNode.index][currentNode.index] += pathFlow; assert flow[lastNode.index][currentNode.index] >= -0.0000001d; } else { flow[currentNode.index][lastNode.index] -= pathFlow; assert flow[currentNode.index][lastNode.index] >= -0.0000001d; } } lastNode = currentNode; } sourceCapacity[shortestPath.getFirst().index] -= pathFlow; assert sourceCapacity[shortestPath.getFirst().index] >= -0.0000001d; targetCapacity[shortestPath.getLast().index] -= pathFlow; assert targetCapacity[shortestPath.getLast().index] >= -0.0000001d; logger.debug("Augmenting flow {} with {}.", shortestPath, pathFlow); logEmdState("Capacities and flows:", sourceCapacity, targetCapacity, flow); } else { break; } } BitSet nonEmptySources = new BitSet(sourceNodes.length); BitSet labeledTargets = new BitSet(targetNodes.length); for (int i = 0; i < sourceNodes.length; i++) { if (!isZero(sourceCapacity[i])) { nonEmptySources.set(i); for (FlowGraphNode targetNode : flowGraphEdges.getOrDefault(new FlowGraphNode(i, true), Collections.emptyList())) { labeledTargets.set(targetNode.index); } } } if (nonEmptySources.isEmpty()) { double emd = 0d; for (int source = 0; source < sourceNodes.length; source++) { for (int target = 0; target < targetNodes.length; target++) { emd += distance[source][target] * flow[source][target]; } } return emd; } else { if (labeledTargets.cardinality() == targetNodes.length) { throw new IllegalStateException(); } double minDistance = Double.POSITIVE_INFINITY; for (int source = nonEmptySources.nextSetBit(0); source != -1; source = nonEmptySources.nextSetBit(source + 1)) { for (int target = labeledTargets.nextClearBit(0); target < targetNodes.length; target = labeledTargets.nextClearBit(target + 1)) { if (distanceCopy[source][target] > 0) minDistance = Math.min(minDistance, distanceCopy[source][target]); } } if (minDistance == 0d || minDistance == Double.POSITIVE_INFINITY) { throw new IllegalStateException("The calculation of earth mover's distance appears to have failed."); } for (int source = nonEmptySources.nextSetBit(0); source != -1; source = nonEmptySources.nextSetBit(source + 1)) { FlowGraphNode sourceNode = new FlowGraphNode(source, true); for (int target = labeledTargets.nextClearBit(0); target < targetNodes.length; target = labeledTargets.nextClearBit(target + 1)) { if (distanceCopy[source][target] > 0 && (distanceCopy[source][target] -= minDistance) == 0d) { FlowGraphNode targetNode = new FlowGraphNode(target, false); flowGraphEdges.computeIfAbsent(sourceNode, k -> new ArrayList<>()).add(targetNode); flowGraphEdges.computeIfAbsent(targetNode, k -> new ArrayList<>()).add(sourceNode); logger.debug("Adding {} -> {}.", sourceNode, targetNode); } } } for (int source = nonEmptySources.nextClearBit(0); source < sourceNodes.length; source = nonEmptySources.nextClearBit(source + 1)) { for (int target = labeledTargets.nextSetBit(0); target != -1; target = labeledTargets.nextSetBit(target + 1)) { if (distanceCopy[source][target] > 0d) { distanceCopy[source][target] += minDistance; } } } } } }  static Collection<Observation<InclusionDependency>> createTrainingSet(
            MetadataStore store,
            ConstraintCollection<InclusionDependency> foreignKeyCC,
            ConstraintCollection<InclusionDependency> indCC,
            ConstraintCollection<ColumnStatistics> columnStatisticCC,
            ConstraintCollection<TextColumnStatistics> textColumnStatisticCC,
            ConstraintCollection<TableSample> tableSampleCC,
            ConstraintCollection<UniqueColumnCombination> uniqueCC); static Collection<Instance<InclusionDependency>> createEstimationSet(
            MetadataStore store,
            ConstraintCollection<InclusionDependency> indCC,
            ConstraintCollection<ColumnStatistics> columnStatisticCC,
            ConstraintCollection<TextColumnStatistics> textColumnStatisticCC,
            ConstraintCollection<TableSample> tableSampleCC); static Collection<Observation<T>> oversample(
            Collection<Observation<T>> instances); static VectorModel trainLogisticRegressionModel(
            MetadataStore store,
            ConstraintCollection<InclusionDependency> foreignKeyCC,
            ConstraintCollection<InclusionDependency> indCC,
            ConstraintCollection<ColumnStatistics> columnStatisticCC,
            ConstraintCollection<TextColumnStatistics> textColumnStatisticCC,
            ConstraintCollection<TableSample> tableSampleCC,
            ConstraintCollection<UniqueColumnCombination> uniqueCC); static Collection<Prediction<InclusionDependency, Double>> estimateForeignKeyProbability(
            MetadataStore store,
            ConstraintCollection<InclusionDependency> indCC,
            ConstraintCollection<ColumnStatistics> columnStatisticCC,
            ConstraintCollection<TextColumnStatistics> textColumnStatisticCC,
            ConstraintCollection<TableSample> tableSampleCC,
            VectorModel logisticRegressionModel); static Collection<InclusionDependency> estimateForeignKeys(
            MetadataStore store,
            ConstraintCollection<InclusionDependency> indCC,
            ConstraintCollection<UniqueColumnCombination> uccCC,
            ConstraintCollection<ColumnStatistics> columnStatisticCC,
            ConstraintCollection<TextColumnStatistics> textColumnStatisticCC,
            ConstraintCollection<TableSample> tableSampleCC,
            VectorModel logisticRegressionModel); static Collection<InclusionDependency> estimateForeignKeys(
            MetadataStore store,
            ConstraintCollection<InclusionDependency> indCC,
            ConstraintCollection<UniqueColumnCombination> uccCC,
            ConstraintCollection<ColumnStatistics> columnStatisticCC,
            ConstraintCollection<TextColumnStatistics> textColumnStatisticCC,
            ConstraintCollection<TableSample> tableSampleCC,
            VectorModel logisticRegressionModel,
            double minFkProbability); static final String[] features; }
@Test public void testEarthMoversDistanceCalculation() { double emd = ForeignKeys.calculateEarthMoverDistance( new double[]{6, 9}, new double[]{4, 6, 5}, new double[][]{ new double[]{3, 5, 7}, new double[]{6, 4, 3} } ); Assert.assertEquals(4 * 3 + 2 * 5 + 4 * 4 + 5 * 3, emd, 0.0001); }