This is an automated email from the ASF dual-hosted git repository.
taiyangli pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 3add1fb66d [GLUTEN-7659][CH] Implement splittable bzip2 decompression
(#7638)
3add1fb66d is described below
commit 3add1fb66d4f786a711cae26dda32970683f20c3
Author: 李扬 <[email protected]>
AuthorDate: Fri Oct 25 16:22:24 2024 +0800
[GLUTEN-7659][CH] Implement splittable bzip2 decompression (#7638)
* clean some codes
* finish refactor
* special process for first block
* finish last block special process
* minor refactors
* finish dev
* enable bzip2 file split
* fix bugs
* fix failed uts
* fix failed uts in hdfs suite
---
.../backendsapi/clickhouse/CHValidatorApi.scala | 2 +-
cpp-ch/local-engine/CMakeLists.txt | 4 +-
cpp-ch/local-engine/Common/GlutenConfig.cpp | 13 +-
cpp-ch/local-engine/Common/GlutenConfig.h | 9 -
.../SparkFunctionDecimalBinaryArithmetic.cpp | 6 +-
.../local-engine/IO/SplittableBzip2ReadBuffer.cpp | 957 +++++++++++++++++++++
cpp-ch/local-engine/IO/SplittableBzip2ReadBuffer.h | 306 +++++++
cpp-ch/local-engine/Parser/CHColumnToSparkRow.cpp | 4 +-
cpp-ch/local-engine/Parser/CHColumnToSparkRow.h | 4 +-
.../SubstraitSource/ExcelTextFormatFile.cpp | 2 +-
.../Storages/SubstraitSource/FormatFile.h | 1 -
.../Storages/SubstraitSource/JSONFormatFile.cpp | 2 +-
.../Storages/SubstraitSource/ReadBufferBuilder.cpp | 244 ++++--
.../Storages/SubstraitSource/ReadBufferBuilder.h | 13 +-
.../Storages/SubstraitSource/TextFormatFile.cpp | 2 +-
15 files changed, 1462 insertions(+), 107 deletions(-)
diff --git
a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHValidatorApi.scala
b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHValidatorApi.scala
index 6081b87e1d..5fe7406946 100644
---
a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHValidatorApi.scala
+++
b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHValidatorApi.scala
@@ -67,7 +67,7 @@ class CHValidatorApi extends ValidatorApi with
AdaptiveSparkPlanHelper with Logg
/** Validate whether the compression method support splittable at clickhouse
backend. */
override def doCompressionSplittableValidate(compressionMethod: String):
Boolean = {
- false
+ compressionMethod == "BZip2Codec"
}
override def doColumnarShuffleExchangeExecValidate(
diff --git a/cpp-ch/local-engine/CMakeLists.txt
b/cpp-ch/local-engine/CMakeLists.txt
index 03e17ec311..4392b55008 100644
--- a/cpp-ch/local-engine/CMakeLists.txt
+++ b/cpp-ch/local-engine/CMakeLists.txt
@@ -63,6 +63,7 @@ add_headers_and_sources(jni jni)
add_headers_and_sources(aggregate_functions AggregateFunctions)
add_headers_and_sources(disks Disks)
add_headers_and_sources(disks Disks/ObjectStorages)
+add_headers_and_sources(io IO)
include_directories(
${JNI_INCLUDE_DIRS}
@@ -101,7 +102,8 @@ add_library(
${operator_sources}
${aggregate_functions_sources}
${jni_sources}
- ${disks_sources})
+ ${disks_sources}
+ ${io_sources})
target_link_libraries(
gluten_clickhouse_backend_libs
diff --git a/cpp-ch/local-engine/Common/GlutenConfig.cpp
b/cpp-ch/local-engine/Common/GlutenConfig.cpp
index 212a87dcfe..44d77cf372 100644
--- a/cpp-ch/local-engine/Common/GlutenConfig.cpp
+++ b/cpp-ch/local-engine/Common/GlutenConfig.cpp
@@ -76,6 +76,7 @@ GraceMergingAggregateConfig
GraceMergingAggregateConfig::loadFromContext(const D
=
context->getConfigRef().getDouble(MAX_ALLOWED_MEMORY_USAGE_RATIO_FOR_AGGREGATE_MERGING,
0.9);
return config;
}
+
StreamingAggregateConfig StreamingAggregateConfig::loadFromContext(const
DB::ContextPtr & context)
{
StreamingAggregateConfig config;
@@ -88,6 +89,7 @@ StreamingAggregateConfig
StreamingAggregateConfig::loadFromContext(const DB::Con
config.enable_streaming_aggregating =
context->getConfigRef().getBool(ENABLE_STREAMING_AGGREGATING, true);
return config;
}
+
JoinConfig JoinConfig::loadFromContext(const DB::ContextPtr & context)
{
JoinConfig config;
@@ -96,6 +98,7 @@ JoinConfig JoinConfig::loadFromContext(const DB::ContextPtr &
context)
=
context->getConfigRef().getUInt64(MULTI_JOIN_ON_CLAUSES_BUILD_SIDE_ROWS_LIMIT,
10000000);
return config;
}
+
ExecutorConfig ExecutorConfig::loadFromContext(const DB::ContextPtr & context)
{
ExecutorConfig config;
@@ -103,16 +106,6 @@ ExecutorConfig ExecutorConfig::loadFromContext(const
DB::ContextPtr & context)
config.use_local_format =
context->getConfigRef().getBool(USE_LOCAL_FORMAT, false);
return config;
}
-HdfsConfig HdfsConfig::loadFromContext(const Poco::Util::AbstractConfiguration
& config, const DB::ReadSettings & read_settings)
-{
- HdfsConfig hdfs;
- if (read_settings.enable_filesystem_cache)
- hdfs.hdfs_async = false;
- else
- hdfs.hdfs_async = config.getBool(HDFS_ASYNC, true);
-
- return hdfs;
-}
S3Config S3Config::loadFromContext(const DB::ContextPtr & context)
{
diff --git a/cpp-ch/local-engine/Common/GlutenConfig.h
b/cpp-ch/local-engine/Common/GlutenConfig.h
index 3ff8f205a1..82402eaafa 100644
--- a/cpp-ch/local-engine/Common/GlutenConfig.h
+++ b/cpp-ch/local-engine/Common/GlutenConfig.h
@@ -108,15 +108,6 @@ struct ExecutorConfig
static ExecutorConfig loadFromContext(const DB::ContextPtr & context);
};
-struct HdfsConfig
-{
- inline static const String HDFS_ASYNC = "hdfs.enable_async_io";
-
- bool hdfs_async;
-
- static HdfsConfig loadFromContext(const Poco::Util::AbstractConfiguration
& config, const DB::ReadSettings & read_settings);
-};
-
struct S3Config
{
inline static const String S3_LOCAL_CACHE_ENABLE =
"s3.local_cache.enabled";
diff --git
a/cpp-ch/local-engine/Functions/SparkFunctionDecimalBinaryArithmetic.cpp
b/cpp-ch/local-engine/Functions/SparkFunctionDecimalBinaryArithmetic.cpp
index a2d5dec739..830fc0e652 100644
--- a/cpp-ch/local-engine/Functions/SparkFunctionDecimalBinaryArithmetic.cpp
+++ b/cpp-ch/local-engine/Functions/SparkFunctionDecimalBinaryArithmetic.cpp
@@ -279,7 +279,7 @@ private:
const ScaleDataType & scale_b,
ColumnUInt8::Container & vec_null_map_to,
const ResultDataType & resultDataType,
- const size_t & max_scale)
+ size_t max_scale)
{
using NativeResultType = NativeType<typename
ResultDataType::FieldType>;
@@ -357,7 +357,7 @@ private:
const ScaleDataType & scale_right,
NativeResultType & res,
const ResultDataType & resultDataType,
- const size_t & max_scale)
+ size_t max_scale)
{
if constexpr (CalculateWith256)
return calculateImpl<Int256>(l, r, scale_left, scale_right, res,
resultDataType, max_scale);
@@ -375,7 +375,7 @@ private:
const ScaleDataType & scale_right,
NativeResultType & res,
const ResultDataType & resultDataType,
- const size_t & max_scale)
+ size_t max_scale)
{
CalcType scaled_l = applyScaled(static_cast<CalcType>(l),
static_cast<CalcType>(scale_left));
CalcType scaled_r = applyScaled(static_cast<CalcType>(r),
static_cast<CalcType>(scale_right));
diff --git a/cpp-ch/local-engine/IO/SplittableBzip2ReadBuffer.cpp
b/cpp-ch/local-engine/IO/SplittableBzip2ReadBuffer.cpp
new file mode 100644
index 0000000000..701918e38d
--- /dev/null
+++ b/cpp-ch/local-engine/IO/SplittableBzip2ReadBuffer.cpp
@@ -0,0 +1,957 @@
+/*
+ * 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.
+ */
+
+#include "SplittableBzip2ReadBuffer.h"
+
+#if USE_BZIP2
+#include <IO/SeekableReadBuffer.h>
+#include <IO/VarInt.h>
+#include <base/find_symbols.h>
+
+
+namespace DB
+{
+
+namespace ErrorCodes
+{
+extern const int LOGICAL_ERROR;
+extern const int POSITION_OUT_OF_BOUND;
+}
+
+std::vector<Int32> & SplittableBzip2ReadBuffer::Data::initTT(Int32 length)
+{
+ if (tt.size() < static_cast<size_t>(length))
+ tt.resize(length);
+ return tt;
+}
+
+template <typename T>
+std::string SplittableBzip2ReadBuffer::Data::arrayToString(const
std::vector<T> & arr)
+{
+ std::string result = "[";
+ for (size_t i = 0; i < arr.size(); i++)
+ {
+ if (i)
+ result += ", ";
+
+ result += std::to_string(static_cast<Int32>(arr[i]));
+ }
+ result += "]";
+ return result;
+}
+
+template <typename T>
+std::string SplittableBzip2ReadBuffer::Data::array2DToString(T
arr[BZip2Constants::N_GROUPS][BZip2Constants::MAX_ALPHA_SIZE])
+{
+ std::string result = "[";
+ for (int i = 0; i < BZip2Constants::N_GROUPS; i++)
+ {
+ if (i)
+ result += ", ";
+
+ result += arrayToString(arr[i], BZip2Constants::MAX_ALPHA_SIZE);
+ }
+ result += "]";
+ return result;
+}
+
+template <typename T>
+std::string SplittableBzip2ReadBuffer::Data::arrayToString(const T * arr,
size_t size)
+{
+ std::string result = "[";
+ for (size_t i = 0; i < size; i++)
+ {
+ if (i)
+ result += ", ";
+
+ if constexpr (std::is_same_v<T, bool>)
+ result += arr[i] ? "true" : "false";
+ else
+ result += std::to_string(static_cast<Int32>(arr[i]));
+ }
+ result += "]";
+ return result;
+}
+
+std::string SplittableBzip2ReadBuffer::Data::toString()
+{
+ std::string result = "Data{";
+ result += "\ninUse=" + arrayToString(inUse, 256);
+ result += "\nseqToUnseq=" + arrayToString(seqToUnseq, 256);
+ result += "\nselector=" + arrayToString(selector,
BZip2Constants::MAX_SELECTORS);
+ result += "\nselectorMtf=" + arrayToString(selectorMtf,
BZip2Constants::MAX_SELECTORS);
+ result += "\nunzftab=" + arrayToString(unzftab, 256);
+ result += "\nlimit=" + array2DToString(limit);
+ result += "\nbase=" + array2DToString(base);
+ result += "\nperm=" + array2DToString(perm);
+ result += "\nminLens=" + arrayToString(minLens, BZip2Constants::N_GROUPS);
+ result += "\ncftab=" + arrayToString(cftab, 257);
+ result += "\ngetAndMoveToFrontDecode_yy=" +
arrayToString(getAndMoveToFrontDecode_yy, 256);
+ result += "\ntemp_charArray2d=" + array2DToString(temp_charArray2d);
+ result += "\nrecvDecodingTables_pos=" +
arrayToString(recvDecodingTables_pos, BZip2Constants::N_GROUPS);
+ result += "\ntt=" + arrayToString(tt);
+ result += "\nll8=" + arrayToString(ll8);
+ result += "}";
+ return result;
+}
+
+
+void SplittableBzip2ReadBuffer::hbCreateDecodeTables(
+ int * __restrict limit,
+ int * __restrict base,
+ int * __restrict perm,
+ const UInt16 * __restrict length,
+ int minLen,
+ int maxLen,
+ int alphaSize)
+{
+ for (int i = minLen, pp = 0; i <= maxLen; i++)
+ {
+ for (int j = 0; j < alphaSize; j++)
+ if (length[j] == i)
+ perm[pp++] = j;
+ }
+ for (int i = BZip2Constants::MAX_CODE_LEN - 1; i > 0; --i)
+ {
+ base[i] = 0;
+ limit[i] = 0;
+ }
+
+ for (int i = 0; i < alphaSize; i++)
+ base[length[i] + 1]++;
+
+ for (int i = 1, b = base[0]; i < BZip2Constants::MAX_CODE_LEN; i++)
+ {
+ b += base[i];
+ base[i] = b;
+ }
+
+ for (int i = minLen, vec = 0, b = base[i]; i <= maxLen; i++)
+ {
+ int nb = base[i + 1];
+ vec += nb - b;
+ b = nb;
+ limit[i] = vec - 1;
+ vec <<= 1;
+ }
+
+ for (int i = minLen + 1; i <= maxLen; i++)
+ base[i] = ((limit[i - 1] + 1) << 1) - base[i];
+}
+
+SplittableBzip2ReadBuffer::SplittableBzip2ReadBuffer(
+ std::unique_ptr<ReadBuffer> in_,
+ bool first_block_need_special_process_,
+ bool last_block_need_special_process_,
+ size_t buf_size,
+ char * existing_memory,
+ size_t alignment)
+ : CompressedReadBufferWrapper(std::move(in_), buf_size, existing_memory,
alignment)
+ , first_block_need_special_process(first_block_need_special_process_)
+ , last_block_need_special_process(last_block_need_special_process_)
+ , is_first_block(true)
+ , blockSize100k(9)
+ , currentState(STATE::NO_PROCESS_STATE)
+ , skipResult(false)
+ , currentChar(0)
+ , storedBlockCRC(0)
+ , blockRandomised(false)
+ , data(nullptr)
+ , computedBlockCRC(0)
+ , storedCombinedCRC(0)
+ , computedCombinedCRC(0)
+ , origPtr(0)
+ , nInUse(0)
+ , bsBuff(0)
+ , bsLive(0)
+ , last(0)
+{
+ auto * seekable = dynamic_cast<SeekableReadBuffer*>(in.get());
+ skipResult = skipToNextMarker(BLOCK_DELIMITER, DELIMITER_BIT_LENGTH);
+ if (seekable && skipResult)
+ {
+ /// Update adjusted_start
+ adjusted_start = seekable->getPosition();
+ }
+ changeStateToProcessABlock();
+}
+
+Int32 SplittableBzip2ReadBuffer::read(char * dest, size_t dest_size, size_t
offs, size_t len)
+{
+ if (offs + len > dest_size)
+ throw Exception(ErrorCodes::POSITION_OUT_OF_BOUND, "offs({}) + len({})
> dest_size({}).", offs, len, dest_size);
+
+ const size_t hi = offs + len;
+ size_t destOffs = offs;
+ Int32 b = 0;
+ for (; (destOffs < hi && (b = read0()) >= 0); ++destOffs)
+ {
+ dest[destOffs] = static_cast<char>(b);
+ }
+
+ Int32 result = static_cast<Int32>(destOffs - offs);
+ if (result == 0)
+ {
+ result = b;
+ skipResult =
skipToNextMarker(SplittableBzip2ReadBuffer::BLOCK_DELIMITER,
DELIMITER_BIT_LENGTH);
+ changeStateToProcessABlock();
+ }
+ return result;
+}
+
+bool SplittableBzip2ReadBuffer::nextImpl()
+{
+ Position dest = internal_buffer.begin();
+ size_t dest_size = internal_buffer.size();
+ size_t offset = 0;
+ Int32 result;
+ do
+ {
+ result = read(dest, dest_size, offset, dest_size - offset);
+ if (result > 0)
+ offset += result;
+ else if (result == BZip2Constants::END_OF_BLOCK && is_first_block &&
first_block_need_special_process)
+ {
+ /// Special processing for the first block
+ /// Notice that row delim could be \n (Unix) or \r\n (DOS/Windows)
or \n\r (Mac OS Classic)
+ is_first_block = false;
+ Position end = dest + offset;
+ auto * pos = find_last_symbols_or_null<'\n'>(dest, end);
+ if (pos)
+ {
+ if (pos == end - 1 || (pos == end - 2 && *(pos + 1) == '\r'))
+ {
+ /// The last row ends with \n or \r\n or \n\r, discard all
lines in internal buffer
+ offset = 0;
+ }
+ else
+ {
+ /// The last row does not end with \n or \r\n or \n\r,
rewrite the last row to internal buffer
+ Position last_line = pos + 1;
+ size_t last_line_size = end - pos - 1;
+ if (*(pos + 1) == '\r')
+ last_line_size--;
+
+ memmove(dest, last_line, last_line_size);
+ offset = last_line_size;
+ }
+ }
+ }
+ else if (result == BZip2Constants::END_OF_STREAM &&
last_block_need_special_process)
+ {
+ /// Special processing for the last block
+ Position end = dest + offset;
+ auto * pos = find_last_symbols_or_null<'\n'>(dest, end);
+
+ if (!pos)
+ {
+ /// Only one incomplete row in the last block, discard it
+ offset = 0;
+ }
+ else
+ {
+ /// Discard the last incomplete row(if there is) in last block.
+ offset = pos - dest + 1;
+ if (pos + 1 < end && *(pos + 1) == '\r')
+ offset++;
+ }
+ }
+ } while (result != BZip2Constants::END_OF_STREAM && offset < dest_size);
+
+ if (offset)
+ {
+ working_buffer.resize(offset);
+ return true;
+ }
+ else
+ {
+ return false;
+ }
+}
+
+Int32 SplittableBzip2ReadBuffer::read0()
+{
+ Int32 retChar = currentChar;
+
+ switch (currentState)
+ {
+ case STATE::END_OF_FILE:
+ return BZip2Constants::END_OF_STREAM;
+ case STATE::NO_PROCESS_STATE:
+ return BZip2Constants::END_OF_BLOCK;
+ case STATE::START_BLOCK_STATE:
+ throw Exception(ErrorCodes::LOGICAL_ERROR, "Wrong state {}",
magic_enum::enum_name(currentState));
+ case STATE::RAND_PART_A_STATE:
+ throw Exception(ErrorCodes::LOGICAL_ERROR, "Wrong state {}",
magic_enum::enum_name(currentState));
+ case STATE::RAND_PART_B_STATE:
+ setupRandPartB();
+ break;
+ case STATE::RAND_PART_C_STATE:
+ setupRandPartC();
+ break;
+ case STATE::NO_RAND_PART_A_STATE:
+ throw Exception(ErrorCodes::LOGICAL_ERROR, "Wrong state {}",
magic_enum::enum_name(currentState));
+ case STATE::NO_RAND_PART_B_STATE:
+ setupNoRandPartB();
+ break;
+ case STATE::NO_RAND_PART_C_STATE:
+ setupNoRandPartC();
+ break;
+ }
+ return retChar;
+}
+
+Int32 SplittableBzip2ReadBuffer::readAByte(ReadBuffer & in_)
+{
+ char c;
+ if (in_.read(c))
+ return static_cast<Int32>(c) & 0xff;
+ else
+ return -1;
+}
+
+bool SplittableBzip2ReadBuffer::skipToNextMarker(Int64 marker, Int32
markerBitLength, ReadBuffer & in_, Int64 & bsBuff_, Int64 & bsLive_)
+{
+ try
+ {
+ if (markerBitLength > 63)
+ throw Exception(ErrorCodes::LOGICAL_ERROR, "skipToNextMarker can
not find patterns greater than 63 bits");
+
+ Int64 bytes = bsR(markerBitLength, in_, bsBuff_, bsLive_);
+ if (bytes == -1)
+ {
+ return false;
+ }
+
+ while (true)
+ {
+ if (bytes == marker)
+ {
+ return true;
+ }
+ else
+ {
+ bytes = bytes << 1;
+ bytes = bytes & ((1L << markerBitLength) - 1);
+ Int32 oneBit = static_cast<Int32>(bsR(1, in_, bsBuff_,
bsLive_));
+ if (oneBit != -1)
+ {
+ bytes = bytes | oneBit;
+ }
+ else
+ {
+ return false;
+ }
+ }
+ }
+ }
+ catch (const Exception &)
+ {
+ return false;
+ }
+}
+
+bool SplittableBzip2ReadBuffer::skipToNextMarker(Int64 marker, Int32
markerBitLength)
+{
+ return skipToNextMarker(marker, markerBitLength, *in, bsBuff, bsLive);
+}
+
+void SplittableBzip2ReadBuffer::reportCRCError()
+{
+ throw Exception(ErrorCodes::LOGICAL_ERROR, "CRC error");
+}
+
+void SplittableBzip2ReadBuffer::makeMaps()
+{
+ Int32 nInUseShadow = 0;
+ for (Int32 i = 0; i < 256; i++)
+ if (data->inUse[i])
+ data->seqToUnseq[nInUseShadow++] = i;
+ nInUse = nInUseShadow;
+}
+
+void SplittableBzip2ReadBuffer::changeStateToProcessABlock()
+{
+ if (skipResult == true)
+ {
+ initBlock();
+ setupBlock();
+ }
+ else
+ {
+ currentState = STATE::END_OF_FILE;
+ }
+}
+
+void SplittableBzip2ReadBuffer::initBlock()
+{
+ storedBlockCRC = bsGetInt();
+ blockRandomised = (bsR(1) == 1);
+
+ if (!data)
+ data = std::make_unique<Data>(blockSize100k);
+
+ getAndMoveToFrontDecode();
+ crc.initialiseCRC();
+ currentState = STATE::START_BLOCK_STATE;
+}
+
+void SplittableBzip2ReadBuffer::endBlock()
+{
+ computedBlockCRC = crc.getFinalCRC();
+ if (storedBlockCRC != computedBlockCRC)
+ {
+ computedCombinedCRC = (storedCombinedCRC << 1) |
(static_cast<UInt32>(storedCombinedCRC) >> 31);
+ computedCombinedCRC ^= storedBlockCRC;
+ reportCRCError();
+ }
+ computedCombinedCRC = (computedCombinedCRC << 1) |
(static_cast<UInt32>(computedCombinedCRC) >> 31);
+ computedCombinedCRC ^= computedBlockCRC;
+}
+
+void SplittableBzip2ReadBuffer::complete()
+{
+ storedCombinedCRC = bsGetInt();
+ currentState = STATE::END_OF_FILE;
+ data = nullptr;
+ if (storedCombinedCRC != computedCombinedCRC)
+ reportCRCError();
+}
+
+Int64 SplittableBzip2ReadBuffer::bsR(Int64 n, ReadBuffer & in_, Int64 &
bsBuff_, Int64 & bsLive_)
+{
+ Int64 bsLiveShadow = bsLive_;
+ Int64 bsBuffShadow = bsBuff_;
+ if (bsLiveShadow < n)
+ {
+ do
+ {
+ Int32 thech = readAByte(in_);
+ if (thech < 0)
+ DB::throwReadAfterEOF();
+
+ bsBuffShadow = (bsBuffShadow << 8) | thech;
+ bsLiveShadow += 8;
+ } while (bsLiveShadow < n);
+
+ bsBuff_ = bsBuffShadow;
+ }
+
+ bsLive_ = bsLiveShadow - n;
+ return (bsBuffShadow >> (bsLiveShadow - n)) & ((1L << n) - 1);
+}
+
+Int64 SplittableBzip2ReadBuffer::bsR(Int64 n)
+{
+ return bsR(n, *in, bsBuff, bsLive);
+}
+
+bool SplittableBzip2ReadBuffer::bsGetBit()
+{
+ Int64 bsLiveShadow = bsLive;
+ Int64 bsBuffShadow = bsBuff;
+ if (bsLiveShadow < 1)
+ {
+ Int32 thech = readAByte(*in);
+ if (thech < 0)
+ DB::throwReadAfterEOF();
+
+ bsBuffShadow = (bsBuffShadow << 8) | thech;
+ bsLiveShadow += 8;
+ bsBuff = bsBuffShadow;
+ }
+ bsLive = bsLiveShadow - 1;
+ return ((bsBuffShadow >> (bsLiveShadow - 1)) & 1) != 0;
+}
+
+void SplittableBzip2ReadBuffer::recvDecodingTables()
+{
+ Data * dataShadow = data.get();
+ bool * inUse = dataShadow->inUse;
+ char * pos = dataShadow->recvDecodingTables_pos;
+ char * selector = dataShadow->selector;
+ char * selectorMtf = dataShadow->selectorMtf;
+
+ Int32 inUse16 = 0;
+ for (Int32 i = 0; i < 16; ++i)
+ if (bsGetBit())
+ inUse16 |= 1 << i;
+
+ for (Int32 i = 255; i >= 0; --i)
+ inUse[i] = false;
+
+ for (Int32 i = 0; i < 16; ++i)
+ {
+ if ((inUse16 & (1 << i)) != 0)
+ {
+ Int32 i16 = i << 4;
+ for (Int32 j = 0; j < 16; j++)
+ if (bsGetBit())
+ inUse[i16 + j] = true;
+ }
+ }
+ makeMaps();
+
+ Int32 alphaSize = nInUse + 2;
+ Int32 nGroups = static_cast<Int32>(bsR(3));
+ Int32 nSelectors = static_cast<Int32>(bsR(15));
+ for (Int32 i = 0; i < nSelectors; ++i)
+ {
+ Int32 j = 0;
+ while (bsGetBit())
+ j++;
+ selectorMtf[i] = j;
+ }
+
+ for (Int32 v = nGroups - 1; v >= 0; --v)
+ pos[v] = v;
+
+ for (Int32 i = 0; i < nSelectors; ++i)
+ {
+ Int32 v = selectorMtf[i] & 0xff;
+ char tmp = pos[v];
+ while (v > 0)
+ {
+ pos[v] = pos[v - 1];
+ v--;
+ }
+ pos[0] = tmp;
+ selector[i] = tmp;
+ }
+
+ auto * len = dataShadow->temp_charArray2d;
+ for (Int32 t = 0; t < nGroups; t++)
+ {
+ Int32 curr = static_cast<Int32>(bsR(5));
+ auto * len_t = len[t];
+ for (Int32 i = 0; i < alphaSize; i++)
+ {
+ while (bsGetBit())
+ curr += bsGetBit() ? -1 : 1;
+ len_t[i] = curr;
+ }
+ }
+
+ createHuffmanDecodingTables(alphaSize, nGroups);
+}
+
+void SplittableBzip2ReadBuffer::createHuffmanDecodingTables(Int32 alphaSize,
Int32 nGroups)
+{
+ Data * dataShadow = data.get();
+ auto * len = dataShadow->temp_charArray2d;
+ auto * minLens = dataShadow->minLens;
+ auto * limit = dataShadow->limit;
+ auto * base = dataShadow->base;
+ auto * perm = dataShadow->perm;
+ for (Int32 t = 0; t < nGroups; t++)
+ {
+ Int32 minLen = 32;
+ Int32 maxLen = 0;
+ auto * len_t = len[t];
+ for (Int32 i = alphaSize - 1; i >= 0; --i)
+ {
+ Int32 lent = len_t[i];
+ if (lent > maxLen)
+ maxLen = lent;
+ if (lent < minLen)
+ minLen = lent;
+ }
+ hbCreateDecodeTables(limit[t], base[t], perm[t], len[t], minLen,
maxLen, alphaSize);
+ minLens[t] = minLen;
+ }
+}
+
+void SplittableBzip2ReadBuffer::getAndMoveToFrontDecode()
+{
+ origPtr = static_cast<Int32>(bsR(24));
+ recvDecodingTables();
+
+ ReadBuffer * inShadow = in.get();
+ Data * dataShadow = data.get();
+ auto & ll8 = dataShadow->ll8;
+ Int32 * unzftab = dataShadow->unzftab;
+ char * selector = dataShadow->selector;
+ auto * seqToUnseq = dataShadow->seqToUnseq;
+ auto * yy = dataShadow->getAndMoveToFrontDecode_yy;
+ Int32 * minLens = dataShadow->minLens;
+ auto * limit = dataShadow->limit;
+ auto * base = dataShadow->base;
+ auto * perm = dataShadow->perm;
+ Int32 limitLast = blockSize100k * 100000;
+
+ for (Int32 i = 256; --i >= 0;)
+ {
+ yy[i] = i;
+ unzftab[i] = 0;
+ }
+
+ Int32 groupNo = 0;
+ Int32 groupPos = BZip2Constants::G_SIZE - 1;
+ Int32 eob = nInUse + 1;
+ Int32 nextSym = getAndMoveToFrontDecode0(0);
+ Int32 bsBuffShadow = static_cast<Int32>(bsBuff);
+ Int32 bsLiveShadow = static_cast<Int32>(bsLive);
+ Int32 lastShadow = -1;
+ Int32 zt = selector[groupNo] & 0xff;
+ Int32 * base_zt = base[zt];
+ Int32 * limit_zt = limit[zt];
+ Int32 * perm_zt = perm[zt];
+ Int32 minLens_zt = minLens[zt];
+
+ while (nextSym != eob)
+ {
+ if ((nextSym == BZip2Constants::RUNA) || (nextSym ==
BZip2Constants::RUNB))
+ {
+ Int32 s = -1;
+ for (Int32 n = 1; true; n <<= 1)
+ {
+ if (nextSym == BZip2Constants::RUNA)
+ s += n;
+ else if (nextSym == BZip2Constants::RUNB)
+ s += n << 1;
+ else
+ break;
+
+ if (groupPos == 0)
+ {
+ groupPos = BZip2Constants::G_SIZE - 1;
+ zt = selector[++groupNo] & 0xff;
+ base_zt = base[zt];
+ limit_zt = limit[zt];
+ perm_zt = perm[zt];
+ minLens_zt = minLens[zt];
+ }
+ else
+ {
+ groupPos--;
+ }
+
+ Int32 zn = minLens_zt;
+ while (bsLiveShadow < zn)
+ {
+ Int32 thech = readAByte(*inShadow);
+ if (thech < 0)
+ DB::throwReadAfterEOF();
+
+ bsBuffShadow = (bsBuffShadow << 8) | thech;
+ bsLiveShadow += 8;
+ }
+
+ Int64 zvec = (bsBuffShadow >> (bsLiveShadow - zn)) & ((1 <<
zn) - 1);
+ bsLiveShadow -= zn;
+ while (zvec > limit_zt[zn])
+ {
+ zn++;
+ while (bsLiveShadow < 1)
+ {
+ Int32 thech = readAByte(*inShadow);
+ if (thech < 0)
+ DB::throwReadAfterEOF();
+
+ bsBuffShadow = (bsBuffShadow << 8) | thech;
+ bsLiveShadow += 8;
+ }
+ bsLiveShadow--;
+ zvec = (zvec << 1) | ((bsBuffShadow >> bsLiveShadow) & 1);
+ }
+ nextSym = perm_zt[static_cast<Int32>(zvec - base_zt[zn])];
+ }
+
+ char ch = seqToUnseq[yy[0]];
+ unzftab[ch & 0xff] += s + 1;
+ while (s-- >= 0)
+ ll8[++lastShadow] = ch;
+ if (lastShadow >= limitLast)
+ throw Exception(ErrorCodes::LOGICAL_ERROR, "Block overrun");
+ }
+ else
+ {
+ if (++lastShadow >= limitLast)
+ throw Exception(ErrorCodes::LOGICAL_ERROR, "Block overrun");
+ auto tmp = yy[nextSym - 1];
+ unzftab[seqToUnseq[tmp] & 0xff]++;
+ ll8[lastShadow] = seqToUnseq[tmp];
+
+ if (nextSym <= 16)
+ for (Int32 j = nextSym - 1; j > 0; --j)
+ yy[j] = yy[j - 1];
+ else
+ memmove(&yy[1], &yy[0], (nextSym - 1) * sizeof(yy[0]));
+ yy[0] = tmp;
+
+ if (groupPos == 0)
+ {
+ groupPos = BZip2Constants::G_SIZE - 1;
+ zt = selector[++groupNo] & 0xff;
+ base_zt = base[zt];
+ limit_zt = limit[zt];
+ perm_zt = perm[zt];
+ minLens_zt = minLens[zt];
+ }
+ else
+ {
+ groupPos--;
+ }
+
+ Int32 zn = minLens_zt;
+ while (bsLiveShadow < zn)
+ {
+ Int32 thech = readAByte(*inShadow);
+ if (thech < 0)
+ DB::throwReadAfterEOF();
+
+ bsBuffShadow = (bsBuffShadow << 8) | thech;
+ bsLiveShadow += 8;
+ }
+
+ Int32 zvec = (bsBuffShadow >> (bsLiveShadow - zn)) & ((1 << zn) -
1);
+ bsLiveShadow -= zn;
+ while (zvec > limit_zt[zn])
+ {
+ zn++;
+ while (bsLiveShadow < 1)
+ {
+ Int32 thech = readAByte(*inShadow);
+ if (thech < 0)
+ DB::throwReadAfterEOF();
+
+ bsBuffShadow = (bsBuffShadow << 8) | thech;
+ bsLiveShadow += 8;
+ }
+ bsLiveShadow--;
+ zvec = ((zvec << 1) | ((bsBuffShadow >> bsLiveShadow) & 1));
+ }
+ nextSym = perm_zt[zvec - base_zt[zn]];
+ }
+ }
+
+ last = lastShadow;
+ bsLive = bsLiveShadow;
+ bsBuff = bsBuffShadow;
+}
+
+Int32 SplittableBzip2ReadBuffer::getAndMoveToFrontDecode0(Int32 groupNo)
+{
+ ReadBuffer * inShadow = in.get();
+ Data * dataShadow = data.get();
+ Int32 zt = dataShadow->selector[groupNo] & 0xff;
+ Int32 * limit_zt = dataShadow->limit[zt];
+ Int32 zn = dataShadow->minLens[zt];
+ Int32 zvec = static_cast<Int32>(bsR(zn));
+ Int32 bsLiveShadow = static_cast<Int32>(bsLive);
+ Int32 bsBuffShadow = static_cast<Int32>(bsBuff);
+ while (zvec > limit_zt[zn])
+ {
+ zn++;
+ while (bsLiveShadow < 1)
+ {
+ Int32 thech = readAByte(*inShadow);
+ if (thech < 0)
+ DB::throwReadAfterEOF();
+
+ bsBuffShadow = (bsBuffShadow << 8) | thech;
+ bsLiveShadow += 8;
+ }
+ bsLiveShadow--;
+ zvec = (zvec << 1) | ((bsBuffShadow >> bsLiveShadow) & 1);
+ }
+
+ bsLive = bsLiveShadow;
+ bsBuff = bsBuffShadow;
+ return dataShadow->perm[zt][zvec - dataShadow->base[zt][zn]];
+}
+
+void SplittableBzip2ReadBuffer::setupBlock()
+{
+ if (!data)
+ return;
+
+ Int32 * cftab = data->cftab;
+ std::vector<Int32> & tt = data->initTT(last + 1);
+ auto & ll8 = data->ll8;
+ cftab[0] = 0;
+ memcpy(&cftab[1], &data->unzftab[0], 256 * sizeof(cftab[0]));
+ for (Int32 i = 1, c = cftab[0]; i <= 256; i++)
+ {
+ c += cftab[i];
+ cftab[i] = c;
+ }
+ for (Int32 i = 0, lastShadow = last; i <= lastShadow; i++)
+ tt[cftab[ll8[i] & 0xff]++] = i;
+
+ if (origPtr < 0 || static_cast<size_t>(origPtr) >= tt.size())
+ throw Exception(ErrorCodes::LOGICAL_ERROR, "Stream corrupted");
+
+ su_tPos = tt[origPtr];
+ su_count = 0;
+ su_i2 = 0;
+ su_ch2 = 256;
+ if (blockRandomised)
+ {
+ su_rNToGo = 0;
+ su_rTPos = 0;
+ setupRandPartA();
+ }
+ else
+ {
+ setupNoRandPartA();
+ }
+}
+
+void SplittableBzip2ReadBuffer::setupRandPartA()
+{
+ if (su_i2 <= last)
+ {
+ su_chPrev = su_ch2;
+ Int32 su_ch2Shadow = data->ll8[su_tPos] & 0xff;
+ su_tPos = data->tt[su_tPos];
+ if (su_rNToGo == 0)
+ {
+ su_rNToGo = BZip2Constants::rNums[su_rTPos] - 1;
+ if (++su_rTPos == 512)
+ su_rTPos = 0;
+ }
+ else
+ {
+ su_rNToGo--;
+ }
+ su_ch2 = ((su_ch2Shadow ^= (su_rNToGo == 1)) ? 1 : 0);
+ su_i2++;
+ currentChar = su_ch2Shadow;
+ currentState = STATE::RAND_PART_B_STATE;
+ crc.updateCRC(su_ch2Shadow);
+ }
+ else
+ {
+ endBlock();
+ currentState = STATE::NO_PROCESS_STATE;
+ }
+}
+
+void SplittableBzip2ReadBuffer::setupNoRandPartA()
+{
+ if (su_i2 <= last)
+ {
+ su_chPrev = su_ch2;
+ Int32 su_ch2Shadow = data->ll8[su_tPos] & 0xff;
+ su_ch2 = su_ch2Shadow;
+ su_tPos = data->tt[su_tPos];
+ su_i2++;
+ currentChar = su_ch2Shadow;
+ currentState = STATE::NO_RAND_PART_B_STATE;
+ crc.updateCRC(su_ch2Shadow);
+ }
+ else
+ {
+ currentState = STATE::NO_RAND_PART_A_STATE;
+ endBlock();
+ currentState = STATE::NO_PROCESS_STATE;
+ }
+}
+
+void SplittableBzip2ReadBuffer::setupRandPartB()
+{
+ if (su_ch2 != su_chPrev)
+ {
+ currentState = STATE::RAND_PART_A_STATE;
+ su_count = 1;
+ setupRandPartA();
+ }
+ else if (++su_count >= 4)
+ {
+ su_z = static_cast<char>(data->ll8[su_tPos] & 0xff);
+ su_tPos = data->tt[su_tPos];
+ if (su_rNToGo == 0)
+ {
+ su_rNToGo = BZip2Constants::rNums[su_rTPos] - 1;
+ if (++su_rTPos == 512)
+ su_rTPos = 0;
+ }
+ else
+ {
+ su_rNToGo--;
+ }
+ su_j2 = 0;
+ currentState = STATE::RAND_PART_C_STATE;
+ if (su_rNToGo == 1)
+ su_z ^= 1;
+ setupRandPartC();
+ }
+ else
+ {
+ currentState = STATE::RAND_PART_A_STATE;
+ setupRandPartA();
+ }
+}
+
+void SplittableBzip2ReadBuffer::setupRandPartC()
+{
+ if (su_j2 < su_z)
+ {
+ currentChar = su_ch2;
+ crc.updateCRC(su_ch2);
+ su_j2++;
+ }
+ else
+ {
+ currentState = STATE::RAND_PART_A_STATE;
+ su_i2++;
+ su_count = 0;
+ setupRandPartA();
+ }
+}
+
+void SplittableBzip2ReadBuffer::setupNoRandPartB()
+{
+ if (su_ch2 != su_chPrev)
+ {
+ su_count = 1;
+ setupNoRandPartA();
+ }
+ else if (++su_count >= 4)
+ {
+ su_z = static_cast<char>(data->ll8[su_tPos] & 0xff);
+ su_tPos = data->tt[su_tPos];
+ su_j2 = 0;
+ setupNoRandPartC();
+ }
+ else
+ {
+ setupNoRandPartA();
+ }
+}
+
+void SplittableBzip2ReadBuffer::setupNoRandPartC()
+{
+ if (su_j2 < su_z)
+ {
+ Int32 su_ch2Shadow = su_ch2;
+ currentChar = su_ch2Shadow;
+ crc.updateCRC(su_ch2Shadow);
+ su_j2++;
+ currentState = STATE::NO_RAND_PART_C_STATE;
+ }
+ else
+ {
+ su_i2++;
+ su_count = 0;
+ setupNoRandPartA();
+ }
+}
+
+}
+#endif
diff --git a/cpp-ch/local-engine/IO/SplittableBzip2ReadBuffer.h
b/cpp-ch/local-engine/IO/SplittableBzip2ReadBuffer.h
new file mode 100644
index 0000000000..e5702c7a7a
--- /dev/null
+++ b/cpp-ch/local-engine/IO/SplittableBzip2ReadBuffer.h
@@ -0,0 +1,306 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include "config.h"
+
+#if USE_BZIP2
+#include <vector>
+#include <IO/CompressedReadBufferWrapper.h>
+
+namespace DB
+{
+
+class BZip2Constants
+{
+public:
+ static constexpr Int32 BASE_BLOCK_SIZE = 100000;
+ static constexpr Int32 MAX_ALPHA_SIZE = 258;
+ static constexpr Int32 MAX_CODE_LEN = 23;
+ static constexpr Int32 RUNA = 0;
+ static constexpr Int32 RUNB = 1;
+ static constexpr Int32 N_GROUPS = 6;
+ static constexpr Int32 G_SIZE = 50;
+ static constexpr Int32 N_ITERS = 4;
+ static constexpr Int32 MAX_SELECTORS = (2 + (900000 / G_SIZE));
+ static constexpr Int32 NUM_OVERSHOOT_BYTES = 20;
+ static constexpr Int32 END_OF_BLOCK = -2;
+ static constexpr Int32 END_OF_STREAM = -1;
+ static constexpr Int32 rNums[]
+ = {619, 720, 127, 481, 931, 816, 813, 233, 566, 247, 985, 724, 205,
454, 863, 491, 741, 242, 949, 214, 733, 859, 335, 708, 621, 574,
+ 73, 654, 730, 472, 419, 436, 278, 496, 867, 210, 399, 680, 480,
51, 878, 465, 811, 169, 869, 675, 611, 697, 867, 561, 862, 687,
+ 507, 283, 482, 129, 807, 591, 733, 623, 150, 238, 59, 379, 684,
877, 625, 169, 643, 105, 170, 607, 520, 932, 727, 476, 693, 425,
+ 174, 647, 73, 122, 335, 530, 442, 853, 695, 249, 445, 515, 909,
545, 703, 919, 874, 474, 882, 500, 594, 612, 641, 801, 220, 162,
+ 819, 984, 589, 513, 495, 799, 161, 604, 958, 533, 221, 400, 386,
867, 600, 782, 382, 596, 414, 171, 516, 375, 682, 485, 911, 276,
+ 98, 553, 163, 354, 666, 933, 424, 341, 533, 870, 227, 730, 475,
186, 263, 647, 537, 686, 600, 224, 469, 68, 770, 919, 190, 373,
+ 294, 822, 808, 206, 184, 943, 795, 384, 383, 461, 404, 758, 839,
887, 715, 67, 618, 276, 204, 918, 873, 777, 604, 560, 951, 160,
+ 578, 722, 79, 804, 96, 409, 713, 940, 652, 934, 970, 447, 318,
353, 859, 672, 112, 785, 645, 863, 803, 350, 139, 93, 354, 99,
+ 820, 908, 609, 772, 154, 274, 580, 184, 79, 626, 630, 742, 653,
282, 762, 623, 680, 81, 927, 626, 789, 125, 411, 521, 938, 300,
+ 821, 78, 343, 175, 128, 250, 170, 774, 972, 275, 999, 639, 495,
78, 352, 126, 857, 956, 358, 619, 580, 124, 737, 594, 701, 612,
+ 669, 112, 134, 694, 363, 992, 809, 743, 168, 974, 944, 375, 748,
52, 600, 747, 642, 182, 862, 81, 344, 805, 988, 739, 511, 655,
+ 814, 334, 249, 515, 897, 955, 664, 981, 649, 113, 974, 459, 893,
228, 433, 837, 553, 268, 926, 240, 102, 654, 459, 51, 686, 754,
+ 806, 760, 493, 403, 415, 394, 687, 700, 946, 670, 656, 610, 738,
392, 760, 799, 887, 653, 978, 321, 576, 617, 626, 502, 894, 679,
+ 243, 440, 680, 879, 194, 572, 640, 724, 926, 56, 204, 700, 707,
151, 457, 449, 797, 195, 791, 558, 945, 679, 297, 59, 87, 824,
+ 713, 663, 412, 693, 342, 606, 134, 108, 571, 364, 631, 212, 174,
643, 304, 329, 343, 97, 430, 751, 497, 314, 983, 374, 822, 928,
+ 140, 206, 73, 263, 980, 736, 876, 478, 430, 305, 170, 514, 364,
692, 829, 82, 855, 953, 676, 246, 369, 970, 294, 750, 807, 827,
+ 150, 790, 288, 923, 804, 378, 215, 828, 592, 281, 565, 555, 710,
82, 896, 831, 547, 261, 524, 462, 293, 465, 502, 56, 661, 821,
+ 976, 991, 658, 869, 905, 758, 745, 193, 768, 550, 608, 933, 378,
286, 215, 979, 792, 961, 61, 688, 793, 644, 986, 403, 106, 366,
+ 905, 644, 372, 567, 466, 434, 645, 210, 389, 550, 919, 135, 780,
773, 635, 389, 707, 100, 626, 958, 165, 504, 920, 176, 193, 713,
+ 857, 265, 203, 50, 668, 108, 645, 990, 626, 197, 510, 357, 358,
850, 858, 364, 936, 638};
+};
+
+
+class SplittableBzip2ReadBuffer : public CompressedReadBufferWrapper
+{
+public:
+ static constexpr Int64 BLOCK_DELIMITER = 0X314159265359L;
+ static constexpr Int64 EOS_DELIMITER = 0X177245385090L;
+ static constexpr Int32 DELIMITER_BIT_LENGTH = 48;
+
+private:
+ struct Data
+ {
+ bool inUse[256] = {false};
+
+ char seqToUnseq[256] = {0};
+ char selector[BZip2Constants::MAX_SELECTORS] = {0};
+ char selectorMtf[BZip2Constants::MAX_SELECTORS] = {0};
+
+ Int32 unzftab[256] = {0};
+
+ Int32 limit[BZip2Constants::N_GROUPS][BZip2Constants::MAX_ALPHA_SIZE]
= {{0}};
+ Int32 base[BZip2Constants::N_GROUPS][BZip2Constants::MAX_ALPHA_SIZE] =
{{0}};
+ Int32 perm[BZip2Constants::N_GROUPS][BZip2Constants::MAX_ALPHA_SIZE] =
{{0}};
+ Int32 minLens[BZip2Constants::N_GROUPS] = {0};
+
+ Int32 cftab[257] = {0};
+ UInt16 getAndMoveToFrontDecode_yy[256] = {0};
+ UInt16
temp_charArray2d[BZip2Constants::N_GROUPS][BZip2Constants::MAX_ALPHA_SIZE] =
{{0}};
+
+ char recvDecodingTables_pos[BZip2Constants::N_GROUPS] = {0};
+
+ std::vector<Int32> tt;
+ std::vector<char> ll8;
+
+ explicit Data(Int32 blockSize100k_) : ll8(blockSize100k_ *
BZip2Constants::BASE_BLOCK_SIZE) { }
+
+ std::vector<Int32> & initTT(Int32 length);
+ std::string toString();
+
+ private:
+ /// Helper static functions for toString()
+ template <typename T>
+ static std::string arrayToString(const std::vector<T> & arr);
+ template <typename T>
+ static std::string array2DToString(T
arr[BZip2Constants::N_GROUPS][BZip2Constants::MAX_ALPHA_SIZE]);
+ template <typename T>
+ static std::string arrayToString(const T * arr, size_t size);
+ };
+
+ class CRC
+ {
+ public:
+ static constexpr UInt32 crc32Table[]
+ = {0x00000000, 0x04c11db7, 0x09823b6e, 0x0d4326d9, 0x130476dc,
0x17c56b6b, 0x1a864db2, 0x1e475005, 0x2608edb8, 0x22c9f00f,
+ 0x2f8ad6d6, 0x2b4bcb61, 0x350c9b64, 0x31cd86d3, 0x3c8ea00a,
0x384fbdbd, 0x4c11db70, 0x48d0c6c7, 0x4593e01e, 0x4152fda9,
+ 0x5f15adac, 0x5bd4b01b, 0x569796c2, 0x52568b75, 0x6a1936c8,
0x6ed82b7f, 0x639b0da6, 0x675a1011, 0x791d4014, 0x7ddc5da3,
+ 0x709f7b7a, 0x745e66cd, 0x9823b6e0, 0x9ce2ab57, 0x91a18d8e,
0x95609039, 0x8b27c03c, 0x8fe6dd8b, 0x82a5fb52, 0x8664e6e5,
+ 0xbe2b5b58, 0xbaea46ef, 0xb7a96036, 0xb3687d81, 0xad2f2d84,
0xa9ee3033, 0xa4ad16ea, 0xa06c0b5d, 0xd4326d90, 0xd0f37027,
+ 0xddb056fe, 0xd9714b49, 0xc7361b4c, 0xc3f706fb, 0xceb42022,
0xca753d95, 0xf23a8028, 0xf6fb9d9f, 0xfbb8bb46, 0xff79a6f1,
+ 0xe13ef6f4, 0xe5ffeb43, 0xe8bccd9a, 0xec7dd02d, 0x34867077,
0x30476dc0, 0x3d044b19, 0x39c556ae, 0x278206ab, 0x23431b1c,
+ 0x2e003dc5, 0x2ac12072, 0x128e9dcf, 0x164f8078, 0x1b0ca6a1,
0x1fcdbb16, 0x018aeb13, 0x054bf6a4, 0x0808d07d, 0x0cc9cdca,
+ 0x7897ab07, 0x7c56b6b0, 0x71159069, 0x75d48dde, 0x6b93dddb,
0x6f52c06c, 0x6211e6b5, 0x66d0fb02, 0x5e9f46bf, 0x5a5e5b08,
+ 0x571d7dd1, 0x53dc6066, 0x4d9b3063, 0x495a2dd4, 0x44190b0d,
0x40d816ba, 0xaca5c697, 0xa864db20, 0xa527fdf9, 0xa1e6e04e,
+ 0xbfa1b04b, 0xbb60adfc, 0xb6238b25, 0xb2e29692, 0x8aad2b2f,
0x8e6c3698, 0x832f1041, 0x87ee0df6, 0x99a95df3, 0x9d684044,
+ 0x902b669d, 0x94ea7b2a, 0xe0b41de7, 0xe4750050, 0xe9362689,
0xedf73b3e, 0xf3b06b3b, 0xf771768c, 0xfa325055, 0xfef34de2,
+ 0xc6bcf05f, 0xc27dede8, 0xcf3ecb31, 0xcbffd686, 0xd5b88683,
0xd1799b34, 0xdc3abded, 0xd8fba05a, 0x690ce0ee, 0x6dcdfd59,
+ 0x608edb80, 0x644fc637, 0x7a089632, 0x7ec98b85, 0x738aad5c,
0x774bb0eb, 0x4f040d56, 0x4bc510e1, 0x46863638, 0x42472b8f,
+ 0x5c007b8a, 0x58c1663d, 0x558240e4, 0x51435d53, 0x251d3b9e,
0x21dc2629, 0x2c9f00f0, 0x285e1d47, 0x36194d42, 0x32d850f5,
+ 0x3f9b762c, 0x3b5a6b9b, 0x0315d626, 0x07d4cb91, 0x0a97ed48,
0x0e56f0ff, 0x1011a0fa, 0x14d0bd4d, 0x19939b94, 0x1d528623,
+ 0xf12f560e, 0xf5ee4bb9, 0xf8ad6d60, 0xfc6c70d7, 0xe22b20d2,
0xe6ea3d65, 0xeba91bbc, 0xef68060b, 0xd727bbb6, 0xd3e6a601,
+ 0xdea580d8, 0xda649d6f, 0xc423cd6a, 0xc0e2d0dd, 0xcda1f604,
0xc960ebb3, 0xbd3e8d7e, 0xb9ff90c9, 0xb4bcb610, 0xb07daba7,
+ 0xae3afba2, 0xaafbe615, 0xa7b8c0cc, 0xa379dd7b, 0x9b3660c6,
0x9ff77d71, 0x92b45ba8, 0x9675461f, 0x8832161a, 0x8cf30bad,
+ 0x81b02d74, 0x857130c3, 0x5d8a9099, 0x594b8d2e, 0x5408abf7,
0x50c9b640, 0x4e8ee645, 0x4a4ffbf2, 0x470cdd2b, 0x43cdc09c,
+ 0x7b827d21, 0x7f436096, 0x7200464f, 0x76c15bf8, 0x68860bfd,
0x6c47164a, 0x61043093, 0x65c52d24, 0x119b4be9, 0x155a565e,
+ 0x18197087, 0x1cd86d30, 0x029f3d35, 0x065e2082, 0x0b1d065b,
0x0fdc1bec, 0x3793a651, 0x3352bbe6, 0x3e119d3f, 0x3ad08088,
+ 0x2497d08d, 0x2056cd3a, 0x2d15ebe3, 0x29d4f654, 0xc5a92679,
0xc1683bce, 0xcc2b1d17, 0xc8ea00a0, 0xd6ad50a5, 0xd26c4d12,
+ 0xdf2f6bcb, 0xdbee767c, 0xe3a1cbc1, 0xe760d676, 0xea23f0af,
0xeee2ed18, 0xf0a5bd1d, 0xf464a0aa, 0xf9278673, 0xfde69bc4,
+ 0x89b8fd09, 0x8d79e0be, 0x803ac667, 0x84fbdbd0, 0x9abc8bd5,
0x9e7d9662, 0x933eb0bb, 0x97ffad0c, 0xafb010b1, 0xab710d06,
+ 0xa6322bdf, 0xa2f33668, 0xbcb4666d, 0xb8757bda, 0xb5365d03,
0xb1f740b4};
+
+ CRC() { initialiseCRC(); }
+ ~CRC() = default;
+
+ void initialiseCRC() { globalCrc = 0xffffffff; }
+ Int32 getFinalCRC() const { return ~globalCrc; }
+ Int32 getGlobalCRC() const { return globalCrc; }
+ void setGlobalCRC(Int32 newCrc) { globalCrc = newCrc; }
+ void updateCRC(Int32 inCh)
+ {
+ Int32 temp = (globalCrc >> 24) ^ inCh;
+ if (temp < 0)
+ temp = 256 + temp;
+ globalCrc = (globalCrc << 8) ^
static_cast<Int32>(crc32Table[temp]);
+ }
+ void updateCRC(Int32 inCh, Int32 repeat)
+ {
+ Int32 globalCrcShadow = globalCrc;
+ while (repeat-- > 0)
+ {
+ Int32 temp = (globalCrcShadow >> 24) ^ inCh;
+ globalCrcShadow = (globalCrcShadow << 8) ^
static_cast<Int32>(crc32Table[(temp >= 0) ? temp : (temp + 256)]);
+ }
+ globalCrc = globalCrcShadow;
+ }
+
+ private:
+ Int32 globalCrc;
+ };
+
+ enum class STATE : int8_t
+ {
+ END_OF_FILE = 0,
+ START_BLOCK_STATE,
+ RAND_PART_A_STATE,
+ RAND_PART_B_STATE,
+ RAND_PART_C_STATE,
+ NO_RAND_PART_A_STATE,
+ NO_RAND_PART_B_STATE,
+ NO_RAND_PART_C_STATE,
+ NO_PROCESS_STATE
+ };
+
+ /// If current bzip2 file split is not the fist one, then
start_need_special_process should be true.
+ /// When it is true, the decompressed result of the first block will be
ignored except the last line.
+ /// Case1:
+ /// e.g. "line1 \n line2 \n line3", line1 and line2 will be ignored
because they are processed in the previous split.
+ /// We are not sure if line3 is a completed line, so line3 will be
concated before the decompressed result of the next block in current split.
+ /// Case2:
+ /// e.g. "line2 \n line2 \n line3 \n", all lines will be ignored because
they are processed in the previous split.
+ const bool first_block_need_special_process;
+
+ /// If current bzip2 file split is not the last one, then
end_need_special_process should be true.
+ /// When it is true, the decompressed result of the last block will be
processed except the last line.
+ /// Case1:
+ /// e.g. "line1 \n line2 \n line3", line3 will be ignored because it is
processed in the next split.
+ /// Case2:
+ /// e.g. "line1 \n line2 \n line3 \n", all lines will be processed because
we are pretty sure that line3 is a completed line.
+ const bool last_block_need_special_process;
+ bool is_first_block;
+
+ Int32 blockSize100k;
+ STATE currentState;
+ bool skipResult;
+ Int32 currentChar;
+ Int32 storedBlockCRC;
+ bool blockRandomised;
+ std::unique_ptr<Data> data;
+ CRC crc;
+ Int32 computedBlockCRC;
+ Int32 storedCombinedCRC;
+ Int32 computedCombinedCRC;
+
+ Int32 origPtr;
+ Int32 nInUse;
+ Int64 bsBuff;
+ Int64 bsLive;
+ Int32 last;
+
+ Int32 su_count;
+ Int32 su_ch2;
+ Int32 su_chPrev;
+ Int32 su_i2;
+ Int32 su_j2;
+ Int32 su_rNToGo;
+ Int32 su_rTPos;
+ Int32 su_tPos;
+ char su_z;
+
+ /// SplittableBzip2ReadBuffer will skip bytes before the first block
header. adjusted_start records file position after skipping.
+ /// It is only valid when input stream is seekable and block header could
be found in input stream.
+ std::optional<size_t> adjusted_start;
+
+ static void hbCreateDecodeTables(
+ Int32 * __restrict limit,
+ Int32 * __restrict base,
+ Int32 * __restrict perm,
+ const UInt16 * __restrict length,
+ Int32 minLen,
+ Int32 maxLen,
+ Int32 alphaSize);
+
+public:
+ explicit SplittableBzip2ReadBuffer(
+ std::unique_ptr<ReadBuffer> in_,
+ bool first_block_need_special_process_ = false,
+ bool last_block_need_special_process_ = false,
+ size_t buf_size = DBMS_DEFAULT_BUFFER_SIZE,
+ char * existing_memory = nullptr,
+ size_t alignment = 0);
+
+ ~SplittableBzip2ReadBuffer() override = default;
+
+ std::optional<size_t> getAdjustedStart() const { return adjusted_start; }
+
+ bool nextImpl() override;
+
+ static bool skipToNextMarker(Int64 marker, Int32 markerBitLength,
ReadBuffer & in_, Int64 & bsBuff_, Int64 & bsLive_);
+
+private:
+ Int32 read(char * dest, size_t dest_size, size_t offs, size_t len);
+ Int32 read0();
+ static Int32 readAByte(ReadBuffer & in_);
+
+ bool skipToNextMarker(Int64 marker, Int32 markerBitLength);
+
+ [[noreturn]] void reportCRCError();
+
+ void makeMaps();
+
+ void changeStateToProcessABlock();
+
+ void initBlock();
+
+ void endBlock();
+
+ void complete();
+
+ static Int64 bsR(Int64 n, ReadBuffer & in_, Int64 & bsBuff_, Int64 &
bsLive_);
+ Int64 bsR(Int64 n);
+ bool bsGetBit();
+ char bsGetUByte() { return bsR(8); }
+ Int32 bsGetInt() { return static_cast<Int32>((((((bsR(8) << 8) | bsR(8))
<< 8) | bsR(8)) << 8) | bsR(8)); }
+
+ void recvDecodingTables();
+
+ void createHuffmanDecodingTables(Int32 alphaSize, Int32 nGroups);
+
+ void getAndMoveToFrontDecode();
+ Int32 getAndMoveToFrontDecode0(Int32 groupNo);
+
+ void setupBlock();
+ void setupRandPartA();
+ void setupNoRandPartA();
+ void setupRandPartB();
+ void setupRandPartC();
+ void setupNoRandPartB();
+ void setupNoRandPartC();
+};
+
+}
+#endif
diff --git a/cpp-ch/local-engine/Parser/CHColumnToSparkRow.cpp
b/cpp-ch/local-engine/Parser/CHColumnToSparkRow.cpp
index 602cd3d683..d5a6c96419 100644
--- a/cpp-ch/local-engine/Parser/CHColumnToSparkRow.cpp
+++ b/cpp-ch/local-engine/Parser/CHColumnToSparkRow.cpp
@@ -299,8 +299,8 @@ static void writeValue(
SparkRowInfo::SparkRowInfo(
const DB::ColumnsWithTypeAndName & cols,
const DB::DataTypes & dataTypes,
- const size_t & col_size,
- const size_t & row_size,
+ size_t col_size,
+ size_t row_size,
const MaskVector & masks)
: types(dataTypes)
, num_rows(masks == nullptr ? row_size : masks->size())
diff --git a/cpp-ch/local-engine/Parser/CHColumnToSparkRow.h
b/cpp-ch/local-engine/Parser/CHColumnToSparkRow.h
index cb46a3850d..6af580850d 100644
--- a/cpp-ch/local-engine/Parser/CHColumnToSparkRow.h
+++ b/cpp-ch/local-engine/Parser/CHColumnToSparkRow.h
@@ -44,8 +44,8 @@ public:
explicit SparkRowInfo(
const DB::ColumnsWithTypeAndName & cols,
const DB::DataTypes & types,
- const size_t & col_size,
- const size_t & row_size,
+ size_t col_size,
+ size_t row_size,
const MaskVector & masks = nullptr);
const DB::DataTypes & getDataTypes() const;
diff --git
a/cpp-ch/local-engine/Storages/SubstraitSource/ExcelTextFormatFile.cpp
b/cpp-ch/local-engine/Storages/SubstraitSource/ExcelTextFormatFile.cpp
index 7e87d11dc7..6e12938e45 100644
--- a/cpp-ch/local-engine/Storages/SubstraitSource/ExcelTextFormatFile.cpp
+++ b/cpp-ch/local-engine/Storages/SubstraitSource/ExcelTextFormatFile.cpp
@@ -65,7 +65,7 @@ bool ExcelTextFormatFile::useThis(const DB::ContextPtr &
context)
FormatFile::InputFormatPtr ExcelTextFormatFile::createInputFormat(const
DB::Block & header)
{
auto res = std::make_shared<FormatFile::InputFormat>();
- res->read_buffer = read_buffer_builder->build(file_info, true);
+ res->read_buffer = read_buffer_builder->build(file_info);
DB::FormatSettings format_settings = createFormatSettings();
size_t max_block_size = file_info.text().max_block_size();
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/FormatFile.h
b/cpp-ch/local-engine/Storages/SubstraitSource/FormatFile.h
index 8ab82f312f..14b53f6842 100644
--- a/cpp-ch/local-engine/Storages/SubstraitSource/FormatFile.h
+++ b/cpp-ch/local-engine/Storages/SubstraitSource/FormatFile.h
@@ -81,7 +81,6 @@ protected:
ReadBufferBuilderPtr read_buffer_builder;
std::vector<String> partition_keys;
std::map<String, String> partition_values;
- // std::optional<SourceFilter> filter;
std::shared_ptr<const DB::KeyCondition> key_condition;
};
using FormatFilePtr = std::shared_ptr<FormatFile>;
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/JSONFormatFile.cpp
b/cpp-ch/local-engine/Storages/SubstraitSource/JSONFormatFile.cpp
index b3e47681a8..e6e7573be7 100644
--- a/cpp-ch/local-engine/Storages/SubstraitSource/JSONFormatFile.cpp
+++ b/cpp-ch/local-engine/Storages/SubstraitSource/JSONFormatFile.cpp
@@ -31,7 +31,7 @@ JSONFormatFile::JSONFormatFile(DB::ContextPtr context_, const
substrait::ReadRel
FormatFile::InputFormatPtr JSONFormatFile::createInputFormat(const DB::Block &
header)
{
auto res = std::make_shared<FormatFile::InputFormat>();
- res->read_buffer =
read_buffer_builder->buildWithCompressionWrapper(file_info, true);
+ res->read_buffer =
read_buffer_builder->buildWithCompressionWrapper(file_info);
DB::FormatSettings format_settings = DB::getFormatSettings(context);
format_settings.with_names_use_header = true;
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/ReadBufferBuilder.cpp
b/cpp-ch/local-engine/Storages/SubstraitSource/ReadBufferBuilder.cpp
index 56205f224a..e3753495b4 100644
--- a/cpp-ch/local-engine/Storages/SubstraitSource/ReadBufferBuilder.cpp
+++ b/cpp-ch/local-engine/Storages/SubstraitSource/ReadBufferBuilder.cpp
@@ -19,7 +19,6 @@
#include <memory>
#include <shared_mutex>
-#include <thread>
#include <Core/Settings.h>
#include <Disks/IO/AsynchronousBoundedReadBuffer.h>
#include <Disks/IO/CachedOnDiskReadBufferFromFile.h>
@@ -29,6 +28,7 @@
#include <IO/ReadBufferFromFile.h>
#include <IO/ReadBufferFromS3.h>
#include <IO/ReadSettings.h>
+#include <IO/SplittableBzip2ReadBuffer.h>
#include <IO/S3/getObjectInfo.h>
#include <IO/S3Common.h>
#include <IO/SeekableReadBuffer.h>
@@ -84,6 +84,7 @@ namespace ErrorCodes
namespace local_engine
{
+using namespace DB;
FileCacheConcurrentMap ReadBufferBuilder::files_cache_time_map;
template <class key_type, class value_type>
@@ -115,9 +116,9 @@ private:
std::shared_mutex rwLock;
};
-static std::pair<size_t, size_t>
adjustFileReadPosition(DB::ReadBufferFromFileBase & buffer, size_t
read_start_pos, size_t read_end_pos)
+static std::pair<size_t, size_t> getAdjustedReadRange(SeekableReadBuffer &
buffer, const std::pair<size_t, size_t> & start_end)
{
- auto get_next_line_pos = [&](DB::ReadBufferFromFileBase & buf) -> size_t
+ auto get_next_line_pos = [&](SeekableReadBuffer & buf) -> size_t
{
while (!buf.eof())
{
@@ -151,6 +152,8 @@ static std::pair<size_t, size_t>
adjustFileReadPosition(DB::ReadBufferFromFileBa
return buf.getPosition();
};
+ size_t read_start_pos = start_end.first;
+ size_t read_end_pos = start_end.second;
std::pair<size_t, size_t> result;
if (read_start_pos == 0)
@@ -172,24 +175,38 @@ static std::pair<size_t, size_t>
adjustFileReadPosition(DB::ReadBufferFromFileBa
return result;
}
-static std::unique_ptr<DB::ReadBufferFromFileBase>
-resetOffset(std::unique_ptr<DB::ReadBufferFromFileBase> read_buffer, const
substrait::ReadRel::LocalFiles::FileOrFiles & file_info)
+static std::unique_ptr<SeekableReadBuffer>
+adjustReadRangeIfNeeded(std::unique_ptr<SeekableReadBuffer> read_buffer, const
substrait::ReadRel::LocalFiles::FileOrFiles & file_info)
{
- auto start_end_pos = adjustFileReadPosition(*read_buffer,
file_info.start(), file_info.start() + file_info.length());
+ /// Skip formats in which rows are not seperated by newline characters.
+ if (!(file_info.has_text() || file_info.has_json()))
+ return std::move(read_buffer);
+
+ /// Skip text/json files with compression.
+ /// TODO implement adjustFileReadPosition when compression method is bzip2
+ Poco::URI file_uri(file_info.uri_file());
+ DB::CompressionMethod compression =
DB::chooseCompressionMethod(file_uri.getPath(), "auto");
+ if (compression != CompressionMethod::None)
+ return std::move(read_buffer);
+
+ std::pair<size_t, size_t> start_end{file_info.start(), file_info.start() +
file_info.length()};
+ start_end = getAdjustedReadRange(*read_buffer, start_end);
+
LOG_DEBUG(
&Poco::Logger::get("ReadBufferBuilder"),
"File read start and end position adjusted from {},{} to {},{}",
file_info.start(),
file_info.start() + file_info.length(),
- start_end_pos.first,
- start_end_pos.second);
+ start_end.first,
+ start_end.second);
+ /// If read buffer doesn't support right bounded reads, wrap it with
BoundedReadBuffer to enable right bounded reads.
if (dynamic_cast<DB::ReadBufferFromHDFS *>(read_buffer.get()) ||
dynamic_cast<DB::ReadBufferFromFile *>(read_buffer.get()))
read_buffer =
std::make_unique<DB::BoundedReadBuffer>(std::move(read_buffer));
- read_buffer->seek(start_end_pos.first, SEEK_SET);
- read_buffer->setReadUntilPosition(start_end_pos.second);
- return read_buffer;
+ read_buffer->seek(start_end.first, SEEK_SET);
+ read_buffer->setReadUntilPosition(start_end.second);
+ return std::move(read_buffer);
}
class LocalFileReadBufferBuilder : public ReadBufferBuilder
@@ -199,7 +216,7 @@ public:
~LocalFileReadBufferBuilder() override = default;
std::unique_ptr<DB::ReadBuffer>
- build(const substrait::ReadRel::LocalFiles::FileOrFiles & file_info, bool
set_read_util_position) override
+ build(const substrait::ReadRel::LocalFiles::FileOrFiles & file_info)
override
{
Poco::URI file_uri(file_info.uri_file());
std::unique_ptr<DB::ReadBufferFromFileBase> read_buffer;
@@ -213,10 +230,7 @@ public:
else
read_buffer = std::make_unique<DB::ReadBufferFromFile>(file_path);
- if (set_read_util_position)
- return resetOffset(std::move(read_buffer), file_info);
-
- return read_buffer;
+ return adjustReadRangeIfNeeded(std::move(read_buffer), file_info);
}
};
@@ -229,71 +243,89 @@ public:
~HDFSFileReadBufferBuilder() override = default;
std::unique_ptr<DB::ReadBuffer>
- build(const substrait::ReadRel::LocalFiles::FileOrFiles & file_info, bool
set_read_util_position) override
+ build(const substrait::ReadRel::LocalFiles::FileOrFiles & file_info)
override
{
DB::ReadSettings read_settings = getReadSettings();
- auto & config = context->getConfigRef();
- auto hdfs_config = HdfsConfig::loadFromContext(config, read_settings);
- Poco::URI file_uri(file_info.uri_file());
- std::string uri_path = "hdfs://" + file_uri.getHost();
- if (file_uri.getPort())
- uri_path += ":" +
std::to_string(static_cast<unsigned>(file_uri.getPort()));
+ const auto & config = context->getConfigRef();
+
+ /// Get hdfs_uri
+ Poco::URI uri(file_info.uri_file());
+ auto hdfs_file_path = uri.getPath();
+ std::string hdfs_uri = "hdfs://" + uri.getHost();
+ if (uri.getPort())
+ hdfs_uri += ":" + std::to_string(uri.getPort());
- size_t file_size = 0;
- size_t modified_time = 0;
+ std::optional<size_t> file_size;
+ std::optional<size_t> modified_time;
if (file_info.has_properties())
{
- file_size = file_info.properties().filesize();
+ if (file_info.properties().filesize() > 0)
+ {
+ /// filesize may be zero, under such condition we should not
set file_size
+ file_size = file_info.properties().filesize();
+ }
+
modified_time = file_info.properties().modificationtime();
}
- std::unique_ptr<DB::ReadBufferFromFileBase> read_buffer;
-
- if (hdfs_config.hdfs_async)
+ std::unique_ptr<SeekableReadBuffer> read_buffer;
+ if (!read_settings.enable_filesystem_cache)
{
- std::optional<size_t> size = std::nullopt;
- if (file_size)
- size = file_size;
+ bool thread_pool_read = read_settings.remote_fs_method ==
DB::RemoteFSReadMethod::threadpool;
+ /// ORC and Parquet reader had already implemented async prefetch.
They don't rely on AsynchronousReadBufferFromHDFS
+ bool use_async_prefetch
+ = read_settings.remote_fs_prefetch && thread_pool_read &&
(file_info.has_text() || file_info.has_json());
+ auto raw_read_buffer = std::make_unique<ReadBufferFromHDFS>(
+ hdfs_uri,
+ hdfs_file_path,
+ config,
+ read_settings,
+ /* read_until_position */ 0,
+ /* use_external_buffer */ true,
+ file_size);
- read_buffer
- = std::make_unique<DB::ReadBufferFromHDFS>(uri_path,
file_uri.getPath(), config, read_settings, file_size, true, size);
- if (read_settings.remote_fs_prefetch)
- {
- auto & pool_reader =
context->getThreadPoolReader(DB::FilesystemReaderType::ASYNCHRONOUS_REMOTE_FS_READER);
- read_buffer =
std::make_unique<DB::AsynchronousBoundedReadBuffer>(std::move(read_buffer),
pool_reader, read_settings);
- }
+ if (use_async_prefetch)
+ read_buffer = std::make_unique<AsynchronousReadBufferFromHDFS>(
+
getThreadPoolReader(FilesystemReaderType::ASYNCHRONOUS_REMOTE_FS_READER),
read_settings, std::move(raw_read_buffer));
+ else
+ read_buffer = std::move(raw_read_buffer);
}
else
{
- if (!file_size)
+ if (!file_size.has_value())
{
// only for spark3.2 file partition not contained file size
// so first compute file size first
- auto read_buffer_impl =
std::make_unique<DB::ReadBufferFromHDFS>(
- uri_path, file_uri.getPath(), config, read_settings, 0,
true);
- file_size = read_buffer_impl->getFileSize();
+ auto tmp_read_buffer =
std::make_unique<DB::ReadBufferFromHDFS>(
+ hdfs_uri,
+ hdfs_file_path,
+ config,
+ read_settings,
+ /* read_until_position */ 0,
+ /* use_external_buffer */ true);
+ file_size = tmp_read_buffer->getFileSize();
}
- ReadBufferCreator hdfs_read_buffer_creator
- = [this, hdfs_uri = uri_path, hdfs_file_path =
file_uri.getPath(), read_settings, &config](
- bool /* restricted_seek */, const DB::StoredObject &
object) -> std::unique_ptr<DB::ReadBufferFromHDFS>
- {
+ if (!modified_time.has_value())
+ modified_time = 0;
+
+ ReadBufferCreator read_buffer_creator
+ = [this, hdfs_uri = hdfs_uri, hdfs_file_path = hdfs_file_path,
read_settings, &config](
+ bool /* restricted_seek */, const DB::StoredObject &
object) -> std::unique_ptr<DB::ReadBufferFromHDFS> {
return std::make_unique<DB::ReadBufferFromHDFS>(
hdfs_uri, hdfs_file_path, config, read_settings, 0, true,
object.bytes_size);
};
- auto remote_path = file_uri.getPath().substr(1);
- DB::StoredObjects stored_objects{DB::StoredObject{remote_path, "",
file_size}};
- auto cache_creator = wrapWithCache(hdfs_read_buffer_creator,
read_settings, remote_path, modified_time, file_size);
+ auto remote_path = uri.getPath().substr(1);
+ DB::StoredObjects stored_objects{DB::StoredObject{remote_path, "",
*file_size}};
+ auto cache_creator = wrapWithCache(
+ read_buffer_creator, read_settings, remote_path,
*modified_time, *file_size);
auto cache_hdfs_read =
std::make_unique<DB::ReadBufferFromRemoteFSGather>(
std::move(cache_creator), stored_objects, read_settings,
nullptr, /* use_external_buffer */ false);
read_buffer = std::move(cache_hdfs_read);
}
- if (set_read_util_position)
- return resetOffset(std::move(read_buffer), file_info);
-
- return read_buffer;
+ return adjustReadRangeIfNeeded(std::move(read_buffer), file_info);
}
private:
@@ -330,7 +362,7 @@ public:
~S3FileReadBufferBuilder() override = default;
std::unique_ptr<DB::ReadBuffer>
- build(const substrait::ReadRel::LocalFiles::FileOrFiles & file_info, bool
set_read_util_position) override
+ build(const substrait::ReadRel::LocalFiles::FileOrFiles & file_info)
override
{
DB::ReadSettings read_settings = getReadSettings();
Poco::URI file_uri(file_info.uri_file());
@@ -382,10 +414,7 @@ public:
if (read_settings.remote_fs_prefetch)
async_reader->prefetch(Priority{});
- if (set_read_util_position)
- return resetOffset(std::move(async_reader), file_info);
-
- return async_reader;
+ return adjustReadRangeIfNeeded(std::move(async_reader), file_info);
}
private:
@@ -570,7 +599,7 @@ public:
explicit AzureBlobReadBuffer(DB::ContextPtr context_) :
ReadBufferBuilder(context_) { }
~AzureBlobReadBuffer() override = default;
- std::unique_ptr<DB::ReadBuffer> build(const
substrait::ReadRel::LocalFiles::FileOrFiles & file_info, bool) override
+ std::unique_ptr<DB::ReadBuffer> build(const
substrait::ReadRel::LocalFiles::FileOrFiles & file_info) override
{
Poco::URI file_uri(file_info.uri_file());
std::unique_ptr<DB::ReadBuffer> read_buffer;
@@ -631,35 +660,110 @@ ReadBufferBuilder::ReadBufferBuilder(DB::ContextPtr
context_) : context(context_
DB::ReadSettings ReadBufferBuilder::getReadSettings() const
{
DB::ReadSettings read_settings = context->getReadSettings();
- if (context->getConfigRef().getBool("gluten_cache.local.enabled", false))
- read_settings.enable_filesystem_cache = true;
- else
- read_settings.enable_filesystem_cache = false;
+ const auto & config = context->getConfigRef();
+
+ /// Override enable_filesystem_cache with gluten config
+ read_settings.enable_filesystem_cache =
config.getBool("gluten_cache.local.enabled", false);
+
+ /// Override remote_fs_prefetch with gluten config
+ read_settings.remote_fs_prefetch = config.getBool("hdfs.enable_async_io",
false);
return read_settings;
}
std::unique_ptr<DB::ReadBuffer>
-ReadBufferBuilder::buildWithCompressionWrapper(const
substrait::ReadRel::LocalFiles::FileOrFiles & file_info, bool
set_read_util_position)
+ReadBufferBuilder::wrapWithBzip2(std::unique_ptr<DB::ReadBuffer> in, const
substrait::ReadRel::LocalFiles::FileOrFiles & file_info)
+{
+ /// Bzip2 compressed file is splittable and we need to adjust read range
for each split
+ auto * seekable = dynamic_cast<SeekableReadBuffer *>(in.release());
+ if (!seekable)
+ throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "ReadBuffer
underlying BZIP2 decompressor must be seekable");
+ std::unique_ptr<SeekableReadBuffer> seekable_in(seekable);
+
+ size_t file_size = getFileSizeFromReadBuffer(*seekable_in);
+ size_t start = file_info.start();
+ size_t end = file_info.start() + file_info.length();
+
+ /// No need to adjust start becuase it is already processed inside
SplittableBzip2ReadBuffer
+ size_t new_start = start;
+
+ /// Extend end to the end of next block.
+ size_t new_end = end;
+ if (end < file_size)
+ {
+ Int64 bs_buff = 0;
+ Int64 bs_live = 0;
+
+ /// From end position skip to the second block delimiter.
+ seekable_in->seek(end, SEEK_SET);
+ for (size_t i = 0; i < 2; ++i)
+ {
+ size_t pos = seekable_in->getPosition();
+ bool ok = SplittableBzip2ReadBuffer::skipToNextMarker(
+ SplittableBzip2ReadBuffer::BLOCK_DELIMITER,
+ SplittableBzip2ReadBuffer::DELIMITER_BIT_LENGTH,
+ *seekable_in,
+ bs_buff,
+ bs_live);
+
+ if (seekable_in->eof())
+ break;
+
+ if (!ok)
+ throw Exception(ErrorCodes::LOGICAL_ERROR, "Can't find next
block delimiter in after offset: {}", pos);
+ }
+ new_end = seekable->eof() ? file_size : seekable_in->getPosition() -
SplittableBzip2ReadBuffer::DELIMITER_BIT_LENGTH / 8 + 1;
+ }
+
+ LOG_DEBUG(
+ &Poco::Logger::get("ReadBufferBuilder"),
+ "File read start and end position adjusted from {},{} to {},{}",
+ start,
+ end,
+ new_start,
+ new_end);
+
+ std::unique_ptr<SeekableReadBuffer> bounded_in;
+ if (dynamic_cast<DB::ReadBufferFromHDFS *>(seekable_in.get()) ||
dynamic_cast<DB::ReadBufferFromFile *>(seekable_in.get()))
+ bounded_in =
std::make_unique<BoundedReadBuffer>(std::move(seekable_in));
+ else
+ bounded_in = std::move(seekable_in);
+
+ bounded_in->seek(new_start, SEEK_SET);
+ bounded_in->setReadUntilPosition(new_end);
+ bool first_block_need_special_process = (new_start > 0);
+ bool last_block_need_special_process = (new_end < file_size);
+ auto decompressed_in = std::make_unique<SplittableBzip2ReadBuffer>(
+ std::move(bounded_in), first_block_need_special_process,
last_block_need_special_process);
+ return std::move(decompressed_in);
+}
+
+std::unique_ptr<DB::ReadBuffer>
+ReadBufferBuilder::buildWithCompressionWrapper(const
substrait::ReadRel::LocalFiles::FileOrFiles & file_info)
{
- auto in = build(file_info, set_read_util_position);
+ auto in = build(file_info);
/// Wrap the read buffer with compression method if exists
Poco::URI file_uri(file_info.uri_file());
DB::CompressionMethod compression =
DB::chooseCompressionMethod(file_uri.getPath(), "auto");
- return compression != DB::CompressionMethod::None ?
DB::wrapReadBufferWithCompressionMethod(std::move(in), compression) :
std::move(in);
+
+ if (compression == CompressionMethod::Bzip2)
+ return wrapWithBzip2(std::move(in), file_info);
+ else
+ return wrapReadBufferWithCompressionMethod(std::move(in), compression);
}
ReadBufferBuilder::ReadBufferCreator ReadBufferBuilder::wrapWithCache(
ReadBufferCreator read_buffer_creator,
DB::ReadSettings & read_settings,
const String & key,
- const size_t & last_modified_time,
- const size_t & file_size)
+ size_t last_modified_time,
+ size_t file_size)
{
const auto & config = context->getConfigRef();
if (!config.getBool("gluten_cache.local.enabled", false))
return read_buffer_creator;
+
read_settings.enable_filesystem_cache = true;
if (!file_cache)
{
@@ -709,7 +813,7 @@ ReadBufferBuilder::ReadBufferCreator
ReadBufferBuilder::wrapWithCache(
};
}
-void ReadBufferBuilder::updateCaches(const String & key, const size_t &
last_modified_time, const size_t & file_size) const
+void ReadBufferBuilder::updateCaches(const String & key, size_t
last_modified_time, size_t file_size) const
{
if (!file_cache)
return;
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/ReadBufferBuilder.h
b/cpp-ch/local-engine/Storages/SubstraitSource/ReadBufferBuilder.h
index 5e63a93e3d..f2c1c10a6f 100644
--- a/cpp-ch/local-engine/Storages/SubstraitSource/ReadBufferBuilder.h
+++ b/cpp-ch/local-engine/Storages/SubstraitSource/ReadBufferBuilder.h
@@ -37,26 +37,29 @@ public:
/// build a new read buffer
virtual std::unique_ptr<DB::ReadBuffer>
- build(const substrait::ReadRel::LocalFiles::FileOrFiles & file_info, bool
set_read_util_position = false) = 0;
+ build(const substrait::ReadRel::LocalFiles::FileOrFiles & file_info) = 0;
/// build a new read buffer, consider compression method
- std::unique_ptr<DB::ReadBuffer> buildWithCompressionWrapper(const
substrait::ReadRel::LocalFiles::FileOrFiles & file_info, bool
set_read_util_position = false);
+ std::unique_ptr<DB::ReadBuffer> buildWithCompressionWrapper(const
substrait::ReadRel::LocalFiles::FileOrFiles & file_info);
protected:
using ReadBufferCreator =
std::function<std::unique_ptr<DB::ReadBufferFromFileBase>(bool restricted_seek,
const DB::StoredObject & object)>;
+ std::unique_ptr<DB::ReadBuffer>
+ wrapWithBzip2(std::unique_ptr<DB::ReadBuffer> in, const
substrait::ReadRel::LocalFiles::FileOrFiles & file_info);
+
ReadBufferCreator wrapWithCache(
ReadBufferCreator read_buffer_creator,
DB::ReadSettings & read_settings,
const String & key,
- const size_t & last_modified_time,
- const size_t & file_size);
+ size_t last_modified_time,
+ size_t file_size);
DB::ReadSettings getReadSettings() const;
DB::ContextPtr context;
private:
- void updateCaches(const String & key, const size_t & last_modified_time,
const size_t & file_size) const;
+ void updateCaches(const String & key, size_t last_modified_time, size_t
file_size) const;
public:
DB::FileCachePtr file_cache = nullptr;
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/TextFormatFile.cpp
b/cpp-ch/local-engine/Storages/SubstraitSource/TextFormatFile.cpp
index b22f883f5d..71362c5b60 100644
--- a/cpp-ch/local-engine/Storages/SubstraitSource/TextFormatFile.cpp
+++ b/cpp-ch/local-engine/Storages/SubstraitSource/TextFormatFile.cpp
@@ -35,7 +35,7 @@ TextFormatFile::TextFormatFile(
FormatFile::InputFormatPtr TextFormatFile::createInputFormat(const DB::Block &
header)
{
auto res = std::make_shared<FormatFile::InputFormat>();
- res->read_buffer =
read_buffer_builder->buildWithCompressionWrapper(file_info, true);
+ res->read_buffer =
read_buffer_builder->buildWithCompressionWrapper(file_info);
/// Initialize format params
size_t max_block_size = file_info.text().max_block_size();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
