This is an automated email from the ASF dual-hosted git repository.
lgbo-ustc pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new b5d0205f26 [FLINK] Support Parquet format for Kafka to Hive streaming
(#12317)
b5d0205f26 is described below
commit b5d0205f268caeaa7497605254e677a966882c8f
Author: kevinyhzou <[email protected]>
AuthorDate: Fri Jul 3 09:36:13 2026 +0800
[FLINK] Support Parquet format for Kafka to Hive streaming (#12317)
Squashed PR #12317 into a single commit.
---
.github/workflows/flink.yml | 2 +-
gluten-flink/docs/Flink.md | 2 +-
.../apache/gluten/velox/FileSystemSinkFactory.java | 281 +++++++++++++---
.../gluten/velox/FuzzerSourceSinkFactory.java | 3 +-
.../apache/gluten/velox/HiveSourceSinkFactory.java | 99 ++++++
.../gluten/velox/KafkaSourceSinkFactory.java | 7 +-
.../org.apache.gluten.velox.VeloxSourceSinkFactory | 1 +
.../api/operators/GlutenStreamSource.java | 16 +-
.../runtime/operators/GlutenOneInputOperator.java | 5 +-
.../GlutenStreamingFileWriterOperator.java | 75 +++++
.../java/org/apache/gluten/util/ReflectUtils.java | 20 ++
.../org/apache/gluten/util/VectorOutputBridge.java | 18 +
gluten-flink/ut/pom.xml | 86 +++++
.../runtime/stream/custom/KafkaToHiveE2ETest.java | 362 +++++++++++++++++++++
14 files changed, 913 insertions(+), 64 deletions(-)
diff --git a/.github/workflows/flink.yml b/.github/workflows/flink.yml
index 656a08d5cf..a222cc7a25 100644
--- a/.github/workflows/flink.yml
+++ b/.github/workflows/flink.yml
@@ -88,7 +88,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
9811c337d5e37ab8d16d5bf0791b48836fcd2afd
+ cd velox4j && git reset --hard
1b1ad0833220476be6a042b83a474cf58f8f076c
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/docs/Flink.md b/gluten-flink/docs/Flink.md
index 87f9a4e24b..9c1900f06b 100644
--- a/gluten-flink/docs/Flink.md
+++ b/gluten-flink/docs/Flink.md
@@ -48,7 +48,7 @@ As some features have not been committed to upstream, you
have to use the follow
## fetch velox4j code
git clone -b gluten-0530 https://github.com/bigo-sg/velox4j.git
cd velox4j
-git reset --hard 9811c337d5e37ab8d16d5bf0791b48836fcd2afd
+git reset --hard 1b1ad0833220476be6a042b83a474cf58f8f076c
mvn clean install -DskipTests -Dgpg.skip -Dspotless.skip=true
```
**Get gluten**
diff --git
a/gluten-flink/planner/src/main/java/org/apache/gluten/velox/FileSystemSinkFactory.java
b/gluten-flink/planner/src/main/java/org/apache/gluten/velox/FileSystemSinkFactory.java
index ff08dcec63..57c9576322 100644
---
a/gluten-flink/planner/src/main/java/org/apache/gluten/velox/FileSystemSinkFactory.java
+++
b/gluten-flink/planner/src/main/java/org/apache/gluten/velox/FileSystemSinkFactory.java
@@ -18,6 +18,7 @@ package org.apache.gluten.velox;
import org.apache.gluten.streaming.api.operators.GlutenOneInputOperatorFactory;
import org.apache.gluten.table.runtime.operators.GlutenOneInputOperator;
+import
org.apache.gluten.table.runtime.operators.GlutenStreamingFileWriterOperator;
import org.apache.gluten.util.LogicalTypeConverter;
import org.apache.gluten.util.PlanNodeIdGenerator;
import org.apache.gluten.util.ReflectUtils;
@@ -38,7 +39,6 @@ import
org.apache.flink.streaming.api.transformations.OneInputTransformation;
import org.apache.flink.streaming.api.transformations.SinkTransformation;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.data.RowData;
-import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import java.util.HashMap;
import java.util.List;
@@ -47,24 +47,20 @@ import java.util.stream.Collectors;
public class FileSystemSinkFactory implements VeloxSourceSinkFactory {
- @SuppressWarnings("unchecked")
+ protected static final String PARTITION_COMMITTER_CLASS =
+ "org.apache.flink.connector.file.table.stream.PartitionCommitter";
+ protected static final String ABSTRACT_STREAMING_WRITER_CLASS =
+ "org.apache.flink.connector.file.table.stream.AbstractStreamingWriter";
+ private static final String HIVE_TABLE_META_STORE_FACTORY_CLASS =
+ "org.apache.flink.connectors.hive.HiveTableMetaStoreFactory";
+ private static final String CONNECTOR_FILESYSTEM = "connector-filesystem";
+
@Override
public boolean match(Transformation<RowData> transformation) {
- if (transformation instanceof SinkTransformation) {
- SinkTransformation<RowData, RowData> sinkTransformation =
- (SinkTransformation<RowData, RowData>) transformation;
- Transformation<RowData> inputTransformation =
- (Transformation<RowData>) sinkTransformation.getInputs().get(0);
- if (inputTransformation instanceof OneInputTransformation
- && inputTransformation.getName().equals("PartitionCommitter")) {
- OneInputTransformation<RowData, RowData> oneInputTransformatin =
- (OneInputTransformation<RowData, RowData>) inputTransformation;
- Transformation<RowData> preInputTransformation =
- (Transformation<RowData>) oneInputTransformatin.getInputs().get(0);
- return preInputTransformation.getName().equals("StreamingFileWriter");
- }
+ if (!isFileSystemSinkTransformation(transformation)) {
+ return false;
}
- return false;
+ return !isHiveConnector(transformation);
}
@Override
@@ -80,28 +76,22 @@ public class FileSystemSinkFactory implements
VeloxSourceSinkFactory {
SinkTransformation<RowData, RowData> sinkTransformation =
(SinkTransformation<RowData, RowData>) transformation;
OneInputTransformation<RowData, RowData> partitionCommitTransformation =
- (OneInputTransformation<RowData, RowData>)
sinkTransformation.getInputs().get(0);
+ getPartitionCommitTransformation(transformation);
OneInputTransformation<RowData, RowData> fileWriterTransformation =
- (OneInputTransformation<RowData, RowData>)
partitionCommitTransformation.getInputs().get(0);
+ getFileWriterTransformation(transformation);
OneInputStreamOperator<?, ?> operator =
fileWriterTransformation.getOperator();
List<String> partitionKeys =
(List<String>) ReflectUtils.getObjectField(operator.getClass(),
operator, "partitionKeys");
- Map<String, String> tableParams = new HashMap<>();
- Configuration tableOptions =
- (Configuration)
- ReflectUtils.getObjectField(
-
"org.apache.flink.connector.file.table.stream.PartitionCommitter",
- partitionCommitTransformation.getOperator(),
- "conf");
- tableParams.putAll(tableOptions.toMap());
-
+ Object partitionCommitter =
+ partitionCommitTransformation == null ? null :
partitionCommitTransformation.getOperator();
+ Map<String, String> tableParams = buildTableParams(partitionCommitter,
operator);
ResolvedSchema schema = (ResolvedSchema)
parameters.get(ResolvedSchema.class.getName());
List<String> columnList = schema.getColumnNames();
List<Integer> partitionIndexes =
partitionKeys.stream().mapToInt(columnList::indexOf).boxed().collect(Collectors.toList());
org.apache.flink.table.types.logical.RowType inputType =
(org.apache.flink.table.types.logical.RowType)
- ((InternalTypeInfo<?>)
fileWriterTransformation.getInputType()).toLogicalType();
+ schema.toPhysicalRowDataType().getLogicalType();
RowType inputDataColumns = (RowType)
LogicalTypeConverter.toVLType(inputType);
FileSystemInsertTableHandle insertTableHandle =
new FileSystemInsertTableHandle(
@@ -117,41 +107,45 @@ public class FileSystemSinkFactory implements
VeloxSourceSinkFactory {
inputDataColumns,
inputDataColumns.getNames(),
null,
- "connector-filesystem",
+ CONNECTOR_FILESYSTEM,
insertTableHandle,
false,
ignore,
CommitStrategy.NO_COMMIT,
List.of(new EmptyNode(inputDataColumns)));
GlutenOneInputOperator onewInputOperator =
- new GlutenOneInputOperator(
+ new GlutenStreamingFileWriterOperator(
new StatefulPlanNode(fileSystemWriteNode.getId(),
fileSystemWriteNode),
PlanNodeIdGenerator.newId(),
inputDataColumns,
Map.of(fileSystemWriteNode.getId(), ignore),
RowData.class,
- RowData.class,
- "FileSystemInsertTable");
- GlutenOneInputOperatorFactory<?, ?> operatorFactory =
- new GlutenOneInputOperatorFactory(onewInputOperator);
- Transformation<RowData> veloxFileWriterTransformation =
+ getSinkDescription());
+ GlutenOneInputOperatorFactory<RowData, ?> operatorFactory =
+ new GlutenOneInputOperatorFactory<>(onewInputOperator);
+ OneInputTransformation<RowData, ?> veloxFileWriterTransformation =
new OneInputTransformation(
fileWriterTransformation.getInputs().get(0),
fileWriterTransformation.getName(),
operatorFactory,
fileWriterTransformation.getOutputType(),
fileWriterTransformation.getParallelism());
- OneInputTransformation<RowData, RowData> newPartitionCommitTransformation =
- new OneInputTransformation(
- veloxFileWriterTransformation,
- partitionCommitTransformation.getName(),
- partitionCommitTransformation.getOperatorFactory(),
- partitionCommitTransformation.getOutputType(),
- partitionCommitTransformation.getParallelism());
+ Transformation<?> newSinkInputTransformation;
+ if (partitionCommitTransformation == null) {
+ newSinkInputTransformation = veloxFileWriterTransformation;
+ } else {
+ newSinkInputTransformation =
+ new OneInputTransformation(
+ veloxFileWriterTransformation,
+ partitionCommitTransformation.getName(),
+ partitionCommitTransformation.getOperatorFactory(),
+ partitionCommitTransformation.getOutputType(),
+ partitionCommitTransformation.getParallelism());
+ }
DataStream<RowData> newInputStream =
new DataStream<RowData>(
sinkTransformation.getInputStream().getExecutionEnvironment(),
- newPartitionCommitTransformation);
+ (Transformation<RowData>) newSinkInputTransformation);
return new SinkTransformation<RowData, RowData>(
newInputStream,
sinkTransformation.getSink(),
@@ -161,4 +155,205 @@ public class FileSystemSinkFactory implements
VeloxSourceSinkFactory {
sinkTransformation.isParallelismConfigured(),
sinkTransformation.getSinkOperatorsUidHashes());
}
+
+ @SuppressWarnings("unchecked")
+ protected boolean isFileSystemSinkTransformation(Transformation<RowData>
transformation) {
+ if (transformation instanceof SinkTransformation) {
+ SinkTransformation<RowData, RowData> sinkTransformation =
+ (SinkTransformation<RowData, RowData>) transformation;
+ Transformation<RowData> inputTransformation =
+ (Transformation<RowData>) sinkTransformation.getInputs().get(0);
+ if (inputTransformation instanceof OneInputTransformation) {
+ if (inputTransformation.getName().equals("StreamingFileWriter")) {
+ return true;
+ }
+ if (inputTransformation.getName().equals("PartitionCommitter")) {
+ OneInputTransformation<RowData, RowData> oneInputTransformation =
+ (OneInputTransformation<RowData, RowData>) inputTransformation;
+ Transformation<RowData> preInputTransformation =
+ (Transformation<RowData>)
oneInputTransformation.getInputs().get(0);
+ return
preInputTransformation.getName().equals("StreamingFileWriter");
+ }
+ }
+ }
+ return false;
+ }
+
+ @SuppressWarnings("unchecked")
+ protected OneInputTransformation<RowData, RowData>
getPartitionCommitTransformation(
+ Transformation<RowData> transformation) {
+ SinkTransformation<RowData, RowData> sinkTransformation =
+ (SinkTransformation<RowData, RowData>) transformation;
+ Transformation<RowData> inputTransformation =
+ (Transformation<RowData>) sinkTransformation.getInputs().get(0);
+ if (inputTransformation.getName().equals("PartitionCommitter")) {
+ return (OneInputTransformation<RowData, RowData>) inputTransformation;
+ }
+ return null;
+ }
+
+ @SuppressWarnings("unchecked")
+ protected OneInputTransformation<RowData, RowData>
getFileWriterTransformation(
+ Transformation<RowData> transformation) {
+ OneInputTransformation<RowData, RowData> partitionCommitTransformation =
+ getPartitionCommitTransformation(transformation);
+ if (partitionCommitTransformation != null) {
+ return (OneInputTransformation<RowData, RowData>)
+ partitionCommitTransformation.getInputs().get(0);
+ }
+ SinkTransformation<RowData, RowData> sinkTransformation =
+ (SinkTransformation<RowData, RowData>) transformation;
+ return (OneInputTransformation<RowData, RowData>)
sinkTransformation.getInputs().get(0);
+ }
+
+ protected Object getPartitionCommitter(Transformation<RowData>
transformation) {
+ OneInputTransformation<RowData, RowData> partitionCommitTransformation =
+ getPartitionCommitTransformation(transformation);
+ return partitionCommitTransformation == null
+ ? null
+ : partitionCommitTransformation.getOperator();
+ }
+
+ protected boolean isHiveConnector(Transformation<RowData> transformation) {
+ Object partitionCommitter = getPartitionCommitter(transformation);
+ if (partitionCommitter != null) {
+ return isHiveConnector(partitionCommitter);
+ }
+ return "hive"
+ .equals(
+
getTableOptions(getFileWriterTransformation(transformation).getOperator())
+ .toMap()
+ .get("connector"));
+ }
+
+ protected boolean isHiveConnector(Object partitionCommitter) {
+ Object metaStoreFactory =
+ ReflectUtils.getObjectField(
+ PARTITION_COMMITTER_CLASS, partitionCommitter, "metaStoreFactory");
+ return
HIVE_TABLE_META_STORE_FACTORY_CLASS.equals(metaStoreFactory.getClass().getName());
+ }
+
+ protected Map<String, String> buildTableParams(
+ Object partitionCommitter, OneInputStreamOperator<?, ?>
fileWriterOperator) {
+ Configuration tableOptions = getTableOptions(partitionCommitter,
fileWriterOperator);
+ Map<String, String> tableParams = new HashMap<>(tableOptions.toMap());
+ tableParams.putIfAbsent("path", getLocationPath(partitionCommitter,
fileWriterOperator));
+ tableParams.putIfAbsent("format", resolveWriteFormat(fileWriterOperator));
+ tableParams.put("connector", "filesystem");
+ return tableParams;
+ }
+
+ protected String getSinkDescription() {
+ return "FileSystemInsertTable";
+ }
+
+ protected String getDefaultFormat() {
+ return "unknown";
+ }
+
+ protected Configuration getTableOptions(OneInputStreamOperator<?, ?>
fileWriterOperator) {
+ return (Configuration)
+ ReflectUtils.getObjectField(fileWriterOperator.getClass(),
fileWriterOperator, "conf");
+ }
+
+ protected Configuration getTableOptions(
+ Object partitionCommitter, OneInputStreamOperator<?, ?>
fileWriterOperator) {
+ if (partitionCommitter != null) {
+ return (Configuration)
+ ReflectUtils.getObjectField(PARTITION_COMMITTER_CLASS,
partitionCommitter, "conf");
+ }
+ return getTableOptions(fileWriterOperator);
+ }
+
+ protected String getLocationPath(
+ Object partitionCommitter, OneInputStreamOperator<?, ?>
fileWriterOperator) {
+ if (partitionCommitter != null) {
+ Object locationPath =
+ ReflectUtils.getObjectField(
+ PARTITION_COMMITTER_CLASS, partitionCommitter, "locationPath");
+ return locationPath.toString();
+ }
+ Object bucketsBuilder =
+ ReflectUtils.getObjectField(
+ ABSTRACT_STREAMING_WRITER_CLASS, fileWriterOperator,
"bucketsBuilder");
+ Object basePath = ReflectUtils.tryGetObjectField(bucketsBuilder,
"basePath");
+ return basePath.toString();
+ }
+
+ protected String resolveWriteFormat(OneInputStreamOperator<?, ?>
fileWriterOperator) {
+ Object bucketsBuilder =
+ ReflectUtils.getObjectField(
+ ABSTRACT_STREAMING_WRITER_CLASS, fileWriterOperator,
"bucketsBuilder");
+ String format = resolveFormatFromBucketsBuilder(bucketsBuilder);
+ if (format != null) {
+ return format;
+ }
+ return getDefaultFormat();
+ }
+
+ protected String resolveFormatFromBucketsBuilder(Object bucketsBuilder) {
+ Class<?> builderClass = bucketsBuilder.getClass();
+ String builderName = builderClass.getName();
+ if (builderName.contains("HadoopPathBasedBulkFormatBuilder")) {
+ Object writerFactory = ReflectUtils.tryGetObjectField(bucketsBuilder,
"writerFactory");
+ if (writerFactory != null) {
+ return resolveFormatFromHadoopBulkWriterFactory(writerFactory);
+ }
+ return inferFormatFromClassName(builderName);
+ }
+ if (builderName.contains("BulkFormatBuilder")) {
+ Object writerFactory = ReflectUtils.tryGetObjectField(bucketsBuilder,
"writerFactory");
+ if (writerFactory != null) {
+ return resolveFormatFromBulkWriterFactory(writerFactory);
+ }
+ return inferFormatFromClassName(builderName);
+ }
+ if (builderName.contains("RowFormatBuilder")) {
+ Object encoder = ReflectUtils.tryGetObjectField(bucketsBuilder,
"encoder");
+ if (encoder != null) {
+ return inferFormatFromClassName(encoder.getClass().getName());
+ }
+ }
+ return inferFormatFromClassName(builderName);
+ }
+
+ protected String resolveFormatFromHadoopBulkWriterFactory(Object
writerFactory) {
+ return inferFormatFromClassName(writerFactory.getClass().getName());
+ }
+
+ protected String resolveFormatFromBulkWriterFactory(Object writerFactory) {
+ String format =
inferFormatFromClassName(writerFactory.getClass().getName());
+ if (format != null) {
+ return format;
+ }
+ Object innerFactory =
+ ReflectUtils.tryGetObjectField(writerFactory.getClass(),
writerFactory, "factory");
+ if (innerFactory != null) {
+ format = inferFormatFromClassName(innerFactory.getClass().getName());
+ if (format != null) {
+ return format;
+ }
+ }
+ return null;
+ }
+
+ protected String inferFormatFromClassName(String className) {
+ String lower = className.toLowerCase();
+ if (lower.contains("parquet")) {
+ return "parquet";
+ }
+ if (lower.contains("orc")) {
+ return "orc";
+ }
+ if (lower.contains("avro")) {
+ return "avro";
+ }
+ if (lower.contains("json")) {
+ return "json";
+ }
+ if (lower.contains("csv")) {
+ return "csv";
+ }
+ return null;
+ }
}
diff --git
a/gluten-flink/planner/src/main/java/org/apache/gluten/velox/FuzzerSourceSinkFactory.java
b/gluten-flink/planner/src/main/java/org/apache/gluten/velox/FuzzerSourceSinkFactory.java
index 6e9c232f04..b16d0e405a 100644
---
a/gluten-flink/planner/src/main/java/org/apache/gluten/velox/FuzzerSourceSinkFactory.java
+++
b/gluten-flink/planner/src/main/java/org/apache/gluten/velox/FuzzerSourceSinkFactory.java
@@ -59,7 +59,8 @@ public class FuzzerSourceSinkFactory implements
VeloxSourceSinkFactory {
(SinkTransformation<RowData, RowData>) transformation;
Transformation<?> inputTransformation =
sinkTransformation.getInputs().get(0);
if (sinkTransformation.getSink() instanceof DiscardingSink
- && !inputTransformation.getName().equals("PartitionCommitter")) {
+ && !inputTransformation.getName().equals("PartitionCommitter")
+ && transformation.getOutputType() instanceof InternalTypeInfo) {
return true;
}
} else if (transformation instanceof LegacySourceTransformation) {
diff --git
a/gluten-flink/planner/src/main/java/org/apache/gluten/velox/HiveSourceSinkFactory.java
b/gluten-flink/planner/src/main/java/org/apache/gluten/velox/HiveSourceSinkFactory.java
new file mode 100644
index 0000000000..56d356f237
--- /dev/null
+++
b/gluten-flink/planner/src/main/java/org/apache/gluten/velox/HiveSourceSinkFactory.java
@@ -0,0 +1,99 @@
+/*
+ * 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.velox;
+
+import org.apache.gluten.util.ReflectUtils;
+
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.table.data.RowData;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class HiveSourceSinkFactory extends FileSystemSinkFactory {
+
+ @Override
+ public boolean match(Transformation<RowData> transformation) {
+ if (!isFileSystemSinkTransformation(transformation)) {
+ return false;
+ }
+ return isHiveConnector(transformation);
+ }
+
+ @Override
+ protected Map<String, String> buildTableParams(
+ Object partitionCommitter, OneInputStreamOperator<?, ?>
fileWriterOperator) {
+ Configuration tableOptions = getTableOptions(partitionCommitter,
fileWriterOperator);
+ Map<String, String> tableParams = new HashMap<>(tableOptions.toMap());
+ tableParams.put("path", getLocationPath(partitionCommitter,
fileWriterOperator));
+ tableParams.putIfAbsent("format", resolveWriteFormat(fileWriterOperator));
+ tableParams.put("connector", "hive");
+ return tableParams;
+ }
+
+ @Override
+ protected String getSinkDescription() {
+ return "HiveInsertTable";
+ }
+
+ @Override
+ protected String getDefaultFormat() {
+ return "hive";
+ }
+
+ @Override
+ protected String resolveFormatFromHadoopBulkWriterFactory(Object
writerFactory) {
+ Class<?> factoryClass = writerFactory.getClass();
+ if (factoryClass.getName().contains("HiveBulkWriterFactory")) {
+ Object hiveWriterFactory =
+ ReflectUtils.getObjectField(factoryClass, writerFactory, "factory");
+ return resolveFormatFromHiveWriterFactory(hiveWriterFactory);
+ }
+ return super.resolveFormatFromHadoopBulkWriterFactory(writerFactory);
+ }
+
+ private String resolveFormatFromHiveWriterFactory(Object hiveWriterFactory) {
+ Class<?> factoryClass = hiveWriterFactory.getClass();
+ Object serDeInfoCached =
+ ReflectUtils.getObjectField(factoryClass, hiveWriterFactory,
"serDeInfo");
+ Object serDeInfo =
+ ReflectUtils.invokeObjectMethod(
+ serDeInfoCached.getClass(),
+ serDeInfoCached,
+ "deserializeValue",
+ new Class<?>[] {},
+ new Object[] {});
+ String serializationLib =
+ (String)
+ ReflectUtils.invokeObjectMethod(
+ serDeInfo.getClass(),
+ serDeInfo,
+ "getSerializationLib",
+ new Class<?>[] {},
+ new Object[] {});
+ String format = inferFormatFromClassName(serializationLib);
+ if (format != null) {
+ return format;
+ }
+ Class<?> outputFormatClz =
+ (Class<?>)
+ ReflectUtils.getObjectField(factoryClass, hiveWriterFactory,
"hiveOutputFormatClz");
+ return inferFormatFromClassName(outputFormatClz.getName());
+ }
+}
diff --git
a/gluten-flink/planner/src/main/java/org/apache/gluten/velox/KafkaSourceSinkFactory.java
b/gluten-flink/planner/src/main/java/org/apache/gluten/velox/KafkaSourceSinkFactory.java
index 589e57b0a2..3e6034faf2 100644
---
a/gluten-flink/planner/src/main/java/org/apache/gluten/velox/KafkaSourceSinkFactory.java
+++
b/gluten-flink/planner/src/main/java/org/apache/gluten/velox/KafkaSourceSinkFactory.java
@@ -31,6 +31,7 @@ import
io.github.zhztheplayer.velox4j.plan.TableScanWithWatermarkNode;
import io.github.zhztheplayer.velox4j.plan.WatermarkPushDownSpec;
import io.github.zhztheplayer.velox4j.type.RowType;
+import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.api.connector.source.Source;
import org.apache.flink.api.dag.Transformation;
import
org.apache.flink.streaming.api.transformations.LegacySourceTransformation;
@@ -54,7 +55,11 @@ public class KafkaSourceSinkFactory implements
VeloxSourceSinkFactory {
@Override
public boolean match(Transformation<RowData> transformation) {
if (transformation instanceof SourceTransformation) {
- Source source = ((SourceTransformation) transformation).getSource();
+ SourceTransformation sourceTransformation = (SourceTransformation)
transformation;
+ if (sourceTransformation.getBoundedness() == Boundedness.BOUNDED) {
+ return false;
+ }
+ Source source = sourceTransformation.getSource();
return source.getClass().getSimpleName().equals("KafkaSource");
}
return false;
diff --git
a/gluten-flink/planner/src/main/resources/META-INF/services/org.apache.gluten.velox.VeloxSourceSinkFactory
b/gluten-flink/planner/src/main/resources/META-INF/services/org.apache.gluten.velox.VeloxSourceSinkFactory
index db57ce53dc..862ffe305c 100644
---
a/gluten-flink/planner/src/main/resources/META-INF/services/org.apache.gluten.velox.VeloxSourceSinkFactory
+++
b/gluten-flink/planner/src/main/resources/META-INF/services/org.apache.gluten.velox.VeloxSourceSinkFactory
@@ -3,5 +3,6 @@ org.apache.gluten.velox.KafkaSourceSinkFactory
org.apache.gluten.velox.PulsarSourceSinkFactory
org.apache.gluten.velox.PrintSinkFactory
org.apache.gluten.velox.NexmarkSourceFactory
+org.apache.gluten.velox.HiveSourceSinkFactory
org.apache.gluten.velox.FileSystemSinkFactory
org.apache.gluten.velox.FuzzerSourceSinkFactory
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/streaming/api/operators/GlutenStreamSource.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/streaming/api/operators/GlutenStreamSource.java
index 349261ff4d..acc6b2e60a 100644
---
a/gluten-flink/runtime/src/main/java/org/apache/gluten/streaming/api/operators/GlutenStreamSource.java
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/streaming/api/operators/GlutenStreamSource.java
@@ -18,13 +18,11 @@ package org.apache.gluten.streaming.api.operators;
import org.apache.gluten.table.runtime.operators.GlutenSourceFunction;
import org.apache.gluten.util.ReflectUtils;
-import org.apache.gluten.util.Utils;
import io.github.zhztheplayer.velox4j.connector.ConnectorSplit;
import io.github.zhztheplayer.velox4j.plan.StatefulPlanNode;
import io.github.zhztheplayer.velox4j.type.RowType;
-import org.apache.flink.api.common.TaskInfo;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.operators.StreamSource;
@@ -91,19 +89,7 @@ public class GlutenStreamSource extends StreamSource
implements GlutenOperator {
@Override
public void notifyCheckpointComplete(long checkpointId) throws Exception {
super.notifyCheckpointComplete(checkpointId);
- String[] committed = sourceFunction.notifyCheckpointComplete(checkpointId);
- SourceFunction.SourceContext sourceContext = getSourceContext();
- TaskInfo taskInfo = getRuntimeContext().getTaskInfo();
- if (sourceContext != null
- && committed != null
- && taskInfo.getTaskName().contains("StreamingFileWriter")) {
- sourceContext.collect(
- Utils.constructCommitInfo(
- checkpointId,
- taskInfo.getIndexOfThisSubtask(),
- taskInfo.getNumberOfParallelSubtasks(),
- committed));
- }
+ sourceFunction.notifyCheckpointComplete(checkpointId);
}
@Override
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 09e2b2bb62..ef9d0d845d 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
@@ -340,8 +340,9 @@ public class GlutenOneInputOperator<IN, OUT> extends
TableStreamOperator<OUT>
@Override
public void notifyCheckpointComplete(long checkpointId) throws Exception {
- // TODO: notify velox
- task.notifyCheckpointComplete(checkpointId);
+ if (!(this instanceof GlutenStreamingFileWriterOperator)) {
+ task.notifyCheckpointComplete(checkpointId);
+ }
super.notifyCheckpointComplete(checkpointId);
}
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenStreamingFileWriterOperator.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenStreamingFileWriterOperator.java
new file mode 100644
index 0000000000..e508fb80c3
--- /dev/null
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenStreamingFileWriterOperator.java
@@ -0,0 +1,75 @@
+/*
+ * 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 org.apache.gluten.util.Utils;
+
+import io.github.zhztheplayer.velox4j.plan.StatefulPlanNode;
+import io.github.zhztheplayer.velox4j.type.RowType;
+
+import org.apache.flink.api.common.TaskInfo;
+import org.apache.flink.connector.file.table.stream.PartitionCommitInfo;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.util.Map;
+
+public class GlutenStreamingFileWriterOperator<IN>
+ extends GlutenOneInputOperator<IN, PartitionCommitInfo> {
+
+ public GlutenStreamingFileWriterOperator(
+ StatefulPlanNode plan,
+ String id,
+ RowType inputType,
+ Map<String, RowType> outputTypes,
+ Class<IN> inClass,
+ String description) {
+ super(plan, id, inputType, outputTypes, inClass,
PartitionCommitInfo.class, description);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public <NIN, NOUT> GlutenOneInputOperator<NIN, NOUT>
cloneWithInputOutputClasses(
+ StatefulPlanNode plan, Class<NIN> inClass, Class<NOUT> outClass) {
+ return (GlutenOneInputOperator<NIN, NOUT>)
+ new GlutenStreamingFileWriterOperator<>(
+ plan, getId(), getInputType(), getOutputTypes(), inClass,
getDescription());
+ }
+
+ @Override
+ public void processWatermark(Watermark mark) throws Exception {
+ super.processWatermark(mark);
+ output.emitWatermark(mark);
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) throws Exception {
+ String[] committed = task.notifyCheckpointComplete(checkpointId);
+ if (committed != null) {
+ TaskInfo taskInfo = getRuntimeContext().getTaskInfo();
+ output.collect(
+ new StreamRecord<PartitionCommitInfo>(
+ (PartitionCommitInfo)
+ Utils.constructCommitInfo(
+ checkpointId,
+ taskInfo.getIndexOfThisSubtask(),
+ taskInfo.getNumberOfParallelSubtasks(),
+ committed)));
+ }
+ super.notifyCheckpointComplete(checkpointId);
+ }
+}
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/util/ReflectUtils.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/util/ReflectUtils.java
index 7dff789ae2..4d5e65ccc8 100644
---
a/gluten-flink/runtime/src/main/java/org/apache/gluten/util/ReflectUtils.java
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/util/ReflectUtils.java
@@ -42,6 +42,26 @@ public class ReflectUtils {
}
}
+ public static Object tryGetObjectField(Class<?> clazz, Object obj, String
fieldName) {
+ try {
+ return getObjectField(clazz, obj, fieldName);
+ } catch (Exception e) {
+ return null;
+ }
+ }
+
+ public static Object tryGetObjectField(Object obj, String fieldName) {
+ Class<?> clazz = obj.getClass();
+ while (clazz != null) {
+ Object value = tryGetObjectField(clazz, obj, fieldName);
+ if (value != null) {
+ return value;
+ }
+ clazz = clazz.getSuperclass();
+ }
+ return null;
+ }
+
public static Object invokeObjectMethod(
Class<?> clazz, Object obj, String methodName, Class<?>[] paramTypes,
Object[] paramValues) {
try {
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/util/VectorOutputBridge.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/util/VectorOutputBridge.java
index 4849f90e94..bbc4ea796c 100644
---
a/gluten-flink/runtime/src/main/java/org/apache/gluten/util/VectorOutputBridge.java
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/util/VectorOutputBridge.java
@@ -19,6 +19,7 @@ package org.apache.gluten.util;
import io.github.zhztheplayer.velox4j.stateful.StatefulRecord;
import io.github.zhztheplayer.velox4j.type.RowType;
+import org.apache.flink.connector.file.table.stream.PartitionCommitInfo;
import org.apache.flink.streaming.api.operators.Output;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -65,6 +66,11 @@ public interface VectorOutputBridge<OUT> extends
Serializable {
@SuppressWarnings("unchecked")
VectorOutputBridge<OUT> bridge = (VectorOutputBridge<OUT>) new
StatefulRecordOutputBridge();
return bridge;
+ } else if (outputClass.isAssignableFrom(PartitionCommitInfo.class)) {
+ @SuppressWarnings("unchecked")
+ VectorOutputBridge<OUT> bridge =
+ (VectorOutputBridge<OUT>) new PartitionCommitInfoOutputBridge();
+ return bridge;
} else {
throw new UnsupportedOperationException(
"Unsupported output class: " + outputClass.getName());
@@ -134,4 +140,16 @@ public interface VectorOutputBridge<OUT> extends
Serializable {
return outputElement;
}
}
+
+ class PartitionCommitInfoOutputBridge implements
VectorOutputBridge<PartitionCommitInfo> {
+
+ @Override
+ public void collect(
+ Output<StreamRecord<PartitionCommitInfo>> collector,
+ StatefulRecord record,
+ BufferAllocator allocator,
+ RowType outputType) {
+ throw new UnsupportedOperationException("PartitionCommitInfo is not
supported");
+ }
+ }
}
diff --git a/gluten-flink/ut/pom.xml b/gluten-flink/ut/pom.xml
index f94d743e3e..c97c3838e1 100644
--- a/gluten-flink/ut/pom.xml
+++ b/gluten-flink/ut/pom.xml
@@ -118,6 +118,12 @@
<artifactId>junit-jupiter-engine</artifactId>
<version>5.9.2</version>
</dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <version>4.13.2</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.assertj</groupId>
<artifactId>assertj-core</artifactId>
@@ -153,6 +159,12 @@
<classifier>tests</classifier>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-sql-gateway-api</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>com.salesforce.kafka.test</groupId>
<artifactId>kafka-junit5</artifactId>
@@ -165,6 +177,30 @@
<version>3.9.1</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-compress</artifactId>
+ <version>1.26.2</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-auth</artifactId>
+ <version>2.7.2</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <version>2.7.2</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-hdfs</artifactId>
+ <version>2.7.2</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka</artifactId>
@@ -183,6 +219,56 @@
<version>${flink.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-hive_${scala.binary.version}</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-hive_${scala.binary.version}</artifactId>
+ <version>${flink.version}</version>
+ <classifier>tests</classifier>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hive</groupId>
+ <artifactId>hive-exec</artifactId>
+ <version>2.3.9</version>
+ <classifier>core</classifier>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.pentaho</groupId>
+ <artifactId>pentaho-aggdesigner-algorithm</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hive</groupId>
+ <artifactId>hive-metastore</artifactId>
+ <version>2.3.9</version>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>jdk.tools</groupId>
+ <artifactId>jdk.tools</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.pentaho</groupId>
+ <artifactId>pentaho-aggdesigner-algorithm</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-json</artifactId>
diff --git
a/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/KafkaToHiveE2ETest.java
b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/KafkaToHiveE2ETest.java
new file mode 100644
index 0000000000..0388cbf9d4
--- /dev/null
+++
b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/KafkaToHiveE2ETest.java
@@ -0,0 +1,362 @@
+/*
+ * 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.stream.custom;
+
+import org.apache.gluten.streaming.api.operators.GlutenOneInputOperatorFactory;
+import
org.apache.gluten.table.runtime.operators.GlutenStreamingFileWriterOperator;
+
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.transformations.OneInputTransformation;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.SqlDialect;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.internal.TableEnvironmentImpl;
+import org.apache.flink.table.catalog.hive.HiveCatalog;
+import org.apache.flink.table.catalog.hive.HiveTestUtils;
+import org.apache.flink.table.operations.ModifyOperation;
+import org.apache.flink.table.operations.Operation;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+
+import com.salesforce.kafka.test.junit5.SharedKafkaTestResource;
+import com.salesforce.kafka.test.listeners.PlainListener;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.lang.reflect.Method;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Enumeration;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class KafkaToHiveE2ETest {
+
+ private static final int KAFKA_PORT = 19092;
+ private static final String VELOX_SOURCE_SINK_FACTORY_SERVICE =
+ "META-INF/services/org.apache.gluten.velox.VeloxSourceSinkFactory";
+
+ @RegisterExtension
+ static final SharedKafkaTestResource KAFKA =
+ new SharedKafkaTestResource()
+ .withBrokerProperty("host.name", "127.0.0.1")
+ .withBrokers(1)
+ .registerListener(new PlainListener().onPorts(KAFKA_PORT));
+
+ @Test
+ void testInsertFromKafkaToHive(@TempDir Path tempDir) throws Exception {
+ String topic = "kafka_to_hive_" +
UUID.randomUUID().toString().replace("-", "");
+ KAFKA.getKafkaTestUtils().createTopic(topic, 1, (short) 1);
+ KAFKA
+ .getKafkaTestUtils()
+ .produceRecords(
+ List.of(
+ jsonRecord(topic, "{\"id\":1,\"name\":\"alice\"}"),
+ jsonRecord(topic, "{\"id\":2,\"name\":\"bob\"}"),
+ jsonRecord(topic, "{\"id\":3,\"name\":\"carol\"}")));
+
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+ StreamTableEnvironment tEnv =
+ StreamTableEnvironment.create(
+ env, EnvironmentSettings.newInstance().inStreamingMode().build());
+
+ createKafkaSource(tEnv, topic);
+ Path sinkPath = tempDir.resolve("hive_sink");
+ createHiveCatalogAndSink(tEnv, sinkPath);
+
+ String insertSql =
+ "INSERT INTO hive_catalog.`default`.hive_sink "
+ + "SELECT id, name FROM
default_catalog.default_database.kafka_source";
+ assertHiveSinkOffloadedToNative(tEnv, insertSql);
+ TableResult insertResult = executeSqlWithoutVeloxSourceSinkFactories(tEnv,
insertSql);
+ JobClient jobClient = insertResult.getJobClient().orElseThrow();
+ jobClient.getJobExecutionResult().get(60000, TimeUnit.MILLISECONDS);
+
+
assertThat(readHiveTextRows(sinkPath)).containsExactlyInAnyOrder("1,alice",
"2,bob", "3,carol");
+ }
+
+ @Test
+ void testCommitHivePartitionWithProcessTime(@TempDir Path tempDir) throws
Exception {
+ testCommitHivePartition(tempDir, "process-time");
+ }
+
+ @Test
+ void testCommitHivePartitionWithPartitionTime(@TempDir Path tempDir) throws
Exception {
+ testCommitHivePartition(tempDir, "partition-time");
+ }
+
+ private void testCommitHivePartition(Path tempDir, String
partitionCommitTrigger)
+ throws Exception {
+ String topic =
+ "kafka_to_partitioned_hive_"
+ + partitionCommitTrigger.replace("-", "_")
+ + "_"
+ + UUID.randomUUID().toString().replace("-", "");
+ KAFKA.getKafkaTestUtils().createTopic(topic, 1, (short) 1);
+ KAFKA
+ .getKafkaTestUtils()
+ .produceRecords(
+ List.of(
+ jsonRecord(
+ topic,
"{\"id\":1,\"name\":\"alice\",\"dt\":\"2026-06-29\",\"hm\":\"10:00\"}"),
+ jsonRecord(
+ topic,
"{\"id\":2,\"name\":\"bob\",\"dt\":\"2026-06-29\",\"hm\":\"10:00\"}"),
+ jsonRecord(
+ topic,
+
"{\"id\":3,\"name\":\"carol\",\"dt\":\"2026-06-29\",\"hm\":\"10:01\"}")));
+
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+ StreamTableEnvironment tEnv =
+ StreamTableEnvironment.create(
+ env, EnvironmentSettings.newInstance().inStreamingMode().build());
+
+ createPartitionedKafkaSource(tEnv, topic);
+ Path sinkPath = tempDir.resolve("partitioned_hive_sink");
+ createHiveCatalogAndPartitionedSink(tEnv, sinkPath,
partitionCommitTrigger);
+
+ TableResult insertResult =
+ executeSqlWithoutVeloxSourceSinkFactories(
+ tEnv,
+ "INSERT INTO hive_catalog.`default`.partitioned_hive_sink "
+ + "SELECT id, name, dt, hm FROM
default_catalog.default_database.kafka_source");
+ JobClient jobClient = insertResult.getJobClient().orElseThrow();
+ jobClient.getJobExecutionResult().get(60000, TimeUnit.MILLISECONDS);
+
+
assertThat(readHiveTextRows(sinkPath)).containsExactlyInAnyOrder("1,alice",
"2,bob", "3,carol");
+ assertCommittedPartitions(tEnv);
+ assertCommittedPartition(sinkPath, "2026-06-29", "10:00");
+ assertCommittedPartition(sinkPath, "2026-06-29", "10:01");
+ }
+
+ private static ProducerRecord<byte[], byte[]> jsonRecord(String topic,
String value) {
+ return new ProducerRecord<>(topic, value.getBytes(StandardCharsets.UTF_8));
+ }
+
+ private void createKafkaSource(StreamTableEnvironment tEnv, String topic) {
+ tEnv.executeSql(
+ "CREATE TABLE kafka_source ("
+ + " id INT,"
+ + " name STRING"
+ + ") WITH ("
+ + " 'connector' = 'kafka',"
+ + " 'topic' = '"
+ + topic
+ + "',"
+ + " 'properties.bootstrap.servers' = '"
+ + "127.0.0.1:"
+ + KAFKA_PORT
+ + "',"
+ + " 'properties.group.id' = 'kafka-to-hive-e2e',"
+ + " 'properties.broker.address.family' = 'v4',"
+ + " 'scan.startup.mode' = 'earliest-offset',"
+ + " 'scan.bounded.mode' = 'latest-offset',"
+ + " 'format' = 'json'"
+ + ")");
+ }
+
+ private void createPartitionedKafkaSource(StreamTableEnvironment tEnv,
String topic) {
+ tEnv.executeSql(
+ "CREATE TABLE kafka_source ("
+ + " id INT,"
+ + " name STRING,"
+ + " dt STRING,"
+ + " hm STRING"
+ + ") WITH ("
+ + " 'connector' = 'kafka',"
+ + " 'topic' = '"
+ + topic
+ + "',"
+ + " 'properties.bootstrap.servers' = '"
+ + "127.0.0.1:"
+ + KAFKA_PORT
+ + "',"
+ + " 'properties.group.id' = 'kafka-to-partitioned-hive-e2e',"
+ + " 'properties.broker.address.family' = 'v4',"
+ + " 'scan.startup.mode' = 'earliest-offset',"
+ + " 'scan.bounded.mode' = 'latest-offset',"
+ + " 'format' = 'json'"
+ + ")");
+ }
+
+ private void createHiveCatalogAndSink(StreamTableEnvironment tEnv, Path
sinkPath) {
+ HiveCatalog hiveCatalog = HiveTestUtils.createHiveCatalog("hive_catalog",
"2.3.9");
+ tEnv.registerCatalog("hive_catalog", hiveCatalog);
+
+ tEnv.useCatalog("hive_catalog");
+ tEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
+ tEnv.executeSql(
+ "CREATE TABLE hive_sink (id INT, name STRING) "
+ + "ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' "
+ + "STORED AS TEXTFILE LOCATION '"
+ + sinkPath.toUri()
+ + "'");
+ tEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
+ tEnv.useCatalog("default_catalog");
+ }
+
+ private void createHiveCatalogAndPartitionedSink(
+ StreamTableEnvironment tEnv, Path sinkPath, String
partitionCommitTrigger) {
+ HiveCatalog hiveCatalog = HiveTestUtils.createHiveCatalog("hive_catalog",
"2.3.9");
+ tEnv.registerCatalog("hive_catalog", hiveCatalog);
+
+ tEnv.useCatalog("hive_catalog");
+ tEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
+ tEnv.executeSql(
+ "CREATE TABLE partitioned_hive_sink (id INT, name STRING) "
+ + "PARTITIONED BY (dt STRING, hm STRING) "
+ + "ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' "
+ + "STORED AS TEXTFILE LOCATION '"
+ + sinkPath.toUri()
+ + "' "
+ + "TBLPROPERTIES ("
+ + " 'sink.partition-commit.trigger' = '"
+ + partitionCommitTrigger
+ + "',"
+ + " 'sink.partition-commit.delay' = '0 s',"
+ + " 'sink.partition-commit.policy.kind' =
'metastore,success-file',"
+ + " 'partition.time-extractor.timestamp-pattern' = '$dt $hm:00'"
+ + ")");
+ tEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
+ tEnv.useCatalog("default_catalog");
+ }
+
+ private static TableResult executeSqlWithoutVeloxSourceSinkFactories(
+ StreamTableEnvironment tEnv, String sql) {
+ ClassLoader originalClassLoader =
Thread.currentThread().getContextClassLoader();
+ Thread.currentThread()
+ .setContextClassLoader(new
VeloxSourceSinkFactoryFilteringClassLoader(originalClassLoader));
+ try {
+ return tEnv.executeSql(sql);
+ } finally {
+ Thread.currentThread().setContextClassLoader(originalClassLoader);
+ }
+ }
+
+ private static class VeloxSourceSinkFactoryFilteringClassLoader extends
ClassLoader {
+ private VeloxSourceSinkFactoryFilteringClassLoader(ClassLoader parent) {
+ super(parent);
+ }
+
+ @Override
+ public URL getResource(String name) {
+ if (VELOX_SOURCE_SINK_FACTORY_SERVICE.equals(name)) {
+ return null;
+ }
+ return super.getResource(name);
+ }
+
+ @Override
+ public Enumeration<URL> getResources(String name) throws IOException {
+ if (VELOX_SOURCE_SINK_FACTORY_SERVICE.equals(name)) {
+ return Collections.emptyEnumeration();
+ }
+ return super.getResources(name);
+ }
+ }
+
+ private static List<String> readHiveTextRows(Path sinkPath) throws Exception
{
+ try (java.util.stream.Stream<Path> paths = Files.walk(sinkPath)) {
+ return paths
+ .filter(Files::isRegularFile)
+ .filter(path -> !path.getFileName().toString().startsWith("."))
+ .filter(path -> !path.getFileName().toString().startsWith("_"))
+ .flatMap(
+ path -> {
+ try {
+ return Files.lines(path, StandardCharsets.UTF_8);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ })
+ .collect(java.util.stream.Collectors.toList());
+ }
+ }
+
+ private static void assertCommittedPartitions(StreamTableEnvironment tEnv)
throws Exception {
+ tEnv.useCatalog("hive_catalog");
+ List<String> partitions = new ArrayList<>();
+ try (CloseableIterator<Row> rows =
+ tEnv.executeSql("SHOW PARTITIONS partitioned_hive_sink").collect()) {
+ while (rows.hasNext()) {
+ partitions.add(String.valueOf(rows.next().getField(0)));
+ }
+ } finally {
+ tEnv.useCatalog("default_catalog");
+ }
+ assertThat(partitions)
+ .containsExactlyInAnyOrder("dt=2026-06-29/hm=10:00",
"dt=2026-06-29/hm=10:01");
+ }
+
+ private static void assertCommittedPartition(Path sinkPath, String dt,
String hm) {
+ Path partitionPath = sinkPath.resolve("dt=" + dt).resolve("hm=" +
hm.replace(":", "%3A"));
+ assertThat(partitionPath).isDirectory();
+ assertThat(partitionPath.resolve("_SUCCESS")).exists();
+ }
+
+ private static void assertHiveSinkOffloadedToNative(StreamTableEnvironment
tEnv, String insertSql)
+ throws Exception {
+ assertThat(
+ translateInsert(tEnv, insertSql).stream()
+ .anyMatch(KafkaToHiveE2ETest::containsNativeHiveSink))
+ .as("Hive HDFS sink should use GlutenStreamingFileWriterOperator")
+ .isTrue();
+ }
+
+ @SuppressWarnings("unchecked")
+ private static List<Transformation<?>> translateInsert(
+ StreamTableEnvironment tEnv, String insertSql) throws Exception {
+ TableEnvironmentImpl tableEnvironment = (TableEnvironmentImpl) tEnv;
+ List<Operation> operations = tableEnvironment.getParser().parse(insertSql);
+ assertThat(operations).hasSize(1);
+ assertThat(operations.get(0)).isInstanceOf(ModifyOperation.class);
+
+ Method translate =
TableEnvironmentImpl.class.getDeclaredMethod("translate", List.class);
+ translate.setAccessible(true);
+ return (List<Transformation<?>>)
+ translate.invoke(tableEnvironment, List.of((ModifyOperation)
operations.get(0)));
+ }
+
+ private static boolean containsNativeHiveSink(Transformation<?>
transformation) {
+ if (transformation instanceof OneInputTransformation) {
+ OneInputTransformation<?, ?> oneInputTransformation =
+ (OneInputTransformation<?, ?>) transformation;
+ if (oneInputTransformation.getOperatorFactory() instanceof
GlutenOneInputOperatorFactory) {
+ GlutenOneInputOperatorFactory<?, ?> operatorFactory =
+ (GlutenOneInputOperatorFactory<?, ?>)
oneInputTransformation.getOperatorFactory();
+ if (operatorFactory.getOperator() instanceof
GlutenStreamingFileWriterOperator) {
+ return true;
+ }
+ }
+ }
+ return
transformation.getInputs().stream().anyMatch(KafkaToHiveE2ETest::containsNativeHiveSink);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
