[FLINK-36929][table] Add SQL connector for keyed savepoint data #26035
Conversation
| } | ||
|
|
||
| public void setPrivateLong(Long privateLong) { | ||
| this.privateLong = privateLong; |
There was a problem hiding this comment.
I am not sure what this example is showing . publicLong is defined but not referenced at all.
There was a problem hiding this comment.
The state contains this POJO serialized. It shows that it can read private members too.
|
It would really help me to review this is the documentation was included in this PR. |
| .stringType() | ||
| .noDefaultValue() | ||
| .withDescription( | ||
| "Defines the state backend which must be used for state reading."); |
There was a problem hiding this comment.
I think the valid values should be included in the description, as per the Flip - I assume with ForsDB as well.
The Flip documents this as an Enum - shouldn't this be an EnumType?
There was a problem hiding this comment.
This is similar like state.backend.type so I would keep it that way.
There was a problem hiding this comment.
Adding more description makes sense though.
|
|
||
| if (operatorUid.isPresent() == operatorUidHash.isPresent()) { | ||
| throw new IllegalArgumentException( | ||
| "Either operator uid or operator hash must be specified."); |
There was a problem hiding this comment.
nit: operator hash -> operator UID hash
gaborgsomogyi
force-pushed
the
FLINK-36929
branch
from
af8e251 to
10b6b73
Compare
|
cc @Zakelly |
gaborgsomogyi
force-pushed
the
FLINK-36929
branch
from
10b6b73 to
73b3c89
Compare
|
|
||
| @Override | ||
| public String factoryIdentifier() { | ||
| return "state"; |
There was a problem hiding this comment.
I think the identifier should be 'savepoint' as the FLIP describes.
There was a problem hiding this comment.
It would be great if also changing the class name to 'SavepointDynamicTableSourceFactory'
There was a problem hiding this comment.
Oh yeah, leftover what I've forgotten. Fixing it...
Yeah, it's intended to be optional. |
|
Not sure why but some of the tests are flaky so rebased to the latest master. Hope this helps. |
gaborgsomogyi
force-pushed
the
FLINK-36929
branch
from
ec89de7 to
2461550
Compare
|
Finally tests passed. |
There was a problem hiding this comment.
I have briefly read this and overall LGTM. I'm not an expert of SQL types so it would be great if @fsk119 or @lvyanquan could take a look.
| private Method getMethod(String rowFieldName, Class objectClass) { | ||
| String upperRowFieldName = | ||
| rowFieldName.substring(0, 1).toUpperCase() + rowFieldName.substring(1); | ||
| Method method; | ||
| try { | ||
| String methodName = "get" + upperRowFieldName; | ||
| method = objectClass.getMethod(methodName); | ||
| } catch (NoSuchMethodException e2) { | ||
| try { | ||
| String methodName = "is" + upperRowFieldName; | ||
| method = objectClass.getMethod(methodName); | ||
| } catch (NoSuchMethodException e3) { | ||
| throw new RuntimeException(e3); | ||
| } | ||
| } | ||
| return method; | ||
| } |
There was a problem hiding this comment.
IIUC this is using the java reflection to access the sub field of a pojo, right? And it is invoked each time an incoming instance should be converted, which is slow. Is it possible to maintain a class/method cache, or a converter, to avoid calling reflection on the fly?
There was a problem hiding this comment.
Just added guava cache to have a limited entries solution.
gaborgsomogyi
force-pushed
the
FLINK-36929
branch
from
2461550 to
4d8c993
Compare
|
Did a rebase to latest master. |
What is the purpose of the change
FLIP-496 has described the SQL connector for keyed savepoint data feature in-depth. In this PR I've added this functionality with automated tests. Documentation is intended to be added after the merge is order to have consumable PR pieces.
Brief change log
Add SQL connector for keyed savepoint data.
Verifying this change
Automated tests + manually on state files.
Does this pull request potentially affect one of the following parts:
@Public(Evolving): noDocumentation