[Flink] Support analysis statistics in flink #4280
Conversation
| try { | ||
| Table table = catalog.getTable(toIdentifier(tablePath)); | ||
| Preconditions.checkArgument( | ||
| table instanceof FileStoreTable, "Can't analyze system table."); |
There was a problem hiding this comment.
Change "Can't analyze system table." to "Now only support analyze FileStoreTable."
Because lots of table implement Table.
| Preconditions.checkArgument( | ||
| table instanceof FileStoreTable, "Can't analyze system table."); | ||
| if (!table.latestSnapshotId().isPresent()) { | ||
| return; |
| TableStatsUtil.createTableStats((FileStoreTable) table, tableStatistics); | ||
|
|
||
| FileStoreCommit commit = | ||
| ((FileStoreTable) table).store().newCommit(UUID.randomUUID().toString()); |
There was a problem hiding this comment.
Use CoreOptions#createCommitUser instead.
| FileStoreCommit commit = | ||
| ((FileStoreTable) table).store().newCommit(UUID.randomUUID().toString()); | ||
| commit.commitStatistics(tableStats, BatchWriteBuilder.COMMIT_IDENTIFIER); | ||
|
|
There was a problem hiding this comment.
Delete the excess blank line.
| if (!ignoreIfNotExists) { | ||
| throw new TableNotExistException(getName(), tablePath); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public final void alterTableColumnStatistics( |
There was a problem hiding this comment.
1、Same suggests with this method and alterTableStatistics.
2、Abstract a new method named alterTableStatisticsInternal, because alterTableStatistics and alterTableColumnStatistics has lots of same code.
| sql("ANALYZE TABLE T COMPUTE STATISTICS"); | ||
| Statistics stats = paimonTable("T").statistics().get(); | ||
|
|
||
| Assertions.assertEquals(2L, stats.mergedRecordCount().getAsLong()); |
There was a problem hiding this comment.
'OptionalLong.getAsLong()' without 'isPresent()' check
|
|
||
| @Test | ||
| public void testAnalyzeTableColumn() throws Catalog.TableNotExistException { | ||
|
|
There was a problem hiding this comment.
Delete the excess blank line.
|
|
||
| sql("ANALYZE TABLE T COMPUTE STATISTICS FOR ALL COLUMNS"); | ||
|
|
||
| Statistics stats = paimonTable("T").statistics().get(); |
There was a problem hiding this comment.
'Optional.get()' without 'isPresent()' check
|
|
||
| Statistics stats = paimonTable("T").statistics().get(); | ||
|
|
||
| Assertions.assertEquals(4L, stats.mergedRecordCount().getAsLong()); |
There was a problem hiding this comment.
'OptionalLong.getAsLong()' without 'isPresent()' check
| colStats.get("bytes_col")); | ||
|
|
||
| Assertions.assertEquals( | ||
| ColStats.newColStats(3, 2L, new Integer(1), new Integer(4), 0L, null, null), |
There was a problem hiding this comment.
Why add two unnecessary boxing here?
|
@wwj6591812 thanks for your review,I have completed the modification. |
herefree
force-pushed
the
support-flink-statistics
branch
from
b5f1983 to
10bebab
Compare
|
+1 |
JingsongLi
changed the title
Purpose
support statistics in flink
Linked issue: close #xxx
Tests
API and Format
Documentation