Enable enhanced tests for spark 4.0 & fix failures

Co-authored-by: Yuan <yuanzhou@apache.org>

Author: infvg

.github/workflows/velox_backend_enhanced.yml

@@ -298,7 +298,7 @@ jobs:
           java -version
           $MVN_CMD clean test -Pspark-4.0 -Pscala-2.13 -Pjava-17 -Pbackends-velox -Piceberg \
           -Pspark-ut -DargLine="-Dspark.test.home=/opt/shims/spark40/spark_home/" \
-          -DtagsToExclude=org.apache.spark.tags.ExtendedSQLTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.gluten.tags.SkipTest
+          -DtagsToExclude=org.apache.spark.tags.ExtendedSQLTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.SkipTest
       - name: Upload test report
         if: always()
         uses: actions/upload-artifact@v4

backends-velox/src-iceberg/main/scala/org/apache/gluten/connector/write/IcebergColumnarBatchDataWriter.scala

@@ -44,6 +44,10 @@ case class IcebergColumnarBatchDataWriter(
   }
 
   override def write(batch: ColumnarBatch): Unit = {
+    // Pass the original batch to native code
+    // The native code will use the schema (writeSchema) we provided during initialization
+    // to determine which columns to write, effectively filtering out metadata columns
+    // like __row_operation, _file, _pos that Spark 4.0 adds
     val batchHandle = ColumnarBatches.getNativeHandle(BackendsApiManager.getBackendName, batch)
     jniWrapper.write(writer, batchHandle)
   }

backends-velox/src-iceberg/main/scala/org/apache/gluten/execution/AbstractIcebergWriteExec.scala

@@ -24,14 +24,23 @@ import org.apache.spark.sql.types.StructType
 import org.apache.iceberg.spark.source.IcebergWriteUtil
 import org.apache.iceberg.types.TypeUtil
 
+import scala.collection.JavaConverters._
+
 abstract class AbstractIcebergWriteExec extends IcebergWriteExec {
 
   // the writer factory works for both batch and streaming
   private def createIcebergDataWriteFactory(schema: StructType): IcebergDataWriteFactory = {
     val writeSchema = IcebergWriteUtil.getWriteSchema(write)
     val nestedField = TypeUtil.visit(writeSchema, new IcebergNestedFieldVisitor)
+    // Filter out metadata columns from the Spark output schema and reorder to match Iceberg schema
+    // Spark 4.0 may include metadata columns in the output schema during UPDATE operations,
+    // but these should not be written to the Iceberg table
+    val schemaFieldMap = schema.fields.map(f => f.name -> f).toMap
+    val filteredFields =
+      writeSchema.columns().asScala.flatMap(icebergCol => schemaFieldMap.get(icebergCol.name()))
+    val filteredSchema = StructType(filteredFields.toArray)
     IcebergDataWriteFactory(
-      schema,
+      filteredSchema,
       getFileFormat(IcebergWriteUtil.getFileFormat(write)),
       IcebergWriteUtil.getDirectory(write),
       getCodec,

backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/enhanced/VeloxIcebergSuite.scala

@@ -383,4 +383,59 @@ class VeloxIcebergSuite extends IcebergSuite {
       }
     }
   }
+
+  test("iceberg read cow table - update after schema evolution") {
+    withTable("iceberg_cow_update_evolved_tb") {
+      spark.sql("""
+                  |create table iceberg_cow_update_evolved_tb (
+                  |  id int,
+                  |  name string,
+                  |  age int
+                  |) using iceberg
+                  |tblproperties (
+                  |  'format-version' = '2',
+                  |  'write.delete.mode' = 'copy-on-write',
+                  |  'write.update.mode' = 'copy-on-write',
+                  |  'write.merge.mode' = 'copy-on-write'
+                  |)
+                  |""".stripMargin)
+
+      spark.sql("""
+                  |alter table iceberg_cow_update_evolved_tb
+                  |add columns (salary decimal(10, 2))
+                  |""".stripMargin)
+
+      spark.sql("""
+                  |insert into table iceberg_cow_update_evolved_tb values
+                  |  (1, 'Name1', 23, 3400.00),
+                  |  (2, 'Name2', 30, 5500.00),
+                  |  (3, 'Name3', 35, 6500.00)
+                  |""".stripMargin)
+
+      val df = spark.sql("""
+                           |update iceberg_cow_update_evolved_tb
+                           |set name = 'Name4'
+                           |where id = 1
+                           |""".stripMargin)
+
+      assert(
+        df.queryExecution.executedPlan
+          .asInstanceOf[CommandResultExec]
+          .commandPhysicalPlan
+          .isInstanceOf[VeloxIcebergReplaceDataExec])
+
+      checkAnswer(
+        spark.sql("""
+                    |select id, name, age, salary
+                    |from iceberg_cow_update_evolved_tb
+                    |order by id
+                    |""".stripMargin),
+        Seq(
+          Row(1, "Name4", 23, new java.math.BigDecimal("3400.00")),
+          Row(2, "Name2", 30, new java.math.BigDecimal("5500.00")),
+          Row(3, "Name1", 35, new java.math.BigDecimal("6500.00"))
+        )
+      )
+    }
+  }
 }

cpp/velox/compute/iceberg/IcebergWriter.cc

@@ -135,6 +135,15 @@ std::shared_ptr<IcebergInsertTableHandle> createIcebergInsertTableHandle(
   for (const auto& field : spec->fields) {
     partitionColumns.push_back(field.name);
   }
+
+  // Validate that nestedField.children size matches columnNames size
+  VELOX_CHECK_EQ(
+      nestedField.children.size(),
+      columnNames.size(),
+      "Mismatch between nestedField children size ({}) and column names size ({})",
+      nestedField.children.size(),
+      columnNames.size());
+
   for (auto i = 0; i < columnNames.size(); ++i) {
     if (std::find(partitionColumns.begin(), partitionColumns.end(), columnNames[i]) != partitionColumns.end()) {
       columnHandles.push_back(
@@ -154,10 +163,10 @@ std::shared_ptr<IcebergInsertTableHandle> createIcebergInsertTableHandle(
               nestedField.children[i]));
     }
   }
-  
+
   auto fileNameGenerator = std::make_shared<const GlutenIcebergFileNameGenerator>(
       partitionId, taskId, operationId, fileFormat);
-  
+
   std::shared_ptr<const connector::hive::LocationHandle> locationHandle =
       std::make_shared<connector::hive::LocationHandle>(
           outputDirectoryPath, outputDirectoryPath, connector::hive::LocationHandle::TableType::kExisting);
@@ -212,7 +221,38 @@ IcebergWriter::IcebergWriter(
 }
 
 void IcebergWriter::write(const VeloxColumnarBatch& batch) {
-  dataSink_->appendData(batch.getRowVector());
+  auto inputRowVector = batch.getRowVector();
+  auto inputRowType = asRowType(inputRowVector->type());
+
+  if (inputRowType->size() != rowType_->size()) {
+    VELOX_CHECK_GT(
+        inputRowType->size(), rowType_->size(), "Input schema has fewer columns than expected output schema");
+
+    std::vector<VectorPtr> filteredChildren;
+    filteredChildren.reserve(rowType_->size());
+    const size_t offset = inputRowType->size() - rowType_->size();
+
+    for (size_t i = 0; i < rowType_->size(); i++) {
+      VELOX_CHECK_EQ(
+          inputRowType->nameOf(i + offset),
+          rowType_->nameOf(i),
+          "Column name mismatch at position {}: expected '{}', got '{}'. "
+          "Metadata columns should be at the beginning.",
+          i + offset,
+          rowType_->nameOf(i),
+          inputRowType->nameOf(i + offset));
+      filteredChildren.push_back(inputRowVector->childAt(i + offset));
+    }
+
+    // Create a new RowVector with filtered columns
+    auto filteredRowVector = std::make_shared<RowVector>(
+        pool_.get(), rowType_, inputRowVector->nulls(), inputRowVector->size(), std::move(filteredChildren));
+
+    dataSink_->appendData(filteredRowVector);
+  } else {
+    // No filtering needed, schemas match
+    dataSink_->appendData(inputRowVector);
+  }
 }
 
 std::vector<std::string> IcebergWriter::commit() {