Add GroupNode and parallelize TableFunctionProcessorNode

example/udf/src/main/java/org/apache/iotdb/udf/table/ExcludeColumnExample.java

@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.udf.table;
+
+import org.apache.iotdb.udf.api.exception.UDFException;
+import org.apache.iotdb.udf.api.relational.TableFunction;
+import org.apache.iotdb.udf.api.relational.table.TableFunctionAnalysis;
+import org.apache.iotdb.udf.api.relational.table.TableFunctionProcessorProvider;
+import org.apache.iotdb.udf.api.relational.table.argument.Argument;
+import org.apache.iotdb.udf.api.relational.table.argument.DescribedSchema;
+import org.apache.iotdb.udf.api.relational.table.argument.ScalarArgument;
+import org.apache.iotdb.udf.api.relational.table.argument.TableArgument;
+import org.apache.iotdb.udf.api.relational.table.processor.TableFunctionDataProcessor;
+import org.apache.iotdb.udf.api.relational.table.specification.ParameterSpecification;
+import org.apache.iotdb.udf.api.relational.table.specification.ScalarParameterSpecification;
+import org.apache.iotdb.udf.api.relational.table.specification.TableParameterSpecification;
+import org.apache.iotdb.udf.api.type.Type;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * This is an internal example of the TableFunction implementation. This function is declared as row
+ * semantic without pass through columns.
+ *
+ * <p>CREATE DATABASE test;
+ *
+ * <p>USE test;
+ *
+ * <p>CREATE TABLE t1(device_id STRING TAG, s1 TEXT FIELD, s2 INT32 FIELD);
+ *
+ * <p>INSERT INTO t1(time, device_id, s1, s2) VALUES (1, 'd1', 'a', 1), (2, 'd1', null, 2), (3,
+ * 'd1', 'c', null);
+ *
+ * <p>CREATE FUNCTION exclude_column AS 'org.apache.iotdb.udf.table.ExcludeColumnExample';
+ *
+ * <p>SHOW FUNCTIONS;
+ *
+ * <p>SELECT * FROM TABLE(exclude_column(TABLE(t1), 's2'));
+ */
+public class ExcludeColumnExample implements TableFunction {
+  private final String TBL_PARAM = "DATA";
+  private final String COL_PARAM = "EXCLUDE";
+
+  @Override
+  public List<ParameterSpecification> getArgumentsSpecifications() {
+    return Arrays.asList(
+        TableParameterSpecification.builder().name(TBL_PARAM).rowSemantics().build(),
+        ScalarParameterSpecification.builder().name(COL_PARAM).type(Type.STRING).build());
+  }
+
+  @Override
+  public TableFunctionAnalysis analyze(Map<String, Argument> arguments) throws UDFException {
+    TableArgument tableArgument = (TableArgument) arguments.get(TBL_PARAM);
+    String excludeColumn = (String) ((ScalarArgument) arguments.get(COL_PARAM)).getValue();
+    List<Integer> requiredColumns = new ArrayList<>();
+    DescribedSchema.Builder schemaBuilder = DescribedSchema.builder();
+    for (int i = 0; i < tableArgument.getFieldNames().size(); i++) {
+      Optional<String> fieldName = tableArgument.getFieldNames().get(i);
+      if (!fieldName.isPresent() || !fieldName.get().equalsIgnoreCase(excludeColumn)) {
+        requiredColumns.add(i);
+        schemaBuilder.addField(fieldName, tableArgument.getFieldTypes().get(i));
+      }
+    }
+    return TableFunctionAnalysis.builder()
+        .properColumnSchema(schemaBuilder.build())
+        .requiredColumns(TBL_PARAM, requiredColumns)
+        .build();
+  }
+
+  @Override
+  public TableFunctionProcessorProvider getProcessorProvider(Map<String, Argument> arguments) {
+    return new TableFunctionProcessorProvider() {
+      @Override
+      public TableFunctionDataProcessor getDataProcessor() {
+        return (input, properColumnBuilders, passThroughIndexBuilder) -> {
+          for (int i = 0; i < input.size(); i++) {
+            if (input.isNull(i)) {
+              properColumnBuilders.get(i).appendNull();
+            } else {
+              properColumnBuilders.get(i).writeObject(input.getObject(i));
+            }
+          }
+        };
+      }
+    };
+  }
+}

example/udf/src/main/java/org/apache/iotdb/udf/table/HOPTableFunction.java

@@ -0,0 +1,158 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.udf.table;
+
+import org.apache.iotdb.udf.api.exception.UDFException;
+import org.apache.iotdb.udf.api.relational.TableFunction;
+import org.apache.iotdb.udf.api.relational.access.Record;
+import org.apache.iotdb.udf.api.relational.table.TableFunctionAnalysis;
+import org.apache.iotdb.udf.api.relational.table.TableFunctionProcessorProvider;
+import org.apache.iotdb.udf.api.relational.table.argument.Argument;
+import org.apache.iotdb.udf.api.relational.table.argument.DescribedSchema;
+import org.apache.iotdb.udf.api.relational.table.argument.ScalarArgument;
+import org.apache.iotdb.udf.api.relational.table.argument.TableArgument;
+import org.apache.iotdb.udf.api.relational.table.processor.TableFunctionDataProcessor;
+import org.apache.iotdb.udf.api.relational.table.specification.ParameterSpecification;
+import org.apache.iotdb.udf.api.relational.table.specification.ScalarParameterSpecification;
+import org.apache.iotdb.udf.api.relational.table.specification.TableParameterSpecification;
+import org.apache.iotdb.udf.api.type.Type;
+
+import org.apache.tsfile.block.column.ColumnBuilder;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+public class HOPTableFunction implements TableFunction {
+
+  private static final String DATA_PARAMETER_NAME = "DATA";
+  private static final String TIMECOL_PARAMETER_NAME = "TIMECOL";
+  private static final String SLIDE_PARAMETER_NAME = "SLIDE";
+  private static final String SIZE_PARAMETER_NAME = "SIZE";
+  private static final String START_PARAMETER_NAME = "START";
+
+  @Override
+  public List<ParameterSpecification> getArgumentsSpecifications() {
+    return Arrays.asList(
+        TableParameterSpecification.builder()
+            .name(DATA_PARAMETER_NAME)
+            .passThroughColumns()
+            .keepWhenEmpty()
+            .build(),
+        ScalarParameterSpecification.builder()
+            .name(TIMECOL_PARAMETER_NAME)
+            .type(Type.STRING)
+            .build(),
+        ScalarParameterSpecification.builder().name(SLIDE_PARAMETER_NAME).type(Type.INT64).build(),
+        ScalarParameterSpecification.builder().name(SIZE_PARAMETER_NAME).type(Type.INT64).build(),
+        ScalarParameterSpecification.builder()
+            .name(START_PARAMETER_NAME)
+            .type(Type.TIMESTAMP)
+            .defaultValue(Long.MIN_VALUE)
+            .build());
+  }
+
+  private int findTimeColumnIndex(TableArgument tableArgument, String expectedFieldName) {
+    int requiredIndex = -1;
+    for (int i = 0; i < tableArgument.getFieldTypes().size(); i++) {
+      Optional<String> fieldName = tableArgument.getFieldNames().get(i);
+      if (fieldName.isPresent() && expectedFieldName.equalsIgnoreCase(fieldName.get())) {
+        requiredIndex = i;
+        break;
+      }
+    }
+    return requiredIndex;
+  }
+
+  @Override
+  public TableFunctionAnalysis analyze(Map<String, Argument> arguments) {
+    TableArgument tableArgument = (TableArgument) arguments.get(DATA_PARAMETER_NAME);
+    String expectedFieldName =
+        (String) ((ScalarArgument) arguments.get(TIMECOL_PARAMETER_NAME)).getValue();
+    int requiredIndex = findTimeColumnIndex(tableArgument, expectedFieldName);
+    if (requiredIndex == -1) {
+      throw new UDFException("The required field is not found in the input table");
+    }
+    DescribedSchema properColumnSchema =
+        new DescribedSchema.Builder()
+            .addField("window_start", Type.TIMESTAMP)
+            .addField("window_end", Type.TIMESTAMP)
+            .build();
+
+    // outputColumnSchema
+    return TableFunctionAnalysis.builder()
+        .properColumnSchema(properColumnSchema)
+        .requiredColumns(DATA_PARAMETER_NAME, Collections.singletonList(requiredIndex))
+        .build();
+  }
+
+  @Override
+  public TableFunctionProcessorProvider getProcessorProvider(Map<String, Argument> arguments) {
+    return new TableFunctionProcessorProvider() {
+      @Override
+      public TableFunctionDataProcessor getDataProcessor() {
+        return new HOPDataProcessor(
+            (Long) ((ScalarArgument) arguments.get(START_PARAMETER_NAME)).getValue(),
+            (Long) ((ScalarArgument) arguments.get(SLIDE_PARAMETER_NAME)).getValue(),
+            (Long) ((ScalarArgument) arguments.get(SIZE_PARAMETER_NAME)).getValue());
+      }
+    };
+  }
+
+  private static class HOPDataProcessor implements TableFunctionDataProcessor {
+
+    private final long slide;
+    private final long size;
+    private long curTime;
+    private long curIndex = 0;
+
+    public HOPDataProcessor(long startTime, long slide, long size) {
+      this.slide = slide;
+      this.size = size;
+      this.curTime = startTime;
+    }
+
+    @Override
+    public void process(
+        Record input,
+        List<ColumnBuilder> properColumnBuilders,
+        ColumnBuilder passThroughIndexBuilder) {
+      long timeValue = input.getLong(0);
+      if (curTime == Long.MIN_VALUE) {
+        curTime = timeValue;
+      }
+      if (curTime + size <= timeValue) {
+        // jump to appropriate window
+        long move = (timeValue - curTime - size) / slide + 1;
+        curTime += move * slide;
+      }
+      long slideTime = curTime;
+      while (slideTime <= timeValue && slideTime + size > timeValue) {
+        properColumnBuilders.get(0).writeLong(slideTime);
+        properColumnBuilders.get(1).writeLong(slideTime + size);
+        passThroughIndexBuilder.writeLong(curIndex);
+        slideTime += slide;
+      }
+      curIndex++;
+    }
+  }
+}

example/udf/src/main/java/org/apache/iotdb/udf/table/RepeatExample.java

@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.udf.table;
+
+import org.apache.iotdb.udf.api.exception.UDFArgumentNotValidException;
+import org.apache.iotdb.udf.api.exception.UDFException;
+import org.apache.iotdb.udf.api.relational.TableFunction;
+import org.apache.iotdb.udf.api.relational.access.Record;
+import org.apache.iotdb.udf.api.relational.table.TableFunctionAnalysis;
+import org.apache.iotdb.udf.api.relational.table.TableFunctionProcessorProvider;
+import org.apache.iotdb.udf.api.relational.table.argument.Argument;
+import org.apache.iotdb.udf.api.relational.table.argument.DescribedSchema;
+import org.apache.iotdb.udf.api.relational.table.argument.ScalarArgument;
+import org.apache.iotdb.udf.api.relational.table.processor.TableFunctionDataProcessor;
+import org.apache.iotdb.udf.api.relational.table.specification.ParameterSpecification;
+import org.apache.iotdb.udf.api.relational.table.specification.ScalarParameterSpecification;
+import org.apache.iotdb.udf.api.relational.table.specification.TableParameterSpecification;
+import org.apache.iotdb.udf.api.type.Type;
+
+import org.apache.tsfile.block.column.ColumnBuilder;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * This is an internal example of the TableFunction implementation. This function is declared as set
+ * semantic with pass through columns.
+ *
+ * <p>CREATE DATABASE test;
+ *
+ * <p>USE test;
+ *
+ * <p>CREATE TABLE t1(device_id STRING TAG, s1 TEXT FIELD, s2 INT32 FIELD);
+ *
+ * <p>INSERT INTO t1(time, device_id, s1, s2) VALUES (1, 'd1', 'a', 1), (2, 'd1', null, 2), (3,
+ * 'd1', 'c', null);
+ *
+ * <p>CREATE FUNCTION repeat AS 'org.apache.iotdb.udf.table.RepeatExample';
+ *
+ * <p>SHOW FUNCTIONS;
+ *
+ * <p>SELECT * FROM TABLE(repeat(TABLE(t1), 2));
+ */
+public class RepeatExample implements TableFunction {
+  private final String TBL_PARAM = "DATA";
+  private final String N_PARAM = "N";
+
+  @Override
+  public List<ParameterSpecification> getArgumentsSpecifications() {
+    return Arrays.asList(
+        TableParameterSpecification.builder().name(TBL_PARAM).passThroughColumns().build(),
+        ScalarParameterSpecification.builder().name(N_PARAM).type(Type.INT32).build());
+  }
+
+  @Override
+  public TableFunctionAnalysis analyze(Map<String, Argument> arguments) throws UDFException {
+
+    ScalarArgument count = (ScalarArgument) arguments.get("N");
+    if (count == null) {
+      throw new UDFArgumentNotValidException("count argument for function repeat() is missing");
+    } else if ((int) count.getValue() <= 0) {
+      throw new UDFArgumentNotValidException(
+          "count argument for function repeat() must be positive");
+    }
+    return TableFunctionAnalysis.builder()
+        .properColumnSchema(DescribedSchema.builder().addField("repeat_index", Type.INT32).build())
+        .requiredColumns(
+            TBL_PARAM,
+            Collections.singletonList(0)) // per spec, function must require at least one column
+        .build();
+  }
+
+  @Override
+  public TableFunctionProcessorProvider getProcessorProvider(Map<String, Argument> arguments) {
+    ScalarArgument count = (ScalarArgument) arguments.get("N");
+    return new TableFunctionProcessorProvider() {
+      @Override
+      public TableFunctionDataProcessor getDataProcessor() {
+        return new TableFunctionDataProcessor() {
+          private final int n = (int) count.getValue();
+          private long recordIndex = 0;
+
+          @Override
+          public void process(
+              Record input,
+              List<ColumnBuilder> properColumnBuilders,
+              ColumnBuilder passThroughIndexBuilder) {
+            properColumnBuilders.get(0).writeInt(0);
+            passThroughIndexBuilder.writeLong(recordIndex++);
+          }
+
+          @Override
+          public void finish(
+              List<ColumnBuilder> columnBuilders, ColumnBuilder passThroughIndexBuilder) {
+            for (int i = 1; i < n; i++) {
+              for (int j = 0; j < recordIndex; j++) {
+                columnBuilders.get(0).writeInt(i);
+                passThroughIndexBuilder.writeLong(j);
+              }
+            }
+          }
+        };
+      }
+    };
+  }
+}