Iceberg/Comet integration POC #9841

huaxingao · 2024-03-01T02:34:25Z

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 · 2024-03-01T02:41:26Z

cc @aokolnychyi @sunchao

...k/src/main/java/org/apache/iceberg/spark/data/vectorized/comet/CometIcebergColumnReader.java

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.

...k/src/main/java/org/apache/iceberg/spark/data/vectorized/comet/CometIcebergColumnReader.java

aokolnychyi · 2024-04-16T03:57:07Z

spark/v3.4/build.gradle

    }

+    compileOnly "org.apache.comet:comet-spark-spark${sparkMajorVersion}_${scalaVersion}:0.1.0-SNAPSHOT"


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

Right. This only contains the reader.

Does it need to be Spark Version Dependent? Just wondering

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

+1 for exploring that.

...ain/java/org/apache/iceberg/spark/data/vectorized/comet/CometIcebergColumnarBatchReader.java

...rk/src/main/java/org/apache/iceberg/spark/data/vectorized/VectorizedSparkParquetReaders.java

...in/java/org/apache/iceberg/spark/data/vectorized/comet/CometIcebergPositionColumnReader.java

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/SparkConfParser.java

...v3.4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/BaseColumnBatchLoader.java

....4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/comet/CometColumnReader.java

api/src/main/java/org/apache/iceberg/ReaderType.java

aokolnychyi · 2024-04-22T22:27:03Z

build.gradle

@@ -45,6 +45,7 @@ buildscript {
  }
 }

+String sparkMajorVersion = '3.4'


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.

spark/v3.4/build.gradle

aokolnychyi · 2024-04-22T22:27:57Z

spark/v3.4/build.gradle

    }

+    compileOnly "org.apache.comet:comet-spark-spark${sparkMajorVersion}_${scalaVersion}:0.1.0-SNAPSHOT"


+1 for exploring that.

gradle.properties

aokolnychyi · 2024-04-23T00:54:35Z

...v3.4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/BaseColumnBatchLoader.java

+import org.apache.spark.sql.vectorized.ColumnVector;
+import org.apache.spark.sql.vectorized.ColumnarBatch;
+
+@SuppressWarnings("checkstyle:VisibilityModifier")


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.

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

...rk/src/main/java/org/apache/iceberg/spark/data/vectorized/VectorizedSparkParquetReaders.java

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkColumnarReaderFactory.java

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/BaseBatchReader.java

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatch.java

huaxingao · 2024-04-29T16:59:05Z

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

aokolnychyi · 2024-04-30T17:27:41Z

Will check today.

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/ParquetReaderType.java

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java

aokolnychyi · 2024-04-30T19:04:36Z

...v3.4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/BaseColumnBatchLoader.java

+import org.apache.spark.sql.vectorized.ColumnVector;
+import org.apache.spark.sql.vectorized.ColumnarBatch;
+
+@SuppressWarnings("checkstyle:VisibilityModifier")


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

...k/v3.4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java

....4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/comet/CometColumnReader.java

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/BaseBatchReader.java

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatch.java

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkColumnarReaderFactory.java

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/BatchReadConf.java

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/SparkReadConf.java

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/CometColumnReader.java

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/BaseBatchReader.java

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/BatchDataReader.java

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatch.java

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/SparkReadConf.java

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/BatchReadConf.java

cornelcreanga · 2024-06-20T14:09:01Z

@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 · 2024-06-20T15:41:53Z

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

PaulLiang1 · 2024-09-04T04:13:51Z

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

huaxingao · 2024-09-04T04:25:24Z

@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 · 2024-09-04T04:53:39Z

@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 · 2024-09-04T05:24:49Z

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

huaxingao · 2024-09-05T04:50:30Z

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

PaulLiang1 · 2024-09-05T05:00:49Z

@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!

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/ParquetReaderType.java

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/CometColumnReader.java

.../spark/src/main/java/org/apache/iceberg/spark/data/vectorized/CometConstantColumnReader.java

aokolnychyi · 2025-01-29T05:22:24Z

.../spark/src/main/java/org/apache/iceberg/spark/data/vectorized/CometConstantColumnReader.java

+    initialized = true;
+  }
+
+  private Object convertToSparkValue(T value) {


This is a fragile place. I'll have to come back with fresh eyes.

These conversions look correct to me - they match the internal type specified in Spark for these datatypes

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/CometColumnReader.java

...4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/CometColumnarBatchReader.java

aokolnychyi

I did a detailed pass. My biggest questions:

Depending directly on shaded classes from Comet. I think this should be hidden by Comet APIs. Not sure it is a blocker, though.
Iceberg reader should be the default and tests have to be parameterized to support Comet. Where tests fail, there must be no correctness issue. Ideally, we would detect that the read cannot be handled by Comet in SparkBatch.

I have very limited understanding on how Comet works so I can't review that part and rely on @huaxingao. I focused primarily on how the integration is done and impact to the existing reader path. @sunchao @parthchandra, you are welcome to review those parts.

aokolnychyi · 2025-01-29T17:15:28Z

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/CometColumnReader.java

+class CometColumnReader implements VectorizedReader<CometVector> {
+  public static final int DEFAULT_BATCH_SIZE = 5000;
+
+  private final DataType sparkType;


Final vars should be grouped.

Fixed. Thanks

huaxingao · 2025-01-29T17:38:31Z

Thanks a lot @aokolnychyi for your detailed review! I will

fix the shade problem
change the default to iceberg in the final version. I default to Comet only for testing purpose to make sure all the tests will pass with Comet.
@parthchandra will help to review the Comet part today.

aokolnychyi · 2025-01-29T18:14:02Z

Sounds good. Other than what was mentioned in the review, the change looks good to me.

We will have to adapt and run our JMH benchmarks as well. This can be done in a separate PR.

parthchandra · 2025-01-29T20:28:53Z

.baseline/checkstyle/checkstyle-suppressions.xml

@@ -48,4 +48,7 @@

    <!-- Referencing guava classes should be allowed in classes within bundled-guava module -->
    <suppress files="org.apache.iceberg.GuavaClasses" id="BanUnrelocatedGuavaClasses"/>
+


@huaxingao can you log an issue in Comet to address this? CometSchemaImporter is a Comet class but is in the org.apache.arrow.c package to overcome access restrictions (Arrow's SchemaImporter is package private). We can create a wrapper class to access the schema importer.
Also, we should ideally use the allocator from BatchReader, but that too can be in the wrapper class, I think. There is no issue with using a new allocator for each column, but the arrow allocator has powerful features in memory accounting that we can take advantage of down the road.

parthchandra · 2025-01-29T20:34:50Z

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/CometColumnReader.java

+      delegate.close();
+    }
+
+    CometSchemaImporter importer = new CometSchemaImporter(new RootAllocator());


Ideally, we should call close on the importer after the column reader is done.

Fixed. Thanks!

parthchandra · 2025-01-29T20:46:40Z

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/CometColumnReader.java

+  private final CometVector vector;
+  private final ColumnDescriptor descriptor;
+  private boolean initialized = false;
+  private int batchSize = DEFAULT_BATCH_SIZE;


Should this natch the Comet batch size? https://github.com/apache/datafusion-comet/blob/e9649477c4f8b4c6906244c3cc6828b83f32f735/common/src/main/scala/org/apache/comet/CometConf.scala#L492

I originally matched the default batch size to Iceberg's default batch size, but you are right, it makes more sense to match to Comet default size, so I have changed it to Comet default size

parthchandra · 2025-01-29T20:47:29Z

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/CometColumnReader.java

+import org.apache.spark.sql.types.StructField;
+
+class CometColumnReader implements VectorizedReader<CometVector> {
+  public static final int DEFAULT_BATCH_SIZE = 5000;


Should this match the Comet batch size? https://github.com/apache/datafusion-comet/blob/e9649477c4f8b4c6906244c3cc6828b83f32f735/common/src/main/scala/org/apache/comet/CometConf.scala#L492

parthchandra · 2025-01-29T20:51:10Z

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/CometColumnReader.java

+
+  private final DataType sparkType;
+  // the delegated column reader from Comet side
+  private AbstractColumnReader delegate;


The interaction between CometColumnarBatchReader.delegate and CometColumnReader.delegate` is a little confusing. A comment explaining it would be useful.

I added comments in CometColumnarBatchReader

parthchandra · 2025-01-29T20:53:40Z

...4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/CometColumnarBatchReader.java

+  private final CometColumnReader[] readers;
+  private final boolean hasIsDeletedColumn;
+  // The delegated batch reader on Comet side
+  private final BatchReader delegate;


Why do we have this and not use the nextBatch call directly (instead we are explicitly calling readBatch on each column reader). A comment to explain why would be helpful.

Thank you for adding the comment !

parthchandra · 2025-01-29T21:02:24Z

.../spark/src/main/java/org/apache/iceberg/spark/data/vectorized/CometConstantColumnReader.java

+    initialized = true;
+  }
+
+  private Object convertToSparkValue(T value) {


These conversions look correct to me - they match the internal type specified in Spark for these datatypes

parthchandra · 2025-01-29T21:06:10Z

.../spark/src/main/java/org/apache/iceberg/spark/data/vectorized/CometConstantColumnReader.java

+    if (dataType == DataTypes.StringType && value instanceof String) {
+      return UTF8String.fromString((String) value);
+    } else if (dataType instanceof DecimalType && value instanceof BigDecimal) {
+      return Decimal.apply((BigDecimal) value);


This looked dangerous at first. Maybe a comment to clarify that this is the Spark Decimal class (which can handle BigDecimal precision)?

Comments are added. Thanks!

parthchandra

One more change needed, otherwise Comet side looks good to me.

parthchandra · 2025-01-30T01:25:38Z

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/CometColumnReader.java

+
+class CometColumnReader implements VectorizedReader<ColumnVector> {
+  // use the Comet default batch size
+  public static final int DEFAULT_BATCH_SIZE = 8192;


+1. In some follow up PR we can try to pass in the configured value (not just the default).

parthchandra · 2025-01-30T01:25:55Z

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/CometColumnReader.java

+  private final CometVector vector;
+  private final ColumnDescriptor descriptor;
+  private boolean initialized = false;
+  private int batchSize = DEFAULT_BATCH_SIZE;


parthchandra · 2025-01-30T01:27:54Z

...4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/CometColumnarBatchReader.java

+  private final CometColumnReader[] readers;
+  private final boolean hasIsDeletedColumn;
+  // The delegated batch reader on Comet side
+  private final BatchReader delegate;


Thank you for adding the comment !

parthchandra · 2025-01-30T01:31:58Z

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/CometColumnReader.java

+   * groups.
+   */
+  public void reset() {
+    if (delegate != null) {


If delegate is not null here importer will also be not null. I think the importer can be reused, so there is no need to close it here, but we overwrite it in the next line, so either we should reuse it, or close it. (Closing it is safer)

Done. Thanks!

aokolnychyi · 2025-01-31T21:31:37Z

I am OK merging the change if we revert the default reader type and Comet experts approve the logic. I won't block this work because of the dependency on shaded APIs.

We will need to parameterize tests to test both reader types and run JMH benchmarks in the future.

parthchandra

Comet side changes look good

aokolnychyi · 2025-01-31T23:32:46Z

Thanks, @huaxingao! Thanks for reviewing, @parthchandra!

huaxingao · 2025-01-31T23:39:45Z

@aokolnychyi Thank you so much for your reviewing and merge this PR! Also thanks @parthchandra and @RussellSpitzer for reviewing!

github-actions bot added spark build labels Mar 1, 2024

huaxingao mentioned this pull request Mar 1, 2024

Dynamically support Spark native engine in Iceberg #9826

Closed

huaxingao mentioned this pull request Mar 5, 2024

Dynamically support Spark native engine in Iceberg #9721

Closed

sunchao mentioned this pull request Mar 7, 2024

Explore integration with Delta Lake apache/datafusion-comet#174

Open

RussellSpitzer reviewed Apr 2, 2024

View reviewed changes

...k/src/main/java/org/apache/iceberg/spark/data/vectorized/comet/CometIcebergColumnReader.java Outdated Show resolved Hide resolved

aokolnychyi reviewed Apr 16, 2024

View reviewed changes

github-actions bot added the API label Apr 18, 2024

RussellSpitzer reviewed Apr 18, 2024

View reviewed changes

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/SparkConfParser.java Outdated Show resolved Hide resolved

RussellSpitzer reviewed Apr 18, 2024

View reviewed changes

...v3.4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/BaseColumnBatchLoader.java Outdated Show resolved Hide resolved

RussellSpitzer reviewed Apr 18, 2024

View reviewed changes

....4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/comet/CometColumnReader.java Outdated Show resolved Hide resolved

aokolnychyi reviewed Apr 23, 2024

View reviewed changes

github-actions bot removed the API label Apr 26, 2024

aokolnychyi reviewed Apr 30, 2024

View reviewed changes

aokolnychyi reviewed May 3, 2024

View reviewed changes

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/SparkReadConf.java Outdated Show resolved Hide resolved

aokolnychyi reviewed May 9, 2024

View reviewed changes

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/BatchReadConf.java Outdated Show resolved Hide resolved

huaxingao closed this Jun 20, 2024

huaxingao reopened this Jun 20, 2024

aokolnychyi reviewed Jan 29, 2025

View reviewed changes

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/ParquetReaderType.java Outdated Show resolved Hide resolved

aokolnychyi reviewed Jan 29, 2025

View reviewed changes

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/CometColumnReader.java Outdated Show resolved Hide resolved

aokolnychyi reviewed Jan 29, 2025

View reviewed changes

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/CometColumnReader.java Outdated Show resolved Hide resolved

aokolnychyi reviewed Jan 29, 2025

View reviewed changes

.../spark/src/main/java/org/apache/iceberg/spark/data/vectorized/CometConstantColumnReader.java Outdated Show resolved Hide resolved

aokolnychyi reviewed Jan 29, 2025

View reviewed changes

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/CometColumnReader.java Outdated Show resolved Hide resolved

aokolnychyi reviewed Jan 29, 2025

View reviewed changes

...4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/CometColumnarBatchReader.java Outdated Show resolved Hide resolved

aokolnychyi reviewed Jan 29, 2025

View reviewed changes

...4/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/CometColumnarBatchReader.java Outdated Show resolved Hide resolved

aokolnychyi reviewed Jan 29, 2025

View reviewed changes

huaxingao added 2 commits January 28, 2025 22:10

address comments

4bf5cbf

remove un-intended change in test

8f34742

aokolnychyi reviewed Jan 29, 2025

View reviewed changes

parthchandra reviewed Jan 29, 2025

View reviewed changes

huaxingao added 2 commits January 29, 2025 15:54

address comments

0d9e974

address comments

46dd439

parthchandra suggested changes Jan 30, 2025

View reviewed changes

close importer in reset

10901b0

revert to iceberg reader

dae79ad

huaxingao force-pushed the comet3 branch from 01d1cd8 to dae79ad Compare January 31, 2025 22:18

parthchandra approved these changes Jan 31, 2025

View reviewed changes

aokolnychyi approved these changes Jan 31, 2025

View reviewed changes

aokolnychyi merged commit 40334f5 into apache:main Jan 31, 2025
46 checks passed

huaxingao deleted the comet3 branch January 31, 2025 23:39

huaxingao mentioned this pull request Feb 1, 2025

Spark 3.5: Iceberg / DataFusion Comet integration #12147

Merged

jbonofre pushed a commit to jbonofre/iceberg that referenced this pull request Feb 3, 2025

Spark 3.4: Support Comet Parquet readers (apache#9841)

3232536

		}

		compileOnly "org.apache.comet:comet-spark-spark${sparkMajorVersion}_${scalaVersion}:0.1.0-SNAPSHOT"

		@@ -48,4 +48,7 @@

		<!-- Referencing guava classes should be allowed in classes within bundled-guava module -->
		<suppress files="org.apache.iceberg.GuavaClasses" id="BanUnrelocatedGuavaClasses"/>

Iceberg/Comet integration POC #9841

Iceberg/Comet integration POC #9841

Conversation

huaxingao commented Mar 1, 2024 • edited Loading

huaxingao commented Mar 1, 2024

aokolnychyi left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

huaxingao commented Apr 29, 2024

aokolnychyi commented Apr 30, 2024

Choose a reason for hiding this comment

cornelcreanga commented Jun 20, 2024

huaxingao commented Jun 20, 2024

PaulLiang1 commented Sep 4, 2024

huaxingao commented Sep 4, 2024

PaulLiang1 commented Sep 4, 2024

huaxingao commented Sep 4, 2024

huaxingao commented Sep 5, 2024

PaulLiang1 commented Sep 5, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aokolnychyi left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

huaxingao commented Jan 29, 2025

aokolnychyi commented Jan 29, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

parthchandra left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aokolnychyi commented Jan 31, 2025 • edited Loading

parthchandra left a comment

Choose a reason for hiding this comment

aokolnychyi commented Jan 31, 2025

huaxingao commented Jan 31, 2025

huaxingao commented Mar 1, 2024 •

edited

Loading

aokolnychyi left a comment •

edited

Loading

aokolnychyi commented Jan 31, 2025 •

edited

Loading