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 4bde8d0fb28 [fix](be) Preserve JSON and WAL rows with empty file
projections (#70154)
4bde8d0fb28 is described below
commit 4bde8d0fb28452c9bb8314febbe05c4d63dde93c
Author: Gabriel <[email protected]>
AuthorDate: Wed Nov 4 14:00:35 2026 +0800
[fix](be) Preserve JSON and WAL rows with empty file projections (#70154)
### What problem does this PR solve?
Issue Number: close #69926
Problem Summary:
Hive JSON queries that only need partition columns can return zero rows
even when the selected files contain records. For example, `SELECT * ...
WHERE dt = 'p1'` returns data while `SELECT COUNT(*) ... WHERE dt =
'p1'` returns zero. The V2 JSON reader derived its row count from the
output block, which has no columns after physical field pruning.
Writer-backed reader tests reproduce the same row-count loss for empty
WAL projections.
Track accepted JSON records independently of materialized columns,
including batch progress, array expansion, error rollback, and
ignore-malformed NULL rows. Preserve JSON-path record selection for
empty projections, and let the existing TableReader materialize
partition values using the returned row count.
For WAL, save the deserialized source row count before materialization
transfers column ownership, then pass it through the existing filter
pipeline. This preserves the row count when no file columns are
requested. The empty-projection reproduction is at the reader interface;
production replay reachability of an empty projection remains
unverified.
Add ten JSON BE tests covering empty file schemas and projections,
arrays, JSON roots/paths, malformed input, batch boundaries, EOF, and
actual TableReader partition materialization. Add eight writer-backed
WAL cases covering empty, single-column, and reordered full projections
across multiple blocks, with and without schema discovery, plus repeated
EOF reads. Add a WAL SQL regression that forces two original async-load
transactions to abort, then verifies row counts, NULLs, aggregates,
filters, and actual values after replay. Add a Hive SQL regression using
the existing partitioned JSON fixture for COUNT, partition projections,
grouping, and file-column filters with external COUNT pushdown enabled
and disabled.
### Release note
Fix incorrect zero counts and missing rows when querying only partition
columns of Hive JSON tables. Preserve logical row counts for empty WAL
reader projections.
### Check List (For Author)
- Test
- [x] Unit Test: 166 ASAN tests passed, including JSON reader,
TableReader, WAL V2 reader, WAL reader/writer, and CSV/Text
empty-projection controls. Before the JSON fix, nine of ten new tests
failed; before the WAL fix, both empty-projection cases failed while six
projection controls passed.
- [x] WAL SQL regression: `test_group_commit_wal_row_count` passed twice
on isolated Doris 4.1.5 with scanner V2 enabled. Generated the `.out`
file with `run-regression-test.sh`, then reran for result comparison.
Execution on master is pending CI.
- [x] Hive SQL regression: added `test_hive_json_partition_only`; Groovy
compilation passed. Full execution against the external Hive fixture is
pending CI.
- Behavior changed:
- [x] Yes. Empty JSON and WAL file projections preserve logical row
counts. Hive JSON partition-only queries produce the correct rows and
aggregates.
- Does this need documentation?
- [x] No.
---
be/src/format_v2/json/json_reader.cpp | 53 +++++--
be/src/format_v2/json/json_reader.h | 3 +
be/src/format_v2/wal/wal_reader.cpp | 5 +-
be/test/format_v2/json/json_reader_test.cpp | 162 +++++++++++++++++++++
be/test/format_v2/wal/wal_reader_test.cpp | 120 +++++++++++++++
.../test_group_commit_wal_row_count.out | 17 +++
.../hive/test_hive_json_partition_only.groovy | 85 +++++++++++
.../test_group_commit_wal_row_count.groovy | 85 +++++++++++
8 files changed, 517 insertions(+), 13 deletions(-)
diff --git a/be/src/format_v2/json/json_reader.cpp
b/be/src/format_v2/json/json_reader.cpp
index 4052bed03d0..db8a8e93c8a 100644
--- a/be/src/format_v2/json/json_reader.cpp
+++ b/be/src/format_v2/json/json_reader.cpp
@@ -385,14 +385,14 @@ Status JsonReader::get_block(Block* file_block, size_t*
rows, bool* eof) {
return Status::InternalError("JSON v2 reader is not open");
}
- const auto batch_size = _runtime_state->batch_size();
+ const auto batch_size = cast_set<size_t>(_runtime_state->batch_size());
const auto max_block_bytes = _runtime_state->preferred_block_size_bytes();
*rows = 0;
*eof = false;
+ _logical_rows = 0;
_missing_fields.clear();
- while (file_block->rows() < batch_size && !_reader_eof &&
- file_block->bytes() < max_block_bytes) {
+ while (_logical_rows < batch_size && !_reader_eof && file_block->bytes() <
max_block_bytes) {
if (_read_json_by_line && _skip_first_line) {
RETURN_IF_ERROR(_line_reader->skip_split_prefix(
_reader_range.start_offset, _line_delimiter, &_reader_eof,
_io_ctx.get()));
@@ -400,7 +400,7 @@ Status JsonReader::get_block(Block* file_block, size_t*
rows, bool* eof) {
continue;
}
- const size_t original_rows = file_block->rows();
+ const size_t original_rows = _logical_rows;
size_t size = 0;
bool is_empty_row = false;
Status st = Status::OK();
@@ -429,12 +429,13 @@ Status JsonReader::get_block(Block* file_block, size_t*
rows, bool* eof) {
}
// An ignored or empty JSON object can produce no row. Avoid spinning
forever on a document
// that was consumed but produced no materialized value.
- if (!is_empty_row && file_block->rows() == original_rows) {
+ if (!is_empty_row && _logical_rows == original_rows) {
break;
}
}
- *rows = file_block->rows();
+ *rows = _logical_rows;
+ DCHECK(file_block->columns() == 0 || file_block->rows() == *rows);
for (auto& [_, mask] : _missing_fields) {
mask.resize(*rows, 0);
}
@@ -810,7 +811,7 @@ Status JsonReader::_append_simple_json_rows(Block* block,
bool* is_empty_row, bo
_array_iter = _array.begin();
while (_array_iter != _array.end()) {
simdjson::ondemand::object object_value =
(*_array_iter).get_object();
- const size_t row_start = block->rows();
+ const size_t row_start = _logical_rows;
const auto status = _set_column_values_from_object(&object_value,
block, &valid);
if (!status.ok()) {
if (!_request->reject_invalid_rows) {
@@ -841,12 +842,12 @@ Status
JsonReader::_append_flat_array_jsonpath_rows(Block* block, bool* is_empty
DORIS_CHECK(block != nullptr);
DORIS_CHECK(is_empty_row != nullptr);
DORIS_CHECK(eof != nullptr);
- const size_t original_rows = block->rows();
+ const size_t original_rows = _logical_rows;
bool valid = true;
_array = _json_value.get_array();
_array_iter = _array.begin();
while (_array_iter != _array.end()) {
- const size_t row_start = block->rows();
+ const size_t row_start = _logical_rows;
simdjson::ondemand::object object_value = (*_array_iter).get_object();
if (!_parsed_from_json_root && !_parsed_json_root.empty()) {
// Reject roots per element for loads, preserving earlier siblings
and their presence
@@ -885,7 +886,7 @@ Status JsonReader::_append_flat_array_jsonpath_rows(Block*
block, bool* is_empty
}
++_array_iter;
}
- *is_empty_row = block->rows() == original_rows;
+ *is_empty_row = _logical_rows == original_rows;
return Status::OK();
}
@@ -909,7 +910,7 @@ Status
JsonReader::_set_column_values_from_object(simdjson::ondemand::object* ob
DORIS_CHECK(block != nullptr);
DORIS_CHECK(valid != nullptr);
std::vector<bool> seen_columns(block->columns(), false);
- const size_t cur_row_count = block->rows();
+ const size_t cur_row_count = _logical_rows;
bool has_valid_value = false;
size_t key_index = 0;
@@ -960,21 +961,46 @@ Status
JsonReader::_set_column_values_from_object(simdjson::ondemand::object* ob
}
}
*valid = true;
+ ++_logical_rows;
if (!has_valid_value) {
return Status::OK();
}
return Status::OK();
}
+Status JsonReader::_has_matching_jsonpath(simdjson::ondemand::object*
object_value,
+ bool* matches) const {
+ *matches = false;
+ for (const auto& path : _parsed_jsonpaths) {
+ // A root path selects the current document without a field lookup.
+ if (JsonFunctions::is_root_path(path)) {
+ *matches = true;
+ continue;
+ }
+ simdjson::ondemand::value value;
+ const auto status = JsonFunctions::extract_from_object(*object_value,
path, &value);
+ if (!status.ok() && !status.is<ErrorCode::NOT_FOUND>()) {
+ return status;
+ }
+ *matches = *matches || status.ok();
+ }
+ return Status::OK();
+}
+
Status JsonReader::_write_columns_by_jsonpath(simdjson::ondemand::object*
object_value,
Block* block, bool* valid) {
DORIS_CHECK(object_value != nullptr);
DORIS_CHECK(block != nullptr);
DORIS_CHECK(valid != nullptr);
bool has_valid_value = false;
- const size_t cur_row_count = block->rows();
+ const size_t cur_row_count = _logical_rows;
_cached_string_values.clear();
+ // Even without projected fields, jsonpaths still decides whether the
object contributes a row.
+ if (_requested_columns.empty()) {
+ RETURN_IF_ERROR(_has_matching_jsonpath(object_value,
&has_valid_value));
+ }
+
for (size_t i = 0; i < _requested_columns.size(); ++i) {
const auto& requested = _requested_columns[i];
auto* column_ptr =
block->get_by_position(i).column->assert_mutable().get();
@@ -1031,6 +1057,7 @@ Status
JsonReader::_write_columns_by_jsonpath(simdjson::ondemand::object* object
return Status::OK();
}
*valid = true;
+ ++_logical_rows;
return Status::OK();
}
@@ -1233,6 +1260,7 @@ Status JsonReader::_append_null_for_malformed_json(Block*
block) {
assert_cast<ColumnNullable*>(column.get())->insert_default();
column_with_type.column = std::move(column);
}
+ ++_logical_rows;
return Status::OK();
}
@@ -1244,6 +1272,7 @@ Status JsonReader::_handle_json_error(const Status&
status, Block* block, size_t
// block to the row count before this document before either surfacing the
error or appending
// the ignore-malformed null row.
_truncate_block_to_rows(block, original_rows);
+ _logical_rows = original_rows;
for (auto& [_, mask] : _missing_fields) {
mask.resize(original_rows, 0);
}
diff --git a/be/src/format_v2/json/json_reader.h
b/be/src/format_v2/json/json_reader.h
index ba2a0d99e2f..b1c413da0e7 100644
--- a/be/src/format_v2/json/json_reader.h
+++ b/be/src/format_v2/json/json_reader.h
@@ -148,6 +148,7 @@ private:
Status _append_nested_jsonpath_row(Block* block, bool* is_empty_row, bool*
eof);
Status _set_column_values_from_object(simdjson::ondemand::object*
object_value, Block* block,
bool* valid);
+ Status _has_matching_jsonpath(simdjson::ondemand::object* object_value,
bool* matches) const;
Status _write_columns_by_jsonpath(simdjson::ondemand::object*
object_value, Block* block,
bool* valid);
template <bool use_string_cache>
@@ -191,6 +192,8 @@ private:
std::unique_ptr<Decompressor> _decompressor;
std::unique_ptr<NewPlainTextLineReader> _line_reader;
int64_t _current_offset = 0;
+ // Partition-only scans have no file columns, so Block::rows() cannot
count their records.
+ size_t _logical_rows = 0;
bool _reader_eof = false;
bool _skip_first_line = false;
bool _single_document_read = false;
diff --git a/be/src/format_v2/wal/wal_reader.cpp
b/be/src/format_v2/wal/wal_reader.cpp
index 55e88796c98..5f57a422ff8 100644
--- a/be/src/format_v2/wal/wal_reader.cpp
+++ b/be/src/format_v2/wal/wal_reader.cpp
@@ -195,8 +195,11 @@ Status WalReader::get_block(Block* file_block, size_t*
rows, bool* eof) {
return Status::Corruption("WAL block has {} columns but header
declares {}",
source_block.columns(), _column_ids.size());
}
+ // Empty projections still contribute every WAL row. Capture the count
before materialization
+ // moves source columns into the output block and can leave the source
with no owned columns.
+ const size_t source_rows = source_block.rows();
RETURN_IF_ERROR(_materialize_requested_columns(&source_block, file_block));
- *rows = file_block->rows();
+ *rows = source_rows;
_record_scan_rows(cast_set<int64_t>(*rows));
RETURN_IF_ERROR(
apply_materialized_reader_filters(_request.get(), _io_ctx.get(),
file_block, rows));
diff --git a/be/test/format_v2/json/json_reader_test.cpp
b/be/test/format_v2/json/json_reader_test.cpp
index b9711973de2..bf82fbcf01d 100644
--- a/be/test/format_v2/json/json_reader_test.cpp
+++ b/be/test/format_v2/json/json_reader_test.cpp
@@ -46,6 +46,7 @@
#include "exprs/vexpr.h"
#include "exprs/vexpr_context.h"
#include "format_v2/column_data.h"
+#include "format_v2/table_reader.h"
#include "io/io_common.h"
#include "runtime/descriptors.h"
#include "runtime/exec_env.h"
@@ -1031,4 +1032,165 @@ TEST(JsonReaderTest,
AppliesDeleteAndNormalConjunctsWithPredicateFilterAccountin
EXPECT_EQ(io_ctx->predicate_filtered_rows, 1);
}
+// Partition-only scans have no file-local output columns, but still need one
logical row per
+// accepted JSON record so TableReader can materialize the partition constants.
+TEST(JsonReaderTest, EmptyFileProjectionPreservesRows) {
+ ObjectPool pool;
+ const auto slots = build_slots(&pool);
+ for (const auto& source_slots : {slots, std::vector<SlotDescriptor*> {}}) {
+ auto result = read_once("empty_projection.jsonl",
"{\"id\":1}\n\n{}\n{\"id\":3}\n",
+ json_scan_params(), source_slots, {}, true,
true);
+ ASSERT_TRUE(result.status.ok()) << result.status;
+ EXPECT_EQ(result.block.columns(), 0);
+ EXPECT_EQ(result.rows, 3);
+ EXPECT_FALSE(result.eof);
+ EXPECT_TRUE(result.second_status.ok());
+ EXPECT_EQ(result.second_rows, 0);
+ EXPECT_TRUE(result.second_eof);
+ }
+}
+
+TEST(JsonReaderTest, EmptyFileProjectionPreservesOuterArrayRows) {
+ auto result = read_once("empty_projection_array.json",
R"([{"id":1},{},{"id":3}])",
+ json_scan_params(false, true), {}, {}, true, true);
+ ASSERT_TRUE(result.status.ok()) << result.status;
+ EXPECT_EQ(result.rows, 3);
+ EXPECT_TRUE(result.second_status.ok());
+ EXPECT_TRUE(result.second_eof);
+}
+
+TEST(JsonReaderTest, EmptyFileProjectionPreservesJsonRootRows) {
+ auto result = read_once("empty_projection_root.jsonl",
+
"{\"payload\":{\"id\":1}}\n{\"payload\":{\"id\":2}}\n",
+ json_scan_params(true, false, "", "$.payload"),
{}, {}, true, true);
+ ASSERT_TRUE(result.status.ok()) << result.status;
+ EXPECT_EQ(result.rows, 2);
+ EXPECT_TRUE(result.second_status.ok());
+ EXPECT_TRUE(result.second_eof);
+}
+
+TEST(JsonReaderTest, EmptyFileProjectionHonorsJsonPaths) {
+ ObjectPool pool;
+ const auto slots = build_slots(&pool);
+ auto result = read_once("empty_projection_jsonpaths.jsonl",
+
"{\"payload\":{\"id\":1}}\n{\"other\":2}\n{\"payload\":{\"id\":3}}\n",
+ json_scan_params(true, false,
R"(["$.payload.id","$.payload.user"])"),
+ slots, {}, true, true);
+ ASSERT_TRUE(result.status.ok()) << result.status;
+ EXPECT_EQ(result.rows, 2);
+ EXPECT_TRUE(result.second_status.ok());
+ EXPECT_TRUE(result.second_eof);
+}
+
+TEST(JsonReaderTest, EmptyFileProjectionPreservesRootJsonPathRows) {
+ auto result = read_once("empty_projection_root_path.jsonl",
"{\"id\":1}\n{}\n",
+ json_scan_params(true, false, R"(["$."])"), {},
{}, true, true);
+ ASSERT_TRUE(result.status.ok()) << result.status;
+ EXPECT_EQ(result.rows, 2);
+ EXPECT_TRUE(result.second_eof);
+}
+
+TEST(JsonReaderTest, EmptyFileProjectionPreservesRootedArrayJsonPathRows) {
+ auto result = read_once("empty_projection_rooted_array.json",
+
R"({"rows":[{"payload":{"id":1}},{"other":2},{"payload":{"id":3}}]})",
+ json_scan_params(false, true,
R"(["$.payload.id"])", "$.rows"), {}, {},
+ true, true);
+ ASSERT_TRUE(result.status.ok()) << result.status;
+ EXPECT_EQ(result.rows, 2);
+ EXPECT_TRUE(result.second_status.ok());
+ EXPECT_TRUE(result.second_eof);
+}
+
+TEST(JsonReaderTest, EmptyFileProjectionPreservesMalformedNullRows) {
+ auto result = read_once("empty_projection_malformed.jsonl",
"not-json\n{\"id\":1}\n",
+ json_scan_params(true, false, "", "", true), {},
{}, true, true);
+ ASSERT_TRUE(result.status.ok()) << result.status;
+ EXPECT_EQ(result.rows, 2);
+ EXPECT_TRUE(result.second_status.ok());
+ EXPECT_TRUE(result.second_eof);
+
+ auto strict = read_once("empty_projection_strict.jsonl", "not-json\n",
json_scan_params(), {},
+ {}, false, true);
+ EXPECT_FALSE(strict.status.ok());
+}
+
+TEST(JsonReaderTest, EmptyFileProjectionPreservesEmptyInput) {
+ for (const std::string content : {"", "\n", "[]\n"}) {
+ auto result = read_once("empty_projection_empty.jsonl", content,
+ json_scan_params(true, true), {}, {}, false,
true);
+ ASSERT_TRUE(result.status.ok()) << result.status;
+ EXPECT_EQ(result.rows, 0);
+ EXPECT_TRUE(result.eof);
+ }
+}
+
+TEST(JsonReaderTest, EmptyFileProjectionHonorsBatchSize) {
+ const auto path = write_json_file("empty_projection_batches.jsonl",
+ "{\"id\":1}\n{\"id\":2}\n{\"id\":3}\n"
+ "{\"id\":4}\n{\"id\":5}\n");
+ auto properties = std::make_shared<io::FileSystemProperties>();
+ properties->system_type = TFileType::FILE_LOCAL;
+ auto desc = file_description(path.string());
+ auto params = json_scan_params();
+ RuntimeProfile profile("json_empty_projection_batches");
+ MockRuntimeState state;
+ state._batch_size = 2;
+ JsonReader reader(properties, desc, std::make_shared<io::IOContext>(),
&profile, ¶ms,
+ file_range(path), {});
+ ASSERT_TRUE(reader.init(&state).ok());
+ ASSERT_TRUE(reader.open(std::make_shared<FileScanRequest>()).ok());
+ for (const size_t expected_rows : {2, 2, 1, 0}) {
+ Block block;
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(reader.get_block(&block, &rows, &eof).ok());
+ EXPECT_EQ(block.columns(), 0);
+ EXPECT_EQ(rows, expected_rows);
+ EXPECT_EQ(eof, expected_rows == 0);
+ }
+}
+
+TEST(JsonReaderTest, EmptyFileProjectionMaterializesPartitionRows) {
+ const auto path =
+ write_json_file("partition_only.jsonl",
"{\"id\":1}\n{\"id\":2}\n{\"id\":3}\n");
+ auto params = json_scan_params();
+ const std::vector<SlotDescriptor*> file_slots;
+ ColumnDefinition partition;
+ partition.identifier = Field::create_field<TYPE_STRING>("dt");
+ partition.local_id = 0;
+ partition.name = "dt";
+ partition.type = std::make_shared<DataTypeString>();
+ partition.is_partition_key = true;
+ RuntimeProfile profile("json_partition_only");
+ MockRuntimeState state;
+ state._batch_size = 2;
+ TableReader reader;
+ ASSERT_TRUE(reader.init({.projected_columns = {partition},
+ .conjuncts = {},
+ .format = FileFormat::JSON,
+ .scan_params = ¶ms,
+ .io_ctx = std::make_shared<io::IOContext>(),
+ .runtime_state = &state,
+ .scanner_profile = &profile,
+ .file_slot_descs = &file_slots})
+ .ok());
+ SplitReadOptions split;
+ split.current_range = file_range(path);
+ split.current_split_format = FileFormat::JSON;
+ split.partition_values.emplace("dt",
Field::create_field<TYPE_STRING>("p1"));
+ ASSERT_TRUE(reader.prepare_split(split).ok());
+ bool eos = false;
+ for (const size_t expected_rows : {2, 1, 0}) {
+ Block block({{partition.type->create_column(), partition.type,
partition.name}});
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ ASSERT_EQ(block.rows(), expected_rows);
+ EXPECT_EQ(eos, expected_rows == 0);
+ const auto column =
block.get_by_position(0).column->convert_to_full_column_if_const();
+ for (size_t row = 0; row < expected_rows; ++row) {
+ EXPECT_EQ(column->get_data_at(row).to_string(), "p1");
+ }
+ }
+ EXPECT_TRUE(reader.close().ok());
+}
+
} // namespace doris::format::json
diff --git a/be/test/format_v2/wal/wal_reader_test.cpp
b/be/test/format_v2/wal/wal_reader_test.cpp
index 77c14afb12f..65e37bd5f89 100644
--- a/be/test/format_v2/wal/wal_reader_test.cpp
+++ b/be/test/format_v2/wal/wal_reader_test.cpp
@@ -22,6 +22,8 @@
#include <chrono>
#include <filesystem>
#include <memory>
+#include <tuple>
+#include <vector>
#include "agent/be_exec_version_manager.h"
#include "core/block/block.h"
@@ -60,8 +62,126 @@ PBlock serialize_block(const Block& block) {
return pblock;
}
+void expect_wal_column(const IColumn& column, const std::string& name, size_t
expected_rows,
+ size_t offset) {
+ ASSERT_EQ(column.size(), expected_rows);
+ for (size_t row = 0; row < expected_rows; ++row) {
+ EXPECT_EQ(column.get_data_at(row).to_string(), name +
std::to_string(offset + row));
+ }
+}
+
+void expect_wal_batch(WalReader* reader, Block* output, size_t
expected_columns,
+ size_t expected_rows, size_t offset) {
+ output->clear_column_data();
+ size_t rows = 0;
+ bool eof = true;
+ ASSERT_TRUE(reader->get_block(output, &rows, &eof).ok());
+ // Projection can remove every output column while the WAL block still
contributes rows.
+ EXPECT_EQ(rows, expected_rows);
+ EXPECT_FALSE(eof);
+ ASSERT_EQ(output->columns(), expected_columns);
+ for (const auto& column : output->get_columns_with_type_and_name()) {
+ ASSERT_NO_FATAL_FAILURE(
+ expect_wal_column(*column.column, column.name, expected_rows,
offset));
+ }
+}
+
+void initialize_wal_reader(WalReader* reader, const std::string& path, bool
discover_schema) {
+ reader->_wal_reader = std::make_shared<WalFileReader>(path);
+ ASSERT_TRUE(reader->_wal_reader->init().ok());
+ std::string encoded_ids;
+ ASSERT_TRUE(reader->_wal_reader->read_header(reader->_version,
encoded_ids).ok());
+ ASSERT_TRUE(parse_wal_column_ids(encoded_ids, &reader->_column_ids).ok());
+ if (discover_schema) {
+ std::vector<ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ ASSERT_EQ(schema.size(), 2);
+ }
+}
+
+void expect_wal_eof(WalReader* reader, Block* output) {
+ output->clear_column_data();
+ size_t rows = 99;
+ bool eof = false;
+ ASSERT_TRUE(reader->get_block(output, &rows, &eof).ok());
+ EXPECT_EQ(rows, 0);
+ EXPECT_TRUE(eof);
+ EXPECT_EQ(output->rows(), 0);
+}
+
+class WalReaderV2ProjectionTest
+ : public testing::TestWithParam<std::tuple<std::vector<int>, bool>> {
+protected:
+ void SetUp() override {
+ _wal_path = temporary_wal_path();
+ const auto type = std::make_shared<DataTypeString>();
+ WalWriter writer(_wal_path);
+ ASSERT_TRUE(writer.init(io::global_local_filesystem()).ok());
+ ASSERT_TRUE(writer.append_header("17,4").ok());
+ size_t offset = 0;
+ for (const size_t batch_rows : {3, 2}) {
+ Block source;
+ for (const auto& name : {"left_value", "right_value"}) {
+ auto column = type->create_column();
+ for (size_t row = 0; row < batch_rows; ++row) {
+ const auto value = std::string(name) +
std::to_string(offset + row);
+ column->insert_data(value.data(), value.size());
+ }
+ source.insert({std::move(column), type, name});
+ }
+ auto pblock = serialize_block(source);
+ ASSERT_TRUE(writer.append_blocks({&pblock}).ok());
+ offset += batch_rows;
+ }
+ ASSERT_TRUE(writer.finalize().ok());
+ }
+
+ void TearDown() override {
+ std::filesystem::remove_all(
+
std::filesystem::path(_wal_path).parent_path().parent_path().parent_path());
+ }
+
+ std::string _wal_path;
+};
+
} // namespace
+TEST_P(WalReaderV2ProjectionTest, PreservesRowsAcrossBlocksAndEof) {
+ const auto& [projected_columns, discover_schema] = GetParam();
+ std::shared_ptr<io::FileSystemProperties> properties;
+ std::unique_ptr<io::FileDescription> description;
+ WalReader reader(properties, description, nullptr, nullptr, {});
+ ASSERT_NO_FATAL_FAILURE(initialize_wal_reader(&reader, _wal_path,
discover_schema));
+
+ auto request = std::make_shared<FileScanRequest>();
+ Block output;
+ const auto type = std::make_shared<DataTypeString>();
+ const std::vector<std::string> names {"left_value", "right_value"};
+ for (size_t position = 0; position < projected_columns.size(); ++position)
{
+ const auto source_position = projected_columns[position];
+ request->local_positions.emplace(LocalColumnId(source_position),
LocalIndex(position));
+ output.insert({type->create_column(), type, names[source_position]});
+ }
+ ASSERT_TRUE(reader.open(std::move(request)).ok());
+
+ size_t offset = 0;
+ for (const size_t expected_rows : {3, 2}) {
+ ASSERT_NO_FATAL_FAILURE(expect_wal_batch(&reader, &output,
projected_columns.size(),
+ expected_rows, offset));
+ offset += expected_rows;
+ }
+ for (int attempt = 0; attempt < 2; ++attempt) {
+ ASSERT_NO_FATAL_FAILURE(expect_wal_eof(&reader, &output));
+ }
+ ASSERT_TRUE(reader.close().ok());
+}
+
+INSTANTIATE_TEST_SUITE_P(WriterBacked, WalReaderV2ProjectionTest,
+ testing::Combine(testing::Values(std::vector<int> {},
std::vector<int> {0},
+ std::vector<int> {1},
+ std::vector<int> {1,
0}),
+ testing::Bool()));
+
TEST(WalReaderV2Test, ParseColumnIdsPreservesHeaderOrder) {
std::vector<int32_t> column_ids;
ASSERT_TRUE(parse_wal_column_ids("17,4,99", &column_ids).ok());
diff --git
a/regression-test/data/insert_p0/group_commit/test_group_commit_wal_row_count.out
b/regression-test/data/insert_p0/group_commit/test_group_commit_wal_row_count.out
new file mode 100644
index 00000000000..0ae4de99333
--- /dev/null
+++
b/regression-test/data/insert_p0/group_commit/test_group_commit_wal_row_count.out
@@ -0,0 +1,17 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !before_replay --
+0
+
+-- !counts --
+5 4 15 120
+
+-- !values --
+1 10
+2 20
+3 \N
+4 40
+5 50
+
+-- !filtered_count --
+3
+
diff --git
a/regression-test/suites/external_table_p0/hive/test_hive_json_partition_only.groovy
b/regression-test/suites/external_table_p0/hive/test_hive_json_partition_only.groovy
new file mode 100644
index 00000000000..69ae572d563
--- /dev/null
+++
b/regression-test/suites/external_table_p0/hive/test_hive_json_partition_only.groovy
@@ -0,0 +1,85 @@
+// 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.
+
+suite("test_hive_json_partition_only", "p0,external,hive") {
+ String enabled = context.config.otherConfigs.get("enableHiveTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable Hive test")
+ return
+ }
+
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String hmsPort = context.config.otherConfigs.get("hive3HmsPort")
+ String catalog = "test_hive_json_partition_only"
+ def previousSettings = sql """SELECT
@@enable_count_push_down_for_external_table,
+ @@enable_sql_cache"""
+ sql """DROP CATALOG IF EXISTS ${catalog}"""
+ try {
+ sql """CREATE CATALOG ${catalog} PROPERTIES (
+ "type" = "hms",
+ "hive.metastore.uris" = "thrift://${externalEnvIp}:${hmsPort}"
+ )"""
+ sql """USE `${catalog}`.`default`"""
+ sql "SET enable_sql_cache=false"
+
+ // Reading a real JSON field provides the row-count oracle for the
same partition files.
+ def baseline = sql "SELECT id, dt FROM json_all_complex_types"
+ assertTrue(!baseline.isEmpty())
+ def partitions = baseline.groupBy { row -> row[1] }
+
+ for (boolean pushdown : [false, true]) {
+ sql "SET enable_count_push_down_for_external_table=${pushdown}"
+ partitions.each { partition, rows ->
+ assertNotNull(partition)
+ String value = partition.toString().replace("'", "''")
+ String predicate = "dt = '${value}'"
+ String countQuery = "SELECT COUNT(*) FROM
json_all_complex_types WHERE ${predicate}"
+ explain {
+ sql countQuery
+ contains "pushdown agg=NONE"
+ }
+ def count = sql countQuery
+ assertEquals(rows.size(), (count[0][0] as Number).intValue())
+
+ // Partition values come from the split, so neither query
needs a JSON file column.
+ def projected = sql "SELECT dt FROM json_all_complex_types
WHERE ${predicate}"
+ assertEquals(rows.size(), projected.size())
+ assertTrue(projected.every { row -> row[0] == partition })
+ def partitionCount = sql "SELECT COUNT(dt) FROM
json_all_complex_types WHERE ${predicate}"
+ assertEquals(rows.size(), (partitionCount[0][0] as
Number).intValue())
+
+ def filtered = sql """SELECT COUNT(*) FROM
json_all_complex_types
+ WHERE ${predicate} AND id IS NOT NULL"""
+ assertEquals(rows.count { row -> row[0] != null },
+ (filtered[0][0] as Number).intValue())
+ }
+
+ def grouped = sql "SELECT dt, COUNT(*) FROM json_all_complex_types
GROUP BY dt"
+ assertEquals(partitions.size(), grouped.size())
+ grouped.each { row ->
+ assertEquals(partitions[row[0]].size(), (row[1] as
Number).intValue())
+ }
+ def empty = sql """SELECT COUNT(*) FROM json_all_complex_types
+ WHERE dt = '__missing_json_partition__'"""
+ assertEquals(0, (empty[0][0] as Number).intValue())
+ }
+ } finally {
+ sql "SET
enable_count_push_down_for_external_table=${previousSettings[0][0]}"
+ sql "SET enable_sql_cache=${previousSettings[0][1]}"
+ sql """DROP CATALOG IF EXISTS ${catalog}"""
+ }
+}
diff --git
a/regression-test/suites/insert_p0/group_commit/test_group_commit_wal_row_count.groovy
b/regression-test/suites/insert_p0/group_commit/test_group_commit_wal_row_count.groovy
new file mode 100644
index 00000000000..ed1a449724a
--- /dev/null
+++
b/regression-test/suites/insert_p0/group_commit/test_group_commit_wal_row_count.groovy
@@ -0,0 +1,85 @@
+// 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.
+
+import org.awaitility.Awaitility
+import static java.util.concurrent.TimeUnit.SECONDS
+
+suite("test_group_commit_wal_row_count", "nonConcurrent") {
+ sql "DROP TABLE IF EXISTS test_group_commit_wal_row_count"
+ sql """
+ CREATE TABLE test_group_commit_wal_row_count (
+ k INT NOT NULL,
+ v INT NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "group_commit_interval_ms" = "2000"
+ )
+ """
+
+ String loadError = "LoadBlockQueue._finish_group_commit_load.load_error"
+ String replayStop = "WalTable.replay_wals.stop"
+ try {
+ GetDebugPoint().enableDebugPointForAllBEs(loadError)
+ GetDebugPoint().enableDebugPointForAllBEs(replayStop)
+
+ // Abort each original transaction before the next load so both WALs
must be replayed.
+ for (String data : ["1,10\n2,20\n3,\\N\n", "4,40\n5,50\n"]) {
+ long txnId = 0
+ streamLoad {
+ table "test_group_commit_wal_row_count"
+ set "column_separator", ","
+ set "group_commit", "async_mode"
+ unset "label"
+ inputStream new ByteArrayInputStream(data.getBytes("UTF-8"))
+ time 10000
+ check { result, exception, startTime, endTime ->
+ assertNull(exception)
+ def response = parseJson(result)
+ assertEquals("Success", response.Status)
+ assertEquals(data.readLines().size(),
response.NumberTotalRows as int)
+ txnId = response.TxnId as long
+ assertTrue(txnId > 0)
+ }
+ }
+ Awaitility.await().atMost(30, SECONDS).pollInterval(1,
SECONDS).until {
+ def txn = sql_return_maparray "SHOW TRANSACTION WHERE ID =
${txnId}"
+ return txn.size() == 1 && txn[0].TransactionStatus == "ABORTED"
+ }
+ }
+ qt_before_replay "SELECT COUNT(*) FROM test_group_commit_wal_row_count"
+
+ // Keep the original-load failure enabled: rows can become visible
only through WAL replay.
+ GetDebugPoint().disableDebugPointForAllBEs(replayStop)
+ Awaitility.await().atMost(90, SECONDS).pollInterval(1, SECONDS).until {
+ def rows = sql "SELECT COUNT(*) FROM
test_group_commit_wal_row_count"
+ return rows[0][0] == 5
+ }
+ qt_counts """SELECT COUNT(*), COUNT(v), SUM(k), SUM(v)
+ FROM test_group_commit_wal_row_count"""
+ qt_values "SELECT k, v FROM test_group_commit_wal_row_count ORDER BY k"
+ qt_filtered_count "SELECT COUNT(*) FROM
test_group_commit_wal_row_count WHERE v >= 20"
+ } finally {
+ try {
+ GetDebugPoint().disableDebugPointForAllBEs(replayStop)
+ } finally {
+ GetDebugPoint().disableDebugPointForAllBEs(loadError)
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
