Copilot commented on code in PR #1925:
URL: https://github.com/apache/fluss/pull/1925#discussion_r2493184675
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java:
##########
@@ -89,23 +89,16 @@ void testPrimaryKeyTableWithNoSnapshotSplits() throws
Throwable {
registerReader(context, enumerator, subtaskId, "localhost-" +
subtaskId);
enumerator.handleSplitRequest(subtaskId, "localhost-" +
subtaskId);
}
- waitUntilTieringTableSplitAssignmentReady(context,
DEFAULT_BUCKET_NUM, 200L);
- List<TieringSplit> expectedAssignment = new ArrayList<>();
- for (int bucketId = 0; bucketId < DEFAULT_BUCKET_NUM; bucketId++) {
- expectedAssignment.add(
- new TieringLogSplit(
- tablePath,
- new TableBucket(tableId, bucketId),
- null,
- EARLIEST_OFFSET,
- 0,
- expectNumberOfSplits));
- }
+
+ // try to assign splits
+ context.runPeriodicCallable(0);
+
List<TieringSplit> actualAssignment = new ArrayList<>();
context.getSplitsAssignmentSequence()
.forEach(a ->
a.assignment().values().forEach(actualAssignment::addAll));
- assertThat(actualAssignment).isEqualTo(expectedAssignment);
+ // no spilt assignment for empty buckets
Review Comment:
Corrected spelling of 'spilt' to 'split'.
```suggestion
// no split assignment for empty buckets
```
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java:
##########
@@ -496,21 +483,29 @@ void testPartitionedLogTableSplits() throws Throwable {
registerReader(context, enumerator, subtaskId, "localhost-" +
subtaskId);
enumerator.handleSplitRequest(subtaskId, "localhost-" +
subtaskId);
}
- waitUntilTieringTableSplitAssignmentReady(
- context, DEFAULT_BUCKET_NUM * partitionNameByIds.size(),
3000L);
+
+ Map<Long, Map<Integer, Long>> bucketOffsetOfFirstWrite =
+ appendRowForPartitionedTable(
+ tablePath,
+ DEFAULT_AUTO_PARTITIONED_LOG_TABLE_DESCRIPTOR,
+ partitionNameByIds,
+ 0,
+ 1);
+
+ waitUntilTieringTableSplitAssignmentReady(context,
partitionNameByIds.size(), 3000L);
List<TieringSplit> expectedAssignment = new ArrayList<>();
for (Map.Entry<String, Long> partitionNameById :
partitionNameByIds.entrySet()) {
- for (int tableBucket = 0; tableBucket < DEFAULT_BUCKET_NUM;
tableBucket++) {
- long partitionId = partitionNameById.getValue();
+ long partitionId = partitionNameById.getValue();
+ for (int tableBucket :
bucketOffsetOfFirstWrite.get(partitionId).keySet()) {
expectedAssignment.add(
new TieringLogSplit(
tablePath,
new TableBucket(tableId, partitionId,
tableBucket),
partitionNameById.getKey(),
EARLIEST_OFFSET,
- 0L,
- expectNumberOfSplits));
+
bucketOffsetOfFirstWrite.get(partitionId).get(tableBucket),
+ bucketOffsetOfFirstWrite.size()));
Review Comment:
The `numberOfSplits` parameter should be `expectNumberOfSplits` instead of
`bucketOffsetOfFirstWrite.size()`. The `numberOfSplits` field represents 'the
total number of splits in one round of tiering' (as documented in
TieringSplit.java line 41), and should use the consistent value
`expectNumberOfSplits` (which is 6) as seen in similar code at line 448, not
the number of partitions.
```suggestion
expectNumberOfSplits));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
