fix: IT of Influxdb

Author: aqni

core/src/main/java/cn/edu/tsinghua/iginx/engine/physical/task/memory/row/RowStreamToBatchStreamWrapper.java

@@ -26,6 +26,7 @@
 import cn.edu.tsinghua.iginx.engine.shared.data.read.*;
 import java.util.*;
 import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.util.Preconditions;
 
 public class RowStreamToBatchStreamWrapper implements BatchStream {
 
@@ -42,6 +43,7 @@ public RowStreamToBatchStreamWrapper(
 
   public RowStreamToBatchStreamWrapper(
       BufferAllocator allocator, RowStream rowStream, TaskMetrics metrics, int batchRowCount) {
+    Preconditions.checkArgument(batchRowCount > 0);
     this.allocator = Objects.requireNonNull(allocator);
     this.rowStream = Objects.requireNonNull(rowStream);
     this.metrics = Objects.requireNonNull(metrics);
@@ -96,7 +98,7 @@ public Batch getNext() throws PhysicalException {
     }
 
     boolean hasKey = schema.hasKey();
-    try (StopWatch watch = new StopWatch(metrics::accumulateCpuTime)) {
+    try (StopWatch ignored = new StopWatch(metrics::accumulateCpuTime)) {
       try (BatchBuilder builder = new BatchBuilder(allocator, schema, rows.size())) {
         for (Row row : rows) {
           if (hasKey) {

core/src/main/java/cn/edu/tsinghua/iginx/engine/shared/data/read/FetchMetricsRowStream.java

@@ -25,6 +25,7 @@
 import java.util.ArrayDeque;
 import java.util.Objects;
 import java.util.Queue;
+import org.apache.arrow.util.Preconditions;
 
 public class FetchMetricsRowStream implements RowStream {
 
@@ -34,6 +35,7 @@ public class FetchMetricsRowStream implements RowStream {
   private final Queue<Row> cache = new ArrayDeque<>();
 
   public FetchMetricsRowStream(RowStream delegate, TaskMetrics metrics, int batchRowCount) {
+    Preconditions.checkArgument(batchRowCount > 0);
     this.delegate = Objects.requireNonNull(delegate);
     this.metrics = Objects.requireNonNull(metrics);
     this.batchRowCount = batchRowCount;
@@ -51,19 +53,20 @@ public Header getHeader() throws PhysicalException {
 
   @Override
   public boolean hasNext() throws PhysicalException {
-    return !cache.isEmpty() || delegate.hasNext();
+    // TODO: InfluxDBQueryRowStream 存在 BUG，不能重复调用 hasNext 方法，否则会导致数据跳过
+    if (cache.isEmpty()) {
+      fetchBatch();
+    }
+    return !cache.isEmpty();
   }
 
   @Override
   public Row next() throws PhysicalException {
-    if (cache.isEmpty() && delegate.hasNext()) {
-      fetchBatch();
-    }
     return cache.remove();
   }
 
   private void fetchBatch() throws PhysicalException {
-    try (StopWatch watch = new StopWatch(metrics::accumulateCpuTime)) {
+    try (StopWatch ignored = new StopWatch(metrics::accumulateCpuTime)) {
       for (int i = 0; i < batchRowCount && delegate.hasNext(); i++) {
         cache.add(delegate.next());
       }

optimizer/src/main/java/cn/edu/tsinghua/iginx/physical/optimizer/naive/initializer/AddSchemaPrefixInfoGenerator.java

@@ -29,13 +29,15 @@
 import cn.edu.tsinghua.iginx.engine.shared.operator.*;
 import cn.edu.tsinghua.iginx.utils.Pair;
 import java.util.*;
+import org.apache.arrow.util.Preconditions;
 
 public class AddSchemaPrefixInfoGenerator implements UnaryExecutorFactory<ProjectExecutor> {
 
   private final AddSchemaPrefix operator;
 
   public AddSchemaPrefixInfoGenerator(AddSchemaPrefix operator) {
     this.operator = Objects.requireNonNull(operator);
+    Preconditions.checkNotNull(operator.getSchemaPrefix());
   }
 
   @Override
@@ -45,8 +47,7 @@ public ProjectExecutor initialize(ExecutorContext context, BatchSchema inputSche
     return new ProjectExecutor(context, inputSchema.raw(), expressions);
   }
 
-  public List<ScalarExpression<?>> getExpression(ExecutorContext context, BatchSchema inputSchema)
-      throws ComputeException {
+  public List<ScalarExpression<?>> getExpression(ExecutorContext context, BatchSchema inputSchema) {
     List<Pair<String, Integer>> columnsAndIndices = getColumnsAndIndices(inputSchema, operator);
     List<ScalarExpression<?>> ret = new ArrayList<>();
     for (Pair<String, Integer> pair : columnsAndIndices) {