[feature](be) Read Lance datasets with FileScannerV2

be/CMakeLists.txt

@@ -894,6 +894,8 @@ else()
         "-framework Foundation"
         "-framework SystemConfiguration"
         "-framework Security"
+        # liblance_c.a pulls in the Rust `sysinfo` crate, whose macOS backend calls IOKit.
+        "-framework IOKit"
     )
     if (USE_JEMALLOC OR (NOT CMAKE_BUILD_TYPE STREQUAL "DEBUG" AND NOT CMAKE_BUILD_TYPE STREQUAL "RELEASE"))
         set(DORIS_LINK_LIBS

be/src/common/config.cpp

@@ -1232,6 +1232,26 @@ DEFINE_Validator(variant_max_json_key_length,
 DEFINE_Validator(variant_storage_parse_mode,
                  [](const int config) -> bool { return config >= 0 && config <= 2; });
 
+// Lance uses one BE-wide session so metadata/index caches and the optional Foyer data-file cache
+// can be shared by all Lance dataset readers.
+DEFINE_Int64(lance_index_cache_size_bytes, "10737418240");   // 10 GiB
+DEFINE_Int64(lance_metadata_cache_size_bytes, "1073741824"); // 1GB
+DEFINE_Bool(enable_lance_data_cache, "true");
+DEFINE_String(lance_data_cache_path, "${DORIS_HOME}/lance_data_cache");
+DEFINE_Int64(lance_data_cache_disk_capacity_bytes, "107374182400"); // 100GB
+DEFINE_Int64(lance_data_cache_read_block_size_bytes, "1048576");    // 1MB
+
+// I/O buffering budget per Lance scanner, not a cap on its total memory usage.
+// Runtime changes apply to newly created scanners.
+DEFINE_mInt64(lance_io_buffer_size_bytes, "2147483648"); // 2 GiB
+DEFINE_Validator(lance_io_buffer_size_bytes, [](int64_t value) { return value > 0; });
+
+// Read-ahead limits per Lance scanner. Runtime changes apply to newly created scanners.
+DEFINE_mInt32(lance_batch_readahead, "5");
+DEFINE_Validator(lance_batch_readahead, [](int32_t value) { return value > 0; });
+DEFINE_mInt32(lance_fragment_readahead, "5");
+DEFINE_Validator(lance_fragment_readahead, [](int32_t value) { return value > 0; });
+
 // block file cache
 DEFINE_Bool(enable_file_cache, "true");
 // ATTENTION: For test only. Keep this enabled in production.

be/src/common/config.h

@@ -1286,6 +1286,21 @@ DECLARE_Bool(enable_debug_points);
 DECLARE_Int32(pipeline_executor_size);
 DECLARE_Int32(blocking_pipeline_executor_size);
 
+// Lance shared session and optional Foyer data-file cache.
+DECLARE_Int64(lance_index_cache_size_bytes);
+DECLARE_Int64(lance_metadata_cache_size_bytes);
+DECLARE_Bool(enable_lance_data_cache);
+DECLARE_String(lance_data_cache_path);
+DECLARE_Int64(lance_data_cache_disk_capacity_bytes);
+DECLARE_Int64(lance_data_cache_read_block_size_bytes);
+
+// I/O buffering budget per Lance scanner, applied when a new scanner is created.
+DECLARE_mInt64(lance_io_buffer_size_bytes);
+
+// Read-ahead limits per Lance scanner, applied when a new scanner is created.
+DECLARE_mInt32(lance_batch_readahead);
+DECLARE_mInt32(lance_fragment_readahead);
+
 // block file cache
 DECLARE_Bool(enable_file_cache);
 DECLARE_mBool(enable_file_cache_write_from_s3_file_writer);

be/src/core/data_type/data_type.cpp

@@ -84,7 +84,9 @@ void IDataType::to_protobuf(PTypeDesc* ptype) const {
     auto node = ptype->add_types();
     node->set_type(TTypeNodeType::SCALAR);
     auto scalar_type = node->mutable_scalar_type();
-    scalar_type->set_type(doris::to_thrift(get_primitive_type()));
+    // NULL uses UInt8 internally; preserve its logical type even inside complex schemas.
+    scalar_type->set_type(is_null_literal() ? TPrimitiveType::NULL_TYPE
+                                            : doris::to_thrift(get_primitive_type()));
     to_protobuf(ptype, node, scalar_type);
 }
 

be/src/core/data_type_serde/data_type_datev2_serde.cpp

@@ -277,11 +277,27 @@ Status DataTypeDateV2SerDe::read_column_from_arrow(IColumn& column, const arrow:
         const auto* base_ptr = reinterpret_cast<const uint8_t*>(concrete_array->raw_values());
         const size_t element_size = sizeof(int32_t);
         for (auto value_i = start; value_i < end; ++value_i) {
+            // A null slot has no calendar value and its payload may be outside the DATE range.
+            if (concrete_array->IsNull(value_i)) {
+                col_data.emplace_back(DateV2Value<DateV2ValueType>());
+                continue;
+            }
             const uint8_t* raw_byte_ptr = base_ptr + value_i * element_size;
             auto date_value = unaligned_load<int32_t>(raw_byte_ptr);
 
+            // Arrow allows any int32 day, Doris DATE only day numbers 1..DATE_MAX_DAYNR
+            // (0000-01-01..9999-12-31). For example, 2932897 is 10000-01-01: its day number
+            // 3652425 makes get_date_from_daynr() fail, and ignoring that result used to append
+            // a non-null 0000-00-00. Add in 64 bits so INT32_MAX cannot overflow, and reject
+            // the value like the DATE64 branch above.
+            const int64_t daynr = static_cast<int64_t>(date_value) + date_threshold;
             DateV2Value<DateV2ValueType> v;
-            v.get_date_from_daynr(date_value + date_threshold);
+            if (daynr <= 0 || daynr > DATE_MAX_DAYNR ||
+                !v.get_date_from_daynr(static_cast<uint64_t>(daynr))) {
+                return Status::InvalidArgument(
+                        "Arrow Date32 value is outside the Doris DATE range: row={}, days={}",
+                        value_i, date_value);
+            }
             col_data.emplace_back(v);
         }
     } else {

be/src/core/data_type_serde/data_type_nullable_serde.cpp

@@ -410,6 +410,18 @@ Status DataTypeNullableSerDe::read_column_from_arrow(IColumn& column,
                                                      const arrow::Array* arrow_array, int64_t start,
                                                      int64_t end,
                                                      const cctz::time_zone& ctz) const {
+    if (arrow_array->type_id() == arrow::Type::NA) {
+        // Arrow Null has neither a validity bitmap nor values, even when sliced. Avoid the
+        // physical SerDe for Doris's UInt8 placeholder and keep nested values and nulls aligned.
+        if (arrow_array->offset() < 0 || start < 0 || end < start || end > arrow_array->length()) {
+            return Status::InvalidArgument(
+                    "Invalid Arrow Null read range: start={}, end={}, "
+                    "length={}, offset={}",
+                    start, end, arrow_array->length(), arrow_array->offset());
+        }
+        column.insert_many_defaults(end - start);
+        return Status::OK();
+    }
     if (config::enable_arrow_input_validation) {
         check_arrow_array_range(*arrow_array, start, end);
         check_arrow_validity_bitmap(*arrow_array);

be/src/exec/operator/file_scan_operator.cpp

@@ -197,9 +197,12 @@ bool FileScanLocalState::should_use_file_scanner_v2(const TQueryOptions& query_o
     // fluss scan carries that same format -- log ranges and wrapped lake splits alike -- and is
     // dispatched inside FlussHybridReader, which only FileScannerV2 builds. Loads are left alone for
     // the same reason as adbc: there is no fluss load path.
-    if (!is_load && scan_params.__isset.table_format_params &&
-        (scan_params.table_format_params.table_format_type == "adbc" ||
-         scan_params.table_format_params.table_format_type == "fluss")) {
+    // Lance has no v1 reader either. Its scans carry FORMAT_LANCE at scan level, whether they come
+    // from a Lance catalog or from a file TVF, and there is no Lance load path.
+    if (!is_load && (scan_params.format_type == TFileFormatType::FORMAT_LANCE ||
+                     (scan_params.__isset.table_format_params &&
+                      (scan_params.table_format_params.table_format_type == "adbc" ||
+                       scan_params.table_format_params.table_format_type == "fluss")))) {
         return true;
     }
     const bool is_transactional_hive =
@@ -224,8 +227,7 @@ bool FileScanLocalState::should_use_file_scanner_v2(const TQueryOptions& query_o
     // after scanner construction. Keep versioned Iceberg scans on V2 without eagerly listing
     // every file.
     return scanner_v2_requested && !is_load &&
-           scan_params.format_type != TFileFormatType::FORMAT_ES_HTTP &&
-           scan_params.format_type != TFileFormatType::FORMAT_LANCE && !is_transactional_hive;
+           scan_params.format_type != TFileFormatType::FORMAT_ES_HTTP && !is_transactional_hive;
 }
 
 Status FileScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {

be/src/exec/operator/materialization_opertor.cpp

@@ -21,6 +21,7 @@
 #include <fmt/format.h>
 #include <gen_cpp/internal_service.pb.h>
 
+#include <cstring>
 #include <set>
 #include <sstream>
 #include <utility>
@@ -72,6 +73,57 @@ constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND_REMOTE_IO_TIME =
 constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND_WRITE_CACHE_IO_TIME =
         "TopNLazyMaterializationSecondPhasePerBackendWriteCacheIOTime";
 
+struct MaterializationRowLocation {
+    ROW_VERSION version = ROW_VERSION::FILE_LOCAL_ROW_ID;
+    int64_t backend_id = 0;
+    uint32_t file_id = 0;
+    uint64_t row_id = 0;
+};
+
+Status decode_global_row_location_v2(const StringRef& encoded, ROW_VERSION version,
+                                     MaterializationRowLocation* decoded) {
+    if (encoded.size != sizeof(GlobalRowLoacationV2)) {
+        return Status::InternalError(
+                "invalid global row location size for version {}: actual={}, expected={}",
+                static_cast<int>(version), encoded.size, sizeof(GlobalRowLoacationV2));
+    }
+    GlobalRowLoacationV2 location(GlobalRowLoacationV2::VERSION, 0, 0, 0);
+    std::memcpy(&location, encoded.data, sizeof(location));
+    decoded->version = version;
+    decoded->backend_id = location.backend_id;
+    switch (version) {
+    case ROW_VERSION::FILE_LOCAL_ROW_ID:
+        decoded->file_id = location.file_local.file_id;
+        decoded->row_id = location.file_local.row_id;
+        return Status::OK();
+    case ROW_VERSION::LANCE_DATASET_ROW_ID:
+        decoded->file_id = location.lance_file_id;
+        decoded->row_id = location.lance_row_id;
+        return Status::OK();
+    }
+    return Status::NotSupported("unsupported V2 global row location version: {}",
+                                static_cast<int>(version));
+}
+
+Status decode_materialization_row_location(const StringRef& encoded,
+                                           MaterializationRowLocation* decoded) {
+    if (encoded.size < sizeof(uint8_t)) {
+        return Status::InternalError("global row location is empty");
+    }
+
+    const auto version_value = static_cast<uint8_t>(encoded.data[0]);
+    const auto version = static_cast<ROW_VERSION>(version_value);
+    // Keep size validation inside each version-family decoder. A future version can therefore
+    // use a different encoded size without being rejected by the current 24-byte V2 contract.
+    switch (version) {
+    case ROW_VERSION::FILE_LOCAL_ROW_ID:
+    case ROW_VERSION::LANCE_DATASET_ROW_ID:
+        return decode_global_row_location_v2(encoded, version, decoded);
+    }
+    return Status::NotSupported("unsupported global row location version: {}, encoded_size={}",
+                                static_cast<int>(version_value), encoded.size);
+}
+
 void update_counter(RuntimeProfile* profile, const std::string& name, TUnit::type unit,
                     int64_t value) {
     COUNTER_UPDATE(ADD_COUNTER_WITH_LEVEL(profile, name, unit, 2), value);
@@ -403,16 +455,20 @@ Status MaterializationSharedState::validate_rpc_results(int node_id) {
 
 void MaterializationSharedState::_update_profile_info(int64_t backend_id,
                                                       RuntimeProfile* response_profile) {
+    DORIS_CHECK(response_profile != nullptr);
     if (!backend_profile_info_string.contains(backend_id)) {
         backend_profile_info_string.emplace(backend_id,
                                             std::map<std::string, fmt::memory_buffer> {});
     }
     auto& info_map = backend_profile_info_string[backend_id];
 
-    auto update_profile_info_key = [&](const std::string& info_key) {
+    auto update_profile_info_key = [&](const std::string& info_key, bool warn_if_missing = true) {
         const auto* info_value = response_profile->get_info_string(info_key);
         if (info_value == nullptr) [[unlikely]] {
-            LOG(WARNING) << "Get row id fetch rpc profile success, but no info key :" << info_key;
+            if (warn_if_missing) {
+                LOG(WARNING) << "Get row id fetch rpc profile success, but no info key :"
+                             << info_key;
+            }
             return;
         }
         if (!info_map.contains(info_key)) {
@@ -427,6 +483,13 @@ void MaterializationSharedState::_update_profile_info(int64_t backend_id,
     update_profile_info_key(RowIdStorageReader::FileReadLinesProfile);
     update_profile_info_key(FileScanner::FileReadBytesProfile);
     update_profile_info_key(FileScanner::FileReadTimeProfile);
+    update_profile_info_key(RowIdStorageReader::LanceDatasetOpenTimeProfile, false);
+    update_profile_info_key(RowIdStorageReader::LanceRowIdTakeReadTimeProfile, false);
+    update_profile_info_key(RowIdStorageReader::LanceArrowToDorisBlockTimeProfile, false);
+    update_profile_info_key(RowIdStorageReader::LanceRowIdFetchTotalTimeProfile, false);
+    for (const auto& [name, unit] : RowIdStorageReader::LanceFetchCountersProfile) {
+        update_profile_info_key(name, false);
+    }
 }
 
 Status MaterializationSharedState::create_muiltget_result(const Columns& columns, bool child_eos) {
@@ -451,19 +514,28 @@ Status MaterializationSharedState::create_muiltget_result(const Columns& columns
 
         for (int j = 0; j < rows; ++j) {
             if (!null_map || !null_map[j]) {
-                DCHECK(column_rowid->get_data_at(j).size == sizeof(GlobalRowLoacationV2));
-                GlobalRowLoacationV2 row_location =
-                        *((GlobalRowLoacationV2*)column_rowid->get_data_at(j).data);
+                MaterializationRowLocation row_location;
+                RETURN_IF_ERROR(decode_materialization_row_location(column_rowid->get_data_at(j),
+                                                                    &row_location));
                 auto rpc_struct = rpc_struct_map.find(row_location.backend_id);
                 if (UNLIKELY(rpc_struct == rpc_struct_map.end())) {
                     return Status::InternalError(
                             "MaterializationSinkOperatorX failed to find rpc_struct, backend_id={}",
                             row_location.backend_id);
                 }
-                rpc_struct->second.request.mutable_request_block_descs(i)->add_row_id(
-                        row_location.row_id);
-                rpc_struct->second.request.mutable_request_block_descs(i)->add_file_id(
-                        row_location.file_id);
+                auto* request_block_desc =
+                        rpc_struct->second.request.mutable_request_block_descs(i);
+                const auto row_location_version = static_cast<uint32_t>(row_location.version);
+                if (request_block_desc->row_id_size() == 0) {
+                    request_block_desc->set_row_location_version(row_location_version);
+                } else if (request_block_desc->row_location_version() != row_location_version) {
+                    return Status::InternalError(
+                            "mixed row location versions in one materialization request: "
+                            "actual={}, expected={}",
+                            row_location_version, request_block_desc->row_location_version());
+                }
+                request_block_desc->add_row_id(row_location.row_id);
+                request_block_desc->add_file_id(row_location.file_id);
                 block_order[j] = row_location.backend_id;
 
                 // Count rows per backend
@@ -593,8 +665,7 @@ Status MaterializationOperator::pull(RuntimeState* state, Block* output_block, b
              local_state._materialization_state.backend_profile_info_string) {
             auto* child_profile = local_state.operator_profile()->create_child(
                     "RowIDFetcher: BackendId:" + std::to_string(backend_id));
-            for (const auto& [info_key, info_value] :
-                 local_state._materialization_state.backend_profile_info_string[backend_id]) {
+            for (const auto& [info_key, info_value] : child_info) {
                 child_profile->add_info_string(info_key, "{" + fmt::to_string(info_value) + "}");
             }
             local_state.operator_profile()->add_child(child_profile, true);