0dunay0 opened a new issue, #5640:
URL: https://github.com/apache/paimon/issues/5640

   ### Search before asking
   
   - [x] I searched in the [issues](https://github.com/apache/paimon/issues) 
and found nothing similar.
   
   
   ### Paimon version
   
   1.1.1
   
   ### Compute Engine
   
   Flink
   
   ### Minimal reproduce step
   
   Consider these two avro schemas where the first is evolved into the second 
by promoting primitives to larger sized ones. 
   
   ```
   {
       "type": "record",
       "name": "primitives_promotion",
       "namespace": "some_ns",
       "doc": "some stuff",
       "fields": [
       {
         "name": "int_and_long",
         "type": "int",
         "doc": "some stuff"
       },
       {
         "name": "float_and_double",
         "type": "float",
         "doc": "some stuff"
       }
     ]
   }
   ```
   
   ```
   {
       "type": "record",
       "name": "primitives_promotion",
       "namespace": "some_ns",
       "doc": "some stuff",
       "fields": [
       {
         "name": "int_and_long",
         "type": "long",   //  <-- int promoted to long (BIGINT in Paimon)
         "doc": "some stuff"
       },
       {
         "name": "float_and_double",
         "type": "double",  // <-- float promoted to double
         "doc": "some stuff"
       }
     ]
   }
   ```
   
   ```
   # This evolution is compatible in Avro
   
   from avro import schema
   from avro.compatibility import SchemaCompatibilityResult, 
ReaderWriterCompatibilityChecker
   
   avsc1 = schema.from_path('s1.json')
   avsc2 = schema.from_path('s2.json')
   
   rwc = ReaderWriterCompatibilityChecker()
   result = rwc.get_compatibility(avsc2, avsc1)
   
   print(result.compatibility)
   SchemaCompatibilityType.compatible
   
   print(result.incompatibilities)
   []
   ```
   
   Start an ingestion job with a kafka topic with the first schema then switch 
to second schema you should get errors like this.
   
   ```
   2025-05-20 03:13:26,649 INFO  
org.apache.paimon.flink.action.cdc.CdcActionCommonUtils      [] - Cannot 
convert field 'int_and_long' from source table type 'BIGINT' to Paimon type 
'INT'.
   2025-05-20 03:13:26,650 ERROR org.apache.flink.client.cli.CliFrontend        
              [] - Error while running the command.
   org.apache.flink.client.program.ProgramInvocationException: The main method 
caused an error: Paimon schema and source table schema are not compatible.
   Paimon fields are: [`int_and_long` INT, `float_and_double` FLOAT].
   Source table fields are: [`int_and_long` BIGINT, `float_and_double` DOUBLE].
   ```
   
   ### What doesn't meet your expectations?
   
   There appears to be a logic error in the schema compatibility check 
implementation in the CDC synchronization functionality. The 
[schemaCompatible](https://github.com/apache/paimon/blob/master/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/CdcActionCommonUtils.java#L95-L96)
 method in `CdcActionCommonUtils` is incorrectly checking whether source field 
types can be converted to Paimon table field types, rather than checking if the 
Paimon table can evolve to accommodate the source field types.
   
   This bug prevents valid schema evolution scenarios. If a Paimon table has an 
`INT` column and the source data has a `BIGINT` column, the current code calls 
`canConvert(BIGINT, INT)`. This returns `IGNORE` because a larger type can't be 
downcast to a smaller one. However, this is a valid evolution case - the Paimon 
table should evolve from `INT` to `BIGINT` to accommodate the source data.
   
   
   ### Anything else?
   
   _No response_
   
   ### Are you willing to submit a PR?
   
   - [x] I'm willing to submit a PR!


-- 
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]

Reply via email to