Iceberg/Comet integration POC

[huaxingao]

This PR shows how I will integrate Comet with iceberg. The PR doesn't compile yet because we haven't released Comet yet, but it shows the ideas how we are going to change iceberg code to integrate Comet. Also, Comet doesn't have Spark3.5 support yet so I am doing this on 3.4, but we will add 3.5 support in Comet.

In VectorizedSparkParquetReaders.buildReader, if Comet library is available, a CometIcebergColumnarBatchReader will be created, which will use Comet batch reader to read data. We can also add a property later to control whether we want to use Comet or not.

The logic in CometIcebergVectorizedReaderBuilder is very similar to VectorizedReaderBuilder. It builds Comet column reader instead of iceberg column reader.

The delete logic in CometIcebergColumnarBatchReader is exactly the same as the one in ColumnarBatchReader. I will extract the common code and put the common code in a base class.

The main motivation of this PR is to improve performance using native execution. Comet's Parquet reader is a hybrid implementation: IO and decompression are done in the JVM while decoding is done natively. There is some performance gain from native decoding, but the gain is not much. However, by switching to the Comet Parquet reader, Comet will recognize that this is a Comet scan and will convert the Spark physical plan into a Comet plan for native execution. The major performance gain will be from this native execution.

[huaxingao]

cc @aokolnychyi @sunchao

[aokolnychyi]

I think this is the right direction to take. I did an initial high-level pass. Looking forward to having a Comet release soon.

[aokolnychyi]

I assume this library will only contain the reader, not the operators.

[huaxingao]

Right. This only contains the reader.

[RussellSpitzer]

Does it need to be Spark Version Dependent? Just wondering

[huaxingao]

We are currently doing some experiments to see if we can provide a Spark Version independent jar.

[aokolnychyi]

+1 for exploring that.

[aokolnychyi]

I hope we can soon have a snapshot for Comet jar independent of Spark to clean up deps here.
We can't have parquet module depend on a jar with any Spark deps.

[aokolnychyi]

+1 for exploring that.

[aokolnychyi]

These changes would require a bit more time to review. I'll do that tomorrow. I think we would want to restructure the original implementation a bit. Not a concern for now.

[aokolnychyi]

We would want to structure this a bit differently. Let me think more.

[huaxingao]

@aokolnychyi I have addressed the comments. Could you please take one more look when you have a moment? Thanks a lot!

[aokolnychyi]

Will check today.

[aokolnychyi]

We would want to structure this a bit differently. Let me think more.

[cornelcreanga]

@huaxingao - Hi, is the Comet Parquet reader able to support page skipping/use page indexes? -eg see #193 for the Iceberg Parquet reader initial issue.

[huaxingao]

@cornelcreanga Comet Parquet reader doesn't support page skipping yet

[PaulLiang1]

hey @huaxingao
we are really interested in this feature, just wonder what can we help to getting this integrated?

[huaxingao]

@PaulLiang1 Thank you for your interest! We are currently working on a binary release of DataFusion Comet. Once the binary release is available, I will proceed with this PR.

[PaulLiang1]

@huaxingao
I think we got a internal version of building DataFusion comet and publish a JAR internally.
Is there anything we can help with on that front?

Thanks

[huaxingao]

@PaulLiang1 Thanks! I'll check with my colleague tomorrow to find out where we are in the binary release process.

[huaxingao]

@PaulLiang1 We are pretty close to this and will have a binary release for Comet soon.

[PaulLiang1]

@PaulLiang1 Thanks! I'll check with my colleague tomorrow to find out where we are in the binary release process.

got it, thanks for letting me know. please feel free to let us know if there is anything we could help on. thanks!

[aokolnychyi]

I think we inherit the existing loading logic that is too complicated. First, we mix isDeleted and rowIdMapping cases. Second, we create ColumnarBatch prior to having all column vectors (e.g. isDeleted array).

What we add methods like this to the util class instead? Only one of them will be needed in a query, right? We either mark records as removed or hide them.

public static Pair<int[], Integer> buildRowIdMapping(
    ColumnVector[] vectors,
    DeleteFilter<InternalRow> deletes,
    long rowStartPosInBatch,
    int batchSize) {
  if (deletes == null) {
    return null;
  }

  PositionDeleteIndex deletedPositions = deletes.deletedRowPositions();
  Predicate<InternalRow> eqDeleteFilter = deletes.eqDeletedRowFilter();
  ColumnarBatchRow row = new ColumnarBatchRow(vectors);
  int[] rowIdMapping = new int[batchSize];
  int liveRowId = 0;

  for (int rowId = 0; rowId < batchSize; rowId++) {
    long pos = rowStartPosInBatch + rowId;
    row.rowId = rowId;
    if (isDeleted(pos, row, deletedPositions, eqDeleteFilter)) {
      deletes.incrementDeleteCount();
    } else {
      rowIdMapping[liveRowId] = rowId;
      liveRowId++;
    }
  }

  return liveRowId == batchSize ? null : Pair.of(rowIdMapping, liveRowId);
}

public static boolean[] buildIsDeleted(
    ColumnVector[] vectors,
    DeleteFilter<InternalRow> deletes,
    long rowStartPosInBatch,
    int batchSize) {
  boolean[] isDeleted = new boolean[batchSize];

  if (deletes == null) {
    return isDeleted;
  }

  PositionDeleteIndex deletedPositions = deletes.deletedRowPositions();
  Predicate<InternalRow> eqDeleteFilter = deletes.eqDeletedRowFilter();
  ColumnarBatchRow row = new ColumnarBatchRow(vectors);

  for (int rowId = 0; rowId < batchSize; rowId++) {
    long pos = rowStartPosInBatch + rowId;
    row.rowId = rowId;
    isDeleted[rowId] = isDeleted(pos, row, deletedPositions, eqDeleteFilter);
  }

  return isDeleted;
}

// use separate if statements to reduce the chance of speculative execution for equality tests
private static boolean isDeleted(
    long pos,
    InternalRow row,
    PositionDeleteIndex deletedPositions,
    Predicate<InternalRow> eqDeleteFilter) {
  if (deletedPositions != null && deletedPositions.isDeleted(pos)) {
    return true;
  }

  if (!eqDeleteFilter.test(row)) {
    return true;
  }

  return false;
}

Then our loading logic can look like:

• Initialize the vector array.
• Load all data vectors (leaving metadata vectors as null).
• If you need to discard deleted records, call buildRowIdMapping and either wrap loaded data vectors into other vectors or mutate them in place via setRowIdMapping.
• If you need to mark deleted records, call buildIsDeleted to compute the flags.
• Load all metadata vectors (we will have the is_deleted array fully populated now).