Core: Avoid per-row path rehashing in toPositionIndexes

[kinolaev]

CharSequenceMap.computeIfAbsent computes a hash code for the provided key on every call. Because rows in a position delete file are sorted by file_path, we only need to call computeIfAbsent when the key changes.

Benchmark                                      (numDataFiles)  (numDeletes)  Mode  Cnt  Score    Error  Units
ToPositionIndexesBenchmark.computeIfAbsent                  1       1000000    ss    5  0.310 ±  0.002   s/op
ToPositionIndexesBenchmark.computeIfAbsent                 10       1000000    ss    5  0.314 ±  0.012   s/op
ToPositionIndexesBenchmark.computeIfAbsent                100       1000000    ss    5  0.311 ±  0.009   s/op
ToPositionIndexesBenchmark.computeIfAbsent               1000       1000000    ss    5  0.133 ±  0.008   s/op
ToPositionIndexesBenchmark.computeIfAbsent              10000       1000000    ss    5  0.316 ±  0.028   s/op
ToPositionIndexesBenchmark.computeIfAbsent             100000       1000000    ss    5  0.196 ±  0.050   s/op
ToPositionIndexesBenchmark.computeIfAbsent            1000000       1000000    ss    5  0.748 ±  0.578   s/op
ToPositionIndexesBenchmark.stringEquals                     1       1000000    ss    5  0.023 ±  0.050   s/op
ToPositionIndexesBenchmark.stringEquals                    10       1000000    ss    5  0.016 ±  0.001   s/op
ToPositionIndexesBenchmark.stringEquals                   100       1000000    ss    5  0.017 ±  0.001   s/op
ToPositionIndexesBenchmark.stringEquals                  1000       1000000    ss    5  0.019 ±  0.011   s/op
ToPositionIndexesBenchmark.stringEquals                 10000       1000000    ss    5  0.027 ±  0.025   s/op
ToPositionIndexesBenchmark.stringEquals                100000       1000000    ss    5  0.068 ±  0.027   s/op
ToPositionIndexesBenchmark.stringEquals               1000000       1000000    ss    5  0.622 ±  0.519   s/op
ToPositionIndexesBenchmark.comparatorsFilePath              1       1000000    ss    5  0.151 ±  0.007   s/op
ToPositionIndexesBenchmark.comparatorsFilePath             10       1000000    ss    5  0.151 ±  0.005   s/op
ToPositionIndexesBenchmark.comparatorsFilePath            100       1000000    ss    5  0.153 ±  0.001   s/op
ToPositionIndexesBenchmark.comparatorsFilePath           1000       1000000    ss    5  0.153 ±  0.004   s/op
ToPositionIndexesBenchmark.comparatorsFilePath          10000       1000000    ss    5  0.277 ±  0.012   s/op
ToPositionIndexesBenchmark.comparatorsFilePath         100000       1000000    ss    5  0.193 ±  0.017   s/op
ToPositionIndexesBenchmark.comparatorsFilePath        1000000       1000000    ss    5  0.612 ±  0.378   s/op

ToPositionIndexesBenchmark.java

package org.apache.iceberg.deletes;

import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.List;
import java.util.Locale;
import java.util.concurrent.TimeUnit;
import org.apache.iceberg.Accessor;
import org.apache.iceberg.DeleteFile;
import org.apache.iceberg.MetadataColumns;
import org.apache.iceberg.Schema;
import org.apache.iceberg.StructLike;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.io.DeleteSchemaUtil;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.types.Comparators;
import org.apache.iceberg.util.CharSequenceMap;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
import org.openjdk.jmh.annotations.Fork;
import org.openjdk.jmh.annotations.Measurement;
import org.openjdk.jmh.annotations.Mode;
import org.openjdk.jmh.annotations.Param;
import org.openjdk.jmh.annotations.Scope;
import org.openjdk.jmh.annotations.Setup;
import org.openjdk.jmh.annotations.State;
import org.openjdk.jmh.annotations.Threads;
import org.openjdk.jmh.annotations.Timeout;
import org.openjdk.jmh.annotations.Warmup;
import org.openjdk.jmh.infra.Blackhole;

@Fork(1)
@State(Scope.Benchmark)
@Warmup(iterations = 3)
@Measurement(iterations = 5)
@BenchmarkMode(Mode.SingleShotTime)
@Timeout(time = 10, timeUnit = TimeUnit.MINUTES)
public class ToPositionIndexesBenchmark {
  private static final Schema POS_DELETE_SCHEMA = DeleteSchemaUtil.pathPosSchema();
  private static final Accessor<StructLike> FILENAME_ACCESSOR =
      POS_DELETE_SCHEMA.accessorForField(MetadataColumns.DELETE_FILE_PATH.fieldId());
  private static final Accessor<StructLike> POSITION_ACCESSOR =
      POS_DELETE_SCHEMA.accessorForField(MetadataColumns.DELETE_FILE_POS.fieldId());

  @Param("1000000")
  private int numDeletes;

  @Param({"1", "10", "100", "1000", "10000", "100000", "1000000"})
  private int numDataFiles;

  private List<StructLike> deletes;
  private String pathTemplate =
      "s3://pretty-long-bucket-name/pretty-long-namespace-name/pretty-long-table-name/data/%07d-data.parquet";

  @Setup
  public void setupBenchmark() {
    deletes = Lists.newArrayListWithExpectedSize(numDeletes);
    int numDeletesPerDataFile = numDeletes / numDataFiles;
    for (int index = 0; index < numDeletes; index++) {
      String path = String.format(Locale.ROOT, pathTemplate, index / numDeletesPerDataFile);
      deletes.add(new PositionDeleteRow(path, index % numDeletesPerDataFile));
    }
  }

  @Benchmark
  @Threads(1)
  public void computeIfAbsent(Blackhole blackhole) {
    blackhole.consume(toPositionIndexesComputeIfAbsent(CloseableIterable.withNoopClose(deletes), null));
  }

  // https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/core/src/main/java/org/apache/iceberg/deletes/Deletes.java#L139
  private static <T extends StructLike> CharSequenceMap<PositionDeleteIndex> toPositionIndexesComputeIfAbsent(
      CloseableIterable<T> posDeletes, DeleteFile file) {
    CharSequenceMap<PositionDeleteIndex> indexes = CharSequenceMap.create();

    try (CloseableIterable<T> deletes = posDeletes) {
      for (T delete : deletes) {
        CharSequence filePath = (CharSequence) FILENAME_ACCESSOR.get(delete);
        long position = (long) POSITION_ACCESSOR.get(delete);
        PositionDeleteIndex index =
            indexes.computeIfAbsent(filePath, key -> new BitmapPositionDeleteIndex(file));
        index.delete(position);
      }
    } catch (IOException e) {
      throw new UncheckedIOException("Failed to close position delete source", e);
    }

    return indexes;
  }
  @Benchmark
  @Threads(1)
  public void comparatorsFilePath(Blackhole blackhole) {
    blackhole.consume(
        toPositionIndexesComparatorsFilePath(CloseableIterable.withNoopClose(deletes), null));
  }

  public static <T extends StructLike>
      CharSequenceMap<PositionDeleteIndex> toPositionIndexesComparatorsFilePath(
          CloseableIterable<T> posDeletes, DeleteFile file) {
    CharSequenceMap<PositionDeleteIndex> indexes = CharSequenceMap.create();

    try (CloseableIterable<T> deletes = posDeletes) {
      CharSequence lastFilePath = null;
      PositionDeleteIndex index = null;
      for (T delete : deletes) {
        CharSequence filePath = (CharSequence) FILENAME_ACCESSOR.get(delete);
        long position = (long) POSITION_ACCESSOR.get(delete);
        if (lastFilePath == null || Comparators.filePath().compare(lastFilePath, filePath) != 0) {
          lastFilePath = filePath;
          index = indexes.computeIfAbsent(filePath, key -> new BitmapPositionDeleteIndex(file));
        }
        index.delete(position);
      }
    } catch (IOException e) {
      throw new UncheckedIOException("Failed to close position delete source", e);
    }

    return indexes;
  }

  @Benchmark
  @Threads(1)
  public void stringEquals(Blackhole blackhole) {
    blackhole.consume(
        toPositionIndexesStringEquals(CloseableIterable.withNoopClose(deletes), null));
  }

  public static <T extends StructLike>
      CharSequenceMap<PositionDeleteIndex> toPositionIndexesStringEquals(
          CloseableIterable<T> posDeletes, DeleteFile file) {
    CharSequenceMap<PositionDeleteIndex> indexes = CharSequenceMap.create();

    try (CloseableIterable<T> deletes = posDeletes) {
      String lastFilePath = null;
      PositionDeleteIndex index = null;
      for (T delete : deletes) {
        CharSequence filePath = ((CharSequence) FILENAME_ACCESSOR.get(delete));
        long position = (long) POSITION_ACCESSOR.get(delete);
        if (lastFilePath == null || !lastFilePath.equals(filePath)) {
          lastFilePath = filePath.toString();
          index = indexes.computeIfAbsent(filePath, key -> new BitmapPositionDeleteIndex(file));
        }
        index.delete(position);
      }
    } catch (IOException e) {
      throw new UncheckedIOException("Failed to close position delete source", e);
    }

    return indexes;
  }

  private record PositionDeleteRow(String path, long position) implements StructLike {
    @Override
    public int size() {
      return 2;
    }

    @Override
    @SuppressWarnings("unchecked")
    public <T> T get(int pos, Class<T> javaClass) {
      switch (pos) {
        case 0:
          return (T) path;
        case 1:
          return (T) (Long) position;
        default:
          throw new UnsupportedOperationException("Unsupported position: " + pos);
      }
    }

    @Override
    public <T> void set(int pos, T value) {
      throw new UnsupportedOperationException("Not supported");
    }
  }
}

Related to #11648.
Extracted from #15714.

@anuragmantri