github-actions[bot] commented on code in PR #68335:
URL: https://github.com/apache/doris/pull/68335#discussion_r4061188386
##########
be/src/format_v2/lance/lance_reader_helper.cpp:
##########
@@ -575,78 +473,92 @@ Status compact_lance_array_if_needed(const
std::shared_ptr<arrow::Array>& array,
} // namespace
-// Normalize Lance extensions and materialize sliced arrays for Doris Arrow
SerDes.
-Status normalize_lance_arrow_array(const std::shared_ptr<arrow::Field>& field,
- const std::shared_ptr<arrow::Array>& array,
- arrow::MemoryPool* memory_pool,
- std::shared_ptr<arrow::Array>* normalized) {
+Status LanceArrowArrayNormalizer::create(const std::shared_ptr<arrow::Field>&
field,
+ LanceArrowArrayNormalizer*
normalizer) {
DORIS_CHECK(field != nullptr);
- DORIS_CHECK(array != nullptr);
- DORIS_CHECK(memory_pool != nullptr);
- DORIS_CHECK(normalized != nullptr);
+ DORIS_CHECK(normalizer != nullptr);
LanceExtensionKind extension_kind;
std::shared_ptr<arrow::DataType> storage_type;
RETURN_IF_ERROR(get_lance_extension(field, &extension_kind,
&storage_type));
+ LanceArrowArrayNormalizer result;
+ result._field_name = field->name();
+ result._storage_type = std::move(storage_type);
+ result._unwrap_registered_extension = field->type()->id() ==
arrow::Type::EXTENSION;
+ result._convert_bfloat16 = extension_kind == LanceExtensionKind::BFLOAT16;
+ result._requires_special_handling =
+ result._unwrap_registered_extension || result._convert_bfloat16;
+ result._child_normalizers.reserve(result._storage_type->num_fields());
+ for (const auto& child_field : result._storage_type->fields()) {
+ LanceArrowArrayNormalizer child_normalizer;
+ RETURN_IF_ERROR(create(child_field, &child_normalizer));
+ result._requires_special_handling |=
child_normalizer._requires_special_handling;
+ result._child_normalizers.emplace_back(std::move(child_normalizer));
+ }
+ *normalizer = std::move(result);
+ return Status::OK();
+}
+
+Status LanceArrowArrayNormalizer::normalize_for_doris(
+ const std::shared_ptr<arrow::Array>& array, arrow::MemoryPool*
memory_pool,
+ std::shared_ptr<arrow::Array>* normalized) const {
+ DORIS_CHECK(array != nullptr);
+ DORIS_CHECK(memory_pool != nullptr);
+ DORIS_CHECK(normalized != nullptr);
+
+ // The schema binding already resolved extension metadata and nested
special types. For common
+ // arrays, runtime work is limited to checking whether the visible slice
needs compaction.
+ if (!_requires_special_handling) {
+ return compact_lance_array_if_needed(array, memory_pool, normalized);
+ }
+
auto storage_array = array;
- if (type_contains_registered_extension(field->type())) {
- std::shared_ptr<arrow::Array> unwrapped_array;
- RETURN_IF_ERROR(unwrap_lance_extension_arrays(field->type(), array,
&unwrapped_array));
- storage_array = std::move(unwrapped_array);
+ if (_unwrap_registered_extension && array->type_id() ==
arrow::Type::EXTENSION) {
+ const auto extension_array =
std::dynamic_pointer_cast<arrow::ExtensionArray>(array);
+ if (extension_array == nullptr) {
+ return Status::InvalidArgument("invalid Arrow extension array: {}",
+ array->type()->ToString());
+ }
+ storage_array = extension_array->storage();
}
- if (storage_array->type_id() != storage_type->id()) {
+ // Extension and BFloat16 arrays are first converted to their physical
storage representation;
+ // nested children are then normalized recursively before Doris reads the
result.
+ if (storage_array->type_id() != _storage_type->id()) {
return Status::InvalidArgument(
- "Lance field '{}' storage type {} does not match array type
{}", field->name(),
- storage_type->ToString(), storage_array->type()->ToString());
+ "Lance field '{}' storage type {} does not match array type
{}", _field_name,
+ _storage_type->ToString(), storage_array->type()->ToString());
}
- if (extension_kind == LanceExtensionKind::BFLOAT16) {
+ if (_convert_bfloat16) {
return convert_bfloat16_array(storage_array, memory_pool, normalized);
}
Review Comment:
Because this compaction now runs before child normalizers, a sliced
LIST/STRUCT whose child is a registered ExtensionType reaches
arrow::MakeBuilder(list<extension>). Arrow 24 returns NotImplemented for
ExtensionType builders, so the existing ReadsRegisteredJsonNestedInSlicedList
path fails instead of unwrapping the child. The old routine unwrapped first.
Please unwrap/rebuild extension children before compacting the parent (or
compact with storage-type builders) and retain this regression coverage.
##########
be/src/format_v2/table/lance_reader.cpp:
##########
@@ -839,7 +766,8 @@ Status LanceTableReader::_open_scanner(const
TFileRangeDesc& range) {
return lance_error("set Lance scanner statistics callback");
}
- if (_global_rowid_output_idx.has_value() &&
lance_scanner_with_row_id(scanner, true) != 0) {
+ if (_record_batch_converter.requires_global_rowid() &&
+ lance_scanner_with_row_id(scanner, true) != 0) {
return lance_error("enable Lance row id output");
Review Comment:
This new null-counter guard makes scanner-less row-ID fetches lose cache
statistics. _init_scanner_profile is only called from _open_scanner, but
read_by_row_ids opens the dataset and calls take_rows without a scanner;
_close_dataset therefore reaches this guard with null counters and skips
lance_dataset_get_data_cache_statistics. Before the refactor the counters were
initialized in init, so phase-two I/O exposed these bytes. Please
initialize/collect the cache counters for take_rows as well, or explicitly wire
those statistics into the row-ID fetch profile.
##########
be/src/format_v2/table/lance_reader.cpp:
##########
@@ -1238,10 +1166,12 @@ void LanceTableReader::_close_dataset() {
_dataset = nullptr;
}
_dataset_schema.reset();
+ _record_batch_converter.reset_schema();
}
void LanceTableReader::_collect_data_cache_statistics() {
- if (_dataset == nullptr) {
+ if (_dataset == nullptr || (_data_cache_bytes_read_from_cache == nullptr &&
Review Comment:
This new null-counter guard makes scanner-less row-ID fetches lose cache
statistics. _init_scanner_profile is only called from _open_scanner, but
read_by_row_ids opens the dataset and calls take_rows without a scanner;
_close_dataset therefore reaches this guard with null counters and skips
lance_dataset_get_data_cache_statistics. Before the refactor the counters were
initialized in init, so phase-two I/O exposed these bytes. Please
initialize/collect the cache counters for take_rows as well, or explicitly wire
those statistics into the row-ID fetch profile.
##########
be/src/format_v2/lance/lance_reader_helper.cpp:
##########
@@ -575,78 +473,92 @@ Status compact_lance_array_if_needed(const
std::shared_ptr<arrow::Array>& array,
} // namespace
-// Normalize Lance extensions and materialize sliced arrays for Doris Arrow
SerDes.
-Status normalize_lance_arrow_array(const std::shared_ptr<arrow::Field>& field,
- const std::shared_ptr<arrow::Array>& array,
- arrow::MemoryPool* memory_pool,
- std::shared_ptr<arrow::Array>* normalized) {
+Status LanceArrowArrayNormalizer::create(const std::shared_ptr<arrow::Field>&
field,
+ LanceArrowArrayNormalizer*
normalizer) {
DORIS_CHECK(field != nullptr);
- DORIS_CHECK(array != nullptr);
- DORIS_CHECK(memory_pool != nullptr);
- DORIS_CHECK(normalized != nullptr);
+ DORIS_CHECK(normalizer != nullptr);
LanceExtensionKind extension_kind;
std::shared_ptr<arrow::DataType> storage_type;
RETURN_IF_ERROR(get_lance_extension(field, &extension_kind,
&storage_type));
+ LanceArrowArrayNormalizer result;
+ result._field_name = field->name();
+ result._storage_type = std::move(storage_type);
+ result._unwrap_registered_extension = field->type()->id() ==
arrow::Type::EXTENSION;
+ result._convert_bfloat16 = extension_kind == LanceExtensionKind::BFLOAT16;
+ result._requires_special_handling =
+ result._unwrap_registered_extension || result._convert_bfloat16;
+ result._child_normalizers.reserve(result._storage_type->num_fields());
+ for (const auto& child_field : result._storage_type->fields()) {
+ LanceArrowArrayNormalizer child_normalizer;
+ RETURN_IF_ERROR(create(child_field, &child_normalizer));
+ result._requires_special_handling |=
child_normalizer._requires_special_handling;
+ result._child_normalizers.emplace_back(std::move(child_normalizer));
+ }
+ *normalizer = std::move(result);
+ return Status::OK();
+}
+
+Status LanceArrowArrayNormalizer::normalize_for_doris(
+ const std::shared_ptr<arrow::Array>& array, arrow::MemoryPool*
memory_pool,
+ std::shared_ptr<arrow::Array>* normalized) const {
+ DORIS_CHECK(array != nullptr);
+ DORIS_CHECK(memory_pool != nullptr);
+ DORIS_CHECK(normalized != nullptr);
+
+ // The schema binding already resolved extension metadata and nested
special types. For common
+ // arrays, runtime work is limited to checking whether the visible slice
needs compaction.
+ if (!_requires_special_handling) {
+ return compact_lance_array_if_needed(array, memory_pool, normalized);
+ }
+
auto storage_array = array;
- if (type_contains_registered_extension(field->type())) {
- std::shared_ptr<arrow::Array> unwrapped_array;
- RETURN_IF_ERROR(unwrap_lance_extension_arrays(field->type(), array,
&unwrapped_array));
- storage_array = std::move(unwrapped_array);
+ if (_unwrap_registered_extension && array->type_id() ==
arrow::Type::EXTENSION) {
+ const auto extension_array =
std::dynamic_pointer_cast<arrow::ExtensionArray>(array);
+ if (extension_array == nullptr) {
+ return Status::InvalidArgument("invalid Arrow extension array: {}",
+ array->type()->ToString());
+ }
+ storage_array = extension_array->storage();
}
- if (storage_array->type_id() != storage_type->id()) {
+ // Extension and BFloat16 arrays are first converted to their physical
storage representation;
+ // nested children are then normalized recursively before Doris reads the
result.
+ if (storage_array->type_id() != _storage_type->id()) {
return Status::InvalidArgument(
- "Lance field '{}' storage type {} does not match array type
{}", field->name(),
- storage_type->ToString(), storage_array->type()->ToString());
+ "Lance field '{}' storage type {} does not match array type
{}", _field_name,
+ _storage_type->ToString(), storage_array->type()->ToString());
}
- if (extension_kind == LanceExtensionKind::BFLOAT16) {
+ if (_convert_bfloat16) {
return convert_bfloat16_array(storage_array, memory_pool, normalized);
}
std::shared_ptr<arrow::Array> compacted_array;
RETURN_IF_ERROR(compact_lance_array_if_needed(storage_array, memory_pool,
&compacted_array));
Review Comment:
Because this compaction now runs before child normalizers, a sliced
LIST/STRUCT whose child is a registered ExtensionType reaches
arrow::MakeBuilder(list<extension>). Arrow 24 returns NotImplemented for
ExtensionType builders, so the existing ReadsRegisteredJsonNestedInSlicedList
path fails instead of unwrapping the child. The old routine unwrapped first.
Please unwrap/rebuild extension children before compacting the parent (or
compact with storage-type builders) and retain this regression coverage.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
