feat(io): Automated Schema Inference and Catalog Exploration for SQL Connectors - #257
feat(io): Automated Schema Inference and Catalog Exploration for SQL Connectors#257thinkapoorv wants to merge 4 commits into
Conversation
123c5f6 to
ed4c727
Compare
Closes pathwaycom#224. This commit introduces dynamic schema exploration for pw.io.postgres.read, pw.io.mysql.read, and pw.io.mssql.read, allowing users to omit the schema parameter when initializing database readers. ### Approach Instead of adding heavy Python-level database drivers (e.g., SQLAlchemy) to query the schemas, this implementation extends the existing internal Rust connectors to extract metadata directly from INFORMATION_SCHEMA and sys. The results are mapped directly to Pathway Schema definitions via schema_builder. ### Key Changes - **Rust Backend**: Exposes postgres_explore_schema, mysql_explore_schema, and mssql_explore_schema via PyO3 in python_api.rs. These functions securely invoke standard metadata queries utilizing internal iberius, mysql, and okio-postgres connections. - **Python Connectors**: Updates __init__.py for Postgres, MySQL, and MSSQL to handle schema=None. When triggered, they fetch schema topology from the Rust backend and construct a dynamic pw.Schema mapping. - **Primary Key Handling**: Automatically explores and applies primary_key=True properties to the corresponding pw.column_definition elements. If no PK is found, the engine logs a visible warning to inform the user about potential CDC/streaming issues. - **User Visibility**: The dynamically inferred schema is logged at startup, allowing developers to easily copy it into their codebase if they require stricter type enforcement down the line. This zero-dependency approach ensures type safety parity while vastly improving the developer experience for database onboarding.
f675d27 to
634ab77
Compare
…ture/automated-schema-exploration
a0433ad to
19a174c
Compare
|
@zxqfd555 Thanks for taking ownership of this one as well. Whenever you have a chance, I'd appreciate a review of the current implementation. I've rebased it onto the latest Looking forward to your feedback on the overall approach. |
zxqfd555
left a comment
zxqfd555
left a comment
There was a problem hiding this comment.
Thank you for the contribution, and for your patience with the review! I've read through the PR and left inline comments. The direction you chose — deducing the schema on the Rust side, with zero new dependencies — is the right one. However, we can't accept the current architecture built around it. To summarize the main points:
Extensibility. With only three databases covered, there is already a lot of duplicated code. The long-term intent is to support schema deduction for a couple dozen connectors, and in its current shape this approach would multiply that duplication across the codebase. The feature needs to be designed so that the common parts are generalized and adding a new source is simple and obvious.
Single place of responsibility. Everything needed to work with a given database (e.g., Postgres) lives in its own module today, and it must stay that way — otherwise, when something breaks, it's unclear where to look.
Abstraction leak. The Python connector shouldn't know anything about the database's type system. If we have to teach the Python layer which types exist in Postgres or MySQL, it duplicates knowledge the Rust layer already owns and spreads schema construction across many files. Instead, Rust can pass Python a mapping from field names to types, making the schema constructor storage-agnostic.
Besides the architecture, there are two more blockers:
- The current code contains bugs; I've pointed out some of them in the inline comments.
- A feature of this complexity requires integration tests under
integration_tests/.
Next steps. The first step here is not code but a careful design that guarantees extensibility and correctness — the code is the last step. So I wouldn't iterate on this code for now; I'll keep the PR open until there is a sketch of the generic design. If prioritization shows the feature is urgent, we may put such a sketch on our side.
| py_type = mapping.get(udt_name_lower, Any) | ||
| if is_nullable and py_type is not Any: | ||
| py_type = Optional[py_type] | ||
|
|
||
| is_pk = col_name in pk_columns | ||
| from pathway.internals.schema import column_definition | ||
|
|
||
| schema_columns[col_name] = column_definition( | ||
| dtype=py_type, | ||
| primary_key=is_pk, | ||
| ) |
There was a problem hiding this comment.
This part is repeated over three files (MySQL, MS SQL Server, Postgres input connectors, __init__.py in all cases).
Conceptually, given that in the long run, the schema is going to be deduced for everything we can, I think that all deduction tech must move into a separate file, not to be fragmented across the multitude of connectors. It's worth designing carefully beforehand, though: how can we store it, make necessary distinctions, and avoid duplications?
| mapping = { | ||
| "tinyint": int, | ||
| "smallint": int, | ||
| "int": int, | ||
| "bigint": int, | ||
| "bit": bool, | ||
| "real": float, | ||
| "float": float, | ||
| "decimal": float, | ||
| "numeric": float, | ||
| "char": str, | ||
| "varchar": str, | ||
| "nchar": str, | ||
| "nvarchar": str, | ||
| "text": str, | ||
| "ntext": str, | ||
| "uniqueidentifier": str, | ||
| } |
There was a problem hiding this comment.
The DB types are an abstraction level that we would generally prefer not to leak in the Python code. Moreover, whenever a new DB type becomes supported by the framework, we would not want to edit several places in order to support it.
Having that given, I would say that the preferred design of the change would be as follows:
- To keep the schema deduction for each particular DB inside its respective module (postgres.rs, mssql.rs, ...).
- Provide a mapping from the field name to the type (e.g.,
Typeas a Rust enum entity) as a dict. - On the Rust -> Python side, provide a mapping not to strings, but to the respective Python types. The schema constructor is then storage-agnostic and avoids duplication. All storage-specific routines are kept in the same place.
I didn't look closer into it, but there will also be cases where the schema probably can't be reconstructed because of column names that will be illegal for Python/Pathway. Needs research; if so, preflight checks must be put in place.
| connection_string: str, | ||
| table_name: str, | ||
| schema: type[Schema], | ||
| schema: type[Schema] | None = None, |
There was a problem hiding this comment.
Docstrings are not updated: schema can be deduced here and in other files, but it isn't mentioned in the docs.
| mapping = { | ||
| "int2": int, | ||
| "int4": int, | ||
| "int8": int, | ||
| "float4": float, | ||
| "float8": float, | ||
| "numeric": float, | ||
| "bool": bool, | ||
| "text": str, | ||
| "varchar": str, | ||
| "bpchar": str, | ||
| "char": str, | ||
| "uuid": str, | ||
| "json": str, | ||
| "jsonb": str, | ||
| } |
There was a problem hiding this comment.
Some of the supported types aren't mentioned here.
There was a problem hiding this comment.
Same for other connectors.
| IndexError, | ||
| match=( | ||
| re.escape(f"Index 2 out of range for a tuple of type {tuple[int,int]}.") | ||
| re.escape(f"Index 2 out of range for a tuple of type {tuple[int, int]}.") |
| match=( | ||
| "(?s)" # make dot match newlines | ||
| + re.escape(f"Index 2 out of range for a tuple of type {tuple[int,int]}. ") | ||
| + re.escape(f"Index 2 out of range for a tuple of type {tuple[int, int]}. ") |
There was a problem hiding this comment.
Also unrelated to this PR.
| def postgres_explore_schema( | ||
| connection_string: str, | ||
| schema_name: str | None, | ||
| table_name: str, | ||
| ssl_mode: str, | ||
| ssl_cert_path: str | None, | ||
| ) -> tuple[list[tuple[str, str, bool]], list[str]]: ... | ||
| def mysql_explore_schema( | ||
| connection_string: str, | ||
| table_name: str, | ||
| ) -> tuple[list[tuple[str, str, bool]], list[str]]: ... | ||
| def mssql_explore_schema( | ||
| connection_string: str, | ||
| full_table_name: str, | ||
| ) -> tuple[list[tuple[str, str, bool]], list[str]]: ... |
There was a problem hiding this comment.
The overall direction is good: I think it is useful to have a function in the PyO3 layer (python_api.rs) rather than to mix the DB interaction layer with the Rust<->Python interaction layer. However, with more and more sources supporting schema deduction, the number of dedicated methods will explode if we have one per source.
I would say that on the code architecture level, the responsibilities must be split the following way:
- Within a database connector (
postgres.rs, etc), we discover the schema and pass it over as a mapping from a field name to the type in its internal representation (enum Type). It knows nothing about the Python layer. - Within a
python_api.rsmethod, we accept in some form (requires thinking, probably just asDataStorage) the connection parameters and dispatch to the right module to call the schema construction. On completion, we use the mapping from internal types into Python types, which is exactly the responsibility zone of the Rust <-> Python layer, and do the required preflight checks, which will be communal, as they mostly concern the names that are forbidden in Python/Pathway and not connected in any way with the underlying sources.
|
|
||
| def postgres_explore_schema( | ||
| connection_string: str, | ||
| schema_name: str | None, |
There was a problem hiding this comment.
There is always a schema in Postgres, so it can't be None ;)
It is often omitted because the default value public is silently used. However, there can't be a table without a schema, and a schema acts like a namespace for a set of tables.
| columns_data, pk_columns = postgres_explore_schema( | ||
| _connection_string_from_settings(owned_postgres_settings), | ||
| schema_name, | ||
| table_name, | ||
| ssl_mode, | ||
| ssl_cert_path, | ||
| ) |
There was a problem hiding this comment.
It doesn't use the TLS settings, so it may differ from how the framework will read an actual table.
2 participants
Introduction
This PR introduces zero-dependency, automated schema exploration for structured database connectors (
pw.io.postgres.read,pw.io.mysql.read, andpw.io.mssql.read). It allows developers to completely omit the explicitpw.Schemaduring pipeline initialization, drastically reducing friction during data exploration and onboarding while strictly preserving Pathway's static type validation.Context
Currently, initializing a Pathway pipeline against an existing SQL database requires developers to manually duplicate the database's schema into a
pw.Schemaclass. For tables with dozens of columns, this is tedious and error-prone. This PR solves this by enabling the Pathway engine to automatically query the target database'sINFORMATION_SCHEMA(orsyscatalog) at startup, mapping SQL types and primary key constraints directly to Pathway types.Architectural Approaches Considered:
Python-Level Extraction via SQLAlchemy / Native Drivers
pyproject.tomlwith heavy, unnecessary dependencies (e.g.,psycopg2,pymysql). It would also duplicate connection/authentication logic outside of Pathway's core engine. (Rejected)Deferred Engine-Level Inference
Rust-Backed Catalog Extraction via PyO3 (Chosen Approach)
tokio-postgres,mysql,tiberius) already powering the Pathway engine.By choosing the Rust-backed approach, we infer the schema at pipeline construction time, bridging dynamic database metadata directly into strict
pw.Schemavalidation before the engine even starts.How has this been tested?
postgres_explore_schema,mysql_explore_schema, andmssql_explore_schematosrc/python_api.rs. Verified that they correctly extractdata_type,is_nullable, andPRIMARY KEYconstraints.__init__.pyfor all three connectors to handleschema=None.tinyint,varchar,uniqueidentifier) correctly map topw.dtypeprimitives, wrapped inOptionalwhere nullable.logging.warningis emitted advising the user about potential CDC stream degradation, rather than crashing the pipeline.blackandflake8against the modified files.Types of changes
Related issue(s):
Checklist:
(Note: I originally had a workaround for a
DuckDbWriterclippy warning here, but since that was recently patched onmain, I synced with upstream and dropped it!)