This is an automated email from the ASF dual-hosted git repository.
KevinyhZou 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 90e9825a89 [GLUTEN-12572][FLINK]Add support for compression in hive
table sink (#12571)
90e9825a89 is described below
commit 90e9825a895cc926d246724cb47140f3f326fbdc
Author: kevinyhzou <[email protected]>
AuthorDate: Tue Jul 28 17:21:55 2026 +0800
[GLUTEN-12572][FLINK]Add support for compression in hive table sink (#12571)
* [GLUTEN-12206][FLINK] Support filesystem writer bucket state snapshot and
restore
* Support compresion type of hive table
---
.github/workflows/flink.yml | 2 +-
gluten-flink/docs/Flink.md | 2 +-
.../apache/gluten/velox/HiveSourceSinkFactory.java | 98 +++++++++++++++--
.../GlutenStreamingFileWriterOperator.java | 14 ++-
.../gluten/velox/HiveSourceSinkFactoryTest.java | 118 +++++++++++++++++++++
5 files changed, 220 insertions(+), 14 deletions(-)
diff --git a/.github/workflows/flink.yml b/.github/workflows/flink.yml
index 878b47145d..9d7906a1e9 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
95e9afe64c38d2e5ba962eee7949ef629fe50197
+ cd velox4j && git reset --hard
feb921eabe637bff88eda0fc753d36bbe422fe87
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 1da1209f20..3fc46f25d1 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 95e9afe64c38d2e5ba962eee7949ef629fe50197
+git reset --hard feb921eabe637bff88eda0fc753d36bbe422fe87
mvn clean install -DskipTests -Dgpg.skip -Dspotless.skip=true
```
**Get gluten**
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
index 56d356f237..b0e5248b6a 100644
---
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
@@ -25,8 +25,13 @@ import org.apache.flink.table.data.RowData;
import java.util.HashMap;
import java.util.Map;
+import java.util.Properties;
public class HiveSourceSinkFactory extends FileSystemSinkFactory {
+ private static final String COMPRESSION_KIND = "sink.file.compression";
+ private static final String[] SUPPORTED_COMPRESSION_TABLE_KEYS = {
+ "orc.compress", "parquet.compression", "parquet.compression.codec",
"parquet.compression-codec"
+ };
@Override
public boolean match(Transformation<RowData> transformation) {
@@ -42,8 +47,10 @@ public class HiveSourceSinkFactory extends
FileSystemSinkFactory {
Configuration tableOptions = getTableOptions(partitionCommitter,
fileWriterOperator);
Map<String, String> tableParams = new HashMap<>(tableOptions.toMap());
tableParams.put("path", getLocationPath(partitionCommitter,
fileWriterOperator));
- tableParams.putIfAbsent("format", resolveWriteFormat(fileWriterOperator));
+ Object bucketsBuilder = getBucketsBuilder(fileWriterOperator);
+ tableParams.putIfAbsent("format", resolveWriteFormat(bucketsBuilder));
tableParams.put("connector", "hive");
+ addHiveCompressionParams(bucketsBuilder, tableParams);
return tableParams;
}
@@ -57,15 +64,21 @@ public class HiveSourceSinkFactory extends
FileSystemSinkFactory {
return "hive";
}
+ private String resolveWriteFormat(Object bucketsBuilder) {
+ String format = resolveFormatFromBucketsBuilder(bucketsBuilder);
+ if (format != null) {
+ return format;
+ }
+ return getDefaultFormat();
+ }
+
@Override
- protected String resolveFormatFromHadoopBulkWriterFactory(Object
writerFactory) {
- Class<?> factoryClass = writerFactory.getClass();
- if (factoryClass.getName().contains("HiveBulkWriterFactory")) {
- Object hiveWriterFactory =
- ReflectUtils.getObjectField(factoryClass, writerFactory, "factory");
+ protected String resolveFormatFromBucketsBuilder(Object bucketsBuilder) {
+ Object hiveWriterFactory =
getHiveWriterFactoryFromBucketsBuilder(bucketsBuilder);
+ if (hiveWriterFactory != null) {
return resolveFormatFromHiveWriterFactory(hiveWriterFactory);
}
- return super.resolveFormatFromHadoopBulkWriterFactory(writerFactory);
+ return super.resolveFormatFromBucketsBuilder(bucketsBuilder);
}
private String resolveFormatFromHiveWriterFactory(Object hiveWriterFactory) {
@@ -96,4 +109,75 @@ public class HiveSourceSinkFactory extends
FileSystemSinkFactory {
ReflectUtils.getObjectField(factoryClass, hiveWriterFactory,
"hiveOutputFormatClz");
return inferFormatFromClassName(outputFormatClz.getName());
}
+
+ private void addHiveCompressionParams(Object bucketsBuilder, Map<String,
String> tableParams) {
+ Object hiveWriterFactory =
getHiveWriterFactoryFromBucketsBuilder(bucketsBuilder);
+ if (hiveWriterFactory == null) {
+ return;
+ }
+ Properties tableProperties =
+ (Properties) ReflectUtils.tryGetObjectField(hiveWriterFactory,
"tableProperties");
+ addNativeCompressionParamFromTableProperties(tableProperties, tableParams);
+ }
+
+ private Object getBucketsBuilder(OneInputStreamOperator<?, ?>
fileWriterOperator) {
+ return ReflectUtils.getObjectField(
+ ABSTRACT_STREAMING_WRITER_CLASS, fileWriterOperator, "bucketsBuilder");
+ }
+
+ private Object getHiveWriterFactoryFromBucketsBuilder(Object bucketsBuilder)
{
+ Object writerFactory = ReflectUtils.tryGetObjectField(bucketsBuilder,
"writerFactory");
+ if (writerFactory == null
+ ||
!writerFactory.getClass().getName().contains("HiveBulkWriterFactory")) {
+ return null;
+ }
+ return ReflectUtils.getObjectField(writerFactory.getClass(),
writerFactory, "factory");
+ }
+
+ static void addNativeCompressionParamFromTableProperties(
+ Properties tableProperties, Map<String, String> tableParams) {
+ if (!isParquetFormat(tableParams.get("format"))) {
+ tableParams.remove(COMPRESSION_KIND);
+ return;
+ }
+ if (tableProperties == null) {
+ return;
+ }
+
+ String compressionKind = resolveCompressionKind(tableProperties);
+ if (compressionKind != null) {
+ tableParams.put(COMPRESSION_KIND, compressionKind);
+ }
+ }
+
+ private static boolean isParquetFormat(String format) {
+ return format != null && "parquet".equalsIgnoreCase(format.trim());
+ }
+
+ private static String resolveCompressionKind(Properties tableProperties) {
+ for (String key : SUPPORTED_COMPRESSION_TABLE_KEYS) {
+ String compressionKind =
normalizeCompressionKind(tableProperties.getProperty(key));
+ if (compressionKind != null) {
+ return compressionKind;
+ }
+ }
+ return null;
+ }
+
+ static String normalizeCompressionKind(String compression) {
+ if (compression == null) {
+ return null;
+ }
+ final Map<String, String> supportedCompressionKinds =
+ Map.ofEntries(
+ Map.entry("snappy", "snappy"),
+ Map.entry("gzip", "gzip"),
+ Map.entry("zstd", "zstd"),
+ Map.entry("zstandard", "zstd"),
+ Map.entry("lz4", "lz4"),
+ Map.entry("lzo", "lzo"),
+ Map.entry("zlib", "zlib"),
+ Map.entry("deflate", "zlib"));
+ return
supportedCompressionKinds.getOrDefault(compression.trim().toLowerCase(), null);
+ }
}
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
index 833b2c452a..0f4eee8411 100644
---
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
@@ -95,11 +95,15 @@ public class GlutenStreamingFileWriterOperator<IN>
}
restoredCheckpointRecords = records.toArray(new String[0]);
}
- LOG.info(
- "Restore native file writer state for operator {}, restored {},
records {}",
- getDescription(),
- context.isRestored(),
- Arrays.toString(restoredCheckpointRecords));
+ if (restoredCheckpointRecords != null) {
+ LOG.info(
+ "Restore native file writer state for operator {}, restored {},
records {}, contents {}",
+ getDescription(),
+ context.isRestored(),
+ restoredCheckpointRecords.length,
+ Arrays.toString(
+ restoredCheckpointRecords != null ? restoredCheckpointRecords :
new String[0]));
+ }
if (task == null) {
initSession();
}
diff --git
a/gluten-flink/ut/src/test/java/org/apache/gluten/velox/HiveSourceSinkFactoryTest.java
b/gluten-flink/ut/src/test/java/org/apache/gluten/velox/HiveSourceSinkFactoryTest.java
new file mode 100644
index 0000000000..1b1dabfbe7
--- /dev/null
+++
b/gluten-flink/ut/src/test/java/org/apache/gluten/velox/HiveSourceSinkFactoryTest.java
@@ -0,0 +1,118 @@
+/*
+ * 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.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class HiveSourceSinkFactoryTest {
+
+ @Test
+ void addNativeCompressionParamMapsSupportedParquetCodecs() {
+ String[][] compressionCodecs = {
+ {"SNAPPY", "snappy"},
+ {"GZIP", "gzip"},
+ {"zstandard", "zstd"},
+ {"LZ4", "lz4"},
+ {"LZO", "lzo"},
+ {"deflate", "zlib"}
+ };
+
+ for (String[] compressionCodec : compressionCodecs) {
+ Properties tableProperties = new Properties();
+ tableProperties.setProperty("parquet.compression", compressionCodec[0]);
+
+ Map<String, String> tableParams = new HashMap<>();
+ tableParams.put("format", "parquet");
+ HiveSourceSinkFactory.addNativeCompressionParamFromTableProperties(
+ tableProperties, tableParams);
+
+ assertThat(tableParams)
+ .containsEntry("sink.file.compression", compressionCodec[1])
+ .doesNotContainKey("parquet.compression");
+ }
+ }
+
+ @Test
+ void addNativeCompressionParamReadsSupportedParquetCompressionKeys() {
+ Properties tableProperties = new Properties();
+ tableProperties.setProperty("parquet.compression.codec", "SNAPPY");
+
+ Map<String, String> tableParams = new HashMap<>();
+ tableParams.put("format", "parquet");
+ HiveSourceSinkFactory.addNativeCompressionParamFromTableProperties(
+ tableProperties, tableParams);
+
+ assertThat(tableParams).containsEntry("sink.file.compression", "snappy");
+ }
+
+ @Test
+ void addNativeCompressionParamDoesNotProduceConfigForUnsupportedFormats() {
+ for (String format : new String[] {"orc", "json", "csv", "hive"}) {
+ Properties tableProperties = new Properties();
+ tableProperties.setProperty("parquet.compression", "SNAPPY");
+
+ Map<String, String> tableParams = new HashMap<>();
+ tableParams.put("format", format);
+ tableParams.put("sink.file.compression", "snappy");
+ HiveSourceSinkFactory.addNativeCompressionParamFromTableProperties(
+ tableProperties, tableParams);
+
+ assertThat(tableParams)
+ .containsEntry("format", format)
+ .doesNotContainKey("sink.file.compression");
+ }
+ }
+
+ @Test
+ void
addNativeCompressionParamDoesNotProduceConfigForUnsupportedParquetCodecs() {
+ for (String compressionCodec : new String[] {"brotli",
"org.example.SnappyCodec"}) {
+ Properties tableProperties = new Properties();
+ tableProperties.setProperty("parquet.compression", compressionCodec);
+
+ Map<String, String> tableParams = new HashMap<>();
+ tableParams.put("format", "parquet");
+ HiveSourceSinkFactory.addNativeCompressionParamFromTableProperties(
+ tableProperties, tableParams);
+
+ assertThat(tableParams)
+ .containsEntry("format", "parquet")
+ .doesNotContainKey("sink.file.compression");
+ }
+ }
+
+ @Test
+ void
addNativeCompressionParamDoesNotProduceConfigForUnsupportedCompressionKeys() {
+ Properties tableProperties = new Properties();
+ tableProperties.setProperty("custom.compress", "SNAPPY");
+ tableProperties.setProperty("custom.codec", "GZIP");
+
+ Map<String, String> tableParams = new HashMap<>();
+ tableParams.put("format", "parquet");
+ HiveSourceSinkFactory.addNativeCompressionParamFromTableProperties(
+ tableProperties, tableParams);
+
+ assertThat(tableParams)
+ .containsEntry("format", "parquet")
+ .doesNotContainKey("sink.file.compression");
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
