[Kernel][Clustering #2] add getPhysicalColumnNameAndDataType util in ColumnMapping (delta-io#4319)

<!--
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

<!--
- Describe what this PR changes.
- Describe why we need the change.
 
If this PR resolves an issue be sure to include "Resolves #XXX" to
correctly link and close the issue upon merge.
-->
Split the main PR delta-io#4265 for faster
review

Add a util func `convertToPhysicalColumnNames` in ColumnMapping to get
the corresponding physical column name for a logical column
## How was this patch tested?

<!--
If tests were added, say they were added here. Please make sure to test
the changes thoroughly including negative and positive cases if
possible.
If the changes were tested in any way other than unit tests, please
clarify how you tested step by step (ideally copy and paste-able, so
that other reviewers can test and check, and descendants can verify in
the future).
If the changes were not tested, please explain why.
-->
Add unit test cases in ColumnMappingSuite.scala

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

<!--
If yes, please clarify the previous behavior and the change this PR
proposes - provide the console output, description and/or an example to
show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change
compared to the released Delta Lake versions or within the unreleased
branches such as master.
If no, write 'No'.
-->

Author: KaiqiJinWow

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

@@ -18,6 +18,7 @@
 import static java.lang.String.format;
 
 import io.delta.kernel.exceptions.*;
+import io.delta.kernel.expressions.Column;
 import io.delta.kernel.internal.actions.DomainMetadata;
 import io.delta.kernel.internal.tablefeatures.TableFeature;
 import io.delta.kernel.types.DataType;
@@ -251,6 +252,11 @@ public static KernelException statsTypeMismatch(
     return new KernelException(format(msgFormat, fieldName, expected, actual));
   }
 
+  public static KernelException columnNotFoundInSchema(Column column, StructType tableSchema) {
+    return new KernelException(
+        format("Column '%s' was not found in the table schema: %s", column, tableSchema));
+  }
+
   /// Start: icebergCompat exceptions
   public static KernelException icebergCompatMissingNumRecordsStats(
       String compatVersion, DataFileStatus dataFileStatus) {

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

@@ -15,10 +15,12 @@
  */
 package io.delta.kernel.internal.util;
 
+import static io.delta.kernel.internal.DeltaErrors.columnNotFoundInSchema;
 import static io.delta.kernel.internal.util.Preconditions.checkArgument;
 import static java.util.Collections.singletonMap;
 
 import io.delta.kernel.exceptions.InvalidConfigurationValueException;
+import io.delta.kernel.expressions.Column;
 import io.delta.kernel.internal.TableConfig;
 import io.delta.kernel.internal.actions.Metadata;
 import io.delta.kernel.types.*;
@@ -157,6 +159,31 @@ public static Optional<Metadata> updateColumnMappingMetadataIfNeeded(
     }
   }
 
+  /** Returns the physical column and data type for a given logical column based on the schema. */
+  public static Tuple2<Column, DataType> getPhysicalColumnNameAndDataType(
+      StructType schema, Column logicalColumn) {
+    List<String> physicalNameParts = new ArrayList<>();
+    DataType currentType = schema;
+
+    // Traverse through each level of the logical name to resolve its corresponding physical name.
+    for (String namePart : logicalColumn.getNames()) {
+      if (!(currentType instanceof StructType)) {
+        throw columnNotFoundInSchema(logicalColumn, schema);
+      }
+
+      StructType structType = (StructType) currentType;
+      // Find the field in the current structure that matches the given name
+      StructField field =
+          structType.fields().stream()
+              .filter(f -> f.getName().equalsIgnoreCase(namePart))
+              .findFirst()
+              .orElseThrow(() -> columnNotFoundInSchema(logicalColumn, schema));
+      physicalNameParts.add(ColumnMapping.getPhysicalName(field));
+      currentType = field.getDataType();
+    }
+    return new Tuple2<>(new Column(physicalNameParts.toArray(new String[0])), currentType);
+  }
+
   ////////////////////////////
   // Private Helper Methods //
   ////////////////////////////

kernel/kernel-api/src/test/scala/io/delta/kernel/internal/util/ColumnMappingSuite.scala

@@ -17,10 +17,12 @@ package io.delta.kernel.internal.util
 
 import java.util
 
+import io.delta.kernel.exceptions.KernelException
+import io.delta.kernel.expressions.Column
 import io.delta.kernel.internal.actions.Metadata
 import io.delta.kernel.internal.util.ColumnMapping._
 import io.delta.kernel.internal.util.ColumnMapping.ColumnMappingMode._
-import io.delta.kernel.types.{ArrayType, FieldMetadata, IntegerType, MapType, StringType, StructField, StructType}
+import io.delta.kernel.types._
 
 import org.assertj.core.api.Assertions.{assertThat, assertThatNoException, assertThatThrownBy}
 import org.assertj.core.util.Maps
@@ -199,6 +201,105 @@ class ColumnMappingSuite extends AnyFunSuite with ColumnMappingSuiteBase {
     assertThat(ColumnMapping.findMaxColumnId(schema)).isEqualTo(12)
   }
 
+  private val testingSchema = new StructType()
+    .add("a", StringType.STRING)
+    .add(
+      "b",
+      new StructType()
+        .add("c", DoubleType.DOUBLE)
+        .add("d", DateType.DATE))
+    .add("e", FloatType.FLOAT)
+    .add(
+      "f",
+      new StructType()
+        .add(
+          "g",
+          new StructType()
+            .add("h", TimestampNTZType.TIMESTAMP_NTZ)))
+    .add("i", new MapType(StringType.STRING, DoubleType.DOUBLE, false))
+    .add("j", new ArrayType(StringType.STRING, false))
+
+  Seq(
+    (Array("a"), StringType.STRING),
+    (Array("b", "c"), DoubleType.DOUBLE),
+    (Array("b", "d"), DateType.DATE),
+    (Array("e"), FloatType.FLOAT),
+    (Array("f", "g", "h"), TimestampNTZType.TIMESTAMP_NTZ),
+    (Array("i"), new MapType(StringType.STRING, DoubleType.DOUBLE, false)),
+    (Array("j"), new ArrayType(StringType.STRING, false))).foreach {
+    case (columnName, expectedType) =>
+      test(s"get physical column name and dataType for $columnName") {
+        // case 1: column mapping disabled
+        val column = new Column(columnName)
+        val resultTuple =
+          ColumnMapping.getPhysicalColumnNameAndDataType(testingSchema, column)
+
+        val actualColumn = resultTuple._1
+        val actualType = resultTuple._2
+        assert(actualColumn == column)
+        assert(actualType == expectedType)
+
+        // case 2: column mapping disabled
+        val metadata: Metadata = updateColumnMappingMetadataIfNeeded(
+          testMetadata(testingSchema).withColumnMappingEnabled("id"),
+          true).orElseGet(() => fail("Metadata should not be empty"))
+
+        val physicalResultTuple = ColumnMapping.getPhysicalColumnNameAndDataType(
+          metadata.getSchema,
+          column)
+        val actualPhysicalColumn = physicalResultTuple._1
+        val actualPhysicalType = physicalResultTuple._2
+        assert(actualPhysicalColumn.getNames.length == columnName.length)
+        assert(actualPhysicalType == expectedType)
+      }
+  }
+
+  Seq(
+    (Array("A"), Array("a"), StringType.STRING),
+    (Array("B", "C"), Array("b", "c"), DoubleType.DOUBLE),
+    (Array("B", "D"), Array("b", "d"), DateType.DATE),
+    (Array("E"), Array("e"), FloatType.FLOAT),
+    (Array("F", "G", "H"), Array("f", "g", "h"), TimestampNTZType.TIMESTAMP_NTZ),
+    (Array("I"), Array("i"), new MapType(StringType.STRING, DoubleType.DOUBLE, false)),
+    (Array("J"), Array("j"), new ArrayType(StringType.STRING, false))).foreach {
+    case (inputColumnName, expectedColumnName, expectedType) =>
+      test(s"get physical column name should respect case of table schema, $inputColumnName") {
+
+        val column = new Column(inputColumnName)
+        val resultTuple =
+          ColumnMapping.getPhysicalColumnNameAndDataType(testingSchema, column)
+
+        val actualColumn = resultTuple._1
+        val actualType = resultTuple._2
+        assert(actualColumn == new Column(expectedColumnName))
+        assert(actualType == expectedType)
+      }
+  }
+
+  test("getPhysicalColumnNameAndDataType: exception expected when column does not exist") {
+    val ex = intercept[KernelException] {
+      ColumnMapping.getPhysicalColumnNameAndDataType(
+        new StructType()
+          .add("A", StringType.STRING)
+          .add("b", IntegerType.INTEGER),
+        new Column("abc"))
+    }
+    assert(ex.getMessage.contains("Column 'column(`abc`)' was not found in the table schema"))
+
+    val ex1 = intercept[KernelException] {
+      ColumnMapping.getPhysicalColumnNameAndDataType(
+        new StructType().add("a", StringType.STRING)
+          .add(
+            "b",
+            new StructType()
+              .add("D", IntegerType.INTEGER)
+              .add("e", IntegerType.INTEGER))
+          .add("c", IntegerType.INTEGER),
+        new Column(Array("Bbb", "d")))
+    }
+    assert(ex1.getMessage.contains("Column 'column(`Bbb`.`d`)' was not found in the table schema"))
+  }
+
   Seq(true, false).foreach { isNewTable =>
     test(s"assign id and physical name to new table: $isNewTable") {
       val schema: StructType = new StructType()