Fix engine bugs and CPU fallbacks found on science datasets

.config/nextest.toml

@@ -501,6 +501,9 @@ binary(/_cpu$/)
     | internal::resolve::tests::validate_no_duplicate_names_rejects_dup
     | io::tests::test_collect_column_names_rejects_negative_count
     | ops::reshape::tests::interleave_output_size_overflow_is_structured
+    | ops::concatenate::tests::concatenated_vector_frames_overflow_at_the_list_child_with_the_width_row_cap
+    | ops::concatenate::tests::concatenation_up_to_the_width_row_cap_is_accepted_and_one_row_more_is_rejected
+    | ops::concatenate::tests::concatenated_flat_rows_overflow_at_the_row_count
     | runtime::device_health::tests::cross_device_teardown_suspends_instead_of_quarantining
     | diagnostics::memory::tests::fixed_pressure_plan_is_bounded_and_preserves_the_documented_reserve
     | runtime::matrix::tests::matrix_element_count_accepts_cudf_size_type_maximum

.config/policies/public-api/algeon-cudf.txt

@@ -3,6 +3,7 @@ pub mod cudf::core
 pub mod cudf::core::column
 pub struct cudf::core::column::Column
 impl cudf::core::Column
+pub const cudf::core::Column::MAX_ELEMENTS: usize
 pub fn cudf::core::Column::data_type(&self) -> cudf::core::error::Result<cudf::core::types::DataType>
 pub fn cudf::core::Column::deep_copy(&self) -> cudf::core::error::Result<Self>
 pub fn cudf::core::Column::device_data_ptr(&self) -> cudf::core::error::Result<*const std::os::raw::c_void>
@@ -1030,6 +1031,7 @@ impl core::fmt::Display for cudf::core::types::TypeId
 pub fn cudf::core::types::TypeId::fmt(&self, &mut core::fmt::Formatter<'_>) -> core::fmt::Result
 pub struct cudf::core::Column
 impl cudf::core::Column
+pub const cudf::core::Column::MAX_ELEMENTS: usize
 pub fn cudf::core::Column::data_type(&self) -> cudf::core::error::Result<cudf::core::types::DataType>
 pub fn cudf::core::Column::deep_copy(&self) -> cudf::core::error::Result<Self>
 pub fn cudf::core::Column::device_data_ptr(&self) -> cudf::core::error::Result<*const std::os::raw::c_void>
@@ -4176,6 +4178,8 @@ impl cudf::Runtime
 pub fn cudf::Runtime::dataframe_from_record_batch(&self, &arrow_array::record_batch::RecordBatch) -> cudf::core::error::Result<cudf::DataFrame>
 pub fn cudf::Runtime::dataframe_from_record_batches(&self, &[arrow_array::record_batch::RecordBatch]) -> cudf::core::error::Result<cudf::DataFrame>
 impl cudf::Runtime
+pub fn cudf::Runtime::empty_dataframe_from_arrow_schema(&self, &arrow_schema::schema::Schema) -> cudf::core::error::Result<cudf::DataFrame>
+impl cudf::Runtime
 pub fn cudf::Runtime::narrow_packed_matrix_to_float32(&self, &cudf::PackedMatrix) -> cudf::core::error::Result<cudf::PackedMatrix>
 pub fn cudf::Runtime::pack_columns(&self, &[&cudf::core::Column], cudf::DenseMatrixLayout) -> cudf::core::error::Result<cudf::PackedMatrix>
 pub fn cudf::Runtime::pack_flat_column(&self, &cudf::core::Column, usize, usize, cudf::DenseMatrixLayout) -> cudf::core::error::Result<cudf::PackedMatrix>
@@ -5707,6 +5711,8 @@ impl cudf::Runtime
 pub fn cudf::Runtime::dataframe_from_record_batch(&self, &arrow_array::record_batch::RecordBatch) -> cudf::core::error::Result<cudf::DataFrame>
 pub fn cudf::Runtime::dataframe_from_record_batches(&self, &[arrow_array::record_batch::RecordBatch]) -> cudf::core::error::Result<cudf::DataFrame>
 impl cudf::Runtime
+pub fn cudf::Runtime::empty_dataframe_from_arrow_schema(&self, &arrow_schema::schema::Schema) -> cudf::core::error::Result<cudf::DataFrame>
+impl cudf::Runtime
 pub fn cudf::Runtime::narrow_packed_matrix_to_float32(&self, &cudf::PackedMatrix) -> cudf::core::error::Result<cudf::PackedMatrix>
 pub fn cudf::Runtime::pack_columns(&self, &[&cudf::core::Column], cudf::DenseMatrixLayout) -> cudf::core::error::Result<cudf::PackedMatrix>
 pub fn cudf::Runtime::pack_flat_column(&self, &cudf::core::Column, usize, usize, cudf::DenseMatrixLayout) -> cudf::core::error::Result<cudf::PackedMatrix>

.config/policies/public-api/algeon-query-engine.txt

@@ -40,6 +40,7 @@ pub fn query_engine::rapids_backend::RapidsBackend::aggregate_output_names(&self
 pub fn query_engine::rapids_backend::RapidsBackend::binary_output_type(&self, query_planning::expr::BinaryOp, &arrow_schema::datatype::DataType, &arrow_schema::datatype::DataType) -> query_planning::error::Result<arrow_schema::datatype::DataType>
 pub fn query_engine::rapids_backend::RapidsBackend::check_expression(&self, &query_planning::expr::Expr, query_planning::backend::ExpressionUse) -> query_planning::error::Result<()>
 pub fn query_engine::rapids_backend::RapidsBackend::check_projection_output(&self, &query_planning::plan::nodes::ProjectionOutput) -> query_planning::error::Result<()>
+pub fn query_engine::rapids_backend::RapidsBackend::max_column_elements(&self) -> u64
 pub fn query_engine::rapids_backend::RapidsBackend::supports_type(&self, &arrow_schema::datatype::DataType) -> bool
 pub fn query_engine::rapids_backend::rapids_analysis_context(query_planning::policy::PlanningPolicy) -> query_planning::backend::AnalysisContext<'static>
 pub const fn query_engine::rapids_backend::rapids_backend() -> &'static query_engine::rapids_backend::RapidsBackend

.config/policies/public-api/algeon-query-planning.txt

@@ -40,6 +40,7 @@ pub fn query_planning::backend::BackendCapabilities::aggregate_output_names(&sel
 pub fn query_planning::backend::BackendCapabilities::binary_output_type(&self, query_planning::expr::BinaryOp, &arrow_schema::datatype::DataType, &arrow_schema::datatype::DataType) -> query_planning::error::Result<arrow_schema::datatype::DataType>
 pub fn query_planning::backend::BackendCapabilities::check_expression(&self, &query_planning::expr::Expr, query_planning::backend::ExpressionUse) -> query_planning::error::Result<()>
 pub fn query_planning::backend::BackendCapabilities::check_projection_output(&self, &query_planning::plan::ProjectionOutput) -> query_planning::error::Result<()>
+pub fn query_planning::backend::BackendCapabilities::max_column_elements(&self) -> u64
 pub fn query_planning::backend::BackendCapabilities::supports_type(&self, &arrow_schema::datatype::DataType) -> bool
 pub mod query_planning::capability
 pub enum query_planning::capability::AggregateExecutionStage
@@ -191,8 +192,6 @@ pub query_planning::capability::Reason::DatafusionExactStatisticalGroupKeyUnsupp
 pub query_planning::capability::Reason::DatafusionExactUngroupedFloatSumUnsupported
 pub query_planning::capability::Reason::ExpressionLineageRequiresResolvedInputColumn
 pub query_planning::capability::Reason::LargeKeyedHashJoinStrategyUnsupported
-pub query_planning::capability::Reason::LeftAntiBuildRequiresDedupAggregate
-pub query_planning::capability::Reason::LeftAntiProbeRequiresDedupProvenInput
 pub query_planning::capability::Reason::MissingEdgeProducer
 pub query_planning::capability::Reason::OutputLineageAggregateFieldIndexOutOfRange
 pub query_planning::capability::Reason::OutputLineageFieldIndexOutOfRange
@@ -381,6 +380,7 @@ pub fn query_planning::chunk::decoded::row_group_decoded_bytes_from_footer(&parq
 pub struct query_planning::chunk::SourceChunk
 pub query_planning::chunk::SourceChunk::chunk_index: usize
 pub query_planning::chunk::SourceChunk::estimated_chunk_bytes: core::option::Option<u64>
+pub query_planning::chunk::SourceChunk::estimated_column_elements: core::option::Option<u64>
 pub query_planning::chunk::SourceChunk::estimated_decoded_bytes: core::option::Option<u64>
 pub query_planning::chunk::SourceChunk::estimated_source_read_bytes: core::option::Option<u64>
 pub query_planning::chunk::SourceChunk::exact_source_read_preflight_bytes: core::option::Option<u64>
@@ -393,6 +393,7 @@ pub query_planning::chunk::SourceChunk::known_pure_filter_uncompressed_bytes: co
 pub query_planning::chunk::SourceChunk::known_row_count: core::option::Option<u64>
 pub query_planning::chunk::SourceChunk::known_row_group_bytes: core::option::Option<u64>
 pub query_planning::chunk::SourceChunk::location: query_planning::source::ParquetSourceLocation
+pub query_planning::chunk::SourceChunk::max_column_elements: u64
 pub query_planning::chunk::SourceChunk::partition_values: alloc::vec::Vec<query_planning::expr::LiteralValue>
 pub query_planning::chunk::SourceChunk::path: std::path::PathBuf
 pub query_planning::chunk::SourceChunk::read_columns: alloc::vec::Vec<alloc::string::String>
@@ -411,6 +412,7 @@ pub query_planning::chunk::SourceChunk::source_file_projected_uncompressed_bytes
 pub query_planning::chunk::SourceChunk::source_file_pure_filter_compressed_bytes: core::option::Option<alloc::vec::Vec<u64>>
 pub query_planning::chunk::SourceChunk::source_file_pure_filter_uncompressed_bytes: core::option::Option<alloc::vec::Vec<u64>>
 pub query_planning::chunk::SourceChunk::source_file_row_group_bytes: core::option::Option<alloc::vec::Vec<u64>>
+pub query_planning::chunk::SourceChunk::source_file_row_group_column_elements: core::option::Option<alloc::vec::Vec<u64>>
 pub query_planning::chunk::SourceChunk::source_file_row_group_decoded_bytes: core::option::Option<alloc::vec::Vec<u64>>
 pub query_planning::chunk::SourceChunk::source_file_row_group_rows: core::option::Option<alloc::vec::Vec<u64>>
 pub query_planning::chunk::SourceChunk::source_file_row_groups: core::option::Option<alloc::vec::Vec<usize>>

components/cudf/algeon-cudf-sys/native/include/cudf_rust/column.h

@@ -74,6 +74,32 @@ cudf_status_t cudf_column_create_empty_ex(cudf_type_id_t type_id,
                                           cudf_exec_options_t const* options,
                                           cudf_column_handle* out);
 
+/// One node of a column type tree laid out in depth-first pre-order. The
+/// `num_children` nodes after a parent describe its children, each followed
+/// by its own subtree. `scale` is read for DECIMAL types only.
+typedef struct cudf_column_type_node_t {
+  cudf_type_id_t type_id;
+  int32_t scale;
+  int32_t num_children;
+} cudf_column_type_node_t;
+
+/// Create a zero-row column whose type, including every nested child type,
+/// is the pre-order tree `nodes[0..num_nodes)`.
+///
+/// A LIST node takes exactly one child, its element type. A STRUCT node takes
+/// one or more children, its field types. Every other node takes none.
+/// EMPTY and DICTIONARY32 are rejected with `CUDF_ERROR_TYPE`; a malformed
+/// tree, including nodes left over after the root subtree, is rejected with
+/// `CUDF_ERROR_INVALID_ARGUMENT`. No device memory is allocated.
+///
+/// `options` must be non-NULL and must name an explicit valid device, a
+/// non-NULL stream, and a non-NULL memory resource. The returned column
+/// records that execution provenance.
+cudf_status_t cudf_column_create_empty_from_type_tree_ex(cudf_column_type_node_t const* nodes,
+                                                         int32_t num_nodes,
+                                                         cudf_exec_options_t const* options,
+                                                         cudf_column_handle* out);
+
 /// Create a numeric column from host data with explicit execution options.
 ///
 /// `options` must be non-NULL and must name an explicit valid device, a

components/cudf/algeon-cudf-sys/native/src/abi_checks.cpp

@@ -4,6 +4,7 @@
 
 #include "cudf_rust/ast.h"
 #include "cudf_rust/binary.h"
+#include "cudf_rust/column.h"
 #include "cudf_rust/copying.h"
 #include "cudf_rust/datetime.h"
 #include "cudf_rust/device.h"
@@ -202,6 +203,12 @@ static_assert(CUDF_AST_IS_NULL == static_cast<int>(cudf::ast::ast_operator::IS_N
 static_assert(CUDF_AST_NOT == static_cast<int>(cudf::ast::ast_operator::NOT));
 static_assert(CUDF_AST_CAST_TO_FLOAT64 == static_cast<int>(cudf::ast::ast_operator::CAST_TO_FLOAT64));
 
+static_assert(std::is_standard_layout_v<cudf_column_type_node_t>, "cudf_column_type_node_t must be standard layout");
+static_assert(offsetof(cudf_column_type_node_t, type_id) == 0, "cudf_column_type_node_t::type_id must be the first field");
+static_assert(offsetof(cudf_column_type_node_t, scale) == sizeof(cudf_type_id_t), "cudf_column_type_node_t::scale must follow type_id");
+static_assert(offsetof(cudf_column_type_node_t, num_children) == offsetof(cudf_column_type_node_t, scale) + sizeof(int32_t), "cudf_column_type_node_t::num_children must follow scale");
+static_assert(sizeof(cudf_column_type_node_t) == 3 * sizeof(int32_t), "cudf_column_type_node_t must be three 32-bit fields");
+
 static_assert(std::is_standard_layout_v<cudf_row_predicate_node_t>, "cudf_row_predicate_node_t must be standard layout");
 static_assert(offsetof(cudf_row_predicate_node_t, kind) == 0, "cudf_row_predicate_node_t::kind must be the first field");
 static_assert(sizeof(static_cast<cudf_row_predicate_node_t*>(nullptr)->lhs_index) == sizeof(int32_t), "cudf_row_predicate_node_t::lhs_index must be 32-bit");

components/cudf/algeon-cudf-sys/native/src/c_header_smoke.c

@@ -72,6 +72,7 @@ static int cudf_rust_c_header_smoke(void) {
   SMOKE_CHECK_FN(cudf_status_t, cudf_column_adopt_device_buffer_nullable, cudf_type_id_t, cudf_device_buffer_handle, cudf_device_buffer_handle, int32_t, int32_t, cudf_column_handle*);
 
   // Column and table creation, copying, and inspection
+  SMOKE_CHECK_FN(cudf_status_t, cudf_column_create_empty_from_type_tree_ex, cudf_column_type_node_t const*, int32_t, cudf_exec_options_t const*, cudf_column_handle*);
   SMOKE_CHECK_FN(cudf_status_t, cudf_column_create_numeric_ex, cudf_type_id_t, void const*, int32_t, uint8_t const*, int32_t, cudf_exec_options_t const*, cudf_column_handle*);
   SMOKE_CHECK_FN(cudf_status_t, cudf_column_create_fixed_width_from_host_ex, cudf_type_id_t, int32_t, void const*, int32_t, uint8_t const*, int32_t, cudf_exec_options_t const*, cudf_column_handle*);
   SMOKE_CHECK_FN(cudf_status_t, cudf_column_create_sequence_ex, cudf_scalar_handle, cudf_scalar_handle, int32_t, cudf_exec_options_t const*, cudf_column_handle*);

components/cudf/algeon-cudf-sys/native/src/column.cpp

@@ -77,6 +77,47 @@ cudf_status_t resolve_column_view_null_count(cudf::column_view const& view,
   return CUDF_SUCCESS;
 }
 
+// Mirrors cudf::empty_like for nested columns: a zero-row LIST holds an empty
+// INT32 offsets child before its element child, and no level owns a buffer.
+std::unique_ptr<cudf::column> make_empty_column_from_type_tree(cudf_column_type_node_t const* nodes,
+                                                               int32_t num_nodes,
+                                                               int32_t& cursor)
+{
+  CUDF_EXPECTS(cursor < num_nodes,
+               "type tree ends before every declared child is described",
+               std::invalid_argument);
+  auto const& node = nodes[cursor++];
+  CUDF_EXPECTS(node.num_children >= 0, "type tree num_children must be >= 0", std::invalid_argument);
+  CUDF_EXPECTS(node.type_id > CUDF_TYPE_EMPTY && node.type_id <= CUDF_TYPE_STRUCT &&
+                 node.type_id != CUDF_TYPE_DICTIONARY32,
+               "type tree node has no empty column construction",
+               cudf::data_type_error);
+
+  std::vector<std::unique_ptr<cudf::column>> children;
+  if (node.type_id == CUDF_TYPE_LIST) {
+    CUDF_EXPECTS(
+      node.num_children == 1, "type tree LIST node requires exactly one child", std::invalid_argument);
+    children.push_back(cudf::make_empty_column(cudf::data_type{cudf::type_id::INT32}));
+    children.push_back(make_empty_column_from_type_tree(nodes, num_nodes, cursor));
+  } else if (node.type_id == CUDF_TYPE_STRUCT) {
+    CUDF_EXPECTS(
+      node.num_children > 0, "type tree STRUCT node requires at least one child", std::invalid_argument);
+    children.reserve(static_cast<size_t>(node.num_children));
+    for (int32_t child = 0; child < node.num_children; ++child) {
+      children.push_back(make_empty_column_from_type_tree(nodes, num_nodes, cursor));
+    }
+  } else {
+    CUDF_EXPECTS(node.num_children == 0, "type tree leaf node must have no children", std::invalid_argument);
+    return cudf::make_empty_column(host_fixed_width_dtype(node.type_id, node.scale));
+  }
+  return std::make_unique<cudf::column>(cudf::data_type{to_cudf_type_id(node.type_id)},
+                                        0,
+                                        rmm::device_buffer{},
+                                        rmm::device_buffer{},
+                                        0,
+                                        std::move(children));
+}
+
 }  // namespace
 
 static cudf_status_t create_column_from_device_data_impl(cudf_type_id_t type_id,
@@ -149,6 +190,35 @@ cudf_status_t cudf_column_create_empty_ex(cudf_type_id_t type_id,
   CUDF_RUST_CATCH
 }
 
+cudf_status_t cudf_column_create_empty_from_type_tree_ex(cudf_column_type_node_t const* nodes,
+                                                         int32_t num_nodes,
+                                                         cudf_exec_options_t const* options,
+                                                         cudf_column_handle* out)
+{
+  CUDF_RUST_NULL_CHECK(out, "out is null");
+  *out = nullptr;
+  CUDF_RUST_NULL_CHECK(nodes, "nodes is null");
+  if (num_nodes <= 0) {
+    set_error(CUDF_ERROR_INVALID_ARGUMENT, "num_nodes must be > 0");
+    return CUDF_ERROR_INVALID_ARGUMENT;
+  }
+  CUDF_RUST_TRY
+  {
+    auto exec                       = std::unique_ptr<cudf_exec_context>{};
+    cudf_status_t const exec_status = cudf_resolve_required_exec_options(options, &exec);
+    if (exec_status != CUDF_SUCCESS) { return exec_status; }
+
+    int32_t cursor = 0;
+    auto col       = make_empty_column_from_type_tree(nodes, num_nodes, cursor);
+    CUDF_EXPECTS(cursor == num_nodes,
+                 "type tree has nodes after the root subtree",
+                 std::invalid_argument);
+    *out = cudf_release_column_handle(std::move(col), cudf_capture_exec_provenance(*exec));
+    return CUDF_SUCCESS;
+  }
+  CUDF_RUST_CATCH
+}
+
 cudf_status_t cudf_column_create_numeric_ex(cudf_type_id_t type_id,
                                             void const* data,
                                             int32_t size,

components/cudf/algeon-cudf-sys/src/bindings.rs

@@ -274,6 +274,14 @@ pub struct cudf_interop_semantic_column_t {
     pub num_child_column_ids: i32,
     pub keys_column_id: u32,
 }
+#[doc = " One node of a column type tree laid out in depth-first pre-order. The\n `num_children` nodes after a parent describe its children, each followed\n by its own subtree. `scale` is read for DECIMAL types only."]
+#[repr(C)]
+#[derive(Debug, Copy, Clone)]
+pub struct cudf_column_type_node_t {
+    pub type_id: cudf_type_id_t,
+    pub scale: i32,
+    pub num_children: i32,
+}
 #[repr(C)]
 #[derive(Debug)]
 pub struct cudf_ast_expression_handle_state {
@@ -1041,6 +1049,13 @@ unsafe extern "C" {
         options: *const cudf_exec_options_t,
         out: *mut cudf_column_handle,
     ) -> cudf_status_t;
+    #[doc = " Create a zero-row column whose type, including every nested child type,\n is the pre-order tree `nodes[0..num_nodes)`.\n\n A LIST node takes exactly one child, its element type. A STRUCT node takes\n one or more children, its field types. Every other node takes none.\n EMPTY and DICTIONARY32 are rejected with `CUDF_ERROR_TYPE`; a malformed\n tree, including nodes left over after the root subtree, is rejected with\n `CUDF_ERROR_INVALID_ARGUMENT`. No device memory is allocated.\n\n `options` must be non-NULL and must name an explicit valid device, a\n non-NULL stream, and a non-NULL memory resource. The returned column\n records that execution provenance."]
+    pub fn cudf_column_create_empty_from_type_tree_ex(
+        nodes: *const cudf_column_type_node_t,
+        num_nodes: i32,
+        options: *const cudf_exec_options_t,
+        out: *mut cudf_column_handle,
+    ) -> cudf_status_t;
     #[doc = " Create a numeric column from host data with explicit execution options.\n\n `options` must be non-NULL and must name an explicit valid device, a\n non-NULL stream, and a non-NULL memory resource. `options->mr` must outlive\n the returned column.\n\n If host-source copies are queued, the selected stream is synchronized\n before return on success or failure so the borrowed host buffers are no\n longer in use."]
     pub fn cudf_column_create_numeric_ex(
         type_id: cudf_type_id_t,