This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 3eebe290108 [fix](iceberg) Support GZIP for iceberg/hive Parquet
writers. (#68854) (#69338)
3eebe290108 is described below
commit 3eebe2901081b139ad6d87d24d884aa93a6b99eb
Author: daidai <[email protected]>
AuthorDate: Thu Oct 22 09:58:37 2026 +0800
[fix](iceberg) Support GZIP for iceberg/hive Parquet writers. (#68854)
(#69338)
pick #68854
---
.../writer/iceberg/viceberg_delete_file_writer.cpp | 28 +++------
.../writer/iceberg/viceberg_partition_writer.cpp | 46 ++++++---------
be/src/exec/sink/writer/vhive_partition_writer.cpp | 34 +++--------
be/src/format/transformer/vparquet_writer.cpp | 19 ++++++
be/src/format/transformer/vparquet_writer.h | 11 ++++
.../iceberg/iceberg_partition_writer_test.cpp | 28 +++++++++
.../org/apache/doris/datasource/hive/HiveUtil.java | 2 +-
.../apache/doris/datasource/hive/HiveUtilTest.java | 47 +++++++++++++++
.../hive/ddl/test_hive_ddl.groovy | 27 ++++++++-
.../test_iceberg_write_parquet_compression.groovy | 69 +++++++++++++++++++---
10 files changed, 228 insertions(+), 83 deletions(-)
diff --git a/be/src/exec/sink/writer/iceberg/viceberg_delete_file_writer.cpp
b/be/src/exec/sink/writer/iceberg/viceberg_delete_file_writer.cpp
index 8ad7751e1ed..5fb4fabb1ce 100644
--- a/be/src/exec/sink/writer/iceberg/viceberg_delete_file_writer.cpp
+++ b/be/src/exec/sink/writer/iceberg/viceberg_delete_file_writer.cpp
@@ -23,6 +23,7 @@
#include "format/table/iceberg/types.h"
#include "format/transformer/viceberg_parquet_writer.h"
#include "format/transformer/vorc_transformer.h"
+#include "format/transformer/vparquet_writer.h"
#include "io/file_factory.h"
#include "runtime/runtime_state.h"
@@ -77,30 +78,16 @@ Status VIcebergDeleteFileWriter::open(RuntimeState* state,
RuntimeProfile* profi
_fs = DORIS_TRY(FileFactory::create_fs(fs_properties, file_description));
io::FileWriterOptions file_writer_options = {.used_by_s3_committer =
false};
- RETURN_IF_ERROR(_fs->create_file(file_description.path, &_file_writer,
&file_writer_options));
+ auto create_file = [&]() {
+ return _fs->create_file(file_description.path, &_file_writer,
&file_writer_options);
+ };
// Create file format transformer based on format type
switch (_file_format) {
case TFileFormatType::FORMAT_PARQUET: {
- TParquetCompressionType::type parquet_compression_type;
- switch (_compress_type) {
- case TFileCompressType::PLAIN:
- parquet_compression_type = TParquetCompressionType::UNCOMPRESSED;
- break;
- case TFileCompressType::SNAPPYBLOCK:
- parquet_compression_type = TParquetCompressionType::SNAPPY;
- break;
- case TFileCompressType::ZSTD:
- parquet_compression_type = TParquetCompressionType::ZSTD;
- break;
- case TFileCompressType::LZ4BLOCK:
- // Hadoop-framed Parquet LZ4 (not LZ4_RAW) for cross-engine
compatibility.
- parquet_compression_type = TParquetCompressionType::LZ4_HADOOP;
- break;
- default:
- return Status::InternalError("Unsupported compress type {} with
parquet",
- to_string(_compress_type));
- }
+ TParquetCompressionType::type parquet_compression_type =
+
DORIS_TRY(ParquetBuildHelper::to_parquet_compression_type(_compress_type));
+ RETURN_IF_ERROR(create_file());
ParquetFileOptions parquet_options = {parquet_compression_type,
TParquetVersion::PARQUET_1_0,
false, false};
@@ -110,6 +97,7 @@ Status VIcebergDeleteFileWriter::open(RuntimeState* state,
RuntimeProfile* profi
return _file_format_transformer->open();
}
case TFileFormatType::FORMAT_ORC: {
+ RETURN_IF_ERROR(create_file());
_file_format_transformer.reset(new VOrcTransformer(state,
_file_writer.get(), output_exprs,
"", column_names,
false, _compress_type,
_position_delete_schema.get()));
diff --git a/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp
b/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp
index 565e47b4686..b84efb0bb06 100644
--- a/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp
+++ b/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp
@@ -25,6 +25,7 @@
#include "format/table/iceberg/schema.h"
#include "format/transformer/viceberg_parquet_writer.h"
#include "format/transformer/vorc_transformer.h"
+#include "format/transformer/vparquet_writer.h"
#include "io/file_factory.h"
#include "runtime/runtime_state.h"
@@ -93,35 +94,16 @@ Status VIcebergPartitionWriter::open(RuntimeState* state,
RuntimeProfile* profil
io::FileDescription file_description = {.path = _path, .fs_name {}};
_fs = DORIS_TRY(FileFactory::create_fs(fs_properties, file_description));
io::FileWriterOptions file_writer_options = {.used_by_s3_committer =
false};
- RETURN_IF_ERROR(_fs->create_file(file_description.path, &_file_writer,
&file_writer_options));
+ auto create_file = [&]() {
+ return _fs->create_file(file_description.path, &_file_writer,
&file_writer_options);
+ };
+ Status open_status;
switch (_file_format_type) {
case TFileFormatType::FORMAT_PARQUET: {
- TParquetCompressionType::type parquet_compression_type;
- switch (_compress_type) {
- case TFileCompressType::PLAIN: {
- parquet_compression_type = TParquetCompressionType::UNCOMPRESSED;
- break;
- }
- case TFileCompressType::SNAPPYBLOCK: {
- parquet_compression_type = TParquetCompressionType::SNAPPY;
- break;
- }
- case TFileCompressType::ZSTD: {
- parquet_compression_type = TParquetCompressionType::ZSTD;
- break;
- }
- case TFileCompressType::LZ4BLOCK: {
- // Map Doris LZ4 to the Hadoop-framed Parquet LZ4 codec (not
LZ4_RAW) so the file
- // stays readable across Spark/Iceberg/Trino. See
ParquetBuildHelper.
- parquet_compression_type = TParquetCompressionType::LZ4_HADOOP;
- break;
- }
- default: {
- return Status::InternalError("Unsupported compress type {} with
parquet",
- to_string(_compress_type));
- }
- }
+ TParquetCompressionType::type parquet_compression_type =
+
DORIS_TRY(ParquetBuildHelper::to_parquet_compression_type(_compress_type));
+ RETURN_IF_ERROR(create_file());
// Parquet 1.0 coerces Arrow nanoseconds to microseconds, including
nested fields.
ParquetFileOptions parquet_options = {
.compression_type = parquet_compression_type,
@@ -133,19 +115,27 @@ Status VIcebergPartitionWriter::open(RuntimeState* state,
RuntimeProfile* profil
_file_format_transformer = std::make_unique<VIcebergParquetWriter>(
state, _file_writer.get(), _write_output_expr_ctxs,
_write_column_names, false,
parquet_options, _iceberg_schema_json, _schema);
- return _file_format_transformer->open();
+ open_status = _file_format_transformer->open();
+ break;
}
case TFileFormatType::FORMAT_ORC: {
+ RETURN_IF_ERROR(create_file());
_file_format_transformer = std::make_unique<VOrcTransformer>(
state, _file_writer.get(), _write_output_expr_ctxs, "",
_write_column_names, false,
_compress_type, &_schema, _fs);
- return _file_format_transformer->open();
+ open_status = _file_format_transformer->open();
+ break;
}
default: {
return Status::InternalError("Unsupported file format type {}",
to_string(_file_format_type));
}
}
+ if (!open_status.ok()) {
+ // A transformer failure happens after object creation, so remove any
published file.
+ WARN_IF_ERROR(_fs->delete_file(_path), "failed to delete Iceberg file
after open error");
+ }
+ return open_status;
}
Status VIcebergPartitionWriter::close(const Status& status) {
diff --git a/be/src/exec/sink/writer/vhive_partition_writer.cpp
b/be/src/exec/sink/writer/vhive_partition_writer.cpp
index 6a116aa9254..a480853bd3f 100644
--- a/be/src/exec/sink/writer/vhive_partition_writer.cpp
+++ b/be/src/exec/sink/writer/vhive_partition_writer.cpp
@@ -24,6 +24,7 @@
#include "format/transformer/vcsv_transformer.h"
#include "format/transformer/vhive_parquet_writer.h"
#include "format/transformer/vorc_transformer.h"
+#include "format/transformer/vparquet_writer.h"
#include "io/file_factory.h"
#include "io/fs/s3_file_writer.h"
#include "runtime/runtime_state.h"
@@ -69,34 +70,15 @@ Status VHivePartitionWriter::open(RuntimeState* state,
RuntimeProfile* operator_
.fs_name {}};
_fs = DORIS_TRY(FileFactory::create_fs(fs_properties, file_description));
io::FileWriterOptions file_writer_options = {.used_by_s3_committer = true};
- RETURN_IF_ERROR(_fs->create_file(file_description.path, &_file_writer,
&file_writer_options));
+ auto create_file = [&]() {
+ return _fs->create_file(file_description.path, &_file_writer,
&file_writer_options);
+ };
switch (_file_format_type) {
case TFileFormatType::FORMAT_PARQUET: {
- TParquetCompressionType::type parquet_compression_type;
- switch (_hive_compress_type) {
- case TFileCompressType::PLAIN: {
- parquet_compression_type = TParquetCompressionType::UNCOMPRESSED;
- break;
- }
- case TFileCompressType::SNAPPYBLOCK: {
- parquet_compression_type = TParquetCompressionType::SNAPPY;
- break;
- }
- case TFileCompressType::ZSTD: {
- parquet_compression_type = TParquetCompressionType::ZSTD;
- break;
- }
- case TFileCompressType::LZ4BLOCK: {
- // Hadoop-framed Parquet LZ4 (not LZ4_RAW) for cross-engine
compatibility.
- parquet_compression_type = TParquetCompressionType::LZ4_HADOOP;
- break;
- }
- default: {
- return Status::InternalError("Unsupported hive compress type {}
with parquet",
- to_string(_hive_compress_type));
- }
- }
+ TParquetCompressionType::type parquet_compression_type =
+
DORIS_TRY(ParquetBuildHelper::to_parquet_compression_type(_hive_compress_type));
+ RETURN_IF_ERROR(create_file());
// TODO: INT96 is kept for Hive 2/3 compatibility. Add an explicit
option before
// changing the default Hive parquet timestamp encoding to standard
logical types.
ParquetFileOptions parquet_options = {parquet_compression_type,
@@ -113,12 +95,14 @@ Status VHivePartitionWriter::open(RuntimeState* state,
RuntimeProfile* operator_
return _file_format_transformer->open();
}
case TFileFormatType::FORMAT_ORC: {
+ RETURN_IF_ERROR(create_file());
_file_format_transformer = std::make_unique<VOrcTransformer>(
state, _file_writer.get(), _write_output_expr_ctxs, "",
_write_column_names, false,
_hive_compress_type);
return _file_format_transformer->open();
}
case TFileFormatType::FORMAT_CSV_PLAIN: {
+ RETURN_IF_ERROR(create_file());
// TODO(syt): support hive csv table, only hive text file is
supportted now
_file_format_transformer = std::make_unique<VCSVTransformer>(
state, _file_writer.get(), _write_output_expr_ctxs, false,
"csv", "",
diff --git a/be/src/format/transformer/vparquet_writer.cpp
b/be/src/format/transformer/vparquet_writer.cpp
index d019e43a8de..7ec66e56b28 100644
--- a/be/src/format/transformer/vparquet_writer.cpp
+++ b/be/src/format/transformer/vparquet_writer.cpp
@@ -108,6 +108,25 @@ void ParquetOutputStream::set_written_len(int64_t
written_len) {
_written_len = written_len;
}
+Result<TParquetCompressionType::type>
ParquetBuildHelper::to_parquet_compression_type(
+ TFileCompressType::type compression_type) {
+ switch (compression_type) {
+ case TFileCompressType::PLAIN:
+ return TParquetCompressionType::UNCOMPRESSED;
+ case TFileCompressType::SNAPPYBLOCK:
+ return TParquetCompressionType::SNAPPY;
+ case TFileCompressType::ZSTD:
+ return TParquetCompressionType::ZSTD;
+ case TFileCompressType::GZ:
+ return TParquetCompressionType::GZIP;
+ case TFileCompressType::LZ4BLOCK:
+ return TParquetCompressionType::LZ4_HADOOP;
+ default:
+ return ResultError(Status::InternalError("Unsupported compress type {}
with parquet",
+ to_string(compression_type)));
+ }
+}
+
void ParquetBuildHelper::build_compression_type(
::parquet::WriterProperties::Builder& builder,
const TParquetCompressionType::type& compression_type) {
diff --git a/be/src/format/transformer/vparquet_writer.h
b/be/src/format/transformer/vparquet_writer.h
index cbe0bf9d5f5..faf7c4aa22c 100644
--- a/be/src/format/transformer/vparquet_writer.h
+++ b/be/src/format/transformer/vparquet_writer.h
@@ -73,6 +73,17 @@ private:
class ParquetBuildHelper {
public:
+ // Maps the Doris file-level compression codec onto the Parquet codec to
write with, so the
+ // Iceberg/Hive writers share one definition instead of repeating the
switch:
+ // PLAIN -> UNCOMPRESSED, SNAPPYBLOCK -> SNAPPY, ZSTD -> ZSTD,
+ // GZ -> GZIP, LZ4BLOCK -> LZ4_HADOOP.
+ // GZ is what Iceberg's `write.parquet.compression-codec=gzip` (the table
default before
+ // Iceberg 1.4) and Hive's `parquet.compression=GZIP` resolve to, and GZIP
is the codec
+ // Spark/Trino write for those properties. LZ4_HADOOP is the Hadoop-framed
LZ4 codec, which
+ // keeps the file readable by engines that do not support LZ4_RAW.
+ static Result<TParquetCompressionType::type> to_parquet_compression_type(
+ TFileCompressType::type compression_type);
+
static void build_compression_type(::parquet::WriterProperties::Builder&
builder,
const TParquetCompressionType::type&
compression_type);
diff --git a/be/test/exec/sink/writer/iceberg/iceberg_partition_writer_test.cpp
b/be/test/exec/sink/writer/iceberg/iceberg_partition_writer_test.cpp
index d453177cf25..424d94bab29 100644
--- a/be/test/exec/sink/writer/iceberg/iceberg_partition_writer_test.cpp
+++ b/be/test/exec/sink/writer/iceberg/iceberg_partition_writer_test.cpp
@@ -20,6 +20,9 @@
#include <optional>
#include "exec/sink/writer/iceberg/viceberg_partition_writer.h"
+#include "io/fs/file_system.h"
+#include "runtime/runtime_state.h"
+#include "util/uid_util.h"
namespace doris {
@@ -104,4 +107,29 @@ TEST_F(VIcebergPartitionWriterTest,
MissingPolicyKeepsCollectionEnabledForRollin
EXPECT_TRUE(collect_column_stats(*writer));
}
+TEST_F(VIcebergPartitionWriterTest, UnsupportedParquetCodecDoesNotCreateFile) {
+ VExprContextSPtrs output_exprs;
+ iceberg::Schema schema(std::vector<iceberg::NestedField> {});
+ std::string schema_json;
+ std::map<std::string, std::string> hadoop_conf;
+ IPartitionWriterBase::WriteInfo write_info;
+ write_info.file_type = TFileType::FILE_LOCAL;
+ write_info.write_path = ".";
+ auto writer = std::make_unique<VIcebergPartitionWriter>(
+ make_table_sink(std::nullopt), std::vector<std::string> {},
output_exprs, schema,
+ &schema_json, std::vector<std::string> {}, std::move(write_info),
+ "unsupported_parquet_codec_" + UniqueId::gen_uid().to_string(), 0,
+ TFileFormatType::FORMAT_PARQUET, TFileCompressType::ZLIB,
hadoop_conf);
+
+ RuntimeState state;
+ Status status = writer->open(&state, nullptr, nullptr);
+
+ EXPECT_FALSE(status.ok());
+ EXPECT_NE(std::string::npos, status.to_string().find("Unsupported compress
type"));
+ ASSERT_NE(nullptr, writer->_fs);
+ bool exists = true;
+ ASSERT_TRUE(writer->_fs->exists(writer->_path, &exists).ok());
+ EXPECT_FALSE(exists);
+}
+
} // namespace doris
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveUtil.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveUtil.java
index f87f7a02c6c..01b90574456 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveUtil.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveUtil.java
@@ -68,7 +68,7 @@ public final class HiveUtil {
public static final Set<String> SUPPORTED_ORC_COMPRESSIONS =
ImmutableSet.of("plain", "zlib", "snappy", "zstd", "lz4");
public static final Set<String> SUPPORTED_PARQUET_COMPRESSIONS =
- ImmutableSet.of("plain", "snappy", "zstd", "lz4");
+ ImmutableSet.of("plain", "snappy", "gzip", "zstd", "lz4");
public static final Set<String> SUPPORTED_TEXT_COMPRESSIONS =
ImmutableSet.of("plain", "gzip", "zstd", "bzip2", "lz4", "snappy");
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/HiveUtilTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/HiveUtilTest.java
new file mode 100644
index 00000000000..45ff5f79c6d
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/HiveUtilTest.java
@@ -0,0 +1,47 @@
+// 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.doris.datasource.hive;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.PrimitiveType;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Maps;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+import java.util.Optional;
+
+public class HiveUtilTest {
+ @Test
+ public void testExplicitParquetGzipCompression() {
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(HiveUtil.COMPRESSION_KEY, "gzip");
+ HiveTableMetadata metadata = new HiveTableMetadata(
+ "db", "table", Optional.empty(),
+ ImmutableList.of(new Column("id", PrimitiveType.INT)),
+ ImmutableList.of(), properties, "parquet", "");
+
+ Table table = HiveUtil.toHiveTable(metadata);
+
+ Assertions.assertEquals("gzip",
table.getParameters().get("parquet.compression"));
+
Assertions.assertFalse(table.getParameters().containsKey(HiveUtil.COMPRESSION_KEY));
+ }
+}
diff --git
a/regression-test/suites/external_table_p0/hive/ddl/test_hive_ddl.groovy
b/regression-test/suites/external_table_p0/hive/ddl/test_hive_ddl.groovy
index 34efc5a86fe..5cf69868e1d 100644
--- a/regression-test/suites/external_table_p0/hive/ddl/test_hive_ddl.groovy
+++ b/regression-test/suites/external_table_p0/hive/ddl/test_hive_ddl.groovy
@@ -419,6 +419,27 @@ suite("test_hive_ddl",
"p0,external,hive,external_docker,external_docker_hive")
}
}
+ // GZIP used to be rejected by the Doris Parquet writer with
+ // "Unsupported compress type GZ with parquet"; the write must
succeed and land on the
+ // GZIP Parquet codec. Asserted instead of `qt_` because the
expectation is one codec name.
+ if (file_format.equals("parquet") && compression.equals("gzip")) {
+ sql """ INSERT INTO tbl_${file_format}_${compression} VALUES
('doris_gzip') """
+ def gzipRows = sql """ SELECT * FROM
tbl_${file_format}_${compression} ORDER BY col """
+ assertEquals(1, gzipRows.size())
+ assertEquals("doris_gzip", gzipRows[0][0].toString())
+ String hdfsPort =
context.config.otherConfigs.get("hive2HdfsPort")
+ String externalEnvIp =
context.config.otherConfigs.get("externalEnvIp")
+ def gzipCodec = sql """
+ SELECT DISTINCT compression
+ FROM parquet_meta(
+ "uri" =
"hdfs://${externalEnvIp}:${hdfsPort}/user/hive/warehouse/test_hive_compress.db/tbl_parquet_gzip/*",
+ "hadoop.username" = "doris",
+ "mode" = "parquet_metadata"
+ )
+ """
+ assertEquals("GZIP", gzipCodec[0][0].toString().toUpperCase())
+ }
+
sql """DROP TABLE `tbl_${file_format}_${compression}`"""
sql """ drop database if exists `test_hive_compress` """;
}
@@ -774,7 +795,7 @@ suite("test_hive_ddl",
"p0,external,hive,external_docker,external_docker_hive")
sql """set enable_fallback_to_original_planner=false;"""
test_db(catalog_name)
test_loc_db(externalEnvIp, hdfs_port, catalog_name)
- def compressions = ["snappy", "zlib", "zstd", "lz4"]
+ def compressions = ["snappy", "zlib", "zstd", "lz4", "gzip"]
for (String file_format in file_formats) {
logger.info("Process file format " + file_format)
test_loc_tbl(file_format, externalEnvIp, hdfs_port,
catalog_name)
@@ -785,6 +806,10 @@ suite("test_hive_ddl",
"p0,external,hive,external_docker,external_docker_hive")
if (file_format.equals("parquet") &&
compression.equals("zlib")) {
continue
}
+ // ORC has no GZIP codec (it spells the same codec
`zlib`), so only test it for Parquet.
+ if (file_format.equals("orc") &&
compression.equals("gzip")) {
+ continue
+ }
logger.info("Process file compression " + compression)
test_tbl_compress(compression, file_format, catalog_name)
}
diff --git
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_parquet_compression.groovy
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_parquet_compression.groovy
index a0bcdafbfd7..e0aea0458ae 100644
---
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_parquet_compression.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_parquet_compression.groovy
@@ -16,7 +16,8 @@
// under the License.
// Verifies that Doris writes Iceberg Parquet LZ4 as Hadoop-framed Parquet
"LZ4"
-// (not LZ4_RAW) and that Iceberg ORC/delete-file LZ4 paths are reachable.
+// (not LZ4_RAW), that Iceberg ORC/delete-file LZ4 paths are reachable, and
that
+// `write.parquet.compression-codec=gzip` is honoured by the Parquet
data/delete writers.
suite("test_iceberg_write_parquet_compression", "p0,external") {
String enabled = context.config.otherConfigs.get("enableIcebergTest")
if (enabled == null || !enabled.equalsIgnoreCase("true")) {
@@ -137,9 +138,10 @@ suite("test_iceberg_write_parquet_compression",
"p0,external") {
ORDER BY compression
"""
- // GZIP is intentionally not yet supported by the Doris Parquet writer;
the INSERT must
- // fail explicitly rather than silently fall back. Remove this block when
GZIP support is
- // added (and update the cross-engine product test expectation in tandem).
+ // GZIP is a standard Iceberg Parquet codec (the table default before
Iceberg 1.4) and used to
+ // be rejected by the Doris Parquet writer with "Unsupported compress type
GZ with parquet".
+ // The assertions below are intentionally not `qt_` cases: the expected
result is a single
+ // codec name, so an explicit check keeps the intent obvious.
sql """ drop table if exists ${db}.tbl_gzip """
sql """
CREATE TABLE ${db}.tbl_gzip (a STRING, b BIGINT) PROPERTIES (
@@ -148,8 +150,59 @@ suite("test_iceberg_write_parquet_compression",
"p0,external") {
'format-version' = '2',
'write.parquet.compression-codec' = 'gzip'
)"""
- test {
- sql """ INSERT INTO ${db}.tbl_gzip VALUES ('doris0', 0) """
- exception "Unsupported compress type GZ with parquet"
- }
+ sql """ INSERT INTO ${db}.tbl_gzip VALUES ('doris0', 0), ('doris1', 1) """
+ def gzipRows = sql """ SELECT a, b FROM ${db}.tbl_gzip ORDER BY b """
+ assertEquals(2, gzipRows.size())
+ assertEquals("doris1", gzipRows[1][0].toString())
+ def gzipDataFiles = sql """ SELECT file_path FROM ${db}.tbl_gzip\$files
ORDER BY file_path """
+ String gzipDataFile = gzipDataFiles[0][0].toString()
+ def gzipDataCodec = sql """
+ SELECT DISTINCT compression
+ FROM parquet_meta(
+ "uri" = "${gzipDataFile}",
+ "s3.access_key" = "admin",
+ "s3.secret_key" = "password",
+ "s3.endpoint" = "http://${externalEnvIp}:${minio_port}";,
+ "s3.region" = "us-east-1",
+ "mode" = "parquet_metadata"
+ )
+ """
+ assertEquals("GZIP", gzipDataCodec[0][0].toString().toUpperCase())
+
+ // The delete-file writer must honour the same table property, otherwise
DELETE/merge-on-read
+ // on a gzip table fails where INSERT succeeds.
+ sql """ drop table if exists ${db}.tbl_gzip_delete """
+ sql """
+ CREATE TABLE ${db}.tbl_gzip_delete (id INT, name STRING) PROPERTIES (
+ 'write-format' = 'parquet',
+ 'write.format.default' = 'parquet',
+ 'format-version' = '2',
+ 'write.parquet.compression-codec' = 'gzip',
+ 'write.delete.mode' = 'merge-on-read',
+ 'write.update.mode' = 'merge-on-read',
+ 'write.merge.mode' = 'merge-on-read'
+ )"""
+ sql """ INSERT INTO ${db}.tbl_gzip_delete VALUES (1, 'a'), (2, 'b'), (3,
'c') """
+ sql """ DELETE FROM ${db}.tbl_gzip_delete WHERE id = 2 """
+ def gzipDeleteRows = sql """ SELECT id, name FROM ${db}.tbl_gzip_delete
ORDER BY id """
+ assertEquals(2, gzipDeleteRows.size())
+ assertEquals("3", gzipDeleteRows[1][0].toString())
+ def gzipDeleteFiles = sql """
+ SELECT file_path
+ FROM ${db}.tbl_gzip_delete\$delete_files
+ ORDER BY file_path
+ """
+ String gzipDeleteFile = gzipDeleteFiles[0][0].toString()
+ def gzipDeleteCodec = sql """
+ SELECT DISTINCT compression
+ FROM parquet_meta(
+ "uri" = "${gzipDeleteFile}",
+ "s3.access_key" = "admin",
+ "s3.secret_key" = "password",
+ "s3.endpoint" = "http://${externalEnvIp}:${minio_port}";,
+ "s3.region" = "us-east-1",
+ "mode" = "parquet_metadata"
+ )
+ """
+ assertEquals("GZIP", gzipDeleteCodec[0][0].toString().toUpperCase())
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
