|
| 1 | +""" |
| 2 | +DataFusion Python FFI provider example. |
| 3 | +
|
| 4 | +Walks the conformance matrix in numbered sections: table provider, functions, |
| 5 | +catalog provider, config extension, codec round-trip, then the same bytes |
| 6 | +decoded a second time. |
| 7 | +""" |
| 8 | +import sys |
| 9 | + |
| 10 | +try: |
| 11 | + from datafusion_ffi_example import ( |
| 12 | + IsNullUDF, |
| 13 | + MyCatalogProvider, |
| 14 | + MyConfig, |
| 15 | + MyLogicalExtensionCodec, |
| 16 | + MyTableProvider, |
| 17 | + ) |
| 18 | +except ImportError: |
| 19 | + sys.exit("build the extension first:\n uv run maturin develop\nSee README.md.") |
| 20 | + |
| 21 | +from datafusion import LogicalPlan, SessionConfig, SessionContext, udf |
| 22 | + |
| 23 | + |
| 24 | + |
| 25 | + |
| 26 | +print("1. table provider") |
| 27 | +ctx = SessionContext() |
| 28 | +ctx.register_table("numbers", MyTableProvider(1, 6, 1)) |
| 29 | +ctx.sql('SELECT "A" FROM numbers').show() |
| 30 | + |
| 31 | +print("\n2. functions") |
| 32 | +ctx.register_udf(udf(IsNullUDF())) |
| 33 | +ctx.sql('SELECT "A", my_custom_is_null("A") AS is_null FROM numbers').show() |
| 34 | + |
| 35 | +print("\n3. catalog provider") |
| 36 | +ctx.register_catalog_provider("ffi_catalog", MyCatalogProvider()) |
| 37 | +ctx.sql("SELECT * FROM ffi_catalog.my_schema.my_table").show() |
| 38 | + |
| 39 | +print("\n4. config extension") |
| 40 | +config = MyConfig() |
| 41 | +config = SessionConfig({"datafusion.catalog.information_schema": "true"}).with_extension(config) |
| 42 | +config.set("my_config.baz_count", "42") |
| 43 | +ctx2 = SessionContext(config) |
| 44 | +ctx2.sql("SHOW my_config.baz_count;").show() |
| 45 | + |
| 46 | +print("\n5. codec round-trip") |
| 47 | +codec = MyLogicalExtensionCodec() |
| 48 | +ctx3 = SessionContext().with_logical_extension_codec(codec) |
| 49 | +ctx3.register_table("numbers", MyTableProvider(1, 4, 1)) |
| 50 | +plan = ctx3.sql('SELECT "A" FROM numbers').logical_plan() |
| 51 | +blob = plan.to_bytes(ctx3) |
| 52 | +restored = LogicalPlan.from_bytes(ctx3, blob) |
| 53 | +ctx3.create_dataframe_from_logical_plan(restored).show() |
| 54 | + |
| 55 | +print("\n6. the same bytes decoded a second time") |
| 56 | +restored2 = LogicalPlan.from_bytes(ctx3, blob) |
| 57 | +ctx3.create_dataframe_from_logical_plan(restored2).show() |
0 commit comments