beyond1920 commented on code in PR #11297:
URL: https://github.com/apache/hudi/pull/11297#discussion_r1616165704
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/HoodieLookupTableReader.java:
##########
@@ -49,15 +54,24 @@ public
HoodieLookupTableReader(SerializableSupplier<InputFormat<RowData, ?>> inp
public void open() throws IOException {
this.inputFormat = inputFormatSupplier.get();
inputFormat.configure(conf);
- InputSplit[] inputSplits = inputFormat.createInputSplits(1);
+ this.inputSplits =
Arrays.stream(inputFormat.createInputSplits(1)).collect(Collectors.toList());
Review Comment:
this.inputSplits = Arrays.stream(inputFormat.createInputSplits(1))
.collect(Collectors.toCollection(LinkedList::new));
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/HoodieLookupTableReader.java:
##########
@@ -41,6 +44,8 @@ public class HoodieLookupTableReader implements Serializable {
private InputFormat inputFormat;
+ private List<InputSplit> inputSplits;
Review Comment:
private Queue<InputSplit> inputSplits;
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/HoodieLookupTableReader.java:
##########
@@ -49,15 +54,24 @@ public
HoodieLookupTableReader(SerializableSupplier<InputFormat<RowData, ?>> inp
public void open() throws IOException {
this.inputFormat = inputFormatSupplier.get();
inputFormat.configure(conf);
- InputSplit[] inputSplits = inputFormat.createInputSplits(1);
+ this.inputSplits =
Arrays.stream(inputFormat.createInputSplits(1)).collect(Collectors.toList());
((RichInputFormat) inputFormat).openInputFormat();
- inputFormat.open(inputSplits[0]);
+ inputFormat.open(inputSplits.remove(0));
Review Comment:
inputFormat.open(inputSplits.poll());
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/HoodieLookupTableReader.java:
##########
@@ -49,15 +54,24 @@ public
HoodieLookupTableReader(SerializableSupplier<InputFormat<RowData, ?>> inp
public void open() throws IOException {
this.inputFormat = inputFormatSupplier.get();
inputFormat.configure(conf);
- InputSplit[] inputSplits = inputFormat.createInputSplits(1);
+ this.inputSplits =
Arrays.stream(inputFormat.createInputSplits(1)).collect(Collectors.toList());
((RichInputFormat) inputFormat).openInputFormat();
- inputFormat.open(inputSplits[0]);
+ inputFormat.open(inputSplits.remove(0));
}
@Nullable
public RowData read(RowData reuse) throws IOException {
if (!inputFormat.reachedEnd()) {
return (RowData) inputFormat.nextRecord(reuse);
+ } else {
+ while (!inputSplits.isEmpty()) {
+ // release the last itr first.
+ inputFormat.close();
+ inputFormat.open(inputSplits.remove(0));
Review Comment:
inputFormat.open(inputSplits.poll());
--
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]
