From cfa09e27041d8c14c65d411def50c2fe577775ec Mon Sep 17 00:00:00 2001
From: Wenchao Wu <60921147+Stephen0421@users.noreply.github.com>
Date: Tue, 2 Jul 2024 10:22:08 +0800
Subject: [PATCH] [hotfix][parquet] Fix dependences and variable name problems
 in FLINK-35702 (#25004)

---
 flink-connectors/flink-connector-hive/pom.xml      |  5 +++++
 flink-formats/flink-parquet/pom.xml                | 14 ++++++++------
 .../parquet/vector/reader/NestedColumnReader.java  |  9 ++++-----
 3 files changed, 17 insertions(+), 11 deletions(-)

diff --git a/flink-connectors/flink-connector-hive/pom.xml b/flink-connectors/flink-connector-hive/pom.xml
index a4e43da013bda..8d54d961e5586 100644
--- a/flink-connectors/flink-connector-hive/pom.xml
+++ b/flink-connectors/flink-connector-hive/pom.xml
@@ -1264,6 +1264,11 @@ under the License.
 								<relocation>
 									<pattern>org.apache.flink.runtime.util</pattern>
 									<shadedPattern>org.apache.flink.hive.shaded.util</shadedPattern>
+									<excludes>
+										<exclude>org.apache.flink.runtime.util.IntArrayList</exclude>
+										<exclude>org.apache.flink.runtime.util.BooleanArrayList</exclude>
+										<exclude>org.apache.flink.runtime.util.LongArrayList</exclude>
+									</excludes>
 								</relocation>
 								<relocation>
 									<pattern>org.apache.flink.formats.parquet</pattern>
diff --git a/flink-formats/flink-parquet/pom.xml b/flink-formats/flink-parquet/pom.xml
index cb7452f607f47..9b8e133a8349b 100644
--- a/flink-formats/flink-parquet/pom.xml
+++ b/flink-formats/flink-parquet/pom.xml
@@ -88,6 +88,14 @@ under the License.
 			<optional>true</optional>
 		</dependency>
 
+		<!-- Runtime -->
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-runtime</artifactId>
+			<version>${project.version}</version>
+			<scope>provided</scope>
+		</dependency>
+
 		<!-- Parquet -->
 
 		<dependency>
@@ -273,12 +281,6 @@ under the License.
 			<scope>test</scope>
 			<type>test-jar</type>
 		</dependency>
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-runtime</artifactId>
-            <version>2.0-SNAPSHOT</version>
-            <scope>compile</scope>
-        </dependency>
     </dependencies>
 
 	<dependencyManagement>
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/NestedColumnReader.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/NestedColumnReader.java
index 0db4b4025820f..036bf99c3b602 100644
--- a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/NestedColumnReader.java
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/NestedColumnReader.java
@@ -27,6 +27,7 @@
 import org.apache.flink.formats.parquet.vector.type.ParquetGroupField;
 import org.apache.flink.formats.parquet.vector.type.ParquetPrimitiveField;
 import org.apache.flink.table.data.columnar.vector.ColumnVector;
+import org.apache.flink.table.data.columnar.vector.heap.AbstractHeapVector;
 import org.apache.flink.table.data.columnar.vector.heap.HeapArrayVector;
 import org.apache.flink.table.data.columnar.vector.heap.HeapMapVector;
 import org.apache.flink.table.data.columnar.vector.heap.HeapRowVector;
@@ -244,13 +245,11 @@ private Tuple2<LevelDelegation, WritableColumnVector> readPrimitive(
         return Tuple2.of(reader.getLevelDelegation(), writableColumnVector);
     }
 
-    private static void setFieldNullFalg(boolean[] rowPosition, WritableColumnVector vector) {
-        int index = 0;
-        for (boolean isNullIndex : rowPosition) {
-            if (isNullIndex) {
+    private static void setFieldNullFalg(boolean[] nullFlags, AbstractHeapVector vector) {
+        for (int index = 0; index < vector.getLen() && index < nullFlags.length; index++) {
+            if (nullFlags[index]) {
                 vector.setNullAt(index);
             }
-            index++;
         }
     }
 }