From 0fbe9b978d5e370bd6d2c52d425bac48a0ba610a Mon Sep 17 00:00:00 2001 From: Allison Portis Date: Thu, 27 Feb 2025 09:14:24 -0800 Subject: [PATCH] [Kernel] Itty bitty refactor (#4188) #### Which Delta project/connector is this regarding? - [ ] Spark - [ ] Standalone - [ ] Flink - [x] Kernel - [ ] Other (fill in here) ## Description Just itty bitty clean up -- reduces changes in add/remove action API prototype. ## How was this patch tested? Existing tests suffice. ## Does this PR introduce _any_ user-facing changes? No --- .../src/main/java/io/delta/kernel/Transaction.java | 6 +++--- .../java/io/delta/kernel/internal/IcebergCompatV2Utils.java | 2 +- .../io/delta/kernel/internal/data/TransactionStateRow.java | 3 +-- 3 files changed, 5 insertions(+), 6 deletions(-) diff --git a/kernel/kernel-api/src/main/java/io/delta/kernel/Transaction.java b/kernel/kernel-api/src/main/java/io/delta/kernel/Transaction.java index b2658bb21f9..457f73f89c1 100644 --- a/kernel/kernel-api/src/main/java/io/delta/kernel/Transaction.java +++ b/kernel/kernel-api/src/main/java/io/delta/kernel/Transaction.java @@ -117,7 +117,7 @@ static CloseableIterator transformLogicalData( // Note: `partitionValues` are not used as of now in this API, but taking the partition // values as input forces the connector to not pass data from multiple partitions this // API in a single call. - StructType tableSchema = getLogicalSchema(engine, transactionState); + StructType tableSchema = getLogicalSchema(transactionState); List partitionColNames = getPartitionColumnsList(transactionState); validateAndSanitizePartitionValues(tableSchema, partitionColNames, partitionValues); @@ -167,7 +167,7 @@ static CloseableIterator transformLogicalData( */ static DataWriteContext getWriteContext( Engine engine, Row transactionState, Map partitionValues) { - StructType tableSchema = getLogicalSchema(engine, transactionState); + StructType tableSchema = getLogicalSchema(transactionState); List partitionColNames = getPartitionColumnsList(transactionState); partitionValues = @@ -209,7 +209,7 @@ static CloseableIterator generateAppendActions( return fileStatusIter.map( dataFileStatus -> { if (isIcebergCompatV2Enabled) { - IcebergCompatV2Utils.validDataFileStatus(dataFileStatus); + IcebergCompatV2Utils.validateDataFileStatus(dataFileStatus); } AddFile addFileRow = AddFile.convertDataFileStatus( diff --git a/kernel/kernel-api/src/main/java/io/delta/kernel/internal/IcebergCompatV2Utils.java b/kernel/kernel-api/src/main/java/io/delta/kernel/internal/IcebergCompatV2Utils.java index 999eb036125..81fc251aeea 100644 --- a/kernel/kernel-api/src/main/java/io/delta/kernel/internal/IcebergCompatV2Utils.java +++ b/kernel/kernel-api/src/main/java/io/delta/kernel/internal/IcebergCompatV2Utils.java @@ -27,7 +27,7 @@ private IcebergCompatV2Utils() {} * * @param dataFileStatus The {@link DataFileStatus} to validate. */ - public static void validDataFileStatus(DataFileStatus dataFileStatus) { + public static void validateDataFileStatus(DataFileStatus dataFileStatus) { if (!dataFileStatus.getStatistics().isPresent()) { // presence of stats means always has a non-null `numRecords` throw DeltaErrors.missingNumRecordsStatsForIcebergCompatV2(dataFileStatus); diff --git a/kernel/kernel-api/src/main/java/io/delta/kernel/internal/data/TransactionStateRow.java b/kernel/kernel-api/src/main/java/io/delta/kernel/internal/data/TransactionStateRow.java index 11c50a2784f..871c3ce6997 100644 --- a/kernel/kernel-api/src/main/java/io/delta/kernel/internal/data/TransactionStateRow.java +++ b/kernel/kernel-api/src/main/java/io/delta/kernel/internal/data/TransactionStateRow.java @@ -60,11 +60,10 @@ private TransactionStateRow(HashMap valueMap) { * Get the logical schema of the table from the transaction state {@link Row} returned by {@link * Transaction#getTransactionState(Engine)}} * - * @param engine {@link Engine} instance to use for parsing the schema * @param transactionState Transaction state state {@link Row} * @return Logical schema of the table as {@link StructType} */ - public static StructType getLogicalSchema(Engine engine, Row transactionState) { + public static StructType getLogicalSchema(Row transactionState) { String serializedSchema = transactionState.getString(COL_NAME_TO_ORDINAL.get("logicalSchemaString")); return DataTypeJsonSerDe.deserializeStructType(serializedSchema);