hudi-client/src/main/java/org/apache/hudi/index/bloom/HoodieGlobalBloomIndex.java
/**
   * Tagging for global index should only consider the record key.
   */
Tagging for global index should only consider the record key.
Tagging for global index should only consider the record key .
@Override
  protected JavaRDD<HoodieRecord<T>> tagLocationBacktoRecords(
      JavaPairRDD<HoodieKey, HoodieRecordLocation> keyLocationPairRDD, JavaRDD<HoodieRecord<T>> recordRDD) {
    JavaPairRDD<String, HoodieRecord<T>> incomingRowKeyRecordPairRDD =
        recordRDD.mapToPair(record -> new Tuple2<>(record.getRecordKey(), record));
    JavaPairRDD<String, Tuple2<HoodieRecordLocation, HoodieKey>> existingRecordKeyToRecordLocationHoodieKeyMap =
        keyLocationPairRDD.mapToPair(p -> new Tuple2<>(p._1.getRecordKey(), new Tuple2<>(p._2, p._1)));
    return incomingRowKeyRecordPairRDD.leftOuterJoin(existingRecordKeyToRecordLocationHoodieKeyMap).values().flatMap(record -> {
      final HoodieRecord<T> hoodieRecord = record._1;
      final Optional<Tuple2<HoodieRecordLocation, HoodieKey>> recordLocationHoodieKeyPair = record._2;
      if (recordLocationHoodieKeyPair.isPresent()) {
        if (config.getBloomIndexUpdatePartitionPath()
            && !recordLocationHoodieKeyPair.get()._2.getPartitionPath().equals(hoodieRecord.getPartitionPath())) {
          HoodieRecord<T> deleteRecord = new HoodieRecord(recordLocationHoodieKeyPair.get()._2,
              new EmptyHoodieRecordPayload());
          deleteRecord.setCurrentLocation(recordLocationHoodieKeyPair.get()._1());
          deleteRecord.seal();
          HoodieRecord<T> insertRecord = HoodieIndexUtils.getTaggedRecord(hoodieRecord, Option.empty());
          return Arrays.asList(deleteRecord, insertRecord).iterator();
        } else {
          return Collections.singletonList(
              (HoodieRecord<T>) HoodieIndexUtils.getTaggedRecord(new HoodieRecord<>(recordLocationHoodieKeyPair.get()._2, hoodieRecord.getData()),
                  Option.ofNullable(recordLocationHoodieKeyPair.get()._1))).iterator();
        }
      } else {
        return Collections.singletonList((HoodieRecord<T>) HoodieIndexUtils.getTaggedRecord(hoodieRecord, Option.empty())).iterator();
      }
    });
  }
@ Override protected JavaRDD < HoodieRecord < T > > tagLocationBacktoRecords ( JavaPairRDD < HoodieKey , HoodieRecordLocation > keyLocationPairRDD , JavaRDD < HoodieRecord < T > > recordRDD ) { JavaPairRDD < String , HoodieRecord < T > > incomingRowKeyRecordPairRDD = recordRDD . mapToPair ( record -> new Tuple2 < > ( record . getRecordKey ( ) , record ) ) ; JavaPairRDD < String , Tuple2 < HoodieRecordLocation , HoodieKey > > existingRecordKeyToRecordLocationHoodieKeyMap = keyLocationPairRDD . mapToPair ( p -> new Tuple2 < > ( p . _1 . getRecordKey ( ) , new Tuple2 < > ( p . _2 , p . _1 ) ) ) ; return incomingRowKeyRecordPairRDD . leftOuterJoin ( existingRecordKeyToRecordLocationHoodieKeyMap ) . values ( ) . flatMap ( record -> { final HoodieRecord < T > hoodieRecord = record . _1 ; final Optional < Tuple2 < HoodieRecordLocation , HoodieKey > > recordLocationHoodieKeyPair = record . _2 ; if ( recordLocationHoodieKeyPair . isPresent ( ) ) { if ( config . getBloomIndexUpdatePartitionPath ( ) && ! recordLocationHoodieKeyPair . get ( ) . _2 . getPartitionPath ( ) . equals ( hoodieRecord . getPartitionPath ( ) ) ) { HoodieRecord < T > deleteRecord = new HoodieRecord ( recordLocationHoodieKeyPair . get ( ) . _2 , new EmptyHoodieRecordPayload ( ) ) ; deleteRecord . setCurrentLocation ( recordLocationHoodieKeyPair . get ( ) . _1 ( ) ) ; deleteRecord . seal ( ) ; HoodieRecord < T > insertRecord = HoodieIndexUtils . getTaggedRecord ( hoodieRecord , Option . empty ( ) ) ; return Arrays . asList ( deleteRecord , insertRecord ) . iterator ( ) ; } else { return Collections . singletonList ( ( HoodieRecord < T > ) HoodieIndexUtils . getTaggedRecord ( new HoodieRecord < > ( recordLocationHoodieKeyPair . get ( ) . _2 , hoodieRecord . getData ( ) ) , Option . ofNullable ( recordLocationHoodieKeyPair . get ( ) . _1 ) ) ) . iterator ( ) ; } } else { return Collections . singletonList ( ( HoodieRecord < T > ) HoodieIndexUtils . getTaggedRecord ( hoodieRecord , Option . empty ( ) ) ) . iterator ( ) ; } } ) ; }
Tagging for global index should only consider the record key.
Tagging for global index should only consider the record key .
// Here as the recordRDD might have more data than rowKeyRDD (some rowKeys' fileId is null), so we do left outer join. // Record key matched to file // Create an empty record to delete the record in the old partition // Tag the incoming record for inserting to the new partition // Ignore the incoming record's partition, regardless of whether it differs from its old partition or not. // When it differs, the record will still be updated at its old partition.