This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new dda6861fb35 [fix](iceberg) Support GZIP for iceberg/hive Parquet
writers. (#68854)
dda6861fb35 is described below
commit dda6861fb35a4becb22b63f416506a4c9b477731
Author: daidai <[email protected]>
AuthorDate: Mon Oct 12 10:13:24 2026 +0800
[fix](iceberg) Support GZIP for iceberg/hive Parquet writers. (#68854)
### What problem does this PR solve?
Issue Number: N/A
Related PR: #64723
Problem Summary:
Iceberg tables with `write.parquet.compression-codec = gzip` (the
default before Iceberg 1.4) and Hive
tables with `parquet.compression = GZIP` could not be written by Doris:
the BE Parquet writers only
accepted PLAIN/SNAPPY/ZSTD/LZ4 and failed with
`[INTERNAL_ERROR]Unsupported compress type GZ with
parquet`, while Spark/Trino wrote the same table fine.
- BE: map `TFileCompressType.GZ` to `TParquetCompressionType.GZIP`, and
move the whole
`TFileCompressType` -> `TParquetCompressionType` mapping into
`ParquetBuildHelper` so the Iceberg
data-file writer, the Iceberg delete-file writer and the Hive Parquet
writer share one copy.
- FE: accept `gzip` for Parquet in the Hive `CREATE TABLE` compression
allow-list.
Tests: the Iceberg Parquet-compression suite now writes and reads gzip
data and merge-on-read delete
files and checks the codec with `parquet_meta()`; the Hive DDL suite
adds gzip to the Parquet
compression matrix; FE unit tests cover the gzip allow-list and the
codec mapping.
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [x] Regression test
- [x] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [x] Yes. <!-- Explain the behavior change --> Iceberg/Hive Parquet
writes with GZIP no longer fail; the file is written with the standard
Parquet GZIP codec, the same codec Spark/Trino write for that table
property.
- Does this need documentation?
- [x] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
.../writer/iceberg/viceberg_delete_file_writer.cpp | 22 +------
.../writer/iceberg/viceberg_partition_writer.cpp | 36 +++--------
be/src/exec/sink/writer/vhive_partition_writer.cpp | 27 +--------
be/src/format/transformer/vparquet_writer.cpp | 19 ++++++
be/src/format/transformer/vparquet_writer.h | 11 ++++
.../connector/hive/HiveWritePlanProviderTest.java | 4 ++
.../doris/connector/hms/HmsWriteConverter.java | 2 +-
.../doris/connector/hms/HmsWriteConverterTest.java | 11 ++++
.../hive/ddl/test_hive_ddl.groovy | 27 ++++++++-
.../test_iceberg_write_parquet_compression.groovy | 69 +++++++++++++++++++---
10 files changed, 147 insertions(+), 81 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..7f7f2c6f2f4 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"
@@ -82,25 +83,8 @@ Status VIcebergDeleteFileWriter::open(RuntimeState* state,
RuntimeProfile* profi
// 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));
ParquetFileOptions parquet_options = {parquet_compression_type,
TParquetVersion::PARQUET_1_0,
false, false};
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 abdde43e637..a4bacb3b01d 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"
@@ -75,36 +76,15 @@ Status VIcebergPartitionWriter::open(RuntimeState* state,
RuntimeProfile* profil
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;
+ auto parquet_compression_type =
+
ParquetBuildHelper::to_parquet_compression_type(_compress_type);
+ if (!parquet_compression_type.has_value()) {
+ // The target file is already created, so let the shared cleanup
below remove it
+ // instead of returning early.
+ open_status = parquet_compression_type.error();
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: {
- open_status = Status::InternalError("Unsupported compress type {}
with parquet",
- to_string(_compress_type));
- break;
- }
- }
- if (!open_status.ok()) {
- break;
- }
- ParquetFileOptions parquet_options = {.compression_type =
parquet_compression_type,
+ ParquetFileOptions parquet_options = {.compression_type =
parquet_compression_type.value(),
.parquet_version =
TParquetVersion::PARQUET_1_0,
.parquet_disable_dictionary =
false,
.enable_int96_timestamps =
false};
diff --git a/be/src/exec/sink/writer/vhive_partition_writer.cpp
b/be/src/exec/sink/writer/vhive_partition_writer.cpp
index 1ef5054e9e3..f28adde9004 100644
--- a/be/src/exec/sink/writer/vhive_partition_writer.cpp
+++ b/be/src/exec/sink/writer/vhive_partition_writer.cpp
@@ -25,6 +25,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_system.h"
#include "io/fs/s3_file_writer.h"
@@ -88,30 +89,8 @@ Status VHivePartitionWriter::open(RuntimeState* state,
RuntimeProfile* operator_
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));
// 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,
diff --git a/be/src/format/transformer/vparquet_writer.cpp
b/be/src/format/transformer/vparquet_writer.cpp
index f8aba44cf12..c5a06867b15 100644
--- a/be/src/format/transformer/vparquet_writer.cpp
+++ b/be/src/format/transformer/vparquet_writer.cpp
@@ -97,6 +97,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 8894fb71421..081d24e31b9 100644
--- a/be/src/format/transformer/vparquet_writer.h
+++ b/be/src/format/transformer/vparquet_writer.h
@@ -72,6 +72,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/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveWritePlanProviderTest.java
b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveWritePlanProviderTest.java
index 6f5f054556a..e4594916d66 100644
---
a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveWritePlanProviderTest.java
+++
b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveWritePlanProviderTest.java
@@ -388,6 +388,10 @@ public class HiveWritePlanProviderTest {
Assertions.assertEquals(TFileCompressType.SNAPPYBLOCK,
compressionFor(PARQUET_INPUT_FORMAT,
Collections.singletonMap("parquet.compression", "snappy"),
Collections.emptyMap()));
+ // GZIP is a valid parquet.compression value; BE maps GZ to the
parquet GZIP codec.
+ Assertions.assertEquals(TFileCompressType.GZ,
+ compressionFor(PARQUET_INPUT_FORMAT,
Collections.singletonMap("parquet.compression", "gzip"),
+ Collections.emptyMap()));
Assertions.assertEquals(TFileCompressType.GZ,
compressionFor(TEXT_INPUT_FORMAT,
Collections.singletonMap("text.compression", "gzip"),
Collections.emptyMap()));
diff --git
a/fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/HmsWriteConverter.java
b/fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/HmsWriteConverter.java
index 9abdb070e8b..1b8b640c8f1 100644
---
a/fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/HmsWriteConverter.java
+++
b/fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/HmsWriteConverter.java
@@ -62,7 +62,7 @@ public final class HmsWriteConverter {
private static final Set<String> SUPPORTED_ORC_COMPRESSIONS =
unmodifiableSet("plain", "zlib", "snappy", "zstd", "lz4");
private static final Set<String> SUPPORTED_PARQUET_COMPRESSIONS =
- unmodifiableSet("plain", "snappy", "zstd", "lz4");
+ unmodifiableSet("plain", "snappy", "gzip", "zstd", "lz4");
private static final Set<String> SUPPORTED_TEXT_COMPRESSIONS =
unmodifiableSet("plain", "gzip", "zstd", "bzip2", "lz4", "snappy");
diff --git
a/fe/fe-connector/fe-connector-hms/src/test/java/org/apache/doris/connector/hms/HmsWriteConverterTest.java
b/fe/fe-connector/fe-connector-hms/src/test/java/org/apache/doris/connector/hms/HmsWriteConverterTest.java
index 9ea4e0f75c5..663acce43db 100644
---
a/fe/fe-connector/fe-connector-hms/src/test/java/org/apache/doris/connector/hms/HmsWriteConverterTest.java
+++
b/fe/fe-connector/fe-connector-hms/src/test/java/org/apache/doris/connector/hms/HmsWriteConverterTest.java
@@ -144,6 +144,17 @@ public class HmsWriteConverterTest {
Assertions.assertFalse(table.getParameters().containsKey("compression"));
}
+ @Test
+ public void testExplicitParquetGzipCompression() {
+ // GZIP is a valid Hive `parquet.compression` value, and the Doris
Parquet writer now
+ // honours it, so the CREATE TABLE allow-list must accept it instead
of rejecting it.
+ Table table = HmsWriteConverter.toHiveTable(
+ baseTable("parquet", Collections.singletonList(col("id",
"INT")),
+
Collections.emptyList()).properties(mutableMap("compression", "gzip")).build());
+ Assertions.assertEquals("gzip",
table.getParameters().get("parquet.compression"));
+
Assertions.assertFalse(table.getParameters().containsKey("compression"));
+ }
+
@Test
public void testUnsupportedCompressionAndFormatThrow() {
Assertions.assertThrows(IllegalArgumentException.class, () ->
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 75f92730664..ce297554471 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") {
}
}
+ // 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` """;
}
@@ -754,7 +775,7 @@ suite("test_hive_ddl", "p0,external") {
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)
@@ -765,6 +786,10 @@ suite("test_hive_ddl", "p0,external") {
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]
