[Kernel][Writes] Initial insert APIs and implementation #2944

vkorukanti · 2024-04-22T16:15:37Z

Description

Implements interfaces described in #2920. Currently supports inserting into partitioned/unpartitioned tables with protocol version (1, 2). It also has transaction identifier support for idempotent writes.

Conflict resolution is not yet available. It will come as part of the follow up PRs.

How was this patch tested?

Integration tests. More are on the way.

…rings (#2973) ## Description (Split from larger PR #2944) Adds a utility method to convert the partition value literal to Delta protocol-compliant string value. ## How was this patch tested? UTs

…#2975) ## Description (Split from larger PR #2944) Utility method to construct the partition data directory. ## How was this patch tested? UTs

(Split from larger PR delta-io#2944) Add `toRow` for each of the actions objects. Also add any missing actions such as `CommitInfo`.

…cts (#2976) (Split from larger PR #2944) Add `toRow` for each of the action objects. Also add any missing actions such as `CommitInfo`. Also * Rename `READ_SCHEMA` to `FULL_SCHEMA` when all columns in an action are present. `READ_SCHEMA` is a term used to read just the subset of columns, but in some actions, it also represents the full schema. * Utility method to create single action `Row` object from a specific actions. Eg. `createMetadataSingleAction(Metadata metadata)` returns a `Row` of single action schema with `metaData` column representing the given `metadata` object.

…rings (Split from larger PR delta-io#2944) Adds a utility method to convert the partition value literal to Delta protocol-compliant string value. UTs

…ctory (Split from larger PR delta-io#2944) Utility method to construct the partition data directory. UTs

… objects (delta-io#2976) (Split from larger PR delta-io#2944) Add `toRow` for each of the action objects. Also add any missing actions such as `CommitInfo`. Also * Rename `READ_SCHEMA` to `FULL_SCHEMA` when all columns in an action are present. `READ_SCHEMA` is a term used to read just the subset of columns, but in some actions, it also represents the full schema. * Utility method to create single action `Row` object from a specific actions. Eg. `createMetadataSingleAction(Metadata metadata)` returns a `Row` of single action schema with `metaData` column representing the given `metadata` object.

scottsand-db

Looks great! Thanks in advance for answering all my questions :)

scottsand-db · 2024-04-26T18:07:44Z

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

Can't this go under some exceptions folder? IMO this shouldn't be next to Table.java

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

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

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

scottsand-db · 2024-04-26T20:40:28Z

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

+        List<Row> metadataActions = new ArrayList<>();
+        metadataActions.add(createCommitInfoSingleAction(generateCommitAction()));
+        metadataActions.add(createMetadataSingleAction(metadata.toRow()));
+        metadataActions.add(createProtocolSingleAction(protocol.toRow()));


why do you write the metadata and protocol action every time? performance improvement?

scottsand-db · 2024-04-26T20:44:06Z

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

+
+        // Create a new CloseableIterator that will return the commit action, metadata action, and
+        // the stagedData
+        CloseableIterator<Row> stagedDataWithCommitInfo = new CloseableIterator<Row>() {


are you basically just unioning two iterators? is there a simpler util method we can use to do that instead?

scottsand-db · 2024-04-26T20:44:27Z

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

+
+        // Create a new CloseableIterator that will return the commit action, metadata action, and
+        // the stagedData
+        CloseableIterator<Row> stagedDataWithCommitInfo = new CloseableIterator<Row>() {


nit: it's more than just commitInfo right? Since above it is also protocol and metadata?

scottsand-db · 2024-04-26T20:45:51Z

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

+    }
+
+    private Row generateCommitAction() {
+        return new CommitInfo(System.currentTimeMillis(), engineInfo, operation).toRow();


Do we insert the kernel-specific engine info? Or do we rely on the connector to do that?

scottsand-db · 2024-04-26T20:48:54Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/fs/FileOperations.java

+     *             child paths.
+     * @return
+     */
+    public static Path tryRelativizePath(Path child, URI root) {


nit: just relativizePath? "try" makes me think that it may or may not return a relativized path, i.e. that it is best effort.

but this method 100% does return a relativezedPath (it just might not have to do any work if it already is relativized, that's okay)

changed to relativizePath

## Description (Split from larger PR #2944) This API allows the creation of an initial delta log directory when the table is created.

## Description (Split from the larger PR #2944) These are utility to make sure the given schema when creating the table is valid (has no duplicate column names or invalid chars). The code/logic is similar to Delta-Spark/Standalone. ## How was this patch tested? Unittests

(Split from larger PR delta-io#2944) This API allows the creation of an initial delta log directory when the table is created.

(Split from the larger PR delta-io#2944) These are utility to make sure the given schema when creating the table is valid (has no duplicate column names or invalid chars). The code/logic is similar to Delta-Spark/Standalone. Unittests

## Description (Split from #2944) APIs and implementation for creating partitioned or unpartitioned tables. No data insertion yet. Will come in the next PR. ## How was this patch tested? Test suite

## Description (Split from #2944) Adds support for inserting data into the table. ## How was this patch tested? Tests for inserting into partitioned and unpartitioned tables with various combinations of the types, partition values etc. Also tests the checkpoint is ready to create.

…o#3016) (Split from delta-io#2944) APIs and implementation for creating partitioned or unpartitioned tables. No data insertion yet. Will come in the next PR. Test suite

…a-io#3030) (Split from delta-io#2944) Adds support for inserting data into the table. Tests for inserting into partitioned and unpartitioned tables with various combinations of the types, partition values etc. Also tests the checkpoint is ready to create.

@param

## Description (Split from #2944) Adds an API on `TransactionBuilder` to take the transaction identifier for idempotent writes ``` /* * Set the transaction identifier for idempotent writes. Incremental processing systems (e.g., * streaming systems) that track progress using their own application-specific versions need to * record what progress has been made, in order to avoid duplicating data in the face of * failures and retries during writes. By setting the transaction identifier, the Delta table * can ensure that the data with same identifier is not written multiple times. For more * information refer to the Delta protocol section <a * href="https://github.com/delta-io/delta/blob/master/PROTOCOL.md#transaction-identifiers"> * Transaction Identifiers</a>. * * @param engine {@link Engine} instance to use. * @param applicationId The application ID that is writing to the table. * @param transactionVersion The version of the transaction. This should be monotonically * increasing with each write for the same application ID. * @return updated {@link TransactionBuilder} instance. */ TransactionBuilder withTransactionId( Engine engine, String applicationId, long transactionVersion); ``` During the transaction build, check the latest txn version of the given AppId. If it is not monotonically increasing throw `ConcurrentTransactionException`. ## How was this patch tested? Added to `DeltaTableWriteSuite.scala`

@param

(Split from #2944) Adds an API on `TransactionBuilder` to take the transaction identifier for idempotent writes ``` /* * Set the transaction identifier for idempotent writes. Incremental processing systems (e.g., * streaming systems) that track progress using their own application-specific versions need to * record what progress has been made, in order to avoid duplicating data in the face of * failures and retries during writes. By setting the transaction identifier, the Delta table * can ensure that the data with same identifier is not written multiple times. For more * information refer to the Delta protocol section <a * href="https://github.com/delta-io/delta/blob/master/PROTOCOL.md#transaction-identifiers"> * Transaction Identifiers</a>. * * @param engine {@link Engine} instance to use. * @param applicationId The application ID that is writing to the table. * @param transactionVersion The version of the transaction. This should be monotonically * increasing with each write for the same application ID. * @return updated {@link TransactionBuilder} instance. */ TransactionBuilder withTransactionId( Engine engine, String applicationId, long transactionVersion); ``` During the transaction build, check the latest txn version of the given AppId. If it is not monotonically increasing throw `ConcurrentTransactionException`. Added to `DeltaTableWriteSuite.scala`

vkorukanti added the kernel label Apr 22, 2024

vkorukanti added this to the 3.2.0 milestone Apr 22, 2024

vkorukanti force-pushed the inserts-end-2-end branch 3 times, most recently from c5b2ee5 to f61d500 Compare April 25, 2024 15:37

vkorukanti mentioned this pull request Apr 25, 2024

[Kernel][Writes] Utility method to serialize partition literals as strings #2973

Merged

vkorukanti mentioned this pull request Apr 25, 2024

[Kernel][Writes] Utility method to construct partition data directory #2975

Merged

vkorukanti mentioned this pull request Apr 25, 2024

[Kernel[Writes] Add methods to encode Delta Log actions as Row objects #2976

Merged

vkorukanti added a commit to vkorukanti/delta that referenced this pull request Apr 25, 2024

[3.2][Kernel][Writes] Utility method to construct partition data dire…

67d72f8

…ctory (Split from larger PR delta-io#2944) Utility method to construct the partition data directory. UTs

vkorukanti force-pushed the inserts-end-2-end branch from f61d500 to 738f0f2 Compare April 26, 2024 15:54

vkorukanti requested review from scottsand-db and allisonport-db April 26, 2024 15:56

scottsand-db requested changes Apr 26, 2024

View reviewed changes

This was referenced May 1, 2024

[Kernel][Writes] Add schema validation utils #3003

Merged

[Kernel][Writes] Add FileSystemClient.mkdirs API #3004

Merged

vkorukanti added a commit that referenced this pull request May 1, 2024

[Kernel][Writes] Add FileSystemClient.mkdirs API (#3004)

8a4a91e

## Description (Split from larger PR #2944) This API allows the creation of an initial delta log directory when the table is created.

vkorukanti force-pushed the inserts-end-2-end branch from 58e39cd to d8ae28b Compare May 1, 2024 19:51

[Kernel] APIs for inserting data into a Delta table

f17450d

vkorukanti force-pushed the inserts-end-2-end branch from d8ae28b to f17450d Compare May 1, 2024 19:54

This was referenced May 1, 2024

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

Merged

[Kernel][Writes] Add support of inserting data into tables #3030

Merged

vkorukanti mentioned this pull request May 6, 2024

[Kernel][Writes] Support idempotent writes #3051

Merged

vkorukanti closed this May 9, 2024

vkorukanti deleted the inserts-end-2-end branch May 9, 2024 02:43

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Kernel][Writes] Initial insert APIs and implementation #2944

[Kernel][Writes] Initial insert APIs and implementation #2944

vkorukanti commented Apr 22, 2024 •

edited

Loading

scottsand-db left a comment

scottsand-db Apr 26, 2024

scottsand-db Apr 26, 2024

scottsand-db Apr 26, 2024

scottsand-db Apr 26, 2024

scottsand-db Apr 26, 2024

scottsand-db Apr 26, 2024

vkorukanti Apr 29, 2024

[Kernel][Writes] Initial insert APIs and implementation #2944

[Kernel][Writes] Initial insert APIs and implementation #2944

Conversation

vkorukanti commented Apr 22, 2024 • edited Loading

Description

How was this patch tested?

scottsand-db left a comment

Choose a reason for hiding this comment

scottsand-db Apr 26, 2024

Choose a reason for hiding this comment

scottsand-db Apr 26, 2024

Choose a reason for hiding this comment

scottsand-db Apr 26, 2024

Choose a reason for hiding this comment

scottsand-db Apr 26, 2024

Choose a reason for hiding this comment

scottsand-db Apr 26, 2024

Choose a reason for hiding this comment

scottsand-db Apr 26, 2024

Choose a reason for hiding this comment

vkorukanti Apr 29, 2024

Choose a reason for hiding this comment

vkorukanti commented Apr 22, 2024 •

edited

Loading