WIP

Author: scottsand-db

kernel/kernel-api/src/main/java/io/delta/kernel/ResolvedTable.java

@@ -17,8 +17,8 @@
 package io.delta.kernel;
 
 import io.delta.kernel.annotation.Experimental;
-import io.delta.kernel.commit.Committer;
 import io.delta.kernel.expressions.Column;
+import io.delta.kernel.transaction.UpdateTableTransactionBuilder;
 import io.delta.kernel.types.StructType;
 import java.util.List;
 import java.util.Optional;
@@ -66,5 +66,5 @@ public interface ResolvedTable {
   /** @return a scan builder for constructing scans to read data from this table */
   ScanBuilder getScanBuilder();
 
-  Committer getCommitter();
+  UpdateTableTransactionBuilder buildUpdateTableTransaction(String engineInfo, Operation operation);
 }

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

@@ -55,14 +55,15 @@
 import io.delta.kernel.internal.util.ColumnMapping.ColumnMappingMode;
 import io.delta.kernel.internal.util.SchemaUtils;
 import io.delta.kernel.internal.util.Tuple2;
+import io.delta.kernel.transaction.UpdateTableTransactionBuilder;
 import io.delta.kernel.types.StringType;
 import io.delta.kernel.types.StructType;
 import java.util.*;
 import java.util.stream.Collectors;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class TransactionBuilderImpl implements TransactionBuilder {
+public class TransactionBuilderImpl implements TransactionBuilder, UpdateTableTransactionBuilder {
   private static final Logger logger = LoggerFactory.getLogger(TransactionBuilderImpl.class);
 
   private final long currentTimeMillis = System.currentTimeMillis();
@@ -104,6 +105,38 @@ public TransactionBuilderImpl(TableImpl table, String engineInfo, Operation oper
     this.operation = operation;
   }
 
+  ///////////////////////////////////////////
+  // UpdateTableTransactionBuilder methods //
+  ///////////////////////////////////////////
+
+  @Override
+  public TransactionBuilderImpl withUpdatedSchema(StructType updatedSchema) {
+    this.schema = Optional.of(updatedSchema); // will be verified as part of the build() call
+    return this;
+  }
+
+  @Override
+  public TransactionBuilderImpl withTableProperties(Map<String, String> properties) {
+    this.tableProperties =
+        Optional.of(Collections.unmodifiableMap(TableConfig.validateDeltaProperties(properties)));
+    return this;
+  }
+
+  @Override
+  public TransactionBuilderImpl withTransactionId(String applicationId, long transactionVersion) {
+    SetTransaction txnId =
+        new SetTransaction(
+            requireNonNull(applicationId, "applicationId is null"),
+            transactionVersion,
+            Optional.of(currentTimeMillis));
+    this.setTxnOpt = Optional.of(txnId);
+    return this;
+  }
+
+  ////////////////////////////////
+  // TransactionBuilder methods //
+  ////////////////////////////////
+
   @Override
   public TransactionBuilder withSchema(Engine engine, StructType newSchema) {
     this.schema = Optional.of(newSchema); // will be verified as part of the build() call
@@ -152,24 +185,16 @@ public TransactionBuilder withClusteringColumns(Engine engine, List<Column> clus
   @Override
   public TransactionBuilder withTransactionId(
       Engine engine, String applicationId, long transactionVersion) {
-    SetTransaction txnId =
-        new SetTransaction(
-            requireNonNull(applicationId, "applicationId is null"),
-            transactionVersion,
-            Optional.of(currentTimeMillis));
-    this.setTxnOpt = Optional.of(txnId);
-    return this;
+    return withTransactionId(applicationId, transactionVersion);
   }
 
   @Override
   public TransactionBuilder withTableProperties(Engine engine, Map<String, String> properties) {
-    this.tableProperties =
-        Optional.of(Collections.unmodifiableMap(TableConfig.validateDeltaProperties(properties)));
-    return this;
+    return withTableProperties(properties);
   }
 
   @Override
-  public TransactionBuilder withTablePropertiesRemoved(Set<String> propertyKeys) {
+  public TransactionBuilderImpl withTablePropertiesRemoved(Set<String> propertyKeys) {
     checkArgument(
         propertyKeys.stream().noneMatch(key -> key.toLowerCase(Locale.ROOT).startsWith("delta.")),
         "Unsetting 'delta.' table properties is currently unsupported");
@@ -198,7 +223,7 @@ public TransactionBuilder withDomainMetadataSupported() {
   }
 
   @Override
-  public Transaction build(Engine engine) {
+  public TransactionImpl build(Engine engine) {
     if (operation == Operation.REPLACE_TABLE) {
       throw new UnsupportedOperationException("REPLACE TABLE is not yet supported");
     }

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

@@ -23,6 +23,8 @@
 import static io.delta.kernel.internal.util.Utils.toCloseableIterator;
 
 import io.delta.kernel.*;
+import io.delta.kernel.commit.CommitPayload;
+import io.delta.kernel.commit.Committer;
 import io.delta.kernel.data.Row;
 import io.delta.kernel.engine.Engine;
 import io.delta.kernel.exceptions.ConcurrentWriteException;
@@ -66,7 +68,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class TransactionImpl implements Transaction {
+public class TransactionImpl implements Transaction, io.delta.kernel.transaction.Transaction {
   private static final Logger logger = LoggerFactory.getLogger(TransactionImpl.class);
 
   public static final int DEFAULT_READ_VERSION = 1;
@@ -132,19 +134,54 @@ public TransactionImpl(
     this.currentCrcInfo = readSnapshot.getCurrentCrcInfo();
   }
 
+  /////////////////////////////////////////////////////
+  // io.delta.kernel.transaction.Transaction methods //
+  /////////////////////////////////////////////////////
+
   @Override
-  public Row getTransactionState(Engine engine) {
+  public StructType getSchema() {
+    return metadata.getSchema();
+  }
+
+  @Override
+  public List<Column> getPartitionColumns() {
+    return VectorUtils.<String>toJavaList(metadata.getPartitionColumns()).stream()
+        .map(Column::new)
+        .collect(Collectors.toList());
+  }
+
+  @Override
+  public Row getTransactionState() {
     return TransactionStateRow.of(metadata, dataPath.toString(), maxRetries);
   }
 
+  @Override
+  public Committer getCommitter() {
+    throw new UnsupportedOperationException("not implemented");
+  }
+
+  @Override
+  public CommitPayload getCommitPayload(CloseableIterator<Row> dataActions) {
+    return null;
+  }
+
+  /////////////////////////////////////////
+  // io.delta.kernel.Transaction methods //
+  /////////////////////////////////////////
+
+  @Override
+  public Row getTransactionState(Engine engine) {
+    return getTransactionState();
+  }
+
   @Override
   public List<String> getPartitionColumns(Engine engine) {
     return VectorUtils.toJavaList(metadata.getPartitionColumns());
   }
 
   @Override
   public StructType getSchema(Engine engine) {
-    return metadata.getSchema();
+    return getSchema();
   }
 
   @Override

kernel/kernel-api/src/main/java/io/delta/kernel/internal/table/ResolvedTableInternalImpl.java

@@ -18,8 +18,8 @@
 
 import static java.util.Objects.requireNonNull;
 
+import io.delta.kernel.Operation;
 import io.delta.kernel.ScanBuilder;
-import io.delta.kernel.commit.Committer;
 import io.delta.kernel.expressions.Column;
 import io.delta.kernel.internal.ScanBuilderImpl;
 import io.delta.kernel.internal.actions.DomainMetadata;
@@ -35,6 +35,7 @@
 import io.delta.kernel.internal.util.Clock;
 import io.delta.kernel.internal.util.VectorUtils;
 import io.delta.kernel.metrics.SnapshotReport;
+import io.delta.kernel.transaction.UpdateTableTransactionBuilder;
 import io.delta.kernel.types.StructType;
 import java.util.List;
 import java.util.Map;
@@ -117,8 +118,9 @@ public ScanBuilder getScanBuilder() {
   }
 
   @Override
-  public Committer getCommitter() {
-    throw new UnsupportedOperationException("not implemented");
+  public UpdateTableTransactionBuilder buildUpdateTableTransaction(
+      String engineInfo, Operation operation) {
+    return null;
   }
 
   ///////////////////////////////////////

kernel/kernel-api/src/main/java/io/delta/kernel/internal/transaction/TransactionSource.java

@@ -0,0 +1 @@
+ 

kernel/kernel-api/src/main/java/io/delta/kernel/transaction/BaseTransactionBuilder.java

@@ -0,0 +1,12 @@
+package io.delta.kernel.transaction;
+
+import io.delta.kernel.engine.Engine;
+import java.util.Map;
+
+public interface BaseTransactionBuilder<T extends BaseTransactionBuilder<T>> {
+  T withTableProperties(Map<String, String> properties);
+
+  T withTransactionId(String applicationId, long transactionVersion);
+
+  Transaction build(Engine engine);
+}

kernel/kernel-api/src/main/java/io/delta/kernel/transaction/Transaction.java

@@ -0,0 +1,28 @@
+package io.delta.kernel.transaction;
+
+import io.delta.kernel.commit.CommitPayload;
+import io.delta.kernel.commit.Committer;
+import io.delta.kernel.data.Row;
+import io.delta.kernel.expressions.Column;
+import io.delta.kernel.types.StructType;
+import io.delta.kernel.utils.CloseableIterator;
+import java.util.List;
+
+public interface Transaction {
+  StructType getSchema();
+
+  List<Column> getPartitionColumns();
+
+  long getReadTableVersion();
+
+  Row getTransactionState();
+
+  Committer getCommitter();
+
+  CommitPayload getCommitPayload(CloseableIterator<Row> dataActions);
+
+  // TODO: CommitPayload resolveConflictsAndRebase(
+  //         CloseableIterator<Row> dataActions,
+  //         List<ParsedDeltaData> winningCommits,
+  //         Optional<Long> latestTableVersion);
+}

kernel/kernel-api/src/main/java/io/delta/kernel/transaction/UpdateTableTransactionBuilder.java

@@ -0,0 +1,12 @@
+package io.delta.kernel.transaction;
+
+import io.delta.kernel.types.StructType;
+import java.util.Set;
+
+public interface UpdateTableTransactionBuilder
+    extends BaseTransactionBuilder<UpdateTableTransactionBuilder> {
+
+  UpdateTableTransactionBuilder withUpdatedSchema(StructType updatedSchema);
+
+  UpdateTableTransactionBuilder withTablePropertiesRemoved(Set<String> propertyKeys);
+}