lgbo-ustc commented on code in PR #12571:
URL: https://github.com/apache/gluten/pull/12571#discussion_r3629149523
##########
gluten-flink/planner/src/main/java/org/apache/gluten/velox/HiveSourceSinkFactory.java:
##########
@@ -96,4 +109,87 @@ private String resolveFormatFromHiveWriterFactory(Object
hiveWriterFactory) {
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) {
Review Comment:
This normalization would be easier to read and maintain as a static map from
lowercase input values to canonical native codec names. Then unsupported values
naturally return null instead of being handled in a long switch. For example:
```java
private static final Map<String, String> SUPPORTED_COMPRESSION_KINDS =
Map.ofEntries(
Map.entry("snappy", "snappy"),
Map.entry("zstd", "zstd"),
Map.entry("zstandard", "zstd"),
Map.entry("lz4", "lz4"));
static String normalizeCompressionKind(String compression) {
if (compression == null) {
return null;
}
return SUPPORTED_COMPRESSION_KINDS.get(compression.trim().toLowerCase());
}
```
This also makes the supported codec set explicit and avoids accidentally
passing unsupported native codecs later.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
