Sketch out gpu physical execution

Cargo.toml

@@ -32,6 +32,7 @@ members = [
     "datafusion/expr-common",
     "datafusion/execution",
     "datafusion/ffi",
+    "datafusion/gpu",
     "datafusion/functions",
     "datafusion/functions-aggregate",
     "datafusion/functions-aggregate-common",
@@ -143,6 +144,7 @@ datafusion-functions-nested = { path = "datafusion/functions-nested", version =
 datafusion-functions-table = { path = "datafusion/functions-table", version = "53.1.0" }
 datafusion-functions-window = { path = "datafusion/functions-window", version = "53.1.0" }
 datafusion-functions-window-common = { path = "datafusion/functions-window-common", version = "53.1.0" }
+datafusion-gpu = { path = "datafusion/gpu", version = "53.1.0" }
 datafusion-macros = { path = "datafusion/macros", version = "53.1.0" }
 datafusion-optimizer = { path = "datafusion/optimizer", version = "53.1.0", default-features = false }
 datafusion-physical-expr = { path = "datafusion/physical-expr", version = "53.1.0", default-features = false }

datafusion/gpu/Cargo.toml

@@ -0,0 +1,47 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[package]
+name = "datafusion-gpu"
+description = "DataFusion GPU physical operators (experimental)"
+keywords = ["datafusion", "gpu", "cuda"]
+version = { workspace = true }
+edition = { workspace = true }
+homepage = { workspace = true }
+repository = { workspace = true }
+license = { workspace = true }
+authors = { workspace = true }
+rust-version = { workspace = true }
+publish = false
+
+# Note: add additional linter rules in lib.rs.
+# Rust does not support workspace + new linter rules in subcrates yet
+# https://github.com/rust-lang/cargo/issues/13157
+[lints]
+workspace = true
+
+[dependencies]
+arrow = { workspace = true }
+datafusion-common = { workspace = true }
+datafusion-execution = { workspace = true }
+datafusion-physical-expr = { workspace = true }
+datafusion-physical-optimizer = { workspace = true }
+datafusion-physical-plan = { workspace = true }
+futures = { workspace = true }
+
+[dev-dependencies]
+tokio = { workspace = true }

datafusion/gpu/src/batch.rs

@@ -0,0 +1,121 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Device-resident column buffers and record batches.
+
+use std::sync::Arc;
+
+use arrow::datatypes::SchemaRef;
+use datafusion_common::{Result, internal_err};
+
+use crate::memory_pool::GpuMemoryReservation;
+
+/// A column buffer resident in GPU device memory.
+///
+/// In this scaffold the underlying storage is a host-side placeholder; the
+/// real implementation will replace [`Backing::HostPlaceholder`] with a CUDA
+/// device allocation (e.g. `cudarc::driver::CudaSlice`). Operators must
+/// treat the bytes as opaque — they are *not* a valid host pointer for kernel
+/// inputs.
+#[derive(Debug)]
+pub struct GpuBuffer {
+    byte_len: usize,
+    #[expect(dead_code, reason = "read once the CUDA backend lands")]
+    backing: Backing,
+}
+
+#[derive(Debug)]
+enum Backing {
+    /// Host-resident placeholder. Held only so the scaffold compiles and
+    /// round-trips data through `GpuUploadExec`/`GpuDownloadExec` once those
+    /// land; never to be dereferenced by GPU code.
+    HostPlaceholder(
+        #[expect(dead_code, reason = "read once the CUDA backend lands")] Vec<u8>,
+    ),
+}
+
+impl GpuBuffer {
+    pub fn from_host_placeholder(bytes: Vec<u8>) -> Self {
+        Self {
+            byte_len: bytes.len(),
+            backing: Backing::HostPlaceholder(bytes),
+        }
+    }
+
+    pub fn byte_len(&self) -> usize {
+        self.byte_len
+    }
+}
+
+/// Arrow-shaped batch whose column buffers live in device memory.
+///
+/// This is the unit of data exchanged between operators implementing
+/// [`crate::GpuExecutionPlan`]. It deliberately does *not* implement any
+/// conversion to `arrow::record_batch::RecordBatch`; crossing the host/device
+/// boundary must go through [`crate::boundary::GpuUploadExec`] /
+/// [`crate::boundary::GpuDownloadExec`].
+#[derive(Debug)]
+pub struct GpuRecordBatch {
+    schema: SchemaRef,
+    columns: Vec<Arc<GpuBuffer>>,
+    num_rows: usize,
+    // Dropped with the batch, releasing the device-memory reservation.
+    _reservation: GpuMemoryReservation,
+}
+
+impl GpuRecordBatch {
+    pub fn try_new(
+        schema: SchemaRef,
+        columns: Vec<Arc<GpuBuffer>>,
+        num_rows: usize,
+        reservation: GpuMemoryReservation,
+    ) -> Result<Self> {
+        if schema.fields().len() != columns.len() {
+            return internal_err!(
+                "GpuRecordBatch: schema has {} fields but {} columns supplied",
+                schema.fields().len(),
+                columns.len()
+            );
+        }
+        Ok(Self {
+            schema,
+            columns,
+            num_rows,
+            _reservation: reservation,
+        })
+    }
+
+    pub fn schema(&self) -> SchemaRef {
+        Arc::clone(&self.schema)
+    }
+
+    pub fn num_rows(&self) -> usize {
+        self.num_rows
+    }
+
+    pub fn num_columns(&self) -> usize {
+        self.columns.len()
+    }
+
+    pub fn column(&self, idx: usize) -> &Arc<GpuBuffer> {
+        &self.columns[idx]
+    }
+
+    pub fn columns(&self) -> &[Arc<GpuBuffer>] {
+        &self.columns
+    }
+}