使用flink cdc 执行数据同步，目标数据是由多张源表进行关联查询得到的，查询语句较为复杂,已验证可以在数据库中查到数据，代码执行到同步数据的部分报空指针，请大佬们帮忙看一下

[luxiao66]

使用flink cdc 执行数据同步，目标数据是由多张源表进行关联查询得到的，查询语句较为复杂,已验证可以在数据库中查到数据，代码执行到同步数据的部分报空指针，请大佬们帮忙看一下

使用flink cdc 执行数据同步，目标数据是由多张源表进行关联查询得到的，查询语句较为复杂,已验证可以在数据库中查到数据，代码执行到同步数据的部分报空指针，请大佬们帮忙分析一下原因
Using Flink CDC to perform data synchronization, the target data is obtained from multiple source tables through associated queries. The query statements are relatively complex and have been verified to be able to find data in the database. When the code is executed, a null pointer is reported in the synchronized data section. Please help analyze the reason。

代码合错误信息如下：
String transformSQL ="INSERT INTO vehicle_info_access SELECT id\n" +
",\n" +
"veh_instance_id,\n" +
"create_time,\n" +
"device_id,\n" +
"vin,\n" +
"load_type,\n" +
"veh_usage AS vehUse,\n" +
"license_color,\n" +
"brand,\n" +
"license,\n" +
"veh_series,\n" +
"veh_cat AS veh_type,\n" +
"engine_number,\n" +
"engine_brand,\n" +
"engine_type,\n" +
"max_horsepower,\n" +
"transmission_model,\n" +
"carriage_length,\n" +
"actuate,\n" +
"rear_axle_speed_ratio,\n" +
"tank_volume,\n" +
"power_type AS energy_type,\n" +
"tyre_type,\n" +
"vehicle_mass,\n" +
"rated_load_mass,\n" +
"urea_volume,\n" +
"TBOX,\n" +
"ADAS \n" +
"FROM\n" +
"\t(\n" +
"\tSELECT\n" +
"\t\tdvi.id,\n" +
"\t\tdvi.veh_instance_id,\n" +
"\t\tdvi.vin,\n" +
"\t\tdvi.veh_usage,\n" +
"\t\tdvi.license,\n" +
"\t\tdvi.create_time,\n" +
"\t\tdvm.brand,\n" +
"\t\tdvm.power_type,\n" +
"\t\tdvm.veh_cat,\n" +
"\t\tdvm.veh_series,\n" +
"\t\tddt.* \n" +
"\tFROM\n" +
"\t\tdev_vehicle_instance dvi\n" +
"\t\tLEFT JOIN dev_vehicle_model dvm ON dvi.dvm_id = dvm.\n" +
"\t\tid LEFT JOIN (\n" +
"\t\tSELECT\n" +
"\t\t\tdevice_id,\n" +
"\t\t\tMAX ( CASE WHEN key = 'load_type' THEN dt.value END ) AS load_type,\n" +
"\t\t\tMAX ( CASE WHEN key = 'license_color' THEN dt.value END ) AS license_color,\n" +
"\t\t\tMAX ( CASE WHEN key = 'engine_number' THEN dt.value END ) AS engine_number,\n" +
"\t\t\tMAX ( CASE WHEN key = 'engine_brand' THEN dt.value END ) AS engine_brand,\n" +
"\t\t\tMAX ( CASE WHEN key = 'engine_type' THEN dt.value END ) AS engine_type,\n" +
"\t\t\tMAX ( CASE WHEN key = 'max_horsepower' THEN dt.value END ) AS max_horsepower,\n" +
"\t\t\tMAX ( CASE WHEN key = 'transmission_model' THEN dt.value END ) AS transmission_model,\n" +
"\t\t\tMAX ( CASE WHEN key = 'carriage_length' THEN dt.value END ) AS carriage_length,\n" +
"\t\t\tMAX ( CASE WHEN key = 'actuate' THEN dt.value END ) AS actuate,\n" +
"\t\t\tMAX ( CASE WHEN key = 'rear_axle_speed_ratio' THEN dt.value END ) AS rear_axle_speed_ratio,\n" +
"\t\t\tMAX ( CASE WHEN key = 'tank_volume' THEN dt.value END ) AS tank_volume,\n" +
"\t\t\tMAX ( CASE WHEN key = 'tyre_type' THEN dt.value END ) AS tyre_type,\n" +
"\t\t\tMAX ( CASE WHEN key = 'vehicle_mass' THEN dt.value END ) AS vehicle_mass,\n" +
"\t\t\tMAX ( CASE WHEN key = 'rated_load_mass' THEN dt.value END ) AS rated_load_mass,\n" +
"\t\t\tMAX ( CASE WHEN key = 'urea_volume' THEN dt.value END ) AS urea_volume,\n" +
"\t\t\tMAX ( CASE WHEN key = 'TBOX' THEN dt.value END ) AS TBOX,\n" +
"\t\t\tMAX ( CASE WHEN key = 'ADAS' THEN dt.value END ) AS ADAS \n" +
"\t\tFROM\n" +
"\t\t\tdev_device_tags dt \n" +
"\t\tGROUP BY\n" +
"\t\t\tdevice_id \n" +
"\t\t) ddt ON dvi.veh_instance_id = ddt.device_id \n" +
"\tWHERE\n" +
"\t\tEXISTS (\n" +
"\t\tSELECT\n" +
"\t\t\t1 \n" +
"\t\tFROM\n" +
"\t\t\tdev_device_instance ddi \n" +
"\t\tWHERE\n" +
"\t\t\tddi.id IN (\n" +
"\t\t\tSELECT\n" +
"\t\t\t\tdie.device_instance_id \n" +
"\t\t\tFROM\n" +
"\t\t\t\tdev_device_instance_ext die\n" +
"\t\t\t\tLEFT JOIN vehicle_device_instance_relation vdir ON die.id = vdir.device_instance_id \n" +
"\t\t\tWHERE\n" +
"\t\t\t\tvdir.veh_instance_id = dvi.id \n" +
"\t\t\t) \n" +
"\t\t\tAND ddi.state IN ( 'online', 'offline' ) \n" +
"\t\t) \n" +
"\t) r\n" +
"\t";

Exception in thread "main" java.lang.NullPointerException
at org.apache.calcite.sql2rel.RelDecorrelator.decorrelateRel(RelDecorrelator.java:1199)
at org.apache.calcite.sql2rel.RelDecorrelator.decorrelateRel(RelDecorrelator.java:1153)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.calcite.util.ReflectUtil$2.invoke(ReflectUtil.java:525)
at org.apache.calcite.sql2rel.RelDecorrelator.getInvoke(RelDecorrelator.java:687)
at org.apache.calcite.sql2rel.RelDecorrelator.decorrelateRel(RelDecorrelator.java:1123)
at org.apache.calcite.sql2rel.RelDecorrelator.decorrelateRel(RelDecorrelator.java:1102)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.calcite.util.ReflectUtil$2.invoke(ReflectUtil.java:525)
at org.apache.calcite.sql2rel.RelDecorrelator.getInvoke(RelDecorrelator.java:687)
at org.apache.calcite.sql2rel.RelDecorrelator.decorrelateRel(RelDecorrelator.java:1123)
at org.apache.calcite.sql2rel.RelDecorrelator.decorrelateRel(RelDecorrelator.java:1102)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.calcite.util.ReflectUtil$2.invoke(ReflectUtil.java:525)
at org.apache.calcite.sql2rel.RelDecorrelator.getInvoke(RelDecorrelator.java:687)
at org.apache.calcite.sql2rel.RelDecorrelator.decorrelateRel(RelDecorrelator.java:734)
at org.apache.calcite.sql2rel.RelDecorrelator.decorrelateRel(RelDecorrelator.java:723)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.calcite.util.ReflectUtil$2.invoke(ReflectUtil.java:525)
at org.apache.calcite.sql2rel.RelDecorrelator.getInvoke(RelDecorrelator.java:687)
at org.apache.calcite.sql2rel.RelDecorrelator.decorrelateRel(RelDecorrelator.java:734)
at org.apache.calcite.sql2rel.RelDecorrelator.decorrelateRel(RelDecorrelator.java:723)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.calcite.util.ReflectUtil$2.invoke(ReflectUtil.java:525)
at org.apache.calcite.sql2rel.RelDecorrelator.getInvoke(RelDecorrelator.java:687)
at org.apache.calcite.sql2rel.RelDecorrelator.decorrelateRel(RelDecorrelator.java:489)
at org.apache.calcite.sql2rel.RelDecorrelator.decorrelateRel(RelDecorrelator.java:472)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.calcite.util.ReflectUtil$2.invoke(ReflectUtil.java:525)
at org.apache.calcite.sql2rel.RelDecorrelator.getInvoke(RelDecorrelator.java:687)
at org.apache.calcite.sql2rel.RelDecorrelator.decorrelateRel(RelDecorrelator.java:1171)
at org.apache.calcite.sql2rel.RelDecorrelator.decorrelateRel(RelDecorrelator.java:1153)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.calcite.util.ReflectUtil$2.invoke(ReflectUtil.java:525)
at org.apache.calcite.sql2rel.RelDecorrelator.getInvoke(RelDecorrelator.java:687)
at org.apache.calcite.sql2rel.RelDecorrelator.decorrelateRel(RelDecorrelator.java:734)
at org.apache.calcite.sql2rel.RelDecorrelator.decorrelateRel(RelDecorrelator.java:723)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.calcite.util.ReflectUtil$2.invoke(ReflectUtil.java:525)
at org.apache.calcite.sql2rel.RelDecorrelator.getInvoke(RelDecorrelator.java:687)
at org.apache.calcite.sql2rel.RelDecorrelator.decorrelateRel(RelDecorrelator.java:734)
at org.apache.calcite.sql2rel.RelDecorrelator.decorrelateRel(RelDecorrelator.java:723)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.calcite.util.ReflectUtil$2.invoke(ReflectUtil.java:525)
at org.apache.calcite.sql2rel.RelDecorrelator.getInvoke(RelDecorrelator.java:687)
at org.apache.calcite.sql2rel.RelDecorrelator.decorrelateRel(RelDecorrelator.java:391)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.calcite.util.ReflectUtil$2.invoke(ReflectUtil.java:525)
at org.apache.calcite.sql2rel.RelDecorrelator.getInvoke(RelDecorrelator.java:687)
at org.apache.calcite.sql2rel.RelDecorrelator.decorrelate(RelDecorrelator.java:276)
at org.apache.calcite.sql2rel.RelDecorrelator.decorrelateQuery(RelDecorrelator.java:200)
at org.apache.calcite.sql2rel.RelDecorrelator.decorrelateQuery(RelDecorrelator.java:165)
at org.apache.flink.table.planner.plan.optimize.program.FlinkDecorrelateProgram.optimize(FlinkDecorrelateProgram.scala:41)
at org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$2(FlinkGroupProgram.scala:63)
at scala.collection.TraversableOnce.$anonfun$foldLeft$1(TraversableOnce.scala:156)
at scala.collection.TraversableOnce.$anonfun$foldLeft$1$adapted(TraversableOnce.scala:156)
at scala.collection.Iterator.foreach(Iterator.scala:937)
at scala.collection.Iterator.foreach$(Iterator.scala:937)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1425)
at scala.collection.IterableLike.foreach(IterableLike.scala:70)
at scala.collection.IterableLike.foreach$(IterableLike.scala:69)
at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:156)
at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:154)
at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:104)
at org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1(FlinkGroupProgram.scala:60)
at org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1$adapted(FlinkGroupProgram.scala:55)
at scala.collection.TraversableOnce.$anonfun$foldLeft$1(TraversableOnce.scala:156)
at scala.collection.TraversableOnce.$anonfun$foldLeft$1$adapted(TraversableOnce.scala:156)
at scala.collection.immutable.Range.foreach(Range.scala:155)
at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:156)
at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:154)
at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:104)
at org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.optimize(FlinkGroupProgram.scala:55)
at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:62)
at scala.collection.TraversableOnce.$anonfun$foldLeft$1(TraversableOnce.scala:156)
at scala.collection.TraversableOnce.$anonfun$foldLeft$1$adapted(TraversableOnce.scala:156)
at scala.collection.Iterator.foreach(Iterator.scala:937)
at scala.collection.Iterator.foreach$(Iterator.scala:937)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1425)
at scala.collection.IterableLike.foreach(IterableLike.scala:70)
at scala.collection.IterableLike.foreach$(IterableLike.scala:69)
at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:156)
at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:154)
at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:104)
at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:58)
at org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:163)
at org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:81)
at org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:77)
at org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:279)
at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:163)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1518)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:740)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:856)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:730)
at cn.cicv.cloud.flink.utils.EnvUtil.executeSqlList(EnvUtil.java:54)
at cn.cicv.cloud.flink.task.VehicleInfoCDCTask.processTask(VehicleInfoCDCTask.java:520)
at cn.cicv.cloud.flink.task.TaskEngine.run(TaskEngine.java:33)
at cn.cicv.cloud.flink.Main.main(Main.java:36)

查询sql.txt

错误信息.txt