Spark Datafusion Comet 向量化Rust Native--Native算子ScanExec以及涉及到的Selection Vectors

Ne0inhk

— 7 min read

背景

Apache Datafusion Comet 是苹果公司开源的加速Spark运行的向量化项目。
本项目采用了 Spark插件化 + Protobuf + Arrow + DataFusion 架构形式
其中

  • Spark插件是 利用 SparkPlugin 插件，其中分为 DriverPlugin 和 ExecutorPlugin ,这两个插件在driver和 Executor启动的时候就会调用
  • Protobuf 是用来序列化 spark对应的表达式以及计划，用来传递给 native 引擎去执行,利用了 体积小，速度快的特性
  • Arrow 是用来 spark 和 native 引擎进行高效的数据交换(native执行的结果或者spark执行的数据结果)，主要在JNI中利用Arrow IPC 列式存储以及零拷贝等特点进行进程间数据交换
  • DataFusion 主要是利用Rust native以及Arrow内存格式实现的向量化执行引擎，Spark中主要offload对应的算子到该引擎中去执行

本文基于 datafusion comet 截止到2026年1月13号的main分支的最新代码(对应的commit为 eef5f28a0727d9aef043fa2b87d6747ff68b827a)
主要分析Rust Native的Spark Datafusion Comet 向量化Rust Native–执行Datafusion计划中ScanExec以及涉及到的Selection Vectors

Selection Vectors

什么是Selection Vectors

Selection Vectors 是向量化查询执行引擎过滤操作中的一种表达，还有另一种表达是 Bitmap :

  1. Bitmap 表达是： 用 BitMap 来标记哪些数据是被过滤选中的
  2. Selection Vectors表达是：用 vector 存储被命中的数据的下标
    两者的区别是Bitmap表达会记录所有的数据，只不过是用不同的0/1代表存在与否，而 Selection Vectors 只记录命中的数据
    具体相关的论文可以参考Filter Representation in Vectorized Query Execution
    针对这两种过滤算子的表达，可以衍生出三种执行策略：
  • BMFull：总是对所有数据处理，未选中的数据的值未定义，优势是能充分发挥向量化的优势
  • BMPartial：只对选中的数据进行处理，无法利用向量化，依然需要遍历所有下标
  • SVPartial：只需要遍历选中的下标，无法利用向量化

ScanExec读取以及涉及到的Select Vectors

    • values 为一列中的所有原始值
    • selectionIndices为选中的数据的下标
      • 假如存在则获取每一列的Selection Vector,否则返回None
        • 首先用对于每一列值构造一个Vec(FFI_ArrowArray)和Vec(FFI_ArrowSchema)类型的数组以及初始化数组，并创建的 FFI_ArrowArray和FFI_ArrowSchema对应的地址插入到该数组中
        • 使用JNIEnv.new_long_array创建Java Long型数组
        • 使用JNIEnv.set_long_array_region新创建的Java Long型数组(也就是对应的FFIArray/Schema地址)赋值给该数组
        • 调用ArrayData::from_spark方法将Spark 端通过 Arrow C Data Interface 传递过来的内存地址转换为 Rust 端的 Arrow ArrayData 对象
          这里主要使用了from_ffi方法，从这些裸指针（Raw Pointers）重建出 Rust 的 ArrayData 结构，这个过程是零拷贝的（Zero-copy），直接复用 Spark 分配的内存；并调用 align_buffers() 确保数据在 Rust 端能被正确、安全地访问（例如 SIMD 操作对内存对齐有要求）
    • allocate_and_fetch_batch方法 及后续说明
    • 调用allocate_and_fetch_batch方法从Java端获取一批数据并赋值到传入的FFI_ArrowArray和FFI_ArrowSchema中
      对于java端的处理和之前的exportSingleVector方法处理一样，主要是使用Data.exportVector方法来进行数据回传
    • 如果存在Selection Vector，则使用Arrow take方法获取到真正的值，否则就是原值
      其中对&Arc使用 &**操作是将类型由引用 &Arc 依次转换为 Arc、T，最后再取引用 &T，用于不转移所有权地读取数据。
    • 如果存在Selection Vector，则返回真正的行数
      最后组装成InputBatch::new(inputs, Some(actual_num_rows)返回

JNI调用JVM的exportSelectionIndices方法

 let _exported_count: i32 = unsafe { jni_call!(env, comet_batch_iterator(iter).export_selection_indices( JValueGen::Object(JObject::from(indices_array_obj).as_ref()), JValueGen::Object(JObject::from(indices_schema_obj).as_ref()) ) -> i32)? };

这里用到的as_ref方法通常用于将高级包装类型（如JObject, JString等）转换为对底层JNI指针（jobject）的共享引用。它使得在不改变对象所有权的情况下，可以安全地将对象传递给JNIEnv函数进行后续操作，
在Java侧的话，主要是NativeUtil.exportSingleVector的调用:

 def exportSingleVector(vector: CometVector, arrayAddr: Long, schemaAddr: Long): Unit = { val valueVector = vector.getValueVector val provider = if (valueVector.getField.getDictionary != null) { vector.getDictionaryProvider } else { null } val arrowSchema = ArrowSchema.wrap(schemaAddr) val arrowArray = ArrowArray.wrap(arrayAddr) Data.exportVector( allocator, getFieldVector(valueVector, "export"), provider, arrowArray, arrowSchema) }

Data.exportVector 使用这个方法使Selection Vector回传到Rust端，

判断是否存在Selection Vectors,通过JNI调用java侧方法hasSelectionVectors：

 jni_call!(env, comet_batch_iterator(iter).has_selection_vectors() -> jni::sys::jboolean

get_selection_indices方法说明

 fn get_selection_indices( env: &mut jni::JNIEnv, iter: &JObject, num_cols: usize, ) -> Result<Option<Vec<ArrayRef>>, CometError> { // Check if all columns have selection vectors let has_selection_vectors_result: jni::sys::jboolean = unsafe { jni_call!(env, comet_batch_iterator(iter).has_selection_vectors() -> jni::sys::jboolean)? }; let has_selection_vectors = has_selection_vectors_result != 0; let selection_indices_arrays = if has_selection_vectors { // Allocate arrays for selection indices export (one per column) let mut indices_array_addrs = Vec::with_capacity(num_cols); let mut indices_schema_addrs = Vec::with_capacity(num_cols); for _ in 0..num_cols { let arrow_array = Rc::new(FFI_ArrowArray::empty()); let arrow_schema = Rc::new(FFI_ArrowSchema::empty()); indices_array_addrs.push(Rc::into_raw(arrow_array) as i64); indices_schema_addrs.push(Rc::into_raw(arrow_schema) as i64); } // Prepare JNI arrays for the export call let indices_array_obj = env.new_long_array(num_cols as jsize)?; let indices_schema_obj = env.new_long_array(num_cols as jsize)?; env.set_long_array_region(&indices_array_obj, 0, &indices_array_addrs)?; env.set_long_array_region(&indices_schema_obj, 0, &indices_schema_addrs)?; // Export selection indices from JVM let _exported_count: i32 = unsafe { jni_call!(env, comet_batch_iterator(iter).export_selection_indices( JValueGen::Object(JObject::from(indices_array_obj).as_ref()), JValueGen::Object(JObject::from(indices_schema_obj).as_ref()) ) -> i32)? }; // Convert to ArrayRef for easier handling let mut selection_arrays = Vec::with_capacity(num_cols); for i in 0..num_cols { let array_data = ArrayData::from_spark((indices_array_addrs[i], indices_schema_addrs[i]))?; selection_arrays.push(make_array(array_data)); // Drop the references to the FFI arrays unsafe { Rc::from_raw(indices_array_addrs[i] as *const FFI_ArrowArray); Rc::from_raw(indices_schema_addrs[i] as *const FFI_ArrowSchema); } } Some(selection_arrays) } else { None }; Ok(selection_indices_arrays) }

Rust侧

 let (num_rows, array_addrs, schema_addrs) = Self::allocate_and_fetch_batch(&mut env, iter, num_cols)?; let mut inputs: Vec<ArrayRef> = Vec::with_capacity(num_cols); // Process each column for i in 0..num_cols { let array_ptr = array_addrs[i]; let schema_ptr = schema_addrs[i]; let array_data = ArrayData::from_spark((array_ptr, schema_ptr))?; // TODO: validate array input data // array_data.validate_full()?; let array = make_array(array_data); // Apply selection if selection vectors exist (applies to all columns) let array = if let Some(ref selection_arrays) = selection_indices_arrays { let indices = &selection_arrays[i]; // Apply the selection using Arrow's take kernel match take(&*array, &**indices, None) { Ok(selected_array) => selected_array, Err(e) => { return Err(CometError::from(ExecutionError::ArrowError(format!( "Failed to apply selection for column {i}: {e}", )))); } } } else { array }; let array = if arrow_ffi_safe { // ownership of this array has been transferred to native // but we still need to unpack dictionary arrays copy_or_unpack_array(&array, &CopyMode::UnpackOrClone)? } else { // it is necessary to copy the array because the contents may be // overwritten on the JVM side in the future copy_array(&array) }; inputs.push(array); // Drop the Arcs to avoid memory leak unsafe { Rc::from_raw(array_ptr as *const FFI_ArrowArray); Rc::from_raw(schema_ptr as *const FFI_ArrowSchema); } } // If selection was applied, determine the actual row count from the selected arrays let actual_num_rows = if let Some(ref selection_arrays) = selection_indices_arrays { if !selection_arrays.is_empty() { // Use the length of the first selection array as the actual row count selection_arrays[0].len() } else { num_rows as usize } } else { num_rows as usize }; Ok(InputBatch::new(inputs, Some(actual_num_rows)))

indices为 java数组selectionIndices 以Arrow vector的表示，便于其他语言能够以零拷贝的方式访问这些数据，后续会被传递给 Native(Rust) 层

 this.indices = CometVector.getVector(indicesVector, values.useDecimal128, values.getDictionaryProvider());

Java侧

 public boolean hasSelectionVectors() { if (currentBatch == null) { return false; } // Check if all columns are CometSelectionVector instances for (int i = 0; i < currentBatch.numCols(); i++) { if (!(currentBatch.column(i) instanceof CometSelectionVector)) { return false; } } return true; }

这其中的 CometSelectionVector就是对应上文中说到的selection vector,具体的代码如下:

 public class CometSelectionVector extends CometVector { /** The original vector containing all values */ private final CometVector values; /** * The valid indices in the values vector. This array is converted into an Arrow vector so we can * transfer the data to native in one JNI call. This is used to represent the rowid mapping used * by Iceberg */ private final int[] selectionIndices; /** * The indices vector containing selection indices. This is currently allocated by the JVM side * unlike the values vector which is allocated on the native side */ private final CometVector indices; /**

参考

Read more

计算机视觉热点：三维人体姿态估计的前沿算法与论文案例

计算机视觉热点：三维人体姿态估计的前沿算法与论文案例

计算机视觉热点：三维人体姿态估计的前沿算法与论文案例 * 一、前言 * 二、三维人体姿态估计概述 * 2.1 定义与目标 * 2.2 应用场景 * 2.3 面临的挑战 * 三、前沿算法介绍 * 3.1 基于深度学习的方法 * 3.2 多视角方法 * 3.3 结合传感器的方法 * 四、算法对比与分析 * 4.1 不同算法的性能比较 * 4.2 适用场景分析 * 五、数据集介绍 * 5.1 常用数据集概述 * 5.2 数据集特点与应用 * 六、未来发展趋势 * 6.1 算法优化方向 * 6.2 新兴技术融合

By Ne0inhk
优选算法——滑动窗口

优选算法——滑动窗口

优选算法——滑动窗口 1.长度最小的子数组 解题原理 📋 解题步骤 第一步：理解题意 * 找一个连续子数组，使其和 ≥ target，且长度最小 * 数组元素都是正整数（关键性质） * 无解返回 0 第二步：分析暴力解法 * 枚举所有子数组：O(n²) 或 O(n³) * 对于 n = 10⁵ 会超时 第三步：寻找优化点 * 正整数 → 窗口扩展时和单调递增 * 可以用滑动窗口优化 第四步：设计滑动窗口 遍历右指针： 扩展窗口 从右边进窗口 判断： 如果 sum >= target： 更新最小长度 收缩窗口 从左边出窗口 第五步：手动模拟 步骤leftright窗口sumresult403[2,

By Ne0inhk
C++：探索哈希表秘密之哈希桶实现哈希

C++：探索哈希表秘密之哈希桶实现哈希

文章目录 * 前言 * 一、链地址法概念 * 二、哈希表扩容 * 三、哈希桶插入逻辑 * 四、析构函数 * 五、删除逻辑 * 六、查找 * 七、链地址法代码实现总结 前言 前面我们用开放定址法代码实现了哈希表： C++：揭秘哈希：提升查找效率的终极技巧_1 对于开放定址法来说，包含以下两种探测插入节点位置方法： 1. 线性探测 2. 二次探测 但是开放定址法的缺点也很明显，开放定址法容易很多数据堆积在一起，大大减少了效率。 为了解决上述问题，引入了第二种方法实现哈希表 ——链地址法（哈希桶法） 一、链地址法概念 开放定址法中，所有的元素都放到哈希表里。 链地址法中，所有的数据不再直接存储在哈希表中。哈希表中存储一个指针，没有数据映射到这个位置时，这个指针为空；有多个数据映射到这个位置时，我们把这些冲突的数据链接成一个链表，挂在哈希表这个位置下面。链地址法也叫做拉链法或者哈希桶。 下⾯演⽰

By Ne0inhk
【数据结构】彻底搞懂二叉树：四种遍历逻辑、经典OJ题与递归性能全解析

【数据结构】彻底搞懂二叉树：四种遍历逻辑、经典OJ题与递归性能全解析

🏠 个人主页:EXtreme35 📚 个人专栏: 专栏名称专栏主题简述《C语言》C语言基础、语法解析与实战应用《数据结构》线性表、树、图等核心数据结构详解《题解思维》算法思路、解题技巧与高效编程实践 目录 * 二叉树全栈进阶指南：从内存布局到递归本质的深度复盘 * 一、二叉树的底层逻辑与核心概念 * 1.1 核心定义与特点 * 1.2 二叉树的五种基本形态 * 1.3 特殊二叉树 * 1.4 二叉树的五条性质 * 1.5 存储结构 * 二、遍历的递归之美 * 2.1 前序遍历 * 2.2 中序遍历 (In-order Traversal) * 2.3 后序遍历 (Post-order Traversal) * 2.

By Ne0inhk