github-actions[bot] commented on code in PR #67325:
URL: https://github.com/apache/doris/pull/67325#discussion_r3985617451
##########
be/src/format_v2/lance/lance_reader_helper.cpp:
##########
@@ -178,8 +270,313 @@ Status arrow_field_to_doris_type(const
std::shared_ptr<arrow::Field>& field,
}
}
+// Determine whether a field subtree contains values that require Lance
normalization.
+Status field_requires_lance_normalization(const std::shared_ptr<arrow::Field>&
field,
+ bool* requires_normalization) {
+ DORIS_CHECK(field != nullptr);
+ DORIS_CHECK(requires_normalization != nullptr);
+
+ LanceExtensionKind extension_kind;
+ std::shared_ptr<arrow::DataType> storage_type;
+ RETURN_IF_ERROR(get_lance_extension(field, &extension_kind,
&storage_type));
+ bool required = extension_kind == LanceExtensionKind::BFLOAT16 ||
+ field->type()->id() == arrow::Type::EXTENSION;
+ for (const auto& child : storage_type->fields()) {
+ bool child_required = false;
+ RETURN_IF_ERROR(field_requires_lance_normalization(child,
&child_required));
+ required |= child_required;
+ }
+ *requires_normalization = required;
+ return Status::OK();
+}
+
+// Widen little-endian Lance BFloat16 values to Arrow Float32 without
precision loss.
+Status convert_bfloat16_array(const std::shared_ptr<arrow::Array>& array,
+ std::shared_ptr<arrow::Array>* normalized) {
+ DORIS_CHECK(array != nullptr);
+ DORIS_CHECK(normalized != nullptr);
+ const auto fixed_binary =
std::dynamic_pointer_cast<arrow::FixedSizeBinaryArray>(array);
+ if (fixed_binary == nullptr || fixed_binary->byte_width() != 2) {
+ return Status::InvalidArgument("invalid Lance BFloat16 array storage:
{}",
+ array->type()->ToString());
+ }
+ if (config::enable_arrow_input_validation) {
+ check_arrow_fixed_width_buffer(*fixed_binary, sizeof(uint16_t));
+ }
+
+ arrow::FloatBuilder builder;
+ auto arrow_status = builder.Reserve(fixed_binary->length());
+ if (!arrow_status.ok()) {
+ return Status::InternalError("reserve Lance BFloat16 output failed:
{}",
+ arrow_status.message());
+ }
+ for (int64_t row = 0; row < fixed_binary->length(); ++row) {
+ if (fixed_binary->IsNull(row)) {
+ arrow_status = builder.AppendNull();
+ } else {
+ const auto bits =
LittleEndian::Load16(fixed_binary->GetValue(row));
+ arrow_status =
builder.Append(std::bit_cast<float>(static_cast<uint32_t>(bits) << 16));
+ }
+ if (!arrow_status.ok()) {
+ return Status::InternalError("append Lance BFloat16 value failed:
{}",
+ arrow_status.message());
+ }
+ }
+ std::shared_ptr<arrow::FloatArray> result;
+ arrow_status = builder.Finish(&result);
+ if (!arrow_status.ok()) {
+ return Status::InternalError("finish Lance BFloat16 conversion failed:
{}",
+ arrow_status.message());
+ }
+ *normalized = std::move(result);
+ return Status::OK();
+}
+
+// Materialize the visible range into an offset-zero Arrow array for Doris
SerDes.
+Status compact_lance_array(const std::shared_ptr<arrow::Array>& array,
+ std::shared_ptr<arrow::Array>* compacted) {
+ const auto validation = array->ValidateFull();
+ if (!validation.ok()) {
+ return Status::InvalidArgument("validate sliced Lance array failed:
{}",
+ validation.message());
+ }
+ auto builder_result = arrow::MakeBuilder(array->type(),
arrow::default_memory_pool());
Review Comment:
[P2] Normalize registered extension children before compacting this parent.
Arrow 24's C Data importer rehydrates the built-in `arrow.json` type
recursively, so a sliced `List`/`Struct`/`Map` can still have an
`ExtensionType` child here. `MakeBuilder(array->type())` recursively asks for
child builders, and Arrow returns `NotImplemented` for extension types; the
scan therefore fails before the later child-unwrapping loop runs. Please
recurse/unwrap before parent compaction (or build from storage-typed children)
and add a sliced nested `arrow.json` import/materialization regression.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceTypeConverter.java:
##########
@@ -37,28 +37,37 @@
public final class LanceTypeConverter {
private static final int MAX_DECIMAL_PRECISION = 76;
private static final String ARROW_EXTENSION_NAME = "ARROW:extension:name";
+ private static final String ARROW_JSON_EXTENSION = "arrow.json";
+ private static final String LANCE_JSON_EXTENSION = "lance.json";
+ private static final String LANCE_BFLOAT16_EXTENSION = "lance.bfloat16";
private LanceTypeConverter() {
}
+ /** Converts Arrow fields exposed by Lance to Doris types. */
public static Type toDorisType(Field field) {
- // Arrow Java exposes unknown extension types through their storage
type and field
- // metadata. Treating the storage type as the logical type would make
DESC report
- // Blob, JSON, or BFloat16 as supported even though the scanner cannot
decode their
- // extension semantics. Dictionary arrays are likewise not decoded by
the BE reader.
+ return toDorisType(field, true);
+ }
+
+ /** Converts an Arrow field, allowing Doris NULL only at the top level. */
+ private static Type toDorisType(Field field, boolean allowNull) {
// TODO(lance): Dataset.getSchema() currently erases the Dictionary
marker, while
// Dataset.getLanceSchema() fails to convert a schema containing
Dictionary in the
// Lance 9.1.0-beta.3 Java SDK. Reject physical Dictionary columns
after that SDK
// conversion is fixed; an unmarked Int16 field cannot be
distinguished safely here.
String extensionName = field.getMetadata() == null
? null : field.getMetadata().get(ARROW_EXTENSION_NAME);
- if (field.getDictionary() != null
- || (extensionName != null && !extensionName.isEmpty())) {
+ if (field.getDictionary() != null) {
return Type.UNSUPPORTED;
}
+ if (extensionName != null && !extensionName.isEmpty()) {
Review Comment:
[P2] Gate catalog/S3 execution on BE support for these new encodings. This
converter can now advertise JSONB/FLOAT from FE metadata, but `LanceScanNode`
still schedules through the ordinary backend policy, including a smooth-upgrade
source BE running the previous reader. BFloat16 then reaches that reader as
`FixedSizeBinary(2)` and is rejected by the FLOAT SerDe (JSON similarly lacks
the new normalization). The local-TVF pin does not cover this FE-discovery
path; please filter/reject old source BEs when these types are projected and
add a mixed-version catalog regression.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/FederationBackendPolicy.java:
##########
@@ -156,13 +156,25 @@ public void init() throws UserException {
}
public void init(List<String> preLocations) throws UserException {
+ init(preLocations, Collections.emptyList());
+ }
+
+ /** Initialize the policy with exactly one eligible backend. */
+ public void initWithBackendId(long backendId) throws UserException {
+ init(Collections.emptyList(), Collections.singletonList(backendId));
+ }
+
+ /** Build the standard external-scan policy with optional location and
backend-ID constraints. */
+ private void init(List<String> preLocations, List<Long> requiredBackendIds)
+ throws UserException {
// scan node is used for query
BeSelectionPolicy.Builder builder = new BeSelectionPolicy.Builder();
builder.needQueryAvailable()
.needLoadAvailable()
.preferComputeNode(Config.prefer_compute_node_for_external_table)
.assignExpectBeNum(Config.min_backend_num_for_external_table)
- .addPreLocations(preLocations);
+ .addPreLocations(preLocations)
+ .addRequiredBackendIds(requiredBackendIds);
Review Comment:
[P2] Select the schema BE from the same eligibility set used here.
Shared-storage local Lance currently picks `backendIdForRequest` from all alive
nodes, but this standard policy additionally rejects compute-only nodes when
`prefer_compute_node_for_external_table=false` (the default), plus
query/load-disabled, decommissioning, blacklisted, or out-of-group nodes. If
one of those served schema discovery, the required-ID filter leaves no
candidates and planning fails with `No available backends`. Please align schema
selection with execution eligibility (or make the exact-ID role semantics
explicit while retaining real availability checks) and cover a
compute-only/ineligible selected BE.
--
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]
