[3.2][Kernel][Writes] APIs and impl. for creating new tables (#3016)

(Split from #2944)

APIs and implementation for creating partitioned or unpartitioned
tables. No data insertion yet. Will come in the next PR.

Test suite

Author: vkorukanti

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

@@ -0,0 +1,53 @@
+/*
+ * Copyright (2023) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.delta.kernel;
+
+import java.util.List;
+import java.util.Map;
+
+import io.delta.kernel.annotation.Evolving;
+import io.delta.kernel.data.Row;
+import io.delta.kernel.engine.Engine;
+import io.delta.kernel.expressions.Column;
+
+/**
+ * Contains the context for writing data to Delta table. The context is created for each partition
+ * for partitioned table or once per table for un-partitioned table. It is created using
+ * {@link Transaction#getWriteContext(Engine, Row, Map)} (String, Map, List)}.
+ *
+ * @since 3.2.0
+ */
+@Evolving
+public interface DataWriteContext {
+    /**
+     * Returns the target directory where the data should be written.
+     *
+     * @return fully qualified path of the target directory
+     */
+    String getTargetDirectory();
+
+    /**
+     * Returns the list of {@link Column} that the connector can optionally collect statistics. Each
+     * {@link Column} is a reference to a top-level or nested column in the table.
+     * <p>
+     * Statistics collections can be skipped or collected for a partial list of the returned
+     * {@link Column}s. When stats are present in the written Delta log, they can be used to
+     * optimize query performance.
+     *
+     * @return schema of the statistics
+     */
+    List<Column> getStatisticsColumns();
+}

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

@@ -0,0 +1,61 @@
+/*
+ * Copyright (2023) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.delta.kernel;
+
+/**
+ * An operation that can be performed on a Delta table.
+ * <p>
+ * An operation is tracked as the first line in commit info action inside the Delta Log
+ * It also shows up when {@code DESCRIBE HISTORY} on the table is executed.
+ */
+public enum Operation {
+
+    /**
+     * Recorded when the table is created.
+     */
+    CREATE_TABLE("CREATE TABLE"),
+
+    /**
+     * Recorded during batch inserts.
+     */
+    WRITE("WRITE"),
+
+    /**
+     * Recorded during streaming inserts.
+     */
+    STREAMING_UPDATE("STREAMING UPDATE"),
+
+    /**
+     * For any operation that doesn't fit the above categories.
+     */
+    MANUAL_UPDATE("Manual Update");
+
+    /**
+     * Actual value that will be recorded in the transaction log
+     */
+    private final String description;
+
+    Operation(String description) {
+        this.description = description;
+    }
+
+    /**
+     * Returns the string that will be recorded in the transaction log.
+     */
+    public String getDescription() {
+        return description;
+    }
+}

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

@@ -120,6 +120,21 @@ Snapshot getSnapshotAsOfVersion(Engine engine, long versionId)
     Snapshot getSnapshotAsOfTimestamp(Engine engine, long millisSinceEpochUTC)
         throws TableNotFoundException;
 
+    /**
+     * Create a {@link TransactionBuilder} which can create a {@link Transaction} object to mutate
+     * the table.
+     *
+     * @param engine {@link Engine} instance to use.
+     * @param engineInfo information about the engine that is making the updates.
+     * @param operation metadata of operation that is being performed. E.g. "insert", "delete".
+     * @return {@link TransactionBuilder} instance to build the transaction.
+     * @since 3.2.0
+     */
+    TransactionBuilder createTransactionBuilder(
+            Engine engine,
+            String engineInfo,
+            Operation operation);
+
     /**
      * Checkpoint the table at given version. It writes a single checkpoint file.
      *

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

@@ -0,0 +1,151 @@
+/*
+ * Copyright (2023) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.delta.kernel;
+
+import java.util.List;
+import java.util.Map;
+
+import io.delta.kernel.annotation.Evolving;
+import io.delta.kernel.data.FilteredColumnarBatch;
+import io.delta.kernel.data.Row;
+import io.delta.kernel.engine.Engine;
+import io.delta.kernel.exceptions.ConcurrentWriteException;
+import io.delta.kernel.expressions.Literal;
+import io.delta.kernel.types.StructType;
+import io.delta.kernel.utils.*;
+
+/**
+ * Represents a transaction to mutate a Delta table.
+ *
+ * @since 3.2.0
+ */
+@Evolving
+public interface Transaction {
+    /**
+     * Get the schema of the table. If the connector is adding any data to the table through this
+     * transaction, it should have the same schema as the table schema.
+     */
+    StructType getSchema(Engine engine);
+
+    /**
+     * Get the list of logical names of the partition columns. This helps the connector to do
+     * physical partitioning of the data before asking the Kernel to stage the data per partition.
+     */
+    List<String> getPartitionColumns(Engine engine);
+
+    /**
+     * Get the state of the transaction. The state helps Kernel do the transformations to logical
+     * data according to the Delta protocol and table features enabled on the table. The engine
+     * should use this at the data writer task to transform the logical data that the engine wants
+     * to write to the table in to physical data that goes in data files using
+     * {@link Transaction#transformLogicalData(Engine, Row, CloseableIterator, Map)}
+     */
+    Row getTransactionState(Engine engine);
+
+    /**
+     * Commit the transaction including the data action rows generated by
+     * {@link Transaction#generateAppendActions}.
+     *
+     * @param engine      {@link Engine} instance.
+     * @param dataActions Iterable of data actions to commit. These data actions are generated by
+     *                    the
+     *                    {@link Transaction#generateAppendActions(Engine, Row, CloseableIterator,
+     *                    DataWriteContext)}. The {@link CloseableIterable} allows the Kernel to
+     *                    access the list of actions multiple times (in case of retries to resolve
+     *                    the conflicts due to other writers to the table). Kernel provides a
+     *                    in-memory based implementation of {@link CloseableIterable} with utility
+     *                    API {@link CloseableIterable#inMemoryIterable(CloseableIterator)}
+     * @return {@link TransactionCommitResult} status of the successful transaction.
+     * @throws ConcurrentWriteException when the transaction has encountered a non-retryable
+     *                                  conflicts or exceeded the maximum number of retries reached.
+     *                                  The connector needs to rerun the query on top of the latest
+     *                                  table state and retry the transaction.
+     */
+    TransactionCommitResult commit(Engine engine, CloseableIterable<Row> dataActions)
+            throws ConcurrentWriteException;
+
+    /**
+     * Given the logical data that needs to be written to the table, convert it into the required
+     * physical data depending upon the table Delta protocol and features enabled on the table.
+     * Kernel takes care of adding any additional column or removing existing columns that doesn't
+     * need to be in physical data files. All these transformations are driven by the Delta protocol
+     * and table features enabled on the table.
+     * <p>
+     * The given data should belong to exactly one partition. It is the job of the connector to do
+     * partitioning of the data before calling the API. Partition values are provided as map of
+     * column name to partition value (as {@link Literal}). If the table is an un-partitioned table,
+     * then map should be empty.
+     *
+     * @param engine           {@link Engine} instance to use.
+     * @param transactionState The transaction state
+     * @param dataIter         Iterator of logical data (with schema same as the table schema)
+     *                         to transform to physical data. All the data n this iterator should
+     *                         belong to one physical partition and it should also include the
+     *                         partition data.
+     * @param partitionValues  The partition values for the data. If the table is un-partitioned,
+     *                         the map should be empty
+     * @return Iterator of physical data to write to the data files.
+     */
+    static CloseableIterator<FilteredColumnarBatch> transformLogicalData(
+            Engine engine,
+            Row transactionState,
+            CloseableIterator<FilteredColumnarBatch> dataIter,
+            Map<String, Literal> partitionValues) {
+        throw new UnsupportedOperationException("Not implemented yet");
+    }
+
+    /**
+     * Get the context for writing data into a table. The context tells the connector where the data
+     * should be written. For partitioned table context is generated per partition. So, the
+     * connector should call this API for each partition. For un-partitioned table, the context is
+     * same for all the data.
+     *
+     * @param engine           {@link Engine} instance to use.
+     * @param transactionState The transaction state
+     * @param partitionValues  The partition values for the data. If the table is un-partitioned,
+     *                         the map should be empty
+     * @return {@link DataWriteContext} containing metadata about where and how the data for
+     * partition should be written.
+     */
+    static DataWriteContext getWriteContext(
+            Engine engine,
+            Row transactionState,
+            Map<String, Literal> partitionValues) {
+        throw new UnsupportedOperationException("Not implemented yet");
+    }
+
+    /**
+     * For given data files, generate Delta actions that can be committed in a transaction.
+     * These data files are the result of writing the data returned by
+     * {@link Transaction#transformLogicalData} with the context returned by
+     * {@link Transaction#getWriteContext}.
+     *
+     * @param engine           {@link Engine} instance.
+     * @param transactionState State of the transaction.
+     * @param fileStatusIter   Iterator of row objects representing each data file written.
+     * @param dataWriteContext The context used when writing the data files given in
+     *                         {@code fileStatusIter}
+     * @return {@link CloseableIterator} of {@link Row} representing the actions to commit using
+     * {@link Transaction#commit}.
+     */
+    static CloseableIterator<Row> generateAppendActions(
+            Engine engine,
+            Row transactionState,
+            CloseableIterator<DataFileStatus> fileStatusIter,
+            DataWriteContext dataWriteContext) {
+        throw new UnsupportedOperationException("Not implemented yet");
+    }
+}

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

@@ -0,0 +1,57 @@
+/*
+ * Copyright (2023) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.delta.kernel;
+
+import java.util.List;
+
+import io.delta.kernel.annotation.Evolving;
+import io.delta.kernel.engine.Engine;
+import io.delta.kernel.types.StructType;
+
+/**
+ * Builder for creating a {@link Transaction} to mutate a Delta table.
+ *
+ * @since 3.2.0
+ */
+@Evolving
+public interface TransactionBuilder {
+    /**
+     * Set the schema of the table when creating a new table.
+     *
+     * @param engine {@link Engine} instance to use.
+     * @param schema The new schema of the table.
+     * @return updated {@link TransactionBuilder} instance.
+     */
+    TransactionBuilder withSchema(Engine engine, StructType schema);
+
+    /**
+     * Set the list of partitions columns when create a new partitioned table.
+     *
+     * @param engine           {@link Engine} instance to use.
+     * @param partitionColumns The partition columns of the table. These should be a subset of the
+     *                         columns in the schema.
+     * @return updated {@link TransactionBuilder} instance.
+     */
+    TransactionBuilder withPartitionColumns(Engine engine, List<String> partitionColumns);
+
+    /**
+     * Build the transaction. Also validates the given info to ensure that a valida transaction
+     * can be created.
+     *
+     * @param engine {@link Engine} instance to use.
+     */
+    Transaction build(Engine engine);
+}