apache · BackendArchitectX · Sep 3, 2026
diff --git a/...park/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussScanBuilder.scala b/...park/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussScanBuilder.scala
@@ -96,14 +96,7 @@ trait FlussSupportsPushDownV2Filters extends FlussSupportsPushDownPartitionFilte
         SparkPredicateConverter.convertPredicates(tableInfo.getRowType, nonPartition.toSeq)
       pushedPredicate = predicate
       acceptedPredicates = accepted.toArray
-    } else if (
-      tableInfo.getTableConfig.isDataLakeEnabled &&
-      tableInfo.getLakeTablePath == tableInfo.getTablePath
-    ) {
-      // TODO: Remove the custom lake path guard when Spark supports reading custom Paimon paths.
-      // Predicate pushdown also runs for log-only fallback and streaming scans. Do not fail those
-      // paths before we know that lake data is needed; actual lake reads are rejected when they
-      // create the lake source.
+    } else if (tableInfo.getTableConfig.isDataLakeEnabled) {
       // Lake-enabled tables: probe the lake source for which predicates it accepts. All predicates
       // (including partition) are offered because the lake source handles both partition pruning
       // and data filtering internally. This recovers the former FlussLakeSupportsPushDownV2Filters

diff --git a/...k/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/lake/FlussLakeUtils.scala b/...k/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/lake/FlussLakeUtils.scala
@@ -53,12 +53,7 @@ object FlussLakeUtils extends Logging {
       tableProperties: util.Map[String, String],
       tablePath: TablePath): LakeSource[LakeSplit] = {
     val tableConfig = Configuration.fromMap(tableProperties)
-    // TODO: Support reading custom Paimon lake table paths in Spark.
-    // See https://github.com/apache/fluss/issues/3832.
-    if (LakeTableUtil.resolveLakeTablePath(tablePath, tableConfig) != tablePath) {
-      throw new UnsupportedOperationException(
-        "Custom lake table path is not supported for Spark lake reads yet.")
-    }
+    val lakeTablePath = LakeTableUtil.resolveLakeTablePath(tablePath, tableConfig)
     val datalakeFormat = tableConfig.get(ConfigOptions.TABLE_DATALAKE_FORMAT)
     val dataLakePrefix = "table.datalake." + datalakeFormat + "."
 
@@ -73,7 +68,7 @@ object FlussLakeUtils extends Logging {
     val lakeStoragePlugin =
       LakeStoragePluginSetUp.fromDataLakeFormat(datalakeFormat.toString, null)
     val lakeStorage = lakeStoragePlugin.createLakeStorage(lakeConfig)
-    lakeStorage.createLakeSource(tablePath).asInstanceOf[LakeSource[LakeSplit]]
+    lakeStorage.createLakeSource(lakeTablePath).asInstanceOf[LakeSource[LakeSplit]]
   }
 
   def lakeProjection(projection: Array[Int]): Array[Array[Int]] = {

diff --git a/...fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/SparkLakeLogTableReadTest.scala b/...fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/SparkLakeLogTableReadTest.scala
@@ -734,30 +734,69 @@ class SparkLakePaimonLogTableReadTest extends SparkLakeLogTableReadTest {
     }
   }
 
-  test("Spark Lake Read: custom lake path rejects an actual lake read") {
-    withTable("t_custom_lake_path_snapshot") {
-      sql(s"""
-             |CREATE TABLE $DEFAULT_DATABASE.t_custom_lake_path_snapshot (id INT, name STRING)
-             | TBLPROPERTIES (
-             |  '${ConfigOptions.TABLE_DATALAKE_ENABLED.key()}' = true,
-             |  '${ConfigOptions.TABLE_DATALAKE_DATABASE_NAME.key()}' = 'custom_lake_db',
-             |  '${ConfigOptions.TABLE_DATALAKE_TABLE_NAME.key()}' = 'custom_lake_table_snapshot',
-             |  '${ConfigOptions.TABLE_DATALAKE_FRESHNESS.key()}' = '1s',
-             |  '${BUCKET_NUMBER.key()}' = 1)
-             |""".stripMargin)
-
-      sql(s"""
-             |INSERT INTO $DEFAULT_DATABASE.t_custom_lake_path_snapshot VALUES
-             |(1, "hello"), (2, "fluss")
-             |""".stripMargin)
-      tierToLake("t_custom_lake_path_snapshot")
-
-      val error = intercept[UnsupportedOperationException] {
-        sql(s"SELECT * FROM $DEFAULT_DATABASE.t_custom_lake_path_snapshot").collect()
+  Seq(
+    (
+      "custom database",
+      "t_custom_lake_database",
+      Some("custom_lake_db"),
+      None
+    ),
+    (
+      "custom table",
+      "t_custom_lake_table",
+      None,
+      Some("mapped_lake_table")
+    ),
+    (
+      "custom database and table",
+      "t_custom_lake_database_table",
+      Some("custom_lake_db_combined"),
+      Some("mapped_lake_table_combined")
+    )
+  ).foreach {
+    case (description, tableName, lakeDatabase, lakeTable) =>
+      test(s"Spark Lake Read: $description mapping supports lake-only and union reads") {
+        withTable(tableName) {
+          val customPathProperties =
+            Seq(
+              lakeDatabase.map(
+                name => s"'${ConfigOptions.TABLE_DATALAKE_DATABASE_NAME.key()}' = '$name'"),
+              lakeTable.map(name => s"'${ConfigOptions.TABLE_DATALAKE_TABLE_NAME.key()}' = '$name'")
+            ).flatten.mkString(",\n  ")
+
+          sql(s"""
+                 |CREATE TABLE $DEFAULT_DATABASE.$tableName (id INT, name STRING)
+                 | TBLPROPERTIES (
+                 |  '${ConfigOptions.TABLE_DATALAKE_ENABLED.key()}' = true,
+                 |  $customPathProperties,
+                 |  '${ConfigOptions.TABLE_DATALAKE_FRESHNESS.key()}' = '1s',
+                 |  '${BUCKET_NUMBER.key()}' = 1)
+                 |""".stripMargin)
+
+          sql(s"""
+                 |INSERT INTO $DEFAULT_DATABASE.$tableName VALUES
+                 |(1, "hello"), (2, "fluss")
+                 |""".stripMargin)
+
+          tierToLake(tableName)
+
+          checkAnswer(
+            sql(s"SELECT * FROM $DEFAULT_DATABASE.$tableName ORDER BY id"),
+            Row(1, "hello") :: Row(2, "fluss") :: Nil
+          )
+
+          sql(s"""
+                 |INSERT INTO $DEFAULT_DATABASE.$tableName VALUES
+                 |(3, "lake"), (4, "union")
+                 |""".stripMargin)
+
+          checkAnswer(
+            sql(s"SELECT * FROM $DEFAULT_DATABASE.$tableName ORDER BY id"),
+            Row(1, "hello") :: Row(2, "fluss") ::
+              Row(3, "lake") :: Row(4, "union") :: Nil
+          )
+        }
       }
-      assert(
-        error.getMessage == "Custom lake table path is not supported for Spark lake reads yet.")
-    }
   }
 
   override protected def flussConf: Configuration = {

diff --git a/...-ut/src/test/scala/org/apache/fluss/spark/lake/SparkLakePrimaryKeyTableReadTestBase.scala b/...-ut/src/test/scala/org/apache/fluss/spark/lake/SparkLakePrimaryKeyTableReadTestBase.scala
@@ -575,4 +575,43 @@ class SparkLakePaimonPrimaryKeyTableReadTest extends SparkLakePrimaryKeyTableRea
     conf.setString("warehouse", warehousePath)
     conf
   }
+
+  test("Spark Lake Read: custom lake path preserves predicate pushdown in union read") {
+    withTable("t_custom_path_union") {
+      sql(s"""
+             |CREATE TABLE $DEFAULT_DATABASE.t_custom_path_union
+             |  (id INT, name STRING, score INT)
+             | TBLPROPERTIES (
+             |  '${ConfigOptions.TABLE_DATALAKE_ENABLED.key()}' = true,
+             |  '${ConfigOptions.TABLE_DATALAKE_DATABASE_NAME.key()}' = 'custom_lake_db',
+             |  '${ConfigOptions.TABLE_DATALAKE_TABLE_NAME.key()}' = 'custom_pk_union',
+             |  '${ConfigOptions.TABLE_DATALAKE_FRESHNESS.key()}' = '1s',
+             |  '${PRIMARY_KEY.key()}' = 'id',
+             |  '${BUCKET_NUMBER.key()}' = 1)
+             |""".stripMargin)
+
+      sql(s"""
+             |INSERT INTO $DEFAULT_DATABASE.t_custom_path_union VALUES
+             |(1, 'alice', 90), (2, 'bob', 85), (3, 'charlie', 95)
+             |""".stripMargin)
+
+      tierToLake("t_custom_path_union")
+
+      sql(s"""
+             |INSERT INTO $DEFAULT_DATABASE.t_custom_path_union VALUES
+             |(4, 'dave', 88), (5, 'eve', 92)
+             |""".stripMargin)
+
+      val query =
+        sql(
+          s"SELECT id, score FROM $DEFAULT_DATABASE.t_custom_path_union " +
+            "WHERE score >= 90 ORDER BY id")
+
+      checkAnswer(
+        query,
+        Row(1, 90) :: Row(3, 95) :: Row(5, 92) :: Nil
+      )
+      assertPushedNames(query, Set(">="))
+    }
+  }
 }