processing/src/main/java/org/apache/druid/segment/IndexMergerV9.java
/**
   * Returns rowNumConversions, if fillRowNumConversions argument is true
   */
Returns rowNumConversions, if fillRowNumConversions argument is true
Returns rowNumConversions if fillRowNumConversions argument is true
private IndexMergeResult mergeIndexesAndWriteColumns(
      final List<IndexableAdapter> adapters,
      final ProgressIndicator progress,
      final TimeAndDimsIterator timeAndDimsIterator,
      final GenericColumnSerializer timeWriter,
      final ArrayList<GenericColumnSerializer> metricWriters,
      final List<DimensionMergerV9> mergers,
      final boolean fillRowNumConversions
  ) throws IOException
  {
    final String section = "walk through and merge rows";
    progress.startSection(section);
    long startTime = System.currentTimeMillis();
    List<IntBuffer> rowNumConversions = null;
    int rowCount = 0;
    if (fillRowNumConversions) {
      rowNumConversions = new ArrayList<>(adapters.size());
      for (IndexableAdapter adapter : adapters) {
        int[] arr = new int[adapter.getNumRows()];
        Arrays.fill(arr, INVALID_ROW);
        rowNumConversions.add(IntBuffer.wrap(arr));
      }
    }
    long time = System.currentTimeMillis();
    while (timeAndDimsIterator.moveToNext()) {
      progress.progress();
      TimeAndDimsPointer timeAndDims = timeAndDimsIterator.getPointer();
      timeWriter.serialize(timeAndDims.timestampSelector);
      for (int metricIndex = 0; metricIndex < timeAndDims.getNumMetrics(); metricIndex++) {
        metricWriters.get(metricIndex).serialize(timeAndDims.getMetricSelector(metricIndex));
      }
      for (int dimIndex = 0; dimIndex < timeAndDims.getNumDimensions(); dimIndex++) {
        DimensionMerger merger = mergers.get(dimIndex);
        if (merger.hasOnlyNulls()) {
          continue;
        }
        merger.processMergedRow(timeAndDims.getDimensionSelector(dimIndex));
      }
      if (timeAndDimsIterator instanceof RowCombiningTimeAndDimsIterator) {
        RowCombiningTimeAndDimsIterator comprisedRows = (RowCombiningTimeAndDimsIterator) timeAndDimsIterator;
        for (int originalIteratorIndex = comprisedRows.nextCurrentlyCombinedOriginalIteratorIndex(0);
             originalIteratorIndex >= 0;
             originalIteratorIndex =
                 comprisedRows.nextCurrentlyCombinedOriginalIteratorIndex(originalIteratorIndex + 1)) {
          IntBuffer conversionBuffer = rowNumConversions.get(originalIteratorIndex);
          int minRowNum = comprisedRows.getMinCurrentlyCombinedRowNumByOriginalIteratorIndex(originalIteratorIndex);
          int maxRowNum = comprisedRows.getMaxCurrentlyCombinedRowNumByOriginalIteratorIndex(originalIteratorIndex);
          for (int rowNum = minRowNum; rowNum <= maxRowNum; rowNum++) {
            while (conversionBuffer.position() < rowNum) {
              conversionBuffer.put(INVALID_ROW);
            }
            conversionBuffer.put(rowCount);
          }
        }
      } else if (timeAndDimsIterator instanceof MergingRowIterator) {
        RowPointer rowPointer = (RowPointer) timeAndDims;
        IntBuffer conversionBuffer = rowNumConversions.get(rowPointer.getIndexNum());
        int rowNum = rowPointer.getRowNum();
        while (conversionBuffer.position() < rowNum) {
          conversionBuffer.put(INVALID_ROW);
        }
        conversionBuffer.put(rowCount);
      } else {
        if (fillRowNumConversions) {
          throw new IllegalStateException(
              "Filling row num conversions is supported only with RowCombining and Merging iterators"
          );
        }
      }
      if ((++rowCount % 500000) == 0) {
        log.debug("walked 500,000/%d rows in %,d millis.", rowCount, System.currentTimeMillis() - time);
        time = System.currentTimeMillis();
      }
    }
    if (rowNumConversions != null) {
      for (IntBuffer rowNumConversion : rowNumConversions) {
        rowNumConversion.rewind();
      }
    }
    log.debug("completed walk through of %,d rows in %,d millis.", rowCount, System.currentTimeMillis() - startTime);
    progress.stopSection(section);
    return new IndexMergeResult(rowNumConversions, rowCount);
  }
private IndexMergeResult mergeIndexesAndWriteColumns ( final List < IndexableAdapter > adapters , final ProgressIndicator progress , final TimeAndDimsIterator timeAndDimsIterator , final GenericColumnSerializer timeWriter , final ArrayList < GenericColumnSerializer > metricWriters , final List < DimensionMergerV9 > mergers , final boolean fillRowNumConversions ) throws IOException { final String section = "walk through and merge rows" ; progress . startSection ( section ) ; long startTime = System . currentTimeMillis ( ) ; List < IntBuffer > rowNumConversions = null ; int rowCount = 0 ; if ( fillRowNumConversions ) { rowNumConversions = new ArrayList < > ( adapters . size ( ) ) ; for ( IndexableAdapter adapter : adapters ) { int [ ] arr = new int [ adapter . getNumRows ( ) ] ; Arrays . fill ( arr , INVALID_ROW ) ; rowNumConversions . add ( IntBuffer . wrap ( arr ) ) ; } } long time = System . currentTimeMillis ( ) ; while ( timeAndDimsIterator . moveToNext ( ) ) { progress . progress ( ) ; TimeAndDimsPointer timeAndDims = timeAndDimsIterator . getPointer ( ) ; timeWriter . serialize ( timeAndDims . timestampSelector ) ; for ( int metricIndex = 0 ; metricIndex < timeAndDims . getNumMetrics ( ) ; metricIndex ++ ) { metricWriters . get ( metricIndex ) . serialize ( timeAndDims . getMetricSelector ( metricIndex ) ) ; } for ( int dimIndex = 0 ; dimIndex < timeAndDims . getNumDimensions ( ) ; dimIndex ++ ) { DimensionMerger merger = mergers . get ( dimIndex ) ; if ( merger . hasOnlyNulls ( ) ) { continue ; } merger . processMergedRow ( timeAndDims . getDimensionSelector ( dimIndex ) ) ; } if ( timeAndDimsIterator instanceof RowCombiningTimeAndDimsIterator ) { RowCombiningTimeAndDimsIterator comprisedRows = ( RowCombiningTimeAndDimsIterator ) timeAndDimsIterator ; for ( int originalIteratorIndex = comprisedRows . nextCurrentlyCombinedOriginalIteratorIndex ( 0 ) ; originalIteratorIndex >= 0 ; originalIteratorIndex = comprisedRows . nextCurrentlyCombinedOriginalIteratorIndex ( originalIteratorIndex + 1 ) ) { IntBuffer conversionBuffer = rowNumConversions . get ( originalIteratorIndex ) ; int minRowNum = comprisedRows . getMinCurrentlyCombinedRowNumByOriginalIteratorIndex ( originalIteratorIndex ) ; int maxRowNum = comprisedRows . getMaxCurrentlyCombinedRowNumByOriginalIteratorIndex ( originalIteratorIndex ) ; for ( int rowNum = minRowNum ; rowNum <= maxRowNum ; rowNum ++ ) { while ( conversionBuffer . position ( ) < rowNum ) { conversionBuffer . put ( INVALID_ROW ) ; } conversionBuffer . put ( rowCount ) ; } } } else if ( timeAndDimsIterator instanceof MergingRowIterator ) { RowPointer rowPointer = ( RowPointer ) timeAndDims ; IntBuffer conversionBuffer = rowNumConversions . get ( rowPointer . getIndexNum ( ) ) ; int rowNum = rowPointer . getRowNum ( ) ; while ( conversionBuffer . position ( ) < rowNum ) { conversionBuffer . put ( INVALID_ROW ) ; } conversionBuffer . put ( rowCount ) ; } else { if ( fillRowNumConversions ) { throw new IllegalStateException ( "Filling row num conversions is supported only with RowCombining and Merging iterators" ) ; } } if ( ( ++ rowCount % 500000 ) == 0 ) { log . debug ( "walked 500,000/%d rows in %,d millis." , rowCount , System . currentTimeMillis ( ) - time ) ; time = System . currentTimeMillis ( ) ; } } if ( rowNumConversions != null ) { for ( IntBuffer rowNumConversion : rowNumConversions ) { rowNumConversion . rewind ( ) ; } } log . debug ( "completed walk through of %,d rows in %,d millis." , rowCount , System . currentTimeMillis ( ) - startTime ) ; progress . stopSection ( section ) ; return new IndexMergeResult ( rowNumConversions , rowCount ) ; }
Returns rowNumConversions, if fillRowNumConversions argument is true
Returns rowNumConversions if fillRowNumConversions argument is true