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 077ba5254e fix code style and respect max_read_buffer_size for bzip2
read buffer (#8342)
077ba5254e is described below
commit 077ba5254e7bb946bfdc8dc5ae1a7961d73d457b
Author: 李扬 <[email protected]>
AuthorDate: Wed Dec 25 18:57:25 2024 +0800
fix code style and respect max_read_buffer_size for bzip2 read buffer
(#8342)
---
cpp-ch/local-engine/IO/SplittableBzip2ReadBuffer.cpp | 7 +++----
.../Storages/SubstraitSource/ParquetFormatFile.cpp | 9 +++++++--
.../Storages/SubstraitSource/ParquetFormatFile.h | 2 +-
.../Storages/SubstraitSource/ReadBufferBuilder.cpp | 4 +++-
cpp-ch/local-engine/tests/gtest_parquet_read.cpp | 20 ++++++++++----------
5 files changed, 24 insertions(+), 18 deletions(-)
diff --git a/cpp-ch/local-engine/IO/SplittableBzip2ReadBuffer.cpp
b/cpp-ch/local-engine/IO/SplittableBzip2ReadBuffer.cpp
index ba36baaf4c..4929c9e75f 100644
--- a/cpp-ch/local-engine/IO/SplittableBzip2ReadBuffer.cpp
+++ b/cpp-ch/local-engine/IO/SplittableBzip2ReadBuffer.cpp
@@ -192,10 +192,11 @@ SplittableBzip2ReadBuffer::SplittableBzip2ReadBuffer(
changeStateToProcessABlock();
LOG_DEBUG(
getLogger("SplittableBzip2ReadBuffer"),
- "adjusted_start:{} first_block_need_special_process:{}
last_block_need_special_process:{}",
+ "adjusted_start:{} first_block_need_special_process:{}
last_block_need_special_process:{} buf_size:{}",
*adjusted_start,
first_block_need_special_process,
- last_block_need_special_process);
+ last_block_need_special_process,
+ buf_size);
}
Int32 SplittableBzip2ReadBuffer::read(char * dest, size_t dest_size, size_t
offs, size_t len)
@@ -444,8 +445,6 @@ void SplittableBzip2ReadBuffer::changeStateToProcessABlock()
void SplittableBzip2ReadBuffer::initBlock()
{
- auto * seekable = dynamic_cast<SeekableReadBuffer*>(in.get());
- size_t position = seekable->getPosition();
storedBlockCRC = bsGetInt();
blockRandomised = (bsR(1) == 1);
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/ParquetFormatFile.cpp
b/cpp-ch/local-engine/Storages/SubstraitSource/ParquetFormatFile.cpp
index fdd7574a06..212a196550 100644
--- a/cpp-ch/local-engine/Storages/SubstraitSource/ParquetFormatFile.cpp
+++ b/cpp-ch/local-engine/Storages/SubstraitSource/ParquetFormatFile.cpp
@@ -95,9 +95,12 @@ FormatFile::InputFormatPtr
ParquetFormatFile::createInputFormat(const DB::Block
const DB::Settings & settings = context->getSettingsRef();
- if (use_pageindex_reader && pageindex_reader_support(header))
+ if (use_pageindex_reader && supportPageindexReader(header))
+ {
res->input =
std::make_shared<VectorizedParquetBlockInputFormat>(*(res->read_buffer),
header, format_settings);
+ }
else
+ {
res->input = std::make_shared<DB::ParquetBlockInputFormat>(
*(res->read_buffer),
header,
@@ -105,6 +108,7 @@ FormatFile::InputFormatPtr
ParquetFormatFile::createInputFormat(const DB::Block
settings[DB::Setting::max_parsing_threads],
settings[DB::Setting::max_download_threads],
8192);
+ }
return res;
}
@@ -128,7 +132,8 @@ std::optional<size_t> ParquetFormatFile::getTotalRows()
return total_rows;
}
}
-bool ParquetFormatFile::pageindex_reader_support(const DB::Block & header)
+
+bool ParquetFormatFile::supportPageindexReader(const DB::Block & header)
{
const auto result = std::ranges::find_if(
header,
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/ParquetFormatFile.h
b/cpp-ch/local-engine/Storages/SubstraitSource/ParquetFormatFile.h
index ba7f28883e..bed0343942 100644
--- a/cpp-ch/local-engine/Storages/SubstraitSource/ParquetFormatFile.h
+++ b/cpp-ch/local-engine/Storages/SubstraitSource/ParquetFormatFile.h
@@ -55,7 +55,7 @@ public:
String getFileFormat() const override { return "Parquet"; }
- static bool pageindex_reader_support(const DB::Block & header);
+ static bool supportPageindexReader(const DB::Block & header);
private:
bool use_pageindex_reader;
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/ReadBufferBuilder.cpp
b/cpp-ch/local-engine/Storages/SubstraitSource/ReadBufferBuilder.cpp
index d7f9a9d5ac..6b2785407d 100644
--- a/cpp-ch/local-engine/Storages/SubstraitSource/ReadBufferBuilder.cpp
+++ b/cpp-ch/local-engine/Storages/SubstraitSource/ReadBufferBuilder.cpp
@@ -75,6 +75,7 @@ extern const SettingsUInt64 s3_retry_attempts;
extern const SettingsMaxThreads max_download_threads;
extern const SettingsUInt64 max_download_buffer_size;
extern const SettingsBool input_format_allow_seeks;
+extern const SettingsUInt64 max_read_buffer_size;
}
namespace ErrorCodes
{
@@ -753,8 +754,9 @@
ReadBufferBuilder::wrapWithBzip2(std::unique_ptr<DB::ReadBuffer> in, const subst
bounded_in->setReadUntilPosition(new_end);
bool first_block_need_special_process = (new_start > 0);
bool last_block_need_special_process = (new_end < file_size);
+ size_t buffer_size =
context->getSettingsRef()[DB::Setting::max_read_buffer_size];
auto decompressed_in = std::make_unique<SplittableBzip2ReadBuffer>(
- std::move(bounded_in), first_block_need_special_process,
last_block_need_special_process);
+ std::move(bounded_in), first_block_need_special_process,
last_block_need_special_process, buffer_size);
return std::move(decompressed_in);
}
diff --git a/cpp-ch/local-engine/tests/gtest_parquet_read.cpp
b/cpp-ch/local-engine/tests/gtest_parquet_read.cpp
index 3421d7f78b..469fd8a64c 100644
--- a/cpp-ch/local-engine/tests/gtest_parquet_read.cpp
+++ b/cpp-ch/local-engine/tests/gtest_parquet_read.cpp
@@ -142,27 +142,27 @@ TEST(ParquetRead, ReadSchema)
TEST(ParquetRead, VerifyPageindexReaderSupport)
{
- EXPECT_FALSE(local_engine::ParquetFormatFile::pageindex_reader_support(
+ EXPECT_FALSE(local_engine::ParquetFormatFile::supportPageindexReader(
toBlockRowType(local_engine::test::readParquetSchema(local_engine::test::data_file("alltypes/alltypes_notnull.parquet")))));
- EXPECT_FALSE(local_engine::ParquetFormatFile::pageindex_reader_support(
+ EXPECT_FALSE(local_engine::ParquetFormatFile::supportPageindexReader(
toBlockRowType(local_engine::test::readParquetSchema(local_engine::test::data_file("alltypes/alltypes_null.parquet")))));
- EXPECT_FALSE(local_engine::ParquetFormatFile::pageindex_reader_support(
+ EXPECT_FALSE(local_engine::ParquetFormatFile::supportPageindexReader(
toBlockRowType(local_engine::test::readParquetSchema(local_engine::test::data_file("array.parquet")))));
- EXPECT_TRUE(local_engine::ParquetFormatFile::pageindex_reader_support(
+ EXPECT_TRUE(local_engine::ParquetFormatFile::supportPageindexReader(
toBlockRowType(local_engine::test::readParquetSchema(local_engine::test::data_file("date.parquet")))));
- EXPECT_TRUE(local_engine::ParquetFormatFile::pageindex_reader_support(
+ EXPECT_TRUE(local_engine::ParquetFormatFile::supportPageindexReader(
toBlockRowType(local_engine::test::readParquetSchema(local_engine::test::data_file("datetime64.parquet")))));
- EXPECT_TRUE(local_engine::ParquetFormatFile::pageindex_reader_support(
+ EXPECT_TRUE(local_engine::ParquetFormatFile::supportPageindexReader(
toBlockRowType(local_engine::test::readParquetSchema(local_engine::test::data_file("decimal.parquet")))));
- EXPECT_TRUE(local_engine::ParquetFormatFile::pageindex_reader_support(
+ EXPECT_TRUE(local_engine::ParquetFormatFile::supportPageindexReader(
toBlockRowType(local_engine::test::readParquetSchema(local_engine::test::data_file("iris.parquet")))));
- EXPECT_FALSE(local_engine::ParquetFormatFile::pageindex_reader_support(
+ EXPECT_FALSE(local_engine::ParquetFormatFile::supportPageindexReader(
toBlockRowType(local_engine::test::readParquetSchema(local_engine::test::data_file("map.parquet")))));
- EXPECT_TRUE(local_engine::ParquetFormatFile::pageindex_reader_support(
+ EXPECT_TRUE(local_engine::ParquetFormatFile::supportPageindexReader(
toBlockRowType(local_engine::test::readParquetSchema(local_engine::test::data_file("sample.parquet")))));
- EXPECT_FALSE(local_engine::ParquetFormatFile::pageindex_reader_support(
+ EXPECT_FALSE(local_engine::ParquetFormatFile::supportPageindexReader(
toBlockRowType(local_engine::test::readParquetSchema(local_engine::test::data_file("struct.parquet")))));
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
