Skip to content

HIVE-29828: Introduce PARTITION__NAME virtual column and refactor Iceberg virtual column plumbing #6711

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
deniskuzZ wants to merge 2 commits into
base: master
Choose a base branch
from
Open

HIVE-29828: Introduce PARTITION__NAME virtual column and refactor Iceberg virtual column plumbing #6711

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
987043a
HIVE-29781: Introduce PARTITION__NAME virtual column and refactor Ice…
deniskuzZ Aug 20, 2026
ced7038
HIVE-29781: Make PARTITION__PROJECTION a merge task record field inst…
deniskuzZ Aug 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 11 additions & 6 deletions iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergSerDe.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -140,8 +140,7 @@ public void initialize(Configuration conf, Properties serDeProperties,
}
}

this.projectedSchema =
projectedSchema(conf, serDeProperties.getProperty(Catalogs.NAME), tableSchema, jobConf);
this.projectedSchema = projectedSchema(conf, serDeProperties, tableSchema, jobConf);

if (!IcebergTableUtil.isFanoutEnabled(serDeProperties::getProperty)) {
// ClusteredWriter requires that records are ordered by partition keys.
Expand All @@ -156,9 +155,11 @@ public void initialize(Configuration conf, Properties serDeProperties,
}
}

private static Schema projectedSchema(Configuration conf, String tableName, Schema tableSchema,
Map<String, String> jobConf) {
private static Schema projectedSchema(Configuration conf, Properties serDeProperties,
Schema tableSchema, Map<String, String> jobConf) {
String tableName = serDeProperties.getProperty(Catalogs.NAME);
Context.Operation operation = HiveCustomStorageHandlerUtils.getWriteOperation(conf::get, tableName);

if (operation == null) {
jobConf.put(InputFormatConfig.CASE_SENSITIVE, "false");
String[] selectedColumns = ColumnProjectionUtils.getReadColumnNames(conf);
Expand All @@ -180,11 +181,15 @@ private static Schema projectedSchema(Configuration conf, String tableName, Sche
}
boolean isCOW = IcebergTableUtil.isCopyOnWriteMode(operation, conf::get);
if (isCOW) {
return getSchemaWithRowLineage(IcebergAcidUtil.createSerdeSchemaForDelete(tableSchema.columns()), conf);
return getSchemaWithRowLineage(
IcebergAcidUtil.createSerdeSchemaForDelete(tableSchema.columns(), false), conf);
}
switch (operation) {
case DELETE:
return IcebergAcidUtil.createSerdeSchemaForDelete(tableSchema.columns());
boolean isMergeTask = HiveCustomStorageHandlerUtils.isMergeTaskEnabled(
key -> serDeProperties.getProperty(key, conf.get(key)),
tableName);
return IcebergAcidUtil.createSerdeSchemaForDelete(tableSchema.columns(), isMergeTask);
case UPDATE:
return IcebergAcidUtil.createSerdeSchemaForUpdate(tableSchema.columns());
case OTHER:
Expand Down
7 changes: 4 additions & 3 deletions ...g/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -214,7 +214,7 @@
import static org.apache.hadoop.hive.ql.metadata.VirtualColumn.FILE_PATH;
import static org.apache.hadoop.hive.ql.metadata.VirtualColumn.LAST_UPDATED_SEQUENCE_NUMBER;
import static org.apache.hadoop.hive.ql.metadata.VirtualColumn.PARTITION_HASH;
import static org.apache.hadoop.hive.ql.metadata.VirtualColumn.PARTITION_PROJECTION;
import static org.apache.hadoop.hive.ql.metadata.VirtualColumn.PARTITION_NAME;
import static org.apache.hadoop.hive.ql.metadata.VirtualColumn.PARTITION_SPEC_ID;
import static org.apache.hadoop.hive.ql.metadata.VirtualColumn.ROW_LINEAGE_ID;
import static org.apache.hadoop.hive.ql.metadata.VirtualColumn.ROW_POSITION;
Expand Down Expand Up @@ -244,7 +244,7 @@ public class HiveIcebergStorageHandler extends DefaultStorageHandler implements
public static final String MERGE_ON_READ = RowLevelOperationMode.MERGE_ON_READ.modeName();

private static final List<VirtualColumn> ACID_VIRTUAL_COLS = ImmutableList.of(
PARTITION_SPEC_ID, PARTITION_HASH, FILE_PATH, ROW_POSITION, PARTITION_PROJECTION);
PARTITION_SPEC_ID, PARTITION_HASH, FILE_PATH, ROW_POSITION);

private static final List<FieldSchema> ACID_VIRTUAL_COLS_AS_FIELD_SCHEMA = schema(ACID_VIRTUAL_COLS);

Expand Down Expand Up @@ -427,7 +427,8 @@ public DecomposedPredicate decomposePredicate(JobConf jobConf, Deserializer dese

List<ExprNodeDesc> subExprNodes = pushedPredicate.getChildren();
Set<String> skipList =
Stream.of(FILE_PATH, PARTITION_SPEC_ID, PARTITION_HASH, ROW_LINEAGE_ID, LAST_UPDATED_SEQUENCE_NUMBER)
Stream.of(FILE_PATH, PARTITION_SPEC_ID, PARTITION_HASH, PARTITION_NAME,
ROW_LINEAGE_ID, LAST_UPDATED_SEQUENCE_NUMBER)
.map(VirtualColumn::getName).collect(Collectors.toSet());

if (subExprNodes.removeIf(nodeDesc -> nodeDesc.getCols() != null &&
Expand Down
155 changes: 76 additions & 79 deletions iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergAcidUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,15 +25,15 @@
import java.util.Objects;
import java.util.Optional;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.ql.io.IOContextMap;
import org.apache.hadoop.hive.ql.io.PositionDeleteInfo;
import org.apache.hadoop.hive.ql.io.RowLineageInfo;
import org.apache.hadoop.hive.ql.lockmgr.HiveTxnManager;
import org.apache.hadoop.hive.ql.metadata.VirtualColumn;
import org.apache.hadoop.hive.ql.session.SessionState;
import org.apache.hadoop.hive.ql.session.SessionStateUtil;
import org.apache.iceberg.ContentFile;
import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.MetadataColumns;
import org.apache.iceberg.PartitionKey;
import org.apache.iceberg.PartitionSpec;
Expand All @@ -51,16 +51,13 @@
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.SerializationUtil;
import org.apache.iceberg.util.StructProjection;

public class IcebergAcidUtil {

private IcebergAcidUtil() {
}

private static final Types.NestedField PARTITION_STRUCT_META_COL = null; // placeholder value in the map
private static final Map<Types.NestedField, Integer> FILE_READ_META_COLS = Maps.newLinkedHashMap();
private static final Map<String, Types.NestedField> VIRTUAL_COLS_TO_META_COLS = Maps.newLinkedHashMap();
public static final String META_TABLE_PROPERTY = "metaTable";
private static final Map<Types.NestedField, Integer> DELETE_FILE_META_COLS = Maps.newLinkedHashMap();
public static final Integer PARTITION_PROJECTION_COLUMN_ID = Integer.MAX_VALUE - 6;
Expand All @@ -71,59 +68,52 @@ private IcebergAcidUtil() {
DELETE_FILE_META_COLS.put(MetadataColumns.ROW_POSITION, 1);

FILE_READ_META_COLS.put(MetadataColumns.SPEC_ID, 0);
FILE_READ_META_COLS.put(PARTITION_STRUCT_META_COL, 1);
FILE_READ_META_COLS.put(MetadataColumns.FILE_PATH, 2);
FILE_READ_META_COLS.put(MetadataColumns.ROW_POSITION, 3);

VIRTUAL_COLS_TO_META_COLS.put(VirtualColumn.PARTITION_SPEC_ID.getName(), MetadataColumns.SPEC_ID);
VIRTUAL_COLS_TO_META_COLS.put(VirtualColumn.PARTITION_HASH.getName(), PARTITION_STRUCT_META_COL);
VIRTUAL_COLS_TO_META_COLS.put(VirtualColumn.FILE_PATH.getName(), MetadataColumns.FILE_PATH);
VIRTUAL_COLS_TO_META_COLS.put(VirtualColumn.ROW_POSITION.getName(), MetadataColumns.ROW_POSITION);
VIRTUAL_COLS_TO_META_COLS.put(VirtualColumn.ROW_LINEAGE_ID.getName(), MetadataColumns.ROW_ID);
VIRTUAL_COLS_TO_META_COLS.put(VirtualColumn.LAST_UPDATED_SEQUENCE_NUMBER.getName(),
MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER);
FILE_READ_META_COLS.put(MetadataColumns.FILE_PATH, 1);
FILE_READ_META_COLS.put(MetadataColumns.ROW_POSITION, 2);
}

private static final Types.NestedField PARTITION_HASH_META_COL = Types.NestedField.required(
MetadataColumns.PARTITION_COLUMN_ID, MetadataColumns.PARTITION_COLUMN_NAME, Types.LongType.get());

public static final Types.NestedField PARTITION_PROJECTION = Types.NestedField.required(
private static final Types.NestedField PARTITION_PROJECTION = Types.NestedField.required(
PARTITION_PROJECTION_COLUMN_ID, PARTITION_PROJECTION_COLUMN_NAME, Types.StringType.get());

private static final Map<Types.NestedField, Integer> SERDE_META_COLS = Maps.newLinkedHashMap();

// a merge task reads delete files, so its writer has no row data to derive the partition key from
private static final Map<Types.NestedField, Integer> MERGE_SERDE_META_COLS = Maps.newLinkedHashMap();

static {
SERDE_META_COLS.put(MetadataColumns.SPEC_ID, 0);
SERDE_META_COLS.put(PARTITION_HASH_META_COL, 1);
SERDE_META_COLS.put(MetadataColumns.FILE_PATH, 2);
SERDE_META_COLS.put(MetadataColumns.ROW_POSITION, 3);
SERDE_META_COLS.put(PARTITION_PROJECTION, 4);

MERGE_SERDE_META_COLS.putAll(SERDE_META_COLS);
MERGE_SERDE_META_COLS.put(PARTITION_PROJECTION, 4);
}

/**
* @param dataCols The columns of the original file read schema
* @param table The table object - it is used for populating the partition struct meta column
* @return The schema for reading files, extended with metadata columns
*/
public static Schema createFileReadSchemaWithVirtualColums(List<Types.NestedField> dataCols, Table table) {
public static Schema createFileReadSchemaWithVirtualColums(List<Types.NestedField> dataCols) {
List<Types.NestedField> cols = Lists.newArrayListWithCapacity(dataCols.size() + FILE_READ_META_COLS.size());
FILE_READ_META_COLS.forEach((metaCol, index) -> {
if (metaCol == PARTITION_STRUCT_META_COL) {
cols.add(MetadataColumns.metadataColumn(table, MetadataColumns.PARTITION_COLUMN_NAME));
} else {
cols.add(metaCol);
}
});
FILE_READ_META_COLS.forEach((metaCol, index) -> cols.add(metaCol));
cols.addAll(dataCols);
return new Schema(cols);
}

/**
* @param dataCols The columns of the serde projection schema
* @param isMergeTask Whether the schema is for a merge task, which also carries the partition key
* @return The schema for SerDe operations, extended with metadata columns needed for deletes
*/
public static Schema createSerdeSchemaForDelete(List<Types.NestedField> dataCols) {
List<Types.NestedField> cols = Lists.newArrayListWithCapacity(dataCols.size() + SERDE_META_COLS.size());
SERDE_META_COLS.forEach((metaCol, index) -> cols.add(metaCol));
public static Schema createSerdeSchemaForDelete(List<Types.NestedField> dataCols, boolean isMergeTask) {
Map<Types.NestedField, Integer> metaCols = isMergeTask ?
MERGE_SERDE_META_COLS : SERDE_META_COLS;
List<Types.NestedField> cols = Lists.newArrayListWithCapacity(dataCols.size() + metaCols.size());
cols.addAll(metaCols.keySet());
cols.addAll(dataCols);
return new Schema(cols);
}
Expand All @@ -133,14 +123,17 @@ public static Schema createSerdeSchemaForDelete(List<Types.NestedField> dataCols
* the field values from `rec`.
* @param rec The record read by the file scan task, which contains both the metadata fields and the row data fields
* @param rowData The record object to populate with the rowData fields only
* @param isMergeTask Whether the record was built by a merge task
* @return The position delete object
*/
public static PositionDelete<Record> getPositionDelete(Record rec, Record rowData) {
public static PositionDelete<Record> getPositionDelete(Record rec, Record rowData, boolean isMergeTask) {
Map<Types.NestedField, Integer> metaCols = isMergeTask ?
MERGE_SERDE_META_COLS : SERDE_META_COLS;
PositionDelete<Record> positionDelete = PositionDelete.create();
String filePath = rec.get(SERDE_META_COLS.get(MetadataColumns.FILE_PATH), String.class);
Long filePosition = rec.get(SERDE_META_COLS.get(MetadataColumns.ROW_POSITION), Long.class);
String filePath = rec.get(metaCols.get(MetadataColumns.FILE_PATH), String.class);
Long filePosition = rec.get(metaCols.get(MetadataColumns.ROW_POSITION), Long.class);

int dataOffset = SERDE_META_COLS.size(); // position in the rec where the actual row data begins
int dataOffset = metaCols.size(); // position in the rec where the actual row data begins
for (int i = dataOffset; i < rec.size(); ++i) {
rowData.set(i - dataOffset, rec.get(i));
}
Expand Down Expand Up @@ -169,14 +162,8 @@ public static int parseSpecId(Record rec) {
return rec.get(FILE_READ_META_COLS.get(MetadataColumns.SPEC_ID), Integer.class);
}

public static long computePartitionHash(Record rec) {
StructProjection part = rec.get(FILE_READ_META_COLS.get(PARTITION_STRUCT_META_COL), StructProjection.class);
// we need to compute a hash value for the partition struct so that it can be used as a sorting key
return computeHash(part);
}

public static PartitionKey parsePartitionKey(Record rec) {
String serializedStr = rec.get(SERDE_META_COLS.get(PARTITION_PROJECTION), String.class);
String serializedStr = rec.get(MERGE_SERDE_META_COLS.get(PARTITION_PROJECTION), String.class);
return SerializationUtil.deserializeFromBase64(serializedStr);
}

Expand All @@ -190,15 +177,11 @@ public static String getSerializedPartitionKey(StructLike structLike, PartitionS
return SerializationUtil.serializeToBase64(partitionKey);
}

public static String parseFilePath(Record rec) {
return rec.get(FILE_READ_META_COLS.get(MetadataColumns.FILE_PATH), String.class);
}

public static String getFilePath(Record rec) {
return rec.get(DELETE_FILE_META_COLS.get(MetadataColumns.FILE_PATH), String.class);
}

public static long parseFilePosition(Record rec) {
public static long getFilePosition(Record rec) {
return rec.get(FILE_READ_META_COLS.get(MetadataColumns.ROW_POSITION), Long.class);
}

Expand Down Expand Up @@ -261,12 +244,23 @@ public static class VirtualColumnAwareIterator<T> implements CloseableIterator<T
private final GenericRecord current;
private final Configuration conf;

public VirtualColumnAwareIterator(
CloseableIterator<T> currentIterator, Schema expectedSchema, Configuration conf) {
private final int specId;
private final long partitionHash;
private final String filePath;

public VirtualColumnAwareIterator(CloseableIterator<T> currentIterator, List<Types.NestedField> columns,
Configuration conf, FileScanTask task) {
this.currentIterator = currentIterator;
this.current = GenericRecord.create(
new Schema(expectedSchema.columns().subList(4, expectedSchema.columns().size())));
new Schema(columns.subList(FILE_READ_META_COLS.size(), columns.size())));
this.conf = conf;

this.specId = task.file().specId();
this.partitionHash = computeHash(task.file().partition());
this.filePath = task.file().location();

IOContextMap.get(conf).setPartitionName(
IcebergTableUtil.toPartitionName(task.spec(), task.file().partition()));
}

@Override
Expand All @@ -285,11 +279,10 @@ public T next() {
GenericRecord rec = (GenericRecord) next;
IcebergAcidUtil.copyFields(rec, FILE_READ_META_COLS.size(), current.size(), current);
PositionDeleteInfo.setIntoConf(conf,
IcebergAcidUtil.parseSpecId(rec),
IcebergAcidUtil.computePartitionHash(rec),
IcebergAcidUtil.parseFilePath(rec),
IcebergAcidUtil.parseFilePosition(rec),
StringUtils.EMPTY);
specId,
partitionHash,
filePath,
IcebergAcidUtil.getFilePosition(rec));
RowLineageInfo.setRowLineageInfoIntoConf(RowLineageReader.readRowId(rec),
RowLineageReader.readLastUpdatedSequenceNumber(rec), conf);
return (T) current;
Expand All @@ -299,17 +292,20 @@ public T next() {
public static class MergeTaskVirtualColumnAwareIterator<T> implements CloseableIterator<T> {

private final CloseableIterator<T> currentIterator;
private final GenericRecordBuilder<T> recordBuilder;
private final PartitionSpec partitionSpec;
private final StructLike partition;
private final MergeTaskRecordBuilder<T> recordBuilder;

public MergeTaskVirtualColumnAwareIterator(
CloseableIterator<T> currentIterator, Schema expectedSchema, ContentFile<?> contentFile, Table table) {
private final int specId;
private final long partitionHash;
private final String serializedPartitionKey;

public MergeTaskVirtualColumnAwareIterator(CloseableIterator<T> currentIterator, Schema expectedSchema,
PartitionSpec spec, ContentFile<?> file) {
this.currentIterator = currentIterator;
this.partition = contentFile.partition();
this.recordBuilder = new GenericRecordBuilder<>(
new Schema(expectedSchema.columns().subList(0, expectedSchema.columns().size())));
this.partitionSpec = table.specs().get(contentFile.specId());
this.recordBuilder = new MergeTaskRecordBuilder<>(expectedSchema);

this.specId = spec.specId();
this.partitionHash = computeHash(file.partition());
this.serializedPartitionKey = getSerializedPartitionKey(file.partition(), spec);
}

@Override
Expand All @@ -326,44 +322,45 @@ public boolean hasNext() {
public T next() {
T next = currentIterator.next();
GenericRecord rec = (GenericRecord) next;
return recordBuilder.withSpecId(partitionSpec.specId())
.withPartitionHash(computeHash(partition))

return recordBuilder.withSpecId(specId)
.withPartitionHash(partitionHash)
.withFilePath(IcebergAcidUtil.getFilePath(rec))
.withFilePosition(IcebergAcidUtil.getDeleteFilePosition(rec))
.withPartitionKey(getSerializedPartitionKey(partition, partitionSpec)).build();
.withPartitionKey(serializedPartitionKey)
.build();
}
}

private static final class GenericRecordBuilder<T> {

private static final class MergeTaskRecordBuilder<T> {
private final GenericRecord current;

GenericRecordBuilder(Schema schema) {
MergeTaskRecordBuilder(Schema schema) {
current = GenericRecord.create(schema);
}

public GenericRecordBuilder<T> withSpecId(int specId) {
current.set(SERDE_META_COLS.get(MetadataColumns.SPEC_ID), specId);
public MergeTaskRecordBuilder<T> withSpecId(int specId) {
current.set(MERGE_SERDE_META_COLS.get(MetadataColumns.SPEC_ID), specId);
return this;
}

public GenericRecordBuilder<T> withPartitionHash(long partitionHash) {
current.set(SERDE_META_COLS.get(PARTITION_HASH_META_COL), partitionHash);
public MergeTaskRecordBuilder<T> withPartitionHash(long partitionHash) {
current.set(MERGE_SERDE_META_COLS.get(PARTITION_HASH_META_COL), partitionHash);
return this;
}

public GenericRecordBuilder<T> withFilePath(String filePath) {
current.set(SERDE_META_COLS.get(MetadataColumns.FILE_PATH), filePath);
public MergeTaskRecordBuilder<T> withFilePath(String filePath) {
current.set(MERGE_SERDE_META_COLS.get(MetadataColumns.FILE_PATH), filePath);
return this;
}

public GenericRecordBuilder<T> withFilePosition(long filePosition) {
current.set(SERDE_META_COLS.get(MetadataColumns.ROW_POSITION), filePosition);
public MergeTaskRecordBuilder<T> withFilePosition(long filePosition) {
current.set(MERGE_SERDE_META_COLS.get(MetadataColumns.ROW_POSITION), filePosition);
return this;
}

public GenericRecordBuilder<T> withPartitionKey(String serializedPartitionKey) {
current.set(SERDE_META_COLS.get(PARTITION_PROJECTION), serializedPartitionKey);
public MergeTaskRecordBuilder<T> withPartitionKey(String serializedPartitionKey) {
current.set(MERGE_SERDE_META_COLS.get(PARTITION_PROJECTION), serializedPartitionKey);
return this;
}

Expand Down
6 changes: 4 additions & 2 deletions iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -493,8 +493,10 @@ public static PartitionData toPartitionData(StructLike sourceKey, Types.StructTy
* name an unpartitioned spec renders. Statistics and partition pruning join on this name, so both must
* render it the same way.
*/
static String toPartitionName(PartitionSpec spec, PartitionData data) {
return StringUtils.defaultIfEmpty(spec.partitionToPath(data), DummyPartition.VOID);
public static String toPartitionName(PartitionSpec spec, StructLike data) {
String path = spec.partitionToPath(data);
// an unpartitioned spec renders nothing: its rows belong to the table-level partition
return path.isEmpty() ? DummyPartition.VOID : path;

@difin difin Aug 25, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Previously it was using StringUtils.defaultIfEmpty(..), which used to handle null, but now path.isEmpty() may throw NPE, it may be a null-safety regression.

@deniskuzZ deniskuzZ Aug 25, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

path comes from Iceberg's PartitionSpec.partitionToPath, which returns StringBuilder.toString() unconditionally — never null

}

/**
Expand Down
Loading
Loading