Add DuckDB plugin

[andreahlert]

Summary

• Add a DuckDB connector plugin for running SQL queries against DuckDB as Flyte tasks
• DuckDB is an embedded analytical database (like SQLite for OLAP) that runs locally and synchronously
• Follows the same architecture and patterns as the Snowflake plugin

Features

• In-memory and file-based database support via DuckDBConfig(database_path=...)
• Parameterized SQL queries with typed inputs
• Extension installation and loading (httpfs, json, spatial, etc.)
• Query results returned as pandas DataFrames via temporary parquet files
• Automatic cleanup of temporary result files on delete

Design

Unlike Snowflake/BigQuery, DuckDB runs locally with no remote service, so the connector pattern is adapted:

• Queries execute synchronously in create() (wrapped in run_in_executor for async compat)
• get() always returns SUCCEEDED since queries complete in create()
• delete() cleans up temporary parquet result files
• No credentials, polling, or dashboard links needed

Test plan

• 25 unit tests covering task creation, config serialization, SQL generation, connector create/get/delete, parameterized queries, extensions, and end-to-end flow
• All tests pass (pytest plugins/duckdb/tests/ -v)
• Linting passes (ruff check plugins/duckdb/)
• Plugin structure matches Snowflake plugin layout

[kumare3]

Where is the data.parquet coming from?
Is this an input of type flyte.io.DataFrame

Then i would love to support

count_rows = DuckDB(
    name="count_rows",
    query_template="SELECT COUNT(*) AS total FROM '{input}'",
    plugin_config=config,
    input=DataFrame,   # This can be implicit
    output_dataframe_type=DataFrame,
)

Then you can pass a parquet, a pandasDataframe, a spark DataFrame or anything to it

[kumare3]

You dont want a connector. connector is useful for connecting to remote services. For this it should just be a Dataframe type or a task plugin

[kumare3]

delete this entire file

[kumare3]

No need of connector mixin. Rather this should implement the execute method. It should be like ContainerTask (not all of it just the execute function)

flyte-sdk/src/flyte/extras/_container.py

Line 235 in 28396f2

async def execute(self, **kwargs) -> Any:

[kumare3]

Once you do this, you can delete the connector. You need this way, because you need guaranteed memory and isolation. connector is ok for shared resources, this is usually api calls

[kumare3]

I think you got it a little wrong

[andreahlert]

I think you got it a little wrong

Thanks for the review! You're right, I should have looked at the existing flytekit DuckDB plugin as reference instead of modeling it after Snowflake. I'll rework this to use TaskTemplate with execute() and add DataFrame input type support.