hudi-client/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndex.java
/**
   * Lookup the location for each record key and return the pair<record_key,location> for all record keys already
   * present and drop the record keys if not present.
   */
Lookup the location for each record key and return the pair for all record keys already
present and drop the record keys if not present.
Lookup the location for each record key and return the pair for all record keys already present and drop the record keys if not present .
private JavaPairRDD<HoodieKey, HoodieRecordLocation> lookupIndex(
      JavaPairRDD<String, String> partitionRecordKeyPairRDD, final JavaSparkContext jsc,
      final HoodieTable hoodieTable) {
    Map<String, Long> recordsPerPartition = partitionRecordKeyPairRDD.countByKey();
    List<String> affectedPartitionPathList = new ArrayList<>(recordsPerPartition.keySet());
    List<Tuple2<String, BloomIndexFileInfo>> fileInfoList =
        loadInvolvedFiles(affectedPartitionPathList, jsc, hoodieTable);
    final Map<String, List<BloomIndexFileInfo>> partitionToFileInfo =
        fileInfoList.stream().collect(groupingBy(Tuple2::_1, mapping(Tuple2::_2, toList())));
    Map<String, Long> comparisonsPerFileGroup =
        computeComparisonsPerFileGroup(recordsPerPartition, partitionToFileInfo, partitionRecordKeyPairRDD);
    int inputParallelism = partitionRecordKeyPairRDD.partitions().size();
    int joinParallelism = Math.max(inputParallelism, config.getBloomIndexParallelism());
    LOG.info("InputParallelism: ${" + inputParallelism + "}, IndexParallelism: ${"
        + config.getBloomIndexParallelism() + "}");
    return findMatchingFilesForRecordKeys(partitionToFileInfo, partitionRecordKeyPairRDD, joinParallelism, hoodieTable,
        comparisonsPerFileGroup);
  }
private JavaPairRDD < HoodieKey , HoodieRecordLocation > lookupIndex ( JavaPairRDD < String , String > partitionRecordKeyPairRDD , final JavaSparkContext jsc , final HoodieTable hoodieTable ) { Map < String , Long > recordsPerPartition = partitionRecordKeyPairRDD . countByKey ( ) ; List < String > affectedPartitionPathList = new ArrayList < > ( recordsPerPartition . keySet ( ) ) ; List < Tuple2 < String , BloomIndexFileInfo > > fileInfoList = loadInvolvedFiles ( affectedPartitionPathList , jsc , hoodieTable ) ; final Map < String , List < BloomIndexFileInfo > > partitionToFileInfo = fileInfoList . stream ( ) . collect ( groupingBy ( Tuple2 :: _1 , mapping ( Tuple2 :: _2 , toList ( ) ) ) ) ; Map < String , Long > comparisonsPerFileGroup = computeComparisonsPerFileGroup ( recordsPerPartition , partitionToFileInfo , partitionRecordKeyPairRDD ) ; int inputParallelism = partitionRecordKeyPairRDD . partitions ( ) . size ( ) ; int joinParallelism = Math . max ( inputParallelism , config . getBloomIndexParallelism ( ) ) ; LOG . info ( "InputParallelism: ${" + inputParallelism + "}, IndexParallelism: ${" + config . getBloomIndexParallelism ( ) + "}" ) ; return findMatchingFilesForRecordKeys ( partitionToFileInfo , partitionRecordKeyPairRDD , joinParallelism , hoodieTable , comparisonsPerFileGroup ) ; }
Lookup the location for each record key and return the pair<record_key,location> for all record keys already
 present and drop the record keys if not present.
Lookup the location for each record key and return the pair<record_key location > for all record keys already present and drop the record keys if not present .
// Obtain records per partition, in the incoming records // Step 2: Load all involved files as <Partition, filename> pairs // Step 3: Obtain a RDD, for each incoming record, that already exists, with the file id, // that contains it.