From 14516646f5eef75748cf407ba7f92a7475caee9d Mon Sep 17 00:00:00 2001 From: Pranay Kadu <96111851+BackendArchitectX@users.noreply.github.com> Date: Thu, 3 Sep 2026 12:48:44 +0000 Subject: [PATCH] [spark] Support custom Paimon lake table paths --- .../fluss/spark/read/FlussScanBuilder.scala | 9 +- .../spark/read/lake/FlussLakeUtils.scala | 9 +- .../lake/SparkLakeLogTableReadTest.scala | 85 ++++++++++++++----- ...SparkLakePrimaryKeyTableReadTestBase.scala | 39 +++++++++ 4 files changed, 104 insertions(+), 38 deletions(-) diff --git a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussScanBuilder.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussScanBuilder.scala index 294a35ef208..3ce15d1b06f 100644 --- a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussScanBuilder.scala +++ b/fluss-spark/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/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/lake/FlussLakeUtils.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/lake/FlussLakeUtils.scala index c0c0f38305b..97d6bcb1aaa 100644 --- a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/lake/FlussLakeUtils.scala +++ b/fluss-spark/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/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/SparkLakeLogTableReadTest.scala b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/SparkLakeLogTableReadTest.scala index a3cc752a9cd..1c3de82d64a 100644 --- a/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/SparkLakeLogTableReadTest.scala +++ b/fluss-spark/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/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/SparkLakePrimaryKeyTableReadTestBase.scala b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/SparkLakePrimaryKeyTableReadTestBase.scala index 22958d883da..fb30f3a3499 100644 --- a/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/SparkLakePrimaryKeyTableReadTestBase.scala +++ b/fluss-spark/fluss-spark-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(">=")) + } + } }