Iceberg/Comet integration POC

.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"/>
+
+    <!-- Suppress checks for CometColumnReader -->
+    <suppress files="org.apache.iceberg.spark.data.vectorized.CometColumnReader" checks="IllegalImport"/>
 </suppressions>

spark/v3.4/build.gradle

@@ -52,6 +52,8 @@ project(":iceberg-spark:iceberg-spark-${sparkMajorVersion}_${scalaVersion}") {
   dependencies {
     implementation project(path: ':iceberg-bundled-guava', configuration: 'shadow')
     api project(':iceberg-api')
+    annotationProcessor libs.immutables.value
+    compileOnly libs.immutables.value
     implementation project(':iceberg-common')
     implementation project(':iceberg-core')
     implementation project(':iceberg-data')
@@ -77,6 +79,8 @@ project(":iceberg-spark:iceberg-spark-${sparkMajorVersion}_${scalaVersion}") {
       exclude group: 'org.roaringbitmap'
     }
 
+    compileOnly "org.apache.datafusion:comet-spark-spark${sparkMajorVersion}_${scalaVersion}:0.3.0"
+
     implementation libs.parquet.column
     implementation libs.parquet.hadoop
 
@@ -189,6 +193,7 @@ project(":iceberg-spark:iceberg-spark-extensions-${sparkMajorVersion}_${scalaVer
     testImplementation libs.avro.avro
     testImplementation libs.parquet.hadoop
     testImplementation libs.junit.vintage.engine
+    testImplementation "org.apache.datafusion:comet-spark-spark${sparkMajorVersion}_${scalaVersion}:0.3.0"
 
     // Required because we remove antlr plugin dependencies from the compile configuration, see note above
     runtimeOnly libs.antlr.runtime

spark/v3.4/spark-runtime/src/integration/java/org/apache/iceberg/spark/SmokeTest.java

@@ -28,6 +28,7 @@
 import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
 import org.junit.Assert;
 import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Test;
 
 public class SmokeTest extends SparkExtensionsTestBase {
@@ -44,7 +45,7 @@ public void dropTable() {
   // Run through our Doc's Getting Started Example
   // TODO Update doc example so that it can actually be run, modifications were required for this
   // test suite to run
-  @Test
+  @Ignore
   public void testGettingStarted() throws IOException {
     // Creating a table
     sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName);

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

@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark;
+
+import java.io.Serializable;
+import org.immutables.value.Value;
+
+@Value.Immutable
+public interface OrcBatchReadConf extends Serializable {
+  int batchSize();
+}

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

@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark;
+
+import java.io.Serializable;
+import org.immutables.value.Value;
+
+@Value.Immutable
+public interface ParquetBatchReadConf extends Serializable {
+  int batchSize();
+
+  ParquetReaderType readerType();
+}

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

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark;
+
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/** Enumerates the types of Parquet readers. */
+public enum ParquetReaderType {
+  /** ICEBERG type utilizes the built-in Parquet reader. */
+  ICEBERG("iceberg"),
+
+  /**
+   * COMET type changes the Parquet reader to the Apache DataFusion Comet Parquet reader. Comet
+   * Parquet reader performs I/O and decompression in the JVM but decodes in native to improve
+   * performance. Additionally, Comet will convert Spark's physical plan into a native physical plan
+   * and execute this plan natively.
+   *
+   * <p>TODO: Implement {@link org.apache.comet.parquet.SupportsComet} in SparkScan to convert Spark
+   * physical plan to native physical plan for native execution.
+   */
+  COMET("comet");
+
+  private final String parquetReaderType;
+
+  ParquetReaderType(String readerType) {
+    this.parquetReaderType = readerType;
+  }
+
+  public static ParquetReaderType fromName(String parquetReaderType) {
+    Preconditions.checkArgument(parquetReaderType != null, "Parquet reader type is null");
+
+    if (ICEBERG.parquetReaderType().equalsIgnoreCase(parquetReaderType)) {
+      return ICEBERG;
+
+    } else if (COMET.parquetReaderType().equalsIgnoreCase(parquetReaderType)) {
+      return COMET;
+
+    } else {
+      throw new IllegalArgumentException("Unknown parquet reader type: " + parquetReaderType);
+    }
+  }
+
+  public String parquetReaderType() {
+    return parquetReaderType;
+  }
+}

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

@@ -359,4 +359,12 @@ public boolean reportColumnStats() {
         .defaultValue(SparkSQLProperties.REPORT_COLUMN_STATS_DEFAULT)
         .parse();
   }
+
+  public ParquetReaderType parquetReaderType() {
+    return confParser
+        .enumConf(ParquetReaderType::fromName)
+        .sessionConf(SparkSQLProperties.PARQUET_READER_TYPE)
+        .defaultValue(SparkSQLProperties.PARQUET_READER_TYPE_DEFAULT)
+        .parse();
+  }
 }

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

@@ -27,6 +27,10 @@ private SparkSQLProperties() {}
   // Controls whether vectorized reads are enabled
   public static final String VECTORIZATION_ENABLED = "spark.sql.iceberg.vectorization.enabled";
 
+  // Controls which Parquet reader implementation to use
+  public static final String PARQUET_READER_TYPE = "spark.sql.iceberg.parquet.reader-type";
+  public static final ParquetReaderType PARQUET_READER_TYPE_DEFAULT = ParquetReaderType.COMET;
+
   // Controls whether reading/writing timestamps without timezones is allowed
   @Deprecated
   public static final String HANDLE_TIMESTAMP_WITHOUT_TIMEZONE =

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

@@ -18,78 +18,133 @@
  */
 package org.apache.iceberg.spark.data.vectorized;
 
-import org.apache.iceberg.arrow.vectorized.VectorHolder;
 import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.vectorized.ColumnVector;
 import org.apache.spark.sql.vectorized.ColumnarArray;
+import org.apache.spark.sql.vectorized.ColumnarMap;
 import org.apache.spark.unsafe.types.UTF8String;
 
-public class ColumnVectorWithFilter extends IcebergArrowColumnVector {
+/**
+ * A column vector implementation that applies row-level filtering.
+ *
+ * <p>This class wraps an existing column vector and uses a row ID mapping array to remap row
+ * indices during data access. Each method that retrieves data for a specific row translates the
+ * provided row index using the mapping array, effectively filtering the original data to only
+ * expose the live subset of rows. This approach allows efficient row-level filtering without
+ * modifying the underlying data.
+ */
+public class ColumnVectorWithFilter extends ColumnVector {
+  private final ColumnVector delegate;
   private final int[] rowIdMapping;
+  private volatile ColumnVectorWithFilter[] children = null;
 
-  public ColumnVectorWithFilter(VectorHolder holder, int[] rowIdMapping) {
-    super(holder);
+  public ColumnVectorWithFilter(ColumnVector delegate, int[] rowIdMapping) {
+    super(delegate.dataType());
+    this.delegate = delegate;
     this.rowIdMapping = rowIdMapping;
   }
 
+  @Override
+  public void close() {
+    delegate.close();
+  }
+
+  @Override
+  public boolean hasNull() {
+    return delegate.hasNull();
+  }
+
+  @Override
+  public int numNulls() {
+    // computing the actual number of nulls with rowIdMapping is expensive
+    // it is OK to overestimate and return the number of nulls in the original vector
+    return delegate.numNulls();
+  }
+
   @Override
   public boolean isNullAt(int rowId) {
-    return nullabilityHolder().isNullAt(rowIdMapping[rowId]) == 1;
+    return delegate.isNullAt(rowIdMapping[rowId]);
   }
 
   @Override
   public boolean getBoolean(int rowId) {
-    return accessor().getBoolean(rowIdMapping[rowId]);
+    return delegate.getBoolean(rowIdMapping[rowId]);
+  }
+
+  @Override
+  public byte getByte(int rowId) {
+    return delegate.getByte(rowIdMapping[rowId]);
+  }
+
+  @Override
+  public short getShort(int rowId) {
+    return delegate.getShort(rowIdMapping[rowId]);
   }
 
   @Override
   public int getInt(int rowId) {
-    return accessor().getInt(rowIdMapping[rowId]);
+    return delegate.getInt(rowIdMapping[rowId]);
   }
 
   @Override
   public long getLong(int rowId) {
-    return accessor().getLong(rowIdMapping[rowId]);
+    return delegate.getLong(rowIdMapping[rowId]);
   }
 
   @Override
   public float getFloat(int rowId) {
-    return accessor().getFloat(rowIdMapping[rowId]);
+    return delegate.getFloat(rowIdMapping[rowId]);
   }
 
   @Override
   public double getDouble(int rowId) {
-    return accessor().getDouble(rowIdMapping[rowId]);
+    return delegate.getDouble(rowIdMapping[rowId]);
   }
 
   @Override
   public ColumnarArray getArray(int rowId) {
-    if (isNullAt(rowId)) {
-      return null;
-    }
-    return accessor().getArray(rowIdMapping[rowId]);
+    return delegate.getArray(rowIdMapping[rowId]);
+  }
+
+  @Override
+  public ColumnarMap getMap(int rowId) {
+    return delegate.getMap(rowIdMapping[rowId]);
   }
 
   @Override
   public Decimal getDecimal(int rowId, int precision, int scale) {
-    if (isNullAt(rowId)) {
-      return null;
-    }
-    return accessor().getDecimal(rowIdMapping[rowId], precision, scale);
+    return delegate.getDecimal(rowIdMapping[rowId], precision, scale);
   }
 
   @Override
   public UTF8String getUTF8String(int rowId) {
-    if (isNullAt(rowId)) {
-      return null;
-    }
-    return accessor().getUTF8String(rowIdMapping[rowId]);
+    return delegate.getUTF8String(rowIdMapping[rowId]);
   }
 
   @Override
   public byte[] getBinary(int rowId) {
-    if (isNullAt(rowId)) {
-      return null;
+    return delegate.getBinary(rowIdMapping[rowId]);
+  }
+
+  @Override
+  public ColumnVector getChild(int ordinal) {
+    if (children == null) {
+      synchronized (this) {
+        if (children == null) {
+          if (dataType() instanceof StructType) {
+            StructType structType = (StructType) dataType();
+            this.children = new ColumnVectorWithFilter[structType.length()];
+            for (int index = 0; index < structType.length(); index++) {
+              children[index] = new ColumnVectorWithFilter(delegate.getChild(index), rowIdMapping);
+            }
+          } else {
+            throw new UnsupportedOperationException("Unsupported nested type: " + dataType());
+          }
+        }
+      }
     }
-    return accessor().getBinary(rowIdMapping[rowId]);
+
+    return children[ordinal];
   }
 }