AliRana30 opened a new pull request, #803: URL: https://github.com/apache/wayang/pull/803
Closes #747 When executing a pipeline involving SQL TableSources, a SQL Join, and downstream Java operators (e.g., `PostgresTableSource -> Join -> SqlToStream -> Java map -> collect`), a runtime `ClassCastException` occurred. ### Root Cause 1. In Wayang, the output of a SQL `JoinOperator` is structured as `Tuple2<Record, Record>`. However, `SqlToStreamOperator` and `GenericSqlToStreamOperator` hardcoded their output to flat `Record` objects, discarding the tuple structure. When downstream Java operators expected `Tuple2<Record, Record>`, they received raw `Record` objects and threw runtime `ClassCastException`. 2. Neither `Channel.addConsumer` nor `DefaultChannelConversion.convert` performed early schema/type validation between channels and consumers, so mismatches were not caught during plan compilation. ### Changes - **Generic Operators**: Made `SqlToStreamOperator`, `SqlToRddOperator`, and `GenericSqlToStreamOperator` generic (`<Type>`) with support for emitting `Tuple2<Record, Record>` via `Tuple2ResultSetIterator` and `rowToTuple2`. - **Dynamic Split Detection**: Implemented `resolveLeftColumnCount` and `getTableColumnCount` using database metadata and operator schemas to accurately partition joined rows into left and right `Record` instances. - **Dynamic Channel Conversions**: Updated `ChannelConversions` across all SQL platforms (`postgres`, `sqlite3`, `trino`, `presto`, `bigquery`, `generic-jdbc`) to pass the channel producer's data type. - **Early Type Validation**: Enhanced `Channel.addConsumer` and `DefaultChannelConversion.convert` to validate `consumerInput.getType().isSupertypeOf(producerSlot.getType())` or adapt types via `adaptType(...)`, throwing an early `IllegalArgumentException` on genuine mismatches. - **Tests**: Added tests for SQL join to Stream `Tuple2` evaluation with Java mapping in `SqlToStreamOperatorTest`, and channel type validation tests in `ChannelTypeValidationTest`. ## Type of Change - [x] Bug fix (non-breaking change which fixes an issue) - [ ] New feature (non-breaking change which adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected) - [ ] Documentation update ## How Has This Been Tested? - `SqlToStreamOperatorTest` (4 tests passed, including `testJoinWithHsqldbYieldsTuple2` and `testTypeAdaptationAndValidation`). - `ChannelTypeValidationTest` (3 tests passed, verifying type mismatch rejection and compatible consumer acceptance). - `JdbcJoinOperatorTest` and `GenericJdbcJoinOperatorTest` passed with 0 failures. - Clean compilation across all SQL platform modules. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
