Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[flink] Fix read partitioned table throw LogScanner is not subscribed any bucket #285

Merged
merged 3 commits into from
Jan 7, 2025
+138 −58
Merged

[flink] Fix read partitioned table throw LogScanner is not subscribed any bucket #285

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c538928
debug
luoyuxia Dec 19, 2024
fda9023
[flink] Fix read partitioned table throw LogScanner is not subscribed…
luoyuxia Dec 27, 2024
142a897
address comments
luoyuxia Jan 7, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions fluss-common/src/main/java/com/alibaba/fluss/metadata/TableDescriptor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,6 +128,11 @@ private TableDescriptor(
.allMatch(e -> e.getKey() != null && e.getValue() != null),
"options cannot have null keys or values.");

if (getAutoPartitionStrategy().isAutoPartitionEnabled() && !isPartitioned()) {
throw new IllegalArgumentException(
"Auto partition is only supported when table is partitioned.");
}

if (hasPrimaryKey()
&& getKvFormat() == KvFormat.COMPACTED
&& getLogFormat() != LogFormat.ARROW) {
Expand Down
13 changes: 13 additions & 0 deletions fluss-common/src/test/java/com/alibaba/fluss/metadata/TableDescriptorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

import com.alibaba.fluss.config.ConfigBuilder;
import com.alibaba.fluss.config.ConfigOption;
import com.alibaba.fluss.config.ConfigOptions;
import com.alibaba.fluss.types.DataTypes;

import org.junit.jupiter.api.Test;
Expand Down Expand Up @@ -331,4 +332,16 @@ void testPartitionedTable() {
.hasMessage(
"Bucket key [f0, f3] shouldn't include any column in partition keys [f0].");
}

@Test
void testAutoPartitionForNonPartitionedTableShouldThrowException() {
assertThatThrownBy(
() ->
TableDescriptor.builder()
.schema(SCHEMA_1)
.property(ConfigOptions.TABLE_AUTO_PARTITION_ENABLED, true)
.build())
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Auto partition is only supported when table is partitioned.");
}
}
41 changes: 22 additions & 19 deletions .../main/java/com/alibaba/fluss/connector/flink/source/enumerator/FlinkSourceEnumerator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,8 +138,8 @@ public FlinkSourceEnumerator(
this(
tablePath,
flussConf,
isPartitioned,
hasPrimaryKey,
isPartitioned,
context,
Collections.emptySet(),
Collections.emptyMap(),
Expand All @@ -151,8 +151,8 @@ public FlinkSourceEnumerator(
public FlinkSourceEnumerator(
TablePath tablePath,
Configuration flussConf,
boolean isPartitioned,
boolean hasPrimaryKey,
boolean isPartitioned,
SplitEnumeratorContext<SourceSplitBase> context,
Set<TableBucket> assignedTableBuckets,
Map<Long, String> assignedPartitions,
Expand Down Expand Up @@ -306,26 +306,29 @@ private PartitionChange getPartitionChange(Set<PartitionInfo> fetchedPartitionIn
}
};

Set<PartitionInfo> assignedOrPendingPartitions = new HashSet<>();
assignedPartitions.forEach(
(partitionId, partitionName) ->
dedupOrMarkAsRemoved.accept(new PartitionInfo(partitionId, partitionName)));
assignedOrPendingPartitions.add(
new PartitionInfo(partitionId, partitionName)));

pendingSplitAssignment.forEach(
(reader, splits) ->
splits.forEach(
split -> {
long partitionId =
checkNotNull(
split.getTableBucket().getPartitionId(),
"partition id shouldn't be null for the splits of partitioned table.");
String partitionName =
checkNotNull(
split.getPartitionName(),
"partition name shouldn't be null for the splits of partitioned table.");
PartitionInfo partitionInfo =
new PartitionInfo(partitionId, partitionName);
dedupOrMarkAsRemoved.accept(partitionInfo);
}));
pendingSplitAssignment.values().stream()
.flatMap(Collection::stream)
.forEach(
split -> {
long partitionId =
checkNotNull(
split.getTableBucket().getPartitionId(),
"partition id shouldn't be null for the splits of partitioned table.");
String partitionName =
checkNotNull(
split.getPartitionName(),
"partition name shouldn't be null for the splits of partitioned table.");
assignedOrPendingPartitions.add(
new PartitionInfo(partitionId, partitionName));
});

assignedOrPendingPartitions.forEach(dedupOrMarkAsRemoved);

if (!removedPartitionIds.isEmpty()) {
LOG.info("Discovered removed partitions: {}", removedPartitionIds);
Expand Down
13 changes: 2 additions & 11 deletions ...c/main/java/com/alibaba/fluss/connector/flink/source/reader/FlinkRecordsWithSplitIds.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,18 +55,9 @@ public class FlinkRecordsWithSplitIds implements RecordsWithSplitIds<RecordAndPo
private @Nullable TableBucket currentTableBucket;
private @Nullable Long currentSplitStoppingOffset;

// for multiple splits
public FlinkRecordsWithSplitIds(
Map<String, CloseableIterator<RecordAndPos>> splitRecords,
Iterator<String> splitIterator,
Iterator<TableBucket> tableBucketIterator,
public static FlinkRecordsWithSplitIds emptyRecords(
FlinkSourceReaderMetrics flinkSourceReaderMetrics) {
this(
splitRecords,
splitIterator,
tableBucketIterator,
new HashSet<>(),
flinkSourceReaderMetrics);
return new FlinkRecordsWithSplitIds(Collections.emptySet(), flinkSourceReaderMetrics);
}

// only for single split
Expand Down
9 changes: 8 additions & 1 deletion ...src/main/java/com/alibaba/fluss/connector/flink/source/reader/FlinkSourceSplitReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -144,9 +144,16 @@ public RecordsWithSplitIds<RecordAndPos> fetch() throws IOException {
} else {
// may need to finish empty log splits
if (!emptyLogSplits.isEmpty()) {
FlinkRecordsWithSplitIds records =
new FlinkRecordsWithSplitIds(
new HashSet<>(emptyLogSplits), flinkSourceReaderMetrics);
emptyLogSplits.clear();
return new FlinkRecordsWithSplitIds(emptyLogSplits, flinkSourceReaderMetrics);
return records;
} else {
// if not subscribe any buckets, just return empty records
if (subscribedBuckets.isEmpty()) {
return FlinkRecordsWithSplitIds.emptyRecords(flinkSourceReaderMetrics);
}
ScanRecords scanRecords = logScanner.poll(POLL_TIMEOUT);
return forLogRecords(scanRecords);
}
Expand Down
6 changes: 3 additions & 3 deletions ...ctor-flink/src/test/java/com/alibaba/fluss/connector/flink/sink/FlinkTableSinkITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -641,16 +641,16 @@ void testDeleteAndUpdateStmtOnPartitionedPkTable() throws Exception {
void testUnsupportedDeleteAndUpdateStmtOnLogTable(boolean isPartitionedTable) {
String tableName =
isPartitionedTable ? "partitioned_log_table_delete_test" : "log_table_delete_test";
String partitionedTableStmt =
" partitioned by (c) with ('table.auto-partition.enabled' = 'true','table.auto-partition.time-unit' = 'year')";
tBatchEnv.executeSql(
String.format(
"create table %s ("
+ " a int not null,"
+ " b bigint, "
+ " c string"
+ ")"
+ (isPartitionedTable ? " partitioned by (c) " : "")
+ "with ('table.auto-partition.enabled' = 'true',"
+ " 'table.auto-partition.time-unit' = 'year')",
+ (isPartitionedTable ? partitionedTableStmt : ""),
tableName));
assertThatThrownBy(
() ->
Expand Down
20 changes: 13 additions & 7 deletions ...t/java/com/alibaba/fluss/connector/flink/source/enumerator/FlinkSourceEnumeratorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -405,19 +405,25 @@ void testDiscoverPartitionsPeriodically(boolean isPrimaryKeyTable) throws Throwa
Map<Long, String> partitionNameByIds =
waitUntilPartitions(zooKeeperClient, DEFAULT_TABLE_PATH);
enumerator.start();
// register all readers
for (int i = 0; i < numSubtasks; i++) {
registerReader(context, enumerator, i);
}

// invoke partition discovery callable again and there should assignments.
// invoke partition discovery callable again and there should be pending assignments.
runPeriodicPartitionDiscovery(context);

// register two readers
registerReader(context, enumerator, 0);
registerReader(context, enumerator, 1);

// invoke partition discovery callable again, shouldn't produce RemovePartitionEvent.
runPeriodicPartitionDiscovery(context);
assertThat(context.getSentSourceEvent()).isEmpty();

// now, register the third reader
registerReader(context, enumerator, 2);

// check the assignments
Map<Integer, List<SourceSplitBase>> expectedAssignment =
expectAssignments(enumerator, tableId, partitionNameByIds);
Map<Integer, List<SourceSplitBase>> actualAssignments =
getLastReadersAssignments(context);
Map<Integer, List<SourceSplitBase>> actualAssignments = getReadersAssignments(context);
checkAssignmentIgnoreOrder(actualAssignments, expectedAssignment);

// now, create a new partition and runPeriodicPartitionDiscovery again,
Expand Down
53 changes: 53 additions & 0 deletions ...test/java/com/alibaba/fluss/connector/flink/source/reader/FlinkSourceSplitReaderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,13 +49,15 @@

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import static com.alibaba.fluss.client.scanner.log.LogScanner.EARLIEST_OFFSET;
import static com.alibaba.fluss.connector.flink.source.testutils.RecordAndPosAssert.assertThatRecordAndPos;
import static com.alibaba.fluss.record.TestData.DATA1_ROW_TYPE;
import static com.alibaba.fluss.testutils.DataTestUtils.compactedRow;
Expand Down Expand Up @@ -268,6 +270,57 @@ void testHandleMixSnapshotLogSplitChangesAndFetch() throws Exception {
}
}

@Test
void testNoSubscribedBucket() throws Exception {
TablePath tablePath = TablePath.of(DEFAULT_DB, "test-no-subscribe-bucket-table");
Schema schema =
Schema.newBuilder()
.column("id", DataTypes.INT())
.column("name", DataTypes.STRING())
.build();
final TableDescriptor tableDescriptor =
TableDescriptor.builder().schema(schema).distributedBy(1).build();
createTable(tablePath, tableDescriptor);

try (FlinkSourceSplitReader splitReader =
createSplitReader(tablePath, schema.toRowType())) {
// fetch shouldn't throw exception
RecordsWithSplitIds<RecordAndPos> records = splitReader.fetch();
assertThat(records.nextSplit()).isNull();
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

assert records.nextRecordFromSplit() is null as well?

Copy link
Collaborator Author

@luoyuxia luoyuxia Jan 7, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If assert records.nextRecordFromSplit() is null, it will throw java.lang.NullPointerException: Make sure nextSplit() did not return null before iterate over the records split. FlinkRecordsWithSplitIds#nextRecordFromSplit will check the currentRecordIterator is not null, but if no records in the FlinkRecordsWithSplitIds, the currentRecordIterator will be null.

I think check nextSplit is enough since if nextSplit is null, flink fetcher won't call method records.nextRecordFromSplit()

}
}

@Test
void testSubscribeEmptySplits() throws Exception {
TablePath tablePath = TablePath.of(DEFAULT_DB, "test-subscribe-empty-splits");
Schema schema =
Schema.newBuilder()
.column("id", DataTypes.INT())
.column("name", DataTypes.STRING())
.build();
long tableId =
createTable(
tablePath,
TableDescriptor.builder().schema(schema).distributedBy(3).build());

// create two empty splits with log start offset equal to end offset
LogSplit split1 = new LogSplit(new TableBucket(tableId, 0), null, 0, 0);
LogSplit split2 = new LogSplit(new TableBucket(tableId, 1), null, 0, 0);
LogSplit split3 = new LogSplit(new TableBucket(tableId, 2), null, EARLIEST_OFFSET);
List<SourceSplitBase> subscribeSplits = Arrays.asList(split1, split2, split3);

try (FlinkSourceSplitReader splitReader =
createSplitReader(tablePath, schema.toRowType())) {
splitReader.handleSplitsChanges(new SplitsAddition<>(subscribeSplits));

// fetch records
RecordsWithSplitIds<RecordAndPos> records = splitReader.fetch();
// finished splits should be split1,split2
assertThat(records.finishedSplits())
.containsExactlyInAnyOrder(split1.splitId(), split2.splitId());
}
}

// ------------------

private void assignSplitsAndFetchUntilRetrieveRecords(
Expand Down
36 changes: 19 additions & 17 deletions ...main/java/com/alibaba/fluss/lakehouse/paimon/source/enumerator/FlinkSourceEnumerator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -591,26 +591,28 @@ private PartitionChange getPartitionChange(Set<PartitionInfo> fetchedPartitionIn
}
};

Set<PartitionInfo> assignedOrPendingPartitions = new HashSet<>();
assignedPartitions.forEach(
(partitionId, partitionName) ->
dedupOrMarkAsRemoved.accept(new PartitionInfo(partitionId, partitionName)));
assignedOrPendingPartitions.add(
new PartitionInfo(partitionId, partitionName)));
pendingSplitAssignment.values().stream()
.flatMap(Collection::stream)
.forEach(
split -> {
long partitionId =
checkNotNull(
split.getTableBucket().getPartitionId(),
"partition id shouldn't be null for the splits of partitioned table.");
String partitionName =
checkNotNull(
split.getPartitionName(),
"partition name shouldn't be null for the splits of partitioned table.");
assignedOrPendingPartitions.add(
new PartitionInfo(partitionId, partitionName));
});

pendingSplitAssignment.forEach(
(reader, splits) ->
splits.forEach(
split -> {
long partitionId =
checkNotNull(
split.getTableBucket().getPartitionId(),
"partition id shouldn't be null for the splits of partitioned table.");
String partitionName =
checkNotNull(
split.getPartitionName(),
"partition name shouldn't be null for the splits of partitioned table.");
PartitionInfo partitionInfo =
new PartitionInfo(partitionId, partitionName);
dedupOrMarkAsRemoved.accept(partitionInfo);
}));
assignedOrPendingPartitions.forEach(dedupOrMarkAsRemoved);

if (!removedPartitionIds.isEmpty()) {
LOG.info("Discovered removed partitions: {}", removedPartitionIds);
Expand Down