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 7f00c963484 [fix](hive) Keep compressed text files in whole scan
ranges (#69167)
7f00c963484 is described below
commit 7f00c96348437f945515897730a99ef1165f8c2a
Author: Gabriel <[email protected]>
AuthorDate: Fri Oct 16 15:53:54 2026 +0800
[fix](hive) Keep compressed text files in whole scan ranges (#69167)
### What problem does this PR solve?
`test_hive_compress_type_large_data` succeeds with the default file
split size but fails after setting `file_split_size=8388608`:
```
[INTERNAL_ERROR] For now we do not support split compressed file
```
Hive's planner used the table-level TEXT/CSV splittability for every
file. For example, a 20 MiB gzip file was assigned ranges starting at 0,
8 MiB and 16 MiB. Both native text/CSV readers reject compressed streams
starting at a nonzero offset. The same partition can contain plain and
compressed files, so a table-wide decision is insufficient.
Determine splittability per file in the common range iterator used by
synchronous, partition-batch and streaming scans. Recognized compressed
TEXT/CSV files now produce one complete range starting at zero. Plain
text still splits; ORC/Parquet keep their existing range planning.
Hadoop codec splittability is not used because the native readers have
different restart capabilities.
Move the existing filename-to-codec inference into a shared,
dependency-free foundation enum and delegate
`Util.inferFileCompressTypeByPath` to it. This keeps planning and the
codec sent to BE on the same suffix rules, without changing supported
suffixes, Hive's LZ4 adjustment, LZO sidecar filtering, or the BE safety
checks.
### Validation
- The new mixed-file planner test fails on the original code (33 ranges
instead of 13) and passes with the fix. It covers TEXT/CSV, ORC/Parquet,
all recognized compression suffixes and uppercase gzip, empty files,
plain-file offsets/lengths, and synchronous/batch/streaming enumeration.
- `mvn -f fe/pom.xml -pl :fe-connector-hive -am test
-Dmaven.build.cache.enabled=false`: 1,462 tests passed, zero
failures/errors/skips; 510 tests in the Hive module.
- Foundation tests cover unchanged codec recognition and unknown-suffix
behavior; connector tests verify the shared enum names map to native
Thrift codecs.
- The existing large-data regression retains its cardinality/filter
assertions, forces synchronous 8 MiB queries, and compares forced
streaming queries with the default-size results. Groovy syntax
compilation passed.
- FE Checkstyle and `git diff --check` passed.
A live Hive environment was not exercised locally. The external
regression and full FE integration build remain for CI.
### Release note
Fix Hive compressed-text queries failing when file split sizes are
smaller than the compressed files.
### Check List (For Author)
- [x] Unit Test
- [x] Regression test extension
- [x] Behavior changed: compressed native text files are read as whole
streams
- [x] No documentation needed
---
.../doris/connector/hive/HiveScanPlanProvider.java | 8 ++-
.../connector/hive/HiveScanBatchModeTest.java | 69 ++++++++++++++++++++++
.../java/org/apache/doris/common/util/Util.java | 22 +------
.../doris/foundation/format/FileCompression.java | 57 ++++++++++++++++++
.../foundation/format/FileCompressionTest.java | 46 +++++++++++++++
.../hive/test_hive_compress_type_large_data.groovy | 22 ++++++-
6 files changed, 200 insertions(+), 24 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java
b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java
index f3e4a916b38..18c4b142adc 100644
---
a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java
+++
b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java
@@ -37,6 +37,7 @@ import
org.apache.doris.connector.spi.scan.ConnectorSplitSource;
import org.apache.doris.connector.spi.scan.ScanNodePropertyKeys;
import org.apache.doris.filesystem.FileEntry;
import org.apache.doris.filesystem.FileSystem;
+import org.apache.doris.foundation.format.FileCompression;
import org.apache.doris.thrift.TFileCompressType;
import org.apache.doris.thrift.TFileScanRangeParams;
import org.apache.doris.thrift.TTableFormatFileDesc;
@@ -909,7 +910,12 @@ public class HiveScanPlanProvider implements
ConnectorScanPlanProvider {
return Collections.emptyIterator();
}
String normalizedPath = normalizeNativeUri(filePath);
- long splitSize = splittable && targetSplitSize > 0 ? targetSplitSize :
fileSize;
+ // Native text/CSV readers cannot restart compressed streams at
nonzero offsets, even for
+ // codecs Hadoop considers splittable. Use the same inference as the
codec sent to BE;
+ // columnar readers handle range boundaries independently of filename
compression suffixes.
+ boolean compressedText = (fileFormat == HiveFileFormat.TEXT ||
fileFormat == HiveFileFormat.CSV)
+ && FileCompression.fromPath(normalizedPath) !=
FileCompression.PLAIN;
+ long splitSize = splittable && !compressedText && targetSplitSize > 0
? targetSplitSize : fileSize;
return new Iterator<ConnectorScanRange>() {
private long offset;
diff --git
a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveScanBatchModeTest.java
b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveScanBatchModeTest.java
index 7af485014df..e4b45ceaa12 100644
---
a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveScanBatchModeTest.java
+++
b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveScanBatchModeTest.java
@@ -35,6 +35,7 @@ import
org.apache.doris.connector.spi.scan.ConnectorScanRequest;
import org.apache.doris.connector.spi.scan.ConnectorSplitSource;
import org.apache.doris.connector.spi.scan.ScanNodePropertyKeys;
import org.apache.doris.filesystem.FileSystem;
+import org.apache.doris.foundation.format.FileCompression;
import org.apache.doris.thrift.TFileCompressType;
import org.apache.doris.thrift.TFileScanRangeParams;
@@ -220,6 +221,74 @@ public class HiveScanBatchModeTest {
Assertions.assertEquals(Collections.singleton("year=2024/month=02"),
lister.callsPerLocation.keySet());
}
+ @Test
+ public void sharedCompressionNamesMapToNativeCodecs() {
+ for (FileCompression codec : FileCompression.values()) {
+ Assertions.assertDoesNotThrow(() ->
TFileCompressType.valueOf(codec.name()));
+ }
+ }
+
+ @Test
+ public void compressedTextFilesStayWholeInEveryPlanningPath() throws
Exception {
+ long splitSize = 8L * 1024 * 1024;
+ long fileSize = 2 * splitSize + 3;
+ List<String> extensions = Arrays.asList(".gz", ".bz2", ".lz4", ".lzo",
".lzo_deflate",
+ ".deflate", ".snappy", ".zst", ".zstd", ".GZ");
+ String partition = "year=2024/month=01";
+ String[] inputs = {TEXT_INPUT_FORMAT, TEXT_INPUT_FORMAT,
PARQUET_INPUT_FORMAT,
+ "org.apache.hadoop.hive.ql.io.orc.OrcInputFormat"};
+ String[] serdes = {LAZY_SIMPLE_SERDE, OPEN_CSV_SERDE, PARQUET_SERDE,
+ "org.apache.hadoop.hive.ql.io.orc.OrcSerde"};
+ for (int format = 0; format < inputs.length; format++) {
+ HiveScanPlanProvider provider = provider(new FakeHmsClient(),
(location, fs) -> {
+ List<HiveFileStatus> files = new ArrayList<>();
+ for (String extension : extensions) {
+ files.add(new HiveFileStatus(location + "/data" +
extension, fileSize, 1));
+ }
+ files.add(new HiveFileStatus(location + "/plain", fileSize,
1));
+ files.add(new HiveFileStatus(location + "/empty.gz", 0, 1));
+ return files;
+ });
+ HiveTableHandle handle = parquetHandle(PART_KEYS).toBuilder()
+
.inputFormat(inputs[format]).serializationLib(serdes[format])
+
.prunedPartitions(Collections.singletonList(part(partition))).build();
+ ConnectorScanRequest request = scanOf(handle);
+ FakeSession session = new
FakeSession(Collections.singletonMap("file_split_size", "" + splitSize));
+ List<ConnectorScanRange> synchronous = provider.planScan(session,
request);
+ List<ConnectorScanRange> batch =
provider.planScanForPartitionBatch(session, request,
+ Collections.singletonList(partition));
+ List<ConnectorScanRange> streaming = new ArrayList<>();
+ try (ConnectorSplitSource source = provider.streamSplits(session,
request)) {
+ while (source.hasNext()) {
+ streaming.add(source.next());
+ }
+ }
+ for (List<ConnectorScanRange> ranges : Arrays.asList(synchronous,
batch, streaming)) {
+ boolean text = format < 2;
+ Assertions.assertEquals(text ? extensions.size() + 3 :
(extensions.size() + 1) * 3,
+ ranges.size(), "only plain text and columnar files
support byte-range splitting");
+ for (String extension : extensions) {
+ List<ConnectorScanRange> fileRanges = new ArrayList<>();
+ for (ConnectorScanRange range : ranges) {
+ if (range.getPath().get().equals(partition + "/data" +
extension)) {
+ fileRanges.add(range);
+ }
+ }
+ Assertions.assertEquals(text ? 1 : 3, fileRanges.size());
+ Assertions.assertEquals(0, fileRanges.get(0).getStart());
+ Assertions.assertEquals(text ? fileSize : splitSize,
fileRanges.get(0).getLength());
+ }
+ int plainStart = ranges.size() - 3;
+ for (int i = 0; i < 3; i++) {
+ ConnectorScanRange range = ranges.get(plainStart + i);
+ Assertions.assertEquals(partition + "/plain",
range.getPath().get());
+ Assertions.assertEquals(i * splitSize, range.getStart());
+ Assertions.assertEquals(i == 2 ? 3 : splitSize,
range.getLength());
+ }
+ }
+ }
+ }
+
@Test
public void
partitionBatchAndStreamingMatchAcrossFormatsAndListingFailures() throws
Exception {
List<String> batch = Arrays.asList("year=2024/month=01",
"year=2024/month=02", "year=2024/month=03");
diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/util/Util.java
b/fe/fe-core/src/main/java/org/apache/doris/common/util/Util.java
index 8679c9029e5..dbe14be0833 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/util/Util.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/util/Util.java
@@ -23,6 +23,7 @@ import org.apache.doris.cloud.security.SecurityChecker;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.FeNameFormat;
import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.foundation.format.FileCompression;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.thrift.TFileCompressType;
import org.apache.doris.thrift.TFileFormatType;
@@ -444,26 +445,7 @@ public class Util {
@NotNull
public static TFileCompressType inferFileCompressTypeByPath(String path) {
- String lowerCasePath = path.toLowerCase();
- if (lowerCasePath.endsWith(".gz")) {
- return TFileCompressType.GZ;
- } else if (lowerCasePath.endsWith(".bz2")) {
- return TFileCompressType.BZ2;
- } else if (lowerCasePath.endsWith(".lz4")) {
- return TFileCompressType.LZ4FRAME;
- } else if (lowerCasePath.endsWith(".lzo")) {
- return TFileCompressType.LZOP;
- } else if (lowerCasePath.endsWith(".lzo_deflate")) {
- return TFileCompressType.LZO;
- } else if (lowerCasePath.endsWith(".deflate")) {
- return TFileCompressType.DEFLATE;
- } else if (lowerCasePath.endsWith(".snappy")) {
- return TFileCompressType.SNAPPYBLOCK;
- } else if (lowerCasePath.endsWith(".zst") ||
lowerCasePath.endsWith(".zstd")) {
- return TFileCompressType.ZSTD;
- } else {
- return TFileCompressType.PLAIN;
- }
+ return
TFileCompressType.valueOf(FileCompression.fromPath(path).name());
}
public static TFileCompressType getFileCompressType(String compressType)
throws AnalysisException {
diff --git
a/fe/fe-foundation/src/main/java/org/apache/doris/foundation/format/FileCompression.java
b/fe/fe-foundation/src/main/java/org/apache/doris/foundation/format/FileCompression.java
new file mode 100644
index 00000000000..ba0bff59dd0
--- /dev/null
+++
b/fe/fe-foundation/src/main/java/org/apache/doris/foundation/format/FileCompression.java
@@ -0,0 +1,57 @@
+// 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.foundation.format;
+
+import java.util.Locale;
+
+/** Filename compression inference shared by native-reader planning and
FE-to-BE codec selection. */
+public enum FileCompression {
+ PLAIN,
+ GZ,
+ BZ2,
+ LZ4FRAME,
+ LZOP,
+ LZO,
+ DEFLATE,
+ SNAPPYBLOCK,
+ ZSTD;
+
+ // Names match TFileCompressType while keeping foundation independent of
the Thrift module.
+ public static FileCompression fromPath(String path) {
+ String lowerCasePath = path.toLowerCase(Locale.ROOT);
+ if (lowerCasePath.endsWith(".gz")) {
+ return GZ;
+ } else if (lowerCasePath.endsWith(".bz2")) {
+ return BZ2;
+ } else if (lowerCasePath.endsWith(".lz4")) {
+ return LZ4FRAME;
+ } else if (lowerCasePath.endsWith(".lzo")) {
+ return LZOP;
+ } else if (lowerCasePath.endsWith(".lzo_deflate")) {
+ return LZO;
+ } else if (lowerCasePath.endsWith(".deflate")) {
+ return DEFLATE;
+ } else if (lowerCasePath.endsWith(".snappy")) {
+ return SNAPPYBLOCK;
+ } else if (lowerCasePath.endsWith(".zst") ||
lowerCasePath.endsWith(".zstd")) {
+ return ZSTD;
+ } else {
+ return PLAIN;
+ }
+ }
+}
diff --git
a/fe/fe-foundation/src/test/java/org/apache/doris/foundation/format/FileCompressionTest.java
b/fe/fe-foundation/src/test/java/org/apache/doris/foundation/format/FileCompressionTest.java
new file mode 100644
index 00000000000..55f318cef5b
--- /dev/null
+++
b/fe/fe-foundation/src/test/java/org/apache/doris/foundation/format/FileCompressionTest.java
@@ -0,0 +1,46 @@
+// 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.foundation.format;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Locale;
+
+public class FileCompressionTest {
+ @Test
+ public void recognizesNativeCompressionSuffixes() {
+ String[] extensions = {"gz", "bz2", "lz4", "lzo", "lzo_deflate",
"deflate", "snappy", "zst", "zstd"};
+ FileCompression[] codecs = {FileCompression.GZ, FileCompression.BZ2,
FileCompression.LZ4FRAME,
+ FileCompression.LZOP, FileCompression.LZO,
FileCompression.DEFLATE,
+ FileCompression.SNAPPYBLOCK, FileCompression.ZSTD,
FileCompression.ZSTD};
+ for (int i = 0; i < extensions.length; i++) {
+ Assertions.assertEquals(codecs[i],
FileCompression.fromPath("s3://bucket/data." + extensions[i]));
+ Assertions.assertEquals(codecs[i],
+ FileCompression.fromPath("hdfs://nameservice/data." +
extensions[i].toUpperCase(Locale.ROOT)));
+ }
+ }
+
+ @Test
+ public void preservesPlainAndUnknownSuffixBehavior() {
+ for (String path : new String[] {"", "data", "data.csv",
"data.parquet", "data.orc", "data.gzip",
+ "dir.gz/data", "data.gz.tmp", "data.gz?key=value",
"data.gz#fragment"}) {
+ Assertions.assertEquals(FileCompression.PLAIN,
FileCompression.fromPath(path), path);
+ }
+ }
+}
diff --git
a/regression-test/suites/external_table_p2/hive/test_hive_compress_type_large_data.groovy
b/regression-test/suites/external_table_p2/hive/test_hive_compress_type_large_data.groovy
index 20236ab7722..042b5439608 100644
---
a/regression-test/suites/external_table_p2/hive/test_hive_compress_type_large_data.groovy
+++
b/regression-test/suites/external_table_p2/hive/test_hive_compress_type_large_data.groovy
@@ -62,13 +62,29 @@ suite("test_hive_compress_type_large_data", "p2,external") {
}
}
- def countMix2 = sql """select count(*) from test_compress_partitioned
where dt="gzip" or dt="mix""""
+ def countMix2 = sql """select /*+
SET_VAR(enable_external_table_batch_mode=false) */ count(*)
+ from test_compress_partitioned where dt='gzip' or dt='mix'"""
assertEquals(600005, countMix2[0][0])
- def countAll2 = sql """select count(*) from
test_compress_partitioned"""
+ def countAll2 = sql """select /*+
SET_VAR(enable_external_table_batch_mode=false) */ count(*)
+ from test_compress_partitioned"""
assertEquals(1510010, countAll2[0][0])
- def countWatchId2 = sql """select count(*) from
test_compress_partitioned where watchid=4611870011201662970"""
+ def countWatchId2 = sql """select /*+
SET_VAR(enable_external_table_batch_mode=false) */ count(*)
+ from test_compress_partitioned where
watchid=4611870011201662970"""
assertEquals(15, countWatchId2[0][0])
+ // Force the streaming producer too: mixed directories must keep
compressed files whole
+ // while plain files still split, regardless of which planner
enumerates the ranges.
+ def streamingHint = "/*+
SET_VAR(enable_external_table_batch_mode=true, " +
+ "num_files_in_batch_mode=1, num_partitions_in_batch_mode=1) */"
+ def countMixStreaming = sql """select ${streamingHint} count(*) from
test_compress_partitioned
+ where dt='gzip' or dt='mix'"""
+ assertEquals(countMix1, countMixStreaming)
+ def countAllStreaming = sql """select ${streamingHint} count(*) from
test_compress_partitioned"""
+ assertEquals(countAll1, countAllStreaming)
+ def countWatchIdStreaming = sql """select ${streamingHint} count(*)
from test_compress_partitioned
+ where watchid=4611870011201662970"""
+ assertEquals(countWatchId1, countWatchIdStreaming)
+
sql """set file_split_size=0"""
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
