[VL] feat(iceberg): Add V3 initial value reads

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

@@ -16,4 +16,52 @@
  */
 package org.apache.gluten.execution
 
-class VeloxIcebergSuite extends IcebergSuite
+import org.apache.gluten.config.GlutenConfig
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog}
+
+import org.apache.iceberg.UpdateSchema
+import org.apache.iceberg.expressions.Literal
+import org.apache.iceberg.spark.source.SparkTable
+import org.apache.iceberg.types.{Type, Types}
+
+class VeloxIcebergSuite extends IcebergSuite {
+  testWithMinSparkVersion("iceberg v3 initial default for an added column", "3.4") {
+    withTable("iceberg_v3_initial_default") {
+      withSQLConf(GlutenConfig.GLUTEN_ENABLED.key -> "false") {
+        spark.sql("""
+                    |CREATE TABLE iceberg_v3_initial_default (id INT)
+                    |USING iceberg
+                    |TBLPROPERTIES ('format-version' = '3')
+                    |""".stripMargin)
+        spark.sql("INSERT INTO iceberg_v3_initial_default VALUES (1), (2)")
+
+        val catalog = spark.sessionState.catalogManager
+          .catalog("spark_catalog")
+          .asInstanceOf[TableCatalog]
+        val updateSchema = catalog
+          .loadTable(Identifier.of(Array("default"), "iceberg_v3_initial_default"))
+          .asInstanceOf[SparkTable]
+          .table()
+          .updateSchema()
+        classOf[UpdateSchema]
+          .getMethod(
+            "addColumn",
+            classOf[String],
+            classOf[Type],
+            classOf[Literal[_]])
+          .invoke(updateSchema, "country", Types.StringType.get(), Literal.of("IN"))
+        updateSchema.commit()
+        spark.catalog.refreshTable("iceberg_v3_initial_default")
+      }
+
+      runQueryAndCompare(
+        "SELECT id, country FROM iceberg_v3_initial_default ORDER BY id") {
+        df =>
+          checkAnswer(df, Seq(Row(1, "IN"), Row(2, "IN")))
+          checkGlutenPlan[IcebergScanTransformer](df)
+      }
+    }
+  }
+}

backends-velox/src/main/resources/org/apache/gluten/proto/IcebergReadExtension.proto

@@ -0,0 +1,18 @@
+// SPDX-License-Identifier: Apache-2.0
+syntax = "proto3";
+
+package gluten;
+
+option java_package = "org.apache.gluten.proto";
+option java_multiple_files = true;
+
+// Iceberg-specific metadata carried in LocalFiles.advanced_extension.
+message IcebergReadExtension {
+  message ColumnDefault {
+    string name = 1;
+    int32 field_id = 2;
+    string initial_default = 3;
+  }
+
+  repeated ColumnDefault column_defaults = 1;
+}

backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala

@@ -571,6 +571,8 @@ object VeloxBackendSettings extends BackendSettingsApi {
 
   override def supportIcebergEqualityDeleteRead(): Boolean = false
 
+  override def supportIcebergInitialDefaultRead(): Boolean = true
+
   override def reorderColumnsForPartitionWrite(): Boolean = true
 
   override def enableEnhancedFeatures(): Boolean = VeloxConfig.get.enableEnhancedFeatures()

backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxTransformerApi.scala

@@ -21,7 +21,7 @@ import org.apache.gluten.exception.GlutenException
 import org.apache.gluten.execution.WriteFilesExecTransformer
 import org.apache.gluten.execution.datasource.GlutenFormatFactory
 import org.apache.gluten.expression.ConverterUtils
-import org.apache.gluten.proto.ConfigMap
+import org.apache.gluten.proto.{ConfigMap, IcebergReadExtension}
 import org.apache.gluten.runtime.Runtimes
 import org.apache.gluten.substrait.SubstraitContext
 import org.apache.gluten.substrait.expression.{ExpressionBuilder, ExpressionNode}
@@ -124,6 +124,26 @@ class VeloxTransformerApi extends TransformerApi with Logging {
 
   override def packPBMessage(message: Message): Any = Any.pack(message, "")
 
+  override def packIcebergReadExtension(
+      fieldIds: JMap[String, Integer],
+      initialDefaults: JMap[String, String]): Any = {
+    val extensionBuilder = IcebergReadExtension.newBuilder()
+    initialDefaults.asScala.toSeq.sortBy(_._1).foreach {
+      case (name, initialDefault) =>
+        val fieldId = fieldIds.get(name)
+        if (fieldId == null) {
+          throw new IllegalArgumentException(s"Missing Iceberg field ID for column $name")
+        }
+        extensionBuilder.addColumnDefaults(
+          IcebergReadExtension.ColumnDefault
+            .newBuilder()
+            .setName(name)
+            .setFieldId(fieldId)
+            .setInitialDefault(initialDefault))
+    }
+    packPBMessage(extensionBuilder.build())
+  }
+
   override def invalidateSQLExecutionResource(executionId: String): Unit = {
     GlutenDriverEndpoint.invalidateResourceRelation(executionId)
   }

cpp/velox/compute/VeloxPlanConverter.cc

@@ -137,9 +137,10 @@ std::shared_ptr<DeltaSplitInfo> parseDeltaSplitInfo(
 
 std::shared_ptr<SplitInfo> parseScanSplitInfo(
     const facebook::velox::config::ConfigBase* veloxCfg,
-    const google::protobuf::RepeatedPtrField<substrait::ReadRel_LocalFiles_FileOrFiles>& fileList) {
+    const substrait::ReadRel_LocalFiles& localFiles) {
   using SubstraitFileFormatCase = ::substrait::ReadRel_LocalFiles_FileOrFiles::FileFormatCase;
 
+  const auto& fileList = localFiles.items();
   auto splitInfo = std::make_shared<SplitInfo>();
   splitInfo->leafType = SplitInfo::LeafType::TABLE_SCAN;
   splitInfo->paths.reserve(fileList.size());
@@ -193,7 +194,8 @@ std::shared_ptr<SplitInfo> parseScanSplitInfo(
         splitInfo->format = dwio::common::FileFormat::TEXT;
         break;
       case SubstraitFileFormatCase::kIceberg:
-        splitInfo = IcebergPlanConverter::parseIcebergSplitInfo(file, std::move(splitInfo));
+        splitInfo =
+            IcebergPlanConverter::parseIcebergSplitInfo(file, localFiles.advanced_extension(), std::move(splitInfo));
         break;
       case SubstraitFileFormatCase::kDelta:
         splitInfo = parseDeltaSplitInfo(file, std::move(splitInfo));
@@ -237,8 +239,7 @@ void parseLocalFileNodes(
   std::vector<std::shared_ptr<SplitInfo>> splitInfos;
   splitInfos.reserve(localFiles.size());
   for (const auto& localFile : localFiles) {
-    const auto& fileList = localFile.items();
-    splitInfos.push_back(parseScanSplitInfo(veloxCfg, fileList));
+    splitInfos.push_back(parseScanSplitInfo(veloxCfg, localFile));
   }
 
   planConverter->setSplitInfos(std::move(splitInfos));

cpp/velox/compute/iceberg/IcebergPlanConverter.cc

@@ -17,10 +17,13 @@
 
 #include "IcebergPlanConverter.h"
 
+#include "IcebergReadExtension.pb.h"
+
 namespace gluten {
 
 std::shared_ptr<IcebergSplitInfo> IcebergPlanConverter::parseIcebergSplitInfo(
     substrait::ReadRel_LocalFiles_FileOrFiles file,
+    const substrait::extensions::AdvancedExtension& extension,
     std::shared_ptr<SplitInfo> splitInfo) {
   using SubstraitFileFormatCase = ::substrait::ReadRel_LocalFiles_FileOrFiles::IcebergReadOptions::FileFormatCase;
   using SubstraitDeleteFileFormatCase =
@@ -40,6 +43,20 @@ std::shared_ptr<IcebergSplitInfo> IcebergPlanConverter::parseIcebergSplitInfo(
       icebergSplitInfo->format = dwio::common::FileFormat::UNKNOWN;
       break;
   }
+
+  if (icebergSplitInfo->columns.empty() && extension.has_enhancement() &&
+      extension.enhancement().Is<::gluten::IcebergReadExtension>()) {
+    ::gluten::IcebergReadExtension icebergExtension;
+    VELOX_USER_CHECK(extension.enhancement().UnpackTo(&icebergExtension), "Failed to unpack Iceberg read extension");
+    for (const auto& column : icebergExtension.column_defaults()) {
+      auto [it, inserted] = icebergSplitInfo->columns.try_emplace(
+          column.name(), IcebergColumnInfo{column.field_id(), column.initial_default()});
+      if (!inserted) {
+        VELOX_USER_CHECK_EQ(
+            it->second.fieldId, column.field_id(), "Conflicting Iceberg field IDs for column '{}'", column.name());
+      }
+    }
+  }
   if (icebergReadOption.delete_files_size() > 0) {
     auto deleteFiles = icebergReadOption.delete_files();
     std::vector<IcebergDeleteFile> deletes;

cpp/velox/compute/iceberg/IcebergPlanConverter.h

@@ -17,14 +17,24 @@
 
 #pragma once
 
+#include <optional>
+#include <string>
+#include <unordered_map>
+
 #include "substrait/SubstraitToVeloxPlan.h"
 #include "velox/connectors/hive/iceberg/IcebergDeleteFile.h"
 
 using namespace facebook::velox::connector::hive::iceberg;
 
 namespace gluten {
+struct IcebergColumnInfo {
+  int32_t fieldId;
+  std::optional<std::string> initialDefault;
+};
+
 struct IcebergSplitInfo : SplitInfo {
   std::vector<std::vector<IcebergDeleteFile>> deleteFilesVec;
+  std::unordered_map<std::string, IcebergColumnInfo> columns;
 
   IcebergSplitInfo(const SplitInfo& splitInfo) : SplitInfo(splitInfo) {
     // Reserve the actual size of the deleteFilesVec.
@@ -36,6 +46,7 @@ class IcebergPlanConverter {
  public:
   static std::shared_ptr<IcebergSplitInfo> parseIcebergSplitInfo(
       substrait::ReadRel_LocalFiles_FileOrFiles file,
+      const substrait::extensions::AdvancedExtension& extension,
       std::shared_ptr<SplitInfo> splitInfo);
 };
 

cpp/velox/substrait/SubstraitToVeloxPlan.cc

@@ -26,6 +26,7 @@
 #include "operators/hashjoin/HashTableBuilder.h"
 #include "operators/plannodes/RowVectorStream.h"
 #include "velox/connectors/hive/HiveDataSink.h"
+#include "velox/connectors/hive/iceberg/IcebergColumnHandle.h"
 #include "velox/exec/TableWriter.h"
 #include "velox/type/Type.h"
 
@@ -1609,11 +1610,41 @@ core::PlanNodePtr SubstraitToVeloxPlanConverter::toVeloxPlan(const ::substrait::
   std::vector<std::string> outNames;
   outNames.reserve(colNameList.size());
   connector::ColumnHandleMap assignments;
+  const auto icebergSplitInfo = std::dynamic_pointer_cast<IcebergSplitInfo>(splitInfo);
   for (int idx = 0; idx < colNameList.size(); idx++) {
     auto outName = SubstraitParser::makeNodeName(planNodeId_, idx);
     auto columnType = columnTypes[idx];
-    assignments[outName] = std::make_shared<connector::hive::HiveColumnHandle>(
-        colNameList[idx], columnType, veloxTypeList[idx], veloxTypeList[idx]);
+    const IcebergColumnInfo* icebergColumn = nullptr;
+    if (icebergSplitInfo) {
+      auto columnIt = icebergSplitInfo->columns.find(colNameList[idx]);
+      if (columnIt != icebergSplitInfo->columns.end()) {
+        icebergColumn = &columnIt->second;
+      } else if (asLowerCase) {
+        for (const auto& [name, column] : icebergSplitInfo->columns) {
+          auto normalizedName = name;
+          folly::toLowerAscii(normalizedName);
+          if (normalizedName == colNameList[idx]) {
+            icebergColumn = &column;
+            break;
+          }
+        }
+      }
+    }
+    // Gluten serializes Iceberg partition dates as ISO strings. Use the
+    // Iceberg handle, which expects epoch-day partition dates, only when its
+    // initial-default metadata is needed by Velox.
+    if (icebergColumn && icebergColumn->initialDefault.has_value()) {
+      assignments[outName] = std::make_shared<connector::hive::iceberg::IcebergColumnHandle>(
+          colNameList[idx],
+          columnType,
+          veloxTypeList[idx],
+          facebook::velox::parquet::ParquetFieldId(icebergColumn->fieldId),
+          std::vector<common::Subfield>{},
+          icebergColumn->initialDefault);
+    } else {
+      assignments[outName] = std::make_shared<connector::hive::HiveColumnHandle>(
+          colNameList[idx], columnType, veloxTypeList[idx], veloxTypeList[idx]);
+    }
     outNames.emplace_back(outName);
   }
   auto outputType = ROW(std::move(outNames), std::move(veloxTypeList));

gluten-iceberg/src-iceberg10/main/java/org/apache/gluten/IcebergDefaultValueUtil.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.gluten;
+
+import org.apache.iceberg.types.Types;
+
+public final class IcebergDefaultValueUtil {
+  private IcebergDefaultValueUtil() {}
+
+  public static Object getInitialDefault(Types.NestedField field) {
+    return field.initialDefault();
+  }
+}

gluten-iceberg/src-iceberg3/main/java/org/apache/gluten/IcebergDefaultValueUtil.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.gluten;
+
+import org.apache.iceberg.types.Types;
+
+public final class IcebergDefaultValueUtil {
+  private IcebergDefaultValueUtil() {}
+
+  public static Object getInitialDefault(Types.NestedField field) {
+    return null;
+  }
+}

gluten-iceberg/src-iceberg5/main/java/org/apache/gluten/IcebergDefaultValueUtil.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.gluten;
+
+import org.apache.iceberg.types.Types;
+
+public final class IcebergDefaultValueUtil {
+  private IcebergDefaultValueUtil() {}
+
+  public static Object getInitialDefault(Types.NestedField field) {
+    return null;
+  }
+}

gluten-iceberg/src/main/java/org/apache/gluten/substrait/rel/IcebergLocalFilesBuilder.java

@@ -31,7 +31,9 @@ public static IcebergLocalFilesNode makeIcebergLocalFiles(
       LocalFilesNode.ReadFileFormat fileFormat,
       List<String> preferredLocations,
       List<List<DeleteFile>> deleteFilesList,
-      List<Map<String, String>> metadataColumns) {
+      List<Map<String, String>> metadataColumns,
+      Map<String, Integer> fieldIds,
+      Map<String, String> initialDefaults) {
     return new IcebergLocalFilesNode(
         index,
         paths,
@@ -41,6 +43,8 @@ public static IcebergLocalFilesNode makeIcebergLocalFiles(
         fileFormat,
         preferredLocations,
         deleteFilesList,
-        metadataColumns);
+        metadataColumns,
+        fieldIds,
+        initialDefaults);
   }
 }