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 testInitContext() {
          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);
  
          ComputeUtils.<Integer, Integer, Integer>initContext(
              ignite,
              upstreamCacheName,
              UpstreamTransformerBuilder.identity(),
              (k, v) -> true,
              datasetCacheName,
              (env, upstream, upstreamSize) -> {
  
                  assertEquals(1, upstreamSize);
  
                  UpstreamEntry<Integer, Integer> e = upstream.next();
                  return e.getKey() + e.getValue();
              },
              TestUtils.testEnvBuilder(),
              0,
              0,
              false,
              DeployingContext.unitialized()
          );
  
          assertEquals(1, datasetCache.size());
          assertEquals(84, datasetCache.get(0).intValue());
      }
test_case.class_method_signature: ComputeUtilsTest.testInitContext()
test_case.full_signature: @Test public void testInitContext()
test_case.identifier: testInitContext
test_case.invocations: localNode cluster grid randomUUID randomUUID setName setAffinity createCache setName setAffinity createCache put initContext identity assertEquals next getKey getValue testEnvBuilder unitialized assertEquals size assertEquals intValue get
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testInitContext()
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> 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) {
  
          affinityCallWithRetries(ignite, Arrays.asList(datasetCacheName, upstreamCacheName), part -> {
              Ignite locIgnite = Ignition.localIgnite();
              LearningEnvironment env = envBuilder.buildForWorker(part);
  
              IgniteCache<K, V> locUpstreamCache = locIgnite.cache(upstreamCacheName);
  
              if (isKeepBinary)
                  locUpstreamCache = locUpstreamCache.withKeepBinary();
  
              ScanQuery<K, V> qry = new ScanQuery<>();
              qry.setLocal(true);
              qry.setPartition(part);
              qry.setFilter(filter);
  
              C ctx;
              UpstreamTransformer transformer = transformerBuilder.build(env);
              UpstreamTransformer transformerCp = Utils.copy(transformer);
  
              long cnt = computeCount(locUpstreamCache, qry, transformer);
  
              try (QueryCursor<UpstreamEntry<K, V>> cursor = locUpstreamCache.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 + ']');
  
                  ctx = ctxBuilder.build(env, iter, cnt);
              }
  
              IgniteCache<Integer, C> datasetCache = locIgnite.cache(datasetCacheName);
  
              datasetCache.put(part, ctx);
  
              return part;
          }, retries, interval, deployingContext);
      }
focal_method.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_method.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_method.identifier: initContext
focal_method.invocations: affinityCallWithRetries asList localIgnite buildForWorker cache withKeepBinary setLocal setPartition setFilter build copy computeCount query getKey getValue iterator transform map asStream iterator map asStream iterator build cache put
focal_method.modifiers: public static
focal_method.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_method.return: void
focal_method.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)