From 0a83c1f822ba819c962af88cee3b59da0a2095c7 Mon Sep 17 00:00:00 2001 From: zhangzhibiao Date: Mon, 6 Jul 2026 16:48:42 +0800 Subject: [PATCH 1/6] Rebase onto upstream/main: ORC sink, finishTask in endInput, close tests - Support ORC filesystem sink format with q10_orc.sql test - Add finishTask() to endInput() for both OneInput and TwoInput operators - Route finishTask through drainOutput for mailbox reentrancy guard - Treat unknown UpIterator.State as terminal (log+return, not throw) - Close() only releases resources, does not drain output - Add close tests for both operators - Use protobuf.version and hadoop.version properties in UT pom - Use dedicated output dir and process trigger for q10_orc - Clean ORC output dir before q10_orc for deterministic verification - Align Kafka port with KafkaToHiveE2ETest (19092) - Update velox4j commit to f8684c84 (includes EmptyOperator NYI fix) - Remove GlutenWatermarkStatuses (not supported by current velox4j) --- .github/workflows/flink.yml | 4 +- .../nodes/exec/stream/StreamExecJoin.java | 35 +--- .../exec/stream/StreamExecWindowJoin.java | 6 +- gluten-flink/pom.xml | 2 + .../client/OffloadedJobGraphGenerator.java | 11 +- .../operators/GlutenOneInputOperator.java | 89 +++++--- .../operators/GlutenTwoInputOperator.java | 176 ++++++++-------- .../operators/GlutenWatermarkStatuses.java | 30 --- .../java/org/apache/gluten/util/Utils.java | 3 - gluten-flink/ut/pom.xml | 24 ++- .../GlutenOneInputOperatorCloseTest.java | 140 +++++++++++++ .../GlutenTwoInputOperatorCloseTest.java | 190 ++++++++++++++++++ .../runtime/stream/custom/NexmarkTest.java | 71 ++++++- .../ut/src/test/resources/nexmark/q10_orc.sql | 23 +++ 14 files changed, 602 insertions(+), 202 deletions(-) delete mode 100644 gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenWatermarkStatuses.java create mode 100644 gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperatorCloseTest.java create mode 100644 gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperatorCloseTest.java create mode 100644 gluten-flink/ut/src/test/resources/nexmark/q10_orc.sql diff --git a/.github/workflows/flink.yml b/.github/workflows/flink.yml index b66bddf4279..821b3377410 100644 --- a/.github/workflows/flink.yml +++ b/.github/workflows/flink.yml @@ -87,8 +87,8 @@ jobs: export VELOX_DEPENDENCY_SOURCE=BUNDLED export fmt_SOURCE=BUNDLED export folly_SOURCE=BUNDLED - git clone -b feature/watermark-status-propagation https://github.com/bigo-sg/velox4j.git - cd velox4j && git reset --hard 6e2046f + git clone -b gluten-0530 https://github.com/bigo-sg/velox4j.git + cd velox4j && git reset --hard f8684c84ac68729f1370be1ee6a80a0fe4738818 git apply $GITHUB_WORKSPACE/gluten-flink/patches/fix-velox4j.patch $GITHUB_WORKSPACE/build/mvn clean install -DskipTests -Dgpg.skip -Dspotless.skip=true cd .. diff --git a/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecJoin.java b/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecJoin.java index 7a638e71faf..cf66cbbee4b 100644 --- a/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecJoin.java +++ b/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecJoin.java @@ -19,7 +19,6 @@ import org.apache.gluten.rexnode.RexConversionContext; import org.apache.gluten.rexnode.RexNodeConverter; import org.apache.gluten.rexnode.Utils; -import org.apache.gluten.streaming.api.operators.GlutenTwoInputOperatorFactory; import org.apache.gluten.table.runtime.operators.GlutenTwoInputOperator; import org.apache.gluten.util.LogicalTypeConverter; import org.apache.gluten.util.PlanNodeIdGenerator; @@ -41,7 +40,6 @@ import org.apache.flink.FlinkVersion; import org.apache.flink.api.dag.Transformation; import org.apache.flink.configuration.ReadableConfig; -import org.apache.flink.streaming.api.operators.StreamOperator; import org.apache.flink.streaming.api.operators.TwoInputStreamOperator; import org.apache.flink.streaming.api.transformations.TwoInputTransformation; import org.apache.flink.table.data.RowData; @@ -325,30 +323,15 @@ protected Transformation translateToPlanInternal( } final RowType returnType = (RowType) getOutputType(); - final TwoInputTransformation transform; - if (operator instanceof GlutenTwoInputOperator) { - transform = - ExecNodeUtil.createTwoInputTransformation( - leftTransform, - rightTransform, - createTransformationMeta(JOIN_TRANSFORMATION, config), - new GlutenTwoInputOperatorFactory( - (StreamOperator) operator), - InternalTypeInfo.of(returnType), - leftTransform.getParallelism(), - 0L, - false); - } else { - transform = - ExecNodeUtil.createTwoInputTransformation( - leftTransform, - rightTransform, - createTransformationMeta(JOIN_TRANSFORMATION, config), - operator, - InternalTypeInfo.of(returnType), - leftTransform.getParallelism(), - false); - } + final TwoInputTransformation transform = + ExecNodeUtil.createTwoInputTransformation( + leftTransform, + rightTransform, + createTransformationMeta(JOIN_TRANSFORMATION, config), + operator, + InternalTypeInfo.of(returnType), + leftTransform.getParallelism(), + false); // set KeyType and Selector for state RowDataKeySelector leftSelect = diff --git a/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecWindowJoin.java b/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecWindowJoin.java index 5a947f1ef54..400a86ccb05 100644 --- a/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecWindowJoin.java +++ b/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecWindowJoin.java @@ -17,7 +17,6 @@ package org.apache.flink.table.planner.plan.nodes.exec.stream; import org.apache.gluten.rexnode.Utils; -import org.apache.gluten.streaming.api.operators.GlutenTwoInputOperatorFactory; import org.apache.gluten.table.runtime.operators.GlutenTwoInputOperator; import org.apache.gluten.util.LogicalTypeConverter; import org.apache.gluten.util.PlanNodeIdGenerator; @@ -37,7 +36,6 @@ import org.apache.flink.FlinkVersion; import org.apache.flink.api.dag.Transformation; import org.apache.flink.configuration.ReadableConfig; -import org.apache.flink.streaming.api.operators.StreamOperator; import org.apache.flink.streaming.api.operators.TwoInputStreamOperator; import org.apache.flink.streaming.api.transformations.TwoInputTransformation; import org.apache.flink.table.api.TableException; @@ -247,11 +245,9 @@ protected Transformation translateToPlanInternal( leftTransform, rightTransform, createTransformationMeta(WINDOW_JOIN_TRANSFORMATION, config), - new GlutenTwoInputOperatorFactory( - (StreamOperator) operator), + operator, InternalTypeInfo.of(returnType), leftTransform.getParallelism(), - 0L, false); // set KeyType and Selector for state diff --git a/gluten-flink/pom.xml b/gluten-flink/pom.xml index 3f436cc6c1c..983a9e71dd7 100644 --- a/gluten-flink/pom.xml +++ b/gluten-flink/pom.xml @@ -31,6 +31,8 @@ 1.19.2 0.1.0-SNAPSHOT + 3.25.5 + 2.7.2 2.12 11 11 diff --git a/gluten-flink/runtime/src/main/java/org/apache/gluten/client/OffloadedJobGraphGenerator.java b/gluten-flink/runtime/src/main/java/org/apache/gluten/client/OffloadedJobGraphGenerator.java index a9fa29557dd..27dc0b1933c 100644 --- a/gluten-flink/runtime/src/main/java/org/apache/gluten/client/OffloadedJobGraphGenerator.java +++ b/gluten-flink/runtime/src/main/java/org/apache/gluten/client/OffloadedJobGraphGenerator.java @@ -16,10 +16,8 @@ */ package org.apache.gluten.client; -import org.apache.gluten.streaming.api.operators.GlutenOneInputOperatorFactory; import org.apache.gluten.streaming.api.operators.GlutenOperator; import org.apache.gluten.streaming.api.operators.GlutenStreamSource; -import org.apache.gluten.streaming.api.operators.GlutenTwoInputOperatorFactory; import org.apache.gluten.table.runtime.keyselector.GlutenKeySelector; import org.apache.gluten.table.runtime.operators.GlutenOneInputOperator; import org.apache.gluten.table.runtime.operators.GlutenSourceFunction; @@ -243,9 +241,7 @@ private void createOffloadedOneInputOperator( Class outClass = supportsVectorOutput ? StatefulRecord.class : RowData.class; GlutenOneInputOperator newOneInputOp = sourceOperator.cloneWithInputOutputClasses(planNode, inClass, outClass); - // setStreamOperator would wrap this in Flink's SimpleOperatorFactory and skip Gluten-specific - // mailbox binding performed by GlutenOneInputOperatorFactory. - offloadedOpConfig.setStreamOperatorFactory(new GlutenOneInputOperatorFactory<>(newOneInputOp)); + offloadedOpConfig.setStreamOperator(newOneInputOp); if (supportsVectorOutput) { setOffloadedOutputSerializer(offloadedOpConfig, sourceOperator); } @@ -283,10 +279,7 @@ private void createOffloadedTwoInputOperator( sourceOperator.getOutputTypes(), inClass, outClass); - // setStreamOperator would wrap this in Flink's SimpleOperatorFactory, which only initializes - // ProcessingTimeService for Flink AbstractStreamOperator. GlutenTwoInputOperator uses - // GlutenAbstractStreamOperator so it needs the Gluten-specific factory. - offloadedOpConfig.setStreamOperatorFactory(new GlutenTwoInputOperatorFactory<>(newTwoInputOp)); + offloadedOpConfig.setStreamOperator(newTwoInputOp); offloadedOpConfig.setStatePartitioner(0, new GlutenKeySelector()); offloadedOpConfig.setStatePartitioner(1, new GlutenKeySelector()); if (supportsVectorOutput) { diff --git a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperator.java b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperator.java index 5729a60b41c..a824f96266f 100644 --- a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperator.java +++ b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperator.java @@ -40,10 +40,10 @@ 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; -import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus; import org.apache.flink.table.runtime.operators.TableStreamOperator; import java.util.List; @@ -51,7 +51,10 @@ /** Calculate operator in gluten, which will call Velox to run. */ public class GlutenOneInputOperator extends TableStreamOperator - implements OneInputStreamOperator, GlutenOperator, NativeCallbackTarget { + implements OneInputStreamOperator, + BoundedOneInput, + GlutenOperator, + NativeCallbackTarget { private final StatefulPlanNode glutenPlan; private final String id; @@ -219,27 +222,55 @@ private void drainTaskOutput() { } 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 if (statefulElement.isWatermarkStatus()) { - output.emitWatermarkStatus( - GlutenWatermarkStatuses.toFlinkWatermarkStatus(statefulElement)); - } 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(); + } + } + + // Drains remaining output from the native task after end-of-input. The + // BLOCKED case calls task.waitFor() which blocks the mailbox thread. This is + // safe as long as unblocking only depends on native-side I/O (e.g. flushing + // to filesystem) and does not require mailbox progress. If future async + // operators introduce mailbox-dependent BLOCKED, this will need to become + // non-blocking (break on BLOCKED and rely on native callback -> + // scheduleProcessElementOnMailbox to continue draining). + 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: + // Treat unknown states as terminal, consistent with GlutenSourceFunction. + LOG.warn("Unexpected Velox task state in finishTask: {}", state); + return; + } + } + } + public GlutenOneInputOperator cloneWithInputOutputClasses( StatefulPlanNode plan, Class newInClass, Class newOutClass) { return new GlutenOneInputOperator<>( @@ -253,12 +284,6 @@ public void processWatermark(Watermark mark) throws Exception { processElementInternal(); } - @Override - public void processWatermarkStatus(WatermarkStatus status) throws Exception { - task.notifyWatermarkStatus(status.isIdle()); - processElementInternal(); - } - @Override public void processWatermark1(Watermark mark) throws Exception { throw new UnsupportedOperationException("Not implemented for GlutenOneInputOperator"); @@ -269,6 +294,19 @@ public void processWatermark2(Watermark mark) throws Exception { throw new UnsupportedOperationException("Not implemented for GlutenOneInputOperator"); } + @Override + public void endInput() throws Exception { + if (inputQueue != null) { + inputQueue.noMoreInput(); + } + // Drain any final buffered output from the native task after signalling + // end-of-input. endInput is called head-to-tail, so downstream operators + // are still alive to receive the final records. + if (task != null) { + drainOutput(this::finishTask); + } + } + @Override public void close() throws Exception { closing = true; @@ -283,11 +321,6 @@ public void close() throws Exception { task.close(); } }, - () -> { - if (inputQueue != null) { - inputQueue.noMoreInput(); - } - }, () -> { if (inputQueue != null) { inputQueue.close(); diff --git a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperator.java b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperator.java index b51ececd9a0..8948578e556 100644 --- a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperator.java +++ b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperator.java @@ -16,7 +16,6 @@ */ package org.apache.gluten.table.runtime.operators; -import org.apache.gluten.streaming.api.operators.GlutenAbstractStreamOperator; import org.apache.gluten.streaming.api.operators.GlutenOperator; import org.apache.gluten.table.runtime.config.VeloxConnectorConfig; import org.apache.gluten.table.runtime.config.VeloxQueryConfig; @@ -37,10 +36,11 @@ 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; -import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -50,15 +50,12 @@ /** * Two input operator in gluten, which will call Velox to run. It receives RowVector from upstream * instead of flink RowData. - * - *

This class intentionally does not extend {@code AbstractStreamOperator}: in the Flink version - * used here, {@code AbstractStreamOperator.processWatermarkStatus1/2(...)} are final and maintain - * Flink's Java-side combined watermark status. Gluten needs to forward each input's status to - * native, which is the source of truth for combined watermark status inside the Gluten operator - * chain. */ -public class GlutenTwoInputOperator extends GlutenAbstractStreamOperator - implements TwoInputStreamOperator, GlutenOperator, NativeCallbackTarget { +public class GlutenTwoInputOperator extends AbstractStreamOperator + implements TwoInputStreamOperator, + BoundedMultiInput, + GlutenOperator, + NativeCallbackTarget { private static final Logger LOG = LoggerFactory.getLogger(GlutenTwoInputOperator.class); @@ -82,7 +79,8 @@ public class GlutenTwoInputOperator extends GlutenAbstractStreamOperato private VectorOutputBridge outputBridge; private String description; private final GlutenMailboxHolder mailboxHolder = new GlutenMailboxHolder(); - private transient boolean stateInitialized; + private boolean leftInputEnded; + private boolean rightInputEnded; public GlutenTwoInputOperator( StatefulPlanNode plan, @@ -206,79 +204,96 @@ private void drainTaskOutput() { } 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 if (element.isWatermarkStatus()) { - output.emitWatermarkStatus(GlutenWatermarkStatuses.toFlinkWatermarkStatus(element)); - } 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(); + } + } + + // Drains remaining output from the native task after both inputs end. The + // BLOCKED case calls task.waitFor() which blocks the mailbox thread. This is + // safe as long as unblocking only depends on native-side I/O (e.g. flushing + // to filesystem) and does not require mailbox progress. If future async + // operators introduce mailbox-dependent BLOCKED, this will need to become + // non-blocking (break on BLOCKED and rely on native callback -> + // scheduleProcessElementOnMailbox to continue draining). + 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: + // Treat unknown states as terminal, consistent with GlutenSourceFunction. + LOG.warn("Unexpected Velox task state in finishTask: {}", state); + return; + } + } + } + @Override public void processWatermark(Watermark mark) throws Exception { - if (task == null) { - throw new NullPointerException( - "task is null in processWatermark; expected to be initialized in open() and " - + "cleared only after close()"); - } task.notifyWatermark(mark.getTimestamp()); processElementInternal(); } @Override public void processWatermark1(Watermark mark) throws Exception { - if (task == null) { - throw new NullPointerException( - "task is null in processWatermark1; expected to be initialized in open() and " - + "cleared only after close()"); - } task.notifyWatermark(mark.getTimestamp(), 0); processElementInternal(); } @Override public void processWatermark2(Watermark mark) throws Exception { - if (task == null) { - throw new NullPointerException( - "task is null in processWatermark2; expected to be initialized in open() and " - + "cleared only after close()"); - } task.notifyWatermark(mark.getTimestamp(), 1); processElementInternal(); } @Override - public void processWatermarkStatus1(WatermarkStatus status) throws Exception { - if (task == null) { - throw new NullPointerException( - "task is null in processWatermarkStatus1; expected to be initialized in open() and " - + "cleared only after 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); } - task.notifyWatermarkStatus(status.isIdle(), 0); - processElementInternal(); - } - - @Override - public void processWatermarkStatus2(WatermarkStatus status) throws Exception { - if (task == null) { - throw new NullPointerException( - "task is null in processWatermarkStatus2; expected to be initialized in open() and " - + "cleared only after close()"); + // When both inputs have signalled end-of-input, drain the native task to + // flush any remaining buffered output before close() releases resources. + if (leftInputEnded && rightInputEnded && task != null) { + drainOutput(this::finishTask); } - task.notifyWatermarkStatus(status.isIdle(), 1); - processElementInternal(); } @Override @@ -286,13 +301,13 @@ public void close() throws Exception { closing = true; GlutenCloseables.runWithCleanup( () -> { - if (leftInputQueue != null) { - leftInputQueue.noMoreInput(); + if (task != null) { + task.unbindNativeCallbackTarget(); } }, () -> { - if (rightInputQueue != null) { - rightInputQueue.noMoreInput(); + if (task != null) { + task.close(); } }, () -> { @@ -305,16 +320,6 @@ public void close() throws Exception { rightInputQueue.close(); } }, - () -> { - if (task != null) { - task.unbindNativeCallbackTarget(); - } - }, - () -> { - if (task != null) { - task.close(); - } - }, () -> { GlutenTaskSessionContext.unregisterSessionResource(getId()); }, @@ -360,37 +365,22 @@ public String getRightId() { @Override public void prepareSnapshotPreBarrier(long checkpointId) throws Exception { // TODO: notify velox - processElementInternal(); super.prepareSnapshotPreBarrier(checkpointId); } @Override public void snapshotState(StateSnapshotContext context) throws Exception { // TODO: implement it - snapshotNativeState(context.getCheckpointId()); + task.snapshotState(0); super.snapshotState(context); } @Override public void initializeState(StateInitializationContext context) throws Exception { - // TODO: implement it - initializeNativeState(); - super.initializeState(context); - } - - private void snapshotNativeState(long checkpointId) { - if (task != null) { - task.snapshotState(checkpointId); - } - } - - private void initializeNativeState() { - if (stateInitialized) { - return; - } initSession(); + // TODO: implement it task.initializeState(0, null); - stateInitialized = true; + super.initializeState(context); } private void initSession() { @@ -424,18 +414,14 @@ private void initSession() { @Override public void notifyCheckpointComplete(long checkpointId) throws Exception { // TODO: notify velox - if (task != null) { - task.notifyCheckpointComplete(checkpointId); - } + task.notifyCheckpointComplete(checkpointId); super.notifyCheckpointComplete(checkpointId); } @Override public void notifyCheckpointAborted(long checkpointId) throws Exception { // TODO: notify velox - if (task != null) { - task.notifyCheckpointAborted(checkpointId); - } + task.notifyCheckpointAborted(checkpointId); super.notifyCheckpointAborted(checkpointId); } } diff --git a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenWatermarkStatuses.java b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenWatermarkStatuses.java deleted file mode 100644 index b45def322fe..00000000000 --- a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenWatermarkStatuses.java +++ /dev/null @@ -1,30 +0,0 @@ -/* - * 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; - -import io.github.zhztheplayer.velox4j.stateful.StatefulElement; - -import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus; - -final class GlutenWatermarkStatuses { - - private GlutenWatermarkStatuses() {} - - static WatermarkStatus toFlinkWatermarkStatus(StatefulElement element) { - return element.asWatermarkStatus().isIdle() ? WatermarkStatus.IDLE : WatermarkStatus.ACTIVE; - } -} diff --git a/gluten-flink/runtime/src/main/java/org/apache/gluten/util/Utils.java b/gluten-flink/runtime/src/main/java/org/apache/gluten/util/Utils.java index 840abcbf3ff..eaa6a7d20ef 100644 --- a/gluten-flink/runtime/src/main/java/org/apache/gluten/util/Utils.java +++ b/gluten-flink/runtime/src/main/java/org/apache/gluten/util/Utils.java @@ -18,7 +18,6 @@ import org.apache.gluten.streaming.api.operators.GlutenOneInputOperatorFactory; import org.apache.gluten.streaming.api.operators.GlutenOperator; -import org.apache.gluten.streaming.api.operators.GlutenTwoInputOperatorFactory; import org.apache.flink.connector.file.table.stream.PartitionCommitInfo; import org.apache.flink.streaming.api.graph.StreamConfig; @@ -65,8 +64,6 @@ public static Optional getGlutenOperator( } } else if (operatorFactory instanceof GlutenOneInputOperatorFactory) { return Optional.of(((GlutenOneInputOperatorFactory) operatorFactory).getOperator()); - } else if (operatorFactory instanceof GlutenTwoInputOperatorFactory) { - return Optional.of(((GlutenTwoInputOperatorFactory) operatorFactory).getOperator()); } return Optional.empty(); } diff --git a/gluten-flink/ut/pom.xml b/gluten-flink/ut/pom.xml index c97c3838e12..7c61dc45b98 100644 --- a/gluten-flink/ut/pom.xml +++ b/gluten-flink/ut/pom.xml @@ -186,19 +186,19 @@ org.apache.hadoop hadoop-auth - 2.7.2 + ${hadoop.version} test org.apache.hadoop hadoop-common - 2.7.2 + ${hadoop.version} test org.apache.hadoop hadoop-hdfs - 2.7.2 + ${hadoop.version} test @@ -281,6 +281,24 @@ ${flink.version} test + + org.apache.flink + flink-orc + ${flink.version} + test + + + com.google.protobuf + protobuf-java + ${protobuf.version} + test + + + org.apache.hadoop + hadoop-client + ${hadoop.version} + test + diff --git a/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperatorCloseTest.java b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperatorCloseTest.java new file mode 100644 index 00000000000..2ced08c0e8d --- /dev/null +++ b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperatorCloseTest.java @@ -0,0 +1,140 @@ +/* + * 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; + +import io.github.zhztheplayer.velox4j.connector.ExternalStreams; +import io.github.zhztheplayer.velox4j.data.RowVector; +import io.github.zhztheplayer.velox4j.iterator.UpIterator; +import io.github.zhztheplayer.velox4j.plan.StatefulPlanNode; +import io.github.zhztheplayer.velox4j.query.SerialTask; +import io.github.zhztheplayer.velox4j.stateful.StatefulElement; +import io.github.zhztheplayer.velox4j.type.BigIntType; +import io.github.zhztheplayer.velox4j.type.RowType; + +import org.apache.flink.table.data.RowData; + +import org.junit.jupiter.api.Test; + +import java.lang.reflect.Field; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +public class GlutenOneInputOperatorCloseTest { + private static final RowType ROW_TYPE = new RowType(List.of("c0"), List.of(new BigIntType())); + + @Test + public void testEndInputFinishesTask() throws Exception { + List calls = new ArrayList<>(); + GlutenOneInputOperator operator = newOperator(); + FakeBlockingQueue inputQueue = new FakeBlockingQueue(1, calls); + FakeSerialTask task = new FakeSerialTask(calls); + setField(operator, "inputQueue", inputQueue); + setField(operator, "task", task); + + operator.endInput(); + + assertThat(calls).containsExactly("input.noMoreInput", "task.advance"); + } + + @Test + public void testCloseOnlyCleansResources() throws Exception { + List calls = new ArrayList<>(); + GlutenOneInputOperator operator = newOperator(); + FakeBlockingQueue inputQueue = new FakeBlockingQueue(1, calls); + FakeSerialTask task = new FakeSerialTask(calls); + setField(operator, "inputQueue", inputQueue); + setField(operator, "task", task); + + operator.close(); + + assertThat(calls).containsExactly("task.unbind", "task.close", "input.close"); + } + + private static GlutenOneInputOperator newOperator() { + StatefulPlanNode plan = new StatefulPlanNode("project", null); + return new GlutenOneInputOperator<>( + plan, "input", ROW_TYPE, Map.of("project", ROW_TYPE), RowData.class, RowData.class); + } + + private static void setField(Object target, String name, Object value) throws Exception { + Field field = GlutenOneInputOperator.class.getDeclaredField(name); + field.setAccessible(true); + field.set(target, value); + } + + private static class FakeBlockingQueue extends ExternalStreams.BlockingQueue { + private final List calls; + + private FakeBlockingQueue(long id, List calls) { + super(id); + this.calls = calls; + } + + @Override + public void put(RowVector vector) { + throw new UnsupportedOperationException(); + } + + @Override + public void noMoreInput() { + calls.add("input.noMoreInput"); + } + + @Override + public void close() { + calls.add("input.close"); + } + } + + private static class FakeSerialTask extends SerialTask { + private final List calls; + + private FakeSerialTask(List calls) { + super(null, 2); + this.calls = calls; + } + + @Override + public void unbindNativeCallbackTarget() { + calls.add("task.unbind"); + } + + @Override + public UpIterator.State advance() { + calls.add("task.advance"); + return UpIterator.State.FINISHED; + } + + @Override + public void waitFor() { + throw new UnsupportedOperationException(); + } + + @Override + public StatefulElement statefulGet() { + throw new UnsupportedOperationException(); + } + + @Override + public void close() { + calls.add("task.close"); + } + } +} diff --git a/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperatorCloseTest.java b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperatorCloseTest.java new file mode 100644 index 00000000000..e2f46048d24 --- /dev/null +++ b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperatorCloseTest.java @@ -0,0 +1,190 @@ +/* + * 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; + +import io.github.zhztheplayer.velox4j.connector.ExternalStreams; +import io.github.zhztheplayer.velox4j.data.RowVector; +import io.github.zhztheplayer.velox4j.iterator.UpIterator; +import io.github.zhztheplayer.velox4j.plan.StatefulPlanNode; +import io.github.zhztheplayer.velox4j.query.SerialTask; +import io.github.zhztheplayer.velox4j.stateful.StatefulElement; +import io.github.zhztheplayer.velox4j.type.BigIntType; +import io.github.zhztheplayer.velox4j.type.RowType; + +import org.apache.flink.table.data.RowData; + +import org.junit.jupiter.api.Test; + +import java.lang.reflect.Field; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +public class GlutenTwoInputOperatorCloseTest { + private static final RowType ROW_TYPE = new RowType(List.of("c0"), List.of(new BigIntType())); + + @Test + public void testEndInputFinishesAfterBothInputsEnd() throws Exception { + List calls = new ArrayList<>(); + GlutenTwoInputOperator operator = newOperator(); + FakeBlockingQueue leftQueue = new FakeBlockingQueue(1, "left", calls); + FakeBlockingQueue rightQueue = new FakeBlockingQueue(2, "right", calls); + FakeSerialTask task = new FakeSerialTask(calls, false); + setField(operator, "leftInputQueue", leftQueue); + setField(operator, "rightInputQueue", rightQueue); + setField(operator, "task", task); + + operator.endInput(1); + + assertThat(calls).containsExactly("left.noMoreInput"); + + operator.endInput(2); + + assertThat(calls).containsExactly("left.noMoreInput", "right.noMoreInput", "task.advance"); + } + + @Test + public void testCloseOnlyCleansResources() throws Exception { + List calls = new ArrayList<>(); + GlutenTwoInputOperator operator = newOperator(); + FakeBlockingQueue leftQueue = new FakeBlockingQueue(1, "left", calls); + FakeBlockingQueue rightQueue = new FakeBlockingQueue(2, "right", calls); + FakeSerialTask task = new FakeSerialTask(calls, false); + setField(operator, "leftInputQueue", leftQueue); + setField(operator, "rightInputQueue", rightQueue); + setField(operator, "task", task); + + operator.close(); + + assertThat(calls).containsExactly("task.unbind", "task.close", "left.close", "right.close"); + } + + @Test + public void testCloseCleansResourcesWhenCloseFails() throws Exception { + List calls = new ArrayList<>(); + GlutenTwoInputOperator operator = newOperator(); + FakeBlockingQueue leftQueue = new FakeBlockingQueue(1, "left", calls, true); + FakeBlockingQueue rightQueue = new FakeBlockingQueue(2, "right", calls); + FakeSerialTask task = new FakeSerialTask(calls, false); + setField(operator, "leftInputQueue", leftQueue); + setField(operator, "rightInputQueue", rightQueue); + setField(operator, "task", task); + + assertThatThrownBy(operator::close) + .isInstanceOf(RuntimeException.class) + .hasMessage("left.close"); + + assertThat(calls).containsExactly("task.unbind", "task.close", "left.close", "right.close"); + } + + private static GlutenTwoInputOperator newOperator() { + StatefulPlanNode plan = new StatefulPlanNode("join", null); + return new GlutenTwoInputOperator<>( + plan, + "left", + "right", + ROW_TYPE, + ROW_TYPE, + Map.of("join", ROW_TYPE), + RowData.class, + RowData.class); + } + + private static void setField(Object target, String name, Object value) throws Exception { + Field field = GlutenTwoInputOperator.class.getDeclaredField(name); + field.setAccessible(true); + field.set(target, value); + } + + private static class FakeBlockingQueue extends ExternalStreams.BlockingQueue { + private final String name; + private final List calls; + private final boolean failClose; + + private FakeBlockingQueue(long id, String name, List calls) { + this(id, name, calls, false); + } + + private FakeBlockingQueue(long id, String name, List calls, boolean failClose) { + super(id); + this.name = name; + this.calls = calls; + this.failClose = failClose; + } + + @Override + public void put(RowVector vector) { + throw new UnsupportedOperationException(); + } + + @Override + public void noMoreInput() { + calls.add(name + ".noMoreInput"); + } + + @Override + public void close() { + calls.add(name + ".close"); + if (failClose) { + throw new RuntimeException(name + ".close"); + } + } + } + + private static class FakeSerialTask extends SerialTask { + private final List calls; + private final boolean failAdvance; + + private FakeSerialTask(List calls, boolean failAdvance) { + super(null, 3); + this.calls = calls; + this.failAdvance = failAdvance; + } + + @Override + public UpIterator.State advance() { + calls.add("task.advance"); + if (failAdvance) { + throw new RuntimeException("finish"); + } + return UpIterator.State.FINISHED; + } + + @Override + public void waitFor() { + throw new UnsupportedOperationException(); + } + + @Override + public StatefulElement statefulGet() { + throw new UnsupportedOperationException(); + } + + @Override + public void unbindNativeCallbackTarget() { + calls.add("task.unbind"); + } + + @Override + public void close() { + calls.add("task.close"); + } + } +} diff --git a/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/NexmarkTest.java b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/NexmarkTest.java index 14b250480bc..1dcd6204769 100644 --- a/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/NexmarkTest.java +++ b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/NexmarkTest.java @@ -40,6 +40,7 @@ import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; +import java.nio.file.attribute.FileTime; import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; @@ -71,7 +72,7 @@ public class NexmarkTest { } }; - private static final int KAFKA_PORT = 19092; + private static final int KAFKA_PORT = Integer.getInteger("gluten.flink.ut.kafka.port", 19092); private static String topicName = "nexmark"; @RegisterExtension @@ -206,6 +207,7 @@ private static void clearEnvironment(StreamTableEnvironment tEnv) { String sql = String.format("drop table if exists %s", tableName); tEnv.executeSql(sql); } + tEnv.executeSql("drop table if exists nexmark_q10_orc"); for (String view : VIEWS) { String sql = String.format("drop view if exists %s", view); tEnv.executeSql(sql); @@ -221,6 +223,29 @@ private static void clearEnvironment(StreamTableEnvironment tEnv) { private void executeQuery(StreamTableEnvironment tEnv, String queryFileName, boolean kafkaSource) throws ExecutionException, InterruptedException, TimeoutException { String queryContent = readSqlFromFile(NEXMARK_RESOURCE_DIR + "/" + queryFileName); + // Clean the ORC output directory before running q10_orc to ensure deterministic verification. + if ("q10_orc.sql".equals(queryFileName)) { + Path orcOutputDir = Paths.get("/tmp/data/output/bid_orc"); + if (Files.exists(orcOutputDir)) { + try { + try (java.util.stream.Stream files = Files.walk(orcOutputDir)) { + files + .sorted(java.util.Comparator.reverseOrder()) + .forEach( + p -> { + try { + Files.deleteIfExists(p); + } catch (IOException e) { + throw new RuntimeException("Failed to delete " + p, e); + } + }); + } + } catch (IOException e) { + throw new RuntimeException("Failed to clean ORC output directory", e); + } + } + } + long queryStartMillis = System.currentTimeMillis(); String[] sqlStatements = queryContent.split(";"); assertThat(sqlStatements.length).isGreaterThanOrEqualTo(2); @@ -242,11 +267,48 @@ private void executeQuery(StreamTableEnvironment tEnv, String queryFileName, boo assertThat(checkJobRunningStatus(insertResult, 30000) == true); } else { waitForJobCompletion(insertResult, 30000); + // Allow filesystem sink to flush and commit partitions after job completion. + Thread.sleep(2000); + if ("q10_orc.sql".equals(queryFileName)) { + verifyQ10OrcOutput(queryStartMillis); + } } } assertTrue(sqlStatements[sqlStatements.length - 1].trim().isEmpty()); } + private void verifyQ10OrcOutput(long queryStartMillis) { + Path outputDir = Paths.get("/tmp/data/output/bid_orc"); + assertTrue("Q10 ORC output directory should exist", Files.exists(outputDir)); + try { + List outputFiles; + try (java.util.stream.Stream files = Files.walk(outputDir)) { + outputFiles = + files + .filter(Files::isRegularFile) + .filter(path -> isModifiedAfter(path, queryStartMillis)) + .filter(path -> !path.getFileName().toString().startsWith("_")) + .sorted() + .collect(Collectors.toList()); + } + assertThat(outputFiles).isNotEmpty(); + for (Path outputFile : outputFiles) { + assertThat(Files.size(outputFile)).isGreaterThan(0L); + } + } catch (IOException e) { + throw new RuntimeException("Failed to read back Q10 ORC output", e); + } + } + + private boolean isModifiedAfter(Path path, long queryStartMillis) { + try { + FileTime lastModifiedTime = Files.getLastModifiedTime(path); + return lastModifiedTime.toMillis() >= queryStartMillis - 1000L; + } catch (IOException e) { + throw new RuntimeException("Failed to inspect output file " + path, e); + } + } + private void waitForJobCompletion(TableResult result, long timeoutMs) throws InterruptedException, ExecutionException, TimeoutException { assertTrue(result.getJobClient().isPresent()); @@ -294,6 +356,13 @@ private List getQueries() { queryFiles.retainAll(selectedQueries); } + String singleQuery = System.getProperty("gluten.flink.ut.nexmark.query"); + if (singleQuery != null && !singleQuery.isEmpty()) { + return queryFiles.stream() + .filter(singleQuery::equals) + .sorted() + .collect(Collectors.toList()); + } return queryFiles.stream().sorted().collect(Collectors.toList()); } catch (URISyntaxException | IOException e) { diff --git a/gluten-flink/ut/src/test/resources/nexmark/q10_orc.sql b/gluten-flink/ut/src/test/resources/nexmark/q10_orc.sql new file mode 100644 index 00000000000..0ddf8bb3930 --- /dev/null +++ b/gluten-flink/ut/src/test/resources/nexmark/q10_orc.sql @@ -0,0 +1,23 @@ +CREATE TABLE nexmark_q10_orc ( + auction BIGINT, + bidder BIGINT, + price BIGINT, + `dateTime` TIMESTAMP(3), + extra VARCHAR, + dt STRING, + hm STRING +) PARTITIONED BY (dt, hm) WITH ( + 'connector' = 'filesystem', + 'path' = 'file:///tmp/data/output/bid_orc/', + 'format' = 'orc', + 'sink.partition-commit.trigger' = 'process', + 'sink.partition-commit.delay' = '0s', + 'sink.partition-commit.policy.kind' = 'success-file', + 'partition.time-extractor.timestamp-pattern' = '$dt $hm:00', + 'sink.rolling-policy.rollover-interval' = '1min', + 'sink.rolling-policy.check-interval' = '1min' +); + +INSERT INTO nexmark_q10_orc +SELECT auction, bidder, price, `dateTime`, extra, DATE_FORMAT(`dateTime`, 'yyyy-MM-dd'), DATE_FORMAT(`dateTime`, 'HH:mm') +FROM bid; From 11bca73824bb460adf82b62f96e63af93c7922cc Mon Sep 17 00:00:00 2001 From: zhangzhibiao Date: Fri, 10 Jul 2026 16:46:53 +0800 Subject: [PATCH 2/6] Fix q10 ORC partition commit trigger --- gluten-flink/ut/src/test/resources/nexmark/q10_orc.sql | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/gluten-flink/ut/src/test/resources/nexmark/q10_orc.sql b/gluten-flink/ut/src/test/resources/nexmark/q10_orc.sql index 0ddf8bb3930..cf89d3474db 100644 --- a/gluten-flink/ut/src/test/resources/nexmark/q10_orc.sql +++ b/gluten-flink/ut/src/test/resources/nexmark/q10_orc.sql @@ -10,7 +10,7 @@ CREATE TABLE nexmark_q10_orc ( 'connector' = 'filesystem', 'path' = 'file:///tmp/data/output/bid_orc/', 'format' = 'orc', - 'sink.partition-commit.trigger' = 'process', + 'sink.partition-commit.trigger' = 'process-time', 'sink.partition-commit.delay' = '0s', 'sink.partition-commit.policy.kind' = 'success-file', 'partition.time-extractor.timestamp-pattern' = '$dt $hm:00', From 649d55b5b9ee001463a611124b313cf2b6bb3417 Mon Sep 17 00:00:00 2001 From: zhangzhibiao Date: Fri, 10 Jul 2026 19:44:40 +0800 Subject: [PATCH 3/6] Fix Flink watermark status reactivation --- .../nodes/exec/stream/StreamExecJoin.java | 5 +- .../exec/stream/StreamExecWindowJoin.java | 5 +- .../operators/GlutenOneInputOperator.java | 14 ++++ .../operators/GlutenTwoInputOperator.java | 78 ++++++++++++++++++- 4 files changed, 95 insertions(+), 7 deletions(-) diff --git a/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecJoin.java b/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecJoin.java index cf66cbbee4b..01c086cedf8 100644 --- a/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecJoin.java +++ b/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecJoin.java @@ -19,6 +19,7 @@ import org.apache.gluten.rexnode.RexConversionContext; import org.apache.gluten.rexnode.RexNodeConverter; import org.apache.gluten.rexnode.Utils; +import org.apache.gluten.streaming.api.operators.GlutenTwoInputOperatorFactory; import org.apache.gluten.table.runtime.operators.GlutenTwoInputOperator; import org.apache.gluten.util.LogicalTypeConverter; import org.apache.gluten.util.PlanNodeIdGenerator; @@ -328,10 +329,10 @@ protected Transformation translateToPlanInternal( leftTransform, rightTransform, createTransformationMeta(JOIN_TRANSFORMATION, config), - operator, + new GlutenTwoInputOperatorFactory<>(operator), InternalTypeInfo.of(returnType), leftTransform.getParallelism(), - false); + 0L); // set KeyType and Selector for state RowDataKeySelector leftSelect = diff --git a/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecWindowJoin.java b/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecWindowJoin.java index 400a86ccb05..ff8698001c6 100644 --- a/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecWindowJoin.java +++ b/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecWindowJoin.java @@ -17,6 +17,7 @@ package org.apache.flink.table.planner.plan.nodes.exec.stream; import org.apache.gluten.rexnode.Utils; +import org.apache.gluten.streaming.api.operators.GlutenTwoInputOperatorFactory; import org.apache.gluten.table.runtime.operators.GlutenTwoInputOperator; import org.apache.gluten.util.LogicalTypeConverter; import org.apache.gluten.util.PlanNodeIdGenerator; @@ -245,10 +246,10 @@ protected Transformation translateToPlanInternal( leftTransform, rightTransform, createTransformationMeta(WINDOW_JOIN_TRANSFORMATION, config), - operator, + new GlutenTwoInputOperatorFactory<>(operator), InternalTypeInfo.of(returnType), leftTransform.getParallelism(), - false); + 0L); // set KeyType and Selector for state RowDataKeySelector leftSelect = diff --git a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperator.java b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperator.java index a824f96266f..3d948bb26ef 100644 --- a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperator.java +++ b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperator.java @@ -44,6 +44,7 @@ import org.apache.flink.streaming.api.operators.OneInputStreamOperator; import org.apache.flink.streaming.api.watermark.Watermark; import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus; import org.apache.flink.table.runtime.operators.TableStreamOperator; import java.util.List; @@ -73,6 +74,7 @@ public class GlutenOneInputOperator extends TableStreamOperator private transient VectorInputBridge inputBridge; private transient VectorOutputBridge outputBridge; private final GlutenMailboxHolder mailboxHolder = new GlutenMailboxHolder(); + private transient boolean inputIdle; public GlutenOneInputOperator( StatefulPlanNode plan, @@ -279,11 +281,23 @@ public GlutenOneInputOperator cloneWithInputOutputClasses @Override public void processWatermark(Watermark mark) throws Exception { + if (inputIdle) { + inputIdle = false; + output.emitWatermarkStatus(WatermarkStatus.ACTIVE); + } task.notifyWatermark(mark.getTimestamp()); // Process any pending elements to ensure watermark-triggered operations complete. processElementInternal(); } + @Override + public void processWatermarkStatus(WatermarkStatus status) throws Exception { + if (inputIdle != status.isIdle()) { + inputIdle = status.isIdle(); + output.emitWatermarkStatus(status); + } + } + @Override public void processWatermark1(Watermark mark) throws Exception { throw new UnsupportedOperationException("Not implemented for GlutenOneInputOperator"); diff --git a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperator.java b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperator.java index 8948578e556..b6ef4fa45be 100644 --- a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperator.java +++ b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperator.java @@ -16,6 +16,7 @@ */ package org.apache.gluten.table.runtime.operators; +import org.apache.gluten.streaming.api.operators.GlutenAbstractStreamOperator; import org.apache.gluten.streaming.api.operators.GlutenOperator; import org.apache.gluten.table.runtime.config.VeloxConnectorConfig; import org.apache.gluten.table.runtime.config.VeloxQueryConfig; @@ -36,11 +37,11 @@ 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; +import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -51,7 +52,7 @@ * Two input operator in gluten, which will call Velox to run. It receives RowVector from upstream * instead of flink RowData. */ -public class GlutenTwoInputOperator extends AbstractStreamOperator +public class GlutenTwoInputOperator extends GlutenAbstractStreamOperator implements TwoInputStreamOperator, BoundedMultiInput, GlutenOperator, @@ -81,6 +82,12 @@ public class GlutenTwoInputOperator extends AbstractStreamOperator private final GlutenMailboxHolder mailboxHolder = new GlutenMailboxHolder(); private boolean leftInputEnded; private boolean rightInputEnded; + private transient boolean leftInputIdle; + private transient boolean rightInputIdle; + private transient boolean outputIdle; + private transient long leftWatermark = Long.MIN_VALUE; + private transient long rightWatermark = Long.MIN_VALUE; + private transient long lastEmittedWatermark = Long.MIN_VALUE; public GlutenTwoInputOperator( StatefulPlanNode plan, @@ -126,6 +133,12 @@ public String getDescription() { @Override public void open() throws Exception { closing = false; + leftInputIdle = false; + rightInputIdle = false; + outputIdle = false; + leftWatermark = Long.MIN_VALUE; + rightWatermark = Long.MIN_VALUE; + lastEmittedWatermark = Long.MIN_VALUE; super.open(); if (!mailboxHolder().get().isMailboxBound()) { ensureMailboxInitialized(getContainingTask()); @@ -216,7 +229,7 @@ private void processAvailableElement() { try { if (element.isWatermark()) { StatefulWatermark watermark = element.asWatermark(); - output.emitWatermark(new Watermark(watermark.getTimestamp())); + emitWatermarkIfAdvanced(watermark.getTimestamp()); } else { outputBridge.collect( output, element.asRecord(), sessionResource.getAllocator(), outputType); @@ -253,6 +266,41 @@ private void finishTask() { } } + private void emitWatermarkIfAdvanced(long timestamp) { + if (timestamp > lastEmittedWatermark) { + lastEmittedWatermark = timestamp; + output.emitWatermark(new Watermark(timestamp)); + } + } + + private void updateInputStatus(int inputIndex, boolean idle) { + boolean wasIdle = outputIdle; + if (inputIndex == 0) { + leftInputIdle = idle; + } else { + rightInputIdle = idle; + } + outputIdle = leftInputIdle && rightInputIdle; + if (wasIdle != outputIdle) { + output.emitWatermarkStatus(outputIdle ? WatermarkStatus.IDLE : WatermarkStatus.ACTIVE); + } + } + + private boolean hasIdleInput() { + return leftInputIdle || rightInputIdle; + } + + private void emitCombinedWatermarkWithIdleInputs() { + if (leftInputIdle && rightInputIdle) { + return; + } + if (leftInputIdle) { + emitWatermarkIfAdvanced(rightWatermark); + } else if (rightInputIdle) { + emitWatermarkIfAdvanced(leftWatermark); + } + } + @Override public void processWatermark(Watermark mark) throws Exception { task.notifyWatermark(mark.getTimestamp()); @@ -261,14 +309,38 @@ public void processWatermark(Watermark mark) throws Exception { @Override public void processWatermark1(Watermark mark) throws Exception { + leftWatermark = mark.getTimestamp(); + if (leftInputIdle) { + updateInputStatus(0, false); + } task.notifyWatermark(mark.getTimestamp(), 0); processElementInternal(); + if (hasIdleInput()) { + emitCombinedWatermarkWithIdleInputs(); + } } @Override public void processWatermark2(Watermark mark) throws Exception { + rightWatermark = mark.getTimestamp(); + if (rightInputIdle) { + updateInputStatus(1, false); + } task.notifyWatermark(mark.getTimestamp(), 1); processElementInternal(); + if (hasIdleInput()) { + emitCombinedWatermarkWithIdleInputs(); + } + } + + @Override + public void processWatermarkStatus1(WatermarkStatus status) { + updateInputStatus(0, status.isIdle()); + } + + @Override + public void processWatermarkStatus2(WatermarkStatus status) { + updateInputStatus(1, status.isIdle()); } @Override From 6d53e12e8f6107b38e1d271fea03681f69fa2c0b Mon Sep 17 00:00:00 2001 From: zhangzhibiao Date: Fri, 10 Jul 2026 23:11:03 +0800 Subject: [PATCH 4/6] [FLINK] Fix bounded stateful task cleanup --- gluten-flink/patches/fix-velox4j.patch | 41 +++++++++++++++++ .../nodes/exec/stream/StreamExecExchange.java | 4 +- .../client/OffloadedJobGraphGenerator.java | 44 ++++++++++--------- .../GlutenKeyGroupStreamPartitioner.java | 31 ++++++++++--- .../FlinkRowToVLVectorConvertor.java | 17 ++++++- 5 files changed, 107 insertions(+), 30 deletions(-) diff --git a/gluten-flink/patches/fix-velox4j.patch b/gluten-flink/patches/fix-velox4j.patch index d6ba34ca0ad..b099138dbca 100644 --- a/gluten-flink/patches/fix-velox4j.patch +++ b/gluten-flink/patches/fix-velox4j.patch @@ -1,3 +1,16 @@ +diff --git a/src/main/cpp/CMakeLists.txt b/src/main/cpp/CMakeLists.txt +index 9ed0551..5ded913 100644 +--- a/src/main/cpp/CMakeLists.txt ++++ b/src/main/cpp/CMakeLists.txt +@@ -82,7 +82,7 @@ else() + GIT_REPOSITORY + ${VELOX_SOURCE_URL} + GIT_TAG +- c9efdf308a94ea1cb846e2440aeb975572497d87) ++ bfa91ab98d6b53fccf2c862cdc926832835db35c) + endif() + if(TARGET gflags::gflags_shared) + get_target_property(GFLAGS_SHARED_LIBRARY gflags::gflags_shared diff --git a/src/main/cpp/main/CMakeLists.txt b/src/main/cpp/main/CMakeLists.txt index ada06f8..7ed80bb 100644 --- a/src/main/cpp/main/CMakeLists.txt @@ -15,3 +28,31 @@ index ada06f8..7ed80bb 100644 message(\"Found all shared 3rd dependencies of velox4j: \${3rd_deps}\") foreach(dep IN LISTS 3rd_deps) get_filename_component(link_name \${dep} NAME) +diff --git a/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc b/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc +index 2357cf2..159e3d6 100644 +--- a/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc ++++ b/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc +@@ -81,15 +81,8 @@ StatefulSerialTask::StatefulSerialTask( + } + + StatefulSerialTask::~StatefulSerialTask() { +- if (task_ != nullptr && task_->isRunning()) { +- // TODO: add a method to finish the task and set state. +- task_->finish(); +- // FIXME: Calling .wait() may take no effect in single thread execution +- // mode. +- task_->requestCancel().wait(); +- } + if (task_ != nullptr) { +- task_->setNativeCallbackBridge(nullptr); ++ task_->close(); + } + task_.reset(); + } +diff --git a/src/main/cpp/velox-ref.txt b/src/main/cpp/velox-ref.txt +index c24a828..47d494e 100644 +--- a/src/main/cpp/velox-ref.txt ++++ b/src/main/cpp/velox-ref.txt +@@ -1 +1 @@ +-c9efdf308a94ea1cb846e2440aeb975572497d87 ++bfa91ab98d6b53fccf2c862cdc926832835db35c diff --git a/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecExchange.java b/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecExchange.java index 17e3966d52a..e9a35707c9e 100644 --- a/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecExchange.java +++ b/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecExchange.java @@ -143,7 +143,6 @@ protected Transformation translateToPlanInternal( // TODO: velox's parallelism need to be set here, as some nodes need it. // should set it when operator init. parallelism = inputTransform.getParallelism(); - keySelector = new GlutenKeySelector(); final ExecEdge inputEdge = getInputEdges().get(0); io.github.zhztheplayer.velox4j.type.RowType glutenInputType = (io.github.zhztheplayer.velox4j.type.RowType) @@ -181,7 +180,8 @@ protected Transformation translateToPlanInternal( parallelism, false); partitioner = - new GlutenKeyGroupStreamPartitioner(keySelector, DEFAULT_LOWER_BOUND_MAX_PARALLELISM); + new GlutenKeyGroupStreamPartitioner( + new GlutenKeySelector(), keySelector, DEFAULT_LOWER_BOUND_MAX_PARALLELISM); } else { parallelism = ExecutionConfig.PARALLELISM_DEFAULT; partitioner = diff --git a/gluten-flink/runtime/src/main/java/org/apache/gluten/client/OffloadedJobGraphGenerator.java b/gluten-flink/runtime/src/main/java/org/apache/gluten/client/OffloadedJobGraphGenerator.java index 27dc0b1933c..39077b6bf35 100644 --- a/gluten-flink/runtime/src/main/java/org/apache/gluten/client/OffloadedJobGraphGenerator.java +++ b/gluten-flink/runtime/src/main/java/org/apache/gluten/client/OffloadedJobGraphGenerator.java @@ -249,6 +249,8 @@ private void createOffloadedOneInputOperator( setOffloadedInputSerializer(offloadedOpConfig, sourceOperator); setOffloadedStatePartitioner( sourceOpConfig, offloadedOpConfig, 0, sourceOperator.getDescription()); + } else { + offloadedOpConfig.setupNetworkInputs(sourceOpConfig.getTypeSerializerIn1(userClassloader)); } } @@ -287,6 +289,10 @@ private void createOffloadedTwoInputOperator( } if (supportsVectorInput) { setOffloadedInputSerializersForTwoInputOperator(offloadedOpConfig, sourceOperator); + } else { + offloadedOpConfig.setupNetworkInputs( + sourceOpConfig.getTypeSerializerIn1(userClassloader), + sourceOpConfig.getTypeSerializerIn2(userClassloader)); } } @@ -479,6 +485,9 @@ private void updateStatePartitioners(StreamConfig sourceConfig, StreamConfig off private boolean supportsVectorOutput( OperatorChainSlice chainSlice, OperatorChainSliceGraph chainSliceGraph, JobVertex jobVertex) { + if (isStreamExecExchange(chainSlice)) { + return false; + } List downstreamSliceIds = chainSlice.getOutputs(); // If chainSlice has downstream in the operator chain, check these downstream @@ -491,11 +500,22 @@ private boolean supportsVectorOutput( } } + private boolean isStreamExecExchange(OperatorChainSlice chainSlice) { + for (StreamConfig opConfig : chainSlice.getOperatorConfigs()) { + Optional glutenOperator = Utils.getGlutenOperator(opConfig, userClassloader); + if (glutenOperator.isPresent() + && "StreamExecExchange".equals(glutenOperator.get().getDescription())) { + return true; + } + } + return false; + } + private boolean checkDownstreamSlicesInChain( OperatorChainSliceGraph chainSliceGraph, List downstreamSliceIds) { for (Integer downstreamSliceId : downstreamSliceIds) { OperatorChainSlice downstreamSlice = chainSliceGraph.getSlice(downstreamSliceId); - if (!downstreamSlice.isOffloadable()) { + if (!downstreamSlice.isOffloadable() || isStreamExecExchange(downstreamSlice)) { return false; } if (!downstreamSlice.getInputs().stream() @@ -514,16 +534,7 @@ private boolean checkDownstreamVerticesFromJobGraph(JobVertex jobVertex) { // conversion) return true; } - // Check if all downstream vertices' operators are gluten operators - for (JobVertex downstreamVertex : downstreamVertices) { - StreamConfig downstreamStreamConfig = new StreamConfig(downstreamVertex.getConfiguration()); - Optional glutenOperator = - Utils.getGlutenOperator(downstreamStreamConfig, userClassloader); - if (!glutenOperator.isPresent()) { - return false; - } - } - return true; + return false; } private boolean supportsVectorInput( @@ -562,16 +573,7 @@ private boolean checkUpstreamVerticesFromJobGraph(JobVertex jobVertex) { // If there is no upstream JobVertex, can input RowVector (no upstream needs conversion) return true; } - // Check if the last operator of all upstream vertices are gluten operators - for (JobVertex upstreamVertex : upstreamVertices) { - StreamConfig lastOperatorConfig = findLastOperatorInChain(upstreamVertex); - Optional glutenOperator = - Utils.getGlutenOperator(lastOperatorConfig, userClassloader); - if (!glutenOperator.isPresent()) { - return false; - } - } - return true; + return false; } /** Get all downstream JobVertices of a JobVertex. */ diff --git a/gluten-flink/runtime/src/main/java/org/apache/gluten/streaming/runtime/partitioner/GlutenKeyGroupStreamPartitioner.java b/gluten-flink/runtime/src/main/java/org/apache/gluten/streaming/runtime/partitioner/GlutenKeyGroupStreamPartitioner.java index 1a30c525f82..2d57c54fb16 100644 --- a/gluten-flink/runtime/src/main/java/org/apache/gluten/streaming/runtime/partitioner/GlutenKeyGroupStreamPartitioner.java +++ b/gluten-flink/runtime/src/main/java/org/apache/gluten/streaming/runtime/partitioner/GlutenKeyGroupStreamPartitioner.java @@ -26,6 +26,7 @@ import org.apache.flink.streaming.runtime.partitioner.ConfigurableStreamPartitioner; import org.apache.flink.streaming.runtime.partitioner.StreamPartitioner; import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.apache.flink.table.data.RowData; import org.apache.flink.util.Preconditions; import java.util.Objects; @@ -41,13 +42,22 @@ public class GlutenKeyGroupStreamPartitioner extends StreamPartitioner keySelector; + private final KeySelector rowDataKeySelector; private int maxParallelism; public GlutenKeyGroupStreamPartitioner( KeySelector keySelector, int maxParallelism) { + this(keySelector, null, maxParallelism); + } + + public GlutenKeyGroupStreamPartitioner( + KeySelector keySelector, + KeySelector rowDataKeySelector, + int maxParallelism) { Preconditions.checkArgument(maxParallelism > 0, "Number of key-groups must be > 0!"); this.keySelector = Preconditions.checkNotNull(keySelector); + this.rowDataKeySelector = rowDataKeySelector; this.maxParallelism = maxParallelism; } @@ -57,12 +67,19 @@ public int getMaxParallelism() { @Override public int selectChannel(SerializationDelegate> record) { + Object value = record.getInstance().getValue(); try { - int channel = keySelector.getKey(record.getInstance().getValue()); - return channel; + if (value instanceof StatefulRecord) { + return keySelector.getKey((StatefulRecord) value); + } + Preconditions.checkState( + rowDataKeySelector != null, + "RowData key selector is required when partitioning non-StatefulRecord values."); + Object key = rowDataKeySelector.getKey((RowData) value); + return KeyGroupRangeAssignment.assignKeyToParallelOperator( + key, maxParallelism, numberOfChannels); } catch (Exception e) { - throw new RuntimeException( - "Could not extract key from " + record.getInstance().getValue(), e); + throw new RuntimeException("Could not extract key from " + value, e); } } @@ -104,11 +121,13 @@ public boolean equals(Object o) { return false; } final GlutenKeyGroupStreamPartitioner that = (GlutenKeyGroupStreamPartitioner) o; - return maxParallelism == that.maxParallelism && keySelector.equals(that.keySelector); + return maxParallelism == that.maxParallelism + && keySelector.equals(that.keySelector) + && Objects.equals(rowDataKeySelector, that.rowDataKeySelector); } @Override public int hashCode() { - return Objects.hash(super.hashCode(), keySelector, maxParallelism); + return Objects.hash(super.hashCode(), keySelector, rowDataKeySelector, maxParallelism); } } diff --git a/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/FlinkRowToVLVectorConvertor.java b/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/FlinkRowToVLVectorConvertor.java index 3abb06c5115..32a2cfec5ea 100644 --- a/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/FlinkRowToVLVectorConvertor.java +++ b/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/FlinkRowToVLVectorConvertor.java @@ -21,10 +21,12 @@ import io.github.zhztheplayer.velox4j.data.RowVector; import io.github.zhztheplayer.velox4j.session.Session; import io.github.zhztheplayer.velox4j.type.RowType; +import io.github.zhztheplayer.velox4j.type.TimestampType; import io.github.zhztheplayer.velox4j.type.Type; import org.apache.flink.table.data.GenericRowData; import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.TimestampData; import org.apache.arrow.memory.BufferAllocator; import org.apache.arrow.vector.FieldVector; @@ -67,7 +69,7 @@ public static List toRowData( for (int j = 0; j < rowVector.getSize(); j++) { Object[] fieldValues = new Object[rowType.size()]; for (int i = 0; i < rowType.size(); i++) { - fieldValues[i] = accessors.get(i).get(j); + fieldValues[i] = toFlinkValue(accessors.get(i).get(j), rowType.getChildren().get(i)); } rowDatas.add(GenericRowData.of(fieldValues)); } @@ -85,6 +87,19 @@ public static List toRowData( } } + private static Object toFlinkValue(Object value, Type type) { + if (value == null || !(type instanceof TimestampType)) { + return value; + } + if (value instanceof TimestampData) { + return value; + } + if (value instanceof Long) { + return TimestampData.fromEpochMillis((Long) value / 1000); + } + return value; + } + private static List buildArrowVectorAccessors(List vectors) { List accessors = new ArrayList<>(vectors.size()); for (int i = 0; i < vectors.size(); ++i) { From 51c7f24e86bbf012dc7d855ca46de2d9981e3eb2 Mon Sep 17 00:00:00 2001 From: zhanglistar Date: Sat, 11 Jul 2026 17:10:30 +0800 Subject: [PATCH 5/6] Thread.sleep(2000) is applied to every non-Kafka Nexmark query. Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- .../gluten/table/runtime/stream/custom/NexmarkTest.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/NexmarkTest.java b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/NexmarkTest.java index 1dcd6204769..7d45496de3b 100644 --- a/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/NexmarkTest.java +++ b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/NexmarkTest.java @@ -267,9 +267,9 @@ private void executeQuery(StreamTableEnvironment tEnv, String queryFileName, boo assertThat(checkJobRunningStatus(insertResult, 30000) == true); } else { waitForJobCompletion(insertResult, 30000); - // Allow filesystem sink to flush and commit partitions after job completion. - Thread.sleep(2000); if ("q10_orc.sql".equals(queryFileName)) { + // Allow filesystem sink to flush and commit partitions after job completion. + Thread.sleep(2000); verifyQ10OrcOutput(queryStartMillis); } } From f8c51e7149bdfe5a4b8c61c2d2d00857fc6239eb Mon Sep 17 00:00:00 2001 From: zhangzhibiao Date: Tue, 14 Jul 2026 15:25:45 +0800 Subject: [PATCH 6/6] [FLINK] Verify q10 ORC sink output --- .../runtime/stream/custom/NexmarkTest.java | 68 ++++++++++++------- 1 file changed, 43 insertions(+), 25 deletions(-) diff --git a/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/NexmarkTest.java b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/NexmarkTest.java index 7d45496de3b..469fa595458 100644 --- a/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/NexmarkTest.java +++ b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/NexmarkTest.java @@ -27,6 +27,12 @@ import com.salesforce.kafka.test.junit5.SharedKafkaTestResource; import com.salesforce.kafka.test.listeners.PlainListener; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch; +import org.apache.orc.OrcFile; +import org.apache.orc.Reader; +import org.apache.orc.RecordReader; +import org.apache.orc.TypeDescription; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; @@ -40,7 +46,6 @@ import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; -import java.nio.file.attribute.FileTime; import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; @@ -245,7 +250,6 @@ private void executeQuery(StreamTableEnvironment tEnv, String queryFileName, boo } } } - long queryStartMillis = System.currentTimeMillis(); String[] sqlStatements = queryContent.split(";"); assertThat(sqlStatements.length).isGreaterThanOrEqualTo(2); @@ -270,43 +274,57 @@ private void executeQuery(StreamTableEnvironment tEnv, String queryFileName, boo if ("q10_orc.sql".equals(queryFileName)) { // Allow filesystem sink to flush and commit partitions after job completion. Thread.sleep(2000); - verifyQ10OrcOutput(queryStartMillis); + verifyQ10OrcOutput(); } } } assertTrue(sqlStatements[sqlStatements.length - 1].trim().isEmpty()); } - private void verifyQ10OrcOutput(long queryStartMillis) { + private void verifyQ10OrcOutput() { Path outputDir = Paths.get("/tmp/data/output/bid_orc"); assertTrue("Q10 ORC output directory should exist", Files.exists(outputDir)); - try { - List outputFiles; - try (java.util.stream.Stream files = Files.walk(outputDir)) { - outputFiles = - files - .filter(Files::isRegularFile) - .filter(path -> isModifiedAfter(path, queryStartMillis)) - .filter(path -> !path.getFileName().toString().startsWith("_")) - .sorted() - .collect(Collectors.toList()); - } - assertThat(outputFiles).isNotEmpty(); - for (Path outputFile : outputFiles) { - assertThat(Files.size(outputFile)).isGreaterThan(0L); + + try (java.util.stream.Stream files = Files.walk(outputDir)) { + List regularFiles = + files.filter(Files::isRegularFile).sorted().collect(Collectors.toList()); + assertThat(regularFiles).allMatch(path -> !path.toString().contains(".inprogress")); + + List partFiles = + regularFiles.stream() + .filter(path -> path.getFileName().toString().startsWith("part-")) + .collect(Collectors.toList()); + assertThat(partFiles).isNotEmpty(); + + long rowCount = 0L; + for (Path partFile : partFiles) { + rowCount += readAndVerifyQ10OrcFile(partFile); } + assertThat(rowCount).isGreaterThan(0L); } catch (IOException e) { - throw new RuntimeException("Failed to read back Q10 ORC output", e); + throw new RuntimeException("Failed to inspect Q10 ORC output", e); } } - private boolean isModifiedAfter(Path path, long queryStartMillis) { - try { - FileTime lastModifiedTime = Files.getLastModifiedTime(path); - return lastModifiedTime.toMillis() >= queryStartMillis - 1000L; - } catch (IOException e) { - throw new RuntimeException("Failed to inspect output file " + path, e); + private long readAndVerifyQ10OrcFile(Path partFile) throws IOException { + Reader reader = + OrcFile.createReader( + new org.apache.hadoop.fs.Path(partFile.toUri()), + OrcFile.readerOptions(new Configuration())); + TypeDescription schema = reader.getSchema(); + assertThat(schema.getCategory()).isEqualTo(TypeDescription.Category.STRUCT); + assertThat(schema.getFieldNames()) + .containsExactly("auction", "bidder", "price", "dateTime", "extra"); + + long rowCount = 0L; + try (RecordReader rows = reader.rows()) { + VectorizedRowBatch batch = schema.createRowBatch(); + while (rows.nextBatch(batch)) { + rowCount += batch.size; + } } + assertThat(rowCount).isEqualTo(reader.getNumberOfRows()); + return rowCount; } private void waitForJobCompletion(TableResult result, long timeoutMs)