test_class.declarator: NODE_COUNT = 10
test_class.modifier: private static final
test_class.original_string: private static final int NODE_COUNT = 10;
test_class.type: int
test_class.var_name: NODE_COUNT
test_class.declarator: ignite
test_class.modifier: private
test_class.original_string: private Ignite ignite;
test_class.type: Ignite
test_class.var_name: ignite
test_class.file: modules/ml/src/test/java/org/apache/ignite/ml/dataset/impl/cache/util/ComputeUtilsTest.java
test_class.identifier: ComputeUtilsTest
test_class.superclass: extends GridCommonAbstractTest
test_case.body: @Test
      public void testGetData() {
          ClusterNode node = grid(1).cluster().localNode();
  
          String upstreamCacheName = "CACHE_1_" + UUID.randomUUID();
          String datasetCacheName = "CACHE_2_" + UUID.randomUUID();
  
          CacheConfiguration<Integer, Integer> upstreamCacheConfiguration = new CacheConfiguration<>();
          upstreamCacheConfiguration.setName(upstreamCacheName);
          upstreamCacheConfiguration.setAffinity(new TestAffinityFunction(node));
          IgniteCache<Integer, Integer> upstreamCache = ignite.createCache(upstreamCacheConfiguration);
  
          CacheConfiguration<Integer, Integer> datasetCacheConfiguration = new CacheConfiguration<>();
          datasetCacheConfiguration.setName(datasetCacheName);
          datasetCacheConfiguration.setAffinity(new TestAffinityFunction(node));
          IgniteCache<Integer, Integer> datasetCache = ignite.createCache(datasetCacheConfiguration);
  
          upstreamCache.put(42, 42);
          datasetCache.put(0, 0);
  
          UUID datasetId = UUID.randomUUID();
  
          IgniteAtomicLong cnt = ignite.atomicLong("CNT_" + datasetId, 0, true);
  
          for (int i = 0; i < 10; i++) {
              Collection<TestPartitionData> data = ComputeUtils.affinityCallWithRetries(
                  ignite,
                  Arrays.asList(datasetCacheName, upstreamCacheName),
                  part -> ComputeUtils.<Integer, Integer, Serializable, TestPartitionData>getData(
                      ignite,
                      upstreamCacheName,
                      (k, v) -> true,
                      UpstreamTransformerBuilder.identity(),
                      datasetCacheName,
                      datasetId,
                      (env, upstream, upstreamSize, ctx) -> {
                          cnt.incrementAndGet();
  
                          assertEquals(1, upstreamSize);
  
                          UpstreamEntry<Integer, Integer> e = upstream.next();
                          return new TestPartitionData(e.getKey() + e.getValue());
                      },
                      TestUtils.testEnvBuilder().buildForWorker(part),
                      false
                  ),
                  0,
                  DeployingContext.unitialized()
              );
  
              assertEquals(1, data.size());
  
              TestPartitionData dataElement = data.iterator().next();
              assertEquals(84, dataElement.val.intValue());
          }
  
          assertEquals(1, cnt.get());
      }
test_case.class_method_signature: ComputeUtilsTest.testGetData()
test_case.full_signature: @Test public void testGetData()
test_case.identifier: testGetData
test_case.invocations: localNode cluster grid randomUUID randomUUID setName setAffinity createCache setName setAffinity createCache put put randomUUID atomicLong affinityCallWithRetries asList getData identity incrementAndGet assertEquals next getKey getValue buildForWorker testEnvBuilder unitialized assertEquals size next iterator assertEquals intValue assertEquals get
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testGetData()
focal_class.declarator: DATA_STORAGE_KEY_TEMPLATE = "part_data_storage_%s"
focal_class.modifier: private static final
focal_class.original_string: private static final String DATA_STORAGE_KEY_TEMPLATE = "part_data_storage_%s";
focal_class.type: String
focal_class.var_name: DATA_STORAGE_KEY_TEMPLATE
focal_class.declarator: ENVIRONMENT_STORAGE_KEY_TEMPLATE = "part_environment_storage_%s"
focal_class.modifier: private static final
focal_class.original_string: private static final String ENVIRONMENT_STORAGE_KEY_TEMPLATE = "part_environment_storage_%s";
focal_class.type: String
focal_class.var_name: ENVIRONMENT_STORAGE_KEY_TEMPLATE
focal_class.file: modules/ml/src/main/java/org/apache/ignite/ml/dataset/impl/cache/util/ComputeUtils.java
focal_class.identifier: ComputeUtils
focal_class.class_method_signature: ComputeUtils.affinityCallWithRetries(Ignite ignite, Collection<String> cacheNames,
          IgniteFunction<Integer, R> fun, int retries, int interval, DeployingContext deployingContext)
focal_class.full_signature: public static Collection<R> affinityCallWithRetries(Ignite ignite, Collection<String> cacheNames,
          IgniteFunction<Integer, R> fun, int retries, int interval, DeployingContext deployingContext)
focal_class.identifier: affinityCallWithRetries
focal_class.modifiers: public static
focal_class.parameters: (Ignite ignite, Collection<String> cacheNames,
          IgniteFunction<Integer, R> fun, int retries, int interval, DeployingContext deployingContext)
focal_class.return: Collection<R>
focal_class.signature: Collection<R> affinityCallWithRetries(Ignite ignite, Collection<String> cacheNames,
          IgniteFunction<Integer, R> fun, int retries, int interval, DeployingContext deployingContext)
focal_class.class_method_signature: ComputeUtils.affinityCallWithRetries(Ignite ignite, Collection<String> cacheNames,
          IgniteFunction<Integer, R> fun, int retries, DeployingContext deployingContext)
focal_class.full_signature: public static Collection<R> affinityCallWithRetries(Ignite ignite, Collection<String> cacheNames,
          IgniteFunction<Integer, R> fun, int retries, DeployingContext deployingContext)
focal_class.identifier: affinityCallWithRetries
focal_class.modifiers: public static
focal_class.parameters: (Ignite ignite, Collection<String> cacheNames,
          IgniteFunction<Integer, R> fun, int retries, DeployingContext deployingContext)
focal_class.return: Collection<R>
focal_class.signature: Collection<R> affinityCallWithRetries(Ignite ignite, Collection<String> cacheNames,
          IgniteFunction<Integer, R> fun, int retries, DeployingContext deployingContext)
focal_class.class_method_signature: ComputeUtils.getLearningEnvironment(Ignite ignite,
          UUID datasetId,
          int part,
          LearningEnvironmentBuilder envBuilder)
focal_class.full_signature: public static LearningEnvironment getLearningEnvironment(Ignite ignite,
          UUID datasetId,
          int part,
          LearningEnvironmentBuilder envBuilder)
focal_class.identifier: getLearningEnvironment
focal_class.modifiers: public static
focal_class.parameters: (Ignite ignite,
          UUID datasetId,
          int part,
          LearningEnvironmentBuilder envBuilder)
focal_class.return: LearningEnvironment
focal_class.signature: LearningEnvironment getLearningEnvironment(Ignite ignite,
          UUID datasetId,
          int part,
          LearningEnvironmentBuilder envBuilder)
focal_class.class_method_signature: ComputeUtils.getData(
          Ignite ignite,
          String upstreamCacheName, IgniteBiPredicate<K, V> filter,
          UpstreamTransformerBuilder transformerBuilder,
          String datasetCacheName, UUID datasetId,
          PartitionDataBuilder<K, V, C, D> partDataBuilder,
          LearningEnvironment env,
          boolean isKeepBinary)
focal_class.full_signature: public static D getData(
          Ignite ignite,
          String upstreamCacheName, IgniteBiPredicate<K, V> filter,
          UpstreamTransformerBuilder transformerBuilder,
          String datasetCacheName, UUID datasetId,
          PartitionDataBuilder<K, V, C, D> partDataBuilder,
          LearningEnvironment env,
          boolean isKeepBinary)
focal_class.identifier: getData
focal_class.modifiers: public static
focal_class.parameters: (
          Ignite ignite,
          String upstreamCacheName, IgniteBiPredicate<K, V> filter,
          UpstreamTransformerBuilder transformerBuilder,
          String datasetCacheName, UUID datasetId,
          PartitionDataBuilder<K, V, C, D> partDataBuilder,
          LearningEnvironment env,
          boolean isKeepBinary)
focal_class.return: D
focal_class.signature: D getData(
          Ignite ignite,
          String upstreamCacheName, IgniteBiPredicate<K, V> filter,
          UpstreamTransformerBuilder transformerBuilder,
          String datasetCacheName, UUID datasetId,
          PartitionDataBuilder<K, V, C, D> partDataBuilder,
          LearningEnvironment env,
          boolean isKeepBinary)
focal_class.class_method_signature: ComputeUtils.removeData(Ignite ignite, UUID datasetId)
focal_class.full_signature: public static void removeData(Ignite ignite, UUID datasetId)
focal_class.identifier: removeData
focal_class.modifiers: public static
focal_class.parameters: (Ignite ignite, UUID datasetId)
focal_class.return: void
focal_class.signature: void removeData(Ignite ignite, UUID datasetId)
focal_class.class_method_signature: ComputeUtils.removeLearningEnv(Ignite ignite, UUID datasetId)
focal_class.full_signature: public static void removeLearningEnv(Ignite ignite, UUID datasetId)
focal_class.identifier: removeLearningEnv
focal_class.modifiers: public static
focal_class.parameters: (Ignite ignite, UUID datasetId)
focal_class.return: void
focal_class.signature: void removeLearningEnv(Ignite ignite, UUID datasetId)
focal_class.class_method_signature: ComputeUtils.initContext(
          Ignite ignite,
          String upstreamCacheName,
          UpstreamTransformerBuilder transformerBuilder,
          IgniteBiPredicate<K, V> filter,
          String datasetCacheName,
          PartitionContextBuilder<K, V, C> ctxBuilder,
          LearningEnvironmentBuilder envBuilder,
          int retries,
          int interval,
          boolean isKeepBinary,
          DeployingContext deployingContext)
focal_class.full_signature: public static void initContext(
          Ignite ignite,
          String upstreamCacheName,
          UpstreamTransformerBuilder transformerBuilder,
          IgniteBiPredicate<K, V> filter,
          String datasetCacheName,
          PartitionContextBuilder<K, V, C> ctxBuilder,
          LearningEnvironmentBuilder envBuilder,
          int retries,
          int interval,
          boolean isKeepBinary,
          DeployingContext deployingContext)
focal_class.identifier: initContext
focal_class.modifiers: public static
focal_class.parameters: (
          Ignite ignite,
          String upstreamCacheName,
          UpstreamTransformerBuilder transformerBuilder,
          IgniteBiPredicate<K, V> filter,
          String datasetCacheName,
          PartitionContextBuilder<K, V, C> ctxBuilder,
          LearningEnvironmentBuilder envBuilder,
          int retries,
          int interval,
          boolean isKeepBinary,
          DeployingContext deployingContext)
focal_class.return: void
focal_class.signature: void initContext(
          Ignite ignite,
          String upstreamCacheName,
          UpstreamTransformerBuilder transformerBuilder,
          IgniteBiPredicate<K, V> filter,
          String datasetCacheName,
          PartitionContextBuilder<K, V, C> ctxBuilder,
          LearningEnvironmentBuilder envBuilder,
          int retries,
          int interval,
          boolean isKeepBinary,
          DeployingContext deployingContext)
focal_class.class_method_signature: ComputeUtils.getContext(Ignite ignite, String datasetCacheName, int part)
focal_class.full_signature: public static C getContext(Ignite ignite, String datasetCacheName, int part)
focal_class.identifier: getContext
focal_class.modifiers: public static
focal_class.parameters: (Ignite ignite, String datasetCacheName, int part)
focal_class.return: C
focal_class.signature: C getContext(Ignite ignite, String datasetCacheName, int part)
focal_class.class_method_signature: ComputeUtils.saveContext(Ignite ignite, String datasetCacheName, int part, C ctx)
focal_class.full_signature: public static void saveContext(Ignite ignite, String datasetCacheName, int part, C ctx)
focal_class.identifier: saveContext
focal_class.modifiers: public static
focal_class.parameters: (Ignite ignite, String datasetCacheName, int part, C ctx)
focal_class.return: void
focal_class.signature: void saveContext(Ignite ignite, String datasetCacheName, int part, C ctx)
focal_class.class_method_signature: ComputeUtils.computeCount(
          IgniteCache<K, V> cache,
          ScanQuery<K, V> qry,
          UpstreamTransformer transformer)
focal_class.full_signature: private static long computeCount(
          IgniteCache<K, V> cache,
          ScanQuery<K, V> qry,
          UpstreamTransformer transformer)
focal_class.identifier: computeCount
focal_class.modifiers: private static
focal_class.parameters: (
          IgniteCache<K, V> cache,
          ScanQuery<K, V> qry,
          UpstreamTransformer transformer)
focal_class.return: long
focal_class.signature: long computeCount(
          IgniteCache<K, V> cache,
          ScanQuery<K, V> qry,
          UpstreamTransformer transformer)
focal_class.class_method_signature: ComputeUtils.computeCount(Iterator<?> iter)
focal_class.full_signature: private static long computeCount(Iterator<?> iter)
focal_class.identifier: computeCount
focal_class.modifiers: private static
focal_class.parameters: (Iterator<?> iter)
focal_class.return: long
focal_class.signature: long computeCount(Iterator<?> iter)
focal_method.body: public static <K, V, C extends Serializable, D extends AutoCloseable> D getData(
          Ignite ignite,
          String upstreamCacheName, IgniteBiPredicate<K, V> filter,
          UpstreamTransformerBuilder transformerBuilder,
          String datasetCacheName, UUID datasetId,
          PartitionDataBuilder<K, V, C, D> partDataBuilder,
          LearningEnvironment env,
          boolean isKeepBinary) {
  
          PartitionDataStorage dataStorage = (PartitionDataStorage)ignite
              .cluster()
              .nodeLocalMap()
              .computeIfAbsent(String.format(DATA_STORAGE_KEY_TEMPLATE, datasetId), key -> new PartitionDataStorage(env.dataTtl()));
  
          final int part = env.partition();
  
          return dataStorage.computeDataIfAbsent(part, () -> {
              IgniteCache<Integer, C> learningCtxCache = ignite.cache(datasetCacheName);
              C ctx = learningCtxCache.get(part);
  
              IgniteCache<K, V> upstreamCache = ignite.cache(upstreamCacheName);
  
              if (isKeepBinary)
                  upstreamCache = upstreamCache.withKeepBinary();
  
              ScanQuery<K, V> qry = new ScanQuery<>();
              qry.setLocal(true);
              qry.setPartition(part);
              qry.setFilter(filter);
  
              UpstreamTransformer transformer = transformerBuilder.build(env);
              UpstreamTransformer transformerCp = Utils.copy(transformer);
  
              long cnt = computeCount(upstreamCache, qry, transformer);
  
              if (cnt > 0) {
                  try (QueryCursor<UpstreamEntry<K, V>> cursor = upstreamCache.query(qry,
                      e -> new UpstreamEntry<>(e.getKey(), e.getValue()))) {
  
                      Iterator<UpstreamEntry<K, V>> it = cursor.iterator();
                      Stream<UpstreamEntry> transformedStream = transformerCp.transform(Utils.asStream(it, cnt).map(x -> (UpstreamEntry)x));
                      it = Utils.asStream(transformedStream.iterator()).map(x -> (UpstreamEntry<K, V>)x).iterator();
  
                      Iterator<UpstreamEntry<K, V>> iter = new IteratorWithConcurrentModificationChecker<>(it, cnt,
                          "Cache expected to be not modified during dataset data building [partition=" + part + ']');
  
                      return partDataBuilder.build(env, iter, cnt, ctx);
                  }
              }
  
              return null;
          });
      }
focal_method.class_method_signature: ComputeUtils.getData(
          Ignite ignite,
          String upstreamCacheName, IgniteBiPredicate<K, V> filter,
          UpstreamTransformerBuilder transformerBuilder,
          String datasetCacheName, UUID datasetId,
          PartitionDataBuilder<K, V, C, D> partDataBuilder,
          LearningEnvironment env,
          boolean isKeepBinary)
focal_method.full_signature: public static D getData(
          Ignite ignite,
          String upstreamCacheName, IgniteBiPredicate<K, V> filter,
          UpstreamTransformerBuilder transformerBuilder,
          String datasetCacheName, UUID datasetId,
          PartitionDataBuilder<K, V, C, D> partDataBuilder,
          LearningEnvironment env,
          boolean isKeepBinary)
focal_method.identifier: getData
focal_method.invocations: computeIfAbsent nodeLocalMap cluster format dataTtl partition computeDataIfAbsent cache get cache withKeepBinary setLocal setPartition setFilter build copy computeCount query getKey getValue iterator transform map asStream iterator map asStream iterator build
focal_method.modifiers: public static
focal_method.parameters: (
          Ignite ignite,
          String upstreamCacheName, IgniteBiPredicate<K, V> filter,
          UpstreamTransformerBuilder transformerBuilder,
          String datasetCacheName, UUID datasetId,
          PartitionDataBuilder<K, V, C, D> partDataBuilder,
          LearningEnvironment env,
          boolean isKeepBinary)
focal_method.return: D
focal_method.signature: D getData(
          Ignite ignite,
          String upstreamCacheName, IgniteBiPredicate<K, V> filter,
          UpstreamTransformerBuilder transformerBuilder,
          String datasetCacheName, UUID datasetId,
          PartitionDataBuilder<K, V, C, D> partDataBuilder,
          LearningEnvironment env,
          boolean isKeepBinary)