[GLUTEN-12572][FLINK]Add support for compression in hive table sink

[KevinyhZou]

What changes are proposed in this pull request?

#12572

How was this patch tested?

UT

Was this patch authored or co-authored using generative AI tooling?

cursor

Related issue: #12202

[lgbo-ustc]

I think this needs changes before merge. The Java side now maps Hive table compression properties into filesystem sink compression configs, but the native filesystem writer currently only supports Parquet compression for snappy/lz4/zstd. This introduces failure cases for common Hive table definitions.

1. ORC compressed Hive sinks can fail after offload. reads and writes , but rejects any non-none compression unless the write format is Parquet. A Hive ORC table with / would be planned as offloaded and then fail in native instead of being supported. Please either add native ORC compression support or avoid setting this config / avoid offload for unsupported ORC compression.

2. Parquet gzip is also mapped to a native config that is rejected. returns �����, but the native filesystem sink only allows snappy/lz4/zstd for Parquet and throws for gzip. Hive/Flink tables configured with Parquet gzip would regress from Java writer behavior to native runtime failure. Please add gzip support or make planner fallback for unsupported codecs.

The added UT only checks that the map contains ; it does not cover whether the generated config is accepted by native for ORC or gzip. Please add coverage for these unsupported/native behavior cases as part of the fix.

[lgbo-ustc]

I think this needs changes before merge. The Java side now maps Hive table compression properties into filesystem sink compression configs, but the native filesystem writer currently only supports Parquet compression for snappy/lz4/zstd. This introduces failure cases for common Hive table definitions.

1. ORC compressed Hive sinks can fail after offload. HiveSourceSinkFactory reads orc.compress and writes sink.file.compression, but FileSystemWriteConfig::getFileCompressionType() rejects any non-none compression unless the write format is Parquet. A Hive ORC table with orc.compress=SNAPPY or ZLIB would be planned as offloaded and then fail in native instead of being supported. Please either add native ORC compression support or avoid setting this config / avoid offload for unsupported ORC compression.

2. Parquet gzip is also mapped to a native config that is rejected. normalizeCompressionKind("GZIP") returns gzip, but the native filesystem sink only allows snappy/lz4/zstd for Parquet and throws for gzip. Hive/Flink tables configured with Parquet gzip would regress from Java writer behavior to native runtime failure. Please add gzip support or make planner fallback for unsupported codecs.

The added UT only checks that the map contains sink.file.compression; it does not cover whether the generated config is accepted by native for ORC or gzip. Please add coverage for these unsupported/native behavior cases as part of the fix.

[lgbo-ustc]

Additional clean-code feedback, excluding the unrelated logging change already discussed separately:

1. HiveSourceSinkFactory.addCompressionParamsFromTableProperties is doing two different jobs: copying compression-like table properties and deriving the native sink.file.compression config. These have different responsibilities and different compatibility constraints. Please split the behavior or narrow this method to only produce the native config that is actually needed.

2. isCompressionProperty is too broad. Matching any key that contains compress or codec makes the planner pass through unrelated or future table properties by accident. A clearer and safer approach is to maintain an explicit allow-list of supported Hive/Flink compression keys.

3. normalizeCompressionKind is too permissive. It uses substring matching and returns unknown normalized values as-is, which lets planner generate configs that native may reject later. It would be cleaner to explicitly map supported values and return empty/null, fallback, or reject for unsupported codecs.

4. The Hive writer factory reflection path is duplicated with the existing format-resolution path. Both depend on bucketsBuilder -> writerFactory -> HiveBulkWriterFactory -> factory. Please consider sharing one small helper so Flink internal field-name changes do not need to be fixed in multiple places.

5. The new unit tests are focused on helper implementation details rather than behavior. The test using orc.compress also makes the expected output look valid even though native filesystem compression currently does not support ORC. Please add behavior-oriented coverage for supported Parquet codecs and unsupported format/codec cases, verifying fallback/no config/error behavior explicitly.

[lgbo-ustc]

Clean-code suggestion: normalizeCompressionKind would be clearer as a static mapping table instead of a switch.

For example, keep a static map from lowercase input values to the canonical native codec name, then implement normalization as trim/lowercase + map lookup, returning null when the value is absent. This makes the supported codec set explicit, avoids a long switch, and makes it harder to accidentally pass through unsupported values later.

[lgbo-ustc]

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:

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.