Override RexNodeJsonDeserializer until loading PTFs via compiled plan is patched in Flink (#278)

Author: ferenc-csaky

flink-sql-runner/pom.xml

@@ -51,9 +51,17 @@
       <scope>provided</scope>
     </dependency>
 
+    <!-- TODO: Switch back to planner loader when PTF loading is patched upstream. -->
+    <!--    <dependency>-->
+    <!--      <groupId>org.apache.flink</groupId>-->
+    <!--      <artifactId>flink-table-planner-loader</artifactId>-->
+    <!--      <version>${flink.version}</version>-->
+    <!--      <scope>provided</scope>-->
+    <!--    </dependency>-->
+
     <dependency>
       <groupId>org.apache.flink</groupId>
-      <artifactId>flink-table-planner-loader</artifactId>
+      <artifactId>flink-table-planner_2.12</artifactId>
       <version>${flink.version}</version>
       <scope>provided</scope>
     </dependency>
@@ -482,6 +490,26 @@
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-dependency-plugin</artifactId>
         <executions>
+          <execution>
+            <id>unpack-flink-table-planner-without-rex-json-deser</id>
+            <goals>
+              <goal>unpack</goal>
+            </goals>
+            <phase>prepare-package</phase>
+            <configuration>
+              <artifactItems>
+                <artifactItem>
+                  <groupId>org.apache.flink</groupId>
+                  <artifactId>flink-table-planner_2.12</artifactId>
+                  <version>${flink.version}</version>
+                  <type>jar</type>
+                  <overWrite>true</overWrite>
+                  <outputDirectory>${project.build.directory}/flink-table-planner-temp</outputDirectory>
+                  <excludes>org/apache/flink/table/planner/plan/nodes/exec/serde/RexNodeJsonDeserializer**</excludes>
+                </artifactItem>
+              </artifactItems>
+            </configuration>
+          </execution>
           <execution>
             <id>unpack-iceberg-aws-bundle-without-logging</id>
             <goals>
@@ -612,12 +640,28 @@
         </executions>
       </plugin>
 
-      <!-- Repackage iceberg-aws-bundle without SLF4J using antrun plugin -->
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-antrun-plugin</artifactId>
         <executions>
           <execution>
+            <!-- Repackage flink-table-planner_2.12 and override RexNodeJsonDeserializer using antrun plugin -->
+            <id>repack-flink-table-planner</id>
+            <goals>
+              <goal>run</goal>
+            </goals>
+            <phase>prepare-package</phase>
+            <configuration>
+              <target>
+                <copy failonerror="true" todir="${project.build.directory}/flink-table-planner-temp/org/apache/flink/table/planner/plan/nodes/exec/serde">
+                  <fileset dir="${project.build.outputDirectory}/org/apache/flink/table/planner/plan/nodes/exec/serde" includes="RexNodeJsonDeserializer*.class"/>
+                </copy>
+                <jar basedir="${project.build.directory}/flink-table-planner-temp" destfile="${project.build.directory}/flink-table-planner_2.12-${flink.version}.jar"/>
+              </target>
+            </configuration>
+          </execution>
+          <execution>
+            <!-- Repackage iceberg-aws-bundle without SLF4J using antrun plugin -->
             <id>repack-iceberg-aws-bundle</id>
             <goals>
               <goal>run</goal>

flink-sql-runner/src/main/docker/Dockerfile

@@ -18,6 +18,7 @@ FROM apache/flink:${flink-base-image}
 
 RUN mkdir -p /opt/flink/plugins/flink-sql-runner
 
+COPY flink-table-planner_2.12-*.jar /opt/flink/lib
 COPY flink-s3-fs-hadoop-*.jar /opt/flink/lib
 COPY hadoop-common-*.jar /opt/flink/lib
 COPY hadoop-hdfs-client-*.jar /opt/flink/lib
@@ -26,8 +27,9 @@ COPY iceberg-flink-runtime-*.jar /opt/flink/lib
 COPY iceberg-aws-bundle-*.jar /opt/flink/lib
 COPY stdlib-utils-*.jar /opt/flink/lib
 COPY flink-sql-runner.uber.jar /opt/flink/plugins/flink-sql-runner
-COPY entrypoint.sh /entrypoint.sh
+COPY --chmod=755 entrypoint.sh /entrypoint.sh
 
-RUN ln -s /opt/flink/plugins/flink-sql-runner/flink-sql-runner.uber.jar /opt/flink/flink-sql-runner.jar
+RUN rm -rf /opt/flink/lib/flink-table-planner-loader-*.jar \
+    && ln -s /opt/flink/plugins/flink-sql-runner/flink-sql-runner.uber.jar /opt/flink/flink-sql-runner.jar
 
 ENTRYPOINT ["/entrypoint.sh"]