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 374d87d81de [perf](csv reader) Reduce V2 CSV parsing overhead (#69875)
374d87d81de is described below
commit 374d87d81de557a8b002cb94a5816b7d8d650c12
Author: hui lai <[email protected]>
AuthorDate: Fri Oct 30 20:02:17 2026 +0800
[perf](csv reader) Reduce V2 CSV parsing overhead (#69875)
### What problem does this PR solve?
Issue Number: None
Related PR: #69716
Problem Summary:
After restoring sequential prefetch, large V2 CSV loads still spend
substantial CPU time in row parsing and string materialization. The
reader compares a general separator at every byte, updates atomic
profile counters per cell, starts three stage timers per row, and
repeats checked string-column casts per cell. It also ignores the row
cap supplied by TableReader.
This change combines four optimizations on master:
- Compare the separator byte directly for plain single-character
separators; retain enclosed, multi-character, Hive/OpenCSV and PROTO
parsing paths.
- Accumulate exact work counters locally and publish at
get_block/count/schema return, including errors. Sample stage timings
every 1024 physical read_line calls, starting with the first, and expose
unscaled SampledReadLineTime, SampledSplitLineTime and
SampledDeserializeTime with sample counts. DelimitedTextReader remains
full elapsed time; split-prefix alignment has its own full timer.
- Validate nullable string column types and reserve offsets/null maps
once per mutable batch, then insert strings directly with the existing
NULL, quoting and escape semantics.
- Honor FileReader::set_batch_size, retaining the existing byte budget
and runtime default when no explicit cap is supplied. The caller still
controls the maximum batch size; this does not force 16352-row blocks.
Shared counters and batch sizing also apply to the Hive text reader.
### Performance benchmark
On TPC-DS SF100, elapsed time decreased from **1,944 seconds to 1,375
seconds**:
| Benchmark | Before | After | Time reduction | Speedup |
| --- | ---: | ---: | ---: | ---: |
| TPC-DS SF100 | 1,944 s | 1,375 s | 29.3% (569 s) | 1.41× |
### Release note
Reduce V2 CSV parsing overhead and honor caller-provided batch sizes.
Delimited-text per-stage profile timers are now explicitly named
Sampled* and contain unscaled sampled durations; use the full
DelimitedTextReader timer for elapsed-time comparisons.
### Check List (For Author)
- Test
- [x] Unit Test: added cases for dynamic/zero batch caps, byte limits,
filtered blocks, cross-batch sampling and exact counters,
single/multi-character separators, NULL and escape behavior, missing
fields, reordered projection and copy-on-write. Extended error,
rejected-row and COUNT counter coverage.
- Execution: run-be-ut.sh was attempted with CsvV2ReaderTest.*,
TextV2ReaderTest.* and LegacyAndV2/PlainCsvSplitTest.*. It exits before
C++ compilation because the local JDK 17 fails to initialize (Failed
setting boot class path), including outside the sandbox. Full build/UT
remain pending. Performance benchmark results are reported above.
- Passed: repository clang-format 16 formatting/check scripts,
build-support/check-build-hygiene.sh and git diff --check.
- Behavior changed: Yes. Existing batch-size contract is honored, and
detailed stage timers are explicitly sampled. CSV value semantics are
unchanged.
- Does this need documentation: No. The profile includes a timing
explanation; no new configuration or SQL syntax.
For further benchmark comparisons, use identical hardware, input,
compression, cache conditions and load batch/byte-budget settings for
the parent and this commit. Compare total load time, Scanner CPU time,
DelimitedTextReader time and RowsRead/BlockNumber. Do not directly
compare Sampled* timers with the previous full stage timers.
---
be/src/format_v2/delimited_text/csv_reader.cpp | 47 ++++--
be/src/format_v2/delimited_text/csv_reader.h | 1 +
.../delimited_text/delimited_text_reader.cpp | 85 ++++++++--
.../delimited_text/delimited_text_reader.h | 26 ++++
.../format_v2/delimited_text/csv_reader_test.cpp | 171 ++++++++++++++++++++-
.../format_v2/delimited_text/text_reader_test.cpp | 32 +++-
6 files changed, 328 insertions(+), 34 deletions(-)
diff --git a/be/src/format_v2/delimited_text/csv_reader.cpp
b/be/src/format_v2/delimited_text/csv_reader.cpp
index 0d3ebf2cfd5..9a28707fd4e 100644
--- a/be/src/format_v2/delimited_text/csv_reader.cpp
+++ b/be/src/format_v2/delimited_text/csv_reader.cpp
@@ -23,6 +23,7 @@
#include "core/assert_cast.h"
#include "core/column/column_nullable.h"
+#include "core/column/column_string.h"
#include "core/data_type/data_type_string.h"
#include "core/data_type_serde/data_type_string_serde.h"
#include "format/file_reader/new_plain_binary_line_reader.h"
@@ -221,6 +222,14 @@ void CsvReader::_split_line(const Slice& line) {
append_value(value_start, separator_position - value_start);
value_start = separator_position + _value_separator.size();
}
+ } else if (_value_separator.size() == 1) {
+ const char separator = _value_separator[0];
+ for (size_t i = 0; i < line.size; ++i) {
+ if (line.data[i] == separator) {
+ append_value(value_start, i - value_start);
+ value_start = i + 1;
+ }
+ }
} else {
for (size_t i = 0; i < line.size;) {
if (starts_with_at(line, i, _value_separator)) {
@@ -236,6 +245,18 @@ void CsvReader::_split_line(const Slice& line) {
append_value(value_start, line.size - value_start);
}
+void CsvReader::_prepare_columns(std::vector<MutableColumnPtr>& columns,
size_t batch_size) {
+ for (const auto& column : _requested_columns) {
+ if (!column.nullable_string_fast_path) {
+ continue;
+ }
+ auto& nullable =
assert_cast<ColumnNullable&>(*columns[column.block_position.value()]);
+ auto& strings =
assert_cast<ColumnString&>(nullable.get_nested_column());
+ strings.get_offsets().reserve(strings.size() + batch_size);
+ nullable.get_null_map_data().reserve(nullable.size() + batch_size);
+ }
+}
+
Status CsvReader::_deserialize_one_cell(const RequestedColumn& column,
IColumn* output,
Slice value) {
DORIS_CHECK(output != nullptr);
@@ -243,18 +264,20 @@ Status CsvReader::_deserialize_one_cell(const
RequestedColumn& column, IColumn*
return _append_null(output);
}
if (column.nullable_string_fast_path) {
- auto& null_column = assert_cast<ColumnNullable&>(*output);
- // String is the hottest CSV type. Avoid the generic nullable serde
wrapper here:
- // deserialize directly into the nested string column and append the
null map bit ourselves.
+ // _prepare_columns validates the concrete types once per batch, after
copy-on-write.
+ auto& null_column = assert_cast<ColumnNullable&,
TypeCheckOnRelease::DISABLE>(*output);
+ auto& string_column = assert_cast<ColumnString&,
TypeCheckOnRelease::DISABLE>(
+ null_column.get_nested_column());
if (_hive_csv_parser) {
// Splitting and decoding are one operation for OpenCSV; a second
CSV serde pass
// would unescape literal data and confuse empty strings with
missing fields.
- null_column.get_nested_column().insert_data(value.data,
value.size);
+ string_column.insert_data(value.data, value.size);
null_column.get_null_map_data().push_back(0);
return Status::OK();
}
if (_empty_field_as_null && value.size == 0) {
- null_column.insert_data(nullptr, 0);
+ string_column.insert_default();
+ null_column.get_null_map_data().push_back(1);
return Status::OK();
}
// CSV keeps empty-field handling separate from null_format matching.
An empty
@@ -263,16 +286,16 @@ Status CsvReader::_deserialize_one_cell(const
RequestedColumn& column, IColumn*
const bool quoted = _options.converted_from_string &&
value.trim_double_quotes();
if (!quoted && _options.null_len > 0 && value.size ==
_options.null_len &&
std::memcmp(value.data, _options.null_format, value.size) == 0) {
- null_column.insert_data(nullptr, 0);
+ string_column.insert_default();
+ null_column.get_null_map_data().push_back(1);
return Status::OK();
}
- static DataTypeStringSerDe string_serde(TYPE_STRING);
- auto status =
string_serde.deserialize_one_cell_from_csv(null_column.get_nested_column(),
- value,
_options);
- if (!status.ok()) {
- null_column.insert_data(nullptr, 0);
- return Status::OK();
+ // Keep the same unescaping as StringSerDe, which always succeeds for
CSV strings.
+ if (_options.escape_char != 0) {
+ escape_string_for_csv(value.data, &value.size,
_options.escape_char,
+ _options.quote_char);
}
+ string_column.insert_data(value.data, value.size);
null_column.get_null_map_data().push_back(0);
return Status::OK();
}
diff --git a/be/src/format_v2/delimited_text/csv_reader.h
b/be/src/format_v2/delimited_text/csv_reader.h
index c8565e6792a..a54ed13ff66 100644
--- a/be/src/format_v2/delimited_text/csv_reader.h
+++ b/be/src/format_v2/delimited_text/csv_reader.h
@@ -60,6 +60,7 @@ private:
Status _create_line_reader() override;
Status _validate_line(const Slice& line) override;
void _split_line(const Slice& line) override;
+ void _prepare_columns(std::vector<MutableColumnPtr>& columns, size_t
batch_size) override;
Status _deserialize_one_cell(const RequestedColumn& column, IColumn*
output,
Slice value) override;
Slice _normalize_value(Slice value) const override;
diff --git a/be/src/format_v2/delimited_text/delimited_text_reader.cpp
b/be/src/format_v2/delimited_text/delimited_text_reader.cpp
index 7aa9dff1edf..42357257ddb 100644
--- a/be/src/format_v2/delimited_text/delimited_text_reader.cpp
+++ b/be/src/format_v2/delimited_text/delimited_text_reader.cpp
@@ -43,6 +43,7 @@
#include "runtime/file_scan_profile.h"
#include "runtime/runtime_state.h"
#include "util/decompressor.h"
+#include "util/defer_op.h"
#include "util/string_util.h"
#include "util/utf8_check.h"
@@ -192,11 +193,21 @@ void DelimitedTextReader::_init_profile() {
_text_profile.create_line_reader_time =
ADD_CHILD_TIMER_WITH_LEVEL(_profile, "CreateLineReaderTime",
DELIMITED_TEXT_PROFILE, 1);
_text_profile.read_line_time =
- ADD_CHILD_TIMER_WITH_LEVEL(_profile, "ReadLineTime",
DELIMITED_TEXT_PROFILE, 1);
+ ADD_CHILD_TIMER_WITH_LEVEL(_profile, "SampledReadLineTime",
DELIMITED_TEXT_PROFILE, 1);
_text_profile.split_line_time =
- ADD_CHILD_TIMER_WITH_LEVEL(_profile, "SplitLineTime",
DELIMITED_TEXT_PROFILE, 1);
- _text_profile.deserialize_time =
- ADD_CHILD_TIMER_WITH_LEVEL(_profile, "DeserializeTime",
DELIMITED_TEXT_PROFILE, 1);
+ ADD_CHILD_TIMER_WITH_LEVEL(_profile, "SampledSplitLineTime",
DELIMITED_TEXT_PROFILE, 1);
+ _text_profile.deserialize_time = ADD_CHILD_TIMER_WITH_LEVEL(_profile,
"SampledDeserializeTime",
+
DELIMITED_TEXT_PROFILE, 1);
+ _text_profile.split_prefix_time =
+ ADD_CHILD_TIMER_WITH_LEVEL(_profile, "SplitPrefixTime",
DELIMITED_TEXT_PROFILE, 1);
+ _text_profile.sampled_read_line_calls = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "SampledReadLineCalls", TUnit::UNIT,
DELIMITED_TEXT_PROFILE, 1);
+ _text_profile.sampled_parsed_rows = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "SampledParsedRows", TUnit::UNIT,
DELIMITED_TEXT_PROFILE, 1);
+ _profile->add_info_string(
+ "DelimitedTextTiming",
+ "Stage timers are unscaled samples of every 1024th read_line call "
+ "(starting with the first); DelimitedTextReader is full elapsed
time");
_text_profile.conjunct_filter_time =
ADD_CHILD_TIMER_WITH_LEVEL(_profile, "ConjunctFilterTime",
DELIMITED_TEXT_PROFILE, 1);
_text_profile.delete_conjunct_filter_time = ADD_CHILD_TIMER_WITH_LEVEL(
@@ -285,6 +296,7 @@ Status DelimitedTextReader::init(RuntimeState* state) {
Status DelimitedTextReader::infer_schema(RuntimeState* state,
std::vector<ColumnDefinition>*
schema) {
+ Defer profile_flush {[this] { _flush_profile(); }};
_schema_only = true;
RETURN_IF_ERROR(init(state));
if (_file_description->range_start_offset != 0) {
@@ -360,8 +372,13 @@ Status
DelimitedTextReader::open(std::shared_ptr<FileScanRequest> request) {
return Status::OK();
}
+void DelimitedTextReader::set_batch_size(size_t batch_size) {
+ _batch_size = std::max(size_t {1}, batch_size);
+}
+
Status DelimitedTextReader::get_block(Block* file_block, size_t* rows, bool*
eof) {
SCOPED_TIMER(_text_profile.total_time);
+ Defer profile_flush {[this] { _flush_profile(); }};
DORIS_CHECK(file_block != nullptr);
DORIS_CHECK(rows != nullptr);
DORIS_CHECK(eof != nullptr);
@@ -369,7 +386,9 @@ Status DelimitedTextReader::get_block(Block* file_block,
size_t* rows, bool* eof
return Status::InternalError("{} v2 reader is not open", _reader_name);
}
- const auto batch_size = _runtime_state != nullptr ?
_runtime_state->batch_size() : 4096;
+ const size_t batch_size =
+ _batch_size != 0 ? _batch_size
+ : (_runtime_state != nullptr ?
_runtime_state->batch_size() : 4096);
const auto max_block_bytes = _runtime_state != nullptr
?
_runtime_state->preferred_block_size_bytes()
: std::numeric_limits<size_t>::max();
@@ -379,6 +398,7 @@ Status DelimitedTextReader::get_block(Block* file_block,
size_t* rows, bool* eof
{
auto columns_guard = file_block->mutate_columns_scoped();
auto& columns = columns_guard.mutable_columns();
+ _prepare_columns(columns, batch_size);
// Delimited text readers are column-pruned but not lazy materialized:
all file-local
// columns requested by TableReader are decoded before file-local
conjuncts are evaluated.
while (*rows < batch_size && !_line_reader_eof &&
@@ -437,6 +457,7 @@ Status DelimitedTextReader::get_block(Block* file_block,
size_t* rows, bool* eof
Status DelimitedTextReader::get_aggregate_result(const FileAggregateRequest&
request,
FileAggregateResult* result) {
SCOPED_TIMER(_text_profile.total_time);
+ Defer profile_flush {[this] { _flush_profile(); }};
DORIS_CHECK(result != nullptr);
if (request.agg_type != TPushAggOp::type::COUNT) {
return Status::NotSupported("{} v2 reader only supports COUNT
aggregate pushdown",
@@ -461,7 +482,7 @@ Status DelimitedTextReader::get_aggregate_result(const
FileAggregateRequest& req
break;
}
if (line.size == 0) {
- update_counter(_text_profile.empty_lines_read, 1);
+ ++_pending_profile.empty_lines_read;
if (_empty_line_as_record() ||
(_runtime_state != nullptr &&
_runtime_state->is_read_csv_empty_line_as_null())) {
++count;
@@ -638,7 +659,7 @@ Status DelimitedTextReader::_read_next_line(Slice* line,
bool* eof) {
DORIS_CHECK(line != nullptr);
DORIS_CHECK(eof != nullptr);
if (_align_split_prefix) {
- SCOPED_TIMER(_text_profile.read_line_time);
+ SCOPED_TIMER(_text_profile.split_prefix_time);
DCHECK_EQ(_skip_lines, 1);
size_t skipped_lines = 0;
auto* text_reader =
assert_cast<NewPlainTextLineReader*>(_line_reader.get());
@@ -648,13 +669,19 @@ Status DelimitedTextReader::_read_next_line(Slice* line,
bool* eof) {
_align_split_prefix = false;
_skip_lines = 0;
_bom_removed = true;
- update_counter(_text_profile.skipped_lines, skipped_lines);
+ _pending_profile.skipped_lines += skipped_lines;
}
while (true) {
const uint8_t* ptr = nullptr;
size_t size = 0;
{
- SCOPED_TIMER(_text_profile.read_line_time);
+ _sample_current_line =
+ _profile != nullptr && (_read_line_calls++ %
PROFILE_SAMPLE_INTERVAL == 0);
+ std::optional<ScopedRawTimer<MonotonicStopWatch, int64_t>> timer;
+ if (_sample_current_line) {
+ ++_pending_profile.sampled_read_line_calls;
+ timer.emplace(&_pending_profile.read_line_time);
+ }
RETURN_IF_ERROR(_line_reader->read_line(&ptr, &size,
&_line_reader_eof, _io_ctx.get()));
}
if (_line_reader_eof && size == 0) {
@@ -670,12 +697,12 @@ Status DelimitedTextReader::_read_next_line(Slice* line,
bool* eof) {
if (_skip_lines > 0) {
--_skip_lines;
_bom_removed = true;
- update_counter(_text_profile.skipped_lines, 1);
+ ++_pending_profile.skipped_lines;
continue;
}
*line = Slice(ptr, size);
*eof = false;
- update_counter(_text_profile.raw_lines_read, 1);
+ ++_pending_profile.raw_lines_read;
return Status::OK();
}
}
@@ -685,12 +712,12 @@ Status DelimitedTextReader::_fill_columns_from_line(const
Slice& line,
size_t* rows) {
DORIS_CHECK(columns != nullptr);
if (line.size == 0) {
- update_counter(_text_profile.empty_lines_read, 1);
+ ++_pending_profile.empty_lines_read;
if (!_empty_line_as_record()) {
if (_runtime_state != nullptr &&
_runtime_state->is_read_csv_empty_line_as_null()) {
for (const auto& column : _requested_columns) {
RETURN_IF_ERROR(_append_null((*columns)[column.block_position.value()].get()));
- update_counter(_text_profile.cells_deserialized, 1);
+ ++_pending_profile.cells_deserialized;
}
++(*rows);
}
@@ -700,10 +727,17 @@ Status DelimitedTextReader::_fill_columns_from_line(const
Slice& line,
RETURN_IF_ERROR(_validate_line(line));
{
- SCOPED_TIMER(_text_profile.split_line_time);
+ std::optional<ScopedRawTimer<MonotonicStopWatch, int64_t>> timer;
+ if (_sample_current_line) {
+ ++_pending_profile.sampled_parsed_rows;
+ timer.emplace(&_pending_profile.split_line_time);
+ }
_split_line(line);
}
- SCOPED_TIMER(_text_profile.deserialize_time);
+ std::optional<ScopedRawTimer<MonotonicStopWatch, int64_t>> timer;
+ if (_sample_current_line) {
+ timer.emplace(&_pending_profile.deserialize_time);
+ }
if (_request->reject_invalid_rows) {
const auto expected =
cast_set<size_t>(_scan_params->num_of_columns_from_file);
const bool ignore_extra =
_scan_params->file_attributes.ignore_csv_redundant_col;
@@ -722,12 +756,31 @@ Status DelimitedTextReader::_fill_columns_from_line(const
Slice& line,
? _split_values[field_index]
: Slice(_options.null_format, _options.null_len);
RETURN_IF_ERROR(_deserialize_one_cell(column, output,
_normalize_value(value)));
- update_counter(_text_profile.cells_deserialized, 1);
+ ++_pending_profile.cells_deserialized;
}
++(*rows);
return Status::OK();
}
+void DelimitedTextReader::_flush_profile() {
+ update_counter(_text_profile.raw_lines_read,
_pending_profile.raw_lines_read);
+ update_counter(_text_profile.empty_lines_read,
_pending_profile.empty_lines_read);
+ update_counter(_text_profile.skipped_lines,
_pending_profile.skipped_lines);
+ update_counter(_text_profile.cells_deserialized,
_pending_profile.cells_deserialized);
+ update_counter(_text_profile.read_line_time,
_pending_profile.read_line_time);
+ update_counter(_text_profile.split_line_time,
_pending_profile.split_line_time);
+ update_counter(_text_profile.deserialize_time,
_pending_profile.deserialize_time);
+ update_counter(_text_profile.sampled_read_line_calls,
_pending_profile.sampled_read_line_calls);
+ update_counter(_text_profile.sampled_parsed_rows,
_pending_profile.sampled_parsed_rows);
+ _pending_profile = {};
+}
+
+void DelimitedTextReader::_prepare_columns(std::vector<MutableColumnPtr>&
columns,
+ size_t batch_size) {
+ (void)columns;
+ (void)batch_size;
+}
+
Status DelimitedTextReader::_validate_line(const Slice& line) {
(void)line;
return Status::OK();
diff --git a/be/src/format_v2/delimited_text/delimited_text_reader.h
b/be/src/format_v2/delimited_text/delimited_text_reader.h
index 645ba26eecd..4b7fc1f4014 100644
--- a/be/src/format_v2/delimited_text/delimited_text_reader.h
+++ b/be/src/format_v2/delimited_text/delimited_text_reader.h
@@ -55,6 +55,7 @@ public:
TableColumnMapperOptions options) const override;
Status open(std::shared_ptr<FileScanRequest> request) override;
Status get_block(Block* file_block, size_t* rows, bool* eof) override;
+ void set_batch_size(size_t batch_size) override;
Status get_aggregate_result(const FileAggregateRequest& request,
FileAggregateResult* result) override;
Status close() override;
@@ -64,6 +65,9 @@ protected:
RuntimeProfile::Counter* total_time = nullptr;
RuntimeProfile::Counter* open_file_time = nullptr;
RuntimeProfile::Counter* create_line_reader_time = nullptr;
+ RuntimeProfile::Counter* split_prefix_time = nullptr;
+ RuntimeProfile::Counter* sampled_read_line_calls = nullptr;
+ RuntimeProfile::Counter* sampled_parsed_rows = nullptr;
RuntimeProfile::Counter* read_line_time = nullptr;
RuntimeProfile::Counter* split_line_time = nullptr;
RuntimeProfile::Counter* deserialize_time = nullptr;
@@ -112,6 +116,8 @@ protected:
// Split one logical line into `_split_values`. The common materialization
path then resolves
// requested field ordinals against `_split_values`.
virtual void _split_line(const Slice& line) = 0;
+ // Prepare mutable output columns once per batch, after Block has resolved
copy-on-write.
+ virtual void _prepare_columns(std::vector<MutableColumnPtr>& columns,
size_t batch_size);
// Deserialize a single normalized field into the requested output column
using the
// format-specific serde API.
virtual Status _deserialize_one_cell(const RequestedColumn& column,
IColumn* output,
@@ -178,6 +184,26 @@ private:
Status _fill_columns_from_line(const Slice& line,
std::vector<MutableColumnPtr>* columns,
size_t* rows);
+ void _flush_profile();
+
+ // Publish exact counts once per API call instead of updating atomic
counters per cell.
+ struct PendingProfile {
+ int64_t raw_lines_read = 0;
+ int64_t empty_lines_read = 0;
+ int64_t skipped_lines = 0;
+ int64_t cells_deserialized = 0;
+ int64_t read_line_time = 0;
+ int64_t split_line_time = 0;
+ int64_t deserialize_time = 0;
+ int64_t sampled_read_line_calls = 0;
+ int64_t sampled_parsed_rows = 0;
+ } _pending_profile;
+ // Keep sampling periodic across batches, including skipped headers and
EOF probes.
+ static constexpr size_t PROFILE_SAMPLE_INTERVAL = 1024;
+ uint64_t _read_line_calls = 0;
+ bool _sample_current_line = false;
+ // Zero means use the runtime default until TableReader supplies its
predicted row cap.
+ size_t _batch_size = 0;
std::string _reader_name;
};
diff --git a/be/test/format_v2/delimited_text/csv_reader_test.cpp
b/be/test/format_v2/delimited_text/csv_reader_test.cpp
index 8493b216e06..6c9682ad151 100644
--- a/be/test/format_v2/delimited_text/csv_reader_test.cpp
+++ b/be/test/format_v2/delimited_text/csv_reader_test.cpp
@@ -752,9 +752,9 @@ TEST_F(CsvV2ReaderTest,
ProfileCountersTrackReadParseDeserializeAndFilter) {
EXPECT_NE(_profile.get_counter("OpenFileTime"), nullptr);
EXPECT_NE(_profile.get_counter("CreateLineReaderTime"), nullptr);
- EXPECT_NE(_profile.get_counter("ReadLineTime"), nullptr);
- EXPECT_NE(_profile.get_counter("SplitLineTime"), nullptr);
- EXPECT_NE(_profile.get_counter("DeserializeTime"), nullptr);
+ EXPECT_NE(_profile.get_counter("SampledReadLineTime"), nullptr);
+ EXPECT_NE(_profile.get_counter("SampledSplitLineTime"), nullptr);
+ EXPECT_NE(_profile.get_counter("SampledDeserializeTime"), nullptr);
EXPECT_NE(_profile.get_counter("ConjunctFilterTime"), nullptr);
EXPECT_NE(_profile.get_counter("DeleteConjunctFilterTime"), nullptr);
EXPECT_EQ(counter_value(&_profile, "RawLinesRead"), 3);
@@ -769,6 +769,162 @@ TEST_F(CsvV2ReaderTest,
ProfileCountersTrackReadParseDeserializeAndFilter) {
EXPECT_EQ(counter_value(&_profile, "CellsDeserialized"), 6);
}
+// A row cap can change between calls. It must override the runtime default
without dropping
+// records, and a zero prediction must still make progress.
+TEST_F(CsvV2ReaderTest, HonorsChangingBatchSizeAndZeroPrediction) {
+ std::ofstream output(_file_path, std::ios::binary);
+ output << "id,name,score\n";
+ for (int i = 1; i <= 7; ++i) {
+ output << i << ",name,10\n";
+ }
+ output.close();
+ _state._batch_size = 2;
+ auto reader = create_reader(_file_path, &_params, _slots, &_state,
&_profile);
+ reader->set_batch_size(1);
+ auto request = std::make_shared<FileScanRequest>();
+ request->local_positions.emplace(LocalColumnId(0), LocalIndex(0));
+ request->local_positions.emplace(LocalColumnId(1), LocalIndex(1));
+ ASSERT_TRUE(reader->open(request).ok());
+ std::vector<ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ int next_id = 1;
+ const std::vector<size_t> caps {1, 3, 0, 2, 2};
+ const std::vector<size_t> expected {1, 3, 1, 2, 0};
+ for (size_t i = 0; i < caps.size(); ++i) {
+ reader->set_batch_size(caps[i]);
+ auto block = make_block(schema, {0, 1});
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ ASSERT_EQ(rows, expected[i]);
+ EXPECT_EQ(eof, rows == 0);
+ for (size_t row = 0; row < rows; ++row) {
+ EXPECT_EQ(nullable_int_at(*block.get_by_position(0).column, row),
next_id++);
+ EXPECT_EQ(nullable_string_at(*block.get_by_position(1).column,
row), "name");
+ }
+ EXPECT_EQ(counter_value(&_profile, "CellsDeserialized"), (next_id - 1)
* 2);
+ EXPECT_EQ(counter_value(&_profile, "RawLinesRead"), next_id - 1);
+ }
+ EXPECT_EQ(next_id, 8);
+ EXPECT_EQ(counter_value(&_profile, "SkippedLines"), 1);
+}
+
+TEST_F(CsvV2ReaderTest, BatchPredictionStillHonorsByteBudget) {
+ const bool saved = config::enable_adaptive_batch_size;
+ Defer restore {[&] { config::enable_adaptive_batch_size = saved; }};
+ config::enable_adaptive_batch_size = true;
+ _state._query_options.__set_preferred_block_size_bytes(1);
+ auto reader = create_reader(_file_path, &_params, _slots, &_state,
&_profile);
+ reader->set_batch_size(100);
+ auto request = std::make_shared<FileScanRequest>();
+ request->local_positions.emplace(LocalColumnId(1), LocalIndex(0));
+ ASSERT_TRUE(reader->open(request).ok());
+ std::vector<ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ for (const std::string expected : {"alice", "bob"}) {
+ auto block = make_block(schema, {1});
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ ASSERT_EQ(rows, 1);
+ EXPECT_EQ(nullable_string_at(*block.get_by_position(0).column, 0),
expected);
+ }
+}
+
+// Sampling spans batch boundaries; exact work counts are published before
every return.
+TEST_F(CsvV2ReaderTest, ProfileSamplingSpansBatchesWithExactCounts) {
+ std::ofstream output(_file_path, std::ios::binary);
+ for (int i = 0; i < 2050; ++i) {
+ output << "1,name,10\n";
+ }
+ output.close();
+ _params.file_attributes.__set_header_type("");
+ auto reader = create_reader(_file_path, &_params, _slots, &_state,
&_profile);
+ reader->set_batch_size(300);
+ auto request = std::make_shared<FileScanRequest>();
+ request->local_positions.emplace(LocalColumnId(1), LocalIndex(0));
+ ASSERT_TRUE(reader->open(request).ok());
+ std::vector<ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ size_t total = 0;
+ bool eof = false;
+ while (!eof) {
+ auto block = make_block(schema, {1});
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ total += rows;
+ EXPECT_EQ(counter_value(&_profile, "RawLinesRead"), total);
+ EXPECT_EQ(counter_value(&_profile, "CellsDeserialized"), total);
+ }
+ EXPECT_EQ(total, 2050);
+ EXPECT_EQ(counter_value(&_profile, "SampledReadLineCalls"), 3);
+ EXPECT_EQ(counter_value(&_profile, "SampledParsedRows"), 3);
+}
+
+// The single-byte split and direct string insertion must preserve NULL,
quoted literals,
+// escaping, trailing empty fields, missing fields and reordered projections
across batches.
+TEST_F(CsvV2ReaderTest,
StringFastPathPreservesValuesForSingleAndMultiByteSeparators) {
+ auto type = make_nullable(std::make_shared<DataTypeString>());
+ std::vector<SlotDescriptor*> slots;
+ for (int i = 0; i < 3; ++i) {
+ slots.push_back(make_test_slot(&_pool, i, i, type, "s" +
std::to_string(i)));
+ }
+ for (const std::string separator : {"|", "||"}) {
+ for (bool empty_as_null : {false, true}) {
+ SCOPED_TRACE(testing::Message() << separator << ", empty_as_null="
<< empty_as_null);
+ std::ofstream output(_file_path, std::ios::binary);
+ output << "NULL" << separator << "\"NULL\"" << separator << "\n";
+ output << "left" << separator << "a\\b" << separator << "tail\n";
+ output << "only\n";
+ output.close();
+ auto params = _params;
+ params.file_attributes.__set_header_type("");
+ params.file_attributes.__set_trim_double_quotes(true);
+
params.file_attributes.text_params.__set_column_separator(separator);
+ params.file_attributes.text_params.__set_null_format("NULL");
+
params.file_attributes.text_params.__set_empty_field_as_null(empty_as_null);
+ params.file_attributes.text_params.__set_escape('\\');
+ auto reader = create_reader(_file_path, ¶ms, slots, &_state,
&_profile);
+ reader->set_batch_size(1);
+ auto request = std::make_shared<FileScanRequest>();
+ request->local_positions.emplace(LocalColumnId(2), LocalIndex(0));
+ request->local_positions.emplace(LocalColumnId(0), LocalIndex(1));
+ request->local_positions.emplace(LocalColumnId(1), LocalIndex(2));
+ ASSERT_TRUE(reader->open(request).ok());
+ std::vector<ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ for (int row = 0; row < 3; ++row) {
+ auto block = make_block(schema, {2, 0, 1});
+ // Hold shared columns while the reader acquires its mutable
batch.
+ auto shared = block.get_by_position(0).column;
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ ASSERT_EQ(rows, 1);
+ EXPECT_EQ(shared->size(), 0);
+ const auto& c0 = *block.get_by_position(0).column;
+ const auto& c1 = *block.get_by_position(1).column;
+ const auto& c2 = *block.get_by_position(2).column;
+ if (row == 0) {
+ EXPECT_EQ(is_null_at(c0, 0), empty_as_null);
+ EXPECT_EQ(nullable_string_at(c0, 0), "");
+ EXPECT_TRUE(is_null_at(c1, 0));
+ EXPECT_FALSE(is_null_at(c2, 0));
+ EXPECT_EQ(nullable_string_at(c2, 0), "NULL");
+ } else if (row == 1) {
+ EXPECT_EQ(nullable_string_at(c0, 0), "tail");
+ EXPECT_EQ(nullable_string_at(c1, 0), "left");
+ EXPECT_EQ(nullable_string_at(c2, 0), "ab");
+ } else {
+ EXPECT_TRUE(is_null_at(c0, 0));
+ EXPECT_EQ(nullable_string_at(c1, 0), "only");
+ EXPECT_TRUE(is_null_at(c2, 0));
+ }
+ }
+ }
+ }
+}
+
// Scenario: CSV has no embedded nested schema, but TableColumnMapper still
needs semantic children
// for complex table columns. The reader synthesizes ARRAY/MAP/STRUCT children
from the slot type
// while keeping the top-level local id as the CSV field ordinal from
column_idxs.
@@ -828,6 +984,8 @@ TEST_F(CsvV2ReaderTest,
LoadRejectsWrongFieldCountsWithoutShiftingFollowingRows)
EXPECT_EQ(nullable_int_at(*block.get_by_position(0).column, 0), 1);
EXPECT_EQ(nullable_int_at(*block.get_by_position(0).column, 1), 4);
EXPECT_EQ(io_ctx->load_filtered_rows, 2);
+ EXPECT_EQ(counter_value(&_profile, "RawLinesRead"), 4);
+ EXPECT_EQ(counter_value(&_profile, "CellsDeserialized"), 6);
}
// Scenario: CSV v2 honors FileScanRequest local positions, so TableReader can
request a subset of
@@ -880,6 +1038,9 @@ TEST_F(CsvV2ReaderTest,
InvalidUtf8FailsWhenValidationEnabled) {
bool eof = false;
const auto status = reader->get_block(&block, &rows, &eof);
EXPECT_FALSE(status.ok());
+ EXPECT_EQ(counter_value(&_profile, "RawLinesRead"), 1);
+ EXPECT_EQ(counter_value(&_profile, "SkippedLines"), 1);
+ EXPECT_EQ(counter_value(&_profile, "CellsDeserialized"), 0);
EXPECT_TRUE(status.to_string().find("Only support csv data in utf8 codec")
!= std::string::npos)
<< status;
}
@@ -1103,6 +1264,10 @@ TEST_F(CsvV2ReaderTest, CountAggregateScansRows) {
FileAggregateResult aggregate_result;
ASSERT_TRUE(reader->get_aggregate_result(aggregate_request,
&aggregate_result).ok());
EXPECT_EQ(aggregate_result.count, 2);
+ EXPECT_EQ(counter_value(&_profile, "RawLinesRead"), 2);
+ EXPECT_EQ(counter_value(&_profile, "SkippedLines"), 1);
+ EXPECT_EQ(counter_value(&_profile, "CellsDeserialized"), 0);
+ EXPECT_EQ(counter_value(&_profile, "SampledParsedRows"), 0);
}
TEST_F(CsvV2ReaderTest, CountNullableColumnFallsBackToRowMaterialization) {
diff --git a/be/test/format_v2/delimited_text/text_reader_test.cpp
b/be/test/format_v2/delimited_text/text_reader_test.cpp
index 4927ee7622e..2fe1aa84bef 100644
--- a/be/test/format_v2/delimited_text/text_reader_test.cpp
+++ b/be/test/format_v2/delimited_text/text_reader_test.cpp
@@ -408,9 +408,9 @@ TEST_F(TextV2ReaderTest,
ProfileCountersTrackReadParseDeserializeAndFilter) {
EXPECT_NE(_profile.get_counter("OpenFileTime"), nullptr);
EXPECT_NE(_profile.get_counter("CreateLineReaderTime"), nullptr);
- EXPECT_NE(_profile.get_counter("ReadLineTime"), nullptr);
- EXPECT_NE(_profile.get_counter("SplitLineTime"), nullptr);
- EXPECT_NE(_profile.get_counter("DeserializeTime"), nullptr);
+ EXPECT_NE(_profile.get_counter("SampledReadLineTime"), nullptr);
+ EXPECT_NE(_profile.get_counter("SampledSplitLineTime"), nullptr);
+ EXPECT_NE(_profile.get_counter("SampledDeserializeTime"), nullptr);
EXPECT_NE(_profile.get_counter("ConjunctFilterTime"), nullptr);
EXPECT_NE(_profile.get_counter("DeleteConjunctFilterTime"), nullptr);
EXPECT_EQ(counter_value(&_profile, "RawLinesRead"), 3);
@@ -424,6 +424,32 @@ TEST_F(TextV2ReaderTest,
ProfileCountersTrackReadParseDeserializeAndFilter) {
EXPECT_EQ(counter_value(&_profile, "CellsDeserialized"), 6);
}
+TEST_F(TextV2ReaderTest, HonorsBatchSizeAcrossEmptyAndFilteredBlocks) {
+ auto reader = create_reader(_file_path, &_params, _slots, &_state,
&_profile);
+ reader->set_batch_size(1);
+ auto request = std::make_shared<FileScanRequest>();
+ request->local_positions.emplace(LocalColumnId(0), LocalIndex(0));
+ request->conjuncts = {
+ prepared_conjunct(&_state,
std::make_shared<NullableIntGreaterThanExpr>(0, 1))};
+ ASSERT_TRUE(reader->open(request).ok());
+ std::vector<ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ for (int batch = 0; batch < 3; ++batch) {
+ auto block = make_block(schema, {0});
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ EXPECT_EQ(rows, batch == 1 ? 1 : 0);
+ EXPECT_EQ(eof, batch == 2);
+ if (rows == 1) {
+ EXPECT_EQ(nullable_int_at(*block.get_by_position(0).column, 0), 2);
+ }
+ }
+ EXPECT_EQ(counter_value(&_profile, "RawLinesRead"), 2);
+ EXPECT_EQ(counter_value(&_profile, "CellsDeserialized"), 2);
+ EXPECT_EQ(counter_value(&_profile, "DelimitedRowsFilteredByConjunct"), 1);
+}
+
TEST_F(TextV2ReaderTest,
FormatProfilesKeepDistinctCountersInBothInitializationOrders) {
auto expect_format_children = [](RuntimeProfile* profile) {
TRuntimeProfileTree tree;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
