[SQL / Core] Preserve tuple schema and validate types between SQL and Java operators

[AliRana30]

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

• 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.

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[AliRana30]

cc @zkaoudi @novatechflow

Just wanted to highlight the impact of this bug in production pipelines:

In cross-platform workflows (e.g., PostgreSQL / BigQuery / Trino ➔ Java Streams / Spark), SQL joins and network transfers are typically the heaviest and most expensive steps. Because there was no plan-time validation, pipelines would spend significant compute time, cluster bandwidth, and cloud warehouse credits running complex joins across large datasets, only to fail at the very final Java step on the first record with a ClassCastException.

Beyond the wasted compute budget and delayed SLA cycles, if records were accessed by numerical field indices without explicit casts, the flattened tuple structure could have silently corrupted downstream analytics by mapping fields to the wrong offsets. Catching this early during channel conversion and properly preserving Tuple2<Record, Record> ensures pipelines fail fast on schema errors and execute reliably across platform boundaries.