This is an automated email from the ASF dual-hosted git repository. timsaucer pushed a commit to branch feat/distributed-extensions-example in repository https://gitbox.apache.org/repos/asf/datafusion-python.git
commit 29502bf39e1db648952ad7f981f2cf49432b4c64 Author: Tim Saucer <[email protected]> AuthorDate: Wed Sep 9 12:27:42 2026 -0400 Add dfx_storage: a codec that writes durable metadata First of three libraries for the multi-library distributed example in #1719, and the reference implementation the extension guide's `extension_codec_durable_metadata` section currently lacks. Every other example codec in this repository parks the live object in a process-global `HashMap` and encodes an integer token into it. The guide says plainly that this is a demonstration and not a pattern, then has nothing to point at that does it properly. This codec is that: it writes the file paths and sizes, the projection, the row limit, and the schema, so decoding needs nothing at all from the encoding process. The provider scans a directory of Parquet files and reports one output partition per file. That is the reason it exists rather than `register_parquet`: it fixes the mapping from partition index to file, so an engine can hand partition `i` to a worker and know which bytes that worker will read. Paths are sorted, because directory iteration order is unspecified and a worker that disagreed with the driver about which file is partition 3 would produce wrong answers silently rather than fail. `PartitionedParquetExec` is a leaf on purpose. A node with children hands them to the framework to encode with the host's codec, which is correct but means the interesting part of a codec — what it writes down — belongs to someone else. Everything this node needs to run is in the node, so that is what goes on the wire. It reuses `DataSourceExec` to do the actual reading; the point is to own the description of the scan across a process boundary, not to reimplement Parquet. Wire format is `DFXSTOR1 | json_len: u32 | json | arrow ipc schema`. JSON for the scalar fields because someone debugging a worker can read it, Arrow IPC for the schema because it is the only encoding that round-trips every Arrow type. The magic carries a version the codec refuses to guess at. The codec claims by downcasting to its own concrete type and hands anything else to the default codec, whose error is the chain's "not mine" signal. Tests pin one fact that makes the narrow claim obviously right: an extension codec is only ever consulted for nodes with no native encoding, so the only nodes that reach it are ones some library owns — a broad claim can only steal from a peer, never pick up slack. Ten tests, the load-bearing one being a genuinely separate interpreter spawned through `sys.executable` that builds its own session, checks the codec id it expects is installed, decodes a plan written by another process, and executes all three partitions. A token registry cannot pass that test, which is the point of writing it first. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]> --- .gitignore | 1 + Cargo.lock | 18 ++ Cargo.toml | 1 + examples/distributed/storage-library/Cargo.toml | 51 +++++ examples/distributed/storage-library/build.rs | 20 ++ .../distributed/storage-library/pyproject.toml | 32 +++ .../python/tests/_test_portable_codec.py | 247 +++++++++++++++++++++ .../storage-library/python/tests/conftest.py | 69 ++++++ examples/distributed/storage-library/src/codec.rs | 226 +++++++++++++++++++ examples/distributed/storage-library/src/exec.rs | 181 +++++++++++++++ .../distributed/storage-library/src/extension.rs | 156 +++++++++++++ examples/distributed/storage-library/src/lib.rs | 40 ++++ .../storage-library/src/table_provider.rs | 182 +++++++++++++++ 13 files changed, 1224 insertions(+) diff --git a/.gitignore b/.gitignore index 614d8232..ef00c0fd 100644 --- a/.gitignore +++ b/.gitignore @@ -36,3 +36,4 @@ docs/mdbook/book .pyo3_build_config +examples/distributed/*/.venv/ diff --git a/Cargo.lock b/Cargo.lock index 6a7f6843..13ac8b64 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1666,6 +1666,24 @@ dependencies = [ "url", ] +[[package]] +name = "dfx-storage" +version = "54.0.0" +dependencies = [ + "arrow", + "async-trait", + "datafusion", + "datafusion-catalog", + "datafusion-common", + "datafusion-ffi", + "datafusion-proto", + "datafusion-python-util", + "pyo3", + "pyo3-build-config", + "pyo3-log", + "serde_json", +] + [[package]] name = "digest" version = "0.10.7" diff --git a/Cargo.toml b/Cargo.toml index 0fabd543..929c3bba 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -32,6 +32,7 @@ members = [ "crates/util", "examples/datafusion-ffi-example", "examples/datafusion-ffi-query-planner-example", + "examples/distributed/storage-library", ] resolver = "3" diff --git a/examples/distributed/storage-library/Cargo.toml b/examples/distributed/storage-library/Cargo.toml new file mode 100644 index 00000000..0e01e052 --- /dev/null +++ b/examples/distributed/storage-library/Cargo.toml @@ -0,0 +1,51 @@ +# 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 = "dfx-storage" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +description = "Example extension library: a partitioned Parquet table provider whose codec encodes durable metadata" +homepage.workspace = true +repository.workspace = true +publish = false + +[dependencies] +arrow = { workspace = true } +async-trait = { workspace = true } +datafusion = { workspace = true } +datafusion-catalog = { workspace = true, default-features = false } +datafusion-common = { workspace = true, default-features = false } +datafusion-ffi = { workspace = true } +datafusion-proto = { workspace = true } +datafusion-python-util.workspace = true +pyo3 = { workspace = true, features = [ + "extension-module", + "abi3", + "abi3-py310", +] } +pyo3-log = { workspace = true } +serde_json = { workspace = true } + +[build-dependencies] +pyo3-build-config = { workspace = true } + +[lib] +name = "dfx_storage" +crate-type = ["cdylib", "rlib"] diff --git a/examples/distributed/storage-library/build.rs b/examples/distributed/storage-library/build.rs new file mode 100644 index 00000000..4878d8b0 --- /dev/null +++ b/examples/distributed/storage-library/build.rs @@ -0,0 +1,20 @@ +// 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. + +fn main() { + pyo3_build_config::add_extension_module_link_args(); +} diff --git a/examples/distributed/storage-library/pyproject.toml b/examples/distributed/storage-library/pyproject.toml new file mode 100644 index 00000000..24bc294f --- /dev/null +++ b/examples/distributed/storage-library/pyproject.toml @@ -0,0 +1,32 @@ +# 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. + +[build-system] +requires = ["maturin>=1.6,<2.0"] +build-backend = "maturin" + +[project] +name = "dfx_storage" +requires-python = ">=3.10" +classifiers = [ + "Programming Language :: Rust", + "Programming Language :: Python :: Implementation :: CPython", +] +dynamic = ["version"] + +[tool.maturin] +features = ["pyo3/extension-module"] diff --git a/examples/distributed/storage-library/python/tests/_test_portable_codec.py b/examples/distributed/storage-library/python/tests/_test_portable_codec.py new file mode 100644 index 00000000..716d92fd --- /dev/null +++ b/examples/distributed/storage-library/python/tests/_test_portable_codec.py @@ -0,0 +1,247 @@ +# 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. + +"""The claim this library exists to make: its plans decode in another process. + +Every other example codec in this repository parks the live object in a +process-global map and encodes a token. These tests are written to fail if +this one ever does that -- the decoding side is a separate interpreter, so a +token would have nothing to look up. +""" + +from __future__ import annotations + +import itertools +import re +import subprocess +import sys +import textwrap +from typing import TYPE_CHECKING + +import pytest +from datafusion import SessionContext +from datafusion.plan import ExecutionPlan +from dfx_storage import DfxStorageExtension, PartitionedParquetTable + +if TYPE_CHECKING: + import pathlib + + +def _configured(directory: pathlib.Path) -> tuple[SessionContext, DfxStorageExtension]: + bundle = DfxStorageExtension() + ctx = SessionContext().with_extensions(bundle) + ctx.register_table("readings", PartitionedParquetTable(str(directory))) + return ctx, bundle + + +def test_provider_reports_one_partition_per_file(readings_dir: pathlib.Path) -> None: + """The file is the partition, which is the axis an engine splits along.""" + ctx, _ = _configured(readings_dir) + plan = ctx.sql("select sensor_id from readings").execution_plan() + + assert plan.partition_count == 3 + # Not `Hash`: rows are grouped by which file they landed in, which says + # nothing about their values. + assert plan.output_partitioning.scheme == "UnknownPartitioning" + assert plan.output_partitioning.hash_expressions is None + + +def test_each_partition_reads_exactly_one_file(readings_dir: pathlib.Path) -> None: + """Partition i reads file i, so two workers never read the same bytes.""" + ctx, _ = _configured(readings_dir) + plan = ctx.sql("select sensor_id from readings").execution_plan() + + per_partition = [ + sorted( + value + for batch in ctx.execute(plan, partition) + for value in batch.to_pyarrow().column("sensor_id").to_pylist() + ) + for partition in range(plan.partition_count) + ] + + assert per_partition == [[0, 1, 2], [100, 101, 102], [200, 201, 202]] + # Disjoint, and together the whole table. + everything = sorted(itertools.chain.from_iterable(per_partition)) + assert everything == [0, 1, 2, 100, 101, 102, 200, 201, 202] + + +def test_this_librarys_codec_carried_the_node(readings_dir: pathlib.Path) -> None: + """Assert *this* codec did the work, not merely that the query succeeded. + + Both codecs being installed does not mean this one saw the node; a codec + installed earlier that claims broadly would have taken it. + """ + ctx, bundle = _configured(readings_dir) + plan = ctx.sql("select sensor_id from readings").execution_plan() + assert bundle.encode_calls() == 0 + + blob = plan.to_bytes(ctx) + assert bundle.encode_calls() == 1 + + restored = ExecutionPlan.from_bytes(ctx, blob) + assert bundle.decode_calls() == 1 + assert "PartitionedParquetExec" in restored.display_indent() + + +def test_the_payload_is_metadata_not_a_token(readings_dir: pathlib.Path) -> None: + """The bytes name the files, so they mean something in another process.""" + ctx, _ = _configured(readings_dir) + blob = ctx.sql("select sensor_id from readings").execution_plan().to_bytes(ctx) + + assert b"DFXSTOR1" in blob + for index in range(3): + assert f"part-{index}.parquet".encode() in blob + + +def test_the_same_bytes_decode_twice(readings_dir: pathlib.Path) -> None: + """A token registry consumes its entry on decode. Durable metadata does not. + + This is what lets one encoded plan fan out to several workers. + """ + ctx, bundle = _configured(readings_dir) + blob = ctx.sql("select sensor_id from readings").execution_plan().to_bytes(ctx) + + first = ExecutionPlan.from_bytes(ctx, blob) + second = ExecutionPlan.from_bytes(ctx, blob) + + assert bundle.decode_calls() == 2 + assert first.partition_count == second.partition_count == 3 + + +def test_a_projection_survives_the_round_trip(readings_dir: pathlib.Path) -> None: + """The projection is part of the descriptor, not re-derived on decode.""" + ctx, _ = _configured(readings_dir) + plan = ctx.sql("select reading from readings").execution_plan() + restored = ExecutionPlan.from_bytes(ctx, plan.to_bytes(ctx)) + + rows = [ + value + for partition in range(restored.partition_count) + for batch in ctx.execute(restored, partition) + for value in batch.to_pyarrow().column("reading").to_pylist() + ] + assert sorted(rows) == [1.5, 1.5, 1.5, 2.5, 2.5, 2.5, 3.5, 3.5, 3.5] + + +def test_stock_nodes_never_reach_this_codec(readings_dir: pathlib.Path) -> None: + """An extension codec is only consulted for nodes with no native encoding. + + The aggregate and filter above the scan all have their own `try_to_proto`, + so the framework encodes them itself and this codec is never offered them. + That is why claiming a broad category is so damaging: the only nodes that + ever arrive here are ones *some* library owns, so a broad claim can only + ever steal from a peer, never pick up slack. + """ + ctx, bundle = _configured(readings_dir) + plan = ctx.sql("select count(*) from readings where reading > 2.0").execution_plan() + + plan.to_bytes(ctx) + + # Exactly one node in that plan is ours, and nothing else was offered. + assert bundle.encode_calls() == 1 + assert bundle.declined_calls() == 0 + + +WORKER = textwrap.dedent( + """ + import sys + from datafusion import SessionContext + from datafusion.plan import ExecutionPlan + from dfx_storage import DfxStorageExtension + + blob_path, expected_id = sys.argv[1], sys.argv[2] + + # A session built from scratch: this process has never registered the + # table, and shares nothing with the one that wrote the plan. + bundle = DfxStorageExtension() + ctx = SessionContext().with_extensions(bundle) + + installed = ctx.physical_extension_codec_ids() + assert expected_id in installed, f"codec id {expected_id} not in {installed}" + + with open(blob_path, "rb") as handle: + plan = ExecutionPlan.from_bytes(ctx, handle.read()) + + total = 0 + for partition in range(plan.partition_count): + for batch in ctx.execute(plan, partition): + total += batch.to_pyarrow().num_rows + print(f"partitions={plan.partition_count} rows={total} decoded={bundle.decode_calls()}") + """ +) + + +def test_a_separate_process_decodes_and_executes_the_plan( + readings_dir: pathlib.Path, tmp_path: pathlib.Path +) -> None: + """The whole point. No shared session, no shared registry, no token. + + Spawned through `sys.executable` rather than `multiprocessing`: the tokio + runtime backing this extension is a process-global, so `fork` is unsafe, + and a hardcoded `python` could differ in minor version from this one. + """ + ctx, _ = _configured(readings_dir) + blob = ctx.sql("select sensor_id from readings").execution_plan().to_bytes(ctx) + blob_path = tmp_path / "plan.bin" + blob_path.write_bytes(blob) + + worker = tmp_path / "worker.py" + worker.write_text(WORKER) + + result = subprocess.run( # noqa: S603 + [sys.executable, str(worker), str(blob_path), "dfx_storage.physical.v1"], + capture_output=True, + text=True, + check=False, + ) + + assert result.returncode == 0, result.stderr + assert "partitions=3 rows=9 decoded=1" in result.stdout + + +def test_a_worker_without_the_codec_says_which_one_is_missing( + readings_dir: pathlib.Path, +) -> None: + """The failure names the codec, which is the whole value of pinned ids.""" + ctx, _ = _configured(readings_dir) + blob = ctx.sql("select sensor_id from readings").execution_plan().to_bytes(ctx) + + # A session with no extension codecs at all, standing in for a worker + # whose bootstrap forgot to install the bundle. + bare = SessionContext() + with pytest.raises( + Exception, match=re.escape("dfx_storage.physical.v1") + ) as excinfo: + ExecutionPlan.from_bytes(bare, blob) + assert "not installed on this session" in str(excinfo.value) + + +def test_the_bundle_is_reusable_across_sessions(readings_dir: pathlib.Path) -> None: + """One bundle object, two sessions: components are built per install.""" + bundle = DfxStorageExtension() + first = SessionContext().with_extensions(bundle) + second = SessionContext().with_extensions(bundle) + + for ctx in (first, second): + ctx.register_table("readings", PartitionedParquetTable(str(readings_dir))) + assert ( + ctx.sql("select count(*) from readings").collect()[0].column(0)[0].as_py() + == 9 + ) + + assert first.__datafusion_codec_id__ != second.__datafusion_codec_id__ diff --git a/examples/distributed/storage-library/python/tests/conftest.py b/examples/distributed/storage-library/python/tests/conftest.py new file mode 100644 index 00000000..e37db8f8 --- /dev/null +++ b/examples/distributed/storage-library/python/tests/conftest.py @@ -0,0 +1,69 @@ +# 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. + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING + +import pyarrow as pa +import pyarrow.parquet as pq +import pytest + +if TYPE_CHECKING: + import pathlib + from collections.abc import Generator + from typing import Any + + +class _FailOnWarning(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + if record.levelno >= logging.WARNING: + err = f"Unexpected log warning from '{record.name}': {self.format(record)}" + raise AssertionError(err) + + [email protected](autouse=True) +def fail_on_log_warnings() -> Generator[None, Any, None]: + handler = _FailOnWarning() + logging.root.addHandler(handler) + yield + logging.root.removeHandler(handler) + + [email protected] +def readings_dir(tmp_path: pathlib.Path) -> pathlib.Path: + """Three Parquet files, so the provider reports three partitions. + + Written as `part-0/1/2` rather than in one file because the file *is* the + partition for this provider, and a single-file table would hide every + partition-routing mistake. + """ + directory = tmp_path / "readings" + directory.mkdir() + for index in range(3): + base = index * 100 + pq.write_table( + pa.table( + { + "sensor_id": [base, base + 1, base + 2], + "reading": [1.5, 2.5, 3.5], + } + ), + directory / f"part-{index}.parquet", + ) + return directory diff --git a/examples/distributed/storage-library/src/codec.rs b/examples/distributed/storage-library/src/codec.rs new file mode 100644 index 00000000..9868e001 --- /dev/null +++ b/examples/distributed/storage-library/src/codec.rs @@ -0,0 +1,226 @@ +// 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. + +//! A physical codec that writes durable metadata. +//! +//! The other example crates in this repository park the live object in a +//! process-global `HashMap` and encode an integer token into it. That makes +//! Rust type identity observable in a test, and it is explicitly not a +//! pattern: the token means the same bytes cannot be decoded twice, one plan +//! cannot fan out to several readers, and a plan that never reaches a decoder +//! leaks. None of that is acceptable for a plan that leaves the process. +//! +//! This codec writes down what a fresh [`PartitionedParquetExec`] can be built +//! from -- the file paths and sizes, the projection, the row limit, and the +//! schema -- so decoding needs nothing from the encoding process. Sending the +//! same bytes to ten workers works, and so does sending them tomorrow. +//! +//! # Wire format +//! +//! ```text +//! DFXSTOR1 | json_len: u32 (LE) | json | arrow ipc schema +//! ``` +//! +//! The magic is checked before anything else is read, and the trailing `1` is +//! a version this codec refuses to guess at. JSON carries the small scalar +//! fields because a human debugging a worker can read it; the schema is Arrow +//! IPC because that is the only encoding guaranteed to round-trip every Arrow +//! type, including extension types and field metadata. + +use std::fmt; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; + +use arrow::datatypes::Schema; +use arrow::ipc::reader::StreamReader; +use arrow::ipc::writer::StreamWriter; +use datafusion::common::{Result, internal_datafusion_err, internal_err}; +use datafusion::execution::TaskContext; +use datafusion::physical_plan::ExecutionPlan; +use datafusion_proto::physical_plan::{ + DefaultPhysicalExtensionCodec, PhysicalExtensionCodec, PhysicalProtoConverterExtension, +}; + +use crate::exec::{FileSlice, PartitionedParquetExec}; + +/// Framing magic. The trailing digit is the payload version. +const MAGIC: &[u8; 8] = b"DFXSTOR1"; + +/// How often this codec claimed one of its own nodes. +/// +/// Exposed to Python so a test can assert that *this* codec carried the node, +/// rather than inferring it from a query that merely succeeded. Both codecs +/// being installed does not mean yours saw the node -- see +/// `extension_codec_order`. +#[derive(Default, Debug)] +pub(crate) struct CodecCounters { + pub(crate) encoded: AtomicUsize, + pub(crate) decoded: AtomicUsize, + pub(crate) declined: AtomicUsize, +} + +pub(crate) struct DfxStoragePhysicalCodec { + /// Anything this library does not own is handed to the default codec, + /// whose error is the chain's "not mine" signal. + inner: DefaultPhysicalExtensionCodec, + pub(crate) counters: Arc<CodecCounters>, +} + +impl DfxStoragePhysicalCodec { + pub(crate) fn new(counters: Arc<CodecCounters>) -> Self { + Self { + inner: DefaultPhysicalExtensionCodec {}, + counters, + } + } +} + +impl fmt::Debug for DfxStoragePhysicalCodec { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("DfxStoragePhysicalCodec") + .finish_non_exhaustive() + } +} + +fn schema_to_ipc_bytes(schema: &Schema) -> Result<Vec<u8>> { + let mut buf: Vec<u8> = Vec::new(); + { + let mut writer = StreamWriter::try_new(&mut buf, schema) + .map_err(|err| internal_datafusion_err!("dfx_storage: writing schema: {err}"))?; + writer + .finish() + .map_err(|err| internal_datafusion_err!("dfx_storage: writing schema: {err}"))?; + } + Ok(buf) +} + +fn schema_from_ipc_bytes(bytes: &[u8]) -> Result<Schema> { + let reader = StreamReader::try_new(std::io::Cursor::new(bytes), None) + .map_err(|err| internal_datafusion_err!("dfx_storage: reading schema: {err}"))?; + Ok(reader.schema().as_ref().clone()) +} + +impl PhysicalExtensionCodec for DfxStoragePhysicalCodec { + fn try_encode( + &self, + node: Arc<dyn ExecutionPlan>, + buf: &mut Vec<u8>, + proto_converter: &dyn PhysicalProtoConverterExtension, + ) -> Result<()> { + // Downcast to our own concrete type. Claiming a broad category -- + // `ForeignExecutionPlan`, say -- would take nodes from every library + // installed after this one, and the query would still succeed, so + // nothing would point at the codec that stole them. + let Some(exec) = node.downcast_ref::<PartitionedParquetExec>() else { + self.counters.declined.fetch_add(1, Ordering::SeqCst); + return self.inner.try_encode(node, buf, proto_converter); + }; + + let descriptor = serde_json::json!({ + "files": exec.files.iter().map(|file| { + serde_json::json!({ "path": file.path, "size": file.size }) + }).collect::<Vec<_>>(), + "projection": exec.projection, + "limit": exec.limit, + }); + let json = serde_json::to_vec(&descriptor) + .map_err(|err| internal_datafusion_err!("dfx_storage: encoding descriptor: {err}"))?; + let schema = schema_to_ipc_bytes(&exec.table_schema)?; + + buf.extend_from_slice(MAGIC); + let json_len = u32::try_from(json.len()) + .map_err(|_| internal_datafusion_err!("dfx_storage: descriptor too large to encode"))?; + buf.extend_from_slice(&json_len.to_le_bytes()); + buf.extend_from_slice(&json); + buf.extend_from_slice(&schema); + + self.counters.encoded.fetch_add(1, Ordering::SeqCst); + Ok(()) + } + + fn try_decode( + &self, + buf: &[u8], + inputs: &[Arc<dyn ExecutionPlan>], + ctx: &TaskContext, + proto_converter: &dyn PhysicalProtoConverterExtension, + ) -> Result<Arc<dyn ExecutionPlan>> { + // The chain routes a framed payload by id, so reaching this codec + // already means the payload is ours. Checking the magic anyway is + // cheap and turns a version skew into a clear error instead of a + // misparse. + let Some(rest) = buf.strip_prefix(MAGIC) else { + self.counters.declined.fetch_add(1, Ordering::SeqCst); + return self.inner.try_decode(buf, inputs, ctx, proto_converter); + }; + if !inputs.is_empty() { + return internal_err!( + "PartitionedParquetExec is a leaf, got {} input(s)", + inputs.len() + ); + } + + let (len_bytes, rest) = rest.split_at_checked(4).ok_or_else(|| { + internal_datafusion_err!("dfx_storage: payload truncated before descriptor length") + })?; + let json_len = u32::from_le_bytes( + len_bytes + .try_into() + .map_err(|_| internal_datafusion_err!("dfx_storage: bad descriptor length"))?, + ) as usize; + let (json, schema_bytes) = rest.split_at_checked(json_len).ok_or_else(|| { + internal_datafusion_err!( + "dfx_storage: descriptor claims {json_len} bytes, {} remain", + rest.len() + ) + })?; + + let descriptor: serde_json::Value = serde_json::from_slice(json) + .map_err(|err| internal_datafusion_err!("dfx_storage: bad descriptor: {err}"))?; + let files = descriptor["files"] + .as_array() + .ok_or_else(|| internal_datafusion_err!("dfx_storage: descriptor has no file list"))? + .iter() + .map(|file| { + let path = file["path"].as_str().ok_or_else(|| { + internal_datafusion_err!("dfx_storage: file entry has no path") + })?; + let size = file["size"].as_u64().ok_or_else(|| { + internal_datafusion_err!("dfx_storage: file entry {path} has no size") + })?; + Ok(FileSlice { + path: path.to_string(), + size, + }) + }) + .collect::<Result<Vec<_>>>()?; + let projection = descriptor["projection"].as_array().map(|indices| { + indices + .iter() + .filter_map(|index| index.as_u64().map(|index| index as usize)) + .collect::<Vec<_>>() + }); + let limit = descriptor["limit"].as_u64().map(|limit| limit as usize); + let schema = Arc::new(schema_from_ipc_bytes(schema_bytes)?); + + self.counters.decoded.fetch_add(1, Ordering::SeqCst); + Ok(Arc::new(PartitionedParquetExec::new( + files, schema, projection, limit, + )?)) + } +} diff --git a/examples/distributed/storage-library/src/exec.rs b/examples/distributed/storage-library/src/exec.rs new file mode 100644 index 00000000..3f09f039 --- /dev/null +++ b/examples/distributed/storage-library/src/exec.rs @@ -0,0 +1,181 @@ +// 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. + +//! This library's own execution plan node. +//! +//! A leaf, and deliberately so. A node with children hands them to the +//! framework to encode with the *host's* codec, which is the right thing but +//! means the interesting part of a codec -- what it writes down -- is somebody +//! else's problem. Everything this node needs to run lives in the node +//! itself: which files, which columns, how many rows. That is what +//! [`crate::codec`] writes to the wire, and it is why a plan built here can be +//! decoded in a process that has never seen this table registered. +//! +//! One output partition per file. That is the axis a distributed engine +//! splits along: partition `i` reads file `i` and nothing else, so two workers +//! never touch the same bytes and no coordination is needed. + +use std::fmt; +use std::sync::Arc; + +use arrow::datatypes::SchemaRef; +use datafusion::common::Result; +use datafusion::common::tree_node::TreeNodeRecursion; +use datafusion::datasource::listing::PartitionedFile; +use datafusion::datasource::physical_plan::{FileScanConfigBuilder, ParquetSource}; +use datafusion::datasource::source::DataSourceExec; +use datafusion::execution::TaskContext; +use datafusion::execution::object_store::ObjectStoreUrl; +use datafusion::physical_expr::{EquivalenceProperties, Partitioning, PhysicalExpr}; +use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType}; +use datafusion::physical_plan::{ + DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties, SendableRecordBatchStream, +}; + +/// One Parquet file, and the size the object store will report for it. +/// +/// The size travels with the path because `PartitionedFile` needs it up front +/// and a decoding process should not have to stat the file to rebuild a plan. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct FileSlice { + pub(crate) path: String, + pub(crate) size: u64, +} + +/// Scans a fixed list of Parquet files, one file per output partition. +#[derive(Debug)] +pub(crate) struct PartitionedParquetExec { + /// Output partition `i` reads `files[i]`. + pub(crate) files: Vec<FileSlice>, + /// The table's full schema, before projection. + pub(crate) table_schema: SchemaRef, + /// Column indices into `table_schema`, or `None` for all of them. + pub(crate) projection: Option<Vec<usize>>, + pub(crate) limit: Option<usize>, + properties: Arc<PlanProperties>, +} + +impl PartitionedParquetExec { + pub(crate) fn new( + files: Vec<FileSlice>, + table_schema: SchemaRef, + projection: Option<Vec<usize>>, + limit: Option<usize>, + ) -> Result<Self> { + let projected_schema = match projection.as_ref() { + Some(indices) => Arc::new(table_schema.project(indices)?), + None => Arc::clone(&table_schema), + }; + // `UnknownPartitioning`, not `Hash`: the rows are split by which file + // they happen to live in, which says nothing about their values. A + // plan that claimed a hash partitioning here would let the optimizer + // skip a repartition it actually needs. + let properties = Arc::new(PlanProperties::new( + EquivalenceProperties::new(projected_schema), + Partitioning::UnknownPartitioning(files.len()), + EmissionType::Incremental, + Boundedness::Bounded, + )); + Ok(Self { + files, + table_schema, + projection, + limit, + properties, + }) + } + + /// Build the stock scan for a single one of our files. + /// + /// Reusing `DataSourceExec` for the actual reading is the point: this node + /// exists to own the *description* of the scan across a process boundary, + /// not to reimplement Parquet. + fn scan_for(&self, partition: usize) -> Result<Arc<DataSourceExec>> { + let slice = self.files.get(partition).ok_or_else(|| { + datafusion::common::internal_datafusion_err!( + "PartitionedParquetExec has {} partition(s), asked for {partition}", + self.files.len() + ) + })?; + let source = Arc::new(ParquetSource::new(Arc::clone(&self.table_schema))); + let config = FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), source) + .with_file(PartitionedFile::new(slice.path.clone(), slice.size)) + .with_projection_indices(self.projection.clone())? + .with_limit(self.limit) + .build(); + Ok(DataSourceExec::from_data_source(config)) + } +} + +impl DisplayAs for PartitionedParquetExec { + fn fmt_as(&self, _t: DisplayFormatType, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "PartitionedParquetExec: files={}", self.files.len())?; + if let Some(projection) = self.projection.as_ref() { + write!(f, ", projection={projection:?}")?; + } + if let Some(limit) = self.limit { + write!(f, ", limit={limit}")?; + } + Ok(()) + } +} + +impl ExecutionPlan for PartitionedParquetExec { + fn name(&self) -> &str { + Self::static_name() + } + + fn properties(&self) -> &Arc<PlanProperties> { + &self.properties + } + + fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> { + vec![] + } + + fn apply_expressions( + &self, + _f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>, + ) -> Result<TreeNodeRecursion> { + // The projection is column indices, not expressions, and any pushed + // down filter is held by the `DataSourceExec` this node builds. + Ok(TreeNodeRecursion::Continue) + } + + fn with_new_children( + self: Arc<Self>, + children: Vec<Arc<dyn ExecutionPlan>>, + ) -> Result<Arc<dyn ExecutionPlan>> { + if !children.is_empty() { + return datafusion::common::internal_err!( + "PartitionedParquetExec is a leaf, got {} children", + children.len() + ); + } + Ok(self) + } + + fn execute( + &self, + partition: usize, + context: Arc<TaskContext>, + ) -> Result<SendableRecordBatchStream> { + // Partition 0 of the single-file scan: each of our partitions is one + // whole file, so the inner scan only ever has one of its own. + self.scan_for(partition)?.execute(0, context) + } +} diff --git a/examples/distributed/storage-library/src/extension.rs b/examples/distributed/storage-library/src/extension.rs new file mode 100644 index 00000000..b42944b3 --- /dev/null +++ b/examples/distributed/storage-library/src/extension.rs @@ -0,0 +1,156 @@ +// 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. + +//! This library's extension bundle: codecs only, no planner. +//! +//! A provider library has no business installing a query planner, so this +//! bundle implements `__datafusion_session_components__` and stops there. +//! `with_extensions` accepts a bundle that implements only one of the two +//! hooks. + +use std::fmt; +use std::sync::Arc; +use std::sync::atomic::Ordering; + +use datafusion_ffi::proto::physical_extension_codec::FFI_PhysicalExtensionCodec; +use datafusion_proto::physical_plan::PhysicalExtensionCodec; +use datafusion_python_util::{ + create_physical_extension_capsule, ffi_task_context_provider_from_pycapsule, get_tokio_runtime, +}; +use pyo3::prelude::*; +use pyo3::types::{PyCapsule, PyDict}; + +use crate::codec::{CodecCounters, DfxStoragePhysicalCodec}; + +/// Wire id this codec's payloads carry. +/// +/// Pinned rather than left to default to the exporting class's import path, +/// because these payloads outlive the process that wrote them: a driver that +/// imports the class as `dfx_storage.BundledPhysicalCodec` and a worker that +/// imports it under any other name would otherwise disagree about the id and +/// every decode would fail. +const PHYSICAL_CODEC_ID: &str = "dfx_storage.physical.v1"; + +/// Carries this library's physical codec as an object rather than a capsule. +/// +/// `with_extensions` requires an object: a codec's wire id is read off the +/// thing it is handed over as, and a capsule has no type to read one from. +#[pyclass(name = "BundledPhysicalCodec", module = "dfx_storage")] +pub(crate) struct BundledPhysicalCodec { + codec: FFI_PhysicalExtensionCodec, +} + +#[pymethods] +impl BundledPhysicalCodec { + #[getter] + fn __datafusion_codec_id__(&self) -> &'static str { + PHYSICAL_CODEC_ID + } + + /// `session` is unused: the codec was bound to its task-context provider + /// when the bundle was installed, which is why the bundle receives the + /// context at all. + #[pyo3(signature = (session=None))] + fn __datafusion_physical_extension_codec__<'py>( + &self, + py: Python<'py>, + session: Option<Bound<'py, PyAny>>, + ) -> PyResult<Bound<'py, PyCapsule>> { + let _ = session; + create_physical_extension_capsule(py, &self.codec) + } +} + +/// Extension bundle for `SessionContext.with_extensions`. +/// +/// Reusable configuration, not bound state: every +/// `__datafusion_session_components__` call builds fresh components against +/// the context it is handed, so one bundle may be installed on several +/// sessions. +#[pyclass(from_py_object, name = "DfxStorageExtension", module = "dfx_storage")] +#[derive(Default, Clone)] +pub(crate) struct DfxStorageExtension { + counters: Arc<CodecCounters>, +} + +impl fmt::Debug for DfxStorageExtension { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("DfxStorageExtension") + .field("counters", &self.counters) + .finish_non_exhaustive() + } +} + +#[pymethods] +impl DfxStorageExtension { + #[new] + fn new() -> Self { + Self::default() + } + + /// How often this codec encoded one of its own nodes. + fn encode_calls(&self) -> usize { + self.counters.encoded.load(Ordering::SeqCst) + } + + /// How often it rebuilt one, which is the half that happens on a worker. + fn decode_calls(&self) -> usize { + self.counters.decoded.load(Ordering::SeqCst) + } + + /// How often it was offered a node it does not own and passed it on. + /// + /// Non-zero is healthy: it means the chain is asking this codec about + /// other libraries' nodes and it is declining them. + fn declined_calls(&self) -> usize { + self.counters.declined.load(Ordering::SeqCst) + } + + /// The wire id, so a driver can put it in a worker's task envelope and + /// the worker can check it before decoding anything. + #[staticmethod] + fn physical_codec_id() -> &'static str { + PHYSICAL_CODEC_ID + } + + fn __datafusion_session_components__<'py>( + &self, + py: Python<'py>, + ctx: Bound<'py, PyAny>, + ) -> PyResult<Bound<'py, PyAny>> { + // Take the provider off the context supplied by the host, so the + // codec's decode callbacks resolve against the session that will run + // the query. + let provider = ffi_task_context_provider_from_pycapsule(&ctx)?; + let runtime = get_tokio_runtime().handle().clone(); + + let codec: Arc<dyn PhysicalExtensionCodec + Send> = + Arc::new(DfxStoragePhysicalCodec::new(Arc::clone(&self.counters))); + let ffi = FFI_PhysicalExtensionCodec::new(codec, Some(runtime), provider); + let physical = Py::new(py, BundledPhysicalCodec { codec: ffi })?; + + // No logical codec: this library defines no logical extension node. + // Its table provider crosses FFI as a provider, not as a plan node. + let components = py + .import("datafusion")? + .getattr("SessionExtensionComponents")?; + let kwargs = PyDict::new(py); + kwargs.set_item("physical_extension_codecs", (physical,))?; + components.call((), Some(&kwargs)) + } +} diff --git a/examples/distributed/storage-library/src/lib.rs b/examples/distributed/storage-library/src/lib.rs new file mode 100644 index 00000000..bb01cf9c --- /dev/null +++ b/examples/distributed/storage-library/src/lib.rs @@ -0,0 +1,40 @@ +// 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. + +//! Example storage library: a partitioned Parquet table provider, its own +//! execution plan node, and a physical codec that writes durable metadata. +//! +//! One of three libraries in `examples/distributed`. This one owns tables. + +use pyo3::prelude::*; + +use crate::extension::{BundledPhysicalCodec, DfxStorageExtension}; +use crate::table_provider::PyPartitionedParquetTable; + +mod codec; +mod exec; +mod extension; +mod table_provider; + +#[pymodule] +fn dfx_storage(m: &Bound<'_, PyModule>) -> PyResult<()> { + pyo3_log::init(); + m.add_class::<BundledPhysicalCodec>()?; + m.add_class::<DfxStorageExtension>()?; + m.add_class::<PyPartitionedParquetTable>()?; + Ok(()) +} diff --git a/examples/distributed/storage-library/src/table_provider.rs b/examples/distributed/storage-library/src/table_provider.rs new file mode 100644 index 00000000..4b980c03 --- /dev/null +++ b/examples/distributed/storage-library/src/table_provider.rs @@ -0,0 +1,182 @@ +// 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. + +//! A table provider over a directory of Parquet files. +//! +//! One output partition per file, which is the whole reason this provider +//! exists rather than `SessionContext.register_parquet`: it fixes the mapping +//! from partition index to file, so a distributed engine can hand partition +//! `i` to a worker and know exactly which bytes that worker will read. + +use std::fs; +use std::path::Path; +use std::sync::Arc; + +use arrow::datatypes::{Schema, SchemaRef}; +use async_trait::async_trait; +use datafusion::catalog::{Session, TableProvider}; +use datafusion::common::{DataFusionError, Result, plan_err}; +use datafusion::datasource::TableType; +use datafusion::logical_expr::{Expr, TableProviderFilterPushDown}; +use datafusion::parquet::arrow::parquet_to_arrow_schema; +use datafusion::parquet::file::reader::{FileReader, SerializedFileReader}; +use datafusion::physical_plan::ExecutionPlan; +use datafusion_ffi::table_provider::FFI_TableProvider; +use datafusion_python_util::ffi_logical_codec_from_pycapsule; +use pyo3::prelude::*; +use pyo3::types::PyCapsule; + +use crate::exec::{FileSlice, PartitionedParquetExec}; + +/// Scans `*.parquet` under `directory`, one partition per file. +#[derive(Debug)] +pub(crate) struct PartitionedParquetTable { + files: Vec<FileSlice>, + schema: SchemaRef, +} + +impl PartitionedParquetTable { + /// Read the directory listing and the first file's schema, once. + /// + /// Sorted by path so that partition `i` means the same file in every + /// process that opens the same directory. Directory iteration order is + /// not specified, and a worker that disagreed with the driver about which + /// file is partition 3 would silently produce wrong answers. + pub(crate) fn try_new(directory: &Path) -> Result<Self> { + let mut paths: Vec<_> = fs::read_dir(directory) + .map_err(|err| DataFusionError::External(Box::new(err)))? + .collect::<std::io::Result<Vec<_>>>() + .map_err(|err| DataFusionError::External(Box::new(err)))? + .into_iter() + .map(|entry| entry.path()) + .filter(|path| path.extension().is_some_and(|ext| ext == "parquet")) + .collect(); + paths.sort(); + + if paths.is_empty() { + return plan_err!("no .parquet files under {}", directory.display()); + } + + let mut files = Vec::with_capacity(paths.len()); + for path in &paths { + let metadata = + fs::metadata(path).map_err(|err| DataFusionError::External(Box::new(err)))?; + let path = path + .to_str() + .ok_or_else(|| DataFusionError::Plan(format!("non-UTF-8 path {path:?}")))?; + files.push(FileSlice { + path: path.to_string(), + size: metadata.len(), + }); + } + + let schema = Self::read_schema(&paths[0])?; + Ok(Self { + files, + schema: Arc::new(schema), + }) + } + + fn read_schema(path: &Path) -> Result<Schema> { + let file = fs::File::open(path).map_err(|err| DataFusionError::External(Box::new(err)))?; + let reader = SerializedFileReader::new(file) + .map_err(|err| DataFusionError::ParquetError(Box::new(err)))?; + let metadata = reader.metadata().file_metadata(); + parquet_to_arrow_schema(metadata.schema_descr(), metadata.key_value_metadata()) + .map_err(|err| DataFusionError::ParquetError(Box::new(err))) + } +} + +#[async_trait] +impl TableProvider for PartitionedParquetTable { + fn schema(&self) -> SchemaRef { + Arc::clone(&self.schema) + } + + fn table_type(&self) -> TableType { + TableType::Base + } + + fn supports_filters_pushdown( + &self, + filters: &[&Expr], + ) -> Result<Vec<TableProviderFilterPushDown>> { + // Every filter is re-applied above the scan. Claiming `Exact` would + // tell the optimizer to drop the `FilterExec`, and this node does not + // pass predicates down to the Parquet reader. + Ok(vec![ + TableProviderFilterPushDown::Unsupported; + filters.len() + ]) + } + + async fn scan( + &self, + _state: &dyn Session, + projection: Option<&Vec<usize>>, + _filters: &[Expr], + limit: Option<usize>, + ) -> Result<Arc<dyn ExecutionPlan>> { + Ok(Arc::new(PartitionedParquetExec::new( + self.files.clone(), + Arc::clone(&self.schema), + projection.cloned(), + limit, + )?)) + } +} + +/// Python handle for [`PartitionedParquetTable`]. +#[pyclass(name = "PartitionedParquetTable", module = "dfx_storage")] +pub(crate) struct PyPartitionedParquetTable { + directory: String, +} + +#[pymethods] +impl PyPartitionedParquetTable { + /// Open every `*.parquet` file under `directory` as one table. + #[new] + fn new(directory: String) -> Self { + Self { directory } + } + + /// Number of files, and so the number of output partitions. + fn partition_count(&self) -> PyResult<usize> { + Ok(self.build()?.files.len()) + } + + fn __datafusion_table_provider__<'py>( + &self, + py: Python<'py>, + session: Bound<'py, PyAny>, + ) -> PyResult<Bound<'py, PyCapsule>> { + let provider = Arc::new(self.build()?); + // The codec comes off the session this provider is being installed + // on, never from a `SessionContext` built here: one built inline is + // already dropped by the time the capsule is used. + let codec = ffi_logical_codec_from_pycapsule(session, None)?; + let ffi = FFI_TableProvider::new_with_ffi_codec(provider, false, None, codec); + PyCapsule::new_with_value(py, ffi, cr"datafusion_table_provider") + } +} + +impl PyPartitionedParquetTable { + fn build(&self) -> PyResult<PartitionedParquetTable> { + PartitionedParquetTable::try_new(Path::new(&self.directory)) + .map_err(|err| pyo3::exceptions::PyValueError::new_err(err.to_string())) + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
