github-advanced-security[bot] commented on code in PR #15470:
URL: https://github.com/apache/druid/pull/15470#discussion_r1412550178
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/WindowOperatorQueryFrameProcessor.java:
##########
@@ -0,0 +1,172 @@
+package org.apache.druid.msq.querykit;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.druid.collections.ResourceHolder;
+import org.apache.druid.frame.Frame;
+import org.apache.druid.frame.channel.ReadableFrameChannel;
+import org.apache.druid.frame.channel.WritableFrameChannel;
+import org.apache.druid.frame.processor.ReturnOrAwait;
+import org.apache.druid.frame.read.FrameReader;
+import org.apache.druid.frame.segment.FrameSegment;
+import org.apache.druid.frame.util.SettableLongVirtualColumn;
+import org.apache.druid.frame.write.FrameWriter;
+import org.apache.druid.frame.write.FrameWriterFactory;
+import org.apache.druid.java.util.common.Unit;
+import org.apache.druid.java.util.common.io.Closer;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.msq.input.ReadableInput;
+import org.apache.druid.msq.input.table.SegmentWithDescriptor;
+import org.apache.druid.msq.querykit.scan.ScanQueryFrameProcessor;
+import org.apache.druid.query.operator.OffsetLimit;
+import org.apache.druid.query.operator.Operator;
+import org.apache.druid.query.operator.OperatorFactory;
+import org.apache.druid.query.operator.SegmentToRowsAndColumnsOperator;
+import org.apache.druid.query.operator.WindowOperatorQuery;
+import org.apache.druid.query.rowsandcols.LazilyDecoratedRowsAndColumns;
+import org.apache.druid.query.rowsandcols.RowsAndColumns;
+import org.apache.druid.query.rowsandcols.concrete.FrameRowsAndColumns;
+import org.apache.druid.query.rowsandcols.concrete.RowBasedFrameRowAndColumns;
+import org.apache.druid.query.scan.ScanQuery;
+import org.apache.druid.segment.Cursor;
+import org.apache.druid.segment.Segment;
+import org.apache.druid.segment.SegmentReference;
+import org.apache.druid.segment.SimpleAscendingOffset;
+import org.apache.druid.segment.SimpleSettableOffset;
+import org.apache.druid.segment.StorageAdapter;
+import org.apache.druid.segment.VirtualColumn;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.timeline.SegmentId;
+
+import javax.annotation.Nullable;
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Function;
+
+public class WindowOperatorQueryFrameProcessor extends BaseLeafFrameProcessor
+{
+
+ private static final Logger log = new Logger(ScanQueryFrameProcessor.class);
+ private final WindowOperatorQuery query;
+ private final ObjectMapper jsonMapper;
+ private final SettableLongVirtualColumn partitionBoostVirtualColumn;
+ private final VirtualColumns frameWriterVirtualColumns;
+ private final Closer closer = Closer.create();
+
+ private Cursor cursor;
+ private Segment segment;
+ private final SimpleSettableOffset cursorOffset = new
SimpleAscendingOffset(Integer.MAX_VALUE);
+ private FrameWriter frameWriter;
+ private long currentAllocatorCapacity; // Used for generating
FrameRowTooLargeException if needed
+
+ public WindowOperatorQueryFrameProcessor(
+ final WindowOperatorQuery query,
+ final ObjectMapper jsonMapper,
+ final ReadableInput baseInput,
+ final Function<SegmentReference, SegmentReference> segmentMapFn,
+ final ResourceHolder<WritableFrameChannel> outputChannelHolder,
+ final ResourceHolder<FrameWriterFactory> frameWriterFactoryHolder
+ )
+ {
+ super(
+ baseInput,
+ segmentMapFn,
+ outputChannelHolder,
+ frameWriterFactoryHolder
+ );
+ this.query = query;
+ this.jsonMapper = jsonMapper;
+ this.partitionBoostVirtualColumn = new
SettableLongVirtualColumn(QueryKitUtils.PARTITION_BOOST_COLUMN);
+
+ final List<VirtualColumn> frameWriterVirtualColumns = new ArrayList<>();
+ frameWriterVirtualColumns.add(partitionBoostVirtualColumn);
+
+ final VirtualColumn segmentGranularityVirtualColumn =
+ QueryKitUtils.makeSegmentGranularityVirtualColumn(jsonMapper, query);
+
+ if (segmentGranularityVirtualColumn != null) {
+ frameWriterVirtualColumns.add(segmentGranularityVirtualColumn);
+ }
+
+ this.frameWriterVirtualColumns =
VirtualColumns.create(frameWriterVirtualColumns);
+ }
+ // deep storage
+ @Override
+ protected ReturnOrAwait<Unit> runWithSegment(SegmentWithDescriptor segment)
throws IOException
+ {
+ return null;
+ }
+
+ // realtime
+ @Override
+ protected ReturnOrAwait<Unit> runWithLoadedSegment(SegmentWithDescriptor
segment) throws IOException
+ {
+ return null;
+ }
+
+ // previous stage output
+ @Override
+ protected ReturnOrAwait<Unit> runWithInputChannel(ReadableFrameChannel
inputChannel, FrameReader inputFrameReader)
+ throws IOException
+ {
+ // Read the frames from the channel
+ // convert to FrameRowsAndColumns
+
+ if (inputChannel.canRead()) {
+ Frame f = inputChannel.read();
+ final FrameSegment frameSegment = new FrameSegment(f, inputFrameReader,
SegmentId.dummy("x"));
Review Comment:
## Unread local variable
Variable 'FrameSegment frameSegment' is never read.
[Show more
details](https://github.com/apache/druid/security/code-scanning/6019)
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
