-
Notifications
You must be signed in to change notification settings - Fork 1.7k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[Kernel] Add coordinated commits interfaces and table properties #3370
Conversation
37b3148
to
210a48f
Compare
kernel/kernel-api/src/main/java/io/delta/kernel/commit/Commit.java
Outdated
Show resolved
Hide resolved
kernel/kernel-api/src/main/java/io/delta/kernel/internal/TransactionImpl.java
Outdated
Show resolved
Hide resolved
...lts/src/main/java/io/delta/kernel/defaults/engine/DefaultCommitCoordinatorClientHandler.java
Outdated
Show resolved
Hide resolved
kernel/kernel-api/src/main/java/io/delta/kernel/commit/actions/AbstractCommitInfo.java
Outdated
Show resolved
Hide resolved
kernel/kernel-api/src/main/java/io/delta/kernel/internal/TableConfig.java
Outdated
Show resolved
Hide resolved
@@ -201,4 +260,35 @@ private void validate(String value) { | |||
private static void addConfig(HashMap<String, TableConfig<?>> configs, TableConfig<?> config) { | |||
configs.put(config.getKey().toLowerCase(Locale.ROOT), config); | |||
} | |||
|
|||
private static Map<String, String> parseStringToJsonMap(String jsonString) { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Could add some tests?
- A few happy path cases
- Handling of escaped characters
- Test the two areas where you throw exception
I would recommend creating some complex static json with unicode characters using a standard serializer and then deserializing the static json in a test.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
yep, and this can be its own separate PR with the design decision doc attached to it.
kernel/kernel-api/src/main/java/io/delta/kernel/internal/TableConfig.java
Outdated
Show resolved
Hide resolved
/* | ||
* This table property is used to track the table managed by commit-coordinator internally. | ||
*/ | ||
public static final TableConfig<Map<String, String>> COORDINATED_COMMITS_TABLE_CONF = |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@vkorukanti should a new type be created here that encapsulates Map<String, String> or is this okay?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is ok.
kernel/kernel-api/src/main/java/io/delta/kernel/commit/Commit.java
Outdated
Show resolved
Hide resolved
kernel/kernel-api/src/main/java/io/delta/kernel/commit/Commit.java
Outdated
Show resolved
Hide resolved
kernel/kernel-api/src/main/java/io/delta/kernel/commit/Commit.java
Outdated
Show resolved
Hide resolved
kernel/kernel-api/src/main/java/io/delta/kernel/commit/Commit.java
Outdated
Show resolved
Hide resolved
kernel/kernel-api/src/main/java/io/delta/kernel/commit/Commit.java
Outdated
Show resolved
Hide resolved
kernel/kernel-api/src/main/java/io/delta/kernel/engine/CommitCoordinatorClientHandler.java
Outdated
Show resolved
Hide resolved
kernel/kernel-api/src/main/java/io/delta/kernel/engine/Engine.java
Outdated
Show resolved
Hide resolved
/* | ||
* This table property is used to track the table managed by commit-coordinator internally. | ||
*/ | ||
public static final TableConfig<Map<String, String>> COORDINATED_COMMITS_TABLE_CONF = |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is ok.
@@ -201,4 +260,35 @@ private void validate(String value) { | |||
private static void addConfig(HashMap<String, TableConfig<?>> configs, TableConfig<?> config) { | |||
configs.put(config.getKey().toLowerCase(Locale.ROOT), config); | |||
} | |||
|
|||
private static Map<String, String> parseStringToJsonMap(String jsonString) { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
yep, and this can be its own separate PR with the design decision doc attached to it.
kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/engine/DefaultEngine.java
Outdated
Show resolved
Hide resolved
...ernel-api/src/main/java/io/delta/kernel/engine/coordinatedcommits/CommitFailedException.java
Outdated
Show resolved
Hide resolved
...l/kernel-api/src/main/java/io/delta/kernel/engine/coordinatedcommits/GetCommitsResponse.java
Show resolved
Hide resolved
...l/kernel-api/src/main/java/io/delta/kernel/engine/coordinatedcommits/GetCommitsResponse.java
Outdated
Show resolved
Hide resolved
...l/kernel-api/src/main/java/io/delta/kernel/engine/coordinatedcommits/GetCommitsResponse.java
Outdated
Show resolved
Hide resolved
kernel/kernel-api/src/main/java/io/delta/kernel/engine/coordinatedcommits/UpdatedActions.java
Outdated
Show resolved
Hide resolved
kernel/kernel-api/src/main/java/io/delta/kernel/engine/coordinatedcommits/UpdatedActions.java
Outdated
Show resolved
Hide resolved
kernel/kernel-api/src/main/java/io/delta/kernel/engine/coordinatedcommits/UpdatedActions.java
Outdated
Show resolved
Hide resolved
kernel/kernel-api/src/main/java/io/delta/kernel/engine/CommitCoordinatorClientHandler.java
Outdated
Show resolved
Hide resolved
kernel/kernel-api/src/main/java/io/delta/kernel/engine/CommitCoordinatorClientHandler.java
Outdated
Show resolved
Hide resolved
kernel/kernel-api/src/main/java/io/delta/kernel/internal/TableConfig.java
Outdated
Show resolved
Hide resolved
...ain/java/io/delta/kernel/defaults/internal/coordinatedcommits/CommitCoordinatorProvider.java
Outdated
Show resolved
Hide resolved
d84b28c
to
8c06239
Compare
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Looks great! Thanks for just posting the interfaces. Easy to read and understand.
kernel/kernel-api/src/main/java/io/delta/kernel/engine/CommitCoordinatorClientHandler.java
Outdated
Show resolved
Hide resolved
kernel/kernel-api/src/main/java/io/delta/kernel/engine/CommitCoordinatorClientHandler.java
Show resolved
Hide resolved
kernel/kernel-api/src/main/java/io/delta/kernel/engine/CommitCoordinatorClientHandler.java
Outdated
Show resolved
Hide resolved
kernel/kernel-api/src/main/java/io/delta/kernel/engine/CommitCoordinatorClientHandler.java
Outdated
Show resolved
Hide resolved
kernel/kernel-api/src/main/java/io/delta/kernel/engine/CommitCoordinatorClientHandler.java
Outdated
Show resolved
Hide resolved
kernel/kernel-api/src/main/java/io/delta/kernel/engine/Engine.java
Outdated
Show resolved
Hide resolved
...ernel-api/src/main/java/io/delta/kernel/engine/coordinatedcommits/CommitFailedException.java
Outdated
Show resolved
Hide resolved
...ernel-api/src/main/java/io/delta/kernel/engine/coordinatedcommits/CommitFailedException.java
Outdated
Show resolved
Hide resolved
import io.delta.kernel.engine.coordinatedcommits.actions.AbstractProtocol; | ||
import io.delta.kernel.utils.CloseableIterator; | ||
|
||
/** |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
general comment to all APIs in this interface: what exception is thrown if the table is not managed by a coordinated commit?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think the commit coordinator assumes that the logPath
passed as an argument represents a table managed by the commit coordinator. For example, the InMemoryCommitCoordinator in Delta Spark doesn't throw a specific exception for this case (will just throw a NullPointerException
). So the user should ensure that they pass a correct logPath
.
@dhruvarya-db Do I understand correctly?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I don't think that there has been any standardization around that. @prakharjain09 any idea?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
No standardization for now.
/** | ||
* Determines whether this CommitCoordinatorClientHandler is semantically equal to another | ||
* CommitCoordinatorClientHandler. | ||
* <p> | ||
* Semantic equality is determined by each CommitCoordinatorClientHandler implementation based | ||
* on whether the two instances can be used interchangeably when invoking any of the | ||
* CommitCoordinatorClientHandler APIs, such as {@link #commit}, {@link #getCommits}, etc. For | ||
* example, both instances might be pointing to the same underlying endpoint. | ||
*/ | ||
Boolean semanticEquals(CommitCoordinatorClientHandler other); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
why Boolean
and not boolean
?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think boolean should work.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
thanks for refactoring these out, very clear!
b7fa65a
to
9ed9ced
Compare
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
LGTM!
Which Delta project/connector is this regarding?
Description
Add coordinated commits interfaces and related table properties in Kernel to prepare for Kernel read and write supported by coordinated commits.
How was this patch tested?
Unit tests
Does this PR introduce any user-facing changes?
Yes. User can define and register their own commit coordinator builder and get the corresponding commit coordinator client handler from engine.