[spark] Reject variant extraction paths containing the metadata delimiter

paimon-common/src/main/java/org/apache/paimon/data/variant/VariantMetadataUtils.java

@@ -27,6 +27,8 @@
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
 /**
  * Utils for marking and identifying variant-originated RowType. Uses description field in DataField
  * to encode variant metadata.
@@ -42,6 +44,11 @@ public class VariantMetadataUtils {
 
     /** Build variant metadata description string. */
     public static String buildVariantMetadata(String path, boolean failOnError, String timeZoneId) {
+        checkArgument(
+                !path.contains(DELIMITER),
+                "Variant extraction path must not contain '%s': %s",
+                DELIMITER,
+                path);
         return METADATA_KEY + path + DELIMITER + failOnError + DELIMITER + timeZoneId;
     }
 

paimon-common/src/test/java/org/apache/paimon/data/variant/VariantMetadataUtilsTest.java

@@ -28,6 +28,7 @@
 
 import static org.apache.paimon.data.variant.VariantMetadataUtils.VariantRowTypeBuilder;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Tests for {@link VariantMetadataUtils}. */
 public class VariantMetadataUtilsTest {
@@ -38,6 +39,13 @@ public void testBuildVariantMetadata() {
         assertThat(metadata).isEqualTo("__VARIANT_METADATA$.a.b;true;UTC");
     }
 
+    @Test
+    public void testBuildVariantMetadataRejectsPathWithDelimiter() {
+        assertThatThrownBy(() -> VariantMetadataUtils.buildVariantMetadata("$.a;b", true, "UTC"))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("must not contain ';'");
+    }
+
     @Test
     public void testIsVariantRow() {
         // Create a row type with variant metadata

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/VariantPushDownUtils.scala

@@ -42,7 +42,7 @@ object VariantPushDownUtils {
     while (i < extractions.length) {
       val (path, info, isVariantTarget) = extractions(i)
       val canThrow = info.failOnError && !info.paimonType.isInstanceOf[VarCharType]
-      if (path.isEmpty || isVariantTarget || canThrow) {
+      if (path.isEmpty || isVariantTarget || canThrow || !canEncodePath(info.path)) {
         if (path.nonEmpty) {
           rejected += path
         }
@@ -68,6 +68,14 @@ object VariantPushDownUtils {
     (out.toMap, accepted)
   }
 
+  /**
+   * The extraction path is encoded into the projected field's description, delimited by
+   * [[VariantMetadataUtils.DELIMITER]] and split back on read with no escaping. A path carrying the
+   * delimiter would decode into a different path, so it cannot be pushed down.
+   */
+  private def canEncodePath(path: String): Boolean =
+    !path.contains(VariantMetadataUtils.DELIMITER)
+
   /**
    * Replace each variant field at a path in `accepted` with a Paimon variant `RowType`; recurse
    * into nested structs.

paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/read/VariantPushDownUtilsTest.scala

@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.read
+
+import org.apache.paimon.types.DataTypes
+
+import org.scalatest.funsuite.AnyFunSuite
+
+/** Tests for {@link VariantPushDownUtils}. */
+class VariantPushDownUtilsTest extends AnyFunSuite {
+
+  private def extraction(path: String) =
+    (Seq("v"), VariantExtractionInfo(DataTypes.STRING(), path, false, "UTC"), false)
+
+  test("accept an extraction path that can be encoded") {
+    val (byPath, accepted) = VariantPushDownUtils.acceptByPath(IndexedSeq(extraction("$.a.b")))
+    assert(accepted === Array(true))
+    assert(byPath.keySet === Set(Seq("v")))
+  }
+
+  test("reject an extraction path containing the metadata delimiter") {
+    Seq("$.a;b", "$[\"a;b\"]").foreach {
+      path =>
+        val (byPath, accepted) = VariantPushDownUtils.acceptByPath(IndexedSeq(extraction(path)))
+        assert(accepted === Array(false))
+        assert(byPath.isEmpty)
+    }
+  }
+}