Flink: add limit pushdown for IcebergTableSource#1822
Conversation
zhangjun0x01
force-pushed
the
limitPushDown
branch
2 times, most recently
from
bd3efd3 to
06ae616
Compare
|
|
||
| public IcebergTableSource(TableLoader loader, TableSchema schema, Map<String, String> properties) { | ||
| this(loader, schema, properties, null); | ||
| this(loader, schema, properties, null, false, -1); |
There was a problem hiding this comment.
Why disable the limit push down for flink table source ?
There was a problem hiding this comment.
I think it may be related to the design of the LimitableTableSource interface in flink 1.11. I looked up some implement classes of LimitableTableSource in flink, such as HiveTableSource. By default, the limit pushdown is disabled
There was a problem hiding this comment.
Oh, the table won't have a limit cause by default, so we should set it disabled by default. It's OK here.
There was a problem hiding this comment.
Could we add the final modifier ? Also no need to initialize it with a default false because the constructor will always assign a given value to it.
There was a problem hiding this comment.
if limit is -1L, then means we've disabled the limit push down , right ? If so, why do we need two fields ?
There was a problem hiding this comment.
I also think we can judge whether to pushdown by the value of limit, but the LimitableTableSource interface provides two methods, isLimitPushedDown and applyLimit. From the method comments, I think the author wants to judge whether to pushdown by the isLimitPushedDown method.
In versions after flink 1.12, a new interface SupportsLimitPushDown is provided. This interface only provides one method. I think we can judge pushdown by the value of limit
There was a problem hiding this comment.
Here the isLimitPushDown is always true, how about use the string String.format(", LimitPushDown: %d", limit) ?
There was a problem hiding this comment.
Do we need this ? we don't parse the limit from properties, right ?
There was a problem hiding this comment.
How about just using the context.limit() in reachedEnd, rather than introducing another new transient field limit ?
There was a problem hiding this comment.
I deleted the field limit
| public void testLimitPushDown() { | ||
| sql("INSERT INTO %s VALUES (1,'a'),(2,'b')", TABLE_NAME); | ||
|
|
||
| String querySql = String.format("SELECT * FROM %s LIMIT 1", TABLE_NAME); |
There was a problem hiding this comment.
Do we need to add those test cases:
Case.1 : SELECT * FROM test LIMIT -1
Case.2 : SELECT * FROM test LIMIT 0
Case.3: SELECT * FROM test LIMIT 3 , means the limit exceeds the total rows in table .
Case.4: SELECT * FROM test WHERE a = 1 AND LIMIT 2 , query data with both limit and filters
etc.
There was a problem hiding this comment.
I added the test case
zhangjun0x01
force-pushed
the
limitPushDown
branch
from
06ae616 to
c526a49
Compare
fix some bugs
zhangjun0x01
force-pushed
the
limitPushDown
branch
from
c526a49 to
3975631
Compare
|
Merged this PR, thanks @zhangjun0x01 for contributing. |
2 participants
related to #1816
now the source implement the interface
LimitableTableSource,after flink 1.12 ,implement the
SupportsLimitPushDown