implement lookup insert if not exists on the client side #2573
Conversation
| .build(); | ||
| TableDescriptor tableDescriptor = TableDescriptor.builder().schema(schema).build(); | ||
| createTable(tablePath, tableDescriptor, false); | ||
| Table invalid_table = conn.getTable(tablePath); |
There was a problem hiding this comment.
Error: src/test/java/org/apache/fluss/client/table/FlussTableITCase.java:[651,15] (naming) LocalVariableName: Name 'invalid_table' must match pattern '^[a-z][a-zA-Z0-9]*$'.
| return request; | ||
| } | ||
|
|
||
| public static LookupRequest makeLookupRequest( |
There was a problem hiding this comment.
May be two functions makeLookupRequestWithInsertIfNotExists and makeLookupRequest can be merged.
There was a problem hiding this comment.
comments addressed. please take a look.
| * | ||
| * @return a new Lookup instance with insert-if-not-exists enabled | ||
| */ | ||
| Lookup enableInsertIfNotExists(); |
There was a problem hiding this comment.
Lookup.enableInsertIfNotExists() doesn't document:
Auto-increment column handling, Nullable column restriction.
platinumhamburg
left a comment
platinumhamburg
left a comment
There was a problem hiding this comment.
LGTM, just a few minor issues.
| .createLookuper()) | ||
| .isInstanceOf(IllegalArgumentException.class) | ||
| .hasMessageContaining("insertIfNotExists can not be used with prefix lookup"); | ||
| assertThatThrownBy( |
There was a problem hiding this comment.
This is a duplicate assertion. Moreover, the test schema contains only one primary key column, 'b', so creating a lookuper with lookupBy("b") should not result in a PrefixKeyLookuper. This issue was not introduced by this PR; rather, the test has exposed an incompleteness in the semantic implementation of TableLookup.
|
|
||
| @Override | ||
| public Lookup enableInsertIfNotExists() { | ||
| if (lookupColumnNames != null) { |
There was a problem hiding this comment.
This check can be uniformly moved to createLookuper().
|
|
||
| @Override | ||
| public Lookup lookupBy(List<String> lookupColumnNames) { | ||
| if (insertIfNotExists) { |
There was a problem hiding this comment.
Ditto, this check can be uniformly moved to createLookuper().
There was a problem hiding this comment.
good suggestion. done.
| @Internal | ||
| public enum LookupType { | ||
| LOOKUP, | ||
| LOOKUP_WITH_INSERT_IF_NOT_EXISTS, |
There was a problem hiding this comment.
I think there's no need to extend the enum definition here.
There was a problem hiding this comment.
We need to group the different lookups in the lookupQueue into different batches in https://github.com/apache/fluss/blob/main/fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupSender.java#L152 so I introduced this new type
| public Lookup lookupBy(List<String> lookupColumnNames) { | ||
| return new TableLookup( | ||
| tableInfo, schemaGetter, metadataUpdater, lookupClient, lookupColumnNames); | ||
| tableInfo, schemaGetter, metadataUpdater, lookupClient, lookupColumnNames, false); |
There was a problem hiding this comment.
We should copy with the existing insertIfNotExists, rather than override it to false.
|
|
||
| @Override | ||
| public Lookup enableInsertIfNotExists() { | ||
| return new TableLookup(tableInfo, schemaGetter, metadataUpdater, lookupClient, null, true); |
There was a problem hiding this comment.
We should copy with the existing lookupColumnNames, rather than override it to null.
4 participants
Purpose
Linked issue: close #2591
Brief change log
Tests
API and Format
Documentation