Extract PartitionedOutputOperator.appendRow
@@ -353,24 +353,14 @@ public class PartitionedOutputOperator
for (int position = 0; position < page.getPositionCount(); position++) {
if (nullChannel.isPresent() && page.getBlock(nullChannel.getAsInt()).isNull(position)) {
for (PageBuilder pageBuilder : pageBuilders) {
- pageBuilder.declarePosition();
-
- for (int channel = 0; channel < sourceTypes.size(); channel++) {
- Type type = sourceTypes.get(channel);
- type.appendTo(page.getBlock(channel), position, pageBuilder.getBlockBuilder(channel));
- }
+ appendRow(pageBuilder, page, position);
}
}
else {
int partition = partitionFunction.getPartition(partitionFunctionArgs, position);
PageBuilder pageBuilder = pageBuilders.get(partition);
- pageBuilder.declarePosition();
-
- for (int channel = 0; channel < sourceTypes.size(); channel++) {
- Type type = sourceTypes.get(channel);
- type.appendTo(page.getBlock(channel), position, pageBuilder.getBlockBuilder(channel));
- }
+ appendRow(pageBuilder, page, position);
}
}
return flush(false);
@@ -391,6 +381,16 @@ public class PartitionedOutputOperator
return new Page(page.getPositionCount(), blocks);
}
+ private void appendRow(PageBuilder pageBuilder, Page page, int position)
+ {
+ pageBuilder.declarePosition();
+
+ for (int channel = 0; channel < sourceTypes.size(); channel++) {
+ Type type = sourceTypes.get(channel);
+ type.appendTo(page.getBlock(channel), position, pageBuilder.getBlockBuilder(channel));
+ }
+ }
+
public ListenableFuture<?> flush(boolean force)
{
// add all full pages to output buffer
