SimilarityStreamGenerator { public Stream<SimilarityResult> stream(T[] inputs, int[] sourceIndexIds, int[] targetIndexIds, double cutoff, int topK) { int concurrency = configuration.getConcurrency(); int length = inputs.length; if (concurrency == 1) { if (topK != 0) { return similarityStreamTopK(inputs, sourceIndexIds, targetIndexIds, length, cutoff, topK, computer, decoderFactory); } else { return similarityStream(inputs, sourceIndexIds, targetIndexIds, length, cutoff, computer, decoderFactory); } } else { if (topK != 0) { return similarityParallelStreamTopK(inputs, sourceIndexIds, targetIndexIds, length, terminationFlag, concurrency, cutoff, topK, computer, decoderFactory); } else { return similarityParallelStream(inputs, sourceIndexIds, targetIndexIds, length, terminationFlag, concurrency, cutoff, computer, decoderFactory); } } } SimilarityStreamGenerator(TerminationFlag terminationFlag, ProcedureConfiguration configuration, Supplier<RleDecoder> decoderFactory, SimilarityComputer<T> computer); Stream<SimilarityResult> stream(T[] inputs, int[] sourceIndexIds, int[] targetIndexIds, double cutoff, int topK); Stream<SimilarityResult> stream(T[] inputs, double cutoff, int topK); static void computeSimilarityForSourceIndex(int sourceId, T[] inputs, int length, double cutoff, SimilarityConsumer consumer, SimilarityComputer<T> computer, RleDecoder decoder);  }
@Test public void allPairs() { ProcedureConfiguration configuration = ProcedureConfiguration.create(MapUtil.map("concurrency", concurrency)); SimilarityStreamGenerator<CategoricalInput> generator = new SimilarityStreamGenerator<>(terminationFlag(), configuration, DECODER, ALL_PAIRS_COMPUTER); CategoricalInput[] ids = new CategoricalInput[3]; ids[0] = new CategoricalInput(0, new long[]{}); ids[1] = new CategoricalInput(1, new long[]{}); ids[2] = new CategoricalInput(2, new long[]{}); Stream<SimilarityResult> stream = generator.stream(ids, -1.0, 0); List<SimilarityResult> rows = stream.collect(Collectors.toList()); assertEquals(3, rows.size()); assertThat(rows, hasItems(similarityResult(0, 1, true, false))); assertThat(rows, hasItems(similarityResult(0, 2, true, false))); assertThat(rows, hasItems(similarityResult(1, 2, true, false))); }
@Test public void allPairsTopK() { ProcedureConfiguration configuration = ProcedureConfiguration.create(MapUtil.map("concurrency", concurrency)); SimilarityStreamGenerator<CategoricalInput> generator = new SimilarityStreamGenerator<>(terminationFlag(), configuration, DECODER, ALL_PAIRS_COMPUTER); CategoricalInput[] ids = new CategoricalInput[3]; ids[0] = new CategoricalInput(0, new long[]{}); ids[1] = new CategoricalInput(1, new long[]{}); ids[2] = new CategoricalInput(2, new long[]{}); Stream<SimilarityResult> stream = generator.stream(ids,-1.0, 1); List<SimilarityResult> rows = stream.collect(Collectors.toList()); assertEquals(3, rows.size()); assertThat(rows, hasItems(similarityResult(0, 1, true, false))); assertThat(rows, hasItems(similarityResult(1, 0, true, true))); assertThat(rows, hasItems(similarityResult(2, 0, true, true))); }
@Test public void sourceSpecifiedTargetSpecified() { ProcedureConfiguration configuration = ProcedureConfiguration.create(MapUtil.map("concurrency", concurrency)); SimilarityStreamGenerator<CategoricalInput> generator = new SimilarityStreamGenerator<>(terminationFlag(), configuration, DECODER, COMPUTER); CategoricalInput[] ids = new CategoricalInput[3]; ids[0] = new CategoricalInput(0, new long[]{}); ids[1] = new CategoricalInput(1, new long[]{}); ids[2] = new CategoricalInput(2, new long[]{}); int[] sourceIndexIds = new int[]{0}; int[] targetIndexIds = new int[]{1, 2}; Stream<SimilarityResult> stream = generator.stream(ids, sourceIndexIds, targetIndexIds, -1.0, 0); List<SimilarityResult> rows = stream.collect(Collectors.toList()); assertEquals(2, rows.size()); assertThat(rows, hasItems(similarityResult(0, 1, false, false))); assertThat(rows, hasItems(similarityResult(0, 2, false, false))); }
@Test public void sourceSpecifiedTargetSpecifiedTopK() { ProcedureConfiguration configuration = ProcedureConfiguration.create(MapUtil.map("concurrency", concurrency)); SimilarityStreamGenerator<CategoricalInput> generator = new SimilarityStreamGenerator<>(terminationFlag(), configuration, DECODER, COMPUTER); CategoricalInput[] ids = new CategoricalInput[3]; ids[0] = new CategoricalInput(0, new long[]{}); ids[1] = new CategoricalInput(1, new long[]{}); ids[2] = new CategoricalInput(2, new long[]{}); int[] sourceIndexIds = new int[]{1}; int[] targetIndexIds = new int[]{0, 2}; Stream<SimilarityResult> stream = generator.stream(ids, sourceIndexIds, targetIndexIds, -1.0, 1); List<SimilarityResult> rows = stream.collect(Collectors.toList()); assertEquals(1, rows.size()); assertThat(rows, hasItems(similarityResult(1, 0, false, false))); }
@Test public void sourceSpecifiedTargetNotSpecified() { ProcedureConfiguration configuration = ProcedureConfiguration.create(MapUtil.map("concurrency", concurrency)); SimilarityStreamGenerator<CategoricalInput> generator = new SimilarityStreamGenerator<>(terminationFlag(), configuration, DECODER, COMPUTER); CategoricalInput[] ids = new CategoricalInput[4]; ids[0] = new CategoricalInput(0, new long[]{}); ids[1] = new CategoricalInput(1, new long[]{}); ids[2] = new CategoricalInput(2, new long[]{}); ids[3] = new CategoricalInput(3, new long[]{}); int[] sourceIndexIds = new int[]{1, 3}; int[] targetIndexIds = new int[]{}; Stream<SimilarityResult> stream = generator.stream(ids, sourceIndexIds, targetIndexIds, -1.0, 0); List<SimilarityResult> rows = stream.collect(Collectors.toList()); for (SimilarityResult row : rows) { System.out.println(row); } assertEquals(6, rows.size()); assertThat(rows, hasItems(similarityResult(1, 0, false, false))); assertThat(rows, hasItems(similarityResult(1, 2, false, false))); assertThat(rows, hasItems(similarityResult(1, 3, false, false))); assertThat(rows, hasItems(similarityResult(3, 0, false, false))); assertThat(rows, hasItems(similarityResult(3, 1, false, false))); assertThat(rows, hasItems(similarityResult(3, 2, false, false))); }
@Test public void sourceSpecifiedTargetNotSpecifiedTopK() { ProcedureConfiguration configuration = ProcedureConfiguration.create(MapUtil.map("concurrency", concurrency)); SimilarityStreamGenerator<CategoricalInput> generator = new SimilarityStreamGenerator<>(terminationFlag(), configuration, DECODER, COMPUTER); CategoricalInput[] ids = new CategoricalInput[4]; ids[0] = new CategoricalInput(0, new long[]{}); ids[1] = new CategoricalInput(1, new long[]{}); ids[2] = new CategoricalInput(2, new long[]{}); ids[3] = new CategoricalInput(3, new long[]{}); int[] sourceIndexIds = new int[]{1, 3}; int[] targetIndexIds = new int[]{}; Stream<SimilarityResult> stream = generator.stream(ids, sourceIndexIds, targetIndexIds, -1.0, 1); List<SimilarityResult> rows = stream.collect(Collectors.toList()); for (SimilarityResult row : rows) { System.out.println(row); } assertEquals(2, rows.size()); assertThat(rows, hasItems(similarityResult(1, 0, false, false))); assertThat(rows, hasItems(similarityResult(3, 0, false, false))); }
@Test public void sourceNotSpecifiedTargetSpecified() { ProcedureConfiguration configuration = ProcedureConfiguration.create(MapUtil.map("concurrency", concurrency)); SimilarityStreamGenerator<CategoricalInput> generator = new SimilarityStreamGenerator<>(terminationFlag(), configuration, DECODER, COMPUTER); CategoricalInput[] ids = new CategoricalInput[4]; ids[0] = new CategoricalInput(0, new long[]{}); ids[1] = new CategoricalInput(1, new long[]{}); ids[2] = new CategoricalInput(2, new long[]{}); ids[3] = new CategoricalInput(3, new long[]{}); int[] sourceIndexIds = new int[]{}; int[] targetIndexIds = new int[]{2, 3}; Stream<SimilarityResult> stream = generator.stream(ids, sourceIndexIds, targetIndexIds, -1.0, 0); List<SimilarityResult> rows = stream.collect(Collectors.toList()); assertEquals(6, rows.size()); assertThat(rows, hasItems(similarityResult(0, 3, false, false))); assertThat(rows, hasItems(similarityResult(1, 3, false, false))); assertThat(rows, hasItems(similarityResult(2, 3, false, false))); assertThat(rows, hasItems(similarityResult(0, 2, false, false))); assertThat(rows, hasItems(similarityResult(1, 2, false, false))); assertThat(rows, hasItems(similarityResult(2, 3, false, false))); }
@Test public void sourceNotSpecifiedTargetSpecifiedTopK() { ProcedureConfiguration configuration = ProcedureConfiguration.create(MapUtil.map("concurrency", concurrency)); SimilarityStreamGenerator<CategoricalInput> generator = new SimilarityStreamGenerator<>(terminationFlag(), configuration, DECODER, COMPUTER); CategoricalInput[] ids = new CategoricalInput[4]; ids[0] = new CategoricalInput(0, new long[]{}); ids[1] = new CategoricalInput(1, new long[]{}); ids[2] = new CategoricalInput(2, new long[]{}); ids[3] = new CategoricalInput(3, new long[]{}); int[] sourceIndexIds = new int[]{}; int[] targetIndexIds = new int[]{2, 3}; Stream<SimilarityResult> stream = generator.stream(ids, sourceIndexIds, targetIndexIds, -1.0, 1); List<SimilarityResult> rows = stream.collect(Collectors.toList()); assertEquals(4, rows.size()); assertThat(rows, hasItems(similarityResult(0, 2, false, false))); assertThat(rows, hasItems(similarityResult(1, 2, false, false))); assertThat(rows, hasItems(similarityResult(2, 3, false, false))); assertThat(rows, hasItems(similarityResult(3, 2, false, false))); }
@Test public void sourceTargetOverlap() { ProcedureConfiguration configuration = ProcedureConfiguration.create(MapUtil.map("concurrency", concurrency)); SimilarityStreamGenerator<CategoricalInput> generator = new SimilarityStreamGenerator<>(terminationFlag(), configuration, DECODER, COMPUTER); CategoricalInput[] ids = new CategoricalInput[4]; ids[0] = new CategoricalInput(5, new long[]{}); ids[1] = new CategoricalInput(6, new long[]{}); ids[2] = new CategoricalInput(7, new long[]{}); ids[3] = new CategoricalInput(8, new long[]{}); int[] sourceIndexIds = new int[]{0, 1, 2}; int[] targetIndexIds = new int[]{1, 2}; Stream<SimilarityResult> stream = generator.stream(ids, sourceIndexIds, targetIndexIds, -1.0, 0); List<SimilarityResult> rows = stream.collect(Collectors.toList()); assertEquals(4, rows.size()); assertThat(rows, hasItems(similarityResult(5, 6, false, false))); assertThat(rows, hasItems(similarityResult(5, 7, false, false))); assertThat(rows, hasItems(similarityResult(6, 7, false, false))); assertThat(rows, hasItems(similarityResult(7, 6, false, false))); }
@Test public void sourceTargetOverlapTopK() { ProcedureConfiguration configuration = ProcedureConfiguration.create(MapUtil.map("concurrency", concurrency)); SimilarityStreamGenerator<CategoricalInput> generator = new SimilarityStreamGenerator<>(terminationFlag(), configuration, DECODER, COMPUTER); CategoricalInput[] ids = new CategoricalInput[4]; ids[0] = new CategoricalInput(5, new long[]{}); ids[1] = new CategoricalInput(6, new long[]{}); ids[2] = new CategoricalInput(7, new long[]{}); ids[3] = new CategoricalInput(8, new long[]{}); int[] sourceIndexIds = new int[]{0, 1, 2}; int[] targetIndexIds = new int[]{1, 2}; Stream<SimilarityResult> stream = generator.stream(ids, sourceIndexIds, targetIndexIds, -1.0, 1); List<SimilarityResult> rows = stream.collect(Collectors.toList()); assertEquals(3, rows.size()); assertThat(rows, hasItems(similarityResult(5, 6, false, false))); assertThat(rows, hasItems(similarityResult(6, 7, false, false))); assertThat(rows, hasItems(similarityResult(7, 6, false, false))); }