[Kernel] Minor refactor to column mapping related code for clarity (delta-io#4348)

<!--
Thanks for sending a pull request!  Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://github.com/delta-io/delta/blob/master/CONTRIBUTING.md
2. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP]
Your PR title ...'.
  3. Be sure to keep the PR description updated to reflect all changes.
  4. Please write your PR title to summarize what this PR proposes.
5. If possible, provide a concise example to reproduce the issue for a
faster review.
6. If applicable, include the corresponding issue number in the PR title
and link it in the body.
-->

#### Which Delta project/connector is this regarding?
<!--
Please add the component selected below to the beginning of the pull
request title
For example: [Spark] Title of my pull request
-->

- [ ] Spark
- [ ] Standalone
- [ ] Flink
- [X] Kernel
- [ ] Other (fill in here)

## Description

Rename some fields, improve docs, and move some code for better clarity
/ readability.

## How was this patch tested?

Existing tests suffice.

## Does this PR introduce _any_ user-facing changes?

No

Author: allisonport-db

kernel/kernel-api/src/main/java/io/delta/kernel/internal/TransactionImpl.java

@@ -47,7 +47,6 @@
 import io.delta.kernel.internal.tablefeatures.TableFeatures;
 import io.delta.kernel.internal.util.*;
 import io.delta.kernel.internal.util.Clock;
-import io.delta.kernel.internal.util.ColumnMapping;
 import io.delta.kernel.internal.util.FileNames;
 import io.delta.kernel.internal.util.InCommitTimestampUtils;
 import io.delta.kernel.internal.util.VectorUtils;
@@ -122,14 +121,7 @@ public TransactionImpl(
 
   @Override
   public Row getTransactionState(Engine engine) {
-    ColumnMapping.ColumnMappingMode mappingMode =
-        ColumnMapping.getColumnMappingMode(metadata.getConfiguration());
-    StructType basePhysicalSchema = metadata.getSchema();
-    StructType physicalSchema =
-        ColumnMapping.convertToPhysicalSchema(
-            metadata.getSchema(), basePhysicalSchema, mappingMode);
-
-    return TransactionStateRow.of(metadata, dataPath.toString(), physicalSchema);
+    return TransactionStateRow.of(metadata, dataPath.toString());
   }
 
   @Override

kernel/kernel-api/src/main/java/io/delta/kernel/internal/actions/Metadata.java

@@ -23,6 +23,7 @@
 import io.delta.kernel.internal.data.GenericRow;
 import io.delta.kernel.internal.lang.Lazy;
 import io.delta.kernel.internal.types.DataTypeJsonSerDe;
+import io.delta.kernel.internal.util.ColumnMapping;
 import io.delta.kernel.internal.util.VectorUtils;
 import io.delta.kernel.types.*;
 import java.util.*;
@@ -239,6 +240,18 @@ public Map<String, String> getConfiguration() {
     return Collections.unmodifiableMap(configuration.get());
   }
 
+  /**
+   * The full schema (including partition columns) with the field names converted to their physical
+   * names (column names used in the data files) based on the table's column mapping mode. When
+   * column mapping mode is ID, fieldId metadata is preserved in the field metadata; all column
+   * metadata is otherwise removed.
+   */
+  public StructType getPhysicalSchema() {
+    ColumnMapping.ColumnMappingMode mappingMode =
+        ColumnMapping.getColumnMappingMode(getConfiguration());
+    return ColumnMapping.convertToPhysicalSchema(schema, schema, mappingMode);
+  }
+
   /**
    * Filter out the key-value pair matches exactly with the old properties.
    *

kernel/kernel-api/src/main/java/io/delta/kernel/internal/data/TransactionStateRow.java

@@ -46,11 +46,11 @@ public class TransactionStateRow extends GenericRow {
           .boxed()
           .collect(toMap(i -> SCHEMA.at(i).getName(), i -> i));
 
-  public static TransactionStateRow of(
-      Metadata metadata, String tablePath, StructType physicalSchema) {
+  public static TransactionStateRow of(Metadata metadata, String tablePath) {
     HashMap<Integer, Object> valueMap = new HashMap<>();
     valueMap.put(COL_NAME_TO_ORDINAL.get("logicalSchemaString"), metadata.getSchemaString());
-    valueMap.put(COL_NAME_TO_ORDINAL.get("physicalSchemaString"), physicalSchema.toJson());
+    valueMap.put(
+        COL_NAME_TO_ORDINAL.get("physicalSchemaString"), metadata.getPhysicalSchema().toJson());
     valueMap.put(COL_NAME_TO_ORDINAL.get("partitionColumns"), metadata.getPartitionColumns());
     valueMap.put(COL_NAME_TO_ORDINAL.get("configuration"), metadata.getConfigurationMapValue());
     valueMap.put(COL_NAME_TO_ORDINAL.get("tablePath"), tablePath);

kernel/kernel-api/src/main/java/io/delta/kernel/internal/util/ColumnMapping.java

@@ -88,21 +88,26 @@ public static ColumnMappingMode getColumnMappingMode(Map<String, String> configu
 
   /**
    * Helper method that converts the logical schema (requested by the connector) to physical schema
-   * of the data stored in data files based on the table's column mapping mode.
+   * of the data stored in data files based on the table's column mapping mode. Field-id column
+   * metadata is preserved when cmMode = ID, all column metadata is otherwise removed.
    *
-   * @param logicalSchema Logical schema of the scan
-   * @param physicalSchema Physical schema of the scan
+   * <p>We require {@code fullSchema} in addition to the pruned schema we want to convert since we
+   * need the complete field metadata as it is stored in the schema in the _delta_log. We cannot be
+   * sure (and do not enforce) that this metadata is preserved by the connector.
+   *
+   * @param prunedSchema the logical read schema requested by the connector
+   * @param fullSchema the full delta schema (with complete metadata) as read from the _delta_log
    * @param columnMappingMode Column mapping mode
    */
   public static StructType convertToPhysicalSchema(
-      StructType logicalSchema, StructType physicalSchema, ColumnMappingMode columnMappingMode) {
+      StructType prunedSchema, StructType fullSchema, ColumnMappingMode columnMappingMode) {
     switch (columnMappingMode) {
       case NONE:
-        return logicalSchema;
+        return prunedSchema;
       case ID: // fall through
       case NAME:
         boolean includeFieldIds = columnMappingMode == ColumnMappingMode.ID;
-        return convertToPhysicalSchema(logicalSchema, physicalSchema, includeFieldIds);
+        return convertToPhysicalSchema(prunedSchema, fullSchema, includeFieldIds);
       default:
         throw new UnsupportedOperationException(
             "Unsupported column mapping mode: " + columnMappingMode);
@@ -239,20 +244,20 @@ private static int findMaxColumnId(StructField field, int maxColumnId) {
   /**
    * Utility method to convert the given logical schema to physical schema, recursively converting
    * sub-types in case of complex types. When {@code includeFieldId} is true, converted physical
-   * schema will have field ids in the metadata.
+   * schema will have field ids in the metadata. Column metadata is otherwise removed.
    */
   private static StructType convertToPhysicalSchema(
-      StructType logicalSchema, StructType physicalSchema, boolean includeFieldId) {
+      StructType prunedSchema, StructType fullSchema, boolean includeFieldId) {
     StructType newSchema = new StructType();
-    for (StructField logicalField : logicalSchema.fields()) {
-      DataType logicalType = logicalField.getDataType();
-      StructField physicalField = physicalSchema.get(logicalField.getName());
+    for (StructField prunedField : prunedSchema.fields()) {
+      StructField completeField = fullSchema.get(prunedField.getName());
       DataType physicalType =
-          convertToPhysicalType(logicalType, physicalField.getDataType(), includeFieldId);
-      String physicalName = physicalField.getMetadata().getString(COLUMN_MAPPING_PHYSICAL_NAME_KEY);
+          convertToPhysicalType(
+              prunedField.getDataType(), completeField.getDataType(), includeFieldId);
+      String physicalName = completeField.getMetadata().getString(COLUMN_MAPPING_PHYSICAL_NAME_KEY);
 
       if (includeFieldId) {
-        Long fieldId = physicalField.getMetadata().getLong(COLUMN_MAPPING_ID_KEY);
+        Long fieldId = completeField.getMetadata().getLong(COLUMN_MAPPING_ID_KEY);
         FieldMetadata.Builder builder =
             FieldMetadata.builder().putLong(PARQUET_FIELD_ID_KEY, fieldId);
 
@@ -261,15 +266,15 @@ private static StructType convertToPhysicalSchema(
         // the 'element' and 'key'/'value' fields of Arrays/Maps haven been written,
         // then IcebergCompatV2 is enabled because the schema we are looking at is from
         // the DeltaLog and has nested field IDs setup
-        if (hasNestedColumnIds(physicalField)) {
+        if (hasNestedColumnIds(completeField)) {
           builder.putFieldMetadata(
-              PARQUET_FIELD_NESTED_IDS_METADATA_KEY, getNestedColumnIds(physicalField));
+              PARQUET_FIELD_NESTED_IDS_METADATA_KEY, getNestedColumnIds(completeField));
         }
 
         newSchema =
-            newSchema.add(physicalName, physicalType, logicalField.isNullable(), builder.build());
+            newSchema.add(physicalName, physicalType, prunedField.isNullable(), builder.build());
       } else {
-        newSchema = newSchema.add(physicalName, physicalType, logicalField.isNullable());
+        newSchema = newSchema.add(physicalName, physicalType, prunedField.isNullable());
       }
     }
     return newSchema;

kernel/kernel-api/src/test/scala/io/delta/kernel/TransactionSuite.scala

@@ -156,7 +156,7 @@ class TransactionSuite extends AnyFunSuite with VectorTestUtils with MockEngineU
         VectorUtils.buildArrayValue(Seq.empty.asJava, StringType.STRING),
         Optional.empty(),
         stringStringMapValue(configMap.asJava))
-      val txnState = TransactionStateRow.of(metadata, "table path", schema)
+      val txnState = TransactionStateRow.of(metadata, "table path")
 
       // Get statistics columns and define expected result
       val statsColumns = TransactionImpl.getStatisticsColumns(txnState)
@@ -262,7 +262,7 @@ object TransactionSuite extends VectorTestUtils with MockEngineUtils {
       Optional.empty(), // createdTime
       stringStringMapValue(configurationMap.asJava) // configurationMap
     )
-    TransactionStateRow.of(metadata, "table path", schema)
+    TransactionStateRow.of(metadata, "table path")
   }
 
   def testStats(numRowsOpt: Option[Long]): Option[DataFileStatistics] = {