[FLINK] Support ORC filesystem sink format

.github/workflows/flink.yml

@@ -70,7 +70,7 @@ jobs:
           export fmt_SOURCE=BUNDLED
           export folly_SOURCE=BUNDLED
           git clone -b gluten-0530 https://github.com/bigo-sg/velox4j.git
-          cd velox4j && git reset --hard 115edf79d265a61c30d45dfcc6ce932ad92378ca
+          cd velox4j && git reset --hard f6ea2d7f9c79a3476827dd7fd4c16a2b67a17cc3
           git apply $GITHUB_WORKSPACE/gluten-flink/patches/fix-velox4j.patch
           $GITHUB_WORKSPACE/build/mvn clean install -DskipTests -Dgpg.skip -Dspotless.skip=true
           cd ..

gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenCloseables.java

@@ -0,0 +1,52 @@
+/*
+ * 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.gluten.table.runtime.operators;
+
+final class GlutenCloseables {
+  private GlutenCloseables() {}
+
+  static void runWithCleanup(ThrowingRunnable action, ThrowingRunnable... cleanups)
+      throws Exception {
+    Exception failure = null;
+    try {
+      action.run();
+    } catch (Exception e) {
+      failure = e;
+    }
+
+    for (ThrowingRunnable cleanup : cleanups) {
+      try {
+        cleanup.run();
+      } catch (Exception e) {
+        if (failure == null) {
+          failure = e;
+        } else {
+          failure.addSuppressed(e);
+        }
+      }
+    }
+
+    if (failure != null) {
+      throw failure;
+    }
+  }
+
+  @FunctionalInterface
+  interface ThrowingRunnable {
+    void run() throws Exception;
+  }
+}

gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperator.java

@@ -39,6 +39,7 @@
 import org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend;
 import org.apache.flink.runtime.state.StateInitializationContext;
 import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
 import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
 import org.apache.flink.streaming.api.watermark.Watermark;
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -49,7 +50,7 @@
 
 /** Calculate operator in gluten, which will call Velox to run. */
 public class GlutenOneInputOperator<IN, OUT> extends TableStreamOperator<OUT>
-    implements OneInputStreamOperator<IN, OUT>, GlutenOperator {
+    implements OneInputStreamOperator<IN, OUT>, BoundedOneInput, GlutenOperator {
 
   private final StatefulPlanNode glutenPlan;
   private final String id;
@@ -195,24 +196,46 @@ private void drainTaskOutput() {
     while (true) {
       UpIterator.State state = task.advance();
       if (state == UpIterator.State.AVAILABLE) {
-        final StatefulElement statefulElement = task.statefulGet();
-        try {
-          if (statefulElement.isWatermark()) {
-            StatefulWatermark watermark = statefulElement.asWatermark();
-            output.emitWatermark(new Watermark(watermark.getTimestamp()));
-          } else {
-            outputBridge.collect(
-                output, statefulElement.asRecord(), sessionResource.getAllocator(), outputType);
-          }
-        } finally {
-          statefulElement.close();
-        }
+        processAvailableElement();
       } else {
         break;
       }
     }
   }
 
+  private void processAvailableElement() {
+    final StatefulElement statefulElement = task.statefulGet();
+    try {
+      if (statefulElement.isWatermark()) {
+        StatefulWatermark watermark = statefulElement.asWatermark();
+        output.emitWatermark(new Watermark(watermark.getTimestamp()));
+      } else {
+        outputBridge.collect(
+            output, statefulElement.asRecord(), sessionResource.getAllocator(), outputType);
+      }
+    } finally {
+      statefulElement.close();
+    }
+  }
+
+  private void finishTask() {
+    while (true) {
+      UpIterator.State state = task.advance();
+      switch (state) {
+        case AVAILABLE:
+          processAvailableElement();
+          break;
+        case BLOCKED:
+          task.waitFor();
+          break;
+        case FINISHED:
+          return;
+        default:
+          throw new IllegalStateException("Unknown Velox task state: " + state);
+      }
+    }
+  }
+
   public <NIN, NOUT> GlutenOneInputOperator<NIN, NOUT> cloneWithInputOutputClasses(
       StatefulPlanNode plan, Class<NIN> newInClass, Class<NOUT> newOutClass) {
     return new GlutenOneInputOperator<>(
@@ -237,19 +260,37 @@ public void processWatermark2(Watermark mark) throws Exception {
   }
 
   @Override
-  public void close() throws Exception {
-    if (task != null) {
-      task.close();
-    }
+  public void endInput() throws Exception {
     if (inputQueue != null) {
       inputQueue.noMoreInput();
-      inputQueue.close();
     }
-    if (sessionResource != null) {
-      sessionResource.close();
+    if (task != null) {
+      finishTask();
     }
   }
 
+  @Override
+  public void close() throws Exception {
+    GlutenCloseables.runWithCleanup(
+        () -> {},
+        () -> {
+          if (inputQueue != null) {
+            inputQueue.close();
+          }
+        },
+        () -> {
+          if (task != null) {
+            task.close();
+          }
+        },
+        () -> {
+          if (sessionResource != null) {
+            sessionResource.close();
+          }
+        },
+        super::close);
+  }
+
   @Override
   public StatefulPlanNode getPlanNode() {
     return glutenPlan;

gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperator.java

@@ -36,6 +36,7 @@
 import org.apache.flink.runtime.state.StateInitializationContext;
 import org.apache.flink.runtime.state.StateSnapshotContext;
 import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedMultiInput;
 import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
 import org.apache.flink.streaming.api.watermark.Watermark;
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -50,7 +51,7 @@
  * instead of flink RowData.
  */
 public class GlutenTwoInputOperator<IN, OUT> extends AbstractStreamOperator<OUT>
-    implements TwoInputStreamOperator<IN, IN, OUT>, GlutenOperator {
+    implements TwoInputStreamOperator<IN, IN, OUT>, BoundedMultiInput, GlutenOperator {
 
   private static final Logger LOG = LoggerFactory.getLogger(GlutenTwoInputOperator.class);
 
@@ -73,6 +74,8 @@ public class GlutenTwoInputOperator<IN, OUT> extends AbstractStreamOperator<OUT>
   private VectorOutputBridge<OUT> outputBridge;
   private String description;
   private final GlutenMailboxHolder mailboxHolder = new GlutenMailboxHolder();
+  private boolean leftInputEnded;
+  private boolean rightInputEnded;
 
   public GlutenTwoInputOperator(
       StatefulPlanNode plan,
@@ -175,24 +178,46 @@ private void drainTaskOutput() {
     while (true) {
       UpIterator.State state = task.advance();
       if (state == UpIterator.State.AVAILABLE) {
-        final StatefulElement element = task.statefulGet();
-        try {
-          if (element.isWatermark()) {
-            StatefulWatermark watermark = element.asWatermark();
-            output.emitWatermark(new Watermark(watermark.getTimestamp()));
-          } else {
-            outputBridge.collect(
-                output, element.asRecord(), sessionResource.getAllocator(), outputType);
-          }
-        } finally {
-          element.close();
-        }
+        processAvailableElement();
       } else {
         break;
       }
     }
   }
 
+  private void processAvailableElement() {
+    final StatefulElement element = task.statefulGet();
+    try {
+      if (element.isWatermark()) {
+        StatefulWatermark watermark = element.asWatermark();
+        output.emitWatermark(new Watermark(watermark.getTimestamp()));
+      } else {
+        outputBridge.collect(
+            output, element.asRecord(), sessionResource.getAllocator(), outputType);
+      }
+    } finally {
+      element.close();
+    }
+  }
+
+  private void finishTask() {
+    while (true) {
+      UpIterator.State state = task.advance();
+      switch (state) {
+        case AVAILABLE:
+          processAvailableElement();
+          break;
+        case BLOCKED:
+          task.waitFor();
+          break;
+        case FINISHED:
+          return;
+        default:
+          throw new IllegalStateException("Unknown Velox task state: " + state);
+      }
+    }
+  }
+
   @Override
   public void processWatermark(Watermark mark) throws Exception {
     task.notifyWatermark(mark.getTimestamp());
@@ -212,21 +237,55 @@ public void processWatermark2(Watermark mark) throws Exception {
   }
 
   @Override
-  public void close() throws Exception {
-    if (leftInputQueue != null) {
-      leftInputQueue.close();
-    }
-    if (rightInputQueue != null) {
-      rightInputQueue.close();
-    }
-    if (task != null) {
-      task.close();
+  public void endInput(int inputId) throws Exception {
+    switch (inputId) {
+      case 1:
+        leftInputEnded = true;
+        if (leftInputQueue != null) {
+          leftInputQueue.noMoreInput();
+        }
+        break;
+      case 2:
+        rightInputEnded = true;
+        if (rightInputQueue != null) {
+          rightInputQueue.noMoreInput();
+        }
+        break;
+      default:
+        throw new IllegalArgumentException("Unknown input id: " + inputId);
     }
-    if (sessionResource != null) {
-      sessionResource.close();
+    if (leftInputEnded && rightInputEnded && task != null) {
+      finishTask();
     }
   }
 
+  @Override
+  public void close() throws Exception {
+    GlutenCloseables.runWithCleanup(
+        () -> {},
+        () -> {
+          if (leftInputQueue != null) {
+            leftInputQueue.close();
+          }
+        },
+        () -> {
+          if (rightInputQueue != null) {
+            rightInputQueue.close();
+          }
+        },
+        () -> {
+          if (task != null) {
+            task.close();
+          }
+        },
+        () -> {
+          if (sessionResource != null) {
+            sessionResource.close();
+          }
+        },
+        super::close);
+  }
+
   @Override
   public StatefulPlanNode getPlanNode() {
     return glutenPlan;

gluten-flink/ut/pom.xml

@@ -194,6 +194,36 @@
       <version>${flink.version}</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-orc</artifactId>
+      <version>${flink.version}</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.orc</groupId>
+      <artifactId>orc-core</artifactId>
+      <version>1.9.2</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.hive</groupId>
+      <artifactId>hive-storage-api</artifactId>
+      <version>2.8.1</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.google.protobuf</groupId>
+      <artifactId>protobuf-java</artifactId>
+      <version>3.25.5</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-client</artifactId>
+      <version>2.7.4</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>