cudf 列重排 API 深度指南：libcudf 的 Partitioning 与 Stream Compaction 双核心

• 数据分析
• 数据工程
• 机器学习

本篇技术文章围绕 libcudf 文档体系中的 “Column Reorder”（列重排）API 分组展开，对应文档页 column_reorder.rst。该分组是 libcudf 中所有“按某种规则重新排列行顺序”能力的总入口，下分两个子分组：Partitioning（行分区重排，文档页 reorder_partition.rst）和 Stream Compaction（流式压缩/行过滤，文档页 reorder_compact.rst）。读完本文后，你将掌握partition、hash_partition、round_robin_partition三种分区重排 API 的签名、参数语义与边界行为，以及drop_nulls、drop_nans、apply_retention_mask、unique、distinct等流式压缩 API 的用法、废弃演进与源码/测试定位方式。

1. Column Reorder 分组在 libcudf 中的定位

column_reorder.rst 本身是一个 Doxygen 分组页面，其内容只有一个指令：

.. doxygengroup:: column_reorder :members:

以及指向两个子页面的 toctree（reorder_partition、reorder_compact）。分组的层级结构在 doxygen_groups.h 中定义：

@defgroup column_reorder Reordering @{ @defgroup reorder_partition Partitioning @defgroup reorder_compact Stream Compaction @}

从源码结构看，这两个@addtogroup标记分别落在两个公开头文件中，也就是说Column Reorder 分组的实际 API 集合 = partitioning.hpp + stream_compaction.hpp：

• cpp/include/cudf/partitioning.hpp 开头即为@addtogroup reorder_partition；
• cpp/include/cudf/stream_compaction.hpp 开头即为@addtogroup reorder_compact。

这两个头文件共同构成了 GPU 上“重排行”的两类原语：一类是把同一分组的行聚拢成连续块（分区），另一类是把不满足条件的行从序列中剔除（压缩）。二者输出都带有明确的“稳定”或“未定义序”语义，这是使用时必须理解的关键点。

2. Reorder Partition：行分区重排

Partitioning 子分组定义于 partitioning.hpp，文件头注释写明其用途为 “Column partitioning APIs”。它提供 4 个公开函数和 1 个枚举。

2.1 哈希函数选择：hash_id

enum class hash_id : int32_t { HASH_IDENTITY = 0, ///< 恒等哈希，直接返回待哈希的 key HASH_MURMUR3 ///< Murmur3 哈希函数 };

定义见 partitioning.hpp#L32-L35。hash_partition的默认哈希函数为HASH_MURMUR3，默认种子为头文件中定义的DEFAULT_HASH_SEED（该常量来自 hashing.hpp，通过#include <cudf/hashing.hpp>引入）。

2.2partition：按显式分区映射重排

签名见 partitioning.hpp#L71-L76：

std::pair<std::unique_ptr<table>, std::vector<size_type>> partition( table_view const& t, column_view const& partition_map, size_type num_partitions, cuda::stream_ref stream = cudf::get_default_stream(), rmm::device_async_resource_ref mr = cudf::get_current_device_resource_ref());

语义要点（均来自头文件 Doxygen 注释）：

• 对t中第i行，partition_map[i]指定该行属于哪个分区；输出的新表中同一分区的行连续放置，整体按分区号升序[0, num_partitions)排列；分区内部的行序未定义。
• 返回num_partitions + 1个偏移量offsets：分区i的行位于[offsets[i], offsets[i+1])。若某个分区值j未出现在partition_map中，则该分区为空，即offsets[j+1] - offsets[j] == 0。
• 约束与异常：
  • partition_map的值必须在[0, num_partitions)内，否则行为未定义；
  • partition_map非整数类型 → 抛cudf::logic_error；
  • partition_map.has_nulls() == true→ 抛cudf::logic_error；
  • partition_map.size() != t.num_rows()→ 抛cudf::logic_error。

这是一个“分区结果已知、只要求重排”的底层原语：上层可以先用hash_partition或任何自定义逻辑算出partition_map，再调用partition完成物理重排。

2.3hash_partition：按列哈希自动分区

hash_partition有两个重载，都返回std::pair<std::unique_ptr<table>, std::vector<size_type>>。

重载一：按列索引哈希（partitioning.hpp#L103-L110）：

std::pair<std::unique_ptr<table>, std::vector<size_type>> hash_partition( table_view const& input, std::vector<size_type> const& columns_to_hash, // 参与哈希的列下标 int num_partitions, hash_id hash_function = hash_id::HASH_MURMUR3, uint32_t seed = DEFAULT_HASH_SEED, cuda::stream_ref stream = cudf::get_default_stream(), rmm::device_async_resource_ref mr = cudf::get_current_device_resource_ref());

重载二：按 keys 表哈希（partitioning.hpp#L138-L145）：

std::pair<std::unique_ptr<table>, std::vector<size_type>> hash_partition( table_view const& input, table_view const& keys, // 指定参与哈希的键表 int num_partitions, hash_id hash_function = hash_id::HASH_MURMUR3, uint32_t seed = DEFAULT_HASH_SEED, ...);

两个重载的行为约定一致：

• 行被分入num_partitions个桶，同一桶的行在输出表中连续存放；偏移向量长度为num_partitions + 1，最后一个偏移恒等于输出表总行数。
• 空columns_to_hash（或零列keys表）被视为空输入：即使input行数非零，也返回空结果（不抛错）。
• 重载一的异常：columns_to_hash中下标非法时抛std::out_of_range；重载二：keys非空且行数与input不一致时抛std::invalid_argument。

分区内行序同样未定义——这是分布式哈希连接、并行 groupby 等场景所需要的“桶均衡”性质，而非有序性保证。

2.4round_robin_partition：轮询分区

签名见 partitioning.hpp#L286-L291：

std::pair<std::unique_ptr<cudf::table>, std::vector<cudf::size_type>> round_robin_partition( table_view const& input, cudf::size_type num_partitions, cudf::size_type start_partition = 0, cuda::stream_ref stream = cudf::get_default_stream(), rmm::device_async_resource_ref mr = cudf::get_current_device_resource_ref());

行按其在表中的行下标以轮询方式分配给各分区。头文件注释给出了一个形象的“发牌”类比（partitioning.hpp#L160-L183）：

• 一副牌 = 表的行；
• 玩家数 = 分区数；
• start_partition= 从哪个玩家开始发牌；
• 结果一：把每位玩家的牌按顺序叠回一副新牌堆（玩家 0 在前）；
• 结果二：一个偏移向量，指示每位玩家的牌堆从哪个位置开始。

异常约束：num_partitions <= 1或start_partition >= num_partitions时抛cudf::logic_error。

当num_partitions > nrows时，从起始玩家开始发牌直到牌发完；没拿到牌的分区表现为offset[i] == offset[i+1]（空分区）。头文件中给出了 9 个完整示例（partitioning.hpp#L185-L275），这里摘录两个典型：

示例 1: 输入: table => col 1 {0, ..., 12} (13 行) num_partitions = 3, start_partition = 0 输出: table => col 1 {0,3,6,9,12, 1,4,7,10, 2,5,8,11} partition_offsets => {0, 5, 9, 13} 示例 6: 输入: table => col 1 {0, ..., 10} (11 行) num_partitions = 15 > num_rows = 11, start_partition = 2 输出: table => col 1 {0,1,2,3,4,5,6,7,8,9,10} (行序不变) partition_offsets => {0,0,0,1,2,3,4,5,6,7,8,9,10,11,11,11}

示例 6 值得注意：当分区数远大于行数时，轮询退化为“每分区至多 1 行”，空分区以连续相等偏移表示。

2.5 Partitioning 的源码与测试位置

从源码结构看，分区重排的实现集中在 cpp/src/partitioning/ 目录：partitioning.cu 承载partition/hash_partition，round_robin.cu 承载轮询分区。对应测试位于 cpp/tests/partitioning/，包括：

• partition_test.cpp
• hash_partition_test.cpp
• round_robin_test.cpp

如果要验证某个分区行为（例如空columns_to_hash的退化语义），直接读对应测试文件即可复现头文件注释中的边界约定。

3. Reorder Compact：流式压缩（行过滤）

Stream Compaction 子分组定义于 stream_compaction.hpp，文件头注释写明 “Column APIs for filtering rows”。它提供两组能力：按 null/NAN 条件过滤，按布尔掩码过滤，以及去重（去重本质是“保留满足去重条件的行”）。

3.1 按 null 过滤：drop_nulls

带阈值版本（stream_compaction.hpp#L73-L78）：

std::unique_ptr<table> drop_nulls( table_view const& input, std::vector<size_type> const& keys, // 参与判定的列下标 cudf::size_type keep_threshold, // 行内至少多少个非 null 字段才保留 cuda::stream_ref stream = cudf::get_default_stream(), rmm::device_async_resource_ref mr = cudf::get_current_device_resource_ref());

头文件给出的伪代码示例（stream_compaction.hpp#L49-L59）：

input {col1: {1, 2, 3, null}, col2: {4, 5, null, null}, col3: {7, null, null, null}} keys = {0, 1, 2} // 全部三列 keep_threshold = 2 output {col1: {1, 2}, col2: {4, 5}, col3: {7, null}}

即：第 2、3 行在 keys 中非 null 字段数分别为 2 和 1，只有满足>= keep_threshold的行被保留。要点：

• 该操作是稳定的：输入行序在输出中保持；
• 输入中不可空（non-nullable）的列被视为“全非 null”；
• 若input零行、或keys为空、或 keys 无 null，不报错，返回空表。

无阈值版本（stream_compaction.hpp#L107-L111）等价于keep_threshold = keys.size()，即 keys 列必须全部非 null才保留该行。

3.2 按 NAN 过滤：drop_nans

签名与drop_nulls完全平行，但判定对象是浮点 NaN：带阈值版（stream_compaction.hpp#L151-L156）与无阈值版（stream_compaction.hpp#L186-L190）。约束：

• keys 列必须为浮点类型，否则抛cudf::logic_error；
• 同样是稳定操作，空输入/空 keys 返回空表。

头文件示例（阈值版，stream_compaction.hpp#L125-L135）：

input {col1: {1.0, 2.0, 3.0, NAN}, col2: {4.0, null, NAN, NAN}, col3: {7.0, NAN, NAN, NAN}} keys = {0, 1, 2}, keep_threshold = 2 output {col1: {1.0, 2.0}, col2: {4.0, null}, col3: {7.0, NAN}}

注意 null 与 NaN 是相互独立的两种“缺失”：drop_nans只按 NaN 计数，null 字段不影响判定（示例中 col2 第 2 行为 null 仍被保留）。

3.3 布尔掩码过滤：apply_retention_mask/apply_deletion_mask

保留掩码（stream_compaction.hpp#L214-L218）：

std::unique_ptr<table> apply_retention_mask( table_view const& input, column_view const& retention_mask, // 必须为 type_id::BOOL8，可空 cuda::stream_ref stream = cudf::get_default_stream(), rmm::device_async_resource_ref mr = cudf::get_current_device_resource_ref());

• 掩码中非 null 且为true的行被复制到输出；掩码为 null 或false的行被剔除。
• 掩码为空或输入零行 → 返回空表；
• 非空输入与掩码行数不一致 → 抛cudf::logic_error；掩码类型不是BOOL8→ 抛cudf::logic_error。

删除掩码（stream_compaction.hpp#L262-L266）语义相反：掩码中非 null 且为false的行被剔除；掩码为空时返回输入的拷贝。

演进提示：旧 APIapply_boolean_mask（stream_compaction.hpp#L233-L237）已标注[[deprecated]]，注释写明自 release 26.10 起弃用，应改用apply_retention_mask。新代码应直接以“保留/删除”两个方向显式表达意图。

3.4 去重：unique/distinct/stable_distinct/distinct_indices

去重语义由枚举duplicate_keep_option（stream_compaction.hpp#L271-L276）控制：

enum class duplicate_keep_option { KEEP_ANY = 0, ///< 保留不指定的某一个重复行 KEEP_FIRST, ///< 保留第一次出现 KEEP_LAST, ///< 保留最后一次出现 KEEP_NONE ///< 重复行全部删除 };

四个 API 的分工：

值得注意的实现约定（来自头文件注释）：

• unique与distinct都支持只按keys列判重，其余列随行拷贝；零列输入视为空，返回空表；
• 性能提示：若输入已预排序，用unique可以得到与distinct相同的输出行集合，但运行时间更短（unique是相邻判重，distinct需要全局哈希/排序）；
• stable_distinct对KEEP_ANY的语义有精确说明：保留的是输入序中的“某个”重复行，例如 keys 列1, 2, 1、values 列3, 4, 5，结果可能是 values3, 4或4, 5，但不会出现4, 3或5, 4（stream_compaction.hpp#L385-L388）。

去重的测试用例分布在 cpp/tests/stream_compaction/ 目录（含unique、distinct、stable_distinct、apply_mask等测试文件，由 ci/discover_libcudf_tests.sh 发现的 CTest 目标覆盖）。

3.5 已弃用的 UDF/ASTfilter系列

stream_compaction.hpp中还有三个filter重载：基于 UDF 的列过滤（L451-L459）、带filter_input（列/标量变体）的扩展版（L506-L514）、以及基于 AST 表达式的表过滤（L542-L546）。三者均标注[[deprecated]]（自 release 26.12 起），官方推荐的替代路径是：

1. 用cudf::transform（或cudf::compute_column）先算出布尔掩码列；
2. 再调用apply_retention_mask或apply_deletion_mask完成过滤。

这一“先掩码、后压缩”的两步式写法是当前推荐的流式压缩主路径，也解释了为什么掩码 API 被拆成 retention/deletion 两个显式方向。

4. 公共约定：stream 与 mr 参数

两个头文件中的所有 API 共享同一套尾部参数：

• cuda::stream_ref stream = cudf::get_default_stream()：设备内存操作与 kernel 启动所用的 CUDA stream，默认值定义见 default_stream.hpp；
• rmm::device_async_resource_ref mr = cudf::get_current_device_resource_ref()：结果设备内存使用的 RMM 资源，默认取当前设备资源（memory_resource.hpp）。

这意味着所有重排/压缩操作都可以无缝接入调用方的 stream 编排与自定义内存资源（如池化分配器），是构建多流管线（如 IO → 分区 → 计算）时的基本前提。

返回类型也有统一模式：分区类 API 返回std::pair<unique_ptr<table>, vector<size_type>>（表 + 偏移向量）；过滤类 API 直接返回unique_ptr<table>；distinct_indices返回unique_ptr<column>（行号列）。偏移向量的统一约定是长度num_partitions + 1、末位等于输出总行数，可用slice/gather等 API 按[offsets[i], offsets[i+1])切出各分区。

5. 小结与延伸阅读

Column Reorder 分组是 libcudf 行级重排原语的集合：分区侧提供partition（显式映射）、hash_partition（列/键表哈希，默认 Murmur3）、round_robin_partition（轮询发牌）三个重排工具；压缩侧提供drop_nulls/drop_nans（阈值化 null/NAN 过滤）、apply_retention_mask/apply_deletion_mask（布尔掩码过滤）、unique/distinct/stable_distinct/distinct_indices（去重）四组过滤工具。所有 API 均遵循stream/mr双参数约定，并在头文件注释中以伪代码示例明确了空输入、空分区等边界语义。

继续深入可参考的路径：

• 文档页：column_reorder.rst、reorder_partition.rst、reorder_compact.rst；
• API 头文件：partitioning.hpp、stream_compaction.hpp；
• 实现：cpp/src/partitioning/、cpp/src/stream_compaction/；
• 测试：cpp/tests/partitioning/、cpp/tests/stream_compaction/。

• 数据分析
• 数据工程
• 机器学习