apache · kokila-19 · Aug 20, 2026
diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergSerDe.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergSerDe.java
@@ -93,11 +93,14 @@ public void initialize(Configuration conf, Properties serDeProperties,
     // executor, but serDeProperties are populated by HiveIcebergStorageHandler.configureInputJobProperties() and
     // the resulting properties are serialized and distributed to the executors
 
+    PartitionSpec partitionSpec = PartitionSpec.unpartitioned();
+
     if (serDeProperties.get(InputFormatConfig.TABLE_SCHEMA) != null) {
       this.tableSchema = SchemaParser.fromJson(serDeProperties.getProperty(InputFormatConfig.TABLE_SCHEMA));
       if (serDeProperties.get(InputFormatConfig.PARTITION_SPEC) != null) {
         PartitionSpec spec =
             PartitionSpecParser.fromJson(tableSchema, serDeProperties.getProperty(InputFormatConfig.PARTITION_SPEC));
+        partitionSpec = spec;
         this.partitionColumns = spec.fields().stream().map(PartitionField::name).collect(Collectors.toList());
       } else {
         this.partitionColumns = ImmutableList.of();
@@ -107,6 +110,7 @@ public void initialize(Configuration conf, Properties serDeProperties,
         Table table = IcebergTableUtil.getTable(conf, serDeProperties);
         // always prefer the original table schema if there is one
         this.tableSchema = table.schema();
+        partitionSpec = table.spec();
         this.partitionColumns = table.spec().fields().stream().map(PartitionField::name).collect(Collectors.toList());
         LOG.info("Using schema from existing table {}", SchemaParser.toJson(tableSchema));
       } catch (Exception e) {
@@ -119,6 +123,7 @@ public void initialize(Configuration conf, Properties serDeProperties,
           try (FileIO fileIO = new HadoopFileIO(conf)) {
             TableMetadata metadata = TableMetadataParser.read(fileIO, serDeProperties.getProperty("metadata_location"));
             this.tableSchema = metadata.schema();
+            partitionSpec = metadata.spec();
             this.partitionColumns =
                 metadata.spec().fields().stream().map(PartitionField::name).collect(Collectors.toList());
             // Validate no schema is provided via create command
@@ -140,7 +145,8 @@ public void initialize(Configuration conf, Properties serDeProperties,
       }
     }
 
-    this.projectedSchema = projectedSchema(conf, serDeProperties, tableSchema, jobConf);
+    this.projectedSchema =
+        projectedSchema(conf, serDeProperties, tableSchema, partitionSpec, jobConf);
 
     if (!IcebergTableUtil.isFanoutEnabled(serDeProperties::getProperty)) {
       // ClusteredWriter requires that records are ordered by partition keys.
@@ -156,7 +162,7 @@ public void initialize(Configuration conf, Properties serDeProperties,
   }
 
   private static Schema projectedSchema(Configuration conf, Properties serDeProperties,
-      Schema tableSchema, Map<String, String> jobConf) {
+      Schema tableSchema, PartitionSpec partitionSpec, Map<String, String> jobConf) {
     String tableName = serDeProperties.getProperty(Catalogs.NAME);
     Context.Operation operation = HiveCustomStorageHandlerUtils.getWriteOperation(conf::get, tableName);
 
@@ -176,7 +182,8 @@ private static Schema projectedSchema(Configuration conf, Properties serDeProper
             MetadataColumns.schemaWithRowLineage(tableSchema) :
             tableSchema;
       } else {
-        return projectedSchema;
+        return IcebergTableUtil.includeIdentityPartitionSourceColumns(
+            projectedSchema, tableSchema, partitionSpec, partitionSpec.identitySourceIds());
       }
     }
     boolean isCOW = IcebergTableUtil.isCopyOnWriteMode(operation, conf::get);

diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java
@@ -27,6 +27,7 @@
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.Iterator;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -56,6 +57,7 @@
 import org.apache.hadoop.hive.metastore.utils.TableFetcher;
 import org.apache.hadoop.hive.ql.Context;
 import org.apache.hadoop.hive.ql.QueryState;
+import org.apache.hadoop.hive.ql.exec.SerializationUtilities;
 import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
 import org.apache.hadoop.hive.ql.metadata.DummyPartition;
 import org.apache.hadoop.hive.ql.metadata.HiveException;
@@ -65,7 +67,9 @@
 import org.apache.hadoop.hive.ql.parse.SemanticException;
 import org.apache.hadoop.hive.ql.parse.TransformSpec;
 import org.apache.hadoop.hive.ql.parse.TransformSpec.TransformType;
+import org.apache.hadoop.hive.ql.plan.ExprNodeGenericFuncDesc;
 import org.apache.hadoop.hive.ql.plan.PlanUtils;
+import org.apache.hadoop.hive.ql.plan.TableScanDesc;
 import org.apache.hadoop.hive.ql.session.SessionState;
 import org.apache.hadoop.hive.ql.session.SessionStateUtil;
 import org.apache.hadoop.util.Sets;
@@ -98,6 +102,7 @@
 import org.apache.iceberg.Transaction;
 import org.apache.iceberg.UpdatePartitionSpec;
 import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.expressions.Binder;
 import org.apache.iceberg.expressions.Evaluator;
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.expressions.Expressions;
@@ -117,10 +122,12 @@
 import org.apache.iceberg.types.Comparators;
 import org.apache.iceberg.types.Conversions;
 import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.TypeUtil;
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.util.ByteBuffers;
 import org.apache.iceberg.util.Pair;
 import org.apache.iceberg.util.PartitionUtil;
+import org.apache.iceberg.util.SerializationUtil;
 import org.apache.iceberg.util.SnapshotUtil;
 import org.apache.iceberg.util.StructProjection;
 import org.slf4j.Logger;
@@ -458,6 +465,96 @@ public static void performMetadataDelete(Table icebergTable, String branchName,
     deleteFiles.deleteFromRowFilter(exp).commit();
   }
 
+  /**
+   * Returns identity partition source column ids that should be included in a read schema.
+   */
+  public static Set<Integer> requiredIdentityPartitionSourceIds(Configuration conf, Schema tableSchema,
+      PartitionSpec spec, String[] selectedColumnNames) {
+    if (spec == null || spec.isUnpartitioned()) {
+      return Collections.emptySet();
+    }
+
+    Set<Integer> required = new LinkedHashSet<>();
+    if (selectedColumnNames != null) {
+      for (String name : selectedColumnNames) {
+        Types.NestedField field = tableSchema.findField(name);
+        if (field != null) {
+          required.add(field.fieldId());
+        }
+      }
+    }
+
+    Expression filter = filterExpressionFromConf(conf);
+    if (filter != null && filter != Expressions.alwaysTrue()) {
+      boolean caseSensitive = conf.getBoolean(
+          InputFormatConfig.CASE_SENSITIVE, InputFormatConfig.CASE_SENSITIVE_DEFAULT);
+      required.addAll(Binder.boundReferences(tableSchema.asStruct(), Collections.singletonList(filter), caseSensitive));
+    }
+
+    Set<Integer> identitySourceIds = spec.identitySourceIds();
+    required.retainAll(identitySourceIds);
+    return required;
+  }
+
+  private static Expression filterExpressionFromConf(Configuration conf) {
+    String encodedFilter = conf.get(InputFormatConfig.FILTER_EXPRESSION);
+    if (encodedFilter != null) {
+      return SerializationUtil.deserializeFromBase64(encodedFilter);
+    }
+
+    String hiveFilter = conf.get(TableScanDesc.FILTER_EXPR_CONF_STR);
+    if (hiveFilter != null) {
+      ExprNodeGenericFuncDesc exprNodeDesc =
+          SerializationUtilities.deserializeObject(hiveFilter, ExprNodeGenericFuncDesc.class);
+      return HiveIcebergInputFormat.getFilterExpr(conf, exprNodeDesc);
+    }
+
+    return null;
+  }
+
+  /**
+   * Include identity partition source columns in the schema so that partition constants can be injected even when the
+   * columns are omitted from the read projection.
+   */
+  public static Schema includeIdentityPartitionSourceColumns(Schema schema, Schema tableSchema, PartitionSpec spec,
+      Set<Integer> requiredSourceIds) {
+    if (spec == null || spec.isUnpartitioned() || requiredSourceIds == null || requiredSourceIds.isEmpty()) {
+      return schema;
+    }
+
+    Set<Integer> projectedIds = schema.columns().stream()
+        .map(Types.NestedField::fieldId)
+        .collect(Collectors.toSet());
+    Set<Integer> identitySourceIds = spec.identitySourceIds();
+    Set<Integer> sourceIdsToAdd = requiredSourceIds.stream()
+        .filter(identitySourceIds::contains)
+        .filter(id -> !projectedIds.contains(id))
+        .collect(Collectors.toCollection(LinkedHashSet::new));
+    if (sourceIdsToAdd.isEmpty()) {
+      return schema;
+    }
+
+    Set<Integer> tableFieldIds = tableSchema.columns().stream()
+        .map(Types.NestedField::fieldId)
+        .collect(Collectors.toSet());
+    List<Types.NestedField> metaColumns = Lists.newArrayList();
+    Set<Integer> tableProjectionIds = new LinkedHashSet<>();
+    for (Types.NestedField field : schema.columns()) {
+      if (tableFieldIds.contains(field.fieldId())) {
+        tableProjectionIds.add(field.fieldId());
+      } else {
+        metaColumns.add(field);
+      }
+    }
+    tableProjectionIds.addAll(sourceIdsToAdd);
+
+    List<Types.NestedField> columns = Lists.newArrayListWithCapacity(
+        metaColumns.size() + tableProjectionIds.size());
+    columns.addAll(metaColumns);
+    columns.addAll(TypeUtil.select(tableSchema, tableProjectionIds).columns());
+    return new Schema(columns);
+  }
+
   /**
    * Parses an Iceberg partition path into a Hive-compatible spec map, representing null partition
    * values with the Hive default partition name.

diff --git a/...rg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vector/HiveBatchIterator.java b/...rg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vector/HiveBatchIterator.java
@@ -22,9 +22,12 @@
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.stream.LongStream;
+import org.apache.hadoop.hive.common.type.DataTypePhysicalVariation;
 import org.apache.hadoop.hive.llap.LlapHiveUtils;
 import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
 import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedBatchUtil;
 import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
 import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatchCtx;
 import org.apache.hadoop.hive.ql.io.RowPositionAwareVectorizedRecordReader;
@@ -101,10 +104,18 @@ private void advance() {
         if (partitionColIndices != null) {
           for (int i = 0; i < partitionColIndices.length; ++i) {
             int colIdx = partitionColIndices[i];
-            // The partition column might not be part of the current projection - in which case no CV is inited
-            if (batch.cols[colIdx] != null) {
-              vrbCtx.addPartitionColsToBatch(batch.cols[colIdx], partitionValues[i], partitionColIndices[i]);
+            ColumnVector partitionColVector = batch.cols[colIdx];
+            // The partition column might not be part of the current projection - allocate a CV and inject constants.
+            if (partitionColVector == null) {
+              DataTypePhysicalVariation[] variations = vrbCtx.getRowdataTypePhysicalVariations();
+              DataTypePhysicalVariation variation = variations != null ?
+                  variations[colIdx] : DataTypePhysicalVariation.NONE;
+              partitionColVector = VectorizedBatchUtil.createColumnVector(
+                  vrbCtx.getRowColumnTypeInfos()[colIdx], variation);
+              partitionColVector.init();
+              batch.cols[colIdx] = partitionColVector;
             }
+            vrbCtx.addPartitionColsToBatch(partitionColVector, partitionValues[i], colIdx);
           }
         }
         // Fill virtual columns

diff --git a/...rg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/AbstractIcebergRecordReader.java b/...rg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/AbstractIcebergRecordReader.java
@@ -19,6 +19,7 @@
 
 package org.apache.iceberg.mr.mapreduce;
 
+import java.util.Set;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.mapreduce.InputSplit;
 import org.apache.hadoop.mapreduce.RecordReader;
@@ -67,19 +68,24 @@ public void initialize(InputSplit split, TaskAttemptContext newContext) {
 
   private static Schema readSchema(Configuration conf, Table table, boolean caseSensitive) {
     Schema readSchema = InputFormatConfig.readSchema(conf);
+    String[] selectedColumns = InputFormatConfig.selectedColumns(conf);
+    Set<Integer> requiredSourceIds = requiredIdentityPartitionSourceIds(conf, table, selectedColumns);
 
     if (readSchema != null) {
-      return readSchema;
+      return IcebergTableUtil.includeIdentityPartitionSourceColumns(
+          readSchema, table.schema(), table.spec(), requiredSourceIds);
     }
 
-    String[] selectedColumns = InputFormatConfig.selectedColumns(conf);
     readSchema = table.schema();
 
     if (selectedColumns != null) {
       readSchema = caseSensitive ?
           readSchema.select(selectedColumns) : readSchema.caseInsensitiveSelect(selectedColumns);
     }
 
+    readSchema = IcebergTableUtil.includeIdentityPartitionSourceColumns(
+        readSchema, table.schema(), table.spec(), requiredSourceIds);
+
     if (InputFormatConfig.fetchVirtualColumns(conf)) {
       readSchema = IcebergAcidUtil.createFileReadSchemaWithVirtualColums(readSchema.columns());
       if (IcebergTableUtil.supportsRowLineage(table.properties())) {
@@ -90,6 +96,20 @@ private static Schema readSchema(Configuration conf, Table table, boolean caseSe
     return readSchema;
   }
 
+  /**
+   * For explicit READ_SCHEMA (Iceberg InputFormat API / unit tests), only expand for columns
+   * referenced by the scan filter. For Hive execution (column names pushed via SELECTED_COLUMNS),
+   * include all identity partition source columns to match HiveIcebergSerDe.
+   */
+  private static Set<Integer> requiredIdentityPartitionSourceIds(Configuration conf, Table table,
+      String[] selectedColumns) {
+    if (InputFormatConfig.readSchema(conf) != null) {
+      return IcebergTableUtil.requiredIdentityPartitionSourceIds(
+          conf, table.schema(), table.spec(), selectedColumns);
+    }
+    return table.spec().identitySourceIds();
+  }
+
   CloseableIterable<T> applyResidualFiltering(CloseableIterable<T> iter, Expression residual,
                                               Schema readSchema) {
     boolean applyResidual = !getContext().getConfiguration()

diff --git a/...rg/iceberg-handler/src/test/queries/positive/update_iceberg_cow_null_identity_partition.q b/...rg/iceberg-handler/src/test/queries/positive/update_iceberg_cow_null_identity_partition.q
@@ -0,0 +1,33 @@
+-- SORT_QUERY_RESULTS
+
+drop table if exists iceberg_cow_partitioned;
+
+create external table iceberg_cow_partitioned (
+  index int,
+  string_col string,
+  boolean_col boolean,
+  str_col string,
+  tinyint_col int
+) partitioned by spec(str_col, tinyint_col)
+stored by iceberg
+tblproperties ('write.update.mode'='copy-on-write');
+
+insert into iceberg_cow_partitioned partition (str_col, tinyint_col)
+  values (1, 'a', true, null, 0);
+
+explain update iceberg_cow_partitioned set str_col = 'UPDATED NULLS' where str_col is null;
+update iceberg_cow_partitioned set str_col = 'UPDATED NULLS' where str_col is null;
+
+select * from iceberg_cow_partitioned;
+
+-- Disable vectorization
+
+set hive.vectorized.execution.enabled=false;
+
+insert into iceberg_cow_partitioned partition (str_col, tinyint_col)
+  values (2, 'b', false, null, 1);
+
+explain update iceberg_cow_partitioned set str_col = 'UPDATED NULLS' where str_col is null;
+update iceberg_cow_partitioned set str_col = 'UPDATED NULLS' where str_col is null;
+
+select * from iceberg_cow_partitioned;