hudi-client/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndex.java
/**
   * Find out <RowKey, filename> pair. All workload grouped by file-level.
   * <p>
   * Join PairRDD(PartitionPath, RecordKey) and PairRDD(PartitionPath, File) & then repartition such that each RDD
   * partition is a file, then for each file, we do (1) load bloom filter, (2) load rowKeys, (3) Tag rowKey
   * <p>
   * Make sure the parallelism is atleast the groupby parallelism for tagging location
   */
Find out  pair. All workload grouped by file-level.

Join PairRDD(PartitionPath, RecordKey) and PairRDD(PartitionPath, File) & then repartition such that each RDD
partition is a file, then for each file, we do (1) load bloom filter, (2) load rowKeys, (3) Tag rowKey

Make sure the parallelism is atleast the groupby parallelism for tagging location
Find out pair . All workload grouped by file - level . Join PairRDD ( PartitionPath RecordKey ) and PairRDD ( PartitionPath File ) & then repartition such that each RDD partition is a file then for each file we do ( 1 ) load bloom filter ( 2 ) load rowKeys ( 3 ) Tag rowKey Make sure the parallelism is atleast the groupby parallelism for tagging location
JavaPairRDD<HoodieKey, HoodieRecordLocation> findMatchingFilesForRecordKeys(
      final Map<String, List<BloomIndexFileInfo>> partitionToFileIndexInfo,
      JavaPairRDD<String, String> partitionRecordKeyPairRDD, int shuffleParallelism, HoodieTable hoodieTable,
      Map<String, Long> fileGroupToComparisons) {
    JavaRDD<Tuple2<String, HoodieKey>> fileComparisonsRDD =
        explodeRecordRDDWithFileComparisons(partitionToFileIndexInfo, partitionRecordKeyPairRDD);
    if (config.useBloomIndexBucketizedChecking()) {
      Partitioner partitioner = new BucketizedBloomCheckPartitioner(shuffleParallelism, fileGroupToComparisons,
          config.getBloomIndexKeysPerBucket());
      fileComparisonsRDD = fileComparisonsRDD.mapToPair(t -> new Tuple2<>(Pair.of(t._1, t._2.getRecordKey()), t))
          .repartitionAndSortWithinPartitions(partitioner).map(Tuple2::_2);
    } else {
      fileComparisonsRDD = fileComparisonsRDD.sortBy(Tuple2::_1, true, shuffleParallelism);
    }
    return fileComparisonsRDD.mapPartitionsWithIndex(new HoodieBloomIndexCheckFunction(hoodieTable, config), true)
        .flatMap(List::iterator).filter(lr -> lr.getMatchingRecordKeys().size() > 0)
        .flatMapToPair(lookupResult -> lookupResult.getMatchingRecordKeys().stream()
            .map(recordKey -> new Tuple2<>(new HoodieKey(recordKey, lookupResult.getPartitionPath()),
                new HoodieRecordLocation(lookupResult.getBaseInstantTime(), lookupResult.getFileId())))
            .collect(Collectors.toList()).iterator());
  }
JavaPairRDD < HoodieKey , HoodieRecordLocation > findMatchingFilesForRecordKeys ( final Map < String , List < BloomIndexFileInfo > > partitionToFileIndexInfo , JavaPairRDD < String , String > partitionRecordKeyPairRDD , int shuffleParallelism , HoodieTable hoodieTable , Map < String , Long > fileGroupToComparisons ) { JavaRDD < Tuple2 < String , HoodieKey > > fileComparisonsRDD = explodeRecordRDDWithFileComparisons ( partitionToFileIndexInfo , partitionRecordKeyPairRDD ) ; if ( config . useBloomIndexBucketizedChecking ( ) ) { Partitioner partitioner = new BucketizedBloomCheckPartitioner ( shuffleParallelism , fileGroupToComparisons , config . getBloomIndexKeysPerBucket ( ) ) ; fileComparisonsRDD = fileComparisonsRDD . mapToPair ( t -> new Tuple2 < > ( Pair . of ( t . _1 , t . _2 . getRecordKey ( ) ) , t ) ) . repartitionAndSortWithinPartitions ( partitioner ) . map ( Tuple2 :: _2 ) ; } else { fileComparisonsRDD = fileComparisonsRDD . sortBy ( Tuple2 :: _1 , true , shuffleParallelism ) ; } return fileComparisonsRDD . mapPartitionsWithIndex ( new HoodieBloomIndexCheckFunction ( hoodieTable , config ) , true ) . flatMap ( List :: iterator ) . filter ( lr -> lr . getMatchingRecordKeys ( ) . size ( ) > 0 ) . flatMapToPair ( lookupResult -> lookupResult . getMatchingRecordKeys ( ) . stream ( ) . map ( recordKey -> new Tuple2 < > ( new HoodieKey ( recordKey , lookupResult . getPartitionPath ( ) ) , new HoodieRecordLocation ( lookupResult . getBaseInstantTime ( ) , lookupResult . getFileId ( ) ) ) ) . collect ( Collectors . toList ( ) ) . iterator ( ) ) ; }
Find out <RowKey, filename> pair.
Find out <RowKey filename > pair .